682 lines
28 KiB
Python
682 lines
28 KiB
Python
"""Headless tests for the Panes TUI (issue #28).
|
||
|
||
Covers:
|
||
- CI-1: Exhaustive state matrix from synthetic stores
|
||
- TUI-1: One dense keyboard-first screen with four normative regions
|
||
- TUI-4: Layout regions normative per register
|
||
- CI-4: Six disclosures verbatim, empty-store greeting, first-run opt-in
|
||
- IN-3: First-run TUI prompt enables timer and opens first period
|
||
|
||
Criteria: TUI-1, TUI-4, CI-1, CI-4, IN-3.
|
||
"""
|
||
import sqlite3
|
||
from xml.etree import ElementTree
|
||
from datetime import datetime, timedelta, timezone
|
||
from pathlib import Path
|
||
from unittest.mock import patch, MagicMock
|
||
|
||
import pytest
|
||
import sys
|
||
|
||
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
|
||
|
||
from textual.app import App
|
||
from textual.pilot import Pilot
|
||
|
||
from fenris.store import init_store, SCHEMA_VERSION
|
||
from fenris.monitoring_periods import ensure_period_open, close_period
|
||
from fenris.projection import (
|
||
ConfidenceState,
|
||
compute_projection,
|
||
DISCLOSURES,
|
||
WARMING_MIN_DAYS,
|
||
STALENESS_HOURS,
|
||
YOUNG_REGIME_DAYS,
|
||
)
|
||
from fenris.status import (
|
||
FRESH_THRESHOLD_S,
|
||
STALENESS_THRESHOLD_S,
|
||
grade_freshness,
|
||
)
|
||
from fenris.tui import (
|
||
FenrisTuiApp,
|
||
_format_remaining,
|
||
_sparkline,
|
||
_habit_bar,
|
||
_query_usage_history,
|
||
_query_drive_health,
|
||
_query_service_facts,
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Helpers
|
||
# ---------------------------------------------------------------------------
|
||
|
||
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"):
|
||
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, "https://example.com/spec", "v1.0", "2026-01-01", model,
|
||
1024000000000, "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, device="/dev/nvme0n1"):
|
||
conn.execute(
|
||
"INSERT INTO samples (ts, device, data_units_written, data_units_read, "
|
||
"percentage_used, bytes_written, bytes_read, power_on_hours) "
|
||
"VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
|
||
(ts, device, 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))
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Unit tests for helpers
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class TestFormatRemaining:
|
||
def test_hours_only(self):
|
||
assert _format_remaining(3600) == "1 h"
|
||
|
||
def test_days_and_hours(self):
|
||
assert _format_remaining(86400) == "1 d 0 h"
|
||
|
||
def test_years(self):
|
||
assert _format_remaining(31557600) == "1 yr 0 d 0 h"
|
||
|
||
def test_zero(self):
|
||
assert _format_remaining(0) == "endurance exhausted"
|
||
|
||
def test_negative(self):
|
||
assert _format_remaining(-100) == "endurance exhausted"
|
||
|
||
|
||
class TestSparkline:
|
||
def test_empty(self):
|
||
assert _sparkline([]) == ""
|
||
|
||
def test_single_value(self):
|
||
result = _sparkline([100.0])
|
||
assert len(result) == 1
|
||
|
||
def test_multiple_values(self):
|
||
result = _sparkline([1.0, 2.0, 3.0, 4.0, 5.0])
|
||
assert len(result) > 0
|
||
assert all(c in " ▁▂▃▄▅▆▇█" for c in result)
|
||
|
||
def test_width_limit(self):
|
||
result = _sparkline([1.0] * 100, width=20)
|
||
assert len(result) <= 20
|
||
|
||
|
||
class TestHabitBar:
|
||
def test_all_active(self):
|
||
result = _habit_bar(1.0, 0.0, 0.0, 0.0)
|
||
assert "active 100%" in result
|
||
|
||
def test_mixed(self):
|
||
result = _habit_bar(0.5, 0.3, 0.1, 0.1)
|
||
assert "active 50%" in result
|
||
assert "idle 30%" in result
|
||
|
||
def test_all_unknown(self):
|
||
result = _habit_bar(0.0, 0.0, 0.0, 1.0)
|
||
assert "unknown 100%" in result
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Data query tests
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class TestQueryUsageHistory:
|
||
def test_empty_store(self, tmp_path):
|
||
conn = init_store(tmp_path / "test.db")
|
||
result = _query_usage_history(conn)
|
||
assert result["num_days"] == 0
|
||
assert result["sparkline"] == ""
|
||
conn.close()
|
||
|
||
def test_with_days(self, tmp_path):
|
||
conn = init_store(tmp_path / "test.db")
|
||
_insert_segment(conn)
|
||
_open_period(conn)
|
||
for i in range(14):
|
||
d = (datetime(2026, 9, 15) + timedelta(days=i)).strftime("%Y-%m-%d")
|
||
_insert_day(conn, d, bw=1024*1024*100)
|
||
result = _query_usage_history(conn)
|
||
assert result["num_days"] == 14
|
||
assert result["sparkline"] != ""
|
||
conn.close()
|
||
|
||
|
||
class TestQueryDriveHealth:
|
||
def test_empty_store(self, tmp_path):
|
||
conn = init_store(tmp_path / "test.db")
|
||
result = _query_drive_health(conn)
|
||
assert result["model"] == "unknown"
|
||
conn.close()
|
||
|
||
def test_with_sample(self, tmp_path):
|
||
conn = init_store(tmp_path / "test.db")
|
||
_insert_sample(conn, "2026-09-30T10:00:00+00:00", pu=10)
|
||
result = _query_drive_health(conn)
|
||
assert result["percentage_used"] == 10
|
||
conn.close()
|
||
|
||
|
||
class TestQueryServiceFacts:
|
||
def test_empty_store(self, tmp_path):
|
||
conn = init_store(tmp_path / "test.db")
|
||
now = _clock()
|
||
result = _query_service_facts(conn, now)
|
||
assert result["freshness"] == "empty"
|
||
conn.close()
|
||
|
||
def test_fresh_sample(self, tmp_path):
|
||
conn = init_store(tmp_path / "test.db")
|
||
_insert_segment(conn)
|
||
_open_period(conn)
|
||
_insert_day(conn, "2026-09-29", bw=1024*1024*100)
|
||
now = _clock()
|
||
ts = (now - timedelta(minutes=2)).isoformat()
|
||
_insert_sample(conn, ts)
|
||
result = _query_service_facts(conn, now)
|
||
assert result["freshness"] == "fresh"
|
||
conn.close()
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# CI-1: Exhaustive state matrix from synthetic stores
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class TestStateMatrix:
|
||
"""TUI renders every realizable combination of confidence state × freshness × baseline tier."""
|
||
|
||
def test_unsupported_no_baseline(self, tmp_path):
|
||
"""No baseline → Unavailable."""
|
||
conn = init_store(tmp_path / "test.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")
|
||
proj = compute_projection(conn, _clock())
|
||
assert proj.confidence_state == ConfidenceState.UNSUPPORTED
|
||
assert proj.headline_remaining_seconds is None
|
||
conn.close()
|
||
|
||
def test_limited_warming(self, tmp_path):
|
||
"""Warming up (< 14 days) → Limited."""
|
||
conn = init_store(tmp_path / "test.db")
|
||
_insert_baseline(conn, tbw_tb=1.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)
|
||
_insert_sample(conn, "2026-09-30T10:00:00+00:00")
|
||
proj = compute_projection(conn, _clock())
|
||
assert proj.confidence_state == ConfidenceState.LIMITED
|
||
assert proj.headline_remaining_seconds is not None
|
||
conn.close()
|
||
|
||
def test_supported_full(self, tmp_path):
|
||
"""Full data → Supported."""
|
||
conn = init_store(tmp_path / "test.db")
|
||
_insert_baseline(conn, tbw_tb=1.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=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.SUPPORTED
|
||
assert proj.headline_remaining_seconds is not None
|
||
conn.close()
|
||
|
||
def test_stale_freshness(self, tmp_path):
|
||
"""Stale data: newest day aggregate > 48h old → Limited or Unsupported."""
|
||
conn = init_store(tmp_path / "test.db")
|
||
_insert_baseline(conn, tbw_tb=1.0, verified=True)
|
||
# Segment opened long ago so days are old
|
||
_insert_segment(conn, opened_at="2026-08-01T00:00:00+00:00")
|
||
_open_period(conn, start="2026-08-01T00:00:00+00:00")
|
||
# Days all end on Aug 30 — 31 days before clock (Sept 30)
|
||
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)
|
||
# Sample also old
|
||
stale_ts = (_clock() - timedelta(days=31)).isoformat()
|
||
_insert_sample(conn, stale_ts)
|
||
proj = compute_projection(conn, _clock())
|
||
# Stale data (> 48h since newest day) → not Supported
|
||
assert proj.confidence_state != ConfidenceState.SUPPORTED
|
||
conn.close()
|
||
|
||
def test_empty_store_state(self, tmp_path):
|
||
"""Empty store → no projection, headline=None."""
|
||
conn = init_store(tmp_path / "test.db")
|
||
proj = compute_projection(conn, _clock())
|
||
assert proj.confidence_state == ConfidenceState.UNSUPPORTED
|
||
assert proj.headline_remaining_seconds is None
|
||
conn.close()
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# TUI-1: One dense keyboard-first screen
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class TestDenseScreen:
|
||
"""TUI renders one dense screen with four normative regions."""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_four_regions_exist(self, tmp_path):
|
||
"""All four normative regions are present in the DOM."""
|
||
conn = init_store(tmp_path / "test.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()
|
||
|
||
app = FenrisTuiApp(store_path=tmp_path / "test.db")
|
||
async with app.run_test() as pilot:
|
||
assert app.query_one("#headline-band") is not None
|
||
assert app.query_one("#usage-history") is not None
|
||
assert app.query_one("#drive-health") is not None
|
||
assert app.query_one("#service-strip") is not None
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_headline_contains_projection(self, tmp_path):
|
||
"""Headline band shows projection headline."""
|
||
conn = init_store(tmp_path / "test.db")
|
||
_insert_baseline(conn, tbw_tb=1.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=1024*1024*100, coverage=0.95, samples=24)
|
||
_insert_sample(conn, "2026-09-30T10:00:00+00:00")
|
||
conn.close()
|
||
|
||
app = FenrisTuiApp(store_path=tmp_path / "test.db")
|
||
async with app.run_test() as pilot:
|
||
headline = str(app.query_one("#headline-band").render())
|
||
assert "remaining" in headline.lower() or "projection" in headline.lower()
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_confidence_rendered_as_evidence(self, tmp_path):
|
||
"""Confidence is state + contributing facts, never a percentage."""
|
||
conn = init_store(tmp_path / "test.db")
|
||
_insert_baseline(conn, tbw_tb=1.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=1024*1024*100, coverage=0.95, samples=24)
|
||
_insert_sample(conn, "2026-09-30T10:00:00+00:00")
|
||
conn.close()
|
||
|
||
app = FenrisTuiApp(store_path=tmp_path / "test.db")
|
||
async with app.run_test() as pilot:
|
||
headline = str(app.query_one("#headline-band").render())
|
||
assert "projection confidence" in headline.lower()
|
||
# Contributing facts shown, never a percentage as confidence
|
||
# (percentage in "95% interval coverage" is allowed as a fact, not as confidence)
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_service_strip_has_four_facts(self, tmp_path):
|
||
"""Service strip has four separate facts (boot, timer, collect, freshness)."""
|
||
conn = init_store(tmp_path / "test.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()
|
||
|
||
app = FenrisTuiApp(store_path=tmp_path / "test.db")
|
||
async with app.run_test() as pilot:
|
||
strip = str(app.query_one("#service-strip").render())
|
||
assert "boot:" in strip
|
||
assert "timer:" in strip
|
||
assert "last collect:" in strip
|
||
assert "freshness:" in strip
|
||
assert "by Bongbetic" in strip
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_service_strip_shows_continuity_and_separate_quit_rail(self, tmp_path):
|
||
"""The visible action footer excludes quit because the rail owns it."""
|
||
conn = init_store(tmp_path / "test.db")
|
||
_open_period(conn)
|
||
conn.close()
|
||
app = FenrisTuiApp(store_path=tmp_path / "test.db")
|
||
|
||
with patch("fenris.tui.query_service_state", return_value={
|
||
"boot_enabled": True, "timer_active": True,
|
||
"last_collect_ok": True, "last_collect_age_s": 60,
|
||
"last_collect_reason": None,
|
||
}):
|
||
async with app.run_test():
|
||
strip = str(app.query_one("#service-strip").render()).lower()
|
||
rail = str(app.query_one("#quit-rail").render())
|
||
usage = app.query_one("#usage-history")
|
||
service = app.query_one("#service-strip")
|
||
quit_rail = app.query_one("#quit-rail")
|
||
assert "continuity" in strip
|
||
assert "monitoring: active in background · persists across reboots" in strip
|
||
assert "p pause · r resume · c collect · d disclosures" in strip
|
||
assert "q quit" not in strip
|
||
assert rail == "q QUIT TUI"
|
||
assert usage.region.y < service.region.y < quit_rail.region.y
|
||
assert usage.region.bottom <= service.region.y
|
||
assert service.region.bottom <= quit_rail.region.y
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_deliberate_pause_banner_is_visible_and_quit_preserves_periods(self, tmp_path):
|
||
"""The Pilot sees the paused block; q leaves persisted monitoring state alone."""
|
||
db = tmp_path / "test.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()
|
||
app = FenrisTuiApp(store_path=db)
|
||
|
||
with patch("fenris.tui.query_service_state", return_value={
|
||
"boot_enabled": False, "timer_active": False,
|
||
"last_collect_ok": None, "last_collect_age_s": None,
|
||
"last_collect_reason": None,
|
||
}), patch("fenris.tui.subprocess.run") as subprocess_run, patch.object(
|
||
app, "_run_helper"
|
||
) as run_helper:
|
||
async with app.run_test(size=(80, 24)) as pilot:
|
||
await pilot.pause()
|
||
paused_banner = app.query_one("#paused-banner")
|
||
main_grid = app.query_one("#main-grid")
|
||
assert str(main_grid.styles.layout) == "<grid>"
|
||
assert main_grid.has_class("paused")
|
||
assert len(main_grid.styles.grid_rows) == 5
|
||
banner = str(paused_banner.render()).lower()
|
||
assert "monitoring: paused — deliberate disable" in banner
|
||
assert "paused time is excluded from your usage habit · resume: fenris monitor resume" in banner
|
||
assert paused_banner.region.height >= 5
|
||
assert paused_banner.region.y < app.query_one("#usage-history").region.y
|
||
assert app.query_one("#usage-history").region.bottom <= app.query_one(
|
||
"#service-strip"
|
||
).region.y
|
||
screenshot = app.export_screenshot()
|
||
visible_text = " ".join(
|
||
"".join(ElementTree.fromstring(screenshot).itertext()).split()
|
||
)
|
||
assert "paused time is excluded from your usage habit" in visible_text
|
||
assert "resume: fenris monitor resume" in visible_text
|
||
assert "monitoring: does not start on next boot" in str(
|
||
app.query_one("#service-strip").render()
|
||
).lower()
|
||
dashboard_scroll = app.query_one("#dashboard-scroll")
|
||
assert dashboard_scroll.max_scroll_y > 0
|
||
dashboard_scroll.focus()
|
||
await pilot.press("end")
|
||
assert dashboard_scroll.scroll_y == dashboard_scroll.max_scroll_y
|
||
footer_text = " ".join(
|
||
"".join(
|
||
ElementTree.fromstring(app.export_screenshot()).itertext()
|
||
).split()
|
||
)
|
||
assert "q QUIT TUI" in footer_text
|
||
await pilot.press("q")
|
||
assert not app.is_running
|
||
subprocess_run.assert_not_called()
|
||
run_helper.assert_not_called()
|
||
|
||
conn = sqlite3.connect(db)
|
||
row = conn.execute(
|
||
"SELECT ended_at, end_cause FROM monitoring_periods"
|
||
).fetchone()
|
||
conn.close()
|
||
assert row == ("2026-09-01T10:00:00+00:00", "user_disabled")
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_quit_preserves_an_active_monitoring_period(self, tmp_path):
|
||
"""Quitting an active dashboard never closes or mutates its period."""
|
||
db = tmp_path / "test.db"
|
||
conn = init_store(db)
|
||
_open_period(conn, "2026-09-01T09:00:00+00:00")
|
||
conn.close()
|
||
app = FenrisTuiApp(store_path=db)
|
||
|
||
with patch("fenris.tui.query_service_state", return_value={
|
||
"boot_enabled": True, "timer_active": True,
|
||
"last_collect_ok": None, "last_collect_age_s": None,
|
||
"last_collect_reason": None,
|
||
}), patch("fenris.tui.subprocess.run") as subprocess_run, patch.object(
|
||
app, "_run_helper"
|
||
) as run_helper:
|
||
async with app.run_test() as pilot:
|
||
await pilot.press("q")
|
||
assert not app.is_running
|
||
subprocess_run.assert_not_called()
|
||
run_helper.assert_not_called()
|
||
|
||
conn = sqlite3.connect(db)
|
||
row = conn.execute(
|
||
"SELECT started_at, ended_at, end_cause FROM monitoring_periods"
|
||
).fetchone()
|
||
conn.close()
|
||
assert row == ("2026-09-01T09:00:00+00:00", None, None)
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_branding_and_one_time_auth_banner(self, tmp_path):
|
||
"""Identity is visible at launch; auth notice clears once per session."""
|
||
app = FenrisTuiApp(
|
||
store_path=tmp_path / "nonexistent.db",
|
||
refresh_interval_s=0.2,
|
||
)
|
||
auth_notice = "privileged actions will prompt for authentication (polkit)"
|
||
|
||
async with app.run_test() as pilot:
|
||
headline = str(app.query_one("#headline-band").render())
|
||
assert "Fenris — NVMe endurance monitor" in headline
|
||
assert auth_notice in headline
|
||
assert "by Bongbetic" in str(app.query_one("#service-strip").render())
|
||
|
||
await pilot.pause(0.25)
|
||
assert auth_notice not in str(app.query_one("#headline-band").render())
|
||
|
||
await pilot.pause(0.25)
|
||
assert auth_notice not in str(app.query_one("#headline-band").render())
|
||
|
||
fresh_app = FenrisTuiApp(
|
||
store_path=tmp_path / "nonexistent.db",
|
||
refresh_interval_s=0.2,
|
||
)
|
||
async with fresh_app.run_test():
|
||
assert auth_notice in str(fresh_app.query_one("#headline-band").render())
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# CI-4: Disclosures, empty-store greeting, first-run
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class TestDisclosuresAndGreeting:
|
||
@pytest.mark.asyncio
|
||
async def test_disclosures_screen(self, tmp_path):
|
||
"""Disclosures view renders six disclosures verbatim."""
|
||
app = FenrisTuiApp(store_path=tmp_path / "nonexistent.db")
|
||
async with app.run_test() as pilot:
|
||
await pilot.press("d")
|
||
# Modal should be pushed
|
||
assert len(app.screen_stack) > 1
|
||
disc_text = str(app.screen.query_one("Static").render())
|
||
for i in range(1, 7):
|
||
assert "%d." % i in disc_text
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_empty_store_greeting(self, tmp_path):
|
||
"""Empty store shows 'no observations yet' with enable hint."""
|
||
app = FenrisTuiApp(store_path=tmp_path / "nonexistent.db")
|
||
async with app.run_test() as pilot:
|
||
headline = str(app.query_one("#headline-band").render())
|
||
assert "no observations yet" in headline.lower()
|
||
assert "enable" in headline.lower() or "resume" in headline.lower()
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# IN-3: First-run opt-in
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class TestFirstRun:
|
||
@pytest.mark.asyncio
|
||
async def test_first_run_prompt(self, tmp_path):
|
||
"""First-run prompt enables timer and opens first period."""
|
||
app = FenrisTuiApp(store_path=tmp_path / "nonexistent.db")
|
||
async with app.run_test() as pilot:
|
||
headline = str(app.query_one("#headline-band").render())
|
||
assert "no observations yet" in headline.lower()
|
||
# The enable hint should mention resume
|
||
assert "resume" in headline.lower()
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# TUI-4: Layout regions normative
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class TestLayoutNormative:
|
||
@pytest.mark.asyncio
|
||
async def test_no_page_navigation(self, tmp_path):
|
||
"""No page navigation keys exist (variant switching was prototype-only)."""
|
||
app = FenrisTuiApp(store_path=tmp_path / "nonexistent.db")
|
||
# Check that only production bindings exist
|
||
binding_keys = {b.key for b in app.BINDINGS}
|
||
assert "left" not in binding_keys
|
||
assert "right" not in binding_keys
|
||
assert "1" not in binding_keys
|
||
assert "2" not in binding_keys
|
||
assert "3" not in binding_keys
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_pause_resume_asymmetry(self, tmp_path):
|
||
"""Pause asks, resume does not (TUI-2)."""
|
||
app = FenrisTuiApp(store_path=tmp_path / "nonexistent.db")
|
||
async with app.run_test() as pilot:
|
||
# Pause should push a confirmation screen
|
||
await pilot.press("p")
|
||
assert len(app.screen_stack) > 1
|
||
# Press n to cancel
|
||
await pilot.press("n")
|
||
assert len(app.screen_stack) == 1
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# CI-1: State matrix exhaustive combinations
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class TestStateMatrixCombinations:
|
||
"""CI-1: confidence × freshness × baseline tier combinations."""
|
||
|
||
def test_unsupported_with_fresh_data(self, tmp_path):
|
||
"""Fresh data but no baseline → Unavailable + fresh."""
|
||
conn = init_store(tmp_path / "test.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)
|
||
# Sample must be within FRESH_THRESHOLD_S of clock
|
||
fresh_ts = (_clock() - timedelta(seconds=FRESH_THRESHOLD_S - 10)).isoformat()
|
||
_insert_sample(conn, fresh_ts)
|
||
proj = compute_projection(conn, _clock())
|
||
assert proj.confidence_state == ConfidenceState.UNSUPPORTED
|
||
freshness = grade_freshness(fresh_ts, _clock())
|
||
assert freshness == "fresh"
|
||
conn.close()
|
||
|
||
def test_limited_with_stale_data(self, tmp_path):
|
||
"""Stale data with baseline → Limited or Unsupported (young regime + stale)."""
|
||
conn = init_store(tmp_path / "test.db")
|
||
_insert_baseline(conn, tbw_tb=1.0, verified=True)
|
||
_insert_segment(conn, opened_at="2026-09-01T00:00:00+00:00")
|
||
_open_period(conn)
|
||
for i in range(20):
|
||
d = (datetime(2026, 9, 1) + timedelta(days=i)).strftime("%Y-%m-%d")
|
||
_insert_day(conn, d, bw=1024*1024*100)
|
||
stale_ts = (_clock() - timedelta(days=3)).isoformat()
|
||
_insert_sample(conn, stale_ts)
|
||
proj = compute_projection(conn, _clock())
|
||
assert proj.confidence_state != ConfidenceState.SUPPORTED
|
||
freshness = grade_freshness(stale_ts, _clock())
|
||
assert freshness == "stale"
|
||
conn.close()
|
||
|
||
def test_supported_with_unverified_baseline(self, tmp_path):
|
||
"""Incomplete provenance → unverified baseline tier."""
|
||
conn = init_store(tmp_path / "test.db")
|
||
# Insert baseline with incomplete provenance (missing source_url)
|
||
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 (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
|
||
(1.0, None, "v1.0", "2026-01-01", "Samsung SSD 970 EVO Plus 1TB",
|
||
1024000000000, None, False,
|
||
"2026-01-01T00:00:00+00:00", "2026-01-01T00:00:00+00:00"),
|
||
)
|
||
conn.commit()
|
||
_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())
|
||
# Incomplete provenance → UNVERIFIED tier
|
||
assert proj.baseline_tier.value == "unverified_override"
|
||
conn.close()
|