diff --git a/src/fenris/status.py b/src/fenris/status.py index 501473d..54b81b7 100644 --- a/src/fenris/status.py +++ b/src/fenris/status.py @@ -687,3 +687,79 @@ def render_status(store_path: Optional[Path] = None, clock_now: Optional[datetim parts.append("") parts.append(format_disclosures()) return "\n\n".join(parts) + + + +# --------------------------------------------------------------------------- +# Shared status composition integration (issue #78) +# --------------------------------------------------------------------------- + +def get_status_composition( + store_path: Optional[Path] = None, + clock_now: Optional[datetime] = None, + query_services: bool = True, + collecting: bool = False, + reduced_motion: bool = False, +) -> 'StatusComposition': + """Get the shared status composition consumed by both TUI and CLI. + + This is the new entry point that centralizes the status lattice. + """ + # Lazy import to avoid circular dependency + from .status_composition import ( + StatusComposition, + compose_status, + ) + + if clock_now is None: + clock_now = datetime.now(timezone.utc) + + # --- Configuration (§8.3) --- + # Config errors are surfaced through the store fault mechanism + + # --- Service state --- + service = {} + if query_services: + try: + service = query_service_state() + except Exception: + service = None + + # --- Store open --- + store_fault = None + newer_schema = None + conn = None + + if store_path is None: + store_path = Path("/var/lib/fenris/observations.db") + + try: + conn = open_store_readonly(store_path) + except StoreFault as e: + store_fault = str(e) + except NewerSchema as e: + newer_schema = str(e) + + # --- Use shared composition --- + if conn is not None: + try: + comp = compose_status( + conn, service, clock_now, + store_fault=store_fault, + newer_schema=newer_schema, + collecting=collecting, + reduced_motion=reduced_motion, + ) + finally: + conn.close() + else: + # Store fault or newer schema - compose without store data + comp = compose_status( + None, service, clock_now, + store_fault=store_fault, + newer_schema=newer_schema, + collecting=collecting, + reduced_motion=reduced_motion, + ) + + return comp diff --git a/src/fenris/status_composition.py b/src/fenris/status_composition.py new file mode 100644 index 0000000..4d0cacc --- /dev/null +++ b/src/fenris/status_composition.py @@ -0,0 +1,526 @@ +"""Shared status composition for TUI and CLI (issue #78, TPH-2/3). + +Centralizes the monitoring status lattice consumed by both surfaces. +States: Monitoring, Collecting, Paused, Waiting, Interrupted, Error, Stale, Unknown. + +Precedence: Error > Interrupted > Paused > Stale > Waiting > Monitoring > Unknown. +Collecting overlays every base except store fault. + +Freshness grading uses shared constants from status.py. +""" +import enum +import sqlite3 +from dataclasses import dataclass, field +from datetime import datetime, timedelta, timezone +from typing import Any, Dict, List, Optional + +# Status poll interval (AC78-6): lightweight 5s systemctl show poll +STATUS_POLL_INTERVAL_S = 5 + +from .status import ( + FRESH_THRESHOLD_S, + STALENESS_THRESHOLD_S, + grade_freshness, + freshness_age_human, + is_deliberately_paused, + monitoring_continuity, + deliberate_pause_lines, + open_store_readonly, + StoreFault, + NewerSchema, +) + + +# --------------------------------------------------------------------------- +# Status state enum with glyph and label +# --------------------------------------------------------------------------- + +class StatusState(enum.Enum): + """The eight monitoring status states (TPH-2).""" + + MONITORING = "monitoring" + COLLECTING = "collecting" + PAUSED = "paused" + WAITING = "waiting" + INTERRUPTED = "interrupted" + ERROR = "error" + STALE = "stale" + UNKNOWN = "unknown" + + @property + def glyph(self) -> str: + _glyphs = { + "monitoring": "●", + "collecting": "◐", + "paused": "‖", + "waiting": "○", + "interrupted": "⊘", + "error": "✖", + "stale": "◌", + "unknown": "?", + } + return _glyphs[self.value] + + @property + def label(self) -> str: + return self.value.capitalize() + + +# Precedence order: higher index = higher precedence +_PRECEDENCE = [ + StatusState.UNKNOWN, + StatusState.MONITORING, + StatusState.WAITING, + StatusState.STALE, + StatusState.PAUSED, + StatusState.INTERRUPTED, + StatusState.ERROR, +] + +_PRECEDENCE_RANK = {s: i for i, s in enumerate(_PRECEDENCE)} + + +# --------------------------------------------------------------------------- +# Status composition result +# --------------------------------------------------------------------------- + +@dataclass +class StatusComposition: + """The composed status result shared between TUI and CLI.""" + + state: StatusState + glyph: str + label: str + explanation: str + + # Separate facts (never folded into the status word) + freshness: str = "unknown" + freshness_age_s: Optional[int] = None + last_collect_ok: Optional[bool] = None + last_collect_age_s: Optional[int] = None + last_collect_reason: Optional[str] = None + boot_enabled: Optional[bool] = None + timer_active: Optional[bool] = None + deliberately_paused: bool = False + pause_age_s: Optional[int] = None + external_stop_reason: Optional[str] = None + + # Store fault / newer schema (suppress store-dependent views) + store_fault: Optional[str] = None + newer_schema: Optional[str] = None + + # Collecting overlay + overlay_base: Optional[StatusState] = None + + # Sample counts for waiting explanations + sample_count: int = 0 + day_count: int = 0 + + # Whether the status dot should blink (only Monitoring) + should_blink: bool = False + + # Continuity line + continuity: str = "" + + # Paused lines (for TUI banner) + paused_lines: List[str] = field(default_factory=list) + + +# --------------------------------------------------------------------------- +# Status composition logic +# --------------------------------------------------------------------------- + +def _determine_base_state( + freshness: str, + sample_count: int, + day_count: int, + boot_enabled: Optional[bool], + timer_active: Optional[bool], + last_collect_ok: Optional[bool], + deliberately_paused: bool, + external_stop_reason: Optional[str], + service_available: bool, +) -> StatusState: + """Determine the base status state from facts. + + Precedence: Error > Interrupted > Paused > Stale > Waiting > Monitoring > Unknown. + """ + # Unknown: service query failed and no store-derived fact places us higher + if not service_available: + return StatusState.UNKNOWN + + # Error: last collect failed + if last_collect_ok is False: + return StatusState.ERROR + + # Interrupted: external stop (not user_disabled) + if external_stop_reason is not None: + return StatusState.INTERRUPTED + + # Paused: deliberate disable + if deliberately_paused: + return StatusState.PAUSED + + # Stale: timer active, no failure, but data ≥ 48h old + if freshness == "stale" and timer_active and last_collect_ok is not False: + return StatusState.STALE + + # Waiting: empty store, single sample, or fresh data but not yet enough evidence + if sample_count == 0: + return StatusState.WAITING + if sample_count <= 1 and day_count == 0: + return StatusState.WAITING + + # Monitoring: everything is fine + return StatusState.MONITORING + + +def _determine_explanation( + state: StatusState, + freshness: str, + freshness_age_s: Optional[int], + last_collect_ok: Optional[bool], + last_collect_reason: Optional[str], + sample_count: int, + deliberately_paused: bool, + store_fault: Optional[str], + newer_schema: Optional[str], +) -> str: + """Determine the explanation line for the status state.""" + if store_fault: + return "observation store unreadable — see journal" + + if newer_schema: + return "observation store written by a newer Fenris — upgrade Fenris" + + if state == StatusState.ERROR: + parts = [] + if last_collect_ok is False: + parts.append("last run failed") + if last_collect_reason: + parts.append("(%s)" % last_collect_reason) + if freshness_age_s is not None and freshness != "empty": + parts.append("· last good sample %s ago" % freshness_age_human(freshness_age_s)) + return " ".join(parts) if parts else "last run failed" + + if state == StatusState.INTERRUPTED: + return "collection stopped outside Fenris — monitoring period still open" + + if state == StatusState.PAUSED: + return "monitoring paused — paused time excluded from your usage habit" + + if state == StatusState.STALE: + if freshness_age_s is not None: + return "last sample %s ago" % freshness_age_human(freshness_age_s) + return "data is stale" + + if state == StatusState.WAITING: + if sample_count == 0: + return "awaiting first sample" + if sample_count <= 1: + return "awaiting another sample" + return "waiting for data" + + if state == StatusState.MONITORING: + if freshness_age_s is not None: + return "last sample %s ago" % freshness_age_human(freshness_age_s) + return "monitoring active" + + if state == StatusState.UNKNOWN: + return "service state unavailable" + + return "" + + +def _determine_collecting_overlay( + base_state: StatusState, + last_collect_ok: Optional[bool], + deliberately_paused: bool, + store_fault: Optional[str], + newer_schema: Optional[str], +) -> str: + """Determine the explanation line when Collecting overlays a base state.""" + if store_fault or newer_schema: + return "run in flight — store fault" + + if base_state == StatusState.PAUSED: + return "run in flight — paused" + if base_state == StatusState.INTERRUPTED: + return "run in flight — interrupted" + if base_state == StatusState.ERROR: + return "run in flight — retry" + if base_state == StatusState.STALE: + return "run in flight — stale data" + if base_state == StatusState.WAITING: + return "run in flight" + return "run in flight" + + +def compose_status( + conn: sqlite3.Connection, + service: Optional[Dict[str, Any]], + clock_now: datetime, + store_fault: Optional[str] = None, + newer_schema: Optional[str] = None, + collecting: bool = False, + reduced_motion: bool = False, +) -> StatusComposition: + """Compose the shared status from service state and store data. + + This is the single entry point consumed by both TUI and CLI. + """ + service_available = service is not None and len(service) > 0 + + # --- Separate facts from service --- + boot_enabled = service.get("boot_enabled") if service_available else None + timer_active = service.get("timer_active") if service_available else None + last_collect_ok = service.get("last_collect_ok") if service_available else None + last_collect_age_s = service.get("last_collect_age_s") if service_available else None + last_collect_reason = service.get("last_collect_reason") if service_available else None + + # --- Freshness from store --- + freshness = "unknown" + freshness_age_s = None + sample_count = 0 + day_count = 0 + deliberately_paused = False + + if store_fault is None and newer_schema is None: + try: + cursor = conn.execute("SELECT ts FROM samples ORDER BY id DESC LIMIT 1") + row = cursor.fetchone() + newest_ts = row[0] if row else None + freshness = grade_freshness(newest_ts, clock_now) + if newest_ts: + try: + ts = datetime.fromisoformat(newest_ts) + if ts.tzinfo is None: + ts = ts.replace(tzinfo=timezone.utc) + else: + ts = ts.astimezone(timezone.utc) + freshness_age_s = int((clock_now - ts).total_seconds()) + except (ValueError, TypeError): + pass + except sqlite3.Error: + freshness = "unknown" + + try: + cursor = conn.execute("SELECT COUNT(*) FROM samples") + sample_count = cursor.fetchone()[0] + cursor = conn.execute("SELECT COUNT(*) FROM day_aggregates") + day_count = cursor.fetchone()[0] + except sqlite3.Error: + pass + + try: + svc_for_pause = service if service_available else {} + deliberately_paused = is_deliberately_paused(conn, svc_for_pause) + except sqlite3.Error: + deliberately_paused = False + + # --- External stop detection --- + # External stop = timer inactive + boot disabled + NOT deliberately paused + # + period still open (the timer was stopped but Fenris didn't close the period) + external_stop_reason = None + if not deliberately_paused and timer_active is False and boot_enabled is False: + # Check if there's an open monitoring period (external stop left it open) + try: + open_period = conn.execute( + "SELECT 1 FROM monitoring_periods WHERE ended_at IS NULL LIMIT 1" + ).fetchone() + if open_period is not None: + external_stop_reason = "external_stop" + except sqlite3.Error: + pass + + # --- Determine base state --- + # Store fault and newer schema always override to ERROR + if store_fault is not None or newer_schema is not None: + base_state = StatusState.ERROR + else: + base_state = _determine_base_state( + freshness=freshness, + sample_count=sample_count, + day_count=day_count, + boot_enabled=boot_enabled, + timer_active=timer_active, + last_collect_ok=last_collect_ok, + deliberately_paused=deliberately_paused, + external_stop_reason=external_stop_reason, + service_available=service_available, + ) + + # --- Apply Collecting overlay --- + state = base_state + overlay_base = None + explanation = "" + + if collecting and store_fault is None and newer_schema is None: + # Collecting overlays every base except store fault + overlay_base = base_state + state = StatusState.COLLECTING + explanation = _determine_collecting_overlay( + base_state, last_collect_ok, deliberately_paused, + store_fault, newer_schema, + ) + else: + explanation = _determine_explanation( + base_state, freshness, freshness_age_s, + last_collect_ok, last_collect_reason, + sample_count, deliberately_paused, + store_fault, newer_schema, + ) + + # --- Continuity line --- + continuity = "" + if service_available: + continuity = monitoring_continuity(service) + + # --- Paused lines --- + paused_lines = [] + if deliberately_paused: + paused_lines = deliberate_pause_lines() + + # Determine blink flag: only Monitoring dot blinks, never text or other states + should_blink = (state == StatusState.MONITORING and not reduced_motion) + + return StatusComposition( + state=state, + glyph=state.glyph, + label=state.label, + explanation=explanation, + should_blink=should_blink, + freshness=freshness, + freshness_age_s=freshness_age_s, + last_collect_ok=last_collect_ok, + last_collect_age_s=last_collect_age_s, + last_collect_reason=last_collect_reason, + boot_enabled=boot_enabled, + timer_active=timer_active, + deliberately_paused=deliberately_paused, + external_stop_reason=external_stop_reason, + store_fault=store_fault, + newer_schema=newer_schema, + overlay_base=overlay_base, + sample_count=sample_count, + day_count=day_count, + continuity=continuity, + paused_lines=paused_lines, + ) + + +# --------------------------------------------------------------------------- +# CLI rendering (static, no styling) +# --------------------------------------------------------------------------- + +def render_status_cli(comp: StatusComposition) -> str: + """Render the status composition as static CLI text.""" + lines = [] + + # Status line + lines.append("%s %s" % (comp.glyph, comp.label)) + + # Explanation + if comp.explanation: + lines.append(comp.explanation) + + lines.append("") + + # Separate facts + facts = [] + if comp.freshness != "unknown": + facts.append("freshness: %s" % comp.freshness) + if comp.freshness_age_s is not None: + facts[-1] += " (%s)" % freshness_age_human(comp.freshness_age_s) if facts else "freshness: %s" % freshness_age_human(comp.freshness_age_s) + if comp.last_collect_ok is True: + facts.append("last collect: ok") + elif comp.last_collect_ok is False: + collect_str = "last collect: FAILED" + if comp.last_collect_reason: + collect_str += " (%s)" % comp.last_collect_reason + facts.append(collect_str) + else: + facts.append("last collect: unknown") + if comp.boot_enabled is not None: + facts.append("boot: %s" % ("enabled" if comp.boot_enabled else "disabled")) + if comp.timer_active is not None: + facts.append("timer: %s" % ("active" if comp.timer_active else "inactive")) + + if facts: + lines.append(" · ".join(facts)) + + # Continuity + if comp.continuity: + lines.append("") + lines.append("CONTINUITY: %s" % comp.continuity) + + # Deliberate pause + if comp.paused_lines: + for pl in comp.paused_lines: + lines.append(pl) + + return "\n".join(lines) + + +# --------------------------------------------------------------------------- +# TUI rendering (with styling tokens) +# --------------------------------------------------------------------------- + +def render_status_tui(comp: StatusComposition) -> str: + """Render the status composition as TUI text with Textual markup.""" + lines = [] + + # Status line with color + color = { + StatusState.MONITORING: "green", + StatusState.COLLECTING: "green", + StatusState.PAUSED: "yellow", + StatusState.WAITING: "yellow", + StatusState.INTERRUPTED: "red", + StatusState.ERROR: "red", + StatusState.STALE: "red", + StatusState.UNKNOWN: "dim", + }[comp.state] + + lines.append("[%s]%s %s[/%s]" % (color, comp.glyph, comp.label, color)) + + # Explanation + if comp.explanation: + lines.append(comp.explanation) + + lines.append("") + + # Separate facts + facts = [] + if comp.freshness != "unknown": + facts.append("freshness: %s" % comp.freshness) + if comp.freshness_age_s is not None and facts: + facts[-1] += " (%s)" % freshness_age_human(comp.freshness_age_s) + if comp.last_collect_ok is True: + facts.append("last collect: ok") + elif comp.last_collect_ok is False: + collect_str = "last collect: FAILED" + if comp.last_collect_reason: + collect_str += " (%s)" % comp.last_collect_reason + facts.append(collect_str) + else: + facts.append("last collect: unknown") + if comp.boot_enabled is not None: + facts.append("boot: %s" % ("enabled" if comp.boot_enabled else "disabled")) + if comp.timer_active is not None: + facts.append("timer: %s" % ("active" if comp.timer_active else "inactive")) + + if facts: + lines.append(" · ".join(facts)) + + # Continuity + if comp.continuity: + lines.append("") + lines.append("[bold]CONTINUITY[/bold] %s" % comp.continuity) + + # Deliberate pause + if comp.paused_lines: + for pl in comp.paused_lines: + lines.append(pl) + + return "\n".join(lines) diff --git a/src/fenris/tui.py b/src/fenris/tui.py index 06adc21..19396ad 100644 --- a/src/fenris/tui.py +++ b/src/fenris/tui.py @@ -52,6 +52,13 @@ from .status import ( read_config, ) from .monitoring_periods import get_open_period +from .status_composition import ( + StatusState, + StatusComposition, + compose_status, + render_status_tui, + STATUS_POLL_INTERVAL_S, +) # --------------------------------------------------------------------------- @@ -1111,22 +1118,23 @@ class FenrisTuiApp(App): # --- Service strip (§7.2 bottom) --- try: svc = _query_service_facts(conn, self._clock_now) - boot = "enabled" if svc.get("boot_enabled") else "disabled" - activity = "active" if svc.get("timer_active") else "inactive" - collect = "ok" if svc.get("last_collect_ok") else "FAILED" - freshness = svc.get("freshness", "unknown") - self.query_one("#service-strip").update( - "boot: %s · timer: %s · last collect: %s · freshness: %s · " - "[dim]by Bongbetic[/dim]\n" - "[bold]CONTINUITY[/bold] %s\n" - "%s\n" - "p pause · r resume · c collect · d disclosures" - % ( - boot, activity, collect, freshness, - monitoring_continuity(svc), svc.get("period", ""), - ) + # Use shared status composition + comp = compose_status( + conn, svc, self._clock_now, + store_fault=None, + newer_schema=None, + reduced_motion=False, ) - self._render_paused_banner(svc) + # Render using the shared composition with TUI styling + status_text = render_status_tui(comp) + # Add TUI-only branding and actions + self.query_one("#service-strip").update( + "%s\n" + "[dim]by Bongbetic[/dim]\n" + "p pause · r resume · c collect · d disclosures" + % status_text + ) + self._render_paused_banner(comp) except Exception: self._hide_paused_banner() self.query_one("#service-strip").update( @@ -1135,13 +1143,24 @@ class FenrisTuiApp(App): "p pause · r resume · c collect · d disclosures" ) - def _render_paused_banner(self, service: Dict[str, Any]) -> None: - """Show the high-contrast Deliberate disable block only when sanctioned.""" + def _render_paused_banner(self, comp) -> None: + """Show the high-contrast Deliberate disable block only when sanctioned. + + Accepts either a StatusComposition or a dict for backward compatibility. + """ banner = self.query_one("#paused-banner") - if service.get("deliberately_paused"): + # Support both StatusComposition and legacy dict + if hasattr(comp, 'deliberately_paused'): + is_paused = comp.deliberately_paused + paused_lines = comp.paused_lines + else: + is_paused = comp.get("deliberately_paused", False) + paused_lines = deliberate_pause_lines() if is_paused else [] + + if is_paused and paused_lines: banner.update( "[bold black on red]%s[/bold black on red]\n%s" - % tuple(deliberate_pause_lines()) + % tuple(paused_lines) ) banner.styles.display = "block" main_grid = self.query_one("#main-grid") diff --git a/tests/test_status_composition.py b/tests/test_status_composition.py new file mode 100644 index 0000000..0a57559 --- /dev/null +++ b/tests/test_status_composition.py @@ -0,0 +1,1072 @@ +"""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