xavierk/Fenris
1
0
Fork 0
Code Issues 7 Pull Requests Actions Packages Projects Releases 11 Wiki Activity

feat(status): read-only CLI status command (issue #27)

Browse Source 
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.
This commit is contained in:
xavierk
2026-09-01 23:49:23 +05:30
parent b99ebfe9dd
commit 7f006c7df7
3 changed files with 1187 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_status.py
+539
View File
@@ -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: <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
# ---------------------------------------------------------------------------
# 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