bvandeusen
3b6e005ed8
Adds a Kuma-style "is everything up?" surface that aggregates heterogeneous monitor types via a status-source registry (steward/core/status.py): each type registers an async source(db) -> [StatusEntry]. Core registers ping/DNS; the http plugin registers its own from setup() so core never imports plugin tables. Per entry: current up/down, last-30 heartbeat bar, uptime % (24h/7d/30d), latest latency + response sparkline, and TLS expiry countdown (HTTP). New /status page (live htmx refresh) + a status_overview dashboard widget + nav link. Pure-function unit tests for registry + sparkline. First deliverable of milestone #68 (task #866). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
72 lines
2.1 KiB
Python
72 lines
2.1 KiB
Python
"""Unit tests for the status-source registry + sparkline (no DB, no Quart).
|
|
|
|
The DB-backed sources (ping/dns/http) are exercised in the integration lane;
|
|
here we cover the registry contract, ordering, failure isolation, and the
|
|
pure sparkline helper.
|
|
"""
|
|
import asyncio
|
|
|
|
from steward.core import status as status_mod
|
|
from steward.core.status import (
|
|
StatusEntry,
|
|
clear_status_sources,
|
|
collect_status,
|
|
register_status_source,
|
|
sparkline_svg,
|
|
)
|
|
|
|
|
|
def _entry(name, st, kind="ping"):
|
|
return StatusEntry(kind=kind, key=name, name=name, status=st)
|
|
|
|
|
|
def test_collect_orders_down_then_pending_then_up_alphabetical():
|
|
clear_status_sources()
|
|
|
|
async def src(db):
|
|
return [_entry("zeta", "up"), _entry("alpha", "down"),
|
|
_entry("beta", "pending"), _entry("gamma", "up")]
|
|
|
|
register_status_source(src)
|
|
entries = asyncio.run(collect_status(None))
|
|
assert [e.status for e in entries] == ["down", "pending", "up", "up"]
|
|
# within the trailing "up" group, alphabetical by name
|
|
assert entries[2].name == "gamma"
|
|
assert entries[3].name == "zeta"
|
|
clear_status_sources()
|
|
|
|
|
|
def test_failing_source_is_skipped_not_fatal():
|
|
clear_status_sources()
|
|
|
|
async def good(db):
|
|
return [_entry("a", "up")]
|
|
|
|
async def bad(db):
|
|
raise RuntimeError("boom")
|
|
|
|
register_status_source(good)
|
|
register_status_source(bad)
|
|
entries = asyncio.run(collect_status(None))
|
|
assert [e.name for e in entries] == ["a"]
|
|
clear_status_sources()
|
|
|
|
|
|
def test_register_is_idempotent_on_same_callable():
|
|
clear_status_sources()
|
|
|
|
async def src(db):
|
|
return []
|
|
|
|
register_status_source(src)
|
|
register_status_source(src)
|
|
assert status_mod._SOURCES.count(src) == 1
|
|
clear_status_sources()
|
|
|
|
|
|
def test_sparkline_svg_handles_edge_cases():
|
|
assert "polyline" not in sparkline_svg([5]) # <2 points → empty svg
|
|
assert "polyline" in sparkline_svg([1, 2, 3])
|
|
assert "polyline" in sparkline_svg([4, 4, 4]) # flat series, no div-by-zero
|
|
assert "polyline" in sparkline_svg([1, None, 3]) # None values filtered
|