689 lines
27 KiB
Python
689 lines
27 KiB
Python
"""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: <reason>."""
|
||
|
||
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
|