Drop the celery provider dependency from airflow-core tests

airflow-core/tests/unit/cli/commands/test_scheduler_command.py

@@ -29,6 +29,7 @@
 from airflow.utils.serve_logs import serve_logs
 
 from tests_common.test_utils.config import conf_vars
+from tests_common.test_utils.markers import skip_if_celery_not_installed
 
 pytestmark = pytest.mark.db_test
 
@@ -41,7 +42,7 @@ def setup_class(cls):
     @pytest.mark.parametrize(
         ("executor", "expect_serve_logs"),
         [
-            ("CeleryExecutor", False),
+            pytest.param("CeleryExecutor", False, marks=skip_if_celery_not_installed),
             ("LocalExecutor", True),
             ("KubernetesExecutor", False),
             ("LocalExecutor,KubernetesExecutor", True),

airflow-core/tests/unit/cli/commands/test_standalone_command.py

@@ -32,11 +32,17 @@
     LOCAL_EXECUTOR,
 )
 
+from tests_common.test_utils.markers import skip_if_celery_not_installed
+
 
 class TestStandaloneCommand:
     @pytest.mark.parametrize(
         "conf_executor_name",
-        [LOCAL_EXECUTOR, CELERY_EXECUTOR, KUBERNETES_EXECUTOR],
+        [
+            LOCAL_EXECUTOR,
+            pytest.param(CELERY_EXECUTOR, marks=skip_if_celery_not_installed),
+            KUBERNETES_EXECUTOR,
+        ],
     )
     def test_calculate_env(self, conf_executor_name):
         """Should always force a local executor compatible with the db."""

airflow-core/tests/unit/cli/conftest.py

@@ -23,7 +23,6 @@
 
 from airflow.dag_processing.dagbag import DagBag
 from airflow.executors import local_executor
-from airflow.providers.celery.executors import celery_executor
 from airflow.providers.cncf.kubernetes.executors import kubernetes_executor
 
 from tests_common.test_utils.config import conf_vars
@@ -32,11 +31,17 @@
     StdoutCaptureManager,
 )
 
+try:
+    from airflow.providers.celery.executors import celery_executor
+except ImportError:
+    celery_executor = None  # type: ignore[assignment]
+
 # Create custom executors here because conftest is imported first
 custom_executor_module = type(sys)("custom_executor")
-custom_executor_module.CustomCeleryExecutor = type(  # type:  ignore
-    "CustomCeleryExecutor", (celery_executor.CeleryExecutor,), {}
-)
+if celery_executor is not None:
+    custom_executor_module.CustomCeleryExecutor = type(  # type:  ignore
+        "CustomCeleryExecutor", (celery_executor.CeleryExecutor,), {}
+    )
 custom_executor_module.CustomLocalExecutor = type(  # type:  ignore
     "CustomLocalExecutor", (local_executor.LocalExecutor,), {}
 )

airflow-core/tests/unit/cli/test_cli_parser.py

@@ -43,6 +43,7 @@
 from airflow.executors import executor_loader
 
 from tests_common.test_utils.config import conf_vars
+from tests_common.test_utils.markers import skip_if_celery_not_installed
 
 pytestmark = pytest.mark.db_test
 
@@ -553,12 +554,16 @@ def test_variables_import_help_message_consistency(self):
     @pytest.mark.parametrize(
         ("executor", "expected_args"),
         [
-            ("CeleryExecutor", ["celery"]),
+            pytest.param("CeleryExecutor", ["celery"], marks=skip_if_celery_not_installed),
             ("KubernetesExecutor", ["kubernetes"]),
             ("LocalExecutor", []),
             # custom executors are mapped to the regular ones in `conftest.py`
             ("custom_executor.CustomLocalExecutor", []),
-            ("custom_executor.CustomCeleryExecutor", ["celery"]),
+            pytest.param(
+                "custom_executor.CustomCeleryExecutor",
+                ["celery"],
+                marks=skip_if_celery_not_installed,
+            ),
             ("custom_executor.CustomKubernetesExecutor", ["kubernetes"]),
         ],
     )

airflow-core/tests/unit/core/test_configuration.py

@@ -57,7 +57,10 @@
     FAKE_CONFIG_BACKEND_PATH,
     FAKE_UNREACHABLE_BACKEND_PATH,
 )
