Fix serialization tests

Author: gopidesupavan

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

@@ -36,6 +36,7 @@
 from tests_common.test_utils.asserts import assert_queries_count
 from tests_common.test_utils.markers import skip_if_force_lowest_dependencies_marker
 from tests_common.test_utils.paths import AIRFLOW_PROVIDERS_ROOT_PATH, AIRFLOW_ROOT_PATH
+from tests_common.test_utils.providers import get_suspended_providers_folders
 
 CURRENT_PYTHON_VERSION = f"{sys.version_info.major}.{sys.version_info.minor}"
 PROVIDERS_PREFIXES = ["providers/"]
@@ -96,19 +97,6 @@ def match_optional_dependencies(distribution_name: str, specifier: str | None) -
     return True, ""
 
 
-def get_suspended_providers_folders() -> list[str]:
-    """
-    Returns a list of suspended providers folders that should be
-    skipped when running tests (without any prefix - for example apache/beam, yandex, google etc.).
-    """
-    suspended_providers = []
-    for provider_path in AIRFLOW_PROVIDERS_ROOT_PATH.rglob("provider.yaml"):
-        provider_yaml = yaml.safe_load(provider_path.read_text())
-        if provider_yaml["state"] == "suspended":
-            suspended_providers.append(provider_path.parent.resolve().as_posix())
-    return suspended_providers
-
-
 def get_python_excluded_providers_folders() -> list[str]:
     """
     Returns a list of providers folders that should be excluded for current Python version and

airflow-core/tests/unit/serialization/test_dag_serialization.py

@@ -104,6 +104,10 @@
     GithubLink,
     MockOperator,
 )
+from tests_common.test_utils.providers import (
+    IGNORE_MODULE_IMPORT_ERRORS,
+    get_suspended_providers_folders,
+)
 from tests_common.test_utils.timetables import (
     CustomSerializationTimetable,
     cron_timetable,
@@ -424,8 +428,8 @@ def get_excluded_patterns() -> Generator[str, None, None]:
         (AIRFLOW_REPO_ROOT_PATH / "generated" / "provider_dependencies.json").read_text()
     )
     for provider, provider_info in all_providers.items():
+        provider_path = provider.replace(".", "/")
         if python_version in provider_info.get("excluded-python-versions"):
-            provider_path = provider.replace(".", "/")
             yield f"providers/{provider_path}"
     current_python_version = sys.version_info[:2]
     if current_python_version >= (3, 13):
@@ -455,9 +459,12 @@ def collect_dags(dag_folder=None):
             patterns = dag_folder
         else:
             patterns = [dag_folder]
+    suspended_providers_path = get_suspended_providers_folders()
+
     excluded_patterns = [
         f"{AIRFLOW_REPO_ROOT_PATH}/{excluded_pattern}" for excluded_pattern in get_excluded_patterns()
-    ]
+    ] + suspended_providers_path
+
     with mock.patch("airflow.dag_processing.dagbag.settings.get_dagbag_import_timeout", return_value=60):
         for pattern in patterns:
             for directory in glob(f"{AIRFLOW_REPO_ROOT_PATH}/{pattern}"):
@@ -551,6 +558,11 @@ def test_serialization(self):
             # This "looks" like a problem, but is just a quirk of the parse-all-dags-in-one-process we do
             # in this test
             if "AirflowDagDuplicatedIdException: Ignoring DAG example_sagemaker" not in error
+            # Ignore module import errors for any suspended provider paths used in example dags
+            if any(
+                f"{ignore_module_import_error}" not in error
+                for ignore_module_import_error in IGNORE_MODULE_IMPORT_ERRORS
+            )
         }
 
         # Let's not be exact about this, but if everything fails to parse we should fail this test too

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

@@ -18,6 +18,9 @@
 from __future__ import annotations
 
 import semver
+import yaml
+
+from tests_common.test_utils.paths import AIRFLOW_PROVIDERS_ROOT_PATH
 
 
 def object_exists(path: str):
@@ -62,3 +65,20 @@ def get_provider_min_airflow_version(provider_name: str) -> tuple[int, ...]:
             f"The provider should have `apache-airflow>=` in their dependencies: {provider_name}"
         )
     return Version(airflow_dep.split(">=")[1]).release
+
+
+# Ignore module import errors for any suspended provider paths used in example dags
+IGNORE_MODULE_IMPORT_ERRORS: list[str] = ["airflow.providers.apache.beam"]
+
+
+def get_suspended_providers_folders() -> list[str]:
+    """
+    Returns a list of suspended providers folders that should be
+    skipped when running tests (without any prefix - for example apache/beam, yandex, google etc.).
+    """
+    suspended_providers = []
+    for provider_path in AIRFLOW_PROVIDERS_ROOT_PATH.rglob("provider.yaml"):
+        provider_yaml = yaml.safe_load(provider_path.read_text())
+        if provider_yaml["state"] == "suspended":
+            suspended_providers.append(provider_path.parent.resolve().as_posix())
+    return suspended_providers