Copilot commented on code in PR #65029:
URL: https://github.com/apache/airflow/pull/65029#discussion_r3069909665
##########
airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_task_instances.py:
##########
@@ -72,6 +72,18 @@
DEFAULT_RENDERED_MAP_INDEX = "test rendered map index"
+def _where_column_keys(statement) -> set[str]:
+ return {criterion.left.key for criterion in statement._where_criteria}
+
+
+def _is_task_instance_update(statement) -> bool:
+ return getattr(statement, "is_update", False) and statement.table.name ==
TaskInstance.__table__.name
+
+
+def _is_select_for_update(statement) -> bool:
+ return getattr(statement, "is_select", False) and getattr(statement,
"_for_update_arg", None) is not None
Review Comment:
These helpers rely on SQLAlchemy private/internal statement attributes
(`_where_criteria`, `_for_update_arg`). That makes the tests brittle across
SQLAlchemy upgrades. Consider using a public surface (e.g., inspecting the
statement’s `whereclause` / `get_children()` traversal, or asserting behavior
via DB effects) so the tests don’t depend on SQLAlchemy internals.
```suggestion
whereclause = getattr(statement, "whereclause", None)
if whereclause is None:
return set()
keys: set[str] = set()
stack = [whereclause]
while stack:
clause = stack.pop()
left = getattr(clause, "left", None)
if getattr(left, "key", None) is not None:
keys.add(left.key)
stack.extend(clause.get_children())
return keys
def _is_task_instance_update(statement) -> bool:
return getattr(statement, "is_update", False) and statement.table.name
== TaskInstance.__table__.name
def _is_select_for_update(statement) -> bool:
return getattr(statement, "is_select", False) and "FOR UPDATE" in
str(statement.compile()).upper()
```
--
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]
