"""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 sqlite3 from pathlib import Path from unittest.mock import patch, MagicMock import pytest import sys sys.path.insert(0, str(Path(__file__).parent.parent / "src")) from fenris.init_system import ( InitSystem, detect_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, 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