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 <noreply@commandcode.ai>
This commit is contained in:
xavierk
2026-09-14 10:47:37 +05:30
co-authored by CommandCodeBot
parent 6917a658cb
commit 5d916ee97f
2 changed files with 1035 additions and 14 deletions
+628 -14
View File
@@ -19,12 +19,13 @@ import subprocess
import sys import sys
from datetime import datetime, timedelta, timezone from datetime import datetime, timedelta, timezone
from pathlib import Path 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.app import App, ComposeResult
from textual.binding import Binding from textual.binding import Binding
from textual.containers import Container, Horizontal, VerticalScroll from textual.containers import Container, Horizontal, VerticalScroll
from textual.screen import ModalScreen from textual.screen import ModalScreen
from textual.widget import Widget
from textual.widgets import Static from textual.widgets import Static
from .projection import ( 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 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 # Data queries for TUI regions
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
@@ -356,7 +952,7 @@ class FenrisTuiApp(App):
with Container(id="main-grid"): with Container(id="main-grid"):
yield Static("", id="headline-band", classes="pane") yield Static("", id="headline-band", classes="pane")
yield Static("", id="paused-banner") 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="drive-health", classes="pane")
yield Static("", id="service-strip", classes="pane") yield Static("", id="service-strip", classes="pane")
yield Static("q QUIT TUI", id="quit-rail") yield Static("q QUIT TUI", id="quit-rail")
@@ -423,7 +1019,7 @@ class FenrisTuiApp(App):
"[bold]No observations yet[/bold]\n\n" "[bold]No observations yet[/bold]\n\n"
"Enable monitoring: fenris monitor resume" "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("#drive-health").update("")
self.query_one("#service-strip").update( self.query_one("#service-strip").update(
"boot: disabled · timer: inactive · last collect: unknown · freshness: empty · " "boot: disabled · timer: inactive · last collect: unknown · freshness: empty · "
@@ -438,7 +1034,7 @@ class FenrisTuiApp(App):
"[bold red]Observation store unreadable[/bold red]\n" "[bold red]Observation store unreadable[/bold red]\n"
"Check journalctl -u fenris-collect.service" "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("#drive-health").update("")
self.query_one("#service-strip").update( self.query_one("#service-strip").update(
"[dim]by Bongbetic[/dim]\n" "[dim]by Bongbetic[/dim]\n"
@@ -474,16 +1070,20 @@ class FenrisTuiApp(App):
except Exception: except Exception:
self._render_headline("[bold]No projection available[/bold]") self._render_headline("[bold]No projection available[/bold]")
# --- Usage-history pane (§7.2 left) --- # --- Usage-history pane (§7.2 left): interactive bar graph ---
history = _query_usage_history(conn) graph = self.query_one("#usage-history")
history_text = "[bold]Usage history[/bold] · %d days · %.1f–%.1f GB/day\n %s\n %s" % ( day_data = _query_daily_graph_data(conn)
history["num_days"],
history["min_gb"], # Preserve drill-down state across refresh if still valid
history["max_gb"], if graph.view_mode == "hourly" and graph.drill_day:
history["sparkline"], saved_drill_day = graph.drill_day
history["habit_bar"], hour_data = _query_hourly_graph_data(conn, saved_drill_day)
) graph.set_data(day_data, on_drill=self._on_graph_drill)
self.query_one("#usage-history").update(history_text) 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) --- # --- Drive-health pane (§7.2 right) ---
health = _query_drive_health(conn) health = _query_drive_health(conn)
@@ -557,6 +1157,20 @@ class FenrisTuiApp(App):
main_grid.remove_class("paused") main_grid.remove_class("paused")
main_grid.refresh(layout=True) 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: def _format_headline(self, proj: ProjectionResult) -> str:
"""Format the lifespan headline (spec §6.11).""" """Format the lifespan headline (spec §6.11)."""
if proj.headline_remaining_seconds is None: if proj.headline_remaining_seconds is None:
+407
View File
@@ -40,12 +40,23 @@ from fenris.status import (
) )
from fenris.tui import ( from fenris.tui import (
FenrisTuiApp, FenrisTuiApp,
DailyBarGraph,
_format_remaining, _format_remaining,
_sparkline, _sparkline,
_habit_bar, _habit_bar,
_query_usage_history, _query_usage_history,
_query_drive_health, _query_drive_health,
_query_service_facts, _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 # Incomplete provenance → UNVERIFIED tier
assert proj.baseline_tier.value == "unverified_override" assert proj.baseline_tier.value == "unverified_override"
conn.close() 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