Files
app/backend/tests/flow/test_alerts.py
T
stroblmeandClaude Opus 5 413501c6ce flow: structured dtypes, and the widgets that read them
A series, record or list message declares its shape instead of riding
DType.JSON, so a widget binds a shape rather than some JSON and a wrong
binding is refused before anything runs. A list declares its item type,
which is what keeps list[float] expressible for a pipeline.

On top of that: an agenda over a list, a notification over a record, and
a dashboard alert channel that publishes engine faults as one — so a
panel can show what went wrong without a flow wiring it by hand.

Also: only None means a node published nothing, a falsy value of the
wrong shape is now the named error it always should have been; and the
gauge's readout says its size is viewBox geometry rather than type scale.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-08-17 15:06:45 +02:00

233 lines
6.3 KiB
Python

"""Alerting: failing loudly once, not thirty-six thousand times."""
import asyncio
import pytest
from app.flow.alerts import (
FLAP_THRESHOLD,
RATE_LIMIT,
Alert,
AlertManager,
AlertsConfig,
Channel,
Rule,
describe,
)
from app.flow.events import EventBus
class Clock:
"""A hand-wound clock, so cooldowns take no real time."""
def __init__(self) -> None:
self.now = 0.0
def __call__(self) -> float:
return self.now
def advance(self, seconds: float) -> None:
self.now += seconds
def manager(clock: Clock) -> tuple[AlertManager, list[Alert]]:
sent: list[Alert] = []
config = AlertsConfig(
channels=[Channel(name="phone", kind="ntfy", config={"topic": "t"})],
rules=[Rule(events=[], channels=["phone"], cooldown_s=900)],
)
alerts = AlertManager(EventBus(), config=config, now=clock)
async def capture(channel, alert):
sent.append(alert)
alerts.send = capture # type: ignore[method-assign]
return alerts, sent
def error_event(node: str = "heating.pump") -> dict:
return {"type": "node_error", "flow": "heating", "node": node, "error": "boom"}
def test_a_failure_reaches_the_channel():
clock = Clock()
alerts, sent = manager(clock)
asyncio.run(alerts.handle(error_event()))
assert [a.title for a in sent] == ["heating.pump failed"]
assert sent[0].body == "boom"
def test_ordinary_traffic_is_not_an_alert():
clock = Clock()
alerts, sent = manager(clock)
asyncio.run(alerts.handle({"type": "message_value", "name": "heating.temp"}))
asyncio.run(alerts.handle({"type": "node_executed", "node": "heating.pump"}))
assert sent == []
def test_the_same_failure_repeating_is_one_alert():
clock = Clock()
alerts, sent = manager(clock)
async def scenario():
for _ in range(50):
await alerts.handle(error_event())
clock.advance(1)
asyncio.run(scenario())
assert len(sent) == 1
def test_the_cooldown_ends_and_says_what_was_missed():
clock = Clock()
alerts, sent = manager(clock)
async def scenario():
await alerts.handle(error_event())
for _ in range(4):
clock.advance(10)
await alerts.handle(error_event())
clock.advance(1000)
await alerts.handle(error_event())
asyncio.run(scenario())
assert len(sent) == 2
assert "4 more since the last alert" in sent[1].body
def test_different_nodes_alert_separately():
clock = Clock()
alerts, sent = manager(clock)
async def scenario():
await alerts.handle(error_event("heating.pump"))
await alerts.handle(error_event("heating.valve"))
asyncio.run(scenario())
assert len(sent) == 2
def test_a_flapping_connection_goes_quiet():
"""A device dropping every few seconds is one story, not one alert each."""
clock = Clock()
alerts, sent = manager(clock)
async def scenario():
for _ in range(FLAP_THRESHOLD * 2 + 6):
await alerts.handle(
{"type": "node_health", "node": "heating.pump", "health": "down"}
)
clock.advance(5)
await alerts.handle(
{"type": "node_health", "node": "heating.pump", "health": "ok"}
)
clock.advance(5)
asyncio.run(scenario())
# The first drop is worth knowing about; the rest is noise.
assert len(sent) == 1
def test_a_storm_is_capped():
clock = Clock()
alerts, sent = manager(clock)
async def scenario():
for i in range(RATE_LIMIT + 20):
await alerts.handle(error_event(f"heating.n{i}"))
clock.advance(1)
asyncio.run(scenario())
assert len(sent) == RATE_LIMIT
def test_nothing_is_sent_without_a_rule():
clock = Clock()
alerts, sent = manager(clock)
alerts.config = AlertsConfig(
channels=[Channel(name="phone", kind="ntfy", config={"topic": "t"})],
rules=[Rule(events=["queue_unavailable"], channels=["phone"])],
)
asyncio.run(alerts.handle(error_event()))
assert sent == []
def test_alerting_can_be_switched_off():
clock = Clock()
alerts, sent = manager(clock)
alerts.config = alerts.config.model_copy(update={"enabled": False})
asyncio.run(alerts.handle(error_event()))
assert sent == []
@pytest.mark.parametrize(
("event", "expected"),
[
(
{"type": "flow_quarantined", "flow": "heating"},
"Flow 'heating' was quarantined",
),
(
{"type": "queue_unavailable", "error": "gone"},
"The work queue is unreachable",
),
({"type": "engine_degraded", "reason": "lag"}, "The engine is struggling"),
({"type": "node_health", "health": "ok"}, None),
# The engine publishes `health`, not `status`: reading the wrong key
# meant a device dropping never alerted anyone.
(
{"type": "node_health", "node": "heating.pump", "health": "down"},
"heating.pump lost its connection",
),
],
)
def test_every_alerting_event_reads_as_a_sentence(event, expected):
alert = describe(event)
assert (alert.title if alert else None) == expected
def test_a_dashboard_channel_publishes_the_alert_as_a_record():
published: list[tuple[str, dict]] = []
config = AlertsConfig(
channels=[
Channel(name="panel", kind="dashboard", config={"message": "house.notice"})
],
rules=[Rule(events=[], channels=["panel"])],
)
alerts = AlertManager(EventBus(), config=config, now=Clock())
alerts.publish = lambda name, value: published.append((name, value))
asyncio.run(alerts.handle(error_event()))
assert len(published) == 1
name, record = published[0]
assert name == "house.notice"
# Flat named scalars, which is what a notification widget binds.
assert record["title"] == "heating.pump failed"
assert record["severity"] == "error"
assert all(isinstance(v, str) for v in record.values())
def test_a_dashboard_channel_without_a_message_says_so():
alerts = AlertManager(EventBus(), now=Clock())
alerts.publish = lambda name, value: None
channel = Channel(name="panel", kind="dashboard")
with pytest.raises(ValueError):
asyncio.run(
alerts.send(channel, Alert(title="t", body="b"), raise_on_error=True)
)