Pin that a per-flow publish deploys the source it just published
The worst thing moving publish off the whole-pipeline rebuild could do is quietly keep running the old code, and nothing was watching for it. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01StpRc2C6au1WJ1EUU7fsfu
This commit is contained in:
@@ -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
|
assert {name: engine.loaded[name].node for name in ("a.io", "b.io")} == before
|
||||||
|
|
||||||
asyncio.run(asyncio.wait_for(scenario(), timeout=10))
|
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))
|
||||||
|
|||||||
Reference in New Issue
Block a user