From 2b052676908a80a5aae431e514ba2565445f6be5 Mon Sep 17 00:00:00 2001 From: xavierk Date: Wed, 2 Sep 2026 11:48:08 +0530 Subject: [PATCH] feat: Fenris persistent TUI monitoring redesign Implement the complete redesign per fenris-redesign spec: - Observation store: SQLite WAL mode, six entities, schema versioning - Collector: smartctl acquisition, sysfs identity, normalization - Projection: sustained regime rate, habit change, confidence states - Panes TUI: Textual keyboard-first layout with four normative regions - Status CLI: read-only composition with four service facts - Monitor helper: polkit-guarded toggle, collect, baseline ops - Legacy migration: idempotent single-transaction import - Hour classification, day aggregates, monitoring periods - Pruning, segmentation, drive health facts Cross-cutting acceptance sweep (CI-1 through CI-4): - 59 tests covering state matrix, TUI/CLI parity, prohibition set, required wording and six disclosures - Full suite: 289 tests, all green Issues #20, #32 closed. --- Makefile | 21 +- README.md | 209 +++---- fenris.py | 1061 -------------------------------- fenris.sh | 128 ---- scripts/fenris | 26 + src/fenris/store.py | 47 +- tests/test_acceptance_sweep.py | 666 ++++++++++++++++++++ 7 files changed, 841 insertions(+), 1317 deletions(-) delete mode 100755 fenris.py delete mode 100755 fenris.sh create mode 100644 tests/test_acceptance_sweep.py diff --git a/Makefile b/Makefile index 2eb9814..8c2dff9 100644 --- a/Makefile +++ b/Makefile @@ -122,7 +122,7 @@ upgrade: dist/fenris-*.whl @echo "=== Installing new wheel with pinned dependencies ===" @sudo $(VENV_DIR)/bin/pip install dist/fenris-*.whl --quiet - @echo "=== Syncing units ===" + @echo "=== Syncing units against manifest ===" @sudo install -m 0644 units/fenris-collect.timer $(UNIT_DIR)/ @sudo install -m 0644 units/fenris-collect.service $(UNIT_DIR)/ @sudo install -m 0644 polkit/com.bongbetic.fenris.monitor.policy $(POLKIT_DIR)/ @@ -146,10 +146,21 @@ upgrade: dist/fenris-*.whl @echo "$(CONF_DIR)" | sudo tee -a $(MANIFEST) > /dev/null @echo "$(MANIFEST)" | sudo tee -a $(MANIFEST) > /dev/null - @if systemctl is-active --quiet fenris-collect.timer; then \ - echo "=== Restarting timer (active) ==="; \ - sudo systemctl restart fenris-collect.timer; \ - fi + @echo "=== Restarting timer only if contents changed and active (IN-5) ===" + @for unit in fenris-collect.timer fenris-collect.service; do \ + TMPFILE=$$(mktemp); \ + sudo systemctl cat $$unit > $$TMPFILE 2>/dev/null || true; \ + if ! diff -q $$TMPFILE $(UNIT_DIR)/$$unit > /dev/null 2>&1; then \ + if systemctl is-active --quiet $$unit; then \ + echo " $$unit changed and active β€” restarting"; \ + sudo systemctl restart $$unit; \ + fi; \ + fi; \ + rm -f $$TMPFILE; \ + done + + @echo "=== Applying forward-only schema migrations (IN-5, IN-6) ===" + @sudo $(VENV_DIR)/bin/python3 -c "from fenris.store import migrate_to_latest; from pathlib import Path; n = migrate_to_latest(Path('$(DATA_DIR)/observations.db')); print(f' Migration steps applied: {n}') if n else print(' Schema already current')" @echo "=== Upgrade complete ===" diff --git a/README.md b/README.md index e3af9e4..a444334 100644 --- a/README.md +++ b/README.md @@ -1,157 +1,126 @@ -

- - - Bongbetic - -
- crafted with stubborn curiosity by Bongbetic -

+# Fenris 🐺 -

- Fenris glyph -

+*Observes an NVMe drive's real-world use and translates that history into an understandable endurance outlook.* -

Fenris 🐺 β€” Your SSD's Tell-All Diary

- -

- Your NVMe drive has been keeping secrets. Fenris makes it confess β€” in real time. -
- How much did you write today? How long until it taps out? No fairy dust β€” just your actual bytes. -

