A batch run skips the nodes waiting on a failed one and ends error; a run whose nodes cannot load fails with the load error before anything executes; cancelling a run stops its pipeline without announcing a pause of the live flow; batch_issues refuses a delay node's interval; the client accepts an engine URL with or without /api/v1; RedisState's compare-and-swap records the names it writes

backend/tests/api/routes/test_runs.py

@@ -19,7 +19,7 @@ from fluksio.core.db import engine as db_engine
 from fluksio.flow.artifacts import ArtifactStore
 from fluksio.flow.controller import FlowController
 from fluksio.flow.messages import DType, MessageSpec
-from fluksio.flow.pipeline import CACHE_SHAPE_SINCE, NodeOutcome
+from fluksio.flow.pipeline import CACHE_SHAPE_SINCE, NodeOutcome, Pipeline
 from fluksio.flow.runs import (
     LEASE_STALE_S,
     OUTPUT_CAP,
@@ -729,6 +729,57 @@ def test_a_draft_run_executes_the_draft(tmp_path):
             session.commit()
 
 
+def test_a_failed_stage_skips_what_waits_on_it_and_fails_the_run(tmp_path):
+    """Stages after a failure were called without their input and raised
+    TypeError, and a run whose only node could not load finished `ok`."""
+    model, score = (MessageSpec(name=n, dtype=DType.FLOAT) for n in ("model", "score"))
+    staged = FlowDef(
+        name="staged",
+        mode="batch",
+        nodes=[
+            NodeDef(id="fit", provides=[model]),
+            NodeDef(id="readout", requires=[model], provides=[score]),
+            NodeDef(id="evaluate", requires=[score]),
+        ],
+    )
+    broken = FlowDef(name="broken", mode="batch", nodes=[NodeDef(id="fit")])
+    store = FlowStore(tmp_path / "flows")
+    store.write_flow(staged)
+    store.write_flow(broken)
+    store.write_node_source("staged", "fit", "def process():\n    raise ValueError\n")
+    store.write_node_source(
+        "staged", "readout", "def process(model):\n    return {'score': model}\n"
+    )
+    store.write_node_source("staged", "evaluate", "def process(score):\n    pass\n")
+    store.write_node_source("broken", "fit", "def process(:\n")
+
+    service = RunService(controller=FlowController(store), queue=_Collect())
+    made = [service.submit(name, {}).id for name in ("staged", "broken")]
+    try:
+        for run_id in made:
+            service._drive(run_id)
+        with Session(db_engine) as session:
+            ran, failed = (session.get(Run, run_id) for run_id in made)
+            assert ran.status == failed.status == "error"
+            assert failed.status_reason.startswith("Node 'broken.fit' could not")
+            rows = session.exec(select(RunNode).where(col(RunNode.run_id).in_(made)))
+            assert {row.node: row.status for row in rows} == {
+                "staged.fit": "error",
+                "staged.readout": "skipped",
+                "staged.evaluate": "skipped",
+                "broken.fit": "error",
+            }
+    finally:
+        with Session(db_engine) as session:
+            for node in session.exec(
+                select(RunNode).where(col(RunNode.run_id).in_(made))
+            ).all():
+                session.delete(node)
+            for run in session.exec(select(Run).where(col(Run.id).in_(made))).all():
+                session.delete(run)
+            session.commit()
+
+
 def test_a_retry_is_a_new_run_that_names_the_one_it_repeats():
     """The way back from a run an engine restart interrupted.
 
@@ -905,6 +956,39 @@ def test_a_run_waiting_for_a_worker_keeps_its_lease_and_can_be_cancelled(
             session.commit()
 
 
+class _Bus:
+    def __init__(self):
+        self.events = []
+
+    def publish(self, event):
+        self.events.append(event)
+
+
+def test_cancelling_a_run_does_not_pause_the_live_flow_it_shares_a_key_with():
+    """A run's pipeline publishes on the engine's bus under the flow's key, so
+    its pause marked the live flow paused in every open editor."""
+
+    class _Idle(_OneFlow):
+        workers = None
+        remote = None
+
+    service = RunService(
+        controller=_Idle(FlowDef(name="study", mode="batch")), queue=_Collect()
+    )
+    run = service.submit("study", {})
+    bus = _Bus()
+    pipeline = Pipeline(events=bus)
+    service._active[run.id] = pipeline
+    try:
+        assert service.cancel(run.id)
+        assert pipeline.is_disabled("study")
+        assert not [e for e in bus.events if e["type"] == "flow_paused"]
+    finally:
+        with Session(db_engine) as session:
+            session.delete(session.get(Run, run.id))
+            session.commit()
+
+
 def test_a_key_nobody_used_submits_normally(
     client, superuser_token_headers, monkeypatch
 ):