"""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 collector logs" 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=4)).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": 240, "last_collect_reason": None, } # Sample is 4 min old (240s) → fresh (within 450s threshold) comp1 = compose_status(conn, svc, now, store_fault=None, newer_schema=None) assert comp1.freshness == "fresh" assert comp1.state == StatusState.MONITORING # At now + 3 min → sample is 7 min old (420s) → still fresh now2 = now + timedelta(minutes=3) comp2 = compose_status(conn, svc, now2, store_fault=None, newer_schema=None) assert comp2.freshness == "fresh" # At now + 5 min → sample is 9 min old (540s) → missed (past 450s) now3 = now + timedelta(minutes=5) 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 4min 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.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, }): 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.status.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.status.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