feat: start, stop, restart, enable and disable services, and list them in one round trip

napalm-device-types' SystemdServicesMixin (2.2.0) now provides get_services()
and manage_service(); this driver supplies only the transport (#7):

- _run_service_command(): unprivileged reads and root logins run as they
  are -- the user is asked once per session with "id -u", so a root login on a
  box without sudo is not prefixed with one. With a sudo password the command
  goes through _sudo(); without one through "sudo -n", which fails at once
  instead of hanging the session on a password prompt until the read timeout.
- get_services(): one round trip instead of an is-enabled and a show per unit
  (6.0 s -> 0.8 s on a 184-unit Ubuntu host). A host without systemd still
  falls back to "service --status-all".
- manage_service(): when sudo wants a password netOrk does not have, the
  failure says how to fix it -- the same hint the apt and SNMP actions give,
  now one constant (_SUDO_PASSWORD_HINT) instead of two copies.

OpenMediaVault and QNAP inherit this driver. OMV gets service control with
it; QNAP opts out (napalm-qnap-qts), since QTS has no systemd.

README: the sudo option is sudo_password, not secret; manage_service and the
systemctl permissions are listed.

Closes #7
This commit is contained in:
Christian Manivong
2026-10-05 13:12:13 +02:00
parent 31b8a37895
commit 84717ff53b
4 changed files with 176 additions and 54 deletions
+102
View File
@@ -1180,3 +1180,105 @@ def test_get_kernel_facts_raises_on_output_without_a_report(driver):
with patch.object(driver, "_send", return_value="sh: base64: not found"):
with pytest.raises(ValueError):
driver.get_kernel_facts()
# ---------------------------------------------------------------------------
# Services: listed in one round trip, controlled through systemctl (#7)
# ---------------------------------------------------------------------------
_REPORT = (
"SVC_BEGIN\n[files]\ncron.service enabled enabled\n[units]\n"
"MainPID=640\nId=cron.service\nNames=cron.service\nLoadState=loaded\n"
"ActiveState=active\nSubState=running\nUnitFileState=enabled\n"
"[generated]\nSVC_END\n"
)
class TestGetServices:
def test_one_command_lists_every_service(self, driver):
driver._device.send_command.return_value = _REPORT
services = driver.get_services()
assert services == [{"name": "cron", "running": True, "enabled": True, "pid": 640}]
assert driver._device.send_command.call_count == 1
assert "systemctl show" in driver._device.send_command.call_args[0][0]
def test_a_host_without_systemd_falls_back_to_service(self, driver):
driver._device.send_command.side_effect = [
"SVC_BEGIN\n[no-systemd]\n[files]\n[units]\n[generated]\nSVC_END\n",
" [ + ] cron\n [ - ] rsync\n",
]
services = driver.get_services()
assert {s["name"]: s["running"] for s in services} == {"cron": True, "rsync": False}
assert "service --status-all" in driver._device.send_command.call_args[0][0]
class TestManageService:
def _sent(self, driver) -> list[str]:
return [c[0][0] for c in driver._device.send_command.call_args_list]
def test_as_root_the_command_runs_as_it_is(self, driver):
driver._device.send_command.side_effect = ["0", "__SVC_RC=0"]
assert driver.manage_service("cron", "restart") == {"success": True, "output": ""}
uid, action = self._sent(driver)
assert uid == "id -u"
assert action.startswith("timeout 45 systemctl --no-ask-password restart -- cron.service")
def test_with_a_sudo_password_it_goes_through_sudo(self, driver):
driver._sudo_password = "pw" # noqa: S105
driver._device.send_command.side_effect = ["1000", "__SVC_RC=0"]
assert driver.manage_service("cron", "stop")["success"] is True
action = self._sent(driver)[1]
assert action.startswith("echo pw | sudo -S")
assert "timeout 45 systemctl --no-ask-password stop -- cron.service" in action
def test_without_one_sudo_never_waits_for_a_password(self, driver):
driver._device.send_command.side_effect = ["1000", "__SVC_RC=0"]
driver.manage_service("cron", "enable")
assert self._sent(driver)[1].startswith("sudo -n timeout 45 systemctl")
def test_a_missing_sudo_password_is_explained(self, driver):
driver._device.send_command.side_effect = [
"1000",
"sudo: a password is required\n__SVC_RC=1",
]
result = driver.manage_service("cron", "restart")
assert result["success"] is False
assert "sudo password" in result["output"]
assert "NOPASSWD" in result["output"]
def test_a_failure_keeps_systemctls_message(self, driver):
driver._device.send_command.side_effect = [
"0",
"Failed to start nope.service: Unit nope.service not found.\n__SVC_RC=5",
]
result = driver.manage_service("nope", "start")
assert result == {
"success": False,
"output": "Failed to start nope.service: Unit nope.service not found.",
}
def test_who_the_user_is_is_asked_once(self, driver):
driver._device.send_command.side_effect = ["0", "__SVC_RC=0", "__SVC_RC=0"]
driver.manage_service("cron", "stop")
driver.manage_service("cron", "start")
assert self._sent(driver).count("id -u") == 1
def test_an_invalid_name_is_refused_before_anything_is_sent(self, driver):
with pytest.raises(ValueError):
driver.manage_service("cron; reboot", "stop")
assert driver._device.send_command.call_count == 0