mirror of
https://github.com/kiviktnm/decman.git
synced 2026-09-19 12:08:28 +00:00
Split plugins to seperate packages
This commit is contained in:
Kivi Kaitaniemi
@@ -0,0 +1,20 @@
|
||||
[project]
|
||||
name = "decman-systemd"
|
||||
version = "1.0.0"
|
||||
requires-python = ">=3.13"
|
||||
dependencies = ["decman==1.0.0"]
|
||||
|
||||
[project.entry-points."decman.plugins"]
|
||||
systemd = "decman.plugins.systemd:Systemd"
|
||||
|
||||
[build-system]
|
||||
requires = ["setuptools>=61.0"]
|
||||
build-backend = "setuptools.build_meta"
|
||||
|
||||
[tool.setuptools.packages.find]
|
||||
where = ["src"]
|
||||
namespaces = true
|
||||
include = ["decman.plugins*"]
|
||||
|
||||
[tool.pytest.ini_options]
|
||||
testpaths = ["tests"]
|
||||
@@ -0,0 +1,253 @@
|
||||
import shutil
|
||||
|
||||
import decman.config as config
|
||||
import decman.core.command as command
|
||||
import decman.core.error as errors
|
||||
import decman.core.module as module
|
||||
import decman.core.output as output
|
||||
import decman.core.store as _store
|
||||
import decman.plugins as plugins
|
||||
|
||||
|
||||
def units(fn):
|
||||
"""
|
||||
Annotate that this function returns a set of systemd unit names that should be enabled.
|
||||
|
||||
Return type of ``fn``: ``set[str]``
|
||||
"""
|
||||
fn.__systemd__units__ = True
|
||||
return fn
|
||||
|
||||
|
||||
def user_units(fn):
|
||||
"""
|
||||
Annotate that this function returns a dict of users and systemd user unit names that should be
|
||||
enabled.
|
||||
|
||||
Return type of ``fn``: ``dict[str, set[str]]``
|
||||
"""
|
||||
fn.__systemd__user__units__ = True
|
||||
return fn
|
||||
|
||||
|
||||
class SystemdCommands:
|
||||
"""
|
||||
Default commands for the Systemd plugin.
|
||||
"""
|
||||
|
||||
def enable_units(self, units: set[str]) -> list[str]:
|
||||
"""
|
||||
Running this command enables the given systemd units.
|
||||
"""
|
||||
return ["systemctl", "enable"] + list(units)
|
||||
|
||||
def disable_units(self, units: set[str]) -> list[str]:
|
||||
"""
|
||||
Running this command disables the given systemd units.
|
||||
"""
|
||||
return ["systemctl", "disable"] + list(units)
|
||||
|
||||
def enable_user_units(self, units: set[str], user: str) -> list[str]:
|
||||
"""
|
||||
Running this command enables the given systemd units for the user.
|
||||
"""
|
||||
return ["systemctl", "--user", "-M", f"{user}@", "enable"] + list(units)
|
||||
|
||||
def disable_user_units(self, units: set[str], user: str) -> list[str]:
|
||||
"""
|
||||
Running this command disables the given systemd units for the user.
|
||||
"""
|
||||
return ["systemctl", "--user", "-M", f"{user}@", "disable"] + list(units)
|
||||
|
||||
def daemon_reload(self) -> list[str]:
|
||||
"""
|
||||
Running this command reloads the systemd daemon.
|
||||
"""
|
||||
return ["systemctl", "daemon-reload"]
|
||||
|
||||
def user_daemon_reload(self, user: str) -> list[str]:
|
||||
"""
|
||||
Running this command reloads the systemd daemon for the given user.
|
||||
"""
|
||||
return ["systemctl", "--user", "-M", f"{user}@", "daemon-reload"]
|
||||
|
||||
|
||||
class Systemd(plugins.Plugin):
|
||||
NAME = "systemd"
|
||||
|
||||
def __init__(self) -> None:
|
||||
self.enabled_units: set[str] = set()
|
||||
self.enabled_user_units: dict[str, set[str]] = {}
|
||||
self.commands = SystemdCommands()
|
||||
|
||||
def available(self) -> bool:
|
||||
return shutil.which("systemctl") is not None
|
||||
|
||||
def process_modules(self, store: _store.Store, modules: set[module.Module]):
|
||||
# These store keys are used to track changes in modules.
|
||||
# This way when these change, module can be marked as changed
|
||||
store.ensure("systemd_units_for_module", {})
|
||||
store.ensure("systemd_user_units_for_module", {})
|
||||
|
||||
for mod in modules:
|
||||
store["systemd_units_for_module"].setdefault(mod.name, set())
|
||||
store["systemd_user_units_for_module"].setdefault(mod.name, {})
|
||||
|
||||
units = plugins.run_method_with_attribute(mod, "__systemd__units__") or set()
|
||||
user_units = plugins.run_method_with_attribute(mod, "__systemd__user__units__") or {}
|
||||
|
||||
if store["systemd_units_for_module"][mod.name] != units:
|
||||
mod._changed = True
|
||||
output.print_debug(
|
||||
f"Module '{mod.name}' set to changed due to modified systemd units."
|
||||
)
|
||||
|
||||
if store["systemd_user_units_for_module"][mod.name] != user_units:
|
||||
mod._changed = True
|
||||
output.print_debug(
|
||||
f"Module '{mod.name}' set to changed due to modified systemd user units."
|
||||
)
|
||||
|
||||
self.enabled_units |= units
|
||||
for user, u_units in user_units.items():
|
||||
self.enabled_user_units.setdefault(user, set()).update(u_units)
|
||||
|
||||
store["systemd_units_for_module"][mod.name] = units
|
||||
store["systemd_user_units_for_module"][mod.name] = user_units
|
||||
|
||||
def apply(
|
||||
self, store: _store.Store, dry_run: bool = False, params: list[str] | None = None
|
||||
) -> bool:
|
||||
store.ensure("systemd_units", set())
|
||||
store.ensure("systemd_user_units", {})
|
||||
|
||||
units_to_enable = set()
|
||||
units_to_disable = set()
|
||||
user_units_to_enable: dict[str, set[str]] = {}
|
||||
user_units_to_disable: dict[str, set[str]] = {}
|
||||
|
||||
for unit in self.enabled_units:
|
||||
if unit not in store["systemd_units"]:
|
||||
units_to_enable.add(unit)
|
||||
|
||||
for unit in store["systemd_units"]:
|
||||
if unit not in self.enabled_units:
|
||||
units_to_disable.add(unit)
|
||||
|
||||
for user, units in self.enabled_user_units.items():
|
||||
store["systemd_user_units"].setdefault(user, set())
|
||||
user_units_to_enable.setdefault(user, set())
|
||||
|
||||
for unit in units:
|
||||
if unit not in store["systemd_user_units"][user]:
|
||||
user_units_to_enable[user].add(unit)
|
||||
|
||||
for user, units in store["systemd_user_units"].items():
|
||||
self.enabled_user_units.setdefault(user, set())
|
||||
user_units_to_disable.setdefault(user, set())
|
||||
|
||||
for unit in units:
|
||||
if unit not in self.enabled_user_units[user]:
|
||||
user_units_to_disable[user].add(unit)
|
||||
|
||||
try:
|
||||
output.print_info("Reloading systemd daemon.")
|
||||
if not dry_run:
|
||||
self.reload_daemon()
|
||||
|
||||
output.print_info("Reloading systemd daemon for users.")
|
||||
if not dry_run:
|
||||
for user in user_units_to_enable.keys() | user_units_to_disable.keys():
|
||||
self.reload_user_daemon(user)
|
||||
|
||||
output.print_list("Enabling systemd units:", list(units_to_enable))
|
||||
if not dry_run:
|
||||
self.enable_units(store, units_to_enable)
|
||||
|
||||
output.print_list("Disabling systemd units:", list(units_to_disable))
|
||||
if not dry_run:
|
||||
self.disable_units(store, units_to_disable)
|
||||
|
||||
for user, units in user_units_to_enable.items():
|
||||
output.print_list(f"Enabling systemd units for {user}:", list(units))
|
||||
if not dry_run:
|
||||
self.enable_user_units(store, units, user)
|
||||
|
||||
for user, units in user_units_to_disable.items():
|
||||
output.print_list(f"Disabling systemd units for {user}:", list(units))
|
||||
if not dry_run:
|
||||
self.disable_user_units(store, units, user)
|
||||
except errors.CommandFailedError as error:
|
||||
output.print_error("Running a systemd command failed.")
|
||||
output.print_error(str(error))
|
||||
if error.output:
|
||||
output.print_command_output(error.output)
|
||||
output.print_traceback()
|
||||
return False
|
||||
return True
|
||||
|
||||
def enable_units(self, store: _store.Store, units: set[str]):
|
||||
"""
|
||||
Enables the given units.
|
||||
"""
|
||||
if not units:
|
||||
return
|
||||
|
||||
cmd = self.commands.enable_units(units)
|
||||
command.prg(cmd, pty=config.debug_output)
|
||||
|
||||
store["systemd_units"] |= units
|
||||
|
||||
def disable_units(self, store: _store.Store, units: set[str]):
|
||||
"""
|
||||
Disables the given units.
|
||||
"""
|
||||
if not units:
|
||||
return
|
||||
|
||||
cmd = self.commands.disable_units(units)
|
||||
command.prg(cmd, pty=config.debug_output)
|
||||
|
||||
store["systemd_units"] -= units
|
||||
|
||||
def enable_user_units(self, store: _store.Store, units: set[str], user: str):
|
||||
"""
|
||||
Enables the given units for the given user.
|
||||
"""
|
||||
if not units:
|
||||
return
|
||||
|
||||
cmd = self.commands.enable_user_units(units, user)
|
||||
command.prg(cmd, pty=config.debug_output)
|
||||
|
||||
store["systemd_user_units"].setdefault(user, set())
|
||||
store["systemd_user_units"][user] |= units
|
||||
|
||||
def disable_user_units(self, store: _store.Store, units: set[str], user: str):
|
||||
"""
|
||||
Disables the given units for the given user.
|
||||
"""
|
||||
if not units:
|
||||
return
|
||||
|
||||
cmd = self.commands.disable_user_units(units, user)
|
||||
command.prg(cmd, pty=config.debug_output)
|
||||
|
||||
store["systemd_user_units"].setdefault(user, set())
|
||||
store["systemd_user_units"][user] -= units
|
||||
|
||||
def reload_user_daemon(self, user: str):
|
||||
"""
|
||||
Reloads the user's systemd daemon.
|
||||
"""
|
||||
|
||||
cmd = self.commands.user_daemon_reload(user)
|
||||
command.prg(cmd, pty=config.debug_output)
|
||||
|
||||
def reload_daemon(self):
|
||||
"""
|
||||
Reloads the systemd daemon.
|
||||
"""
|
||||
|
||||
cmd = self.commands.daemon_reload()
|
||||
command.prg(cmd, pty=config.debug_output)
|
||||
@@ -0,0 +1,354 @@
|
||||
import pytest
|
||||
|
||||
from decman.plugins import systemd as systemd_mod
|
||||
|
||||
|
||||
class DummyStore(dict):
|
||||
def ensure(self, key, default):
|
||||
if key not in self:
|
||||
self[key] = default
|
||||
|
||||
|
||||
class DummyModule:
|
||||
def __init__(self, name: str):
|
||||
self.name = name
|
||||
self._changed = False
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def store():
|
||||
return DummyStore()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def systemd():
|
||||
return systemd_mod.Systemd()
|
||||
|
||||
|
||||
def test_units_decorator_sets_attribute():
|
||||
@systemd_mod.units
|
||||
def fn():
|
||||
pass
|
||||
|
||||
assert getattr(fn, "__systemd__units__", False) is True
|
||||
|
||||
|
||||
def test_user_units_decorator_sets_attribute():
|
||||
@systemd_mod.user_units
|
||||
def fn():
|
||||
pass
|
||||
|
||||
assert getattr(fn, "__systemd__user__units__", False) is True
|
||||
|
||||
|
||||
def test_available_true_if_systemctl_found(monkeypatch, systemd):
|
||||
called = {}
|
||||
|
||||
def fake_which(name):
|
||||
called["name"] = name
|
||||
return "/bin/systemctl"
|
||||
|
||||
monkeypatch.setattr(systemd_mod.shutil, "which", fake_which)
|
||||
assert systemd.available() is True
|
||||
assert called["name"] == "systemctl"
|
||||
|
||||
|
||||
def test_available_false_if_systemctl_missing(monkeypatch, systemd):
|
||||
monkeypatch.setattr(systemd_mod.shutil, "which", lambda name: None)
|
||||
assert systemd.available() is False
|
||||
|
||||
|
||||
def test_process_modules_marks_changed_and_updates_store(monkeypatch, store, systemd):
|
||||
# initial store empty; ensure keys will be created
|
||||
m1 = DummyModule("mod1")
|
||||
m2 = DummyModule("mod2")
|
||||
|
||||
def fake_run_method(mod, attr):
|
||||
if mod is m1 and attr == "__systemd__units__":
|
||||
return {"a.service"}
|
||||
if mod is m1 and attr == "__systemd__user__units__":
|
||||
return {"alice": {"u1.service"}}
|
||||
# m2 has no units
|
||||
return None
|
||||
|
||||
monkeypatch.setattr(systemd_mod.plugins, "run_method_with_attribute", fake_run_method)
|
||||
|
||||
systemd.process_modules(store, {m1, m2})
|
||||
|
||||
# m1 changed from default -> marked _changed
|
||||
assert m1._changed is True
|
||||
# m2 had no units
|
||||
assert m2._changed is False
|
||||
|
||||
# enabled units aggregated
|
||||
assert systemd.enabled_units == {"a.service"}
|
||||
assert systemd.enabled_user_units == {"alice": {"u1.service"}}
|
||||
|
||||
# store updated per module
|
||||
assert store["systemd_units_for_module"]["mod1"] == {"a.service"}
|
||||
assert store["systemd_user_units_for_module"]["mod1"] == {"alice": {"u1.service"}}
|
||||
assert store["systemd_units_for_module"]["mod2"] == set()
|
||||
assert store["systemd_user_units_for_module"]["mod2"] == {}
|
||||
|
||||
|
||||
def test_process_modules_no_change_second_run(monkeypatch, store, systemd):
|
||||
m1 = DummyModule("mod1")
|
||||
|
||||
def fake_run_method(mod, attr):
|
||||
if attr == "__systemd__units__":
|
||||
return {"a.service"}
|
||||
if attr == "__systemd__user__units__":
|
||||
return {"alice": {"u1.service"}}
|
||||
return None
|
||||
|
||||
monkeypatch.setattr(systemd_mod.plugins, "run_method_with_attribute", fake_run_method)
|
||||
|
||||
# first run populates store
|
||||
systemd.process_modules(store, {m1})
|
||||
m1._changed = False
|
||||
|
||||
# new instance (fresh per-process in real usage)
|
||||
systemd2 = systemd_mod.Systemd()
|
||||
monkeypatch.setattr(systemd_mod.plugins, "run_method_with_attribute", fake_run_method)
|
||||
|
||||
systemd2.process_modules(store, {m1})
|
||||
|
||||
# values in store are same -> _changed stays False
|
||||
assert m1._changed is False
|
||||
|
||||
|
||||
def test_apply_enables_and_disables_units_and_user_units(store):
|
||||
s = systemd_mod.Systemd()
|
||||
|
||||
# Current enabled according to modules
|
||||
s.enabled_units = {"new.service"}
|
||||
s.enabled_user_units = {"alice": {"newuser.service"}}
|
||||
|
||||
# Store says we had an old unit enabled before
|
||||
store["systemd_units"] = {"old.service"}
|
||||
store["systemd_user_units"] = {"alice": {"olduser.service"}}
|
||||
|
||||
calls = []
|
||||
|
||||
def fake_reload_daemon():
|
||||
calls.append(("reload_daemon",))
|
||||
|
||||
def fake_reload_user_daemon(user):
|
||||
calls.append(("reload_user_daemon", user))
|
||||
|
||||
def fake_enable_units(store_arg, units_arg):
|
||||
calls.append(("enable_units", frozenset(units_arg)))
|
||||
store_arg["systemd_units"] |= units_arg
|
||||
|
||||
def fake_disable_units(store_arg, units_arg):
|
||||
calls.append(("disable_units", frozenset(units_arg)))
|
||||
store_arg["systemd_units"] -= units_arg
|
||||
|
||||
def fake_enable_user_units(store_arg, units_arg, user):
|
||||
calls.append(("enable_user_units", user, frozenset(units_arg)))
|
||||
store_arg["systemd_user_units"].setdefault(user, set()).update(units_arg)
|
||||
|
||||
def fake_disable_user_units(store_arg, units_arg, user):
|
||||
calls.append(("disable_user_units", user, frozenset(units_arg)))
|
||||
store_arg["systemd_user_units"].setdefault(user, set()).difference_update(units_arg)
|
||||
|
||||
# patch instance methods (no self parameter expected)
|
||||
s.reload_daemon = fake_reload_daemon
|
||||
s.reload_user_daemon = fake_reload_user_daemon
|
||||
s.enable_units = fake_enable_units
|
||||
s.disable_units = fake_disable_units
|
||||
s.enable_user_units = fake_enable_user_units
|
||||
s.disable_user_units = fake_disable_user_units
|
||||
|
||||
result = s.apply(store, dry_run=False, params=None)
|
||||
|
||||
# reloads called once
|
||||
assert ("reload_daemon",) in calls
|
||||
assert ("reload_user_daemon", "alice") in calls
|
||||
|
||||
# enable/disable correct units
|
||||
assert ("enable_units", frozenset({"new.service"})) in calls
|
||||
assert ("disable_units", frozenset({"old.service"})) in calls
|
||||
assert ("enable_user_units", "alice", frozenset({"newuser.service"})) in calls
|
||||
assert ("disable_user_units", "alice", frozenset({"olduser.service"})) in calls
|
||||
|
||||
# store reconciled
|
||||
assert store["systemd_units"] == {"new.service"}
|
||||
assert store["systemd_user_units"]["alice"] == {"newuser.service"}
|
||||
|
||||
|
||||
def test_apply_dry_run_does_not_mutate_store_or_call_commands(store):
|
||||
s = systemd_mod.Systemd()
|
||||
s.enabled_units = {"new.service"}
|
||||
s.enabled_user_units = {"alice": {"newuser.service"}}
|
||||
|
||||
store["systemd_units"] = {"old.service"}
|
||||
store["systemd_user_units"] = {"alice": {"olduser.service"}}
|
||||
|
||||
called = {"reload": False, "enable": False, "disable": False}
|
||||
|
||||
s.reload_daemon = lambda: called.__setitem__("reload", True) or True
|
||||
s.reload_user_daemon = lambda user: called.__setitem__("reload", True) or True
|
||||
s.enable_units = lambda st, u: called.__setitem__("enable", True) or True
|
||||
s.disable_units = lambda st, u: called.__setitem__("disable", True) or True
|
||||
s.enable_user_units = lambda st, u, user: called.__setitem__("enable", True) or True
|
||||
s.disable_user_units = lambda st, u, user: called.__setitem__("disable", True) or True
|
||||
|
||||
result = s.apply(store, dry_run=True, params=None)
|
||||
assert result is True
|
||||
|
||||
# no commands should be called
|
||||
assert called == {"reload": False, "enable": False, "disable": False}
|
||||
|
||||
# store unchanged
|
||||
assert store["systemd_units"] == {"old.service"}
|
||||
assert store["systemd_user_units"]["alice"] == {"olduser.service"}
|
||||
|
||||
|
||||
def test_enable_units_success(monkeypatch, store, systemd):
|
||||
store["systemd_units"] = {"old.service"}
|
||||
|
||||
def fake_run(cmd, **kwargs):
|
||||
assert cmd[0] == "systemctl"
|
||||
assert cmd[1] == "enable"
|
||||
assert "new.service" in cmd[2:]
|
||||
return 0, "ok"
|
||||
|
||||
monkeypatch.setattr(systemd_mod.command, "run", fake_run)
|
||||
|
||||
systemd.enable_units(store, {"new.service"})
|
||||
assert store["systemd_units"] == {"old.service", "new.service"}
|
||||
|
||||
|
||||
def test_enable_units_failure_does_not_update_store(monkeypatch, store, systemd):
|
||||
store["systemd_units"] = {"old.service"}
|
||||
|
||||
def fake_run(cmd, **kwargs):
|
||||
return 1, "error"
|
||||
|
||||
monkeypatch.setattr(systemd_mod.command, "run", fake_run)
|
||||
|
||||
with pytest.raises(systemd_mod.errors.CommandFailedError):
|
||||
systemd.enable_units(store, {"new.service"})
|
||||
# unchanged
|
||||
assert store["systemd_units"] == {"old.service"}
|
||||
|
||||
|
||||
def test_disable_units_success(monkeypatch, store, systemd):
|
||||
store["systemd_units"] = {"old.service", "new.service"}
|
||||
|
||||
def fake_run(cmd, **kwargs):
|
||||
assert cmd[0] == "systemctl"
|
||||
assert cmd[1] == "disable"
|
||||
assert "new.service" in cmd[2:]
|
||||
return 0, "ok"
|
||||
|
||||
monkeypatch.setattr(systemd_mod.command, "run", fake_run)
|
||||
|
||||
systemd.disable_units(store, {"new.service"})
|
||||
assert store["systemd_units"] == {"old.service"}
|
||||
|
||||
|
||||
def test_disable_units_failure_does_not_update_store(monkeypatch, store, systemd):
|
||||
store["systemd_units"] = {"old.service", "new.service"}
|
||||
|
||||
def fake_run(cmd, **kwargs):
|
||||
return 1, "error"
|
||||
|
||||
monkeypatch.setattr(systemd_mod.command, "run", fake_run)
|
||||
|
||||
with pytest.raises(systemd_mod.errors.CommandFailedError):
|
||||
systemd.disable_units(store, {"new.service"})
|
||||
assert store["systemd_units"] == {"old.service", "new.service"}
|
||||
|
||||
|
||||
def test_enable_user_units_success(monkeypatch, store, systemd):
|
||||
store["systemd_user_units"] = {"alice": {"olduser.service"}}
|
||||
|
||||
def fake_run(cmd, **kwargs):
|
||||
assert cmd[0] == "systemctl"
|
||||
assert "--user" in cmd
|
||||
assert "enable" in cmd
|
||||
assert "newuser.service" in cmd
|
||||
return 0, "ok"
|
||||
|
||||
monkeypatch.setattr(systemd_mod.command, "run", fake_run)
|
||||
|
||||
systemd.enable_user_units(store, {"newuser.service"}, "alice")
|
||||
assert store["systemd_user_units"]["alice"] == {
|
||||
"olduser.service",
|
||||
"newuser.service",
|
||||
}
|
||||
|
||||
|
||||
def test_enable_user_units_failure_does_not_update_store(monkeypatch, store, systemd):
|
||||
store["systemd_user_units"] = {"alice": {"olduser.service"}}
|
||||
|
||||
def fake_run(cmd, **kwargs):
|
||||
return 1, "error"
|
||||
|
||||
monkeypatch.setattr(systemd_mod.command, "run", fake_run)
|
||||
|
||||
with pytest.raises(systemd_mod.errors.CommandFailedError):
|
||||
systemd.enable_user_units(store, {"newuser.service"}, "alice")
|
||||
assert store["systemd_user_units"]["alice"] == {"olduser.service"}
|
||||
|
||||
|
||||
def test_disable_user_units_success(monkeypatch, store, systemd):
|
||||
store["systemd_user_units"] = {"alice": {"olduser.service", "newuser.service"}}
|
||||
|
||||
def fake_run(cmd, **kwargs):
|
||||
assert cmd[0] == "systemctl"
|
||||
assert "--user" in cmd
|
||||
assert "disable" in cmd
|
||||
assert "newuser.service" in cmd
|
||||
return 0, "ok"
|
||||
|
||||
monkeypatch.setattr(systemd_mod.command, "run", fake_run)
|
||||
|
||||
systemd.disable_user_units(store, {"newuser.service"}, "alice")
|
||||
assert store["systemd_user_units"]["alice"] == {"olduser.service"}
|
||||
|
||||
|
||||
def test_disable_user_units_failure_does_not_update_store(monkeypatch, store, systemd):
|
||||
store["systemd_user_units"] = {"alice": {"olduser.service", "newuser.service"}}
|
||||
|
||||
def fake_run(cmd, **kwargs):
|
||||
return 1, "error"
|
||||
|
||||
monkeypatch.setattr(systemd_mod.command, "run", fake_run)
|
||||
|
||||
with pytest.raises(systemd_mod.errors.CommandFailedError):
|
||||
systemd.disable_user_units(store, {"newuser.service"}, "alice")
|
||||
|
||||
assert store["systemd_user_units"]["alice"] == {
|
||||
"olduser.service",
|
||||
"newuser.service",
|
||||
}
|
||||
|
||||
|
||||
def test_reload_daemon_uses_command_run(monkeypatch, systemd):
|
||||
called = {}
|
||||
|
||||
def fake_run(cmd, **kwargs):
|
||||
called["cmd"] = cmd
|
||||
return 0, "ok"
|
||||
|
||||
monkeypatch.setattr(systemd_mod.command, "run", fake_run)
|
||||
systemd.reload_daemon()
|
||||
assert called["cmd"][:2] == ["systemctl", "daemon-reload"]
|
||||
|
||||
|
||||
def test_reload_user_daemon_uses_command_run(monkeypatch, systemd):
|
||||
called = {}
|
||||
|
||||
def fake_run(cmd, **kwargs):
|
||||
called["cmd"] = cmd
|
||||
return 0, "ok"
|
||||
|
||||
monkeypatch.setattr(systemd_mod.command, "run", fake_run)
|
||||
systemd.reload_user_daemon("alice")
|
||||
cmd = called["cmd"]
|
||||
assert cmd[0] == "systemctl"
|
||||
assert "--user" in cmd
|
||||
assert "daemon-reload" in cmd
|
||||
Reference in New Issue
Block a user