xavierk/Fenris
1
0
Fork 0
Code Issues 7 Pull Requests Actions Packages Projects Releases 11 Wiki Activity
Files
e27052d09a384535bc460d7aa4f164d456514fa7
Fenris/tests/test_init_system.py
T
xavierkandCommandCodeBot e27052d09a Implement runit support for Fenris monitoring (issue #84) 
Deliver end-to-end native monitoring path under runit with existing
CLI/TUI controls and truthful status, preserving systemd behavior.

Changes:
- Add init system abstraction layer (src/fenris/init_system.py) that
  detects systemd vs runit and provides unified interface for timer
  control, on-demand collection, and service state queries
- Create runit service files (units/runit/) with completion-relative
  5-minute cadence, 2-minute boot delay, bounded execution (90s),
  no catch-up, and serialized runs via flock
- Update monitor.py to use abstraction layer instead of direct systemctl
- Update status.py to use abstraction layer for service state queries
- Update all packaging scripts (deb, rpm) for init-system-aware setup
- Update Makefile to install runit service files alongside systemd units
- Add 46 tests for init system abstraction layer

Spec: ADR 0008, §8.4, §8.5, §8.6, §8.7, §8.8
Closes #84

Co-authored-by: CommandCodeBot <noreply@commandcode.ai>
2026-09-15 00:15:24 +05:30

660 lines
25 KiB
Python
Raw Blame History

"""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 = {}
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):
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
# ---------------------------------------------------------------------------
# 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)
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.Path") as mock_path:
mock_path.return_value.exists.return_value = False
with pytest.raises(SystemExit) as exc_info:
_runit_collect()
assert exc_info.value.code == 1
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 False
assert result["timer_active"] is False
Reference in New Issue View Git Blame Copy Permalink