"""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