Skip to content

Commit 81a80be

Browse files
committed
fixup! Run loop gates and advance iterations atomically
From this layer a Dag run creates its loop's region, and the mapped region of a pass, when it creates the pass. The edge3 log tests inserted those regions by hand, so from here they collided on slot_key and failed. They now take the run's regions through get_or_create, which still creates the regions a run has not made.
1 parent 3f16d57 commit 81a80be

1 file changed

Lines changed: 8 additions & 6 deletions

File tree

  • providers/edge3/tests/unit/edge3/worker_api/routes

‎providers/edge3/tests/unit/edge3/worker_api/routes/test_logs.py‎

Lines changed: 8 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -75,8 +75,9 @@ def body():
7575
create_loop(body, max_iterations=3)
7676
dr = dag_maker.create_dagrun(run_id="loop_run")
7777
ti = next(ti for ti in dr.task_instances if ti.task_id == "body.work")
78-
region = DynamicRegion(dag_id=dr.dag_id, run_id=dr.run_id, node_id="body")
79-
session.add(region)
78+
region = DynamicRegion.get_or_create(
79+
dag_id=dr.dag_id, run_id=dr.run_id, node_id="body", session=session
80+
)
8081
session.flush()
8182
ti.region_id, ti.region_index = region.id, 2
8283
session.add(
@@ -118,19 +119,20 @@ def body():
118119
create_loop(body, max_iterations=3)
119120
dr = dag_maker.create_dagrun(run_id="mapped_loop_run")
120121
first = next(ti for ti in dr.task_instances if ti.task_id == "body.mapped")
121-
loop_region = DynamicRegion(dag_id=dr.dag_id, run_id=dr.run_id, node_id="body")
122-
session.add(loop_region)
122+
loop_region = DynamicRegion.get_or_create(
123+
dag_id=dr.dag_id, run_id=dr.run_id, node_id="body", session=session
124+
)
123125
session.flush()
124126
tis = []
125127
for iteration in range(2):
126-
region = DynamicRegion(
128+
region = DynamicRegion.get_or_create(
127129
dag_id=dr.dag_id,
128130
run_id=dr.run_id,
129131
node_id="body.mapped",
130132
parent_region_id=loop_region.id,
131133
parent_region_index=iteration,
134+
session=session,
132135
)
133-
session.add(region)
134136
session.flush()
135137
ti = (
136138
first

0 commit comments

Comments
 (0)