426 lines
17 KiB
Python
426 lines
17 KiB
Python
"""Tests for live activity graph (issue #91).
|
||
|
||
Covers:
|
||
- Live interval volume query from raw samples
|
||
- LiveActivityGraph widget rendering, toggle, and inspection
|
||
- Three-minute cadence constants
|
||
- TUI integration with live graph
|
||
"""
|
||
import sys
|
||
from datetime import datetime, timedelta, timezone
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
|
||
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
|
||
|
||
from fenris.activity_selection import IntervalIdentity
|
||
from fenris.monitoring_periods import ensure_period_open
|
||
from fenris.status import ACCURACY_SEC, CADENCE_DEFAULT_S, FRESH_THRESHOLD_S
|
||
from fenris.store import init_store
|
||
from fenris.tui import (
|
||
LIVE_WINDOW_H,
|
||
FenrisTuiApp,
|
||
LiveActivityGraph,
|
||
_query_live_graph_data,
|
||
)
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Helpers
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def _clock(year=2026, month=9, day=30, hour=12):
|
||
return datetime(year, month, day, hour, 0, 0, tzinfo=timezone.utc)
|
||
|
||
|
||
def _insert_segment(conn, opened_at="2026-09-01T00:00:00+00:00",
|
||
identity_key="nqn.test", degraded=False,
|
||
mn="Samsung SSD 970 EVO Plus 1TB"):
|
||
conn.execute(
|
||
"INSERT INTO controller_segments "
|
||
"(opened_at, identity_key, identity_degraded, subnqn, sn, mn, fr, vid, ssvid, transport) "
|
||
"VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
|
||
(opened_at, identity_key, degraded, "nqn.test", "SN123", mn, "FW1",
|
||
"0x144d", "0x144d", "pcie"),
|
||
)
|
||
conn.commit()
|
||
|
||
|
||
def _insert_sample(conn, ts, bw=512000000000, br=256000000000,
|
||
segment_id=None, pu=5, device="/dev/nvme0n1"):
|
||
conn.execute(
|
||
"INSERT INTO samples (ts, device, data_units_written, data_units_read, "
|
||
"percentage_used, bytes_written, bytes_read, power_on_hours, segment_id) "
|
||
"VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
|
||
(ts, device, 1000000, 500000, pu, bw, br, 8765, segment_id),
|
||
)
|
||
conn.commit()
|
||
|
||
|
||
def _open_period(conn, start="2026-09-01T00:00:00+00:00"):
|
||
ensure_period_open(conn, datetime.fromisoformat(start))
|
||
|
||
|
||
def _live_intervals(count=4):
|
||
first = datetime(2026, 9, 28, 12, 0, tzinfo=timezone.utc)
|
||
intervals = []
|
||
for index in range(count):
|
||
start = first + timedelta(minutes=3 * index)
|
||
end = start + timedelta(minutes=3)
|
||
intervals.append({
|
||
"start_ts": start.isoformat(),
|
||
"end_ts": end.isoformat(),
|
||
"start_label": start.strftime("%H:%M"),
|
||
"end_label": end.strftime("%H:%M"),
|
||
"bytes_written": (index + 1) * 1000,
|
||
"bytes_read": (index + 1) * 500,
|
||
"elapsed_s": 180,
|
||
"is_gap": False,
|
||
"is_zero": False,
|
||
"is_segment_boundary": False,
|
||
})
|
||
return intervals
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Cadence constants (issue #91 AC1)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class TestCadenceConstants:
|
||
def test_cadence_default_is_3_minutes(self):
|
||
"""Default collection cadence is 3 minutes (180 s)."""
|
||
assert CADENCE_DEFAULT_S == 180
|
||
|
||
def test_fresh_threshold_uses_3min_cadence(self):
|
||
"""Fresh threshold = 2 × 3min + AccuracySec + 60s = 450s."""
|
||
expected = 2 * 180 + ACCURACY_SEC + 60
|
||
assert FRESH_THRESHOLD_S == expected
|
||
|
||
def test_live_window_is_3_hours(self):
|
||
"""Live graph window is 3 hours."""
|
||
assert LIVE_WINDOW_H == 3
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Live graph query (issue #91 AC2-5)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class TestQueryLiveData:
|
||
def test_empty_store_returns_empty(self, tmp_path):
|
||
"""No samples → empty list."""
|
||
conn = init_store(tmp_path / "test.db")
|
||
now = _clock()
|
||
result = _query_live_graph_data(conn, now)
|
||
assert result == []
|
||
conn.close()
|
||
|
||
def test_single_sample_returns_empty(self, tmp_path):
|
||
"""One sample (no pair) → empty list."""
|
||
conn = init_store(tmp_path / "test.db")
|
||
now = _clock()
|
||
_insert_sample(conn, now.isoformat())
|
||
result = _query_live_graph_data(conn, now)
|
||
assert result == []
|
||
conn.close()
|
||
|
||
def test_two_samples_in_window(self, tmp_path):
|
||
"""Two samples within 3h window produce one interval."""
|
||
conn = init_store(tmp_path / "test.db")
|
||
now = _clock()
|
||
t1 = (now - timedelta(minutes=3)).isoformat()
|
||
t2 = (now - timedelta(minutes=0)).isoformat()
|
||
_insert_sample(conn, t1, bw=1000, br=500)
|
||
_insert_sample(conn, t2, bw=1500, br=700)
|
||
result = _query_live_graph_data(conn, now)
|
||
assert len(result) == 1
|
||
assert result[0]["bytes_written"] == 500
|
||
assert result[0]["bytes_read"] == 200
|
||
assert result[0]["elapsed_s"] == 180
|
||
assert result[0]["is_gap"] is False
|
||
assert result[0]["is_zero"] is False
|
||
conn.close()
|
||
|
||
def test_samples_outside_window_ignored(self, tmp_path):
|
||
"""Samples older than 3h are excluded."""
|
||
conn = init_store(tmp_path / "test.db")
|
||
now = _clock()
|
||
old = (now - timedelta(hours=4)).isoformat()
|
||
recent = (now - timedelta(minutes=1)).isoformat()
|
||
_insert_sample(conn, old, bw=1000, br=500)
|
||
_insert_sample(conn, recent, bw=1500, br=700)
|
||
result = _query_live_graph_data(conn, now)
|
||
assert len(result) == 0
|
||
conn.close()
|
||
|
||
def test_gap_detection(self, tmp_path):
|
||
"""Interval > 3× cadence is flagged as a gap."""
|
||
conn = init_store(tmp_path / "test.db")
|
||
now = _clock()
|
||
t1 = (now - timedelta(minutes=13)).isoformat()
|
||
t2 = (now - timedelta(minutes=3)).isoformat()
|
||
_insert_sample(conn, t1, bw=1000, br=500)
|
||
_insert_sample(conn, t2, bw=1500, br=700)
|
||
result = _query_live_graph_data(conn, now)
|
||
assert len(result) == 1
|
||
assert result[0]["is_gap"] is True
|
||
conn.close()
|
||
|
||
def test_zero_interval(self, tmp_path):
|
||
"""No byte delta → is_zero flag."""
|
||
conn = init_store(tmp_path / "test.db")
|
||
now = _clock()
|
||
t1 = (now - timedelta(minutes=3)).isoformat()
|
||
t2 = (now - timedelta(minutes=0)).isoformat()
|
||
_insert_sample(conn, t1, bw=1000, br=500)
|
||
_insert_sample(conn, t2, bw=1000, br=500)
|
||
result = _query_live_graph_data(conn, now)
|
||
assert len(result) == 1
|
||
assert result[0]["bytes_written"] == 0
|
||
assert result[0]["bytes_read"] == 0
|
||
assert result[0]["is_zero"] is True
|
||
conn.close()
|
||
|
||
def test_segment_boundary_resets_counters(self, tmp_path):
|
||
"""Counter discontinuity across segments → zero delta."""
|
||
conn = init_store(tmp_path / "test.db")
|
||
_insert_segment(conn, identity_key="key_a")
|
||
_insert_segment(conn, opened_at="2026-09-30T11:00:00+00:00",
|
||
identity_key="key_b")
|
||
now = _clock()
|
||
t1 = (now - timedelta(minutes=6)).isoformat()
|
||
t2 = (now - timedelta(minutes=3)).isoformat()
|
||
_insert_sample(conn, t1, bw=1000, br=500, segment_id=1)
|
||
_insert_sample(conn, t2, bw=500, br=200, segment_id=2)
|
||
result = _query_live_graph_data(conn, now)
|
||
assert len(result) == 1
|
||
assert result[0]["bytes_written"] == 0
|
||
assert result[0]["bytes_read"] == 0
|
||
assert result[0]["is_segment_boundary"] is True
|
||
conn.close()
|
||
|
||
def test_actual_timestamps_used(self, tmp_path):
|
||
"""Points use actual sample timestamps, not 3-min spacing."""
|
||
conn = init_store(tmp_path / "test.db")
|
||
now = _clock()
|
||
t1 = (now - timedelta(minutes=5)).isoformat()
|
||
t2 = (now - timedelta(minutes=1)).isoformat()
|
||
_insert_sample(conn, t1, bw=1000, br=500)
|
||
_insert_sample(conn, t2, bw=2000, br=800)
|
||
result = _query_live_graph_data(conn, now)
|
||
assert len(result) == 1
|
||
assert result[0]["start_ts"] == t1
|
||
assert result[0]["end_ts"] == t2
|
||
assert result[0]["elapsed_s"] == 240
|
||
conn.close()
|
||
|
||
def test_multiple_intervals_ordered_oldest_first(self, tmp_path):
|
||
"""Multiple intervals are returned in chronological order."""
|
||
conn = init_store(tmp_path / "test.db")
|
||
now = _clock()
|
||
for i in range(5):
|
||
ts = (now - timedelta(minutes=15 - i * 3)).isoformat()
|
||
_insert_sample(conn, ts, bw=1000 * (i + 1), br=500 * (i + 1))
|
||
result = _query_live_graph_data(conn, now)
|
||
assert len(result) == 4
|
||
for i in range(len(result) - 1):
|
||
assert result[i]["end_ts"] <= result[i + 1]["start_ts"]
|
||
conn.close()
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# LiveActivityGraph widget (issue #91 AC3)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class TestLiveActivityGraph:
|
||
@pytest.mark.asyncio
|
||
async def test_toggle_measure(self, tmp_path):
|
||
"""Toggle switches between written and read."""
|
||
app = FenrisTuiApp(store_path=tmp_path / "test.db")
|
||
async with app.run_test() as pilot:
|
||
await pilot.pause()
|
||
graph = app.query_one("#live-activity")
|
||
assert graph.measure == "written"
|
||
graph.toggle_measure()
|
||
assert graph.measure == "read"
|
||
graph.toggle_measure()
|
||
assert graph.measure == "written"
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_empty_data_renders_awaiting(self, tmp_path):
|
||
"""Empty data shows awaiting message."""
|
||
app = FenrisTuiApp(store_path=tmp_path / "test.db")
|
||
async with app.run_test() as pilot:
|
||
await pilot.pause()
|
||
graph = app.query_one("#live-activity")
|
||
graph.set_data([])
|
||
await pilot.pause()
|
||
render = str(graph.query_one("#live-render").render())
|
||
assert "Awaiting intervals" in render
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_data_renders_header(self, tmp_path):
|
||
"""Data updates the header with interval count."""
|
||
app = FenrisTuiApp(store_path=tmp_path / "test.db")
|
||
async with app.run_test() as pilot:
|
||
await pilot.pause()
|
||
graph = app.query_one("#live-activity")
|
||
data = [
|
||
{
|
||
"start_ts": "2026-09-30T11:57:00+00:00",
|
||
"end_ts": "2026-09-30T12:00:00+00:00",
|
||
"start_label": "11:57",
|
||
"end_label": "12:00",
|
||
"bytes_written": 500000000,
|
||
"bytes_read": 200000000,
|
||
"elapsed_s": 180,
|
||
"is_gap": False,
|
||
"is_zero": False,
|
||
"is_segment_boundary": False,
|
||
},
|
||
]
|
||
graph.set_data(data)
|
||
await pilot.pause()
|
||
header = str(graph.query_one("#live-header").render())
|
||
assert "1 intervals" in header
|
||
assert "w W/R" in str(app.query_one("#activity-tools").render())
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# TUI integration (issue #91 AC2, AC6-9)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class TestTUILiveIntegration:
|
||
@pytest.mark.asyncio
|
||
async def test_live_graph_exists_in_compose(self, tmp_path):
|
||
"""Live activity graph is composed in the TUI."""
|
||
app = FenrisTuiApp(store_path=tmp_path / "test.db")
|
||
async with app.run_test() as pilot:
|
||
await pilot.pause()
|
||
assert app.query_one("#live-activity") is not None
|
||
assert isinstance(app.query_one("#live-activity"), LiveActivityGraph)
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_live_graph_has_border_title(self, tmp_path):
|
||
"""Live activity graph has a border title."""
|
||
app = FenrisTuiApp(store_path=tmp_path / "test.db")
|
||
async with app.run_test() as pilot:
|
||
await pilot.pause()
|
||
live = app.query_one("#live-activity")
|
||
assert live.border_title == "Live activity"
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_today_action_resets_drill(self, tmp_path):
|
||
"""Action today returns from hourly drill to daily view."""
|
||
app = FenrisTuiApp(store_path=tmp_path / "test.db")
|
||
async with app.run_test() as pilot:
|
||
await pilot.pause()
|
||
graph = app.query_one("#usage-history")
|
||
graph.view_mode = "hourly"
|
||
graph.drill_day = "2026-09-30"
|
||
app.action_today()
|
||
await pilot.pause()
|
||
assert graph.view_mode == "daily"
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_toggle_measure_binding(self, tmp_path):
|
||
"""The w key toggles the live graph measure."""
|
||
app = FenrisTuiApp(store_path=tmp_path / "test.db")
|
||
async with app.run_test() as pilot:
|
||
await pilot.pause()
|
||
graph = app.query_one("#live-activity")
|
||
assert graph.measure == "written"
|
||
await pilot.press("w")
|
||
await pilot.pause()
|
||
assert graph.measure == "read"
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_constrained_mode_hides_live_graph(self, tmp_path):
|
||
"""Constrained terminal hides the live graph."""
|
||
app = FenrisTuiApp(store_path=tmp_path / "test.db")
|
||
async with app.run_test(size=(70, 20)) as pilot:
|
||
await pilot.pause()
|
||
main_grid = app.query_one("#main-grid")
|
||
assert main_grid.has_class("constrained")
|
||
|
||
@pytest.mark.asyncio
|
||
@pytest.mark.parametrize("size", [(100, 40), (70, 20)])
|
||
async def test_follow_pin_expiry_and_today_use_visible_readout(self, tmp_path, size):
|
||
"""Live selection follows, pins, expires, and resets at both layout sizes."""
|
||
app = FenrisTuiApp(store_path=tmp_path / "test.db", refresh_interval_s=999)
|
||
intervals = _live_intervals()
|
||
async with app.run_test(size=size) as pilot:
|
||
await pilot.pause()
|
||
graph = app.query_one("#live-activity")
|
||
graph.set_data(intervals[:3])
|
||
|
||
readout = str(app.query_one("#live-readout").render())
|
||
assert "12:09 UTC" in readout
|
||
assert graph.selection.following_live
|
||
assert graph.selection.selected_live_interval == IntervalIdentity(
|
||
intervals[2]["start_ts"], intervals[2]["end_ts"]
|
||
)
|
||
|
||
# Refresh while following selects the newly measured newest point.
|
||
graph.set_data(intervals)
|
||
assert "12:12 UTC" in str(app.query_one("#live-readout").render())
|
||
|
||
# Live graph receives focus at launch; first arrow pins immediately.
|
||
await pilot.press("left")
|
||
pinned = graph.selection.selected_live_interval
|
||
readout = str(app.query_one("#live-readout").render())
|
||
assert "12:09 UTC" in readout
|
||
assert not graph.selection.following_live
|
||
|
||
graph.set_data(intervals)
|
||
readout = str(app.query_one("#live-readout").render())
|
||
assert "12:09 UTC" in readout
|
||
assert graph.selection.selected_live_interval == pinned
|
||
|
||
# Read/write changes presentation only; pinned point stays stable.
|
||
await pilot.press("w")
|
||
graph.set_data(intervals)
|
||
assert graph.measure == "read"
|
||
assert graph.selection.selected_live_interval == pinned
|
||
assert "12:09 UTC" in str(app.query_one("#live-readout").render())
|
||
|
||
# Remove pinned interval from rolling window. Explain expiration.
|
||
graph.set_data(intervals[3:])
|
||
readout = str(app.query_one("#live-readout").render())
|
||
assert "Inspected interval expired; following live" in readout
|
||
assert "12:12 UTC" in readout
|
||
assert graph.selection.following_live
|
||
|
||
# Today action clears expiration and resumes newest-point following.
|
||
await pilot.press("t")
|
||
assert graph.selection.following_live
|
||
assert not graph.selection.live_interval_expired
|
||
graph.set_data(intervals)
|
||
readout = str(app.query_one("#live-readout").render())
|
||
assert "12:12 UTC" in readout
|
||
assert "Inspected interval expired" not in readout
|
||
assert graph.selection.following_live
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_mouse_inspection_pins_same_interval_identity(self, tmp_path):
|
||
"""Mouse inspection pins the interval at its visible graph location."""
|
||
app = FenrisTuiApp(store_path=tmp_path / "test.db", refresh_interval_s=999)
|
||
intervals = _live_intervals()
|
||
async with app.run_test(size=(100, 40)) as pilot:
|
||
await pilot.pause()
|
||
graph = app.query_one("#live-activity")
|
||
graph.set_data(intervals[:3])
|
||
await pilot.pause()
|
||
render = app.query_one("#live-render")
|
||
await pilot.click(
|
||
"#live-render",
|
||
offset=(8, max(0, render.content_size.height // 2)),
|
||
)
|
||
await pilot.pause()
|
||
|
||
expected = IntervalIdentity(
|
||
intervals[0]["start_ts"], intervals[0]["end_ts"]
|
||
)
|
||
assert graph.selection.selected_live_interval == expected
|
||
assert not graph.selection.following_live
|
||
assert "12:03 UTC" in str(app.query_one("#live-readout").render())
|