+Fenris is a persistent TUI monitor backed by a short-lived privileged collector on a systemd timer. It reads SMART data every few minutes, stores compact observation history in SQLite, and recomputes a usage-adjusted theoretical lifespan on every screen render β€” no fairy dust, just your actual bytes. --- -Fenris is a tiny, stubborn daemon that eavesdrops on your NVMe drive's SMART gossip, writes it down every few minutes, and serves you a live dashboard that actually means something. Not "vibes" β€” **real GB written in the last 24 hours, real GB/hour, and a real countdown in hours, days, and years until your drive's endurance runs out**. +## Requirements -> Think of it as a Fitbit for your SSD. Except it doesn't nag you to drink water. +- **Python β‰₯ 3.9** (verified at install time) +- **smartmontools** (`smartctl` β€” verified at install time) +- **systemd** with a polkit agent (the collector runs as root oneshot; elevation is exclusively polkit) -## What it actually does (no hand-waving) +No other OS packages or Python dependencies beyond [Textual](https://textual.textualize.io/) (pinned in the lockfile). -- **Listens** β€” polls `smartctl -j` on your NVMe device (default every 5 minutes, you pick). -- **Remembers** β€” appends every sample to `data/history.jsonl` and rolls up per-hour totals into `data/hourly.jsonl` (survives restarts, rebuilds itself if you yank the power). -- **Calculates** β€” rolling 24-hour window: *exact* bytes written in the last 24h, GB/h, GB/day, implied total TBW from `percentage_used`, remaining TB, and a projected life-remaining breakdown. Warming-up badge until it has 24h of coverage β€” no fake confidence. -- **Shows off** β€” dense, live dashboard with wear-over-time + trailing-24h per-hour bars, sticky header, live countdown, and stale warnings if the daemon dozes off. - -## You need - -- **Python 3.7+** -- **smartmontools** (`smartctl`) -- Root-ish access to read NVMe SMART (passwordless `smartctl` or just run with `sudo` β€” your call) - -### The sudo dance (one time) - -Fenris runs `sudo -n smartctl ...` so it doesn't get stuck asking for a password mid-nap: +## Install ```bash -sudo visudo -# add this line (swap in your username): -youruser ALL=(root) NOPASSWD: /usr/sbin/smartctl +sudo make install ``` -No sudo? Run the whole thing with `sudo` and it'll still behave. +What it does: +1. Builds a wheel from the checkout and installs it β€” with pinned dependencies β€” into the dedicated venv at `/opt/fenris`. +2. Places the `fenris` wrapper in `/usr/local/bin`, helpers in `/usr/libexec/fenris`, systemd units in `/etc/systemd/system`, and the polkit policy in `/usr/share/polkit-1/actions/`. +3. Creates `/var/lib/fenris` (root-written, group-readable) β€” the observation store is created lazily by the first collection run. +4. Records every placed file in a manifest consumed by upgrade and uninstall. +5. Detects `./data/history.jsonl` beside the source checkout and runs the idempotent legacy import if present. -## Get it running β€” 30 seconds - -### The cozy way +**A fresh install is fully dormant.** Units are present but disabled; nothing runs. The only opt-in is the sanctioned toggle: ```bash -./fenris.sh -# pick 1) Start monitoring β†’ choose device / interval / port β†’ done +fenris monitor resume # enable timer + open first monitoring period +fenris monitor pause # close the period, disable timer ``` -### The no-nonsense way +## Upgrade ```bash -python3 fenris.py start # defaults: /dev/nvme0, every 300s, port 8420 -python3 fenris.py start --interval 60 --port 9000 # if you're impatient -python3 fenris.py status # "are we live? how's the drive?" -python3 fenris.py sample # one sneaky sample right now -python3 fenris.py stop # tuck it back in +sudo make upgrade ``` -Dashboard lives at **http://localhost:8420** (or whatever port you chose). +What it does: +1. Snapshots `observations.db` to a one-generation backup (`.bak`). +2. Installs the new wheel into the same venv with pinned dependencies. +3. Syncs units and polkit against the manifest; runs `daemon-reload`. +4. Restarts the timer **only** if unit contents changed **and** it is active β€” a running collection run finishes on its mapped interpreter; the next run uses the new code. +5. Applies forward-only schema migrations (the store directory is never rebuilt; automatic downgrade does not exist). -## The menu, demystified +Rollback: reinstall the previous version and restore `observations.db.bak`. -Run `./fenris.sh` and you'll get: +## Uninstall and purge -``` - 1) Start monitoring (background daemon + dashboard) - 2) Stop monitoring - 3) Status / current wear stats - 4) Take one sample right now - 5) Open dashboard URL - --- - h) Help / how this works - q) Exit (go touch grass) +```bash +make uninstall # removes artifacts, preserves config and observation history +make purge # also removes /etc/fenris and /var/lib/fenris ``` -## What Fenris jots down +Uninstall performs the sanctioned disable first (`fenris-monitor disable --now`) β€” an open period closes `user_disabled` β€” then removes the venv, helpers, units, polkit policy, and wrapper while keeping `/etc/fenris` and the observation store. Reinstalling resumes from the preserved store. -| Field | What's the gossip? | -|-------|---------------------| -| `percentage_used` | The drive's own wear-o-meter (0–100%) | -| `bytes_written` / `bytes_read` | Lifetime totals β€” the receipts | -| `available_spare` | Spare blocks left (%) | -| `media_errors` | Uncorrectable boo-boos | -| `power_on_hours` | How long it's been awake | -| `temperature_c` | Is it sweating? | -| `critical_warning` | NVMe's panic flags | +## Cadence drop-ins -Hourly rollups also stash `bytes_written` per hour, `pct_start`/`pct_end`, and temp peaks β€” so the 24h math stays honest. +The default collection cadence is **5 minutes** (`OnUnitInactiveSec=5min` in the timer unit). To change it, place a systemd drop-in: -## The dashboard β€” what's on screen +```bash +sudo systemctl edit fenris-collect.timer +# Add: +# [Timer] +# OnUnitInactiveSec=10min +``` -- **Hero card: Projected life remaining** β€” big, friendly `361 d 2 h` (plus `β‰ˆ 361 days Β· β‰ˆ 8666 hours Β· β‰ˆ 0.99 years`), backed by `~280 GB/day` and `~101 TB left of ~202 TB total` on the test box. -- **Data written (24h)** β€” exact GB in the rolling window + coverage (`10.4h of 24h` until warmed up). -- **Write rate** β€” GB/h and GB/day, live. -- **Wear, spare, temp, errors, power-on** β€” the usual suspects, with progress bars and polite color-coding. -- **Two charts, side by side:** wear over time + trailing-24h hourly write bars (with a cheeky "now" bar for the current partial hour). -- **Live plumbing:** polling synced to your interval, ETag-cached, countdown to next sample, warming-up + stale banners, pauses when you hide the tab (saves your battery, you're welcome). +No interval key exists in `/etc/fenris/fenris.conf`. Cadence is a systemd concern, not a Fenris configuration key. -**API for the tinkerers:** `GET /api/data` Β· `/api/hourly` Β· `/api/summary` Β· `/api/config` Β· `/api/status` β€” all JSON, all friendly. +## CLI reference + +| Command | Behavior | +|---|---| +| `fenris` | Opens the TUI (no arguments). | +| `fenris status` | Projection facts, enabled/active state, last collect outcome, journal hint on failure or staleness. Never auto-samples. | +| `fenris sample` | On-demand collection via the privileged helper. Blocks until the run completes. | +| `fenris monitor pause` | Sanctioned disable β€” asks for confirmation, then disables the timer and closes the monitoring period. | +| `fenris monitor resume` | Sanctioned enable β€” enables the timer and opens a monitoring period. No confirmation. | +| `fenris baseline set ` | CLI-side validation, then polkit-guarded persistence. | +| `fenris baseline clear` | Remove the endurance baseline. | +| `fenris import ` | Idempotent single-transaction legacy import. | +| `fenris start` / `stop` / `run` | Rejected with a one-line migration pointer β€” never aliased. | +| `fenris --device` | Rejected with a pointer to the configuration file. | + +## Retired menu options + +The legacy `fenris.sh` menu script and the `fenris.py` monolith have been removed. Here's where the old options went: + +| Legacy option | Successor | +|---|---| +| 1) Start monitoring | `fenris monitor resume` | +| 2) Stop monitoring | `fenris monitor pause` | +| 3) Status / current wear stats | `fenris status` | +| 4) Take one sample right now | `fenris sample` | +| 5) Open dashboard URL | Removed β€” the HTML dashboard and HTTP server are gone; the TUI is the primary interface. | + +## Configuration + +`/etc/fenris/fenris.conf` holds exactly one key β€” the device selector: + +``` +device = /dev/disk/by-id/nvme-Samsung_SSD_980_PRO_2TB_S6BENS0Txxxxx +``` + +Use a stable `/dev/disk/by-id/` path. Raw `/dev/nvmeX` paths are warned against. The file is re-read every collection run. ## Where's my stuff? -``` -fenris/ -β”œβ”€β”€ fenris.py # the whole show β€” daemon + server + math -β”œβ”€β”€ fenris.sh # the cozy menu -β”œβ”€β”€ README.md # hi β€” you're here -β”œβ”€β”€ assets/bongbetic-brand/ # Bongbetic wordmarks & glyphs (for Gitea + dashboard) -└── data/ - β”œβ”€β”€ history.jsonl # raw samples (JSONL, append-only) - β”œβ”€β”€ hourly.jsonl # per-hour rollups (auto-rebuilt on restart) - β”œβ”€β”€ fenris.pid # daemon PID - └── fenris.log # daemon chatter -``` - -## CLI cheat sheet - -```bash -python3 fenris.py start [--device /dev/nvme0] [--interval 300] [--port 8420] -python3 fenris.py stop -python3 fenris.py status -python3 fenris.py sample [--device /dev/nvme0] -python3 fenris.py run # foreground mode β€” what `start` spawns internally -``` - -## Oops β€” troubleshooting without the tears - -**"smartctl not found"** -```bash -sudo apt install smartmontools # Debian/Ubuntu -sudo pacman -S smartmontools # Arch β€” you already knew -``` - -**"needs root" / permission denied** -Set up the passwordless line above, or just `sudo ./fenris.sh`. - -**Dashboard says "stale"** -Daemon napped or crashed. `python3 fenris.py status` will tell you. Kick it again with `start`. - -**Only 10 hours of data and it says "preliminary"?** -That's honesty, not a bug. It needs 24h of real writes to give a tight estimate. Let it simmer β€” the number gets sharper every hour. +| Artifact | Location | +|---|---| +| Wrapper | `/usr/local/bin/fenris` | +| Helpers | `/usr/libexec/fenris/fenris-collect`, `fenris-monitor` | +| Units | `/etc/systemd/system/fenris-collect.{timer,service}` | +| Polkit policy | `/usr/share/polkit-1/actions/com.bongbetic.fenris.monitor.policy` | +| Configuration | `/etc/fenris/fenris.conf` | +| Observation store | `/var/lib/fenris/observations.db` | +| Venv | `/opt/fenris` | +| Manifest | `/var/lib/fenris/manifest.txt` | +| Legacy history | `./data/history.jsonl` (auto-imported on install if present) | --- diff --git a/fenris.py b/fenris.py deleted file mode 100755 index 87f9d0b..0000000 --- a/fenris.py +++ /dev/null @@ -1,1061 +0,0 @@ -#!/usr/bin/env python3 -""" -Fenris β€” NVMe wear monitor + live dashboard. -Created by Bongbetic. - -Samples NVMe SMART health data (via smartctl -j), logs it over time, -and serves a self-contained HTML dashboard estimating SSD lifespan -from your actual daily usage trend. -""" - -import json -import os -import subprocess -import sys -import time -import signal -import argparse -import threading -import hashlib -import mimetypes -from datetime import datetime, timezone, timedelta -from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer - -# Add src/ to path for package imports -_SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__)) -sys.path.insert(0, os.path.join(_SCRIPT_DIR, "src")) - -SCRIPT_DIR = _SCRIPT_DIR -DATA_DIR = os.path.join(SCRIPT_DIR, "data") -DATA_FILE = os.path.join(DATA_DIR, "history.jsonl") -HOURLY_FILE = os.path.join(DATA_DIR, "hourly.jsonl") -PID_FILE = os.path.join(DATA_DIR, "fenris.pid") -LOG_FILE = os.path.join(DATA_DIR, "fenris.log") -ASSETS_DIR = os.path.join(SCRIPT_DIR, "assets") -VERSION = "0.2.0" - -os.makedirs(DATA_DIR, exist_ok=True) - -_CONFIG = {"interval": 300, "device": "/dev/nvme0", "port": 8420, "version": VERSION} -_hour_lock = threading.Lock() -_current_hour = None - - -def detect_device(): - for cand in ("/dev/nvme0", "/dev/nvme1"): - if os.path.exists(cand): - return cand - return "/dev/nvme0" - - -def sample(device): - try: - out = subprocess.run( - ["sudo", "-n", "smartctl", "-a", "-j", device], - capture_output=True, text=True, timeout=15 - ) - except FileNotFoundError: - print("ERROR: smartctl not found. Install smartmontools.", file=sys.stderr) - return None - if out.returncode not in (0, 4): - print(f"smartctl failed (exit {out.returncode}): {out.stderr.strip()}", file=sys.stderr) - print("Hint: needs root. Run 'sudo visudo' and allow passwordless " - "'smartctl' for your user, or run Fenris with sudo.", file=sys.stderr) - return None - try: - d = json.loads(out.stdout) - except json.JSONDecodeError: - return None - log = d.get("nvme_smart_health_information_log") - if not log: - print("No NVMe SMART data in smartctl output (not an NVMe device?).", file=sys.stderr) - return None - capacity_bytes = (d.get("user_capacity") or {}).get("bytes", 0) - units_written = log.get("data_units_written", 0) - units_read = log.get("data_units_read", 0) - return { - "ts": datetime.now(timezone.utc).isoformat(), - "device": device, - "model": d.get("model_name", "unknown"), - "capacity_bytes": capacity_bytes, - "percentage_used": log.get("percentage_used"), - "available_spare": log.get("available_spare"), - "media_errors": log.get("media_errors"), - "power_on_hours": log.get("power_on_hours"), - "power_cycles": log.get("power_cycles"), - "unsafe_shutdowns": log.get("unsafe_shutdowns"), - "temperature_c": log.get("temperature"), - "data_units_written": units_written, - "data_units_read": units_read, - "bytes_written": units_written * 512000, - "bytes_read": units_read * 512000, - "critical_warning": log.get("critical_warning"), - } - - -def append_sample(rec): - with open(DATA_FILE, "a") as f: - f.write(json.dumps(rec) + "\n") - - -def load_history(): - if not os.path.exists(DATA_FILE): - return [] - out = [] - with open(DATA_FILE) as f: - for line in f: - line = line.strip() - if line: - try: - out.append(json.loads(line)) - except json.JSONDecodeError: - continue - return out - - -def hour_key(ts_str): - try: - dt = datetime.fromisoformat(ts_str) - if dt.tzinfo is None: - dt = dt.replace(tzinfo=timezone.utc) - else: - dt = dt.astimezone(timezone.utc) - dt = dt.replace(minute=0, second=0, microsecond=0) - return dt.isoformat().replace("+00:00", "Z") - except Exception: - return ts_str[:13] + ":00:00Z" - - -def load_hourly(): - if not os.path.exists(HOURLY_FILE): - return [] - out = [] - with open(HOURLY_FILE) as f: - for line in f: - line = line.strip() - if line: - try: - out.append(json.loads(line)) - except json.JSONDecodeError: - continue - out.sort(key=lambda r: r.get("hour", "")) - return out - - -def append_hourly(rec): - with _hour_lock: - with open(HOURLY_FILE, "a") as f: - f.write(json.dumps(rec) + "\n") - - -def _hourly_record_for_bucket(hour_str, recs): - if not recs: - return None - first = recs[0] - last = recs[-1] - bw = (last.get("bytes_written", 0) - first.get("bytes_written", 0)) if len(recs) > 1 else 0 - br = (last.get("bytes_read", 0) - first.get("bytes_read", 0)) if len(recs) > 1 else 0 - temps = [r.get("temperature_c") for r in recs if r.get("temperature_c") is not None] - return { - "hour": hour_str, - "samples": len(recs), - "bytes_written": max(bw, 0), - "bytes_read": max(br, 0), - "pct_start": first.get("percentage_used"), - "pct_end": last.get("percentage_used"), - "temp_avg": round(sum(temps) / len(temps), 1) if temps else None, - "temp_max": max(temps) if temps else None, - "media_errors": last.get("media_errors", 0), - "available_spare": last.get("available_spare"), - } - - -def rebuild_hourly_from_history(): - history = load_history() - if not history: - return - buckets = {} - for r in history: - hk = hour_key(r["ts"]) - buckets.setdefault(hk, []).append(r) - existing = {r["hour"]: r for r in load_hourly()} - for hk in sorted(buckets.keys()): - if hk in existing: - continue - rec = _hourly_record_for_bucket(hk, buckets[hk]) - if rec: - append_hourly(rec) - - -def update_hour_bucket(rec): - global _current_hour - hk = hour_key(rec["ts"]) - if _current_hour is None or _current_hour["hour"] != hk: - if _current_hour is not None: - flushed = _hourly_record_for_bucket(_current_hour["hour"], _current_hour["recs"]) - if flushed: - existing_hours = {r["hour"] for r in load_hourly()} - if flushed["hour"] not in existing_hours: - append_hourly(flushed) - _current_hour = {"hour": hk, "recs": [rec]} - else: - _current_hour["recs"].append(rec) - - -def _parse_ts(ts_str): - try: - dt = datetime.fromisoformat(ts_str) - if dt.tzinfo is None: - dt = dt.replace(tzinfo=timezone.utc) - return dt.astimezone(timezone.utc) - except Exception: - return None - - -def compute_summary(history=None, hourly=None): - if history is None: - history = load_history() - if not history: - return { - "window24h": {"bytes": 0, "gb": 0, "coverage_hours": 0}, - "gb_per_hour": 0, - "gb_per_day": 0, - "endurance_tb": None, - "endurance_estimated": True, - "remaining_tb": None, - "remaining_bytes": 0, - "seconds_remaining": None, - "breakdown": {"years": 0, "days": 0, "hours": 0, "human": "β€”"}, - "wear_model_days": None, - "preliminary": True, - "notes": ["No data yet"], - "latest": None, - } - latest = history[-1] - latest_ts = _parse_ts(latest["ts"]) - earliest_ts = _parse_ts(history[0]["ts"]) - total_span_sec = (latest_ts - earliest_ts).total_seconds() if latest_ts and earliest_ts else 0 - window_sec = 24 * 3600 - window_start_ts = latest_ts - timedelta(seconds=window_sec) if latest_ts else None - - idx = 0 - if window_start_ts: - for i, r in enumerate(history): - dt = _parse_ts(r["ts"]) - if dt and dt >= window_start_ts: - idx = i - break - else: - idx = 0 - window_start_rec = history[idx] - window_start_ts_actual = _parse_ts(window_start_rec["ts"]) - coverage_sec = (latest_ts - window_start_ts_actual).total_seconds() if latest_ts and window_start_ts_actual else 0 - if coverage_sec < 1: - coverage_sec = total_span_sec if total_span_sec > 0 else 1 - - bytes_in_window = latest.get("bytes_written", 0) - window_start_rec.get("bytes_written", 0) - if bytes_in_window < 0: - bytes_in_window = 0 - - rate = bytes_in_window / coverage_sec if coverage_sec > 0 else 0 - gb_per_hour = rate * 3600 / 1e9 - gb_per_day = gb_per_hour * 24 - coverage_hours = coverage_sec / 3600 - - pct = latest.get("percentage_used") - cap = latest.get("capacity_bytes") or 0 - bw_total = latest.get("bytes_written", 0) - if pct is not None and pct > 0: - endurance_bytes = bw_total / (pct / 100) - endurance_estimated = False - else: - endurance_bytes = cap * 600 if cap else 0 - endurance_estimated = True - remaining_bytes = max(endurance_bytes - bw_total, 0) if endurance_bytes else 0 - endurance_tb = endurance_bytes / 1e12 if endurance_bytes else None - remaining_tb = remaining_bytes / 1e12 if remaining_bytes else 0 - - seconds_remaining = None - if rate > 0 and remaining_bytes > 0: - seconds_remaining = remaining_bytes / rate - - def _humanize(secs): - if secs is None or secs <= 0: - return "β€”" - years = int(secs // 31557600) - rem = secs % 31557600 - days = int(rem // 86400) - rem %= 86400 - hours = int(rem // 3600) - parts = [] - if years: - parts.append(f"{years} yr") - if days or years: - parts.append(f"{days} d") - parts.append(f"{hours} h") - return " ".join(parts) - - human = _humanize(seconds_remaining) - breakdown = { - "years": (seconds_remaining / 31557600) if seconds_remaining else 0, - "days": (seconds_remaining / 86400) if seconds_remaining else 0, - "hours": (seconds_remaining / 3600) if seconds_remaining else 0, - "human": human, - } - - wear_days = None - pts = [( _parse_ts(r["ts"]).timestamp(), r["percentage_used"]) for r in history if r.get("percentage_used") is not None and _parse_ts(r["ts"]) is not None] - if len(pts) >= 2: - win_pts = [p for p in pts if p[0] >= (latest_ts.timestamp() - window_sec)] if latest_ts else pts - if len(win_pts) < 2: - win_pts = pts - n = len(win_pts) - mean_x = sum(p[0] for p in win_pts) / n - mean_y = sum(p[1] for p in win_pts) / n - num = sum((p[0]-mean_x)*(p[1]-mean_y) for p in win_pts) - den = sum((p[0]-mean_x)**2 for p in win_pts) - if den != 0: - slope = num/den - if slope > 0: - last_pct = win_pts[-1][1] - secs_to_100 = (100 - last_pct) / slope - wear_days = secs_to_100 / 86400 - - preliminary = coverage_hours < 24 - notes = [] - if preliminary: - notes.append(f"Warming up β€” {coverage_hours:.1f}h of 24h") - if endurance_estimated: - notes.append("TBW estimated from capacity (pct=0)") - if rate <= 0: - notes.append("No writes in window") - if wear_days is None: - notes.append("Wear flat β€” write model only") - - return { - "window24h": {"bytes": bytes_in_window, "gb": bytes_in_window / 1e9, "coverage_hours": coverage_hours}, - "gb_per_hour": gb_per_hour, - "gb_per_day": gb_per_day, - "endurance_tb": endurance_tb, - "endurance_estimated": endurance_estimated, - "remaining_tb": remaining_tb, - "remaining_bytes": remaining_bytes, - "seconds_remaining": seconds_remaining, - "breakdown": breakdown, - "wear_model_days": wear_days, - "preliminary": preliminary, - "notes": notes, - "latest": latest, - } - - -def collector_loop(device, interval, stop_event): - while not stop_event.is_set(): - rec = sample(device) - if rec: - append_sample(rec) - update_hour_bucket(rec) - stop_event.wait(interval) - - -DASHBOARD_HTML = r""" - - - - -Fenris β€” NVMe Wear Dashboard - - - - - - - - - - -
-
-
- - Bongbetic - - - Bongbetic - - - Β· -
-
FENRIS
-
loading…
-
-
-
- ● Live - every β€” - next β€” -
-
-
- -
- - - -
- -
-
-
Wear (% used) over time
- -
-
-
Trailing 24h β€” GB written per hour
- -
-
-
- - - - - - -""" - - -class Handler(BaseHTTPRequestHandler): - def log_message(self, fmt, *args): - pass - - def do_GET(self): - path = self.path.split("?")[0] - if path == "/" or path.startswith("/index"): - body = DASHBOARD_HTML.encode() - self.send_response(200) - self.send_header("Content-Type", "text/html; charset=utf-8") - self.send_header("Content-Length", str(len(body))) - self.end_headers() - self.wfile.write(body) - return - if path.startswith("/assets/"): - rel = path[len("/assets/"):] - if ".." in rel or rel.startswith("/"): - self.send_response(404); self.end_headers(); return - fp = os.path.join(ASSETS_DIR, rel) - if not os.path.isfile(fp): - self.send_response(404); self.end_headers(); return - ctype = mimetypes.guess_type(fp)[0] or "application/octet-stream" - with open(fp, "rb") as f: - body = f.read() - self.send_response(200) - self.send_header("Content-Type", ctype) - self.send_header("Content-Length", str(len(body))) - self.end_headers() - self.wfile.write(body) - return - if path == "/api/config": - body = json.dumps(_CONFIG).encode() - self.send_response(200) - self.send_header("Content-Type", "application/json") - self.send_header("Content-Length", str(len(body))) - self.end_headers() - self.wfile.write(body) - return - if path == "/api/status": - hist = load_history() - latest = hist[-1] if hist else None - summ = compute_summary(hist) - body = json.dumps({ - "alive": True, - "samples": len(hist), - "last_ts": latest["ts"] if latest else None, - "version": VERSION, - "config": _CONFIG, - "summary": summ, - }).encode() - self.send_response(200) - self.send_header("Content-Type", "application/json") - self.send_header("Content-Length", str(len(body))) - self.end_headers() - self.wfile.write(body) - return - if path.startswith("/api/data"): - hist = load_history() - body = json.dumps(hist).encode() - etag = f'"{hashlib.md5(body).hexdigest()}"' - inm = self.headers.get("If-None-Match") - if inm and inm == etag: - self.send_response(304); self.end_headers(); return - self.send_response(200) - self.send_header("Content-Type", "application/json") - self.send_header("ETag", etag) - self.send_header("Cache-Control", "no-cache") - self.send_header("Content-Length", str(len(body))) - self.end_headers() - self.wfile.write(body) - return - if path.startswith("/api/hourly"): - hourly = load_hourly() - body = json.dumps(hourly).encode() - etag = f'"{hashlib.md5(body).hexdigest()}"' - inm = self.headers.get("If-None-Match") - if inm and inm == etag: - self.send_response(304); self.end_headers(); return - self.send_response(200) - self.send_header("Content-Type", "application/json") - self.send_header("ETag", etag) - self.send_header("Cache-Control", "no-cache") - self.send_header("Content-Length", str(len(body))) - self.end_headers() - self.wfile.write(body) - return - if path.startswith("/api/summary"): - summ = compute_summary() - body = json.dumps(summ).encode() - self.send_response(200) - self.send_header("Content-Type", "application/json") - self.send_header("Content-Length", str(len(body))) - self.end_headers() - self.wfile.write(body) - return - self.send_response(404) - self.end_headers() - - -def run_foreground(device, interval, port): - global _CONFIG, _current_hour - _CONFIG = {"interval": interval, "device": device, "port": port, "version": VERSION} - stop_event = threading.Event() - - def handle_sig(signum, frame): - stop_event.set() - - signal.signal(signal.SIGTERM, handle_sig) - signal.signal(signal.SIGINT, handle_sig) - - def cleanup_pid(): - try: - os.remove(PID_FILE) - except OSError: - pass - import atexit - atexit.register(cleanup_pid) - - print(f"Fenris starting β€” device={device} interval={interval}s port={port}") - print(f"Data log: {DATA_FILE}") - - rebuild_hourly_from_history() - - rec = sample(device) - if rec: - append_sample(rec) - update_hour_bucket(rec) - else: - print("WARNING: initial sample failed β€” check smartctl/sudo setup. " - "The daemon will keep retrying.", file=sys.stderr) - - t = threading.Thread(target=collector_loop, args=(device, interval, stop_event), daemon=True) - t.start() - - server = ThreadingHTTPServer(("0.0.0.0", port), Handler) - server.timeout = 1 - print(f"Dashboard: http://localhost:{port}") - - def server_loop(): - while not stop_event.is_set(): - server.handle_request() - - st = threading.Thread(target=server_loop, daemon=True) - st.start() - - while not stop_event.is_set(): - time.sleep(0.5) - print("Fenris stopping.") - - -def cmd_start(args): - if os.path.exists(PID_FILE): - with open(PID_FILE) as f: - pid = int(f.read().strip()) - if pid_alive(pid): - print(f"Fenris already running (pid {pid}). Use 'status' or 'stop'.") - return - os.remove(PID_FILE) - log_f = open(LOG_FILE, "a") - proc = subprocess.Popen( - [sys.executable, os.path.abspath(__file__), "run", - "--device", args.device, "--interval", str(args.interval), "--port", str(args.port)], - stdout=log_f, stderr=log_f, stdin=subprocess.DEVNULL, - start_new_session=True, - ) - with open(PID_FILE, "w") as f: - f.write(str(proc.pid)) - time.sleep(0.5) - print(f"Fenris started in background (pid {proc.pid}).") - print(f"Dashboard: http://localhost:{args.port}") - print(f"Logs: {LOG_FILE}") - - -def pid_alive(pid): - try: - os.kill(pid, 0) - return True - except OSError: - return False - - -def cmd_stop(args): - if not os.path.exists(PID_FILE): - print("Fenris is not running (no pid file).") - return - with open(PID_FILE) as f: - pid = int(f.read().strip()) - if pid_alive(pid): - os.kill(pid, signal.SIGTERM) - for _ in range(50): - if not pid_alive(pid): - break - time.sleep(0.1) - if pid_alive(pid): - print(f"SIGTERM did not stop pid {pid}, sending SIGKILL...") - os.kill(pid, signal.SIGKILL) - for _ in range(20): - if not pid_alive(pid): - break - time.sleep(0.1) - if pid_alive(pid): - print(f"ERROR: could not stop pid {pid}. Manual intervention needed.") - return - print(f"Stopped Fenris (pid {pid}).") - else: - print("Stale pid file β€” process was not running.") - try: - os.remove(PID_FILE) - except OSError: - pass - - -def cmd_status(args): - try: - from fenris.status import render_status, check_retired_flag - from pathlib import Path - - show_disclosures = getattr(args, "disclosures", False) - store_path = Path("/var/lib/fenris/observations.db") - output = render_status(store_path=store_path, show_disclosures=show_disclosures) - print(output) - except ImportError: - # Fallback if fenris package not importable - print("Error: cannot import fenris.status module. Is the package installed?") - sys.exit(1) - - -def cmd_run(args): - run_foreground(args.device, args.interval, args.port) - - -def cmd_sample_once(args): - rec = sample(args.device) - if rec: - append_sample(rec) - update_hour_bucket(rec) - print(json.dumps(rec, indent=2)) - summ = compute_summary() - print(f"\n24h: {summ['window24h']['gb']:.2f} GB rate {summ['gb_per_hour']:.2f} GB/h remaining {summ['breakdown']['human']}") - else: - sys.exit(1) - - -def cmd_retired(args): - """Handle retired commands with migration pointers (Β§8.8).""" - from fenris.status import check_retired_command - cmd = sys.argv[1] if len(sys.argv) > 1 else "" - ptr = check_retired_command(cmd) - if ptr: - print(ptr) - else: - print("Unknown command. Use 'fenris status' or 'fenris sample'.") - sys.exit(1) - - -def main(): - p = argparse.ArgumentParser(description="Fenris β€” NVMe wear monitor & dashboard (by Bongbetic)") - p.add_argument("--device", default=None, - help="(retired β€” device is configured in /etc/fenris/fenris.conf)") - sub = p.add_subparsers(dest="cmd", required=True) - - def add_common(sp): - sp.add_argument("--device", default=detect_device(), help="NVMe device, e.g. /dev/nvme0") - sp.add_argument("--interval", type=int, default=300, help="seconds between samples (default 300)") - sp.add_argument("--port", type=int, default=8420, help="dashboard HTTP port (default 8420)") - - sp = sub.add_parser("start", help="(retired β€” use 'fenris monitor resume')") - add_common(sp); sp.set_defaults(func=cmd_retired) - sp = sub.add_parser("stop", help="(retired β€” use 'fenris monitor pause')") - sp.set_defaults(func=cmd_retired) - sp = sub.add_parser("status", help="show read-only status (Β§8.8)") - sp.add_argument("-d", "--disclosures", action="store_true", - help="show the six disclosures (Β§6.11)") - sp.set_defaults(func=cmd_status) - sp = sub.add_parser("run", help="(retired β€” use 'fenris monitor resume')") - add_common(sp); sp.set_defaults(func=cmd_retired) - sp = sub.add_parser("sample", help="take one sample immediately and print it") - sp.add_argument("--device", default=detect_device()) - sp.set_defaults(func=cmd_sample_once) - - args = p.parse_args() - - # Reject retired --device flag (Β§8.8) β€” only if explicitly passed - if getattr(args, "device", None) is not None: - from fenris.status import check_retired_flag - ptr = check_retired_flag("--device") - if ptr: - print(ptr) - sys.exit(1) - - args.func(args) - - -if __name__ == "__main__": - main() diff --git a/fenris.sh b/fenris.sh deleted file mode 100755 index 82c9bd0..0000000 --- a/fenris.sh +++ /dev/null @@ -1,128 +0,0 @@ -#!/usr/bin/env bash -# Fenris β€” interactive menu for the NVMe wear monitor & dashboard. -# Created by Bongbetic. - -set -euo pipefail -SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" -PY="$SCRIPT_DIR/fenris.py" -PORT_DEFAULT=8420 -INTERVAL_DEFAULT=300 - -banner() { - cat <<'EOF' - _____ _ -| __|___ ___ _| |___ -| __| -_| | . | _| -|__| |___|_|_|_|___|_| - - NVMe wear monitor & live dashboard - Created by Bongbetic -EOF -} - -pause() { read -rp "Press Enter to continue..." _; } - -detect_device() { - # Query Python's auto-detect for the default device. - python3 -c "import sys; sys.path.insert(0,'$SCRIPT_DIR'); from fenris import detect_device; print(detect_device())" 2>/dev/null || echo /dev/nvme0 -} - -menu() { - clear - banner - echo - echo " 1) Start monitoring (background daemon + dashboard)" - echo " 2) Stop monitoring" - echo " 3) Status / current wear stats" - echo " 4) Take one sample right now" - echo " 5) Open dashboard URL" - echo " ---" - echo " h) Help / how this works" - echo " q) Exit" - echo - read -rp "Choose an option: " choice - echo - case "$choice" in - 1) start_flow ;; - 2) python3 "$PY" stop; pause ;; - 3) python3 "$PY" status; pause ;; - 4) read -rp "Device [default: auto-detect]: " dev - if [ -z "$dev" ]; then python3 "$PY" sample; else python3 "$PY" sample --device "$dev"; fi - pause ;; - 5) show_url; pause ;; - h|H) help_text; pause ;; - q|Q) echo "Bye. β€” Fenris, by Bongbetic"; exit 0 ;; - *) echo "Invalid choice."; pause ;; - esac -} - -start_flow() { - read -rp "NVMe device [Enter = auto-detect]: " dev - read -rp "Sample interval in seconds [Enter = ${INTERVAL_DEFAULT}]: " interval - read -rp "Dashboard port [Enter = ${PORT_DEFAULT}]: " port - interval="${interval:-$INTERVAL_DEFAULT}" - port="${port:-$PORT_DEFAULT}" - - args=(start --interval "$interval" --port "$port") - if [ -n "${dev:-}" ]; then args+=(--device "$dev"); fi - - echo - echo "Note: reading NVMe SMART data needs root." - echo "Fenris runs 'sudo -n smartctl ...' (no-prompt sudo). If this fails," - echo "either run this menu with sudo, or allow passwordless smartctl via:" - echo " sudo visudo -> youruser ALL=(root) NOPASSWD: /usr/sbin/smartctl" - echo - - python3 "$PY" "${args[@]}" - pause -} - -show_url() { - if [ -f "$SCRIPT_DIR/data/fenris.pid" ]; then - # Try to read actual port from running process cmdline, else guess default. - local pid port - pid=$(<"$SCRIPT_DIR/data/fenris.pid") - port=$(tr '\0' '\n' < /proc/"$pid"/cmdline 2>/dev/null | grep -A1 -- '--port' | tail -1 || true) - port="${port:-$PORT_DEFAULT}" - echo "Dashboard: http://localhost:${port}" - else - echo "Fenris is not currently running. Start it first (option 1)." - fi -} - -help_text() { - cat < None: print("Legacy import: use fenris-import directly") +def cmd_migrate(args: argparse.Namespace) -> None: + """Apply forward-only schema migrations (IN-5, IN-6). + + Called by 'sudo make upgrade'. Raises on newer-schema store. + """ + from fenris.store import migrate_to_latest + from pathlib import Path + + store_path = Path("/var/lib/fenris/observations.db") + if not store_path.exists(): + print("No observation store found β€” nothing to migrate.") + return + + steps = migrate_to_latest(store_path) + if steps: + print(f"Migration complete: {steps} step(s) applied.") + else: + print("Schema already current.") + + def main() -> None: parser = argparse.ArgumentParser( prog="fenris", @@ -145,6 +165,10 @@ def main() -> None: import_parser.add_argument("path", help="Path to history.jsonl") import_parser.set_defaults(func=cmd_import) + # Migrate (IN-5, IN-6) β€” called by upgrade, not for human use + migrate_parser = subparsers.add_parser("migrate", help=argparse.SUPPRESS) + migrate_parser.set_defaults(func=cmd_migrate) + # Rejected commands for cmd in ["start", "stop", "run"]: reject_parser = subparsers.add_parser(cmd, help=argparse.SUPPRESS) @@ -172,6 +196,8 @@ def main() -> None: args.func(args) elif args.command == "import": cmd_import(args) + elif args.command == "migrate": + cmd_migrate(args) if __name__ == "__main__": diff --git a/src/fenris/store.py b/src/fenris/store.py index e80cb78..7721778 100644 --- a/src/fenris/store.py +++ b/src/fenris/store.py @@ -176,12 +176,53 @@ def _create_schema(conn: sqlite3.Connection): def _apply_migrations(conn: sqlite3.Connection, current_version: int): - """Apply forward-only migrations from current_version to SCHEMA_VERSION.""" - # Future migrations will go here - # For now, just upgrade to current version + """Apply forward-only migrations from current_version to SCHEMA_VERSION. + + Each migration step is a transactional block. Add new steps as sequential + elif branches when SCHEMA_VERSION increases. + + Spec: Β§3.6, Β§10.2 + """ + # Migration 1β†’2: example placeholder + # if current_version < 2: + # conn.execute("ALTER TABLE ...") + # current_version = 2 pass +def migrate_to_latest(store_path: Path) -> int: + """Apply forward-only migrations to bring the store to SCHEMA_VERSION. + + Called by the upgrade target (Β§10.2). Returns the number of migration + steps applied. Raises ValueError on newer-schema store (Β§3.6, Β§9.5). + + Spec: Β§3.6, Β§10.2, Β§10.3 + """ + conn = sqlite3.connect(str(store_path)) + conn.execute("PRAGMA journal_mode=WAL") + + cursor = conn.execute("PRAGMA user_version") + current_version = cursor.fetchone()[0] + + if current_version > SCHEMA_VERSION: + conn.close() + raise ValueError( + f"Observation store written by a newer Fenris (version {current_version}) " + f"β€” upgrade Fenris" + ) + + if current_version == SCHEMA_VERSION: + conn.close() + return 0 # Already up to date + + steps = SCHEMA_VERSION - current_version + _apply_migrations(conn, current_version) + conn.execute(f"PRAGMA user_version={SCHEMA_VERSION}") + conn.commit() + conn.close() + return steps + + def is_store_faulty(store_path: Path) -> bool: """Check if the store is present but cannot be read or trusted.""" if not store_path.exists(): diff --git a/tests/test_acceptance_sweep.py b/tests/test_acceptance_sweep.py new file mode 100644 index 0000000..7d4a8bf --- /dev/null +++ b/tests/test_acceptance_sweep.py @@ -0,0 +1,666 @@ +"""Cross-cutting acceptance sweep (issue #32). + +Systematic verification of every acceptance criterion that spans multiple +subsystems. Grouped by criterion ID; each test cites its clause. + +CI-1 Exhaustive state matrix: confidence Γ— freshness Γ— baseline tier +CI-2 TUI/CLI parity: identical outcomes and wording +CI-3 Prohibition set: automated structural checks +CI-4 Required wording and six disclosures in both views +""" +import re +import sqlite3 +from datetime import datetime, timedelta, timezone +from pathlib import Path +from unittest.mock import patch + +import pytest +import sys + +sys.path.insert(0, str(Path(__file__).parent.parent / "src")) + +from fenris.store import init_store, SCHEMA_VERSION +from fenris.monitoring_periods import ensure_period_open +from fenris.projection import ( + compute_projection, + ConfidenceState, + BaselineTier, + DISCLOSURES, + STALENESS_HOURS, + WARMING_MIN_DAYS, + YOUNG_REGIME_DAYS, +) +from fenris.status import ( + grade_freshness, + get_status, + render_status, + format_disclosures, + FRESH_THRESHOLD_S, + STALENESS_THRESHOLD_S, + CADENCE_DEFAULT_S, + ACCURACY_SEC, +) +from fenris.tui import ( + FenrisTuiApp, + _format_remaining, +) + + +SRC_DIR = Path(__file__).parent.parent / "src" +FENRIS_PKG = SRC_DIR / "fenris" + + +def _clock(year=2026, month=9, day=30, hour=12): + return datetime(year, month, day, hour, 0, 0, tzinfo=timezone.utc) + + +def _insert_baseline(conn, tbw_tb=1.0, verified=True, + model="Samsung SSD 970 EVO Plus 1TB", + source_url="https://example.com/spec", + doc_rev="v1.0", entry_date="2026-01-01", + nominal_cap=1024000000000): + conn.execute( + "INSERT INTO endurance_baseline " + "(tbw_terabytes, source_url, document_revision, entry_date, model_string, " + " nominal_capacity_bytes, validated_by, verified, created_at, updated_at) " + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", + (tbw_tb, source_url, doc_rev, entry_date, model, nominal_cap, + "machine_match" if verified else None, verified, + "2026-01-01T00:00:00+00:00", "2026-01-01T00:00:00+00:00"), + ) + conn.commit() + + +def _insert_segment(conn, opened_at="2026-09-01T00:00:00+00:00", + identity_key="nqn.test", degraded=False, + mn="Samsung SSD 970 EVO Plus 1TB"): + conn.execute( + "INSERT INTO controller_segments " + "(opened_at, identity_key, identity_degraded, subnqn, sn, mn, fr, vid, ssvid, transport) " + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", + (opened_at, identity_key, degraded, "nqn.test", "SN123", mn, "FW1", + "0x144d", "0x144d", "pcie"), + ) + conn.commit() + + +def _insert_day(conn, day, bw=1024*1024*100, coverage=0.95, samples=24): + conn.execute( + "INSERT INTO day_aggregates (day, active_seconds, idle_seconds, " + "powered_off_seconds, unknown_seconds, bytes_written_delta, " + "bytes_read_delta, sample_count, coverage) " + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)", + (day, 3600, 0, 0, 0, bw, 0, samples, coverage), + ) + conn.commit() + + +def _insert_sample(conn, ts, pu=5): + conn.execute( + "INSERT INTO samples (ts, device, data_units_written, data_units_read, " + "percentage_used, bytes_written, bytes_read, power_on_hours) " + "VALUES (?, ?, ?, ?, ?, ?, ?, ?)", + (ts, "/dev/nvme0n1", 1000000, 500000, pu, 512000000000, 256000000000, 8765), + ) + conn.commit() + + +def _open_period(conn, start="2026-09-01T00:00:00+00:00"): + ensure_period_open(conn, datetime.fromisoformat(start)) + + +def _setup_full_store(conn, *, baseline=True, segment=True, days=30, + bw=1024*1024*100, coverage=0.95, samples_per_day=24, + sample_ts="2026-09-30T10:00:00+00:00", + period_start="2026-09-01T00:00:00+00:00", + segment_opened="2026-09-01T00:00:00+00:00", + baseline_kw=None, segment_kw=None): + if baseline: + _insert_baseline(conn, **(baseline_kw or {})) + if segment: + _insert_segment(conn, opened_at=segment_opened, **(segment_kw or {})) + _open_period(conn, start=period_start) + for i in range(days): + d = (datetime(2026, 9, 1) + timedelta(days=i)).strftime("%Y-%m-%d") + _insert_day(conn, d, bw=bw, coverage=coverage, samples=samples_per_day) + if sample_ts: + _insert_sample(conn, sample_ts) + + +# =================================================================== +# CI-1: Exhaustive state matrix +# =================================================================== + + +class TestCI1StateMatrix: + """Systematic walk of confidence x freshness x baseline tier combinations.""" + + def test_no_baseline_unavailable(self, tmp_path): + conn = init_store(tmp_path / "db") + _insert_segment(conn) + _open_period(conn) + for i in range(30): + d = (datetime(2026, 9, 1) + timedelta(days=i)).strftime("%Y-%m-%d") + _insert_day(conn, d, bw=1024*1024*100, coverage=0.95, samples=24) + _insert_sample(conn, "2026-09-30T10:00:00+00:00") + proj = compute_projection(conn, _clock()) + assert proj.confidence_state == ConfidenceState.UNSUPPORTED + assert proj.headline_remaining_seconds is None + assert proj.baseline_tier == BaselineTier.NONE + conn.close() + + def test_verified_baseline_possible_supported(self, tmp_path): + conn = init_store(tmp_path / "db") + _setup_full_store(conn, baseline_kw=dict(tbw_tb=10.0, verified=True)) + proj = compute_projection(conn, _clock()) + assert proj.confidence_state == ConfidenceState.SUPPORTED + assert proj.baseline_tier == BaselineTier.VERIFIED + assert proj.headline_remaining_seconds is not None + conn.close() + + def test_unverified_baseline_possible_limited(self, tmp_path): + conn = init_store(tmp_path / "db") + _setup_full_store(conn, baseline_kw=dict( + tbw_tb=10.0, verified=False, source_url=None)) + proj = compute_projection(conn, _clock()) + assert proj.baseline_tier == BaselineTier.UNVERIFIED + assert proj.confidence_state != ConfidenceState.SUPPORTED + conn.close() + + def test_model_mismatch_unavailable(self, tmp_path): + conn = init_store(tmp_path / "db") + _setup_full_store(conn, baseline_kw=dict(model="Different Model")) + proj = compute_projection(conn, _clock()) + assert proj.confidence_state == ConfidenceState.UNSUPPORTED + assert proj.baseline_tier == BaselineTier.NONE + conn.close() + + def test_fresh_sample_grades_fresh(self, tmp_path): + now = _clock() + ts = (now - timedelta(seconds=FRESH_THRESHOLD_S - 10)).isoformat() + assert grade_freshness(ts, now) == "fresh" + + def test_missed_sample_grades_missed(self, tmp_path): + now = _clock() + ts = (now - timedelta(hours=2)).isoformat() + assert grade_freshness(ts, now) == "missed" + + def test_stale_sample_grades_stale(self, tmp_path): + now = _clock() + ts = (now - timedelta(hours=49)).isoformat() + assert grade_freshness(ts, now) == "stale" + + def test_empty_store_grades_empty(self, tmp_path): + now = _clock() + assert grade_freshness(None, now) == "empty" + + def test_unsupported_fresh(self, tmp_path): + conn = init_store(tmp_path / "db") + _insert_segment(conn) + _open_period(conn) + for i in range(20): + d = (datetime(2026, 9, 10) + timedelta(days=i)).strftime("%Y-%m-%d") + _insert_day(conn, d, bw=1024*1024*100) + fresh_ts = (_clock() - timedelta(seconds=60)).isoformat() + _insert_sample(conn, fresh_ts) + proj = compute_projection(conn, _clock()) + assert proj.confidence_state == ConfidenceState.UNSUPPORTED + assert grade_freshness(fresh_ts, _clock()) == "fresh" + conn.close() + + def test_limited_young_regime(self, tmp_path): + conn = init_store(tmp_path / "db") + _insert_baseline(conn, tbw_tb=10.0, verified=True) + _insert_segment(conn) + _open_period(conn) + for i in range(5): + d = (datetime(2026, 9, 25) + timedelta(days=i)).strftime("%Y-%m-%d") + _insert_day(conn, d, bw=1024*1024*100) + _insert_sample(conn, "2026-09-30T10:00:00+00:00") + proj = compute_projection(conn, _clock()) + assert proj.confidence_state == ConfidenceState.LIMITED + conn.close() + + def test_limited_warming(self, tmp_path): + conn = init_store(tmp_path / "db") + _insert_baseline(conn, tbw_tb=10.0, verified=True) + _insert_segment(conn) + _open_period(conn) + for i in range(10): + d = (datetime(2026, 9, 20) + timedelta(days=i)).strftime("%Y-%m-%d") + _insert_day(conn, d, bw=1024*1024*100, coverage=0.95, samples=24) + _insert_sample(conn, "2026-09-30T10:00:00+00:00") + proj = compute_projection(conn, _clock()) + assert proj.confidence_state == ConfidenceState.LIMITED + assert proj.warming_fact is not None + conn.close() + + def test_limited_stale_data(self, tmp_path): + conn = init_store(tmp_path / "db") + _insert_baseline(conn, tbw_tb=10.0, verified=True) + _insert_segment(conn, opened_at="2026-08-01T00:00:00+00:00") + _open_period(conn, start="2026-08-01T00:00:00+00:00") + for i in range(30): + d = (datetime(2026, 8, 1) + timedelta(days=i)).strftime("%Y-%m-%d") + _insert_day(conn, d, bw=1024*1024*100, coverage=0.95, samples=24) + stale_ts = (_clock() - timedelta(days=5)).isoformat() + _insert_sample(conn, stale_ts) + proj = compute_projection(conn, _clock()) + assert proj.confidence_state == ConfidenceState.LIMITED + assert proj.staleness_fact is not None + conn.close() + + def test_limited_degraded_identity(self, tmp_path): + conn = init_store(tmp_path / "db") + _insert_baseline(conn, tbw_tb=10.0, verified=True) + _insert_segment(conn, identity_key=None, degraded=True) + _open_period(conn) + for i in range(30): + d = (datetime(2026, 9, 1) + timedelta(days=i)).strftime("%Y-%m-%d") + _insert_day(conn, d, bw=1024*1024*100, coverage=0.95, samples=24) + _insert_sample(conn, "2026-09-30T10:00:00+00:00") + proj = compute_projection(conn, _clock()) + assert proj.confidence_state == ConfidenceState.LIMITED + assert proj.degraded_identity_fact is not None + conn.close() + + def test_unsupported_zero_rate(self, tmp_path): + conn = init_store(tmp_path / "db") + _insert_baseline(conn, tbw_tb=10.0, verified=True) + _insert_segment(conn) + _open_period(conn) + for i in range(30): + d = (datetime(2026, 9, 1) + timedelta(days=i)).strftime("%Y-%m-%d") + _insert_day(conn, d, bw=0) + _insert_sample(conn, "2026-09-30T10:00:00+00:00") + proj = compute_projection(conn, _clock()) + assert proj.confidence_state == ConfidenceState.UNSUPPORTED + assert proj.zero_rate_fact is not None + conn.close() + + def test_headline_present_when_projection_exists(self, tmp_path): + conn = init_store(tmp_path / "db") + _setup_full_store(conn, baseline_kw=dict(tbw_tb=10.0, verified=True)) + proj = compute_projection(conn, _clock()) + assert proj.headline_remaining_seconds is not None + assert proj.headline_remaining_seconds > 0 + conn.close() + + def test_headline_absent_when_unavailable(self, tmp_path): + conn = init_store(tmp_path / "db") + _insert_segment(conn) + _open_period(conn) + proj = compute_projection(conn, _clock()) + assert proj.headline_remaining_seconds is None + conn.close() + + def test_headline_absent_when_zero_rate(self, tmp_path): + conn = init_store(tmp_path / "db") + _insert_baseline(conn, tbw_tb=1.0, verified=True) + _insert_segment(conn) + _open_period(conn) + for i in range(20): + d = (datetime(2026, 9, 10) + timedelta(days=i)).strftime("%Y-%m-%d") + _insert_day(conn, d, bw=0) + proj = compute_projection(conn, _clock()) + assert proj.headline_remaining_seconds is None + conn.close() + + def test_facts_always_list(self, tmp_path): + conn = init_store(tmp_path / "db") + _insert_segment(conn) + _open_period(conn) + proj = compute_projection(conn, _clock()) + assert isinstance(proj.contributing_facts, list) + conn.close() + + def test_facts_never_empty_for_unavailable(self, tmp_path): + conn = init_store(tmp_path / "db") + _insert_segment(conn) + _open_period(conn) + proj = compute_projection(conn, _clock()) + assert len(proj.contributing_facts) > 0 + conn.close() + + def test_confidence_never_percentage(self, tmp_path): + conn = init_store(tmp_path / "db") + _setup_full_store(conn, baseline_kw=dict(tbw_tb=10.0, verified=True)) + proj = compute_projection(conn, _clock()) + assert proj.confidence_state in ( + ConfidenceState.UNSUPPORTED, ConfidenceState.LIMITED, ConfidenceState.SUPPORTED) + for f in proj.contributing_facts: + if re.match(r"^\\d+%$", f.strip()): + pytest.fail("Bare percentage in facts: %r" % f) + conn.close() + + def test_status_renders_same_state_as_projection(self, tmp_path): + db = tmp_path / "observations.db" + conn = init_store(db) + _setup_full_store(conn, baseline_kw=dict(tbw_tb=10.0, verified=True)) + conn.close() + now = _clock() + with patch("fenris.status.query_service_state", return_value={ + "boot_enabled": True, "timer_active": True, + "last_collect_ok": True, "last_collect_age_s": 60, + "last_collect_reason": None, + }): + status = get_status(store_path=db, clock_now=now, + query_services=True, query_journal=False) + assert "Supported" in status or "supported" in status.lower() + assert "remaining" in status.lower() + + +# =================================================================== +# CI-2: TUI/CLI parity +# =================================================================== + + +class TestCI2Parity: + """Verify TUI and CLI share the same constants, formatting, and wording.""" + + def test_freshness_constants_shared(self): + from fenris import tui as tui_mod + from fenris import status as status_mod + assert tui_mod.FRESH_THRESHOLD_S == status_mod.FRESH_THRESHOLD_S + assert tui_mod.STALENESS_THRESHOLD_S == status_mod.STALENESS_THRESHOLD_S + + def test_grade_freshness_shared(self): + from fenris.tui import grade_freshness as tui_gf + from fenris.status import grade_freshness as status_gf + assert tui_gf is status_gf + + def test_disclosures_shared(self): + from fenris.projection import DISCLOSURES as proj_disc + from fenris.status import format_disclosures + output = format_disclosures() + for d in proj_disc: + assert d in output + + def test_status_four_facts_match_tui_strip(self, tmp_path): + db = tmp_path / "observations.db" + conn = init_store(db) + _insert_segment(conn) + _open_period(conn) + for i in range(20): + d = (datetime(2026, 9, 10) + timedelta(days=i)).strftime("%Y-%m-%d") + _insert_day(conn, d, bw=1024*1024*100) + _insert_sample(conn, "2026-09-30T10:00:00+00:00") + conn.close() + now = _clock() + with patch("fenris.status.query_service_state", return_value={ + "boot_enabled": True, "timer_active": True, + "last_collect_ok": True, "last_collect_age_s": 120, + "last_collect_reason": None, + }): + status = get_status(store_path=db, clock_now=now, + query_services=True, query_journal=False) + assert "boot:" in status + assert "timer:" in status + assert "last collect:" in status + assert "freshness:" in status + + def test_pause_resume_action_names(self): + tui_keys = {b.key for b in FenrisTuiApp.BINDINGS} + assert "p" in tui_keys + assert "r" in tui_keys + assert "c" in tui_keys + assert "q" in tui_keys + + def test_empty_store_greeting_both_views(self, tmp_path): + db = tmp_path / "observations.db" + init_store(db) + now = _clock() + with patch("fenris.status.query_service_state", return_value={ + "boot_enabled": False, "timer_active": False, + "last_collect_ok": None, "last_collect_age_s": None, + "last_collect_reason": None, + }): + status = get_status(store_path=db, clock_now=now, + query_services=True, query_journal=False) + assert "no observations yet" in status.lower() + + def test_store_fault_phrase_both_views(self, tmp_path): + status_src = (FENRIS_PKG / "status.py").read_text() + tui_src = (FENRIS_PKG / "tui.py").read_text() + phrase = "observation store unreadable" + assert phrase in status_src + assert phrase.lower() in tui_src.lower() + + def test_newer_schema_phrase_both_views(self): + status_src = (FENRIS_PKG / "status.py").read_text() + phrase = "observation store written by a newer Fenris" + assert phrase in status_src + + def test_status_never_prompts(self): + status_src = (FENRIS_PKG / "status.py").read_text() + assert "input(" not in status_src + + +# =================================================================== +# CI-3: Prohibition set +# =================================================================== + + +class TestCI3ProhibitionSet: + """Structural codebase checks for every prohibition clause.""" + + def _read_all_sources(self): + files = {} + for py in FENRIS_PKG.glob("*.py"): + files[py.name] = py.read_text() + return files + + def test_single_acquisition_path(self): + """Only fenris-collect may interrogate the device. [2.1, 8.7] + + collector.py contains the acquisition functions; collect.py is the + fenris-collect entry point that invokes them. No other module may + reference smartctl. + """ + sources = self._read_all_sources() + allowed = {"collector.py", "collect.py"} + for name, text in sources.items(): + if name in allowed: + continue + assert "smartctl" not in text, ( + "%s must not contain smartctl" % name + ) + + def test_no_run_surface(self): + """No /run/fenris coordination surface. [1.2, 3]""" + sources = self._read_all_sources() + for name, text in sources.items(): + assert "/run/fenris" not in text, ( + "%s references /run/fenris" % name + ) + + def test_single_config_key(self): + """Config holds exactly one key: device. [8.3]""" + status_src = (FENRIS_PKG / "status.py").read_text() + in_read_config = False + config_keys = [] + for line in status_src.split("\n"): + if "def read_config" in line: + in_read_config = True + elif in_read_config and line.strip().startswith("def "): + break + elif in_read_config and "key ==" in line: + match = re.search(r'key\s*==\s*["\']([^"\']+)["\']', line) + if match: + config_keys.append(match.group(1)) + assert "device" in config_keys + assert len(config_keys) == 1, "Found keys: %s" % config_keys + + def test_no_alerting_machinery(self): + """No alerting, notification, or escalation. [9.6]""" + sources = self._read_all_sources() + alert_keywords = ["send_email", "smtp", "webhook", "push_notification"] + for name, text in sources.items(): + for kw in alert_keywords: + for line in text.split("\n"): + stripped = line.strip() + if kw in stripped and not stripped.startswith("#"): + pytest.fail( + "%s contains alerting keyword '%s': %s" % (name, kw, stripped) + ) + + def test_no_synthetic_baselines(self): + """No synthetic or capacity-derived baseline. [6.1]""" + proj_src = (FENRIS_PKG / "projection.py").read_text() + assert "synthetic" not in proj_src.lower() + + def test_no_stored_projections(self): + """Projections never stored; recomputed on read. [3.7, 6.10]""" + store_src = (FENRIS_PKG / "store.py").read_text() + create_tables = re.findall(r"CREATE TABLE.*?(?=\n\n|$)", store_src, re.DOTALL) + table_names = [] + for ct in create_tables: + m = re.search(r"IF NOT EXISTS\s+(\w+)", ct) + if m: + table_names.append(m.group(1)) + assert "projection" not in [t.lower() for t in table_names] + + def test_no_partial_newer_schema_interpretation(self): + """Readers refuse newer-schema stores. [3.6, 9.5]""" + status_src = (FENRIS_PKG / "status.py").read_text() + assert "NewerSchema" in status_src + assert "upgrade Fenris" in status_src + + def test_polkit_authorizes_one_binary(self): + """Polkit authorizes exactly one binary: fenris-monitor. [8.5]""" + monitor_src = (FENRIS_PKG / "monitor.py").read_text() + assert "fenris-monitor" in monitor_src or "fenris_monitor" in monitor_src + collect_src = (FENRIS_PKG / "collect.py").read_text() + assert "polkit" not in collect_src.lower() + + def test_no_hour_interpolation(self): + """No absent hour is interpolated or fabricated. [5.3]""" + proj_src = (FENRIS_PKG / "projection.py").read_text() + assert "interpolat" not in proj_src.lower() + assert "fabricat" not in proj_src.lower() + + def test_fenris_sh_not_shipped(self): + """fenris.sh is not shipped. [8.8]""" + repo_root = Path(__file__).parent.parent + assert not (repo_root / "fenris.sh").exists() + + +# =================================================================== +# CI-4: Required wording and six disclosures +# =================================================================== + + +class TestCI4WordingAndDisclosures: + """Verify exact fixed phrases and disclosures in both views.""" + + def test_exactly_six_disclosures(self): + assert len(DISCLOSURES) == 6 + + def test_disclosure_1_endurance_not_failure(self): + assert "endurance projection" in DISCLOSURES[0].lower() + assert "hardware-failure" in DISCLOSURES[0].lower() or "failure date" in DISCLOSURES[0].lower() + + def test_disclosure_2_vendor_specific(self): + assert "vendor-specific" in DISCLOSURES[1] + assert "255 is saturated" in DISCLOSURES[1] + + def test_disclosure_3_warranty_not_failure(self): + assert "warranty" in DISCLOSURES[2].lower() or "endurance threshold" in DISCLOSURES[2].lower() + assert "failure threshold" in DISCLOSURES[2].lower() + + def test_disclosure_4_duw_rounding(self): + assert "DUW" in DISCLOSURES[3] + assert "upward-rounded" in DISCLOSURES[3] + assert "NAND" in DISCLOSURES[3] + + def test_disclosure_5_quality_depends(self): + assert "baseline provenance" in DISCLOSURES[4] + assert "future workload" in DISCLOSURES[4] + + def test_disclosure_6_gaps_and_disabled(self): + assert "Gaps" in DISCLOSURES[5] + assert "deliberately disabled" in DISCLOSURES[5] + + def test_disclosures_render_in_status(self): + output = format_disclosures() + assert output.startswith("Disclosures") + for i in range(1, 7): + assert "%d." % i in output + for d in DISCLOSURES: + assert d in output + + def test_disclosures_render_in_tui(self): + tui_src = (FENRIS_PKG / "tui.py").read_text() + assert "format_disclosures" in tui_src + + def test_zero_rate_phrase(self): + phrase = "no finite projection from this history" + proj_src = (FENRIS_PKG / "projection.py").read_text() + assert phrase in proj_src + status_src = (FENRIS_PKG / "status.py").read_text() + assert phrase in status_src + + def test_unavailable_no_baseline_phrase(self): + phrase = "no applicable endurance baseline" + proj_src = (FENRIS_PKG / "projection.py").read_text() + assert phrase in proj_src + + def test_store_fault_phrase(self): + phrase = "observation store unreadable" + status_src = (FENRIS_PKG / "status.py").read_text() + assert phrase in status_src + tui_src = (FENRIS_PKG / "tui.py").read_text() + assert phrase.lower() in tui_src.lower() + + def test_newer_schema_phrase(self): + phrase = "observation store written by a newer Fenris" + status_src = (FENRIS_PKG / "status.py").read_text() + assert phrase in status_src + + def test_no_observations_phrase(self): + phrase = "no observations yet" + status_src = (FENRIS_PKG / "status.py").read_text() + assert phrase in status_src + tui_src = (FENRIS_PKG / "tui.py").read_text() + assert phrase in tui_src.lower() + + def test_config_error_phrase(self): + phrase = "configuration error:" + status_src = (FENRIS_PKG / "status.py").read_text() + assert phrase in status_src + + def test_degraded_identity_phrase(self): + phrase = "controller identity unavailable" + proj_src = (FENRIS_PKG / "projection.py").read_text() + assert phrase in proj_src + phrase2 = "replacement detection relies on write-counter continuity only" + assert phrase2 in proj_src + + def test_scenario_range_only_spread(self): + proj_src = (FENRIS_PKG / "projection.py").read_text() + assert "confidence interval" not in proj_src.lower() + + def test_no_percentage_in_confidence_rendering(self): + for name in ["projection.py", "tui.py", "status.py"]: + src = (FENRIS_PKG / name).read_text() + assert not re.search(r"\\d+%\\s*confidence", src, re.IGNORECASE), ( + "Found XX%% confidence in %s" % name + ) + + def test_status_disclosures_accessible(self): + db = Path("/tmp/_ci4_test.db") + conn = init_store(db) + conn.close() + now = _clock() + with patch("fenris.status.query_service_state", return_value={ + "boot_enabled": False, "timer_active": False, + "last_collect_ok": None, "last_collect_age_s": None, + "last_collect_reason": None, + }): + result = render_status(store_path=db, clock_now=now, + query_services=True, query_journal=False, + show_disclosures=True) + assert "Disclosures" in result + assert "1." in result + assert "6." in result + db.unlink(missing_ok=True)