Merge branch 'main' of git.stroblme.de:Fluksio/app

backend/tests/flow/test_house_vocabulary.py

@@ -0,0 +1,132 @@
+"""What the house port needed from the built-in nodes.
+
+Three additions, each with a device behind it: an MQTT filter that actually
+routes what it subscribed to, a value lifted out of the object a device wraps
+it in, and a wait that differs per message because a shutter takes longer to
+come up than to go down.
+"""
+
+from fluksio.flow.messages import DType, MessageSpec
+from fluksio.flow.nodes import MqttNode
+from fluksio.flow.nodes.mqtt import topic_matches
+from fluksio.flow.nodes.trigger import TriggerNode
+
+# ── MQTT topic filters ────────────────────────────────────────────────────
+
+
+def test_a_filter_covers_what_the_broker_would_send_it():
+    assert topic_matches("sensors/#", "sensors/living/temp")
+    assert topic_matches("sensors/#", "sensors")
+    assert topic_matches("sensors/+/temp", "sensors/living/temp")
+    assert topic_matches("#", "anything/at/all")
+    assert topic_matches("shelly/status", "shelly/status")
+
+
+def test_a_filter_does_not_cover_a_neighbouring_topic():
+    assert not topic_matches("sensors/+/temp", "sensors/living/kitchen/temp")
+    assert not topic_matches("sensors/+/temp", "sensors/living/hum")
+    assert not topic_matches("sensors/#", "actors/living/temp")
+    assert not topic_matches("sensors/living/temp", "sensors/living")
+
+
+def _subscriber(topic, **params) -> MqttNode:
+    node = MqttNode(
+        provides=[MessageSpec(name="reading", port="reading", dtype=DType.FLOAT)],
+        params={"topic": {"reading": topic}, **params},
+    )
+    node.assign_flow("house", "in")
+    return node
+
+
+def test_a_wildcard_subscription_routes_to_its_port():
+    """It used to subscribe and then drop every message it was sent."""
+    node = _subscriber("shellypm/status/#")
+    assert node._ports_for("shellypm/status/pm1:0") == ["reading"]
+    assert node._ports_for("shellypmc/status/pm1:0") == []
+
+
+def test_an_exact_topic_still_wins_over_a_filter():
+    node = MqttNode(
+        provides=[
+            MessageSpec(name="soc", port="soc", dtype=DType.FLOAT),
+            MessageSpec(name="rest", port="rest", dtype=DType.JSON),
+        ],
+        params={"topic": {"soc": "N/x/battery/Soc", "rest": "N/x/#"}},
+    )
+    node.assign_flow("power", "in")
+    assert node._ports_for("N/x/battery/Soc") == ["soc"]
+    assert node._ports_for("N/x/vebus/P") == ["rest"]
+
+
+# ── json_key ──────────────────────────────────────────────────────────────
+
+
+def test_json_key_lifts_the_value_a_device_wraps():
+    """Victron publishes {"value": 47} on every one of its topics."""
+    node = _subscriber("N/x/battery/Soc", json_key="value")
+    assert node.json_keys == {"reading": "value"}
+
+
+def test_without_json_key_the_port_name_is_the_key_as_before():
+    node = _subscriber("sensors/temp")
+    assert node.json_keys == {}
+
+
+# ── trigger: passthrough and a per-message wait ───────────────────────────
+
+
+class _Pipeline:
+    """Just enough pipeline to record what was deferred and for how long."""
+
+    def __init__(self):
+        self.deferred = []
+        self.state = {}
+
+    def defer(self, node, outputs, seconds, guard=None, kind="cascade"):
+        self.deferred.append((outputs, seconds, guard))
+        return True
+
+
+def _trigger(**params) -> tuple[TriggerNode, _Pipeline]:
+    node = TriggerNode(
+        requires=[
+            MessageSpec(name="run", port="run", dtype=DType.STR),
+            MessageSpec(name="run_for", port="run_for", dtype=DType.FLOAT),
+        ],
+        provides=[MessageSpec(name="cmd", port="cmd", dtype=DType.STR)],
+        params={"then": "STOP", "passthrough": True, "wait_port": "run_for", **params},
+    )
+    node.assign_flow("shutters", "door")
+    pipeline = _Pipeline()
+    node._pipeline = pipeline
+    return node, pipeline
+
+
+def test_the_incoming_value_passes_through_and_the_wait_comes_from_a_port():
+    node, pipeline = _trigger()
+
+    assert node._trigger({}, run="DOWN", run_for=28.0) == {"cmd": "DOWN"}
+    (outputs, seconds, _guard) = pipeline.deferred[-1]
+    assert outputs == {"shutters.cmd": "STOP"} and seconds == 28.0
+
+    node._trigger({}, run="UP", run_for=26.0)
+    assert pipeline.deferred[-1][1] == 26.0
+
+
+def test_a_second_command_invalidates_the_stop_the_first_one_scheduled():
+    node, pipeline = _trigger()
+    node._trigger({}, run="DOWN", run_for=28.0)
+    first = pipeline.deferred[-1][2]
+    node._trigger({}, run="UP", run_for=26.0)
+    assert pipeline.deferred[-1][2] != first
+
+
+def test_a_wait_of_zero_sends_nothing_afterwards():
+    """STOP is commanded once. Nothing follows it, forever."""
+    node, pipeline = _trigger()
+    node._trigger({}, run="DOWN", run_for=28.0)
+    scheduled = len(pipeline.deferred)
+
+    assert node._trigger({}, run="STOP", run_for=0.0) == {"cmd": "STOP"}
+    assert len(pipeline.deferred) == scheduled
+    assert node.recall("armed", 0) == 0