Compare commits

...
Author SHA1 Message Date
xavierk ef9e7fa7a8 fix: satisfy semgrep on constant-column SQL; use injectable clock for baseline entry_date
CI / security (pull_request) Successful in 20s
CI / lint (pull_request) Successful in 39s
CI / ai-review (pull_request) Successful in 8m44s
2026-10-06 13:01:34 +05:30
xavierk 670671cb25 chore: prepare v0.7.0 release
Release / release (push) Successful in 1m28s
CI / security (pull_request) Failing after 21s
CI / lint (pull_request) Successful in 36s
CI / ai-review (pull_request) Successful in 2m52s
Bump pyproject.toml to 0.7.0 and move the Unreleased section under
[0.7.0] - 2026-10-06.
2026-10-06 10:14:35 +05:30
xavierk 8c8710fe5a Merge origin/main into main
Reconcile the staged-projection and Diagnostics feature work with the CI
quality-gate changes from origin/main. The origin side of every conflict was
lint-only, so the feature imports were kept and the ruff fixes re-applied:

- keep the branch's added imports (timefmt, activity_readout, projection
  ladder helpers) and drop the unused ones ruff reported
- restore `noqa: E402` on the post-bootstrap imports in fenris-monitor and
  keep the `noqa: F401` re-exports in the TUI that tests assert on
- make the TUI monitoring-glyph test clock-relative instead of pinned to a
  fixed date
2026-10-06 10:14:09 +05:30
xavierk ab1d894adc feat(collector): capture additional smartctl diagnostics
Record host read and write commands, controller busy time, error log
entries, warning and critical temperature time, and thermal management
transitions from the existing smartctl -a -j acquisition. Schema 7 adds
nullable columns to samples so legacy rows read as unknown, and the
Diagnostics panel shows each counter with its change.
2026-10-05 21:12:25 +05:30
xavierk 948e097952 feat(tui): add Diagnostics panel for stored evidence
Toggle a Diagnostics panel with i, built on the Activity readout. It swaps
in for the Drive pane and shows the decoded critical_warning, power-on
hours, power cycles, unsafe shutdowns, media errors and available spare
with their change over the selection, and per-hour (Day) or per-day
(History) temperature min/avg/max, active/idle/powered-off/unknown split
and coverage. History bars draw reads beside writes while it is open, and a
set critical_warning shows as a badge in the endurance outlook without
affecting the projection. Counter deltas are withheld across a controller
segment change.
2026-10-05 21:05:44 +05:30
xavierk 6c517e4ecb feat(tui): unify keyboard and mouse navigation
Declare every dashboard key as a Binding on the widget that owns it, and
generate the footer, activity tool chips and help screen from those
bindings so they cannot drift. The footer collapses to "? help · q quit"
under 80x24.

