app/backend/tests/flow/test_provenance.py

"""Who caused a value.

The canvas draws an edge per producer of a message. Without knowing which one
actually published, it pulses all of them — and when the cause is a dashboard
control or another flow, it pulses a node that did nothing at all.
"""

from app.flow.dashboards import (
    DashboardDef,
    DashboardStore,
    PageDef,
    SectionDef,
    WidgetDef,
)
from app.flow.events import EventBus
from app.flow.messages import DType, MessageSpec
from app.flow.nodes import Node
from app.flow.pipeline import Pipeline, ValueSource
from app.flow.state import MemoryState
from app.flow.store import FlowStore


def collect(bus: EventBus) -> list[dict]:
    events: list[dict] = []
    bus.publish = events.append  # type: ignore[method-assign]
    return events


def temp_node() -> Node:
    node = Node(
        f=lambda params: {"temp": 21.0},
        provides=[MessageSpec(name="temp", port="temp", dtype=DType.FLOAT)],
        name="sensor",
    )
    node.assign_flow("house", "sensor")
    return node


def test_a_value_a_node_produced_names_that_node():
    bus = EventBus()
    events = collect(bus)
    node = temp_node()
    pipeline = Pipeline(nodes=[node], state=MemoryState(), events=bus)

    pipeline.run()

    published = [e for e in events if e["type"] == "message_value"]
    assert published[0]["source"] == {
        "kind": "node",
        "id": "house.sensor",
        "label": "sensor",
        "detail": "",
    }


def test_a_value_a_node_injected_names_that_node():
    """An MQTT message or a webhook arrives this way rather than by executing."""
    bus = EventBus()
    events = collect(bus)
    node = temp_node()
    pipeline = Pipeline(nodes=[node], state=MemoryState(), events=bus)

    pipeline.apply_outputs(node, {"house.temp": 19.0})

    published = [e for e in events if e["type"] == "message_value"]
    assert published[0]["source"]["id"] == "house.sensor"


def test_a_value_from_a_dashboard_says_so_rather_than_blaming_a_node():
    """The bug this exists for: a slider must not light up a node's edge."""
    bus = EventBus()
    events = collect(bus)
    pipeline = Pipeline(nodes=[temp_node()], state=MemoryState(), events=bus)

    pipeline.publish(
        {"house.temp": 25.0},
        ValueSource(kind="dashboard", id="panel", label="Setpoint", detail="slider"),
    )

    published = [e for e in events if e["type"] == "message_value"]
    assert published[0]["source"]["kind"] == "dashboard"
    assert published[0]["source"]["label"] == "Setpoint"


def test_a_value_from_nowhere_in_particular_is_still_attributed():
    bus = EventBus()
    events = collect(bus)
    pipeline = Pipeline(nodes=[temp_node()], state=MemoryState(), events=bus)

    pipeline.publish({"house.temp": 25.0})

    published = [e for e in events if e["type"] == "message_value"]
    assert published[0]["source"]["kind"] == "api"


# ---------------------------------------------------------------------------
# What the canvas draws for it
# ---------------------------------------------------------------------------


def test_the_widgets_wired_into_a_flow_are_reported(tmp_path):
    store = DashboardStore(FlowStore(tmp_path / "flows"))
    store.write(
        DashboardDef(
            name="panel",
            title="Panel",
            pages=[
                PageDef(
                    id="main",
                    sections=[
                        SectionDef(
                            id="main",
                            widgets=[
                                WidgetDef(
                                    id="setpoint",
                                    type="slider",
                                    title="Setpoint",
                                    config={"target": "house.setpoint"},
                                ),
                                WidgetDef(
                                    id="reading",
                                    type="stat",
                                    title="Reading",
                                    config={"message": "house.temp"},
                                ),
                                # Another flow's message: not this flow's business.
                                WidgetDef(
                                    id="elsewhere",
                                    type="stat",
                                    config={"message": "garage.temp"},
                                ),
                            ],
                        )
                    ],
                )
            ],
        )
    )

    bindings = store.bindings_for("house")

    assert [b["widget"] for b in bindings] == ["setpoint", "reading"]
    setpoint = bindings[0]
    assert setpoint["provides"] == "house.setpoint"
    assert setpoint["requires"] == []
    assert bindings[1]["requires"] == ["house.temp"]


def test_a_flow_nothing_points_at_has_no_endpoints(tmp_path):
    store = DashboardStore(FlowStore(tmp_path / "flows"))

    assert store.bindings_for("house") == []