Guard provider-dependent tests in airflow-core for standalone core testing

airflow-core/pyproject.toml

@@ -286,15 +286,7 @@ dev = [
     "apache-airflow-ctl",
     "apache-airflow-devel-common",
     "apache-airflow-task-sdk",
-    # TODO(potiuk): eventually we do not want any providers nor apache-airflow extras to be needed for
-    # airflow-core tests
     "apache-airflow[pandas,polars]",
-    "apache-airflow-providers-amazon",
-    "apache-airflow-providers-celery",
-    "apache-airflow-providers-cncf-kubernetes",
-    "apache-airflow-providers-fab>=2.2.0",
-    "apache-airflow-providers-git",
-    "apache-airflow-providers-ftp",
 ]
 
 # To build docs:

airflow-core/tests/integration/cli/commands/test_celery_command.py

@@ -24,6 +24,8 @@
 
 from airflow.cli import cli_parser
 from airflow.executors import executor_loader
+
+pytest.importorskip("airflow.providers.celery", reason="celery provider required")
 from airflow.providers.celery.cli import celery_command
 
 from tests_common.test_utils.config import conf_vars

airflow-core/tests/unit/always/test_example_dags.py

@@ -217,6 +217,10 @@ def test_should_be_importable(example: str, patch_get_dagbag_import_timeout):
             f"Skipping {example} because it requires an optional provider feature that is not installed."
         )
     assert len(dagbag.import_errors) == 0, f"import_errors={str(dagbag.import_errors)}"
+    if len(dagbag.dag_ids) == 0:
+        pytest.skip(
+            f"Skipping {example} because it produced no DAGs (likely due to missing optional dependencies)."
+        )
     assert len(dagbag.dag_ids) >= 1
 
 

airflow-core/tests/unit/always/test_providers_manager.py

@@ -41,6 +41,15 @@
 
 from tests_common.test_utils.markers import skip_if_force_lowest_dependencies_marker
 
+
+def _provider_installed(module_path: str) -> bool:
+    try:
+        __import__(module_path)
+        return True
+    except ImportError:
+        return False
+
+
 if TYPE_CHECKING:
     from unittest.mock import MagicMock
 
@@ -416,6 +425,9 @@ def test_add_customized_fields(self):
         assert behaviour["relabeling"] == {"login": "Email Address"}
         assert behaviour["placeholders"]["host"] == "smtp.gmail.com"
 
+    @pytest.mark.skipif(
+        not _provider_installed("airflow.providers.http"), reason="http provider not installed"
+    )
     def test_load_ui_for_http_provider(self):
         """Test that HTTP provider ui metadata is loaded from provider info."""
         pm = ProvidersManager()
@@ -428,6 +440,9 @@ def test_load_ui_for_http_provider(self):
         assert "relabeling" in behaviour
         assert "placeholders" in behaviour
 
+    @pytest.mark.skipif(
+        not _provider_installed("airflow.providers.http"), reason="http provider not installed"
+    )
     def test_ui_metadata_loading_without_hook_import(self):
         """Test that UI metadata loads from provider info without importing hook classes."""
         with patch("airflow.providers_manager.import_string") as mock_import:

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

@@ -29,7 +29,6 @@
 from airflow.api_fastapi.auth.managers.simple.user import SimpleAuthManagerUser
 from airflow.dag_processing.bundles.manager import DagBundlesManager
 from airflow.models import Connection
-from airflow.providers.git.bundles.git import GitDagBundle
 from airflow.providers.standard.operators.empty import EmptyOperator
 
 from tests_common.test_utils.config import conf_vars
@@ -127,6 +126,9 @@ def create_test_client(apps="all"):
 
 @pytest.fixture
 def configure_git_connection_for_dag_bundle(session):
