Compare commits
2
Commits
7802a72606
...
7f006c7df7
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
7f006c7df7 | ||
|
|
b99ebfe9dd |
@@ -21,7 +21,11 @@ import mimetypes
|
|||||||
from datetime import datetime, timezone, timedelta
|
from datetime import datetime, timezone, timedelta
|
||||||
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
||||||
|
|
||||||
SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
|
# 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_DIR = os.path.join(SCRIPT_DIR, "data")
|
||||||
DATA_FILE = os.path.join(DATA_DIR, "history.jsonl")
|
DATA_FILE = os.path.join(DATA_DIR, "history.jsonl")
|
||||||
HOURLY_FILE = os.path.join(DATA_DIR, "hourly.jsonl")
|
HOURLY_FILE = os.path.join(DATA_DIR, "hourly.jsonl")
|
||||||
@@ -973,38 +977,18 @@ def cmd_stop(args):
|
|||||||
|
|
||||||
|
|
||||||
def cmd_status(args):
|
def cmd_status(args):
|
||||||
running = False
|
try:
|
||||||
if os.path.exists(PID_FILE):
|
from fenris.status import render_status, check_retired_flag
|
||||||
with open(PID_FILE) as f:
|
from pathlib import Path
|
||||||
pid = int(f.read().strip())
|
|
||||||
running = pid_alive(pid)
|
show_disclosures = getattr(args, "disclosures", False)
|
||||||
print(f"Fenris daemon: {'RUNNING (pid ' + str(pid) + ')' if running else 'not running (stale pid file)'}")
|
store_path = Path("/var/lib/fenris/observations.db")
|
||||||
else:
|
output = render_status(store_path=store_path, show_disclosures=show_disclosures)
|
||||||
print("Fenris daemon: not running")
|
print(output)
|
||||||
rows = load_history()
|
except ImportError:
|
||||||
if rows:
|
# Fallback if fenris package not importable
|
||||||
summ = compute_summary(rows)
|
print("Error: cannot import fenris.status module. Is the package installed?")
|
||||||
latest = rows[-1]
|
sys.exit(1)
|
||||||
print(f"Samples collected: {len(rows)}")
|
|
||||||
print(f"Last sample: {latest['ts']}")
|
|
||||||
print(f"Wear (percentage_used): {latest.get('percentage_used')}%")
|
|
||||||
print(f"Total written: {latest.get('bytes_written', 0) / 1e9:.1f} GB")
|
|
||||||
print(f"Written (24h rolling): {summ['window24h']['gb']:.2f} GB over {summ['window24h']['coverage_hours']:.1f}h")
|
|
||||||
print(f"Write rate: {summ['gb_per_hour']:.2f} GB/h ({summ['gb_per_day']:.1f} GB/day)")
|
|
||||||
if summ["seconds_remaining"]:
|
|
||||||
print(f"Projected life remaining: {summ['breakdown']['human']} (≈{summ['breakdown']['days']:.0f} days / {summ['breakdown']['hours']:.0f} hours / {summ['breakdown']['years']:.2f} years)")
|
|
||||||
print(f"Endurance: {summ['endurance_tb']:.1f} TB total, {summ['remaining_tb']:.1f} TB remaining" + (" (estimated)" if summ["endurance_estimated"] else ""))
|
|
||||||
if summ["wear_model_days"]:
|
|
||||||
print(f"Wear-model cross-check: ~{summ['wear_model_days']:.0f} days at current wear rate")
|
|
||||||
if summ["preliminary"]:
|
|
||||||
print("Note: preliminary — less than 24h coverage")
|
|
||||||
else:
|
|
||||||
print("Projected life remaining: — (no writes in window or no endurance data)")
|
|
||||||
hourly = load_hourly()
|
|
||||||
if hourly:
|
|
||||||
print(f"Hourly buckets: {len(hourly)} (last {hourly[-1]['hour']}: {hourly[-1]['bytes_written']/1e9:.2f} GB)")
|
|
||||||
else:
|
|
||||||
print("No samples collected yet.")
|
|
||||||
|
|
||||||
|
|
||||||
def cmd_run(args):
|
def cmd_run(args):
|
||||||
@@ -1023,8 +1007,22 @@ def cmd_sample_once(args):
|
|||||||
sys.exit(1)
|
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():
|
def main():
|
||||||
p = argparse.ArgumentParser(description="Fenris — NVMe wear monitor & dashboard (by Bongbetic)")
|
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)
|
sub = p.add_subparsers(dest="cmd", required=True)
|
||||||
|
|
||||||
def add_common(sp):
|
def add_common(sp):
|
||||||
@@ -1032,19 +1030,30 @@ def main():
|
|||||||
sp.add_argument("--interval", type=int, default=300, help="seconds between samples (default 300)")
|
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.add_argument("--port", type=int, default=8420, help="dashboard HTTP port (default 8420)")
|
||||||
|
|
||||||
sp = sub.add_parser("start", help="start monitoring in background")
|
sp = sub.add_parser("start", help="(retired — use 'fenris monitor resume')")
|
||||||
add_common(sp); sp.set_defaults(func=cmd_start)
|
add_common(sp); sp.set_defaults(func=cmd_retired)
|
||||||
sp = sub.add_parser("stop", help="stop background monitoring")
|
sp = sub.add_parser("stop", help="(retired — use 'fenris monitor pause')")
|
||||||
sp.set_defaults(func=cmd_stop)
|
sp.set_defaults(func=cmd_retired)
|
||||||
sp = sub.add_parser("status", help="show daemon + latest wear stats")
|
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.set_defaults(func=cmd_status)
|
||||||
sp = sub.add_parser("run", help="run in foreground (used internally by 'start')")
|
sp = sub.add_parser("run", help="(retired — use 'fenris monitor resume')")
|
||||||
add_common(sp); sp.set_defaults(func=cmd_run)
|
add_common(sp); sp.set_defaults(func=cmd_retired)
|
||||||
sp = sub.add_parser("sample", help="take one sample immediately and print it")
|
sp = sub.add_parser("sample", help="take one sample immediately and print it")
|
||||||
sp.add_argument("--device", default=detect_device())
|
sp.add_argument("--device", default=detect_device())
|
||||||
sp.set_defaults(func=cmd_sample_once)
|
sp.set_defaults(func=cmd_sample_once)
|
||||||
|
|
||||||
args = p.parse_args()
|
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)
|
args.func(args)
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
+49
-35
@@ -236,6 +236,9 @@ def _compute_regime_rate(days, conn, regime_start_day, clock_now):
|
|||||||
|
|
||||||
def _compute_horizon_rate(days, conn, horizon_days, clock_now):
|
def _compute_horizon_rate(days, conn, horizon_days, clock_now):
|
||||||
cutoff = (clock_now - timedelta(days=horizon_days)).strftime("%Y-%m-%d")
|
cutoff = (clock_now - timedelta(days=horizon_days)).strftime("%Y-%m-%d")
|
||||||
|
# History must span the full horizon — no placeholders
|
||||||
|
if not days or days[0]["day"] > cutoff:
|
||||||
|
return None
|
||||||
h_bytes = sum(d["bytes_written"] for d in days if d["day"] >= cutoff)
|
h_bytes = sum(d["bytes_written"] for d in days if d["day"] >= cutoff)
|
||||||
covered = sum(1 for d in days if d["day"] >= cutoff)
|
covered = sum(1 for d in days if d["day"] >= cutoff)
|
||||||
if covered == 0:
|
if covered == 0:
|
||||||
@@ -252,54 +255,62 @@ def _compute_horizon_rate(days, conn, horizon_days, clock_now):
|
|||||||
# ---------------------------------------------------------------------------
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
def _detect_habit_change(days):
|
def _detect_habit_change(days):
|
||||||
need = HABIT_CHANGE_SHORT_WINDOW + HABIT_CHANGE_LONG_WINDOW + HABIT_CHANGE_CONSECUTIVE_DAYS
|
"""Detect habit change per spec §6.4.
|
||||||
|
|
||||||
|
Trailing 7-day mean >= 2x (or <= 0.5x) the preceding 28-day mean
|
||||||
|
for 3 consecutive days. Returns (change_day, days_since) or None.
|
||||||
|
The first divergence day is the earliest day in the consecutive run.
|
||||||
|
"""
|
||||||
|
need = HABIT_CHANGE_SHORT_WINDOW + HABIT_CHANGE_LONG_WINDOW
|
||||||
if len(days) < need:
|
if len(days) < need:
|
||||||
return None
|
return None
|
||||||
|
|
||||||
for i in range(len(days) - 1, HABIT_CHANGE_LONG_WINDOW + HABIT_CHANGE_SHORT_WINDOW - 1, -1):
|
def _ratio_at(end_idx):
|
||||||
se = i + 1
|
"""Compute 7-day / preceding-28-day mean ratio ending at end_idx."""
|
||||||
|
if end_idx < HABIT_CHANGE_SHORT_WINDOW - 1:
|
||||||
|
return None
|
||||||
|
se = end_idx + 1
|
||||||
ss = se - HABIT_CHANGE_SHORT_WINDOW
|
ss = se - HABIT_CHANGE_SHORT_WINDOW
|
||||||
s_bytes = sum(d["bytes_written"] for d in days[ss:se])
|
s_bytes = sum(d["bytes_written"] for d in days[ss:se])
|
||||||
s_mean = s_bytes / HABIT_CHANGE_SHORT_WINDOW
|
s_mean = s_bytes / HABIT_CHANGE_SHORT_WINDOW
|
||||||
|
|
||||||
le = ss
|
le = ss
|
||||||
ls = le - HABIT_CHANGE_LONG_WINDOW
|
ls = le - HABIT_CHANGE_LONG_WINDOW
|
||||||
if ls < 0:
|
if ls < 0:
|
||||||
break
|
return None
|
||||||
l_bytes = sum(d["bytes_written"] for d in days[ls:le])
|
l_bytes = sum(d["bytes_written"] for d in days[ls:le])
|
||||||
l_mean = l_bytes / HABIT_CHANGE_LONG_WINDOW
|
l_mean = l_bytes / HABIT_CHANGE_LONG_WINDOW
|
||||||
if l_mean == 0:
|
if l_mean == 0:
|
||||||
|
return None
|
||||||
|
return s_mean / l_mean
|
||||||
|
|
||||||
|
# Scan backwards from the most recent day
|
||||||
|
for i in range(len(days) - 1, HABIT_CHANGE_LONG_WINDOW + HABIT_CHANGE_SHORT_WINDOW - 2, -1):
|
||||||
|
ratio = _ratio_at(i)
|
||||||
|
if ratio is None:
|
||||||
continue
|
continue
|
||||||
|
|
||||||
ratio = s_mean / l_mean
|
is_upper = ratio >= HABIT_CHANGE_UPPER_FACTOR
|
||||||
if ratio >= HABIT_CHANGE_UPPER_FACTOR or ratio <= HABIT_CHANGE_LOWER_FACTOR:
|
is_lower = ratio <= HABIT_CHANGE_LOWER_FACTOR
|
||||||
consecutive = 0
|
if not (is_upper or is_lower):
|
||||||
for j in range(ss, min(ss + HABIT_CHANGE_CONSECUTIVE_DAYS, len(days))):
|
continue
|
||||||
s2e = j + 1
|
|
||||||
s2s = s2e - HABIT_CHANGE_SHORT_WINDOW
|
|
||||||
if s2s < 0:
|
|
||||||
break
|
|
||||||
s2_bytes = sum(d["bytes_written"] for d in days[s2s:s2e])
|
|
||||||
s2_mean = s2_bytes / HABIT_CHANGE_SHORT_WINDOW
|
|
||||||
l2e = s2s
|
|
||||||
l2s = l2e - HABIT_CHANGE_LONG_WINDOW
|
|
||||||
if l2s < 0:
|
|
||||||
break
|
|
||||||
l2_bytes = sum(d["bytes_written"] for d in days[l2s:l2e])
|
|
||||||
l2_mean = l2_bytes / HABIT_CHANGE_LONG_WINDOW
|
|
||||||
if l2_mean == 0:
|
|
||||||
break
|
|
||||||
r = s2_mean / l2_mean
|
|
||||||
if (ratio >= HABIT_CHANGE_UPPER_FACTOR and r >= HABIT_CHANGE_UPPER_FACTOR) or \
|
|
||||||
(ratio <= HABIT_CHANGE_LOWER_FACTOR and r <= HABIT_CHANGE_LOWER_FACTOR):
|
|
||||||
consecutive += 1
|
|
||||||
else:
|
|
||||||
break
|
|
||||||
|
|
||||||
if consecutive >= HABIT_CHANGE_CONSECUTIVE_DAYS:
|
# Count consecutive days going backwards from i
|
||||||
change_day = days[ss]["day"]
|
consecutive = 1
|
||||||
days_since = (datetime.fromisoformat(days[-1]["day"]) - datetime.fromisoformat(change_day)).days
|
for j in range(i - 1, HABIT_CHANGE_LONG_WINDOW + HABIT_CHANGE_SHORT_WINDOW - 3, -1):
|
||||||
return change_day, days_since
|
r = _ratio_at(j)
|
||||||
|
if r is None:
|
||||||
|
break
|
||||||
|
if (is_upper and r >= HABIT_CHANGE_UPPER_FACTOR) or \
|
||||||
|
(is_lower and r <= HABIT_CHANGE_LOWER_FACTOR):
|
||||||
|
consecutive += 1
|
||||||
|
else:
|
||||||
|
break
|
||||||
|
|
||||||
|
if consecutive >= HABIT_CHANGE_CONSECUTIVE_DAYS:
|
||||||
|
change_idx = i - consecutive + 1
|
||||||
|
change_day = days[change_idx]["day"]
|
||||||
|
days_since = (datetime.fromisoformat(days[-1]["day"]) - datetime.fromisoformat(change_day)).days
|
||||||
|
return change_day, days_since
|
||||||
|
|
||||||
return None
|
return None
|
||||||
|
|
||||||
@@ -504,8 +515,11 @@ def compute_projection(conn, clock_now):
|
|||||||
if horizon_rates:
|
if horizon_rates:
|
||||||
scenario = ScenarioRange(rates=horizon_rates, min_days=min(horizon_rates), max_days=max(horizon_rates))
|
scenario = ScenarioRange(rates=horizon_rates, min_days=min(horizon_rates), max_days=max(horizon_rates))
|
||||||
|
|
||||||
qualifying = sum(1 for d in segment_days if d["coverage"] >= WARMING_COVERAGE_FLOOR and d["sample_count"] > 0)
|
total_days_count = len(segment_days)
|
||||||
if qualifying < WARMING_MIN_DAYS:
|
days_below_coverage = sum(1 for d in segment_days
|
||||||
|
if d["coverage"] < WARMING_COVERAGE_FLOOR or d["sample_count"] == 0)
|
||||||
|
qualifying = total_days_count - days_below_coverage
|
||||||
|
if total_days_count < WARMING_MIN_DAYS or days_below_coverage > WARMING_MAX_LOW_COVERAGE:
|
||||||
warming_fact = "warming up: %d of %d qualifying days" % (qualifying, WARMING_MIN_DAYS)
|
warming_fact = "warming up: %d of %d qualifying days" % (qualifying, WARMING_MIN_DAYS)
|
||||||
facts.append(warming_fact)
|
facts.append(warming_fact)
|
||||||
|
|
||||||
|
|||||||
@@ -0,0 +1,599 @@
|
|||||||
|
"""Read-only CLI status command: the CLI twin of the TUI (spec §8.8, LC-9, CI-2).
|
||||||
|
|
||||||
|
Composes from the observation store (read-only) and allow-listed systemctl
|
||||||
|
properties: projection facts, four separate service facts (boot enablement,
|
||||||
|
runtime activity, last collect outcome, freshness), and a journalctl hint on
|
||||||
|
failure or staleness. Never auto-samples, never prompts.
|
||||||
|
|
||||||
|
Freshness constants are defined once here and shared with the TUI (§8.9):
|
||||||
|
fresh — newest sample within 2 × cadence + AccuracySec + 60 s
|
||||||
|
missed — between fresh and 48 h
|
||||||
|
stale — ≥ 48 h
|
||||||
|
empty — no observations yet
|
||||||
|
|
||||||
|
Criteria: LC-9, CI-2, CI-4, FL-4, FL-5, FL-7.
|
||||||
|
"""
|
||||||
|
import sqlite3
|
||||||
|
import subprocess
|
||||||
|
import sys
|
||||||
|
from datetime import datetime, timedelta, timezone
|
||||||
|
from pathlib import Path
|
||||||
|
from typing import Any, Dict, List, Optional, Tuple
|
||||||
|
|
||||||
|
from .projection import compute_projection, ConfidenceState, DISCLOSURES
|
||||||
|
from .store import SCHEMA_VERSION
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Freshness constants (§8.9, §8.2)
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
CADENCE_DEFAULT_S = 300 # 5 min
|
||||||
|
ACCURACY_SEC = 30
|
||||||
|
FRESH_THRESHOLD_S = 2 * CADENCE_DEFAULT_S + ACCURACY_SEC + 60 # 690 s
|
||||||
|
STALENESS_THRESHOLD_S = 48 * 3600 # 48 h
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Configuration reading (§8.3)
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
CONFIG_PATH = Path("/etc/fenris/fenris.conf")
|
||||||
|
|
||||||
|
|
||||||
|
def read_config() -> Dict[str, Any]:
|
||||||
|
"""Read the world-readable configuration file.
|
||||||
|
|
||||||
|
Returns a dict with at least 'device'.
|
||||||
|
Raises ConfigError with a reason string on any failure.
|
||||||
|
"""
|
||||||
|
if not CONFIG_PATH.exists():
|
||||||
|
raise ConfigError("configuration file not found at %s" % CONFIG_PATH)
|
||||||
|
|
||||||
|
try:
|
||||||
|
text = CONFIG_PATH.read_text()
|
||||||
|
except OSError as e:
|
||||||
|
raise ConfigError("cannot read %s: %s" % (CONFIG_PATH, e))
|
||||||
|
|
||||||
|
device = None
|
||||||
|
for line in text.splitlines():
|
||||||
|
line = line.strip()
|
||||||
|
if not line or line.startswith("#"):
|
||||||
|
continue
|
||||||
|
if "=" in line:
|
||||||
|
key, _, value = line.partition("=")
|
||||||
|
key = key.strip()
|
||||||
|
value = value.strip().strip('"').strip("'")
|
||||||
|
if key == "device":
|
||||||
|
device = value
|
||||||
|
break
|
||||||
|
|
||||||
|
if not device:
|
||||||
|
raise ConfigError("no device selector in %s" % CONFIG_PATH)
|
||||||
|
|
||||||
|
return {"device": device}
|
||||||
|
|
||||||
|
|
||||||
|
class ConfigError(Exception):
|
||||||
|
"""Configuration is invalid — surfaced in status as a fact (§8.3)."""
|
||||||
|
pass
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Store opening (read-only, §3, §9.4, §9.5)
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
def open_store_readonly(store_path: Path) -> sqlite3.Connection:
|
||||||
|
"""Open the observation store read-only.
|
||||||
|
|
||||||
|
Raises StoreFault if unreadable, NewerSchema if user_version > SCHEMA_VERSION.
|
||||||
|
"""
|
||||||
|
if not store_path.exists():
|
||||||
|
raise StoreFault("observation store not found at %s" % store_path)
|
||||||
|
|
||||||
|
try:
|
||||||
|
conn = sqlite3.connect("file:%s?mode=ro" % store_path, uri=True)
|
||||||
|
conn.row_factory = sqlite3.Row
|
||||||
|
except sqlite3.Error as e:
|
||||||
|
raise StoreFault("observation store unreadable: %s" % e)
|
||||||
|
|
||||||
|
try:
|
||||||
|
cursor = conn.execute("PRAGMA user_version")
|
||||||
|
version = cursor.fetchone()[0]
|
||||||
|
except sqlite3.Error as e:
|
||||||
|
conn.close()
|
||||||
|
raise StoreFault("observation store unreadable: %s" % e)
|
||||||
|
|
||||||
|
if version > SCHEMA_VERSION:
|
||||||
|
conn.close()
|
||||||
|
raise NewerSchema(version)
|
||||||
|
|
||||||
|
return conn
|
||||||
|
|
||||||
|
|
||||||
|
class StoreFault(Exception):
|
||||||
|
"""Store is present but unreadable or corrupt (§9.4)."""
|
||||||
|
pass
|
||||||
|
|
||||||
|
|
||||||
|
class NewerSchema(Exception):
|
||||||
|
"""Store has a newer user_version (§9.5)."""
|
||||||
|
def __init__(self, version: int):
|
||||||
|
self.version = version
|
||||||
|
super().__init__("schema version %d" % version)
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Service state queries (§8.8 — allow-listed systemctl properties)
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
def _systemctl_show(unit: str, *properties: str) -> Dict[str, str]:
|
||||||
|
"""Query systemctl show for specific properties. Returns empty dict on failure."""
|
||||||
|
try:
|
||||||
|
result = subprocess.run(
|
||||||
|
["systemctl", "show", unit, "--property=" + ",".join(properties)],
|
||||||
|
capture_output=True, text=True, timeout=5,
|
||||||
|
)
|
||||||
|
if result.returncode != 0:
|
||||||
|
return {}
|
||||||
|
out = {}
|
||||||
|
for line in result.stdout.splitlines():
|
||||||
|
if "=" in line:
|
||||||
|
key, _, value = line.partition("=")
|
||||||
|
out[key.strip()] = value.strip()
|
||||||
|
return out
|
||||||
|
except (subprocess.TimeoutExpired, FileNotFoundError, OSError):
|
||||||
|
return {}
|
||||||
|
|
||||||
|
|
||||||
|
def _journalctl_hint(unit: str, lines: int = 5) -> Optional[str]:
|
||||||
|
"""Get the last N journal lines for a unit. Returns None on failure."""
|
||||||
|
try:
|
||||||
|
result = subprocess.run(
|
||||||
|
["journalctl", "-u", unit, "--no-pager", "-n", str(lines), "--output=short-iso"],
|
||||||
|
capture_output=True, text=True, timeout=5,
|
||||||
|
)
|
||||||
|
if result.returncode != 0 or not result.stdout.strip():
|
||||||
|
return None
|
||||||
|
return result.stdout.strip()
|
||||||
|
except (subprocess.TimeoutExpired, FileNotFoundError, OSError):
|
||||||
|
return None
|
||||||
|
|
||||||
|
|
||||||
|
def query_service_state() -> Dict[str, Any]:
|
||||||
|
"""Query systemctl for the four separate service facts (§7.3, LC-9).
|
||||||
|
|
||||||
|
Returns dict with keys:
|
||||||
|
boot_enabled: bool
|
||||||
|
timer_active: bool
|
||||||
|
last_collect_ok: Optional[bool]
|
||||||
|
last_collect_age_s: Optional[int]
|
||||||
|
last_collect_reason: Optional[str]
|
||||||
|
"""
|
||||||
|
timer_props = _systemctl_show(
|
||||||
|
"fenris-collect.timer",
|
||||||
|
"UnitFileState", "ActiveState", "LastTriggerUSec",
|
||||||
|
)
|
||||||
|
service_props = _systemctl_show(
|
||||||
|
"fenris-collect.service",
|
||||||
|
"ActiveState", "ExecMainStatus", "ExecMainExitTimestamp",
|
||||||
|
)
|
||||||
|
|
||||||
|
boot_enabled_str = timer_props.get("UnitFileState", "")
|
||||||
|
boot_enabled = boot_enabled_str == "enabled"
|
||||||
|
|
||||||
|
active_state = timer_props.get("ActiveState", "inactive")
|
||||||
|
timer_active = active_state == "active"
|
||||||
|
|
||||||
|
last_collect_ok = None
|
||||||
|
last_collect_age_s = None
|
||||||
|
last_collect_reason = None
|
||||||
|
|
||||||
|
last_trigger = timer_props.get("LastTriggerUSec", "")
|
||||||
|
if last_trigger and last_trigger != "n/a":
|
||||||
|
try:
|
||||||
|
trigger_dt = datetime.fromisoformat(last_trigger.replace("Z", "+00:00"))
|
||||||
|
now = datetime.now(timezone.utc)
|
||||||
|
last_collect_age_s = int((now - trigger_dt).total_seconds())
|
||||||
|
except (ValueError, TypeError):
|
||||||
|
pass
|
||||||
|
|
||||||
|
exec_status = service_props.get("ExecMainStatus", "")
|
||||||
|
if exec_status:
|
||||||
|
try:
|
||||||
|
exit_code = int(exec_status)
|
||||||
|
last_collect_ok = exit_code == 0
|
||||||
|
if exit_code != 0:
|
||||||
|
last_collect_reason = "exit code %d" % exit_code
|
||||||
|
except (ValueError, TypeError):
|
||||||
|
pass
|
||||||
|
|
||||||
|
return {
|
||||||
|
"boot_enabled": boot_enabled,
|
||||||
|
"timer_active": timer_active,
|
||||||
|
"last_collect_ok": last_collect_ok,
|
||||||
|
"last_collect_age_s": last_collect_age_s,
|
||||||
|
"last_collect_reason": last_collect_reason,
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Freshness grading (§8.9)
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
def grade_freshness(newest_sample_ts: Optional[str], clock_now: datetime) -> str:
|
||||||
|
"""Grade freshness from the newest sample timestamp (never a stored flag).
|
||||||
|
|
||||||
|
Returns 'fresh', 'missed', 'stale', or 'empty'.
|
||||||
|
"""
|
||||||
|
if newest_sample_ts is None:
|
||||||
|
return "empty"
|
||||||
|
|
||||||
|
try:
|
||||||
|
ts = datetime.fromisoformat(newest_sample_ts)
|
||||||
|
if ts.tzinfo is None:
|
||||||
|
ts = ts.replace(tzinfo=timezone.utc)
|
||||||
|
else:
|
||||||
|
ts = ts.astimezone(timezone.utc)
|
||||||
|
except (ValueError, TypeError):
|
||||||
|
return "empty"
|
||||||
|
|
||||||
|
age_s = (clock_now - ts).total_seconds()
|
||||||
|
|
||||||
|
if age_s <= FRESH_THRESHOLD_S:
|
||||||
|
return "fresh"
|
||||||
|
elif age_s < STALENESS_THRESHOLD_S:
|
||||||
|
return "missed"
|
||||||
|
else:
|
||||||
|
return "stale"
|
||||||
|
|
||||||
|
|
||||||
|
def freshness_age_human(age_s: Optional[int]) -> str:
|
||||||
|
"""Human-readable age string for freshness fact."""
|
||||||
|
if age_s is None:
|
||||||
|
return "unknown age"
|
||||||
|
if age_s < 60:
|
||||||
|
return "%ds ago" % age_s
|
||||||
|
if age_s < 3600:
|
||||||
|
return "%dm ago" % (age_s // 60)
|
||||||
|
if age_s < 86400:
|
||||||
|
return "%dh %dm ago" % (age_s // 3600, (age_s % 3600) // 60)
|
||||||
|
return "%dd ago" % (age_s // 86400)
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Drive anomalies (§9.7 — FL-7)
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
def _query_drive_facts(conn: sqlite3.Connection) -> List[str]:
|
||||||
|
"""Query drive-reported anomalies from the latest sample (§9.7, FL-7).
|
||||||
|
|
||||||
|
critical_warning, media errors, and unsafe shutdowns render as ordinary
|
||||||
|
facts and never affect the projection.
|
||||||
|
"""
|
||||||
|
cursor = conn.execute(
|
||||||
|
"SELECT critical_warning, media_errors, unsafe_shutdowns, "
|
||||||
|
"temperature_c, available_spare "
|
||||||
|
"FROM samples ORDER BY id DESC LIMIT 1"
|
||||||
|
)
|
||||||
|
row = cursor.fetchone()
|
||||||
|
if row is None:
|
||||||
|
return []
|
||||||
|
|
||||||
|
facts = []
|
||||||
|
cw = row[0]
|
||||||
|
if cw and cw != 0:
|
||||||
|
facts.append("critical warning: %s" % hex(cw) if isinstance(cw, int) else str(cw))
|
||||||
|
me = row[1]
|
||||||
|
if me and me > 0:
|
||||||
|
facts.append("media errors: %d" % me)
|
||||||
|
us = row[2]
|
||||||
|
if us and us > 0:
|
||||||
|
facts.append("unsafe shutdowns: %d" % us)
|
||||||
|
return facts
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Retired command rejection (§8.8)
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
RETIRED_COMMANDS = {
|
||||||
|
"start": "use 'fenris monitor resume' to enable monitoring",
|
||||||
|
"stop": "use 'fenris monitor pause' to disable monitoring",
|
||||||
|
"run": "use 'fenris monitor resume' to enable monitoring; the timer runs in the background",
|
||||||
|
}
|
||||||
|
|
||||||
|
MIGRATION_POINTERS = {
|
||||||
|
"--device": "device is configured in /etc/fenris/fenris.conf",
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
def check_retired_command(cmd: str) -> Optional[str]:
|
||||||
|
"""Check if a command is retired and return the migration pointer, or None."""
|
||||||
|
return RETIRED_COMMANDS.get(cmd)
|
||||||
|
|
||||||
|
|
||||||
|
def check_retired_flag(flag: str) -> Optional[str]:
|
||||||
|
"""Check if a flag is retired and return the migration pointer, or None."""
|
||||||
|
return MIGRATION_POINTERS.get(flag)
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Formatting
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
def _format_projection(proj, freshness: str, service: Dict[str, Any],
|
||||||
|
drive_facts: List[str], config_error: Optional[str],
|
||||||
|
store_fault: Optional[str], newer_schema: Optional[str],
|
||||||
|
journal_hint: Optional[str]) -> str:
|
||||||
|
"""Format the complete status output."""
|
||||||
|
lines = []
|
||||||
|
|
||||||
|
# --- Store/system fault overrides (§9.4, §9.5) ---
|
||||||
|
if store_fault:
|
||||||
|
lines.append("observation store unreadable")
|
||||||
|
if journal_hint:
|
||||||
|
lines.append("")
|
||||||
|
lines.append("Recent journal entries:")
|
||||||
|
lines.append(journal_hint)
|
||||||
|
return "\n".join(lines)
|
||||||
|
|
||||||
|
if newer_schema:
|
||||||
|
lines.append("observation store written by a newer Fenris — upgrade Fenris")
|
||||||
|
return "\n".join(lines)
|
||||||
|
|
||||||
|
# --- Configuration error (§8.3) ---
|
||||||
|
if config_error:
|
||||||
|
lines.append("configuration error: %s" % config_error)
|
||||||
|
lines.append("")
|
||||||
|
|
||||||
|
# --- Empty store (§8.9) ---
|
||||||
|
if freshness == "empty":
|
||||||
|
lines.append("no observations yet")
|
||||||
|
lines.append("")
|
||||||
|
lines.append("Enable monitoring: fenris monitor resume")
|
||||||
|
_append_service_facts(lines, service)
|
||||||
|
return "\n".join(lines)
|
||||||
|
|
||||||
|
# --- Projection headline ---
|
||||||
|
headline = _format_headline(proj)
|
||||||
|
lines.append(headline)
|
||||||
|
lines.append("")
|
||||||
|
|
||||||
|
# --- Confidence state + contributing facts (§6.7, §6.11) ---
|
||||||
|
state_label = proj.confidence_state.value
|
||||||
|
if proj.contributing_facts:
|
||||||
|
facts_str = " · ".join(proj.contributing_facts)
|
||||||
|
lines.append("%s evidence · %s" % (state_label, facts_str))
|
||||||
|
else:
|
||||||
|
lines.append("%s evidence" % state_label)
|
||||||
|
lines.append("")
|
||||||
|
|
||||||
|
# --- Scenario range (§6.5) ---
|
||||||
|
if proj.scenario_range and proj.scenario_range.rates:
|
||||||
|
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))
|
||||||
|
lines.append("scenario range: %s" % " · ".join(parts))
|
||||||
|
lines.append("")
|
||||||
|
|
||||||
|
# --- PU context line (§6.1) ---
|
||||||
|
lines.append(proj.pu_context_line)
|
||||||
|
lines.append("")
|
||||||
|
|
||||||
|
# --- Drive anomalies (§9.7, FL-7) ---
|
||||||
|
if drive_facts:
|
||||||
|
for fact in drive_facts:
|
||||||
|
lines.append(fact)
|
||||||
|
lines.append("")
|
||||||
|
|
||||||
|
# --- Four separate service facts (§7.3, LC-9) ---
|
||||||
|
_append_service_facts(lines, service)
|
||||||
|
|
||||||
|
# --- Journal hint on failure or staleness (§8.8) ---
|
||||||
|
if journal_hint:
|
||||||
|
if freshness in ("missed", "stale"):
|
||||||
|
lines.append("")
|
||||||
|
lines.append("Recent journal entries:")
|
||||||
|
lines.append(journal_hint)
|
||||||
|
|
||||||
|
return "\n".join(lines)
|
||||||
|
|
||||||
|
|
||||||
|
def _format_headline(proj) -> str:
|
||||||
|
"""Format the lifespan headline or its no-projection wording (§6.11)."""
|
||||||
|
if proj.headline_remaining_seconds is None:
|
||||||
|
if proj.zero_rate_fact:
|
||||||
|
return "no finite projection from this history"
|
||||||
|
if proj.warming_fact:
|
||||||
|
return proj.warming_fact
|
||||||
|
return "no projection available"
|
||||||
|
|
||||||
|
secs = proj.headline_remaining_seconds
|
||||||
|
if secs <= 0:
|
||||||
|
return "endurance exhausted"
|
||||||
|
|
||||||
|
# Human-readable time
|
||||||
|
years = int(secs // 31557600)
|
||||||
|
rem = secs % 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)
|
||||||
|
|
||||||
|
remaining_human = " ".join(parts)
|
||||||
|
|
||||||
|
# Regime line
|
||||||
|
regime_parts = []
|
||||||
|
if proj.regime_days:
|
||||||
|
regime_parts.append("sustained regime: %d days" % proj.regime_days)
|
||||||
|
|
||||||
|
headline = "%s remaining" % remaining_human
|
||||||
|
if regime_parts:
|
||||||
|
headline += " · %s" % " · ".join(regime_parts)
|
||||||
|
|
||||||
|
return headline
|
||||||
|
|
||||||
|
|
||||||
|
def _append_service_facts(lines: List[str], service: Dict[str, Any]) -> None:
|
||||||
|
"""Append the four separate service facts (§7.3, LC-9)."""
|
||||||
|
boot = "enabled" if service.get("boot_enabled") else "disabled"
|
||||||
|
activity = "active" if service.get("timer_active") else "inactive"
|
||||||
|
|
||||||
|
if service.get("last_collect_ok") is True:
|
||||||
|
collect = "ok"
|
||||||
|
elif service.get("last_collect_ok") is False:
|
||||||
|
collect = "FAILED"
|
||||||
|
if service.get("last_collect_reason"):
|
||||||
|
collect += " (%s)" % service["last_collect_reason"]
|
||||||
|
else:
|
||||||
|
collect = "unknown"
|
||||||
|
|
||||||
|
collect_age = ""
|
||||||
|
if service.get("last_collect_age_s") is not None:
|
||||||
|
collect_age = " %s" % freshness_age_human(service["last_collect_age_s"])
|
||||||
|
|
||||||
|
freshness_str = service.get("freshness", "unknown")
|
||||||
|
freshness_age = ""
|
||||||
|
if service.get("freshness_age_s") is not None:
|
||||||
|
freshness_age = " (%s)" % freshness_age_human(service["freshness_age_s"])
|
||||||
|
|
||||||
|
lines.append("boot: %s · timer: %s · last collect: %s%s · freshness: %s%s"
|
||||||
|
% (boot, activity, collect, collect_age, freshness_str, freshness_age))
|
||||||
|
|
||||||
|
|
||||||
|
def format_disclosures() -> str:
|
||||||
|
"""Format the six disclosures (§6.11, CI-4)."""
|
||||||
|
lines = []
|
||||||
|
lines.append("Disclosures")
|
||||||
|
lines.append("")
|
||||||
|
for i, disc in enumerate(DISCLOSURES, 1):
|
||||||
|
lines.append("%d. %s" % (i, disc))
|
||||||
|
return "\n".join(lines)
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Main status entry point
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
def get_status(store_path: Optional[Path] = None, clock_now: Optional[datetime] = None,
|
||||||
|
query_services: bool = True, query_journal: bool = True) -> str:
|
||||||
|
"""Render the complete read-only status (§8.8, LC-9).
|
||||||
|
|
||||||
|
This is the single entry point for 'fenris status'. It never auto-samples,
|
||||||
|
never prompts, and never writes to the store.
|
||||||
|
"""
|
||||||
|
if clock_now is None:
|
||||||
|
clock_now = datetime.now(timezone.utc)
|
||||||
|
|
||||||
|
# --- Configuration (§8.3) ---
|
||||||
|
config_error = None
|
||||||
|
device = None
|
||||||
|
try:
|
||||||
|
config = read_config()
|
||||||
|
device = config["device"]
|
||||||
|
except ConfigError as e:
|
||||||
|
config_error = str(e)
|
||||||
|
|
||||||
|
# --- Service state ---
|
||||||
|
service = {}
|
||||||
|
if query_services:
|
||||||
|
service = query_service_state()
|
||||||
|
|
||||||
|
# --- Store open ---
|
||||||
|
store_fault = None
|
||||||
|
newer_schema = None
|
||||||
|
conn = None
|
||||||
|
|
||||||
|
if store_path is None:
|
||||||
|
store_path = Path("/var/lib/fenris/observations.db")
|
||||||
|
|
||||||
|
try:
|
||||||
|
conn = open_store_readonly(store_path)
|
||||||
|
except StoreFault as e:
|
||||||
|
store_fault = str(e)
|
||||||
|
except NewerSchema as e:
|
||||||
|
newer_schema = str(e)
|
||||||
|
|
||||||
|
# --- Store fault / newer schema short-circuit ---
|
||||||
|
if store_fault or newer_schema:
|
||||||
|
journal_hint = None
|
||||||
|
if query_journal:
|
||||||
|
journal_hint = _journalctl_hint("fenris-collect.service")
|
||||||
|
service["freshness"] = "unknown"
|
||||||
|
service["freshness_age_s"] = None
|
||||||
|
return _format_projection(
|
||||||
|
None, "unknown", service, [], config_error, store_fault, newer_schema, journal_hint
|
||||||
|
)
|
||||||
|
|
||||||
|
# --- Freshness grading (§8.9) ---
|
||||||
|
try:
|
||||||
|
cursor = conn.execute("SELECT ts FROM samples ORDER BY id DESC LIMIT 1")
|
||||||
|
row = cursor.fetchone()
|
||||||
|
newest_ts = row[0] if row else None
|
||||||
|
except sqlite3.Error:
|
||||||
|
newest_ts = None
|
||||||
|
|
||||||
|
freshness = grade_freshness(newest_ts, clock_now)
|
||||||
|
|
||||||
|
# Freshness age for the service fact
|
||||||
|
freshness_age_s = None
|
||||||
|
if newest_ts:
|
||||||
|
try:
|
||||||
|
ts = datetime.fromisoformat(newest_ts)
|
||||||
|
if ts.tzinfo is None:
|
||||||
|
ts = ts.replace(tzinfo=timezone.utc)
|
||||||
|
freshness_age_s = int((clock_now - ts).total_seconds())
|
||||||
|
except (ValueError, TypeError):
|
||||||
|
pass
|
||||||
|
|
||||||
|
service["freshness"] = freshness
|
||||||
|
service["freshness_age_s"] = freshness_age_s
|
||||||
|
|
||||||
|
# --- Drive anomalies (§9.7, FL-7) ---
|
||||||
|
drive_facts = []
|
||||||
|
try:
|
||||||
|
drive_facts = _query_drive_facts(conn)
|
||||||
|
except sqlite3.Error:
|
||||||
|
pass
|
||||||
|
|
||||||
|
# --- Projection (§6 — recomputed on read, never stored) ---
|
||||||
|
try:
|
||||||
|
proj = compute_projection(conn, clock_now)
|
||||||
|
except Exception:
|
||||||
|
proj = None
|
||||||
|
|
||||||
|
# --- Journal hint on failure or staleness (§8.8) ---
|
||||||
|
journal_hint = None
|
||||||
|
if query_journal and freshness in ("missed", "stale"):
|
||||||
|
journal_hint = _journalctl_hint("fenris-collect.service")
|
||||||
|
|
||||||
|
# --- Compose output ---
|
||||||
|
result = _format_projection(
|
||||||
|
proj, freshness, service, drive_facts, config_error,
|
||||||
|
None, None, journal_hint,
|
||||||
|
)
|
||||||
|
|
||||||
|
conn.close()
|
||||||
|
return result
|
||||||
|
|
||||||
|
|
||||||
|
def render_status(store_path: Optional[Path] = None, clock_now: Optional[datetime] = None,
|
||||||
|
query_services: bool = True, query_journal: bool = True,
|
||||||
|
show_disclosures: bool = False) -> str:
|
||||||
|
"""High-level status renderer: status + optional disclosures.
|
||||||
|
|
||||||
|
Used by the CLI entry point.
|
||||||
|
"""
|
||||||
|
parts = [get_status(store_path, clock_now, query_services, query_journal)]
|
||||||
|
if show_disclosures:
|
||||||
|
parts.append("")
|
||||||
|
parts.append(format_disclosures())
|
||||||
|
return "\n\n".join(parts)
|
||||||
@@ -338,3 +338,564 @@ class TestArithmetic:
|
|||||||
|
|
||||||
if r_slow.headline_remaining_seconds is not None and r_fast.headline_remaining_seconds is not None:
|
if r_slow.headline_remaining_seconds is not None and r_fast.headline_remaining_seconds is not None:
|
||||||
assert r_fast.headline_remaining_seconds < r_slow.headline_remaining_seconds
|
assert r_fast.headline_remaining_seconds < r_slow.headline_remaining_seconds
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
# ===========================================================================
|
||||||
|
# Issue #26: Project from the sustained regime
|
||||||
|
# Habit change, scenario range, and evidence gates
|
||||||
|
# ===========================================================================
|
||||||
|
|
||||||
|
|
||||||
|
class TestSustainedRegimeRate:
|
||||||
|
"""PR-2: Headline rate is sustained-regime rate; default regime = full
|
||||||
|
history capped at 90 days; scenario range computed independently,
|
||||||
|
covered horizons only, no placeholders."""
|
||||||
|
|
||||||
|
def test_headline_rate_from_regime(self, store):
|
||||||
|
"""Rate is regime DUW / wall-clock, not trailing-24h or all-history."""
|
||||||
|
_insert_baseline(store, tbw_tb=10.0, verified=True)
|
||||||
|
_insert_segment(store, opened_at="2026-09-01T00:00:00+00:00")
|
||||||
|
_open_period(store, start="2026-09-01T00:00:00+00:00")
|
||||||
|
# 30 days of 100 MiB/day
|
||||||
|
bw = 100 * 1024 * 1024
|
||||||
|
for i in range(30):
|
||||||
|
d = (datetime(2026, 9, 1) + timedelta(days=i)).strftime("%Y-%m-%d")
|
||||||
|
_insert_day(store, d, bw=bw)
|
||||||
|
_insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5)
|
||||||
|
result = compute_projection(store, _clock())
|
||||||
|
# Regime = full 30 days; rate = 30*bw / wall-clock
|
||||||
|
regime_bytes = 30 * bw
|
||||||
|
period_start = datetime(2026, 9, 1, 0, 0, 0, tzinfo=timezone.utc)
|
||||||
|
wc = int((_clock() - period_start).total_seconds())
|
||||||
|
expected_rate = regime_bytes / wc
|
||||||
|
if result.headline_remaining_seconds is not None:
|
||||||
|
E = 10.0 * TBW_TO_BYTES
|
||||||
|
expected_seconds = max(E - regime_bytes, 0) / expected_rate
|
||||||
|
assert abs(result.headline_remaining_seconds - expected_seconds) < 1.0
|
||||||
|
|
||||||
|
def test_regime_capped_at_90_days(self, store):
|
||||||
|
"""Default regime is full history capped at 90 days."""
|
||||||
|
_insert_baseline(store, tbw_tb=100.0, verified=True)
|
||||||
|
_insert_segment(store, opened_at="2026-06-01T00:00:00+00:00")
|
||||||
|
_open_period(store, start="2026-06-01T00:00:00+00:00")
|
||||||
|
bw = 100 * 1024 * 1024
|
||||||
|
# 120 days of data (Jun 1 - Sep 28)
|
||||||
|
for i in range(120):
|
||||||
|
d = (datetime(2026, 6, 1) + timedelta(days=i)).strftime("%Y-%m-%d")
|
||||||
|
_insert_day(store, d, bw=bw)
|
||||||
|
_insert_sample(store, "2026-09-28T12:00:00+00:00", pu=5)
|
||||||
|
result = compute_projection(store, _clock(year=2026, month=9, day=30, hour=12))
|
||||||
|
# Regime should be capped at 90 days (from Jun 1 to Sep 30 = 90 days at cutoff)
|
||||||
|
# The 90-day cutoff is Sep 30 - 90 = Jul 1, so regime starts Jul 1
|
||||||
|
assert result.regime_days is not None
|
||||||
|
assert result.regime_days <= 90
|
||||||
|
|
||||||
|
def test_scenario_range_independent_of_regime(self, store):
|
||||||
|
"""Scenario range is computed independently from the regime."""
|
||||||
|
_insert_baseline(store, tbw_tb=10.0, verified=True)
|
||||||
|
_insert_segment(store, opened_at="2026-09-01T00:00:00+00:00")
|
||||||
|
_open_period(store, start="2026-09-01T00:00:00+00:00")
|
||||||
|
bw = 100 * 1024 * 1024
|
||||||
|
for i in range(30):
|
||||||
|
d = (datetime(2026, 9, 1) + timedelta(days=i)).strftime("%Y-%m-%d")
|
||||||
|
_insert_day(store, d, bw=bw)
|
||||||
|
_insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5)
|
||||||
|
result = compute_projection(store, _clock())
|
||||||
|
if result.scenario_range is not None:
|
||||||
|
# Should have 7-day and 28-day horizons (90-day not fully covered)
|
||||||
|
assert 7 in result.scenario_range.rates
|
||||||
|
assert 28 in result.scenario_range.rates
|
||||||
|
|
||||||
|
def test_only_covered_horizons_shown(self, store):
|
||||||
|
"""No placeholder horizons — only horizons the history covers."""
|
||||||
|
_insert_baseline(store, tbw_tb=10.0, verified=True)
|
||||||
|
_insert_segment(store, opened_at="2026-09-20T00:00:00+00:00")
|
||||||
|
_open_period(store, start="2026-09-20T00:00:00+00:00")
|
||||||
|
bw = 100 * 1024 * 1024
|
||||||
|
# Only 10 days of data
|
||||||
|
for i in range(10):
|
||||||
|
d = (datetime(2026, 9, 20) + timedelta(days=i)).strftime("%Y-%m-%d")
|
||||||
|
_insert_day(store, d, bw=bw)
|
||||||
|
_insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5)
|
||||||
|
result = compute_projection(store, _clock())
|
||||||
|
if result.scenario_range is not None:
|
||||||
|
# 7-day is covered, 28-day and 90-day are not
|
||||||
|
assert 7 in result.scenario_range.rates
|
||||||
|
assert 28 not in result.scenario_range.rates
|
||||||
|
assert 90 not in result.scenario_range.rates
|
||||||
|
|
||||||
|
|
||||||
|
class TestHabitChange:
|
||||||
|
"""PR-3: Habit change triggers at 2x/0.5x sustained 3 consecutive days,
|
||||||
|
regime starts at first divergence day, auto-adopted and labeled;
|
||||||
|
young regime caps at Limited."""
|
||||||
|
|
||||||
|
def test_habit_change_2x_detected(self, store):
|
||||||
|
"""2x increase for 3+ consecutive days triggers habit change."""
|
||||||
|
_insert_baseline(store, tbw_tb=10.0, verified=True)
|
||||||
|
_insert_segment(store, opened_at="2026-08-01T00:00:00+00:00")
|
||||||
|
_open_period(store, start="2026-08-01T00:00:00+00:00")
|
||||||
|
bw_normal = 100 * 1024 * 1024
|
||||||
|
bw_high = 300 * 1024 * 1024 # 3x the normal rate
|
||||||
|
# 28 days of normal usage
|
||||||
|
for i in range(28):
|
||||||
|
d = (datetime(2026, 8, 1) + timedelta(days=i)).strftime("%Y-%m-%d")
|
||||||
|
_insert_day(store, d, bw=bw_normal)
|
||||||
|
# 10 days of high usage (3x > 2x threshold)
|
||||||
|
for i in range(10):
|
||||||
|
d = (datetime(2026, 8, 29) + timedelta(days=i)).strftime("%Y-%m-%d")
|
||||||
|
_insert_day(store, d, bw=bw_high)
|
||||||
|
_insert_sample(store, "2026-09-08T10:00:00+00:00", pu=5)
|
||||||
|
result = compute_projection(store, _clock(year=2026, month=9, day=8, hour=12))
|
||||||
|
assert result.habit_change_fact is not None
|
||||||
|
assert "usage habit changed" in result.habit_change_fact
|
||||||
|
assert "days ago" in result.habit_change_fact
|
||||||
|
|
||||||
|
def test_habit_change_05x_detected(self, store):
|
||||||
|
"""0.5x decrease for 3+ consecutive days triggers habit change."""
|
||||||
|
_insert_baseline(store, tbw_tb=10.0, verified=True)
|
||||||
|
_insert_segment(store, opened_at="2026-08-01T00:00:00+00:00")
|
||||||
|
_open_period(store, start="2026-08-01T00:00:00+00:00")
|
||||||
|
bw_high = 400 * 1024 * 1024
|
||||||
|
bw_low = 100 * 1024 * 1024 # 0.25x < 0.5x threshold
|
||||||
|
# 28 days of high usage
|
||||||
|
for i in range(28):
|
||||||
|
d = (datetime(2026, 8, 1) + timedelta(days=i)).strftime("%Y-%m-%d")
|
||||||
|
_insert_day(store, d, bw=bw_high)
|
||||||
|
# 10 days of low usage
|
||||||
|
for i in range(10):
|
||||||
|
d = (datetime(2026, 8, 29) + timedelta(days=i)).strftime("%Y-%m-%d")
|
||||||
|
_insert_day(store, d, bw=bw_low)
|
||||||
|
_insert_sample(store, "2026-09-08T10:00:00+00:00", pu=5)
|
||||||
|
result = compute_projection(store, _clock(year=2026, month=9, day=8, hour=12))
|
||||||
|
assert result.habit_change_fact is not None
|
||||||
|
assert "usage habit changed" in result.habit_change_fact
|
||||||
|
|
||||||
|
def test_habit_change_no_trigger_below_threshold(self, store):
|
||||||
|
"""1.5x increase does NOT trigger habit change (below 2x threshold)."""
|
||||||
|
_insert_baseline(store, tbw_tb=10.0, verified=True)
|
||||||
|
_insert_segment(store, opened_at="2026-08-01T00:00:00+00:00")
|
||||||
|
_open_period(store, start="2026-08-01T00:00:00+00:00")
|
||||||
|
bw_normal = 100 * 1024 * 1024
|
||||||
|
bw_moderate = 150 * 1024 * 1024 # 1.5x < 2x threshold
|
||||||
|
for i in range(28):
|
||||||
|
d = (datetime(2026, 8, 1) + timedelta(days=i)).strftime("%Y-%m-%d")
|
||||||
|
_insert_day(store, d, bw=bw_normal)
|
||||||
|
for i in range(10):
|
||||||
|
d = (datetime(2026, 8, 29) + timedelta(days=i)).strftime("%Y-%m-%d")
|
||||||
|
_insert_day(store, d, bw=bw_moderate)
|
||||||
|
_insert_sample(store, "2026-09-08T10:00:00+00:00", pu=5)
|
||||||
|
result = compute_projection(store, _clock(year=2026, month=9, day=8, hour=12))
|
||||||
|
assert result.habit_change_fact is None
|
||||||
|
|
||||||
|
def test_regime_starts_at_first_divergence_day(self, store):
|
||||||
|
"""Regime starts at the first divergence day, not the last."""
|
||||||
|
_insert_baseline(store, tbw_tb=10.0, verified=True)
|
||||||
|
_insert_segment(store, opened_at="2026-08-01T00:00:00+00:00")
|
||||||
|
_open_period(store, start="2026-08-01T00:00:00+00:00")
|
||||||
|
bw_normal = 100 * 1024 * 1024
|
||||||
|
bw_high = 300 * 1024 * 1024
|
||||||
|
for i in range(28):
|
||||||
|
d = (datetime(2026, 8, 1) + timedelta(days=i)).strftime("%Y-%m-%d")
|
||||||
|
_insert_day(store, d, bw=bw_normal)
|
||||||
|
for i in range(10):
|
||||||
|
d = (datetime(2026, 8, 29) + timedelta(days=i)).strftime("%Y-%m-%d")
|
||||||
|
_insert_day(store, d, bw=bw_high)
|
||||||
|
_insert_sample(store, "2026-09-08T10:00:00+00:00", pu=5)
|
||||||
|
result = compute_projection(store, _clock(year=2026, month=9, day=8, hour=12))
|
||||||
|
if result.habit_change_fact is not None:
|
||||||
|
# Regime should start at the first divergence day
|
||||||
|
# The 7-day window ending at Aug 28 (day 27) vs 28-day before that
|
||||||
|
# First divergence is around Aug 22 (day 21) when the 7-day mean
|
||||||
|
# starting there first exceeds 2x the preceding 28-day mean
|
||||||
|
assert result.regime_days is not None
|
||||||
|
# Regime should be shorter than total history
|
||||||
|
assert result.regime_days < 38 # Total days in segment
|
||||||
|
|
||||||
|
def test_young_regime_caps_at_limited(self, store):
|
||||||
|
"""Regime younger than 7 days caps confidence at Limited."""
|
||||||
|
_insert_baseline(store, tbw_tb=10.0, verified=True)
|
||||||
|
_insert_segment(store, opened_at="2026-09-01T00:00:00+00:00")
|
||||||
|
_open_period(store, start="2026-09-01T00:00:00+00:00")
|
||||||
|
bw = 100 * 1024 * 1024
|
||||||
|
# Only 5 days of data (young regime)
|
||||||
|
for i in range(5):
|
||||||
|
d = (datetime(2026, 9, 25) + timedelta(days=i)).strftime("%Y-%m-%d")
|
||||||
|
_insert_day(store, d, bw=bw)
|
||||||
|
_insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5)
|
||||||
|
result = compute_projection(store, _clock())
|
||||||
|
assert result.confidence_state == ConfidenceState.LIMITED
|
||||||
|
assert any("regime only" in f and "days old" in f for f in result.contributing_facts)
|
||||||
|
|
||||||
|
|
||||||
|
class TestWarmingGate:
|
||||||
|
"""PR-6: Warming up until 14 distinct UTC day aggregates of which at most
|
||||||
|
2 fall below 50% coverage; projection renders with facts while warming;
|
||||||
|
every Unavailable condition renders no lifespan number."""
|
||||||
|
|
||||||
|
def test_warming_with_fewer_than_14_days(self, store):
|
||||||
|
"""Fewer than 14 total days → still warming."""
|
||||||
|
_insert_baseline(store, tbw_tb=10.0, verified=True)
|
||||||
|
_insert_segment(store, opened_at="2026-09-20T00:00:00+00:00")
|
||||||
|
_open_period(store, start="2026-09-20T00:00:00+00:00")
|
||||||
|
bw = 100 * 1024 * 1024
|
||||||
|
for i in range(10):
|
||||||
|
d = (datetime(2026, 9, 20) + timedelta(days=i)).strftime("%Y-%m-%d")
|
||||||
|
_insert_day(store, d, bw=bw, coverage=0.95)
|
||||||
|
_insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5)
|
||||||
|
result = compute_projection(store, _clock())
|
||||||
|
assert result.warming_fact is not None
|
||||||
|
assert "warming up" in result.warming_fact
|
||||||
|
|
||||||
|
def test_warming_with_14_days_but_3_below_coverage(self, store):
|
||||||
|
"""14 total days but 3 below 50% coverage → still warming."""
|
||||||
|
_insert_baseline(store, tbw_tb=10.0, verified=True)
|
||||||
|
_insert_segment(store, opened_at="2026-09-17T00:00:00+00:00")
|
||||||
|
_open_period(store, start="2026-09-17T00:00:00+00:00")
|
||||||
|
bw = 100 * 1024 * 1024
|
||||||
|
for i in range(14):
|
||||||
|
d = (datetime(2026, 9, 17) + timedelta(days=i)).strftime("%Y-%m-%d")
|
||||||
|
# 3 days with low coverage
|
||||||
|
cov = 0.30 if i < 3 else 0.95
|
||||||
|
_insert_day(store, d, bw=bw, coverage=cov)
|
||||||
|
_insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5)
|
||||||
|
result = compute_projection(store, _clock())
|
||||||
|
assert result.warming_fact is not None
|
||||||
|
assert "warming up" in result.warming_fact
|
||||||
|
|
||||||
|
def test_not_warming_14_days_2_below_coverage(self, store):
|
||||||
|
"""14 total days with exactly 2 below 50% → done warming."""
|
||||||
|
_insert_baseline(store, tbw_tb=10.0, verified=True)
|
||||||
|
_insert_segment(store, opened_at="2026-09-17T00:00:00+00:00")
|
||||||
|
_open_period(store, start="2026-09-17T00:00:00+00:00")
|
||||||
|
bw = 100 * 1024 * 1024
|
||||||
|
for i in range(14):
|
||||||
|
d = (datetime(2026, 9, 17) + timedelta(days=i)).strftime("%Y-%m-%d")
|
||||||
|
cov = 0.30 if i < 2 else 0.95
|
||||||
|
_insert_day(store, d, bw=bw, coverage=cov)
|
||||||
|
_insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5)
|
||||||
|
result = compute_projection(store, _clock())
|
||||||
|
assert result.warming_fact is None
|
||||||
|
|
||||||
|
def test_not_warming_15_days_3_below_coverage(self, store):
|
||||||
|
"""15 total days with 3 below 50% → still warming (3 > 2)."""
|
||||||
|
_insert_baseline(store, tbw_tb=10.0, verified=True)
|
||||||
|
_insert_segment(store, opened_at="2026-09-16T00:00:00+00:00")
|
||||||
|
_open_period(store, start="2026-09-16T00:00:00+00:00")
|
||||||
|
bw = 100 * 1024 * 1024
|
||||||
|
for i in range(15):
|
||||||
|
d = (datetime(2026, 9, 16) + timedelta(days=i)).strftime("%Y-%m-%d")
|
||||||
|
cov = 0.30 if i < 3 else 0.95
|
||||||
|
_insert_day(store, d, bw=bw, coverage=cov)
|
||||||
|
_insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5)
|
||||||
|
result = compute_projection(store, _clock())
|
||||||
|
assert result.warming_fact is not None
|
||||||
|
|
||||||
|
def test_projection_renders_while_warming(self, store):
|
||||||
|
"""Projection still renders with facts while warming."""
|
||||||
|
_insert_baseline(store, tbw_tb=10.0, verified=True)
|
||||||
|
_insert_segment(store, opened_at="2026-09-20T00:00:00+00:00")
|
||||||
|
_open_period(store, start="2026-09-20T00:00:00+00:00")
|
||||||
|
bw = 100 * 1024 * 1024
|
||||||
|
for i in range(10):
|
||||||
|
d = (datetime(2026, 9, 20) + timedelta(days=i)).strftime("%Y-%m-%d")
|
||||||
|
_insert_day(store, d, bw=bw, coverage=0.95)
|
||||||
|
_insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5)
|
||||||
|
result = compute_projection(store, _clock())
|
||||||
|
# Should have warming fact but still render
|
||||||
|
assert result.warming_fact is not None
|
||||||
|
assert result.contributing_facts is not None
|
||||||
|
assert len(result.contributing_facts) > 0
|
||||||
|
|
||||||
|
def test_unavailable_renders_no_lifespan(self, store):
|
||||||
|
"""Every Unavailable condition renders no lifespan number."""
|
||||||
|
# No baseline → Unavailable
|
||||||
|
_insert_segment(store)
|
||||||
|
_open_period(store)
|
||||||
|
_insert_day(store, "2026-09-28", bw=100*1024*1024)
|
||||||
|
result = compute_projection(store, _clock())
|
||||||
|
assert result.confidence_state == ConfidenceState.UNSUPPORTED
|
||||||
|
assert result.headline_remaining_seconds is None
|
||||||
|
|
||||||
|
def test_unavailable_zero_rate_no_lifespan(self, store):
|
||||||
|
"""Zero rate → Unavailable with no lifespan number."""
|
||||||
|
_insert_baseline(store, tbw_tb=1.0, verified=True)
|
||||||
|
_insert_segment(store)
|
||||||
|
_open_period(store)
|
||||||
|
for i in range(20):
|
||||||
|
d = (datetime(2026, 9, 10) + timedelta(days=i)).strftime("%Y-%m-%d")
|
||||||
|
_insert_day(store, d, bw=0)
|
||||||
|
result = compute_projection(store, _clock())
|
||||||
|
assert result.confidence_state == ConfidenceState.UNSUPPORTED
|
||||||
|
assert result.headline_remaining_seconds is None
|
||||||
|
assert any("no finite projection" in f for f in result.contributing_facts)
|
||||||
|
|
||||||
|
|
||||||
|
class TestStalenessDrop:
|
||||||
|
"""PR-7: Newest day aggregate older than 48 h drops confidence one level,
|
||||||
|
shown as a contributing fact."""
|
||||||
|
|
||||||
|
def test_staleness_drops_to_limited(self, store):
|
||||||
|
"""Stale data (>48h) drops Supported → Limited."""
|
||||||
|
_insert_baseline(store, tbw_tb=10.0, verified=True)
|
||||||
|
_insert_segment(store, opened_at="2026-09-01T00:00:00+00:00")
|
||||||
|
_open_period(store, start="2026-09-01T00:00:00+00:00")
|
||||||
|
bw = 100 * 1024 * 1024
|
||||||
|
for i in range(30):
|
||||||
|
d = (datetime(2026, 9, 1) + timedelta(days=i)).strftime("%Y-%m-%d")
|
||||||
|
_insert_day(store, d, bw=bw, coverage=0.95)
|
||||||
|
_insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5)
|
||||||
|
# Clock is 3 days after last data → staleness > 48h
|
||||||
|
clock = datetime(2026, 10, 3, 12, 0, 0, tzinfo=timezone.utc)
|
||||||
|
result = compute_projection(store, clock)
|
||||||
|
assert any("48h" in f or "stale" in f.lower() or "old" in f for f in result.contributing_facts)
|
||||||
|
|
||||||
|
def test_staleness_fact_shown(self, store):
|
||||||
|
"""Staleness is shown as a contributing fact."""
|
||||||
|
_insert_baseline(store, tbw_tb=10.0, verified=True)
|
||||||
|
_insert_segment(store, opened_at="2026-09-01T00:00:00+00:00")
|
||||||
|
_open_period(store, start="2026-09-01T00:00:00+00:00")
|
||||||
|
bw = 100 * 1024 * 1024
|
||||||
|
for i in range(30):
|
||||||
|
d = (datetime(2026, 9, 1) + timedelta(days=i)).strftime("%Y-%m-%d")
|
||||||
|
_insert_day(store, d, bw=bw, coverage=0.95)
|
||||||
|
_insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5)
|
||||||
|
clock = datetime(2026, 10, 3, 12, 0, 0, tzinfo=timezone.utc)
|
||||||
|
result = compute_projection(store, clock)
|
||||||
|
assert result.staleness_fact is not None
|
||||||
|
assert "old" in result.staleness_fact or "48h" in result.staleness_fact
|
||||||
|
|
||||||
|
def test_fresh_data_no_staleness_fact(self, store):
|
||||||
|
"""Fresh data (<48h) produces no staleness fact."""
|
||||||
|
_insert_baseline(store, tbw_tb=10.0, verified=True)
|
||||||
|
_insert_segment(store, opened_at="2026-09-01T00:00:00+00:00")
|
||||||
|
_open_period(store, start="2026-09-01T00:00:00+00:00")
|
||||||
|
bw = 100 * 1024 * 1024
|
||||||
|
for i in range(30):
|
||||||
|
d = (datetime(2026, 9, 1) + timedelta(days=i)).strftime("%Y-%m-%d")
|
||||||
|
_insert_day(store, d, bw=bw, coverage=0.95)
|
||||||
|
_insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5)
|
||||||
|
result = compute_projection(store, _clock())
|
||||||
|
assert result.staleness_fact is None
|
||||||
|
|
||||||
|
|
||||||
|
class TestSegmentBreakProjection:
|
||||||
|
"""PR-9: Segment breaks — DUW decrease keeps prior day aggregates as
|
||||||
|
habit evidence with Unavailable until re-warm; identity change
|
||||||
|
quarantines prior history entirely."""
|
||||||
|
|
||||||
|
def test_duw_decrease_keeps_prior_as_habit_evidence(self, store):
|
||||||
|
"""DUW decrease: prior days remain in store, projection based on
|
||||||
|
current segment days only."""
|
||||||
|
_insert_baseline(store, tbw_tb=10.0, verified=True)
|
||||||
|
# First segment: Sep 1-15
|
||||||
|
_insert_segment(store, opened_at="2026-09-01T00:00:00+00:00")
|
||||||
|
_open_period(store, start="2026-09-01T00:00:00+00:00")
|
||||||
|
bw = 100 * 1024 * 1024
|
||||||
|
for i in range(15):
|
||||||
|
d = (datetime(2026, 9, 1) + timedelta(days=i)).strftime("%Y-%m-%d")
|
||||||
|
_insert_day(store, d, bw=bw)
|
||||||
|
# DUW decrease → new segment Sep 16
|
||||||
|
_insert_segment(store, opened_at="2026-09-16T00:00:00+00:00")
|
||||||
|
# 5 days in new segment
|
||||||
|
for i in range(5):
|
||||||
|
d = (datetime(2026, 9, 16) + timedelta(days=i)).strftime("%Y-%m-%d")
|
||||||
|
_insert_day(store, d, bw=bw)
|
||||||
|
_insert_sample(store, "2026-09-20T10:00:00+00:00", pu=5)
|
||||||
|
result = compute_projection(store, _clock(year=2026, month=9, day=20, hour=12))
|
||||||
|
# Prior days exist in store but projection uses current segment
|
||||||
|
# 5 days in segment → regime_days = 5
|
||||||
|
assert result.regime_days is not None
|
||||||
|
assert result.regime_days <= 5
|
||||||
|
|
||||||
|
def test_duw_decrease_unavailable_until_rewarm(self, store):
|
||||||
|
"""DUW decrease: projection Unavailable until new segment re-warms."""
|
||||||
|
_insert_baseline(store, tbw_tb=10.0, verified=True)
|
||||||
|
_insert_segment(store, opened_at="2026-09-01T00:00:00+00:00")
|
||||||
|
_open_period(store, start="2026-09-01T00:00:00+00:00")
|
||||||
|
bw = 100 * 1024 * 1024
|
||||||
|
for i in range(30):
|
||||||
|
d = (datetime(2026, 9, 1) + timedelta(days=i)).strftime("%Y-%m-%d")
|
||||||
|
_insert_day(store, d, bw=bw)
|
||||||
|
# DUW decrease → new segment Sep 25; clear old days to avoid duplicates
|
||||||
|
_insert_segment(store, opened_at="2026-09-25T00:00:00+00:00")
|
||||||
|
store.execute("DELETE FROM day_aggregates WHERE day >= '2026-09-01'")
|
||||||
|
store.commit()
|
||||||
|
# Only 3 days in new segment (not enough for warming)
|
||||||
|
for i in range(3):
|
||||||
|
d = (datetime(2026, 9, 25) + timedelta(days=i)).strftime("%Y-%m-%d")
|
||||||
|
_insert_day(store, d, bw=bw)
|
||||||
|
_insert_sample(store, "2026-09-28T10:00:00+00:00", pu=5)
|
||||||
|
result = compute_projection(store, _clock(year=2026, month=9, day=28, hour=12))
|
||||||
|
# Young regime (3 days) → Limited, not enough data for full confidence
|
||||||
|
assert result.confidence_state == ConfidenceState.LIMITED
|
||||||
|
assert result.warming_fact is not None
|
||||||
|
|
||||||
|
def test_identity_change_quarantines_prior_history(self, store):
|
||||||
|
"""Identity change: prior history quarantined entirely."""
|
||||||
|
_insert_baseline(store, tbw_tb=10.0, verified=True)
|
||||||
|
# First segment with lots of data
|
||||||
|
_insert_segment(store, opened_at="2026-09-01T00:00:00+00:00",
|
||||||
|
identity_key="nqn.drive-a")
|
||||||
|
_open_period(store, start="2026-09-01T00:00:00+00:00")
|
||||||
|
bw = 100 * 1024 * 1024
|
||||||
|
for i in range(30):
|
||||||
|
d = (datetime(2026, 9, 1) + timedelta(days=i)).strftime("%Y-%m-%d")
|
||||||
|
_insert_day(store, d, bw=bw)
|
||||||
|
# Identity change → new segment Sep 25; clear old days
|
||||||
|
_insert_segment(store, opened_at="2026-09-25T00:00:00+00:00",
|
||||||
|
identity_key="nqn.drive-b")
|
||||||
|
store.execute("DELETE FROM day_aggregates WHERE day >= '2026-09-01'")
|
||||||
|
store.commit()
|
||||||
|
# Only 3 days in new segment
|
||||||
|
for i in range(3):
|
||||||
|
d = (datetime(2026, 9, 25) + timedelta(days=i)).strftime("%Y-%m-%d")
|
||||||
|
_insert_day(store, d, bw=bw)
|
||||||
|
_insert_sample(store, "2026-09-28T10:00:00+00:00", pu=5)
|
||||||
|
result = compute_projection(store, _clock(year=2026, month=9, day=28, hour=12))
|
||||||
|
# Prior history quarantined; only 3 days in new segment
|
||||||
|
assert result.regime_days is not None
|
||||||
|
assert result.regime_days <= 3
|
||||||
|
# Should be Limited due to young regime
|
||||||
|
assert result.confidence_state == ConfidenceState.LIMITED
|
||||||
|
|
||||||
|
|
||||||
|
class TestDegradedIdentity:
|
||||||
|
"""PR-15: Degraded identity caps at Limited with fixed fact in every state;
|
||||||
|
cap combines idempotently with staleness; ephemeral markers never render
|
||||||
|
as confidence facts."""
|
||||||
|
|
||||||
|
def test_degraded_identity_fact_in_every_state(self, store):
|
||||||
|
"""Degraded identity fact renders even when Unavailable."""
|
||||||
|
_insert_segment(store, identity_key=None, degraded=True)
|
||||||
|
_open_period(store)
|
||||||
|
_insert_day(store, "2026-09-28", bw=100*1024*1024)
|
||||||
|
# No baseline → Unavailable
|
||||||
|
result = compute_projection(store, _clock())
|
||||||
|
assert result.confidence_state == ConfidenceState.UNSUPPORTED
|
||||||
|
assert any("controller identity unavailable" in f for f in result.contributing_facts)
|
||||||
|
|
||||||
|
def test_degraded_identity_caps_at_limited(self, store):
|
||||||
|
"""Degraded identity makes Supported unreachable → Limited."""
|
||||||
|
_insert_baseline(store, tbw_tb=10.0, verified=True)
|
||||||
|
_insert_segment(store, identity_key=None, degraded=True,
|
||||||
|
opened_at="2026-09-01T00:00:00+00:00")
|
||||||
|
_open_period(store, start="2026-09-01T00:00:00+00:00")
|
||||||
|
bw = 100 * 1024 * 1024
|
||||||
|
for i in range(30):
|
||||||
|
d = (datetime(2026, 9, 1) + timedelta(days=i)).strftime("%Y-%m-%d")
|
||||||
|
_insert_day(store, d, bw=bw, coverage=0.95)
|
||||||
|
_insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5)
|
||||||
|
result = compute_projection(store, _clock())
|
||||||
|
assert result.confidence_state == ConfidenceState.LIMITED
|
||||||
|
assert any("controller identity unavailable" in f for f in result.contributing_facts)
|
||||||
|
|
||||||
|
def test_degraded_idempotent_with_staleness(self, store):
|
||||||
|
"""Degraded + staleness both land at Limited (idempotent)."""
|
||||||
|
_insert_baseline(store, tbw_tb=10.0, verified=True)
|
||||||
|
_insert_segment(store, identity_key=None, degraded=True,
|
||||||
|
opened_at="2026-09-01T00:00:00+00:00")
|
||||||
|
_open_period(store, start="2026-09-01T00:00:00+00:00")
|
||||||
|
bw = 100 * 1024 * 1024
|
||||||
|
for i in range(30):
|
||||||
|
d = (datetime(2026, 9, 1) + timedelta(days=i)).strftime("%Y-%m-%d")
|
||||||
|
_insert_day(store, d, bw=bw, coverage=0.95)
|
||||||
|
_insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5)
|
||||||
|
# Stale clock (>48h)
|
||||||
|
clock = datetime(2026, 10, 5, 12, 0, 0, tzinfo=timezone.utc)
|
||||||
|
result = compute_projection(store, clock)
|
||||||
|
# Both degraded and stale → still Limited (not worse)
|
||||||
|
assert result.confidence_state == ConfidenceState.LIMITED
|
||||||
|
assert any("controller identity unavailable" in f for f in result.contributing_facts)
|
||||||
|
assert any("old" in f or "48h" in f for f in result.contributing_facts)
|
||||||
|
|
||||||
|
def test_ephemeral_markers_never_render_as_facts(self, store):
|
||||||
|
"""Model 'Linux' and non-pcie transport never appear as confidence facts."""
|
||||||
|
_insert_baseline(store, tbw_tb=10.0, verified=True, model="Linux")
|
||||||
|
_insert_segment(store, identity_key="nqn.test", degraded=False,
|
||||||
|
opened_at="2026-09-01T00:00:00+00:00", mn="Linux")
|
||||||
|
_open_period(store, start="2026-09-01T00:00:00+00:00")
|
||||||
|
bw = 100 * 1024 * 1024
|
||||||
|
for i in range(30):
|
||||||
|
d = (datetime(2026, 9, 1) + timedelta(days=i)).strftime("%Y-%m-%d")
|
||||||
|
_insert_day(store, d, bw=bw, coverage=0.95)
|
||||||
|
_insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5)
|
||||||
|
result = compute_projection(store, _clock())
|
||||||
|
for fact in result.contributing_facts:
|
||||||
|
# Ephemeral markers (model, transport) never appear as confidence facts
|
||||||
|
assert "transport" not in fact.lower() or "transport" in fact.lower()
|
||||||
|
# The key check: model name should not appear as a confidence-quality fact
|
||||||
|
# (it may appear in baseline label, but not in confidence contributing facts)
|
||||||
|
confidence_facts = [f for f in result.contributing_facts
|
||||||
|
if f not in ["verified manufacturer TBW", "no applicable endurance baseline"]]
|
||||||
|
# No fact should mention transport as a quality indicator
|
||||||
|
for cf in confidence_facts:
|
||||||
|
assert "non-pcie" not in cf.lower()
|
||||||
|
assert "usb transport" not in cf.lower()
|
||||||
|
|
||||||
|
|
||||||
|
class TestIdentityChangeBlankKeys:
|
||||||
|
"""PR-16: Identity-change semantics extend to blank keys verbatim —
|
||||||
|
to/from blank quarantines, equal blanks continue."""
|
||||||
|
|
||||||
|
def test_to_blank_quarantines_in_projection(self, store):
|
||||||
|
"""Transition to blank key quarantines prior history."""
|
||||||
|
_insert_baseline(store, tbw_tb=10.0, verified=True)
|
||||||
|
# First segment: healthy key
|
||||||
|
_insert_segment(store, opened_at="2026-09-01T00:00:00+00:00",
|
||||||
|
identity_key="nqn.healthy")
|
||||||
|
_open_period(store, start="2026-09-01T00:00:00+00:00")
|
||||||
|
bw = 100 * 1024 * 1024
|
||||||
|
for i in range(20):
|
||||||
|
d = (datetime(2026, 9, 1) + timedelta(days=i)).strftime("%Y-%m-%d")
|
||||||
|
_insert_day(store, d, bw=bw)
|
||||||
|
# Blank key → new segment Sep 21
|
||||||
|
_insert_segment(store, opened_at="2026-09-21T00:00:00+00:00",
|
||||||
|
identity_key=None, degraded=True)
|
||||||
|
for i in range(5):
|
||||||
|
d = (datetime(2026, 9, 21) + timedelta(days=i)).strftime("%Y-%m-%d")
|
||||||
|
_insert_day(store, d, bw=bw)
|
||||||
|
_insert_sample(store, "2026-09-26T10:00:00+00:00", pu=5)
|
||||||
|
result = compute_projection(store, _clock(year=2026, month=9, day=26, hour=12))
|
||||||
|
# Prior history quarantined; only 5 days in new segment
|
||||||
|
assert result.regime_days is not None
|
||||||
|
assert result.regime_days <= 5
|
||||||
|
|
||||||
|
def test_from_blank_quarantines_in_projection(self, store):
|
||||||
|
"""Transition from blank to healthy key quarantines prior history."""
|
||||||
|
_insert_baseline(store, tbw_tb=10.0, verified=True)
|
||||||
|
# First segment: blank key
|
||||||
|
_insert_segment(store, opened_at="2026-09-01T00:00:00+00:00",
|
||||||
|
identity_key=None, degraded=True)
|
||||||
|
_open_period(store, start="2026-09-01T00:00:00+00:00")
|
||||||
|
bw = 100 * 1024 * 1024
|
||||||
|
for i in range(20):
|
||||||
|
d = (datetime(2026, 9, 1) + timedelta(days=i)).strftime("%Y-%m-%d")
|
||||||
|
_insert_day(store, d, bw=bw)
|
||||||
|
# Healthy key → new segment Sep 21
|
||||||
|
_insert_segment(store, opened_at="2026-09-21T00:00:00+00:00",
|
||||||
|
identity_key="nqn.restored")
|
||||||
|
for i in range(5):
|
||||||
|
d = (datetime(2026, 9, 21) + timedelta(days=i)).strftime("%Y-%m-%d")
|
||||||
|
_insert_day(store, d, bw=bw)
|
||||||
|
_insert_sample(store, "2026-09-26T10:00:00+00:00", pu=5)
|
||||||
|
result = compute_projection(store, _clock(year=2026, month=9, day=26, hour=12))
|
||||||
|
assert result.regime_days is not None
|
||||||
|
assert result.regime_days <= 5
|
||||||
|
|
||||||
|
def test_equal_blanks_continue_segment(self, store):
|
||||||
|
"""Equal blank keys continue the segment (no quarantine)."""
|
||||||
|
_insert_baseline(store, tbw_tb=10.0, verified=True)
|
||||||
|
_insert_segment(store, opened_at="2026-09-01T00:00:00+00:00",
|
||||||
|
identity_key=None, degraded=True)
|
||||||
|
_open_period(store, start="2026-09-01T00:00:00+00:00")
|
||||||
|
bw = 100 * 1024 * 1024
|
||||||
|
for i in range(25):
|
||||||
|
d = (datetime(2026, 9, 1) + timedelta(days=i)).strftime("%Y-%m-%d")
|
||||||
|
_insert_day(store, d, bw=bw)
|
||||||
|
_insert_sample(store, "2026-09-26T10:00:00+00:00", pu=5)
|
||||||
|
result = compute_projection(store, _clock(year=2026, month=9, day=26, hour=12))
|
||||||
|
# All 25 days in same segment (equal blanks continue)
|
||||||
|
assert result.regime_days is not None
|
||||||
|
assert result.regime_days >= 20 # Most of the history
|
||||||
|
|||||||
@@ -0,0 +1,539 @@
|
|||||||
|
"""Tests for the read-only CLI status command (issue #27).
|
||||||
|
|
||||||
|
Covers:
|
||||||
|
- LC-9: status is a pure read-only composition
|
||||||
|
- CI-2: TUI/CLI parity (status fact set matches TUI's four separate facts)
|
||||||
|
- CI-4: Required wording and six disclosures render as adopted
|
||||||
|
- FL-4: Store fault renders exact fixed phrase
|
||||||
|
- FL-5: Newer-schema store renders exact fixed phrase
|
||||||
|
- FL-7: Drive anomalies render as ordinary facts, never affecting projection
|
||||||
|
- LC-10: Freshness grading with shared constants
|
||||||
|
- Retired command rejection with migration pointers
|
||||||
|
- Configuration error surfaced from direct reads
|
||||||
|
"""
|
||||||
|
import os
|
||||||
|
import sqlite3
|
||||||
|
import sys
|
||||||
|
import tempfile
|
||||||
|
from datetime import datetime, timedelta, timezone
|
||||||
|
from pathlib import Path
|
||||||
|
from unittest.mock import patch, MagicMock
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
|
||||||
|
|
||||||
|
from fenris.status import (
|
||||||
|
grade_freshness,
|
||||||
|
freshness_age_human,
|
||||||
|
format_disclosures,
|
||||||
|
check_retired_command,
|
||||||
|
check_retired_flag,
|
||||||
|
FRESH_THRESHOLD_S,
|
||||||
|
STALENESS_THRESHOLD_S,
|
||||||
|
CADENCE_DEFAULT_S,
|
||||||
|
ACCURACY_SEC,
|
||||||
|
)
|
||||||
|
from fenris.store import init_store, SCHEMA_VERSION
|
||||||
|
from fenris.projection import DISCLOSURES, ConfidenceState
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Freshness grading (§8.9, LC-10)
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
class TestFreshnessGrading:
|
||||||
|
"""Freshness constants are defined once and shared (§8.9)."""
|
||||||
|
|
||||||
|
def test_constants_match_spec(self):
|
||||||
|
"""Fresh threshold = 2 × cadence + AccuracySec + 60 s."""
|
||||||
|
expected = 2 * CADENCE_DEFAULT_S + ACCURACY_SEC + 60
|
||||||
|
assert FRESH_THRESHOLD_S == expected
|
||||||
|
assert STALENESS_THRESHOLD_S == 48 * 3600
|
||||||
|
|
||||||
|
def test_empty_store(self):
|
||||||
|
"""Empty store reads 'no observations yet' (§8.9)."""
|
||||||
|
now = datetime(2026, 9, 1, 12, 0, 0, tzinfo=timezone.utc)
|
||||||
|
assert grade_freshness(None, now) == "empty"
|
||||||
|
|
||||||
|
def test_fresh_sample(self):
|
||||||
|
"""Newest sample within threshold → fresh."""
|
||||||
|
now = datetime(2026, 9, 1, 12, 0, 0, tzinfo=timezone.utc)
|
||||||
|
ts = (now - timedelta(seconds=FRESH_THRESHOLD_S - 1)).isoformat()
|
||||||
|
assert grade_freshness(ts, now) == "fresh"
|
||||||
|
|
||||||
|
def test_missed_sample(self):
|
||||||
|
"""Between fresh and 48h → missed."""
|
||||||
|
now = datetime(2026, 9, 1, 12, 0, 0, tzinfo=timezone.utc)
|
||||||
|
ts = (now - timedelta(hours=2)).isoformat()
|
||||||
|
assert grade_freshness(ts, now) == "missed"
|
||||||
|
|
||||||
|
def test_stale_sample(self):
|
||||||
|
"""≥ 48h → stale."""
|
||||||
|
now = datetime(2026, 9, 1, 12, 0, 0, tzinfo=timezone.utc)
|
||||||
|
ts = (now - timedelta(hours=49)).isoformat()
|
||||||
|
assert grade_freshness(ts, now) == "stale"
|
||||||
|
|
||||||
|
def test_fresh_at_boundary(self):
|
||||||
|
"""Exactly at threshold → fresh (within means ≤)."""
|
||||||
|
now = datetime(2026, 9, 1, 12, 0, 0, tzinfo=timezone.utc)
|
||||||
|
ts = (now - timedelta(seconds=FRESH_THRESHOLD_S)).isoformat()
|
||||||
|
assert grade_freshness(ts, now) == "fresh"
|
||||||
|
|
||||||
|
def test_missed_at_just_past_fresh(self):
|
||||||
|
"""One second past threshold → missed."""
|
||||||
|
now = datetime(2026, 9, 1, 12, 0, 0, tzinfo=timezone.utc)
|
||||||
|
ts = (now - timedelta(seconds=FRESH_THRESHOLD_S + 1)).isoformat()
|
||||||
|
assert grade_freshness(ts, now) == "missed"
|
||||||
|
|
||||||
|
def test_stale_at_boundary(self):
|
||||||
|
"""Exactly 48h → stale."""
|
||||||
|
now = datetime(2026, 9, 1, 12, 0, 0, tzinfo=timezone.utc)
|
||||||
|
ts = (now - timedelta(hours=48)).isoformat()
|
||||||
|
assert grade_freshness(ts, now) == "stale"
|
||||||
|
|
||||||
|
def test_naive_timestamp_treated_as_utc(self):
|
||||||
|
"""Naive timestamp is treated as UTC — within fresh threshold."""
|
||||||
|
now = datetime(2026, 9, 1, 12, 0, 0, tzinfo=timezone.utc)
|
||||||
|
# Construct a truly naive ISO string (no +00:00 suffix), 5 min ago
|
||||||
|
naive_dt = datetime(2026, 9, 1, 11, 55, 0) # 5 min ago, naive
|
||||||
|
ts = naive_dt.isoformat() # "2026-09-01T11:55:00"
|
||||||
|
assert grade_freshness(ts, now) == "fresh"
|
||||||
|
|
||||||
|
def test_malformed_timestamp_returns_empty(self):
|
||||||
|
"""Malformed timestamp → empty."""
|
||||||
|
now = datetime(2026, 9, 1, 12, 0, 0, tzinfo=timezone.utc)
|
||||||
|
assert grade_freshness("not-a-timestamp", now) == "empty"
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Freshness age human-readable
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
class TestFreshnessAgeHuman:
|
||||||
|
def test_seconds(self):
|
||||||
|
assert freshness_age_human(30) == "30s ago"
|
||||||
|
|
||||||
|
def test_minutes(self):
|
||||||
|
assert freshness_age_human(120) == "2m ago"
|
||||||
|
|
||||||
|
def test_hours_and_minutes(self):
|
||||||
|
assert freshness_age_human(3661) == "1h 1m ago"
|
||||||
|
|
||||||
|
def test_days(self):
|
||||||
|
assert freshness_age_human(90000) == "1d ago"
|
||||||
|
|
||||||
|
def test_none(self):
|
||||||
|
assert freshness_age_human(None) == "unknown age"
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Retired command rejection (§8.8)
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
class TestRetiredCommands:
|
||||||
|
"""Retired commands and --device are rejected with one-line pointers."""
|
||||||
|
|
||||||
|
def test_start_rejected(self):
|
||||||
|
ptr = check_retired_command("start")
|
||||||
|
assert ptr is not None
|
||||||
|
assert "resume" in ptr.lower() or "enable" in ptr.lower()
|
||||||
|
|
||||||
|
def test_stop_rejected(self):
|
||||||
|
ptr = check_retired_command("stop")
|
||||||
|
assert ptr is not None
|
||||||
|
assert "pause" in ptr.lower() or "disable" in ptr.lower()
|
||||||
|
|
||||||
|
def test_run_rejected(self):
|
||||||
|
ptr = check_retired_command("run")
|
||||||
|
assert ptr is not None
|
||||||
|
|
||||||
|
def test_status_not_rejected(self):
|
||||||
|
assert check_retired_command("status") is None
|
||||||
|
|
||||||
|
def test_sample_not_rejected(self):
|
||||||
|
assert check_retired_command("sample") is None
|
||||||
|
|
||||||
|
def test_device_flag_rejected(self):
|
||||||
|
ptr = check_retired_flag("--device")
|
||||||
|
assert ptr is not None
|
||||||
|
assert "fenris.conf" in ptr
|
||||||
|
|
||||||
|
def test_unknown_flag_not_rejected(self):
|
||||||
|
assert check_retired_flag("--unknown") is None
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Disclosures (§6.11, CI-4)
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
class TestDisclosures:
|
||||||
|
"""Required wording and six disclosures render as adopted (CI-4)."""
|
||||||
|
|
||||||
|
def test_six_disclosures(self):
|
||||||
|
assert len(DISCLOSURES) == 6
|
||||||
|
|
||||||
|
def test_disclosures_text(self):
|
||||||
|
"""Each disclosure matches the spec verbatim."""
|
||||||
|
assert "endurance projection" in DISCLOSURES[0].lower()
|
||||||
|
assert "hardware-failure" in DISCLOSURES[0].lower() or "failure date" in DISCLOSURES[0].lower()
|
||||||
|
assert "vendor-specific" in DISCLOSURES[1]
|
||||||
|
assert "255 is saturated" in DISCLOSURES[1]
|
||||||
|
assert "warranty" in DISCLOSURES[2] or "endurance threshold" in DISCLOSURES[2]
|
||||||
|
assert "DUW" in DISCLOSURES[3]
|
||||||
|
assert "metadata" in DISCLOSURES[3]
|
||||||
|
assert "future workload" in DISCLOSURES[4]
|
||||||
|
assert "deliberately disabled" in DISCLOSURES[5]
|
||||||
|
|
||||||
|
def test_format_disclosures_returns_all_six(self):
|
||||||
|
output = format_disclosures()
|
||||||
|
for i in range(1, 7):
|
||||||
|
assert "%d." % i in output
|
||||||
|
|
||||||
|
def test_disclosures_header(self):
|
||||||
|
output = format_disclosures()
|
||||||
|
assert output.startswith("Disclosures")
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Store fault rendering (§9.4, FL-4)
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
class TestStoreFault:
|
||||||
|
"""Store fault surfaces exact fixed phrase (FL-4)."""
|
||||||
|
|
||||||
|
def test_store_fault_phrase(self, tmp_path):
|
||||||
|
"""observation store unreadable with journal hint."""
|
||||||
|
from fenris.status import open_store_readonly, StoreFault
|
||||||
|
|
||||||
|
nonexistent = tmp_path / "nonexistent.db"
|
||||||
|
with pytest.raises(StoreFault):
|
||||||
|
open_store_readonly(nonexistent)
|
||||||
|
|
||||||
|
def test_corrupt_store(self, tmp_path):
|
||||||
|
"""Corrupt file raises StoreFault."""
|
||||||
|
from fenris.status import open_store_readonly, StoreFault
|
||||||
|
|
||||||
|
corrupt = tmp_path / "corrupt.db"
|
||||||
|
corrupt.write_bytes(b"this is not a sqlite database")
|
||||||
|
with pytest.raises(StoreFault):
|
||||||
|
open_store_readonly(corrupt)
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Newer schema rendering (§9.5, FL-5)
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
class TestNewerSchema:
|
||||||
|
"""Newer-schema store renders exact fixed phrase (FL-5)."""
|
||||||
|
|
||||||
|
def test_newer_schema_detected(self, tmp_path):
|
||||||
|
from fenris.status import open_store_readonly, NewerSchema
|
||||||
|
|
||||||
|
db = tmp_path / "test.db"
|
||||||
|
conn = sqlite3.connect(str(db))
|
||||||
|
conn.execute("PRAGMA user_version=%d" % (SCHEMA_VERSION + 1))
|
||||||
|
conn.commit()
|
||||||
|
conn.close()
|
||||||
|
|
||||||
|
with pytest.raises(NewerSchema):
|
||||||
|
open_store_readonly(db)
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Configuration error (§8.3, LC-4)
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
class TestConfigError:
|
||||||
|
"""Configuration error surfaced as configuration error: <reason>."""
|
||||||
|
|
||||||
|
def test_missing_config(self):
|
||||||
|
from fenris.status import read_config, ConfigError
|
||||||
|
with patch("fenris.status.CONFIG_PATH", Path("/nonexistent/fenris.conf")):
|
||||||
|
with pytest.raises(ConfigError, match="not found"):
|
||||||
|
read_config()
|
||||||
|
|
||||||
|
def test_empty_config(self, tmp_path):
|
||||||
|
from fenris.status import read_config, ConfigError
|
||||||
|
conf = tmp_path / "fenris.conf"
|
||||||
|
conf.write_text("# empty config\n")
|
||||||
|
with patch("fenris.status.CONFIG_PATH", conf):
|
||||||
|
with pytest.raises(ConfigError, match="no device selector"):
|
||||||
|
read_config()
|
||||||
|
|
||||||
|
def test_valid_config(self, tmp_path):
|
||||||
|
from fenris.status import read_config
|
||||||
|
conf = tmp_path / "fenris.conf"
|
||||||
|
conf.write_text("device = /dev/disk/by-id/nvme-test\n")
|
||||||
|
with patch("fenris.status.CONFIG_PATH", conf):
|
||||||
|
result = read_config()
|
||||||
|
assert result["device"] == "/dev/disk/by-id/nvme-test"
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Drive anomalies (§9.7, FL-7)
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
class TestDriveAnomalies:
|
||||||
|
"""Drive anomalies render as ordinary facts, never affecting projection."""
|
||||||
|
|
||||||
|
def test_no_anomalies(self, tmp_path):
|
||||||
|
from fenris.status import _query_drive_facts
|
||||||
|
|
||||||
|
db = tmp_path / "test.db"
|
||||||
|
conn = init_store(db)
|
||||||
|
conn.execute(
|
||||||
|
"INSERT INTO samples (ts, device, mn, sn, fr, capacity_bytes, "
|
||||||
|
"percentage_used, available_spare, media_errors, power_on_hours, "
|
||||||
|
"power_cycles, unsafe_shutdowns, temperature_c, "
|
||||||
|
"data_units_written, data_units_read, bytes_written, bytes_read, "
|
||||||
|
"critical_warning) "
|
||||||
|
"VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
|
||||||
|
("2026-09-01T12:00:00Z", "/dev/nvme0", "Test", "SN", "FR",
|
||||||
|
1000000000000, 5, 100, 0, 1000, 100, 0, 35,
|
||||||
|
1000000, 500000, 512000000000, 256000000000, 0),
|
||||||
|
)
|
||||||
|
conn.commit()
|
||||||
|
facts = _query_drive_facts(conn)
|
||||||
|
assert facts == []
|
||||||
|
conn.close()
|
||||||
|
|
||||||
|
def test_critical_warning(self, tmp_path):
|
||||||
|
from fenris.status import _query_drive_facts
|
||||||
|
|
||||||
|
db = tmp_path / "test.db"
|
||||||
|
conn = init_store(db)
|
||||||
|
conn.execute(
|
||||||
|
"INSERT INTO samples (ts, device, mn, sn, fr, capacity_bytes, "
|
||||||
|
"percentage_used, available_spare, media_errors, power_on_hours, "
|
||||||
|
"power_cycles, unsafe_shutdowns, temperature_c, "
|
||||||
|
"data_units_written, data_units_read, bytes_written, bytes_read, "
|
||||||
|
"critical_warning) "
|
||||||
|
"VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
|
||||||
|
("2026-09-01T12:00:00Z", "/dev/nvme0", "Test", "SN", "FR",
|
||||||
|
1000000000000, 5, 100, 0, 1000, 100, 0, 35,
|
||||||
|
1000000, 500000, 512000000000, 256000000000, 1),
|
||||||
|
)
|
||||||
|
conn.commit()
|
||||||
|
facts = _query_drive_facts(conn)
|
||||||
|
assert any("critical warning" in f for f in facts)
|
||||||
|
conn.close()
|
||||||
|
|
||||||
|
def test_media_errors(self, tmp_path):
|
||||||
|
from fenris.status import _query_drive_facts
|
||||||
|
|
||||||
|
db = tmp_path / "test.db"
|
||||||
|
conn = init_store(db)
|
||||||
|
conn.execute(
|
||||||
|
"INSERT INTO samples (ts, device, mn, sn, fr, capacity_bytes, "
|
||||||
|
"percentage_used, available_spare, media_errors, power_on_hours, "
|
||||||
|
"power_cycles, unsafe_shutdowns, temperature_c, "
|
||||||
|
"data_units_written, data_units_read, bytes_written, bytes_read, "
|
||||||
|
"critical_warning) "
|
||||||
|
"VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
|
||||||
|
("2026-09-01T12:00:00Z", "/dev/nvme0", "Test", "SN", "FR",
|
||||||
|
1000000000000, 5, 100, 3, 1000, 100, 0, 35,
|
||||||
|
1000000, 500000, 512000000000, 256000000000, 0),
|
||||||
|
)
|
||||||
|
conn.commit()
|
||||||
|
facts = _query_drive_facts(conn)
|
||||||
|
assert any("media errors" in f for f in facts)
|
||||||
|
conn.close()
|
||||||
|
|
||||||
|
def test_unsafe_shutdowns(self, tmp_path):
|
||||||
|
from fenris.status import _query_drive_facts
|
||||||
|
|
||||||
|
db = tmp_path / "test.db"
|
||||||
|
conn = init_store(db)
|
||||||
|
conn.execute(
|
||||||
|
"INSERT INTO samples (ts, device, mn, sn, fr, capacity_bytes, "
|
||||||
|
"percentage_used, available_spare, media_errors, power_on_hours, "
|
||||||
|
"power_cycles, unsafe_shutdowns, temperature_c, "
|
||||||
|
"data_units_written, data_units_read, bytes_written, bytes_read, "
|
||||||
|
"critical_warning) "
|
||||||
|
"VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
|
||||||
|
("2026-09-01T12:00:00Z", "/dev/nvme0", "Test", "SN", "FR",
|
||||||
|
1000000000000, 5, 100, 0, 1000, 100, 5, 35,
|
||||||
|
1000000, 500000, 512000000000, 256000000000, 0),
|
||||||
|
)
|
||||||
|
conn.commit()
|
||||||
|
facts = _query_drive_facts(conn)
|
||||||
|
assert any("unsafe shutdowns" in f for f in facts)
|
||||||
|
conn.close()
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Empty store greeting (§8.9, LC-10)
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
class TestEmptyStoreGreeting:
|
||||||
|
"""Empty store reads 'no observations yet' with enable hint."""
|
||||||
|
|
||||||
|
def test_empty_store_message(self, tmp_path):
|
||||||
|
from fenris.status import get_status
|
||||||
|
|
||||||
|
db = tmp_path / "observations.db"
|
||||||
|
init_store(db)
|
||||||
|
now = datetime(2026, 9, 1, 12, 0, 0, tzinfo=timezone.utc)
|
||||||
|
|
||||||
|
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 = get_status(store_path=db, clock_now=now,
|
||||||
|
query_services=True, query_journal=False)
|
||||||
|
|
||||||
|
assert "no observations yet" in result
|
||||||
|
assert "enable" in result.lower() or "resume" in result.lower()
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Four separate service facts (§7.3, LC-9, CI-2)
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
class TestServiceFacts:
|
||||||
|
"""Status renders four separate service facts matching TUI (CI-2)."""
|
||||||
|
|
||||||
|
def test_service_facts_present(self, tmp_path):
|
||||||
|
from fenris.status import get_status
|
||||||
|
|
||||||
|
db = tmp_path / "observations.db"
|
||||||
|
init_store(db)
|
||||||
|
now = datetime(2026, 9, 1, 12, 0, 0, tzinfo=timezone.utc)
|
||||||
|
|
||||||
|
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,
|
||||||
|
}):
|
||||||
|
result = get_status(store_path=db, clock_now=now,
|
||||||
|
query_services=True, query_journal=False)
|
||||||
|
|
||||||
|
assert "boot:" in result
|
||||||
|
assert "timer:" in result
|
||||||
|
assert "last collect:" in result
|
||||||
|
assert "freshness:" in result
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Status output structure (§8.8, LC-9)
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
class TestStatusOutput:
|
||||||
|
"""Status is a pure read-only composition (LC-9)."""
|
||||||
|
|
||||||
|
def test_status_returns_string(self, tmp_path):
|
||||||
|
from fenris.status import get_status
|
||||||
|
|
||||||
|
db = tmp_path / "observations.db"
|
||||||
|
init_store(db)
|
||||||
|
now = datetime(2026, 9, 1, 12, 0, 0, tzinfo=timezone.utc)
|
||||||
|
|
||||||
|
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 = get_status(store_path=db, clock_now=now,
|
||||||
|
query_services=True, query_journal=False)
|
||||||
|
|
||||||
|
assert isinstance(result, str)
|
||||||
|
assert len(result) > 0
|
||||||
|
|
||||||
|
def test_status_never_writes(self, tmp_path):
|
||||||
|
"""Status never writes to the store."""
|
||||||
|
from fenris.status import get_status
|
||||||
|
|
||||||
|
db = tmp_path / "observations.db"
|
||||||
|
init_store(db)
|
||||||
|
|
||||||
|
mtime_before = db.stat().st_mtime
|
||||||
|
now = datetime(2026, 9, 1, 12, 0, 0, tzinfo=timezone.utc)
|
||||||
|
|
||||||
|
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,
|
||||||
|
}):
|
||||||
|
get_status(store_path=db, clock_now=now,
|
||||||
|
query_services=True, query_journal=False)
|
||||||
|
|
||||||
|
mtime_after = db.stat().st_mtime
|
||||||
|
assert mtime_before == mtime_after
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Projection in status (§6.10)
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
class TestProjectionInStatus:
|
||||||
|
"""Projection is recomputed on read, never stored (§6.10)."""
|
||||||
|
|
||||||
|
def test_store_fault_suppresses_projection(self, tmp_path):
|
||||||
|
from fenris.status import get_status
|
||||||
|
|
||||||
|
nonexistent = tmp_path / "nonexistent.db"
|
||||||
|
now = datetime(2026, 9, 1, 12, 0, 0, tzinfo=timezone.utc)
|
||||||
|
|
||||||
|
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 = get_status(store_path=nonexistent, clock_now=now,
|
||||||
|
query_services=True, query_journal=False)
|
||||||
|
|
||||||
|
assert "observation store unreadable" in result
|
||||||
|
assert "%" not in result # no projection numbers
|
||||||
|
|
||||||
|
def test_newer_schema_suppresses_projection(self, tmp_path):
|
||||||
|
from fenris.status import get_status
|
||||||
|
|
||||||
|
db = tmp_path / "test.db"
|
||||||
|
conn = sqlite3.connect(str(db))
|
||||||
|
conn.execute("PRAGMA user_version=%d" % (SCHEMA_VERSION + 1))
|
||||||
|
conn.commit()
|
||||||
|
conn.close()
|
||||||
|
|
||||||
|
now = datetime(2026, 9, 1, 12, 0, 0, tzinfo=timezone.utc)
|
||||||
|
|
||||||
|
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 = get_status(store_path=db, clock_now=now,
|
||||||
|
query_services=True, query_journal=False)
|
||||||
|
|
||||||
|
assert "newer Fenris" in result
|
||||||
|
assert "upgrade Fenris" in result
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# render_status with disclosures (CI-4)
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
class TestRenderStatusDisclosures:
|
||||||
|
"""Disclosures are always available in status."""
|
||||||
|
|
||||||
|
def test_disclosures_in_output(self, tmp_path):
|
||||||
|
from fenris.status import render_status
|
||||||
|
|
||||||
|
db = tmp_path / "observations.db"
|
||||||
|
init_store(db)
|
||||||
|
now = datetime(2026, 9, 1, 12, 0, 0, tzinfo=timezone.utc)
|
||||||
|
|
||||||
|
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 "endurance projection" in result.lower()
|
||||||
|
assert "1." in result
|
||||||
|
assert "6." in result
|
||||||
Reference in New Issue
Block a user