"""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") == []