1073 lines
41 KiB
Python
1073 lines
41 KiB
Python
"""Tests for shared status composition (issue #78).
|
|
|
|
Covers:
|
|
- TPH-2: Status lattice with approved glyphs, labels, precedence
|
|
- TPH-3: CLI/status parity, separate facts, timing
|
|
- AC78-1: Shared status composition renders in both surfaces
|
|
- AC78-2: Base precedence Error > Interrupted > Paused > Stale > Waiting > Monitoring > Unknown
|
|
- AC78-3: Collecting overlays every base except store fault
|
|
- AC78-4: Failed service queries → Unknown, not manufactured facts
|
|
- AC78-5: Separate freshness, last outcome, boot enablement, activity facts
|
|
- AC78-6: 5s poll reevaluates freshness from cached clock
|
|
"""
|
|
import sqlite3
|
|
from datetime import datetime, timedelta, timezone
|
|
from pathlib import Path
|
|
from unittest.mock import patch, MagicMock
|
|
|
|
import pytest
|
|
import sys
|
|
|
|
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
|
|
|
|
from fenris.store import init_store
|
|
from fenris.status_composition import (
|
|
StatusState,
|
|
StatusComposition,
|
|
compose_status,
|
|
render_status_cli,
|
|
render_status_tui,
|
|
FRESH_THRESHOLD_S,
|
|
STALENESS_THRESHOLD_S,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _clock(year=2026, month=9, day=30, hour=12):
|
|
return datetime(year, month, day, hour, 0, 0, tzinfo=timezone.utc)
|
|
|
|
|
|
def _insert_sample(conn, ts, device="/dev/nvme0n1"):
|
|
"""Insert a minimal sample row."""
|
|
conn.execute(
|
|
"INSERT INTO samples (ts, device, data_units_written, data_units_read, "
|
|
"percentage_used, bytes_written, bytes_read, power_on_hours) "
|
|
"VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
|
|
(ts, device, 1000000, 500000, 5, 512000000000, 256000000000, 8765),
|
|
)
|
|
conn.commit()
|
|
|
|
|
|
def _insert_segment(conn, opened_at="2026-09-01T00:00:00+00:00",
|
|
identity_key="nqn.test", degraded=False,
|
|
mn="Samsung SSD 970 EVO Plus 1TB"):
|
|
conn.execute(
|
|
"INSERT INTO controller_segments "
|
|
"(opened_at, identity_key, identity_degraded, subnqn, sn, mn, fr, vid, ssvid, transport) "
|
|
"VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
|
|
(opened_at, identity_key, degraded, "nqn.test", "SN123", mn, "FW1",
|
|
"0x144d", "0x144d", "pcie"),
|
|
)
|
|
conn.commit()
|
|
|
|
|
|
def _insert_day(conn, day, bw=1024*1024*100, coverage=0.95, samples=24):
|
|
conn.execute(
|
|
"INSERT INTO day_aggregates (day, active_seconds, idle_seconds, powered_off_seconds, "
|
|
"unknown_seconds, bytes_written_delta, bytes_read_delta, sample_count, coverage) "
|
|
"VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
|
|
(day, 3600, 0, 0, 0, bw, 0, samples, coverage),
|
|
)
|
|
conn.commit()
|
|
|
|
|
|
def _open_period(conn, start="2026-09-01T00:00:00+00:00"):
|
|
from fenris.monitoring_periods import ensure_period_open
|
|
ensure_period_open(conn, datetime.fromisoformat(start))
|
|
|
|
|
|
def _close_period(conn, started_at="2026-09-01T09:00:00+00:00",
|
|
ended_at="2026-09-01T10:00:00+00:00", cause="user_disabled"):
|
|
"""Insert a closed monitoring period directly."""
|
|
conn.execute(
|
|
"INSERT INTO monitoring_periods (started_at, ended_at, end_cause) "
|
|
"VALUES (?, ?, ?)",
|
|
(started_at, ended_at, cause),
|
|
)
|
|
conn.commit()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# StatusState enum tests (TPH-2)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestStatusState:
|
|
"""StatusState has exactly 8 states with correct properties."""
|
|
|
|
def test_all_states_exist(self):
|
|
expected = {
|
|
"monitoring", "collecting", "paused", "waiting",
|
|
"interrupted", "error", "stale", "unknown",
|
|
}
|
|
actual = {s.value for s in StatusState}
|
|
assert actual == expected
|
|
|
|
def test_monitoring_glyph(self):
|
|
assert StatusState.MONITORING.glyph == "●"
|
|
|
|
def test_collecting_glyph(self):
|
|
assert StatusState.COLLECTING.glyph == "◐"
|
|
|
|
def test_paused_glyph(self):
|
|
assert StatusState.PAUSED.glyph == "‖"
|
|
|
|
def test_waiting_glyph(self):
|
|
assert StatusState.WAITING.glyph == "○"
|
|
|
|
def test_interrupted_glyph(self):
|
|
assert StatusState.INTERRUPTED.glyph == "⊘"
|
|
|
|
def test_error_glyph(self):
|
|
assert StatusState.ERROR.glyph == "✖"
|
|
|
|
def test_stale_glyph(self):
|
|
assert StatusState.STALE.glyph == "◌"
|
|
|
|
def test_unknown_glyph(self):
|
|
assert StatusState.UNKNOWN.glyph == "?"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Precedence tests (TPH-2, AC78-2)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestPrecedence:
|
|
"""Base precedence is Error > Interrupted > Paused > Stale > Waiting > Monitoring > Unknown."""
|
|
|
|
def test_error_beats_all(self, tmp_path):
|
|
"""Error outranks every other base state."""
|
|
conn = init_store(tmp_path / "test.db")
|
|
_insert_segment(conn)
|
|
_open_period(conn)
|
|
# Active timer + fresh sample + error condition
|
|
now = _clock()
|
|
ts = (now - timedelta(minutes=2)).isoformat()
|
|
_insert_sample(conn, ts)
|
|
for i in range(20):
|
|
d = (datetime(2026, 9, 10) + timedelta(days=i)).strftime("%Y-%m-%d")
|
|
_insert_day(conn, d, bw=1024*1024*100)
|
|
|
|
svc = {
|
|
"boot_enabled": True, "timer_active": True,
|
|
"last_collect_ok": False, "last_collect_age_s": 60,
|
|
"last_collect_reason": "exit code 3",
|
|
}
|
|
comp = compose_status(conn, svc, now, store_fault=None, newer_schema=None)
|
|
assert comp.state == StatusState.ERROR
|
|
conn.close()
|
|
|
|
def test_interrupted_beats_paused_stale_waiting(self, tmp_path):
|
|
"""Interrupted outranks Paused, Stale, Waiting.
|
|
|
|
External stop = timer inactive + boot disabled + period still open
|
|
(not closed by user_disabled).
|
|
"""
|
|
conn = init_store(tmp_path / "test.db")
|
|
_insert_segment(conn)
|
|
# Open a monitoring period that stays open (external stop didn't close it)
|
|
_open_period(conn, start="2026-09-28T10:00:00+00:00")
|
|
now = _clock()
|
|
ts = (now - timedelta(days=3)).isoformat()
|
|
_insert_sample(conn, ts)
|
|
|
|
svc = {
|
|
"boot_enabled": False, "timer_active": False,
|
|
"last_collect_ok": None, "last_collect_age_s": None,
|
|
"last_collect_reason": None,
|
|
}
|
|
comp = compose_status(conn, svc, now, store_fault=None, newer_schema=None)
|
|
assert comp.state == StatusState.INTERRUPTED
|
|
conn.close()
|
|
|
|
def test_paused_beats_stale(self, tmp_path):
|
|
"""Paused outranks Stale — a drive paused for 49 days is Paused, not Stale."""
|
|
conn = init_store(tmp_path / "test.db")
|
|
_insert_segment(conn)
|
|
_close_period(conn, started_at="2026-09-28T10:00:00+00:00",
|
|
ended_at="2026-09-29T10:00:00+00:00", cause="user_disabled")
|
|
now = _clock()
|
|
# Stale sample (49 days old)
|
|
ts = (now - timedelta(days=49)).isoformat()
|
|
_insert_sample(conn, ts)
|
|
|
|
svc = {
|
|
"boot_enabled": False, "timer_active": False,
|
|
"last_collect_ok": None, "last_collect_age_s": None,
|
|
"last_collect_reason": None,
|
|
}
|
|
comp = compose_status(conn, svc, now, store_fault=None, newer_schema=None)
|
|
assert comp.state == StatusState.PAUSED
|
|
conn.close()
|
|
|
|
def test_stale_beats_waiting(self, tmp_path):
|
|
"""Stale outranks Waiting."""
|
|
conn = init_store(tmp_path / "test.db")
|
|
_insert_segment(conn)
|
|
_open_period(conn)
|
|
now = _clock()
|
|
ts = (now - timedelta(hours=49)).isoformat()
|
|
_insert_sample(conn, ts)
|
|
|
|
svc = {
|
|
"boot_enabled": True, "timer_active": True,
|
|
"last_collect_ok": True, "last_collect_age_s": 300,
|
|
"last_collect_reason": None,
|
|
}
|
|
comp = compose_status(conn, svc, now, store_fault=None, newer_schema=None)
|
|
assert comp.state == StatusState.STALE
|
|
conn.close()
|
|
|
|
def test_waiting_beats_monitoring(self, tmp_path):
|
|
"""Waiting outranks Monitoring (fresh but awaiting data)."""
|
|
conn = init_store(tmp_path / "test.db")
|
|
_insert_segment(conn)
|
|
_open_period(conn)
|
|
now = _clock()
|
|
ts = (now - timedelta(minutes=2)).isoformat()
|
|
_insert_sample(conn, ts)
|
|
|
|
svc = {
|
|
"boot_enabled": True, "timer_active": True,
|
|
"last_collect_ok": True, "last_collect_age_s": 120,
|
|
"last_collect_reason": None,
|
|
}
|
|
# Only 1 sample → Waiting
|
|
comp = compose_status(conn, svc, now, store_fault=None, newer_schema=None)
|
|
assert comp.state == StatusState.WAITING
|
|
conn.close()
|
|
|
|
def test_monitoring_is_lowest_active(self, tmp_path):
|
|
"""Monitoring is the lowest active state."""
|
|
conn = init_store(tmp_path / "test.db")
|
|
_insert_segment(conn)
|
|
_open_period(conn)
|
|
now = _clock()
|
|
ts = (now - timedelta(minutes=2)).isoformat()
|
|
_insert_sample(conn, ts)
|
|
for i in range(20):
|
|
d = (datetime(2026, 9, 10) + timedelta(days=i)).strftime("%Y-%m-%d")
|
|
_insert_day(conn, d, bw=1024*1024*100)
|
|
|
|
svc = {
|
|
"boot_enabled": True, "timer_active": True,
|
|
"last_collect_ok": True, "last_collect_age_s": 120,
|
|
"last_collect_reason": None,
|
|
}
|
|
comp = compose_status(conn, svc, now, store_fault=None, newer_schema=None)
|
|
assert comp.state == StatusState.MONITORING
|
|
conn.close()
|
|
|
|
def test_unknown_is_lowest(self, tmp_path):
|
|
"""Unknown is the lowest state, used only when service query fails."""
|
|
conn = init_store(tmp_path / "test.db")
|
|
now = _clock()
|
|
|
|
# Service query failed → no facts available
|
|
svc = None
|
|
comp = compose_status(conn, svc, now, store_fault=None, newer_schema=None)
|
|
assert comp.state == StatusState.UNKNOWN
|
|
conn.close()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Collecting overlay tests (TPH-2, AC78-3)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestCollectingOverlay:
|
|
"""Collecting overlays every base except store fault."""
|
|
|
|
def test_collecting_overlays_monitoring(self, tmp_path):
|
|
"""Collecting overlays Monitoring."""
|
|
conn = init_store(tmp_path / "test.db")
|
|
_insert_segment(conn)
|
|
_open_period(conn)
|
|
now = _clock()
|
|
ts = (now - timedelta(minutes=2)).isoformat()
|
|
_insert_sample(conn, ts)
|
|
for i in range(20):
|
|
d = (datetime(2026, 9, 10) + timedelta(days=i)).strftime("%Y-%m-%d")
|
|
_insert_day(conn, d, bw=1024*1024*100)
|
|
|
|
svc = {
|
|
"boot_enabled": True, "timer_active": True,
|
|
"last_collect_ok": True, "last_collect_age_s": 120,
|
|
"last_collect_reason": None,
|
|
}
|
|
comp = compose_status(conn, svc, now, store_fault=None, newer_schema=None,
|
|
collecting=True)
|
|
assert comp.state == StatusState.COLLECTING
|
|
assert comp.overlay_base == StatusState.MONITORING
|
|
conn.close()
|
|
|
|
def test_collecting_overlays_paused(self, tmp_path):
|
|
"""Collecting overlays Paused, naming the underlying state."""
|
|
conn = init_store(tmp_path / "test.db")
|
|
_insert_segment(conn)
|
|
_close_period(conn, started_at="2026-09-28T10:00:00+00:00",
|
|
ended_at="2026-09-29T10:00:00+00:00", cause="user_disabled")
|
|
now = _clock()
|
|
ts = (now - timedelta(minutes=2)).isoformat()
|
|
_insert_sample(conn, ts)
|
|
|
|
svc = {
|
|
"boot_enabled": False, "timer_active": False,
|
|
"last_collect_ok": None, "last_collect_age_s": None,
|
|
"last_collect_reason": None,
|
|
}
|
|
comp = compose_status(conn, svc, now, store_fault=None, newer_schema=None,
|
|
collecting=True)
|
|
assert comp.state == StatusState.COLLECTING
|
|
assert comp.overlay_base == StatusState.PAUSED
|
|
assert "paused" in comp.explanation.lower()
|
|
conn.close()
|
|
|
|
def test_collecting_overlays_error(self, tmp_path):
|
|
"""Collecting overlays Error with retry naming."""
|
|
conn = init_store(tmp_path / "test.db")
|
|
_insert_segment(conn)
|
|
_open_period(conn)
|
|
now = _clock()
|
|
ts = (now - timedelta(minutes=2)).isoformat()
|
|
_insert_sample(conn, ts)
|
|
for i in range(20):
|
|
d = (datetime(2026, 9, 10) + timedelta(days=i)).strftime("%Y-%m-%d")
|
|
_insert_day(conn, d, bw=1024*1024*100)
|
|
|
|
svc = {
|
|
"boot_enabled": True, "timer_active": True,
|
|
"last_collect_ok": False, "last_collect_age_s": 60,
|
|
"last_collect_reason": "exit code 3",
|
|
}
|
|
comp = compose_status(conn, svc, now, store_fault=None, newer_schema=None,
|
|
collecting=True)
|
|
assert comp.state == StatusState.COLLECTING
|
|
assert comp.overlay_base == StatusState.ERROR
|
|
assert "retry" in comp.explanation.lower()
|
|
conn.close()
|
|
|
|
def test_collecting_does_not_overlay_store_fault(self, tmp_path):
|
|
"""Collecting does NOT overlay a store fault."""
|
|
conn = init_store(tmp_path / "test.db")
|
|
now = _clock()
|
|
|
|
svc = {
|
|
"boot_enabled": True, "timer_active": True,
|
|
"last_collect_ok": True, "last_collect_age_s": 120,
|
|
"last_collect_reason": None,
|
|
}
|
|
comp = compose_status(conn, svc, now, store_fault="observation store unreadable",
|
|
newer_schema=None, collecting=True)
|
|
assert comp.state == StatusState.ERROR
|
|
assert "store" in comp.explanation.lower()
|
|
conn.close()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Separate facts tests (TPH-3, AC78-5)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestSeparateFacts:
|
|
"""Freshness, last outcome, boot enablement, and activity are separate facts."""
|
|
|
|
def test_fresh_data_shows_fresh_fact(self, tmp_path):
|
|
conn = init_store(tmp_path / "test.db")
|
|
_insert_segment(conn)
|
|
_open_period(conn)
|
|
now = _clock()
|
|
ts = (now - timedelta(minutes=2)).isoformat()
|
|
_insert_sample(conn, ts)
|
|
for i in range(20):
|
|
d = (datetime(2026, 9, 10) + timedelta(days=i)).strftime("%Y-%m-%d")
|
|
_insert_day(conn, d, bw=1024*1024*100)
|
|
|
|
svc = {
|
|
"boot_enabled": True, "timer_active": True,
|
|
"last_collect_ok": True, "last_collect_age_s": 120,
|
|
"last_collect_reason": None,
|
|
}
|
|
comp = compose_status(conn, svc, now, store_fault=None, newer_schema=None)
|
|
assert comp.freshness == "fresh"
|
|
assert comp.last_collect_ok is True
|
|
assert comp.boot_enabled is True
|
|
assert comp.timer_active is True
|
|
conn.close()
|
|
|
|
def test_stale_data_shows_stale_fact(self, tmp_path):
|
|
conn = init_store(tmp_path / "test.db")
|
|
_insert_segment(conn)
|
|
_open_period(conn)
|
|
now = _clock()
|
|
ts = (now - timedelta(hours=49)).isoformat()
|
|
_insert_sample(conn, ts)
|
|
|
|
svc = {
|
|
"boot_enabled": True, "timer_active": True,
|
|
"last_collect_ok": True, "last_collect_age_s": 176400,
|
|
"last_collect_reason": None,
|
|
}
|
|
comp = compose_status(conn, svc, now, store_fault=None, newer_schema=None)
|
|
assert comp.freshness == "stale"
|
|
assert comp.freshness_age_s is not None
|
|
assert comp.freshness_age_s > 48 * 3600
|
|
conn.close()
|
|
|
|
def test_boot_disabled_is_separate_fact(self, tmp_path):
|
|
conn = init_store(tmp_path / "test.db")
|
|
_insert_segment(conn)
|
|
_open_period(conn)
|
|
now = _clock()
|
|
ts = (now - timedelta(minutes=2)).isoformat()
|
|
_insert_sample(conn, ts)
|
|
|
|
svc = {
|
|
"boot_enabled": False, "timer_active": True,
|
|
"last_collect_ok": True, "last_collect_age_s": 120,
|
|
"last_collect_reason": None,
|
|
}
|
|
comp = compose_status(conn, svc, now, store_fault=None, newer_schema=None)
|
|
assert comp.boot_enabled is False
|
|
conn.close()
|
|
|
|
def test_pause_age_is_separate_fact(self, tmp_path):
|
|
conn = init_store(tmp_path / "test.db")
|
|
_insert_segment(conn)
|
|
_close_period(conn, started_at="2026-09-28T10:00:00+00:00",
|
|
ended_at="2026-09-29T10:00:00+00:00", cause="user_disabled")
|
|
now = _clock()
|
|
ts = (now - timedelta(hours=49)).isoformat()
|
|
_insert_sample(conn, ts)
|
|
|
|
svc = {
|
|
"boot_enabled": False, "timer_active": False,
|
|
"last_collect_ok": None, "last_collect_age_s": None,
|
|
"last_collect_reason": None,
|
|
}
|
|
comp = compose_status(conn, svc, now, store_fault=None, newer_schema=None)
|
|
assert comp.deliberately_paused is True
|
|
assert comp.state == StatusState.PAUSED
|
|
conn.close()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Unknown from failed service query (AC78-4)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestUnknownFromFailedQuery:
|
|
"""Failed service queries produce Unknown, not manufactured facts."""
|
|
|
|
def test_no_service_query_gives_unknown(self, tmp_path):
|
|
conn = init_store(tmp_path / "test.db")
|
|
_insert_segment(conn)
|
|
_open_period(conn)
|
|
now = _clock()
|
|
ts = (now - timedelta(minutes=2)).isoformat()
|
|
_insert_sample(conn, ts)
|
|
|
|
comp = compose_status(conn, None, now, store_fault=None, newer_schema=None)
|
|
assert comp.state == StatusState.UNKNOWN
|
|
# No manufactured facts
|
|
assert comp.boot_enabled is None
|
|
assert comp.timer_active is None
|
|
conn.close()
|
|
|
|
def test_empty_service_dict_gives_unknown(self, tmp_path):
|
|
conn = init_store(tmp_path / "test.db")
|
|
_insert_segment(conn)
|
|
_open_period(conn)
|
|
now = _clock()
|
|
ts = (now - timedelta(minutes=2)).isoformat()
|
|
_insert_sample(conn, ts)
|
|
|
|
comp = compose_status(conn, {}, now, store_fault=None, newer_schema=None)
|
|
assert comp.state == StatusState.UNKNOWN
|
|
conn.close()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# CLI rendering tests (AC78-1, TPH-3)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestCLIStatusRendering:
|
|
"""CLI renders the same status vocabulary statically."""
|
|
|
|
def test_cli_monitoring_state(self, tmp_path):
|
|
conn = init_store(tmp_path / "test.db")
|
|
_insert_segment(conn)
|
|
_open_period(conn)
|
|
now = _clock()
|
|
ts = (now - timedelta(minutes=2)).isoformat()
|
|
_insert_sample(conn, ts)
|
|
for i in range(20):
|
|
d = (datetime(2026, 9, 10) + timedelta(days=i)).strftime("%Y-%m-%d")
|
|
_insert_day(conn, d, bw=1024*1024*100)
|
|
|
|
svc = {
|
|
"boot_enabled": True, "timer_active": True,
|
|
"last_collect_ok": True, "last_collect_age_s": 120,
|
|
"last_collect_reason": None,
|
|
}
|
|
comp = compose_status(conn, svc, now, store_fault=None, newer_schema=None)
|
|
cli_text = render_status_cli(comp)
|
|
assert "● Monitoring" in cli_text
|
|
assert comp.state == StatusState.MONITORING
|
|
conn.close()
|
|
|
|
def test_cli_error_state(self, tmp_path):
|
|
conn = init_store(tmp_path / "test.db")
|
|
_insert_segment(conn)
|
|
_open_period(conn)
|
|
now = _clock()
|
|
ts = (now - timedelta(minutes=2)).isoformat()
|
|
_insert_sample(conn, ts)
|
|
for i in range(20):
|
|
d = (datetime(2026, 9, 10) + timedelta(days=i)).strftime("%Y-%m-%d")
|
|
_insert_day(conn, d, bw=1024*1024*100)
|
|
|
|
svc = {
|
|
"boot_enabled": True, "timer_active": True,
|
|
"last_collect_ok": False, "last_collect_age_s": 60,
|
|
"last_collect_reason": "exit code 3",
|
|
}
|
|
comp = compose_status(conn, svc, now, store_fault=None, newer_schema=None)
|
|
cli_text = render_status_cli(comp)
|
|
assert "✖ Error" in cli_text
|
|
assert "exit code 3" in cli_text
|
|
conn.close()
|
|
|
|
def test_cli_paused_state(self, tmp_path):
|
|
conn = init_store(tmp_path / "test.db")
|
|
_insert_segment(conn)
|
|
_close_period(conn, started_at="2026-09-28T10:00:00+00:00",
|
|
ended_at="2026-09-29T10:00:00+00:00", cause="user_disabled")
|
|
now = _clock()
|
|
ts = (now - timedelta(minutes=2)).isoformat()
|
|
_insert_sample(conn, ts)
|
|
|
|
svc = {
|
|
"boot_enabled": False, "timer_active": False,
|
|
"last_collect_ok": None, "last_collect_age_s": None,
|
|
"last_collect_reason": None,
|
|
}
|
|
comp = compose_status(conn, svc, now, store_fault=None, newer_schema=None)
|
|
cli_text = render_status_cli(comp)
|
|
assert "‖ Paused" in cli_text
|
|
assert "monitoring paused" in cli_text.lower()
|
|
conn.close()
|
|
|
|
def test_cli_no_tui_only_strings(self, tmp_path):
|
|
"""CLI never renders TUI-only identity or preferences."""
|
|
conn = init_store(tmp_path / "test.db")
|
|
now = _clock()
|
|
|
|
svc = {
|
|
"boot_enabled": True, "timer_active": True,
|
|
"last_collect_ok": True, "last_collect_age_s": 120,
|
|
"last_collect_reason": None,
|
|
}
|
|
comp = compose_status(conn, svc, now, store_fault=None, newer_schema=None)
|
|
cli_text = render_status_cli(comp)
|
|
assert "Fenris — NVMe endurance monitor" not in cli_text
|
|
assert "by Bongbetic" not in cli_text
|
|
assert "privileged actions" not in cli_text
|
|
conn.close()
|
|
|
|
def test_cli_store_fault(self, tmp_path):
|
|
conn = init_store(tmp_path / "test.db")
|
|
now = _clock()
|
|
|
|
svc = {
|
|
"boot_enabled": True, "timer_active": True,
|
|
"last_collect_ok": True, "last_collect_age_s": 120,
|
|
"last_collect_reason": None,
|
|
}
|
|
comp = compose_status(conn, svc, now, store_fault="observation store unreadable",
|
|
newer_schema=None)
|
|
cli_text = render_status_cli(comp)
|
|
assert "observation store unreadable" in cli_text
|
|
assert "see journal" in cli_text.lower()
|
|
conn.close()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# TUI rendering tests (AC78-1, TPH-2)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestTUIStatusRendering:
|
|
"""TUI renders the same status vocabulary with styling."""
|
|
|
|
def test_tui_monitoring_state(self, tmp_path):
|
|
conn = init_store(tmp_path / "test.db")
|
|
_insert_segment(conn)
|
|
_open_period(conn)
|
|
now = _clock()
|
|
ts = (now - timedelta(minutes=2)).isoformat()
|
|
_insert_sample(conn, ts)
|
|
for i in range(20):
|
|
d = (datetime(2026, 9, 10) + timedelta(days=i)).strftime("%Y-%m-%d")
|
|
_insert_day(conn, d, bw=1024*1024*100)
|
|
|
|
svc = {
|
|
"boot_enabled": True, "timer_active": True,
|
|
"last_collect_ok": True, "last_collect_age_s": 120,
|
|
"last_collect_reason": None,
|
|
}
|
|
comp = compose_status(conn, svc, now, store_fault=None, newer_schema=None)
|
|
tui_text = render_status_tui(comp)
|
|
assert "● Monitoring" in tui_text
|
|
conn.close()
|
|
|
|
def test_tui_collecting_overlay(self, tmp_path):
|
|
conn = init_store(tmp_path / "test.db")
|
|
_insert_segment(conn)
|
|
_open_period(conn)
|
|
now = _clock()
|
|
ts = (now - timedelta(minutes=2)).isoformat()
|
|
_insert_sample(conn, ts)
|
|
for i in range(20):
|
|
d = (datetime(2026, 9, 10) + timedelta(days=i)).strftime("%Y-%m-%d")
|
|
_insert_day(conn, d, bw=1024*1024*100)
|
|
|
|
svc = {
|
|
"boot_enabled": True, "timer_active": True,
|
|
"last_collect_ok": True, "last_collect_age_s": 120,
|
|
"last_collect_reason": None,
|
|
}
|
|
comp = compose_status(conn, svc, now, store_fault=None, newer_schema=None,
|
|
collecting=True)
|
|
tui_text = render_status_tui(comp)
|
|
assert "◐ Collecting" in tui_text
|
|
conn.close()
|
|
|
|
def test_tui_store_fault(self, tmp_path):
|
|
conn = init_store(tmp_path / "test.db")
|
|
now = _clock()
|
|
|
|
svc = {
|
|
"boot_enabled": True, "timer_active": True,
|
|
"last_collect_ok": True, "last_collect_age_s": 120,
|
|
"last_collect_reason": None,
|
|
}
|
|
comp = compose_status(conn, svc, now, store_fault="observation store unreadable",
|
|
newer_schema=None)
|
|
tui_text = render_status_tui(comp)
|
|
assert "observation store unreadable" in tui_text
|
|
conn.close()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Store fault suppresses store-dependent views (TPH-3)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestStoreFaultSuppression:
|
|
def test_store_fault_suppresses_projection(self, tmp_path):
|
|
conn = init_store(tmp_path / "test.db")
|
|
now = _clock()
|
|
|
|
svc = {
|
|
"boot_enabled": True, "timer_active": True,
|
|
"last_collect_ok": True, "last_collect_age_s": 120,
|
|
"last_collect_reason": None,
|
|
}
|
|
comp = compose_status(conn, svc, now, store_fault="observation store unreadable",
|
|
newer_schema=None)
|
|
# Store fault should suppress projection
|
|
assert comp.state == StatusState.ERROR
|
|
assert comp.store_fault is not None
|
|
# CLI rendering should show store fault
|
|
cli_text = render_status_cli(comp)
|
|
assert "observation store unreadable" in cli_text
|
|
conn.close()
|
|
|
|
def test_newer_schema_suppresses_projection(self, tmp_path):
|
|
import fenris.store as store_mod
|
|
conn = init_store(tmp_path / "test.db")
|
|
now = _clock()
|
|
|
|
svc = {
|
|
"boot_enabled": True, "timer_active": True,
|
|
"last_collect_ok": True, "last_collect_age_s": 120,
|
|
"last_collect_reason": None,
|
|
}
|
|
comp = compose_status(conn, svc, now, store_fault=None,
|
|
newer_schema="schema version 99")
|
|
assert comp.state == StatusState.ERROR
|
|
assert comp.newer_schema is not None
|
|
cli_text = render_status_cli(comp)
|
|
assert "newer Fenris" in cli_text
|
|
conn.close()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Empty and single-sample evidence (AC78-5)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestEmptyAndSingleSample:
|
|
def test_empty_store_gives_waiting(self, tmp_path):
|
|
conn = init_store(tmp_path / "test.db")
|
|
now = _clock()
|
|
|
|
svc = {
|
|
"boot_enabled": True, "timer_active": True,
|
|
"last_collect_ok": True, "last_collect_age_s": 120,
|
|
"last_collect_reason": None,
|
|
}
|
|
comp = compose_status(conn, svc, now, store_fault=None, newer_schema=None)
|
|
assert comp.state == StatusState.WAITING
|
|
assert "awaiting first sample" in comp.explanation.lower()
|
|
conn.close()
|
|
|
|
def test_single_sample_gives_waiting(self, tmp_path):
|
|
conn = init_store(tmp_path / "test.db")
|
|
_insert_segment(conn)
|
|
_open_period(conn)
|
|
now = _clock()
|
|
ts = (now - timedelta(minutes=2)).isoformat()
|
|
_insert_sample(conn, ts)
|
|
|
|
svc = {
|
|
"boot_enabled": True, "timer_active": True,
|
|
"last_collect_ok": True, "last_collect_age_s": 120,
|
|
"last_collect_reason": None,
|
|
}
|
|
comp = compose_status(conn, svc, now, store_fault=None, newer_schema=None)
|
|
assert comp.state == StatusState.WAITING
|
|
assert "awaiting another sample" in comp.explanation.lower()
|
|
conn.close()
|
|
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Monitoring dot blink animation (TPH-2, AC78-6)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestMonitoringDotBlink:
|
|
"""Only Monitoring dot blinks 750ms on/750ms off. Never text or other states."""
|
|
|
|
def test_monitoring_has_blink_flag(self, tmp_path):
|
|
"""StatusComposition indicates when dot should blink."""
|
|
conn = init_store(tmp_path / "test.db")
|
|
_insert_segment(conn)
|
|
_open_period(conn)
|
|
now = _clock()
|
|
ts = (now - timedelta(minutes=2)).isoformat()
|
|
_insert_sample(conn, ts)
|
|
for i in range(20):
|
|
d = (datetime(2026, 9, 10) + timedelta(days=i)).strftime("%Y-%m-%d")
|
|
_insert_day(conn, d, bw=1024*1024*100)
|
|
|
|
svc = {
|
|
"boot_enabled": True, "timer_active": True,
|
|
"last_collect_ok": True, "last_collect_age_s": 120,
|
|
"last_collect_reason": None,
|
|
}
|
|
comp = compose_status(conn, svc, now, store_fault=None, newer_schema=None)
|
|
assert comp.state == StatusState.MONITORING
|
|
assert comp.should_blink is True
|
|
conn.close()
|
|
|
|
def test_collecting_does_not_blink(self, tmp_path):
|
|
"""Collecting state does not blink."""
|
|
conn = init_store(tmp_path / "test.db")
|
|
_insert_segment(conn)
|
|
_open_period(conn)
|
|
now = _clock()
|
|
ts = (now - timedelta(minutes=2)).isoformat()
|
|
_insert_sample(conn, ts)
|
|
for i in range(20):
|
|
d = (datetime(2026, 9, 10) + timedelta(days=i)).strftime("%Y-%m-%d")
|
|
_insert_day(conn, d, bw=1024*1024*100)
|
|
|
|
svc = {
|
|
"boot_enabled": True, "timer_active": True,
|
|
"last_collect_ok": True, "last_collect_age_s": 120,
|
|
"last_collect_reason": None,
|
|
}
|
|
comp = compose_status(conn, svc, now, store_fault=None, newer_schema=None,
|
|
collecting=True)
|
|
assert comp.state == StatusState.COLLECTING
|
|
assert comp.should_blink is False
|
|
conn.close()
|
|
|
|
def test_paused_does_not_blink(self, tmp_path):
|
|
"""Paused state does not blink."""
|
|
conn = init_store(tmp_path / "test.db")
|
|
_insert_segment(conn)
|
|
_close_period(conn, started_at="2026-09-28T10:00:00+00:00",
|
|
ended_at="2026-09-29T10:00:00+00:00", cause="user_disabled")
|
|
now = _clock()
|
|
ts = (now - timedelta(minutes=2)).isoformat()
|
|
_insert_sample(conn, ts)
|
|
|
|
svc = {
|
|
"boot_enabled": False, "timer_active": False,
|
|
"last_collect_ok": None, "last_collect_age_s": None,
|
|
"last_collect_reason": None,
|
|
}
|
|
comp = compose_status(conn, svc, now, store_fault=None, newer_schema=None)
|
|
assert comp.state == StatusState.PAUSED
|
|
assert comp.should_blink is False
|
|
conn.close()
|
|
|
|
def test_error_does_not_blink(self, tmp_path):
|
|
"""Error state does not blink."""
|
|
conn = init_store(tmp_path / "test.db")
|
|
_insert_segment(conn)
|
|
_open_period(conn)
|
|
now = _clock()
|
|
ts = (now - timedelta(minutes=2)).isoformat()
|
|
_insert_sample(conn, ts)
|
|
for i in range(20):
|
|
d = (datetime(2026, 9, 10) + timedelta(days=i)).strftime("%Y-%m-%d")
|
|
_insert_day(conn, d, bw=1024*1024*100)
|
|
|
|
svc = {
|
|
"boot_enabled": True, "timer_active": True,
|
|
"last_collect_ok": False, "last_collect_age_s": 60,
|
|
"last_collect_reason": "exit code 3",
|
|
}
|
|
comp = compose_status(conn, svc, now, store_fault=None, newer_schema=None)
|
|
assert comp.state == StatusState.ERROR
|
|
assert comp.should_blink is False
|
|
conn.close()
|
|
|
|
def test_reduced_motion_makes_monitoring_steady(self, tmp_path):
|
|
"""Reduced motion disables blink even for Monitoring."""
|
|
conn = init_store(tmp_path / "test.db")
|
|
_insert_segment(conn)
|
|
_open_period(conn)
|
|
now = _clock()
|
|
ts = (now - timedelta(minutes=2)).isoformat()
|
|
_insert_sample(conn, ts)
|
|
for i in range(20):
|
|
d = (datetime(2026, 9, 10) + timedelta(days=i)).strftime("%Y-%m-%d")
|
|
_insert_day(conn, d, bw=1024*1024*100)
|
|
|
|
svc = {
|
|
"boot_enabled": True, "timer_active": True,
|
|
"last_collect_ok": True, "last_collect_age_s": 120,
|
|
"last_collect_reason": None,
|
|
}
|
|
comp = compose_status(conn, svc, now, store_fault=None, newer_schema=None,
|
|
reduced_motion=True)
|
|
assert comp.state == StatusState.MONITORING
|
|
assert comp.should_blink is False
|
|
conn.close()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Refresh timing (TPH-3, AC78-6)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestRefreshTiming:
|
|
"""5s poll reevaluates freshness from cached clock."""
|
|
|
|
def test_freshness_ages_without_store_read(self, tmp_path):
|
|
"""Freshness transitions from fresh to missed as time passes."""
|
|
conn = init_store(tmp_path / "test.db")
|
|
_insert_segment(conn)
|
|
_open_period(conn)
|
|
now = _clock()
|
|
ts = (now - timedelta(minutes=10)).isoformat()
|
|
_insert_sample(conn, ts)
|
|
for i in range(20):
|
|
d = (datetime(2026, 9, 10) + timedelta(days=i)).strftime("%Y-%m-%d")
|
|
_insert_day(conn, d, bw=1024*1024*100)
|
|
|
|
svc = {
|
|
"boot_enabled": True, "timer_active": True,
|
|
"last_collect_ok": True, "last_collect_age_s": 600,
|
|
"last_collect_reason": None,
|
|
}
|
|
|
|
# Sample is 10 min old (600s) → fresh (within 690s threshold)
|
|
comp1 = compose_status(conn, svc, now, store_fault=None, newer_schema=None)
|
|
assert comp1.freshness == "fresh"
|
|
assert comp1.state == StatusState.MONITORING
|
|
|
|
# At now + 1 min → sample is 11 min old (660s) → still fresh
|
|
now2 = now + timedelta(minutes=1)
|
|
comp2 = compose_status(conn, svc, now2, store_fault=None, newer_schema=None)
|
|
assert comp2.freshness == "fresh"
|
|
|
|
# At now + 2 min → sample is 12 min old (720s) → missed (past 690s)
|
|
now3 = now + timedelta(minutes=2)
|
|
comp3 = compose_status(conn, svc, now3, store_fault=None, newer_schema=None)
|
|
assert comp3.freshness == "missed"
|
|
assert comp3.state == StatusState.MONITORING
|
|
|
|
# At now + 49 hours → sample is 49h 10min old → stale
|
|
now4 = now + timedelta(hours=49)
|
|
comp4 = compose_status(conn, svc, now4, store_fault=None, newer_schema=None)
|
|
assert comp4.freshness == "stale"
|
|
assert comp4.state == StatusState.STALE
|
|
conn.close()
|
|
|
|
def test_stale_triggers_when_timer_active(self, tmp_path):
|
|
"""Stale state triggers only when timer is active."""
|
|
conn = init_store(tmp_path / "test.db")
|
|
_insert_segment(conn)
|
|
_open_period(conn)
|
|
now = _clock()
|
|
ts = (now - timedelta(hours=49)).isoformat()
|
|
_insert_sample(conn, ts)
|
|
for i in range(20):
|
|
d = (datetime(2026, 9, 10) + timedelta(days=i)).strftime("%Y-%m-%d")
|
|
_insert_day(conn, d, bw=1024*1024*100)
|
|
|
|
# Timer active + stale → Stale state
|
|
svc_active = {
|
|
"boot_enabled": True, "timer_active": True,
|
|
"last_collect_ok": True, "last_collect_age_s": 176400,
|
|
"last_collect_reason": None,
|
|
}
|
|
comp = compose_status(conn, svc_active, now, store_fault=None, newer_schema=None)
|
|
assert comp.state == StatusState.STALE
|
|
assert comp.freshness == "stale"
|
|
conn.close()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 5-second poll constant (AC78-6)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestPollConstant:
|
|
"""The status poll interval is 5 seconds."""
|
|
|
|
def test_poll_interval_exists(self):
|
|
from fenris.status_composition import STATUS_POLL_INTERVAL_S
|
|
assert STATUS_POLL_INTERVAL_S == 5
|
|
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# TUI integration with status composition (AC78-1)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestTUIIntegration:
|
|
"""TUI renders status composition with correct styling."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_tui_renders_monitoring_glyph(self, tmp_path):
|
|
"""TUI shows the Monitoring glyph in the service strip."""
|
|
from fenris.tui import FenrisTuiApp
|
|
|
|
conn = init_store(tmp_path / "test.db")
|
|
_insert_segment(conn)
|
|
_open_period(conn)
|
|
now = _clock()
|
|
ts = (now - timedelta(minutes=2)).isoformat()
|
|
_insert_sample(conn, ts)
|
|
for i in range(20):
|
|
d = (datetime(2026, 9, 10) + timedelta(days=i)).strftime("%Y-%m-%d")
|
|
_insert_day(conn, d, bw=1024*1024*100)
|
|
conn.close()
|
|
|
|
app = FenrisTuiApp(store_path=tmp_path / "test.db")
|
|
with patch("fenris.tui.query_service_state", return_value={
|
|
"boot_enabled": True, "timer_active": True,
|
|
"last_collect_ok": True, "last_collect_age_s": 120,
|
|
"last_collect_reason": None,
|
|
}):
|
|
async with app.run_test() as pilot:
|
|
strip = str(app.query_one("#service-strip").render())
|
|
assert "●" in strip
|
|
assert "Monitoring" in strip or "monitoring" in strip.lower()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_tui_renders_paused_state(self, tmp_path):
|
|
"""TUI shows Paused state with correct glyph."""
|
|
from fenris.tui import FenrisTuiApp
|
|
|
|
db = tmp_path / "test.db"
|
|
conn = init_store(db)
|
|
_insert_segment(conn)
|
|
_close_period(conn, started_at="2026-09-28T10:00:00+00:00",
|
|
ended_at="2026-09-29T10:00:00+00:00", cause="user_disabled")
|
|
now = _clock()
|
|
ts = (now - timedelta(minutes=2)).isoformat()
|
|
_insert_sample(conn, ts)
|
|
conn.close()
|
|
|
|
app = FenrisTuiApp(store_path=db)
|
|
with patch("fenris.tui.query_service_state", return_value={
|
|
"boot_enabled": False, "timer_active": False,
|
|
"last_collect_ok": None, "last_collect_age_s": None,
|
|
"last_collect_reason": None,
|
|
}):
|
|
async with app.run_test() as pilot:
|
|
strip = str(app.query_one("#service-strip").render())
|
|
# Should show Paused state
|
|
assert "‖" in strip or "Paused" in strip or "paused" in strip.lower()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_tui_renders_error_state(self, tmp_path):
|
|
"""TUI shows Error state with correct glyph."""
|
|
from fenris.tui import FenrisTuiApp
|
|
|
|
conn = init_store(tmp_path / "test.db")
|
|
_insert_segment(conn)
|
|
_open_period(conn)
|
|
now = _clock()
|
|
ts = (now - timedelta(minutes=2)).isoformat()
|
|
_insert_sample(conn, ts)
|
|
for i in range(20):
|
|
d = (datetime(2026, 9, 10) + timedelta(days=i)).strftime("%Y-%m-%d")
|
|
_insert_day(conn, d, bw=1024*1024*100)
|
|
conn.close()
|
|
|
|
app = FenrisTuiApp(store_path=tmp_path / "test.db")
|
|
with patch("fenris.tui.query_service_state", return_value={
|
|
"boot_enabled": True, "timer_active": True,
|
|
"last_collect_ok": False, "last_collect_age_s": 60,
|
|
"last_collect_reason": "exit code 3",
|
|
}):
|
|
async with app.run_test() as pilot:
|
|
strip = str(app.query_one("#service-strip").render())
|
|
# Should show Error state
|
|
assert "✖" in strip or "Error" in strip or "error" in strip.lower()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# CLI integration with status composition (AC78-1, TPH-3)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestCLIIntegration:
|
|
"""CLI renders status composition correctly."""
|
|
|
|
def test_cli_uses_composition(self, tmp_path):
|
|
"""CLI status uses the shared composition layer."""
|
|
from fenris.status import get_status_composition
|
|
from fenris.status_composition import render_status_cli
|
|
|
|
conn = init_store(tmp_path / "test.db")
|
|
_insert_segment(conn)
|
|
_open_period(conn)
|
|
now = _clock()
|
|
ts = (now - timedelta(minutes=2)).isoformat()
|
|
_insert_sample(conn, ts)
|
|
for i in range(20):
|
|
d = (datetime(2026, 9, 10) + timedelta(days=i)).strftime("%Y-%m-%d")
|
|
_insert_day(conn, d, bw=1024*1024*100)
|
|
conn.close()
|
|
|
|
with patch("fenris.status.query_service_state", return_value={
|
|
"boot_enabled": True, "timer_active": True,
|
|
"last_collect_ok": True, "last_collect_age_s": 120,
|
|
"last_collect_reason": None,
|
|
}):
|
|
comp = get_status_composition(
|
|
store_path=tmp_path / "test.db",
|
|
clock_now=now,
|
|
query_services=True,
|
|
)
|
|
cli_text = render_status_cli(comp)
|
|
|
|
# Should show Monitoring state
|
|
assert "● Monitoring" in cli_text
|
|
# Should NOT show TUI-only strings
|
|
assert "Fenris — NVMe endurance monitor" not in cli_text
|
|
assert "by Bongbetic" not in cli_text
|