"""The poll loop: isolation, last-good-value, and events that mark transitions only. These three behaviours are why the dashboard can be trusted when something is broken, so they are asserted directly rather than through HTTP: * a source that throws must not take the loop, the hub, or its neighbours down; * a failed poll keeps the previous payload and is marked stale — never zeroed; * ok→fail and fail→ok each write exactly one event; 500 more failures write nothing; * maintenance failing is logged, because a silent rollup failure means every long-window chart is quietly empty (that bug is what the logging was added for). """ from __future__ import annotations import asyncio import logging import time from collections.abc import Iterator from typing import Any import pytest from dashboards.poller import Hub, Job, Result from dashboards.rates import RateEstimator from dashboards.store import Event, Sample, SampleStore @pytest.fixture def store(tmp_path: Any) -> Iterator[SampleStore]: instance = SampleStore(tmp_path / "poller.sqlite3") try: yield instance finally: instance.close() def ok_result(value: float = 1.0) -> Result: return Result(payload={"value": value}, samples=[Sample("host", "gw", "cpu.usage_pct", value, time.time())]) async def failing() -> Result: raise ConnectionError("agent is down") async def test_a_successful_poll_stores_payload_samples_and_health(store) -> None: job = Job("host:gw", "system", "gw", 0.5, lambda: asyncio.sleep(0, ok_result(7.0))) outcome = await job.run_once(store) assert outcome is True, job.state.error assert job.state.ok is True and job.state.error is None assert job.state.payload == {"value": 7.0} assert job.state.successes == 1 and job.state.samples_written == 1 assert job.state.latency_ms is not None assert store.recent_events(limit=5) == [], "a first successful poll is not a recovery" async def test_a_failing_poll_keeps_the_last_good_payload(store) -> None: """The UI shows `stale 12s` against real numbers; it must not show zeros or n/a.""" job = Job("host:gw", "system", "gw", 0.5, lambda: asyncio.sleep(0, ok_result(7.0))) await job.run_once(store) job.fn = failing assert await job.run_once(store) is False assert job.state.ok is False assert job.state.payload == {"value": 7.0}, "a dead host must not blank its last values" assert job.state.error == "ConnectionError: agent is down" assert job.state.as_json()["age_seconds"] is not None async def test_transitions_write_one_event_each_and_spam_writes_none(store) -> None: job = Job("host:gw", "system", "gw", 0.5, lambda: asyncio.sleep(0, ok_result()), grace_failures=1) await job.run_once(store) job.fn = failing for _ in range(5): await job.run_once(store) events = store.recent_events(limit=50) down = [event for event in events if event.kind == "unreachable"] assert len(down) == 1, "five consecutive failures must not log five events" assert down[0].host == "gw", "events carry the display label, not the internal job key" assert down[0].severity == "error" job.fn = lambda: asyncio.sleep(0, ok_result()) await job.run_once(store) events = store.recent_events(limit=50) up = [event for event in events if event.kind == "reachable"] assert len(up) == 1 assert up[0].message == "gw is answering again after 5 failures" async def test_one_failure_reads_as_a_failure_not_failures(store) -> None: job = Job("host:gw", "system", "gw", 0.5, lambda: asyncio.sleep(0, ok_result())) await job.run_once(store) job.fn = failing await job.run_once(store) job.fn = lambda: asyncio.sleep(0, ok_result()) await job.run_once(store) up = [event for event in store.recent_events(limit=50) if event.kind == "reachable"] assert up[0].message.endswith("after 1 failure") async def test_grace_allows_a_blip_without_an_event(store) -> None: """One failure inside the grace window must not page anybody.""" job = Job("host:gw", "system", "gw", 0.5, failing, grace_failures=2) await job.run_once(store) assert store.recent_events(limit=10) == [] await job.run_once(store) events = store.recent_events(limit=10) assert [event.kind for event in events] == ["unreachable"] assert events[0].severity == "warn", "never answered is a config problem, not an outage" assert "never answered" in events[0].message for _ in range(4): await job.run_once(store) assert len(store.recent_events(limit=10)) == 1, "one event per outage, not per poll" async def test_a_slow_poll_never_overlaps_itself(store) -> None: calls = 0 gate = asyncio.Event() async def slow() -> Result: nonlocal calls calls += 1 await gate.wait() return ok_result() job = Job("host:gw", "system", "gw", 0.5, slow) first = asyncio.create_task(job.run_once(store)) await asyncio.sleep(0.01) assert await job.run_once(store) is False, "an overlapping poll double-counts rates" gate.set() assert await first is True assert calls == 1 async def test_a_counter_that_resets_is_reported_not_silently_negative() -> None: """The reset path is what ships: an agent restart must not read as negative throughput.""" rates = RateEstimator(span=10.0) first, _ = rates.rate("cpu.total", 1000.0, now=100.0) assert first is None, "one sample cannot produce a rate" second, window = rates.rate("cpu.total", 2000.0, now=110.0) assert second == pytest.approx(100.0) and window == pytest.approx(10.0) after, _ = rates.rate("cpu.total", 5.0, now=120.0) # agent restarted assert after is None, "a reset counter must read n/a, never as negative throughput" async def test_the_hub_runs_every_job_and_stops_cleanly(store) -> None: counts = {"a": 0, "b": 0} async def good(key: str) -> Result: counts[key] += 1 return ok_result() hub = Hub(store) hub.add(Job("host:a", "system", "a", 0.5, lambda: good("a"))) hub.add(Job("host:b", "system", "b", 0.5, failing)) hub.add(Job("llm:off", "llm", "off", 0.5, failing, enabled=False)) hub.start() try: await asyncio.sleep(1.2) finally: await hub.stop() assert counts["a"] >= 2, "a neighbour that throws must not slow the healthy one" states = {state["key"]: state for state in hub.states()} assert states["host:a"]["ok"] is True assert states["host:b"]["ok"] is False assert states["llm:off"]["configured"] is False, "not configured is not the same as broken" assert hub.jobs["host:b"].state.failures >= 1 async def test_states_can_be_filtered_to_one_dashboard(store) -> None: hub = Hub(store) hub.add(Job("host:a", "system", "a", 0.5, lambda: asyncio.sleep(0, ok_result()))) hub.add(Job("nut", "ups", "NUT server", 0.5, lambda: asyncio.sleep(0, ok_result()))) assert [s["key"] for s in hub.states("ups")] == ["nut"] async def test_maintenance_failure_is_logged_not_swallowed(store, monkeypatch, caplog: pytest.LogCaptureFixture) -> None: """A broken rollup means every chart past the raw window is empty. Retrying quietly turned a data-loss bug into nothing at all — this is the regression that added the log.""" def boom() -> dict[str, int]: raise RuntimeError("no such table: sample_rollup") monkeypatch.setattr(store, "maintain", boom) hub = Hub(store) task = asyncio.create_task(hub._maintain_loop(interval=0.02)) try: with caplog.at_level(logging.ERROR, logger="dashboards.poller"): await asyncio.sleep(0.15) finally: task.cancel() with pytest.raises(asyncio.CancelledError): await task assert "store maintenance failed" in caplog.text assert "no such table" in caplog.text, "the log line must carry the real reason" async def test_maintenance_runs_from_the_hub_and_records_its_pass(store) -> None: hub = Hub(store) calls = [] store.maintain = lambda: calls.append(1) or {"to_minute": 0, "to_hour": 0, # type: ignore[method-assign] "pruned": 0} task = asyncio.create_task(hub._maintain_loop(interval=0.02)) try: await asyncio.sleep(0.1) finally: task.cancel() with pytest.raises(asyncio.CancelledError): await task assert calls, "the maintenance loop must actually call maintain" async def test_events_written_by_a_job_land_in_the_store(store) -> None: event = Event(1_700_000_000.0, "ups", "cyberpower", "status", "warn", "OL → OB") job = Job("nut", "ups", "NUT server", 0.5, lambda: asyncio.sleep(0, Result(payload={}, events=[event]))) await job.run_once(store) assert [e.message for e in store.recent_events(limit=5)] == ["OL → OB"]