Melvin StroblandClaude Fable 5
3724b68f23
Connectors are the device-facing node class third parties write, so the surface they build against is versioned and documented: ConnectorNode carries a declared contract version, a polling loop that publishes only what changed and reports health around it, and parameters whose credential fields are marked x-secret so the editor offers the secrets store instead of a text box. They are found through the fluksio.node_types entry point group, with the package's own metadata as the manifest. docs/connectors/ has the contract and the authoring guide; connector-skeleton/ is a working one to copy. The controller no longer knows what any node type is: start, stop and report_health are protocol methods on Node, and the built-ins were migrated to them first, so the hooks a connector implements are the ones the engine has been driving all along. Marking a node reusable moves its source to _lib/ and points the node at it by name. Other flows instantiate it with their own ports and settings, one fix reaches all of them, and a shared source still in use cannot be deleted. Ports gained an interval: an output publishes, and an input wakes its node, at most every n seconds. State keeps the latest value, so only the delivery is skipped, and pressing Run is never throttled. Also fixes autosave sending no version on its first save of a session, which made every flow saved more than once conflict with itself. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
157 lines
4.5 KiB
Python
157 lines
4.5 KiB
Python
"""The connector contract: polling, deduplication, health and discovery."""
|
|
|
|
import asyncio
|
|
from typing import Any
|
|
|
|
from app.flow.connector import CONTRACT_VERSION, ConnectorNode
|
|
from app.flow.controller import NODE_TYPES
|
|
from app.flow.messages import DType, MessageSpec
|
|
from app.flow.nodes import Node
|
|
from app.flow.pipeline import Pipeline
|
|
from app.flow.plugins import load_plugins
|
|
|
|
|
|
class Sensor(ConnectorNode):
|
|
contract = CONTRACT_VERSION
|
|
title = "Test sensor"
|
|
description = "Reads whatever it is told to."
|
|
|
|
class Params(ConnectorNode.Params):
|
|
secret_token: str | None = None
|
|
|
|
def __init__(self, readings: list[Any] | None = None, **kwargs: Any) -> None:
|
|
super().__init__(**kwargs)
|
|
self._readings = list(readings or [])
|
|
self.polls = 0
|
|
|
|
async def poll(self) -> dict[str, Any] | None:
|
|
self.polls += 1
|
|
if not self._readings:
|
|
return None
|
|
value = self._readings.pop(0)
|
|
if isinstance(value, Exception):
|
|
raise value
|
|
return {"reading": value}
|
|
|
|
|
|
def a_sensor(readings: list[Any], **params: Any) -> Sensor:
|
|
node = Sensor(
|
|
readings=readings,
|
|
provides=[MessageSpec(name="reading", dtype=DType.FLOAT)],
|
|
params={"poll_interval": 0.01, **params},
|
|
)
|
|
node.assign_flow("demo", "sensor")
|
|
return node
|
|
|
|
|
|
def run_briefly(node: ConnectorNode, seconds: float = 0.12) -> None:
|
|
"""Start the poll loop, let it tick a few times, stop it."""
|
|
|
|
async def cycle() -> None:
|
|
await node.start()
|
|
await asyncio.sleep(seconds)
|
|
await node.stop()
|
|
|
|
asyncio.run(cycle())
|
|
|
|
|
|
def test_polling_publishes_what_it_reads():
|
|
node = a_sensor([21.5])
|
|
pipeline = Pipeline(nodes=[node])
|
|
|
|
run_briefly(node)
|
|
|
|
assert pipeline.state["demo.reading"] == 21.5
|
|
|
|
|
|
def test_an_unchanged_reading_is_not_republished():
|
|
node = a_sensor([21.5, 21.5, 21.5])
|
|
consumer_ran: list[float] = []
|
|
|
|
def consume(reading, params):
|
|
consumer_ran.append(reading)
|
|
return None
|
|
|
|
consumer = Node(
|
|
f=consume,
|
|
requires=[MessageSpec(name="reading", dtype=DType.FLOAT)],
|
|
name="consumer",
|
|
)
|
|
consumer.assign_flow("demo", "consumer")
|
|
Pipeline(nodes=[node, consumer])
|
|
|
|
run_briefly(node)
|
|
|
|
# Polled repeatedly, but the value never changed, so downstream ran once.
|
|
assert node.polls > 1
|
|
assert consumer_ran == [21.5]
|
|
|
|
|
|
def test_a_failing_poll_reports_down_and_keeps_going():
|
|
health: list[tuple[str, str | None]] = []
|
|
node = a_sensor([RuntimeError("device unplugged"), 21.5])
|
|
node._on_health = lambda _node, status, detail: health.append((status, detail))
|
|
Pipeline(nodes=[node])
|
|
|
|
run_briefly(node)
|
|
|
|
assert ("down", "RuntimeError: device unplugged") in health
|
|
# It recovered rather than giving up.
|
|
assert health[-1][0] == "ok"
|
|
|
|
|
|
def test_a_credential_param_is_marked_for_the_editor():
|
|
schema = Sensor.Params.model_json_schema()
|
|
assert schema["properties"]["poll_interval"]["default"] == 0
|
|
assert "secret_token" in schema["properties"]
|
|
|
|
|
|
def test_a_connector_is_discovered_from_its_entry_point(monkeypatch):
|
|
class FakeDist:
|
|
name = "fluksio-connector-test"
|
|
version = "0.1.0"
|
|
|
|
class FakeEntry:
|
|
name = "test_sensor"
|
|
dist = FakeDist()
|
|
|
|
def load(self):
|
|
return Sensor
|
|
|
|
monkeypatch.setattr("app.flow.plugins.entry_points", lambda group: [FakeEntry()])
|
|
try:
|
|
assert load_plugins() == ["test_sensor"]
|
|
assert NODE_TYPES["test_sensor"].plugin == "fluksio-connector-test 0.1.0"
|
|
assert NODE_TYPES["test_sensor"].title == "Test sensor"
|
|
finally:
|
|
NODE_TYPES.pop("test_sensor", None)
|
|
|
|
|
|
def test_a_connector_written_for_another_contract_is_refused(monkeypatch):
|
|
class Outdated(ConnectorNode):
|
|
contract = CONTRACT_VERSION + 1
|
|
|
|
class FakeEntry:
|
|
name = "outdated"
|
|
dist = None
|
|
|
|
def load(self):
|
|
return Outdated
|
|
|
|
monkeypatch.setattr("app.flow.plugins.entry_points", lambda group: [FakeEntry()])
|
|
assert load_plugins() == []
|
|
assert "outdated" not in NODE_TYPES
|
|
|
|
|
|
def test_a_connector_may_not_take_over_a_built_in_type(monkeypatch):
|
|
class FakeEntry:
|
|
name = "mqtt"
|
|
dist = None
|
|
|
|
def load(self): # pragma: no cover - never reached
|
|
raise AssertionError("should not be loaded")
|
|
|
|
monkeypatch.setattr("app.flow.plugins.entry_points", lambda group: [FakeEntry()])
|
|
assert load_plugins() == []
|
|
assert NODE_TYPES["mqtt"].plugin is None
|