Files
Fenris/tests/test_tui.py
T

1098 lines
43 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""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,
DailyBarGraph,
_format_remaining,
_sparkline,
_habit_bar,
_query_usage_history,
_query_drive_health,
_query_service_facts,
_query_daily_graph_data,
_query_hourly_graph_data,
_RANGE_OPTIONS,
_RANGE_DEFAULT,
_BAR_HEIGHT,
_GLYPH_ALLOCATED,
_GLYPH_UNALLOCATED,
_GLYPH_GAP,
_GLYPH_ZERO,
_GLYPH_PARTIAL,
)
# ---------------------------------------------------------------------------
# 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).
Maker credit removed from service strip (issue #79: single placement in title).
"""
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
# Maker credit now in titlebox only (issue #79)
assert "by Bongbetic" not 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.
Maker credit in titlebox only (issue #79: Fenris by Bongbetic).
"""
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())
# Identity now shows Fenris by Bongbetic (issue #79)
assert "Fenris by Bongbetic" in headline
assert auth_notice in headline
# Maker credit removed from service strip (issue #79)
assert "by Bongbetic" not 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()
# ---------------------------------------------------------------------------
# Hour observation helper for graph tests
# ---------------------------------------------------------------------------
def _insert_hour(conn, hour, bw=0, active=3600, idle=0, powered_off=0,
unknown=0, coverage=1.0, samples=1):
"""Insert an hour observation row."""
conn.execute(
"INSERT INTO hour_observations "
"(hour, active_seconds, idle_seconds, powered_off_seconds, unknown_seconds, "
" bytes_written_delta, bytes_read_delta, sample_count, coverage) "
"VALUES (?, ?, ?, ?, ?, ?, 0, ?, ?)",
(hour, active, idle, powered_off, unknown, bw, samples, coverage),
)
conn.commit()
def _insert_day_with_unattributed(conn, day, bw=0, unattributed=0,
coverage=0.95, samples=24):
"""Insert a day aggregate with explicit unattributed bytes."""
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, "
" unattributed_bytes_written, unattributed_bytes_read) "
"VALUES (?, 3600, 0, 0, 0, ?, 0, ?, ?, ?, 0)",
(day, bw, samples, coverage, unattributed),
)
conn.commit()
# ---------------------------------------------------------------------------
# Graph data query tests (issue #75)
# ---------------------------------------------------------------------------
class TestQueryDailyGraphData:
def test_empty_store(self, tmp_path):
conn = init_store(tmp_path / "test.db")
result = _query_daily_graph_data(conn)
assert result == []
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_daily_graph_data(conn)
assert len(result) == 14
assert result[0]["total_bytes"] == 1024*1024*100
assert result[0]["allocated_bytes"] == 1024*1024*100
assert result[0]["unallocated_bytes"] == 0
assert result[0]["is_zero"] is False
conn.close()
def test_zero_day(self, tmp_path):
conn = init_store(tmp_path / "test.db")
_insert_segment(conn)
_open_period(conn)
_insert_day(conn, "2026-09-20", bw=0, coverage=0.95, samples=24)
result = _query_daily_graph_data(conn)
assert len(result) == 1
assert result[0]["is_zero"] is True
assert result[0]["total_bytes"] == 0
conn.close()
def test_unattributed_bytes(self, tmp_path):
conn = init_store(tmp_path / "test.db")
_insert_segment(conn)
_open_period(conn)
_insert_day_with_unattributed(
conn, "2026-09-20", bw=500, unattributed=300
)
result = _query_daily_graph_data(conn)
assert len(result) == 1
assert result[0]["allocated_bytes"] == 500
assert result[0]["unallocated_bytes"] == 300
assert result[0]["total_bytes"] == 800
conn.close()
def test_gap_day(self, tmp_path):
conn = init_store(tmp_path / "test.db")
_insert_segment(conn)
_open_period(conn)
# Day with no hours but unknown seconds (gap in monitoring period)
conn.execute(
"INSERT INTO day_aggregates "
"(day, active_seconds, idle_seconds, powered_off_seconds, unknown_seconds, "
" bytes_written_delta, sample_count, coverage) "
"VALUES (?, 0, 0, 0, 86400, 0, 0, 0.0)",
("2026-09-20",),
)
conn.commit()
result = _query_daily_graph_data(conn)
assert len(result) == 1
assert result[0]["is_gap"] is True
conn.close()
class TestQueryHourlyGraphData:
def test_empty_day(self, tmp_path):
conn = init_store(tmp_path / "test.db")
result = _query_hourly_graph_data(conn, "2026-09-20")
assert result == []
conn.close()
def test_with_hours(self, tmp_path):
conn = init_store(tmp_path / "test.db")
for h in range(24):
hour = "2026-09-20T%02d:00:00+00:00" % h
bw = 1024*1024*100 if h in (10, 14) else 0
_insert_hour(conn, hour, bw=bw)
result = _query_hourly_graph_data(conn, "2026-09-20")
assert len(result) == 24
assert result[10]["bytes_written"] == 1024*1024*100
assert result[10]["is_zero"] is False
assert result[0]["is_zero"] is True
conn.close()
def test_hour_labels(self, tmp_path):
conn = init_store(tmp_path / "test.db")
_insert_hour(conn, "2026-09-20T12:00:00+00:00", bw=1000)
result = _query_hourly_graph_data(conn, "2026-09-20")
assert len(result) == 1
assert result[0]["local_label"] == "12"
conn.close()
# ---------------------------------------------------------------------------
# DailyBarGraph widget unit tests (issue #75)
# ---------------------------------------------------------------------------
class TestDailyBarGraph:
def test_empty_data(self, tmp_path):
"""Empty data shows awaiting message."""
app = FenrisTuiApp(store_path=tmp_path / "nonexistent.db")
# We test the widget directly via the app's compose
graph = DailyBarGraph()
# Simulate setting empty data
from textual.app import App as TextualApp
class _TestApp(TextualApp):
def compose(self):
yield graph
# We can't easily test widget lifecycle outside an app, so test the data
assert graph._all_day_data == []
assert graph.range_days == _RANGE_DEFAULT
def test_range_default(self):
"""Default range is 14 days."""
graph = DailyBarGraph()
assert graph.range_days == _RANGE_DEFAULT
assert graph.selected_index == -1
assert graph.view_mode == "daily"
def test_trim_to_range(self):
"""Data is trimmed to the selected range."""
graph = DailyBarGraph()
data = [{"day": "2026-09-%02d" % d, "total_bytes": d * 100}
for d in range(1, 31)]
graph._all_day_data = data
graph.range_days = 7
graph._trim_to_range()
assert len(graph._day_data) == 7
assert graph._day_data[0]["day"] == "2026-09-24"
def test_range_all_fits(self):
"""When data fits within range, all days are shown."""
graph = DailyBarGraph()
data = [{"day": "2026-09-%02d" % d, "total_bytes": d * 100}
for d in range(1, 8)]
graph._all_day_data = data
graph.range_days = 14
graph._trim_to_range()
assert len(graph._day_data) == 7
# ---------------------------------------------------------------------------
# Bar graph headless TUI tests (issue #75)
# ---------------------------------------------------------------------------
class TestBarGraphTUI:
"""Headless tests for the interactive bar graph."""
@pytest.mark.asyncio
async def test_graph_renders_with_data(self, tmp_path):
"""Graph widget renders with day data."""
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)
_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(size=(80, 24)) as pilot:
graph = app.query_one("#usage-history")
assert isinstance(graph, DailyBarGraph)
assert len(graph._day_data) > 0
# Legend should be visible
legend = str(app.query_one("#bar-legend").render())
assert "alloc" in legend
assert "zero" in legend
@pytest.mark.asyncio
async def test_arrow_selection(self, tmp_path):
"""Left/right arrows move the selection."""
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)
_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(size=(80, 24)) as pilot:
graph = app.query_one("#usage-history")
app.set_focus(graph)
await pilot.pause()
await pilot.pause()
# Right arrow selects first bar
await pilot.press("right")
await pilot.pause()
assert graph.selected_index == 0
# Right again moves to second
await pilot.press("right")
await pilot.pause()
assert graph.selected_index == 1
# Left moves back
await pilot.press("left")
await pilot.pause()
assert graph.selected_index == 0
@pytest.mark.asyncio
async def test_range_switching(self, tmp_path):
"""1-4 keys switch the visible range."""
conn = init_store(tmp_path / "test.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)
_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(size=(80, 24)) as pilot:
graph = app.query_one("#usage-history")
app.set_focus(graph)
await pilot.pause()
await pilot.pause()
# Default is 14
assert graph.range_days == 14
assert len(graph._day_data) == 14
# Press 1 for 7-day range
await pilot.press("1")
await pilot.pause()
assert graph.range_days == 7
assert len(graph._day_data) == 7
# Press 3 for 28-day range
await pilot.press("3")
await pilot.pause()
assert graph.range_days == 28
assert len(graph._day_data) == 28
# Press 4 for 90-day range (only 30 days available)
await pilot.press("4")
await pilot.pause()
assert graph.range_days == 90
assert len(graph._day_data) == 30
@pytest.mark.asyncio
async def test_readout_updates_on_selection(self, tmp_path):
"""Readout shows selected day info."""
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)
_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(size=(80, 24)) as pilot:
graph = app.query_one("#usage-history")
app.set_focus(graph)
await pilot.pause()
await pilot.pause()
# No selection initially
readout = str(app.query_one("#bar-readout").render())
assert "select" in readout.lower()
# Select first bar
await pilot.press("right")
await pilot.pause()
readout = str(app.query_one("#bar-readout").render())
assert "2026-09" in readout
assert "GB" in readout
@pytest.mark.asyncio
async def test_hourly_drill_down_and_back(self, tmp_path):
"""Enter drills into hourly view, Esc returns to daily."""
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)
# Insert hours for each day
for h in range(24):
hour = "%sT%02d:00:00+00:00" % (d, h)
bw_h = 1024*1024*10 if h == 12 else 0
_insert_hour(conn, hour, bw=bw_h)
_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(size=(80, 24)) as pilot:
graph = app.query_one("#usage-history")
app.set_focus(graph)
await pilot.pause()
await pilot.pause()
# Select a day
await pilot.press("right")
await pilot.pause()
assert graph.selected_index == 0
assert graph.view_mode == "daily"
# Enter drill-down
await pilot.press("enter")
await pilot.pause()
assert graph.view_mode == "hourly"
assert graph.drill_day is not None
assert len(graph._hour_data) == 24
# Esc returns to daily
await pilot.press("escape")
await pilot.pause()
assert graph.view_mode == "daily"
assert graph.drill_day is None
@pytest.mark.asyncio
async def test_empty_store_graph(self, tmp_path):
"""Empty store shows awaiting message in graph."""
app = FenrisTuiApp(store_path=tmp_path / "nonexistent.db")
async with app.run_test(size=(80, 24)) as pilot:
graph = app.query_one("#usage-history")
assert isinstance(graph, DailyBarGraph)
render = str(app.query_one("#bar-render").render())
assert "awaiting" in render.lower()
@pytest.mark.asyncio
async def test_graph_border_title(self, tmp_path):
"""Graph pane has a border title."""
app = FenrisTuiApp(store_path=tmp_path / "nonexistent.db")
async with app.run_test(size=(80, 24)) as pilot:
graph = app.query_one("#usage-history")
assert graph.border_title == "usage history"
@pytest.mark.asyncio
async def test_glyphs_in_legend(self, tmp_path):
"""Legend shows all required glyphs."""
conn = init_store(tmp_path / "test.db")
_insert_segment(conn)
_open_period(conn)
for i in range(7):
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")
conn.close()
app = FenrisTuiApp(store_path=tmp_path / "test.db")
async with app.run_test(size=(80, 24)) as pilot:
await pilot.pause()
legend = str(app.query_one("#bar-legend").render())
assert _GLYPH_ALLOCATED in legend
assert _GLYPH_UNALLOCATED in legend
assert _GLYPH_GAP in legend
assert _GLYPH_ZERO in legend
assert _GLYPH_PARTIAL in legend