Suspend Apache Beam Provider due to grpcio limitation (#61926)

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/"]
@@ -62,6 +63,23 @@
     # Deprecated Operators/Hooks, which replaced by common.sql Operators/Hooks
 )
 
+IGNORE_EXAMPLE_DAGS: tuple[str, ...] = (
+    # These example dags require suspended providers, eg: google dataflow dependent on the Apache Beam provider,
+    # but it's in the suspended list, we can't import the dag
+    "providers/google/tests/system/google/cloud/dataflow/example_dataflow_go.py",
+    "providers/google/tests/system/google/cloud/dataflow/example_dataflow_java_streaming.py",
+    "providers/google/tests/system/google/cloud/dataflow/example_dataflow_native_java.py",
+    "providers/google/tests/system/google/cloud/dataflow/example_dataflow_native_python.py",
+    "providers/google/tests/system/google/cloud/dataflow/example_dataflow_native_python_async.py",
+    "providers/google/tests/system/google/cloud/dataflow/example_dataflow_pipeline.py",
+    "providers/google/tests/system/google/cloud/dataflow/example_dataflow_pipeline_streaming.py",
+    "providers/google/tests/system/google/cloud/dataflow/example_dataflow_sensors_deferrable.py",
+    "providers/google/tests/system/google/cloud/dataflow/example_dataflow_streaming_python.py",
+    "providers/google/tests/system/google/cloud/dataflow/example_dataflow_template.py",
+    "providers/google/tests/system/google/cloud/dataflow/example_dataflow_yaml.py",
+    "providers/google/tests/system/google/cloud/gcs/example_firestore.py",
+)
+
 LONGER_IMPORT_TIMEOUTS: dict[str, float] = {
     "providers/google/tests/system/google/cloud/gen_ai/example_gen_ai_generative_model.py": 60
 }
@@ -79,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
@@ -127,6 +132,13 @@ def example_not_excluded_dags(xfail_db_exception: bool = False):
             if candidate.startswith(tuple(suspended_providers_folders)):
                 param_marks.append(pytest.mark.skip(reason="Suspended provider"))
 
+            if candidate.endswith(IGNORE_EXAMPLE_DAGS):
+                param_marks.append(
+                    pytest.mark.skip(
+                        reason="Excluded from imports, these examples require suspended providers"
+                    )
+                )
+
             if candidate.startswith(tuple(current_python_excluded_providers_folders)):
                 param_marks.append(
                     pytest.mark.skip(reason=f"Not supported for Python {CURRENT_PYTHON_VERSION}")

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

@@ -105,6 +105,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,
@@ -426,8 +430,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):
@@ -457,9 +461,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}"):
@@ -553,6 +560,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

dev/breeze/doc/images/output_build-docs.svg

@@ -1 +1 @@
-9cab0b55ae6a31b4cd43f8edb5d69521
+6f8557a503649e72517ecd140dadf2f6

dev/breeze/doc/images/output_build-docs.txt

@@ -1 +1 @@
-244b18c58ed980a094c555a81698da9b
+c4f3137fd042c7fe7f6cf21479523c5d