"""History store: rollups, retention, and the honesty of an empty chart. The rollup queries are plain SQL with positional bindings, and a mismatch there used to raise on every maintenance tick while the poller swallowed it — so every window past the raw retention was silently empty. These tests fail loudly if that ever happens again. """ from __future__ import annotations import time from typing import Any import pytest from dashboards.store import ( HOUR, MINUTE, RES_HOUR, RES_MIN, RES_RAW, Event, Sample, SampleStore, samples_from_flat, select_window, ) def add_series(db, source: str, host: str, metric: str, values, *, start: float, step: float, res: str = RES_RAW) -> None: for i, value in enumerate(values): # the private handle is the point: writing at an explicit resolution is what # the resolution-choice tests need db._db.execute( "INSERT INTO sample (source, host, metric, res, ts, value) VALUES (?,?,?,?,?,?)", (source, host, metric, res, start + i * step, float(value)), ) db._db.commit() def test_store_starts_in_wal_mode(db) -> None: """Readers poll while the writer commits; without WAL the API would hit locked errors.""" assert db._db.execute("PRAGMA journal_mode").fetchone()[0].lower() == "wal" def test_add_and_query_roundtrip(db) -> None: now = time.time() db.add("host", "driveripper", "cpu.usage_pct", 42.5, ts=now) points = db.query("host", "driveripper", "cpu.usage_pct", start=now - 60) assert points == [(now, 42.5)] def test_query_is_bounded_and_ascending(db) -> None: base = 1_700_000_000.0 add_series(db, "host", "h", "m", [3, 1, 2], start=base, step=10) assert db.query("host", "h", "m", start=base - 1) == [(base, 3.0), (base + 10, 1.0), (base + 20, 2.0)] assert len(db.query("host", "h", "m", start=base - 1, limit=2)) == 2 def test_query_tail_keeps_the_newest_rows(db) -> None: base = 1_700_000_000.0 add_series(db, "host", "h", "m", range(100), start=base, step=10) head = db.query("host", "h", "m", start=base - 1, limit=3) tail = db.query("host", "h", "m", start=base - 1, limit=3, tail=True) assert [v for _, v in head] == [0.0, 1.0, 2.0] assert [v for _, v in tail] == [97.0, 98.0, 99.0] assert tail == sorted(tail), "tail must still come back oldest-first" def test_samples_from_flat_drops_unknown_and_none() -> None: payload: dict[str, Any] = {"a": 1.0, "b": None, "c": "nope"} samples = samples_from_flat("host", "gw", payload, ts=5.0) assert [(s.metric, s.value) for s in samples] == [("a", 1.0)] def test_maintain_rolls_raw_into_minutes(db) -> None: """Regression: the rollup statement had 7 placeholders and 6 bindings.""" now = time.time() # two full, aligned minute buckets — see the hour test for why alignment matters minute = int(now // MINUTE) * MINUTE add_series(db, "host", "h", "cpu.usage_pct", [10, 20, 30, 40], start=minute - 4 * MINUTE, step=1) add_series(db, "host", "h", "cpu.usage_pct", [1, 1, 1, 1], start=minute - 3 * MINUTE, step=1) counts = db.maintain(now=now) assert counts["to_minute"] > 0 minutes = db.query("host", "h", "cpu.usage_pct", res=RES_MIN, start=now - 3600) assert minutes, "no 1m rollup was written" assert [round(value) for _ts, value in minutes] == [25, 1] def test_maintain_is_idempotent(db) -> None: """Rollups run on a timer: running twice must not double-count or duplicate rows.""" now = time.time() minute = int(now // MINUTE) * MINUTE add_series(db, "host", "h", "m", [10, 20], start=minute - 4 * MINUTE, step=1) first = db.maintain(now=now) rows = db.query("host", "h", "m", res=RES_MIN, start=now - 3600) count = db.count() second = db.maintain(now=now) assert rows == db.query("host", "h", "m", res=RES_MIN, start=now - 3600) assert first["to_minute"] == second["to_minute"] assert db.count() == count, "re-running maintenance duplicated samples" def test_maintain_rolls_minutes_into_hours(db) -> None: now = time.time() # buckets are aligned to the epoch (`ts/3600*3600`), not to "hours from now", so the # fixtures start on a boundary or the two hours mix into a third, averaged bucket hour = int(now // HOUR) * HOUR # two *complete* hours of minute buckets: the current and previous hour are still # filling in, and maintenance must not touch them (see the partial-bucket test) add_series(db, "host", "h", "m", [1] * 60, start=hour - 5 * HOUR, step=60, res=RES_MIN) add_series(db, "host", "h", "m", [3] * 60, start=hour - 4 * HOUR, step=60, res=RES_MIN) counts = db.maintain(now=now) assert counts["to_hour"] > 0 hours = db.query("host", "h", "m", res=RES_HOUR, start=now - 86_400) assert sorted(round(v) for _, v in hours) == [1.0, 3.0] def test_maintain_prunes_expired_raw(db) -> None: now = time.time() old = now - (db.raw_keep_seconds + 7200) add_series(db, "host", "h", "m", [1, 2], start=old, step=60) add_series(db, "host", "h", "fresh", [1], start=now - 60, step=1) counts = db.maintain(now=now) assert counts["pruned"] >= 2 assert db.query("host", "h", "m", start=0) == [] def test_maintain_records_timestamp_in_stats(db) -> None: db.maintain() stats = db.stats() assert stats["last_maintain"] is not None assert "samples_by_res" in stats and "path" in stats def test_pick_res_prefers_the_finest_resolution_that_fits(db) -> None: now = time.time() add_series(db, "host", "h", "m", list(range(500)), start=now - 3600, step=1) assert db.pick_res("host", "h", "m", 3600, 600) == RES_RAW add_series(db, "host", "h", "m", list(range(4000)), start=now - 7200, step=1) assert db.pick_res("host", "h", "m", 7200, 600) == RES_RAW, "raw still fits" add_series(db, "host", "h", "m", list(range(120)), start=now - 7200, step=60, res=RES_MIN) # 4500 raw points blow the 600 budget, so the minute rollup takes over assert db.pick_res("host", "h", "m", 7200, 600) == RES_MIN def test_pick_res_never_picks_an_empty_resolution(db) -> None: """A fresh install has raw only: choosing 1h because it 'fits' would chart nothing.""" now = time.time() add_series(db, "host", "h", "m", [1] * 2000, start=now - 7200, step=1) assert db.pick_res("host", "h", "m", 86_400, 600) == RES_RAW def test_history_falls_back_to_raw_when_rollups_lag(db) -> None: now = time.time() add_series(db, "host", "h", "m", list(range(1500)), start=now - 1500, step=1) points = db.history("host", "h", "m", span=3600, max_points=600) assert len(points) > 0, "empty chart while raw data exists" assert len(points) <= 600 def test_history_always_keeps_the_newest_point(db) -> None: now = time.time() values = list(range(2000)) add_series(db, "host", "h", "m", values, start=now - 2000, step=1) points = db.history("host", "h", "m", span=3600, max_points=50) assert points[-1][1] == pytest.approx(values[-1]) def test_series_names_and_hosts(db) -> None: db.add("host", "a", "cpu.usage_pct", 1.0) db.add("host", "a", "mem.used_pct", 2.0) db.add("host", "b", "cpu.usage_pct", 3.0) db.add("llm", "gufo", "llm.decode_tokens_per_second", 4.0) assert db.series_names("host", "a") == ["cpu.usage_pct", "mem.used_pct"] assert db.hosts("host") == ["a", "b"] assert db.series_names("llm") == ["llm.decode_tokens_per_second"] def test_delete_series_cleans_up(db) -> None: db.add("host", "a", "gone.metric", 1.0) db.add("host", "a", "keep.metric", 2.0) assert db.delete_series("host", "a", "gone.metric") == 1 assert db.series_names("host", "a") == ["keep.metric"] def test_events_roundtrip_and_filter(db) -> None: now = time.time() db.log_event(Event(ts=now, source="host", host="a", kind="down", severity="error", message="a is not answering", data={"fails": 3})) db.log_event(Event(ts=now + 1, source="nut", host="cyberpower", kind="state", severity="warn", message="OL → OB")) everything = db.recent_events(10) assert [e.host for e in everything] == ["cyberpower", "a"] assert everything[-1].data == {"fails": 3} assert [e.source for e in db.recent_events(10, source="nut")] == ["nut"] assert len(db.recent_events(1)) == 1 # the API hands out plain JSON, so the boundary has to be serialisable assert everything[0].as_json()["message"] == "OL → OB" def test_event_as_json_is_serialisable(db) -> None: event = Event(ts=1.0, source="host", host="a", kind="up", severity="info", message="ok") payload = event.as_json() assert payload["severity"] == "info" and payload["data"] is None def test_count_filters_by_resolution(db) -> None: now = time.time() db.add("host", "h", "m", 1.0, ts=now) add_series(db, "host", "h", "m", [2.0], start=now, step=1, res=RES_MIN) assert db.count() == 2 assert db.count(res=RES_RAW) == 1 assert db.count(res=RES_HOUR) == 0 def test_add_many_reports_inserted_rows(db) -> None: now = time.time() rows = [Sample("host", "h", f"m{i}", float(i), now) for i in range(5)] assert db.add_many(rows) == 5 assert db.count() == 5 def test_store_reopens_without_losing_history(tmp_path) -> None: path = tmp_path / "reopen.sqlite3" first = SampleStore(path) now = time.time() first.add("host", "h", "m", 7.0, ts=now) first.close() second = SampleStore(path) try: assert second.query("host", "h", "m", start=now - 5) == [(now, 7.0)] finally: second.close() def test_select_window_maps_range_onto_presets() -> None: assert select_window(60, [300, 3600, 86_400]) == 300 assert select_window(3600, [300, 3600, 86_400]) == 3600 assert select_window(86_401, [300, 3600, 86_400]) == 86_400 def test_rollup_does_not_average_a_partial_minute(db) -> None: """A bucket is only rolled once its whole window is in the past, or it gets rewritten.""" now = time.time() add_series(db, "host", "h", "m", [1, 1, 1], start=now - 5, step=1) # current minute db.maintain(now=now) assert db.query("host", "h", "m", res=RES_MIN, start=now - 600) == [] def test_hour_constant_is_used_in_retention(db) -> None: assert db.raw_keep_seconds == pytest.approx(48 * HOUR)