"""Exercise shared status acquisition through the CLI and running TUI.""" import sqlite3 import sys from datetime import datetime, timezone from pathlib import Path from unittest.mock import patch import pytest sys.path.insert(0, str(Path(__file__).parent.parent / "src")) from fenris.status import get_status, get_status_composition, read_status from fenris.status_composition import StatusState from fenris.store import init_store, SCHEMA_VERSION from fenris.tui import FenrisTuiApp NOW = datetime(2026, 9, 16, 12, tzinfo=timezone.utc) ACTIVE = {"boot_enabled": True, "timer_active": True, "last_collect_ok": True} DISABLED = {"boot_enabled": False, "timer_active": False, "last_collect_ok": None} @pytest.mark.asyncio @pytest.mark.parametrize("kind", ["missing", "corrupt", "newer", "incomplete", "denied"]) @pytest.mark.parametrize("service", [ACTIVE, DISABLED, {}]) async def test_cli_tui_acquisition_parity(tmp_path, kind, service): path = tmp_path / "observations.db" if kind == "corrupt": path.write_bytes(b"Not a SQLite database") elif kind == "incomplete": sqlite3.connect(path).close() elif kind != "missing": conn = init_store(path) if kind == "newer": conn.execute("PRAGMA user_version = %d" % (SCHEMA_VERSION + 1)) conn.close() original_exists = Path.exists def exists(target): if kind == "denied" and target == path: raise PermissionError("Observation store access denied") return original_exists(target) before = path.read_bytes() if original_exists(path) else None with patch("fenris.status.query_service_state", return_value=service), \ patch("fenris.status._journalctl_hint", return_value="[bold]Native collector log[/bold]"), \ patch("fenris.tui._journalctl_hint", return_value="[bold]Native collector log[/bold]"), \ patch.object(Path, "exists", exists), \ patch("fenris.tui.compute_projection") as projection: comp = get_status_composition(path, NOW) cli = get_status(path, NOW).lower() app = FenrisTuiApp(store_path=path) async with app.run_test(size=(80, 24)) as pilot: headline = str(app.query_one("#headline-band").render()).lower() strip = str(app.query_one("#service-strip").render()).lower() if kind == "missing": assert comp.store_fault is None assert comp.freshness == "empty" assert "no observations yet" in cli and "no observations yet" in headline else: assert comp.state == StatusState.ERROR assert comp.freshness == "unknown" phrase = "newer fenris — upgrade fenris" if kind == "newer" else "observation store unreadable" assert phrase in cli and phrase in headline if kind != "newer": assert str(app.query_one("#drive-health").render()) == "[bold]Native collector log[/bold]" assert not app.query_one("#main-grid").has_class("paused") await pilot.resize_terminal(60, 18) await pilot.pause() assert phrase in str(app.query_one("#live-render").render()).lower() for text in (cli, strip): assert ("boot: enabled" if service == ACTIVE else "boot: disabled" if service == DISABLED else "boot: unknown") in text assert ("timer: active" if service == ACTIVE else "timer: inactive" if service == DISABLED else "timer: unknown") in text if not service: assert "does not start on next boot" not in text projection.assert_not_called() assert (path.read_bytes() if original_exists(path) else None) == before def test_reader_owns_readonly_snapshot_and_closes_on_error(tmp_path): path = tmp_path / "observations.db" writer = init_store(path) with patch("fenris.status.query_service_state", return_value=ACTIVE) as query: with pytest.raises(RuntimeError, match="renderer failed"): with read_status(path, NOW) as (conn, comp): assert comp.sample_count == 0 assert conn.in_transaction with pytest.raises(sqlite3.OperationalError, match="readonly"): conn.execute("DELETE FROM samples") writer.execute("INSERT INTO monitoring_periods (started_at) VALUES ('2026-09-16')") writer.commit() assert conn.execute("SELECT COUNT(*) FROM monitoring_periods").fetchone()[0] == 0 raise RuntimeError("renderer failed") query.assert_called_once() with pytest.raises(sqlite3.ProgrammingError, match="closed"): conn.execute("SELECT 1") writer.close() def test_monitoring_query_failure_is_unknown(tmp_path): path = tmp_path / "observations.db" init_store(path).close() with patch("fenris.status.query_service_state", side_effect=OSError("Unavailable")): comp = get_status_composition(path, NOW) cli = get_status(path, NOW, query_journal=False) assert comp.state == StatusState.UNKNOWN assert comp.boot_enabled is None and comp.timer_active is None assert "boot: unknown" in cli and "timer: unknown" in cli assert "does not start on next boot" not in cli def test_native_systemd_query_failure_is_unknown(tmp_path): from fenris.init_system import InitSystem with patch("fenris.init_system.get_init_system", return_value=InitSystem.SYSTEMD), \ patch("fenris.init_system._systemctl_show", return_value={}): comp = get_status_composition(tmp_path / "missing.db", NOW) assert comp.state == StatusState.UNKNOWN assert comp.boot_enabled is None and comp.timer_active is None @pytest.mark.asyncio async def test_new_store_fault_clears_paused_views_and_can_recover(tmp_path): path = tmp_path / "observations.db" conn = init_store(path) conn.execute( "INSERT INTO monitoring_periods (started_at, ended_at, end_cause) " "VALUES ('2026-09-15', '2026-09-16', 'user_disabled')" ) conn.commit() conn.close() with patch("fenris.status.query_service_state", return_value=DISABLED): app = FenrisTuiApp(store_path=path) async with app.run_test() as pilot: assert app.query_one("#main-grid").has_class("paused") writer = sqlite3.connect(path) writer.execute("PRAGMA user_version = %d" % (SCHEMA_VERSION + 1)) writer.close() app.on_refresh_tick() await pilot.pause() assert not app.query_one("#main-grid").has_class("paused") assert str(app.query_one("#drive-health").render()) == "" assert "upgrade Fenris" in str(app.query_one("#headline-band").render()) assert app.query_one("#usage-history")._day_data == [] writer = sqlite3.connect(path) writer.execute("PRAGMA user_version = %d" % SCHEMA_VERSION) writer.close() app.on_refresh_tick() await pilot.pause() assert app.query_one("#main-grid").has_class("paused")