+    pytest.importorskip("airflow.providers.git.bundles.git")
+    from airflow.providers.git.bundles.git import GitDagBundle
+
     clear_db_connections(False)
     # Git connection is required for the bundles to have a url.
     connection = Connection(

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

@@ -30,6 +30,15 @@
 
 from tests_common.test_utils.config import conf_vars
 
+
+def _provider_installed(module_path: str) -> bool:
+    try:
+        __import__(module_path)
+        return True
+    except ImportError:
+        return False
+
+
 STATSD_CONFIG_BEGIN_WITH = "# `StatsD <https://github.com/statsd/statsd>`"
 
 
@@ -220,6 +229,10 @@ def test_cli_show_changed_defaults_when_overridden_in_env(self, stdout_capture):
         lines = output.splitlines()
         assert any(line.startswith("hostname_callable = test_env") for line in lines if line)
 
+    @pytest.mark.skipif(
+        not _provider_installed("airflow.providers.celery"),
+        reason="celery provider not installed",
+    )
     def test_cli_has_providers(self, stdout_capture):
         with stdout_capture as temp_stdout:
             config_command.show_config(

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

@@ -1060,6 +1060,7 @@ def setup_class(self):
         clear_db_connections()
 
     def test_cli_connections_test_success(self, mocker, stdout_capture):
+        pytest.importorskip("airflow.providers.http")
         mocker.patch.dict(os.environ, {"AIRFLOW__CORE__TEST_CONNECTION": "Enabled"})
         mock_test_conn = mocker.patch("airflow.providers.http.hooks.http.HttpHook.test_connection")
         conn_id = "http_default"
@@ -1069,6 +1070,7 @@ def test_cli_connections_test_success(self, mocker, stdout_capture):
             assert "Connection success!" in stdout.getvalue()
 
     def test_cli_connections_test_fail(self, mocker, stdout_capture):
+        pytest.importorskip("airflow.providers.http")
         mocker.patch.dict(os.environ, {"AIRFLOW__CORE__TEST_CONNECTION": "Enabled"})
         mock_test_conn = mocker.patch("airflow.providers.http.hooks.http.HttpHook.test_connection")
         conn_id = "http_default"

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

@@ -55,6 +55,7 @@
     clear_db_runs,
     parse_and_sync_to_db,
 )
+from tests_common.test_utils.markers import skip_if_force_lowest_dependencies_marker
 from unit.models import TEST_DAGS_FOLDER
 
 DEFAULT_DATE = timezone.make_aware(datetime(2015, 1, 1), timezone=timezone.utc)
@@ -491,6 +492,7 @@ def test_pause(self):
         dag_command.dag_unpause(args)
         assert not DagModel.get_dagmodel("example_bash_operator").is_paused
 
+    @skip_if_force_lowest_dependencies_marker
     @mock.patch("airflow.cli.commands.dag_command.ask_yesno")
     def test_pause_regex(self, mock_yesno):
         args = self.parser.parse_args(["dags", "pause", "^example_.*$", "--treat-dag-id-as-regex"])
@@ -506,6 +508,7 @@ def test_pause_regex(self, mock_yesno):
         assert not DagModel.get_dagmodel("example_kubernetes_executor").is_paused
         assert not DagModel.get_dagmodel("example_xcom_args").is_paused
 
+    @skip_if_force_lowest_dependencies_marker
     @mock.patch("airflow.cli.commands.dag_command.ask_yesno")
     def test_pause_regex_operation_cancelled(self, ask_yesno, capsys):
         args = self.parser.parse_args(["dags", "pause", "example_bash_operator", "--treat-dag-id-as-regex"])

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

@@ -34,6 +34,14 @@
 from tests_common.test_utils.config import conf_vars
 
 
+def _provider_installed(module_path: str) -> bool:
+    try:
+        __import__(module_path)
+        return True
+    except ImportError:
+        return False
+
+
 def capture_show_output(instance):
     console = Console()
     with console.capture() as capture:
@@ -90,6 +98,10 @@ def teardown_class(cls) -> None:
     def unique_items(items):
         return {i[0] for i in items}
 
+    @pytest.mark.skipif(
+        not _provider_installed("airflow.providers.amazon"),
+        reason="amazon provider not installed",
+    )
     @conf_vars(
         {
             ("core", "executor"): "TEST_EXECUTOR",

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

@@ -30,6 +30,15 @@
 
 from tests_common.test_utils.config import conf_vars
 
+
+def _provider_installed(module_path: str) -> bool:
+    try:
+        __import__(module_path)
+        return True
+    except ImportError:
+        return False
+
+
 pytestmark = pytest.mark.db_test
 
 
@@ -41,9 +50,23 @@ def setup_class(cls):
     @pytest.mark.parametrize(
         ("executor", "expect_serve_logs"),
         [
-            ("CeleryExecutor", False),
+            pytest.param(
+                "CeleryExecutor",
+                False,
+                marks=pytest.mark.skipif(
+                    not _provider_installed("airflow.providers.celery"),
+                    reason="celery provider not installed",
+                ),
+            ),
             ("LocalExecutor", True),
-            ("KubernetesExecutor", False),
+            pytest.param(
+                "KubernetesExecutor",
+                False,
+                marks=pytest.mark.skipif(
+                    not _provider_installed("airflow.providers.cncf.kubernetes"),
+                    reason="cncf.kubernetes provider not installed",
+                ),
+            ),
         ],
     )
     @mock.patch("airflow.cli.commands.scheduler_command.SchedulerJobRunner")

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

@@ -31,10 +31,34 @@
 )
 
 
+def _provider_installed(module_path: str) -> bool:
+    try:
+        __import__(module_path)
+        return True
+    except ImportError:
+        return False
+
+
 class TestStandaloneCommand:
     @pytest.mark.parametrize(
         "conf_executor_name",
-        [LOCAL_EXECUTOR, CELERY_EXECUTOR, KUBERNETES_EXECUTOR],
+        [
+            LOCAL_EXECUTOR,
+            pytest.param(
+                CELERY_EXECUTOR,
+                marks=pytest.mark.skipif(
+                    not _provider_installed("airflow.providers.celery"),
+                    reason="celery provider not installed",
+                ),
+            ),
+            pytest.param(
+                KUBERNETES_EXECUTOR,
+                marks=pytest.mark.skipif(
+                    not _provider_installed("airflow.providers.cncf.kubernetes"),
+                    reason="cncf.kubernetes provider not installed",
+                ),
+            ),
+        ],
     )
     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,8 +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
 from tests_common.test_utils.stream_capture_manager import (
@@ -36,15 +34,25 @@
 
 # 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,), {}
-)
 custom_executor_module.CustomLocalExecutor = type(  # type:  ignore
     "CustomLocalExecutor", (local_executor.LocalExecutor,), {}
 )
-custom_executor_module.CustomKubernetesExecutor = type(  # type:  ignore
-    "CustomKubernetesExecutor", (kubernetes_executor.KubernetesExecutor,), {}
-)
+try:
+    from airflow.providers.celery.executors import celery_executor
+
+    custom_executor_module.CustomCeleryExecutor = type(  # type:  ignore
+        "CustomCeleryExecutor", (celery_executor.CeleryExecutor,), {}
+    )
+except ImportError:
+    pass
+try:
+    from airflow.providers.cncf.kubernetes.executors import kubernetes_executor
+
+    custom_executor_module.CustomKubernetesExecutor = type(  # type:  ignore
+        "CustomKubernetesExecutor", (kubernetes_executor.KubernetesExecutor,), {}
+    )
+except ImportError:
+    pass
 sys.modules["custom_executor"] = custom_executor_module
 
 

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

@@ -44,6 +44,18 @@
 
 from tests_common.test_utils.config import conf_vars
 
+
+def _provider_installed(module_path: str) -> bool:
+    try:
+        __import__(module_path)
+        return True
+    except ImportError:
+        return False
+
+
+_celery_installed = _provider_installed("airflow.providers.celery")
+_kubernetes_installed = _provider_installed("airflow.providers.cncf.kubernetes")
+
 pytestmark = pytest.mark.db_test
 
 # Can not be `--snake_case` or contain uppercase letter
@@ -533,13 +545,33 @@ def test_variables_import_help_message_consistency(self):
     @pytest.mark.parametrize(
         ("executor", "expected_args"),
         [
-            ("CeleryExecutor", ["celery"]),
-            ("KubernetesExecutor", ["kubernetes"]),
+            pytest.param(
+                "CeleryExecutor",
+                ["celery"],
+                marks=pytest.mark.skipif(not _celery_installed, reason="celery provider not installed"),
+            ),
+            pytest.param(
+                "KubernetesExecutor",
+                ["kubernetes"],
+                marks=pytest.mark.skipif(
+                    not _kubernetes_installed, reason="cncf.kubernetes provider not installed"
+                ),
+            ),
             ("LocalExecutor", []),
             # custom executors are mapped to the regular ones in `conftest.py`
             ("custom_executor.CustomLocalExecutor", []),
-            ("custom_executor.CustomCeleryExecutor", ["celery"]),
-            ("custom_executor.CustomKubernetesExecutor", ["kubernetes"]),
+            pytest.param(
+                "custom_executor.CustomCeleryExecutor",
+                ["celery"],
+                marks=pytest.mark.skipif(not _celery_installed, reason="celery provider not installed"),
+            ),
+            pytest.param(
+                "custom_executor.CustomKubernetesExecutor",
+                ["kubernetes"],
+                marks=pytest.mark.skipif(
+                    not _kubernetes_installed, reason="cncf.kubernetes provider not installed"
+                ),
+            ),
         ],
     )
     def test_cli_parser_executors(self, executor, expected_args):
@@ -637,6 +669,7 @@ def test_cli_run_time(self):
         # Minimum run time of Airflow CLI should at least be within 5s
         assert timing_result < threshold
 
+    @pytest.mark.skipif(not _celery_installed, reason="celery provider not installed")
     def test_airflow_config_contains_providers(self):
         """
         Test that airflow config has providers included by default.
@@ -655,6 +688,7 @@ def test_airflow_config_contains_providers(self):
         assert CONFIG_FILE.exists()
         assert "celery_config_options" in CONFIG_FILE.read_text()
 
+    @pytest.mark.skipif(not _celery_installed, reason="celery provider not installed")
     def test_airflow_config_output_contains_providers_by_default(self):
         """Test that airflow config has providers excluded in config list when asked for it."""
         CONFIG_FILE.parent.mkdir(parents=True, exist_ok=True)