720 lines
29 KiB
Python
720 lines
29 KiB
Python
"""Cross-cutting acceptance sweep (issue #32).
|
||
|
||
Systematic verification of every acceptance criterion that spans multiple
|
||
subsystems. Grouped by criterion ID; each test cites its clause.
|
||
|
||
CI-1 Exhaustive state matrix: confidence × freshness × baseline tier
|
||
CI-2 TUI/CLI parity: identical outcomes and wording
|
||
CI-3 Prohibition set: automated structural checks
|
||
CI-4 Required wording and six disclosures in both views
|
||
"""
|
||
import re
|
||
import sqlite3
|
||
from datetime import datetime, timedelta, timezone
|
||
from pathlib import Path
|
||
from unittest.mock import patch
|
||
|
||
import pytest
|
||
import sys
|
||
|
||
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
|
||
|
||
from fenris.store import init_store, SCHEMA_VERSION
|
||
from fenris.monitoring_periods import ensure_period_open
|
||
from fenris.projection import (
|
||
compute_projection,
|
||
ConfidenceState,
|
||
BaselineTier,
|
||
DISCLOSURES,
|
||
STALENESS_HOURS,
|
||
WARMING_MIN_DAYS,
|
||
YOUNG_REGIME_DAYS,
|
||
)
|
||
from fenris.status import (
|
||
grade_freshness,
|
||
get_status,
|
||
render_status,
|
||
format_disclosures,
|
||
FRESH_THRESHOLD_S,
|
||
STALENESS_THRESHOLD_S,
|
||
CADENCE_DEFAULT_S,
|
||
ACCURACY_SEC,
|
||
)
|
||
from fenris.tui import (
|
||
FenrisTuiApp,
|
||
_format_remaining,
|
||
)
|
||
|
||
|
||
SRC_DIR = Path(__file__).parent.parent / "src"
|
||
FENRIS_PKG = SRC_DIR / "fenris"
|
||
|
||
|
||
def _clock(year=2026, month=9, day=30, hour=12):
|
||
return datetime(year, month, day, hour, 0, 0, tzinfo=timezone.utc)
|
||
|
||
|
||
def _insert_baseline(conn, tbw_tb=1.0, verified=True,
|
||
model="Samsung SSD 970 EVO Plus 1TB",
|
||
source_url="https://example.com/spec",
|
||
doc_rev="v1.0", entry_date="2026-01-01",
|
||
nominal_cap=1024000000000):
|
||
conn.execute(
|
||
"INSERT INTO endurance_baseline "
|
||
"(tbw_terabytes, source_url, document_revision, entry_date, model_string, "
|
||
" nominal_capacity_bytes, validated_by, verified, created_at, updated_at) "
|
||
"VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
|
||
(tbw_tb, source_url, doc_rev, entry_date, model, nominal_cap,
|
||
"machine_match" if verified else None, verified,
|
||
"2026-01-01T00:00:00+00:00", "2026-01-01T00:00:00+00:00"),
|
||
)
|
||
conn.commit()
|
||
|
||
|
||
def _insert_segment(conn, opened_at="2026-09-01T00:00:00+00:00",
|
||
identity_key="nqn.test", degraded=False,
|
||
mn="Samsung SSD 970 EVO Plus 1TB"):
|
||
conn.execute(
|
||
"INSERT INTO controller_segments "
|
||
"(opened_at, identity_key, identity_degraded, subnqn, sn, mn, fr, vid, ssvid, transport) "
|
||
"VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
|
||
(opened_at, identity_key, degraded, "nqn.test", "SN123", mn, "FW1",
|
||
"0x144d", "0x144d", "pcie"),
|
||
)
|
||
conn.commit()
|
||
|
||
|
||
def _insert_day(conn, day, bw=1024*1024*100, coverage=0.95, samples=24):
|
||
conn.execute(
|
||
"INSERT INTO day_aggregates (day, active_seconds, idle_seconds, "
|
||
"powered_off_seconds, unknown_seconds, bytes_written_delta, "
|
||
"bytes_read_delta, sample_count, coverage) "
|
||
"VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
|
||
(day, 3600, 0, 0, 0, bw, 0, samples, coverage),
|
||
)
|
||
conn.commit()
|
||
|
||
|
||
def _insert_sample(conn, ts, pu=5):
|
||
conn.execute(
|
||
"INSERT INTO samples (ts, device, data_units_written, data_units_read, "
|
||
"percentage_used, bytes_written, bytes_read, power_on_hours) "
|
||
"VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
|
||
(ts, "/dev/nvme0n1", 1000000, 500000, pu, 512000000000, 256000000000, 8765),
|
||
)
|
||
conn.commit()
|
||
|
||
|
||
def _open_period(conn, start="2026-09-01T00:00:00+00:00"):
|
||
ensure_period_open(conn, datetime.fromisoformat(start))
|
||
|
||
|
||
def _setup_full_store(conn, *, baseline=True, segment=True, days=30,
|
||
bw=1024*1024*100, coverage=0.95, samples_per_day=24,
|
||
sample_ts="2026-09-30T10:00:00+00:00",
|
||
period_start="2026-09-01T00:00:00+00:00",
|
||
segment_opened="2026-09-01T00:00:00+00:00",
|
||
baseline_kw=None, segment_kw=None):
|
||
if baseline:
|
||
_insert_baseline(conn, **(baseline_kw or {}))
|
||
if segment:
|
||
_insert_segment(conn, opened_at=segment_opened, **(segment_kw or {}))
|
||
_open_period(conn, start=period_start)
|
||
for i in range(days):
|
||
d = (datetime(2026, 9, 1) + timedelta(days=i)).strftime("%Y-%m-%d")
|
||
_insert_day(conn, d, bw=bw, coverage=coverage, samples=samples_per_day)
|
||
if sample_ts:
|
||
_insert_sample(conn, sample_ts)
|
||
|
||
|
||
# ===================================================================
|
||
# CI-1: Exhaustive state matrix
|
||
# ===================================================================
|
||
|
||
|
||
class TestCI1StateMatrix:
|
||
"""Systematic walk of confidence x freshness x baseline tier combinations."""
|
||
|
||
def test_no_baseline_unavailable(self, tmp_path):
|
||
conn = init_store(tmp_path / "db")
|
||
_insert_segment(conn)
|
||
_open_period(conn)
|
||
for i in range(30):
|
||
d = (datetime(2026, 9, 1) + timedelta(days=i)).strftime("%Y-%m-%d")
|
||
_insert_day(conn, d, bw=1024*1024*100, coverage=0.95, samples=24)
|
||
_insert_sample(conn, "2026-09-30T10:00:00+00:00")
|
||
proj = compute_projection(conn, _clock())
|
||
assert proj.confidence_state == ConfidenceState.UNSUPPORTED
|
||
assert proj.headline_remaining_seconds is None
|
||
assert proj.baseline_tier == BaselineTier.NONE
|
||
conn.close()
|
||
|
||
def test_verified_baseline_possible_supported(self, tmp_path):
|
||
conn = init_store(tmp_path / "db")
|
||
_setup_full_store(conn, baseline_kw=dict(tbw_tb=10.0, verified=True))
|
||
proj = compute_projection(conn, _clock())
|
||
assert proj.confidence_state == ConfidenceState.SUPPORTED
|
||
assert proj.baseline_tier == BaselineTier.VERIFIED
|
||
assert proj.headline_remaining_seconds is not None
|
||
conn.close()
|
||
|
||
def test_unverified_baseline_possible_limited(self, tmp_path):
|
||
conn = init_store(tmp_path / "db")
|
||
_setup_full_store(conn, baseline_kw=dict(
|
||
tbw_tb=10.0, verified=False, source_url=None))
|
||
proj = compute_projection(conn, _clock())
|
||
assert proj.baseline_tier == BaselineTier.UNVERIFIED
|
||
assert proj.confidence_state != ConfidenceState.SUPPORTED
|
||
conn.close()
|
||
|
||
def test_model_mismatch_unavailable(self, tmp_path):
|
||
conn = init_store(tmp_path / "db")
|
||
_setup_full_store(conn, baseline_kw=dict(model="Different Model"))
|
||
proj = compute_projection(conn, _clock())
|
||
assert proj.confidence_state == ConfidenceState.UNSUPPORTED
|
||
assert proj.baseline_tier == BaselineTier.NONE
|
||
conn.close()
|
||
|
||
def test_fresh_sample_grades_fresh(self, tmp_path):
|
||
now = _clock()
|
||
ts = (now - timedelta(seconds=FRESH_THRESHOLD_S - 10)).isoformat()
|
||
assert grade_freshness(ts, now) == "fresh"
|
||
|
||
def test_missed_sample_grades_missed(self, tmp_path):
|
||
now = _clock()
|
||
ts = (now - timedelta(hours=2)).isoformat()
|
||
assert grade_freshness(ts, now) == "missed"
|
||
|
||
def test_stale_sample_grades_stale(self, tmp_path):
|
||
now = _clock()
|
||
ts = (now - timedelta(hours=49)).isoformat()
|
||
assert grade_freshness(ts, now) == "stale"
|
||
|
||
def test_empty_store_grades_empty(self, tmp_path):
|
||
now = _clock()
|
||
assert grade_freshness(None, now) == "empty"
|
||
|
||
def test_unsupported_fresh(self, tmp_path):
|
||
conn = init_store(tmp_path / "db")
|
||
_insert_segment(conn)
|
||
_open_period(conn)
|
||
for i in range(20):
|
||
d = (datetime(2026, 9, 10) + timedelta(days=i)).strftime("%Y-%m-%d")
|
||
_insert_day(conn, d, bw=1024*1024*100)
|
||
fresh_ts = (_clock() - timedelta(seconds=60)).isoformat()
|
||
_insert_sample(conn, fresh_ts)
|
||
proj = compute_projection(conn, _clock())
|
||
assert proj.confidence_state == ConfidenceState.UNSUPPORTED
|
||
assert grade_freshness(fresh_ts, _clock()) == "fresh"
|
||
conn.close()
|
||
|
||
def test_limited_young_regime(self, tmp_path):
|
||
conn = init_store(tmp_path / "db")
|
||
_insert_baseline(conn, tbw_tb=10.0, verified=True)
|
||
_insert_segment(conn)
|
||
_open_period(conn)
|
||
for i in range(5):
|
||
d = (datetime(2026, 9, 25) + timedelta(days=i)).strftime("%Y-%m-%d")
|
||
_insert_day(conn, d, bw=1024*1024*100)
|
||
_insert_sample(conn, "2026-09-30T10:00:00+00:00")
|
||
proj = compute_projection(conn, _clock())
|
||
assert proj.confidence_state == ConfidenceState.LIMITED
|
||
conn.close()
|
||
|
||
def test_limited_warming(self, tmp_path):
|
||
conn = init_store(tmp_path / "db")
|
||
_insert_baseline(conn, tbw_tb=10.0, verified=True)
|
||
_insert_segment(conn)
|
||
_open_period(conn)
|
||
for i in range(10):
|
||
d = (datetime(2026, 9, 20) + timedelta(days=i)).strftime("%Y-%m-%d")
|
||
_insert_day(conn, d, bw=1024*1024*100, coverage=0.95, samples=24)
|
||
_insert_sample(conn, "2026-09-30T10:00:00+00:00")
|
||
proj = compute_projection(conn, _clock())
|
||
assert proj.confidence_state == ConfidenceState.LIMITED
|
||
assert proj.warming_fact is not None
|
||
conn.close()
|
||
|
||
def test_limited_stale_data(self, tmp_path):
|
||
conn = init_store(tmp_path / "db")
|
||
_insert_baseline(conn, tbw_tb=10.0, verified=True)
|
||
_insert_segment(conn, opened_at="2026-08-01T00:00:00+00:00")
|
||
_open_period(conn, start="2026-08-01T00:00:00+00:00")
|
||
for i in range(30):
|
||
d = (datetime(2026, 8, 1) + timedelta(days=i)).strftime("%Y-%m-%d")
|
||
_insert_day(conn, d, bw=1024*1024*100, coverage=0.95, samples=24)
|
||
stale_ts = (_clock() - timedelta(days=5)).isoformat()
|
||
_insert_sample(conn, stale_ts)
|
||
proj = compute_projection(conn, _clock())
|
||
assert proj.confidence_state == ConfidenceState.LIMITED
|
||
assert proj.staleness_fact is not None
|
||
conn.close()
|
||
|
||
def test_limited_degraded_identity(self, tmp_path):
|
||
conn = init_store(tmp_path / "db")
|
||
_insert_baseline(conn, tbw_tb=10.0, verified=True)
|
||
_insert_segment(conn, identity_key=None, degraded=True)
|
||
_open_period(conn)
|
||
for i in range(30):
|
||
d = (datetime(2026, 9, 1) + timedelta(days=i)).strftime("%Y-%m-%d")
|
||
_insert_day(conn, d, bw=1024*1024*100, coverage=0.95, samples=24)
|
||
_insert_sample(conn, "2026-09-30T10:00:00+00:00")
|
||
proj = compute_projection(conn, _clock())
|
||
assert proj.confidence_state == ConfidenceState.LIMITED
|
||
assert proj.degraded_identity_fact is not None
|
||
conn.close()
|
||
|
||
def test_unsupported_zero_rate(self, tmp_path):
|
||
conn = init_store(tmp_path / "db")
|
||
_insert_baseline(conn, tbw_tb=10.0, verified=True)
|
||
_insert_segment(conn)
|
||
_open_period(conn)
|
||
for i in range(30):
|
||
d = (datetime(2026, 9, 1) + timedelta(days=i)).strftime("%Y-%m-%d")
|
||
_insert_day(conn, d, bw=0)
|
||
_insert_sample(conn, "2026-09-30T10:00:00+00:00")
|
||
proj = compute_projection(conn, _clock())
|
||
assert proj.confidence_state == ConfidenceState.UNSUPPORTED
|
||
assert proj.zero_rate_fact is not None
|
||
conn.close()
|
||
|
||
def test_headline_present_when_projection_exists(self, tmp_path):
|
||
conn = init_store(tmp_path / "db")
|
||
_setup_full_store(conn, baseline_kw=dict(tbw_tb=10.0, verified=True))
|
||
proj = compute_projection(conn, _clock())
|
||
assert proj.headline_remaining_seconds is not None
|
||
assert proj.headline_remaining_seconds > 0
|
||
conn.close()
|
||
|
||
def test_headline_absent_when_unavailable(self, tmp_path):
|
||
conn = init_store(tmp_path / "db")
|
||
_insert_segment(conn)
|
||
_open_period(conn)
|
||
proj = compute_projection(conn, _clock())
|
||
assert proj.headline_remaining_seconds is None
|
||
conn.close()
|
||
|
||
def test_headline_absent_when_zero_rate(self, tmp_path):
|
||
conn = init_store(tmp_path / "db")
|
||
_insert_baseline(conn, tbw_tb=1.0, verified=True)
|
||
_insert_segment(conn)
|
||
_open_period(conn)
|
||
for i in range(20):
|
||
d = (datetime(2026, 9, 10) + timedelta(days=i)).strftime("%Y-%m-%d")
|
||
_insert_day(conn, d, bw=0)
|
||
proj = compute_projection(conn, _clock())
|
||
assert proj.headline_remaining_seconds is None
|
||
conn.close()
|
||
|
||
def test_facts_always_list(self, tmp_path):
|
||
conn = init_store(tmp_path / "db")
|
||
_insert_segment(conn)
|
||
_open_period(conn)
|
||
proj = compute_projection(conn, _clock())
|
||
assert isinstance(proj.contributing_facts, list)
|
||
conn.close()
|
||
|
||
def test_facts_never_empty_for_unavailable(self, tmp_path):
|
||
conn = init_store(tmp_path / "db")
|
||
_insert_segment(conn)
|
||
_open_period(conn)
|
||
proj = compute_projection(conn, _clock())
|
||
assert len(proj.contributing_facts) > 0
|
||
conn.close()
|
||
|
||
def test_confidence_never_percentage(self, tmp_path):
|
||
conn = init_store(tmp_path / "db")
|
||
_setup_full_store(conn, baseline_kw=dict(tbw_tb=10.0, verified=True))
|
||
proj = compute_projection(conn, _clock())
|
||
assert proj.confidence_state in (
|
||
ConfidenceState.UNSUPPORTED, ConfidenceState.LIMITED, ConfidenceState.SUPPORTED)
|
||
for f in proj.contributing_facts:
|
||
if re.match(r"^\\d+%$", f.strip()):
|
||
pytest.fail("Bare percentage in facts: %r" % f)
|
||
conn.close()
|
||
|
||
def test_status_renders_same_state_as_projection(self, tmp_path):
|
||
db = tmp_path / "observations.db"
|
||
conn = init_store(db)
|
||
_setup_full_store(conn, baseline_kw=dict(tbw_tb=10.0, verified=True))
|
||
conn.close()
|
||
now = _clock()
|
||
with patch("fenris.status.query_service_state", return_value={
|
||
"boot_enabled": True, "timer_active": True,
|
||
"last_collect_ok": True, "last_collect_age_s": 60,
|
||
"last_collect_reason": None,
|
||
}):
|
||
status = get_status(store_path=db, clock_now=now,
|
||
query_services=True, query_journal=False)
|
||
assert "Supported" in status or "supported" in status.lower()
|
||
assert "remaining" in status.lower()
|
||
|
||
|
||
# ===================================================================
|
||
# CI-2: TUI/CLI parity
|
||
# ===================================================================
|
||
|
||
|
||
class TestCI2Parity:
|
||
"""Verify TUI and CLI share the same constants, formatting, and wording."""
|
||
|
||
def test_freshness_constants_shared(self):
|
||
from fenris import tui as tui_mod
|
||
from fenris import status as status_mod
|
||
assert tui_mod.FRESH_THRESHOLD_S == status_mod.FRESH_THRESHOLD_S
|
||
assert tui_mod.STALENESS_THRESHOLD_S == status_mod.STALENESS_THRESHOLD_S
|
||
|
||
def test_grade_freshness_shared(self):
|
||
from fenris.tui import grade_freshness as tui_gf
|
||
from fenris.status import grade_freshness as status_gf
|
||
assert tui_gf is status_gf
|
||
|
||
def test_disclosures_shared(self):
|
||
from fenris.projection import DISCLOSURES as proj_disc
|
||
from fenris.status import format_disclosures
|
||
output = format_disclosures()
|
||
for d in proj_disc:
|
||
assert d in output
|
||
|
||
def test_status_four_facts_match_tui_strip(self, tmp_path):
|
||
db = tmp_path / "observations.db"
|
||
conn = init_store(db)
|
||
_insert_segment(conn)
|
||
_open_period(conn)
|
||
for i in range(20):
|
||
d = (datetime(2026, 9, 10) + timedelta(days=i)).strftime("%Y-%m-%d")
|
||
_insert_day(conn, d, bw=1024*1024*100)
|
||
_insert_sample(conn, "2026-09-30T10:00:00+00:00")
|
||
conn.close()
|
||
now = _clock()
|
||
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,
|
||
}):
|
||
status = get_status(store_path=db, clock_now=now,
|
||
query_services=True, query_journal=False)
|
||
assert "boot:" in status
|
||
assert "timer:" in status
|
||
assert "last collect:" in status
|
||
assert "freshness:" in status
|
||
|
||
def test_dashboard_clarity_parity_strings_have_one_status_source(self):
|
||
"""DC-2/DC-3 wording originates in status and the TUI imports it."""
|
||
status_src = (FENRIS_PKG / "status.py").read_text()
|
||
tui_src = (FENRIS_PKG / "tui.py").read_text()
|
||
for wording in (
|
||
"monitoring: active in background · persists across reboots",
|
||
"monitoring: does not start on next boot",
|
||
"monitoring: paused — deliberate disable",
|
||
"paused time is excluded from your usage habit · resume: fenris monitor resume",
|
||
):
|
||
assert status_src.count(wording) == 1
|
||
assert wording not in tui_src
|
||
|
||
@pytest.mark.asyncio
|
||
@pytest.mark.parametrize(
|
||
("state", "service", "expected_lines"),
|
||
[
|
||
(
|
||
"active_enabled",
|
||
{"boot_enabled": True, "timer_active": True},
|
||
["monitoring: active in background · persists across reboots"],
|
||
),
|
||
(
|
||
"boot_disabled",
|
||
{"boot_enabled": False, "timer_active": False},
|
||
["monitoring: does not start on next boot"],
|
||
),
|
||
(
|
||
"deliberately_paused",
|
||
{"boot_enabled": False, "timer_active": False},
|
||
[
|
||
"monitoring: does not start on next boot",
|
||
"monitoring: paused — deliberate disable",
|
||
"paused time is excluded from your usage habit · resume: fenris monitor resume",
|
||
],
|
||
),
|
||
],
|
||
)
|
||
async def test_dashboard_clarity_monitoring_lines_match_both_views(
|
||
self, tmp_path, state, service, expected_lines
|
||
):
|
||
"""CI-2 synthetic-store sweep covers active, disabled, and paused states."""
|
||
db = tmp_path / (state + ".db")
|
||
conn = init_store(db)
|
||
if state == "active_enabled":
|
||
ensure_period_open(conn, _clock())
|
||
elif state == "deliberately_paused":
|
||
conn.execute(
|
||
"INSERT INTO monitoring_periods (started_at, ended_at, end_cause) "
|
||
"VALUES (?, ?, ?)",
|
||
("2026-09-30T09:00:00+00:00", "2026-09-30T10:00:00+00:00", "user_disabled"),
|
||
)
|
||
conn.commit()
|
||
conn.close()
|
||
service_state = {
|
||
**service,
|
||
"last_collect_ok": None,
|
||
"last_collect_age_s": None,
|
||
"last_collect_reason": None,
|
||
}
|
||
|
||
with patch("fenris.status.query_service_state", return_value=service_state), patch(
|
||
"fenris.status.query_service_state", return_value=service_state
|
||
):
|
||
status = get_status(
|
||
store_path=db, clock_now=_clock(), query_services=True, query_journal=False
|
||
).lower()
|
||
app = FenrisTuiApp(store_path=db)
|
||
async with app.run_test(size=(100, 40)):
|
||
tui_text = "\n".join(
|
||
(
|
||
str(app.query_one("#service-strip").render()),
|
||
str(app.query_one("#paused-banner").render()),
|
||
)
|
||
).lower()
|
||
|
||
for expected in expected_lines:
|
||
assert expected in status
|
||
assert expected in tui_text
|
||
if state != "deliberately_paused":
|
||
assert "monitoring: paused — deliberate disable" not in status
|
||
assert "monitoring: paused — deliberate disable" not in tui_text
|
||
|
||
def test_pause_resume_action_names(self):
|
||
tui_keys = {b.key for b in FenrisTuiApp.BINDINGS}
|
||
assert "p" in tui_keys
|
||
assert "r" in tui_keys
|
||
assert "c" in tui_keys
|
||
assert "q" in tui_keys
|
||
|
||
def test_empty_store_greeting_both_views(self, tmp_path):
|
||
db = tmp_path / "observations.db"
|
||
init_store(db)
|
||
now = _clock()
|
||
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,
|
||
}):
|
||
status = get_status(store_path=db, clock_now=now,
|
||
query_services=True, query_journal=False)
|
||
assert "no observations yet" in status.lower()
|
||
|
||
def test_status_never_prompts(self):
|
||
status_src = (FENRIS_PKG / "status.py").read_text()
|
||
assert "input(" not in status_src
|
||
|
||
|
||
# ===================================================================
|
||
# CI-3: Prohibition set
|
||
# ===================================================================
|
||
|
||
|
||
class TestCI3ProhibitionSet:
|
||
"""Structural codebase checks for every prohibition clause."""
|
||
|
||
def _read_all_sources(self):
|
||
files = {}
|
||
for py in FENRIS_PKG.glob("*.py"):
|
||
files[py.name] = py.read_text()
|
||
return files
|
||
|
||
def test_single_acquisition_path(self):
|
||
"""Only fenris-collect may interrogate the device. [2.1, 8.7]
|
||
|
||
collector.py contains the acquisition functions; collect.py is the
|
||
fenris-collect entry point that invokes them. No other module may
|
||
reference smartctl.
|
||
"""
|
||
sources = self._read_all_sources()
|
||
allowed = {"collector.py", "collect.py"}
|
||
for name, text in sources.items():
|
||
if name in allowed:
|
||
continue
|
||
assert "smartctl" not in text, (
|
||
"%s must not contain smartctl" % name
|
||
)
|
||
|
||
def test_no_run_surface(self):
|
||
"""No /run/fenris coordination surface. [1.2, 3]"""
|
||
sources = self._read_all_sources()
|
||
for name, text in sources.items():
|
||
assert "/run/fenris" not in text, (
|
||
"%s references /run/fenris" % name
|
||
)
|
||
|
||
def test_single_config_key(self):
|
||
"""Config holds exactly one key: device. [8.3]"""
|
||
status_src = (FENRIS_PKG / "status.py").read_text()
|
||
in_read_config = False
|
||
config_keys = []
|
||
for line in status_src.split("\n"):
|
||
if "def read_config" in line:
|
||
in_read_config = True
|
||
elif in_read_config and line.strip().startswith("def "):
|
||
break
|
||
elif in_read_config and "key ==" in line:
|
||
match = re.search(r'key\s*==\s*["\']([^"\']+)["\']', line)
|
||
if match:
|
||
config_keys.append(match.group(1))
|
||
assert "device" in config_keys
|
||
assert len(config_keys) == 1, "Found keys: %s" % config_keys
|
||
|
||
def test_no_alerting_machinery(self):
|
||
"""No alerting, notification, or escalation. [9.6]"""
|
||
sources = self._read_all_sources()
|
||
alert_keywords = ["send_email", "smtp", "webhook", "push_notification"]
|
||
for name, text in sources.items():
|
||
for kw in alert_keywords:
|
||
for line in text.split("\n"):
|
||
stripped = line.strip()
|
||
if kw in stripped and not stripped.startswith("#"):
|
||
pytest.fail(
|
||
"%s contains alerting keyword '%s': %s" % (name, kw, stripped)
|
||
)
|
||
|
||
def test_no_synthetic_baselines(self):
|
||
"""No synthetic or capacity-derived baseline. [6.1]"""
|
||
proj_src = (FENRIS_PKG / "projection.py").read_text()
|
||
assert "synthetic" not in proj_src.lower()
|
||
|
||
def test_no_stored_projections(self):
|
||
"""Projections never stored; recomputed on read. [3.7, 6.10]"""
|
||
store_src = (FENRIS_PKG / "store.py").read_text()
|
||
create_tables = re.findall(r"CREATE TABLE.*?(?=\n\n|$)", store_src, re.DOTALL)
|
||
table_names = []
|
||
for ct in create_tables:
|
||
m = re.search(r"IF NOT EXISTS\s+(\w+)", ct)
|
||
if m:
|
||
table_names.append(m.group(1))
|
||
assert "projection" not in [t.lower() for t in table_names]
|
||
|
||
def test_polkit_authorizes_one_binary(self):
|
||
"""Polkit authorizes exactly one binary: fenris-monitor. [8.5]"""
|
||
monitor_src = (FENRIS_PKG / "monitor.py").read_text()
|
||
assert "fenris-monitor" in monitor_src or "fenris_monitor" in monitor_src
|
||
collect_src = (FENRIS_PKG / "collect.py").read_text()
|
||
assert "polkit" not in collect_src.lower()
|
||
|
||
def test_no_hour_interpolation(self):
|
||
"""No absent hour is interpolated or fabricated. [5.3]"""
|
||
proj_src = (FENRIS_PKG / "projection.py").read_text()
|
||
assert "interpolat" not in proj_src.lower()
|
||
assert "fabricat" not in proj_src.lower()
|
||
|
||
def test_fenris_sh_not_shipped(self):
|
||
"""fenris.sh is not shipped. [8.8]"""
|
||
repo_root = Path(__file__).parent.parent
|
||
assert not (repo_root / "fenris.sh").exists()
|
||
|
||
|
||
# ===================================================================
|
||
# CI-4: Required wording and six disclosures
|
||
# ===================================================================
|
||
|
||
|
||
class TestCI4WordingAndDisclosures:
|
||
"""Verify exact fixed phrases and disclosures in both views."""
|
||
|
||
def test_exactly_six_disclosures(self):
|
||
assert len(DISCLOSURES) == 6
|
||
|
||
def test_disclosure_1_endurance_not_failure(self):
|
||
assert "endurance projection" in DISCLOSURES[0].lower()
|
||
assert "hardware-failure" in DISCLOSURES[0].lower() or "failure date" in DISCLOSURES[0].lower()
|
||
|
||
def test_disclosure_2_vendor_specific(self):
|
||
assert "vendor-specific" in DISCLOSURES[1]
|
||
assert "255 is saturated" in DISCLOSURES[1]
|
||
|
||
def test_disclosure_3_warranty_not_failure(self):
|
||
assert "warranty" in DISCLOSURES[2].lower() or "endurance threshold" in DISCLOSURES[2].lower()
|
||
assert "failure threshold" in DISCLOSURES[2].lower()
|
||
|
||
def test_disclosure_4_duw_rounding(self):
|
||
assert "DUW" in DISCLOSURES[3]
|
||
assert "upward-rounded" in DISCLOSURES[3]
|
||
assert "NAND" in DISCLOSURES[3]
|
||
|
||
def test_disclosure_5_quality_depends(self):
|
||
assert "baseline provenance" in DISCLOSURES[4]
|
||
assert "future workload" in DISCLOSURES[4]
|
||
|
||
def test_disclosure_6_gaps_and_disabled(self):
|
||
assert "Gaps" in DISCLOSURES[5]
|
||
assert "deliberately disabled" in DISCLOSURES[5]
|
||
|
||
def test_disclosures_render_in_status(self):
|
||
output = format_disclosures()
|
||
assert output.startswith("Disclosures")
|
||
for i in range(1, 7):
|
||
assert "%d." % i in output
|
||
for d in DISCLOSURES:
|
||
assert d in output
|
||
|
||
def test_disclosures_render_in_tui(self):
|
||
tui_src = (FENRIS_PKG / "tui.py").read_text()
|
||
assert "format_disclosures" in tui_src
|
||
|
||
def test_zero_rate_phrase(self):
|
||
phrase = "no finite projection from this history"
|
||
proj_src = (FENRIS_PKG / "projection.py").read_text()
|
||
assert phrase in proj_src
|
||
status_src = (FENRIS_PKG / "status.py").read_text()
|
||
assert phrase in status_src
|
||
|
||
def test_unavailable_no_baseline_phrase(self):
|
||
phrase = "no applicable endurance baseline"
|
||
proj_src = (FENRIS_PKG / "projection.py").read_text()
|
||
assert phrase in proj_src
|
||
|
||
def test_no_observations_phrase(self):
|
||
phrase = "no observations yet"
|
||
status_src = (FENRIS_PKG / "status.py").read_text()
|
||
assert phrase in status_src
|
||
tui_src = (FENRIS_PKG / "tui.py").read_text()
|
||
assert phrase in tui_src.lower()
|
||
|
||
def test_config_error_phrase(self):
|
||
phrase = "configuration error:"
|
||
status_src = (FENRIS_PKG / "status.py").read_text()
|
||
assert phrase in status_src
|
||
|
||
def test_degraded_identity_phrase(self):
|
||
phrase = "controller identity unavailable"
|
||
proj_src = (FENRIS_PKG / "projection.py").read_text()
|
||
assert phrase in proj_src
|
||
phrase2 = "replacement detection relies on write-counter continuity only"
|
||
assert phrase2 in proj_src
|
||
|
||
def test_scenario_range_only_spread(self):
|
||
proj_src = (FENRIS_PKG / "projection.py").read_text()
|
||
assert "confidence interval" not in proj_src.lower()
|
||
|
||
def test_no_percentage_in_confidence_rendering(self):
|
||
for name in ["projection.py", "tui.py", "status.py"]:
|
||
src = (FENRIS_PKG / name).read_text()
|
||
assert not re.search(r"\\d+%\\s*confidence", src, re.IGNORECASE), (
|
||
"Found XX%% confidence in %s" % name
|
||
)
|
||
|
||
def test_status_disclosures_accessible(self):
|
||
db = Path("/tmp/_ci4_test.db")
|
||
conn = init_store(db)
|
||
conn.close()
|
||
now = _clock()
|
||
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 "1." in result
|
||
assert "6." in result
|
||
db.unlink(missing_ok=True)
|