Add Home/End and PgUp/PgDn (one week) navigation, mouse-wheel selection on
the graphs, click-to-focus on every panel, and clickable footer chips.
Tab and Shift+Tab now visit only the dashboard panels.
2026-10-05 20:42:04 +05:30
xavierk 79ebb10dea refactor(tui): extract Activity readout and show local time
Move activity data assembly out of tui.py into activity_readout, behind
readout(conn, selection, now). Dashboard and fenris-monitor enable now show
local time with a timezone label, inspected points keep a secondary UTC line,
and the legacy UTC-relabelled day and hour branches are gone. Time formatting
(including freshness_age_human) lives in timefmt. Hour offsets now render as
+05:30 to match day labels.
2026-10-05 20:21:22 +05:30
xavierk ef61ececc1 feat(tui): enter rated TBW from the dashboard
Add a b key and a no-baseline banner that open a form for rated TBW and an
optional source URL. The value is persisted as an unverified override through
fenris-monitor baseline set under the existing polkit action, and the outcome
is reported synchronously. Verified overrides stay CLI-only (ADR 0012 section 7).
2026-10-05 20:01:32 +05:30
xavierk e26a851115 feat(projection): show staged projection with evidence ladder
Report a projection stage, an evidence ladder and the sustained-regime
write rate from compute_projection, per ADR 0012. A provisional lifespan
appears after 3 observed hours and carries the hours observed and a
short-horizon spread; the complete-local-day condition becomes a fact
instead of a gate, and the write rate replaces the lifespan when no
baseline applies. The TUI and CLI render the stage, ladder count and
full ladder, and drop their own two-sample pre-gates.
2026-10-05 19:57:46 +05:30
xavierk 96fd1702fa fix(projection): align projection with ADR 0002
Use lifetime written bytes from the current segment's newest published
sample in the headline formula, select the Percentage-Used-implied
baseline after two increments in the segment (otherwise show the
too-coarse fact), and judge horizon agreement across existing horizons.
Make the vacuous projection tests assert unconditionally and add
coverage for implied-baseline gating and single-horizon agreement.
2026-10-05 19:45:38 +05:30
xavierk cd053cf125 Merge pull request 'ci: add quality-gate workflow (semgrep, ruff, jscpd, advisory PR-Agent)' (#112) from ci/quality-gates into main
CI / lint (push) Successful in 29s
CI / security (push) Successful in 19s
CI / ai-review (push) Skipped
Release / release (push) Failing after 6s
Reviewed-on: #112
2026-10-05 14:14:59 +00:00
xavierk 117082af8c docs: record staged projection decisions (#105)
Add ADR 0012 amending ADR 0002 and ADR 0003: projection stages,
provisional lifespan after 3 observed hours, short scenario horizons,
evidence ladder, and TUI unverified TBW entry. Add Projection stage,
Provisional projection, Evidence ladder and Activity readout to the
glossary.
2026-10-03 08:13:12 +05:30
41 changed files with 4177 additions and 1201 deletions
+20
View File
@@ -9,6 +9,26 @@ backfill releases from before this changelog.
## [Unreleased] ## [Unreleased]
## [0.7.0] - 2026-10-06
### Added
- Show a provisional usage-adjusted theoretical lifespan after 3 observed hours, with the hours observed and a short-horizon spread, instead of waiting for a full local day.
- Show the projection stage and an evidence ladder of Supported conditions, each met or unmet with its reason, so confidence visibly strengthens with each collection run.
- Show the sustained-regime write rate when no endurance baseline applies.
- Enter a rated TBW from the dashboard with `b`, saved as an unverified override, so the lifespan appears without the CLI.
- Jump to the first or last point with Home and End on every graph, and a week at a time with PgUp and PgDn in the Day and History views.
- Move through graph points with the mouse wheel, focus any panel by clicking it, and click the footer and activity tool chips to run their actions.
- Open a Diagnostics panel with `i`: the decoded critical warning, power-on hours, power cycles, unsafe shutdowns, media errors and available spare with their change over the selection, plus hourly or daily temperature, active/idle/powered-off/unknown split and coverage. History bars draw reads beside writes while it is open, and a set critical warning shows as a badge in the endurance outlook.
- Record more drive evidence from each collection run (host read and write commands, controller busy time, error log entries, warning and critical temperature time, and thermal management transitions where the drive reports them) and show it in the Diagnostics panel. Existing stores migrate automatically; earlier samples show n/a for the new counters.
### Changed
- Generate the footer and help screen from the dashboard's key bindings so they always agree, and collapse the footer to "? help · q quit" under 80x24.
- Make Tab and Shift+Tab visit only the dashboard panels.
- Show every dashboard time in local time with its timezone, keep a secondary UTC line when a point is inspected, and report the monitoring period start in local time from `fenris-monitor enable`.
- Treat the complete-local-day requirement as a contributing fact rather than a gate, and stop the TUI and CLI from deciding when a projection can be shown.
## [0.6.0] - 2026-09-29 ## [0.6.0] - 2026-09-29
### Added ### Added
+17 -1
View File
@@ -20,6 +20,18 @@ _Avoid_: Future life, actual lifespan, failure date
The degree to which the observation history is sufficiently long, complete, and stable to support the usage-adjusted theoretical lifespan. The degree to which the observation history is sufficiently long, complete, and stable to support the usage-adjusted theoretical lifespan.
_Avoid_: Accuracy percentage, certainty _Avoid_: Accuracy percentage, certainty
**Projection stage**:
Where a usage-adjusted theoretical lifespan stands on its way to support: no observations yet, provisional, warming, limited, supported, or unavailable. Each stage is shown with the facts that place it there.
_Avoid_: Loading state, progress percentage
**Provisional projection**:
A lifespan shown from only a few hours of observation history, before a full daily cycle has been seen; it is labelled with the hours observed and its spread, and is expected to move.
_Avoid_: Estimate, preview, guess
**Evidence ladder**:
The conditions that supported projection confidence requires, each shown as met or unmet with its reason, so confidence visibly strengthens as each collection run adds evidence.
_Avoid_: Confidence score, accuracy meter
**Monitoring period**: **Monitoring period**:
A span during which Fenris monitoring is enabled; powered-off time remains part of the usage habit, while deliberately disabled time does not. A span during which Fenris monitoring is enabled; powered-off time remains part of the usage habit, while deliberately disabled time does not.
_Avoid_: Daemon uptime, calibration window _Avoid_: Daemon uptime, calibration window
@@ -56,6 +68,10 @@ _Avoid_: Missing bytes, estimated midnight split
The read/write measurement, date and point being inspected in live or historical drive activity. Live activity follows the newest point until deliberate inspection pins a point; historical selection survives refresh, while an expired live pin is explained before following resumes. The read/write measurement, date and point being inspected in live or historical drive activity. Live activity follows the newest point until deliberate inspection pins a point; historical selection survives refresh, while an expired live pin is explained before following resumes.
_Avoid_: Projection window, current usage habit _Avoid_: Projection window, current usage habit
**Activity readout**:
The ready-to-read account of an activity selection: measured read and write volume, thermal evidence, usage-habit split, coverage and drive warnings for the selected live interval, local hour or local day, expressed in local time with its timezone.
_Avoid_: Graph data, view model
**Usage-history window**: **Usage-history window**:
An exact consecutive span of UTC calendar days ending today, shown from day aggregates; a day without trustworthy evidence remains an explicit gap rather than disappearing or being estimated. An exact consecutive span of UTC calendar days ending today, shown from day aggregates; a day without trustworthy evidence remains an explicit gap rather than disappearing or being estimated.
_Avoid_: Available records, dataset range _Avoid_: Available records, dataset range
@@ -93,7 +109,7 @@ A sustained divergence between recent and earlier daily write rates that starts
_Avoid_: Spike, anomaly _Avoid_: Spike, anomaly
**Scenario range**: **Scenario range**:
The spread of lifespan projections computed from the 7-, 28-, and 90-day horizons of the observation history, shown in place of a statistical interval. The spread of lifespan projections computed from the horizons the observation history covers — 24-hour and 3-day horizons until a 7-day horizon exists, then 7-, 28-, and 90-day horizons — shown in place of a statistical interval.
_Avoid_: Confidence interval, error bar _Avoid_: Confidence interval, error bar
**Coverage**: **Coverage**:
+18 -6
View File
@@ -257,8 +257,9 @@ is retried at the next interval; it never fabricates missing observations.
Preview uses synthetic observations, not measurements from a real drive. Preview uses synthetic observations, not measurements from a real drive.
The Chalktone dashboard opens with a large **Live** activity plot. **Day** shows The Chalktone dashboard opens with a large **Live** activity plot. **Day** shows
hourly evidence for a selected date, labelled UTC, and **History** shows daily hourly evidence for a selected date, and **History** shows daily evidence. Every
evidence. Local-day totals keep their recorded timezone. Dotted traces show time is local, labelled with its timezone, and the selected-point readout adds
a secondary UTC line. Local-day totals keep their recorded timezone. Dotted traces show
measured read/write volumes, not transfer speed; missing evidence breaks the measured read/write volumes, not transfer speed; missing evidence breaks the
trace. `?` marks a gap, `~` a partial total, and `u` unallocated daily volume. trace. `?` marks a gap, `~` a partial total, and `u` unallocated daily volume.
Use the selected-point readout for exact values and evidence state. Use the selected-point readout for exact values and evidence state.
@@ -270,16 +271,27 @@ shows incomplete or unavailable dates without treating them as zero. Older
UTC-only summaries cannot establish exact local-day totals. UTC-only summaries cannot establish exact local-day totals.
- `v` cycles Live / Day / History; the tabs are also clickable. - `v` cycles Live / Day / History; the tabs are also clickable.
- `←` / `→` inspect points; `w` switches read/write volume in every view. - `←` / `→` inspect points, `Home` / `End` jump to the first or last point, and
`PgUp` / `PgDn` jump a week in Day and History (the live window is three
hours); the mouse wheel moves the selection too. `w`
switches read/write volume in every view.
- `i` swaps the Drive pane for a Diagnostics panel built from the stored
evidence: the decoded `critical_warning`, power-on hours, power cycles,
unsafe shutdowns, media errors and available spare with their change over the
selection, and per-hour (Day) or per-day (History) temperature min/avg/max,
active / idle / powered-off / unknown split and coverage. History bars then
draw reads beside writes. A set `critical_warning` also shows as a badge in
the endurance outlook; it never changes the projection.
- `[` / `]` browse dates, `g` enters a date, and `t` returns to today/live. - `[` / `]` browse dates, `g` enters a date, and `t` returns to today/live.
- `Tab` / `Shift+Tab` move focus; `z` expands the focused panel, and `z` or - `Tab` / `Shift+Tab` move focus between panels (clicking a panel focuses it);
`Esc` restores it. Monitoring status and controls remain visible. `z` expands the focused panel, and `z` or `Esc` restores it. Monitoring status and controls remain visible.
- `s` cycles Chalktone, Amber, Nord, and High Contrast; saved theme preferences - `s` cycles Chalktone, Amber, Nord, and High Contrast; saved theme preferences
survive upgrades. `m` toggles reduced motion. survive upgrades. `m` toggles reduced motion.
On smaller terminals, textual summaries and scrollable panels keep evidence On smaller terminals, textual summaries and scrollable panels keep evidence
accessible. Pause, resume, collect, disclosures, help, and quit remain available accessible. Pause, resume, collect, disclosures, help, and quit remain available
in the fixed control row. in the fixed control row, which collapses to `? help · q quit` under 80×24. `?`
lists every key; the footer and help are generated from the same bindings.
Run `fenris` as your normal user to open the TUI dashboard. The dashboard does Run `fenris` as your normal user to open the TUI dashboard. The dashboard does
not need `sudo`. Use `sudo` for package installation and system configuration; not need `sudo`. Use `sudo` for package installation and system configuration;
@@ -0,0 +1,29 @@
# 12. Staged projection with an evidence ladder
Status: Accepted. Amends [ADR 0002](0002-projection-model-sustained-regime.md) §2, §6, §8 and §11, and [ADR 0003](0003-service-lifecycle-and-sanctioned-toggle.md) §4 and §8 for the TUI baseline entry.
## Context
Under ADR 0002 as implemented, a fresh install showed no usage-adjusted theoretical lifespan until a manual baseline existed and a complete local observation day had passed. Confidence then stayed Limited until two scenario horizons existed, about 28 days. Users read this as Fenris being slow and its confidence never moving. The architecture review of 2026-10-02 also found that the TUI and CLI each pre-gated the projection with rules the projection itself does not use.
## Decision
1. **Projection stages.** The projection reports one projection stage: no observations yet, provisional, warming, limited, supported, or unavailable. Provisional and warming add to ADR 0002's three confidence states, which remain the categorical judgement once warm-up ends. Callers render the stage. They never pre-gate.
2. **First number after 3 observed hours.** A provisional projection appears once at least 3 hours of hour observations with samples exist. Until 24 hours are observed, it carries the fact "daily cycle not yet seen". From 24 hours until ADR 0002's 14-qualifying-day warm-up completes, the stage is warming. The complete-local-day condition is a contributing fact, not a gate. The rate formula is unchanged; only the evidence span is shorter.
3. **Short horizons.** Until a 7-day horizon is covered, the scenario range uses 24-hour and 3-day horizons, each shown only once covered. They drop out when the 7-day horizon exists. The scenario range remains the only spread shown, and still no statistical interval appears.
4. **Horizon agreement over existing horizons.** As ADR 0002 §8 already says, agreement is judged across the horizons that exist. A single 7-day horizon passes, so Supported is reachable at 14 qualifying days when every other condition holds.
5. **Evidence ladder.** Each Supported condition from ADR 0002 §8 is reported as met or unmet with its reason. The headline shows a count of conditions met, and the outlook shows the full ladder. Confidence is never a percentage.
6. **Write rate without a baseline.** When no endurance baseline applies, the sustained-regime write rate is shown in place of a lifespan. No baseline is synthesized.
7. **TUI baseline entry.** The TUI offers an unverified-override entry: rated TBW plus an optional source URL. It is persisted through `fenris-monitor baseline set` under the existing polkit action. Verified overrides with full provenance stay CLI-only.
## Considered options
- **Numeric confidence percentage.** Rejected: it implies precision the evidence cannot support, and ADR 0002 §8 already rejected it.
- **Capacity-based default baseline.** Rejected: ADR 0002 dropped `capacity × 600`, and a fabricated baseline would mislabel the lifespan.
- **Keep the complete-local-day gate.** Rejected: it delays the first number by up to 34 hours while adding nothing the provisional label and spread do not already disclose.
## Consequences
- Early lifespans swing widely. The provisional label, the hours observed and the short-horizon spread carry that honestly.
- The projection module's interface becomes the single test surface for every gating rule.
- ADR 0002 §10's implied baseline still needs two Percentage Used increments. Entering a rated TBW remains the fast path to a lifespan.
+4 -3
View File
@@ -95,9 +95,10 @@ finance, email, or gallery features.
reference does not require squeezing its entire density into small terminals. reference does not require squeezing its entire density into small terminals.
These defaults implement the confirmed choices. Date entry and inspection These defaults implement the confirmed choices. Date entry and inspection
retain existing evidence precision: hourly/daily UTC evidence is labelled UTC, retain existing evidence precision: every time is shown in local time with its
while local-day totals keep their recorded timezone. This visual redesign does timezone label and a secondary UTC line on inspection, daily bars read
not manufacture finer or local-hour precision from coarse UTC evidence. local-day evidence, and local-day totals keep their recorded timezone. This
visual redesign does not manufacture finer precision than the evidence holds.
## Smallest sufficient implementation proof ## Smallest sufficient implementation proof
+1 -1
View File
@@ -1,6 +1,6 @@
[project] [project]
name = "fenris" name = "fenris"
version = "0.6.0" version = "0.7.0"
description = "NVMe wear monitor with persistent TUI" description = "NVMe wear monitor with persistent TUI"
requires-python = ">=3.10" requires-python = ">=3.10"
license = {file = "LICENSE"} license = {file = "LICENSE"}
+35 -8
View File
@@ -21,13 +21,19 @@ class VolumePoint:
def volume_plot( def volume_plot(
points: Sequence[VolumePoint], width: int, height: int, points: Sequence[VolumePoint], width: int, height: int,
selected: int, colors: Mapping[str, str], selected: int, colors: Mapping[str, str],
secondary: Sequence[VolumePoint] | None = None,
) -> tuple[Text, list[int], str]: ) -> tuple[Text, list[int], str]:
"""Render bounded axes and a dotted trace, without resampling evidence.""" """Render bounded axes and a dotted trace, without resampling evidence.
*secondary* is a second series over the same time axis (reads beside
writes); it shares the volume scale and draws in its own colour.
"""
width, height = max(12, width), max(4, height) width, height = max(12, width), max(4, height)
axis_width = 8 axis_width = 8
columns, rows = width - axis_width, height - 2 columns, rows = width - axis_width, height - 2
pixel_width, pixel_height = columns * 2, rows * 4 pixel_width, pixel_height = columns * 2, rows * 4
maximum = max((p.volume or 0 for p in points), default=0) maximum = max(
(p.volume or 0 for p in (*points, *(secondary or ()))), default=0)
scale, unit = (1e12, "TB") if maximum >= 1e12 else ( scale, unit = (1e12, "TB") if maximum >= 1e12 else (
(1e9, "GB") if maximum >= 1e9 else (1e9, "GB") if maximum >= 1e9 else
(1e6, "MB") if maximum >= 1e6 else (1e6, "MB") if maximum >= 1e6 else
@@ -39,11 +45,16 @@ def volume_plot(
xs = [round((p.time - start) / span * (pixel_width - 1)) if span else 0 xs = [round((p.time - start) / span * (pixel_width - 1)) if span else 0
for p in points] for p in points]
cells = [[0] * columns for _ in range(rows)] cells = [[0] * columns for _ in range(rows)]
second = [[0] * columns for _ in range(rows)]
# Unicode braille dot numbering, indexed by y within cell and then x. # Unicode braille dot numbering, indexed by y within cell and then x.
bits = ((1, 8), (2, 16), (4, 32), (64, 128)) bits = ((1, 8), (2, 16), (4, 32), (64, 128))
def dot(x: int, y: int) -> None: def dot(x: int, y: int, grid: list[list[int]] = cells) -> None:
cells[y // 4][x // 2] |= bits[y % 4][x % 2] grid[y // 4][x // 2] |= bits[y % 4][x % 2]
def y_of(volume: int) -> int:
y = round((1 - volume / ceiling) * (pixel_height - 1))
return min(pixel_height - 1, max(0, y))
previous = None previous = None
markers = {} markers = {}
@@ -53,8 +64,7 @@ def volume_plot(
if point.state != "future": if point.state != "future":
markers[x // 2] = "?" markers[x // 2] = "?"
continue continue
y = round((1 - point.volume / ceiling) * (pixel_height - 1)) y = y_of(point.volume)
y = min(pixel_height - 1, max(0, y))
if previous is not None and point.state == "measured": if previous is not None and point.state == "measured":
px, py = previous px, py = previous
steps = max(abs(x - px), abs(y - py), 1) steps = max(abs(x - px), abs(y - py), 1)
@@ -68,6 +78,21 @@ def volume_plot(
elif point.volume == 0: elif point.volume == 0:
markers.setdefault(x // 2, "·") markers.setdefault(x // 2, "·")
previous = None
for point, x in zip(secondary or (), xs):
if point.volume is None:
previous = None
continue
y = y_of(point.volume)
if previous is not None and point.state == "measured":
px, py = previous
steps = max(abs(x - px), abs(y - py), 1)
for step in range(steps + 1):
dot(round(px + (x - px) * step / steps),
round(py + (y - py) * step / steps), second)
dot(x, y, second)
previous = (x, y) if point.state == "measured" else None
selected_column = xs[selected] // 2 if 0 <= selected < len(xs) else -1 selected_column = xs[selected] // 2 if 0 <= selected < len(xs) else -1
result = Text(no_wrap=True, overflow="crop") result = Text(no_wrap=True, overflow="crop")
ticks = {0, rows // 2, rows - 1} ticks = {0, rows // 2, rows - 1}
@@ -76,8 +101,10 @@ def volume_plot(
label = f"{value:6.2f}"[-6:] if row in ticks else " " label = f"{value:6.2f}"[-6:] if row in ticks else " "
result.append(label + " │", style=colors["muted"]) result.append(label + " │", style=colors["muted"])
for col, value in enumerate(values): for col, value in enumerate(values):
char = chr(0x2800 + value) if value else " " extra = second[row][col]
style = colors["allocated"] char = chr(0x2800 + (value | extra)) if value | extra else " "
style = colors["allocated"] if value or not extra else colors["secondary"]
value |= extra
if col == selected_column: if col == selected_column:
style = "bold " + colors["selection"] style = "bold " + colors["selection"]
if not value: if not value:
File diff suppressed because it is too large Load Diff
+8
View File
@@ -3,6 +3,8 @@ from collections.abc import Sequence
from dataclasses import dataclass from dataclasses import dataclass
HISTORY_RANGE_DEFAULT = 14 HISTORY_RANGE_DEFAULT = 14
HISTORY_RANGE_OPTIONS = (7, 14, 30, 90)
HISTORY_RANGE_MAX = max(HISTORY_RANGE_OPTIONS)
@dataclass(frozen=True) @dataclass(frozen=True)
@@ -33,6 +35,7 @@ class ActivitySelection:
self.selected_history_hour: str | None = None self.selected_history_hour: str | None = None
self.history_range_days = HISTORY_RANGE_DEFAULT self.history_range_days = HISTORY_RANGE_DEFAULT
self.measure = "written" self.measure = "written"
self.diagnostics = False
self.following_live = True self.following_live = True
self.selected_live_interval: IntervalIdentity | None = None self.selected_live_interval: IntervalIdentity | None = None
self.live_interval_expired = False self.live_interval_expired = False
@@ -72,6 +75,11 @@ class ActivitySelection:
self.measure = "read" if self.measure == "written" else "written" self.measure = "read" if self.measure == "written" else "written"
return self.measure return self.measure
def toggle_diagnostics(self) -> bool:
"""Show or hide the Diagnostics panel and read-beside-write bars."""
self.diagnostics = not self.diagnostics
return self.diagnostics
def follow_live(self) -> None: def follow_live(self) -> None:
"""Resume following the newest interval and clear stale-pin notices.""" """Resume following the newest interval and clear stale-pin notices."""
self.view = "live" self.view = "live"
+11 -4
View File
@@ -19,7 +19,7 @@ from typing import Any, Dict, Optional
from .derive import derive_hours_from_interval, find_previous_sample from .derive import derive_hours_from_interval, find_previous_sample
from .monitoring_periods import ensure_period_open from .monitoring_periods import ensure_period_open
from .store import init_store, get_store_path from .store import DIAGNOSTIC_SAMPLE_COLUMNS, init_store, get_store_path
class AcquisitionError(Exception): class AcquisitionError(Exception):
@@ -103,6 +103,8 @@ def acquire_from_smartctl(smartctl_data: Dict[str, Any]) -> Dict[str, Any]:
"bytes_written": log["data_units_written"] * 512000, "bytes_written": log["data_units_written"] * 512000,
"bytes_read": log["data_units_read"] * 512000, "bytes_read": log["data_units_read"] * 512000,
"critical_warning": log.get("critical_warning", 0), "critical_warning": log.get("critical_warning", 0),
# Diagnostic counters (issue #111); absent keys stay None, never zero.
**{column: log.get(column) for column in DIAGNOSTIC_SAMPLE_COLUMNS},
} }
@@ -252,6 +254,7 @@ def write_sample(
segment_id = current_segment["id"] if current_segment else None segment_id = current_segment["id"] if current_segment else None
# Insert sample with segment_id # Insert sample with segment_id
# nosemgrep: python.sqlalchemy.security.sqlalchemy-execute-raw-query.sqlalchemy-execute-raw-query -- columns come from module constants, values are bound
cursor = conn.execute( cursor = conn.execute(
""" """
INSERT INTO samples ( INSERT INTO samples (
@@ -259,9 +262,10 @@ def write_sample(
percentage_used, available_spare, media_errors, power_on_hours, percentage_used, available_spare, media_errors, power_on_hours,
power_cycles, unsafe_shutdowns, temperature_c, power_cycles, unsafe_shutdowns, temperature_c,
data_units_written, data_units_read, bytes_written, bytes_read, data_units_written, data_units_read, bytes_written, bytes_read,
critical_warning, segment_id, local_tz critical_warning, segment_id, local_tz, %s
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, %s)
""", """ % (", ".join(DIAGNOSTIC_SAMPLE_COLUMNS),
", ".join("?" * len(DIAGNOSTIC_SAMPLE_COLUMNS))),
( (
sample["ts"], sample["ts"],
sample["device"], sample["device"],
@@ -284,6 +288,8 @@ def write_sample(
sample["critical_warning"], sample["critical_warning"],
segment_id, segment_id,
sample.get("local_tz"), sample.get("local_tz"),
# Samples staged by an earlier version lack these keys.
*(sample.get(column) for column in DIAGNOSTIC_SAMPLE_COLUMNS),
), ),
) )
@@ -382,6 +388,7 @@ def _stage_observation(
separators=(",", ":"), separators=(",", ":"),
sort_keys=True, sort_keys=True,
) )
# nosemgrep: python.sqlalchemy.security.sqlalchemy-execute-raw-query.sqlalchemy-execute-raw-query -- columns come from module constants, values are bound
cursor = conn.execute( cursor = conn.execute(
"INSERT INTO pending_publications (sample_ts, payload) VALUES (?, ?)", "INSERT INTO pending_publications (sample_ts, payload) VALUES (?, ?)",
(sample["ts"], payload), (sample["ts"], payload),
+2 -1
View File
@@ -32,6 +32,7 @@ if VENV_DIR.exists():
sys.path.insert(0, str(site_packages)) sys.path.insert(0, str(site_packages))
from fenris.store import init_store # noqa: E402 -- must follow the sys.path bootstrap above from fenris.store import init_store # noqa: E402 -- must follow the sys.path bootstrap above
from fenris.timefmt import local_timestamp, local_zone_name # noqa: E402 -- sys.path bootstrap above
from fenris.monitoring_periods import ( # noqa: E402 -- sys.path bootstrap above from fenris.monitoring_periods import ( # noqa: E402 -- sys.path bootstrap above
ensure_period_open, ensure_period_open,
close_period, close_period,
@@ -70,7 +71,7 @@ def cmd_enable(args: argparse.Namespace) -> None:
if open_period is None: if open_period is None:
ensure_period_open(conn, now) ensure_period_open(conn, now)
conn.commit() conn.commit()
print("Monitoring period opened at", now.isoformat()) print("Monitoring period opened at", local_timestamp(now, local_zone_name()))
else: else:
print("Monitoring period already open (id=%d)" % open_period["id"]) print("Monitoring period already open (id=%d)" % open_period["id"])
+250 -109
View File
@@ -20,6 +20,11 @@ Arithmetic (§6.3):
E_rated = entered_TBW × 10¹² bytes E_rated = entered_TBW × 10¹² bytes
E_implied = 100 · W_t / p (1 ≤ p ≤ 254) E_implied = 100 · W_t / p (1 ≤ p ≤ 254)
Staging (ADR 0012): the result carries a projection stage, the evidence
ladder (each Supported condition met or unmet with its reason) and the
sustained-regime write rate, so callers render the stage and never
pre-gate the projection themselves.
Criteria: PR-1–PR-17, CI-4. Criteria: PR-1–PR-17, CI-4.
""" """
import sqlite3 import sqlite3
@@ -35,6 +40,9 @@ from .monitoring_periods import interval_within_one_monitoring_period
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
HORIZON_DAYS = (7, 28, 90) HORIZON_DAYS = (7, 28, 90)
SHORT_HORIZON_DAYS = (1, 3)
PROVISIONAL_MIN_HOURS = 3
DAILY_CYCLE_HOURS = 24
TBW_TO_BYTES = 10 ** 12 TBW_TO_BYTES = 10 ** 12
IMPLIED_P_MIN = 1 IMPLIED_P_MIN = 1
IMPLIED_P_MAX = 254 IMPLIED_P_MAX = 254
@@ -62,6 +70,15 @@ class ConfidenceState(Enum):
SUPPORTED = "Supported" SUPPORTED = "Supported"
class ProjectionStage(Enum):
NO_OBSERVATIONS = "No observations yet"
PROVISIONAL = "Provisional"
WARMING = "Warming"
LIMITED = "Limited"
SUPPORTED = "Supported"
UNAVAILABLE = "Unavailable"
class BaselineTier(Enum): class BaselineTier(Enum):
VERIFIED = "verified_override" VERIFIED = "verified_override"
UNVERIFIED = "unverified_override" UNVERIFIED = "unverified_override"
@@ -77,6 +94,14 @@ class ScenarioRange:
horizon_reasons: Dict[int, str] = field(default_factory=dict) horizon_reasons: Dict[int, str] = field(default_factory=dict)
@dataclass(frozen=True)
class LadderCondition:
"""One Supported condition from ADR 0002 §8, met or unmet with its reason."""
name: str
met: bool
reason: str
@dataclass(frozen=True) @dataclass(frozen=True)
class ProjectionResult: class ProjectionResult:
confidence_state: ConfidenceState confidence_state: ConfidenceState
@@ -94,6 +119,25 @@ class ProjectionResult:
degraded_identity_fact: Optional[str] degraded_identity_fact: Optional[str]
zero_rate_fact: Optional[str] zero_rate_fact: Optional[str]
qualifying_days_progress: Optional[str] = None # Issue #77: honest qualifying-day progress qualifying_days_progress: Optional[str] = None # Issue #77: honest qualifying-day progress
stage: ProjectionStage = ProjectionStage.NO_OBSERVATIONS
evidence_ladder: List[LadderCondition] = field(default_factory=list)
write_rate: Optional[float] = None # sustained-regime rate, bytes per second
hours_observed: int = 0
@property
def stage_label(self) -> str:
"""Display wording; observations that are too few are not "none"."""
if self.stage == ProjectionStage.NO_OBSERVATIONS and self.hours_observed > 0:
return "Collecting observations"
return self.stage.value
@property
def ladder_met(self) -> int:
return sum(1 for c in self.evidence_ladder if c.met)
@property
def ladder_total(self) -> int:
return len(self.evidence_ladder)
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
@@ -150,16 +194,40 @@ def _get_all_days(conn):
for r in cursor.fetchall()] for r in cursor.fetchall()]
def _get_latest_published_sample(conn, segment_id):
"""Newest published sample in the segment: (lifetime bytes written, PU).
W_t is the lifetime counter, not a regime delta. Pending staging lives
in a separate table (ADR 0011), so only published samples are seen here.
bytes_written is DUW x 512000 as stored by the collector; fall back to
that conversion if only the raw counter is present.
"""
if segment_id is not None:
cursor = conn.execute(
"SELECT COALESCE(bytes_written, data_units_written * 512000), percentage_used "
"FROM samples WHERE segment_id = ? ORDER BY id DESC LIMIT 1",
(segment_id,),
)
else:
cursor = conn.execute(
"SELECT COALESCE(bytes_written, data_units_written * 512000), percentage_used "
"FROM samples ORDER BY id DESC LIMIT 1"
)
row = cursor.fetchone()
return (row[0], row[1]) if row else (None, None)
def _get_latest_pu(conn): def _get_latest_pu(conn):
cursor = conn.execute("SELECT percentage_used FROM samples ORDER BY id DESC LIMIT 1") cursor = conn.execute("SELECT percentage_used FROM samples ORDER BY id DESC LIMIT 1")
row = cursor.fetchone() row = cursor.fetchone()
return row[0] if row else None return row[0] if row else None
def _get_pu_increments_in_segment(conn, segment_opened_at): def _get_pu_increments_in_segment(conn, segment_id):
cursor = conn.execute( cursor = conn.execute(
"SELECT COUNT(DISTINCT percentage_used) FROM samples WHERE ts >= ?", "SELECT COUNT(DISTINCT percentage_used) FROM samples "
(segment_opened_at,), "WHERE segment_id = ? AND percentage_used IS NOT NULL",
(segment_id,),
) )
row = cursor.fetchone() row = cursor.fetchone()
return max(0, (row[0] if row else 0) - 1) return max(0, (row[0] if row else 0) - 1)
@@ -182,15 +250,50 @@ def _wall_clock_in_range(conn, start, end):
return total return total
def _get_hours_observed(conn, segment_opened_at, days):
"""Hours of sampled evidence in the current segment.
Counted from hour observations. A day aggregate without any hour rows
(imported or pruned history) stands for a whole observed day.
"""
hour_floor = segment_opened_at[:13] if segment_opened_at else ""
cursor = conn.execute(
"SELECT substr(hour, 1, 10), COUNT(*) FROM hour_observations "
"WHERE sample_count > 0 AND hour >= ? GROUP BY 1",
(hour_floor,),
)
by_day = dict(cursor.fetchall())
total = 0
for d in days:
if d["day"] in by_day:
total += by_day.pop(d["day"])
elif d["sample_count"] > 0:
total += DAILY_CYCLE_HOURS
return total + sum(by_day.values())
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# Baseline resolution (§6.1, §6.2) # Baseline resolution (§6.1, §6.2)
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
def _resolve_implied_baseline(conn, current_segment, facts):
"""No override exists: Percentage-Used-implied tier, or unavailable (§10)."""
if current_segment is not None:
_, p = _get_latest_published_sample(conn, current_segment["id"])
increments = _get_pu_increments_in_segment(conn, current_segment["id"])
if (increments >= IMPLIED_MIN_PU_INCREMENTS and p is not None
and IMPLIED_P_MIN <= p <= IMPLIED_P_MAX):
return (BaselineTier.IMPLIED, None,
"implied from vendor wear (%d%% used) — coarse" % p, facts)
facts.append("vendor wear estimate too coarse to imply endurance")
return BaselineTier.NONE, None, "no baseline", facts
def _resolve_baseline(conn, current_segment): def _resolve_baseline(conn, current_segment):
baseline = _get_baseline(conn) baseline = _get_baseline(conn)
facts: list[str] = [] facts: list[str] = []
if baseline is None: if baseline is None:
return BaselineTier.NONE, None, "no baseline", facts return _resolve_implied_baseline(conn, current_segment, facts)
mandatory = [baseline["source_url"], baseline["document_revision"], mandatory = [baseline["source_url"], baseline["document_revision"],
baseline["entry_date"], baseline["model_string"], baseline["entry_date"], baseline["model_string"],
@@ -341,84 +444,88 @@ def _detect_habit_change(days):
# Confidence rule table (§6.7) # Confidence rule table (§6.7)
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
def _evaluate_confidence(tier, rate, regime_days, days, current_segment, DEGRADED_IDENTITY_FACT = "controller identity unavailable — replacement detection relies on write-counter continuity only"
clock_now, warming_days, warming_low_coverage,
habit_change, staleness_hours, scenario_range):
facts = []
if tier == BaselineTier.NONE:
facts.append("no applicable endurance baseline")
return ConfidenceState.UNSUPPORTED, facts
if rate is None or rate <= 0: def _build_evidence_ladder(tier, regime_days, days, current_segment,
facts.append("no finite projection from this history") staleness_hours, scenario_range):
return ConfidenceState.UNSUPPORTED, facts """The eight Supported conditions of ADR 0002 §8, each met or unmet."""
ladder = []
supported_facts = [] def add(name, met, met_reason, unmet_reason):
failing = False ladder.append(LadderCondition(name, met, met_reason if met else unmet_reason))
# 1. Verified baseline add("verified baseline", tier == BaselineTier.VERIFIED,
if tier != BaselineTier.VERIFIED: "verified manufacturer TBW",
failing = True {BaselineTier.UNVERIFIED: "baseline is unverified (user-supplied)",
else: BaselineTier.IMPLIED: "baseline is implied from vendor wear — coarse"}.get(
supported_facts.append("verified manufacturer TBW") tier, "no applicable endurance baseline"))
# 2. >= 14 qualifying days qualifying = sum(1 for d in days
qualifying = sum(1 for d in days if d["coverage"] >= WARMING_COVERAGE_FLOOR and d["sample_count"] > 0) if d["coverage"] >= WARMING_COVERAGE_FLOOR and d["sample_count"] > 0)
if qualifying < WARMING_MIN_DAYS: add("14 qualifying days", qualifying >= WARMING_MIN_DAYS,
failing = True "%d calendar days" % qualifying,
else: "%d of %d qualifying days" % (qualifying, WARMING_MIN_DAYS))
supported_facts.append("%d calendar days" % qualifying)
# 3. Coverage >= 80%
total_wc = len(days) * 86400 total_wc = len(days) * 86400
total_known = sum(int(d["coverage"] * 86400) for d in days) total_known = sum(int(d["coverage"] * 86400) for d in days)
avg_cov = total_known / total_wc if total_wc > 0 else 0.0 avg_cov = total_known / total_wc if total_wc > 0 else 0.0
if avg_cov < SUPPORTED_COVERAGE_FLOOR: add("coverage at least 80%", avg_cov >= SUPPORTED_COVERAGE_FLOOR,
failing = True "%d%% interval coverage" % int(avg_cov * 100),
"interval coverage %d%% (≥%d%% required)"
% (int(avg_cov * 100), int(SUPPORTED_COVERAGE_FLOOR * 100)))
if staleness_hours is None:
add("fresh data", False, "", "no observation days yet")
else: else:
supported_facts.append("%d%% interval coverage" % int(avg_cov * 100)) add("fresh data", staleness_hours <= STALENESS_HOURS, "recent data",
"newest data %dh old (≥%dh)" % (staleness_hours, STALENESS_HOURS))
# 4. Fresh (< 48h) if scenario_range is None or not scenario_range.rates:
if staleness_hours is not None and staleness_hours > STALENESS_HOURS: add("horizons agree", False, "", "no scenario horizon covered yet")
failing = True else:
elif staleness_hours is not None:
supported_facts.append("recent data")
# 5. Horizon agreement
if scenario_range is not None and len(scenario_range.rates) >= 2:
rl = list(scenario_range.rates.values()) rl = list(scenario_range.rates.values())
if min(rl) > 0 and max(rl) / min(rl) > HORIZON_AGREEMENT_FACTOR: spread = max(rl) / min(rl) if min(rl) > 0 else None
failing = True agree = spread is None or spread <= HORIZON_AGREEMENT_FACTOR
else: add("horizons agree", agree, "%d weekly cycles" % len(rl),
supported_facts.append("%d weekly cycles" % len(scenario_range.rates)) "horizon rates differ by %.1f× (≤%d× required)"
else: % (spread or 0.0, HORIZON_AGREEMENT_FACTOR))
failing = True
# 6. Burst guard recent = days[-BURST_GUARD_LOOKBACK:]
if not failing and len(days) >= BURST_GUARD_LOOKBACK: t28 = sum(d["bytes_written"] for d in recent)
t28 = sum(d["bytes_written"] for d in days[-BURST_GUARD_LOOKBACK:]) burst = t28 > 0 and any(d["bytes_written"] >= BURST_GUARD_FRACTION * t28 for d in recent)
for d in days[-BURST_GUARD_LOOKBACK:]: add("no burst day", bool(recent) and not burst, "no burst days",
if t28 > 0 and d["bytes_written"] >= BURST_GUARD_FRACTION * t28: "one day carries ≥%d%% of trailing writes" % int(BURST_GUARD_FRACTION * 100)
failing = True if burst else "no observation days yet")
break
if not failing:
supported_facts.append("no burst days")
# 7. Regime >= 7 days add("regime at least 7 days", regime_days >= YOUNG_REGIME_DAYS,
if regime_days < YOUNG_REGIME_DAYS: "regime %d days" % regime_days,
failing = True "regime only %d days old (≥%d required)" % (regime_days, YOUNG_REGIME_DAYS))
# 8. Degraded identity degraded = bool(current_segment and current_segment.get("identity_degraded"))
if current_segment and current_segment.get("identity_degraded"): add("controller identity", not degraded, "controller identity available",
failing = True DEGRADED_IDENTITY_FACT)
facts.append("controller identity unavailable — replacement detection relies on write-counter continuity only") return ladder
if not failing:
return ConfidenceState.SUPPORTED, supported_facts
# Limited def _evaluate_confidence(tier, rate, regime_days, days, current_segment,
limited_facts = list(supported_facts) habit_change, staleness_hours, scenario_range):
"""Return (confidence state, facts, evidence ladder)."""
ladder = _build_evidence_ladder(tier, regime_days, days, current_segment,
staleness_hours, scenario_range)
if tier == BaselineTier.NONE:
return ConfidenceState.UNSUPPORTED, ["no applicable endurance baseline"], ladder
if rate is None or rate <= 0:
return ConfidenceState.UNSUPPORTED, ["no finite projection from this history"], ladder
met_facts = [c.reason for c in ladder if c.met]
if all(c.met for c in ladder):
return ConfidenceState.SUPPORTED, met_facts, ladder
# Limited: met facts plus the contributing failures
limited_facts = list(met_facts)
if staleness_hours is not None and staleness_hours > STALENESS_HOURS: if staleness_hours is not None and staleness_hours > STALENESS_HOURS:
limited_facts.append("newest data %dh old (≥48h)" % staleness_hours) limited_facts.append("newest data %dh old (≥48h)" % staleness_hours)
if habit_change is not None: if habit_change is not None:
@@ -426,11 +533,27 @@ def _evaluate_confidence(tier, rate, regime_days, days, current_segment,
if regime_days < YOUNG_REGIME_DAYS: if regime_days < YOUNG_REGIME_DAYS:
limited_facts.append("regime only %d days old (≥7 required)" % regime_days) limited_facts.append("regime only %d days old (≥7 required)" % regime_days)
if current_segment and current_segment.get("identity_degraded"): if current_segment and current_segment.get("identity_degraded"):
degraded_fact = "controller identity unavailable — replacement detection relies on write-counter continuity only" if DEGRADED_IDENTITY_FACT not in limited_facts:
if degraded_fact not in limited_facts: limited_facts.append(DEGRADED_IDENTITY_FACT)
limited_facts.append(degraded_fact) return ConfidenceState.LIMITED, limited_facts, ladder
return ConfidenceState.LIMITED, limited_facts
def ladder_count_text(result):
"""Headline form of the evidence ladder: how many conditions are met."""
return "%d of %d conditions met" % (result.ladder_met, result.ladder_total)
def ladder_lines(result):
"""Full evidence ladder for the outlook, one line per condition."""
return ["[%s] %s — %s" % ("x" if c.met else " ", c.name, c.reason)
for c in result.evidence_ladder]
def write_rate_gb_day(result):
"""The sustained-regime write rate in GB/day, or None when not yet known."""
if result.write_rate is None:
return None
return result.write_rate * 86400 / 1e9
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
@@ -511,6 +634,21 @@ def _has_complete_local_day(conn):
# Main projection function # Main projection function
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
def _determine_stage(state, hours_observed, total_days, days_below_coverage):
"""Where the projection stands on its way to support (ADR 0012 §1, §2)."""
if hours_observed < PROVISIONAL_MIN_HOURS:
return ProjectionStage.NO_OBSERVATIONS
if state == ConfidenceState.UNSUPPORTED:
return ProjectionStage.UNAVAILABLE
if hours_observed < DAILY_CYCLE_HOURS:
return ProjectionStage.PROVISIONAL
if total_days < WARMING_MIN_DAYS or days_below_coverage > WARMING_MAX_LOW_COVERAGE:
return ProjectionStage.WARMING
if state == ConfidenceState.SUPPORTED:
return ProjectionStage.SUPPORTED
return ProjectionStage.LIMITED
def compute_projection(conn, clock_now): def compute_projection(conn, clock_now):
facts = [] facts = []
habit_change_fact = None habit_change_fact = None
@@ -523,32 +661,14 @@ def compute_projection(conn, clock_now):
tier, baseline, baseline_label, baseline_facts = _resolve_baseline(conn, current_segment) tier, baseline, baseline_label, baseline_facts = _resolve_baseline(conn, current_segment)
facts.extend(baseline_facts) facts.extend(baseline_facts)
# --- Complete observation day gate (issue #94) --- # The complete-local-day condition is a fact, not a gate (ADR 0012 §2).
# An endurance outlook requires at least one complete local if not _has_complete_local_day(conn):
# midnight-to-midnight calendar day with usable observation evidence.
has_complete_day = _has_complete_local_day(conn)
if not has_complete_day:
facts.append("waiting for a full local observation day") facts.append("waiting for a full local observation day")
return ProjectionResult(
confidence_state=ConfidenceState.UNSUPPORTED,
contributing_facts=facts,
headline_remaining_seconds=None,
scenario_range=None,
pu_context_line="Percentage Used: unknown" if _get_latest_pu(conn) is None else "Percentage Used: %d%%" % (_get_latest_pu(conn) or 0),
disclosure_text=list(DISCLOSURES),
baseline_tier=tier,
baseline_label=baseline_label,
regime_days=None,
habit_change_fact=None,
warming_fact=None,
staleness_fact=None,
degraded_identity_fact=None,
zero_rate_fact=None,
qualifying_days_progress=None,
)
segment_days = _get_days_in_segment(conn, current_segment["opened_at"]) if current_segment else _get_all_days(conn) segment_days = _get_days_in_segment(conn, current_segment["opened_at"]) if current_segment else _get_all_days(conn)
all_days = _get_all_days(conn) all_days = _get_all_days(conn)
hours_observed = _get_hours_observed(
conn, current_segment["opened_at"] if current_segment else None, segment_days)
regime_start_day = None regime_start_day = None
habit_change = None habit_change = None
@@ -578,12 +698,15 @@ def compute_projection(conn, clock_now):
scenario = None scenario = None
horizon_rates = {} horizon_rates = {}
horizon_reasons = {} horizon_reasons = {}
for h in HORIZON_DAYS: for h in SHORT_HORIZON_DAYS + HORIZON_DAYS:
hr, reason = _compute_horizon_rate(all_days, conn, h, clock_now) hr, reason = _compute_horizon_rate(all_days, conn, h, clock_now)
if hr is not None: if hr is not None:
horizon_rates[h] = hr horizon_rates[h] = hr
else: elif h in HORIZON_DAYS:
horizon_reasons[h] = reason horizon_reasons[h] = reason
if HORIZON_DAYS[0] in horizon_rates:
for h in SHORT_HORIZON_DAYS:
horizon_rates.pop(h, None)
if horizon_rates: if horizon_rates:
scenario = ScenarioRange( scenario = ScenarioRange(
rates=horizon_rates, rates=horizon_rates,
@@ -619,33 +742,47 @@ def compute_projection(conn, clock_now):
facts.append(staleness_fact) facts.append(staleness_fact)
if current_segment and current_segment.get("identity_degraded"): if current_segment and current_segment.get("identity_degraded"):
degraded_identity_fact = "controller identity unavailable — replacement detection relies on write-counter continuity only" degraded_identity_fact = DEGRADED_IDENTITY_FACT
facts.append(degraded_identity_fact) facts.append(degraded_identity_fact)
state, conf_facts = _evaluate_confidence( state, conf_facts, ladder = _evaluate_confidence(
tier, rate, regime_days_count, segment_days, current_segment, clock_now, tier, rate, regime_days_count, segment_days, current_segment,
qualifying, 0, habit_change, staleness_hours, scenario, habit_change, staleness_hours, scenario,
) )
stage = _determine_stage(state, hours_observed, total_days_count, days_below_coverage)
if stage == ProjectionStage.NO_OBSERVATIONS:
# Too little evidence for any number: no rate, no lifespan.
state = ConfidenceState.UNSUPPORTED
rate = None
scenario = None
if hours_observed > 0:
facts.insert(0, "%d of %d hours observed before the first projection"
% (hours_observed, PROVISIONAL_MIN_HOURS))
elif stage == ProjectionStage.PROVISIONAL:
# Short-horizon spread: the rate over the hours observed so far.
scenario = ScenarioRange(rates={SHORT_HORIZON_DAYS[0]: rate},
min_days=SHORT_HORIZON_DAYS[0],
max_days=SHORT_HORIZON_DAYS[0])
facts[:0] = ["%d hours observed" % hours_observed, "daily cycle not yet seen"]
all_facts = list(facts) all_facts = list(facts)
for cf in conf_facts: for cf in conf_facts:
if cf not in all_facts: if cf not in all_facts:
all_facts.append(cf) all_facts.append(cf)
headline_seconds = None headline_seconds = None
if state != ConfidenceState.UNSUPPORTED and rate is not None and rate > 0 and baseline is not None: if state != ConfidenceState.UNSUPPORTED and rate is not None and rate > 0:
if tier in (BaselineTier.VERIFIED, BaselineTier.UNVERIFIED): W_t, p = _get_latest_published_sample(
conn, current_segment["id"] if current_segment else None)
E_baseline = None
if tier in (BaselineTier.VERIFIED, BaselineTier.UNVERIFIED) and baseline is not None:
E_baseline = baseline["tbw_terabytes"] * TBW_TO_BYTES E_baseline = baseline["tbw_terabytes"] * TBW_TO_BYTES
elif tier == BaselineTier.IMPLIED: elif (tier == BaselineTier.IMPLIED and W_t is not None and p is not None
p = _get_latest_pu(conn) and IMPLIED_P_MIN <= p <= IMPLIED_P_MAX):
if p is not None and IMPLIED_P_MIN <= p <= IMPLIED_P_MAX: E_baseline = 100 * W_t / p
E_baseline = 100 * regime_bytes / p if E_baseline is not None and W_t is not None:
else: headline_seconds = max(E_baseline - W_t, 0) / rate
E_baseline = None
else:
E_baseline = None
if E_baseline is not None:
headline_seconds = max(E_baseline - regime_bytes, 0) / rate
pu_line = _build_pu_context_line(conn, rate, segment_days, clock_now) pu_line = _build_pu_context_line(conn, rate, segment_days, clock_now)
@@ -665,4 +802,8 @@ def compute_projection(conn, clock_now):
degraded_identity_fact=degraded_identity_fact, degraded_identity_fact=degraded_identity_fact,
zero_rate_fact=zero_rate_fact, zero_rate_fact=zero_rate_fact,
qualifying_days_progress=qualifying_days_progress, qualifying_days_progress=qualifying_days_progress,
stage=stage,
evidence_ladder=ladder,
write_rate=rate,
hours_observed=hours_observed,
) )
+19 -29
View File
@@ -22,7 +22,10 @@ from typing import Any, Dict, Iterator, List, Optional, Tuple, TYPE_CHECKING
if TYPE_CHECKING: if TYPE_CHECKING:
from .status_composition import StatusComposition from .status_composition import StatusComposition
from .projection import compute_projection, DISCLOSURES from .projection import (
compute_projection, DISCLOSURES, ProjectionStage, ladder_count_text,
ladder_lines, write_rate_gb_day,
)
from .store import SCHEMA_VERSION from .store import SCHEMA_VERSION
from .init_system import ( from .init_system import (
query_service_state as _init_query_service_state, query_service_state as _init_query_service_state,
@@ -179,19 +182,6 @@ def grade_freshness(newest_sample_ts: Optional[str], clock_now: datetime) -> str
return "stale" 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) # Drive anomalies (§9.7 — FL-7)
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
@@ -254,8 +244,7 @@ def check_retired_flag(flag: str) -> Optional[str]:
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
def _format_projection(proj, freshness: str, drive_facts: List[str], def _format_projection(proj, freshness: str, drive_facts: List[str],
config_error: Optional[str], config_error: Optional[str]) -> str:
sample_count: int = 0, day_count: int = 0) -> str:
"""Format projection details; monitoring status has its own renderer.""" """Format projection details; monitoring status has its own renderer."""
lines = [] lines = []
@@ -271,15 +260,6 @@ def _format_projection(proj, freshness: str, drive_facts: List[str],
lines.append("Enable monitoring: fenris monitor resume") lines.append("Enable monitoring: fenris monitor resume")
return "\n".join(lines) return "\n".join(lines)
# --- Single sample: awaiting another sample (issue #73 AC3) ---
# Only show awaiting state when there are no day aggregates (e.g., legacy import
# or hand-crafted stores can have 1 sample but sufficient day data for projection)
if sample_count <= 1 and day_count == 0:
lines.append("awaiting another sample")
lines.append("")
lines.append("Collecting usage data — the first projection requires at least two samples.")
return "\n".join(lines)
if proj is None: if proj is None:
lines.append("no projection available") lines.append("no projection available")
return "\n".join(lines) return "\n".join(lines)
@@ -289,6 +269,9 @@ def _format_projection(proj, freshness: str, drive_facts: List[str],
lines.append(headline) lines.append(headline)
lines.append("") lines.append("")
# --- Stage and ladder count (ADR 0012) ---
lines.append("Projection stage: %s · %s" % (proj.stage_label, ladder_count_text(proj)))
# --- Confidence state + contributing facts (§6.7, §6.11) --- # --- Confidence state + contributing facts (§6.7, §6.11) ---
state_label = proj.confidence_state.value state_label = proj.confidence_state.value
facts_list = list(proj.contributing_facts) if proj.contributing_facts else [] facts_list = list(proj.contributing_facts) if proj.contributing_facts else []
@@ -304,6 +287,11 @@ def _format_projection(proj, freshness: str, drive_facts: List[str],
lines.append("%s evidence" % state_label) lines.append("%s evidence" % state_label)
lines.append("") lines.append("")
# --- Full evidence ladder in the outlook (ADR 0012 §5) ---
lines.append("Evidence ladder")
lines.extend(" " + line for line in ladder_lines(proj))
lines.append("")
# --- Scenario range (§6.5) with horizon reasons --- # --- Scenario range (§6.5) with horizon reasons ---
if proj.scenario_range: if proj.scenario_range:
parts = [] parts = []
@@ -332,8 +320,13 @@ def _format_projection(proj, freshness: str, drive_facts: List[str],
def _format_headline(proj) -> str: def _format_headline(proj) -> str:
"""Format the lifespan headline or its no-projection wording (§6.11).""" """Format the lifespan headline or its no-projection wording (§6.11)."""
if proj.headline_remaining_seconds is None: if proj.headline_remaining_seconds is None:
if proj.stage == ProjectionStage.NO_OBSERVATIONS:
return "no projection yet — collecting observations"
if proj.zero_rate_fact: if proj.zero_rate_fact:
return "no finite projection from this history" return "no finite projection from this history"
rate = write_rate_gb_day(proj)
if rate is not None:
return "write rate: %.2f GB/day · no lifespan without an endurance baseline" % rate
if proj.warming_fact: if proj.warming_fact:
return proj.warming_fact return proj.warming_fact
return "no projection available" return "no projection available"
@@ -506,10 +499,7 @@ def get_status(store_path: Optional[Path] = None, clock_now: Optional[datetime]
proj = compute_projection(conn, clock_now) proj = compute_projection(conn, clock_now)
except (sqlite3.Error, ValueError, TypeError): except (sqlite3.Error, ValueError, TypeError):
pass pass
parts.append(_format_projection( parts.append(_format_projection(proj, comp.freshness, drive_facts, config_error))
proj, comp.freshness, drive_facts, config_error,
comp.sample_count, comp.day_count,
))
if query_journal and ( if query_journal and (
comp.store_fault or comp.last_collect_ok is False comp.store_fault or comp.last_collect_ok is False
or comp.freshness in ("missed", "stale") or comp.freshness in ("missed", "stale")
+1 -1
View File
@@ -16,11 +16,11 @@ from typing import Any, Dict, List, Optional
from .status import ( from .status import (
grade_freshness, grade_freshness,
freshness_age_human,
is_deliberately_paused, is_deliberately_paused,
monitoring_continuity, monitoring_continuity,
deliberate_pause_lines, deliberate_pause_lines,
) )
from .timefmt import freshness_age_human
# Status poll interval (AC78-6): lightweight 5s systemctl show poll # Status poll interval (AC78-6): lightweight 5s systemctl show poll
STATUS_POLL_INTERVAL_S = 5 STATUS_POLL_INTERVAL_S = 5
+46 -2
View File
@@ -10,7 +10,23 @@ from pathlib import Path
# Schema version - increment on each migration # Schema version - increment on each migration
SCHEMA_VERSION = 6 SCHEMA_VERSION = 7
# Diagnostic counters added in schema 7 (issue #111), named after the
# NVMe health-log keys they come from. All nullable: legacy rows and drives that
# do not report a counter stay unknown rather than zero.
DIAGNOSTIC_SAMPLE_COLUMNS = (
"host_reads",
"host_writes",
"controller_busy_time",
"num_err_log_entries",
"warning_temp_time",
"critical_comp_time",
"thm_temp1_trans_count",
"thm_temp1_total_time",
"thm_temp2_trans_count",
"thm_temp2_total_time",
)
# Packaged default placement (spec §8.3). The config may override it, but a # Packaged default placement (spec §8.3). The config may override it, but a
@@ -108,7 +124,17 @@ def _create_schema(conn: sqlite3.Connection):
bytes_read INTEGER, bytes_read INTEGER,
critical_warning INTEGER, critical_warning INTEGER,
segment_id INTEGER, segment_id INTEGER,
local_tz TEXT local_tz TEXT,
host_reads INTEGER,
host_writes INTEGER,
controller_busy_time INTEGER,
num_err_log_entries INTEGER,
warning_temp_time INTEGER,
critical_comp_time INTEGER,
thm_temp1_trans_count INTEGER,
thm_temp1_total_time INTEGER,
thm_temp2_trans_count INTEGER,
thm_temp2_total_time INTEGER
) )
""") """)
@@ -302,6 +328,7 @@ def _apply_migrations(conn: sqlite3.Connection, current_version: int):
4: _migrate_3_to_4, 4: _migrate_3_to_4,
5: _migrate_4_to_5, 5: _migrate_4_to_5,
6: _migrate_5_to_6, 6: _migrate_5_to_6,
7: _migrate_6_to_7,
} }
while current_version < SCHEMA_VERSION: while current_version < SCHEMA_VERSION:
target_version = current_version + 1 target_version = current_version + 1
@@ -407,6 +434,23 @@ def _migrate_5_to_6(conn: sqlite3.Connection) -> None:
repair_legacy_local_day_evidence(conn) repair_legacy_local_day_evidence(conn)
def _migrate_6_to_7(conn: sqlite3.Connection) -> None:
"""Add nullable diagnostic counters to samples (issue #111).
Additive only: existing rows are untouched and read NULL (unknown).
Rollback (SQLite 3.35+, with the collector stopped): for each column in
DIAGNOSTIC_SAMPLE_COLUMNS run ``ALTER TABLE samples DROP COLUMN <name>``,
then ``PRAGMA user_version=6``. Dropping loses only the new counters.
"""
cols = {row[1] for row in conn.execute("PRAGMA table_info(samples)").fetchall()}
if not cols:
return
for column in DIAGNOSTIC_SAMPLE_COLUMNS:
if column not in cols:
# nosemgrep: python.lang.security.audit.formatted-sql-query.formatted-sql-query, python.sqlalchemy.security.sqlalchemy-execute-raw-query.sqlalchemy-execute-raw-query -- column is from DIAGNOSTIC_SAMPLE_COLUMNS, a constant
conn.execute(f"ALTER TABLE samples ADD COLUMN {column} INTEGER")
def migrate_to_latest(store_path: Path) -> int: def migrate_to_latest(store_path: Path) -> int:
"""Apply forward-only migrations to bring the store to SCHEMA_VERSION. """Apply forward-only migrations to bring the store to SCHEMA_VERSION.
+3 -2
View File
@@ -161,14 +161,15 @@ def get_theme(name: str) -> Theme:
def get_graph_colors(theme_name: str) -> Dict[str, str]: def get_graph_colors(theme_name: str) -> Dict[str, str]:
"""Return the graph colour roles for a theme preset. """Return the graph colour roles for a theme preset.
Returns a dict with keys: allocated, unallocated, gap, zero, partial, Returns a dict with keys: allocated, unallocated, secondary (the second
selection and muted text. Falls back to Chalktone for unknown names. series), gap, zero, partial, selection and muted text. Falls back to Chalktone for unknown names.
""" """
theme = get_theme(theme_name) theme = get_theme(theme_name)
variables = theme.variables or {} variables = theme.variables or {}
return { return {
"allocated": variables.get("graph-allocated", "#d4a017"), "allocated": variables.get("graph-allocated", "#d4a017"),
"unallocated": variables.get("graph-unallocated", "#8b6914"), "unallocated": variables.get("graph-unallocated", "#8b6914"),
"secondary": variables.get("graph-unallocated", "#8b6914"),
"gap": variables.get("graph-gap", "#554422"), "gap": variables.get("graph-gap", "#554422"),
"zero": variables.get("graph-zero", "#665533"), "zero": variables.get("graph-zero", "#665533"),
"partial": variables.get("graph-partial", "#aa8822"), "partial": variables.get("graph-partial", "#aa8822"),
+63
View File
@@ -0,0 +1,63 @@
"""Time formatting for the dashboard and CLI.
One home for every user-facing time string: local clock times labelled with
their timezone, the secondary UTC form shown beside them, and relative ages.
Stored evidence stays UTC; only presentation is localised.
"""
from datetime import datetime, timezone
from typing import Optional
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
from .tz_util import detect_system_tz, get_tz_offset_str
def local_zone_name() -> str:
"""Return the system timezone name, falling back to UTC when unusable."""
name = detect_system_tz()
try:
ZoneInfo(name)
except (ZoneInfoNotFoundError, ValueError, OSError):
return "UTC"
return name
def tz_label(when: datetime, zone_name: str) -> str:
"""Timezone name with its UTC offset at *when*, e.g. ``Asia/Kolkata +05:30``."""
return "%s %s" % (zone_name, get_tz_offset_str(when, zone_name))
def local_clock(when: datetime, zone_name: str) -> str:
"""Local wall-clock ``HH:MM`` for *when*."""
return when.astimezone(ZoneInfo(zone_name)).strftime("%H:%M")
def local_timestamp(when: datetime, zone_name: str) -> str:
"""Local date, clock time and timezone label for CLI output."""
local = when.astimezone(ZoneInfo(zone_name))
return "%s %s" % (local.strftime("%Y-%m-%d %H:%M"), tz_label(when, zone_name))
def utc_stamp_label(when: datetime) -> str:
"""Secondary UTC form of a single instant."""
return when.astimezone(timezone.utc).strftime("%Y-%m-%d %H:%M UTC")
def utc_range_label(start: datetime, end: datetime) -> str:
"""Secondary UTC form of an interval."""
return "%s → %s UTC" % (
start.astimezone(timezone.utc).strftime("%H:%M"),
end.astimezone(timezone.utc).strftime("%H:%M"),
)
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)
+430 -910
View File
File diff suppressed because it is too large Load Diff
+2 -2
View File
@@ -90,8 +90,8 @@ def _insert_day(conn, day, bw=1024*1024*100, coverage=0.95, samples=24):
def _insert_sample(conn, ts, pu=5): def _insert_sample(conn, ts, pu=5):
conn.execute( conn.execute(
"INSERT INTO samples (ts, device, data_units_written, data_units_read, " "INSERT INTO samples (ts, device, data_units_written, data_units_read, "
"percentage_used, bytes_written, bytes_read, power_on_hours) " "percentage_used, bytes_written, bytes_read, power_on_hours, segment_id) "
"VALUES (?, ?, ?, ?, ?, ?, ?, ?)", "VALUES (?, ?, ?, ?, ?, ?, ?, ?, 1)",
(ts, "/dev/nvme0n1", 1000000, 500000, pu, 512000000000, 256000000000, 8765), (ts, "/dev/nvme0n1", 1000000, 500000, pu, 512000000000, 256000000000, 8765),
) )
conn.commit() conn.commit()
+22
View File
@@ -50,3 +50,25 @@ def test_many_intervals_and_selection_fit_small_plot_without_losing_points():
assert len(columns) == len(points) assert len(columns) == len(points)
assert text.plain.splitlines()[-2][columns[100]] == "▼" assert text.plain.splitlines()[-2][columns[100]] == "▼"
assert all(len(row) == 32 for row in text.plain.splitlines()) assert all(len(row) == 32 for row in text.plain.splitlines())
def test_secondary_series_shares_the_axis_and_colours_its_own_dots():
colors = get_graph_colors("chalktone")
writes = [VolumePoint(t, v, str(t)) for t, v in ((0, 10), (1, 10), (2, 10))]
reads = [VolumePoint(t, v, str(t)) for t, v in ((0, 100), (1, 100), (2, 100))]
text, columns, unit = volume_plot(
writes, 48, 10, -1, colors, secondary=reads)
assert "100.00" in text.plain # the larger read volume sets the ceiling
styles = {span.style for span in text.spans}
assert colors["allocated"] in styles and colors["secondary"] in styles
def test_secondary_gap_does_not_draw_or_connect():
colors = get_graph_colors("chalktone")
points = [VolumePoint(t, 5, str(t)) for t in range(3)]
reads = [VolumePoint(0, 9, "0"), VolumePoint(1, None, "1", "gap"),
VolumePoint(2, 9, "2")]
text, columns, _ = volume_plot(points, 48, 10, -1, colors, secondary=reads)
rows = text.plain.splitlines()[:-2]
top = next(r for r in rows if any(0x2801 <= ord(c) <= 0x28ff for c in r))
assert all(c == " " for c in top[columns[0] + 1:columns[-1]])
+483
View File
@@ -0,0 +1,483 @@
"""Tests for the Activity readout module (issue #108).
The readout is exercised against SQLite fixtures only; no Textual is involved.
The local timezone is pinned through the TZ environment variable.
"""
from datetime import datetime, timezone
from pathlib import Path
import sys
import pytest
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
from fenris.activity_readout import (
critical_warning_badge,
decode_critical_warning,
diagnostics_text,
day_readout_text,
hour_readout_text,
local_day_text,
interval_readout_text,
readout,
selected_local_day,
)
from fenris.activity_selection import (
HISTORY_RANGE_MAX,
ActivitySelection,
HistoryDayIdentity,
)
from fenris.store import init_store
from fenris.timefmt import (
freshness_age_human,
local_timestamp,
utc_range_label,
utc_stamp_label,
)
NOW = datetime(2026, 9, 30, 20, 0, tzinfo=timezone.utc) # 01:30 on 10-01 in Kolkata
@pytest.fixture
def kolkata(monkeypatch):
monkeypatch.setenv("TZ", "Asia/Kolkata")
@pytest.fixture
def conn(tmp_path):
connection = init_store(tmp_path / "observations.db")
yield connection
connection.close()
def _local_day(conn, local_date, utc_start, utc_end, tz="Asia/Kolkata",
offset="+05:30", written=1_000_000_000, read=500_000_000,
intervals=1, seconds=86400, incomplete=0, precision="measured"):
conn.execute(
"INSERT INTO local_days (local_date, tz_name, tz_offset, utc_start, "
"utc_end, bytes_written, bytes_read, coverage, sample_count, complete, "
"activity_seconds, activity_intervals, activity_incomplete, "
"activity_precision) VALUES (?,?,?,?,?,?,?,1.0,24,1,?,?,?,?)",
(local_date, tz, offset, utc_start, utc_end, written, read,
seconds, intervals, incomplete, precision),
)
conn.commit()
def _kolkata_day(conn, local_date, **kwargs):
day = datetime.fromisoformat(local_date)
start = datetime(day.year, day.month, day.day, tzinfo=timezone.utc)
start_utc = start.timestamp() - 19800
end_utc = start_utc + 86400
_local_day(
conn, local_date,
datetime.fromtimestamp(start_utc, timezone.utc).isoformat(),
datetime.fromtimestamp(end_utc, timezone.utc).isoformat(),
**kwargs,
)
def _hour(conn, hour, written=0, read=0, active=0, idle=0, off=0, unknown=0,
samples=1, coverage=1.0, tmin=None, tmax=None, tavg=None):
conn.execute(
"INSERT INTO hour_observations (hour, active_seconds, idle_seconds, "
"powered_off_seconds, unknown_seconds, bytes_written_delta, "
"bytes_read_delta, temperature_min, temperature_max, temperature_avg, "
"sample_count, coverage) VALUES (?,?,?,?,?,?,?,?,?,?,?,?)",
(hour, active, idle, off, unknown, written, read, tmin, tmax, tavg,
samples, coverage),
)
conn.commit()
def _sample(conn, ts, written, read, segment=1, temp=35, critical=0):
conn.execute(
"INSERT INTO samples (ts, device, data_units_written, data_units_read, "
"percentage_used, bytes_written, bytes_read, power_on_hours, "
"segment_id, temperature_c, critical_warning) "
"VALUES (?,?,?,?,?,?,?,?,?,?,?)",
(ts, "/dev/nvme0n1", 1, 1, 1, written, read, 100, segment, temp,
critical),
)
conn.commit()
class TestHistoryDays:
def test_daily_bars_use_local_dates_not_utc_aggregates(self, conn, kolkata):
# A UTC aggregate for 09-30 must never be relabelled as a local day.
conn.execute(
"INSERT INTO day_aggregates (day, bytes_written_delta, "
"sample_count, coverage) VALUES ('2026-09-30', 9000000000, 24, 1.0)"
)
conn.commit()
_kolkata_day(conn, "2026-10-01")
result = readout(conn, ActivitySelection(), NOW)
assert result.days[-1]["day"] == "2026-10-01"
assert result.days[-1]["timezone_label"] == "Asia/Kolkata +05:30"
by_day = {d["day"]: d for d in result.days}
assert by_day["2026-09-30"]["is_gap"]
assert by_day["2026-09-30"]["total_written"] is None
def test_window_is_exact_and_missing_days_stay_explicit_gaps(self, conn, kolkata):
_kolkata_day(conn, "2026-09-29")
result = readout(conn, ActivitySelection(), NOW)
assert len(result.days) == HISTORY_RANGE_MAX
gaps = [d for d in result.days if d["is_gap"]]
assert len(gaps) == HISTORY_RANGE_MAX - 1 # only 09-29 has evidence
known = next(d for d in result.days if d["day"] == "2026-09-29")
assert known["activity_state"] == "complete"
assert known["total_written"] == 1_000_000_000
def test_current_local_day_is_partial_and_incomplete_day_is_flagged(self, conn, kolkata):
_kolkata_day(conn, "2026-10-01", written=100)
_kolkata_day(conn, "2026-09-30", written=200, seconds=3600)
by_day = {d["day"]: d for d in readout(conn, ActivitySelection(), NOW).days}
assert by_day["2026-10-01"]["activity_state"] == "so_far"
assert by_day["2026-10-01"]["is_partial"]
assert by_day["2026-09-30"]["activity_state"] == "incomplete"
assert by_day["2026-09-30"]["is_partial"]
def test_shared_local_day_evidence_is_kept_apart_not_prorated(self, conn, kolkata):
_kolkata_day(conn, "2026-09-29", written=7_000_000_000, read=3_000_000_000)
conn.execute(
"INSERT INTO local_day_unallocated_evidence (start_sample_id, "
"end_sample_id, start_local_date, end_local_date, start_tz_name, "
"end_tz_name, started_at, ended_at, bytes_written, bytes_read, "
"reason) VALUES (1,2,'2026-09-29','2026-09-30','Asia/Kolkata',"
"'Asia/Kolkata','2026-09-29T18:00:00+00:00',"
"'2026-09-29T18:40:00+00:00',1000000000,500000000,'local_midnight')"
)
conn.commit()
day = next(
d for d in readout(conn, ActivitySelection(), NOW).days
if d["day"] == "2026-09-29"
)
assert day["total_written"] == 7_000_000_000
assert day["shared_bytes_written"] == 1_000_000_000
assert day["shared_bytes_read"] == 500_000_000
text = day_readout_text(day)
assert "W 7.000 GB known" in text
assert "shared at midnight W 1.000 GB · R 0.500 GB" in text
def test_unrecorded_selected_date_is_an_unavailable_gap(self, conn, kolkata):
selection = ActivitySelection()
selection.select_history_date("2026-06-01")
result = readout(conn, selection, NOW)
assert result.local_date == "2026-06-01"
assert result.local_day is None
picked = [d for d in result.days if d["day"] == "2026-06-01"]
assert picked and picked[0]["is_gap"]
assert "local-day evidence unavailable" in day_readout_text(picked[0])
class TestLocalDayStrip:
def test_current_day_text_has_zone_state_and_habit(self, conn, kolkata):
_kolkata_day(conn, "2026-10-01", written=2_000_000_000, read=1_000_000_000)
_hour(conn, "2026-09-30T19:00:00+00:00", active=3600, tmin=30, tmax=44)
local_date, local = selected_local_day(conn, ActivitySelection(), NOW)
text = local_day_text(local_date, local)
assert "2026-10-01" in text and "Asia/Kolkata +05:30" in text
assert "totals so far" in text
assert "W 2.000 GB known · R 1.000 GB known" in text
assert "active 1h 00m" in text and "30–44°C" in text
def test_missing_day_text_is_unavailable(self, conn, kolkata):
local_date, local = selected_local_day(conn, ActivitySelection(), NOW)
assert local is None
assert "local-day evidence unavailable" in local_day_text(local_date, local)
class TestHours:
def _selected_day(self, conn):
_kolkata_day(conn, "2026-09-29")
selection = ActivitySelection()
selection.select_history_date(
"2026-09-29",
HistoryDayIdentity(
"2026-09-29", "Asia/Kolkata",
"2026-09-28T18:30:00+00:00", "2026-09-29T18:30:00+00:00",
),
)
selection.set_view("day")
return selection
def test_hours_only_in_day_view_with_local_and_secondary_utc_labels(self, conn, kolkata):
selection = self._selected_day(conn)
_hour(conn, "2026-09-28T19:00:00+00:00", written=4_000_000_000,
active=600, idle=2400, off=0, unknown=600, tmin=31, tmax=40,
coverage=0.8)
result = readout(conn, selection, NOW)
first = result.hours[0]
assert first["local_label"] == "00:30"
assert first["timezone_label"] == "Asia/Kolkata +05:30"
assert first["utc_label"] == "2026-09-28 19:00 UTC"
assert first["bytes_written"] == 4_000_000_000
text = hour_readout_text(first, "2026-09-29")
assert "00:30 Asia/Kolkata +05:30" in text
assert text.splitlines()[-1] == "2026-09-28 19:00 UTC"
assert "active 10m · idle 40m · off 0m · unknown 10m" in text
assert "31–40°C" in text
def test_live_view_has_no_hours(self, conn, kolkata):
_kolkata_day(conn, "2026-09-29")
assert readout(conn, ActivitySelection(), NOW).hours == ()
def test_missing_hour_is_gap_and_future_hour_is_not(self, conn, kolkata):
_kolkata_day(conn, "2026-10-01")
selection = ActivitySelection()
selection.select_history_date("2026-10-01")
selection.set_view("day")
result = readout(conn, selection, NOW)
past = next(h for h in result.hours if h["local_label"] == "00:30")
future = next(h for h in result.hours if h["local_label"] == "06:30")
assert past["is_gap"] and not past["is_future"]
assert future["is_future"] and not future["is_gap"]
assert "W unavailable · R unavailable" in hour_readout_text(past, "2026-10-01")
class TestLiveIntervals:
def test_interval_labels_are_local_with_secondary_utc_line(self, conn, kolkata):
_sample(conn, "2026-09-30T19:30:00+00:00", 1000, 500)
_sample(conn, "2026-09-30T19:33:00+00:00", 3000, 900, temp=41)
result = readout(conn, ActivitySelection(), NOW)
point = result.live[0]
assert point["start_label"] == "01:00"
assert point["end_label"] == "01:03"
assert point["timezone_label"] == "Asia/Kolkata +05:30"
assert point["bytes_written"] == 2000
assert point["temperature_c"] == 41
text = interval_readout_text(point)
assert "01:00 → 01:03 Asia/Kolkata +05:30" in text
assert text.splitlines()[-1] == "19:30 → 19:33 UTC"
assert "41°C" in text
def test_drive_warning_is_part_of_the_readout(self, conn, kolkata):
_sample(conn, "2026-09-30T19:30:00+00:00", 1000, 500)
_sample(conn, "2026-09-30T19:33:00+00:00", 3000, 900, critical=4)
point = readout(conn, ActivitySelection(), NOW).live[0]
assert "critical warning" in interval_readout_text(point)
def test_segment_boundary_is_unavailable(self, conn, kolkata):
_sample(conn, "2026-09-30T19:30:00+00:00", 1000, 500, segment=1)
_sample(conn, "2026-09-30T19:33:00+00:00", 5, 5, segment=2)
point = readout(conn, ActivitySelection(), NOW).live[0]
assert "segment boundary" in interval_readout_text(point)
assert "W unavailable · R unavailable" in interval_readout_text(point)
class TestTimezone:
def test_utc_zone_still_carries_a_label(self, conn, monkeypatch):
monkeypatch.setenv("TZ", "UTC")
_sample(conn, "2026-09-30T19:30:00+00:00", 1000, 500)
_sample(conn, "2026-09-30T19:33:00+00:00", 3000, 900)
result = readout(conn, ActivitySelection(), NOW)
assert result.timezone_name == "UTC"
assert result.timezone_label == "UTC +00:00"
assert result.live[0]["start_label"] == "19:30"
assert result.live[0]["timezone_label"] == "UTC +00:00"
def test_unknown_zone_falls_back_to_utc(self, conn, monkeypatch):
monkeypatch.setenv("TZ", "Not/AZone")
assert readout(conn, ActivitySelection(), NOW).timezone_name == "UTC"
def test_health_comes_from_the_latest_sample(self, conn, kolkata):
assert readout(conn, ActivitySelection(), NOW).health["model"] == "unknown"
_sample(conn, "2026-09-30T19:30:00+00:00", 1000, 500, temp=38)
assert readout(conn, ActivitySelection(), NOW).health["temp"] == 38
class TestTimeFormatting:
def test_freshness_age_human_lives_here(self):
assert freshness_age_human(30) == "30s ago"
assert freshness_age_human(None) == "unknown age"
def test_local_timestamp_names_the_zone(self):
when = datetime(2026, 9, 30, 20, 0, tzinfo=timezone.utc)
assert local_timestamp(when, "Asia/Kolkata") == (
"2026-10-01 01:30 Asia/Kolkata +05:30"
)
def test_utc_labels(self):
a = datetime(2026, 9, 30, 19, 30, tzinfo=timezone.utc)
b = datetime(2026, 9, 30, 19, 33, tzinfo=timezone.utc)
assert utc_range_label(a, b) == "19:30 → 19:33 UTC"
assert utc_stamp_label(a) == "2026-09-30 19:30 UTC"
def _counters(conn, ts, poh, cycles, unsafe, media, spare, critical=0):
conn.execute(
"INSERT INTO samples (ts, device, data_units_written, data_units_read, "
"percentage_used, bytes_written, bytes_read, power_on_hours, "
"power_cycles, unsafe_shutdowns, media_errors, available_spare, "
"segment_id, temperature_c, critical_warning) "
"VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)",
(ts, "/dev/nvme0n1", 1, 1, 1, 1, 1, poh, cycles, unsafe, media, spare,
1, 35, critical),
)
conn.commit()
class TestCriticalWarning:
def test_zero_decodes_to_nothing_and_each_bit_to_its_meaning(self):
assert decode_critical_warning(0) == ()
assert decode_critical_warning(None) == ()
assert decode_critical_warning(0x01) == ("available spare below threshold",)
assert decode_critical_warning(0x05) == (
"available spare below threshold", "reliability degraded",
)
assert "temperature outside threshold" in decode_critical_warning(0x02)
assert "media in read-only mode" in decode_critical_warning(0x08)
assert "volatile memory backup failed" in decode_critical_warning(0x10)
assert "persistent memory region read-only" in decode_critical_warning(0x20)
assert decode_critical_warning(0x40) == ("unrecognised bits 0x40",)
def test_health_carries_latest_critical_warning_and_badge_follows_it(self, conn, kolkata):
_counters(conn, "2026-09-30T19:00:00+00:00", 10, 1, 0, 0, 100, critical=0)
assert critical_warning_badge(
readout(conn, ActivitySelection(), NOW).health) == ""
_counters(conn, "2026-09-30T19:30:00+00:00", 11, 1, 0, 0, 100, critical=0x04)
health = readout(conn, ActivitySelection(), NOW).health
assert health["critical_warning"] == 0x04
badge = critical_warning_badge(health)
assert "0x04" in badge and "reliability degraded" in badge
def test_no_samples_means_no_badge(self, conn, kolkata):
assert critical_warning_badge(
readout(conn, ActivitySelection(), NOW).health) == ""
class TestDiagnosticsReadout:
def test_diagnostics_are_absent_until_selected(self, conn, kolkata):
assert readout(conn, ActivitySelection(), NOW).diagnostics is None
selection = ActivitySelection()
selection.toggle_diagnostics()
assert selection.diagnostics
assert readout(conn, selection, NOW).diagnostics is not None
selection.toggle_diagnostics()
assert not selection.diagnostics
def test_live_counters_have_deltas_over_the_live_window(self, conn, kolkata):
_counters(conn, "2026-09-30T16:00:00+00:00", 90, 5, 1, 0, 100) # before window
_counters(conn, "2026-09-30T18:00:00+00:00", 100, 6, 1, 0, 100)
_counters(conn, "2026-09-30T19:50:00+00:00", 103, 8, 3, 2, 98, critical=0x02)
selection = ActivitySelection()
selection.toggle_diagnostics()
diag = readout(conn, selection, NOW).diagnostics
assert diag["window_label"] == "last 3h"
assert diag["critical_warning"] == 0x02
assert diag["critical_warning_flags"] == ("temperature outside threshold",)
counters = diag["counters"]
assert counters["power_on_hours"] == {"value": 103, "delta": 13}
assert counters["power_cycles"] == {"value": 8, "delta": 3}
assert counters["unsafe_shutdowns"] == {"value": 3, "delta": 2}
assert counters["media_errors"] == {"value": 2, "delta": 2}
assert counters["available_spare"] == {"value": 98, "delta": -2}
def test_no_delta_across_a_controller_segment_change(self, conn, kolkata):
_counters(conn, "2026-09-30T18:00:00+00:00", 100, 6, 1, 0, 100)
_counters(conn, "2026-09-30T19:50:00+00:00", 5, 1, 0, 0, 100)
conn.execute("UPDATE samples SET segment_id = 2 WHERE ts > '2026-09-30T19'")
conn.commit()
selection = ActivitySelection()
selection.toggle_diagnostics()
counters = readout(conn, selection, NOW).diagnostics["counters"]
assert counters["power_on_hours"] == {"value": 5, "delta": None}
def test_single_sample_has_no_delta(self, conn, kolkata):
_counters(conn, "2026-09-30T19:50:00+00:00", 103, 8, 3, 2, 98)
selection = ActivitySelection()
selection.toggle_diagnostics()
counters = readout(conn, selection, NOW).diagnostics["counters"]
assert counters["power_on_hours"] == {"value": 103, "delta": None}
def test_day_selection_measures_deltas_across_that_local_day(self, conn, kolkata):
selection = TestHours()._selected_day(conn)
_counters(conn, "2026-09-28T18:00:00+00:00", 50, 2, 0, 0, 100)
_counters(conn, "2026-09-29T06:00:00+00:00", 56, 2, 1, 0, 100)
_counters(conn, "2026-09-29T20:00:00+00:00", 99, 9, 9, 9, 50) # after the day
selection.toggle_diagnostics()
diag = readout(conn, selection, NOW).diagnostics
assert diag["window_label"] == "day 2026-09-29"
assert diag["counters"]["power_on_hours"] == {"value": 56, "delta": 6}
assert diag["counters"]["unsafe_shutdowns"]["delta"] == 1
def test_hours_carry_temperature_average(self, conn, kolkata):
selection = TestHours()._selected_day(conn)
_hour(conn, "2026-09-29T00:00:00+00:00", active=1800, idle=1800,
tmin=31, tavg=36.5, tmax=44)
hour = next(h for h in readout(conn, selection, NOW).hours
if h["hour"] == "2026-09-29T00:00:00+00:00")
assert (hour["temperature_min"], hour["temperature_avg"],
hour["temperature_max"]) == (31, 36.5, 44)
def test_history_days_carry_split_and_temperature_range(self, conn, kolkata):
_kolkata_day(conn, "2026-09-29")
_hour(conn, "2026-09-29T00:00:00+00:00", active=1800, idle=1200, off=600,
tmin=30, tavg=34.0, tmax=40)
_hour(conn, "2026-09-29T01:00:00+00:00", active=0, idle=3000, unknown=600,
tmin=28, tavg=30.0, tmax=33)
selection = ActivitySelection()
selection.set_view("history")
selection.toggle_diagnostics()
days = {d["day"]: d for d in readout(conn, selection, NOW).diagnostics["days"]}
day = days["2026-09-29"]
assert (day["active_seconds"], day["idle_seconds"],
day["powered_off_seconds"], day["unknown_seconds"]) == (1800, 4200, 600, 600)
assert (day["temperature_min"], day["temperature_max"]) == (28, 40)
assert day["temperature_avg"] == pytest.approx(32.0)
assert "2026-09-30" not in days # no evidence, no invented split
class TestDiagnosticsText:
def _diag(self, conn, selection):
selection.toggle_diagnostics()
return readout(conn, selection, NOW)
def test_panel_shows_decoded_warning_counters_and_deltas(self, conn, kolkata):
_counters(conn, "2026-09-30T18:00:00+00:00", 100, 6, 1, 0, 100)
_counters(conn, "2026-09-30T19:50:00+00:00", 103, 8, 3, 2, 98, critical=0x05)
text = diagnostics_text(self._diag(conn, ActivitySelection()))
assert "last 3h" in text
assert "0x05" in text and "reliability degraded" in text
assert "Power-on hours" in text and "103" in text and "+3" in text
assert "Unsafe shutdowns" in text and "+2" in text
assert "Available spare" in text and "98%" in text and "-2" in text
def test_no_warning_says_none_and_missing_delta_says_so(self, conn, kolkata):
_counters(conn, "2026-09-30T19:50:00+00:00", 103, 8, 3, 2, 98)
text = diagnostics_text(self._diag(conn, ActivitySelection()))
assert "Critical warning none" in text
assert "Δ n/a" in text
def test_day_view_lists_each_hour_with_temperature_split_and_coverage(self, conn, kolkata):
selection = TestHours()._selected_day(conn)
_hour(conn, "2026-09-29T00:00:00+00:00", active=1800, idle=1200, off=300,
unknown=300, tmin=31, tavg=36.4, tmax=44, coverage=0.75)
text = diagnostics_text(self._diag(conn, selection))
row = next(line for line in text.splitlines() if line.startswith("05:30"))
assert "31/36/44°C" in row
assert "A30m I20m Off5m ?5m" in row
assert "75%" in row
assert any("gap" in line for line in text.splitlines()
if line.startswith("06:30"))
def test_history_view_lists_each_day_with_split_and_temperature(self, conn, kolkata):
_kolkata_day(conn, "2026-09-29")
_hour(conn, "2026-09-29T00:00:00+00:00", active=7200, idle=3600, off=0,
tmin=30, tavg=34.0, tmax=40)
selection = ActivitySelection()
selection.set_view("history")
text = diagnostics_text(self._diag(conn, selection))
row = next(line for line in text.splitlines() if line.startswith("09-29"))
assert "A2h 00m" in row and "I1h 00m" in row and "30/34/40°C" in row
def test_live_view_summarises_interval_temperature(self, conn, kolkata):
for index, (minute, temp) in enumerate(((0, 30), (3, 36), (6, 42))):
conn.execute(
"INSERT INTO samples (ts, device, bytes_written, bytes_read, "
"segment_id, temperature_c) VALUES (?,?,?,?,?,?)",
("2026-09-30T19:%02d:00+00:00" % (50 + minute), "/dev/n", index, index, 1, temp),
)
conn.commit()
text = diagnostics_text(self._diag(conn, ActivitySelection()))
assert "36/39/42°C" in text and "2 intervals" in text # interval ends
+1 -1
View File
@@ -152,7 +152,7 @@ class TestSchemaMigration:
# Migrate # Migrate
steps = migrate_to_latest(db) steps = migrate_to_latest(db)
assert steps == 5 # v1→v2→v3→v4→v5→v6 assert steps == 6 # v1→v2→v3→v4→v5→v6→v7
# Verify data preserved # Verify data preserved
conn = sqlite3.connect(str(db)) conn = sqlite3.connect(str(db))
+24 -22
View File
@@ -1,16 +1,18 @@
"""Complete observation day gate tests (issue #94). """Complete observation day condition tests (issues #94, #106).
Verifies that the endurance projection is withheld until at least one Originally a gate (issue #94); ADR 0012 turned it into a contributing
complete local calendar day has been observed within a monitoring period. fact. The projection is no longer withheld until one complete local
calendar day has been observed within a monitoring period, but it says
so while that condition is unmet.
Seams: Seams:
- compute_projection() → gate check via local_days table - compute_projection() → gate check via local_days table
- ProjectionResult.contributing_facts → "waiting for a full local observation day" - ProjectionResult.contributing_facts → "waiting for a full local observation day"
Acceptance criteria: Acceptance criteria:
- Gate-1: No complete local day → UNSUPPORTED with waiting fact - Gate-1: No complete local day → projection renders with waiting fact
- Gate-2: One complete local day → Limited confidence (if other conditions met) - Gate-2: One complete local day → no waiting fact
- Gate-3: Partial days don't satisfy the gate - Gate-3: Partial days don't satisfy the condition
- Gate-4: CLI and TUI share the same gate via compute_projection() - Gate-4: CLI and TUI share the same gate via compute_projection()
""" """
from datetime import datetime, timedelta, timezone from datetime import datetime, timedelta, timezone
@@ -76,8 +78,8 @@ def _insert_day(conn, day, bw=1024*1024*100, coverage=0.95, samples=24):
def _insert_sample(conn, ts, pu=5): def _insert_sample(conn, ts, pu=5):
conn.execute( conn.execute(
"INSERT INTO samples (ts, device, data_units_written, data_units_read, " "INSERT INTO samples (ts, device, data_units_written, data_units_read, "
"percentage_used, bytes_written, bytes_read, power_on_hours) " "percentage_used, bytes_written, bytes_read, power_on_hours, segment_id) "
"VALUES (?, ?, ?, ?, ?, ?, ?, ?)", "VALUES (?, ?, ?, ?, ?, ?, ?, ?, 1)",
(ts, "/dev/nvme0n1", 1000000, 500000, pu, 512000000000, 256000000000, 8765), (ts, "/dev/nvme0n1", 1000000, 500000, pu, 512000000000, 256000000000, 8765),
) )
conn.commit() conn.commit()
@@ -109,14 +111,14 @@ def _open_period(conn, start="2026-09-01T00:00:00+00:00"):
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# Gate-1: No complete local day → UNSUPPORTED with waiting fact # Gate-1: No complete local day → projection renders with waiting fact
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
class TestGateNoCompleteDay: class TestGateNoCompleteDay:
"""Projection is unavailable before any complete local observation day.""" """Projection still renders before any complete local observation day."""
def test_no_local_days_unsupported(self, store): def test_no_local_days_states_waiting_fact(self, store):
"""With no local_days entries, projection is UNSUPPORTED.""" """With no local_days entries, the projection renders and says it is waiting."""
_insert_baseline(store) _insert_baseline(store)
_insert_segment(store) _insert_segment(store)
_open_period(store) _open_period(store)
@@ -126,11 +128,11 @@ class TestGateNoCompleteDay:
_insert_day(store, d, bw=1024*1024*100) _insert_day(store, d, bw=1024*1024*100)
_insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5) _insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5)
result = compute_projection(store, _clock()) result = compute_projection(store, _clock())
assert result.confidence_state == ConfidenceState.UNSUPPORTED assert result.confidence_state != ConfidenceState.UNSUPPORTED
assert any("full local observation day" in f for f in result.contributing_facts) assert any("full local observation day" in f for f in result.contributing_facts)
assert result.headline_remaining_seconds is None assert result.headline_remaining_seconds is not None
def test_only_partial_local_days_unsupported(self, store): def test_only_partial_local_days_state_waiting_fact(self, store):
"""Partial (incomplete) local days don't satisfy the gate.""" """Partial (incomplete) local days don't satisfy the gate."""
_insert_baseline(store) _insert_baseline(store)
_insert_segment(store) _insert_segment(store)
@@ -144,7 +146,7 @@ class TestGateNoCompleteDay:
d = (datetime(2026, 9, 25) + timedelta(days=i)).strftime("%Y-%m-%d") d = (datetime(2026, 9, 25) + timedelta(days=i)).strftime("%Y-%m-%d")
_insert_local_day(store, d, complete=False, coverage=0.3) _insert_local_day(store, d, complete=False, coverage=0.3)
result = compute_projection(store, _clock()) result = compute_projection(store, _clock())
assert result.confidence_state == ConfidenceState.UNSUPPORTED assert result.confidence_state != ConfidenceState.UNSUPPORTED
assert any("full local observation day" in f for f in result.contributing_facts) assert any("full local observation day" in f for f in result.contributing_facts)
def test_gate_before_warming_check(self, store): def test_gate_before_warming_check(self, store):
@@ -164,7 +166,7 @@ class TestGateNoCompleteDay:
# The key assertion: gate message should NOT appear when gate IS met # The key assertion: gate message should NOT appear when gate IS met
assert not any("full local observation day" in f for f in result.contributing_facts) assert not any("full local observation day" in f for f in result.contributing_facts)
def test_complete_legacy_day_without_trusted_activity_does_not_open_gate( def test_complete_legacy_day_without_trusted_activity_keeps_waiting_fact(
self, store, self, store,
): ):
"""A complete flag cannot make unavailable local activity qualify.""" """A complete flag cannot make unavailable local activity qualify."""
@@ -181,11 +183,11 @@ class TestGateNoCompleteDay:
"WHERE local_date = '2026-09-29'" "WHERE local_date = '2026-09-29'"
) )
result = compute_projection(store, _clock()) result = compute_projection(store, _clock())
assert result.confidence_state == ConfidenceState.UNSUPPORTED assert result.confidence_state != ConfidenceState.UNSUPPORTED
assert any("full local observation day" in fact assert any("full local observation day" in fact
for fact in result.contributing_facts) for fact in result.contributing_facts)
def test_day_split_by_deliberate_pause_does_not_open_gate(self, store): def test_day_split_by_deliberate_pause_keeps_waiting_fact(self, store):
"""A complete-looking summary cannot span separate monitoring periods.""" """A complete-looking summary cannot span separate monitoring periods."""
_insert_baseline(store) _insert_baseline(store)
_insert_segment(store) _insert_segment(store)
@@ -214,7 +216,7 @@ class TestGateNoCompleteDay:
result = compute_projection(store, _clock()) result = compute_projection(store, _clock())
assert result.confidence_state == ConfidenceState.UNSUPPORTED assert result.confidence_state != ConfidenceState.UNSUPPORTED
assert any("full local observation day" in fact assert any("full local observation day" in fact
for fact in result.contributing_facts) for fact in result.contributing_facts)
@@ -288,7 +290,7 @@ class TestGateOneCompleteDay:
class TestGatePartialFirstDay: class TestGatePartialFirstDay:
"""Starting monitoring at noon means the partial first day doesn't count.""" """Starting monitoring at noon means the partial first day doesn't count."""
def test_partial_first_day_not_enough(self, store): def test_partial_first_day_keeps_waiting_fact(self, store):
"""A single incomplete local day (started at noon) doesn't open the gate.""" """A single incomplete local day (started at noon) doesn't open the gate."""
_insert_baseline(store) _insert_baseline(store)
_insert_segment(store, opened_at="2026-09-29T12:00:00+00:00") _insert_segment(store, opened_at="2026-09-29T12:00:00+00:00")
@@ -301,7 +303,7 @@ class TestGatePartialFirstDay:
_insert_local_day(store, "2026-09-29", complete=False, coverage=0.5) _insert_local_day(store, "2026-09-29", complete=False, coverage=0.5)
_insert_local_day(store, "2026-09-30", complete=False, coverage=0.5) _insert_local_day(store, "2026-09-30", complete=False, coverage=0.5)
result = compute_projection(store, _clock()) result = compute_projection(store, _clock())
assert result.confidence_state == ConfidenceState.UNSUPPORTED assert result.confidence_state != ConfidenceState.UNSUPPORTED
assert any("full local observation day" in f for f in result.contributing_facts) assert any("full local observation day" in f for f in result.contributing_facts)
+35 -6
View File
@@ -20,6 +20,7 @@ NOW = datetime(2026, 9, 19, 12, tzinfo=timezone.utc)
@pytest.fixture @pytest.fixture
def dashboard(tmp_path, monkeypatch): def dashboard(tmp_path, monkeypatch):
monkeypatch.setenv("XDG_CONFIG_HOME", str(tmp_path / "prefs")) monkeypatch.setenv("XDG_CONFIG_HOME", str(tmp_path / "prefs"))
monkeypatch.setenv("TZ", "UTC")
class Clock(datetime): class Clock(datetime):
@classmethod @classmethod
@@ -124,6 +125,19 @@ async def test_mouse_inspection_matches_time_axis(dashboard):
assert "09:00 → 09:03 UTC" in text assert "09:00 → 09:03 UTC" in text
@pytest.mark.asyncio
async def test_live_readout_shows_local_time_with_secondary_utc_line(dashboard, monkeypatch):
monkeypatch.setenv("TZ", "Asia/Kolkata")
app = dashboard
async with app.run_test(size=(100, 36)) as pilot:
await pilot.pause()
await pilot.press("left")
text = str(app.query_one("#live-readout").render())
assert "17:24 → 17:27 Asia/Kolkata +05:30" in text
assert text.splitlines()[-1] == "11:54 → 11:57 UTC"
assert "local time" in str(app.query_one("#live-legend").render())
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_theme_control_persists_choice_and_quit_never_pauses(dashboard, monkeypatch): async def test_theme_control_persists_choice_and_quit_never_pauses(dashboard, monkeypatch):
calls = [] calls = []
@@ -144,14 +158,27 @@ async def test_unallocated_volume_survives_day_with_missing_coverage(dashboard,
await pilot.click("#view-history") await pilot.click("#view-history")
graph = dashboard.query_one("#usage-history") graph = dashboard.query_one("#usage-history")
graph.set_data([{ graph.set_data([{
"day": "2026-09-18", "total_bytes": 2_000_000_000, "day": "2026-09-18", "local_label": "2026-09-18",
"unallocated_bytes": 2_000_000_000, "is_gap": True, "timezone_label": "UTC +00:00", "identity": None,
"is_partial": True, "total_bytes": None, "total_written": None, "total_read": None,
"shared_bytes_written": 2_000_000_000, "shared_bytes_read": 0,
"unallocated_bytes_written": 2_000_000_000,
"unallocated_bytes_read": 0, "activity_state": "incomplete",
"is_gap": True, "is_partial": True,
}]) }])
readout = str(dashboard.query_one("#bar-readout").render()) readout = str(dashboard.query_one("#bar-readout").render())
assert "W 2.000 GB unallocated" in readout assert "unallocated W 2.000 GB" in readout
assert "R unavailable" in readout and "gap" in readout assert "W unavailable · R unavailable" in readout
if size[0] >= 80: if size[0] >= 80:
graph.set_data([{
"day": "2026-09-18", "local_label": "2026-09-18",
"timezone_label": "UTC +00:00", "identity": None,
"total_bytes": 2_000_000_000, "total_written": 2_000_000_000,
"total_read": None, "shared_bytes_written": 2_000_000_000,
"shared_bytes_read": 0, "unallocated_bytes_written": 0,
"unallocated_bytes_read": 0, "activity_state": "incomplete",
"is_gap": False, "is_partial": True,
}])
plot = str(dashboard.query_one("#bar-render").render()) plot = str(dashboard.query_one("#bar-render").render())
assert any(0x2801 <= ord(c) <= 0x28ff for c in plot) assert any(0x2801 <= ord(c) <= 0x28ff for c in plot)
graph.measure = "read" graph.measure = "read"
@@ -167,7 +194,9 @@ async def test_small_terminal_hour_readout_does_not_invent_zero(dashboard, state
await pilot.click("#view-day") await pilot.click("#view-day")
graph = dashboard.query_one("#usage-history") graph = dashboard.query_one("#usage-history")
graph.set_hour_data([{ graph.set_hour_data([{
"hour": "2026-09-18T12:00:00+00:00", "local_label": "12", "hour": "2026-09-18T12:00:00+00:00", "local_label": "12:00",
"timezone_label": "UTC +00:00",
"utc_label": "2026-09-18 12:00 UTC",
"is_gap": state == "gap", "is_future": state == "future", "is_gap": state == "gap", "is_future": state == "future",
}]) }])
graph._hourly_selected = 0 graph._hourly_selected = 0
+251
View File
@@ -0,0 +1,251 @@
"""Additional smartctl diagnostics (issue #111).
Field list fixed from real `smartctl -a -j` output (smartctl 7.5, NVMe 1.4):
nvme_smart_health_information_log carries warning_temp_time, critical_comp_time,
num_err_log_entries, host_reads, host_writes and controller_busy_time. The
thermal-management transition fields (thm_temp1/2_trans_count, thm_temp1/2_total_time)
appear only on drives that implement them, so they stay nullable.
"""
import sqlite3
import sys
from datetime import datetime, timezone
from pathlib import Path
import pytest
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
from fenris.activity_readout import diagnostics_text, readout
from fenris.activity_selection import ActivitySelection
from fenris.collector import acquire_from_smartctl, run_collection
from fenris.store import SCHEMA_VERSION, init_store, migrate_to_latest
NEW_COLUMNS = (
"host_reads", "host_writes", "controller_busy_time", "num_err_log_entries",
"warning_temp_time", "critical_comp_time",
"thm_temp1_trans_count", "thm_temp1_total_time",
"thm_temp2_trans_count", "thm_temp2_total_time",
)
def _smartctl(**extra_log):
"""Shape of real smartctl -a -j output for an NVMe drive."""
log = {
"critical_warning": 0, "temperature": 37, "available_spare": 100,
"available_spare_threshold": 5, "percentage_used": 52,
"data_units_read": 207986339, "data_units_written": 202978899,
"host_reads": 3543447796, "host_writes": 2458441042,
"controller_busy_time": 61626, "power_cycles": 5324,
"power_on_hours": 11906, "unsafe_shutdowns": 159, "media_errors": 0,
"num_err_log_entries": 0, "warning_temp_time": 0, "critical_comp_time": 0,
"temperature_sensors": [37],
}
log.update(extra_log)
return {
"nvme_smart_health_information_log": log,
"user_capacity": {"bytes": 512110190592},
"model_name": "Drive", "serial_number": "SN1", "firmware_version": "FW1",
}
class TestAcquisition:
def test_real_fields_are_captured(self):
got = acquire_from_smartctl(_smartctl())
assert got["host_reads"] == 3543447796
assert got["host_writes"] == 2458441042
assert got["controller_busy_time"] == 61626
assert got["num_err_log_entries"] == 0
assert got["warning_temp_time"] == 0
assert got["critical_comp_time"] == 0
def test_absent_thermal_management_fields_are_none_not_zero(self):
got = acquire_from_smartctl(_smartctl())
for key in ("thm_temp1_trans_count", "thm_temp1_total_time",
"thm_temp2_trans_count", "thm_temp2_total_time"):
assert got[key] is None
def test_thermal_management_fields_captured_when_reported(self):
got = acquire_from_smartctl(_smartctl(
thm_temp1_trans_count=3, thm_temp1_total_time=120,
thm_temp2_trans_count=1, thm_temp2_total_time=45))
assert (got["thm_temp1_trans_count"], got["thm_temp1_total_time"]) == (3, 120)
assert (got["thm_temp2_trans_count"], got["thm_temp2_total_time"]) == (1, 45)
def test_missing_optional_counters_do_not_fail_the_run(self):
data = _smartctl()
for key in ("host_reads", "host_writes", "controller_busy_time",
"num_err_log_entries", "warning_temp_time", "critical_comp_time"):
del data["nvme_smart_health_information_log"][key]
got = acquire_from_smartctl(data)
assert got["host_reads"] is None and got["warning_temp_time"] is None
@pytest.fixture
def sysfs(tmp_path):
ctrl = tmp_path / "nvme0"
ctrl.mkdir()
for name, value in (("subsysnqn", "nqn.test"), ("model", "Drive"),
("serial", "SN1"), ("firmware_rev", "FW1")):
(ctrl / name).write_text(value + "\n")
return ctrl
class _Clock:
def utcnow(self):
return datetime(2026, 9, 1, 12, 0, tzinfo=timezone.utc)
def test_collection_persists_new_fields(tmp_path, sysfs):
config = {"device": "/dev/nvme0", "store_path": str(tmp_path / "o.db")}
result = run_collection(smartctl_data=_smartctl(thm_temp1_trans_count=3),
sysfs_path=sysfs, config=config, clock=_Clock())
assert result["ok"], result
conn = sqlite3.connect(config["store_path"])
conn.row_factory = sqlite3.Row
row = conn.execute("SELECT * FROM samples").fetchone()
assert row["host_reads"] == 3543447796
assert row["controller_busy_time"] == 61626
assert row["thm_temp1_trans_count"] == 3
assert row["thm_temp2_trans_count"] is None
def _legacy_store(path):
"""A populated schema-6 store, as shipped before this change."""
conn = init_store(path)
for column in reversed(NEW_COLUMNS):
conn.execute(f"ALTER TABLE samples DROP COLUMN {column}")
conn.execute("PRAGMA user_version=6")
for i in range(3):
conn.execute(
"INSERT INTO samples (ts, device, bytes_written, bytes_read, "
"power_on_hours, critical_warning, segment_id) VALUES (?,?,?,?,?,?,1)",
("2026-09-30T1%d:00:00+00:00" % i, "/dev/nvme0", 10 + i, 5, 100 + i, 0))
conn.commit()
conn.close()
class TestMigration:
def test_schema_version_is_seven(self):
assert SCHEMA_VERSION == 7
def test_populated_legacy_store_keeps_rows_and_gains_null_columns(self, tmp_path):
db = tmp_path / "o.db"
_legacy_store(db)
before = sqlite3.connect(db).execute(
"SELECT id, ts, device, bytes_written, power_on_hours FROM samples "
"ORDER BY id").fetchall()
assert migrate_to_latest(db) == 1
conn = sqlite3.connect(db)
assert conn.execute("PRAGMA user_version").fetchone()[0] == 7
assert conn.execute(
"SELECT id, ts, device, bytes_written, power_on_hours FROM samples "
"ORDER BY id").fetchall() == before
cols = {r[1] for r in conn.execute("PRAGMA table_info(samples)")}
assert set(NEW_COLUMNS) <= cols
nulls = conn.execute(
"SELECT %s FROM samples" % " OR ".join(c + " IS NOT NULL" for c in NEW_COLUMNS)
).fetchall()
assert nulls == [(0,)] * 3 # legacy rows stay unknown, never zero
def test_migration_is_idempotent_and_init_store_applies_it(self, tmp_path):
db = tmp_path / "o.db"
_legacy_store(db)
conn = init_store(db)
assert conn.execute("PRAGMA user_version").fetchone()[0] == 7
conn.close()
assert migrate_to_latest(db) == 0
def test_pending_payload_without_new_fields_still_publishes(self, tmp_path):
"""Observations staged by the previous version lack the new keys."""
db = tmp_path / "o.db"
conn = init_store(db)
from fenris.collector import write_sample
sample = {k: v for k, v in acquire_from_smartctl(_smartctl()).items()
if k not in NEW_COLUMNS}
sample.update(ts="2026-09-01T12:00:00+00:00", device="/dev/nvme0")
write_sample(sample, {"subnqn": "nqn.test"}, conn,
datetime(2026, 9, 1, 12, tzinfo=timezone.utc))
assert conn.execute("SELECT host_reads FROM samples").fetchone() == (None,)
def test_rollback_notes_procedure_restores_prior_schema(self, tmp_path):
"""Documented rollback: drop the columns, set user_version back to 6."""
db = tmp_path / "o.db"
_legacy_store(db)
migrate_to_latest(db)
conn = sqlite3.connect(db)
conn.execute(
"UPDATE samples SET host_reads = 1 WHERE id = 1")
for column in NEW_COLUMNS:
conn.execute(f"ALTER TABLE samples DROP COLUMN {column}")
conn.execute("PRAGMA user_version=6")
conn.commit()
assert conn.execute("SELECT COUNT(*) FROM samples").fetchone() == (3,)
cols = {r[1] for r in conn.execute("PRAGMA table_info(samples)")}
assert not set(NEW_COLUMNS) & cols
NOW = datetime(2026, 9, 30, 20, 0, tzinfo=timezone.utc)
def _diag_sample(conn, ts, **values):
cols = ["ts", "device", "bytes_written", "bytes_read", "segment_id"] + list(values)
conn.execute(
"INSERT INTO samples (%s) VALUES (%s)" % (
",".join(cols), ",".join("?" * len(cols))),
[ts, "/dev/nvme0n1", 1, 1, 1] + list(values.values()))
conn.commit()
class TestDiagnosticsPanel:
@pytest.fixture
def conn(self, tmp_path):
c = init_store(tmp_path / "o.db")
yield c
c.close()
def _text(self, conn):
selection = ActivitySelection()
selection.toggle_diagnostics()
return diagnostics_text(readout(conn, selection, NOW))
def test_new_counters_show_value_and_delta(self, conn):
_diag_sample(conn, "2026-09-30T18:00:00+00:00", host_reads=1000,
host_writes=500, controller_busy_time=60,
num_err_log_entries=0, warning_temp_time=2,
critical_comp_time=0)
_diag_sample(conn, "2026-09-30T19:50:00+00:00", host_reads=1300,
host_writes=550, controller_busy_time=65,
num_err_log_entries=1, warning_temp_time=2,
critical_comp_time=0)
text = self._text(conn)
assert "Host reads 1300 Δ +300" in text
assert "Host writes 550 Δ +50" in text
assert "Controller busy time 65 min Δ +5" in text
assert "Error log entries 1 Δ +1" in text
assert "Warning temperature time 2 min Δ ±0" in text
assert "Critical temperature time 0 min Δ ±0" in text
def test_legacy_rows_read_n_a_and_absent_thermal_management_is_hidden(self, conn):
_diag_sample(conn, "2026-09-30T19:50:00+00:00", power_on_hours=5)
text = self._text(conn)
assert "Host reads n/a" in text
assert "Thermal management" not in text
def test_thermal_management_shown_when_the_drive_reports_it(self, conn):
_diag_sample(conn, "2026-09-30T19:50:00+00:00", thm_temp1_trans_count=3,
thm_temp1_total_time=120, thm_temp2_trans_count=1,
thm_temp2_total_time=45)
text = self._text(conn)
assert "Thermal management 1 transitions 3 Δ n/a" in text
assert "Thermal management 1 time 120 s Δ n/a" in text
assert "Thermal management 2 transitions 1 Δ n/a" in text
def test_unmigrated_store_still_renders_diagnostics(self, tmp_path):
"""The dashboard may open a store before the upgrade migrates it."""
db = tmp_path / "legacy.db"
_legacy_store(db)
conn = sqlite3.connect(db)
selection = ActivitySelection()
selection.toggle_diagnostics()
text = diagnostics_text(readout(conn, selection, NOW))
assert "Host reads n/a" in text and "Power-on hours" in text
+1 -1
View File
@@ -18,7 +18,7 @@ from fenris.monitoring_periods import ensure_period_open, close_period
from fenris.derive import derive_hours_from_interval from fenris.derive import derive_hours_from_interval
from fenris.day_aggregate import derive_day from fenris.day_aggregate import derive_day
from fenris.projection import compute_projection from fenris.projection import compute_projection
from fenris.tui import _query_live_graph_data from fenris.activity_readout import _query_live_graph_data
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
+3 -5
View File
@@ -817,19 +817,17 @@ class TestHistoricalSelectionIdentity:
# Move from final local hour to the second 01:00 occurrence. # Move from final local hour to the second 01:00 occurrence.
await pilot.press(*(["left"] * 22)) await pilot.press(*(["left"] * 22))
second_occurrence = str(app.query_one("#bar-readout").render()) second_occurrence = str(app.query_one("#bar-readout").render())
assert "01:00 -0500" in second_occurrence assert "01:00 America/New_York -05:00" in second_occurrence
assert "America/New_York -0500" in second_occurrence
assert "06:00 UTC" in second_occurrence assert "06:00 UTC" in second_occurrence
await pilot.press("left") await pilot.press("left")
first_occurrence = str(app.query_one("#bar-readout").render()) first_occurrence = str(app.query_one("#bar-readout").render())
assert "01:00 -0400" in first_occurrence assert "01:00 America/New_York -04:00" in first_occurrence
assert "America/New_York -0400" in first_occurrence
assert "05:00 UTC" in first_occurrence assert "05:00 UTC" in first_occurrence
await pilot.press("left") await pilot.press("left")
incomplete_hour = str(app.query_one("#bar-readout").render()) incomplete_hour = str(app.query_one("#bar-readout").render())
assert "00:00 -0400" in incomplete_hour assert "00:00 America/New_York -04:00" in incomplete_hour
assert "50% coverage · incomplete" in incomplete_hour assert "50% coverage · incomplete" in incomplete_hour
finally: finally:
patcher.stop() patcher.stop()
+2 -2
View File
@@ -12,7 +12,7 @@ from fenris import collector
from fenris.projection import compute_projection from fenris.projection import compute_projection
from fenris.status import _query_drive_facts, get_status, read_status from fenris.status import _query_drive_facts, get_status, read_status
from fenris.status_composition import render_status_tui from fenris.status_composition import render_status_tui
from fenris.store import init_store from fenris.store import SCHEMA_VERSION, init_store
class FakeClock: class FakeClock:
@@ -288,7 +288,7 @@ def test_v3_readers_ignore_pending_table_until_store_migrates(tmp_path):
migrated = init_store(store_path) migrated = init_store(store_path)
try: try:
assert migrated.execute("PRAGMA user_version").fetchone()[0] == 6 assert migrated.execute("PRAGMA user_version").fetchone()[0] == SCHEMA_VERSION
assert migrated.execute("SELECT COUNT(*) FROM pending_publications").fetchone()[0] == 0 assert migrated.execute("SELECT COUNT(*) FROM pending_publications").fetchone()[0] == 0
finally: finally:
migrated.close() migrated.close()
+8 -9
View File
@@ -18,12 +18,9 @@ from fenris.activity_selection import IntervalIdentity
from fenris.monitoring_periods import ensure_period_open from fenris.monitoring_periods import ensure_period_open
from fenris.status import ACCURACY_SEC, CADENCE_DEFAULT_S, FRESH_THRESHOLD_S from fenris.status import ACCURACY_SEC, CADENCE_DEFAULT_S, FRESH_THRESHOLD_S
from fenris.store import init_store from fenris.store import init_store
from fenris.tui import ( from fenris.timefmt import local_clock, tz_label, utc_range_label
LIVE_WINDOW_H, from fenris.activity_readout import LIVE_WINDOW_H, _query_live_graph_data
FenrisTuiApp, from fenris.tui import FenrisTuiApp, LiveActivityGraph
LiveActivityGraph,
_query_live_graph_data,
)
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# Helpers # Helpers
@@ -70,8 +67,10 @@ def _live_intervals(count=4):
intervals.append({ intervals.append({
"start_ts": start.isoformat(), "start_ts": start.isoformat(),
"end_ts": end.isoformat(), "end_ts": end.isoformat(),
"start_label": start.strftime("%H:%M"), "start_label": local_clock(start, "UTC"),
"end_label": end.strftime("%H:%M"), "end_label": local_clock(end, "UTC"),
"timezone_label": tz_label(end, "UTC"),
"utc_label": utc_range_label(start, end),
"bytes_written": (index + 1) * 1000, "bytes_written": (index + 1) * 1000,
"bytes_read": (index + 1) * 500, "bytes_read": (index + 1) * 500,
"elapsed_s": 180, "elapsed_s": 180,
@@ -282,7 +281,7 @@ class TestLiveActivityGraph:
await pilot.pause() await pilot.pause()
header = str(graph.query_one("#live-header").render()) header = str(graph.query_one("#live-header").render())
assert "1 intervals" in header assert "1 intervals" in header
assert "w W/R" in str(app.query_one("#activity-tools").render()) assert "w Toggle read/write" in str(app.query_one("#activity-tools").render())
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
+4 -1
View File
@@ -624,6 +624,7 @@ class TestCollectorIntegration:
assert summary["activity_state"] in ("incomplete", "so_far") assert summary["activity_state"] in ("incomplete", "so_far")
conn.close() conn.close()
from fenris.activity_readout import selected_local_day
from fenris.status import read_status from fenris.status import read_status
from fenris.tui import FenrisTuiApp from fenris.tui import FenrisTuiApp
@@ -634,7 +635,9 @@ class TestCollectorIntegration:
Path(store), second_at, query_services=False Path(store), second_at, query_services=False
) as (reader, _): ) as (reader, _):
assert reader is not None assert reader is not None
app._render_local_day(reader) app._render_local_day(*selected_local_day(
reader, app._activity_selection, second_at,
))
visible = str(app.query_one("#local-day").render()) visible = str(app.query_one("#local-day").render())
assert "2026-09-02" in visible assert "2026-09-02" in visible
assert "Asia/Kolkata +05:30" in visible assert "Asia/Kolkata +05:30" in visible
+1 -1
View File
@@ -22,7 +22,7 @@ sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
from fenris.collector import run_collection from fenris.collector import run_collection
from fenris.day_aggregate import derive_day from fenris.day_aggregate import derive_day
from fenris.tui import _query_daily_graph_data, _query_hourly_graph_data from fenris.activity_readout import _query_daily_graph_data, _query_hourly_graph_data
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
+19
View File
@@ -83,6 +83,25 @@ class TestEnableIdempotentMatrix:
assert cursor.fetchone() is not None assert cursor.fetchone() is not None
conn.close() conn.close()
def test_enable_reports_period_start_in_local_time(
self, store_path, monkeypatch, capsys,
):
"""The opened-at message is local time with its timezone label."""
monkeypatch.setenv("TZ", "Asia/Kolkata")
args = MagicMock(now=True, store_path=store_path)
conn = init_store(store_path)
conn.execute("DELETE FROM monitoring_periods")
conn.commit()
conn.close()
with patch("fenris.monitor.enable_timer"):
cmd_enable(args)
out = capsys.readouterr().out
assert "Monitoring period opened at" in out
assert "Asia/Kolkata +05:30" in out
assert "+00:00" not in out
def test_resume_with_open_period_noop(self, store_path): def test_resume_with_open_period_noop(self, store_path):
"""Resume with open period: no-op (gap stays inside as unknown).""" """Resume with open period: no-op (gap stays inside as unknown)."""
# Initialize store and open a period # Initialize store and open a period
+106 -18
View File
@@ -72,12 +72,13 @@ def _insert_day(conn, day, bw=1024*1024*100, coverage=0.95, samples=24):
conn.commit() conn.commit()
def _insert_sample(conn, ts, pu=5): def _insert_sample(conn, ts, pu=5, bytes_written=512000000000, segment_id=1):
conn.execute( conn.execute(
"INSERT INTO samples (ts, device, data_units_written, data_units_read, " "INSERT INTO samples (ts, device, data_units_written, data_units_read, "
"percentage_used, bytes_written, bytes_read, power_on_hours) " "percentage_used, bytes_written, bytes_read, power_on_hours, segment_id) "
"VALUES (?, ?, ?, ?, ?, ?, ?, ?)", "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
(ts, "/dev/nvme0n1", 1000000, 500000, pu, 512000000000, 256000000000, 8765), (ts, "/dev/nvme0n1", bytes_written // 512000, 500000, pu, bytes_written,
256000000000, 8765, segment_id),
) )
conn.commit() conn.commit()
@@ -221,6 +222,88 @@ class TestImpliedBaseline:
result = compute_projection(store, _clock()) result = compute_projection(store, _clock())
assert result.baseline_tier == BaselineTier.VERIFIED assert result.baseline_tier == BaselineTier.VERIFIED
def _setup_implied(self, store, pus):
"""No override baseline; one sample per PU value inside the segment."""
_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=1024*1024*100)
for i, pu in enumerate(pus):
_insert_sample(store, "2026-09-%02dT10:00:00+00:00" % (20 + i), pu=pu)
_insert_complete_local_days(store, "2026-09-29", 1)
@pytest.mark.parametrize("pus", [[5], [5, 5, 5], [5, 6]])
def test_implied_unavailable_before_two_increments(self, store, pus):
self._setup_implied(store, pus)
result = compute_projection(store, _clock())
assert result.confidence_state == ConfidenceState.UNSUPPORTED
assert result.baseline_tier == BaselineTier.NONE
assert result.headline_remaining_seconds is None
assert "vendor wear estimate too coarse to imply endurance" in result.contributing_facts
def test_implied_unavailable_ignores_increments_in_other_segments(self, store):
_insert_segment(store, opened_at="2026-09-01T00:00:00+00:00")
_insert_segment(store, opened_at="2026-09-15T00:00:00+00:00", identity_key="nqn.new")
_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=1024*1024*100)
# Two increments in the old segment, none in the current one
_insert_sample(store, "2026-09-02T10:00:00+00:00", pu=3, segment_id=1)
_insert_sample(store, "2026-09-03T10:00:00+00:00", pu=4, segment_id=1)
_insert_sample(store, "2026-09-20T10:00:00+00:00", pu=5, segment_id=2)
_insert_complete_local_days(store, "2026-09-29", 1)
result = compute_projection(store, _clock())
assert result.baseline_tier == BaselineTier.NONE
assert "vendor wear estimate too coarse to imply endurance" in result.contributing_facts
def test_implied_unavailable_when_pu_out_of_range(self, store):
self._setup_implied(store, [0, 1, 2])
store.execute("UPDATE samples SET percentage_used = 255 WHERE id = 3")
store.commit()
result = compute_projection(store, _clock())
assert result.baseline_tier == BaselineTier.NONE
assert "vendor wear estimate too coarse to imply endurance" in result.contributing_facts
def test_implied_eligible_after_two_increments(self, store):
self._setup_implied(store, [5, 6, 7])
W_t = 512000000000
result = compute_projection(store, _clock())
assert result.baseline_tier == BaselineTier.IMPLIED
# Implied baseline is never Supported-eligible
assert result.confidence_state == ConfidenceState.LIMITED
assert "vendor wear estimate too coarse to imply endurance" not in result.contributing_facts
E_implied = 100 * W_t / 7
regime_bytes = 20 * 1024 * 1024 * 100
regime_start = datetime(2026, 9, 10, 0, 0, 0, tzinfo=timezone.utc)
rate = regime_bytes / int((_clock() - regime_start).total_seconds())
assert abs(result.headline_remaining_seconds - max(E_implied - W_t, 0) / rate) < 1.0
def test_override_wins_over_implied(self, store):
_insert_baseline(store, tbw_tb=1.0, verified=True)
self._setup_implied(store, [5, 6, 7])
result = compute_projection(store, _clock())
assert result.baseline_tier == BaselineTier.VERIFIED
class TestHorizonAgreement:
def test_single_horizon_passes_agreement(self, store):
"""ADR 0002 §8: agreement is judged across existing horizons; one passes."""
_insert_baseline(store, tbw_tb=1.0, verified=True)
_insert_segment(store)
_open_period(store)
# 14 days covers only the 7-day horizon
for i in range(14):
d = (datetime(2026, 9, 17) + timedelta(days=i)).strftime("%Y-%m-%d")
_insert_day(store, d, bw=1024*1024*100)
_insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5)
_insert_complete_local_days(store, "2026-09-29", 1)
result = compute_projection(store, _clock())
assert list(result.scenario_range.rates) == [7]
assert not any("regime only" in f for f in result.contributing_facts)
assert result.confidence_state == ConfidenceState.SUPPORTED
class TestZeroRate: class TestZeroRate:
def test_zero_rate_fixed_phrase(self, store): def test_zero_rate_fixed_phrase(self, store):
@@ -313,15 +396,15 @@ class TestArithmetic:
_insert_day(store, d, bw=1024*1024*100) _insert_day(store, d, bw=1024*1024*100)
_insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5) _insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5)
result = compute_projection(store, _clock()) result = compute_projection(store, _clock())
if result.headline_remaining_seconds is not None:
E_rated = 1.0 * TBW_TO_BYTES E_rated = 1.0 * TBW_TO_BYTES
W_t = 512000000000 # lifetime bytes written of the newest published sample
regime_bytes = 30 * 1024 * 1024 * 100 regime_bytes = 30 * 1024 * 1024 * 100
# Actual wall-clock: Sep 1 00:00 -> Sep 30 12:00 = 29.5 days # Actual wall-clock: Sep 1 00:00 -> Sep 30 12:00 = 29.5 days
period_start = datetime(2026, 9, 1, 0, 0, 0, tzinfo=timezone.utc) period_start = datetime(2026, 9, 1, 0, 0, 0, tzinfo=timezone.utc)
period_end = _clock() actual_wc = int((_clock() - period_start).total_seconds())
actual_wc = int((period_end - period_start).total_seconds())
rate = regime_bytes / actual_wc rate = regime_bytes / actual_wc
expected = max(E_rated - regime_bytes, 0) / rate expected = max(E_rated - W_t, 0) / rate
assert result.headline_remaining_seconds is not None
assert abs(result.headline_remaining_seconds - expected) < 1.0 assert abs(result.headline_remaining_seconds - expected) < 1.0
def test_implied_baseline_formula(self, store): def test_implied_baseline_formula(self, store):
@@ -337,16 +420,18 @@ class TestArithmetic:
for i in range(30): for i in range(30):
d = (datetime(2026, 9, 1) + timedelta(days=i)).strftime("%Y-%m-%d") d = (datetime(2026, 9, 1) + timedelta(days=i)).strftime("%Y-%m-%d")
_insert_day(store, d, bw=1024*1024*100) _insert_day(store, d, bw=1024*1024*100)
_insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5) W_t = 512000000000 # lifetime written bytes of the newest sample
_insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5, bytes_written=W_t)
_insert_complete_local_days(store, "2026-09-29", 1)
result = compute_projection(store, _clock()) result = compute_projection(store, _clock())
if result.headline_remaining_seconds is not None: assert result.headline_remaining_seconds is not None
E_rated = 2.0 * TBW_TO_BYTES E_rated = 2.0 * TBW_TO_BYTES
regime_bytes = 30 * 1024 * 1024 * 100 regime_bytes = 30 * 1024 * 1024 * 100
period_start = datetime(2026, 9, 1, 0, 0, 0, tzinfo=timezone.utc) period_start = datetime(2026, 9, 1, 0, 0, 0, tzinfo=timezone.utc)
period_end = _clock() actual_wc = int((_clock() - period_start).total_seconds())
actual_wc = int((period_end - period_start).total_seconds())
rate = regime_bytes / actual_wc rate = regime_bytes / actual_wc
expected = max(E_rated - regime_bytes, 0) / rate # W_t is lifetime written bytes, not the regime delta
expected = max(E_rated - W_t, 0) / rate
assert abs(result.headline_remaining_seconds - expected) < 1.0 assert abs(result.headline_remaining_seconds - expected) < 1.0
def test_wearing_rate_proportional(self, store): def test_wearing_rate_proportional(self, store):
@@ -357,6 +442,7 @@ class TestArithmetic:
d = (datetime(2026, 9, 1) + timedelta(days=i)).strftime("%Y-%m-%d") d = (datetime(2026, 9, 1) + timedelta(days=i)).strftime("%Y-%m-%d")
_insert_day(store, d, bw=1024*1024*100) _insert_day(store, d, bw=1024*1024*100)
_insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5) _insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5)
_insert_complete_local_days(store, "2026-09-29", 1)
r_slow = compute_projection(store, _clock()) r_slow = compute_projection(store, _clock())
store.execute("DELETE FROM day_aggregates") store.execute("DELETE FROM day_aggregates")
@@ -366,11 +452,11 @@ class TestArithmetic:
_insert_day(store, d, bw=2*1024*1024*100) _insert_day(store, d, bw=2*1024*1024*100)
r_fast = compute_projection(store, _clock()) r_fast = compute_projection(store, _clock())
if r_slow.headline_remaining_seconds is not None and r_fast.headline_remaining_seconds is not None: assert r_slow.headline_remaining_seconds is not None
assert 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 # Issue #26: Project from the sustained regime
# Habit change, scenario range, and evidence gates # Habit change, scenario range, and evidence gates
@@ -392,16 +478,18 @@ class TestSustainedRegimeRate:
for i in range(30): for i in range(30):
d = (datetime(2026, 9, 1) + timedelta(days=i)).strftime("%Y-%m-%d") d = (datetime(2026, 9, 1) + timedelta(days=i)).strftime("%Y-%m-%d")
_insert_day(store, d, bw=bw) _insert_day(store, d, bw=bw)
_insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5) W_t = 512000000000
_insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5, bytes_written=W_t)
_insert_complete_local_days(store, "2026-09-29", 1)
result = compute_projection(store, _clock()) result = compute_projection(store, _clock())
# Regime = full 30 days; rate = 30*bw / wall-clock # Regime = full 30 days; rate = 30*bw / wall-clock
regime_bytes = 30 * bw regime_bytes = 30 * bw
period_start = datetime(2026, 9, 1, 0, 0, 0, tzinfo=timezone.utc) period_start = datetime(2026, 9, 1, 0, 0, 0, tzinfo=timezone.utc)
wc = int((_clock() - period_start).total_seconds()) wc = int((_clock() - period_start).total_seconds())
expected_rate = regime_bytes / wc expected_rate = regime_bytes / wc
if result.headline_remaining_seconds is not None: assert result.headline_remaining_seconds is not None
E = 10.0 * TBW_TO_BYTES E = 10.0 * TBW_TO_BYTES
expected_seconds = max(E - regime_bytes, 0) / expected_rate expected_seconds = max(E - W_t, 0) / expected_rate
assert abs(result.headline_remaining_seconds - expected_seconds) < 1.0 assert abs(result.headline_remaining_seconds - expected_seconds) < 1.0
def test_regime_capped_at_90_days(self, store): def test_regime_capped_at_90_days(self, store):
+221
View File
@@ -0,0 +1,221 @@
"""Staged projection with an evidence ladder (issue #106, ADR 0012).
Every gate is exercised through compute_projection() against SQLite
fixtures built by seed_evidence(): N hours or days of complete evidence
written as hour observations, then derived into day aggregates the same
way a collection run does.
"""
import sys
from datetime import datetime, timedelta, timezone
from pathlib import Path
import pytest
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
from fenris.day_aggregate import derive_all_days, persist_day_aggregate
from fenris.monitoring_periods import ensure_period_open
from fenris.projection import (
BaselineTier, ConfidenceState, ProjectionStage, compute_projection,
)
from fenris.store import init_store
START = datetime(2026, 9, 1, 0, 0, 0, tzinfo=timezone.utc)
HOUR_BYTES = 10 * 1024 ** 3
MODEL = "Samsung SSD 970 EVO Plus 1TB"
@pytest.fixture
def store(tmp_path):
conn = init_store(tmp_path / "test.db")
yield conn
conn.close()
def seed_evidence(conn, hours, baseline="verified", start=START, degraded=False):
"""Seed *hours* hours of complete evidence and return the clock.
Monitoring opens at *start*; each hour is fully sampled and active.
The returned clock is the end of the last observed hour.
"""
conn.execute(
"INSERT INTO controller_segments "
"(opened_at, identity_key, identity_degraded, subnqn, sn, mn, fr, vid, ssvid, transport) "
"VALUES (?, ?, ?, 'nqn.test', 'SN1', ?, 'FW1', '0x144d', '0x144d', 'pcie')",
(start.isoformat(), "" if degraded else "nqn.test", degraded, MODEL),
)
if baseline:
conn.execute(
"INSERT INTO endurance_baseline "
"(tbw_terabytes, source_url, document_revision, entry_date, model_string, "
" nominal_capacity_bytes, validated_by, verified, created_at, updated_at) "
"VALUES (600, ?, 'v1', '2026-01-01', ?, 1024000000000, ?, ?, "
" '2026-01-01T00:00:00+00:00', '2026-01-01T00:00:00+00:00')",
("https://example.com/spec" if baseline == "verified" else None, MODEL,
"machine_match" if baseline == "verified" else None,
baseline == "verified"),
)
ensure_period_open(conn, start)
for h in range(hours):
hour = start + timedelta(hours=h)
conn.execute(
"INSERT INTO hour_observations "
"(hour, active_seconds, idle_seconds, powered_off_seconds, unknown_seconds, "
" bytes_written_delta, bytes_read_delta, sample_count, coverage) "
"VALUES (?, 3600, 0, 0, 0, ?, 0, 1, 1.0)",
(hour.strftime("%Y-%m-%dT%H:00:00Z"), HOUR_BYTES),
)
written = HOUR_BYTES * (h + 1)
conn.execute(
"INSERT INTO samples (ts, device, data_units_written, data_units_read, "
"percentage_used, bytes_written, bytes_read, power_on_hours, segment_id) "
"VALUES (?, '/dev/nvme0n1', ?, 0, 1, ?, 0, 100, 1)",
((hour + timedelta(hours=1)).isoformat(), written // 512000, written),
)
for aggregate in derive_all_days(conn):
persist_day_aggregate(conn, aggregate)
conn.commit()
return start + timedelta(hours=hours)
class TestHourStages:
def test_empty_store_has_no_observations(self, store):
result = compute_projection(store, START)
assert result.stage == ProjectionStage.NO_OBSERVATIONS
assert result.headline_remaining_seconds is None
def test_two_hours_is_too_few_for_a_number(self, store):
clock = seed_evidence(store, 2)
result = compute_projection(store, clock)
assert result.stage == ProjectionStage.NO_OBSERVATIONS
assert result.headline_remaining_seconds is None
assert result.hours_observed == 2
def test_three_hours_is_provisional_with_a_number(self, store):
clock = seed_evidence(store, 3)
result = compute_projection(store, clock)
assert result.stage == ProjectionStage.PROVISIONAL
assert result.headline_remaining_seconds is not None
assert result.headline_remaining_seconds > 0
assert result.hours_observed == 3
assert "3 hours observed" in result.contributing_facts
assert "daily cycle not yet seen" in result.contributing_facts
def test_three_hours_shows_short_horizon_spread(self, store):
clock = seed_evidence(store, 3)
result = compute_projection(store, clock)
assert result.scenario_range is not None
assert set(result.scenario_range.rates) <= {1, 3}
assert 1 in result.scenario_range.rates
def test_provisional_rate_is_the_observed_write_rate(self, store):
clock = seed_evidence(store, 3)
result = compute_projection(store, clock)
assert result.write_rate == pytest.approx(HOUR_BYTES / 3600, rel=0.01)
def test_complete_local_day_is_a_fact_not_a_gate(self, store):
clock = seed_evidence(store, 3)
result = compute_projection(store, clock)
assert result.headline_remaining_seconds is not None
assert "waiting for a full local observation day" in result.contributing_facts
def test_provisional_until_24_hours_then_warming(self, store):
clock = seed_evidence(store, 23)
assert compute_projection(store, clock).stage == ProjectionStage.PROVISIONAL
def test_twenty_four_hours_is_warming(self, store):
clock = seed_evidence(store, 24)
result = compute_projection(store, clock)
assert result.stage == ProjectionStage.WARMING
assert "daily cycle not yet seen" not in result.contributing_facts
class TestDayStages:
def test_warming_until_fourteen_qualifying_days(self, store):
clock = seed_evidence(store, 24 * 13)
assert compute_projection(store, clock).stage == ProjectionStage.WARMING
def test_fourteen_days_with_verified_baseline_is_supported(self, store):
clock = seed_evidence(store, 24 * 14)
result = compute_projection(store, clock)
assert result.stage == ProjectionStage.SUPPORTED
assert result.confidence_state == ConfidenceState.SUPPORTED
def test_unverified_baseline_is_limited_after_warm_up(self, store):
clock = seed_evidence(store, 24 * 14, baseline="unverified")
result = compute_projection(store, clock)
assert result.stage == ProjectionStage.LIMITED
assert result.baseline_tier == BaselineTier.UNVERIFIED
def test_degraded_identity_is_limited(self, store):
clock = seed_evidence(store, 24 * 14, degraded=True)
assert compute_projection(store, clock).stage == ProjectionStage.LIMITED
def test_stale_history_is_limited(self, store):
clock = seed_evidence(store, 24 * 14) + timedelta(days=4)
assert compute_projection(store, clock).stage == ProjectionStage.LIMITED
class TestNoBaseline:
def test_write_rate_replaces_lifespan(self, store):
clock = seed_evidence(store, 3, baseline=None)
result = compute_projection(store, clock)
assert result.stage == ProjectionStage.UNAVAILABLE
assert result.headline_remaining_seconds is None
assert result.write_rate == pytest.approx(HOUR_BYTES / 3600, rel=0.01)
def test_no_baseline_is_not_synthesised(self, store):
clock = seed_evidence(store, 24 * 14, baseline=None)
result = compute_projection(store, clock)
assert result.baseline_tier == BaselineTier.NONE
assert result.headline_remaining_seconds is None
assert result.write_rate is not None
class TestShortHorizons:
def test_short_horizons_appear_once_covered(self, store):
clock = seed_evidence(store, 24 * 2)
rates = compute_projection(store, clock).scenario_range.rates
assert 1 in rates
assert 3 not in rates
def test_three_day_horizon_covered_at_four_days(self, store):
clock = seed_evidence(store, 24 * 4)
rates = compute_projection(store, clock).scenario_range.rates
assert {1, 3} <= set(rates)
def test_short_horizons_drop_when_seven_day_covered(self, store):
clock = seed_evidence(store, 24 * 9)
rates = compute_projection(store, clock).scenario_range.rates
assert 7 in rates
assert 1 not in rates and 3 not in rates
class TestEvidenceLadder:
def test_ladder_lists_each_supported_condition_with_reason(self, store):
clock = seed_evidence(store, 3)
ladder = compute_projection(store, clock).evidence_ladder
assert len(ladder) == 8
assert all(c.reason for c in ladder)
assert all(isinstance(c.met, bool) for c in ladder)
def test_ladder_all_met_when_supported(self, store):
clock = seed_evidence(store, 24 * 14)
result = compute_projection(store, clock)
assert result.ladder_met == result.ladder_total == 8
def test_ladder_count_rises_as_evidence_accrues(self, store, tmp_path):
counts = []
for hours in (3, 24 * 2, 24 * 9, 24 * 14):
conn = init_store(tmp_path / ("ladder-%d.db" % hours))
clock = seed_evidence(conn, hours)
counts.append(compute_projection(conn, clock).ladder_met)
conn.close()
assert counts == sorted(counts)
assert counts[0] < counts[-1]
assert len(set(counts)) >= 3
def test_unmet_condition_names_its_reason(self, store):
clock = seed_evidence(store, 3)
unmet = [c for c in compute_projection(store, clock).evidence_ladder if not c.met]
assert any("qualifying days" in c.reason for c in unmet)
+112
View File
@@ -0,0 +1,112 @@
"""CLI and TUI render the projection stage, never pre-gate it (issue #106).
Seams: get_status() and the TUI headline band. Both must show what
compute_projection() reports: the stage, the ladder count, and the full
ladder in the outlook; no caller decides what the projection can show.
"""
import sys
from datetime import datetime, timedelta, timezone
from pathlib import Path
from unittest.mock import patch
import pytest
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
sys.path.insert(0, str(Path(__file__).parent))
from fenris.status import get_status
from fenris.store import init_store
from fenris.tui import FenrisTuiApp
from test_staged_projection import seed_evidence
SERVICES = {
"boot_enabled": False, "timer_active": False,
"last_collect_ok": None, "last_collect_age_s": None,
"last_collect_reason": None,
}
def _status(db, clock):
with patch("fenris.status.query_service_state", return_value=SERVICES):
return get_status(store_path=db, clock_now=clock,
query_services=False, query_journal=False)
@pytest.fixture
def db(tmp_path):
return tmp_path / "observations.db"
class TestCliStatus:
def test_provisional_stage_ladder_count_and_full_ladder(self, db):
conn = init_store(db)
clock = seed_evidence(conn, 3)
conn.close()
out = _status(db, clock)
assert "Provisional" in out
assert "3 hours observed" in out
assert "daily cycle not yet seen" in out
assert "remaining" in out
assert "conditions met" in out
assert "Evidence ladder" in out
assert "[ ] 14 qualifying days" in out
assert "[x] verified baseline" in out
def test_single_sample_is_not_pre_gated(self, db):
conn = init_store(db)
seed_evidence(conn, 1)
conn.close()
out = _status(db, datetime(2026, 9, 1, 1, 0, tzinfo=timezone.utc))
assert "requires at least two samples" not in out
assert "1 of 3 hours observed" in out
def test_no_baseline_shows_write_rate_not_lifespan(self, db):
conn = init_store(db)
clock = seed_evidence(conn, 3, baseline=None)
conn.close()
out = _status(db, clock)
assert "write rate" in out.lower()
assert "GB/day" in out
def test_supported_stage(self, db):
conn = init_store(db)
clock = seed_evidence(conn, 24 * 14)
conn.close()
out = _status(db, clock)
assert "Supported" in out
assert "8 of 8 conditions met" in out
class TestTuiHeadline:
@pytest.mark.asyncio
async def test_provisional_headline_and_ladder(self, db):
now = datetime.now(timezone.utc).replace(minute=0, second=0, microsecond=0)
conn = init_store(db)
seed_evidence(conn, 3, start=now - timedelta(hours=3))
conn.close()
with patch("fenris.status.query_service_state", return_value=SERVICES):
app = FenrisTuiApp(store_path=db)
async with app.run_test(size=(120, 50)):
text = str(app.query_one("#headline-band").render())
assert "Provisional" in text
assert "3 hours observed" in text
assert "conditions met" in text
assert "Evidence ladder" in text
@pytest.mark.asyncio
async def test_single_sample_is_not_pre_gated(self, db):
now = datetime.now(timezone.utc).replace(minute=0, second=0, microsecond=0)
conn = init_store(db)
seed_evidence(conn, 1, start=now - timedelta(hours=1))
conn.close()
with patch("fenris.status.query_service_state", return_value=SERVICES):
app = FenrisTuiApp(store_path=db)
async with app.run_test(size=(120, 50)):
text = str(app.query_one("#headline-band").render())
assert "at least two samples" not in text
def test_callers_carry_no_projection_pre_gate():
root = Path(__file__).parent.parent / "src" / "fenris"
for name in ("tui.py", "status.py"):
assert "at least two samples" not in (root / name).read_text()
+1 -1
View File
@@ -23,7 +23,6 @@ sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
from fenris.status import ( from fenris.status import (
grade_freshness, grade_freshness,
freshness_age_human,
format_disclosures, format_disclosures,
check_retired_command, check_retired_command,
check_retired_flag, check_retired_flag,
@@ -34,6 +33,7 @@ from fenris.status import (
) )
from fenris.store import init_store, SCHEMA_VERSION from fenris.store import init_store, SCHEMA_VERSION
from fenris.projection import DISCLOSURES from fenris.projection import DISCLOSURES
from fenris.timefmt import freshness_age_human
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
+1 -2
View File
@@ -952,8 +952,7 @@ class TestTUIIntegration:
conn = init_store(tmp_path / "test.db") conn = init_store(tmp_path / "test.db")
_insert_segment(conn) _insert_segment(conn)
_open_period(conn) _open_period(conn)
now = _clock() ts = (datetime.now(timezone.utc) - timedelta(minutes=2)).isoformat()
ts = (now - timedelta(minutes=2)).isoformat()
_insert_sample(conn, ts) _insert_sample(conn, ts)
for i in range(20): for i in range(20):
d = (datetime(2026, 9, 10) + timedelta(days=i)).strftime("%Y-%m-%d") d = (datetime(2026, 9, 10) + timedelta(days=i)).strftime("%Y-%m-%d")
+9 -17
View File
@@ -31,13 +31,15 @@ from fenris.status import (
FRESH_THRESHOLD_S, FRESH_THRESHOLD_S,
grade_freshness, grade_freshness,
) )
from fenris.activity_readout import (
_query_daily_graph_data,
_query_drive_health,
_query_hourly_graph_data,
)
from fenris.tui import ( from fenris.tui import (
FenrisTuiApp, FenrisTuiApp,
HistoryGraph, HistoryGraph,
_format_remaining, _format_remaining,
_query_drive_health,
_query_daily_graph_data,
_query_hourly_graph_data,
_RANGE_DEFAULT, _RANGE_DEFAULT,
_MIN_WIDTH, _MIN_WIDTH,
_MIN_HEIGHT, _MIN_HEIGHT,
@@ -99,8 +101,8 @@ def _insert_day(conn, day, bw=1024*1024*100, coverage=0.95, samples=24):
def _insert_sample(conn, ts, pu=5, device="/dev/nvme0n1"): def _insert_sample(conn, ts, pu=5, device="/dev/nvme0n1"):
conn.execute( conn.execute(
"INSERT INTO samples (ts, device, data_units_written, data_units_read, " "INSERT INTO samples (ts, device, data_units_written, data_units_read, "
"percentage_used, bytes_written, bytes_read, power_on_hours) " "percentage_used, bytes_written, bytes_read, power_on_hours, segment_id) "
"VALUES (?, ?, ?, ?, ?, ?, ?, ?)", "VALUES (?, ?, ?, ?, ?, ?, ?, ?, 1)",
(ts, device, 1000000, 500000, pu, 512000000000, 256000000000, 8765), (ts, device, 1000000, 500000, pu, 512000000000, 256000000000, 8765),
) )
conn.commit() conn.commit()
@@ -476,7 +478,6 @@ class TestDenseScreen:
auth_notice = "Open with fenris (no sudo). Actions authenticate via polkit. ? Help" auth_notice = "Open with fenris (no sudo). Actions authenticate via polkit. ? Help"
async with app.run_test() as pilot: async with app.run_test() as pilot:
str(app.query_one("#headline-band").render())
# Identity now shows Fenris by Bongbetic (issue #79) # Identity now shows Fenris by Bongbetic (issue #79)
assert "Fenris by Bongbetic" in str(app.query_one("#identity").render()) assert "Fenris by Bongbetic" in str(app.query_one("#identity").render())
assert auth_notice in str(app.query_one("#auth-notice").render()) assert auth_notice in str(app.query_one("#auth-notice").render())
@@ -585,7 +586,6 @@ class TestFirstRun:
app = FenrisTuiApp(store_path=tmp_path / "nonexistent.db") app = FenrisTuiApp(store_path=tmp_path / "nonexistent.db")
async with app.run_test(): async with app.run_test():
headline = str(app.query_one("#headline-band").render()) headline = str(app.query_one("#headline-band").render())
str(app.query_one("#service-strip").render())
assert "no observations yet" in headline.lower() assert "no observations yet" in headline.lower()
assert "r resume — enable monitoring and future boots" in headline.lower() assert "r resume — enable monitoring and future boots" in headline.lower()
assert "r resume" in str(app.query_one("#action-rail").render()).lower() assert "r resume" in str(app.query_one("#action-rail").render()).lower()
@@ -858,7 +858,6 @@ class TestQueryHourlyGraphData:
class TestHistoryGraph: class TestHistoryGraph:
def test_empty_data(self, tmp_path): def test_empty_data(self, tmp_path):
"""Empty data shows awaiting message.""" """Empty data shows awaiting message."""
FenrisTuiApp(store_path=tmp_path / "nonexistent.db")
# We test the widget directly via the app's compose # We test the widget directly via the app's compose
graph = HistoryGraph() graph = HistoryGraph()
# Simulate setting empty data # Simulate setting empty data
@@ -1121,7 +1120,6 @@ class TestBarGraphTUI:
app = FenrisTuiApp(store_path=tmp_path / "nonexistent.db") app = FenrisTuiApp(store_path=tmp_path / "nonexistent.db")
async with app.run_test(size=(80, 24)) as pilot: async with app.run_test(size=(80, 24)) as pilot:
await pilot.press("v", "v") await pilot.press("v", "v")
app.query_one("#usage-history")
assert app.query_one("#activity-panel").border_title == "Drive activity" assert app.query_one("#activity-panel").border_title == "Drive activity"
@pytest.mark.asyncio @pytest.mark.asyncio
@@ -1347,15 +1345,9 @@ class TestConstrainedLayout:
headline = str(app.query_one("#headline-band").render()) headline = str(app.query_one("#headline-band").render())
assert "remaining" in headline.lower() or "projection" in headline.lower() assert "remaining" in headline.lower() or "projection" in headline.lower()
# Service strip should have actions # Under 80x24 the footer collapses to help and quit (issue #110)
strip = str(app.query_one("#action-rail").render())
assert "p Pause" in strip
assert "r Resume" in strip
assert "q Quit TUI" in strip
# Quit rail should be visible
rail = str(app.query_one("#action-rail").render()) rail = str(app.query_one("#action-rail").render())
assert "Quit" in rail assert rail == "? help · q quit"
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_constrained_long_reasons_visible(self, tmp_path): async def test_constrained_long_reasons_visible(self, tmp_path):
+214
View File
@@ -0,0 +1,214 @@
"""Enter rated TBW from the TUI (issue #107, ADR 0012 §7).
Seam: the app's helper invocation (`_run_helper`), faked here the way the
other TUI tests fake it. The form builds an unverified override and hands it
to `fenris-monitor baseline set`; it never offers verified provenance.
"""
import json
import sys
from datetime import datetime, timedelta, timezone
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import patch
import pytest
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
sys.path.insert(0, str(Path(__file__).parent))
from fenris.monitor import cmd_baseline_set
from fenris.store import init_store
from fenris.tui import FenrisTuiApp
from test_staged_projection import seed_evidence
SERVICES = {
"boot_enabled": False, "timer_active": False,
"last_collect_ok": None, "last_collect_age_s": None,
"last_collect_reason": None,
}
@pytest.fixture
def db(tmp_path):
return tmp_path / "observations.db"
def _app(db):
return FenrisTuiApp(store_path=db)
def _seed_without_baseline(db):
now = datetime.now(timezone.utc).replace(minute=0, second=0, microsecond=0)
conn = init_store(db)
seed_evidence(conn, 3, baseline=None, start=now - timedelta(hours=3))
conn.close()
def _text(app, selector):
return str(app.query_one(selector).render())
async def _open_form(pilot):
await pilot.press("b")
await pilot.pause()
async def _fill(pilot, app, tbw, url=""):
app.screen.query_one("#tbw-input").value = tbw
app.screen.query_one("#source-input").value = url
await pilot.press("enter")
await pilot.pause()
class TestForm:
@pytest.mark.asyncio
async def test_b_opens_form_with_required_tbw_and_optional_source(self, db):
_seed_without_baseline(db)
with patch("fenris.status.query_service_state", return_value=SERVICES):
app = _app(db)
async with app.run_test(size=(120, 50)) as pilot:
await _open_form(pilot)
assert len(app.screen_stack) == 2
assert app.screen.query_one("#tbw-input")
assert app.screen.query_one("#source-input")
text = str(app.screen.query_one("#baseline-header").render())
assert "unverified" in text.lower()
@pytest.mark.asyncio
@pytest.mark.parametrize("value", ["", "abc", "0", "-5", "nan", "inf"])
async def test_invalid_tbw_is_rejected_without_calling_helper(self, db, value):
_seed_without_baseline(db)
with patch("fenris.status.query_service_state", return_value=SERVICES):
app = _app(db)
with patch.object(app, "_run_helper") as run_helper:
async with app.run_test(size=(120, 50)) as pilot:
await _open_form(pilot)
await _fill(pilot, app, value)
assert len(app.screen_stack) == 2
assert str(app.screen.query_one("#baseline-error").render()).strip()
run_helper.assert_not_called()
@pytest.mark.asyncio
async def test_invalid_source_url_is_rejected(self, db):
_seed_without_baseline(db)
with patch("fenris.status.query_service_state", return_value=SERVICES):
app = _app(db)
with patch.object(app, "_run_helper") as run_helper:
async with app.run_test(size=(120, 50)) as pilot:
await _open_form(pilot)
await _fill(pilot, app, "600", "not a url")
assert len(app.screen_stack) == 2
assert "url" in str(app.screen.query_one("#baseline-error").render()).lower()
run_helper.assert_not_called()
@pytest.mark.asyncio
async def test_escape_cancels_without_calling_helper(self, db):
_seed_without_baseline(db)
with patch("fenris.status.query_service_state", return_value=SERVICES):
app = _app(db)
with patch.object(app, "_run_helper") as run_helper:
async with app.run_test(size=(120, 50)) as pilot:
await _open_form(pilot)
await pilot.press("escape")
await pilot.pause()
assert len(app.screen_stack) == 1
run_helper.assert_not_called()
class TestSubmit:
@pytest.mark.asyncio
async def test_submit_routes_unverified_override_to_baseline_set(self, db):
_seed_without_baseline(db)
with patch("fenris.status.query_service_state", return_value=SERVICES):
app = _app(db)
with patch.object(app, "_run_helper", return_value=True) as run_helper:
async with app.run_test(size=(120, 50)) as pilot:
await _open_form(pilot)
await _fill(pilot, app, "600", "https://example.com/spec")
assert len(app.screen_stack) == 1
run_helper.assert_called_once()
operation, args = run_helper.call_args.args
assert operation == "baseline"
assert args[0] == "set"
payload = json.loads(args[1])
assert payload["tbw_terabytes"] == 600
assert payload["source_url"] == "https://example.com/spec"
assert payload["verified"] is False
assert payload["validated_by"] == "user"
# Incomplete provenance keeps the override unverified.
assert payload["document_revision"] is None
@pytest.mark.asyncio
async def test_blank_source_is_stored_as_absent(self, db):
_seed_without_baseline(db)
with patch("fenris.status.query_service_state", return_value=SERVICES):
app = _app(db)
with patch.object(app, "_run_helper", return_value=True) as run_helper:
async with app.run_test(size=(120, 50)) as pilot:
await _open_form(pilot)
await _fill(pilot, app, "1.5")
payload = json.loads(run_helper.call_args.args[1][1])
assert payload["tbw_terabytes"] == 1.5
assert payload["source_url"] is None
@pytest.mark.asyncio
async def test_headline_changes_from_write_rate_to_lifespan(self, db):
_seed_without_baseline(db)
def fake_helper(operation, extra_args=None):
cmd_baseline_set(SimpleNamespace(store_path=db, baseline_json=extra_args[1]))
return True
with patch("fenris.status.query_service_state", return_value=SERVICES):
app = _app(db)
with patch.object(app, "_run_helper", side_effect=fake_helper):
async with app.run_test(size=(120, 50)) as pilot:
before = _text(app, "#headline-band")
assert "Write rate" in before
assert "remaining" not in before
await _open_form(pilot)
await _fill(pilot, app, "600")
app.on_refresh_tick()
await pilot.pause()
after = _text(app, "#headline-band")
assert "remaining" in after
assert "Write rate" not in after
class TestBanner:
@pytest.mark.asyncio
async def test_banner_shown_when_no_baseline_applies(self, db):
_seed_without_baseline(db)
with patch("fenris.status.query_service_state", return_value=SERVICES):
app = _app(db)
async with app.run_test(size=(120, 50)):
banner = app.query_one("#baseline-banner")
assert banner.display
assert "b" in str(banner.render()).lower()
assert "rated tbw" in str(banner.render()).lower()
@pytest.mark.asyncio
async def test_banner_shown_on_first_run(self, tmp_path):
with patch("fenris.status.query_service_state", return_value=SERVICES):
app = _app(tmp_path / "missing.db")
async with app.run_test(size=(120, 50)):
assert app.query_one("#baseline-banner").display
@pytest.mark.asyncio
async def test_banner_hidden_once_baseline_exists(self, db):
now = datetime.now(timezone.utc).replace(minute=0, second=0, microsecond=0)
conn = init_store(db)
seed_evidence(conn, 3, start=now - timedelta(hours=3))
conn.close()
with patch("fenris.status.query_service_state", return_value=SERVICES):
app = _app(db)
async with app.run_test(size=(120, 50)):
assert not app.query_one("#baseline-banner").display
@pytest.mark.asyncio
async def test_banner_hidden_for_unreadable_store(self, db):
db.write_bytes(b"not a sqlite database" * 100)
with patch("fenris.status.query_service_state", return_value=SERVICES):
app = _app(db)
async with app.run_test(size=(120, 50)):
assert not app.query_one("#baseline-banner").display
+152
View File
@@ -0,0 +1,152 @@
"""Diagnostics panel on the dashboard (issue #109): toggle, render, badge."""
import sqlite3
import pytest
from test_dashboard_design import dashboard, visible # noqa: F401 (fixtures)
from fenris.tui import FenrisTuiApp, HelpScreen
def _store(app):
return sqlite3.connect(app.store_path)
def _set_latest_warning(app, value):
conn = _store(app)
conn.execute(
"UPDATE samples SET critical_warning = ?, power_on_hours = id, "
"unsafe_shutdowns = id / 20 WHERE 1", (value,))
conn.commit()
conn.close()
async def _settle(app, pilot):
app.on_refresh_tick()
await pilot.pause()
def _text(app, selector):
return str(app.query_one(selector).render())
def test_i_is_a_declared_binding_and_d_stays_disclosures():
keys = {b.key: b.action for b in FenrisTuiApp.BINDINGS}
assert keys["i"] == "toggle_diagnostics"
assert keys["d"] == "disclose"
@pytest.mark.asyncio
@pytest.mark.parametrize("size", [(140, 44), (100, 30)])
async def test_i_toggles_the_diagnostics_panel_and_hides_the_drive_pane(dashboard, size): # noqa: F811
app = dashboard
_set_latest_warning(app, 0)
async with app.run_test(size=size) as pilot:
await _settle(app, pilot)
assert not app.query_one("#diagnostics-panel").display
assert app.query_one("#drive-panel").display
await pilot.press("i")
await _settle(app, pilot)
assert app.query_one("#diagnostics-panel").display
assert not app.query_one("#drive-panel").display
body = _text(app, "#diagnostics")
assert "Diagnostics" in body and "Critical warning none" in body
assert "Power-on hours 61 Δ +60" in body
await pilot.press("i")
await _settle(app, pilot)
assert not app.query_one("#diagnostics-panel").display
assert app.query_one("#drive-panel").display
@pytest.mark.asyncio
async def test_footer_and_help_list_the_diagnostics_key(dashboard): # noqa: F811
app = dashboard
async with app.run_test(size=(140, 40)) as pilot:
await _settle(app, pilot)
assert "i Diagnostics" in _text(app, "#action-rail")
await pilot.press("?")
assert isinstance(app.screen, HelpScreen)
assert "Diagnostics" in str(app.screen.query_one("#help-text").render())
@pytest.mark.asyncio
async def test_diagnostics_follow_the_activity_view(dashboard): # noqa: F811
app = dashboard
async with app.run_test(size=(140, 44)) as pilot:
await _settle(app, pilot)
await pilot.press("i")
await _settle(app, pilot)
assert "last 3h" in _text(app, "#diagnostics")
await pilot.press("v") # Day
await _settle(app, pilot)
body = _text(app, "#diagnostics")
assert "· day 20" in body and "Hours ·" in body
await pilot.press("v") # History
await _settle(app, pilot)
assert "last 14 days" in _text(app, "#diagnostics")
@pytest.mark.asyncio
async def test_critical_warning_badge_shows_in_the_outlook_without_the_panel(dashboard): # noqa: F811
app = dashboard
_set_latest_warning(app, 0x04)
async with app.run_test(size=(140, 44)) as pilot:
await _settle(app, pilot)
assert not app.query_one("#diagnostics-panel").display
band = _text(app, "#headline-band")
assert "0x04" in band and "reliability degraded" in band
@pytest.mark.asyncio
async def test_no_badge_when_the_drive_reports_no_warning(dashboard): # noqa: F811
app = dashboard
_set_latest_warning(app, 0)
async with app.run_test(size=(140, 44)) as pilot:
await _settle(app, pilot)
assert "critical warning" not in _text(app, "#headline-band").lower()
@pytest.mark.asyncio
async def test_diagnostics_work_in_constrained_mode(dashboard): # noqa: F811
app = dashboard
_set_latest_warning(app, 0x01)
async with app.run_test(size=(60, 20)) as pilot:
await _settle(app, pilot)
assert app._is_constrained_mode
await pilot.press("i")
await _settle(app, pilot)
assert app.query_one("#diagnostics-panel").display
body = _text(app, "#diagnostics")
assert "available spare below threshold" in body
assert "Power-on hours" in body
await pilot.press("i")
await _settle(app, pilot)
assert app.query_one("#drive-panel").display
@pytest.mark.asyncio
async def test_toggling_while_zoomed_restores_the_grid_and_keeps_focus(dashboard): # noqa: F811
app = dashboard
async with app.run_test(size=(140, 44)) as pilot:
await _settle(app, pilot)
await pilot.press("z")
await pilot.press("i")
await _settle(app, pilot)
assert app._zoomed_panel is None
assert app.focused is not None
@pytest.mark.asyncio
async def test_history_bars_draw_reads_alongside_writes_only_with_diagnostics(dashboard): # noqa: F811
app = dashboard
async with app.run_test(size=(140, 44)) as pilot:
await _settle(app, pilot)
await pilot.press("v", "v") # History
await _settle(app, pilot)
assert "Reads" not in _text(app, "#bar-legend")
await pilot.press("i")
await _settle(app, pilot)
legend = _text(app, "#bar-legend")
assert "Writes" in legend and "Reads" in legend
await pilot.press("i")
await _settle(app, pilot)
assert "Reads" not in _text(app, "#bar-legend")
+456
View File
@@ -0,0 +1,456 @@
"""Unified keyboard and mouse navigation (issue #110).
Every action has one declaration: a Binding. The footer, the help screen and
the clickable chips are all generated from those bindings, and the keyboard,
mouse and wheel paths reach the same actions. Pilot drives each path.
"""
import re
from datetime import date, timedelta
from unittest.mock import patch
import pytest
from test_dashboard_design import dashboard # noqa: F401 (pytest fixture)
from textual import events
from textual.geometry import Region
from fenris.tui import (
ConfirmPause,
DatePickerScreen,
BaselineScreen,
DisclosuresScreen,
FenrisTuiApp,
HelpScreen,
HistoryGraph,
LiveActivityGraph,
)
BINDING_SOURCES = (FenrisTuiApp, LiveActivityGraph, HistoryGraph)
# -- helpers ---------------------------------------------------------------
async def _settle(app, pilot):
app.on_refresh_tick()
await pilot.pause()
def _chip(app, label):
"""Offset of a footer or tool chip label inside its Static."""
for selector in ("#action-rail", "#activity-tools"):
widget = app.query_one(selector)
region = Region(0, 0, widget.size.width, widget.size.height)
for y, strip in enumerate(widget.render_lines(region)):
x = strip.text.find(label)
if x >= 0:
return selector, (x + 1, y)
raise AssertionError("no chip %r in footer" % label)
async def _click_chip(app, pilot, label):
selector, offset = _chip(app, label)
await pilot.click(selector, offset=offset)
await pilot.pause()
def _scroll(widget, direction):
event = (events.MouseScrollUp if direction < 0 else events.MouseScrollDown)(
widget, 3, 1, 0, direction, 0, False, False, False,
)
widget.post_message(event)
def _help_rows(app):
text = str(app.screen.query_one("#help-text").render())
section = text.split("Keyboard controls\n", 1)[1].split("\n\n", 1)[0]
rows = {}
for line in section.splitlines():
keys, description = re.match(r"^(\S.*?)\s{2,}(\S.*)$", line).groups()
rows[description] = set(keys.split(" / "))
return rows
def _binding_rows(app):
rows = {}
for source in BINDING_SOURCES:
for binding in source.BINDINGS:
rows.setdefault(binding.description, set()).add(app.get_key_display(binding))
return rows
# -- one source of truth ---------------------------------------------------
def test_graphs_declare_bindings_instead_of_key_handlers():
for graph in (HistoryGraph, LiveActivityGraph):
assert "on_key" not in vars(graph)
assert graph.BINDINGS
@pytest.mark.asyncio
async def test_help_content_equals_binding_set(dashboard): # noqa: F811
async with dashboard.run_test(size=(100, 36)) as pilot:
await pilot.press("?")
assert isinstance(dashboard.screen, HelpScreen)
assert _help_rows(dashboard) == _binding_rows(dashboard)
assert "The dashboard does not need sudo" in str(
dashboard.screen.query_one("#help-text").render())
@pytest.mark.asyncio
async def test_footer_chips_are_generated_from_visible_bindings(dashboard): # noqa: F811
async with dashboard.run_test(size=(120, 36)) as pilot:
await pilot.pause()
footer = str(dashboard.query_one("#action-rail").render())
shown = [b for b in FenrisTuiApp.BINDINGS if b.show]
assert shown
for binding in shown:
assert "%s %s" % (dashboard.get_key_display(binding), binding.description) in footer
for binding in FenrisTuiApp.BINDINGS:
if not binding.show:
label = "%s %s" % (dashboard.get_key_display(binding), binding.description)
assert label not in footer
@pytest.mark.asyncio
@pytest.mark.parametrize("size", [(79, 24), (80, 23), (60, 18)])
async def test_footer_collapses_under_80x24(dashboard, size): # noqa: F811
async with dashboard.run_test(size=size) as pilot:
await pilot.pause()
assert str(dashboard.query_one("#action-rail").render()) == "? help · q quit"
await _click_chip(dashboard, pilot, "? help")
assert isinstance(dashboard.screen, HelpScreen)
await pilot.press("escape")
await _click_chip(dashboard, pilot, "q quit")
assert not dashboard.is_running
@pytest.mark.asyncio
async def test_footer_is_full_at_80x24(dashboard): # noqa: F811
async with dashboard.run_test(size=(80, 24)) as pilot:
await pilot.pause()
footer = str(dashboard.query_one("#action-rail").render())
assert "p Pause" in footer and "q Quit TUI" in footer
# -- keyboard: live graph --------------------------------------------------
@pytest.mark.asyncio
async def test_live_home_end_and_arrows(dashboard): # noqa: F811
app = dashboard
async with app.run_test(size=(100, 36)) as pilot:
await _settle(app, pilot)
graph = app.query_one("#live-activity")
last = len(graph._data) - 1
assert last > 3
await pilot.press("home")
assert graph._selected_index == 0
await pilot.press("right")
assert graph._selected_index == 1
await pilot.press("end")
assert graph._selected_index == last
await pilot.press("left")
assert graph._selected_index == last - 1
# -- keyboard: history graph -----------------------------------------------
async def _open_history(app, pilot):
await pilot.click("#view-history")
await _settle(app, pilot)
graph = app.query_one("#usage-history")
assert app.focused is graph
return graph
@pytest.mark.asyncio
async def test_history_home_end_and_week_jumps(dashboard): # noqa: F811
app = dashboard
async with app.run_test(size=(100, 36)) as pilot:
graph = await _open_history(app, pilot)
last = len(graph._day_data) - 1
assert last >= 13
await pilot.press("home")
assert graph.selected_index == 0
await pilot.press("end")
assert graph.selected_index == last
assert app._browse_date == graph._day_data[last]["day"]
await pilot.press("pageup")
assert graph.selected_index == last - 7
assert app._browse_date == graph._day_data[last - 7]["day"]
await pilot.press("pagedown")
assert graph.selected_index == last
await pilot.press("pagedown")
assert graph.selected_index == last
await pilot.press("home", "pageup")
assert graph.selected_index == 0
@pytest.mark.asyncio
async def test_history_range_keys_and_drill_bindings(dashboard): # noqa: F811
app = dashboard
async with app.run_test(size=(100, 36)) as pilot:
graph = await _open_history(app, pilot)
await pilot.press("1")
assert graph.range_days == 7
await pilot.press("3")
assert graph.range_days == 30
await pilot.press("left", "enter")
assert app._activity_view == "day" and graph.view_mode == "hourly"
await pilot.press("1")
assert graph.range_days == 30 # range keys only apply to the day graph
await pilot.press("escape")
assert graph.view_mode == "daily" and app._activity_view == "history"
await pilot.press("enter")
await pilot.press("backspace")
assert graph.view_mode == "daily"
@pytest.mark.asyncio
async def test_hour_graph_home_end_arrows_and_week_jump(dashboard): # noqa: F811
app = dashboard
async with app.run_test(size=(100, 36)) as pilot:
graph = await _open_history(app, pilot)
await pilot.press("enter")
day = app._browse_date
await pilot.press("home")
assert graph._hourly_selected == 0
await pilot.press("right")
assert graph._hourly_selected == 1
await pilot.press("end")
assert graph._hourly_selected == len(graph._hour_data) - 1
await pilot.press("left")
assert graph._hourly_selected == len(graph._hour_data) - 2
await pilot.press("pageup")
assert app._activity_view == "day"
assert app._browse_date == (date.fromisoformat(day) - timedelta(days=7)).isoformat()
@pytest.mark.asyncio
async def test_escape_in_zoomed_hour_view_restores_zoom_first(dashboard): # noqa: F811
app = dashboard
async with app.run_test(size=(100, 36)) as pilot:
graph = await _open_history(app, pilot)
await pilot.press("enter", "z")
assert app._zoomed_panel == "activity-panel"
await pilot.press("escape")
assert app._zoomed_panel is None and graph.view_mode == "hourly"
await pilot.press("escape")
assert graph.view_mode == "daily"
# -- keyboard: focus ring --------------------------------------------------
@pytest.mark.asyncio
async def test_tab_cycles_panels_with_visible_focus_ring(dashboard): # noqa: F811
app = dashboard
async with app.run_test(size=(120, 40)) as pilot:
await pilot.pause()
order = [app.focused.id]
for _ in range(3):
await pilot.press("tab")
order.append(app.focused.id)
assert order == ["live-activity", "drive-panel", "endurance-panel", "live-activity"]
await pilot.press("shift+tab")
assert app.focused.id == "endurance-panel"
await pilot.press("shift+tab")
assert app.focused.id == "drive-panel"
def ring(pane_id):
return app.query_one("#" + pane_id).styles.border_top[1]
assert ring("drive-panel") != ring("endurance-panel")
await pilot.press("tab")
assert app.focused.id == "endurance-panel"
assert ring("endurance-panel") != ring("drive-panel")
await pilot.press("tab")
assert ring("activity-panel") != ring("drive-panel")
# -- mouse -----------------------------------------------------------------
@pytest.mark.asyncio
async def test_click_focuses_each_panel(dashboard): # noqa: F811
app = dashboard
async with app.run_test(size=(120, 40)) as pilot:
await pilot.pause()
await pilot.click("#drive-health")
assert app.focused.id == "drive-panel"
await pilot.click("#headline-band")
assert app.focused.id == "endurance-panel"
await pilot.click("#local-day")
assert app.focused.id == "live-activity"
await pilot.click("#view-history")
await pilot.click("#drive-health")
await pilot.click("#local-day")
assert app.focused.id == "usage-history"
@pytest.mark.asyncio
async def test_view_tabs_are_clickable(dashboard): # noqa: F811
app = dashboard
async with app.run_test(size=(100, 36)) as pilot:
for view in ("history", "day", "live"):
await pilot.click("#view-" + view)
assert app._activity_view == view
assert app.query_one("#activity-tabs").active == "view-" + view
@pytest.mark.asyncio
async def test_action_chips_fire_their_bindings(dashboard): # noqa: F811
app = dashboard
calls = []
async with app.run_test(size=(120, 40)) as pilot:
await pilot.pause()
with patch.object(app, "_run_helper", side_effect=lambda *a: calls.append(a) or True):
await _click_chip(app, pilot, "p Pause")
assert isinstance(app.screen, ConfirmPause)
await pilot.press("n")
await _click_chip(app, pilot, "r Resume")
assert calls == [("enable", ["--now"])]
await _click_chip(app, pilot, "c Collect")
assert calls[-1] == ("collect",)
await _click_chip(app, pilot, "b Rated TBW")
assert isinstance(app.screen, BaselineScreen)
await pilot.press("escape")
await _click_chip(app, pilot, "s Theme")
assert app.theme == "fenris-amber"
await _click_chip(app, pilot, "d Disclosures")
assert isinstance(app.screen, DisclosuresScreen)
await pilot.press("escape")
await _click_chip(app, pilot, "z Zoom")
assert app._zoomed_panel is not None
await _click_chip(app, pilot, "z Zoom")
assert app._zoomed_panel is None
await _click_chip(app, pilot, "v View")
assert app._activity_view == "day"
await _click_chip(app, pilot, "m Motion")
assert app._reduced_motion is True
@pytest.mark.asyncio
async def test_activity_tool_chips_are_generated_from_bindings(dashboard): # noqa: F811
app = dashboard
async with app.run_test(size=(120, 40)) as pilot:
await pilot.pause()
tools = str(app.query_one("#activity-tools").render())
for action in ("prev_day", "next_day", "go_to_date", "today", "toggle_measure"):
binding = next(b for b in FenrisTuiApp.BINDINGS if b.action == action)
assert "%s %s" % (app.get_key_display(binding), binding.description) in tools
await _click_chip(app, pilot, "w Toggle read/write")
assert app.query_one("#live-activity").measure == "read"
await _click_chip(app, pilot, "g Go to date")
assert isinstance(app.screen, DatePickerScreen)
await pilot.press("escape")
await _click_chip(app, pilot, "Prev day")
assert app._activity_view == "day"
await _click_chip(app, pilot, "t Today")
assert app._activity_view == "live"
@pytest.mark.asyncio
async def test_wheel_moves_live_selection(dashboard): # noqa: F811
app = dashboard
async with app.run_test(size=(100, 36)) as pilot:
await _settle(app, pilot)
graph = app.query_one("#live-activity")
last = len(graph._data) - 1
_scroll(graph, -1)
await pilot.pause()
assert graph._selected_index == last - 1
_scroll(graph, -1)
await pilot.pause()
assert graph._selected_index == last - 2
_scroll(graph, 1)
await pilot.pause()
assert graph._selected_index == last - 1
@pytest.mark.asyncio
async def test_wheel_moves_history_day_and_hour_selection(dashboard): # noqa: F811
app = dashboard
async with app.run_test(size=(100, 36)) as pilot:
graph = await _open_history(app, pilot)
last = len(graph._day_data) - 1
_scroll(graph, -1)
await pilot.pause()
assert graph.selected_index == last - 1
_scroll(graph, 1)
await pilot.pause()
assert graph.selected_index == last
await pilot.press("enter", "end")
end = graph._hourly_selected
_scroll(graph, -1)
await pilot.pause()
assert graph._hourly_selected == end - 1
@pytest.mark.asyncio
async def test_click_to_inspect_stays(dashboard): # noqa: F811
app = dashboard
async with app.run_test(size=(100, 36)) as pilot:
await pilot.pause()
graph = app.query_one("#live-activity")
await pilot.click("#live-render", offset=(graph._point_columns[0], 1))
assert graph._selected_index == 0
# -- modal keys ------------------------------------------------------------
@pytest.mark.asyncio
async def test_confirm_pause_y_n_and_escape(dashboard): # noqa: F811
app = dashboard
async with app.run_test(size=(100, 36)) as pilot:
with patch.object(app, "_run_helper") as helper:
await pilot.press("p", "n")
assert not isinstance(app.screen, ConfirmPause)
await pilot.press("p", "escape")
assert not isinstance(app.screen, ConfirmPause)
helper.assert_not_called()
await pilot.press("p", "y")
assert not isinstance(app.screen, ConfirmPause)
helper.assert_called_once_with("disable", ["--now"])
@pytest.mark.asyncio
async def test_date_picker_keys(dashboard): # noqa: F811
app = dashboard
async with app.run_test(size=(100, 36)) as pilot:
await _settle(app, pilot)
await pilot.press("g", "escape")
assert not isinstance(app.screen, DatePickerScreen)
await pilot.press("g", "enter")
assert isinstance(app.screen, DatePickerScreen)
assert "Please enter a date" in str(app.screen.query_one("#date-error").render())
await pilot.press(*"2026-13-45", "enter")
assert "Invalid date format" in str(app.screen.query_one("#date-error").render())
app.screen.query_one("#date-input").value = ""
await pilot.press(*"2999-01-01", "enter")
assert "future" in str(app.screen.query_one("#date-error").render())
app.screen.query_one("#date-input").value = ""
await pilot.press(*"2026-09-17", "enter")
assert not isinstance(app.screen, DatePickerScreen)
assert app._browse_date == "2026-09-17"
assert app._activity_view == "day"
@pytest.mark.asyncio
async def test_disclosures_and_help_toggle_keys(dashboard): # noqa: F811
app = dashboard
async with app.run_test(size=(100, 36)) as pilot:
await pilot.press("d")
assert isinstance(app.screen, DisclosuresScreen)
await pilot.press("d")
assert not isinstance(app.screen, DisclosuresScreen)
await pilot.press("d", "escape")
assert not isinstance(app.screen, DisclosuresScreen)
await pilot.press("?")
assert isinstance(app.screen, HelpScreen)
await pilot.press("?")
assert not isinstance(app.screen, HelpScreen)
await pilot.press("?", "escape")
assert not isinstance(app.screen, HelpScreen)
@pytest.mark.asyncio
async def test_single_dashboard_has_no_page_navigation_bindings(dashboard): # noqa: F811
keys = {b.key for b in FenrisTuiApp.BINDINGS}
assert not keys & {"left", "right", "1", "2", "3", "4", "home", "end", "pageup", "pagedown"}