158 lines
5.5 KiB
Python
158 lines
5.5 KiB
Python
"""The engine runs a flow by its key and finds it by its id.
|
|
|
|
Every test runs twice, once with the id equal to the key the way a flow from
|
|
before ids has it, and once with them apart. A mix-up of the two is invisible
|
|
in the first run, which is why the second one exists.
|
|
"""
|
|
|
|
import asyncio
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from fluksio.flow.controller import HOOK_PREFIX, FlowController
|
|
from fluksio.flow.messages import DType, MessageSpec
|
|
from fluksio.flow.queue import MemoryWorkQueue, WorkItem
|
|
from fluksio.flow.schemas import FlowDef, NodeDef
|
|
from fluksio.flow.store import FlowStore
|
|
|
|
SENSOR = "def process():\n return {'reading': 21.5}\n"
|
|
DOUBLE = "def process(reading):\n return {'doubled': reading * 2}\n"
|
|
|
|
|
|
@pytest.fixture(params=["house", "x_house"], ids=["id=key", "id!=key"])
|
|
def flow_id(request: pytest.FixtureRequest) -> str:
|
|
return str(request.param)
|
|
|
|
|
|
@pytest.fixture
|
|
def store(tmp_path: Path, flow_id: str) -> FlowStore:
|
|
store = FlowStore(tmp_path / "flows")
|
|
store.write_flow(
|
|
FlowDef(
|
|
id=flow_id,
|
|
name="house",
|
|
nodes=[
|
|
NodeDef(
|
|
id="sensor",
|
|
provides=[MessageSpec(name="reading", dtype=DType.FLOAT)],
|
|
),
|
|
NodeDef(
|
|
id="double",
|
|
requires=[MessageSpec(name="reading", dtype=DType.FLOAT)],
|
|
provides=[MessageSpec(name="doubled", dtype=DType.FLOAT)],
|
|
),
|
|
NodeDef(
|
|
id="hook",
|
|
type="http",
|
|
params={"url": "/tick", "method": "POST"},
|
|
provides=[MessageSpec(name="tick", dtype=DType.FLOAT)],
|
|
),
|
|
],
|
|
)
|
|
)
|
|
store.write_node_source(flow_id, "sensor", SENSOR)
|
|
store.write_node_source(flow_id, "double", DOUBLE)
|
|
return store
|
|
|
|
|
|
def test_a_flow_runs_under_its_key_from_the_code_its_id_holds(
|
|
store: FlowStore, flow_id: str
|
|
):
|
|
engine = FlowController(store)
|
|
asyncio.run(engine.reload())
|
|
|
|
entry = engine.loaded["house.double"]
|
|
assert (entry.flow, entry.flow_id) == ("house", flow_id)
|
|
# Read by the id: by the key it would be the template, and missing.
|
|
assert not entry.missing_source and entry.status.value == "active"
|
|
assert engine.loaded["house.hook"].node.url == f"{HOOK_PREFIX}/{flow_id}/tick"
|
|
|
|
engine.run_flow("house")
|
|
|
|
assert engine.values()["house.doubled"]["value"] == 43.0
|
|
assert {s.id for s in engine.node_statuses("house")} >= {"house.double"}
|
|
|
|
|
|
def test_stopping_writes_under_the_id_and_gates_the_key(store: FlowStore, flow_id: str):
|
|
engine = FlowController(store)
|
|
|
|
async def scenario() -> None:
|
|
await engine.reload()
|
|
await engine.set_enabled(flow_id, "house", False)
|
|
assert engine.disabled == {"house"}
|
|
assert not store.read_enabled(flow_id)
|
|
# Read back by id when everything is built again.
|
|
await engine.reload()
|
|
assert engine.disabled == {"house"}
|
|
|
|
asyncio.run(scenario())
|
|
assert (store.root / flow_id / "runtime.json").exists()
|
|
|
|
|
|
def test_a_flow_rebuilt_by_id_is_taken_out_by_its_key_once_gone(
|
|
store: FlowStore, flow_id: str
|
|
):
|
|
engine = FlowController(store)
|
|
|
|
async def scenario() -> None:
|
|
await engine.reload()
|
|
assert "house.sensor" in engine.loaded
|
|
await engine.reload_flow(flow_id)
|
|
assert "house.sensor" in engine.loaded
|
|
|
|
store.delete_flow(flow_id)
|
|
await engine.reload_flow(flow_id)
|
|
|
|
asyncio.run(scenario())
|
|
assert not [node for node in engine.loaded if node.startswith("house.")]
|
|
assert engine.pipeline is not None and not engine.pipeline.flow_nodes("house")
|
|
|
|
|
|
def test_a_draft_previews_by_id_and_reports_by_key(store: FlowStore, flow_id: str):
|
|
draft = store.read_flow(flow_id).model_copy(update={"title": "Edited"})
|
|
store.write_draft(draft, draft.version)
|
|
engine = FlowController(store)
|
|
|
|
preview = engine.preview(flow_id)
|
|
|
|
assert {node.id for node in preview.nodes} == {
|
|
"house.sensor",
|
|
"house.double",
|
|
"house.hook",
|
|
}
|
|
assert all(node.status == "active" for node in preview.nodes)
|
|
engine.run_preview(flow_id)
|
|
assert engine.state.get("house.doubled") == 43.0
|
|
|
|
|
|
def test_what_a_key_leaves_behind_is_found_and_forgotten(store: FlowStore):
|
|
engine = FlowController(store)
|
|
queue = MemoryWorkQueue()
|
|
engine.execution = SimpleNamespace(queue=queue) # type: ignore[assignment]
|
|
engine.state.set("house.reading", 1.0)
|
|
engine.state.set("__node__:house.double:last", 2)
|
|
engine.state.set("housework.reading", 3.0)
|
|
queue.add_delayed(WorkItem(kind="flush", node="house.double", flow="house"), 9e9)
|
|
queue.park("house", WorkItem(kind="cascade", node="house.sensor", flow="house"))
|
|
# Another flow's work, carrying one of this key's messages.
|
|
queue.add(
|
|
WorkItem(
|
|
kind="cascade", node="garden.pump", flow="garden", outputs={"house.x": 1}
|
|
)
|
|
)
|
|
|
|
found = engine.key_residue("house")
|
|
|
|
assert "house.reading" in found and "__node__:house.double:last" in found
|
|
assert "housework.reading" not in found
|
|
assert sum("queued work" in what for what in found) == 3
|
|
assert engine.key_residue("garden") == ["queued work of 'garden.pump'"]
|
|
|
|
engine.forget_flow("house")
|
|
|
|
# What the other flow queued is its own, and runs as it would have.
|
|
assert engine.key_residue("house") == ["queued work of 'garden.pump'"]
|
|
assert engine.state.get("housework.reading") == 3.0
|