From 948e097952f2e4a348389b8db12b63f54547519c Mon Sep 17 00:00:00 2001 From: xavierk Date: Mon, 5 Oct 2026 21:05:44 +0530 Subject: [PATCH] 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. --- CHANGELOG.md | 1 + README.md | 7 + src/fenris/activity_plot.py | 43 +++++- src/fenris/activity_readout.py | 255 +++++++++++++++++++++++++++++-- src/fenris/activity_selection.py | 6 + src/fenris/themes.py | 5 +- src/fenris/tui.py | 90 ++++++++--- tests/test_activity_plot.py | 22 +++ tests/test_activity_readout.py | 190 ++++++++++++++++++++++- tests/test_tui_diagnostics.py | 154 +++++++++++++++++++ 10 files changed, 730 insertions(+), 43 deletions(-) create mode 100644 tests/test_tui_diagnostics.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 75050da..ba33792 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -17,6 +17,7 @@ backfill releases from before this changelog. - 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. ### Changed diff --git a/README.md b/README.md index d1fa1b2..e483374 100644 --- a/README.md +++ b/README.md @@ -275,6 +275,13 @@ UTC-only summaries cannot establish exact local-day totals. `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. - `Tab` / `Shift+Tab` move focus between panels (clicking a panel focuses it); `z` expands the focused panel, and `z` or `Esc` restores it. Monitoring status and controls remain visible. diff --git a/src/fenris/activity_plot.py b/src/fenris/activity_plot.py index f65d443..b82e48e 100644 --- a/src/fenris/activity_plot.py +++ b/src/fenris/activity_plot.py @@ -21,13 +21,19 @@ class VolumePoint: def volume_plot( points: Sequence[VolumePoint], width: int, height: int, selected: int, colors: Mapping[str, str], + secondary: Sequence[VolumePoint] | None = None, ) -> 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) axis_width = 8 columns, rows = width - axis_width, height - 2 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 ( (1e9, "GB") if maximum >= 1e9 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 for p in points] 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. bits = ((1, 8), (2, 16), (4, 32), (64, 128)) - def dot(x: int, y: int) -> None: - cells[y // 4][x // 2] |= bits[y % 4][x % 2] + def dot(x: int, y: int, grid: list[list[int]] = cells) -> None: + 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 markers = {} @@ -53,8 +64,7 @@ def volume_plot( if point.state != "future": markers[x // 2] = "?" continue - y = round((1 - point.volume / ceiling) * (pixel_height - 1)) - y = min(pixel_height - 1, max(0, y)) + 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) @@ -68,6 +78,21 @@ def volume_plot( elif point.volume == 0: 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 result = Text(no_wrap=True, overflow="crop") ticks = {0, rows // 2, rows - 1} @@ -76,8 +101,10 @@ def volume_plot( label = f"{value:6.2f}"[-6:] if row in ticks else " " result.append(label + " │", style=colors["muted"]) for col, value in enumerate(values): - char = chr(0x2800 + value) if value else " " - style = colors["allocated"] + extra = second[row][col] + 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: style = "bold " + colors["selection"] if not value: diff --git a/src/fenris/activity_readout.py b/src/fenris/activity_readout.py index cbfdea4..3acc7b5 100644 --- a/src/fenris/activity_readout.py +++ b/src/fenris/activity_readout.py @@ -33,6 +33,26 @@ from .timefmt import ( LIVE_WINDOW_H = 3 # Rolling window for the live graph +# NVMe SMART critical warning bits (NVMe base specification, Get Log Page 02h). +CRITICAL_WARNING_FLAGS = ( + (0x01, "available spare below threshold"), + (0x02, "temperature outside threshold"), + (0x04, "reliability degraded"), + (0x08, "media in read-only mode"), + (0x10, "volatile memory backup failed"), + (0x20, "persistent memory region read-only"), +) +_KNOWN_WARNING_BITS = sum(bit for bit, _ in CRITICAL_WARNING_FLAGS) + +# Drive counters shown with their change over the selection. +DIAGNOSTIC_COUNTERS = ( + ("power_on_hours", "Power-on hours", ""), + ("power_cycles", "Power cycles", ""), + ("unsafe_shutdowns", "Unsafe shutdowns", ""), + ("media_errors", "Media errors", ""), + ("available_spare", "Available spare", "%"), +) + LOCAL_ACTIVITY_STATE_LABELS = { "so_far": "totals so far", "incomplete": "incomplete", @@ -55,6 +75,7 @@ class ActivityReadout: live: tuple[Dict[str, Any], ...] health: Dict[str, Any] local_day: Optional[Dict[str, Any]] + diagnostics: Optional[Dict[str, Any]] = None def readout( @@ -68,18 +89,24 @@ def readout( hours: List[Dict[str, Any]] = [] if selection.view == "day" and selection.browse_date is not None: hours = _query_local_hourly_graph_data(conn, local_day, now) + days = tuple(_query_local_day_graph_data( + conn, now, zone_name, selected_date=selection.browse_date, + )) + health = _query_drive_health(conn) return ActivityReadout( view=selection.view, timezone_name=zone_name, timezone_label=tz_label(now, zone_name), local_date=local_date, - days=tuple(_query_local_day_graph_data( - conn, now, zone_name, selected_date=selection.browse_date, - )), + days=days, hours=tuple(hours), live=tuple(_query_live_graph_data(conn, now, zone_name)), - health=_query_drive_health(conn), + health=health, local_day=local_day, + diagnostics=( + _query_diagnostics(conn, selection, now, local_day, days, health) + if selection.diagnostics else None + ), ) @@ -234,7 +261,8 @@ def _query_local_hourly_graph_data( rows = conn.execute( "SELECT hour, bytes_written_delta, bytes_read_delta, coverage, " "sample_count, unknown_seconds, active_seconds, idle_seconds, " - "powered_off_seconds, temperature_min, temperature_max " + "powered_off_seconds, temperature_min, temperature_max, " + "temperature_avg " "FROM hour_observations WHERE hour >= ? AND hour <= ? ORDER BY hour", (first_hour.isoformat(), last_hour.isoformat()), ).fetchall() @@ -251,7 +279,7 @@ def _query_local_hourly_graph_data( is_future = hour_start > current identity = hour_start.isoformat() active = idle = off = 0 - temp_min = temp_max = None + temp_min = temp_avg = temp_max = None if row is None: written = read = coverage = sample_count = 0 unknown_seconds = 0 @@ -263,7 +291,7 @@ def _query_local_hourly_graph_data( sample_count = row[4] or 0 unknown_seconds = row[5] or 0 active, idle, off = row[6] or 0, row[7] or 0, row[8] or 0 - temp_min, temp_max = row[9], row[10] + temp_min, temp_max, temp_avg = row[9], row[10], row[11] is_gap = sample_count == 0 and unknown_seconds > 0 result.append({ "hour": identity, @@ -280,6 +308,7 @@ def _query_local_hourly_graph_data( "powered_off_seconds": off, "unknown_seconds": unknown_seconds, "temperature_min": temp_min, + "temperature_avg": temp_avg, "temperature_max": temp_max, "is_zero": written == 0 and read == 0 and not is_gap, "is_gap": is_gap, @@ -306,7 +335,7 @@ def _day_habit(conn: sqlite3.Connection, summary: Dict[str, Any]) -> Optional[Di "SELECT COUNT(*), COALESCE(SUM(active_seconds), 0), " "COALESCE(SUM(idle_seconds), 0), COALESCE(SUM(powered_off_seconds), 0), " "COALESCE(SUM(unknown_seconds), 0), MIN(temperature_min), " - "MAX(temperature_max) FROM hour_observations " + "MAX(temperature_max), AVG(temperature_avg) FROM hour_observations " "WHERE hour >= ? AND hour <= ?", (first_hour.isoformat(), (day_end - timedelta(hours=1)).isoformat()), ).fetchone() @@ -319,6 +348,7 @@ def _day_habit(conn: sqlite3.Connection, summary: Dict[str, Any]) -> Optional[Di "unknown_seconds": row[4], "temperature_min": row[5], "temperature_max": row[6], + "temperature_avg": row[7], } @@ -586,7 +616,7 @@ def _query_drive_health(conn: sqlite3.Connection) -> Dict[str, Any]: cursor = conn.execute( "SELECT mn, sn, fr, temperature_c, available_spare, media_errors, " "power_on_hours, power_cycles, unsafe_shutdowns, capacity_bytes, " - "percentage_used, data_units_written " + "percentage_used, data_units_written, critical_warning " "FROM samples ORDER BY id DESC LIMIT 1" ) row = cursor.fetchone() @@ -602,6 +632,7 @@ def _query_drive_health(conn: sqlite3.Connection) -> Dict[str, Any]: "capacity": "unknown", "percentage_used": 0, "written_tb": 0, + "critical_warning": 0, } capacity = row[9] @@ -619,6 +650,124 @@ def _query_drive_health(conn: sqlite3.Connection) -> Dict[str, Any]: "capacity": capacity_str, "percentage_used": row[10] or 0, "written_tb": written_tb, + "critical_warning": row[12] or 0, + } + + +# --------------------------------------------------------------------------- +# Diagnostics (issue #109) +# --------------------------------------------------------------------------- + +def decode_critical_warning(value: int | None) -> tuple[str, ...]: + """Name each NVMe critical-warning bit that is set; reserved bits are flagged.""" + value = value or 0 + flags = [name for bit, name in CRITICAL_WARNING_FLAGS if value & bit] + if value & ~_KNOWN_WARNING_BITS: + flags.append("unrecognised bits 0x%02x" % (value & ~_KNOWN_WARNING_BITS)) + return tuple(flags) + + +def critical_warning_badge(health: Dict[str, Any]) -> str: + """Outlook badge for a set critical warning; empty when the drive reports none. + + A drive-reported fact only (ADR 0005): it never changes the projection. + """ + value = health.get("critical_warning") or 0 + if not value: + return "" + return "[bold yellow]![/bold yellow] [bold]Drive reports critical warning 0x%02x[/bold] · %s" % ( + value, ", ".join(decode_critical_warning(value)), + ) + + +def _counter_deltas( + conn: sqlite3.Connection, start: datetime | None, end: datetime, +) -> Dict[str, Dict[str, Optional[int]]]: + """Latest counter values and their change across a window of raw samples. + + The baseline is the last sample at or before *start*, else the first sample + inside the window; with fewer than two distinct samples, or across a + controller-segment change, there is no delta. + """ + columns = ", ".join(key for key, _, _ in DIAGNOSTIC_COUNTERS) + columns += ", segment_id" + last = conn.execute( + "SELECT id, %s FROM samples WHERE ts <= ? ORDER BY ts DESC, id DESC " + "LIMIT 1" % columns, (end.isoformat(),), + ).fetchone() + first = None + if start is not None and last is not None: + first = conn.execute( + "SELECT id, %s FROM samples WHERE ts <= ? ORDER BY ts DESC, id DESC " + "LIMIT 1" % columns, (start.isoformat(),), + ).fetchone() or conn.execute( + "SELECT id, %s FROM samples WHERE ts > ? AND ts <= ? " + "ORDER BY ts, id LIMIT 1" % columns, + (start.isoformat(), end.isoformat()), + ).fetchone() + result: Dict[str, Dict[str, Optional[int]]] = {} + for offset, (key, _, _) in enumerate(DIAGNOSTIC_COUNTERS, start=1): + value = last[offset] if last is not None else None + delta = None + if first is not None and first[0] != last[0] and first[-1] == last[-1] and None not in ( + value, first[offset], + ): + delta = value - first[offset] + result[key] = {"value": value, "delta": delta} + return result + + +def _diagnostics_window( + selection: ActivitySelection, + now: datetime, + local_day: Optional[Dict[str, Any]], + days: tuple[Dict[str, Any], ...], +) -> tuple[str, datetime | None, datetime]: + """Label and UTC bounds of what the selection covers.""" + if selection.view == "live": + return "last %dh" % LIVE_WINDOW_H, now - timedelta(hours=LIVE_WINDOW_H), now + if selection.view == "day": + label = "day %s" % (selection.browse_date or "") + if local_day is None: + return label, None, now + end = min(_parse_ts(local_day["utc_end"]), now) + return label, _parse_ts(local_day["utc_start"]), end + start = next(( + _parse_ts(day["identity"].utc_start) + for day in days if day["identity"] is not None + ), None) + return "last %d days" % selection.history_range_days, start, now + + +def _query_diagnostics( + conn: sqlite3.Connection, + selection: ActivitySelection, + now: datetime, + local_day: Optional[Dict[str, Any]], + days: tuple[Dict[str, Any], ...], + health: Dict[str, Any], +) -> Dict[str, Any]: + """Stored diagnostic evidence for the selection: warning, counters, habit.""" + visible = days[-selection.history_range_days:] + label, start, end = _diagnostics_window(selection, now, local_day, visible) + warning = health["critical_warning"] + habit_days: List[Dict[str, Any]] = [] + if selection.view == "history": + for day in visible: + identity = day["identity"] + if identity is None: + continue + summary = query_local_day_summary( + conn, identity.local_date, identity.timezone, now) + habit = _day_habit(conn, summary) if summary is not None else None + if habit is not None: + habit_days.append(dict(habit, day=identity.local_date)) + return { + "window_label": label, + "critical_warning": warning, + "critical_warning_flags": decode_critical_warning(warning), + "counters": _counter_deltas(conn, start, end), + "days": tuple(habit_days), } @@ -792,3 +941,91 @@ def local_day_text(local_date: str, local: Dict[str, Any] | None) -> str: habit["temperature_min"], habit["temperature_max"], )) return "\n".join(lines) + + +def _temperature_triplet( + low: float | None, avg: float | None, high: float | None, +) -> str: + """Temperature as min/avg/max, or a plain note when none was recorded.""" + if low is None or high is None: + return "no temp" + mid = "%d" % round(avg) if avg is not None else "-" + return "%d/%s/%d°C" % (low, mid, high) + + +def _signed(delta: int) -> str: + """Counter change with an explicit sign; zero reads as no change.""" + return "%+d" % delta if delta else "±0" + + +def _hour_diagnostic_line(hour: Dict[str, Any]) -> str: + """One compact row: hour, temperature, state split in minutes, coverage.""" + head = hour.get("local_label", "") + if hour.get("is_future"): + return "%s future" % head + if hour.get("is_gap"): + return "%s gap" % head + minutes = lambda key: (hour.get(key) or 0) // 60 # noqa: E731 + return "%s %s A%dm I%dm Off%dm ?%dm %d%%" % ( + head, + _temperature_triplet( + hour.get("temperature_min"), hour.get("temperature_avg"), + hour.get("temperature_max")), + minutes("active_seconds"), minutes("idle_seconds"), + minutes("powered_off_seconds"), minutes("unknown_seconds"), + hour.get("coverage", 0) * 100, + ) + + +def _day_diagnostic_line(day: Dict[str, Any]) -> str: + """One compact row: day, state split, temperature.""" + return "%s A%s I%s Off%s ?%s %s" % ( + day["day"][5:], + _duration(day["active_seconds"]), _duration(day["idle_seconds"]), + _duration(day["powered_off_seconds"]), _duration(day["unknown_seconds"]), + _temperature_triplet( + day["temperature_min"], day["temperature_avg"], day["temperature_max"]), + ) + + +def _live_temperature_line(live: tuple[Dict[str, Any], ...]) -> str: + """Interval count and interval-end temperature range for the live window.""" + temps = [p["temperature_c"] for p in live if p.get("temperature_c") is not None] + line = "%d intervals" % len(live) + if temps: + line = "%d intervals · %s" % (len(live), _temperature_triplet( + min(temps), sum(temps) / len(temps), max(temps))) + warned = sum(1 for p in live if p.get("critical_warning")) + if warned: + line += " · %d with critical warning" % warned + return line + + +def diagnostics_text(activity: ActivityReadout) -> str: + """Dense Diagnostics panel text for the readout's selection.""" + diag = activity.diagnostics + if diag is None: + return "Diagnostics unavailable" + warning = diag["critical_warning"] + lines = [ + "[bold]Diagnostics[/bold] · %s" % diag["window_label"], + "Critical warning " + ( + "0x%02x · %s" % (warning, ", ".join(diag["critical_warning_flags"])) + if warning else "none"), + ] + for key, label, unit in DIAGNOSTIC_COUNTERS: + counter = diag["counters"][key] + if counter["value"] is None: + lines.append("%s n/a" % label) + continue + delta = "Δ n/a" if counter["delta"] is None else "Δ " + _signed(counter["delta"]) + lines.append("%s %d%s %s" % (label, counter["value"], unit, delta)) + if activity.view == "live": + lines += ["", _live_temperature_line(activity.live)] + elif activity.view == "day": + lines += ["", "Hours · temp min/avg/max · A active I idle Off off ? unknown · coverage"] + lines += [_hour_diagnostic_line(hour) for hour in activity.hours] + else: + lines += ["", "Days · A active I idle Off off ? unknown · temp min/avg/max"] + lines += [_day_diagnostic_line(day) for day in diag["days"]] + return "\n".join(lines) diff --git a/src/fenris/activity_selection.py b/src/fenris/activity_selection.py index 1e468ba..414d1f6 100644 --- a/src/fenris/activity_selection.py +++ b/src/fenris/activity_selection.py @@ -35,6 +35,7 @@ class ActivitySelection: self.selected_history_hour: str | None = None self.history_range_days = HISTORY_RANGE_DEFAULT self.measure = "written" + self.diagnostics = False self.following_live = True self.selected_live_interval: IntervalIdentity | None = None self.live_interval_expired = False @@ -74,6 +75,11 @@ class ActivitySelection: self.measure = "read" if self.measure == "written" else "written" 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: """Resume following the newest interval and clear stale-pin notices.""" self.view = "live" diff --git a/src/fenris/themes.py b/src/fenris/themes.py index 0480712..bf60891 100644 --- a/src/fenris/themes.py +++ b/src/fenris/themes.py @@ -161,14 +161,15 @@ def get_theme(name: str) -> Theme: def get_graph_colors(theme_name: str) -> Dict[str, str]: """Return the graph colour roles for a theme preset. - Returns a dict with keys: allocated, unallocated, gap, zero, partial, - selection and muted text. Falls back to Chalktone for unknown names. + Returns a dict with keys: allocated, unallocated, secondary (the second + series), gap, zero, partial, selection and muted text. Falls back to Chalktone for unknown names. """ theme = get_theme(theme_name) variables = theme.variables or {} return { "allocated": variables.get("graph-allocated", "#d4a017"), "unallocated": variables.get("graph-unallocated", "#8b6914"), + "secondary": variables.get("graph-unallocated", "#8b6914"), "gap": variables.get("graph-gap", "#554422"), "zero": variables.get("graph-zero", "#665533"), "partial": variables.get("graph-partial", "#aa8822"), diff --git a/src/fenris/tui.py b/src/fenris/tui.py index 61ce51a..10ef963 100644 --- a/src/fenris/tui.py +++ b/src/fenris/tui.py @@ -34,7 +34,9 @@ from textual.widgets import Input, Static, Tab, Tabs from .activity_readout import ( LIVE_WINDOW_H, + critical_warning_badge, day_readout_text, + diagnostics_text, hour_readout_text, interval_readout_text, local_day_text, @@ -460,19 +462,19 @@ class HistoryGraph(Widget): ) self.query_one("#bar-range").update(label) - def _render_bars(self) -> None: + def _day_points(self, measure: str) -> list[VolumePoint]: points = [] for index, day in enumerate(self._day_data): if "activity_state" in day: - volume = day.get("total_read") if self.measure == "read" else day.get("total_bytes") + volume = day.get("total_read") if measure == "read" else day.get("total_bytes") unallocated = day.get( - "shared_bytes_read" if self.measure == "read" else "shared_bytes_written", + "shared_bytes_read" if measure == "read" else "shared_bytes_written", 0, ) else: - volume = day.get("total_read", 0) if self.measure == "read" else day.get("total_bytes", 0) + volume = day.get("total_read", 0) if measure == "read" else day.get("total_bytes", 0) unallocated = day.get( - "unallocated_read" if self.measure == "read" else "unallocated_bytes", + "unallocated_read" if measure == "read" else "unallocated_bytes", 0, ) state = "unallocated" if unallocated else "gap" if day.get("is_gap") else ( @@ -480,19 +482,35 @@ class HistoryGraph(Widget): ) points.append(VolumePoint(index, None if state == "gap" else volume, day.get("day", "")[5:], state)) - self._plot(points, self.selected_index) + return points - def _plot(self, points: list[VolumePoint], selected: int) -> None: + def _render_bars(self) -> None: + self._plot( + self._day_points(self.measure), self.selected_index, + self._day_points("written" if self.measure == "read" else "read"), + ) + + def _plot( + self, points: list[VolumePoint], selected: int, + other: list[VolumePoint] | None = None, + ) -> None: + """Draw *points*; with Diagnostics on, *other* is drawn beside them.""" + both = self.selection.diagnostics and other is not None target = self.query_one("#bar-render") colors = get_graph_colors(getattr(self.app, "_current_theme_name", "chalktone")) chart, self._point_columns, unit = volume_plot( points, target.content_size.width or 40, target.content_size.height or 8, - selected, colors, + selected, colors, other if both else None, ) target.update(chart) + names = ("Reads", "Writes") if self.measure == "read" else ("Writes", "Reads") + series = ( + "[%s]%s[/] · [%s]%s[/]" % ( + colors["allocated"], names[0], colors["secondary"], names[1]) + if both else names[0] + ) self.query_one("#bar-legend").update( - ("Reads" if self.measure == "read" else "Writes") - + " (%s) · ? Gap · · Zero · ~ Partial · u Unalloc" % unit + series + " (%s) · ? Gap · · Zero · ~ Partial · u Unalloc" % unit ) def _render_readout(self) -> None: @@ -530,15 +548,21 @@ class HistoryGraph(Widget): self._show_constrained_hourly_summary() return - points = [] - for index, hour in enumerate(self._hour_data): - state = "future" if hour.get("is_future") else ( - "gap" if hour.get("is_gap") else "partial" if hour.get("is_partial") else "measured" - ) - volume = hour.get("bytes_read" if self.measure == "read" else "bytes_written", 0) - points.append(VolumePoint(index, None if state in ("future", "gap") else volume, - hour.get("local_label", ""), state)) - self._plot(points, self._hourly_selected) + def hour_points(measure: str) -> list[VolumePoint]: + points = [] + for index, hour in enumerate(self._hour_data): + state = "future" if hour.get("is_future") else ( + "gap" if hour.get("is_gap") else "partial" if hour.get("is_partial") else "measured" + ) + volume = hour.get("bytes_read" if measure == "read" else "bytes_written", 0) + points.append(VolumePoint(index, None if state in ("future", "gap") else volume, + hour.get("local_label", ""), state)) + return points + + self._plot( + hour_points(self.measure), self._hourly_selected, + hour_points("written" if self.measure == "read" else "read"), + ) self._render_hourly_readout() @@ -1241,6 +1265,7 @@ class FenrisTuiApp(App): border-title-color: $accent; } #activity-panel { row-span: 2; height: 100%; } + #diagnostics-panel { display: none; } #headline-band, #drive-health { height: auto; } #activity-tabs { height: 1; margin: 0; } #activity-tabs Underline { display: none; } @@ -1267,6 +1292,7 @@ class FenrisTuiApp(App): height: auto; min-height: 0; } #main-grid.constrained #endurance-panel { max-height: 16; } + #main-grid.constrained #diagnostics-panel { max-height: 20; } #main-grid.constrained.zoomed .zoom-target { height: 24; } #confirm-text, #disc-text, #help-text { padding: 1 2; } """ @@ -1277,6 +1303,7 @@ class FenrisTuiApp(App): Binding("c", "collect", "Collect"), Binding("b", "baseline", "Rated TBW"), Binding("d", "disclose", "Disclosures"), + Binding("i", "toggle_diagnostics", "Diagnostics"), Binding("z", "toggle_zoom", "Zoom"), Binding("escape", "exit_zoom", "Restore zoom", show=False), Binding("v", "next_view", "View"), @@ -1324,6 +1351,7 @@ class FenrisTuiApp(App): self._is_constrained_mode = False self._zoomed_panel: Optional[str] = None + self._outlook_badge = "" @property def _browse_date(self) -> Optional[str]: @@ -1367,6 +1395,8 @@ class FenrisTuiApp(App): ) with VerticalScroll(id="drive-panel", classes="dashboard-pane"): yield Static("", id="drive-health") + with VerticalScroll(id="diagnostics-panel", classes="dashboard-pane"): + yield Static("", id="diagnostics") with VerticalScroll(id="status-area"): yield Static("", id="service-strip") yield Static("", id="action-rail") @@ -1376,6 +1406,7 @@ class FenrisTuiApp(App): self.query_one("#endurance-panel").border_title = "Endurance outlook" self.query_one("#activity-panel").border_title = "Drive activity" self.query_one("#drive-panel").border_title = "Drive" + self.query_one("#diagnostics-panel").border_title = "Diagnostics" self.query_one("#live-activity").border_title = "Live activity" # Tab visits panels only: scroll hosts and the view tabs stay out of the order. for selector in ("#dashboard-scroll", "#status-area", "#activity-tabs"): @@ -1494,6 +1525,8 @@ class FenrisTuiApp(App): self._refresh() def _render_headline(self, body: str = "") -> None: + if self._outlook_badge: + body = self._outlook_badge + "\n" + body self.query_one("#identity").update(_IDENTITY_FULL if _can_render_wolf(self.size.width) else _IDENTITY_FALLBACK) self.query_one("#auth-notice").display = self._show_auth_notice self.query_one("#headline-band").update(body) @@ -1535,6 +1568,8 @@ class FenrisTuiApp(App): def _render_empty_or_fault(self, comp: StatusComposition) -> None: """Render empty history or the shared store-fault classification.""" + self._outlook_badge = "" + self.query_one("#diagnostics").update("Diagnostics unavailable: no observations to read") self._show_baseline_banner(not (comp.newer_schema or comp.store_fault)) if comp.newer_schema or comp.store_fault: message = comp.explanation[:1].upper() + comp.explanation[1:] @@ -1571,6 +1606,8 @@ class FenrisTuiApp(App): def _render_all_regions(self, conn: sqlite3.Connection, comp: StatusComposition) -> None: """Render all four regions from live store data.""" + activity = readout(conn, self._activity_selection, self._clock_now) + self._outlook_badge = critical_warning_badge(activity.health) # --- Headline band (§7.2) --- if comp.sample_count == 0 and comp.day_count == 0: self._show_baseline_banner( @@ -1596,7 +1633,6 @@ class FenrisTuiApp(App): self._render_headline("[bold]No projection available[/bold]") # --- Activity panel: live and historical volume plots --- - activity = readout(conn, self._activity_selection, self._clock_now) graph = self.query_one("#usage-history") graph.set_data( list(activity.days), on_drill=self._on_graph_drill, @@ -1627,6 +1663,7 @@ class FenrisTuiApp(App): health["written_tb"], ) self.query_one("#drive-health").update(health_text) + self.query_one("#diagnostics").update(diagnostics_text(activity)) # --- Live activity graph (issue #91) --- self.query_one("#live-activity").set_data(list(activity.live)) @@ -1694,6 +1731,7 @@ class FenrisTuiApp(App): activity = readout(conn, self._activity_selection, self._clock_now) graph.set_hour_data(list(activity.hours)) self._render_local_day(activity.local_date, activity.local_day) + self.query_one("#diagnostics").update(diagnostics_text(activity)) def _format_headline(self, proj: ProjectionResult) -> str: """Format the lifespan headline (spec §6.11).""" @@ -1884,6 +1922,18 @@ class FenrisTuiApp(App): """Show disclosures (spec §6.11, CI-4).""" self.push_screen(DisclosuresScreen()) + def action_toggle_diagnostics(self) -> None: + """Swap the Drive pane for the Diagnostics panel and back.""" + self.action_exit_zoom() + showing = self._activity_selection.toggle_diagnostics() + self.query_one("#diagnostics-panel").display = showing + self.query_one("#drive-panel").display = not showing + hidden = self.query_one("#drive-panel" if showing else "#diagnostics-panel") + if self.focused is None or hidden in self.focused.ancestors_with_self: + self.query_one("#live-activity" if self._activity_view == "live" else "#usage-history").focus( + scroll_visible=not self._is_constrained_mode) + self._refresh() + def action_help(self) -> None: """Show keyboard controls and sudo guidance.""" self.push_screen(HelpScreen()) diff --git a/tests/test_activity_plot.py b/tests/test_activity_plot.py index 673cb8b..d31af93 100644 --- a/tests/test_activity_plot.py +++ b/tests/test_activity_plot.py @@ -50,3 +50,25 @@ def test_many_intervals_and_selection_fit_small_plot_without_losing_points(): assert len(columns) == len(points) assert text.plain.splitlines()[-2][columns[100]] == "▼" 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]]) diff --git a/tests/test_activity_readout.py b/tests/test_activity_readout.py index 588e2a7..3f605b2 100644 --- a/tests/test_activity_readout.py +++ b/tests/test_activity_readout.py @@ -13,6 +13,9 @@ 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, @@ -76,13 +79,13 @@ def _kolkata_day(conn, local_date, **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): + 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, sample_count, " - "coverage) VALUES (?,?,?,?,?,?,?,?,?,?,?)", - (hour, active, idle, off, unknown, written, read, tmin, tmax, + "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() @@ -300,3 +303,182 @@ class TestTimeFormatting: 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 diff --git a/tests/test_tui_diagnostics.py b/tests/test_tui_diagnostics.py new file mode 100644 index 0000000..4ab092f --- /dev/null +++ b/tests/test_tui_diagnostics.py @@ -0,0 +1,154 @@ +"""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.activity_plot import VolumePoint +from fenris.tui import FenrisTuiApp, HelpScreen, HistoryGraph + + +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) + graph = app.query_one(HistoryGraph) + 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")