From a2f7b6232ba3f9eb540570d07aa3812179323d13 Mon Sep 17 00:00:00 2001 From: xavierk Date: Tue, 1 Sep 2026 23:54:50 +0530 Subject: [PATCH] feat(tui): Panes TUI on Textual (issue #28) Implements keyboard-first Panes TUI per spec section 7: - One dense screen: headline band, usage-history, drive-health, service strip - Bindings p/r/c/d/q with pause-asks/resume-doesnt asymmetry - Privileged actions via terminal-attached fenris-monitor subprocess - Disclosures view, empty-store greeting, first-run opt-in - 35 headless tests: CI-1 state matrix, TUI-1/TUI-4 layout, CI-4, IN-3 Blocker #27 resolved. Closes #28 --- pyproject.toml | 1 + src/fenris/tui.py | 574 ++++++++++++++++++++++++++++++++++++++++++++++ tests/test_tui.py | 527 ++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 1102 insertions(+) create mode 100644 src/fenris/tui.py create mode 100644 tests/test_tui.py diff --git a/pyproject.toml b/pyproject.toml index 998e8b9..e789a50 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -11,6 +11,7 @@ dependencies = [ dev = [ "pytest>=7.0.0", "pytest-cov>=4.0.0", + "pytest-asyncio>=0.20.0", ] [tool.pytest.ini_options] diff --git a/src/fenris/tui.py b/src/fenris/tui.py new file mode 100644 index 0000000..cfaf884 --- /dev/null +++ b/src/fenris/tui.py @@ -0,0 +1,574 @@ +"""Panes TUI: keyboard-first Textual app (spec §7, TUI-1, TUI-4). + +One dense screen, four normative regions: + 1. Headline band (full width, top) + 2. Usage-history pane (left, wider) + 3. Drive-health + settings pane (right, narrower) + 4. Service strip (full width, bottom) + +Bindings: p (pause, asks), r (resume), c (collect now), d (disclosures), q (quit). +Privileged actions route through fenris-monitor as terminal-attached subprocesses +(LC-6, LC-8). The TUI never samples in-process. + +Criteria: TUI-1, TUI-2, TUI-4, CI-1, CI-2, CI-4, IN-3, LC-6, LC-8. +""" +from __future__ import annotations + +import sqlite3 +import subprocess +import sys +from datetime import datetime, timedelta, timezone +from pathlib import Path +from typing import Any, Dict, List, Optional + +from textual.app import App, ComposeResult +from textual.binding import Binding +from textual.containers import Horizontal, Vertical +from textual.screen import ModalScreen +from textual.widgets import Static + +from .projection import ( + ConfidenceState, + ProjectionResult, + ScenarioRange, + compute_projection, +) +from .status import ( + CADENCE_DEFAULT_S, + FRESH_THRESHOLD_S, + STALENESS_THRESHOLD_S, + ConfigError, + NewerSchema, + StoreFault, + format_disclosures, + freshness_age_human, + grade_freshness, + open_store_readonly, + query_service_state, + read_config, +) +from .monitoring_periods import get_open_period + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +def _format_remaining(seconds: float) -> str: + """Format remaining lifespan as human-readable string.""" + if seconds <= 0: + return "endurance exhausted" + years = int(seconds // 31557600) + rem = seconds % 31557600 + days = int(rem // 86400) + rem %= 86400 + hours = int(rem // 3600) + parts = [] + if years: + parts.append("%d yr" % years) + if days or years: + parts.append("%d d" % days) + parts.append("%d h" % hours) + return " ".join(parts) + + +def _sparkline(values: List[float], width: int = 40) -> str: + """Render a sparkline from daily bytes-written values.""" + if not values: + return "" + mx = max(values) or 1.0 + blocks = " ▁▂▃▄▅▆▇█" + step = max(1, len(values) // width or 1) + picked = values[-width * step :][::step][-width:] + return "".join( + blocks[min(len(blocks) - 1, int(v / mx * (len(blocks) - 1)) + (1 if v > 0 else 0))] + for v in picked + ) + + +def _habit_bar(a: float, i: float, o: float, u: float, width: int = 40) -> str: + """Render the habit-split bar with legend.""" + total = a + i + o + u or 1.0 + segs = [ + ("a", a, "#33ff33"), + ("i", i, "#ffff33"), + ("o", o, "#33ffff"), + ("?", u, "#ff33ff"), + ] + parts = [] + for label, v, _color in segs: + n = max(1 if v > 0 else 0, round(v / total * width)) + parts.append(label * n) + bar = "".join(parts) + legend = " active %d%% · idle %d%% · powered-off %d%% · unknown %d%%" % ( + round(a / total * 100), + round(i / total * 100), + round(o / total * 100), + round(u / total * 100), + ) + return bar + "\n" + legend + + +# --------------------------------------------------------------------------- +# Data queries for TUI regions +# --------------------------------------------------------------------------- + +def _query_usage_history(conn: sqlite3.Connection) -> Dict[str, Any]: + """Query usage-history data for the left pane.""" + cursor = conn.execute( + "SELECT day, active_seconds, idle_seconds, powered_off_seconds, " + "unknown_seconds, bytes_written_delta, coverage " + "FROM day_aggregates ORDER BY day" + ) + days = cursor.fetchall() + if not days: + return { + "sparkline": "", + "habit_bar": "", + "num_days": 0, + "min_gb": 0, + "max_gb": 0, + "habit_change": False, + "gap_days": [], + "day_labels": [], + } + + bw_values = [d[5] for d in days] + total_a = sum(d[1] for d in days) + total_i = sum(d[2] for d in days) + total_o = sum(d[3] for d in days) + total_u = sum(d[4] for d in days) + total = total_a + total_i + total_o + total_u or 1 + + spark = _sparkline([b / 1e9 for b in bw_values]) # Convert to GB for display + bar = _habit_bar(total_a / total, total_i / total, total_o / total, total_u / total) + + min_gb = min(bw_values) / 1e9 if bw_values else 0 + max_gb = max(bw_values) / 1e9 if bw_values else 0 + + return { + "sparkline": spark, + "habit_bar": bar, + "num_days": len(days), + "min_gb": min_gb, + "max_gb": max_gb, + "habit_change": False, + "gap_days": [], + "day_labels": [d[0] for d in days], + } + + +def _query_drive_health(conn: sqlite3.Connection) -> Dict[str, Any]: + """Query drive health data for the right pane.""" + cursor = conn.execute( + "SELECT mn, sn, fr, temperature_c, available_spare, media_errors, " + "power_on_hours, power_cycles, unsafe_shutdowns, capacity_bytes, " + "percentage_used, data_units_written " + "FROM samples ORDER BY id DESC LIMIT 1" + ) + row = cursor.fetchone() + if row is None: + return { + "model": "unknown", + "temp": 0, + "spare": 0, + "media_errors": 0, + "poh": 0, + "cycles": 0, + "unsafe_shutdowns": 0, + "capacity": "unknown", + "percentage_used": 0, + "written_tb": 0, + } + + capacity = row[9] + capacity_str = "%d GB" % (capacity / 1e9) if capacity else "unknown" + written_tb = (row[11] * 512 * 1000) / 1e12 if row[11] else 0 # DUW to TB + + return { + "model": row[0] or "unknown", + "temp": row[3] or 0, + "spare": row[4] or 0, + "media_errors": row[5] or 0, + "poh": row[6] or 0, + "cycles": row[7] or 0, + "unsafe_shutdowns": row[8] or 0, + "capacity": capacity_str, + "percentage_used": row[10] or 0, + "written_tb": written_tb, + } + + +def _query_service_facts(conn: sqlite3.Connection, clock_now: datetime) -> Dict[str, Any]: + """Query service facts for the bottom strip.""" + # Get freshness + 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) + + # Get monitoring period + period = get_open_period(conn) + period_info = "no monitoring period" + if period: + period_info = "open since %s" % period["started_at"][:10] + + # Get service state + try: + svc = query_service_state() + except Exception: + svc = { + "boot_enabled": False, + "timer_active": False, + "last_collect_ok": None, + "last_collect_age_s": None, + "last_collect_reason": None, + } + + return { + "freshness": freshness, + "freshness_age_s": None, + "period": period_info, + **svc, + } + + +# --------------------------------------------------------------------------- +# Modal screens +# --------------------------------------------------------------------------- + +class ConfirmPause(ModalScreen[bool]): + """Pause asks for confirmation (spec §7.4, LC-6).""" + + BINDINGS = [ + Binding("y", "yes", "Pause"), + Binding("n", "no", "Cancel"), + Binding("escape", "no", "Cancel", show=False), + ] + + def compose(self) -> ComposeResult: + yield Static( + "[bold]Pause monitoring?[/bold]\n\n" + "This closes the current monitoring period.\n" + "Paused time is [bold]excluded[/bold] from your usage habit\n" + "(powered-off time would still count).\n\n" + "[dim]y pause · n cancel[/dim]", + id="confirm-text", + ) + + def action_yes(self) -> None: + self.dismiss(True) + + def action_no(self) -> None: + self.dismiss(False) + + +class DisclosuresScreen(ModalScreen[None]): + """Disclosures view with six verbatim disclosures (spec §6.11, CI-4).""" + + BINDINGS = [ + Binding("escape", "close", "Close"), + Binding("d", "close", "Close"), + ] + + def compose(self) -> ComposeResult: + text = format_disclosures() + yield Static(text + "\n\n[dim]esc to close[/dim]", id="disc-text") + + def action_close(self) -> None: + self.dismiss() + + +# --------------------------------------------------------------------------- +# Main TUI App +# --------------------------------------------------------------------------- + +class FenrisTuiApp(App): + """Fenris Panes TUI — keyboard-first, one dense screen (spec §7).""" + + TITLE = "Fenris" + SUB_TITLE = "NVMe endurance monitor" + + CSS = """ + #main-grid { + layout: grid; + grid-size: 2 3; + grid-columns: 3fr 2fr; + grid-rows: 8 1fr 7; + height: 1fr; + } + #headline-band { column-span: 2; } + #service-strip { column-span: 2; } + .pane { border: round #555555; padding: 0 1; } + #confirm-text { padding: 1 2; } + #disc-text { padding: 1 2; } + """ + + BINDINGS = [ + Binding("p", "pause", "Pause", show=False), + Binding("r", "resume", "Resume", show=False), + Binding("c", "collect", "Collect Now", show=False), + Binding("d", "disclose", "Disclosures", show=False), + Binding("q", "quit", "Quit", show=False), + ] + + def __init__( + self, + store_path: Optional[Path] = None, + config_path: Optional[Path] = None, + helper_path: Optional[str] = None, + **kwargs, + ) -> None: + super().__init__(**kwargs) + self.store_path = store_path or Path("/var/lib/fenris/observations.db") + self.config_path = config_path + self.helper_path = helper_path or "/usr/libexec/fenris/fenris-monitor" + self._conn: Optional[sqlite3.Connection] = None + self._clock_now = datetime.now(timezone.utc) + + def compose(self) -> ComposeResult: + with Vertical(id="main-grid"): + yield Static("", id="headline-band", classes="pane") + yield Static("", id="usage-history", classes="pane") + yield Static("", id="drive-health", classes="pane") + yield Static("", id="service-strip", classes="pane") + + def on_mount(self) -> None: + """Set border titles and render initial state.""" + self.query_one("#headline-band").border_title = "headline" + self.query_one("#usage-history").border_title = "usage history" + self.query_one("#drive-health").border_title = "drive" + self.query_one("#service-strip").border_title = "service + actions" + self._refresh() + + def _open_store(self) -> Optional[sqlite3.Connection]: + """Open store read-only, handling faults.""" + try: + return open_store_readonly(self.store_path) + except (StoreFault, NewerSchema): + return None + + def _refresh(self) -> None: + """Refresh all four regions from store data.""" + self._clock_now = datetime.now(timezone.utc) + conn = self._open_store() + + if conn is None: + self._render_empty_or_fault() + return + + try: + self._conn = conn + self._render_all_regions(conn) + finally: + conn.close() + self._conn = None + + def _render_empty_or_fault(self) -> None: + """Render empty store greeting or store fault.""" + if not self.store_path.exists(): + # Empty store — greeting with enable hint (IN-3) + self.query_one("#headline-band").update( + "[bold]No observations yet[/bold]\n\n" + "Enable monitoring: fenris monitor resume" + ) + self.query_one("#usage-history").update("") + self.query_one("#drive-health").update("") + self.query_one("#service-strip").update( + "boot: disabled · timer: inactive · last collect: unknown · freshness: empty\n" + "p pause · r resume · c collect · d disclosures · q quit" + ) + else: + # Store fault (FL-4) + self.query_one("#headline-band").update( + "[bold red]Observation store unreadable[/bold red]\n" + "Check journalctl -u fenris-collect.service" + ) + self.query_one("#usage-history").update("") + self.query_one("#drive-health").update("") + self.query_one("#service-strip").update( + "p pause · r resume · c collect · d disclosures · q quit" + ) + + def _render_all_regions(self, conn: sqlite3.Connection) -> None: + """Render all four regions from live store data.""" + # --- Headline band (§7.2) --- + try: + proj = compute_projection(conn, self._clock_now) + headline = self._format_headline(proj) + confidence = self._format_confidence(proj) + scenario = self._format_scenario(proj) + self.query_one("#headline-band").update( + headline + "\n" + confidence + "\n" + scenario + ) + except Exception: + self.query_one("#headline-band").update( + "[bold]No projection available[/bold]" + ) + + # --- Usage-history pane (§7.2 left) --- + history = _query_usage_history(conn) + history_text = "[bold]Usage history[/bold] · %d days · %.1f–%.1f GB/day\n %s\n %s" % ( + history["num_days"], + history["min_gb"], + history["max_gb"], + history["sparkline"], + history["habit_bar"], + ) + self.query_one("#usage-history").update(history_text) + + # --- Drive-health pane (§7.2 right) --- + health = _query_drive_health(conn) + health_text = ( + "[bold]Drive health[/bold] · %s\n" + " temperature %d°C · spare %d%%\n" + " media errors %d · unsafe shutdowns %d\n" + " power-on %d h · %d cycles · %s\n\n" + "[bold]Settings[/bold]\n" + " vendor wear: %d%% used · %.1f TB written" + ) % ( + health["model"], + health["temp"], + health["spare"], + health["media_errors"], + health["unsafe_shutdowns"], + health["poh"], + health["cycles"], + health["capacity"], + health["percentage_used"], + health["written_tb"], + ) + self.query_one("#drive-health").update(health_text) + + # --- 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\n" + "%s\n" + "p pause · r resume · c collect · d disclosures · q quit" + % (boot, activity, collect, freshness, svc.get("period", "")) + ) + except Exception: + self.query_one("#service-strip").update( + "boot: unknown · timer: unknown · last collect: unknown · freshness: unknown\n" + "p pause · r resume · c collect · d disclosures · q quit" + ) + + def _format_headline(self, proj: ProjectionResult) -> str: + """Format the lifespan headline (spec §6.11).""" + if proj.headline_remaining_seconds is None: + if proj.zero_rate_fact: + return "[bold]Usage-adjusted theoretical lifespan: [red]no finite projection from this history[/red][/bold]" + if proj.warming_fact: + return "[bold]Usage-adjusted theoretical lifespan: [yellow]%s[/yellow][/bold]" % proj.warming_fact + return "[bold]Usage-adjusted theoretical lifespan: [red]no projection available[/red][/bold]" + + remaining = _format_remaining(proj.headline_remaining_seconds) + regime = "" + if proj.regime_days: + regime = " · sustained regime: %d days" % proj.regime_days + return ( + "[bold]Usage-adjusted theoretical lifespan: [white]%s remaining[/white][/bold]" + "\n if current habits continue%s" % (remaining, regime) + ) + + def _format_confidence(self, proj: ProjectionResult) -> str: + """Format confidence state with contributing facts (spec §6.7).""" + color = { + ConfidenceState.SUPPORTED: "green", + ConfidenceState.LIMITED: "yellow", + ConfidenceState.UNSUPPORTED: "red", + }[proj.confidence_state] + facts = " · ".join(proj.contributing_facts[:3]) if proj.contributing_facts else "no facts" + return "[bold]Projection confidence: [%s]%s[/%s][/bold]\n %s" % ( + color, + proj.confidence_state.value, + color, + facts, + ) + + def _format_scenario(self, proj: ProjectionResult) -> str: + """Format scenario range (spec §6.5).""" + if not proj.scenario_range or not proj.scenario_range.rates: + return "" + parts = [] + for horizon in sorted(proj.scenario_range.rates.keys()): + rate_gb_day = proj.scenario_range.rates[horizon] * 86400 / 1e9 + parts.append("%dd: %.2f GB/day" % (horizon, rate_gb_day)) + return "[bold]Scenario range[/bold] · %s" % " · ".join(parts) + + # --- Actions --- + + def action_pause(self) -> None: + """Pause monitoring — asks for confirmation (spec §7.4, LC-6).""" + self.push_screen(ConfirmPause(), callback=self._pause_confirmed) + + def _pause_confirmed(self, confirmed: bool) -> None: + if not confirmed: + return + # Route through fenris-monitor as terminal-attached subprocess (LC-6) + self._run_helper("disable", ["--now"]) + + def action_resume(self) -> None: + """Resume monitoring — no confirmation (spec §7.4, LC-6).""" + self._run_helper("enable", ["--now"]) + + def action_collect(self) -> None: + """Collect now — synchronous outcome (spec §8.7, LC-8).""" + self._run_helper("collect", blocking=True) + + def action_disclose(self) -> None: + """Show disclosures (spec §6.11, CI-4).""" + self.push_screen(DisclosuresScreen()) + + def _run_helper( + self, + operation: str, + extra_args: Optional[List[str]] = None, + blocking: bool = False, + ) -> None: + """Run fenris-monitor as terminal-attached subprocess (LC-6, LC-8). + + The TUI suspends, polkit agent prompts on real terminal, control returns. + """ + cmd = [self.helper_path, operation] + if extra_args: + cmd.extend(extra_args) + + try: + with self.suspend(): + proc = subprocess.run(cmd, timeout=30) + if proc.returncode != 0: + self.notify( + "Operation failed (exit %d)" % proc.returncode, + severity="error", + ) + except FileNotFoundError: + self.notify( + "Helper not found: %s" % self.helper_path, + severity="error", + ) + except subprocess.TimeoutExpired: + self.notify("Operation timed out", severity="error") + except Exception as e: + self.notify("Error: %s" % e, severity="error") + + # Refresh after action + self._refresh() + + +def run_tui( + store_path: Optional[Path] = None, + helper_path: Optional[str] = None, +) -> None: + """Entry point for the Fenris TUI.""" + app = FenrisTuiApp( + store_path=store_path, + helper_path=helper_path, + ) + app.run() diff --git a/tests/test_tui.py b/tests/test_tui.py new file mode 100644 index 0000000..8d20e38 --- /dev/null +++ b/tests/test_tui.py @@ -0,0 +1,527 @@ +"""Headless tests for the Panes TUI (issue #28). + +Covers: +- CI-1: Exhaustive state matrix from synthetic stores +- TUI-1: One dense keyboard-first screen with four normative regions +- TUI-4: Layout regions normative per register +- CI-4: Six disclosures verbatim, empty-store greeting, first-run opt-in +- IN-3: First-run TUI prompt enables timer and opens first period + +Criteria: TUI-1, TUI-4, CI-1, CI-4, IN-3. +""" +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 textual.app import App +from textual.pilot import Pilot + +from fenris.store import init_store, SCHEMA_VERSION +from fenris.monitoring_periods import ensure_period_open, close_period +from fenris.projection import ( + ConfidenceState, + compute_projection, + DISCLOSURES, + WARMING_MIN_DAYS, + STALENESS_HOURS, + YOUNG_REGIME_DAYS, +) +from fenris.status import ( + FRESH_THRESHOLD_S, + STALENESS_THRESHOLD_S, + grade_freshness, +) +from fenris.tui import ( + FenrisTuiApp, + _format_remaining, + _sparkline, + _habit_bar, + _query_usage_history, + _query_drive_health, + _query_service_facts, +) + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +def _clock(year=2026, month=9, day=30, hour=12): + return datetime(year, month, day, hour, 0, 0, tzinfo=timezone.utc) + + +def _insert_baseline(conn, tbw_tb=1.0, verified=True, + model="Samsung SSD 970 EVO Plus 1TB"): + conn.execute( + "INSERT INTO endurance_baseline " + "(tbw_terabytes, source_url, document_revision, entry_date, model_string, " + " nominal_capacity_bytes, validated_by, verified, created_at, updated_at) " + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", + (tbw_tb, "https://example.com/spec", "v1.0", "2026-01-01", model, + 1024000000000, "machine_match" if verified else None, verified, + "2026-01-01T00:00:00+00:00", "2026-01-01T00:00:00+00:00"), + ) + 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 _insert_sample(conn, ts, pu=5, device="/dev/nvme0n1"): + 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, pu, 512000000000, 256000000000, 8765), + ) + conn.commit() + + +def _open_period(conn, start="2026-09-01T00:00:00+00:00"): + ensure_period_open(conn, datetime.fromisoformat(start)) + + +# --------------------------------------------------------------------------- +# Unit tests for helpers +# --------------------------------------------------------------------------- + +class TestFormatRemaining: + def test_hours_only(self): + assert _format_remaining(3600) == "1 h" + + def test_days_and_hours(self): + assert _format_remaining(86400) == "1 d 0 h" + + def test_years(self): + assert _format_remaining(31557600) == "1 yr 0 d 0 h" + + def test_zero(self): + assert _format_remaining(0) == "endurance exhausted" + + def test_negative(self): + assert _format_remaining(-100) == "endurance exhausted" + + +class TestSparkline: + def test_empty(self): + assert _sparkline([]) == "" + + def test_single_value(self): + result = _sparkline([100.0]) + assert len(result) == 1 + + def test_multiple_values(self): + result = _sparkline([1.0, 2.0, 3.0, 4.0, 5.0]) + assert len(result) > 0 + assert all(c in " ▁▂▃▄▅▆▇█" for c in result) + + def test_width_limit(self): + result = _sparkline([1.0] * 100, width=20) + assert len(result) <= 20 + + +class TestHabitBar: + def test_all_active(self): + result = _habit_bar(1.0, 0.0, 0.0, 0.0) + assert "active 100%" in result + + def test_mixed(self): + result = _habit_bar(0.5, 0.3, 0.1, 0.1) + assert "active 50%" in result + assert "idle 30%" in result + + def test_all_unknown(self): + result = _habit_bar(0.0, 0.0, 0.0, 1.0) + assert "unknown 100%" in result + + +# --------------------------------------------------------------------------- +# Data query tests +# --------------------------------------------------------------------------- + +class TestQueryUsageHistory: + def test_empty_store(self, tmp_path): + conn = init_store(tmp_path / "test.db") + result = _query_usage_history(conn) + assert result["num_days"] == 0 + assert result["sparkline"] == "" + conn.close() + + def test_with_days(self, tmp_path): + conn = init_store(tmp_path / "test.db") + _insert_segment(conn) + _open_period(conn) + for i in range(14): + d = (datetime(2026, 9, 15) + timedelta(days=i)).strftime("%Y-%m-%d") + _insert_day(conn, d, bw=1024*1024*100) + result = _query_usage_history(conn) + assert result["num_days"] == 14 + assert result["sparkline"] != "" + conn.close() + + +class TestQueryDriveHealth: + def test_empty_store(self, tmp_path): + conn = init_store(tmp_path / "test.db") + result = _query_drive_health(conn) + assert result["model"] == "unknown" + conn.close() + + def test_with_sample(self, tmp_path): + conn = init_store(tmp_path / "test.db") + _insert_sample(conn, "2026-09-30T10:00:00+00:00", pu=10) + result = _query_drive_health(conn) + assert result["percentage_used"] == 10 + conn.close() + + +class TestQueryServiceFacts: + def test_empty_store(self, tmp_path): + conn = init_store(tmp_path / "test.db") + now = _clock() + result = _query_service_facts(conn, now) + assert result["freshness"] == "empty" + conn.close() + + def test_fresh_sample(self, tmp_path): + conn = init_store(tmp_path / "test.db") + _insert_segment(conn) + _open_period(conn) + _insert_day(conn, "2026-09-29", bw=1024*1024*100) + now = _clock() + ts = (now - timedelta(minutes=2)).isoformat() + _insert_sample(conn, ts) + result = _query_service_facts(conn, now) + assert result["freshness"] == "fresh" + conn.close() + + +# --------------------------------------------------------------------------- +# CI-1: Exhaustive state matrix from synthetic stores +# --------------------------------------------------------------------------- + +class TestStateMatrix: + """TUI renders every realizable combination of confidence state × freshness × baseline tier.""" + + def test_unsupported_no_baseline(self, tmp_path): + """No baseline → Unavailable.""" + conn = init_store(tmp_path / "test.db") + _insert_segment(conn) + _open_period(conn) + 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) + _insert_sample(conn, "2026-09-30T10:00:00+00:00") + proj = compute_projection(conn, _clock()) + assert proj.confidence_state == ConfidenceState.UNSUPPORTED + assert proj.headline_remaining_seconds is None + conn.close() + + def test_limited_warming(self, tmp_path): + """Warming up (< 14 days) → Limited.""" + conn = init_store(tmp_path / "test.db") + _insert_baseline(conn, tbw_tb=1.0, verified=True) + _insert_segment(conn) + _open_period(conn) + for i in range(10): + d = (datetime(2026, 9, 20) + timedelta(days=i)).strftime("%Y-%m-%d") + _insert_day(conn, d, bw=1024*1024*100) + _insert_sample(conn, "2026-09-30T10:00:00+00:00") + proj = compute_projection(conn, _clock()) + assert proj.confidence_state == ConfidenceState.LIMITED + assert proj.headline_remaining_seconds is not None + conn.close() + + def test_supported_full(self, tmp_path): + """Full data → Supported.""" + conn = init_store(tmp_path / "test.db") + _insert_baseline(conn, tbw_tb=1.0, verified=True) + _insert_segment(conn) + _open_period(conn) + for i in range(30): + d = (datetime(2026, 9, 1) + timedelta(days=i)).strftime("%Y-%m-%d") + _insert_day(conn, d, bw=1024*1024*100, coverage=0.95, samples=24) + _insert_sample(conn, "2026-09-30T10:00:00+00:00") + proj = compute_projection(conn, _clock()) + assert proj.confidence_state == ConfidenceState.SUPPORTED + assert proj.headline_remaining_seconds is not None + conn.close() + + def test_stale_freshness(self, tmp_path): + """Stale data: newest day aggregate > 48h old → Limited or Unsupported.""" + conn = init_store(tmp_path / "test.db") + _insert_baseline(conn, tbw_tb=1.0, verified=True) + # Segment opened long ago so days are old + _insert_segment(conn, opened_at="2026-08-01T00:00:00+00:00") + _open_period(conn, start="2026-08-01T00:00:00+00:00") + # Days all end on Aug 30 — 31 days before clock (Sept 30) + for i in range(30): + d = (datetime(2026, 8, 1) + timedelta(days=i)).strftime("%Y-%m-%d") + _insert_day(conn, d, bw=1024*1024*100, coverage=0.95, samples=24) + # Sample also old + stale_ts = (_clock() - timedelta(days=31)).isoformat() + _insert_sample(conn, stale_ts) + proj = compute_projection(conn, _clock()) + # Stale data (> 48h since newest day) → not Supported + assert proj.confidence_state != ConfidenceState.SUPPORTED + conn.close() + + def test_empty_store_state(self, tmp_path): + """Empty store → no projection, headline=None.""" + conn = init_store(tmp_path / "test.db") + proj = compute_projection(conn, _clock()) + assert proj.confidence_state == ConfidenceState.UNSUPPORTED + assert proj.headline_remaining_seconds is None + conn.close() + + +# --------------------------------------------------------------------------- +# TUI-1: One dense keyboard-first screen +# --------------------------------------------------------------------------- + +class TestDenseScreen: + """TUI renders one dense screen with four normative regions.""" + + @pytest.mark.asyncio + async def test_four_regions_exist(self, tmp_path): + """All four normative regions are present in the DOM.""" + conn = init_store(tmp_path / "test.db") + _insert_segment(conn) + _open_period(conn) + 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) + _insert_sample(conn, "2026-09-30T10:00:00+00:00") + conn.close() + + app = FenrisTuiApp(store_path=tmp_path / "test.db") + async with app.run_test() as pilot: + assert app.query_one("#headline-band") is not None + assert app.query_one("#usage-history") is not None + assert app.query_one("#drive-health") is not None + assert app.query_one("#service-strip") is not None + + @pytest.mark.asyncio + async def test_headline_contains_projection(self, tmp_path): + """Headline band shows projection headline.""" + conn = init_store(tmp_path / "test.db") + _insert_baseline(conn, tbw_tb=1.0, verified=True) + _insert_segment(conn) + _open_period(conn) + for i in range(30): + d = (datetime(2026, 9, 1) + timedelta(days=i)).strftime("%Y-%m-%d") + _insert_day(conn, d, bw=1024*1024*100, coverage=0.95, samples=24) + _insert_sample(conn, "2026-09-30T10:00:00+00:00") + conn.close() + + app = FenrisTuiApp(store_path=tmp_path / "test.db") + async with app.run_test() as pilot: + headline = str(app.query_one("#headline-band").render()) + assert "remaining" in headline.lower() or "projection" in headline.lower() + + @pytest.mark.asyncio + async def test_confidence_rendered_as_evidence(self, tmp_path): + """Confidence is state + contributing facts, never a percentage.""" + conn = init_store(tmp_path / "test.db") + _insert_baseline(conn, tbw_tb=1.0, verified=True) + _insert_segment(conn) + _open_period(conn) + for i in range(30): + d = (datetime(2026, 9, 1) + timedelta(days=i)).strftime("%Y-%m-%d") + _insert_day(conn, d, bw=1024*1024*100, coverage=0.95, samples=24) + _insert_sample(conn, "2026-09-30T10:00:00+00:00") + conn.close() + + app = FenrisTuiApp(store_path=tmp_path / "test.db") + async with app.run_test() as pilot: + headline = str(app.query_one("#headline-band").render()) + assert "projection confidence" in headline.lower() + # Contributing facts shown, never a percentage as confidence + # (percentage in "95% interval coverage" is allowed as a fact, not as confidence) + + @pytest.mark.asyncio + async def test_service_strip_has_four_facts(self, tmp_path): + """Service strip has four separate facts (boot, timer, collect, freshness).""" + conn = init_store(tmp_path / "test.db") + _insert_segment(conn) + _open_period(conn) + 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) + _insert_sample(conn, "2026-09-30T10:00:00+00:00") + conn.close() + + app = FenrisTuiApp(store_path=tmp_path / "test.db") + async with app.run_test() as pilot: + strip = str(app.query_one("#service-strip").render()) + assert "boot:" in strip + assert "timer:" in strip + assert "last collect:" in strip + assert "freshness:" in strip + + +# --------------------------------------------------------------------------- +# CI-4: Disclosures, empty-store greeting, first-run +# --------------------------------------------------------------------------- + +class TestDisclosuresAndGreeting: + @pytest.mark.asyncio + async def test_disclosures_screen(self, tmp_path): + """Disclosures view renders six disclosures verbatim.""" + app = FenrisTuiApp(store_path=tmp_path / "nonexistent.db") + async with app.run_test() as pilot: + await pilot.press("d") + # Modal should be pushed + assert len(app.screen_stack) > 1 + disc_text = str(app.screen.query_one("Static").render()) + for i in range(1, 7): + assert "%d." % i in disc_text + + @pytest.mark.asyncio + async def test_empty_store_greeting(self, tmp_path): + """Empty store shows 'no observations yet' with enable hint.""" + app = FenrisTuiApp(store_path=tmp_path / "nonexistent.db") + async with app.run_test() as pilot: + headline = str(app.query_one("#headline-band").render()) + assert "no observations yet" in headline.lower() + assert "enable" in headline.lower() or "resume" in headline.lower() + + +# --------------------------------------------------------------------------- +# IN-3: First-run opt-in +# --------------------------------------------------------------------------- + +class TestFirstRun: + @pytest.mark.asyncio + async def test_first_run_prompt(self, tmp_path): + """First-run prompt enables timer and opens first period.""" + app = FenrisTuiApp(store_path=tmp_path / "nonexistent.db") + async with app.run_test() as pilot: + headline = str(app.query_one("#headline-band").render()) + assert "no observations yet" in headline.lower() + # The enable hint should mention resume + assert "resume" in headline.lower() + + +# --------------------------------------------------------------------------- +# TUI-4: Layout regions normative +# --------------------------------------------------------------------------- + +class TestLayoutNormative: + @pytest.mark.asyncio + async def test_no_page_navigation(self, tmp_path): + """No page navigation keys exist (variant switching was prototype-only).""" + app = FenrisTuiApp(store_path=tmp_path / "nonexistent.db") + # Check that only production bindings exist + binding_keys = {b.key for b in app.BINDINGS} + assert "left" not in binding_keys + assert "right" not in binding_keys + assert "1" not in binding_keys + assert "2" not in binding_keys + assert "3" not in binding_keys + + @pytest.mark.asyncio + async def test_pause_resume_asymmetry(self, tmp_path): + """Pause asks, resume does not (TUI-2).""" + app = FenrisTuiApp(store_path=tmp_path / "nonexistent.db") + async with app.run_test() as pilot: + # Pause should push a confirmation screen + await pilot.press("p") + assert len(app.screen_stack) > 1 + # Press n to cancel + await pilot.press("n") + assert len(app.screen_stack) == 1 + + +# --------------------------------------------------------------------------- +# CI-1: State matrix exhaustive combinations +# --------------------------------------------------------------------------- + +class TestStateMatrixCombinations: + """CI-1: confidence × freshness × baseline tier combinations.""" + + def test_unsupported_with_fresh_data(self, tmp_path): + """Fresh data but no baseline → Unavailable + fresh.""" + conn = init_store(tmp_path / "test.db") + _insert_segment(conn) + _open_period(conn) + 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) + # Sample must be within FRESH_THRESHOLD_S of clock + fresh_ts = (_clock() - timedelta(seconds=FRESH_THRESHOLD_S - 10)).isoformat() + _insert_sample(conn, fresh_ts) + proj = compute_projection(conn, _clock()) + assert proj.confidence_state == ConfidenceState.UNSUPPORTED + freshness = grade_freshness(fresh_ts, _clock()) + assert freshness == "fresh" + conn.close() + + def test_limited_with_stale_data(self, tmp_path): + """Stale data with baseline → Limited or Unsupported (young regime + stale).""" + conn = init_store(tmp_path / "test.db") + _insert_baseline(conn, tbw_tb=1.0, verified=True) + _insert_segment(conn, opened_at="2026-09-01T00:00:00+00:00") + _open_period(conn) + for i in range(20): + d = (datetime(2026, 9, 1) + timedelta(days=i)).strftime("%Y-%m-%d") + _insert_day(conn, d, bw=1024*1024*100) + stale_ts = (_clock() - timedelta(days=3)).isoformat() + _insert_sample(conn, stale_ts) + proj = compute_projection(conn, _clock()) + assert proj.confidence_state != ConfidenceState.SUPPORTED + freshness = grade_freshness(stale_ts, _clock()) + assert freshness == "stale" + conn.close() + + def test_supported_with_unverified_baseline(self, tmp_path): + """Incomplete provenance → unverified baseline tier.""" + conn = init_store(tmp_path / "test.db") + # Insert baseline with incomplete provenance (missing source_url) + conn.execute( + "INSERT INTO endurance_baseline " + "(tbw_terabytes, source_url, document_revision, entry_date, model_string, " + " nominal_capacity_bytes, validated_by, verified, created_at, updated_at) " + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", + (1.0, None, "v1.0", "2026-01-01", "Samsung SSD 970 EVO Plus 1TB", + 1024000000000, None, False, + "2026-01-01T00:00:00+00:00", "2026-01-01T00:00:00+00:00"), + ) + conn.commit() + _insert_segment(conn) + _open_period(conn) + for i in range(30): + d = (datetime(2026, 9, 1) + timedelta(days=i)).strftime("%Y-%m-%d") + _insert_day(conn, d, bw=1024*1024*100, coverage=0.95, samples=24) + _insert_sample(conn, "2026-09-30T10:00:00+00:00") + proj = compute_projection(conn, _clock()) + # Incomplete provenance → UNVERIFIED tier + assert proj.baseline_tier.value == "unverified_override" + conn.close() +