This is an automated email from the ASF dual-hosted git repository.
shahar1 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 2e9ae2a1999 Split mixed OpenLineage tests into unit and integration
(#75642)
2e9ae2a1999 is described below
commit 2e9ae2a19997a72a31c73f26e697344b65d5952c
Author: Dev-iL <[email protected]>
AuthorDate: Fri Nov 6 10:16:49 2026 +0200
Split mixed OpenLineage tests into unit and integration (#75642)
The hook-lineage exclusion tests used gs:// assets, so they silently
depended on the google provider registering its asset converter. Where google
is not installed, translation produces no datasets and the tests fail for
reasons unrelated to dataset exclusion. The exclusion logic is now tested with
file:// assets, whose converter comes from common.io, a declared dev
dependency. Translation through real provider converters (gs and s3) is covered
by a separate test that runs only when t [...]
---
.../src/tests_common/test_utils/markers.py | 19 +++++++++
providers/openlineage/pyproject.toml | 1 +
.../unit/openlineage/extractors/test_manager.py | 49 +++++++++++++++-------
uv.lock | 2 +
4 files changed, 57 insertions(+), 14 deletions(-)
diff --git a/devel-common/src/tests_common/test_utils/markers.py
b/devel-common/src/tests_common/test_utils/markers.py
index 2eb91e3faf2..40b1f9bd871 100644
--- a/devel-common/src/tests_common/test_utils/markers.py
+++ b/devel-common/src/tests_common/test_utils/markers.py
@@ -17,6 +17,7 @@
from __future__ import annotations
+import importlib.util
import os
import pytest
@@ -31,3 +32,21 @@ skip_if_not_on_main = pytest.mark.skipif(
os.environ.get("DEFAULT_BRANCH", "main") != "main",
reason="This test is only run on main branch in CI",
)
+
+
+def skip_if_not_installed(*modules: str):
+ """
+ Skip the decorated test while one of the given modules is not installed.
+
+ For integration tests with a distribution the tested one does not depend
on, such as a provider that
+ lowest-dependency runs leave out.
+ """
+ missing = [module for module in modules if not _is_installed(module)]
+ return pytest.mark.skipif(bool(missing), reason=f"{', '.join(missing)} is
not installed")
+
+
+def _is_installed(module: str) -> bool:
+ try:
+ return importlib.util.find_spec(module) is not None
+ except (ImportError, ValueError):
+ return False
diff --git a/providers/openlineage/pyproject.toml
b/providers/openlineage/pyproject.toml
index 8a0310a9571..22cbb5a93b3 100644
--- a/providers/openlineage/pyproject.toml
+++ b/providers/openlineage/pyproject.toml
@@ -82,6 +82,7 @@ dev = [
"apache-airflow-providers-common-sql",
# Additional devel dependencies (do not remove this line and add extra
development dependencies)
"uuid6>=2024.7.10",
+ "apache-airflow-providers-common-io",
"apache-airflow-providers-apache-kafka",
"apache-airflow-providers-google",
"apache-airflow-providers-common-sql[pandas]",
diff --git
a/providers/openlineage/tests/unit/openlineage/extractors/test_manager.py
b/providers/openlineage/tests/unit/openlineage/extractors/test_manager.py
index 7361a7be49d..1bbb8298d76 100644
--- a/providers/openlineage/tests/unit/openlineage/extractors/test_manager.py
+++ b/providers/openlineage/tests/unit/openlineage/extractors/test_manager.py
@@ -46,7 +46,7 @@ from airflow.providers.openlineage.utils.utils import Asset
from airflow.utils.state import State, TaskInstanceState
from tests_common.test_utils.compat import DateTimeSensor, PythonOperator
-from tests_common.test_utils.markers import
skip_if_force_lowest_dependencies_marker
+from tests_common.test_utils.markers import
skip_if_force_lowest_dependencies_marker, skip_if_not_installed
from tests_common.test_utils.version_compat import AIRFLOW_V_3_0_PLUS,
AIRFLOW_V_3_2_PLUS
if TYPE_CHECKING:
@@ -941,32 +941,53 @@ def _hook_exclusion(hook_cls, patterns: list[str]) ->
DatasetFilter:
)
-# gs:// because translating an asset needs the converter of the provider
owning the scheme, and
-# google is a dev dependency of this provider while amazon is not.
-def _collect_gs_parts(collector):
+# file:// because translating an asset needs the converter of the provider
owning the scheme, and
+# common.io is a dev dependency of this provider; other providers' converters
are tested only when installed.
+def _collect_file_parts(collector):
storage_hook, warehouse_hook = _FakeStorageHook(), _FakeWarehouseHook()
- collector.add_output_asset(context=storage_hook,
uri="gs://bucket/staging/part_0")
- collector.add_output_asset(context=storage_hook,
uri="gs://bucket/final/summary")
- collector.add_output_asset(context=warehouse_hook,
uri="gs://bucket/staging/part_1")
+ collector.add_output_asset(context=storage_hook,
uri="file://host/staging/part_0")
+ collector.add_output_asset(context=storage_hook,
uri="file://host/final/summary")
+ collector.add_output_asset(context=warehouse_hook,
uri="file://host/staging/part_1")
def
test_get_hook_lineage_excludes_datasets_per_reporting_hook(hook_lineage_collector):
- _collect_gs_parts(hook_lineage_collector)
+ _collect_file_parts(hook_lineage_collector)
result = ExtractorManager().get_hook_lineage(
- controls=_controls(_hook_exclusion(_FakeStorageHook,
["gs://bucket/staging/.*"]))
+ controls=_controls(_hook_exclusion(_FakeStorageHook,
["file://host/staging/.*"]))
)
assert result == OperatorLineage(
outputs=[
- OpenLineageDataset(namespace="gs://bucket", name="final/summary"),
- OpenLineageDataset(namespace="gs://bucket", name="staging/part_1"),
+ OpenLineageDataset(namespace="file://host", name="/final/summary"),
+ OpenLineageDataset(namespace="file://host",
name="/staging/part_1"),
]
)
[email protected](
+ "scheme",
+ [
+ pytest.param("gs",
marks=skip_if_not_installed("airflow.providers.google"), id="google"),
+ pytest.param("s3",
marks=skip_if_not_installed("airflow.providers.amazon"), id="amazon"),
+ ],
+)
+def
test_get_hook_lineage_excludes_datasets_translated_by_provider_converter(hook_lineage_collector,
scheme):
+ storage_hook = _FakeStorageHook()
+ hook_lineage_collector.add_output_asset(context=storage_hook,
uri=f"{scheme}://bucket/staging/part_0")
+ hook_lineage_collector.add_output_asset(context=storage_hook,
uri=f"{scheme}://bucket/final/summary")
+
+ result = ExtractorManager().get_hook_lineage(
+ controls=_controls(_hook_exclusion(_FakeStorageHook,
[f"{scheme}://bucket/staging/.*"]))
+ )
+
+ assert result == OperatorLineage(
+ outputs=[OpenLineageDataset(namespace=f"{scheme}://bucket",
name="final/summary")]
+ )
+
+
def
test_get_hook_lineage_returns_empty_lineage_when_everything_excluded(hook_lineage_collector):
- _collect_gs_parts(hook_lineage_collector)
+ _collect_file_parts(hook_lineage_collector)
result = ExtractorManager().get_hook_lineage(
controls=_controls(DatasetFilter(authoring_patterns=(".*",)))
@@ -998,7 +1019,7 @@ def
test_extract_metadata_excluded_hook_lineage_does_not_fall_back_to_inlets(hoo
def execute(self, context: Context) -> Any:
pass
- _collect_gs_parts(hook_lineage_collector)
+ _collect_file_parts(hook_lineage_collector)
task = FakeUnsupportedOperator(
task_id="unsupported_task", inlets=[OpenLineageDataset(namespace="ns",
name="inlet")]
)
@@ -1008,7 +1029,7 @@ def
test_extract_metadata_excluded_hook_lineage_does_not_fall_back_to_inlets(hoo
task=task,
task_instance_state=None,
task_instance=MagicMock(),
- controls=_controls(DatasetFilter(authoring_patterns=("gs://.*",))),
+ controls=_controls(DatasetFilter(authoring_patterns=("file://.*",))),
)
assert metadata.inputs == []
diff --git a/uv.lock b/uv.lock
index 4a3aa09d35d..462192969ba 100644
--- a/uv.lock
+++ b/uv.lock
@@ -7329,6 +7329,7 @@ dev = [
{ name = "apache-airflow-devel-common" },
{ name = "apache-airflow-providers-apache-kafka" },
{ name = "apache-airflow-providers-common-compat" },
+ { name = "apache-airflow-providers-common-io" },
{ name = "apache-airflow-providers-common-sql", extra = ["pandas"] },
{ name = "apache-airflow-providers-google" },
{ name = "apache-airflow-providers-openlineage", extra = ["sqlalchemy"] },
@@ -7357,6 +7358,7 @@ dev = [
{ name = "apache-airflow-devel-common", editable = "devel-common" },
{ name = "apache-airflow-providers-apache-kafka", editable =
"providers/apache/kafka" },
{ name = "apache-airflow-providers-common-compat", editable =
"providers/common/compat" },
+ { name = "apache-airflow-providers-common-io", editable =
"providers/common/io" },
{ name = "apache-airflow-providers-common-sql", editable =
"providers/common/sql" },
{ name = "apache-airflow-providers-common-sql", extras = ["pandas"],
editable = "providers/common/sql" },
{ name = "apache-airflow-providers-google", editable = "providers/google"
},
