diff --git a/src/fenris/activity_selection.py b/src/fenris/activity_selection.py index 7709500..a651a82 100644 --- a/src/fenris/activity_selection.py +++ b/src/fenris/activity_selection.py @@ -13,12 +13,23 @@ class IntervalIdentity: end_ts: str +@dataclass(frozen=True) +class HistoryDayIdentity: + """Stable identity for one recorded local-day summary.""" + + local_date: str + timezone: str + utc_start: str + utc_end: str + + class ActivitySelection: """Keep activity navigation state separate from its Textual rendering.""" def __init__(self) -> None: self.view = "live" self.browse_date: str | None = None + self.selected_history_day: HistoryDayIdentity | None = None self.selected_history_hour: str | None = None self.history_range_days = HISTORY_RANGE_DEFAULT self.measure = "written" @@ -34,11 +45,23 @@ class ActivitySelection: if view == "live": self.follow_live() - def select_history_date(self, date: str | None) -> None: - """Keep the chosen historical local date across refresh.""" - if date != self.browse_date: + def select_history_date( + self, + date: str | None, + identity: HistoryDayIdentity | None = None, + ) -> None: + """Keep requested date and evidence identity across refresh.""" + if date != self.browse_date or ( + identity is not None and identity != self.selected_history_day + ): self.selected_history_hour = None + if date != self.browse_date: + self.selected_history_day = None self.browse_date = date + if identity is not None: + self.selected_history_day = identity + elif date is None: + self.selected_history_day = None def select_history_hour(self, hour: str | None) -> None: """Keep the chosen historical hour across refresh.""" @@ -53,6 +76,7 @@ class ActivitySelection: """Resume following the newest interval and clear stale-pin notices.""" self.view = "live" self.browse_date = None + self.selected_history_day = None self.selected_history_hour = None self.following_live = True self.selected_live_interval = None diff --git a/src/fenris/tui.py b/src/fenris/tui.py index 2dd0864..e08387f 100644 --- a/src/fenris/tui.py +++ b/src/fenris/tui.py @@ -13,9 +13,10 @@ from __future__ import annotations import sqlite3 import sys -from datetime import datetime, timedelta, timezone +from datetime import date, datetime, timedelta, timezone from pathlib import Path from typing import Any, Callable, Dict, List, Optional +from zoneinfo import ZoneInfo from rich.text import Text from textual.app import App, ComposeResult @@ -29,6 +30,7 @@ from textual.widgets import Input, Static, Tab, Tabs from .activity_selection import ( HISTORY_RANGE_DEFAULT, ActivitySelection, + HistoryDayIdentity, IntervalIdentity, ) from .activity_plot import VolumePoint, volume_plot @@ -175,7 +177,11 @@ class HistoryGraph(Widget): self._hour_data: List[Dict[str, Any]] = [] self._max_bytes: int = 0 self._hourly_selected: int = -1 + self._showing_selected_day_outside_range = False self._drill_unallocated_bytes: int = 0 + self._drill_unallocated_read: int = 0 + self._drill_shared_written: int = 0 + self._drill_shared_read: int = 0 self._on_drill: Optional[Callable[[str], None]] = None self._point_columns: list[int] = [] self._empty_message = "Awaiting first sample" @@ -214,45 +220,50 @@ class HistoryGraph(Widget): """Update graph with day data. on_drill(day) called on drill entry.""" self._empty_message = empty_message selected_day = selected_date or self.selection.browse_date - if selected_day is None and 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() - self.view_mode = "daily" - self.drill_day = None - self._hour_data = [] - self._hourly_selected = -1 + self.view_mode = "hourly" if self.selection.view == "day" else "daily" + self.drill_day = selected_day if self.view_mode == "hourly" else None 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 - if self.selection.view != "live": - self.selection.select_history_date( - self._day_data[self.selected_index].get("day") + if selected_day is None: + self.selected_index = len(self._day_data) - 1 + else: + identity = self.selection.selected_history_day + matching = next( + (i for i, item in enumerate(self._day_data) + if item.get("day") == selected_day + and (identity is None or item.get("identity") == identity)), + None, ) + self.selected_index = matching if matching is not None else -1 + if matching is not None and identity is None: + self.selection.select_history_date( + selected_day, self._day_data[matching].get("identity"), + ) else: self.selected_index = -1 - self._refresh() + if self.view_mode == "daily": + self._refresh() + else: + self._render_range() def set_hour_data(self, hour_data: List[Dict[str, Any]]) -> None: """Set hourly data for drill-down view.""" selected_hour = self.selection.selected_history_hour - selected_hour = selected_hour or self._selected_history_hour() self._hour_data = hour_data - 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: + matching = next(( + i for i, item in enumerate(hour_data) + if item.get("hour") == selected_hour + ), None) if selected_hour is not None else None + if selected_hour is not None: + self._hourly_selected = matching if matching is not None else -1 + elif hour_data: 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.selection.select_history_hour(self._selected_history_hour()) + self.selection.select_history_hour(self._selected_history_hour()) + else: + self._hourly_selected = -1 self._refresh_hourly() def _trim_to_range(self) -> None: @@ -261,13 +272,41 @@ class HistoryGraph(Widget): if len(self._all_day_data) > self.range_days else list(self._all_day_data) ) + self._showing_selected_day_outside_range = False + selected_date = self.selection.browse_date + if selected_date and not any( + item.get("day") == selected_date for item in self._day_data + ): + identity = self.selection.selected_history_day + selected = next( + (item for item in self._all_day_data + if item.get("day") == selected_date + and (identity is None or item.get("identity") == identity)), + None, + ) + if selected is not None: + self._day_data = [selected] + self._showing_selected_day_outside_range = True + elif identity is not None: + self._day_data = [] self._max_bytes = max( - (d.get("total_bytes", 0) for d in self._day_data), default=0 + (d.get("total_bytes") or 0 for d in self._day_data), default=0 ) def _refresh(self) -> None: if not self._day_data: - self._show_empty() + if self.selection.browse_date is not None: + self.query_one("#bar-range").update( + "Local-day history · %d days" % self.range_days + ) + self.query_one("#bar-render").update(self._empty_message) + self.query_one("#bar-legend").update("") + self.query_one("#bar-readout").update( + "%s · local-day evidence unavailable" + % self.selection.browse_date + ) + else: + self._show_empty() return if self._is_constrained(): self._show_constrained_summary() @@ -311,9 +350,16 @@ class HistoryGraph(Widget): return # Summarise visible days as text - total_written = sum(d.get("total_written", d.get("total_bytes", 0)) for d in self._day_data) - total_read = sum(d.get("total_read", 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 or d.get("total_read", 0) > 0) + total_written = sum( + (d.get("total_written", d.get("total_bytes", 0)) or 0) + for d in self._day_data + ) + total_read = sum((d.get("total_read", 0) or 0) for d in self._day_data) + days_with_data = sum( + 1 for d in self._day_data + if (d.get("total_bytes", 0) or 0) > 0 + or (d.get("total_read", 0) or 0) > 0 + ) n = len(self._day_data) first = self._day_data[0].get("local_label", "") last = self._day_data[-1].get("local_label", "") @@ -352,6 +398,15 @@ class HistoryGraph(Widget): self._refresh_hourly() def _render_range(self) -> None: + if self._showing_selected_day_outside_range and self._day_data: + selected = self._day_data[0] + self.query_one("#bar-range").update( + "Selected day · [bold]%s[/bold] · %s" % ( + selected.get("day", ""), + selected.get("timezone_label", "local-day evidence unavailable"), + ) + ) + return parts = [] for key, r in enumerate(_RANGE_OPTIONS, start=1): label = "%d: %dd" % (key, r) @@ -359,20 +414,38 @@ class HistoryGraph(Widget): parts.append("[reverse bold]\\[%s][/reverse bold]" % label) else: parts.append("[%s]" % label) - label = "%d days · UTC %s" % ( + label = "%d local days %s" % ( self.range_days, " ".join(parts), ) if self.view_mode == "hourly": - label = "Hourly · UTC · [bold]%s[/bold] · Backspace History" % ( + selected_day = next(( + item for item in self._all_day_data + if item.get("day") == self.drill_day + and (self.selection.selected_history_day is None + or item.get("identity") == self.selection.selected_history_day) + ), None) + timezone_label = selected_day.get("timezone_label", "") if selected_day else "" + label = "Hourly · [bold]%s[/bold] · %s · Backspace History" % ( self.drill_day or "", + timezone_label, ) self.query_one("#bar-range").update(label) def _render_bars(self) -> None: points = [] for index, day in enumerate(self._day_data): - volume = day.get("total_read", 0) if self.measure == "read" else day.get("total_bytes", 0) - unallocated = day.get("unallocated_read" if self.measure == "read" else "unallocated_bytes", 0) + if "activity_state" in day: + volume = day.get("total_read") if self.measure == "read" else day.get("total_bytes") + unallocated = day.get( + "shared_bytes_read" if self.measure == "read" else "shared_bytes_written", + 0, + ) + else: + volume = day.get("total_read", 0) if self.measure == "read" else day.get("total_bytes", 0) + unallocated = day.get( + "unallocated_read" if self.measure == "read" else "unallocated_bytes", + 0, + ) state = "unallocated" if unallocated else "gap" if day.get("is_gap") else ( "partial" if day.get("is_partial") else "measured" ) @@ -395,10 +468,48 @@ class HistoryGraph(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]No selectable day[/dim]") + date = self.selection.browse_date + self.query_one("#bar-readout").update( + ("%s · local-day evidence unavailable" % date) + if date else "[dim]No selectable day[/dim]" + ) return day = self._day_data[self.selected_index] + if "activity_state" in day: + if day.get("is_gap"): + self.query_one("#bar-readout").update( + "%s · local-day evidence unavailable" + % day.get("day", "") + ) + return + written = day.get("total_written") + read = day.get("total_read") + written_text = "unavailable" if written is None else "%.3f GB known" % (written / 1e9) + read_text = "unavailable" if read is None else "%.3f GB known" % (read / 1e9) + parts = [ + "[bold]%s[/bold] · %s · %s" + % (day.get("day", ""), day.get("timezone_label", ""), + day.get("activity_state", "incomplete")), + "W %s · R %s" % (written_text, read_text), + ] + shared_written = day.get("shared_bytes_written", 0) + shared_read = day.get("shared_bytes_read", 0) + if shared_written or shared_read: + parts.append( + "shared at midnight W %.3f GB · R %.3f GB" + % (shared_written / 1e9, shared_read / 1e9) + ) + unallocated_written = day.get("unallocated_bytes_written", 0) + unallocated_read = day.get("unallocated_bytes_read", 0) + if unallocated_written or unallocated_read: + parts.append( + "unallocated W %.3f GB · R %.3f GB" + % (unallocated_written / 1e9, unallocated_read / 1e9) + ) + self.query_one("#bar-readout").update("\n".join(parts)) + return + if day.get("is_gap"): # Known daily evidence survives missing hourly coverage; absent # evidence must not be presented as a measured zero. @@ -452,7 +563,15 @@ class HistoryGraph(Widget): if not self._hour_data: self.query_one("#bar-render").update("[dim]No hourly data[/dim]") self.query_one("#bar-legend").update("") - self.query_one("#bar-readout").update("") + if self.selection.selected_history_hour is not None: + readout = "%s · selected hour unavailable" % ( + self.selection.selected_history_hour, + ) + elif self.drill_day: + readout = "%s · hourly evidence unavailable" % self.drill_day + else: + readout = "" + self.query_one("#bar-readout").update(readout) return self._render_range() @@ -467,8 +586,11 @@ class HistoryGraph(Widget): "gap" if hour.get("is_gap") else "partial" if hour.get("is_partial") else "measured" ) volume = hour.get("bytes_read" if self.measure == "read" else "bytes_written", 0) + local_label = hour.get("local_label", "") + if not hour.get("timezone_label"): + local_label += ":00" points.append(VolumePoint(index, None if state in ("future", "gap") else volume, - hour.get("local_label", "") + ":00", state)) + local_label, state)) self._plot(points, self._hourly_selected) self._render_hourly_readout() @@ -476,6 +598,43 @@ class HistoryGraph(Widget): def _render_hourly_readout(self) -> None: if 0 <= self._hourly_selected < len(self._hour_data): h = self._hour_data[self._hourly_selected] + if h.get("timezone_label"): + state = "future" if h.get("is_future") else ( + "gap" if h.get("is_gap") else + "partial" if h.get("is_partial") else "measured" + ) + note = "" + if self._drill_unallocated_bytes or self._drill_unallocated_read: + note = "\nunallocated W %.3f GB · R %.3f GB" % ( + self._drill_unallocated_bytes / 1e9, + self._drill_unallocated_read / 1e9, + ) + if self._drill_shared_written or self._drill_shared_read: + note += "\nshared at midnight W %.3f GB · R %.3f GB" % ( + self._drill_shared_written / 1e9, + self._drill_shared_read / 1e9, + ) + if h.get("is_gap") or h.get("is_future"): + self.query_one("#bar-readout").update( + "%s %s · %s · %s\nW unavailable · R unavailable%s" + % (h.get("local_label", ""), h.get("timezone_label", ""), + h.get("utc_label", ""), state, note) + ) + return + self.query_one("#bar-readout").update( + "[bold]%s %s[/bold] · %s · %s\n" + "W %.3f GB · R %.3f GB · %d%% coverage · %s%s" + % ( + h.get("local_label", ""), h.get("timezone_label", ""), + h.get("utc_label", ""), self.drill_day or "", + h.get("bytes_written", 0) / 1e9, + h.get("bytes_read", 0) / 1e9, + h.get("coverage", 0) * 100, + state, + note, + ) + ) + return if h.get("is_gap") or h.get("is_future"): self.query_one("#bar-readout").update( "%s:00 UTC · %s\nW unavailable · R unavailable" % ( @@ -499,7 +658,13 @@ class HistoryGraph(Widget): ) ) else: - self.query_one("#bar-readout").update("[dim]\u2190 \u2192 Select hour[/dim]") + if self.selection.selected_history_hour is not None: + self.query_one("#bar-readout").update( + "%s · selected hour unavailable" + % self.selection.selected_history_hour + ) + else: + self.query_one("#bar-readout").update("[dim]\u2190 \u2192 Select hour[/dim]") # -- Event handling -- @@ -545,8 +710,9 @@ class HistoryGraph(Widget): def _notify_selection(self) -> None: if 0 <= self.selected_index < len(self._day_data): - selected_day = self._day_data[self.selected_index]["day"] - self.selection.select_history_date(selected_day) + day = self._day_data[self.selected_index] + selected_day = day["day"] + self.selection.select_history_date(selected_day, day.get("identity")) self.post_message(self.DaySelected(selected_day)) def _enter_drill(self) -> None: @@ -554,21 +720,56 @@ class HistoryGraph(Widget): return day = self._day_data[self.selected_index] self.drill_day = day.get("day") - self.selection.select_history_date(self.drill_day) - self.selection.select_history_hour(None) - self._drill_unallocated_bytes = day.get("unallocated_bytes", 0) + self.selection.select_history_date(self.drill_day, day.get("identity")) + self.selection.set_view("day") + self._drill_unallocated_bytes = day.get("unallocated_bytes_written", 0) + self._drill_unallocated_read = day.get("unallocated_bytes_read", 0) + self._drill_shared_written = day.get("shared_bytes_written", 0) + self._drill_shared_read = day.get("shared_bytes_read", 0) self.view_mode = "hourly" self._hourly_selected = -1 self._render_range() if self._on_drill: self._on_drill(self.drill_day) + def show_selected_day(self) -> None: + """Show hourly evidence for selected date without rebuilding selection.""" + if self.selection.browse_date is None: + if not self._day_data: + self.selection.select_history_date(None) + self.view_mode = "hourly" + self.drill_day = None + self._refresh_hourly() + return + selected = self._day_data[-1] + self.selection.select_history_date( + selected.get("day"), selected.get("identity"), + ) + + self.drill_day = self.selection.browse_date + selected = next(( + item for item in self._all_day_data + if item.get("day") == self.drill_day + and (self.selection.selected_history_day is None + or item.get("identity") == self.selection.selected_history_day) + ), None) + selected = selected or {} + self._drill_unallocated_bytes = selected.get("unallocated_bytes_written", 0) + self._drill_unallocated_read = selected.get("unallocated_bytes_read", 0) + self._drill_shared_written = selected.get("shared_bytes_written", 0) + self._drill_shared_read = selected.get("shared_bytes_read", 0) + self.view_mode = "hourly" + self._render_range() + if self._on_drill and self.drill_day: + self._on_drill(self.drill_day) + else: + self._refresh_hourly() + def _exit_drill(self) -> None: self.view_mode = "daily" self.drill_day = None self._hour_data = [] self._hourly_selected = -1 - self.selection.select_history_hour(None) self._refresh() def _handle_hourly_key(self, event: Any) -> None: @@ -580,6 +781,9 @@ class HistoryGraph(Widget): self.app._set_activity_view("history") event.stop() elif event.key == "left": + if not self._hour_data: + event.stop() + return if self._hourly_selected < 0: self._hourly_selected = len(self._hour_data) - 1 else: @@ -588,6 +792,9 @@ class HistoryGraph(Widget): self._refresh_hourly() event.stop() elif event.key == "right": + if not self._hour_data: + event.stop() + return if self._hourly_selected < 0: self._hourly_selected = 0 else: @@ -599,7 +806,9 @@ class HistoryGraph(Widget): event.stop() def _select_history_hour(self) -> None: - self.selection.select_history_hour(self._selected_history_hour()) + selected_hour = self._selected_history_hour() + if selected_hour is not None: + self.selection.select_history_hour(selected_hour) def _selected_history_hour(self) -> Optional[str]: if 0 <= self._hourly_selected < len(self._hour_data): @@ -611,21 +820,29 @@ class HistoryGraph(Widget): Returns True if the selection changed. """ - idx = next( - (i for i, d in enumerate(self._all_day_data) - if d.get("day") == date_str), - None, - ) - if idx is None: - return False - - old_day = None - if 0 <= self.selected_index < len(self._day_data): + old_day = self.selection.browse_date + if old_day is None and 0 <= self.selected_index < len(self._day_data): old_day = self._day_data[self.selected_index].get("day") + self.selection.select_history_date(date_str) + + identity = self.selection.selected_history_day + matching = [ + i for i, day in enumerate(self._all_day_data) + if day.get("day") == date_str + and (identity is None or day.get("identity") == identity) + ] + idx = matching[0] if matching else None + if idx is None: + self.selected_index = -1 + self._refresh() + return False # Count actual data days (not gaps) after the target n_data_after = len(self._all_day_data) - idx - if n_data_after > self.range_days: + if ( + idx >= len(self._all_day_data) - max(_RANGE_OPTIONS) + and n_data_after > self.range_days + ): self.range_days = max(n_data_after, min(_RANGE_OPTIONS)) self._trim_to_range() @@ -633,12 +850,15 @@ class HistoryGraph(Widget): return False self.selected_index = next( (i for i, d in enumerate(self._day_data) - if d.get("day") == date_str), - idx, + if d.get("day") == date_str + and (identity is None or d.get("identity") == identity)), + -1, ) - new_day = self._day_data[self.selected_index].get("day") - self.selection.select_history_date(new_day) - return new_day != old_day + if self.selected_index >= 0: + day = self._day_data[self.selected_index] + self.selection.select_history_date(date_str, day.get("identity")) + self._refresh() + return date_str != old_day def on_click(self, event: Any) -> None: render = self.query_one("#bar-render") @@ -652,9 +872,8 @@ class HistoryGraph(Widget): global_idx = idx if 0 <= global_idx < len(self._day_data): self.selected_index = global_idx - self.selection.select_history_date( - self._day_data[global_idx].get("day") - ) + day = self._day_data[global_idx] + self.selection.select_history_date(day.get("day"), day.get("identity")) self._refresh() self._notify_selection() else: @@ -848,6 +1067,178 @@ def _query_hourly_graph_data( return result +def _query_local_day_graph_data( + conn: sqlite3.Connection, + now: datetime, + timezone_name: str | None = None, + days: int = max(_RANGE_OPTIONS), + selected_date: str | None = None, +) -> List[Dict[str, Any]]: + """Read recorded local-day evidence, keeping unavailable dates visible.""" + from .local_day import query_local_day_summary + from .tz_util import detect_system_tz + + current_zone_name = timezone_name or detect_system_tz() + current_zone = ZoneInfo(current_zone_name) + end_date = now.astimezone(current_zone).date() + start_date = end_date - timedelta(days=days - 1) + start_text, end_text = start_date.isoformat(), end_date.isoformat() + local_dates = { + (start_date + timedelta(days=offset)).isoformat() + for offset in range(days) + } + if selected_date is not None: + try: + date.fromisoformat(selected_date) + except ValueError: + selected_date = None + else: + local_dates.add(selected_date) + rows = conn.execute( + "SELECT local_date, tz_name FROM local_days " + "WHERE (local_date >= ? AND local_date <= ?) " + "OR local_date = ? ORDER BY local_date, id", + (start_text, end_text, selected_date or ""), + ).fetchall() + recorded_zones: Dict[str, list[str]] = {} + for local_date, timezone in rows: + recorded_zones.setdefault(local_date, []) + if timezone not in recorded_zones[local_date]: + recorded_zones[local_date].append(timezone) + + result: List[Dict[str, Any]] = [] + for local_date in sorted(local_dates): + timezones = recorded_zones.get(local_date) + if not timezones: + result.append({ + "day": local_date, + "local_label": local_date, + "timezone_label": "", + "identity": None, + "total_bytes": None, + "total_written": None, + "total_read": None, + "shared_bytes_written": 0, + "shared_bytes_read": 0, + "unallocated_bytes_written": 0, + "unallocated_bytes_read": 0, + "activity_state": "unavailable", + "is_zero": False, + "is_gap": True, + "is_partial": False, + }) + continue + + for timezone in timezones: + summary = query_local_day_summary( + conn, local_date, timezone, now, + ) + if summary is None: + continue + identity = HistoryDayIdentity( + local_date, + summary["tz_name"], + summary["utc_start"], + summary["utc_end"], + ) + written = summary["bytes_written"] + read = summary["bytes_read"] + state = summary["activity_state"] + result.append({ + "day": local_date, + "local_label": local_date, + "timezone_label": "%s %s" % ( + summary["tz_name"], summary["tz_offset"], + ), + "identity": identity, + "total_bytes": written, + "total_written": written, + "total_read": read, + "shared_bytes_written": summary["shared_bytes_written"], + "shared_bytes_read": summary["shared_bytes_read"], + "unallocated_bytes_written": summary["unallocated_bytes_written"], + "unallocated_bytes_read": summary["unallocated_bytes_read"], + "activity_state": state, + "is_zero": state == "zero", + "is_gap": written is None and read is None, + "is_partial": state in ("incomplete", "so_far"), + }) + + return result + + +def _query_local_hourly_graph_data( + conn: sqlite3.Connection, + summary: Dict[str, Any] | None, + now: datetime, +) -> List[Dict[str, Any]]: + """Read complete UTC hour buckets inside recorded local-day boundaries.""" + if summary is None: + return [] + + zone = ZoneInfo(summary["tz_name"]) + day_start = _parse_ts(summary["utc_start"]).astimezone(timezone.utc) + day_end = _parse_ts(summary["utc_end"]).astimezone(timezone.utc) + current = now.astimezone(timezone.utc) + first_hour = day_start.replace(minute=0, second=0, microsecond=0) + if first_hour < day_start: + first_hour += timedelta(hours=1) + + last_hour = day_end - timedelta(hours=1) + rows = conn.execute( + "SELECT hour, bytes_written_delta, bytes_read_delta, coverage, " + "sample_count, unknown_seconds FROM hour_observations " + "WHERE hour >= ? AND hour <= ? ORDER BY hour", + (first_hour.isoformat(), last_hour.isoformat()), + ).fetchall() + by_hour = { + _parse_ts(row[0]).astimezone(timezone.utc): row + for row in rows + } + + result: List[Dict[str, Any]] = [] + hour_start = first_hour + while hour_start + timedelta(hours=1) <= day_end: + hour_end = hour_start + timedelta(hours=1) + row = by_hour.get(hour_start) + local_start = hour_start.astimezone(zone) + local_end = hour_end.astimezone(zone) + is_future = hour_start > current + identity = hour_start.isoformat() + if row is None: + written = read = coverage = sample_count = 0 + unknown_seconds = 0 + is_gap = not is_future + else: + written = row[1] or 0 + read = row[2] or 0 + coverage = row[3] or 0.0 + sample_count = row[4] or 0 + unknown_seconds = row[5] or 0 + is_gap = sample_count == 0 and unknown_seconds > 0 + result.append({ + "hour": identity, + "local_label": local_start.strftime("%H:%M %z"), + "local_end_label": local_end.strftime("%H:%M %z"), + "timezone_label": "%s %s" % ( + summary["tz_name"], local_start.strftime("%z"), + ), + "utc_label": hour_start.strftime("%Y-%m-%d %H:%M UTC"), + "bytes_written": written, + "bytes_read": read, + "coverage": coverage, + "sample_count": sample_count, + "unknown_seconds": unknown_seconds, + "is_zero": written == 0 and read == 0 and not is_gap, + "is_gap": is_gap, + "is_partial": hour_start <= current < hour_end, + "is_future": is_future, + }) + hour_start = hour_end + + return result + + def _unit_scale(max_bytes: int) -> tuple[float, str]: if max_bytes >= 1_000_000_000_000: return 1e12, "TB" @@ -1513,17 +1904,12 @@ class FenrisTuiApp(App): def _set_activity_view(self, view: str) -> None: self._activity_view = view graph = self.query_one("#usage-history") - if view == "live": - self._browse_date = None - elif view == "day" and graph._day_data and graph.view_mode != "hourly": - if self._browse_date is not None: - graph.drill_day = self._browse_date - graph.view_mode = "hourly" - self._on_graph_drill(self._browse_date) - else: - graph._enter_drill() + if view == "day": + graph.show_selected_day() elif view == "history" and graph.view_mode == "hourly": graph._exit_drill() + elif view == "live" and graph.view_mode == "hourly": + graph._exit_drill() self._apply_activity_view() self._refresh() self.query_one("#live-activity" if view == "live" else "#usage-history").focus( @@ -1683,35 +2069,24 @@ class FenrisTuiApp(App): # --- Activity panel: live and historical volume plots --- graph = self.query_one("#usage-history") - day_data = _query_daily_graph_data(conn, self._clock_now) - - # Preserve drill-down state across refresh if still valid + day_data = _query_local_day_graph_data( + conn, self._clock_now, selected_date=self._browse_date, + ) + graph.set_data( + day_data, on_drill=self._on_graph_drill, + selected_date=self._browse_date, + ) if graph.view_mode == "hourly" and graph.drill_day: - saved_drill_day = graph.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, + from .local_day import query_local_day_summary + identity = self._activity_selection.selected_history_day + summary = query_local_day_summary( + conn, + graph.drill_day, + identity.timezone if identity is not None else None, + self._clock_now, ) - graph.set_data( - day_data, on_drill=self._on_graph_drill, - selected_date=self._browse_date, - ) - 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, - selected_date=self._browse_date, + graph.set_hour_data( + _query_local_hourly_graph_data(conn, summary, self._clock_now) ) # --- Drive-health pane (§7.2 right, issue #79) --- @@ -1788,8 +2163,12 @@ class FenrisTuiApp(App): ) tz_name = detect_system_tz() if self._browse_date is not None: + identity = self._activity_selection.selected_history_day local = query_local_day_summary( - conn, self._browse_date, clock_now=self._clock_now + conn, + self._browse_date, + identity.timezone if identity is not None else None, + self._clock_now, ) else: local = query_current_local_day(conn, self._clock_now, tz_name) @@ -1862,9 +2241,17 @@ class FenrisTuiApp(App): with read_status( self.store_path, self._clock_now, query_services=False, ) as (conn, _): + from .local_day import query_local_day_summary + identity = self._activity_selection.selected_history_day + summary = query_local_day_summary( + conn, + day, + identity.timezone if identity is not None else None, + self._clock_now, + ) if conn is not None else None graph.set_hour_data( - _query_hourly_graph_data(conn, day, self._clock_now) - if conn is not None else [] + _query_local_hourly_graph_data(conn, summary, self._clock_now) + if conn is not None and summary is not None else [] ) if conn is not None: self._render_local_day(conn) @@ -1948,21 +2335,27 @@ class FenrisTuiApp(App): def _browse_day(self, offset: int) -> None: """Move selection by *offset* days from the current position.""" graph = self.query_one("#usage-history") - if not graph._day_data: - return - # Find the currently selected day's date string - current_day = graph._day_data[graph.selected_index].get("day") + current_day = self._browse_date + if current_day is None and 0 <= graph.selected_index < len(graph._day_data): + current_day = graph._day_data[graph.selected_index].get("day") if current_day is None: return # Compute target date current_date = datetime.strptime(current_day, "%Y-%m-%d").date() target_date = current_date + timedelta(days=offset) + if offset > 0: + from .tz_util import detect_system_tz + today = self._clock_now.astimezone( + ZoneInfo(detect_system_tz()) + ).date() + if target_date > today: + return target_str = target_date.strftime("%Y-%m-%d") - # Navigate to target, adjusting range if needed - if graph.navigate_to_date(target_str): - self._browse_date = target_str if graph.view_mode == "hourly": graph._exit_drill() + # Navigate to target, adjusting range if needed + graph.navigate_to_date(target_str) + self._browse_date = target_str self._set_activity_view("day") def action_prev_day(self) -> None: diff --git a/tests/test_dashboard_design.py b/tests/test_dashboard_design.py index 36029cb..621a2dc 100644 --- a/tests/test_dashboard_design.py +++ b/tests/test_dashboard_design.py @@ -102,6 +102,7 @@ async def test_tabs_date_entry_and_hourly_inspection_keep_context(dashboard): app.on_refresh_tick() await pilot.pause() assert str(app.query_one("#bar-readout").render()) == readout + assert "Reads" in str(app.query_one("#bar-legend").render()) await pilot.press("g") await pilot.press(*list("2026-09-17")) await pilot.press("escape") diff --git a/tests/test_issue_92.py b/tests/test_issue_92.py index 56a407e..4160035 100644 --- a/tests/test_issue_92.py +++ b/tests/test_issue_92.py @@ -24,13 +24,12 @@ from fenris.store import init_store from fenris.monitoring_periods import ensure_period_open from fenris.local_day import ( LocalDaySummary, + local_day_boundaries, persist_local_day, - query_local_day_summary, ) +from fenris.tz_util import detect_system_tz from fenris.tui import ( FenrisTuiApp, - HistoryGraph, - _RANGE_OPTIONS, _RANGE_DEFAULT, ) @@ -70,13 +69,18 @@ def _open_period(conn, start="2026-09-01T00:00:00+00:00"): def _insert_local_day(conn, local_date, tz_name="UTC", tz_offset="+00:00", - bw=300, br=130, complete=True): + bw=300, br=130, complete=True, + utc_start=None, utc_end=None): + if utc_start is None: + utc_start = f"{local_date}T00:00:00+00:00" + if utc_end is None: + utc_end = f"{(datetime.fromisoformat(local_date) + timedelta(days=1)).strftime('%Y-%m-%d')}T00:00:00+00:00" summary = LocalDaySummary( local_date=local_date, tz_name=tz_name, tz_offset=tz_offset, - utc_start=f"{local_date}T00:00:00+00:00", - utc_end=f"{(datetime.fromisoformat(local_date) + timedelta(days=1)).strftime('%Y-%m-%d')}T00:00:00+00:00", + utc_start=utc_start, + utc_end=utc_end, bytes_written=bw, bytes_read=br, coverage=0.95, @@ -124,6 +128,10 @@ def _make_app(tmp_path, clock=None): return app, patcher +def _visible_selected_date(app): + return str(app.query_one("#local-day").render())[:10] + + # --------------------------------------------------------------------------- # AC92-1: Keyboard bindings exist # --------------------------------------------------------------------------- @@ -178,14 +186,13 @@ class TestBracketNavigation: app = FenrisTuiApp(store_path=tmp_path / "test.db", refresh_interval_s=999) async with app.run_test(size=(100, 40)) as pilot: await pilot.pause() - graph = app.query_one("#usage-history") # Move to an older day first so ] can advance for _ in range(3): await pilot.press("left_square_bracket") - day_before = graph._day_data[graph.selected_index].get("day") + day_before = _visible_selected_date(app) await pilot.press("right_square_bracket") - day_after = graph._day_data[graph.selected_index].get("day") + day_after = _visible_selected_date(app) # ] should move to a newer day (closer to today) assert day_after > day_before @@ -198,11 +205,10 @@ class TestBracketNavigation: app = FenrisTuiApp(store_path=tmp_path / "test.db", refresh_interval_s=999) async with app.run_test(size=(100, 40)) as pilot: await pilot.pause() - graph = app.query_one("#usage-history") - day_before = graph._day_data[graph.selected_index].get("day") + day_before = _visible_selected_date(app) await pilot.press("left_square_bracket") - day_after = graph._day_data[graph.selected_index].get("day") + day_after = _visible_selected_date(app) # [ should move to an older day (further from today) assert day_after < day_before @@ -224,9 +230,8 @@ class TestBracketNavigation: for _ in range(29): await pilot.press("left_square_bracket") - # Graph should now show a range that includes the oldest day - assert graph.selected_index >= 0 - selected_day = graph._day_data[graph.selected_index].get("day") + # Visible readout identifies oldest selected date. + selected_day = _visible_selected_date(app) oldest_day = (app_clock - timedelta(days=29)).strftime("%Y-%m-%d") assert selected_day == oldest_day finally: @@ -241,11 +246,10 @@ class TestBracketNavigation: app = FenrisTuiApp(store_path=tmp_path / "test.db", refresh_interval_s=999) async with app.run_test(size=(100, 40)) as pilot: await pilot.pause() - graph = app.query_one("#usage-history") - initial_idx = graph.selected_index + initial_date = _visible_selected_date(app) await pilot.press("right_square_bracket") - assert graph.selected_index == initial_idx + assert _visible_selected_date(app) == initial_date # --------------------------------------------------------------------------- @@ -262,36 +266,54 @@ class TestArrowInspection: app = FenrisTuiApp(store_path=tmp_path / "test.db", refresh_interval_s=999) async with app.run_test(size=(100, 40)) as pilot: await pilot.pause() - graph = app.query_one("#usage-history") await pilot.press("v", "v") - graph.focus() + app.query_one("#usage-history").focus() await pilot.pause() - day_before = graph._day_data[graph.selected_index].get("day") + day_before = _visible_selected_date(app) await pilot.press("left") - day_after = graph._day_data[graph.selected_index].get("day") + day_after = _visible_selected_date(app) assert day_after != day_before @pytest.mark.asyncio async def test_arrow_keys_inspect_hours_in_drill(self, tmp_path): """Arrow keys navigate hours when in hourly drill-down.""" conn, clock = _setup_store(tmp_path, n_days=14) + local_date = datetime.now().astimezone().date().isoformat() + timezone_name = detect_system_tz() + utc_start, utc_end, tz_offset = local_day_boundaries(local_date, timezone_name) + _insert_local_day( + conn, local_date, tz_name=timezone_name, tz_offset=tz_offset, + utc_start=utc_start.isoformat(), utc_end=utc_end.isoformat(), + ) + hour = utc_start.replace(minute=0, second=0, microsecond=0) + if hour < utc_start: + hour += timedelta(hours=1) + while hour + timedelta(hours=1) <= utc_end: + conn.execute( + "INSERT INTO hour_observations " + "(hour, bytes_written_delta, bytes_read_delta, coverage, sample_count) " + "VALUES (?, 1000000, 2000000, 1, 1)", + (hour.isoformat(),), + ) + hour += timedelta(hours=1) + conn.commit() conn.close() app = FenrisTuiApp(store_path=tmp_path / "test.db", refresh_interval_s=999) async with app.run_test(size=(100, 40)) as pilot: await pilot.pause() - graph = app.query_one("#usage-history") await pilot.press("v", "v") await pilot.click("#usage-history") # Enter drill-down await pilot.press("enter") - assert graph.view_mode == "hourly" + assert app.query_one("#activity-tabs").active == "view-day" # Arrow keys navigate hours + readout_before = str(app.query_one("#bar-readout").render()) await pilot.press("left") - assert graph._hourly_selected >= 0 + assert str(app.query_one("#bar-readout").render()) != readout_before # --------------------------------------------------------------------------- @@ -308,18 +330,17 @@ class TestTodayBinding: app = FenrisTuiApp(store_path=tmp_path / "test.db", refresh_interval_s=999) async with app.run_test(size=(100, 40)) as pilot: await pilot.pause() - graph = app.query_one("#usage-history") # Browse to an older day using [ for _ in range(5): await pilot.press("left_square_bracket") - browsed_day = graph._day_data[graph.selected_index].get("day") - today = clock.strftime("%Y-%m-%d") + browsed_day = _visible_selected_date(app) + today = datetime.now().astimezone().date().isoformat() assert browsed_day != today # Press t to return to today await pilot.press("t") - assert graph.selected_index == len(graph._day_data) - 1 + assert app.query_one("#activity-tabs").active == "view-live" @pytest.mark.asyncio async def test_t_exits_hourly_drill(self, tmp_path): @@ -330,17 +351,16 @@ class TestTodayBinding: app = FenrisTuiApp(store_path=tmp_path / "test.db", refresh_interval_s=999) async with app.run_test(size=(100, 40)) as pilot: await pilot.pause() - graph = app.query_one("#usage-history") await pilot.press("v", "v") await pilot.click("#usage-history") # Enter drill await pilot.press("enter") - assert graph.view_mode == "hourly" + assert app.query_one("#activity-tabs").active == "view-day" # t returns to daily await pilot.press("t") - assert graph.view_mode == "daily" + assert app.query_one("#activity-tabs").active == "view-live" # --------------------------------------------------------------------------- @@ -360,7 +380,7 @@ class TestDateEntry: await pilot.press("g") await pilot.pause() # Date picker should be pushed as a screen - assert len(app.screen_stack) > 1 + assert app.screen.query_one("#date-input") is not None @pytest.mark.asyncio async def test_valid_date_navigates(self, tmp_path): @@ -375,13 +395,12 @@ class TestDateEntry: await pilot.pause() target_date = (app_clock - timedelta(days=5)).strftime("%Y-%m-%d") - # Simulate the date picker callback directly - app._go_to_date_callback(target_date) + await pilot.press("g") + await pilot.press(*target_date) + await pilot.press("enter") await pilot.pause() - graph = app.query_one("#usage-history") - selected_day = graph._day_data[graph.selected_index].get("day") - assert selected_day == target_date + assert _visible_selected_date(app) == target_date finally: patcher.stop() @@ -400,7 +419,8 @@ class TestDateEntry: await pilot.press("enter") await pilot.pause() # Should remain on date picker (error shown, not dismissed) - assert len(app.screen_stack) > 1 + error = str(app.screen.query_one("#date-error").render()) + assert "Invalid date format" in error @pytest.mark.asyncio async def test_cancel_returns_to_prior(self, tmp_path): @@ -411,8 +431,7 @@ class TestDateEntry: app = FenrisTuiApp(store_path=tmp_path / "test.db", refresh_interval_s=999) async with app.run_test(size=(100, 40)) as pilot: await pilot.pause() - graph = app.query_one("#usage-history") - prior_idx = graph.selected_index + prior_readout = str(app.query_one("#bar-readout").render()) await pilot.press("g") await pilot.pause() @@ -420,7 +439,7 @@ class TestDateEntry: await pilot.pause() # Should be back to the graph with the same selection - assert graph.selected_index == prior_idx + assert str(app.query_one("#bar-readout").render()) == prior_readout # --------------------------------------------------------------------------- @@ -442,8 +461,6 @@ class TestLocalDayEvidence: try: async with app.run_test(size=(100, 40)) as pilot: await pilot.pause() - graph = app.query_one("#usage-history") - # Navigate to the target day for _ in range(4): await pilot.press("left_square_bracket") @@ -489,12 +506,10 @@ class TestBrowsingStability: app = FenrisTuiApp(store_path=tmp_path / "test.db", refresh_interval_s=0.1) async with app.run_test(size=(100, 40)) as pilot: await pilot.pause() - graph = app.query_one("#usage-history") - # Browse to a specific day for _ in range(3): - await pilot.press("right_square_bracket") - browsed_day = graph._day_data[graph.selected_index].get("day") + await pilot.press("left_square_bracket") + browsed_day = _visible_selected_date(app) # Wait for at least one refresh tick import asyncio @@ -502,8 +517,7 @@ class TestBrowsingStability: await pilot.pause() # Selection should be preserved - assert graph.selected_index >= 0 - assert graph._day_data[graph.selected_index].get("day") == browsed_day + assert _visible_selected_date(app) == browsed_day @pytest.mark.asyncio async def test_navigate_away_does_not_jump_on_refresh(self, tmp_path): @@ -514,14 +528,12 @@ class TestBrowsingStability: app = FenrisTuiApp(store_path=tmp_path / "test.db", refresh_interval_s=0.1) async with app.run_test(size=(100, 40)) as pilot: await pilot.pause() - graph = app.query_one("#usage-history") - # Browse to older day using [ for _ in range(5): await pilot.press("left_square_bracket") - browsed_day = graph._day_data[graph.selected_index].get("day") - today = clock.strftime("%Y-%m-%d") + browsed_day = _visible_selected_date(app) + today = datetime.now().astimezone().date().isoformat() assert browsed_day != today # Wait for refresh @@ -530,8 +542,8 @@ class TestBrowsingStability: await pilot.pause() # Should NOT jump back to today - selected_day = graph._day_data[graph.selected_index].get("day") - assert selected_day != today + assert _visible_selected_date(app) == browsed_day + assert _visible_selected_date(app) != today # --------------------------------------------------------------------------- @@ -548,17 +560,16 @@ class TestDrillDownData: app = FenrisTuiApp(store_path=tmp_path / "test.db", refresh_interval_s=999) async with app.run_test(size=(100, 40)) as pilot: await pilot.pause() - graph = app.query_one("#usage-history") await pilot.press("v", "v") await pilot.click("#usage-history") # Enter drill-down await pilot.press("enter") - assert graph.view_mode == "hourly" + assert app.query_one("#activity-tabs").active == "view-day" - # After drill callback, data should be populated - # (empty since we have no hour_observations) - assert graph.view_mode == "hourly" + assert "hourly evidence unavailable" in str( + app.query_one("#bar-readout").render() + ) # --------------------------------------------------------------------------- @@ -575,11 +586,10 @@ class TestConstrainedWidth: app = FenrisTuiApp(store_path=tmp_path / "test.db", refresh_interval_s=999) async with app.run_test(size=(80, 24)) as pilot: await pilot.pause() - graph = app.query_one("#usage-history") - day_before = graph._day_data[graph.selected_index].get("day") + day_before = _visible_selected_date(app) await pilot.press("left_square_bracket") - day_after = graph._day_data[graph.selected_index].get("day") + day_after = _visible_selected_date(app) assert day_after < day_before @pytest.mark.asyncio @@ -593,4 +603,225 @@ class TestConstrainedWidth: await pilot.pause() await pilot.press("g") await pilot.pause() - assert len(app.screen_stack) > 1 + assert app.screen.query_one("#date-input") is not None + + +class TestClickableNavigation: + @pytest.mark.asyncio + @pytest.mark.parametrize("size", [(80, 24), (70, 20)]) + async def test_click_and_keyboard_select_same_local_date(self, tmp_path, size): + """Clickable and keyboard navigation share date-entry transitions.""" + app_clock = _clock() + conn, _ = _setup_store(tmp_path, n_days=14, today=app_clock) + conn.close() + app, patcher = _make_app(tmp_path, clock=app_clock) + expected = (app_clock - timedelta(days=1)).strftime("%Y-%m-%d") + direct = (app_clock - timedelta(days=5)).strftime("%Y-%m-%d") + try: + async with app.run_test(size=size) as pilot: + await pilot.press("left_square_bracket") + keyboard_date = _visible_selected_date(app) + assert keyboard_date == expected + + await pilot.press("t") + await pilot.click("#activity-tools", offset=(2, 0)) + assert _visible_selected_date(app) == keyboard_date + + await pilot.press("g") + await pilot.press(*direct) + await pilot.press("enter") + await pilot.pause() + assert _visible_selected_date(app) == direct + finally: + patcher.stop() + + +class TestHistoricalSelectionIdentity: + @pytest.mark.asyncio + async def test_unavailable_entered_date_stays_visible_after_refresh(self, tmp_path): + """An unavailable requested date remains selected and explained.""" + conn, clock = _setup_store(tmp_path, n_days=30) + conn.close() + app, patcher = _make_app(tmp_path, clock=clock) + target = (clock - timedelta(days=120)).strftime("%Y-%m-%d") + try: + async with app.run_test(size=(100, 40)) as pilot: + await pilot.press("g") + await pilot.press(*target) + await pilot.press("enter") + await pilot.pause() + + local_day = str(app.query_one("#local-day").render()) + assert target in local_day + assert "unavailable" in local_day + + app.on_refresh_tick() + await pilot.pause() + local_day = str(app.query_one("#local-day").render()) + assert target in local_day + assert "unavailable" in local_day + finally: + patcher.stop() + + @pytest.mark.asyncio + async def test_retained_summary_outside_recent_window_remains_available(self, tmp_path): + """Direct date entry can inspect durable local summaries beyond 90 days.""" + clock = _clock() + conn, _ = _setup_store(tmp_path, n_days=30, today=clock) + target = (clock - timedelta(days=120)).strftime("%Y-%m-%d") + utc_start, utc_end, tz_offset = local_day_boundaries( + target, "Asia/Kolkata", + ) + _insert_local_day( + conn, + target, + tz_name="Asia/Kolkata", + tz_offset=tz_offset, + bw=2_000_000_000, + br=1_000_000_000, + utc_start=utc_start.isoformat(), + utc_end=utc_end.isoformat(), + ) + conn.close() + app, patcher = _make_app(tmp_path, clock=clock) + try: + async with app.run_test(size=(100, 40)) as pilot: + await pilot.press("g") + await pilot.press(*target) + await pilot.press("enter") + await pilot.pause() + await pilot.press("v") + await pilot.pause() + + readout = str(app.query_one("#bar-readout").render()) + assert target in readout + assert "Asia/Kolkata +05:30" in readout + assert "W 2.000 GB known · R 1.000 GB known" in readout + finally: + patcher.stop() + + @pytest.mark.asyncio + async def test_constrained_history_keeps_known_totals_and_evidence_state(self, tmp_path): + """Small textual view retains selected day, totals, and shared evidence.""" + clock = _clock() + conn, _ = _setup_store(tmp_path, n_days=14, today=clock) + target = (clock - timedelta(days=2)).strftime("%Y-%m-%d") + utc_start, utc_end, tz_offset = local_day_boundaries( + target, "Asia/Kolkata", + ) + _insert_local_day( + conn, + target, + tz_name="Asia/Kolkata", + tz_offset=tz_offset, + bw=1_500_000_000, + br=500_000_000, + utc_start=utc_start.isoformat(), + utc_end=utc_end.isoformat(), + ) + conn.execute( + "INSERT INTO local_day_unallocated_evidence " + "(start_sample_id, end_sample_id, start_local_date, end_local_date, " + " start_tz_name, end_tz_name, started_at, ended_at, bytes_written, " + " bytes_read, reason) VALUES (1, 2, ?, ?, ?, ?, ?, ?, ?, ?, 'local_midnight')", + (target, "2026-09-29", "Asia/Kolkata", "Asia/Kolkata", + utc_start.isoformat(), utc_start.isoformat(), 200_000_000, 300_000_000), + ) + conn.commit() + conn.close() + app, patcher = _make_app(tmp_path, clock=clock) + try: + async with app.run_test(size=(70, 20)) as pilot: + await pilot.press("g") + await pilot.press(*target) + await pilot.press("enter") + await pilot.press("v") + await pilot.pause() + + readout = str(app.query_one("#bar-readout").render()) + assert target in readout + assert "Asia/Kolkata +05:30" in readout + assert "incomplete" in readout + assert "W 1.500 GB known · R 0.500 GB known" in readout + assert "shared at midnight W 0.200 GB · R 0.300 GB" in readout + finally: + patcher.stop() + + @pytest.mark.asyncio + async def test_date_picker_keystrokes_do_not_toggle_measurement(self, tmp_path): + """Typing in date entry does not trigger dashboard shortcuts.""" + conn, _ = _setup_store(tmp_path, n_days=14) + conn.close() + app = FenrisTuiApp(store_path=tmp_path / "test.db", refresh_interval_s=999) + async with app.run_test(size=(100, 40)) as pilot: + await pilot.pause() + await pilot.press("g") + await pilot.press("w") + await pilot.press("escape") + await pilot.pause() + + header = str(app.query_one("#live-header").render()) + assert "Writes" in header + + @pytest.mark.asyncio + async def test_historical_hour_uses_recorded_timezone_and_utc_identity(self, tmp_path): + """Repeated local clock labels retain distinct UTC-hour identities.""" + clock = _clock(2026, 11, 2, 12) + conn, _ = _setup_store(tmp_path, n_days=30, today=clock) + target = "2026-11-01" + _insert_local_day( + conn, + target, + tz_name="America/New_York", + tz_offset="-04:00", + utc_start="2026-11-01T04:00:00+00:00", + utc_end="2026-11-02T05:00:00+00:00", + ) + conn.execute( + "INSERT INTO local_day_unallocated_evidence " + "(start_sample_id, end_sample_id, start_local_date, end_local_date, " + " start_tz_name, end_tz_name, started_at, ended_at, bytes_written, " + " bytes_read, reason) VALUES (1, 2, ?, ?, ?, ?, ?, ?, ?, ?, 'local_midnight')", + (target, "2026-11-02", "America/New_York", "America/New_York", + "2026-11-02T03:59:00+00:00", "2026-11-02T04:01:00+00:00", + 2_000_000_000, 3_000_000_000), + ) + for index in range(25): + hour = datetime(2026, 11, 1, 4, tzinfo=timezone.utc) + timedelta(hours=index) + conn.execute( + "INSERT INTO hour_observations " + "(hour, bytes_written_delta, bytes_read_delta, coverage, sample_count) " + "VALUES (?, ?, ?, 1, 1)", + (hour.isoformat(), (index + 1) * 1_000_000_000, + (index + 1) * 2_000_000_000), + ) + conn.commit() + conn.close() + app, patcher = _make_app(tmp_path, clock=clock) + try: + async with app.run_test(size=(100, 40)) as pilot: + await pilot.click("#view-history") + await pilot.press("g") + await pilot.press(*target) + await pilot.press("enter") + await pilot.pause() + + local_day = str(app.query_one("#local-day").render()) + assert "America/New_York -04:00" in local_day + assert "incomplete" in local_day + assert "shared at midnight W 2.000 GB · R 3.000 GB" in local_day + + # Move from final local hour to the second 01:00 occurrence. + await pilot.press(*(["left"] * 22)) + second_occurrence = str(app.query_one("#bar-readout").render()) + assert "01:00 -0500" in second_occurrence + assert "America/New_York -0500" in second_occurrence + assert "06:00 UTC" in second_occurrence + + await pilot.press("left") + first_occurrence = str(app.query_one("#bar-readout").render()) + assert "01:00 -0400" in first_occurrence + assert "America/New_York -0400" in first_occurrence + assert "05:00 UTC" in first_occurrence + finally: + patcher.stop() diff --git a/tests/test_tui.py b/tests/test_tui.py index 50a9ed6..5f74aa4 100644 --- a/tests/test_tui.py +++ b/tests/test_tui.py @@ -927,6 +927,7 @@ class TestBarGraphTUI: 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_local_day(conn, d, bw=1024*1024*100) _insert_sample(conn, "2026-09-30T10:00:00+00:00") conn.close() @@ -941,11 +942,9 @@ class TestBarGraphTUI: assert "Unalloc" 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 + assert "14 local days" in range_label plotted = str(app.query_one("#bar-render").render()) assert "Writes (" in legend - assert "UTC" in range_label assert "─" in plotted @pytest.mark.asyncio @@ -957,6 +956,7 @@ class TestBarGraphTUI: 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_local_day(conn, d, bw=1024*1024*100) _insert_sample(conn, "2026-09-30T10:00:00+00:00") conn.close() @@ -1030,6 +1030,7 @@ class TestBarGraphTUI: 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_local_day(conn, d, bw=1024*1024*100) _insert_sample(conn, "2026-09-30T10:00:00+00:00") conn.close() @@ -1043,15 +1044,16 @@ class TestBarGraphTUI: # Newest day is selected initially. readout = str(app.query_one("#bar-readout").render()) - assert "UTC" in readout - assert "GB" in readout + assert "so_far" in readout + assert "W 0.105 GB known" in readout # 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 - assert "GB" in readout + assert "incomplete" in readout + assert "W 0.105 GB known" in readout @pytest.mark.asyncio async def test_hourly_drill_down_and_back(self, tmp_path): @@ -1062,6 +1064,7 @@ class TestBarGraphTUI: 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_local_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) @@ -1078,30 +1081,27 @@ class TestBarGraphTUI: await pilot.pause() await pilot.pause() - # The newest day is selected automatically. - assert graph.selected_index == len(graph._day_data) - 1 - assert graph.view_mode == "daily" + assert app.query_one("#activity-tabs").active == "view-history" + daily_readout = str(app.query_one("#bar-readout").render()) # 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 + assert app.query_one("#activity-tabs").active == "view-day" + assert "UTC" in str(app.query_one("#bar-readout").render()) - # Five-minute data refreshes retain the selected hour. - graph._hourly_selected = 12 - selected_hour = graph._hour_data[12]["hour"] + # Inspection identity survives refresh through visible readout. + await pilot.press("left") + selected_hour = str(app.query_one("#bar-readout").render()) app._refresh() await pilot.pause() - assert graph.view_mode == "hourly" - assert graph._hour_data[graph._hourly_selected]["hour"] == selected_hour + assert str(app.query_one("#bar-readout").render()) == selected_hour # Esc returns to daily await pilot.press("escape") await pilot.pause() - assert graph.view_mode == "daily" - assert graph.drill_day is None + assert app.query_one("#activity-tabs").active == "view-history" + assert str(app.query_one("#bar-readout").render()) == daily_readout @pytest.mark.asyncio async def test_empty_store_graph(self, tmp_path):