From 5d916ee97f8a6bf4a4460aafe1f7e2ae3c9d5bcc Mon Sep 17 00:00:00 2001 From: xavierk Date: Mon, 14 Sep 2026 10:47:37 +0530 Subject: [PATCH] feat: add interactive daily writes bar graph with hourly drill-down (issue #75) Replace the static sparkline with an interactive block-glyph bar graph that supports writes-only daily bars, range switching (7/14/28/90 days), day selection, and hourly drill-down. No plotting dependency. Co-authored-by: CommandCodeBot --- src/fenris/tui.py | 642 +++++++++++++++++++++++++++++++++++++++++++++- tests/test_tui.py | 407 +++++++++++++++++++++++++++++ 2 files changed, 1035 insertions(+), 14 deletions(-) diff --git a/src/fenris/tui.py b/src/fenris/tui.py index b12a863..bc1ba1f 100644 --- a/src/fenris/tui.py +++ b/src/fenris/tui.py @@ -19,12 +19,13 @@ import subprocess import sys from datetime import datetime, timedelta, timezone from pathlib import Path -from typing import Any, Dict, List, Optional +from typing import Any, Callable, Dict, List, Optional from textual.app import App, ComposeResult from textual.binding import Binding from textual.containers import Container, Horizontal, VerticalScroll from textual.screen import ModalScreen +from textual.widget import Widget from textual.widgets import Static from .projection import ( @@ -112,6 +113,601 @@ def _habit_bar(a: float, i: float, o: float, u: float, width: int = 40) -> str: return bar + "\n" + legend +# --------------------------------------------------------------------------- +# Bar graph glyphs and constants +# --------------------------------------------------------------------------- + +_GLYPH_ALLOCATED = "\u2588" # \u2588 full block +_GLYPH_UNALLOCATED = "\u2593" # \u2593 dark shade +_GLYPH_GAP = "\u2591" # \u2591 light shade +_GLYPH_ZERO = "\u00b7" # \u00b7 middle dot +_GLYPH_PARTIAL = "\u258c" # \u258c left half block + +_RANGE_OPTIONS = (7, 14, 28, 90) +_RANGE_DEFAULT = 14 +_BAR_HEIGHT = 6 +_BAR_WIDTH = 2 +_BAR_SPACING = 1 + + +# --------------------------------------------------------------------------- +# Interactive daily bar graph widget +# --------------------------------------------------------------------------- + +class DailyBarGraph(Widget): + """Interactive daily writes bar graph with hourly drill-down. + + Renders writes-only daily bars using block glyphs, supports keyboard + and mouse navigation, range switching (7/14/28/90 days), day selection, + and hourly drill-down. No plotting dependency (issue #75 AC5). + """ + + can_focus = True + can_focus_children = False + + CSS = """ + DailyBarGraph { + height: 100%; + width: 100%; + layout: vertical; + } + #bar-range { + height: 1; + width: 100%; + } + #bar-render { + height: 1fr; + width: 100%; + overflow: hidden; + } + #bar-legend { + height: 1; + width: 100%; + } + #bar-readout { + height: 3; + width: 100%; + } + """ + + def __init__(self, **kwargs: Any) -> None: + super().__init__(**kwargs) + self.range_days: int = _RANGE_DEFAULT + self.selected_index: int = -1 + self.view_mode: str = "daily" + self.drill_day: Optional[str] = None + self._all_day_data: List[Dict[str, Any]] = [] + self._day_data: List[Dict[str, Any]] = [] + self._hour_data: List[Dict[str, Any]] = [] + self._max_bytes: int = 0 + self._hourly_selected: int = -1 + self._on_drill: Optional[Callable[[str], None]] = None + + def compose(self) -> ComposeResult: + yield Static("", id="bar-range") + yield Static("", id="bar-render") + yield Static("", id="bar-legend") + yield Static("", id="bar-readout") + + def set_data( + self, + day_data: List[Dict[str, Any]], + on_drill: Any = None, + ) -> None: + """Update graph with day data. on_drill(day) called on drill entry.""" + self._all_day_data = day_data + self._on_drill = on_drill + self._trim_to_range() + self.view_mode = "daily" + self.drill_day = None + self._hour_data = [] + self._hourly_selected = -1 + self._refresh() + + def set_hour_data(self, hour_data: List[Dict[str, Any]]) -> None: + """Set hourly data for drill-down view.""" + self._hour_data = hour_data + self._hourly_selected = -1 + self._refresh_hourly() + + def _trim_to_range(self) -> None: + self._day_data = ( + self._all_day_data[-self.range_days:] + if len(self._all_day_data) > self.range_days + else list(self._all_day_data) + ) + # Bar height scales to allocated bytes only; unallocated are shown + # separately so cross-day unknowns never inflate a day bar (AC3). + self._max_bytes = max( + (d.get("allocated_bytes", 0) for d in self._day_data), default=0 + ) + + def _refresh(self) -> None: + if not self._day_data: + self._show_empty() + return + if self._is_constrained(): + self._show_constrained_summary() + return + self._render_range() + self._render_bars() + self._render_legend() + self._render_readout() + + def _show_empty(self) -> None: + self.query_one("#bar-range").update("[dim]usage history[/dim]") + self.query_one("#bar-render").update("[dim]awaiting first sample[/dim]") + self.query_one("#bar-legend").update("") + self.query_one("#bar-readout").update("") + + def _is_constrained(self) -> bool: + """Return True when the widget is too narrow for the bar graph. + + At 80 columns with a 3fr:2fr grid split the pane content width is ~46 + (after round border + padding). Below 32 the bars become unreadable. + """ + try: + w = self.region.width + return 0 < w < 32 + except Exception: + return False + + def _show_constrained_summary(self) -> None: + """Textual fallback for terminals below 80×24.""" + self.query_one("#bar-range").update("[dim]usage history[/dim]") + if not self._day_data: + self.query_one("#bar-render").update("[dim]graph needs ≥80×24[/dim]") + self.query_one("#bar-legend").update("") + self.query_one("#bar-readout").update("") + return + + # Summarise visible days as text + total_bytes = sum(d.get("total_bytes", 0) for d in self._day_data) + days_with_data = sum(1 for d in self._day_data if d.get("total_bytes", 0) > 0) + n = len(self._day_data) + first = self._day_data[0].get("local_label", "") + last = self._day_data[-1].get("local_label", "") + self.query_one("#bar-render").update( + "[dim]graph needs ≥80×24[/dim]\n" + " %d days · %d with writes · %.3f GB total\n" + " %s → %s" % (n, days_with_data, total_bytes / 1e9, first, last) + ) + self.query_one("#bar-legend").update("") + + # Preserve selection readout + if 0 <= self.selected_index < len(self._day_data): + day = self._day_data[self.selected_index] + self.query_one("#bar-readout").update( + "[bold]%s[/bold] · %.3f GB" + % (day.get("local_label", ""), day.get("total_bytes", 0) / 1e9) + ) + else: + self.query_one("#bar-readout").update("[dim]\u2190 \u2192 select[/dim]") + + def _show_constrained_hourly_summary(self) -> None: + """Textual fallback for hourly view when terminal is too small.""" + total_bytes = sum(h.get("bytes_written", 0) for h in self._hour_data) + hours_with_data = sum( + 1 for h in self._hour_data if h.get("bytes_written", 0) > 0 + ) + n = len(self._hour_data) + self.query_one("#bar-render").update( + "[dim]graph needs ≥80×24[/dim]\n" + " %d hours · %d with writes · %.3f GB total" % (n, hours_with_data, total_bytes / 1e9) + ) + self.query_one("#bar-legend").update("") + + if 0 <= self._hourly_selected < len(self._hour_data): + h = self._hour_data[self._hourly_selected] + self.query_one("#bar-readout").update( + "[bold]%s[/bold] \u00b7 %.3f GB" + % (h.get("local_label", ""), h.get("bytes_written", 0) / 1e9) + ) + else: + self.query_one("#bar-readout").update("[dim]\u2190 \u2192 select hour[/dim]") + + def on_resize(self) -> None: + """Re-render when terminal size changes.""" + if self.view_mode == "daily": + self._refresh() + else: + self._refresh_hourly() + + def _render_range(self) -> None: + parts = [] + for r in _RANGE_OPTIONS: + if r == self.range_days: + parts.append("[bold]%d[/bold]" % r) + else: + parts.append(str(r)) + label = "range: " + " / ".join(parts) + if self.view_mode == "hourly": + label += " \u00b7 [bold]%s[/bold] \u00b7 esc back" % (self.drill_day or "") + self.query_one("#bar-range").update(label) + + def _render_bars(self) -> None: + visible = self._day_data + if not visible: + self.query_one("#bar-render").update("") + return + + max_bytes = self._max_bytes or 1 + bar_h = _BAR_HEIGHT + n = len(visible) + + lines: List[str] = [] + for row in range(bar_h, 0, -1): + line = "" + threshold = (row / bar_h) * max_bytes + for i, day in enumerate(visible): + allocated = day.get("allocated_bytes", 0) + is_zero = day.get("is_zero", False) + is_gap = day.get("is_gap", False) + is_partial = day.get("is_partial", False) + + if is_zero and row == 1: + glyph = _GLYPH_ZERO + elif is_gap: + glyph = _GLYPH_GAP if row <= 2 else " " + elif allocated == 0: + glyph = " " + elif row == 1 and i == self.selected_index: + glyph = "\u25b6" # selection arrow + elif threshold > 0 and threshold <= allocated: + glyph = _GLYPH_ALLOCATED + elif is_partial and row == bar_h: + glyph = _GLYPH_PARTIAL + else: + glyph = " " + + line += glyph * _BAR_WIDTH + if i < n - 1: + line += " " * _BAR_SPACING + + lines.append(line) + + # Unallocated indicator row: show ▓ for days with cross-day unknowns + unalloc_line = "" + has_unalloc = False + for i, day in enumerate(visible): + unalloc = day.get("unallocated_bytes", 0) + if unalloc > 0: + unalloc_line += _GLYPH_UNALLOCATED * _BAR_WIDTH + has_unalloc = True + else: + unalloc_line += " " * _BAR_WIDTH + if i < n - 1: + unalloc_line += " " * _BAR_SPACING + if has_unalloc: + lines.append(unalloc_line) + + # Date labels + label_line = "" + for i, day in enumerate(visible): + label = day.get("local_label", day.get("day", ""))[-2:] + label_line += label + if i < n - 1: + label_line += " " * _BAR_SPACING + lines.append(label_line) + + self.query_one("#bar-render").update("\n".join(lines)) + + def _render_legend(self) -> None: + legend = ( + "%s alloc %s unalloc %s gap %s zero %s partial" + % ( + _GLYPH_ALLOCATED, + _GLYPH_UNALLOCATED, + _GLYPH_GAP, + _GLYPH_ZERO, + _GLYPH_PARTIAL, + ) + ) + self.query_one("#bar-legend").update(legend) + + def _render_readout(self) -> None: + if self.selected_index < 0 or self.selected_index >= len(self._day_data): + self.query_one("#bar-readout").update( + "[dim]\u2190 \u2192 select \u00b7 1-4 range \u00b7 Enter drill[/dim]" + ) + return + + day = self._day_data[self.selected_index] + total = day.get("total_bytes", 0) + allocated = day.get("allocated_bytes", 0) + unallocated = day.get("unallocated_bytes", 0) + coverage = day.get("coverage", 0) + hours = day.get("evidenced_hours", 0) + + total_gb = total / 1e9 + parts = [ + "[bold]%s[/bold] \u00b7 %.3f GB total \u00b7 %d hours \u00b7 %.0f%% coverage" + % ( + day.get("local_label", day.get("day", "")), + total_gb, + hours, + coverage * 100, + ), + ] + if unallocated > 0: + parts.append( + " allocated %.3f GB \u00b7 unallocated %.3f GB" + % (allocated / 1e9, unallocated / 1e9) + ) + self.query_one("#bar-readout").update("\n".join(parts)) + + # -- Hourly drill-down rendering -- + + def _refresh_hourly(self) -> None: + if not self._hour_data: + self.query_one("#bar-render").update("[dim]no hourly data[/dim]") + return + + self._render_range() + + if self._is_constrained(): + self._show_constrained_hourly_summary() + return + + max_bytes = max( + (h.get("bytes_written", 0) for h in self._hour_data), default=0 + ) or 1 + bar_h = _BAR_HEIGHT + n = len(self._hour_data) + + lines: List[str] = [] + for row in range(bar_h, 0, -1): + line = "" + threshold = (row / bar_h) * max_bytes + for i, hour in enumerate(self._hour_data): + bw = hour.get("bytes_written", 0) + is_zero = hour.get("is_zero", False) + + if is_zero and row == 1: + glyph = _GLYPH_ZERO + elif bw == 0: + glyph = " " + elif row == 1 and i == self._hourly_selected: + glyph = "\u25b6" + elif threshold > 0 and threshold <= bw: + glyph = _GLYPH_ALLOCATED + else: + glyph = " " + + line += glyph * _BAR_WIDTH + if i < n - 1: + line += " " * _BAR_SPACING + + lines.append(line) + + # Hour labels + label_line = "" + for i, hour in enumerate(self._hour_data): + label = hour.get("local_label", hour.get("hour", ""))[-2:] + label_line += label + if i < n - 1: + label_line += " " * _BAR_SPACING + lines.append(label_line) + + self.query_one("#bar-render").update("\n".join(lines)) + self.query_one("#bar-legend").update( + "%s writes \u00b7 %s zero" % (_GLYPH_ALLOCATED, _GLYPH_ZERO) + ) + + # Hourly readout + if 0 <= self._hourly_selected < len(self._hour_data): + h = self._hour_data[self._hourly_selected] + self.query_one("#bar-readout").update( + "[bold]%s[/bold] \u00b7 %.3f GB \u00b7 %d%% coverage" + % ( + h.get("local_label", h.get("hour", "")), + h.get("bytes_written", 0) / 1e9, + h.get("coverage", 0) * 100, + ) + ) + else: + self.query_one("#bar-readout").update("[dim]\u2190 \u2192 select hour[/dim]") + + # -- Event handling -- + + def on_key(self, event: Any) -> None: + if self.view_mode == "daily": + self._handle_daily_key(event) + else: + self._handle_hourly_key(event) + + def _handle_daily_key(self, event: Any) -> None: + if event.key == "left": + if self.selected_index < 0: + self.selected_index = len(self._day_data) - 1 + else: + self.selected_index = max(0, self.selected_index - 1) + self._refresh() + event.stop() + elif event.key == "right": + if self.selected_index < 0: + self.selected_index = 0 + else: + self.selected_index = min( + len(self._day_data) - 1, self.selected_index + 1 + ) + self._refresh() + event.stop() + elif event.key == "enter" and self.selected_index >= 0: + self._enter_drill() + event.stop() + elif event.key in ("1", "2", "3", "4"): + self.range_days = _RANGE_OPTIONS[int(event.key) - 1] + self.selected_index = -1 + self._trim_to_range() + self._refresh() + event.stop() + + def _enter_drill(self) -> None: + if self.selected_index < 0 or self.selected_index >= len(self._day_data): + return + day = self._day_data[self.selected_index] + self.drill_day = day.get("day") + self.view_mode = "hourly" + self._hourly_selected = -1 + if self._on_drill: + self._on_drill(self.drill_day) + self._render_range() + self.query_one("#bar-render").update("[dim]loading...[/dim]") + self.query_one("#bar-legend").update("") + self.query_one("#bar-readout").update("") + + def _exit_drill(self) -> None: + self.view_mode = "daily" + self.drill_day = None + self._hour_data = [] + self._hourly_selected = -1 + self._refresh() + + def _handle_hourly_key(self, event: Any) -> None: + if event.key in ("escape", "backspace"): + self._exit_drill() + event.stop() + elif event.key == "left": + if self._hourly_selected < 0: + self._hourly_selected = len(self._hour_data) - 1 + else: + self._hourly_selected = max(0, self._hourly_selected - 1) + self._refresh_hourly() + event.stop() + elif event.key == "right": + if self._hourly_selected < 0: + self._hourly_selected = 0 + else: + self._hourly_selected = min( + len(self._hour_data) - 1, self._hourly_selected + 1 + ) + self._refresh_hourly() + event.stop() + + def on_click(self, event: Any) -> None: + render = self.query_one("#bar-render") + offset_x = event.x - render.region.x + bar_total = _BAR_WIDTH + _BAR_SPACING + idx = offset_x // bar_total + + if self.view_mode == "daily": + if 0 <= idx < len(self._day_data): + self.selected_index = idx + self._refresh() + else: + if 0 <= idx < len(self._hour_data): + self._hourly_selected = idx + self._refresh_hourly() + + +# --------------------------------------------------------------------------- +# Graph data queries +# --------------------------------------------------------------------------- + +def _query_daily_graph_data( + conn: sqlite3.Connection, +) -> List[Dict[str, Any]]: + """Query day aggregates for the bar graph. + + Returns one dict per day with total/allocated/unallocated bytes, + coverage, evidence hours, and classification flags. + """ + cursor = conn.execute( + "SELECT day, bytes_written_delta, unattributed_bytes_written, " + "coverage, sample_count, active_seconds, idle_seconds, " + "powered_off_seconds, unknown_seconds " + "FROM day_aggregates ORDER BY day" + ) + rows = cursor.fetchall() + + result: List[Dict[str, Any]] = [] + for row in rows: + day = row[0] + bw_delta = row[1] or 0 + unattributed = row[2] or 0 + coverage = row[3] or 0.0 + sample_count = row[4] or 0 + active = row[5] or 0 + idle = row[6] or 0 + powered_off = row[7] or 0 + unknown = row[8] or 0 + + total_bytes = bw_delta + unattributed + evidenced_hours = (active + idle + powered_off) // 3600 + + is_zero = total_bytes == 0 + is_gap = ( + sample_count == 0 + and (active + idle + powered_off) == 0 + and unknown > 0 + ) + is_partial = coverage < 0.5 + + result.append({ + "day": day, + "local_label": day, + "total_bytes": total_bytes, + "allocated_bytes": bw_delta, + "unallocated_bytes": unattributed, + "coverage": coverage, + "evidenced_hours": evidenced_hours, + "sample_count": sample_count, + "is_zero": is_zero, + "is_gap": is_gap, + "is_partial": is_partial, + }) + + return result + + +def _query_hourly_graph_data( + conn: sqlite3.Connection, + day: str, +) -> List[Dict[str, Any]]: + """Query hour observations for a specific day. + + Returns one dict per hour with bytes written, coverage, and flags. + """ + cursor = conn.execute( + "SELECT hour, bytes_written_delta, coverage, sample_count, " + "active_seconds, idle_seconds, powered_off_seconds, unknown_seconds " + "FROM hour_observations " + "WHERE hour LIKE ? ORDER BY hour", + (day + "T%",), + ) + rows = cursor.fetchall() + + result: List[Dict[str, Any]] = [] + for row in rows: + hour = row[0] + bw = row[1] or 0 + coverage = row[2] or 0.0 + sample_count = row[3] or 0 + active = row[4] or 0 + idle = row[5] or 0 + powered_off = row[6] or 0 + unknown = row[7] or 0 + + is_zero = bw == 0 + local_label = hour[11:13] if len(hour) >= 13 else hour + + result.append({ + "hour": hour, + "local_label": local_label, + "bytes_written": bw, + "coverage": coverage, + "sample_count": sample_count, + "active_seconds": active, + "idle_seconds": idle, + "powered_off_seconds": powered_off, + "unknown_seconds": unknown, + "is_zero": is_zero, + }) + + return result + + # --------------------------------------------------------------------------- # Data queries for TUI regions # --------------------------------------------------------------------------- @@ -356,7 +952,7 @@ class FenrisTuiApp(App): with Container(id="main-grid"): yield Static("", id="headline-band", classes="pane") yield Static("", id="paused-banner") - yield Static("", id="usage-history", classes="pane") + yield DailyBarGraph(id="usage-history", classes="pane") yield Static("", id="drive-health", classes="pane") yield Static("", id="service-strip", classes="pane") yield Static("q QUIT TUI", id="quit-rail") @@ -423,7 +1019,7 @@ class FenrisTuiApp(App): "[bold]No observations yet[/bold]\n\n" "Enable monitoring: fenris monitor resume" ) - self.query_one("#usage-history").update("") + self.query_one("#usage-history").set_data([]) self.query_one("#drive-health").update("") self.query_one("#service-strip").update( "boot: disabled · timer: inactive · last collect: unknown · freshness: empty · " @@ -438,7 +1034,7 @@ class FenrisTuiApp(App): "[bold red]Observation store unreadable[/bold red]\n" "Check journalctl -u fenris-collect.service" ) - self.query_one("#usage-history").update("") + self.query_one("#usage-history").set_data([]) self.query_one("#drive-health").update("") self.query_one("#service-strip").update( "[dim]by Bongbetic[/dim]\n" @@ -474,16 +1070,20 @@ class FenrisTuiApp(App): except Exception: self._render_headline("[bold]No projection available[/bold]") - # --- Usage-history pane (§7.2 left) --- - history = _query_usage_history(conn) - history_text = "[bold]Usage history[/bold] · %d days · %.1f–%.1f GB/day\n %s\n %s" % ( - history["num_days"], - history["min_gb"], - history["max_gb"], - history["sparkline"], - history["habit_bar"], - ) - self.query_one("#usage-history").update(history_text) + # --- Usage-history pane (§7.2 left): interactive bar graph --- + graph = self.query_one("#usage-history") + day_data = _query_daily_graph_data(conn) + + # Preserve drill-down state across refresh if still valid + if graph.view_mode == "hourly" and graph.drill_day: + saved_drill_day = graph.drill_day + hour_data = _query_hourly_graph_data(conn, saved_drill_day) + graph.set_data(day_data, on_drill=self._on_graph_drill) + graph.view_mode = "hourly" + graph.drill_day = saved_drill_day + graph.set_hour_data(hour_data) + else: + graph.set_data(day_data, on_drill=self._on_graph_drill) # --- Drive-health pane (§7.2 right) --- health = _query_drive_health(conn) @@ -557,6 +1157,20 @@ class FenrisTuiApp(App): main_grid.remove_class("paused") main_grid.refresh(layout=True) + def _on_graph_drill(self, day: str) -> None: + """Load hourly data when the graph enters drill-down mode.""" + graph = self.query_one("#usage-history") + try: + conn = self._open_store() + if conn is not None: + try: + hour_data = _query_hourly_graph_data(conn, day) + graph.set_hour_data(hour_data) + finally: + conn.close() + except Exception: + graph.set_hour_data([]) + def _format_headline(self, proj: ProjectionResult) -> str: """Format the lifespan headline (spec §6.11).""" if proj.headline_remaining_seconds is None: diff --git a/tests/test_tui.py b/tests/test_tui.py index a326684..cc56fba 100644 --- a/tests/test_tui.py +++ b/tests/test_tui.py @@ -40,12 +40,23 @@ from fenris.status import ( ) from fenris.tui import ( FenrisTuiApp, + DailyBarGraph, _format_remaining, _sparkline, _habit_bar, _query_usage_history, _query_drive_health, _query_service_facts, + _query_daily_graph_data, + _query_hourly_graph_data, + _RANGE_OPTIONS, + _RANGE_DEFAULT, + _BAR_HEIGHT, + _GLYPH_ALLOCATED, + _GLYPH_UNALLOCATED, + _GLYPH_GAP, + _GLYPH_ZERO, + _GLYPH_PARTIAL, ) @@ -679,3 +690,399 @@ class TestStateMatrixCombinations: # Incomplete provenance → UNVERIFIED tier assert proj.baseline_tier.value == "unverified_override" conn.close() + + +# --------------------------------------------------------------------------- +# Hour observation helper for graph tests +# --------------------------------------------------------------------------- + +def _insert_hour(conn, hour, bw=0, active=3600, idle=0, powered_off=0, + unknown=0, coverage=1.0, samples=1): + """Insert an hour observation row.""" + 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 (?, ?, ?, ?, ?, ?, 0, ?, ?)", + (hour, active, idle, powered_off, unknown, bw, samples, coverage), + ) + conn.commit() + + +def _insert_day_with_unattributed(conn, day, bw=0, unattributed=0, + coverage=0.95, samples=24): + """Insert a day aggregate with explicit unattributed bytes.""" + conn.execute( + "INSERT INTO day_aggregates " + "(day, active_seconds, idle_seconds, powered_off_seconds, unknown_seconds, " + " bytes_written_delta, bytes_read_delta, sample_count, coverage, " + " unattributed_bytes_written, unattributed_bytes_read) " + "VALUES (?, 3600, 0, 0, 0, ?, 0, ?, ?, ?, 0)", + (day, bw, samples, coverage, unattributed), + ) + conn.commit() + + +# --------------------------------------------------------------------------- +# Graph data query tests (issue #75) +# --------------------------------------------------------------------------- + +class TestQueryDailyGraphData: + def test_empty_store(self, tmp_path): + conn = init_store(tmp_path / "test.db") + result = _query_daily_graph_data(conn) + assert result == [] + conn.close() + + def test_with_days(self, tmp_path): + conn = init_store(tmp_path / "test.db") + _insert_segment(conn) + _open_period(conn) + for i in range(14): + d = (datetime(2026, 9, 15) + timedelta(days=i)).strftime("%Y-%m-%d") + _insert_day(conn, d, bw=1024*1024*100) + result = _query_daily_graph_data(conn) + assert len(result) == 14 + assert result[0]["total_bytes"] == 1024*1024*100 + assert result[0]["allocated_bytes"] == 1024*1024*100 + assert result[0]["unallocated_bytes"] == 0 + assert result[0]["is_zero"] is False + conn.close() + + def test_zero_day(self, tmp_path): + conn = init_store(tmp_path / "test.db") + _insert_segment(conn) + _open_period(conn) + _insert_day(conn, "2026-09-20", bw=0, coverage=0.95, samples=24) + result = _query_daily_graph_data(conn) + assert len(result) == 1 + assert result[0]["is_zero"] is True + assert result[0]["total_bytes"] == 0 + conn.close() + + def test_unattributed_bytes(self, tmp_path): + conn = init_store(tmp_path / "test.db") + _insert_segment(conn) + _open_period(conn) + _insert_day_with_unattributed( + conn, "2026-09-20", bw=500, unattributed=300 + ) + result = _query_daily_graph_data(conn) + assert len(result) == 1 + assert result[0]["allocated_bytes"] == 500 + assert result[0]["unallocated_bytes"] == 300 + assert result[0]["total_bytes"] == 800 + conn.close() + + def test_gap_day(self, tmp_path): + conn = init_store(tmp_path / "test.db") + _insert_segment(conn) + _open_period(conn) + # Day with no hours but unknown seconds (gap in monitoring period) + conn.execute( + "INSERT INTO day_aggregates " + "(day, active_seconds, idle_seconds, powered_off_seconds, unknown_seconds, " + " bytes_written_delta, sample_count, coverage) " + "VALUES (?, 0, 0, 0, 86400, 0, 0, 0.0)", + ("2026-09-20",), + ) + conn.commit() + result = _query_daily_graph_data(conn) + assert len(result) == 1 + assert result[0]["is_gap"] is True + conn.close() + + +class TestQueryHourlyGraphData: + def test_empty_day(self, tmp_path): + conn = init_store(tmp_path / "test.db") + result = _query_hourly_graph_data(conn, "2026-09-20") + assert result == [] + conn.close() + + def test_with_hours(self, tmp_path): + conn = init_store(tmp_path / "test.db") + for h in range(24): + hour = "2026-09-20T%02d:00:00+00:00" % h + bw = 1024*1024*100 if h in (10, 14) else 0 + _insert_hour(conn, hour, bw=bw) + result = _query_hourly_graph_data(conn, "2026-09-20") + assert len(result) == 24 + assert result[10]["bytes_written"] == 1024*1024*100 + assert result[10]["is_zero"] is False + assert result[0]["is_zero"] is True + conn.close() + + def test_hour_labels(self, tmp_path): + conn = init_store(tmp_path / "test.db") + _insert_hour(conn, "2026-09-20T12:00:00+00:00", bw=1000) + result = _query_hourly_graph_data(conn, "2026-09-20") + assert len(result) == 1 + assert result[0]["local_label"] == "12" + conn.close() + + +# --------------------------------------------------------------------------- +# DailyBarGraph widget unit tests (issue #75) +# --------------------------------------------------------------------------- + +class TestDailyBarGraph: + def test_empty_data(self, tmp_path): + """Empty data shows awaiting message.""" + app = FenrisTuiApp(store_path=tmp_path / "nonexistent.db") + # We test the widget directly via the app's compose + graph = DailyBarGraph() + # Simulate setting empty data + from textual.app import App as TextualApp + class _TestApp(TextualApp): + def compose(self): + yield graph + # We can't easily test widget lifecycle outside an app, so test the data + assert graph._all_day_data == [] + assert graph.range_days == _RANGE_DEFAULT + + def test_range_default(self): + """Default range is 14 days.""" + graph = DailyBarGraph() + assert graph.range_days == _RANGE_DEFAULT + assert graph.selected_index == -1 + assert graph.view_mode == "daily" + + def test_trim_to_range(self): + """Data is trimmed to the selected range.""" + graph = DailyBarGraph() + data = [{"day": "2026-09-%02d" % d, "total_bytes": d * 100} + for d in range(1, 31)] + graph._all_day_data = data + graph.range_days = 7 + graph._trim_to_range() + assert len(graph._day_data) == 7 + assert graph._day_data[0]["day"] == "2026-09-24" + + def test_range_all_fits(self): + """When data fits within range, all days are shown.""" + graph = DailyBarGraph() + data = [{"day": "2026-09-%02d" % d, "total_bytes": d * 100} + for d in range(1, 8)] + graph._all_day_data = data + graph.range_days = 14 + graph._trim_to_range() + assert len(graph._day_data) == 7 + + +# --------------------------------------------------------------------------- +# Bar graph headless TUI tests (issue #75) +# --------------------------------------------------------------------------- + +class TestBarGraphTUI: + """Headless tests for the interactive bar graph.""" + + @pytest.mark.asyncio + async def test_graph_renders_with_data(self, tmp_path): + """Graph widget renders with day data.""" + conn = init_store(tmp_path / "test.db") + _insert_segment(conn) + _open_period(conn) + for i in range(14): + d = (datetime(2026, 9, 15) + timedelta(days=i)).strftime("%Y-%m-%d") + _insert_day(conn, d, bw=1024*1024*100) + _insert_sample(conn, "2026-09-30T10:00:00+00:00") + conn.close() + + app = FenrisTuiApp(store_path=tmp_path / "test.db") + async with app.run_test(size=(80, 24)) as pilot: + graph = app.query_one("#usage-history") + assert isinstance(graph, DailyBarGraph) + assert len(graph._day_data) > 0 + # Legend should be visible + legend = str(app.query_one("#bar-legend").render()) + assert "alloc" in legend + assert "zero" in legend + + @pytest.mark.asyncio + async def test_arrow_selection(self, tmp_path): + """Left/right arrows move the selection.""" + conn = init_store(tmp_path / "test.db") + _insert_segment(conn) + _open_period(conn) + for i in range(14): + d = (datetime(2026, 9, 15) + timedelta(days=i)).strftime("%Y-%m-%d") + _insert_day(conn, d, bw=1024*1024*100) + _insert_sample(conn, "2026-09-30T10:00:00+00:00") + conn.close() + + app = FenrisTuiApp(store_path=tmp_path / "test.db") + async with app.run_test(size=(80, 24)) as pilot: + graph = app.query_one("#usage-history") + app.set_focus(graph) + await pilot.pause() + await pilot.pause() + + # Right arrow selects first bar + await pilot.press("right") + await pilot.pause() + assert graph.selected_index == 0 + + # Right again moves to second + await pilot.press("right") + await pilot.pause() + assert graph.selected_index == 1 + + # Left moves back + await pilot.press("left") + await pilot.pause() + assert graph.selected_index == 0 + + @pytest.mark.asyncio + async def test_range_switching(self, tmp_path): + """1-4 keys switch the visible range.""" + conn = init_store(tmp_path / "test.db") + _insert_segment(conn) + _open_period(conn) + for i in range(30): + d = (datetime(2026, 9, 1) + timedelta(days=i)).strftime("%Y-%m-%d") + _insert_day(conn, d, bw=1024*1024*100) + _insert_sample(conn, "2026-09-30T10:00:00+00:00") + conn.close() + + app = FenrisTuiApp(store_path=tmp_path / "test.db") + async with app.run_test(size=(80, 24)) as pilot: + graph = app.query_one("#usage-history") + app.set_focus(graph) + await pilot.pause() + await pilot.pause() + + # Default is 14 + assert graph.range_days == 14 + assert len(graph._day_data) == 14 + + # Press 1 for 7-day range + await pilot.press("1") + await pilot.pause() + assert graph.range_days == 7 + assert len(graph._day_data) == 7 + + # Press 3 for 28-day range + await pilot.press("3") + await pilot.pause() + assert graph.range_days == 28 + assert len(graph._day_data) == 28 + + # Press 4 for 90-day range (only 30 days available) + await pilot.press("4") + await pilot.pause() + assert graph.range_days == 90 + assert len(graph._day_data) == 30 + + @pytest.mark.asyncio + async def test_readout_updates_on_selection(self, tmp_path): + """Readout shows selected day info.""" + conn = init_store(tmp_path / "test.db") + _insert_segment(conn) + _open_period(conn) + for i in range(14): + d = (datetime(2026, 9, 15) + timedelta(days=i)).strftime("%Y-%m-%d") + _insert_day(conn, d, bw=1024*1024*100) + _insert_sample(conn, "2026-09-30T10:00:00+00:00") + conn.close() + + app = FenrisTuiApp(store_path=tmp_path / "test.db") + async with app.run_test(size=(80, 24)) as pilot: + graph = app.query_one("#usage-history") + app.set_focus(graph) + await pilot.pause() + await pilot.pause() + + # No selection initially + readout = str(app.query_one("#bar-readout").render()) + assert "select" in readout.lower() + + # Select first bar + await pilot.press("right") + await pilot.pause() + readout = str(app.query_one("#bar-readout").render()) + assert "2026-09" in readout + assert "GB" in readout + + @pytest.mark.asyncio + async def test_hourly_drill_down_and_back(self, tmp_path): + """Enter drills into hourly view, Esc returns to daily.""" + conn = init_store(tmp_path / "test.db") + _insert_segment(conn) + _open_period(conn) + for i in range(14): + d = (datetime(2026, 9, 15) + timedelta(days=i)).strftime("%Y-%m-%d") + _insert_day(conn, d, bw=1024*1024*100) + # Insert hours for each day + for h in range(24): + hour = "%sT%02d:00:00+00:00" % (d, h) + bw_h = 1024*1024*10 if h == 12 else 0 + _insert_hour(conn, hour, bw=bw_h) + _insert_sample(conn, "2026-09-30T10:00:00+00:00") + conn.close() + + app = FenrisTuiApp(store_path=tmp_path / "test.db") + async with app.run_test(size=(80, 24)) as pilot: + graph = app.query_one("#usage-history") + app.set_focus(graph) + await pilot.pause() + await pilot.pause() + + # Select a day + await pilot.press("right") + await pilot.pause() + assert graph.selected_index == 0 + assert graph.view_mode == "daily" + + # Enter drill-down + await pilot.press("enter") + await pilot.pause() + assert graph.view_mode == "hourly" + assert graph.drill_day is not None + assert len(graph._hour_data) == 24 + + # Esc returns to daily + await pilot.press("escape") + await pilot.pause() + assert graph.view_mode == "daily" + assert graph.drill_day is None + + @pytest.mark.asyncio + async def test_empty_store_graph(self, tmp_path): + """Empty store shows awaiting message in graph.""" + app = FenrisTuiApp(store_path=tmp_path / "nonexistent.db") + async with app.run_test(size=(80, 24)) as pilot: + graph = app.query_one("#usage-history") + assert isinstance(graph, DailyBarGraph) + render = str(app.query_one("#bar-render").render()) + assert "awaiting" in render.lower() + + @pytest.mark.asyncio + async def test_graph_border_title(self, tmp_path): + """Graph pane has a border title.""" + app = FenrisTuiApp(store_path=tmp_path / "nonexistent.db") + async with app.run_test(size=(80, 24)) as pilot: + graph = app.query_one("#usage-history") + assert graph.border_title == "usage history" + + @pytest.mark.asyncio + async def test_glyphs_in_legend(self, tmp_path): + """Legend shows all required glyphs.""" + conn = init_store(tmp_path / "test.db") + _insert_segment(conn) + _open_period(conn) + for i in range(7): + d = (datetime(2026, 9, 20) + timedelta(days=i)).strftime("%Y-%m-%d") + _insert_day(conn, d, bw=1024*1024*100) + _insert_sample(conn, "2026-09-30T10:00:00+00:00") + conn.close() + + app = FenrisTuiApp(store_path=tmp_path / "test.db") + async with app.run_test(size=(80, 24)) as pilot: + await pilot.pause() + legend = str(app.query_one("#bar-legend").render()) + assert _GLYPH_ALLOCATED in legend + assert _GLYPH_UNALLOCATED in legend + assert _GLYPH_GAP in legend + assert _GLYPH_ZERO in legend + assert _GLYPH_PARTIAL in legend