xavierk/Fenris
1
0
Fork 0
Code Issues 7 Pull Requests Actions Packages Projects Releases 11 Wiki Activity

Preserve historical activity selection (#103)

Browse Source
This commit is contained in:
xavierk
2026-09-28 17:46:18 +05:30
parent f780212a02
commit c4524cf654
5 changed files with 847 additions and 198 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_issue_92.py
+296 -65
View File
@@ -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()