-from tests_common.test_utils.markers import skip_if_force_lowest_dependencies_marker
+from tests_common.test_utils.markers import (
+    skip_if_celery_not_installed,
+    skip_if_force_lowest_dependencies_marker,
+)
 from tests_common.test_utils.reset_warning_registry import reset_warning_registry
 from unit.utils.test_config import (
     remove_all_configurations,
@@ -1908,18 +1911,18 @@ def test_provider_configuration_toggle_with_context_manager():
 
     assert conf._use_providers_configuration is True
     # With providers enabled, the provider value is returned via the fallback lookup chain.
-    assert conf.get("celery", "celery_app_name") == "airflow.providers.celery.executors.celery_executor"
+    assert conf.get("standard", "venv_install_method") == "auto"
 
     with conf.make_sure_configuration_loaded(with_providers=False):
         assert conf._use_providers_configuration is False
         with pytest.raises(
             AirflowConfigException,
-            match=re.escape("section/key [celery/celery_app_name] not found in config"),
+            match=re.escape("section/key [standard/venv_install_method] not found in config"),
         ):
-            conf.get("celery", "celery_app_name")
+            conf.get("standard", "venv_install_method")
     # After the context manager exits, provider config is restored.
     assert conf._use_providers_configuration is True
-    assert conf.get("celery", "celery_app_name") == "airflow.providers.celery.executors.celery_executor"
+    assert conf.get("standard", "venv_install_method") == "auto"
 
 
 @skip_if_force_lowest_dependencies_marker
@@ -1955,14 +1958,34 @@ def test_validate_sqlite3_version(sqlite_version_info, expect_error, monkeypatch
             test_conf._validate_sqlite3_version()
 
 
+# Sections a provider contributes; their metadata is only registered when that provider is installed.
+PROVIDER_SECTION_SKIP_MARKS: dict[str, pytest.MarkDecorator] = {
+    "celery": skip_if_celery_not_installed,
+    "celery_kubernetes_executor": skip_if_celery_not_installed,
+}
+
+
+def build_provider_metadata_params(rows: list[tuple]):
+    """Build parametrize entries that skip rows whose provider is not installed."""
+    return [
+        pytest.param(
+            section,
+            option,
+            *rest,
+            id=f"{section}.{option}",
+            marks=PROVIDER_SECTION_SKIP_MARKS.get(section, ()),
+        )
+        for section, option, *rest in rows
+    ]
+
+
 @skip_if_force_lowest_dependencies_marker
 class TestProviderConfigPriority:
     """Tests that conf.get and conf.has_option respect provider metadata and cfg fallbacks with correct priority."""
 
     @pytest.mark.parametrize(
         ("section", "option", "expected"),
-        PROVIDER_METADATA_CONFIG_OPTIONS,
-        ids=[f"{s}.{o}" for s, o, _ in PROVIDER_METADATA_CONFIG_OPTIONS],
+        build_provider_metadata_params(PROVIDER_METADATA_CONFIG_OPTIONS),
     )
     def test_get_returns_provider_metadata_value(self, section, option, expected):
         """conf.get returns provider metadata (provider.yaml) values."""
@@ -1983,8 +2006,7 @@ def test_cfg_fallback_has_expected_value(self, section, option, expected):
 
     @pytest.mark.parametrize(
         ("section", "option", "expected"),
-        PROVIDER_METADATA_CONFIG_OPTIONS,
-        ids=[f"{s}.{o}" for s, o, _ in PROVIDER_METADATA_CONFIG_OPTIONS],
+        build_provider_metadata_params(PROVIDER_METADATA_CONFIG_OPTIONS),
     )
     def test_has_option_true_for_provider_metadata(self, section, option, expected):
         """conf.has_option returns True for options defined in provider metadata."""
@@ -2011,8 +2033,7 @@ def test_has_option_false_for_nonexistent_option(self):
 
     @pytest.mark.parametrize(
         ("section", "option", "metadata_value", "cfg_value"),
-        PROVIDER_METADATA_OVERRIDES_CFG_FALLBACK,
-        ids=[f"{s}.{o}" for s, o, _, _ in PROVIDER_METADATA_OVERRIDES_CFG_FALLBACK],
+        build_provider_metadata_params(PROVIDER_METADATA_OVERRIDES_CFG_FALLBACK),
     )
     def test_provider_metadata_overrides_cfg_fallback(self, section, option, metadata_value, cfg_value):
         """Provider metadata values take priority over provider_config_fallback_defaults.cfg values."""
@@ -2023,8 +2044,7 @@ def test_provider_metadata_overrides_cfg_fallback(self, section, option, metadat
 
     @pytest.mark.parametrize(
         ("section", "option", "metadata_value", "cfg_value"),
-        PROVIDER_METADATA_OVERRIDES_CFG_FALLBACK,
-        ids=[f"{s}.{o}" for s, o, _, _ in PROVIDER_METADATA_OVERRIDES_CFG_FALLBACK],
+        build_provider_metadata_params(PROVIDER_METADATA_OVERRIDES_CFG_FALLBACK),
     )
     def test_get_default_value_priority(self, section, option, metadata_value, cfg_value):
         """get_default_value checks provider metadata before cfg fallback."""
@@ -2098,6 +2118,7 @@ def test_user_config_overrides_provider_values(self):
         with conf_vars({("celery", "celery_app_name"): custom_value}):
             assert conf.get("celery", "celery_app_name") == custom_value
 
+    @skip_if_celery_not_installed
     def test_getsection_returns_env_var_only_provider_section(self, monkeypatch):
         """Env vars are picked up for a provider section whose keys all default to None."""
         from airflow.settings import conf

airflow-core/tests/unit/executors/test_executor_loader.py

@@ -27,13 +27,13 @@
 from airflow.executors.local_executor import LocalExecutor
 
 from tests_common.test_utils.config import conf_vars
+from tests_common.test_utils.markers import skip_if_celery_not_installed
 
 
 class FakeExecutor:
     pass
 
 
-celery_executor = pytest.importorskip("airflow.providers.celery.executors.celery_executor")
 ecs_executor = pytest.importorskip("airflow.providers.amazon.aws.executors.ecs.ecs_executor")
 
 
@@ -55,7 +55,7 @@ def test_empty_executor_configured(self):
     @pytest.mark.parametrize(
         "executor_name",
         [
-            "CeleryExecutor",
+            pytest.param("CeleryExecutor", marks=skip_if_celery_not_installed),
             "KubernetesExecutor",
             "LocalExecutor",
         ],
@@ -316,6 +316,7 @@ def test_get_hybrid_executors_from_configs(self, executor_config, expected_execu
                 executors = executor_loader.ExecutorLoader._get_executor_names()
                 assert executors == expected_executors_list
 
+    @skip_if_celery_not_installed
     def test_init_executors(self):
         from airflow.providers.celery.executors.celery_executor import CeleryExecutor
 
@@ -363,10 +364,12 @@ def test_get_hybrid_executors_from_config_core_executors_bad_config_format(self,
     @pytest.mark.parametrize(
         ("executor_config", "expected_value"),
         [
-            ("CeleryExecutor", "CeleryExecutor"),
+            pytest.param("CeleryExecutor", "CeleryExecutor", marks=skip_if_celery_not_installed),
             ("KubernetesExecutor", "KubernetesExecutor"),
             ("LocalExecutor", "LocalExecutor"),
-            ("CeleryExecutor, LocalExecutor", "CeleryExecutor"),
+            pytest.param(
+                "CeleryExecutor, LocalExecutor", "CeleryExecutor", marks=skip_if_celery_not_installed
+            ),
             ("LocalExecutor, CeleryExecutor", "LocalExecutor"),
         ],
     )

devel-common/src/tests_common/test_utils/markers.py

@@ -18,6 +18,7 @@
 from __future__ import annotations
 
 import os
+from importlib.util import find_spec
 
 import pytest
 
@@ -31,3 +32,16 @@
     os.environ.get("DEFAULT_BRANCH", "main") != "main",
     reason="This test is only run on main branch in CI",
 )
+
+
+def skip_if_not_installed(module: str) -> pytest.MarkDecorator:
+    """Skip the test when ``module`` cannot be imported."""
+    try:
+        installed = find_spec(module) is not None
+    except ModuleNotFoundError:
+        # find_spec imports the parent packages, so an absent provider raises instead of returning None
+        installed = False
+    return pytest.mark.skipif(not installed, reason=f"{module} is not installed")
+
+
+skip_if_celery_not_installed = skip_if_not_installed("airflow.providers.celery")

airflow-core/tests/integration/cli/commands/test_celery_command.py → providers/celery/tests/integration/celery/cli/test_celery_command.py

@@ -58,8 +58,8 @@ def test_serve_logs_on_worker_start(self):
     @pytest.mark.parametrize(
         ("skip", "expected"),
         [
-            (True, ["bundle_cleanup_main"]),
-            (False, ["serve_logs", "bundle_cleanup_main"]),
+            (True, ["_bundle_cleanup_main"]),
+            (False, ["serve_logs", "_bundle_cleanup_main"]),
         ],
     )
     def test_skip_serve_logs_on_worker_start(self, skip, expected):