app/backend/tests/flow/test_runtime_control.py

"""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))