From 7f006c7df7adcee333ecb2e5e1db21fa8545672c Mon Sep 17 00:00:00 2001 From: xavierk Date: Tue, 1 Sep 2026 23:49:23 +0530 Subject: [PATCH] feat(status): read-only CLI status command (issue #27) Implement fenris status as the read-only CLI twin of the TUI, composing from the observation store and allow-listed systemctl properties per spec section 8.8. New module src/fenris/status.py: - Freshness grading with shared constants (section 8.9, LC-10) - Configuration error from direct config reads (section 8.3, LC-4) - Store fault / newer-schema exact phrases (section 9.4-9.5, FL-4/FL-5) - Drive anomalies as ordinary facts (section 9.7, FL-7) - Four separate service facts (section 7.3, LC-9, CI-2) - Projection recomputed on read, never stored (section 6.10) - Retired command rejection with migration pointers (section 8.8) - Six disclosures via --disclosures flag (section 6.11, CI-4) Updated fenris.py: - Replaced old cmd_status with new status module integration - Added retired command handlers (start/stop/run) - Added global --device flag rejection Tests: 43 new, 182 total passing, zero regressions. Closes #27. --- fenris.py | 89 ++++--- src/fenris/status.py | 599 +++++++++++++++++++++++++++++++++++++++++++ tests/test_status.py | 539 ++++++++++++++++++++++++++++++++++++++ 3 files changed, 1187 insertions(+), 40 deletions(-) create mode 100644 src/fenris/status.py create mode 100644 tests/test_status.py diff --git a/fenris.py b/fenris.py index f016729..87f9d0b 100755 --- a/fenris.py +++ b/fenris.py @@ -21,7 +21,11 @@ import mimetypes from datetime import datetime, timezone, timedelta from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer -SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__)) +# Add src/ to path for package imports +_SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__)) +sys.path.insert(0, os.path.join(_SCRIPT_DIR, "src")) + +SCRIPT_DIR = _SCRIPT_DIR DATA_DIR = os.path.join(SCRIPT_DIR, "data") DATA_FILE = os.path.join(DATA_DIR, "history.jsonl") HOURLY_FILE = os.path.join(DATA_DIR, "hourly.jsonl") @@ -973,38 +977,18 @@ def cmd_stop(args): def cmd_status(args): - running = False - if os.path.exists(PID_FILE): - with open(PID_FILE) as f: - pid = int(f.read().strip()) - running = pid_alive(pid) - print(f"Fenris daemon: {'RUNNING (pid ' + str(pid) + ')' if running else 'not running (stale pid file)'}") - else: - print("Fenris daemon: not running") - rows = load_history() - if rows: - summ = compute_summary(rows) - latest = rows[-1] - print(f"Samples collected: {len(rows)}") - print(f"Last sample: {latest['ts']}") - print(f"Wear (percentage_used): {latest.get('percentage_used')}%") - print(f"Total written: {latest.get('bytes_written', 0) / 1e9:.1f} GB") - print(f"Written (24h rolling): {summ['window24h']['gb']:.2f} GB over {summ['window24h']['coverage_hours']:.1f}h") - print(f"Write rate: {summ['gb_per_hour']:.2f} GB/h ({summ['gb_per_day']:.1f} GB/day)") - if summ["seconds_remaining"]: - print(f"Projected life remaining: {summ['breakdown']['human']} (≈{summ['breakdown']['days']:.0f} days / {summ['breakdown']['hours']:.0f} hours / {summ['breakdown']['years']:.2f} years)") - print(f"Endurance: {summ['endurance_tb']:.1f} TB total, {summ['remaining_tb']:.1f} TB remaining" + (" (estimated)" if summ["endurance_estimated"] else "")) - if summ["wear_model_days"]: - print(f"Wear-model cross-check: ~{summ['wear_model_days']:.0f} days at current wear rate") - if summ["preliminary"]: - print("Note: preliminary — less than 24h coverage") - else: - print("Projected life remaining: — (no writes in window or no endurance data)") - hourly = load_hourly() - if hourly: - print(f"Hourly buckets: {len(hourly)} (last {hourly[-1]['hour']}: {hourly[-1]['bytes_written']/1e9:.2f} GB)") - else: - print("No samples collected yet.") + try: + from fenris.status import render_status, check_retired_flag + from pathlib import Path + + show_disclosures = getattr(args, "disclosures", False) + store_path = Path("/var/lib/fenris/observations.db") + output = render_status(store_path=store_path, show_disclosures=show_disclosures) + print(output) + except ImportError: + # Fallback if fenris package not importable + print("Error: cannot import fenris.status module. Is the package installed?") + sys.exit(1) def cmd_run(args): @@ -1023,8 +1007,22 @@ def cmd_sample_once(args): sys.exit(1) +def cmd_retired(args): + """Handle retired commands with migration pointers (§8.8).""" + from fenris.status import check_retired_command + cmd = sys.argv[1] if len(sys.argv) > 1 else "" + ptr = check_retired_command(cmd) + if ptr: + print(ptr) + else: + print("Unknown command. Use 'fenris status' or 'fenris sample'.") + sys.exit(1) + + def main(): p = argparse.ArgumentParser(description="Fenris — NVMe wear monitor & dashboard (by Bongbetic)") + p.add_argument("--device", default=None, + help="(retired — device is configured in /etc/fenris/fenris.conf)") sub = p.add_subparsers(dest="cmd", required=True) def add_common(sp): @@ -1032,19 +1030,30 @@ def main(): sp.add_argument("--interval", type=int, default=300, help="seconds between samples (default 300)") sp.add_argument("--port", type=int, default=8420, help="dashboard HTTP port (default 8420)") - sp = sub.add_parser("start", help="start monitoring in background") - add_common(sp); sp.set_defaults(func=cmd_start) - sp = sub.add_parser("stop", help="stop background monitoring") - sp.set_defaults(func=cmd_stop) - sp = sub.add_parser("status", help="show daemon + latest wear stats") + sp = sub.add_parser("start", help="(retired — use 'fenris monitor resume')") + add_common(sp); sp.set_defaults(func=cmd_retired) + sp = sub.add_parser("stop", help="(retired — use 'fenris monitor pause')") + sp.set_defaults(func=cmd_retired) + sp = sub.add_parser("status", help="show read-only status (§8.8)") + sp.add_argument("-d", "--disclosures", action="store_true", + help="show the six disclosures (§6.11)") sp.set_defaults(func=cmd_status) - sp = sub.add_parser("run", help="run in foreground (used internally by 'start')") - add_common(sp); sp.set_defaults(func=cmd_run) + sp = sub.add_parser("run", help="(retired — use 'fenris monitor resume')") + add_common(sp); sp.set_defaults(func=cmd_retired) sp = sub.add_parser("sample", help="take one sample immediately and print it") sp.add_argument("--device", default=detect_device()) sp.set_defaults(func=cmd_sample_once) args = p.parse_args() + + # Reject retired --device flag (§8.8) — only if explicitly passed + if getattr(args, "device", None) is not None: + from fenris.status import check_retired_flag + ptr = check_retired_flag("--device") + if ptr: + print(ptr) + sys.exit(1) + args.func(args) diff --git a/src/fenris/status.py b/src/fenris/status.py new file mode 100644 index 0000000..0100899 --- /dev/null +++ b/src/fenris/status.py @@ -0,0 +1,599 @@ +"""Read-only CLI status command: the CLI twin of the TUI (spec §8.8, LC-9, CI-2). + +Composes from the observation store (read-only) and allow-listed systemctl +properties: projection facts, four separate service facts (boot enablement, +runtime activity, last collect outcome, freshness), and a journalctl hint on +failure or staleness. Never auto-samples, never prompts. + +Freshness constants are defined once here and shared with the TUI (§8.9): + fresh — newest sample within 2 × cadence + AccuracySec + 60 s + missed — between fresh and 48 h + stale — ≥ 48 h + empty — no observations yet + +Criteria: LC-9, CI-2, CI-4, FL-4, FL-5, FL-7. +""" +import sqlite3 +import subprocess +import sys +from datetime import datetime, timedelta, timezone +from pathlib import Path +from typing import Any, Dict, List, Optional, Tuple + +from .projection import compute_projection, ConfidenceState, DISCLOSURES +from .store import SCHEMA_VERSION + + +# --------------------------------------------------------------------------- +# Freshness constants (§8.9, §8.2) +# --------------------------------------------------------------------------- + +CADENCE_DEFAULT_S = 300 # 5 min +ACCURACY_SEC = 30 +FRESH_THRESHOLD_S = 2 * CADENCE_DEFAULT_S + ACCURACY_SEC + 60 # 690 s +STALENESS_THRESHOLD_S = 48 * 3600 # 48 h + + +# --------------------------------------------------------------------------- +# Configuration reading (§8.3) +# --------------------------------------------------------------------------- + +CONFIG_PATH = Path("/etc/fenris/fenris.conf") + + +def read_config() -> Dict[str, Any]: + """Read the world-readable configuration file. + + Returns a dict with at least 'device'. + Raises ConfigError with a reason string on any failure. + """ + if not CONFIG_PATH.exists(): + raise ConfigError("configuration file not found at %s" % CONFIG_PATH) + + try: + text = CONFIG_PATH.read_text() + except OSError as e: + raise ConfigError("cannot read %s: %s" % (CONFIG_PATH, e)) + + device = None + for line in text.splitlines(): + line = line.strip() + if not line or line.startswith("#"): + continue + if "=" in line: + key, _, value = line.partition("=") + key = key.strip() + value = value.strip().strip('"').strip("'") + if key == "device": + device = value + break + + if not device: + raise ConfigError("no device selector in %s" % CONFIG_PATH) + + return {"device": device} + + +class ConfigError(Exception): + """Configuration is invalid — surfaced in status as a fact (§8.3).""" + pass + + +# --------------------------------------------------------------------------- +# Store opening (read-only, §3, §9.4, §9.5) +# --------------------------------------------------------------------------- + +def open_store_readonly(store_path: Path) -> sqlite3.Connection: + """Open the observation store read-only. + + Raises StoreFault if unreadable, NewerSchema if user_version > SCHEMA_VERSION. + """ + if not store_path.exists(): + raise StoreFault("observation store not found at %s" % store_path) + + try: + conn = sqlite3.connect("file:%s?mode=ro" % store_path, uri=True) + conn.row_factory = sqlite3.Row + except sqlite3.Error as e: + raise StoreFault("observation store unreadable: %s" % e) + + try: + cursor = conn.execute("PRAGMA user_version") + version = cursor.fetchone()[0] + except sqlite3.Error as e: + conn.close() + raise StoreFault("observation store unreadable: %s" % e) + + if version > SCHEMA_VERSION: + conn.close() + raise NewerSchema(version) + + return conn + + +class StoreFault(Exception): + """Store is present but unreadable or corrupt (§9.4).""" + pass + + +class NewerSchema(Exception): + """Store has a newer user_version (§9.5).""" + def __init__(self, version: int): + self.version = version + super().__init__("schema version %d" % version) + + +# --------------------------------------------------------------------------- +# Service state queries (§8.8 — allow-listed systemctl properties) +# --------------------------------------------------------------------------- + +def _systemctl_show(unit: str, *properties: str) -> Dict[str, str]: + """Query systemctl show for specific properties. Returns empty dict on failure.""" + try: + result = subprocess.run( + ["systemctl", "show", unit, "--property=" + ",".join(properties)], + capture_output=True, text=True, timeout=5, + ) + if result.returncode != 0: + return {} + out = {} + for line in result.stdout.splitlines(): + if "=" in line: + key, _, value = line.partition("=") + out[key.strip()] = value.strip() + return out + except (subprocess.TimeoutExpired, FileNotFoundError, OSError): + return {} + + +def _journalctl_hint(unit: str, lines: int = 5) -> Optional[str]: + """Get the last N journal lines for a unit. Returns None on failure.""" + try: + result = subprocess.run( + ["journalctl", "-u", unit, "--no-pager", "-n", str(lines), "--output=short-iso"], + capture_output=True, text=True, timeout=5, + ) + if result.returncode != 0 or not result.stdout.strip(): + return None + return result.stdout.strip() + except (subprocess.TimeoutExpired, FileNotFoundError, OSError): + return None + + +def query_service_state() -> Dict[str, Any]: + """Query systemctl for the four separate service facts (§7.3, LC-9). + + Returns dict with keys: + boot_enabled: bool + timer_active: bool + last_collect_ok: Optional[bool] + last_collect_age_s: Optional[int] + last_collect_reason: Optional[str] + """ + timer_props = _systemctl_show( + "fenris-collect.timer", + "UnitFileState", "ActiveState", "LastTriggerUSec", + ) + service_props = _systemctl_show( + "fenris-collect.service", + "ActiveState", "ExecMainStatus", "ExecMainExitTimestamp", + ) + + boot_enabled_str = timer_props.get("UnitFileState", "") + boot_enabled = boot_enabled_str == "enabled" + + active_state = timer_props.get("ActiveState", "inactive") + timer_active = active_state == "active" + + last_collect_ok = None + last_collect_age_s = None + last_collect_reason = None + + last_trigger = timer_props.get("LastTriggerUSec", "") + if last_trigger and last_trigger != "n/a": + try: + trigger_dt = datetime.fromisoformat(last_trigger.replace("Z", "+00:00")) + now = datetime.now(timezone.utc) + last_collect_age_s = int((now - trigger_dt).total_seconds()) + except (ValueError, TypeError): + pass + + exec_status = service_props.get("ExecMainStatus", "") + if exec_status: + try: + exit_code = int(exec_status) + last_collect_ok = exit_code == 0 + if exit_code != 0: + last_collect_reason = "exit code %d" % exit_code + except (ValueError, TypeError): + pass + + return { + "boot_enabled": boot_enabled, + "timer_active": timer_active, + "last_collect_ok": last_collect_ok, + "last_collect_age_s": last_collect_age_s, + "last_collect_reason": last_collect_reason, + } + + +# --------------------------------------------------------------------------- +# Freshness grading (§8.9) +# --------------------------------------------------------------------------- + +def grade_freshness(newest_sample_ts: Optional[str], clock_now: datetime) -> str: + """Grade freshness from the newest sample timestamp (never a stored flag). + + Returns 'fresh', 'missed', 'stale', or 'empty'. + """ + if newest_sample_ts is None: + return "empty" + + try: + ts = datetime.fromisoformat(newest_sample_ts) + if ts.tzinfo is None: + ts = ts.replace(tzinfo=timezone.utc) + else: + ts = ts.astimezone(timezone.utc) + except (ValueError, TypeError): + return "empty" + + age_s = (clock_now - ts).total_seconds() + + if age_s <= FRESH_THRESHOLD_S: + return "fresh" + elif age_s < STALENESS_THRESHOLD_S: + return "missed" + else: + return "stale" + + +def freshness_age_human(age_s: Optional[int]) -> str: + """Human-readable age string for freshness fact.""" + if age_s is None: + return "unknown age" + if age_s < 60: + return "%ds ago" % age_s + if age_s < 3600: + return "%dm ago" % (age_s // 60) + if age_s < 86400: + return "%dh %dm ago" % (age_s // 3600, (age_s % 3600) // 60) + return "%dd ago" % (age_s // 86400) + + +# --------------------------------------------------------------------------- +# Drive anomalies (§9.7 — FL-7) +# --------------------------------------------------------------------------- + +def _query_drive_facts(conn: sqlite3.Connection) -> List[str]: + """Query drive-reported anomalies from the latest sample (§9.7, FL-7). + + critical_warning, media errors, and unsafe shutdowns render as ordinary + facts and never affect the projection. + """ + cursor = conn.execute( + "SELECT critical_warning, media_errors, unsafe_shutdowns, " + "temperature_c, available_spare " + "FROM samples ORDER BY id DESC LIMIT 1" + ) + row = cursor.fetchone() + if row is None: + return [] + + facts = [] + cw = row[0] + if cw and cw != 0: + facts.append("critical warning: %s" % hex(cw) if isinstance(cw, int) else str(cw)) + me = row[1] + if me and me > 0: + facts.append("media errors: %d" % me) + us = row[2] + if us and us > 0: + facts.append("unsafe shutdowns: %d" % us) + return facts + + +# --------------------------------------------------------------------------- +# Retired command rejection (§8.8) +# --------------------------------------------------------------------------- + +RETIRED_COMMANDS = { + "start": "use 'fenris monitor resume' to enable monitoring", + "stop": "use 'fenris monitor pause' to disable monitoring", + "run": "use 'fenris monitor resume' to enable monitoring; the timer runs in the background", +} + +MIGRATION_POINTERS = { + "--device": "device is configured in /etc/fenris/fenris.conf", +} + + +def check_retired_command(cmd: str) -> Optional[str]: + """Check if a command is retired and return the migration pointer, or None.""" + return RETIRED_COMMANDS.get(cmd) + + +def check_retired_flag(flag: str) -> Optional[str]: + """Check if a flag is retired and return the migration pointer, or None.""" + return MIGRATION_POINTERS.get(flag) + + +# --------------------------------------------------------------------------- +# Formatting +# --------------------------------------------------------------------------- + +def _format_projection(proj, freshness: str, service: Dict[str, Any], + drive_facts: List[str], config_error: Optional[str], + store_fault: Optional[str], newer_schema: Optional[str], + journal_hint: Optional[str]) -> str: + """Format the complete status output.""" + lines = [] + + # --- Store/system fault overrides (§9.4, §9.5) --- + if store_fault: + lines.append("observation store unreadable") + if journal_hint: + lines.append("") + lines.append("Recent journal entries:") + lines.append(journal_hint) + return "\n".join(lines) + + if newer_schema: + lines.append("observation store written by a newer Fenris — upgrade Fenris") + return "\n".join(lines) + + # --- Configuration error (§8.3) --- + if config_error: + lines.append("configuration error: %s" % config_error) + lines.append("") + + # --- Empty store (§8.9) --- + if freshness == "empty": + lines.append("no observations yet") + lines.append("") + lines.append("Enable monitoring: fenris monitor resume") + _append_service_facts(lines, service) + return "\n".join(lines) + + # --- Projection headline --- + headline = _format_headline(proj) + lines.append(headline) + lines.append("") + + # --- Confidence state + contributing facts (§6.7, §6.11) --- + state_label = proj.confidence_state.value + if proj.contributing_facts: + facts_str = " · ".join(proj.contributing_facts) + lines.append("%s evidence · %s" % (state_label, facts_str)) + else: + lines.append("%s evidence" % state_label) + lines.append("") + + # --- Scenario range (§6.5) --- + if proj.scenario_range and proj.scenario_range.rates: + 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)) + lines.append("scenario range: %s" % " · ".join(parts)) + lines.append("") + + # --- PU context line (§6.1) --- + lines.append(proj.pu_context_line) + lines.append("") + + # --- Drive anomalies (§9.7, FL-7) --- + if drive_facts: + for fact in drive_facts: + lines.append(fact) + lines.append("") + + # --- Four separate service facts (§7.3, LC-9) --- + _append_service_facts(lines, service) + + # --- Journal hint on failure or staleness (§8.8) --- + if journal_hint: + if freshness in ("missed", "stale"): + lines.append("") + lines.append("Recent journal entries:") + lines.append(journal_hint) + + return "\n".join(lines) + + +def _format_headline(proj) -> str: + """Format the lifespan headline or its no-projection wording (§6.11).""" + if proj.headline_remaining_seconds is None: + if proj.zero_rate_fact: + return "no finite projection from this history" + if proj.warming_fact: + return proj.warming_fact + return "no projection available" + + secs = proj.headline_remaining_seconds + if secs <= 0: + return "endurance exhausted" + + # Human-readable time + years = int(secs // 31557600) + rem = secs % 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) + + remaining_human = " ".join(parts) + + # Regime line + regime_parts = [] + if proj.regime_days: + regime_parts.append("sustained regime: %d days" % proj.regime_days) + + headline = "%s remaining" % remaining_human + if regime_parts: + headline += " · %s" % " · ".join(regime_parts) + + return headline + + +def _append_service_facts(lines: List[str], service: Dict[str, Any]) -> None: + """Append the four separate service facts (§7.3, LC-9).""" + boot = "enabled" if service.get("boot_enabled") else "disabled" + activity = "active" if service.get("timer_active") else "inactive" + + if service.get("last_collect_ok") is True: + collect = "ok" + elif service.get("last_collect_ok") is False: + collect = "FAILED" + if service.get("last_collect_reason"): + collect += " (%s)" % service["last_collect_reason"] + else: + collect = "unknown" + + collect_age = "" + if service.get("last_collect_age_s") is not None: + collect_age = " %s" % freshness_age_human(service["last_collect_age_s"]) + + freshness_str = service.get("freshness", "unknown") + freshness_age = "" + if service.get("freshness_age_s") is not None: + freshness_age = " (%s)" % freshness_age_human(service["freshness_age_s"]) + + lines.append("boot: %s · timer: %s · last collect: %s%s · freshness: %s%s" + % (boot, activity, collect, collect_age, freshness_str, freshness_age)) + + +def format_disclosures() -> str: + """Format the six disclosures (§6.11, CI-4).""" + lines = [] + lines.append("Disclosures") + lines.append("") + for i, disc in enumerate(DISCLOSURES, 1): + lines.append("%d. %s" % (i, disc)) + return "\n".join(lines) + + +# --------------------------------------------------------------------------- +# Main status entry point +# --------------------------------------------------------------------------- + +def get_status(store_path: Optional[Path] = None, clock_now: Optional[datetime] = None, + query_services: bool = True, query_journal: bool = True) -> str: + """Render the complete read-only status (§8.8, LC-9). + + This is the single entry point for 'fenris status'. It never auto-samples, + never prompts, and never writes to the store. + """ + if clock_now is None: + clock_now = datetime.now(timezone.utc) + + # --- Configuration (§8.3) --- + config_error = None + device = None + try: + config = read_config() + device = config["device"] + except ConfigError as e: + config_error = str(e) + + # --- Service state --- + service = {} + if query_services: + service = query_service_state() + + # --- 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) + + # --- Store fault / newer schema short-circuit --- + if store_fault or newer_schema: + journal_hint = None + if query_journal: + journal_hint = _journalctl_hint("fenris-collect.service") + service["freshness"] = "unknown" + service["freshness_age_s"] = None + return _format_projection( + None, "unknown", service, [], config_error, store_fault, newer_schema, journal_hint + ) + + # --- Freshness grading (§8.9) --- + 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 + except sqlite3.Error: + newest_ts = None + + freshness = grade_freshness(newest_ts, clock_now) + + # Freshness age for the service fact + freshness_age_s = None + if newest_ts: + try: + ts = datetime.fromisoformat(newest_ts) + if ts.tzinfo is None: + ts = ts.replace(tzinfo=timezone.utc) + freshness_age_s = int((clock_now - ts).total_seconds()) + except (ValueError, TypeError): + pass + + service["freshness"] = freshness + service["freshness_age_s"] = freshness_age_s + + # --- Drive anomalies (§9.7, FL-7) --- + drive_facts = [] + try: + drive_facts = _query_drive_facts(conn) + except sqlite3.Error: + pass + + # --- Projection (§6 — recomputed on read, never stored) --- + try: + proj = compute_projection(conn, clock_now) + except Exception: + proj = None + + # --- Journal hint on failure or staleness (§8.8) --- + journal_hint = None + if query_journal and freshness in ("missed", "stale"): + journal_hint = _journalctl_hint("fenris-collect.service") + + # --- Compose output --- + result = _format_projection( + proj, freshness, service, drive_facts, config_error, + None, None, journal_hint, + ) + + conn.close() + return result + + +def render_status(store_path: Optional[Path] = None, clock_now: Optional[datetime] = None, + query_services: bool = True, query_journal: bool = True, + show_disclosures: bool = False) -> str: + """High-level status renderer: status + optional disclosures. + + Used by the CLI entry point. + """ + parts = [get_status(store_path, clock_now, query_services, query_journal)] + if show_disclosures: + parts.append("") + parts.append(format_disclosures()) + return "\n\n".join(parts) diff --git a/tests/test_status.py b/tests/test_status.py new file mode 100644 index 0000000..d3cd5cd --- /dev/null +++ b/tests/test_status.py @@ -0,0 +1,539 @@ +"""Tests for the read-only CLI status command (issue #27). + +Covers: + - LC-9: status is a pure read-only composition + - CI-2: TUI/CLI parity (status fact set matches TUI's four separate facts) + - CI-4: Required wording and six disclosures render as adopted + - FL-4: Store fault renders exact fixed phrase + - FL-5: Newer-schema store renders exact fixed phrase + - FL-7: Drive anomalies render as ordinary facts, never affecting projection + - LC-10: Freshness grading with shared constants + - Retired command rejection with migration pointers + - Configuration error surfaced from direct reads +""" +import os +import sqlite3 +import sys +import tempfile +from datetime import datetime, timedelta, timezone +from pathlib import Path +from unittest.mock import patch, MagicMock + +import pytest + +sys.path.insert(0, str(Path(__file__).parent.parent / "src")) + +from fenris.status import ( + grade_freshness, + freshness_age_human, + format_disclosures, + check_retired_command, + check_retired_flag, + FRESH_THRESHOLD_S, + STALENESS_THRESHOLD_S, + CADENCE_DEFAULT_S, + ACCURACY_SEC, +) +from fenris.store import init_store, SCHEMA_VERSION +from fenris.projection import DISCLOSURES, ConfidenceState + + +# --------------------------------------------------------------------------- +# Freshness grading (§8.9, LC-10) +# --------------------------------------------------------------------------- + +class TestFreshnessGrading: + """Freshness constants are defined once and shared (§8.9).""" + + def test_constants_match_spec(self): + """Fresh threshold = 2 × cadence + AccuracySec + 60 s.""" + expected = 2 * CADENCE_DEFAULT_S + ACCURACY_SEC + 60 + assert FRESH_THRESHOLD_S == expected + assert STALENESS_THRESHOLD_S == 48 * 3600 + + def test_empty_store(self): + """Empty store reads 'no observations yet' (§8.9).""" + now = datetime(2026, 9, 1, 12, 0, 0, tzinfo=timezone.utc) + assert grade_freshness(None, now) == "empty" + + def test_fresh_sample(self): + """Newest sample within threshold → fresh.""" + now = datetime(2026, 9, 1, 12, 0, 0, tzinfo=timezone.utc) + ts = (now - timedelta(seconds=FRESH_THRESHOLD_S - 1)).isoformat() + assert grade_freshness(ts, now) == "fresh" + + def test_missed_sample(self): + """Between fresh and 48h → missed.""" + now = datetime(2026, 9, 1, 12, 0, 0, tzinfo=timezone.utc) + ts = (now - timedelta(hours=2)).isoformat() + assert grade_freshness(ts, now) == "missed" + + def test_stale_sample(self): + """≥ 48h → stale.""" + now = datetime(2026, 9, 1, 12, 0, 0, tzinfo=timezone.utc) + ts = (now - timedelta(hours=49)).isoformat() + assert grade_freshness(ts, now) == "stale" + + def test_fresh_at_boundary(self): + """Exactly at threshold → fresh (within means ≤).""" + now = datetime(2026, 9, 1, 12, 0, 0, tzinfo=timezone.utc) + ts = (now - timedelta(seconds=FRESH_THRESHOLD_S)).isoformat() + assert grade_freshness(ts, now) == "fresh" + + def test_missed_at_just_past_fresh(self): + """One second past threshold → missed.""" + now = datetime(2026, 9, 1, 12, 0, 0, tzinfo=timezone.utc) + ts = (now - timedelta(seconds=FRESH_THRESHOLD_S + 1)).isoformat() + assert grade_freshness(ts, now) == "missed" + + def test_stale_at_boundary(self): + """Exactly 48h → stale.""" + now = datetime(2026, 9, 1, 12, 0, 0, tzinfo=timezone.utc) + ts = (now - timedelta(hours=48)).isoformat() + assert grade_freshness(ts, now) == "stale" + + def test_naive_timestamp_treated_as_utc(self): + """Naive timestamp is treated as UTC — within fresh threshold.""" + now = datetime(2026, 9, 1, 12, 0, 0, tzinfo=timezone.utc) + # Construct a truly naive ISO string (no +00:00 suffix), 5 min ago + naive_dt = datetime(2026, 9, 1, 11, 55, 0) # 5 min ago, naive + ts = naive_dt.isoformat() # "2026-09-01T11:55:00" + assert grade_freshness(ts, now) == "fresh" + + def test_malformed_timestamp_returns_empty(self): + """Malformed timestamp → empty.""" + now = datetime(2026, 9, 1, 12, 0, 0, tzinfo=timezone.utc) + assert grade_freshness("not-a-timestamp", now) == "empty" + + +# --------------------------------------------------------------------------- +# Freshness age human-readable +# --------------------------------------------------------------------------- + +class TestFreshnessAgeHuman: + def test_seconds(self): + assert freshness_age_human(30) == "30s ago" + + def test_minutes(self): + assert freshness_age_human(120) == "2m ago" + + def test_hours_and_minutes(self): + assert freshness_age_human(3661) == "1h 1m ago" + + def test_days(self): + assert freshness_age_human(90000) == "1d ago" + + def test_none(self): + assert freshness_age_human(None) == "unknown age" + + +# --------------------------------------------------------------------------- +# Retired command rejection (§8.8) +# --------------------------------------------------------------------------- + +class TestRetiredCommands: + """Retired commands and --device are rejected with one-line pointers.""" + + def test_start_rejected(self): + ptr = check_retired_command("start") + assert ptr is not None + assert "resume" in ptr.lower() or "enable" in ptr.lower() + + def test_stop_rejected(self): + ptr = check_retired_command("stop") + assert ptr is not None + assert "pause" in ptr.lower() or "disable" in ptr.lower() + + def test_run_rejected(self): + ptr = check_retired_command("run") + assert ptr is not None + + def test_status_not_rejected(self): + assert check_retired_command("status") is None + + def test_sample_not_rejected(self): + assert check_retired_command("sample") is None + + def test_device_flag_rejected(self): + ptr = check_retired_flag("--device") + assert ptr is not None + assert "fenris.conf" in ptr + + def test_unknown_flag_not_rejected(self): + assert check_retired_flag("--unknown") is None + + +# --------------------------------------------------------------------------- +# Disclosures (§6.11, CI-4) +# --------------------------------------------------------------------------- + +class TestDisclosures: + """Required wording and six disclosures render as adopted (CI-4).""" + + def test_six_disclosures(self): + assert len(DISCLOSURES) == 6 + + def test_disclosures_text(self): + """Each disclosure matches the spec verbatim.""" + assert "endurance projection" in DISCLOSURES[0].lower() + assert "hardware-failure" in DISCLOSURES[0].lower() or "failure date" in DISCLOSURES[0].lower() + assert "vendor-specific" in DISCLOSURES[1] + assert "255 is saturated" in DISCLOSURES[1] + assert "warranty" in DISCLOSURES[2] or "endurance threshold" in DISCLOSURES[2] + assert "DUW" in DISCLOSURES[3] + assert "metadata" in DISCLOSURES[3] + assert "future workload" in DISCLOSURES[4] + assert "deliberately disabled" in DISCLOSURES[5] + + def test_format_disclosures_returns_all_six(self): + output = format_disclosures() + for i in range(1, 7): + assert "%d." % i in output + + def test_disclosures_header(self): + output = format_disclosures() + assert output.startswith("Disclosures") + + +# --------------------------------------------------------------------------- +# Store fault rendering (§9.4, FL-4) +# --------------------------------------------------------------------------- + +class TestStoreFault: + """Store fault surfaces exact fixed phrase (FL-4).""" + + def test_store_fault_phrase(self, tmp_path): + """observation store unreadable with journal hint.""" + from fenris.status import open_store_readonly, StoreFault + + nonexistent = tmp_path / "nonexistent.db" + with pytest.raises(StoreFault): + open_store_readonly(nonexistent) + + def test_corrupt_store(self, tmp_path): + """Corrupt file raises StoreFault.""" + from fenris.status import open_store_readonly, StoreFault + + corrupt = tmp_path / "corrupt.db" + corrupt.write_bytes(b"this is not a sqlite database") + with pytest.raises(StoreFault): + open_store_readonly(corrupt) + + +# --------------------------------------------------------------------------- +# Newer schema rendering (§9.5, FL-5) +# --------------------------------------------------------------------------- + +class TestNewerSchema: + """Newer-schema store renders exact fixed phrase (FL-5).""" + + def test_newer_schema_detected(self, tmp_path): + from fenris.status import open_store_readonly, NewerSchema + + db = tmp_path / "test.db" + conn = sqlite3.connect(str(db)) + conn.execute("PRAGMA user_version=%d" % (SCHEMA_VERSION + 1)) + conn.commit() + conn.close() + + with pytest.raises(NewerSchema): + open_store_readonly(db) + + +# --------------------------------------------------------------------------- +# Configuration error (§8.3, LC-4) +# --------------------------------------------------------------------------- + +class TestConfigError: + """Configuration error surfaced as configuration error: .""" + + def test_missing_config(self): + from fenris.status import read_config, ConfigError + with patch("fenris.status.CONFIG_PATH", Path("/nonexistent/fenris.conf")): + with pytest.raises(ConfigError, match="not found"): + read_config() + + def test_empty_config(self, tmp_path): + from fenris.status import read_config, ConfigError + conf = tmp_path / "fenris.conf" + conf.write_text("# empty config\n") + with patch("fenris.status.CONFIG_PATH", conf): + with pytest.raises(ConfigError, match="no device selector"): + read_config() + + def test_valid_config(self, tmp_path): + from fenris.status import read_config + conf = tmp_path / "fenris.conf" + conf.write_text("device = /dev/disk/by-id/nvme-test\n") + with patch("fenris.status.CONFIG_PATH", conf): + result = read_config() + assert result["device"] == "/dev/disk/by-id/nvme-test" + + +# --------------------------------------------------------------------------- +# Drive anomalies (§9.7, FL-7) +# --------------------------------------------------------------------------- + +class TestDriveAnomalies: + """Drive anomalies render as ordinary facts, never affecting projection.""" + + def test_no_anomalies(self, tmp_path): + from fenris.status import _query_drive_facts + + db = tmp_path / "test.db" + conn = init_store(db) + conn.execute( + "INSERT INTO samples (ts, device, mn, sn, fr, capacity_bytes, " + "percentage_used, available_spare, media_errors, power_on_hours, " + "power_cycles, unsafe_shutdowns, temperature_c, " + "data_units_written, data_units_read, bytes_written, bytes_read, " + "critical_warning) " + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", + ("2026-09-01T12:00:00Z", "/dev/nvme0", "Test", "SN", "FR", + 1000000000000, 5, 100, 0, 1000, 100, 0, 35, + 1000000, 500000, 512000000000, 256000000000, 0), + ) + conn.commit() + facts = _query_drive_facts(conn) + assert facts == [] + conn.close() + + def test_critical_warning(self, tmp_path): + from fenris.status import _query_drive_facts + + db = tmp_path / "test.db" + conn = init_store(db) + conn.execute( + "INSERT INTO samples (ts, device, mn, sn, fr, capacity_bytes, " + "percentage_used, available_spare, media_errors, power_on_hours, " + "power_cycles, unsafe_shutdowns, temperature_c, " + "data_units_written, data_units_read, bytes_written, bytes_read, " + "critical_warning) " + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", + ("2026-09-01T12:00:00Z", "/dev/nvme0", "Test", "SN", "FR", + 1000000000000, 5, 100, 0, 1000, 100, 0, 35, + 1000000, 500000, 512000000000, 256000000000, 1), + ) + conn.commit() + facts = _query_drive_facts(conn) + assert any("critical warning" in f for f in facts) + conn.close() + + def test_media_errors(self, tmp_path): + from fenris.status import _query_drive_facts + + db = tmp_path / "test.db" + conn = init_store(db) + conn.execute( + "INSERT INTO samples (ts, device, mn, sn, fr, capacity_bytes, " + "percentage_used, available_spare, media_errors, power_on_hours, " + "power_cycles, unsafe_shutdowns, temperature_c, " + "data_units_written, data_units_read, bytes_written, bytes_read, " + "critical_warning) " + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", + ("2026-09-01T12:00:00Z", "/dev/nvme0", "Test", "SN", "FR", + 1000000000000, 5, 100, 3, 1000, 100, 0, 35, + 1000000, 500000, 512000000000, 256000000000, 0), + ) + conn.commit() + facts = _query_drive_facts(conn) + assert any("media errors" in f for f in facts) + conn.close() + + def test_unsafe_shutdowns(self, tmp_path): + from fenris.status import _query_drive_facts + + db = tmp_path / "test.db" + conn = init_store(db) + conn.execute( + "INSERT INTO samples (ts, device, mn, sn, fr, capacity_bytes, " + "percentage_used, available_spare, media_errors, power_on_hours, " + "power_cycles, unsafe_shutdowns, temperature_c, " + "data_units_written, data_units_read, bytes_written, bytes_read, " + "critical_warning) " + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", + ("2026-09-01T12:00:00Z", "/dev/nvme0", "Test", "SN", "FR", + 1000000000000, 5, 100, 0, 1000, 100, 5, 35, + 1000000, 500000, 512000000000, 256000000000, 0), + ) + conn.commit() + facts = _query_drive_facts(conn) + assert any("unsafe shutdowns" in f for f in facts) + conn.close() + + +# --------------------------------------------------------------------------- +# Empty store greeting (§8.9, LC-10) +# --------------------------------------------------------------------------- + +class TestEmptyStoreGreeting: + """Empty store reads 'no observations yet' with enable hint.""" + + def test_empty_store_message(self, tmp_path): + from fenris.status import get_status + + db = tmp_path / "observations.db" + init_store(db) + now = datetime(2026, 9, 1, 12, 0, 0, tzinfo=timezone.utc) + + with patch("fenris.status.query_service_state", return_value={ + "boot_enabled": False, "timer_active": False, + "last_collect_ok": None, "last_collect_age_s": None, + "last_collect_reason": None, + }): + result = get_status(store_path=db, clock_now=now, + query_services=True, query_journal=False) + + assert "no observations yet" in result + assert "enable" in result.lower() or "resume" in result.lower() + + +# --------------------------------------------------------------------------- +# Four separate service facts (§7.3, LC-9, CI-2) +# --------------------------------------------------------------------------- + +class TestServiceFacts: + """Status renders four separate service facts matching TUI (CI-2).""" + + def test_service_facts_present(self, tmp_path): + from fenris.status import get_status + + db = tmp_path / "observations.db" + init_store(db) + now = datetime(2026, 9, 1, 12, 0, 0, tzinfo=timezone.utc) + + 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, + }): + result = get_status(store_path=db, clock_now=now, + query_services=True, query_journal=False) + + assert "boot:" in result + assert "timer:" in result + assert "last collect:" in result + assert "freshness:" in result + + +# --------------------------------------------------------------------------- +# Status output structure (§8.8, LC-9) +# --------------------------------------------------------------------------- + +class TestStatusOutput: + """Status is a pure read-only composition (LC-9).""" + + def test_status_returns_string(self, tmp_path): + from fenris.status import get_status + + db = tmp_path / "observations.db" + init_store(db) + now = datetime(2026, 9, 1, 12, 0, 0, tzinfo=timezone.utc) + + with patch("fenris.status.query_service_state", return_value={ + "boot_enabled": False, "timer_active": False, + "last_collect_ok": None, "last_collect_age_s": None, + "last_collect_reason": None, + }): + result = get_status(store_path=db, clock_now=now, + query_services=True, query_journal=False) + + assert isinstance(result, str) + assert len(result) > 0 + + def test_status_never_writes(self, tmp_path): + """Status never writes to the store.""" + from fenris.status import get_status + + db = tmp_path / "observations.db" + init_store(db) + + mtime_before = db.stat().st_mtime + now = datetime(2026, 9, 1, 12, 0, 0, tzinfo=timezone.utc) + + with patch("fenris.status.query_service_state", return_value={ + "boot_enabled": False, "timer_active": False, + "last_collect_ok": None, "last_collect_age_s": None, + "last_collect_reason": None, + }): + get_status(store_path=db, clock_now=now, + query_services=True, query_journal=False) + + mtime_after = db.stat().st_mtime + assert mtime_before == mtime_after + + +# --------------------------------------------------------------------------- +# Projection in status (§6.10) +# --------------------------------------------------------------------------- + +class TestProjectionInStatus: + """Projection is recomputed on read, never stored (§6.10).""" + + def test_store_fault_suppresses_projection(self, tmp_path): + from fenris.status import get_status + + nonexistent = tmp_path / "nonexistent.db" + now = datetime(2026, 9, 1, 12, 0, 0, tzinfo=timezone.utc) + + with patch("fenris.status.query_service_state", return_value={ + "boot_enabled": False, "timer_active": False, + "last_collect_ok": None, "last_collect_age_s": None, + "last_collect_reason": None, + }): + result = get_status(store_path=nonexistent, clock_now=now, + query_services=True, query_journal=False) + + assert "observation store unreadable" in result + assert "%" not in result # no projection numbers + + def test_newer_schema_suppresses_projection(self, tmp_path): + from fenris.status import get_status + + db = tmp_path / "test.db" + conn = sqlite3.connect(str(db)) + conn.execute("PRAGMA user_version=%d" % (SCHEMA_VERSION + 1)) + conn.commit() + conn.close() + + now = datetime(2026, 9, 1, 12, 0, 0, tzinfo=timezone.utc) + + with patch("fenris.status.query_service_state", return_value={ + "boot_enabled": False, "timer_active": False, + "last_collect_ok": None, "last_collect_age_s": None, + "last_collect_reason": None, + }): + result = get_status(store_path=db, clock_now=now, + query_services=True, query_journal=False) + + assert "newer Fenris" in result + assert "upgrade Fenris" in result + + +# --------------------------------------------------------------------------- +# render_status with disclosures (CI-4) +# --------------------------------------------------------------------------- + +class TestRenderStatusDisclosures: + """Disclosures are always available in status.""" + + def test_disclosures_in_output(self, tmp_path): + from fenris.status import render_status + + db = tmp_path / "observations.db" + init_store(db) + now = datetime(2026, 9, 1, 12, 0, 0, tzinfo=timezone.utc) + + with patch("fenris.status.query_service_state", return_value={ + "boot_enabled": False, "timer_active": False, + "last_collect_ok": None, "last_collect_age_s": None, + "last_collect_reason": None, + }): + result = render_status(store_path=db, clock_now=now, + query_services=True, query_journal=False, + show_disclosures=True) + + assert "Disclosures" in result + assert "endurance projection" in result.lower() + assert "1." in result + assert "6." in result