diff --git a/backend/tests/flow/test_runtime_control.py b/backend/tests/flow/test_runtime_control.py index 680dde4..a6fa4aa 100644 --- a/backend/tests/flow/test_runtime_control.py +++ b/backend/tests/flow/test_runtime_control.py @@ -332,3 +332,36 @@ def test_stopping_one_flow_does_not_touch_the_others( assert {name: engine.loaded[name].node for name in ("a.io", "b.io")} == before asyncio.run(asyncio.wait_for(scenario(), timeout=10)) + + +NEW_SOURCE = "def process():\n return {'reading': 2.0}\n" + + +def test_a_per_flow_rebuild_runs_the_source_that_was_just_published(tmp_path: Path): + """The one thing moving publish off the full rebuild must not break.""" + store = FlowStore(tmp_path / "flows") + store.write_flow( + FlowDef( + name="probe", + nodes=[NodeDef(id="sensor", type="python", provides=[spec("reading")])], + ) + ) + store.write_node_source( + "probe", "sensor", "def process():\n return {'reading': 1.0}\n", draft=False + ) + engine = FlowController(store) + + async def scenario() -> None: + await engine.reload() + engine.run_flow("probe") + assert engine.pipeline is not None + assert engine.pipeline.state["probe.reading"] == 1.0 + + store.write_node_source("probe", "sensor", NEW_SOURCE, draft=True) + store.publish_flow("probe", store.read_flow("probe", draft=True).version) + await engine.reload_flow("probe") + + engine.run_flow("probe") + assert engine.pipeline.state["probe.reading"] == 2.0 + + asyncio.run(asyncio.wait_for(scenario(), timeout=20))