stroblmeandClaude Opus 5
60d7ec81c0
A wheel whose top-level module is `app` collides with anything else in a user's venv, so the package that is about to be published takes the name it is published under. Only the Python package moves; the repo, the Docker WORKDIR and the compose project keep theirs. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
153 lines
4.9 KiB
Python
153 lines
4.9 KiB
Python
"""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 fluksio.flow.dashboards import (
|
|
DashboardDef,
|
|
DashboardStore,
|
|
PageDef,
|
|
SectionDef,
|
|
WidgetDef,
|
|
)
|
|
from fluksio.flow.events import EventBus
|
|
from fluksio.flow.messages import DType, MessageSpec
|
|
from fluksio.flow.nodes import Node
|
|
from fluksio.flow.pipeline import Pipeline, ValueSource
|
|
from fluksio.flow.state import MemoryState
|
|
from fluksio.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") == []
|