feat: implement systemd service commands (install, uninstall, start, stop, restart, status, logs)
This commit is contained in:
Brian Krabach
+59
-8
@@ -1,6 +1,8 @@
|
||||
"""muxplex/service.py — System service management (systemd on Linux, launchd on macOS)."""
|
||||
|
||||
import os
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
@@ -15,6 +17,22 @@ _LAUNCHD_PLIST_DIR: Path = Path.home() / "Library" / "LaunchAgents"
|
||||
_LAUNCHD_PLIST_PATH: Path = _LAUNCHD_PLIST_DIR / "com.muxplex.plist"
|
||||
_LAUNCHD_LABEL: str = "com.muxplex"
|
||||
|
||||
_SYSTEMD_UNIT_TEMPLATE = """\
|
||||
[Unit]
|
||||
Description=muxplex
|
||||
After=network.target
|
||||
|
||||
[Service]
|
||||
Type=simple
|
||||
ExecStart={exec_start}
|
||||
Restart=on-failure
|
||||
RestartSec=5s
|
||||
Environment=PATH={safe_path}
|
||||
|
||||
[Install]
|
||||
WantedBy=default.target
|
||||
"""
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Platform detection
|
||||
# ---------------------------------------------------------------------------
|
||||
@@ -38,36 +56,69 @@ def _resolve_muxplex_bin() -> str:
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Private stubs — systemd (Linux)
|
||||
# Helper
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _prompt_host_if_localhost() -> None:
|
||||
"""Prompt the user to change host from 127.0.0.1 to 0.0.0.0 for service use."""
|
||||
from muxplex.settings import load_settings, patch_settings
|
||||
|
||||
settings = load_settings()
|
||||
if settings["host"] == "127.0.0.1":
|
||||
answer = input(
|
||||
"Host is 127.0.0.1 — change to 0.0.0.0 so the service is reachable? [Y/n] "
|
||||
)
|
||||
if answer.strip().lower() in ("y", ""):
|
||||
patch_settings({"host": "0.0.0.0"})
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Private implementations — systemd (Linux)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _systemd_install() -> None:
|
||||
raise NotImplementedError("systemd install not implemented")
|
||||
muxplex_bin = _resolve_muxplex_bin()
|
||||
safe_path = os.environ.get("PATH", "/usr/local/bin:/usr/bin:/bin")
|
||||
exec_start = f"{muxplex_bin} serve"
|
||||
unit_content = _SYSTEMD_UNIT_TEMPLATE.format(
|
||||
exec_start=exec_start, safe_path=safe_path
|
||||
)
|
||||
_SYSTEMD_UNIT_DIR.mkdir(parents=True, exist_ok=True)
|
||||
_SYSTEMD_UNIT_PATH.write_text(unit_content)
|
||||
subprocess.run(["systemctl", "--user", "daemon-reload"], check=True)
|
||||
subprocess.run(["systemctl", "--user", "enable", "--now", "muxplex"], check=True)
|
||||
_prompt_host_if_localhost()
|
||||
|
||||
|
||||
def _systemd_uninstall() -> None:
|
||||
raise NotImplementedError("systemd uninstall not implemented")
|
||||
subprocess.run(["systemctl", "--user", "stop", "muxplex"], check=True)
|
||||
subprocess.run(["systemctl", "--user", "disable", "muxplex"], check=True)
|
||||
_SYSTEMD_UNIT_PATH.unlink(missing_ok=True)
|
||||
subprocess.run(["systemctl", "--user", "daemon-reload"], check=True)
|
||||
|
||||
|
||||
def _systemd_start() -> None:
|
||||
raise NotImplementedError("systemd start not implemented")
|
||||
subprocess.run(["systemctl", "--user", "start", "muxplex"], check=True)
|
||||
|
||||
|
||||
def _systemd_stop() -> None:
|
||||
raise NotImplementedError("systemd stop not implemented")
|
||||
subprocess.run(["systemctl", "--user", "stop", "muxplex"], check=True)
|
||||
|
||||
|
||||
def _systemd_restart() -> None:
|
||||
raise NotImplementedError("systemd restart not implemented")
|
||||
subprocess.run(["systemctl", "--user", "restart", "muxplex"], check=True)
|
||||
|
||||
|
||||
def _systemd_status() -> None:
|
||||
raise NotImplementedError("systemd status not implemented")
|
||||
subprocess.run(
|
||||
["systemctl", "--user", "status", "muxplex", "--no-pager"], check=True
|
||||
)
|
||||
|
||||
|
||||
def _systemd_logs() -> None:
|
||||
raise NotImplementedError("systemd logs not implemented")
|
||||
subprocess.run(["journalctl", "--user", "-u", "muxplex", "-f"], check=True)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
"""Tests for muxplex/service.py — system service management module."""
|
||||
|
||||
import subprocess
|
||||
import sys
|
||||
|
||||
|
||||
@@ -34,3 +35,116 @@ def test_resolve_muxplex_bin():
|
||||
result = _resolve_muxplex_bin()
|
||||
assert isinstance(result, str)
|
||||
assert "muxplex" in result or "python" in result
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# systemd tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_systemd_install_writes_unit_and_enables(monkeypatch, tmp_path):
|
||||
"""_systemd_install writes unit file with 'muxplex serve' (no --host/--port)
|
||||
and calls daemon-reload + enable --now."""
|
||||
import muxplex.service as svc
|
||||
|
||||
unit_dir = tmp_path / "systemd" / "user"
|
||||
unit_path = unit_dir / "muxplex.service"
|
||||
|
||||
monkeypatch.setattr(svc, "_SYSTEMD_UNIT_DIR", unit_dir)
|
||||
monkeypatch.setattr(svc, "_SYSTEMD_UNIT_PATH", unit_path)
|
||||
|
||||
calls = []
|
||||
monkeypatch.setattr(subprocess, "run", lambda cmd, **kw: calls.append(list(cmd)))
|
||||
|
||||
# Avoid interactive prompt
|
||||
monkeypatch.setattr(svc, "_prompt_host_if_localhost", lambda: None)
|
||||
|
||||
svc._systemd_install()
|
||||
|
||||
# Unit file must exist and contain the right content
|
||||
assert unit_path.exists(), "unit file was not written"
|
||||
content = unit_path.read_text()
|
||||
assert "muxplex" in content, "unit file must mention 'muxplex'"
|
||||
assert "serve" in content, "unit file ExecStart must include 'serve'"
|
||||
assert "--host" not in content, "ExecStart must NOT contain --host"
|
||||
assert "--port" not in content, "ExecStart must NOT contain --port"
|
||||
|
||||
# daemon-reload must be called
|
||||
assert ["systemctl", "--user", "daemon-reload"] in calls, "daemon-reload not called"
|
||||
# enable --now must be called
|
||||
assert ["systemctl", "--user", "enable", "--now", "muxplex"] in calls, (
|
||||
"enable --now not called"
|
||||
)
|
||||
|
||||
|
||||
def test_systemd_uninstall_stops_disables_removes(monkeypatch, tmp_path):
|
||||
"""_systemd_uninstall calls stop, disable, daemon-reload and deletes the unit file."""
|
||||
import muxplex.service as svc
|
||||
|
||||
unit_dir = tmp_path / "systemd" / "user"
|
||||
unit_dir.mkdir(parents=True)
|
||||
unit_path = unit_dir / "muxplex.service"
|
||||
unit_path.write_text("[Unit]\nDescription=muxplex\n")
|
||||
|
||||
monkeypatch.setattr(svc, "_SYSTEMD_UNIT_DIR", unit_dir)
|
||||
monkeypatch.setattr(svc, "_SYSTEMD_UNIT_PATH", unit_path)
|
||||
|
||||
calls = []
|
||||
monkeypatch.setattr(subprocess, "run", lambda cmd, **kw: calls.append(list(cmd)))
|
||||
|
||||
svc._systemd_uninstall()
|
||||
|
||||
assert ["systemctl", "--user", "stop", "muxplex"] in calls, "stop not called"
|
||||
assert ["systemctl", "--user", "disable", "muxplex"] in calls, "disable not called"
|
||||
assert ["systemctl", "--user", "daemon-reload"] in calls, "daemon-reload not called"
|
||||
assert not unit_path.exists(), "unit file was not deleted"
|
||||
|
||||
|
||||
def test_systemd_start_calls_systemctl(monkeypatch):
|
||||
"""_systemd_start runs ['systemctl', '--user', 'start', 'muxplex']."""
|
||||
import muxplex.service as svc
|
||||
|
||||
calls = []
|
||||
monkeypatch.setattr(subprocess, "run", lambda cmd, **kw: calls.append(list(cmd)))
|
||||
svc._systemd_start()
|
||||
assert ["systemctl", "--user", "start", "muxplex"] in calls
|
||||
|
||||
|
||||
def test_systemd_stop_calls_systemctl(monkeypatch):
|
||||
"""_systemd_stop runs ['systemctl', '--user', 'stop', 'muxplex']."""
|
||||
import muxplex.service as svc
|
||||
|
||||
calls = []
|
||||
monkeypatch.setattr(subprocess, "run", lambda cmd, **kw: calls.append(list(cmd)))
|
||||
svc._systemd_stop()
|
||||
assert ["systemctl", "--user", "stop", "muxplex"] in calls
|
||||
|
||||
|
||||
def test_systemd_restart_calls_systemctl(monkeypatch):
|
||||
"""_systemd_restart runs ['systemctl', '--user', 'restart', 'muxplex']."""
|
||||
import muxplex.service as svc
|
||||
|
||||
calls = []
|
||||
monkeypatch.setattr(subprocess, "run", lambda cmd, **kw: calls.append(list(cmd)))
|
||||
svc._systemd_restart()
|
||||
assert ["systemctl", "--user", "restart", "muxplex"] in calls
|
||||
|
||||
|
||||
def test_systemd_status_calls_systemctl(monkeypatch):
|
||||
"""_systemd_status runs ['systemctl', '--user', 'status', 'muxplex', '--no-pager']."""
|
||||
import muxplex.service as svc
|
||||
|
||||
calls = []
|
||||
monkeypatch.setattr(subprocess, "run", lambda cmd, **kw: calls.append(list(cmd)))
|
||||
svc._systemd_status()
|
||||
assert ["systemctl", "--user", "status", "muxplex", "--no-pager"] in calls
|
||||
|
||||
|
||||
def test_systemd_logs_calls_journalctl(monkeypatch):
|
||||
"""_systemd_logs runs ['journalctl', '--user', '-u', 'muxplex', '-f']."""
|
||||
import muxplex.service as svc
|
||||
|
||||
calls = []
|
||||
monkeypatch.setattr(subprocess, "run", lambda cmd, **kw: calls.append(list(cmd)))
|
||||
svc._systemd_logs()
|
||||
assert ["journalctl", "--user", "-u", "muxplex", "-f"] in calls
|
||||
|
||||
Reference in New Issue
Block a user