"""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 sqlite3 import sys from datetime import datetime, timedelta, timezone from pathlib import Path from unittest.mock import patch 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 # --------------------------------------------------------------------------- # 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 def test_continuity_reports_boot_enabled_independently_of_runtime(self, tmp_path): """Status names reboot continuity while retaining the timer fact.""" 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": 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 "monitoring: active in background · persists across reboots" in result assert "timer: inactive" in result def test_continuity_and_deliberate_pause_are_reported_separately(self, tmp_path): """Only a sanctioned user_disabled period renders the paused wording.""" from fenris.status import get_status db = tmp_path / "observations.db" conn = init_store(db) conn.execute( "INSERT INTO monitoring_periods (started_at, ended_at, end_cause) " "VALUES (?, ?, ?)", ("2026-09-01T09:00:00+00:00", "2026-09-01T10:00:00+00:00", "user_disabled"), ) 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 "monitoring: does not start on next boot" in result assert "monitoring: paused — deliberate disable" in result assert "paused time is excluded from your usage habit · resume: fenris monitor resume" in result def test_raw_system_state_without_user_disabled_is_not_a_deliberate_pause(self, tmp_path): """A non-sanctioned stop never acquires the deliberate-disable label.""" from fenris.status import get_status db = tmp_path / "observations.db" conn = init_store(db) conn.execute( "INSERT INTO monitoring_periods (started_at, ended_at, end_cause) " "VALUES (?, ?, ?)", ("2026-09-01T09:00:00+00:00", "2026-09-01T10:00:00+00:00", "migrated"), ) 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 "monitoring: paused — deliberate disable" not in result def test_resumed_open_period_clears_a_previous_deliberate_pause(self, tmp_path): """A later sanctioned resume takes precedence over an older pause.""" from fenris.status import get_status db = tmp_path / "observations.db" conn = init_store(db) conn.execute( "INSERT INTO monitoring_periods (started_at, ended_at, end_cause) " "VALUES (?, ?, ?)", ("2026-09-01T09:00:00+00:00", "2026-09-01T10:00:00+00:00", "user_disabled"), ) conn.execute( "INSERT INTO monitoring_periods (started_at) VALUES (?)", ("2026-09-01T11:00:00+00:00",), ) 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": True, "timer_active": True, "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 "monitoring: paused — deliberate disable" not in result def test_live_enabled_service_suppresses_a_stale_pause_marker(self, tmp_path): """A raw re-enable cannot leave a contradictory paused presentation.""" from fenris.status import get_status db = tmp_path / "observations.db" conn = init_store(db) conn.execute( "INSERT INTO monitoring_periods (started_at, ended_at, end_cause) " "VALUES (?, ?, ?)", ("2026-09-01T09:00:00+00:00", "2026-09-01T10:00:00+00:00", "user_disabled"), ) 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": True, "timer_active": True, "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 "monitoring: paused — deliberate disable" not 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_excludes_tui_identity_and_auth_notice(self, tmp_path): """CLI status never renders TUI-only identity or launch guidance.""" 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) for tui_only in ( "Fenris — NVMe endurance monitor", "by Bongbetic", "privileged actions will prompt for authentication (polkit)", ): assert tui_only not in result 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" nonexistent.write_bytes(b"not a SQLite database") 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