Skip to content

Commit eee7d4f

Browse files
committed
Check import errors only in each mode's Lang-SDK Dag files
The readiness gate and the import-error checks looked at every file of the Dags folder, so a stock example Dag that cannot import in the image (example_event_driven.py without the Kafka provider) would time out the gate and fail the checks in all three Lang-SDK modes. They now check only the mode's own Lang-SDK Dag files.
1 parent 9ba69ef commit eee7d4f

6 files changed

Lines changed: 25 additions & 16 deletions

File tree

‎airflow-e2e-tests/tests/airflow_e2e_tests/conftest.py‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1011,6 +1011,12 @@ def airflow_dags_path():
10111011
return _E2ETestState.airflow_dags_path
10121012

10131013

1014+
@pytest.fixture(scope="session")
1015+
def lang_sdk_dag_files():
1016+
"""The ``dags-folder`` files of the Lang SDK that the stack was given, in the Lang-SDK modes."""
1017+
return _E2ETestState.lang_sdk_dag_files
1018+
1019+
10141020
@pytest.fixture(scope="session")
10151021
def go_dynamic_dag_ids():
10161022
"""The Dag ids that ``go_test_dags.py`` generates from ``E2E_GO_DYNAMIC_DAG_IDS``, in the go_sdk mode."""

‎airflow-e2e-tests/tests/airflow_e2e_tests/e2e_test_utils/lang_sdk.py‎

Lines changed: 6 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -106,11 +106,12 @@ def get_task_handler_artifacts(compose: DockerCompose) -> dict[ArtifactRef, str]
106106
return {ArtifactRef(bundle_name, rel_path): probed_at for bundle_name, rel_path, probed_at in rows}
107107

108108

109-
def get_import_errors(client: AirflowClient) -> dict[str, str]:
110-
"""Return the text of the import error of each ``dags-folder`` file that has one, by file."""
109+
def get_import_errors(client: AirflowClient, files: Collection[str]) -> dict[str, str]:
110+
"""Return the text of the import error of each of the ``dags-folder`` *files* that has one, by file."""
111111
return {
112112
error["filename"]: error["stack_trace"]
113113
for error in client.list_import_errors(bundle_name=DAGS_BUNDLE_NAME)
114+
if error["filename"] in files
114115
}
115116

116117

