diff --git a/airflow-core/tests/integration/cli/commands/__init__.py b/airflow-core/tests/integration/cli/commands/__init__.py deleted file mode 100644 index 13a83393a9124..0000000000000 --- a/airflow-core/tests/integration/cli/commands/__init__.py +++ /dev/null @@ -1,16 +0,0 @@ -# Licensed to the Apache Software Foundation (ASF) under one -# or more contributor license agreements. See the NOTICE file -# distributed with this work for additional information -# regarding copyright ownership. The ASF licenses this file -# to you under the Apache License, Version 2.0 (the -# "License"); you may not use this file except in compliance -# with the License. You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, -# software distributed under the License is distributed on an -# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY -# KIND, either express or implied. See the License for the -# specific language governing permissions and limitations -# under the License. diff --git a/airflow-core/tests/unit/cli/commands/test_scheduler_command.py b/airflow-core/tests/unit/cli/commands/test_scheduler_command.py index 1089edc12e89f..4333f370bdf09 100644 --- a/airflow-core/tests/unit/cli/commands/test_scheduler_command.py +++ b/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), diff --git a/airflow-core/tests/unit/cli/commands/test_standalone_command.py b/airflow-core/tests/unit/cli/commands/test_standalone_command.py index 993f51ca2d72d..0df5f4f37388e 100644 --- a/airflow-core/tests/unit/cli/commands/test_standalone_command.py +++ b/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.""" diff --git a/airflow-core/tests/unit/cli/conftest.py b/airflow-core/tests/unit/cli/conftest.py index 7676a103b5363..e011a660250d1 100644 --- a/airflow-core/tests/unit/cli/conftest.py +++ b/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,), {} ) diff --git a/airflow-core/tests/unit/cli/test_cli_parser.py b/airflow-core/tests/unit/cli/test_cli_parser.py index 2a0680f08ef00..c00f1cc67ad22 100644 --- a/airflow-core/tests/unit/cli/test_cli_parser.py +++ b/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"]), ], ) diff --git a/airflow-core/tests/unit/core/test_configuration.py b/airflow-core/tests/unit/core/test_configuration.py index e173215234ac4..9fbf9b8815cfa 100644 --- a/airflow-core/tests/unit/core/test_configuration.py +++ b/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 diff --git a/airflow-core/tests/unit/executors/test_executor_loader.py b/airflow-core/tests/unit/executors/test_executor_loader.py index c0c0a2f6331aa..7fcfa593ddf4f 100644 --- a/airflow-core/tests/unit/executors/test_executor_loader.py +++ b/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"), ], ) diff --git a/devel-common/src/tests_common/test_utils/markers.py b/devel-common/src/tests_common/test_utils/markers.py index 2eb91e3faf202..ba8d11deccbab 100644 --- a/devel-common/src/tests_common/test_utils/markers.py +++ b/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") diff --git a/airflow-core/tests/integration/cli/__init__.py b/providers/celery/tests/integration/celery/cli/__init__.py similarity index 100% rename from airflow-core/tests/integration/cli/__init__.py rename to providers/celery/tests/integration/celery/cli/__init__.py diff --git a/airflow-core/tests/integration/cli/commands/test_celery_command.py b/providers/celery/tests/integration/celery/cli/test_celery_command.py similarity index 96% rename from airflow-core/tests/integration/cli/commands/test_celery_command.py rename to providers/celery/tests/integration/celery/cli/test_celery_command.py index 20cc664f997e5..e82637e35ffea 100644 --- a/airflow-core/tests/integration/cli/commands/test_celery_command.py +++ b/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):