From 69e08d9d3af89229d2974413b5dbc7243bf2352c Mon Sep 17 00:00:00 2001 From: xavierk Date: Wed, 16 Sep 2026 08:53:38 +0530 Subject: [PATCH] Release Fenris 0.3.7 --- CHANGELOG.md | 6 + CONTEXT.md | 8 + pyproject.toml | 2 +- src/fenris/__init__.py | 2 +- src/fenris/tui.py | 334 +++++++++++++++++++++++++++++++---------- tests/test_tui.py | 118 +++++++++++---- 6 files changed, 352 insertions(+), 118 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index ff60431..e20c37a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,12 @@ backfill releases from before this changelog. ## [Unreleased] +## [0.3.7] - 2026-09-16 + +### Changed + +- Make usage-history axes, units, UTC boundaries, active 7/14/30/90-day window, gaps, partial periods, stacked write attribution, and hourly drill-down explicit; keep live graph data refreshed on the existing five-minute cadence. + ## [0.3.6] - 2026-09-16 ### Changed diff --git a/CONTEXT.md b/CONTEXT.md index 5558469..6a598c9 100644 --- a/CONTEXT.md +++ b/CONTEXT.md @@ -40,6 +40,14 @@ _Avoid_: Hourly record, hourly.jsonl entry One row per UTC day derived from hour observations; the grain at which usage-habit evidence is judged. _Avoid_: Daily summary, daily stats +**Usage-history window**: +An exact consecutive span of UTC calendar days ending today, shown from day aggregates; a day without trustworthy evidence remains an explicit gap rather than disappearing or being estimated. +_Avoid_: Available records, dataset range + +**Unallocated write evidence**: +Writes known to belong to a UTC day but which cannot be assigned honestly to a particular hour; they contribute to that day's total but are never distributed across hourly bars. +_Avoid_: Missing writes, estimated hourly writes + **Controller segment**: A span of observation history within which the drive's controller identity is unchanged and counters are monotonic; write deltas are never computed across a segment boundary. _Avoid_: Counter reset handling, drive swap detection diff --git a/pyproject.toml b/pyproject.toml index e14db06..3535dab 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "fenris" -version = "0.3.6" +version = "0.3.7" description = "NVMe wear monitor with persistent TUI" requires-python = ">=3.10" license = {file = "LICENSE"} diff --git a/src/fenris/__init__.py b/src/fenris/__init__.py index 4aff423..dc89599 100644 --- a/src/fenris/__init__.py +++ b/src/fenris/__init__.py @@ -1,2 +1,2 @@ """Fenris: NVMe wear monitor with persistent TUI.""" -__version__ = "0.3.6" +__version__ = "0.3.7" diff --git a/src/fenris/tui.py b/src/fenris/tui.py index 2e5249b..fa7dea9 100644 --- a/src/fenris/tui.py +++ b/src/fenris/tui.py @@ -108,12 +108,13 @@ def _can_render_wolf(width: int) -> bool: # --------------------------------------------------------------------------- _GLYPH_ALLOCATED = "\u2588" # \u2588 full block -_GLYPH_UNALLOCATED = "\u2593" # \u2593 dark shade +_GLYPH_UNALLOCATED = "\u2592" # \u2592 medium shade _GLYPH_GAP = "\u2591" # \u2591 light shade _GLYPH_ZERO = "\u00b7" # \u00b7 middle dot -_GLYPH_PARTIAL = "\u258c" # \u258c left half block +_GLYPH_PARTIAL = "\u2504" # \u2504 partial-period cap +_GLYPH_SELECTED = "\u25bc" # \u25bc selection marker -_RANGE_OPTIONS = (7, 14, 28, 90) +_RANGE_OPTIONS = (7, 14, 30, 90) _RANGE_DEFAULT = 14 _BAR_HEIGHT = 6 _BAR_WIDTH = 2 @@ -132,7 +133,7 @@ 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 mouse navigation, range switching (7/14/30/90 days), day selection, and hourly drill-down. No plotting dependency (issue #75 AC5). """ @@ -159,7 +160,7 @@ class DailyBarGraph(Widget): width: 100%; } #bar-readout { - height: 3; + height: 2; width: 100%; } """ @@ -175,6 +176,8 @@ class DailyBarGraph(Widget): self._hour_data: List[Dict[str, Any]] = [] self._max_bytes: int = 0 self._hourly_selected: int = -1 + self._drill_unallocated_bytes: int = 0 + self._visible_start: int = 0 self._on_drill: Optional[Callable[[str], None]] = None def compose(self) -> ComposeResult: @@ -189,6 +192,9 @@ class DailyBarGraph(Widget): on_drill: Any = None, ) -> None: """Update graph with day data. on_drill(day) called on drill entry.""" + selected_day = None + if 0 <= self.selected_index < len(self._day_data): + selected_day = self._day_data[self.selected_index].get("day") self._all_day_data = day_data self._on_drill = on_drill self._trim_to_range() @@ -196,12 +202,32 @@ class DailyBarGraph(Widget): self.drill_day = None self._hour_data = [] self._hourly_selected = -1 + if self._day_data: + matching = next( + (i for i, item in enumerate(self._day_data) + if item.get("day") == selected_day), + None, + ) + self.selected_index = matching if matching is not None else len(self._day_data) - 1 + else: + self.selected_index = -1 self._refresh() def set_hour_data(self, hour_data: List[Dict[str, Any]]) -> None: """Set hourly data for drill-down view.""" + selected_hour = None + if 0 <= self._hourly_selected < len(self._hour_data): + selected_hour = self._hour_data[self._hourly_selected].get("hour") self._hour_data = hour_data - self._hourly_selected = -1 + matching = next( + (i for i, item in enumerate(hour_data) if item.get("hour") == selected_hour), + None, + ) + if matching is not None: + self._hourly_selected = matching + else: + selectable = [i for i, item in enumerate(hour_data) if not item.get("is_future")] + self._hourly_selected = selectable[-1] if selectable else -1 self._refresh_hourly() def _trim_to_range(self) -> None: @@ -210,10 +236,8 @@ class DailyBarGraph(Widget): 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 + (d.get("total_bytes", 0) for d in self._day_data), default=0 ) def _refresh(self) -> None: @@ -229,7 +253,9 @@ class DailyBarGraph(Widget): self._render_readout() def _show_empty(self) -> None: - self.query_one("#bar-range").update("[dim]Usage history[/dim]") + self.query_one("#bar-range").update( + "[dim]Usage history · %d days · UTC[/dim]" % self.range_days + ) self.query_one("#bar-render").update("[dim]Awaiting first sample[/dim]") self.query_one("#bar-legend").update("") self.query_one("#bar-readout").update("") @@ -251,7 +277,7 @@ class DailyBarGraph(Widget): def _show_constrained_summary(self) -> None: """Textual fallback for terminals below 80×24.""" - self.query_one("#bar-range").update("[dim]Usage history[/dim]") + self._render_range() if not self._day_data: self.query_one("#bar-render").update("[dim]Graph needs ≥80×24[/dim]") self.query_one("#bar-legend").update("") @@ -279,7 +305,7 @@ class DailyBarGraph(Widget): % (day.get("local_label", ""), day.get("total_bytes", 0) / 1e9) ) else: - self.query_one("#bar-readout").update("[dim]\u2190 \u2192 Select[/dim]") + self.query_one("#bar-readout").update("[dim]No selectable day[/dim]") def _show_constrained_hourly_summary(self) -> None: """Textual fallback for hourly view when terminal is too small.""" @@ -312,48 +338,88 @@ class DailyBarGraph(Widget): def _render_range(self) -> None: parts = [] - for r in _RANGE_OPTIONS: + for key, r in enumerate(_RANGE_OPTIONS, start=1): + label = "%d: %dd" % (key, r) if r == self.range_days: - parts.append("[bold]%d[/bold]" % r) + parts.append("[reverse bold]\\[%s][/reverse bold]" % label) else: - parts.append(str(r)) - label = "Range: " + " / ".join(parts) + parts.append("[%s]" % label) + label = "Usage history · %d days · UTC %s" % ( + self.range_days, " ".join(parts), + ) if self.view_mode == "hourly": - label += " \u00b7 [bold]%s[/bold] \u00b7 Esc Back" % (self.drill_day or "") + label = "Usage history · Hourly · UTC · [bold]%s[/bold] · Esc Back" % ( + self.drill_day or "", + ) self.query_one("#bar-range").update(label) + def _visible_items(self, items: List[Dict[str, Any]], selected: int) -> tuple[int, List[Dict[str, Any]]]: + """Return the readable viewport containing the selected item.""" + width = getattr(getattr(self, "region", None), "width", 0) or 52 + capacity = max(1, (width - 12) // (_BAR_WIDTH + _BAR_SPACING)) + capacity = min(capacity, len(items)) + if capacity >= len(items): + return 0, items + selected = max(0, selected) + start = min(max(0, selected - capacity + 1), len(items) - capacity) + return start, items[start:start + capacity] + + @staticmethod + def _unit_scale(max_bytes: int) -> tuple[float, str]: + if max_bytes >= 1_000_000_000_000: + return 1e12, "TB" + if max_bytes >= 1_000_000_000: + return 1e9, "GB" + return 1e6, "MB" + def _render_bars(self) -> None: - visible = self._day_data + start, visible = self._visible_items(self._day_data, self.selected_index) + self._visible_start = start if not visible: self.query_one("#bar-render").update("") return max_bytes = self._max_bytes or 1 - bar_h = _BAR_HEIGHT + render_height = self.query_one("#bar-render").region.height + bar_h = max(1, min(_BAR_HEIGHT, render_height - 2)) n = len(visible) - lines: List[str] = [] + scale, unit = self._unit_scale(max_bytes) + lines: List[str] = [ + "Writes (%s) · %s → %s UTC" % ( + unit, visible[0].get("day", ""), visible[-1].get("day", ""), + ) + ] for row in range(bar_h, 0, -1): - line = "" + if row == bar_h: + axis = "%7.2f " % (max_bytes / scale) + elif row == (bar_h + 1) // 2 and row != 1: + axis = "%7.2f " % (max_bytes / scale / 2) + elif row == 1: + axis = "%7s " % "0" + else: + axis = " " + line = axis threshold = (row / bar_h) * max_bytes for i, day in enumerate(visible): allocated = day.get("allocated_bytes", 0) + total = day.get("total_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: + elif total == 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: + elif day.get("is_partial") and row == max( + 1, (total * bar_h + max_bytes - 1) // max_bytes, + ): glyph = _GLYPH_PARTIAL + elif threshold <= allocated: + glyph = _GLYPH_ALLOCATED + elif threshold <= total: + glyph = _GLYPH_UNALLOCATED else: glyph = " " @@ -363,25 +429,12 @@ class DailyBarGraph(Widget): 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 = "" + label_line = "Day UTC " for i, day in enumerate(visible): label = day.get("local_label", day.get("day", ""))[-2:] + if start + i == self.selected_index: + label = _GLYPH_SELECTED + label[-1:] label_line += label if i < n - 1: label_line += " " * _BAR_SPACING @@ -404,9 +457,7 @@ class DailyBarGraph(Widget): 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 Hourly view[/dim]" - ) + self.query_one("#bar-readout").update("[dim]No selectable day[/dim]") return day = self._day_data[self.selected_index] @@ -417,13 +468,17 @@ class DailyBarGraph(Widget): hours = day.get("evidenced_hours", 0) total_gb = total / 1e9 + state = " · partial" if day.get("is_partial") else ( + " · gap" if day.get("is_gap") else "" + ) parts = [ - "[bold]%s[/bold] \u00b7 %.3f GB total \u00b7 %d hours \u00b7 %.0f%% coverage" + "[bold]%s UTC[/bold] \u00b7 %.3f GB total \u00b7 %d hours \u00b7 %.0f%% coverage%s" % ( day.get("local_label", day.get("day", "")), total_gb, hours, coverage * 100, + state, ), ] if unallocated > 0: @@ -449,57 +504,89 @@ class DailyBarGraph(Widget): 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) + start, visible = self._visible_items(self._hour_data, self._hourly_selected) + self._visible_start = start + render_height = self.query_one("#bar-render").region.height + bar_h = max(1, min(_BAR_HEIGHT, render_height - 2)) + n = len(visible) + scale, unit = self._unit_scale(max_bytes) - lines: List[str] = [] + lines: List[str] = [ + "Writes (%s) · %s:00 → %s:00 UTC" % ( + unit, + visible[0].get("local_label", ""), + visible[-1].get("local_label", ""), + ) + ] for row in range(bar_h, 0, -1): - line = "" + if row == bar_h: + axis = "%7.2f " % (max_bytes / scale) + elif row == (bar_h + 1) // 2 and row != 1: + axis = "%7.2f " % (max_bytes / scale / 2) + elif row == 1: + axis = "%7s " % "0" + else: + axis = " " + line = axis threshold = (row / bar_h) * max_bytes - for i, hour in enumerate(self._hour_data): + for i, hour in enumerate(visible): bw = hour.get("bytes_written", 0) - is_zero = hour.get("is_zero", False) - - if is_zero and row == 1: + if hour.get("is_future"): + glyph = " " + elif hour.get("is_gap"): + glyph = _GLYPH_GAP if row <= 2 else " " + elif hour.get("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: + elif hour.get("is_partial") and row == max( + 1, (bw * bar_h + max_bytes - 1) // max_bytes, + ): + glyph = _GLYPH_PARTIAL + elif 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_line = "Hour UTC" + for i, hour in enumerate(visible): label = hour.get("local_label", hour.get("hour", ""))[-2:] - label_line += label + if start + i == self._hourly_selected: + label = _GLYPH_SELECTED + label[-1:] + label_line += " " + label if i < n - 1: - label_line += " " * _BAR_SPACING + label_line += " " * max(0, _BAR_SPACING - 1) 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) + "%s Writes · %s Gap · %s Zero · %s Partial" + % (_GLYPH_ALLOCATED, _GLYPH_GAP, _GLYPH_ZERO, _GLYPH_PARTIAL) ) # Hourly readout if 0 <= self._hourly_selected < len(self._hour_data): h = self._hour_data[self._hourly_selected] + state = " · partial" if h.get("is_partial") else ( + " · gap" if h.get("is_gap") else "" + ) + note = "" + if self._drill_unallocated_bytes: + note = "\n%.3f GB could not be assigned to an hour" % ( + self._drill_unallocated_bytes / 1e9, + ) self.query_one("#bar-readout").update( - "[bold]%s[/bold] \u00b7 %.3f GB \u00b7 %d%% coverage" + "[bold]%s:00 UTC[/bold] \u00b7 %.3f GB \u00b7 %d%% coverage%s%s" % ( h.get("local_label", h.get("hour", "")), h.get("bytes_written", 0) / 1e9, h.get("coverage", 0) * 100, + state, + note, ) ) else: @@ -535,8 +622,12 @@ class DailyBarGraph(Widget): event.stop() elif event.key in ("1", "2", "3", "4"): self.range_days = _RANGE_OPTIONS[int(event.key) - 1] - self.selected_index = -1 + selected_day = self._day_data[self.selected_index].get("day") if self._day_data else None self._trim_to_range() + self.selected_index = next( + (i for i, item in enumerate(self._day_data) if item.get("day") == selected_day), + len(self._day_data) - 1, + ) self._refresh() event.stop() @@ -545,6 +636,7 @@ class DailyBarGraph(Widget): return day = self._day_data[self.selected_index] self.drill_day = day.get("day") + self._drill_unallocated_bytes = day.get("unallocated_bytes", 0) self.view_mode = "hourly" self._hourly_selected = -1 if self._on_drill: @@ -586,15 +678,17 @@ class DailyBarGraph(Widget): render = self.query_one("#bar-render") offset_x = event.x - render.region.x bar_total = _BAR_WIDTH + _BAR_SPACING - idx = offset_x // bar_total + idx = max(0, offset_x - 8) // bar_total if self.view_mode == "daily": - if 0 <= idx < len(self._day_data): - self.selected_index = idx + global_idx = self._visible_start + idx + if 0 <= global_idx < len(self._day_data): + self.selected_index = global_idx self._refresh() else: - if 0 <= idx < len(self._hour_data): - self._hourly_selected = idx + global_idx = self._visible_start + idx + if 0 <= global_idx < len(self._hour_data): + self._hourly_selected = global_idx self._refresh_hourly() @@ -604,6 +698,7 @@ class DailyBarGraph(Widget): def _query_daily_graph_data( conn: sqlite3.Connection, + window_end: Optional[datetime] = None, ) -> List[Dict[str, Any]]: """Query day aggregates for the bar graph. @@ -618,7 +713,7 @@ def _query_daily_graph_data( ) rows = cursor.fetchall() - result: List[Dict[str, Any]] = [] + by_day: Dict[str, Dict[str, Any]] = {} for row in rows: day = row[0] bw_delta = row[1] or 0 @@ -641,7 +736,7 @@ def _query_daily_graph_data( ) is_partial = coverage < 0.5 - result.append({ + by_day[day] = { "day": day, "local_label": day, "total_bytes": total_bytes, @@ -653,7 +748,32 @@ def _query_daily_graph_data( "is_zero": is_zero, "is_gap": is_gap, "is_partial": is_partial, + } + + if window_end is None: + return list(by_day.values()) + + end_date = window_end.astimezone(timezone.utc).date() + result: List[Dict[str, Any]] = [] + for offset in range(max(_RANGE_OPTIONS) - 1, -1, -1): + day = (end_date - timedelta(days=offset)).isoformat() + entry = by_day.get(day, { + "day": day, + "local_label": day, + "total_bytes": 0, + "allocated_bytes": 0, + "unallocated_bytes": 0, + "coverage": 0.0, + "evidenced_hours": 0, + "sample_count": 0, + "is_zero": False, + "is_gap": True, + "is_partial": False, }) + if day == end_date.isoformat(): + entry = dict(entry) + entry["is_partial"] = True + result.append(entry) return result @@ -661,6 +781,7 @@ def _query_daily_graph_data( def _query_hourly_graph_data( conn: sqlite3.Connection, day: str, + now: Optional[datetime] = None, ) -> List[Dict[str, Any]]: """Query hour observations for a specific day. @@ -675,7 +796,7 @@ def _query_hourly_graph_data( ) rows = cursor.fetchall() - result: List[Dict[str, Any]] = [] + by_hour: Dict[int, Dict[str, Any]] = {} for row in rows: hour = row[0] bw = row[1] or 0 @@ -689,7 +810,8 @@ def _query_hourly_graph_data( is_zero = bw == 0 local_label = hour[11:13] if len(hour) >= 13 else hour - result.append({ + hour_number = int(local_label) + by_hour[hour_number] = { "hour": hour, "local_label": local_label, "bytes_written": bw, @@ -700,7 +822,40 @@ def _query_hourly_graph_data( "powered_off_seconds": powered_off, "unknown_seconds": unknown, "is_zero": is_zero, - }) + "is_gap": False, + "is_partial": False, + "is_future": False, + } + + current = (now or datetime.now(timezone.utc)).astimezone(timezone.utc) + selected_date = datetime.fromisoformat(day).date() + result: List[Dict[str, Any]] = [] + for hour_number in range(24): + hour_start = datetime.combine( + selected_date, datetime.min.time(), tzinfo=timezone.utc, + ) + timedelta(hours=hour_number) + is_future = hour_start > current + entry = by_hour.get(hour_number) + if entry is None: + entry = { + "hour": hour_start.isoformat(), + "local_label": "%02d" % hour_number, + "bytes_written": 0, + "coverage": 0.0, + "sample_count": 0, + "active_seconds": 0, + "idle_seconds": 0, + "powered_off_seconds": 0, + "unknown_seconds": 0, + "is_zero": False, + "is_gap": not is_future, + "is_partial": hour_start <= current < hour_start + timedelta(hours=1), + "is_future": is_future, + } + else: + entry["is_partial"] = hour_start <= current < hour_start + timedelta(hours=1) + entry["is_future"] = is_future + result.append(entry) return result @@ -1103,16 +1258,28 @@ class FenrisTuiApp(App): # --- Usage-history pane (§7.2 left): interactive bar graph --- graph = self.query_one("#usage-history") - day_data = _query_daily_graph_data(conn) + day_data = _query_daily_graph_data(conn, self._clock_now) # 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) + saved_hour = None + if 0 <= graph._hourly_selected < len(graph._hour_data): + saved_hour = graph._hour_data[graph._hourly_selected].get("hour") + hour_data = _query_hourly_graph_data( + conn, saved_drill_day, self._clock_now, + ) 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) + if saved_hour is not None: + graph._hourly_selected = next( + (i for i, item in enumerate(hour_data) + if item.get("hour") == saved_hour), + graph._hourly_selected, + ) + graph._refresh_hourly() else: graph.set_data(day_data, on_drill=self._on_graph_drill) @@ -1183,7 +1350,8 @@ class FenrisTuiApp(App): self.store_path, self._clock_now, query_services=False, ) as (conn, _): graph.set_hour_data( - _query_hourly_graph_data(conn, day) if conn is not None else [] + _query_hourly_graph_data(conn, day, self._clock_now) + if conn is not None else [] ) def _format_headline(self, proj: ProjectionResult) -> str: diff --git a/tests/test_tui.py b/tests/test_tui.py index f5302c2..95a3d1c 100644 --- a/tests/test_tui.py +++ b/tests/test_tui.py @@ -771,12 +771,29 @@ class TestQueryDailyGraphData: assert result[0]["is_gap"] is True conn.close() + def test_window_is_consecutive_utc_days_with_explicit_gaps(self, tmp_path): + conn = init_store(tmp_path / "test.db") + _insert_day(conn, "2026-09-14", bw=500) + result = _query_daily_graph_data( + conn, datetime(2026, 9, 16, 12, tzinfo=timezone.utc), + ) + assert len(result) == 90 + assert result[-3]["day"] == "2026-09-14" + assert result[-2]["day"] == "2026-09-15" + assert result[-2]["is_gap"] is True + assert result[-1]["day"] == "2026-09-16" + assert result[-1]["is_partial"] 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 == [] + result = _query_hourly_graph_data( + conn, "2026-09-20", datetime(2026, 9, 21, tzinfo=timezone.utc), + ) + assert len(result) == 24 + assert all(item["is_gap"] for item in result) conn.close() def test_with_hours(self, tmp_path): @@ -795,9 +812,25 @@ class TestQueryHourlyGraphData: 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" + result = _query_hourly_graph_data( + conn, "2026-09-20", datetime(2026, 9, 21, tzinfo=timezone.utc), + ) + assert len(result) == 24 + assert result[12]["local_label"] == "12" + assert result[12]["bytes_written"] == 1000 + conn.close() + + def test_current_hour_is_partial_and_future_hours_are_not_gaps(self, tmp_path): + conn = init_store(tmp_path / "test.db") + result = _query_hourly_graph_data( + conn, "2026-09-20", + datetime(2026, 9, 20, 14, 30, tzinfo=timezone.utc), + ) + assert result[13]["is_gap"] is True + assert result[14]["is_partial"] is True + assert result[14]["is_gap"] is True + assert result[15]["is_future"] is True + assert result[15]["is_gap"] is False conn.close() @@ -848,6 +881,17 @@ class TestDailyBarGraph: graph._trim_to_range() assert len(graph._day_data) == 7 + def test_scale_includes_allocated_and_unallocated_writes(self): + graph = DailyBarGraph() + graph._all_day_data = [{ + "day": "2026-09-16", + "total_bytes": 800, + "allocated_bytes": 500, + "unallocated_bytes": 300, + }] + graph._trim_to_range() + assert graph._max_bytes == 800 + # --------------------------------------------------------------------------- # Bar graph headless TUI tests (issue #75) @@ -877,6 +921,13 @@ class TestBarGraphTUI: legend = str(app.query_one("#bar-legend").render()) assert "Alloc" in legend assert "Zero" in legend + range_label = str(app.query_one("#bar-range").render()) + assert "14 days" in range_label + assert "UTC" in range_label + plotted = str(app.query_one("#bar-render").render()) + assert "Writes (" in plotted + assert "Day UTC" in plotted + assert "→" in plotted @pytest.mark.asyncio async def test_arrow_selection(self, tmp_path): @@ -897,20 +948,16 @@ class TestBarGraphTUI: 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 + # The newest day is selected by default; left moves backward. + initial = graph.selected_index await pilot.press("left") await pilot.pause() - assert graph.selected_index == 0 + assert graph.selected_index == initial - 1 + + # Right returns to the newest day. + await pilot.press("right") + await pilot.pause() + assert graph.selected_index == initial @pytest.mark.asyncio async def test_range_switching(self, tmp_path): @@ -941,17 +988,17 @@ class TestBarGraphTUI: assert graph.range_days == 7 assert len(graph._day_data) == 7 - # Press 3 for 28-day range + # Press 3 for 30-day range await pilot.press("3") await pilot.pause() - assert graph.range_days == 28 - assert len(graph._day_data) == 28 + assert graph.range_days == 30 + assert len(graph._day_data) == 30 - # Press 4 for 90-day range (only 30 days available) + # Exact windows retain gap slots instead of compressing time. await pilot.press("4") await pilot.pause() assert graph.range_days == 90 - assert len(graph._day_data) == 30 + assert len(graph._day_data) == 90 @pytest.mark.asyncio async def test_readout_updates_on_selection(self, tmp_path): @@ -972,12 +1019,13 @@ class TestBarGraphTUI: await pilot.pause() await pilot.pause() - # No selection initially + # Newest day is selected initially. readout = str(app.query_one("#bar-readout").render()) - assert "select" in readout.lower() + assert "UTC" in readout + assert "GB" in readout - # Select first bar - await pilot.press("right") + # Moving left updates the selected-day readout. + await pilot.press("left") await pilot.pause() readout = str(app.query_one("#bar-readout").render()) assert "2026-09" in readout @@ -1007,10 +1055,8 @@ class TestBarGraphTUI: await pilot.pause() await pilot.pause() - # Select a day - await pilot.press("right") - await pilot.pause() - assert graph.selected_index == 0 + # The newest day is selected automatically. + assert graph.selected_index == len(graph._day_data) - 1 assert graph.view_mode == "daily" # Enter drill-down @@ -1020,6 +1066,14 @@ class TestBarGraphTUI: assert graph.drill_day is not None assert len(graph._hour_data) == 24 + # Five-minute data refreshes retain the selected hour. + graph._hourly_selected = 12 + selected_hour = graph._hour_data[12]["hour"] + app._refresh() + await pilot.pause() + assert graph.view_mode == "hourly" + assert graph._hour_data[graph._hourly_selected]["hour"] == selected_hour + # Esc returns to daily await pilot.press("escape") await pilot.pause() @@ -1208,9 +1262,7 @@ class TestConstrainedLayout: # Focus and select a day app.set_focus(graph) await pilot.pause() - await pilot.press("right") - await pilot.pause() - assert graph.selected_index == 0 + assert graph.selected_index == len(graph._day_data) - 1 # Resize to constrained — pilot.resize_terminal changes terminal size await pilot.resize_terminal(79, 24)