151 lines
7.0 KiB
Python
151 lines
7.0 KiB
Python
"""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")
|