stroblme
bdcea05de6
The brain graph counts node_executed events client-side and the websocket is torn down on every shell change, so anything a flow published during the navigation gap was lost. The bus now keeps a session tally per qualified node and the snapshot hands it back, letting a reconnecting client catch up. Both the route and the tunnel connector build that snapshot from one helper so the portal cannot drift from the direct connection.
61 lines
1.9 KiB
Python
61 lines
1.9 KiB
Python
"""What each node has emitted, kept so a reconnecting client is not reset.
|
|
|
|
The brain graph pulses a neuron per emission and tallies them in the browser.
|
|
Moving between the shell and the canvas tears the websocket down, and the bus
|
|
has no replay — so the count is kept here and handed back in the snapshot.
|
|
"""
|
|
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from app.api.routes.flows import snapshot_payload
|
|
from app.flow.controller import FlowController
|
|
from app.flow.events import EventBus, event_bus
|
|
from app.flow.messages import DType, MessageSpec
|
|
from app.flow.nodes import Node
|
|
from app.flow.pipeline import Pipeline
|
|
from app.flow.store import FlowStore
|
|
|
|
|
|
def make_node(node_id: str, f, provides=()) -> Node:
|
|
node = Node(f=f, provides=list(provides), name=node_id)
|
|
node.assign_flow("house", node_id)
|
|
return node
|
|
|
|
|
|
def test_a_node_that_publishes_advances_its_own_count():
|
|
bus = EventBus()
|
|
node = make_node(
|
|
"sensor",
|
|
lambda params: {"temp": 21.0},
|
|
provides=[MessageSpec(name="temp", dtype=DType.FLOAT)],
|
|
)
|
|
pipeline = Pipeline(nodes=[node], events=bus)
|
|
|
|
pipeline.run()
|
|
pipeline.run()
|
|
|
|
# Keyed the way `brain_graph` names its members, which is what the graph
|
|
# looks a count up by.
|
|
assert bus.emits == {"house.sensor": 2}
|
|
|
|
|
|
def test_a_node_that_publishes_nothing_is_not_counted():
|
|
"""Ran, but emitted nothing — the same thing the canvas declines to pulse."""
|
|
bus = EventBus()
|
|
pipeline = Pipeline(nodes=[make_node("quiet", lambda params: None)], events=bus)
|
|
|
|
pipeline.run()
|
|
|
|
assert bus.emits == {}
|
|
|
|
|
|
def test_the_snapshot_carries_what_every_node_has_emitted(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
):
|
|
monkeypatch.setattr(event_bus, "emits", {"house.sensor": 3})
|
|
controller = FlowController(FlowStore(tmp_path / "flows"))
|
|
|
|
assert snapshot_payload(controller)["emits"] == {"house.sensor": 3}
|