feat: list systemd services in one round trip and control them, once for every driver

Listing a host's services and starting or stopping one is the same on every host that runs systemd, so the command, its parse, the check of a unit name and the reading of an action's exit status live here once, and a driver supplies only the transport (napalm-linux#7, napalm-proxmox#6): - SYSTEMD_SERVICES_COMMAND: one read-only POSIX sh line. list-unit-files, then one systemctl show over every loaded service unit (Id, Names, LoadState, ActiveState, SubState, UnitFileState, MainPID), and is-enabled only for generated units, whose boot state lives in a SysV script's rc links. Framed; [no-systemd] when /run/systemd/system is missing. Replaces an is-enabled and a show per unit: 0.8 s instead of 6 s on a 180-unit Ubuntu host. - parse_systemd_services(): loaded units except not-found, plus installed unit files that are not loaded; no templates, no aliases (also not the ones older systemd lists as "enabled"). enabled = UnitFileState enabled or enabled-runtime, read from systemctl show and never from list-unit-files' second column, which has had a preset column after it since systemd 245. A report whose end is missing raises ValueError, so a list cut short never reads as services that went away; a host without systemd raises SystemdUnavailable, a NotImplementedError, so a driver can fall back. - unit_name() / service_action_command() / parse_action_result(): template instances, dots, colons and \xHH escapes accepted; a bare template, a leading "-" and anything a shell reads refused. The action runs as "timeout 45 systemctl --no-ask-password <action> -- <unit>.service" with its exit status printed after it; only that status decides, 124 is not called done, and terminal colour codes are dropped. The marker also keeps the output from ever being empty, which a transport that retries on an empty answer would take as a reason to run the action twice. - SystemdServicesMixin, in the template form: get_services() and manage_service() are concrete, _run_service_command(command, *, privileged, timeout) is the driver's hook. Mixed in by the drivers whose host runs systemd, not by OSDriver. Version 2.2.0.
2026-10-05 13:11:44 +02:00
parent d55b036a8e
commit 17d8dabb4d
5 changed files with 764 additions and 1 deletions
@@ -0,0 +1,412 @@
+"""systemd services: listing them in one round trip, and starting and stopping them.
+
+What systemd reports, and how a unit is started or stopped, is the same on
+every host that runs it -- so the command, its parse, the name check and the
+reading of the exit status live here once, and a driver only carries a command
+across (napalm-linux#7, napalm-proxmox#6).
+"""
+
+from __future__ import annotations
+
+import os
+import subprocess
+
+import pytest
+
+from napalm_device_types import OSDriver
+from napalm_device_types.systemd import (
+    ACTION_TIMEOUT,
+    SERVICE_ACTIONS,
+    SYSTEMD_SERVICES_COMMAND,
+    SystemdServicesMixin,
+    SystemdUnavailable,
+    parse_action_result,
+    parse_systemd_services,
+    service_action_command,
+    unit_name,
+)
+
+FILES = """\
+apparmor.service                             enabled         enabled
+ssh.service                                  enabled         enabled
+sshd.service                                 alias           -
+getty@.service                               enabled         enabled
+rsync.service                                disabled        enabled
+cups.service                                 indirect        enabled
+plymouth-quit.service                        static          -
+systemd-networkd-wait-online.service         enabled-runtime enabled
+nfs-server.service                           masked          enabled
+"""
+
+UNITS = """\
+MainPID=812
+Id=ssh.service
+Names=ssh.service sshd.service
+LoadState=loaded
+ActiveState=active
+SubState=running
+UnitFileState=enabled
+
+MainPID=0
+Id=apparmor.service
+Names=apparmor.service
+LoadState=loaded
+ActiveState=active
+SubState=exited
+UnitFileState=enabled
+
+MainPID=900
+Id=getty@tty1.service
+Names=getty@tty1.service
+LoadState=loaded
+ActiveState=active
+SubState=running
+UnitFileState=enabled
+
+MainPID=0
+Id=systemd-fsck@dev-disk-by\\x2dlabel-BOOT.service
+Names=systemd-fsck@dev-disk-by\\x2dlabel-BOOT.service
+LoadState=loaded
+ActiveState=inactive
+SubState=dead
+UnitFileState=static
+
+MainPID=0
+Id=display-manager.service
+Names=display-manager.service
+LoadState=not-found
+ActiveState=inactive
+SubState=dead
+UnitFileState=
+
+MainPID=0
+Id=nfs-server.service
+Names=nfs-server.service
+LoadState=masked
+ActiveState=inactive
+SubState=dead
+UnitFileState=masked
+
+MainPID=0
+Id=systemd-networkd-wait-online.service
+Names=systemd-networkd-wait-online.service
+LoadState=loaded
+ActiveState=active
+SubState=exited
+UnitFileState=enabled-runtime
+"""
+
+
+GENERATED_UNIT = """
+MainPID=0
+Id=rrdcached.service
+Names=rrdcached.service
+LoadState=loaded
+ActiveState=active
+SubState=running
+UnitFileState=generated
+"""
+
+
+def _wire(
+    files: str = FILES,
+    units: str = UNITS,
+    *,
+    generated: str = "",
+    noise: str = "",
+    end: bool = True,
+) -> str:
+    """The report as the command prints it, framed."""
+    tail = "SVC_END\n" if end else ""
+    return f"{noise}SVC_BEGIN\n[files]\n{files}[units]\n{units}[generated]\n{generated}{tail}"
+
+
+def _by_name(services):
+    return {s["name"]: s for s in services}
+
+
+class TestParseSystemdServices:
+    def test_a_loaded_unit_is_read_with_its_state(self):
+        services = _by_name(parse_systemd_services(_wire()))
+
+        assert services["ssh"] == {"name": "ssh", "running": True, "enabled": True, "pid": 812}
+        assert services["apparmor"] == {
+            "name": "apparmor",
+            "running": False,
+            "enabled": True,
+            "pid": 0,
+        }
+
+    def test_enabled_means_enabled_now_not_merely_installed(self):
+        services = _by_name(parse_systemd_services(_wire()))
+
+        assert services["systemd-networkd-wait-online"]["enabled"] is True
+        assert services[r"systemd-fsck@dev-disk-by\x2dlabel-BOOT"]["enabled"] is False
+        assert services["nfs-server"]["enabled"] is False
+
+    def test_a_generated_unit_takes_its_boot_state_from_is_enabled(self):
+        """A SysV script's unit is generated; only is-enabled knows its rc links."""
+        raw = _wire(units=UNITS + GENERATED_UNIT, generated="rrdcached.service enabled\n")
+
+        assert _by_name(parse_systemd_services(raw))["rrdcached"]["enabled"] is True
+
+    def test_a_generated_unit_is_not_enabled_unless_is_enabled_says_so(self):
+        raw = _wire(units=UNITS + GENERATED_UNIT, generated="rrdcached.service disabled\n")
+
+        assert _by_name(parse_systemd_services(raw))["rrdcached"]["enabled"] is False
+
+    def test_a_unit_that_is_not_there_is_left_out(self):
+        assert "display-manager" not in _by_name(parse_systemd_services(_wire()))
+
+    def test_an_installed_unit_that_is_not_loaded_is_listed(self):
+        services = _by_name(parse_systemd_services(_wire()))
+
+        assert services["rsync"] == {"name": "rsync", "running": False, "enabled": False, "pid": 0}
+        assert services["cups"]["enabled"] is False
+
+    def test_templates_and_static_files_that_are_not_loaded_are_not(self):
+        services = _by_name(parse_systemd_services(_wire()))
+
+        assert "getty@" not in services
+        assert "plymouth-quit" not in services
+        assert services["getty@tty1"]["running"] is True
+
+    def test_an_alias_never_appears_beside_its_unit(self):
+        assert "sshd" not in _by_name(parse_systemd_services(_wire()))
+
+    def test_an_alias_that_older_systemd_calls_enabled_does_not_either(self):
+        files = (
+            "\n".join(
+                "sshd.service enabled enabled" if line.startswith("sshd.service") else line
+                for line in FILES.splitlines()
+            )
+            + "\n"
+        )
+
+        assert "sshd" not in _by_name(parse_systemd_services(_wire(files=files)))
+
+    def test_an_escaped_name_survives(self):
+        assert r"systemd-fsck@dev-disk-by\x2dlabel-BOOT" in _by_name(
+            parse_systemd_services(_wire())
+        )
+
+    def test_blocks_run_together_are_still_told_apart(self):
+        """xargs may split the unit list across two systemctl runs."""
+        units = UNITS.replace(
+            "UnitFileState=enabled\n\nMainPID=0\nId=apparmor",
+            "UnitFileState=enabled\nMainPID=0\nId=apparmor",
+        )
+
+        services = _by_name(parse_systemd_services(_wire(units=units)))
+
+        assert services["ssh"]["pid"] == 812
+        assert services["apparmor"]["running"] is False
+
+    def test_the_list_is_sorted_by_name(self):
+        names = [s["name"] for s in parse_systemd_services(_wire())]
+
+        assert names == sorted(names)
+
+    def test_terminal_colours_in_the_report_are_dropped(self):
+        files = FILES.replace(
+            "rsync.service                                disabled",
+            "rsync.service \x1b[0;1;31mdisabled\x1b[0m",
+        )
+
+        assert "rsync" in _by_name(parse_systemd_services(_wire(files=files)))
+
+    def test_whatever_surrounds_the_frame_is_ignored(self):
+        noisy = _wire(noise="user@host:~$ systemctl ...\n") + "user@host:~$ "
+
+        assert "ssh" in _by_name(parse_systemd_services(noisy))
+
+    def test_a_cut_short_report_raises(self):
+        """A missing tail must not read as services that went away."""
+        with pytest.raises(ValueError):
+            parse_systemd_services(_wire(end=False))
+
+    def test_output_without_the_frame_raises(self):
+        with pytest.raises(ValueError):
+            parse_systemd_services("bash: systemctl: command not found\n")
+
+    def test_a_host_without_systemd_says_so(self):
+        raw = "SVC_BEGIN\n[no-systemd]\n[files]\n[units]\nSVC_END\n"
+
+        with pytest.raises(SystemdUnavailable):
+            parse_systemd_services(raw)
+
+    def test_no_systemd_is_a_not_implemented_error(self):
+        assert issubclass(SystemdUnavailable, NotImplementedError)
+
+
+class TestTheCommand:
+    def test_the_frame_is_not_in_the_command_itself(self):
+        """An echoing transport must not show the end marker early."""
+        assert "SVC_END" not in SYSTEMD_SERVICES_COMMAND
+        assert "SVC_BEGIN" not in SYSTEMD_SERVICES_COMMAND
+
+    def test_it_changes_nothing(self):
+        for verb in ("start", "stop", "restart", "enable", "disable", "mask"):
+            assert f"systemctl {verb}" not in SYSTEMD_SERVICES_COMMAND
+
+    @pytest.mark.skipif(not os.path.isdir("/run/systemd/system"), reason="needs systemd")
+    def test_it_runs_and_parses_on_this_host(self):
+        out = subprocess.run(
+            ["sh", "-c", SYSTEMD_SERVICES_COMMAND], capture_output=True, text=True, timeout=60
+        ).stdout
+
+        services = _by_name(parse_systemd_services(out))
+
+        assert "systemd-journald" in services
+        assert services["systemd-journald"]["running"] is True
+
+
+class TestUnitName:
+    @pytest.mark.parametrize(
+        ("raw", "name"),
+        [
+            ("ssh", "ssh"),
+            ("ssh.service", "ssh"),
+            ("getty@tty1", "getty@tty1"),
+            ("wg-quick@wg0", "wg-quick@wg0"),
+            ("snapd.apparmor", "snapd.apparmor"),
+            ("systemd-backlight@backlight:acpi_video0", "systemd-backlight@backlight:acpi_video0"),
+            (r"systemd-fsck@dev-disk-by\x2dlabel-BOOT", r"systemd-fsck@dev-disk-by\x2dlabel-BOOT"),
+        ],
+    )
+    def test_a_unit_name_is_accepted(self, raw, name):
+        assert unit_name(raw) == name
+
+    @pytest.mark.parametrize(
+        "raw",
+        [
+            "",
+            "-x",
+            "foo@",
+            "foo@.service",
+            "a b",
+            "a;b",
+            "$(id)",
+            "a/b",
+            r"bad\x2",
+            "ssh\n",
+            "x" * 256,
+        ],
+    )
+    def test_anything_else_is_refused(self, raw):
+        with pytest.raises(ValueError):
+            unit_name(raw)
+
+
+class TestServiceActionCommand:
+    def test_the_command_is_bounded_and_never_asks(self):
+        cmd = service_action_command("getty@tty1", "restart")
+
+        assert cmd.startswith(f"timeout {ACTION_TIMEOUT} systemctl --no-ask-password restart -- ")
+        assert "getty@tty1.service" in cmd
+
+    def test_an_escaped_name_is_quoted_for_the_shell(self):
+        cmd = service_action_command(r"systemd-fsck@dev-disk-by\x2dlabel-BOOT", "stop")
+
+        assert r"'systemd-fsck@dev-disk-by\x2dlabel-BOOT.service'" in cmd
+
+    def test_its_exit_status_is_printed_after_it(self):
+        assert service_action_command("ssh", "start").endswith("; echo __SVC_RC=$?")
+
+    def test_the_actions(self):
+        assert SERVICE_ACTIONS == ("start", "stop", "restart", "enable", "disable")
+
+    def test_an_unknown_action_is_refused(self):
+        with pytest.raises(ValueError):
+            service_action_command("ssh", "mask")
+
+    def test_an_invalid_name_is_refused(self):
+        with pytest.raises(ValueError):
+            service_action_command("ssh; reboot", "stop")
+
+
+class TestParseActionResult:
+    def test_exit_status_zero_is_success(self):
+        assert parse_action_result("__SVC_RC=0\n") == {"success": True, "output": ""}
+
+    def test_what_systemctl_printed_comes_back_without_the_marker(self):
+        raw = (
+            "Created symlink /etc/systemd/system/multi-user.target.wants/cron.service.\n__SVC_RC=0"
+        )
+
+        result = parse_action_result(raw)
+
+        assert result["success"] is True
+        assert result["output"].startswith("Created symlink")
+        assert "__SVC_RC" not in result["output"]
+
+    def test_terminal_colours_are_dropped(self):
+        """systemctl colours its errors when a transport gives it a terminal."""
+        raw = (
+            "\x1b[0;1;31mFailed to restart x.service: Unit x.service not found.\x1b[0m\n"
+            "__SVC_RC=5\n"
+        )
+
+        assert parse_action_result(raw)["output"] == (
+            "Failed to restart x.service: Unit x.service not found."
+        )
+
+    def test_a_failure_keeps_its_message(self):
+        raw = "Failed to start foo.service: Unit foo.service not found.\n__SVC_RC=5\n"
+
+        assert parse_action_result(raw) == {
+            "success": False,
+            "output": "Failed to start foo.service: Unit foo.service not found.",
+        }
+
+    def test_a_job_still_running_at_the_timeout_is_not_called_done(self):
+        result = parse_action_result("__SVC_RC=124\n")
+
+        assert result["success"] is False
+        assert str(ACTION_TIMEOUT) in result["output"]
+
+    def test_no_exit_status_is_no_success(self):
+        assert parse_action_result("Connection reset\n")["success"] is False
+
+    def test_the_echoed_command_is_not_taken_for_the_status(self):
+        raw = "timeout 45 systemctl restart -- cron.service 2>&1; echo __SVC_RC=$?\n__SVC_RC=1\n"
+
+        assert parse_action_result(raw)["success"] is False
+
+
+class _Driver(SystemdServicesMixin):
+    def __init__(self, reply: str) -> None:
+        self.reply = reply
+        self.calls: list = []
+
+    def _run_service_command(self, command: str, *, privileged: bool, timeout: int) -> str:
+        self.calls.append((command, privileged, timeout))
+        return self.reply
+
+
+class TestSystemdServicesMixin:
+    def test_listing_runs_the_command_unprivileged(self):
+        driver = _Driver(_wire())
+
+        assert "ssh" in _by_name(driver.get_services())
+        assert driver.calls == [(SYSTEMD_SERVICES_COMMAND, False, ACTION_TIMEOUT + 15)]
+
+    def test_an_action_runs_privileged_and_reports_its_outcome(self):
+        driver = _Driver("__SVC_RC=0\n")
+
+        assert driver.manage_service("cron", "restart") == {"success": True, "output": ""}
+        command, privileged, timeout = driver.calls[0]
+        assert command == service_action_command("cron", "restart")
+        assert privileged is True
+        assert timeout > ACTION_TIMEOUT
+
+    def test_an_invalid_request_is_refused_before_anything_is_sent(self):
+        driver = _Driver("__SVC_RC=0\n")
+
+        with pytest.raises(ValueError):
+            driver.manage_service("cron;reboot", "stop")
+        assert driver.calls == []
+
+    def test_not_every_os_driver_has_it(self):
+        assert not hasattr(OSDriver, "manage_service")
+        assert callable(getattr(SystemdServicesMixin, "manage_service"))