kacpermuda commented on code in PR #65727:
URL: https://github.com/apache/airflow/pull/65727#discussion_r3160679463
##########
providers/openlineage/tests/unit/openlineage/extractors/test_manager.py:
##########
@@ -523,3 +523,206 @@ def
test_get_hook_lineage_passes_failed_state(hook_lineage_collector):
sql_extras = mock_sql_fn.call_args.kwargs["sql_extras"]
assert len(sql_extras) == 1
assert
sql_extras[0].value[SqlJobHookLineageExtra.VALUE__SQL_STATEMENT.value] ==
"SELECT 1"
+
+
+def test_extract_inlets_and_outlets_converts_asset_inlet_outlet():
+ """An Airflow ``Asset`` entry goes through ``translate_airflow_asset``
into an OL Dataset."""
+ extractor_manager = ExtractorManager()
+ task = PythonOperator(
+ task_id="task_id",
+ python_callable=lambda x: x,
+ inlets=[Asset(uri="s3://bucket/in.txt", extra={})],
+ outlets=[Asset(uri="s3://bucket/out.txt", extra={})],
+ )
+ lineage = OperatorLineage()
+
+ translated = [
+ OpenLineageDataset(namespace="s3://bucket", name="in.txt"),
+ OpenLineageDataset(namespace="s3://bucket", name="out.txt"),
+ ]
+ with patch(
+
"airflow.providers.openlineage.extractors.manager.translate_airflow_asset",
+ side_effect=translated,
+ ):
+ extractor_manager.extract_inlets_and_outlets(lineage, task)
+
+ assert lineage.inputs == [OpenLineageDataset(namespace="s3://bucket",
name="in.txt")]
+ assert lineage.outputs == [OpenLineageDataset(namespace="s3://bucket",
name="out.txt")]
+
+
+def test_extract_inlets_and_outlets_appends_runtime_outlets():
+ """Runtime outlets (alias resolutions + dynamic events) land on outputs."""
+ extractor_manager = ExtractorManager()
+ task = PythonOperator(task_id="t", python_callable=lambda: 1, outlets=[])
+ lineage = OperatorLineage()
+
+ runtime_assets = [(Asset(uri="s3://bucket/runtime.txt", extra={}), None)]
+ with (
+ patch(
+
"airflow.providers.openlineage.extractors.manager.get_runtime_outlet_assets",
+ return_value=runtime_assets,
+ ),
+ patch(
+
"airflow.providers.openlineage.extractors.manager.translate_airflow_asset",
+ return_value=OpenLineageDataset(namespace="s3://bucket",
name="runtime.txt"),
+ ),
+ ):
+ extractor_manager.extract_inlets_and_outlets(lineage, task,
task_instance=MagicMock())
Review Comment:
These are just minimal tests to check manager logic, and utils functions
just patched here. The utils themselves are tested using `_make_runtime_ti` in
the test_utils.py, so I think it's a nice separation - keeping these manager
tests only about manager logic, with no extra overhead of complex patching.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
