"""Tests for the init system abstraction layer (issue #84). Covers: - Init system detection (systemd vs runit) - systemd backend functions (enable, disable, collect, query state) - runit backend functions (enable, disable, collect, query state) - Public API dispatching to correct backend - Edge cases (already enabled/disabled, missing files, timeouts) Spec: ADR 0008, §8.4, §8.5, §8.6, §8.7, §8.8 """ import os import sqlite3 import tempfile from datetime import datetime, timedelta, timezone from pathlib import Path from unittest.mock import patch, MagicMock, PropertyMock import pytest import sys sys.path.insert(0, str(Path(__file__).parent.parent / "src")) from fenris.init_system import ( InitSystem, detect_init_system, get_init_system, reset_init_system_cache, _systemd_enable, _systemd_disable, _systemd_collect, _systemd_query_state, _runit_enable, _runit_disable, _runit_collect, _runit_query_state, _runit_is_enabled, _runit_is_running, enable_timer, disable_timer, collect_now, query_service_state, journal_hint, FENRIS_SV_DIR, FENRIS_SERVICE_LINK, COLLECT_TIMEOUT_S, ) from fenris.store import init_store @pytest.fixture(autouse=True) def reset_cache(): """Reset the init system cache before each test.""" reset_init_system_cache() yield reset_init_system_cache() # --------------------------------------------------------------------------- # Init system detection # --------------------------------------------------------------------------- class TestInitSystemDetection: """Test init system detection logic.""" def test_detect_systemd_by_run_directory(self): """Systemd detected via /run/systemd/system directory.""" with patch("pathlib.Path.exists") as mock_exists: mock_exists.return_value = True reset_init_system_cache() result = detect_init_system() assert result == InitSystem.SYSTEMD def test_detect_systemd_by_pid1(self): """Systemd detected via PID 1 name.""" original_exists = Path.exists original_read_text = Path.read_text def mock_exists(self_path): if str(self_path) == "/run/systemd/system": return False return original_exists(self_path) def mock_read_text(self_path): if str(self_path) == "/proc/1/comm": return "systemd" return original_read_text(self_path) with patch("pathlib.Path.exists", mock_exists), \ patch("pathlib.Path.read_text", mock_read_text): reset_init_system_cache() result = detect_init_system() assert result == InitSystem.SYSTEMD def test_detect_runit_by_pid1(self): """Runit detected via PID 1 name.""" original_exists = Path.exists original_read_text = Path.read_text def mock_exists(self_path): if str(self_path) == "/run/systemd/system": return False return original_exists(self_path) def mock_read_text(self_path): if str(self_path) == "/proc/1/comm": return "runsv" return original_read_text(self_path) with patch("pathlib.Path.exists", mock_exists), \ patch("pathlib.Path.read_text", mock_read_text): reset_init_system_cache() result = detect_init_system() assert result == InitSystem.RUNIT def test_detect_runit_by_etc_sv(self): """Runit detected via /etc/sv directory.""" original_exists = Path.exists original_read_text = Path.read_text def mock_exists(self_path): if str(self_path) == "/run/systemd/system": return False if str(self_path) == "/etc/sv": return True return original_exists(self_path) def mock_read_text(self_path): if str(self_path) == "/proc/1/comm": raise OSError("no such file") return original_read_text(self_path) with patch("pathlib.Path.exists", mock_exists), \ patch("pathlib.Path.read_text", mock_read_text): reset_init_system_cache() result = detect_init_system() assert result == InitSystem.RUNIT def test_default_to_systemd(self): """Default to systemd when no detection matches.""" original_exists = Path.exists original_read_text = Path.read_text original_is_dir = Path.is_dir def mock_exists(self_path): if str(self_path) == "/run/systemd/system": return False return original_exists(self_path) def mock_is_dir(self_path): if str(self_path) == "/etc/sv": return False return original_is_dir(self_path) def mock_read_text(self_path): if str(self_path) == "/proc/1/comm": raise OSError("no such file") return original_read_text(self_path) with patch("pathlib.Path.exists", mock_exists), \ patch("pathlib.Path.is_dir", mock_is_dir), \ patch("pathlib.Path.read_text", mock_read_text): reset_init_system_cache() result = detect_init_system() assert result == InitSystem.SYSTEMD # --------------------------------------------------------------------------- # systemd backend # --------------------------------------------------------------------------- class TestSystemdEnable: """Test systemd enable function.""" def test_enable_now(self): """Enable with --now flag.""" with patch("fenris.init_system.subprocess") as mock_sub: mock_sub.run.return_value = MagicMock(returncode=0) _systemd_enable(now=True) mock_sub.run.assert_called_once_with( ["systemctl", "enable", "--now", "fenris-collect.timer"], capture_output=True, text=True, ) def test_enable_without_now(self): """Enable without --now flag.""" with patch("fenris.init_system.subprocess") as mock_sub: mock_sub.run.return_value = MagicMock(returncode=0) _systemd_enable(now=False) mock_sub.run.assert_called_once_with( ["systemctl", "enable", "fenris-collect.timer"], capture_output=True, text=True, ) def test_enable_failure_exits(self): """Enable failure exits with error.""" with patch("fenris.init_system.subprocess") as mock_sub: mock_sub.run.return_value = MagicMock( returncode=1, stderr="Unit not found" ) with pytest.raises(SystemExit) as exc_info: _systemd_enable(now=True) assert exc_info.value.code == 1 class TestSystemdDisable: """Test systemd disable function.""" def test_disable_now(self): """Disable with --now flag.""" with patch("fenris.init_system.subprocess") as mock_sub: mock_sub.run.return_value = MagicMock(returncode=0) _systemd_disable(now=True) mock_sub.run.assert_called_once_with( ["systemctl", "disable", "--now", "fenris-collect.timer"], capture_output=True, text=True, ) def test_disable_without_now(self): """Disable without --now flag.""" with patch("fenris.init_system.subprocess") as mock_sub: mock_sub.run.return_value = MagicMock(returncode=0) _systemd_disable(now=False) mock_sub.run.assert_called_once_with( ["systemctl", "disable", "fenris-collect.timer"], capture_output=True, text=True, ) class TestSystemdCollect: """Test systemd collect function.""" def test_collect_success(self): """Successful collection.""" with patch("fenris.init_system.subprocess") as mock_sub: mock_sub.run.return_value = MagicMock(returncode=0) _systemd_collect() mock_sub.run.assert_called_once_with( ["systemctl", "start", "fenris-collect.service"], capture_output=True, text=True, ) def test_collect_failure_exits(self): """Collection failure exits with error.""" with patch("fenris.init_system.subprocess") as mock_sub: mock_sub.run.return_value = MagicMock( returncode=1, stderr="Unit not found" ) with pytest.raises(SystemExit) as exc_info: _systemd_collect() assert exc_info.value.code == 1 class TestSystemdQueryState: """Test systemd query state function.""" def test_query_state_enabled_active(self): """Query state for enabled and active timer.""" with patch("fenris.init_system._systemctl_show") as mock_show: def mock_show_fn(unit, *props): if unit == "fenris-collect.timer": return { "UnitFileState": "enabled", "ActiveState": "active", "LastTriggerUSec": "2026-09-01T12:00:00Z", } elif unit == "fenris-collect.service": return { "ActiveState": "inactive", "ExecMainStatus": "0", "ExecMainExitTimestamp": "2026-09-01T12:00:30Z", } return {} mock_show.side_effect = mock_show_fn result = _systemd_query_state() assert result["boot_enabled"] is True assert result["timer_active"] is True assert result["last_collect_ok"] is True assert result["last_collect_age_s"] is not None 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 = {"UnitFileState": "disabled", "ActiveState": "inactive"} result = _systemd_query_state() assert result["boot_enabled"] is False assert result["timer_active"] is False assert result["last_collect_ok"] is None # --------------------------------------------------------------------------- # runit backend # --------------------------------------------------------------------------- class TestRunitEnable: """Test runit enable function.""" def test_enable_creates_symlink(self, tmp_path): """Enable creates symlink to service directory.""" sv_dir = tmp_path / "sv" / "fenris-collect" sv_dir.mkdir(parents=True) service_link = tmp_path / "service" / "fenris-collect" service_link.parent.mkdir(parents=True) with patch("fenris.init_system.FENRIS_SV_DIR", sv_dir), \ patch("fenris.init_system.FENRIS_SERVICE_LINK", service_link): _runit_enable(_now=True) assert service_link.exists() assert service_link.is_symlink() assert service_link.resolve() == sv_dir def test_enable_removes_down_file(self, tmp_path): """Enable removes the 'down' file if present.""" sv_dir = tmp_path / "sv" / "fenris-collect" sv_dir.mkdir(parents=True) (sv_dir / "down").touch() service_link = tmp_path / "service" / "fenris-collect" service_link.parent.mkdir(parents=True) with patch("fenris.init_system.FENRIS_SV_DIR", sv_dir), \ patch("fenris.init_system.FENRIS_SERVICE_LINK", service_link): _runit_enable(_now=True) assert not (sv_dir / "down").exists() def test_enable_idempotent(self, tmp_path): """Enable is idempotent when already enabled.""" sv_dir = tmp_path / "sv" / "fenris-collect" sv_dir.mkdir(parents=True) service_link = tmp_path / "service" / "fenris-collect" service_link.parent.mkdir(parents=True) service_link.symlink_to(sv_dir) with patch("fenris.init_system.FENRIS_SV_DIR", sv_dir), \ patch("fenris.init_system.FENRIS_SERVICE_LINK", service_link): _runit_enable(_now=True) assert service_link.exists() class TestRunitDisable: """Test runit disable function.""" def test_disable_removes_symlink(self, tmp_path): """Disable removes the service symlink.""" sv_dir = tmp_path / "sv" / "fenris-collect" sv_dir.mkdir(parents=True) service_link = tmp_path / "service" / "fenris-collect" service_link.parent.mkdir(parents=True) service_link.symlink_to(sv_dir) with patch("fenris.init_system.FENRIS_SV_DIR", sv_dir), \ patch("fenris.init_system.FENRIS_SERVICE_LINK", service_link): _runit_disable(_now=True) assert not service_link.exists() assert (sv_dir / "down").exists() def test_disable_idempotent(self, tmp_path): """Disable is idempotent when already disabled.""" sv_dir = tmp_path / "sv" / "fenris-collect" sv_dir.mkdir(parents=True) service_link = tmp_path / "service" / "fenris-collect" service_link.parent.mkdir(parents=True) with patch("fenris.init_system.FENRIS_SV_DIR", sv_dir), \ patch("fenris.init_system.FENRIS_SERVICE_LINK", service_link): _runit_disable(_now=True) assert not service_link.exists() class TestRunitCollect: """Test runit collect function.""" def test_collect_uses_flock(self): """Collect uses flock for serialization.""" with patch("fenris.init_system.subprocess") as mock_sub, \ patch("fenris.init_system.Path") as mock_path: mock_path.return_value.exists.return_value = True mock_sub.run.return_value = MagicMock(returncode=0) _runit_collect() # Verify flock was used call_args = mock_sub.run.call_args[0][0] assert "flock" in call_args def test_collect_timeout_exits(self): """Collection timeout exits with error.""" with patch("fenris.init_system.subprocess") as mock_sub, \ patch("fenris.init_system.Path") as mock_path: mock_path.return_value.exists.return_value = True mock_sub.run.return_value = MagicMock(returncode=124) with pytest.raises(SystemExit) as exc_info: _runit_collect() assert exc_info.value.code == 1 class TestRunitQueryState: """Test runit query state function.""" def test_query_state_enabled_running(self, tmp_path): """Query state for enabled and running service.""" sv_dir = tmp_path / "sv" / "fenris-collect" sv_dir.mkdir(parents=True) supervise_dir = sv_dir / "supervise" supervise_dir.mkdir(parents=True) (supervise_dir / "pid").write_text("12345") service_link = tmp_path / "service" / "fenris-collect" service_link.parent.mkdir(parents=True) service_link.symlink_to(sv_dir) with patch("fenris.init_system.FENRIS_SV_DIR", sv_dir), \ patch("fenris.init_system.FENRIS_SERVICE_LINK", service_link), \ patch("os.kill") as mock_kill: mock_kill.return_value = True # Process exists result = _runit_query_state() assert result["boot_enabled"] is True assert result["timer_active"] is True def test_query_state_disabled_not_running(self, tmp_path): """Query state for disabled and not running service.""" sv_dir = tmp_path / "sv" / "fenris-collect" sv_dir.mkdir(parents=True) service_link = tmp_path / "service" / "fenris-collect" service_link.parent.mkdir(parents=True) with patch("fenris.init_system.FENRIS_SV_DIR", sv_dir), \ patch("fenris.init_system.FENRIS_SERVICE_LINK", service_link): result = _runit_query_state() assert result["boot_enabled"] is False assert result["timer_active"] is False class TestRunitIsEnabled: """Test runit is_enabled check.""" def test_is_enabled_true(self, tmp_path): """Service is enabled when symlink exists.""" sv_dir = tmp_path / "sv" / "fenris-collect" sv_dir.mkdir(parents=True) service_link = tmp_path / "service" / "fenris-collect" service_link.parent.mkdir(parents=True) service_link.symlink_to(sv_dir) with patch("fenris.init_system.FENRIS_SERVICE_LINK", service_link): assert _runit_is_enabled() is True def test_is_enabled_false(self, tmp_path): """Service is disabled when symlink does not exist.""" service_link = tmp_path / "service" / "fenris-collect" with patch("fenris.init_system.FENRIS_SERVICE_LINK", service_link): assert _runit_is_enabled() is False class TestRunitIsRunning: """Test runit is_running check.""" def test_is_running_true(self, tmp_path): """Service is running when PID file exists and process is alive.""" sv_dir = tmp_path / "sv" / "fenris-collect" supervise_dir = sv_dir / "supervise" supervise_dir.mkdir(parents=True) (supervise_dir / "pid").write_text("12345") with patch("fenris.init_system.FENRIS_SV_DIR", sv_dir), \ patch("os.kill") as mock_kill: mock_kill.return_value = True assert _runit_is_running() is True def test_is_running_false_no_pid(self, tmp_path): """Service is not running when PID file does not exist.""" sv_dir = tmp_path / "sv" / "fenris-collect" sv_dir.mkdir(parents=True) with patch("fenris.init_system.FENRIS_SV_DIR", sv_dir), \ patch("subprocess.run") as mock_run: mock_run.return_value.returncode = 1 assert _runit_is_running() is False def test_is_running_false_dead_process(self, tmp_path): """Service is not running when process is dead.""" sv_dir = tmp_path / "sv" / "fenris-collect" supervise_dir = sv_dir / "supervise" supervise_dir.mkdir(parents=True) (supervise_dir / "pid").write_text("12345") with patch("fenris.init_system.FENRIS_SV_DIR", sv_dir), \ patch("os.kill", side_effect=OSError("No such process")): assert _runit_is_running() is False def test_is_running_uses_process_table_when_supervise_is_unreadable(self, tmp_path): """Void keeps runit's supervise directory root-only for normal users.""" sv_dir = tmp_path / "sv" / "fenris-collect" supervise_dir = sv_dir / "supervise" supervise_dir.mkdir(parents=True) pid_file = supervise_dir / "pid" pid_file.write_text("12345") service_link = tmp_path / "service" / "fenris-collect" service_link.parent.mkdir(parents=True) service_link.symlink_to(sv_dir) original_read_text = Path.read_text def deny_pid(path, *args, **kwargs): if path == pid_file: raise PermissionError("supervise is root-only") return original_read_text(path, *args, **kwargs) with patch("fenris.init_system.FENRIS_SV_DIR", sv_dir), \ patch("fenris.init_system.FENRIS_SERVICE_LINK", service_link), \ patch.object(Path, "read_text", autospec=True, side_effect=deny_pid), \ patch("subprocess.run") as mock_run: mock_run.return_value.returncode = 0 assert _runit_is_running() is True mock_run.assert_called_once_with( ["pgrep", "-f", "^runsv fenris-collect$"], capture_output=True, text=True, timeout=5, ) # --------------------------------------------------------------------------- # Public API dispatching # --------------------------------------------------------------------------- class TestPublicAPI: """Test public API dispatches to correct backend.""" def test_enable_dispatches_to_systemd(self): """enable_timer dispatches to systemd on systemd system.""" with patch("fenris.init_system.get_init_system", return_value=InitSystem.SYSTEMD), \ patch("fenris.init_system._systemd_enable") as mock_enable: enable_timer(now=True) mock_enable.assert_called_once_with(True) def test_enable_dispatches_to_runit(self): """enable_timer dispatches to runit on runit system.""" with patch("fenris.init_system.get_init_system", return_value=InitSystem.RUNIT), \ patch("fenris.init_system._runit_enable") as mock_enable: enable_timer(now=True) mock_enable.assert_called_once_with(True) def test_disable_dispatches_to_systemd(self): """disable_timer dispatches to systemd on systemd system.""" with patch("fenris.init_system.get_init_system", return_value=InitSystem.SYSTEMD), \ patch("fenris.init_system._systemd_disable") as mock_disable: disable_timer(now=False) mock_disable.assert_called_once_with(False) def test_disable_dispatches_to_runit(self): """disable_timer dispatches to runit on runit system.""" with patch("fenris.init_system.get_init_system", return_value=InitSystem.RUNIT), \ patch("fenris.init_system._runit_disable") as mock_disable: disable_timer(now=False) mock_disable.assert_called_once_with(False) def test_collect_dispatches_to_systemd(self): """collect_now dispatches to systemd on systemd system.""" with patch("fenris.init_system.get_init_system", return_value=InitSystem.SYSTEMD), \ patch("fenris.init_system._systemd_collect") as mock_collect: collect_now() mock_collect.assert_called_once() def test_collect_dispatches_to_runit(self): """collect_now dispatches to runit on runit system.""" with patch("fenris.init_system.get_init_system", return_value=InitSystem.RUNIT), \ patch("fenris.init_system._runit_collect") as mock_collect: collect_now() mock_collect.assert_called_once() def test_query_state_dispatches_to_systemd(self): """query_service_state dispatches to systemd on systemd system.""" with patch("fenris.init_system.get_init_system", return_value=InitSystem.SYSTEMD), \ patch("fenris.init_system._systemd_query_state") as mock_query: query_service_state() mock_query.assert_called_once() def test_query_state_dispatches_to_runit(self): """query_service_state dispatches to runit on runit system.""" with patch("fenris.init_system.get_init_system", return_value=InitSystem.RUNIT), \ patch("fenris.init_system._runit_query_state") as mock_query: query_service_state() mock_query.assert_called_once() def test_journal_hint_dispatches_to_systemd(self): """journal_hint dispatches to systemd on systemd system.""" with patch("fenris.init_system.get_init_system", return_value=InitSystem.SYSTEMD), \ patch("fenris.init_system._systemd_journal_hint") as mock_hint: journal_hint(lines=3, unit="fenris-collect.service") mock_hint.assert_called_once_with(3) def test_journal_hint_dispatches_to_runit(self): """journal_hint dispatches to runit on runit system.""" with patch("fenris.init_system.get_init_system", return_value=InitSystem.RUNIT), \ patch("fenris.init_system._runit_journal_hint") as mock_hint: journal_hint(lines=3, unit="fenris-collect.service") mock_hint.assert_called_once_with(3) # --------------------------------------------------------------------------- # Constants and configuration # --------------------------------------------------------------------------- class TestConstants: """Test constants match spec requirements.""" def test_collect_timeout(self): """Collection timeout is 90 seconds (bounded execution).""" assert COLLECT_TIMEOUT_S == 90 def test_init_system_enum(self): """InitSystem enum has systemd and runit variants.""" assert InitSystem.SYSTEMD.value == "systemd" assert InitSystem.RUNIT.value == "runit" # --------------------------------------------------------------------------- # Integration with monitor.py # --------------------------------------------------------------------------- class TestMonitorIntegration: """Test that monitor.py uses the abstraction layer correctly.""" def test_cmd_enable_uses_init_system(self, tmp_path): """cmd_enable uses init_system.enable_timer.""" from fenris.monitor import cmd_enable from argparse import Namespace store_path = tmp_path / "observations.db" init_store(store_path) args = Namespace(now=True, store_path=store_path) with patch("fenris.monitor.enable_timer") as mock_enable: cmd_enable(args) mock_enable.assert_called_once_with(True) conn = sqlite3.connect(store_path) try: assert conn.execute( "SELECT COUNT(*) FROM monitoring_periods WHERE ended_at IS NULL" ).fetchone()[0] == 1 finally: conn.close() def test_cmd_disable_uses_init_system(self, tmp_path): """cmd_disable uses init_system.disable_timer.""" from fenris.monitor import cmd_disable from argparse import Namespace store_path = tmp_path / "observations.db" init_store(store_path) args = Namespace(now=True, store_path=store_path) with patch("fenris.monitor.disable_timer") as mock_disable: cmd_disable(args) mock_disable.assert_called_once_with(True) def test_cmd_collect_uses_init_system(self): """cmd_collect uses init_system.collect_now.""" from fenris.monitor import cmd_collect from argparse import Namespace args = Namespace() with patch("fenris.monitor.collect_now") as mock_collect: cmd_collect(args) mock_collect.assert_called_once() # --------------------------------------------------------------------------- # Edge cases # --------------------------------------------------------------------------- class TestEdgeCases: """Test edge cases and error handling.""" def test_systemd_enable_failure_produces_stderr(self): """Systemd enable failure produces error message on stderr.""" with patch("fenris.init_system.subprocess") as mock_sub: mock_sub.run.return_value = MagicMock( returncode=1, stderr="Permission denied" ) with pytest.raises(SystemExit): _systemd_enable(now=True) def test_runit_collect_missing_script(self): """Runit collect exits when fenris-collect script not found.""" with patch("fenris.init_system.subprocess") as mock_sub, \ patch("fenris.init_system.Path") as mock_path: missing = MagicMock() missing.exists.return_value = False # Fallback path is built via Path(...).parent... / "src" / ... missing.parent = missing missing.__truediv__.return_value = missing mock_path.return_value = missing mock_sub.run.return_value = MagicMock(returncode=1) with pytest.raises(SystemExit) as exc_info: _runit_collect() assert exc_info.value.code == 1 mock_sub.run.assert_not_called() def test_systemd_query_state_timeout(self): """Systemd query state handles subprocess timeout.""" with patch("fenris.init_system._systemctl_show") as mock_show: mock_show.return_value = {} result = _systemd_query_state() assert result["boot_enabled"] is None assert result["timer_active"] is None