feat(tui): Panes TUI on Textual (issue #28)
Implements keyboard-first Panes TUI per spec section 7: - One dense screen: headline band, usage-history, drive-health, service strip - Bindings p/r/c/d/q with pause-asks/resume-doesnt asymmetry - Privileged actions via terminal-attached fenris-monitor subprocess - Disclosures view, empty-store greeting, first-run opt-in - 35 headless tests: CI-1 state matrix, TUI-1/TUI-4 layout, CI-4, IN-3 Blocker #27 resolved. Closes #28
This commit is contained in:
xavierk
@@ -11,6 +11,7 @@ dependencies = [
|
||||
dev = [
|
||||
"pytest>=7.0.0",
|
||||
"pytest-cov>=4.0.0",
|
||||
"pytest-asyncio>=0.20.0",
|
||||
]
|
||||
|
||||
[tool.pytest.ini_options]
|
||||
|
||||
@@ -0,0 +1,574 @@
|
||||
"""Panes TUI: keyboard-first Textual app (spec §7, TUI-1, TUI-4).
|
||||
|
||||
One dense screen, four normative regions:
|
||||
1. Headline band (full width, top)
|
||||
2. Usage-history pane (left, wider)
|
||||
3. Drive-health + settings pane (right, narrower)
|
||||
4. Service strip (full width, bottom)
|
||||
|
||||
Bindings: p (pause, asks), r (resume), c (collect now), d (disclosures), q (quit).
|
||||
Privileged actions route through fenris-monitor as terminal-attached subprocesses
|
||||
(LC-6, LC-8). The TUI never samples in-process.
|
||||
|
||||
Criteria: TUI-1, TUI-2, TUI-4, CI-1, CI-2, CI-4, IN-3, LC-6, LC-8.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import sqlite3
|
||||
import subprocess
|
||||
import sys
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
from textual.app import App, ComposeResult
|
||||
from textual.binding import Binding
|
||||
from textual.containers import Horizontal, Vertical
|
||||
from textual.screen import ModalScreen
|
||||
from textual.widgets import Static
|
||||
|
||||
from .projection import (
|
||||
ConfidenceState,
|
||||
ProjectionResult,
|
||||
ScenarioRange,
|
||||
compute_projection,
|
||||
)
|
||||
from .status import (
|
||||
CADENCE_DEFAULT_S,
|
||||
FRESH_THRESHOLD_S,
|
||||
STALENESS_THRESHOLD_S,
|
||||
ConfigError,
|
||||
NewerSchema,
|
||||
StoreFault,
|
||||
format_disclosures,
|
||||
freshness_age_human,
|
||||
grade_freshness,
|
||||
open_store_readonly,
|
||||
query_service_state,
|
||||
read_config,
|
||||
)
|
||||
from .monitoring_periods import get_open_period
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _format_remaining(seconds: float) -> str:
|
||||
"""Format remaining lifespan as human-readable string."""
|
||||
if seconds <= 0:
|
||||
return "endurance exhausted"
|
||||
years = int(seconds // 31557600)
|
||||
rem = seconds % 31557600
|
||||
days = int(rem // 86400)
|
||||
rem %= 86400
|
||||
hours = int(rem // 3600)
|
||||
parts = []
|
||||
if years:
|
||||
parts.append("%d yr" % years)
|
||||
if days or years:
|
||||
parts.append("%d d" % days)
|
||||
parts.append("%d h" % hours)
|
||||
return " ".join(parts)
|
||||
|
||||
|
||||
def _sparkline(values: List[float], width: int = 40) -> str:
|
||||
"""Render a sparkline from daily bytes-written values."""
|
||||
if not values:
|
||||
return ""
|
||||
mx = max(values) or 1.0
|
||||
blocks = " ▁▂▃▄▅▆▇█"
|
||||
step = max(1, len(values) // width or 1)
|
||||
picked = values[-width * step :][::step][-width:]
|
||||
return "".join(
|
||||
blocks[min(len(blocks) - 1, int(v / mx * (len(blocks) - 1)) + (1 if v > 0 else 0))]
|
||||
for v in picked
|
||||
)
|
||||
|
||||
|
||||
def _habit_bar(a: float, i: float, o: float, u: float, width: int = 40) -> str:
|
||||
"""Render the habit-split bar with legend."""
|
||||
total = a + i + o + u or 1.0
|
||||
segs = [
|
||||
("a", a, "#33ff33"),
|
||||
("i", i, "#ffff33"),
|
||||
("o", o, "#33ffff"),
|
||||
("?", u, "#ff33ff"),
|
||||
]
|
||||
parts = []
|
||||
for label, v, _color in segs:
|
||||
n = max(1 if v > 0 else 0, round(v / total * width))
|
||||
parts.append(label * n)
|
||||
bar = "".join(parts)
|
||||
legend = " active %d%% · idle %d%% · powered-off %d%% · unknown %d%%" % (
|
||||
round(a / total * 100),
|
||||
round(i / total * 100),
|
||||
round(o / total * 100),
|
||||
round(u / total * 100),
|
||||
)
|
||||
return bar + "\n" + legend
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Data queries for TUI regions
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _query_usage_history(conn: sqlite3.Connection) -> Dict[str, Any]:
|
||||
"""Query usage-history data for the left pane."""
|
||||
cursor = conn.execute(
|
||||
"SELECT day, active_seconds, idle_seconds, powered_off_seconds, "
|
||||
"unknown_seconds, bytes_written_delta, coverage "
|
||||
"FROM day_aggregates ORDER BY day"
|
||||
)
|
||||
days = cursor.fetchall()
|
||||
if not days:
|
||||
return {
|
||||
"sparkline": "",
|
||||
"habit_bar": "",
|
||||
"num_days": 0,
|
||||
"min_gb": 0,
|
||||
"max_gb": 0,
|
||||
"habit_change": False,
|
||||
"gap_days": [],
|
||||
"day_labels": [],
|
||||
}
|
||||
|
||||
bw_values = [d[5] for d in days]
|
||||
total_a = sum(d[1] for d in days)
|
||||
total_i = sum(d[2] for d in days)
|
||||
total_o = sum(d[3] for d in days)
|
||||
total_u = sum(d[4] for d in days)
|
||||
total = total_a + total_i + total_o + total_u or 1
|
||||
|
||||
spark = _sparkline([b / 1e9 for b in bw_values]) # Convert to GB for display
|
||||
bar = _habit_bar(total_a / total, total_i / total, total_o / total, total_u / total)
|
||||
|
||||
min_gb = min(bw_values) / 1e9 if bw_values else 0
|
||||
max_gb = max(bw_values) / 1e9 if bw_values else 0
|
||||
|
||||
return {
|
||||
"sparkline": spark,
|
||||
"habit_bar": bar,
|
||||
"num_days": len(days),
|
||||
"min_gb": min_gb,
|
||||
"max_gb": max_gb,
|
||||
"habit_change": False,
|
||||
"gap_days": [],
|
||||
"day_labels": [d[0] for d in days],
|
||||
}
|
||||
|
||||
|
||||
def _query_drive_health(conn: sqlite3.Connection) -> Dict[str, Any]:
|
||||
"""Query drive health data for the right pane."""
|
||||
cursor = conn.execute(
|
||||
"SELECT mn, sn, fr, temperature_c, available_spare, media_errors, "
|
||||
"power_on_hours, power_cycles, unsafe_shutdowns, capacity_bytes, "
|
||||
"percentage_used, data_units_written "
|
||||
"FROM samples ORDER BY id DESC LIMIT 1"
|
||||
)
|
||||
row = cursor.fetchone()
|
||||
if row is None:
|
||||
return {
|
||||
"model": "unknown",
|
||||
"temp": 0,
|
||||
"spare": 0,
|
||||
"media_errors": 0,
|
||||
"poh": 0,
|
||||
"cycles": 0,
|
||||
"unsafe_shutdowns": 0,
|
||||
"capacity": "unknown",
|
||||
"percentage_used": 0,
|
||||
"written_tb": 0,
|
||||
}
|
||||
|
||||
capacity = row[9]
|
||||
capacity_str = "%d GB" % (capacity / 1e9) if capacity else "unknown"
|
||||
written_tb = (row[11] * 512 * 1000) / 1e12 if row[11] else 0 # DUW to TB
|
||||
|
||||
return {
|
||||
"model": row[0] or "unknown",
|
||||
"temp": row[3] or 0,
|
||||
"spare": row[4] or 0,
|
||||
"media_errors": row[5] or 0,
|
||||
"poh": row[6] or 0,
|
||||
"cycles": row[7] or 0,
|
||||
"unsafe_shutdowns": row[8] or 0,
|
||||
"capacity": capacity_str,
|
||||
"percentage_used": row[10] or 0,
|
||||
"written_tb": written_tb,
|
||||
}
|
||||
|
||||
|
||||
def _query_service_facts(conn: sqlite3.Connection, clock_now: datetime) -> Dict[str, Any]:
|
||||
"""Query service facts for the bottom strip."""
|
||||
# Get freshness
|
||||
cursor = conn.execute("SELECT ts FROM samples ORDER BY id DESC LIMIT 1")
|
||||
row = cursor.fetchone()
|
||||
newest_ts = row[0] if row else None
|
||||
freshness = grade_freshness(newest_ts, clock_now)
|
||||
|
||||
# Get monitoring period
|
||||
period = get_open_period(conn)
|
||||
period_info = "no monitoring period"
|
||||
if period:
|
||||
period_info = "open since %s" % period["started_at"][:10]
|
||||
|
||||
# Get service state
|
||||
try:
|
||||
svc = query_service_state()
|
||||
except Exception:
|
||||
svc = {
|
||||
"boot_enabled": False,
|
||||
"timer_active": False,
|
||||
"last_collect_ok": None,
|
||||
"last_collect_age_s": None,
|
||||
"last_collect_reason": None,
|
||||
}
|
||||
|
||||
return {
|
||||
"freshness": freshness,
|
||||
"freshness_age_s": None,
|
||||
"period": period_info,
|
||||
**svc,
|
||||
}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Modal screens
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class ConfirmPause(ModalScreen[bool]):
|
||||
"""Pause asks for confirmation (spec §7.4, LC-6)."""
|
||||
|
||||
BINDINGS = [
|
||||
Binding("y", "yes", "Pause"),
|
||||
Binding("n", "no", "Cancel"),
|
||||
Binding("escape", "no", "Cancel", show=False),
|
||||
]
|
||||
|
||||
def compose(self) -> ComposeResult:
|
||||
yield Static(
|
||||
"[bold]Pause monitoring?[/bold]\n\n"
|
||||
"This closes the current monitoring period.\n"
|
||||
"Paused time is [bold]excluded[/bold] from your usage habit\n"
|
||||
"(powered-off time would still count).\n\n"
|
||||
"[dim]y pause · n cancel[/dim]",
|
||||
id="confirm-text",
|
||||
)
|
||||
|
||||
def action_yes(self) -> None:
|
||||
self.dismiss(True)
|
||||
|
||||
def action_no(self) -> None:
|
||||
self.dismiss(False)
|
||||
|
||||
|
||||
class DisclosuresScreen(ModalScreen[None]):
|
||||
"""Disclosures view with six verbatim disclosures (spec §6.11, CI-4)."""
|
||||
|
||||
BINDINGS = [
|
||||
Binding("escape", "close", "Close"),
|
||||
Binding("d", "close", "Close"),
|
||||
]
|
||||
|
||||
def compose(self) -> ComposeResult:
|
||||
text = format_disclosures()
|
||||
yield Static(text + "\n\n[dim]esc to close[/dim]", id="disc-text")
|
||||
|
||||
def action_close(self) -> None:
|
||||
self.dismiss()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Main TUI App
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class FenrisTuiApp(App):
|
||||
"""Fenris Panes TUI — keyboard-first, one dense screen (spec §7)."""
|
||||
|
||||
TITLE = "Fenris"
|
||||
SUB_TITLE = "NVMe endurance monitor"
|
||||
|
||||
CSS = """
|
||||
#main-grid {
|
||||
layout: grid;
|
||||
grid-size: 2 3;
|
||||
grid-columns: 3fr 2fr;
|
||||
grid-rows: 8 1fr 7;
|
||||
height: 1fr;
|
||||
}
|
||||
#headline-band { column-span: 2; }
|
||||
#service-strip { column-span: 2; }
|
||||
.pane { border: round #555555; padding: 0 1; }
|
||||
#confirm-text { padding: 1 2; }
|
||||
#disc-text { padding: 1 2; }
|
||||
"""
|
||||
|
||||
BINDINGS = [
|
||||
Binding("p", "pause", "Pause", show=False),
|
||||
Binding("r", "resume", "Resume", show=False),
|
||||
Binding("c", "collect", "Collect Now", show=False),
|
||||
Binding("d", "disclose", "Disclosures", show=False),
|
||||
Binding("q", "quit", "Quit", show=False),
|
||||
]
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
store_path: Optional[Path] = None,
|
||||
config_path: Optional[Path] = None,
|
||||
helper_path: Optional[str] = None,
|
||||
**kwargs,
|
||||
) -> None:
|
||||
super().__init__(**kwargs)
|
||||
self.store_path = store_path or Path("/var/lib/fenris/observations.db")
|
||||
self.config_path = config_path
|
||||
self.helper_path = helper_path or "/usr/libexec/fenris/fenris-monitor"
|
||||
self._conn: Optional[sqlite3.Connection] = None
|
||||
self._clock_now = datetime.now(timezone.utc)
|
||||
|
||||
def compose(self) -> ComposeResult:
|
||||
with Vertical(id="main-grid"):
|
||||
yield Static("", id="headline-band", classes="pane")
|
||||
yield Static("", id="usage-history", classes="pane")
|
||||
yield Static("", id="drive-health", classes="pane")
|
||||
yield Static("", id="service-strip", classes="pane")
|
||||
|
||||
def on_mount(self) -> None:
|
||||
"""Set border titles and render initial state."""
|
||||
self.query_one("#headline-band").border_title = "headline"
|
||||
self.query_one("#usage-history").border_title = "usage history"
|
||||
self.query_one("#drive-health").border_title = "drive"
|
||||
self.query_one("#service-strip").border_title = "service + actions"
|
||||
self._refresh()
|
||||
|
||||
def _open_store(self) -> Optional[sqlite3.Connection]:
|
||||
"""Open store read-only, handling faults."""
|
||||
try:
|
||||
return open_store_readonly(self.store_path)
|
||||
except (StoreFault, NewerSchema):
|
||||
return None
|
||||
|
||||
def _refresh(self) -> None:
|
||||
"""Refresh all four regions from store data."""
|
||||
self._clock_now = datetime.now(timezone.utc)
|
||||
conn = self._open_store()
|
||||
|
||||
if conn is None:
|
||||
self._render_empty_or_fault()
|
||||
return
|
||||
|
||||
try:
|
||||
self._conn = conn
|
||||
self._render_all_regions(conn)
|
||||
finally:
|
||||
conn.close()
|
||||
self._conn = None
|
||||
|
||||
def _render_empty_or_fault(self) -> None:
|
||||
"""Render empty store greeting or store fault."""
|
||||
if not self.store_path.exists():
|
||||
# Empty store — greeting with enable hint (IN-3)
|
||||
self.query_one("#headline-band").update(
|
||||
"[bold]No observations yet[/bold]\n\n"
|
||||
"Enable monitoring: fenris monitor resume"
|
||||
)
|
||||
self.query_one("#usage-history").update("")
|
||||
self.query_one("#drive-health").update("")
|
||||
self.query_one("#service-strip").update(
|
||||
"boot: disabled · timer: inactive · last collect: unknown · freshness: empty\n"
|
||||
"p pause · r resume · c collect · d disclosures · q quit"
|
||||
)
|
||||
else:
|
||||
# Store fault (FL-4)
|
||||
self.query_one("#headline-band").update(
|
||||
"[bold red]Observation store unreadable[/bold red]\n"
|
||||
"Check journalctl -u fenris-collect.service"
|
||||
)
|
||||
self.query_one("#usage-history").update("")
|
||||
self.query_one("#drive-health").update("")
|
||||
self.query_one("#service-strip").update(
|
||||
"p pause · r resume · c collect · d disclosures · q quit"
|
||||
)
|
||||
|
||||
def _render_all_regions(self, conn: sqlite3.Connection) -> None:
|
||||
"""Render all four regions from live store data."""
|
||||
# --- Headline band (§7.2) ---
|
||||
try:
|
||||
proj = compute_projection(conn, self._clock_now)
|
||||
headline = self._format_headline(proj)
|
||||
confidence = self._format_confidence(proj)
|
||||
scenario = self._format_scenario(proj)
|
||||
self.query_one("#headline-band").update(
|
||||
headline + "\n" + confidence + "\n" + scenario
|
||||
)
|
||||
except Exception:
|
||||
self.query_one("#headline-band").update(
|
||||
"[bold]No projection available[/bold]"
|
||||
)
|
||||
|
||||
# --- Usage-history pane (§7.2 left) ---
|
||||
history = _query_usage_history(conn)
|
||||
history_text = "[bold]Usage history[/bold] · %d days · %.1f–%.1f GB/day\n %s\n %s" % (
|
||||
history["num_days"],
|
||||
history["min_gb"],
|
||||
history["max_gb"],
|
||||
history["sparkline"],
|
||||
history["habit_bar"],
|
||||
)
|
||||
self.query_one("#usage-history").update(history_text)
|
||||
|
||||
# --- Drive-health pane (§7.2 right) ---
|
||||
health = _query_drive_health(conn)
|
||||
health_text = (
|
||||
"[bold]Drive health[/bold] · %s\n"
|
||||
" temperature %d°C · spare %d%%\n"
|
||||
" media errors %d · unsafe shutdowns %d\n"
|
||||
" power-on %d h · %d cycles · %s\n\n"
|
||||
"[bold]Settings[/bold]\n"
|
||||
" vendor wear: %d%% used · %.1f TB written"
|
||||
) % (
|
||||
health["model"],
|
||||
health["temp"],
|
||||
health["spare"],
|
||||
health["media_errors"],
|
||||
health["unsafe_shutdowns"],
|
||||
health["poh"],
|
||||
health["cycles"],
|
||||
health["capacity"],
|
||||
health["percentage_used"],
|
||||
health["written_tb"],
|
||||
)
|
||||
self.query_one("#drive-health").update(health_text)
|
||||
|
||||
# --- Service strip (§7.2 bottom) ---
|
||||
try:
|
||||
svc = _query_service_facts(conn, self._clock_now)
|
||||
boot = "enabled" if svc.get("boot_enabled") else "disabled"
|
||||
activity = "active" if svc.get("timer_active") else "inactive"
|
||||
collect = "ok" if svc.get("last_collect_ok") else "FAILED"
|
||||
freshness = svc.get("freshness", "unknown")
|
||||
self.query_one("#service-strip").update(
|
||||
"boot: %s · timer: %s · last collect: %s · freshness: %s\n"
|
||||
"%s\n"
|
||||
"p pause · r resume · c collect · d disclosures · q quit"
|
||||
% (boot, activity, collect, freshness, svc.get("period", ""))
|
||||
)
|
||||
except Exception:
|
||||
self.query_one("#service-strip").update(
|
||||
"boot: unknown · timer: unknown · last collect: unknown · freshness: unknown\n"
|
||||
"p pause · r resume · c collect · d disclosures · q quit"
|
||||
)
|
||||
|
||||
def _format_headline(self, proj: ProjectionResult) -> str:
|
||||
"""Format the lifespan headline (spec §6.11)."""
|
||||
if proj.headline_remaining_seconds is None:
|
||||
if proj.zero_rate_fact:
|
||||
return "[bold]Usage-adjusted theoretical lifespan: [red]no finite projection from this history[/red][/bold]"
|
||||
if proj.warming_fact:
|
||||
return "[bold]Usage-adjusted theoretical lifespan: [yellow]%s[/yellow][/bold]" % proj.warming_fact
|
||||
return "[bold]Usage-adjusted theoretical lifespan: [red]no projection available[/red][/bold]"
|
||||
|
||||
remaining = _format_remaining(proj.headline_remaining_seconds)
|
||||
regime = ""
|
||||
if proj.regime_days:
|
||||
regime = " · sustained regime: %d days" % proj.regime_days
|
||||
return (
|
||||
"[bold]Usage-adjusted theoretical lifespan: [white]%s remaining[/white][/bold]"
|
||||
"\n if current habits continue%s" % (remaining, regime)
|
||||
)
|
||||
|
||||
def _format_confidence(self, proj: ProjectionResult) -> str:
|
||||
"""Format confidence state with contributing facts (spec §6.7)."""
|
||||
color = {
|
||||
ConfidenceState.SUPPORTED: "green",
|
||||
ConfidenceState.LIMITED: "yellow",
|
||||
ConfidenceState.UNSUPPORTED: "red",
|
||||
}[proj.confidence_state]
|
||||
facts = " · ".join(proj.contributing_facts[:3]) if proj.contributing_facts else "no facts"
|
||||
return "[bold]Projection confidence: [%s]%s[/%s][/bold]\n %s" % (
|
||||
color,
|
||||
proj.confidence_state.value,
|
||||
color,
|
||||
facts,
|
||||
)
|
||||
|
||||
def _format_scenario(self, proj: ProjectionResult) -> str:
|
||||
"""Format scenario range (spec §6.5)."""
|
||||
if not proj.scenario_range or not proj.scenario_range.rates:
|
||||
return ""
|
||||
parts = []
|
||||
for horizon in sorted(proj.scenario_range.rates.keys()):
|
||||
rate_gb_day = proj.scenario_range.rates[horizon] * 86400 / 1e9
|
||||
parts.append("%dd: %.2f GB/day" % (horizon, rate_gb_day))
|
||||
return "[bold]Scenario range[/bold] · %s" % " · ".join(parts)
|
||||
|
||||
# --- Actions ---
|
||||
|
||||
def action_pause(self) -> None:
|
||||
"""Pause monitoring — asks for confirmation (spec §7.4, LC-6)."""
|
||||
self.push_screen(ConfirmPause(), callback=self._pause_confirmed)
|
||||
|
||||
def _pause_confirmed(self, confirmed: bool) -> None:
|
||||
if not confirmed:
|
||||
return
|
||||
# Route through fenris-monitor as terminal-attached subprocess (LC-6)
|
||||
self._run_helper("disable", ["--now"])
|
||||
|
||||
def action_resume(self) -> None:
|
||||
"""Resume monitoring — no confirmation (spec §7.4, LC-6)."""
|
||||
self._run_helper("enable", ["--now"])
|
||||
|
||||
def action_collect(self) -> None:
|
||||
"""Collect now — synchronous outcome (spec §8.7, LC-8)."""
|
||||
self._run_helper("collect", blocking=True)
|
||||
|
||||
def action_disclose(self) -> None:
|
||||
"""Show disclosures (spec §6.11, CI-4)."""
|
||||
self.push_screen(DisclosuresScreen())
|
||||
|
||||
def _run_helper(
|
||||
self,
|
||||
operation: str,
|
||||
extra_args: Optional[List[str]] = None,
|
||||
blocking: bool = False,
|
||||
) -> None:
|
||||
"""Run fenris-monitor as terminal-attached subprocess (LC-6, LC-8).
|
||||
|
||||
The TUI suspends, polkit agent prompts on real terminal, control returns.
|
||||
"""
|
||||
cmd = [self.helper_path, operation]
|
||||
if extra_args:
|
||||
cmd.extend(extra_args)
|
||||
|
||||
try:
|
||||
with self.suspend():
|
||||
proc = subprocess.run(cmd, timeout=30)
|
||||
if proc.returncode != 0:
|
||||
self.notify(
|
||||
"Operation failed (exit %d)" % proc.returncode,
|
||||
severity="error",
|
||||
)
|
||||
except FileNotFoundError:
|
||||
self.notify(
|
||||
"Helper not found: %s" % self.helper_path,
|
||||
severity="error",
|
||||
)
|
||||
except subprocess.TimeoutExpired:
|
||||
self.notify("Operation timed out", severity="error")
|
||||
except Exception as e:
|
||||
self.notify("Error: %s" % e, severity="error")
|
||||
|
||||
# Refresh after action
|
||||
self._refresh()
|
||||
|
||||
|
||||
def run_tui(
|
||||
store_path: Optional[Path] = None,
|
||||
helper_path: Optional[str] = None,
|
||||
) -> None:
|
||||
"""Entry point for the Fenris TUI."""
|
||||
app = FenrisTuiApp(
|
||||
store_path=store_path,
|
||||
helper_path=helper_path,
|
||||
)
|
||||
app.run()
|
||||
@@ -0,0 +1,527 @@
|
||||
"""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 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
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 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()
|
||||
|
||||
Reference in New Issue
Block a user