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>
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 fluksio.api.routes.flows import snapshot_payload
|
|
from fluksio.flow.controller import FlowController
|
|
from fluksio.flow.events import EventBus, event_bus
|
|
from fluksio.flow.messages import DType, MessageSpec
|
|
from fluksio.flow.nodes import Node
|
|
from fluksio.flow.pipeline import Pipeline
|
|
from fluksio.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}
|