uranusjr commented on code in PR #70972:
URL: https://github.com/apache/airflow/pull/70972#discussion_r3711639219
##########
airflow-core/tests/unit/assets/test_manager.py:
##########
@@ -212,23 +212,43 @@ def test_register_asset_change_no_downstreams(self,
session, mock_task_instance)
)
assert
session.scalar(select(func.count()).select_from(AssetDagRunQueue)) == 0
- @pytest.mark.parametrize(
- ("dialect_name", "expected_helper"),
- [
- ("postgresql", "_queue_dagruns_nonpartitioned_conflict_update"),
- ("mysql", "_queue_dagruns_nonpartitioned_mysql"),
- ("sqlite", "_queue_dagruns_nonpartitioned_conflict_update"),
- ],
- )
- def test_queue_dagruns_routes_by_dialect(self, dialect_name,
expected_helper):
- """Test that _queue_dagruns routes to the dialect-appropriate queue
helper."""
+ def test_register_asset_change_is_atomic_on_caller_session(self, session,
mock_task_instance):
+ """The AssetEvent is written on the caller's session: visible before
commit, gone after rollback.
+
+ Under the old side-session behaviour the event was committed
independently and would be
+ orphaned if the caller's transaction rolled back. Registration is now
atomic with the caller.
+ """
+ asset_manager = AssetManager()
+
+ asset = Asset(uri="test://atomic1", name="atomic_asset")
+ asm = AssetModel(uri="test://atomic1/", name="atomic_asset",
group="asset")
+ session.add(asm)
+ session.execute(delete(AssetDagRunQueue))
Review Comment:
Reasonable. I’ll change this.
--
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]
