stroblmeandClaude Opus 5
148d50f2bd
- A node's height follows the ports on its busiest side. It is a function of the document, so `layoutGraph` reserves exactly what is drawn and nothing measured is fed back into the layout. - The three status controls now sit in slots that are there whether the control is or not. A node running many times a second mounted and unmounted the stop button on every execution, resizing the card each time. - A port bound to another flow's message is drawn as a label, naming the node at the far end and its type. Only the opposite direction was answered before. The scan behind both is now cached on the store's commit counter rather than reading every flow per request. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_016ZeGnqVsf5VHQqvz4HdUhN
112 lines
3.8 KiB
Python
112 lines
3.8 KiB
Python
"""What a flow's canvas draws for the wiring that leaves it.
|
|
|
|
Both directions matter and only one used to be answered: an outsider reaching
|
|
into this flow was drawn, and this flow reaching out was not — so a port bound
|
|
to another flow's message had nothing on the canvas to account for it.
|
|
"""
|
|
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from fluksio.api.routes.flows import _endpoints
|
|
from fluksio.flow.controller import FlowController
|
|
from fluksio.flow.messages import DType, MessageSpec
|
|
from fluksio.flow.schemas import FlowDef, NodeDef
|
|
from fluksio.flow.store import FlowStore
|
|
|
|
|
|
def spec(name: str) -> MessageSpec:
|
|
return MessageSpec(name=name, dtype=DType.FLOAT)
|
|
|
|
|
|
@pytest.fixture
|
|
def controller(tmp_path: Path) -> FlowController:
|
|
store = FlowStore(tmp_path / "flows")
|
|
store.write_flow(
|
|
FlowDef(
|
|
name="house",
|
|
nodes=[
|
|
NodeDef(id="sensor", type="mqtt", provides=[spec("temp")]),
|
|
# Reaches into `garage`, which is the direction that was drawn.
|
|
NodeDef(id="relay", type="change", provides=[spec("garage.open")]),
|
|
],
|
|
)
|
|
)
|
|
store.write_flow(
|
|
FlowDef(
|
|
name="garage",
|
|
nodes=[
|
|
NodeDef(id="door", type="python", requires=[spec("garage.open")]),
|
|
# Reaches out into `house`, which was drawn nowhere.
|
|
NodeDef(id="watch", type="python", requires=[spec("house.temp")]),
|
|
],
|
|
)
|
|
)
|
|
return FlowController(store)
|
|
|
|
|
|
def labels(
|
|
controller: FlowController, definition: FlowDef
|
|
) -> dict[str, tuple[str, str]]:
|
|
return {
|
|
endpoint.id: (endpoint.label, endpoint.detail)
|
|
for endpoint in _endpoints(controller, definition)
|
|
}
|
|
|
|
|
|
def test_a_port_bound_to_another_flow_names_the_node_at_the_far_end(controller):
|
|
endpoints = labels(controller, controller.store.read_flow("garage"))
|
|
|
|
# `house.relay` publishes `garage.open`, so it reaches in — drawn before.
|
|
# `house.sensor` publishes `house.temp`, which `garage.watch` reads, so
|
|
# this flow reaches out — and the label names the node and its type, the
|
|
# way a dashboard's label names the widget and its type.
|
|
assert endpoints == {
|
|
"flow:house.relay": ("house.relay", "flow"),
|
|
"flow:house.sensor": ("house.sensor", "mqtt"),
|
|
}
|
|
|
|
|
|
def test_a_node_reaching_into_and_out_of_one_flow_is_one_label(controller):
|
|
endpoints = _endpoints(controller, controller.store.read_flow("house"))
|
|
|
|
# `garage.door` reads what `house.relay` publishes, and `garage.watch`
|
|
# reads `house.temp` — two directions, two nodes, two labels, no collision.
|
|
assert {endpoint.id for endpoint in endpoints} == {
|
|
"flow:garage.watch",
|
|
"flow:garage.door",
|
|
}
|
|
door = next(one for one in endpoints if one.id == "flow:garage.door")
|
|
assert door.requires == ["garage.open"]
|
|
|
|
|
|
def test_a_name_no_published_flow_declares_is_drawn_as_the_name(controller):
|
|
definition = controller.store.read_flow("house")
|
|
definition.nodes.append(
|
|
NodeDef(id="guess", type="python", requires=[spec("shed.humidity")])
|
|
)
|
|
|
|
endpoints = labels(controller, definition)
|
|
|
|
assert endpoints["flow:shed.humidity"] == ("shed.humidity", "flow")
|
|
|
|
|
|
def test_the_wiring_index_is_read_once_per_store_revision(controller):
|
|
reads = 0
|
|
original = controller.store.read_all
|
|
|
|
def counted():
|
|
nonlocal reads
|
|
reads += 1
|
|
return original()
|
|
|
|
controller.store.read_all = counted # type: ignore[method-assign]
|
|
_endpoints(controller, controller.store.read_flow("house"))
|
|
_endpoints(controller, controller.store.read_flow("garage"))
|
|
assert reads == 1
|
|
|
|
controller.store.write_flow(FlowDef(name="shed", nodes=[]))
|
|
_endpoints(controller, controller.store.read_flow("house"))
|
|
assert reads == 2
|