stroblmeandClaude Opus 5
56396732a4
A stopped flow's nodes are built like any other flow's — being stopped means having no subscriptions, schedules or webhooks, not being absent — so a toggle only ever needed the lifecycle call and the gate that goes with it. It was doing a whole-pipeline rebuild instead, which on a populated installation is every node in every flow reconnecting. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01StpRc2C6au1WJ1EUU7fsfu
335 lines
10 KiB
Python
335 lines
10 KiB
Python
"""Stopping a flow takes it off the engine; pausing holds its nodes."""
|
|
|
|
import asyncio
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import pytest
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
|
|
from fluksio.api.deps import get_current_user
|
|
from fluksio.api.routes.flows import router
|
|
from fluksio.flow.controller import (
|
|
NODE_TYPES,
|
|
FlowController,
|
|
LoadedNode,
|
|
NodeType,
|
|
)
|
|
from fluksio.flow.messages import DType, MessageSpec
|
|
from fluksio.flow.nodes import Node
|
|
from fluksio.flow.pipeline import Pipeline
|
|
from fluksio.flow.schemas import FlowDef, FlowInput, NodeDef
|
|
from fluksio.flow.store import FlowStore
|
|
|
|
|
|
def spec(name: str) -> MessageSpec:
|
|
return MessageSpec(name=name, dtype=DType.FLOAT)
|
|
|
|
|
|
def make_node(node_id: str, flow: str, f, requires=(), provides=()) -> Node:
|
|
node = Node(f=f, requires=list(requires), provides=list(provides), name=node_id)
|
|
node.assign_flow(flow, node_id)
|
|
return node
|
|
|
|
|
|
def a_chain(flow: str, ran: list[str]) -> list[Node]:
|
|
"""source → middle: two nodes, so stepping has somewhere to stop."""
|
|
|
|
def source(params):
|
|
ran.append(f"{flow}.source")
|
|
return {"temp": 20.0}
|
|
|
|
def middle(temp, params):
|
|
ran.append(f"{flow}.middle")
|
|
return {"warm": temp > 10}
|
|
|
|
return [
|
|
make_node("source", flow, source, provides=[spec("temp")]),
|
|
make_node(
|
|
"middle",
|
|
flow,
|
|
middle,
|
|
requires=[spec("temp")],
|
|
provides=[MessageSpec(name="warm", dtype=DType.BOOL)],
|
|
),
|
|
]
|
|
|
|
|
|
def test_a_stopped_flow_runs_nothing_and_its_neighbours_carry_on():
|
|
ran: list[str] = []
|
|
pipeline = Pipeline(
|
|
nodes=a_chain("stopped", ran) + a_chain("running", ran),
|
|
disabled_flows={"stopped"},
|
|
)
|
|
|
|
pipeline.run({})
|
|
|
|
assert not any(name.startswith("stopped.") for name in ran)
|
|
assert {"running.source", "running.middle"} <= set(ran)
|
|
|
|
|
|
def test_a_stopped_flow_ignores_a_trigger_from_its_own_nodes():
|
|
ran: list[str] = []
|
|
nodes = a_chain("stopped", ran)
|
|
pipeline = Pipeline(nodes=nodes, disabled_flows={"stopped"})
|
|
|
|
# What an MQTT subscription or a webhook would do.
|
|
nodes[0].inject({"temp": 20.0})
|
|
|
|
assert ran == []
|
|
assert "stopped.temp" not in pipeline.state
|
|
|
|
|
|
def test_a_paused_flow_still_takes_values_but_acts_on_none_of_them():
|
|
ran: list[str] = []
|
|
nodes = a_chain("demo", ran)
|
|
pipeline = Pipeline(nodes=nodes)
|
|
|
|
pipeline.pause("demo")
|
|
nodes[0].inject({"temp": 20.0})
|
|
|
|
# The value is there to look at; nothing downstream of it ran.
|
|
assert pipeline.state["demo.temp"] == 20.0
|
|
assert ran == []
|
|
|
|
|
|
def test_resuming_runs_what_was_held_back():
|
|
ran: list[str] = []
|
|
pipeline = Pipeline(nodes=a_chain("demo", ran))
|
|
|
|
pipeline.pause("demo")
|
|
pipeline.run({})
|
|
assert ran == []
|
|
|
|
pipeline.resume("demo")
|
|
assert ran == ["demo.source", "demo.middle"]
|
|
assert pipeline.paused_flows() == []
|
|
|
|
|
|
def test_stopped_survives_a_restart(tmp_path: Path):
|
|
store = FlowStore(tmp_path / "flows")
|
|
assert store.read_enabled("heating") is True
|
|
|
|
store.write_enabled("heating", False)
|
|
assert store.read_enabled("heating") is False
|
|
|
|
# A second store over the same directory is what a restart looks like.
|
|
assert FlowStore(tmp_path / "flows").read_enabled("heating") is False
|
|
|
|
|
|
class _StubController:
|
|
"""Only what the step route touches: the store, and the step itself."""
|
|
|
|
def __init__(self, store: FlowStore, stepped: str | None) -> None:
|
|
self.store = store
|
|
self.stepped = stepped
|
|
self.calls: list[str] = []
|
|
|
|
def step_flow(self, flow: str) -> str | None:
|
|
self.calls.append(flow)
|
|
return self.stepped
|
|
|
|
|
|
def _stub(tmp_path: Path, stepped: str | None) -> _StubController:
|
|
"""A controller over a store holding one flow, called 'heating'."""
|
|
store = FlowStore(tmp_path / "flows")
|
|
store.write_flow(FlowDef(name="heating"))
|
|
return _StubController(store, stepped)
|
|
|
|
|
|
def _client(controller: Any) -> TestClient:
|
|
app = FastAPI()
|
|
app.include_router(router, prefix="/api/v1")
|
|
app.state.flow_controller = controller
|
|
app.dependency_overrides[get_current_user] = lambda: None
|
|
return TestClient(app)
|
|
|
|
|
|
def test_the_step_endpoint_reports_what_it_ran(tmp_path: Path):
|
|
controller = _stub(tmp_path, "heating.sensor")
|
|
|
|
response = _client(controller).post("/api/v1/flows/heating/step")
|
|
|
|
assert response.status_code == 200
|
|
assert response.json() == {"message": "Stepped 'heating.sensor'"}
|
|
assert controller.calls == ["heating"]
|
|
|
|
|
|
def test_stepping_a_flow_with_nothing_parked_answers_plainly(tmp_path: Path):
|
|
"""A button pressed once too often is not an error."""
|
|
controller = _stub(tmp_path, None)
|
|
|
|
response = _client(controller).post("/api/v1/flows/heating/step")
|
|
|
|
assert response.status_code == 200
|
|
assert response.json() == {"message": "Nothing held back in flow 'heating'"}
|
|
|
|
|
|
def test_stepping_an_unknown_flow_is_a_404(tmp_path: Path):
|
|
controller = _stub(tmp_path, None)
|
|
|
|
response = _client(controller).post("/api/v1/flows/nosuchflow/step")
|
|
|
|
assert response.status_code == 404
|
|
assert controller.calls == []
|
|
|
|
|
|
def test_a_node_that_will_not_stop_does_not_wedge_the_rebuild(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
):
|
|
"""A node closing a connection nobody answers used to hold the lock forever.
|
|
|
|
Everything that deploys rebuilds, so the next publish, start or stop then
|
|
waited on a lock that was never given back.
|
|
"""
|
|
monkeypatch.setattr("fluksio.flow.controller.NODE_STOP_TIMEOUT", 0.05)
|
|
|
|
class NeverStops(Node):
|
|
async def stop(self, app: FastAPI | None = None) -> None:
|
|
await asyncio.Event().wait()
|
|
|
|
node = NeverStops(f=lambda params: None, name="stuck")
|
|
node.assign_flow("heating", "stuck")
|
|
engine = FlowController(FlowStore(tmp_path / "flows"))
|
|
engine.loaded = {
|
|
"heating.stuck": LoadedNode(id="heating.stuck", flow="heating", node=node)
|
|
}
|
|
|
|
async def scenario() -> None:
|
|
# The first rebuild gives up on the node it cannot stop, and the second
|
|
# is not left queueing behind a lock the first never released.
|
|
await engine.reload()
|
|
await engine.reload()
|
|
|
|
asyncio.run(asyncio.wait_for(scenario(), timeout=5))
|
|
|
|
|
|
class _Lifecycle(Node):
|
|
"""A node type that writes down every lifecycle call a rebuild makes."""
|
|
|
|
calls: list[str] = []
|
|
|
|
def __init__(self, **kwargs: Any) -> None:
|
|
super().__init__(f=lambda params: None, **kwargs)
|
|
|
|
async def start(self, app: FastAPI | None = None) -> None:
|
|
_Lifecycle.calls.append(f"start {self.id}")
|
|
|
|
async def stop(self, app: FastAPI | None = None) -> None:
|
|
_Lifecycle.calls.append(f"stop {self.id}")
|
|
|
|
|
|
@pytest.fixture
|
|
def lifecycle(monkeypatch: pytest.MonkeyPatch) -> list[str]:
|
|
"""A node type on the registry, and the log of what it was asked to do."""
|
|
monkeypatch.setitem(
|
|
NODE_TYPES,
|
|
"lifecycle",
|
|
NodeType(
|
|
title="Lifecycle", description="Counts start and stop.", cls=_Lifecycle
|
|
),
|
|
)
|
|
_Lifecycle.calls.clear()
|
|
return _Lifecycle.calls
|
|
|
|
|
|
def test_rebuilding_one_flow_leaves_another_flows_node_running(
|
|
tmp_path: Path, lifecycle: list[str]
|
|
):
|
|
"""The whole point: reconnecting one flow's nodes, not the installation's."""
|
|
store = FlowStore(tmp_path / "flows")
|
|
for name in ("a", "b"):
|
|
store.write_flow(FlowDef(name=name, nodes=[NodeDef(id="io", type="lifecycle")]))
|
|
engine = FlowController(store)
|
|
|
|
async def scenario() -> None:
|
|
await engine.reload()
|
|
untouched = engine.loaded["b.io"].node
|
|
lifecycle.clear()
|
|
|
|
await engine.reload_flow("a")
|
|
|
|
# Not rebuilt, not restarted, and the very same object still bound.
|
|
assert engine.loaded["b.io"].node is untouched
|
|
assert lifecycle == ["stop a.io", "start a.io"]
|
|
assert engine.loaded["a.io"].node is not None
|
|
|
|
asyncio.run(asyncio.wait_for(scenario(), timeout=10))
|
|
|
|
|
|
def test_a_deleted_flow_leaves_its_consumers_reporting_a_missing_input(
|
|
tmp_path: Path, lifecycle: list[str]
|
|
):
|
|
"""A flow is not a subgraph, so deleting one is another flow's problem."""
|
|
store = FlowStore(tmp_path / "flows")
|
|
store.write_flow(
|
|
FlowDef(
|
|
name="a",
|
|
nodes=[NodeDef(id="meter", type="lifecycle", provides=[spec("temp")])],
|
|
inputs=[FlowInput(spec=spec("spare"), initial=1.0)],
|
|
)
|
|
)
|
|
store.write_flow(
|
|
FlowDef(
|
|
name="b",
|
|
nodes=[
|
|
NodeDef(id="load", type="lifecycle", requires=[spec("a.temp")]),
|
|
NodeDef(id="watch", type="lifecycle", requires=[spec("a.spare")]),
|
|
],
|
|
)
|
|
)
|
|
engine = FlowController(store)
|
|
|
|
async def scenario() -> None:
|
|
await engine.reload()
|
|
assert engine.issues == []
|
|
|
|
store.delete_flow("a")
|
|
await engine.reload_flow("a")
|
|
|
|
# Both the producer and the declared input went with the flow.
|
|
assert sorted((issue.code, issue.node) for issue in engine.issues) == [
|
|
("unconnected_input", "b.load"),
|
|
("unconnected_input", "b.watch"),
|
|
]
|
|
assert engine.pipeline is not None
|
|
assert engine.pipeline.get_node_by_id("a.meter") is None
|
|
|
|
asyncio.run(asyncio.wait_for(scenario(), timeout=10))
|
|
|
|
|
|
def test_stopping_one_flow_does_not_touch_the_others(
|
|
tmp_path: Path, lifecycle: list[str]
|
|
):
|
|
"""A toggle is a lifecycle call, not a rebuild.
|
|
|
|
A stopped flow's nodes are still built — being stopped means having no
|
|
subscriptions, not being absent — so nothing is compiled and no other
|
|
flow's node is asked to reconnect. That is where the seconds went.
|
|
"""
|
|
store = FlowStore(tmp_path / "flows")
|
|
for name in ("a", "b"):
|
|
store.write_flow(FlowDef(name=name, nodes=[NodeDef(id="io", type="lifecycle")]))
|
|
engine = FlowController(store)
|
|
|
|
async def scenario() -> None:
|
|
await engine.reload()
|
|
before = {name: engine.loaded[name].node for name in ("a.io", "b.io")}
|
|
lifecycle.clear()
|
|
|
|
await engine.set_enabled("a", False)
|
|
|
|
assert lifecycle == ["stop a.io"]
|
|
# Still built, still the same objects: only the lifecycle changed.
|
|
assert {name: engine.loaded[name].node for name in ("a.io", "b.io")} == before
|
|
assert engine.disabled == {"a"}
|
|
|
|
lifecycle.clear()
|
|
await engine.set_enabled("a", True)
|
|
|
|
assert lifecycle == ["start a.io"]
|
|
assert {name: engine.loaded[name].node for name in ("a.io", "b.io")} == before
|
|
|
|
asyncio.run(asyncio.wait_for(scenario(), timeout=10))
|