@@ -148,7 +149,8 @@ def wait_until_stub_tasks_are_bound(
148149
149150
:param queues: The queues that the Dag processor routes to a Lang-SDK coordinator.
150151
:param dag_files: The ``dags-folder`` files of the Lang SDK that the stack was given. Until the Dag
151-
processor has parsed them, no stub task is there to be unbound.
152+
processor has parsed them, no stub task is there to be unbound. Only their import errors are checked:
153+
another file of the Dags folder, such as a stock example Dag, may fail to import for other reasons.
152154
:param expected_import_errors: The ``dags-folder`` files that fail to import by design.
153155
:raises TimeoutError: when a Dag file is still not parsed, a stub task is still unbound or an import error
154156
is still missing or unexpected after *timeout* seconds, with what is missing and the text of the
@@ -159,7 +161,7 @@ def wait_until_stub_tasks_are_bound(
159161
while True:
160162
bindings = get_task_handler_bindings(compose)
161163
unbound = sorted(task for task in get_routed_stub_tasks(client, queues) if task not in bindings)
162-
import_errors = get_import_errors(client)
164+
import_errors = get_import_errors(client, dag_files)
163165
missing = sorted(set(expected_import_errors) - set(import_errors))
164166
unexpected = {
165167
name: text for name, text in import_errors.items() if name not in expected_import_errors

‎airflow-e2e-tests/tests/airflow_e2e_tests/go_sdk_tests/test_go_sdk_task_handler_parse.py‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -146,7 +146,7 @@ def test_problems_are_one_import_error_of_their_dag_file(client: AirflowClient,
146146
The error is the one of the file, and lists each problem on a line of its own. Its Dag is serialized but
147147
marked as having import errors, and none of its stub tasks is bound.
148148
"""
149-
assert get_import_errors(client).get(_FAILURE_FILE) == _FAILURE_IMPORT_ERROR
149+
assert get_import_errors(client, [_FAILURE_FILE]).get(_FAILURE_FILE) == _FAILURE_IMPORT_ERROR
150150

151151
failing_dags = client.list_dags(bundle_name=DAGS_BUNDLE_NAME, exclude_stale=False, has_import_errors=True)
152152
assert [dag["dag_id"] for dag in failing_dags] == ["go_task_handler_failures"]
@@ -191,6 +191,6 @@ def test_named_mismatches_are_warnings_in_the_parse_log(airflow_logs_path):
191191
)
192192

193193

194-
def test_only_the_failing_dag_file_has_an_import_error(client: AirflowClient):
195-
"""No other Dag file of the Dags folder fails to import, so no fixture hides behind the expected error."""
196-
assert set(get_import_errors(client)) == {_FAILURE_FILE}
194+
def test_only_the_failing_dag_file_has_an_import_error(client: AirflowClient, lang_sdk_dag_files):
195+
"""No other Lang-SDK Dag file fails to import, so no fixture hides behind the expected error."""
196+
assert set(get_import_errors(client, lang_sdk_dag_files)) == {_FAILURE_FILE}

‎airflow-e2e-tests/tests/airflow_e2e_tests/java_sdk_tests/test_java_sdk_task_handler_parse.py‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -100,6 +100,6 @@ def test_a_later_parse_probes_nothing(client: AirflowClient, compose_instance, a
100100
)
101101

102102

103-
def test_no_dag_file_has_an_import_error(client: AirflowClient):
104-
"""Every Dag file of the Dags folder imports: each stub task has exactly one handler that fits it."""
105-
assert get_import_errors(client) == {}
103+
def test_no_dag_file_has_an_import_error(client: AirflowClient, lang_sdk_dag_files):
104+
"""Every Lang-SDK Dag file imports: each stub task has exactly one handler that fits it."""
105+
assert get_import_errors(client, lang_sdk_dag_files) == {}

‎airflow-e2e-tests/tests/airflow_e2e_tests/ts_sdk_tests/test_ts_sdk_task_handler_parse.py‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -74,6 +74,6 @@ def test_a_later_parse_probes_nothing(client: AirflowClient, compose_instance, a
7474
)
7575

7676

77-
def test_no_dag_file_has_an_import_error(client: AirflowClient):
78-
"""Every Dag file of the Dags folder imports: each stub task has exactly one handler that fits it."""
79-
assert get_import_errors(client) == {}
77+
def test_no_dag_file_has_an_import_error(client: AirflowClient, lang_sdk_dag_files):
78+
"""Every Lang-SDK Dag file imports: each stub task has exactly one handler that fits it."""
79+
assert get_import_errors(client, lang_sdk_dag_files) == {}

‎contributing-docs/testing/airflow_e2e_tests.rst‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -64,8 +64,9 @@ SDK. Run one with ``breeze testing airflow-e2e-tests --e2e-test-mode go_sdk``. E
6464
toolchain container, or with the host toolchain when ``LANG_SDK_NATIVE_TOOLCHAIN=true`` is set, as CI does.
6565

6666
Before the first test, the tests wait until the Dag processor has bound each stub task on a queue it routes to the
67-
artifact that registers its task handler, and until the Dag files that fail to import are exactly the ones that are
68-
meant to. A Dag file with an import error is left out of the binding check. If that does not happen in time, the run
67+
artifact that registers its task handler, and until the mode's Language SDK Dag files that fail to import are exactly
68+
the ones that are meant to. Import errors in other files of the Dags folder, such as the stock example Dags, are not
69+
checked. A Dag file with an import error is left out of the binding check. If that does not happen in time, the run
6970
fails with the stub tasks that have no binding and the text of any import error nobody expected.
7071

7172
A Dag file that must fail goes in a test bundle, not in a user-facing example, which the tests pin. The Go test

0 commit comments

Comments
 (0)