@@ -461,7 +461,7 @@ class TestCI2Parity:
|
||||
}
|
||||
|
||||
with patch("fenris.status.query_service_state", return_value=service_state), patch(
|
||||
"fenris.tui.query_service_state", return_value=service_state
|
||||
"fenris.status.query_service_state", return_value=service_state
|
||||
):
|
||||
status = get_status(
|
||||
store_path=db, clock_now=_clock(), query_services=True, query_journal=False
|
||||
@@ -502,18 +502,6 @@ class TestCI2Parity:
|
||||
query_services=True, query_journal=False)
|
||||
assert "no observations yet" in status.lower()
|
||||
|
||||
def test_store_fault_phrase_both_views(self, tmp_path):
|
||||
status_src = (FENRIS_PKG / "status.py").read_text()
|
||||
tui_src = (FENRIS_PKG / "tui.py").read_text()
|
||||
phrase = "observation store unreadable"
|
||||
assert phrase in status_src
|
||||
assert phrase.lower() in tui_src.lower()
|
||||
|
||||
def test_newer_schema_phrase_both_views(self):
|
||||
status_src = (FENRIS_PKG / "status.py").read_text()
|
||||
phrase = "observation store written by a newer Fenris"
|
||||
assert phrase in status_src
|
||||
|
||||
def test_status_never_prompts(self):
|
||||
status_src = (FENRIS_PKG / "status.py").read_text()
|
||||
assert "input(" not in status_src
|
||||
@@ -603,12 +591,6 @@ class TestCI3ProhibitionSet:
|
||||
table_names.append(m.group(1))
|
||||
assert "projection" not in [t.lower() for t in table_names]
|
||||
|
||||
def test_no_partial_newer_schema_interpretation(self):
|
||||
"""Readers refuse newer-schema stores. [3.6, 9.5]"""
|
||||
status_src = (FENRIS_PKG / "status.py").read_text()
|
||||
assert "NewerSchema" in status_src
|
||||
assert "upgrade Fenris" in status_src
|
||||
|
||||
def test_polkit_authorizes_one_binary(self):
|
||||
"""Polkit authorizes exactly one binary: fenris-monitor. [8.5]"""
|
||||
monitor_src = (FENRIS_PKG / "monitor.py").read_text()
|
||||
@@ -688,18 +670,6 @@ class TestCI4WordingAndDisclosures:
|
||||
proj_src = (FENRIS_PKG / "projection.py").read_text()
|
||||
assert phrase in proj_src
|
||||
|
||||
def test_store_fault_phrase(self):
|
||||
phrase = "observation store unreadable"
|
||||
status_src = (FENRIS_PKG / "status.py").read_text()
|
||||
assert phrase in status_src
|
||||
tui_src = (FENRIS_PKG / "tui.py").read_text()
|
||||
assert phrase.lower() in tui_src.lower()
|
||||
|
||||
def test_newer_schema_phrase(self):
|
||||
phrase = "observation store written by a newer Fenris"
|
||||
status_src = (FENRIS_PKG / "status.py").read_text()
|
||||
assert phrase in status_src
|
||||
|
||||
def test_no_observations_phrase(self):
|
||||
phrase = "no observations yet"
|
||||
status_src = (FENRIS_PKG / "status.py").read_text()
|
||||
|
||||
@@ -412,7 +412,7 @@ class TestDisplayStates:
|
||||
|
||||
def test_one_sample_awaiting_another_in_tui(self, tmp_path):
|
||||
"""One sample → TUI shows awaiting state."""
|
||||
from fenris.tui import FenrisTuiApp, _query_service_facts
|
||||
from fenris.tui import FenrisTuiApp
|
||||
db = tmp_path / "test.db"
|
||||
conn = init_store(db)
|
||||
conn.execute(
|
||||
|
||||
@@ -0,0 +1,77 @@
|
||||
"""The CLI and TUI cross the same terminal-attached action interface."""
|
||||
import argparse
|
||||
import runpy
|
||||
import shlex
|
||||
import subprocess
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
|
||||
from fenris.control import MONITOR_HELPER, MonitorError, run_monitor
|
||||
from fenris.tui import FenrisTuiApp
|
||||
|
||||
|
||||
@pytest.mark.parametrize("uid", [0, 1000])
|
||||
@pytest.mark.parametrize("args", [("enable", "--now"), ("disable", "--now"), ("collect",)])
|
||||
def test_fixed_helper_and_terminal_attachment(uid, args):
|
||||
with patch("fenris.control.os.geteuid", return_value=uid), \
|
||||
patch("fenris.control.subprocess.run", return_value=subprocess.CompletedProcess([], 0)) as run:
|
||||
run_monitor(*args)
|
||||
expected = [MONITOR_HELPER, *args]
|
||||
if uid:
|
||||
expected.insert(0, "pkexec")
|
||||
# Inherited stdin/out/err keep authentication on the user's terminal.
|
||||
# No frontend deadline can cut short a valid 90-second Collection run.
|
||||
run.assert_called_once_with(expected)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("code", [1, 126, 127, -15])
|
||||
def test_failure_is_not_retried_and_root_equivalent_preserves_arguments(code):
|
||||
args = ("baseline", "set", '{"model": "Drive $(whoami)", "tbw": 100}')
|
||||
with patch("fenris.control.os.geteuid", return_value=1000), \
|
||||
patch("fenris.control.subprocess.run", return_value=subprocess.CompletedProcess([], code)) as run:
|
||||
with pytest.raises(MonitorError) as failure:
|
||||
run_monitor(*args)
|
||||
run.assert_called_once()
|
||||
assert failure.value.exit_code == (code if code > 0 else 128 - code)
|
||||
root_command = str(failure.value).split("run in your terminal: ")[1]
|
||||
assert shlex.split(root_command) == ["sudo", MONITOR_HELPER, *args]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("uid,missing", [(0, MONITOR_HELPER), (1000, "pkexec")])
|
||||
def test_missing_command_is_identified(uid, missing):
|
||||
with patch("fenris.control.os.geteuid", return_value=uid), \
|
||||
patch("fenris.control.subprocess.run", side_effect=FileNotFoundError(2, "Missing", missing)):
|
||||
with pytest.raises(MonitorError, match="Command not found") as failure:
|
||||
run_monitor("collect")
|
||||
assert missing in str(failure.value)
|
||||
assert failure.value.exit_code == 127
|
||||
assert ("sudo" in str(failure.value)) == bool(uid)
|
||||
|
||||
|
||||
def test_interrupt_reports_uncertain_outcome():
|
||||
with patch("fenris.control.subprocess.run", side_effect=KeyboardInterrupt):
|
||||
with pytest.raises(MonitorError, match="Check fenris status") as failure:
|
||||
run_monitor("collect")
|
||||
assert failure.value.exit_code == 130
|
||||
|
||||
|
||||
def test_cli_and_tui_show_the_same_failure(tmp_path, capsys):
|
||||
# The launcher may prepend an installed runtime while loading; isolate it.
|
||||
with patch.object(sys, "path", sys.path.copy()):
|
||||
cli = runpy.run_path(str(Path(__file__).parent.parent / "scripts" / "fenris"))
|
||||
with patch("fenris.control.os.geteuid", return_value=1000), \
|
||||
patch("fenris.control.subprocess.run", return_value=subprocess.CompletedProcess([], 126)):
|
||||
with pytest.raises(SystemExit) as exit_info:
|
||||
cli["cmd_monitor_resume"](argparse.Namespace())
|
||||
assert exit_info.value.code == 126
|
||||
cli_message = capsys.readouterr().err.strip()
|
||||
app = FenrisTuiApp(store_path=tmp_path / "missing.db")
|
||||
with patch.object(app, "suspend"), patch.object(app, "_refresh") as refresh, \
|
||||
patch.object(app, "notify") as notify:
|
||||
app.action_resume()
|
||||
notify.assert_called_once_with(cli_message, severity="error")
|
||||
refresh.assert_called_once()
|
||||
@@ -283,7 +283,7 @@ class TestSystemdQueryState:
|
||||
def test_query_state_disabled_inactive(self):
|
||||
"""Query state for disabled and inactive timer."""
|
||||
with patch("fenris.init_system._systemctl_show") as mock_show:
|
||||
mock_show.return_value = {}
|
||||
mock_show.return_value = {"UnitFileState": "disabled", "ActiveState": "inactive"}
|
||||
result = _systemd_query_state()
|
||||
assert result["boot_enabled"] is False
|
||||
assert result["timer_active"] is False
|
||||
@@ -688,5 +688,5 @@ class TestEdgeCases:
|
||||
with patch("fenris.init_system._systemctl_show") as mock_show:
|
||||
mock_show.return_value = {}
|
||||
result = _systemd_query_state()
|
||||
assert result["boot_enabled"] is False
|
||||
assert result["timer_active"] is False
|
||||
assert result["boot_enabled"] is None
|
||||
assert result["timer_active"] is None
|
||||
|
||||
@@ -260,7 +260,7 @@ class TestPreservedBehavior:
|
||||
conn.close()
|
||||
app = FenrisTuiApp(store_path=tmp_path / "test.db")
|
||||
|
||||
with patch("fenris.tui.query_service_state", return_value={
|
||||
with patch("fenris.status.query_service_state", return_value={
|
||||
"boot_enabled": True, "timer_active": True,
|
||||
"last_collect_ok": True, "last_collect_age_s": 60,
|
||||
"last_collect_reason": None,
|
||||
@@ -272,11 +272,11 @@ class TestPreservedBehavior:
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_quit_rail_preserved(self, tmp_path):
|
||||
"""Separate q QUIT TUI rail preserved."""
|
||||
"""Separate q Quit TUI rail preserved."""
|
||||
app = FenrisTuiApp(store_path=tmp_path / "nonexistent.db")
|
||||
async with app.run_test(size=(120, 24)) as pilot:
|
||||
rail = str(app.query_one("#quit-rail").render())
|
||||
assert "q QUIT TUI" in rail
|
||||
assert "q Quit TUI" in rail
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_auth_banner_preserved(self, tmp_path):
|
||||
@@ -307,7 +307,7 @@ class TestPreservedBehavior:
|
||||
conn.close()
|
||||
app = FenrisTuiApp(store_path=db)
|
||||
|
||||
with patch("fenris.tui.query_service_state", return_value={
|
||||
with patch("fenris.status.query_service_state", return_value={
|
||||
"boot_enabled": False, "timer_active": False,
|
||||
"last_collect_ok": None, "last_collect_age_s": None,
|
||||
"last_collect_reason": None,
|
||||
|
||||
+18
-10
@@ -135,7 +135,7 @@ RPM_TARGETS = [
|
||||
]
|
||||
|
||||
XBPS_TARGETS = [
|
||||
("voidlinux/void-linux:latest", "xbps"),
|
||||
("ghcr.io/void-linux/void-glibc-full:latest", "xbps"),
|
||||
]
|
||||
|
||||
ALL_TARGETS = DEB_TARGETS + RPM_TARGETS + XBPS_TARGETS
|
||||
@@ -176,9 +176,8 @@ def _build_xbps_dockerfile(image: str, pkg_name: str) -> str:
|
||||
"""Dockerfile for testing xbps installation."""
|
||||
return textwrap.dedent(f"""\
|
||||
FROM {image}
|
||||
RUN xbps-install -Suyn void-repo-nonfree && \\
|
||||
xbps-install -Syun python3 smartmontools runit && \\
|
||||
xbps-remove -Oy void-repo-nonfree || true
|
||||
RUN xbps-install -Sy python3 smartmontools runit shadow && \\
|
||||
xbps-remove -O
|
||||
COPY dist/{pkg_name} /pkg/{pkg_name}
|
||||
""")
|
||||
|
||||
@@ -1713,7 +1712,7 @@ def test_xbps_dormant_install(skip_no_docker, version):
|
||||
check=True,
|
||||
)
|
||||
(build_dir / "Dockerfile").write_text(
|
||||
_build_xbps_dockerfile("voidlinux/void-linux:latest", pkg_name)
|
||||
_build_xbps_dockerfile("ghcr.io/void-linux/void-glibc-full:latest", pkg_name)
|
||||
)
|
||||
|
||||
tag = "fenris-test-xbps"
|
||||
@@ -1731,8 +1730,17 @@ def test_xbps_dormant_install(skip_no_docker, version):
|
||||
)
|
||||
|
||||
try:
|
||||
rc, out = _xbps_extract_and_install(container, pkg_name, action="install")
|
||||
# Exercise XBPS itself, including its file ownership and lifecycle hooks.
|
||||
# The temporary local repository does not need release signatures.
|
||||
rc, out = _container_exec(
|
||||
container,
|
||||
f"mkdir -p /tmp/fenris-repo && cp /pkg/{pkg_name} /tmp/fenris-repo/ && "
|
||||
f"xbps-rindex -a /tmp/fenris-repo/{pkg_name} && "
|
||||
"xbps-install -y -R /tmp/fenris-repo fenris",
|
||||
)
|
||||
assert rc == 0, f"XBPS install failed: {out}"
|
||||
rc, out = _container_exec(container, "xbps-query -p pkgver fenris")
|
||||
assert rc == 0 and out.strip() == f"fenris-{version}_1"
|
||||
_assert_runit_dormant_layout(container, version)
|
||||
finally:
|
||||
subprocess.run(
|
||||
@@ -1755,7 +1763,7 @@ def test_xbps_migration_guard(skip_no_docker, version):
|
||||
check=True,
|
||||
)
|
||||
(build_dir / "Dockerfile").write_text(
|
||||
_build_xbps_dockerfile("voidlinux/void-linux:latest", pkg_name)
|
||||
_build_xbps_dockerfile("ghcr.io/void-linux/void-glibc-full:latest", pkg_name)
|
||||
)
|
||||
|
||||
tag = "fenris-guard-xbps"
|
||||
@@ -1803,7 +1811,7 @@ def test_xbps_upgrade_semantics(skip_no_docker, version):
|
||||
check=True,
|
||||
)
|
||||
(build_dir / "Dockerfile").write_text(
|
||||
_build_xbps_dockerfile("voidlinux/void-linux:latest", pkg_name)
|
||||
_build_xbps_dockerfile("ghcr.io/void-linux/void-glibc-full:latest", pkg_name)
|
||||
)
|
||||
|
||||
tag = "fenris-upgrade-xbps"
|
||||
@@ -1889,7 +1897,7 @@ def test_xbps_remove_preserves_config_and_store(skip_no_docker, version):
|
||||
check=True,
|
||||
)
|
||||
(build_dir / "Dockerfile").write_text(
|
||||
_build_xbps_dockerfile("voidlinux/void-linux:latest", pkg_name)
|
||||
_build_xbps_dockerfile("ghcr.io/void-linux/void-glibc-full:latest", pkg_name)
|
||||
)
|
||||
|
||||
tag = "fenris-remove-xbps"
|
||||
@@ -1960,7 +1968,7 @@ def test_xbps_purge_removes_everything(skip_no_docker, version):
|
||||
check=True,
|
||||
)
|
||||
(build_dir / "Dockerfile").write_text(
|
||||
_build_xbps_dockerfile("voidlinux/void-linux:latest", pkg_name)
|
||||
_build_xbps_dockerfile("ghcr.io/void-linux/void-glibc-full:latest", pkg_name)
|
||||
)
|
||||
|
||||
tag = "fenris-purge-xbps"
|
||||
|
||||
@@ -622,6 +622,7 @@ class TestProjectionInStatus:
|
||||
from fenris.status import get_status
|
||||
|
||||
nonexistent = tmp_path / "nonexistent.db"
|
||||
nonexistent.write_bytes(b"not a SQLite database")
|
||||
now = datetime(2026, 9, 1, 12, 0, 0, tzinfo=timezone.utc)
|
||||
|
||||
with patch("fenris.status.query_service_state", return_value={
|
||||
|
||||
@@ -587,7 +587,7 @@ class TestCLIStatusRendering:
|
||||
newer_schema=None)
|
||||
cli_text = render_status_cli(comp)
|
||||
assert "observation store unreadable" in cli_text
|
||||
assert "see journal" in cli_text.lower()
|
||||
assert "see collector logs" in cli_text.lower()
|
||||
conn.close()
|
||||
|
||||
|
||||
@@ -653,7 +653,7 @@ class TestTUIStatusRendering:
|
||||
comp = compose_status(conn, svc, now, store_fault="observation store unreadable",
|
||||
newer_schema=None)
|
||||
tui_text = render_status_tui(comp)
|
||||
assert "observation store unreadable" in tui_text
|
||||
assert "Observation store unreadable" in tui_text
|
||||
conn.close()
|
||||
|
||||
|
||||
@@ -966,7 +966,7 @@ class TestTUIIntegration:
|
||||
conn.close()
|
||||
|
||||
app = FenrisTuiApp(store_path=tmp_path / "test.db")
|
||||
with patch("fenris.tui.query_service_state", return_value={
|
||||
with patch("fenris.status.query_service_state", return_value={
|
||||
"boot_enabled": True, "timer_active": True,
|
||||
"last_collect_ok": True, "last_collect_age_s": 120,
|
||||
"last_collect_reason": None,
|
||||
@@ -992,7 +992,7 @@ class TestTUIIntegration:
|
||||
conn.close()
|
||||
|
||||
app = FenrisTuiApp(store_path=db)
|
||||
with patch("fenris.tui.query_service_state", return_value={
|
||||
with patch("fenris.status.query_service_state", return_value={
|
||||
"boot_enabled": False, "timer_active": False,
|
||||
"last_collect_ok": None, "last_collect_age_s": None,
|
||||
"last_collect_reason": None,
|
||||
@@ -1019,7 +1019,7 @@ class TestTUIIntegration:
|
||||
conn.close()
|
||||
|
||||
app = FenrisTuiApp(store_path=tmp_path / "test.db")
|
||||
with patch("fenris.tui.query_service_state", return_value={
|
||||
with patch("fenris.status.query_service_state", return_value={
|
||||
"boot_enabled": True, "timer_active": True,
|
||||
"last_collect_ok": False, "last_collect_age_s": 60,
|
||||
"last_collect_reason": "exit code 3",
|
||||
|
||||
@@ -0,0 +1,150 @@
|
||||
"""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("#constrained-summary").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")
|
||||
+60
-109
@@ -42,11 +42,7 @@ from fenris.tui import (
|
||||
FenrisTuiApp,
|
||||
DailyBarGraph,
|
||||
_format_remaining,
|
||||
_sparkline,
|
||||
_habit_bar,
|
||||
_query_usage_history,
|
||||
_query_drive_health,
|
||||
_query_service_facts,
|
||||
_query_daily_graph_data,
|
||||
_query_hourly_graph_data,
|
||||
_RANGE_OPTIONS,
|
||||
@@ -142,64 +138,6 @@ class TestFormatRemaining:
|
||||
assert _format_remaining(-100) == "endurance exhausted"
|
||||
|
||||
|
||||
class TestSparkline:
|
||||
def test_empty(self):
|
||||
assert _sparkline([]) == ""
|
||||
|
||||
def test_single_value(self):
|
||||
result = _sparkline([100.0])
|
||||
assert len(result) == 1
|
||||
|
||||
def test_multiple_values(self):
|
||||
result = _sparkline([1.0, 2.0, 3.0, 4.0, 5.0])
|
||||
assert len(result) > 0
|
||||
assert all(c in " ▁▂▃▄▅▆▇█" for c in result)
|
||||
|
||||
def test_width_limit(self):
|
||||
result = _sparkline([1.0] * 100, width=20)
|
||||
assert len(result) <= 20
|
||||
|
||||
|
||||
class TestHabitBar:
|
||||
def test_all_active(self):
|
||||
result = _habit_bar(1.0, 0.0, 0.0, 0.0)
|
||||
assert "active 100%" in result
|
||||
|
||||
def test_mixed(self):
|
||||
result = _habit_bar(0.5, 0.3, 0.1, 0.1)
|
||||
assert "active 50%" in result
|
||||
assert "idle 30%" in result
|
||||
|
||||
def test_all_unknown(self):
|
||||
result = _habit_bar(0.0, 0.0, 0.0, 1.0)
|
||||
assert "unknown 100%" in result
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Data query tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestQueryUsageHistory:
|
||||
def test_empty_store(self, tmp_path):
|
||||
conn = init_store(tmp_path / "test.db")
|
||||
result = _query_usage_history(conn)
|
||||
assert result["num_days"] == 0
|
||||
assert result["sparkline"] == ""
|
||||
conn.close()
|
||||
|
||||
def test_with_days(self, tmp_path):
|
||||
conn = init_store(tmp_path / "test.db")
|
||||
_insert_segment(conn)
|
||||
_open_period(conn)
|
||||
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)
|
||||
result = _query_usage_history(conn)
|
||||
assert result["num_days"] == 14
|
||||
assert result["sparkline"] != ""
|
||||
conn.close()
|
||||
|
||||
|
||||
class TestQueryDriveHealth:
|
||||
def test_empty_store(self, tmp_path):
|
||||
conn = init_store(tmp_path / "test.db")
|
||||
@@ -215,27 +153,6 @@ class TestQueryDriveHealth:
|
||||
conn.close()
|
||||
|
||||
|
||||
class TestQueryServiceFacts:
|
||||
def test_empty_store(self, tmp_path):
|
||||
conn = init_store(tmp_path / "test.db")
|
||||
now = _clock()
|
||||
result = _query_service_facts(conn, now)
|
||||
assert result["freshness"] == "empty"
|
||||
conn.close()
|
||||
|
||||
def test_fresh_sample(self, tmp_path):
|
||||
conn = init_store(tmp_path / "test.db")
|
||||
_insert_segment(conn)
|
||||
_open_period(conn)
|
||||
_insert_day(conn, "2026-09-29", bw=1024*1024*100)
|
||||
now = _clock()
|
||||
ts = (now - timedelta(minutes=2)).isoformat()
|
||||
_insert_sample(conn, ts)
|
||||
result = _query_service_facts(conn, now)
|
||||
assert result["freshness"] == "fresh"
|
||||
conn.close()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# CI-1: Exhaustive state matrix from synthetic stores
|
||||
# ---------------------------------------------------------------------------
|
||||
@@ -397,10 +314,10 @@ class TestDenseScreen:
|
||||
app = FenrisTuiApp(store_path=tmp_path / "test.db")
|
||||
async with app.run_test() as pilot:
|
||||
strip = str(app.query_one("#service-strip").render())
|
||||
assert "boot:" in strip
|
||||
assert "timer:" in strip
|
||||
assert "last collect:" in strip
|
||||
assert "freshness:" in strip
|
||||
assert "Boot:" in strip
|
||||
assert "Timer:" in strip
|
||||
assert "Last collect:" in strip
|
||||
assert "Freshness:" in strip
|
||||
# Maker credit now in titlebox only (issue #79)
|
||||
assert "by Bongbetic" not in strip
|
||||
|
||||
@@ -412,7 +329,7 @@ class TestDenseScreen:
|
||||
conn.close()
|
||||
app = FenrisTuiApp(store_path=tmp_path / "test.db")
|
||||
|
||||
with patch("fenris.tui.query_service_state", return_value={
|
||||
with patch("fenris.status.query_service_state", return_value={
|
||||
"boot_enabled": True, "timer_active": True,
|
||||
"last_collect_ok": True, "last_collect_age_s": 60,
|
||||
"last_collect_reason": None,
|
||||
@@ -427,7 +344,7 @@ class TestDenseScreen:
|
||||
assert "monitoring: active in background · persists across reboots" in strip
|
||||
assert "r resume — enable monitoring and future boots" in strip
|
||||
assert "q quit" not in strip
|
||||
assert rail == "q QUIT TUI"
|
||||
assert rail == "q Quit TUI"
|
||||
assert usage.region.y < service.region.y < quit_rail.region.y
|
||||
assert usage.region.bottom <= service.region.y
|
||||
assert service.region.bottom <= quit_rail.region.y
|
||||
@@ -446,11 +363,11 @@ class TestDenseScreen:
|
||||
conn.close()
|
||||
app = FenrisTuiApp(store_path=db)
|
||||
|
||||
with patch("fenris.tui.query_service_state", return_value={
|
||||
with patch("fenris.status.query_service_state", return_value={
|
||||
"boot_enabled": False, "timer_active": False,
|
||||
"last_collect_ok": None, "last_collect_age_s": None,
|
||||
"last_collect_reason": None,
|
||||
}), patch("fenris.tui.subprocess.run") as subprocess_run, patch.object(
|
||||
}), patch("fenris.control.subprocess.run") as subprocess_run, patch.object(
|
||||
app, "_run_helper"
|
||||
) as run_helper:
|
||||
async with app.run_test(size=(80, 24)) as pilot:
|
||||
@@ -472,7 +389,7 @@ class TestDenseScreen:
|
||||
visible_text = " ".join(
|
||||
"".join(ElementTree.fromstring(screenshot).itertext()).split()
|
||||
)
|
||||
assert "paused time is excluded from your usage habit" in visible_text
|
||||
assert "Paused time is excluded from your usage habit" in visible_text
|
||||
assert "resume: fenris monitor resume" in visible_text
|
||||
assert "monitoring: does not start on next boot" in str(
|
||||
app.query_one("#service-strip").render()
|
||||
@@ -487,7 +404,7 @@ class TestDenseScreen:
|
||||
ElementTree.fromstring(app.export_screenshot()).itertext()
|
||||
).split()
|
||||
)
|
||||
assert "q QUIT TUI" in footer_text
|
||||
assert "q Quit TUI" in footer_text
|
||||
await pilot.press("q")
|
||||
assert not app.is_running
|
||||
subprocess_run.assert_not_called()
|
||||
@@ -509,11 +426,11 @@ class TestDenseScreen:
|
||||
conn.close()
|
||||
app = FenrisTuiApp(store_path=db)
|
||||
|
||||
with patch("fenris.tui.query_service_state", return_value={
|
||||
with patch("fenris.status.query_service_state", return_value={
|
||||
"boot_enabled": True, "timer_active": True,
|
||||
"last_collect_ok": None, "last_collect_age_s": None,
|
||||
"last_collect_reason": None,
|
||||
}), patch("fenris.tui.subprocess.run") as subprocess_run, patch.object(
|
||||
}), patch("fenris.control.subprocess.run") as subprocess_run, patch.object(
|
||||
app, "_run_helper"
|
||||
) as run_helper:
|
||||
async with app.run_test() as pilot:
|
||||
@@ -539,7 +456,7 @@ class TestDenseScreen:
|
||||
store_path=tmp_path / "nonexistent.db",
|
||||
refresh_interval_s=0.2,
|
||||
)
|
||||
auth_notice = "privileged actions will prompt for authentication (polkit)"
|
||||
auth_notice = "Open with fenris (no sudo). Actions authenticate via polkit. ? Help"
|
||||
|
||||
async with app.run_test() as pilot:
|
||||
headline = str(app.query_one("#headline-band").render())
|
||||
@@ -595,12 +512,47 @@ class TestDisclosuresAndGreeting:
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestFirstRun:
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("size", [(80, 24), (60, 18)])
|
||||
async def test_sudo_help_remains_available_after_launch(self, tmp_path, size):
|
||||
app = FenrisTuiApp(store_path=tmp_path / "missing.db")
|
||||
with patch.object(app, "_run_helper") as run_helper:
|
||||
async with app.run_test(size=size) as pilot:
|
||||
app.on_refresh_tick()
|
||||
await pilot.press("?")
|
||||
help_text = str(app.screen.query_one("#help-text").render())
|
||||
assert "The dashboard does not need sudo" in help_text
|
||||
assert "sudo fenris monitor resume" in help_text
|
||||
assert "sudo fenris monitor pause" in help_text
|
||||
assert "sudo fenris sample" in help_text
|
||||
assert 'sudo usermod -aG fenris "$USER"' in help_text
|
||||
assert "log out and back in" in help_text
|
||||
await pilot.press("end")
|
||||
visible = " ".join("".join(
|
||||
ElementTree.fromstring(app.export_screenshot()).itertext()
|
||||
).split())
|
||||
assert "Esc Close" in visible
|
||||
await pilot.press("escape")
|
||||
assert len(app.screen_stack) == 1
|
||||
run_helper.assert_not_called()
|
||||
|
||||
def test_missing_polkit_identifies_command_and_points_to_help(self, tmp_path):
|
||||
app = FenrisTuiApp(store_path=tmp_path / "missing.db")
|
||||
with patch("fenris.control.os.geteuid", return_value=1000), \
|
||||
patch("fenris.control.subprocess.run", side_effect=FileNotFoundError(2, "Missing", "pkexec")), \
|
||||
patch.object(app, "suspend"), patch.object(app, "_refresh"), \
|
||||
patch.object(app, "notify") as notify:
|
||||
app.action_resume()
|
||||
message = notify.call_args.args[0]
|
||||
assert "Command not found: pkexec" in message
|
||||
assert "sudo /usr/libexec/fenris/fenris-monitor enable --now" in message
|
||||
|
||||
def test_unprivileged_resume_uses_polkit(self, tmp_path):
|
||||
"""TUI controls use the same authenticated path as the CLI."""
|
||||
app = FenrisTuiApp(store_path=tmp_path / "nonexistent.db")
|
||||
|
||||
with patch("fenris.tui.os.geteuid", return_value=1000), \
|
||||
patch("fenris.tui.subprocess.run") as run, \
|
||||
with patch("fenris.control.os.geteuid", return_value=1000), \
|
||||
patch("fenris.control.subprocess.run") as run, \
|
||||
patch.object(app, "suspend"), \
|
||||
patch.object(app, "_refresh"):
|
||||
run.return_value.returncode = 0
|
||||
@@ -608,7 +560,6 @@ class TestFirstRun:
|
||||
|
||||
run.assert_called_once_with(
|
||||
["pkexec", "/usr/libexec/fenris/fenris-monitor", "enable", "--now"],
|
||||
timeout=30,
|
||||
)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@@ -924,8 +875,8 @@ class TestBarGraphTUI:
|
||||
assert len(graph._day_data) > 0
|
||||
# Legend should be visible
|
||||
legend = str(app.query_one("#bar-legend").render())
|
||||
assert "alloc" in legend
|
||||
assert "zero" in legend
|
||||
assert "Alloc" in legend
|
||||
assert "Zero" in legend
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_arrow_selection(self, tmp_path):
|
||||
@@ -1091,7 +1042,7 @@ class TestBarGraphTUI:
|
||||
app = FenrisTuiApp(store_path=tmp_path / "nonexistent.db")
|
||||
async with app.run_test(size=(80, 24)) as pilot:
|
||||
graph = app.query_one("#usage-history")
|
||||
assert graph.border_title == "usage history"
|
||||
assert graph.border_title == "Usage history"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_glyphs_in_legend(self, tmp_path):
|
||||
@@ -1172,7 +1123,7 @@ class TestConstrainedLayout:
|
||||
# Constrained summary should be visible
|
||||
summary = app.query_one("#constrained-summary")
|
||||
summary_text = str(summary.render())
|
||||
assert "graph needs ≥80×24" in summary_text
|
||||
assert "Graph needs ≥80×24" in summary_text
|
||||
assert "days" in summary_text
|
||||
assert "GB total" in summary_text
|
||||
|
||||
@@ -1196,7 +1147,7 @@ class TestConstrainedLayout:
|
||||
# Constrained summary should be visible
|
||||
summary = app.query_one("#constrained-summary")
|
||||
summary_text = str(summary.render())
|
||||
assert "graph needs ≥80×24" in summary_text
|
||||
assert "Graph needs ≥80×24" in summary_text
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_resize_from_constrained_to_normal(self, tmp_path):
|
||||
@@ -1273,7 +1224,7 @@ class TestConstrainedLayout:
|
||||
# Constrained summary should show selected day context
|
||||
summary = app.query_one("#constrained-summary")
|
||||
summary_text = str(summary.render())
|
||||
assert "graph needs ≥80×24" in summary_text
|
||||
assert "Graph needs ≥80×24" in summary_text
|
||||
# Selected day context should survive
|
||||
assert "2026-09" in summary_text
|
||||
|
||||
@@ -1285,7 +1236,7 @@ class TestConstrainedLayout:
|
||||
assert app._is_constrained_mode
|
||||
summary = app.query_one("#constrained-summary")
|
||||
summary_text = str(summary.render())
|
||||
assert "graph needs ≥80×24" in summary_text
|
||||
assert "Graph needs ≥80×24" in summary_text
|
||||
assert "awaiting" in summary_text.lower()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@@ -1315,13 +1266,13 @@ class TestConstrainedLayout:
|
||||
|
||||
# Service strip should have actions
|
||||
strip = str(app.query_one("#service-strip").render())
|
||||
assert "p pause" in strip
|
||||
assert "r resume" in strip
|
||||
assert "p Pause" in strip
|
||||
assert "r Resume" in strip
|
||||
assert "q quit" not in strip # Quit is in quit-rail
|
||||
|
||||
# Quit rail should be visible
|
||||
rail = str(app.query_one("#quit-rail").render())
|
||||
assert "QUIT" in rail
|
||||
assert "Quit" in rail
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_constrained_long_reasons_visible(self, tmp_path):
|
||||
|
||||
Reference in New Issue
Block a user