ephraimbuddy commented on code in PR #71621:
URL: https://github.com/apache/airflow/pull/71621#discussion_r3854193594
##########
airflow-core/tests/unit/dag_processing/test_collection.py:
##########
@@ -1437,6 +1438,105 @@ def
test_max_consecutive_failed_dag_runs_defaults_from_conf_when_none(
assert orm_dag.max_consecutive_failed_dag_runs == 6
[email protected]_test
+class TestUpdateImportErrors:
+ """Tests for the ``_update_import_errors`` helper."""
+
+ @pytest.fixture(autouse=True)
+ def clean_import_errors(self):
+ clear_db_import_errors()
+ yield
+ clear_db_import_errors()
+
+ @pytest.fixture
+ def import_error_statements(self, session):
+ """
+ Collect every SQL statement issued against the ``import_error`` table.
+
+ Matching on the bare table name would also catch statements naming
``dag.has_import_errors``,
+ so match the positions where the table itself can appear.
+ """
+ statements: list[str] = []
+
+ def _capture(conn, cursor, statement, parameters, context,
executemany):
+ lowered = statement.lower()
+ if any(f"{keyword} import_error" in lowered for keyword in
("from", "into", "update")):
+ statements.append(statement)
+
+ bind = session.get_bind()
+ event.listen(bind, "before_cursor_execute", _capture)
+ yield statements
+ event.remove(bind, "before_cursor_execute", _capture)
+
+ @staticmethod
+ def _selects(statements: list[str]) -> list[str]:
+ return [stmt for stmt in statements if
stmt.lower().lstrip().startswith("select")]
+
+ def test_no_lookup_when_there_are_no_import_errors(self, session,
import_error_statements):
+ session.add(ParseImportError(filename="broken.py",
bundle_name="testing", stacktrace="boom"))
+ session.flush()
+ import_error_statements.clear()
+
+ # files_parsed is empty so no DELETE runs either: on backends without
DELETE...RETURNING
+ # its synchronize_session fallback would emit a SELECT of its own and
muddy the assertion.
+ _update_import_errors(
+ files_parsed=set(),
+ import_errors={},
+ session=session,
+ )
+
+ assert self._selects(import_error_statements) == []
+
+ @patch.object(ParseImportError, "full_file_path", return_value="broken.py")
+ def test_existing_error_lookup_is_bounded(self, _mock_full_path, session,
import_error_statements):
+ session.add_all(
+ [
+ ParseImportError(filename="broken.py", bundle_name="testing",
stacktrace="old"),
+ ParseImportError(filename="untouched.py", bundle_name="other",
stacktrace="unrelated"),
+ ]
+ )
+ session.flush()
+ import_error_statements.clear()
+
+ _update_import_errors(
+ files_parsed={("testing", "broken.py")},
+ import_errors={("testing", "broken.py"): "new"},
+ session=session,
+ )
+
+ selects = self._selects(import_error_statements)
+ assert selects, "expected the existing-error lookup to run"
+ assert all("where" in stmt.lower() for stmt in selects), (
+ f"import_error must never be scanned unfiltered, got: {selects}"
+ )
+
+ # A list, not a dict: import_error has no unique constraint on
(bundle_name, filename), so
+ # keying by it would silently collapse a duplicate row the lookup
failed to match.
Review Comment:
Yeah, there's no unique constraint but duplicate won't be possible still
with this change except down below where there's a listener lookup. We can
tighten it later in a different PR
--
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]
