"""A device node: never run by the engine, degraded while its device is away.""" import asyncio from pathlib import Path import pytest from fluksio.flow.alerts import Alert, AlertManager, AlertsConfig, Channel, Rule from fluksio.flow.controller import HEALTH_FLOOR, FlowController from fluksio.flow.devices import presence from fluksio.flow.events import EventBus from fluksio.flow.messages import DType, MessageSpec from fluksio.flow.schemas import FlowDef, FlowInput, NodeDef from fluksio.flow.store import FlowStore def _controller(tmp_path: Path) -> FlowController: store = FlowStore(tmp_path / "flows") store.write_flow( FlowDef( name="hall", inputs=[ FlowInput(spec=MessageSpec(name="level", dtype=DType.INT), initial=50) ], nodes=[ NodeDef( id="panel", type="device", params={"device": "wall"}, provides=[ MessageSpec(name="bell", port="gpio.4", dtype=DType.BOOL) ], requires=[ MessageSpec( name="level", port="backlight", dtype=DType.INT, trigger=False, ) ], ), NodeDef( id="ring", type="change", requires=[MessageSpec(name="bell", dtype=DType.BOOL)], provides=[MessageSpec(name="rung", dtype=DType.BOOL)], ), ], ) ) return FlowController(store) def test_the_engine_never_runs_a_device_node(tmp_path: Path) -> None: controller = _controller(tmp_path) asyncio.run(controller.reload()) assert controller.issues == [] node = controller.get_node("hall.panel") assert node is not None and node.passive ran: list[object] = [] node.f = lambda **kwargs: ran.append(kwargs) # type: ignore[method-assign] # The message its out port follows changes, and a manual run of the flow. controller.publish_message("hall.level", 80) controller.run_flow("hall") assert ran == [] # What the device publishes on its in port runs what reads it. controller.publish_message("hall.bell", True) assert controller.values("hall")["hall.rung"]["value"] is True asyncio.run(controller._teardown()) def test_a_device_node_is_degraded_while_its_device_is_away(tmp_path: Path) -> None: controller = _controller(tmp_path) async def scenario() -> list[tuple[str, str | None]]: await controller.reload() entry = controller.loaded["hall.panel"] seen = [(entry.health, entry.health_detail)] presence.opened("wall") seen.append((entry.health, entry.health_detail)) presence.closed("wall") seen.append((entry.health, entry.health_detail)) await controller._teardown() # Stopped, it no longer follows the device. presence.opened("wall") seen.append((entry.health, entry.health_detail)) presence.closed("wall") return seen # Degraded, not down: a switched-off panel is a warning, never a fault. assert asyncio.run(scenario()) == [ ("degraded", "Offline"), ("ok", None), ("degraded", "Offline"), ("degraded", "Offline"), ] presence.forget("wall") def test_an_offline_device_is_a_warning_not_a_fault(tmp_path: Path) -> None: controller = _controller(tmp_path) async def scenario() -> list: await controller.reload() # Past the floor that holds back a device rebooting. controller.loaded["hall.panel"].health_since -= 3600 found = ( controller.unhealthy_nodes(), [e.id for e in controller.unhealthy_nodes(health="degraded")], [(i.code, i.advisory) for i in controller.flow_issues("hall")], ) await controller._teardown() return list(found) assert asyncio.run(scenario()) == [ [], ["hall.panel"], [("node_degraded", True)], ] def test_a_device_away_after_a_restart_alerts_nobody(tmp_path: Path) -> None: """No push, and degraded on screen only once the floor has passed. A restart starts every device node without its device, which comes back within seconds, and a rule on every node's health is what sends phones a push. """ controller = _controller(tmp_path) events: list[dict] = [] original = controller._publish controller._publish = lambda event: (events.append(event), original(event)) # type: ignore[method-assign] sent: list[Alert] = [] alerts = AlertManager( EventBus(), AlertsConfig( channels=[Channel(name="phone", kind="ntfy", config={"topic": "t"})], rules=[Rule(events=["node_health"], channels=["phone"])], ), ) async def capture(channel: Channel, alert: Alert) -> bool: sent.append(alert) return True alerts.send = capture # type: ignore[method-assign] async def scenario() -> list[list[str]]: await controller.reload() # the restart: the device is not back yet surfaced = [[e.id for e in controller.unhealthy_nodes(health="degraded")]] presence.opened("wall") presence.closed("wall") # and it drops again surfaced.append([e.id for e in controller.unhealthy_nodes(health="degraded")]) controller.loaded["hall.panel"].health_since -= HEALTH_FLOOR surfaced.append([e.id for e in controller.unhealthy_nodes(health="degraded")]) for event in events: await alerts.handle(event) await controller._teardown() return surfaced assert asyncio.run(scenario()) == [[], [], ["hall.panel"]] assert [e["health"] for e in events if e["type"] == "node_health"] == [ "degraded", "ok", "degraded", ] assert sent == [] presence.forget("wall") def test_a_device_gone_for_the_grace_reads_false( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: """A flag such as `person` stops saying someone is there once the device has been away for the grace, and a device back in time keeps it.""" from fluksio.flow.nodes import device as device_node monkeypatch.setattr(device_node, "OFFLINE_GRACE_S", 0.1) controller = _controller(tmp_path) def bell() -> object: return controller.values("hall")["hall.bell"]["value"] async def scenario() -> list[object]: await controller.reload() presence.opened("wall") controller.publish_message("hall.bell", True) seen = [bell()] # Back within the grace: nothing changes. presence.closed("wall") presence.opened("wall") await asyncio.sleep(0.25) seen.append(bell()) # Away for good. presence.closed("wall") await asyncio.sleep(0.05) seen.append(bell()) await asyncio.sleep(0.25) seen.append(bell()) # What reads it hears so too. seen.append(controller.values("hall")["hall.rung"]["value"]) await controller._teardown() presence.forget("wall") return seen assert asyncio.run(scenario()) == [True, True, True, False, False]