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

README.md

@@ -48,7 +48,8 @@ with Driver(
     optional_args={
         # "port": 22,
         # "pkg_manager": "apt",   # force package manager; auto-detected by default
-        # "secret": "sudo-pass",  # password for sudo / enable (defaults to login password)
+        # "sudo_password": "sudo-pass",  # for commands that need root; without it,
+        #                                # `sudo -n` (passwordless sudo) is tried
         # "debugging": True,      # enable verbose logging
     },
 ) as dev:
@@ -69,6 +70,10 @@ with Driver(
 
     # Upgrade everything with pending updates
     result = dev.apply_updates([])
+
+    # Restart a service (start, stop, restart, enable, disable)
+    result = dev.manage_service("cron", "restart")
+    print(result)  # {"success": True, "output": ""}
 ```
 
 ## Supported NAPALM methods
@@ -99,7 +104,8 @@ with Driver(
 | `get_packages()` | ✅ | apt, dnf, yum, apk, pacman |
 | `get_pending_updates()` | ✅ | apt, dnf, yum, apk, pacman |
 | `apply_updates(packages)` | ✅ | apt, dnf, yum, apk, pacman |
-| `get_services()` | ✅ | systemd (fallback: SysV `service`) |
+| `get_services()` | ✅ | systemd, one round trip (fallback: SysV `service`) |
+| `manage_service(name, action)` | ✅ | systemd: start, stop, restart, enable, disable |
 | `get_users()` | ✅ | `/etc/passwd` + `/etc/group` |
 | `get_processes()` | ✅ | `ps axo` |
 | `get_cron_jobs()` | ✅ | user crontabs + `/etc/cron.d/` |
@@ -127,13 +133,25 @@ The SSH user needs read access to:
 | `/etc/passwd`, `/etc/group` | world-readable (default) |
 | `/proc/uptime`, `/sys/class/dmi/…` | world-readable (default) |
 | User crontabs (`/var/spool/cron/…`) | `root` or `sudo` required |
-| `systemctl is-enabled <unit>` | unprivileged on most distros |
+| `systemctl list-unit-files`, `systemctl show` | unprivileged |
+| `systemctl start/stop/restart/enable/disable` | `root`, or `sudo` (with `sudo_password`, or passwordless) |
 | `apt list --upgradable` | may require `apt-get update` (root) |
 | `dnf check-update` / `yum check-update` | unprivileged, but slower without cache |
 
 For full functionality it is recommended to run as `root` or grant passwordless `sudo` for
 the above commands.
 
+`get_services()` and `manage_service()` come from napalm-device-types'
+`SystemdServicesMixin`; this driver supplies only the transport. An action runs as
+`timeout 45 systemctl --no-ask-password <action> -- <unit>.service`, so a unit that hangs
+on its way up or down cannot hold the session, and only the exit status decides whether it
+succeeded. Without a sudo password it uses `sudo -n`, which fails at once instead of
+waiting for a password prompt.
+
+On OpenMediaVault (napalm-openmediavault inherits this driver), enabling or disabling a
+unit that OMV manages itself — Samba, NFS, SSH — may be reverted the next time OMV applies
+its configuration.
+
 ## Tested distributions
 
 | Distribution | Version | Package manager | Tested |

napalm_linux/linux.py

@@ -31,7 +31,13 @@ from netmiko.exceptions import (
 )
 from napalm.base.exceptions import ConnectionException, ConnectionClosedException
 from napalm.base.netmiko_helpers import netmiko_args
-from napalm_device_types import FingerprintRule, OSDriver
+from napalm_device_types import (
+    FingerprintRule,
+    KernelFactsMixin,
+    OSDriver,
+    SystemdServicesMixin,
+    SystemdUnavailable,
+)
 from napalm_device_types.models import (
     ApplyUpdatesResultDict,
     CronJobDict,
@@ -50,6 +56,19 @@ logger = logging.getLogger("napalm_linux")
 # Package managers in detection order
 _PKG_MANAGERS = ["apt", "dnf", "yum", "apk", "pacman"]
 
+#: Printed after a command by ``_sudo_status`` so its exit status survives the
+#: trip through an interactive shell. Matched only on a line of its own with a
+#: number after it — an echoed command line carries the literal ``$?`` instead.
+_RC_MARKER = "__NETORK_RC="
+_RC_MARKER_RE = re.compile(rf"^{_RC_MARKER}(\d+)\s*$", re.MULTILINE)
+
+#: What to do when sudo wants a password netOrk does not have.
+_SUDO_PASSWORD_HINT = (
+    "sudo requires a password on this device but none is configured in netOrk. "
+    "Please add the sudo password to a Credential Profile assigned to this device, "
+    "or configure passwordless sudo (NOPASSWD) for this user."
+)
+
 # DMI field values that carry no useful information (OEM defaults, blanks)
 _BAD_DMI: frozenset[str] = frozenset({
     "", "none", "n/a", "not specified", "not applicable",
@@ -132,7 +151,7 @@ def _short_image_id(raw: str) -> str:
     return raw.strip().removeprefix("sha256:")[:12]
 
 
-class LinuxDriver(OSDriver):
+class LinuxDriver(KernelFactsMixin, SystemdServicesMixin, OSDriver):
     """NAPALM driver for generic Linux systems.
 
     Connects via SSH (netmiko ``linux`` device type) and auto-detects the
@@ -267,6 +286,47 @@ class LinuxDriver(OSDriver):
             return self._send(wrapped, read_timeout=read_timeout)
         return self._send(f'sudo {command}', read_timeout=read_timeout)
 
+    def _sudo_status(self, command: str, read_timeout: float = 100) -> tuple[str, int | None]:
+        """Run *command* via sudo and return ``(output, exit_status)``.
+
+        ``_sudo`` callers append ``|| true`` so a failing command yields output
+        instead of an error, which throws the exit status away. This variant
+        echoes ``$?`` straight after the sudo pipeline instead — sudo passes
+        the command's status through, and a failed password is non-zero too.
+
+        The status is ``None`` when the marker never arrived (output cut short),
+        so a caller can tell "unknown" from "succeeded".
+        """
+        raw = self._sudo(f"{command}; echo {_RC_MARKER}$?", read_timeout=read_timeout)
+        matches = list(_RC_MARKER_RE.finditer(raw))
+        if not matches:
+            return raw, None
+        last = matches[-1]
+        output = (raw[: last.start()] + raw[last.end():]).strip()
+        return output, int(last.group(1))
+
+    def _is_root(self) -> bool:
+        """Whether the SSH user is root, asked once per session.
+
+        A root login on a box without sudo (an LXC container, a minimal Debian)
+        must not have its commands prefixed with a sudo that is not there.
+        """
+        if getattr(self, "_root", None) is None:
+            self._root = self._send("id -u") == "0"
+        return bool(self._root)
+
+    def _run_service_command(self, command: str, *, privileged: bool, timeout: int) -> str:
+        """The transport for :class:`SystemdServicesMixin`.
+
+        Without a sudo password, ``sudo -n`` fails at once where a prompt would
+        otherwise hang the session until the read timeout.
+        """
+        if not privileged or self._is_root():
+            return self._send(command, read_timeout=timeout)
+        if self._sudo_password:
+            return self._sudo(command, read_timeout=timeout)
+        return self._send(f"sudo -n {command}", read_timeout=timeout)
+
     def _detect_pkg_manager(self) -> str | None:
         """Return the first package manager binary found on PATH."""
         for pm in _PKG_MANAGERS:
@@ -815,6 +875,14 @@ class LinuxDriver(OSDriver):
             }
         }
 
+    # ------------------------------------------------------------------
+    # KernelFactsMixin – the transport for get_kernel_facts
+    # ------------------------------------------------------------------
+
+    def _run_kernel_facts_command(self, command: str) -> str:
+        """The transport for ``KernelFactsMixin.get_kernel_facts``: read-only, no sudo."""
+        return self._send(command, read_timeout=60)
+
     # ------------------------------------------------------------------
     # OSDriver – package management
     # ------------------------------------------------------------------
@@ -1046,6 +1114,7 @@ class LinuxDriver(OSDriver):
         return {"success": success, "output": raw.strip()}
 
     #: Words in a package manager's output that mean it did not do the job.
+    #: Only consulted when the exit status is unknown; see ``_uninstall_failed``.
     _UNINSTALL_FAILED = ("error:", "failed", "not found", "is not installed", "no packages")
 
     def uninstall_package(self, name: str, purge: bool = False) -> dict[str, Any]:
@@ -1077,42 +1146,49 @@ class LinuxDriver(OSDriver):
         pm = self._pkg_manager
         if pm == "apt":
             action = "purge" if purge else "remove"
-            raw = self._sudo(
+            cmd = f"DEBIAN_FRONTEND=noninteractive apt-get {action} -y {safe} 2>&1"
-                f"DEBIAN_FRONTEND=noninteractive apt-get {action} -y {safe} 2>&1 || true"
-            )
         elif pm in ("dnf", "yum"):
-            raw = self._sudo(f"{pm} remove -y {safe} 2>&1 || true")
+            cmd = f"{pm} remove -y {safe} 2>&1"
         elif pm == "apk":
             # apk and pacman have no separate purge; asking for one is not an
             # error, it simply has nothing extra to do.
-            raw = self._sudo(f"apk del {safe} 2>&1 || true")
+            cmd = f"apk del {safe} 2>&1"
         elif pm == "pacman":
-            raw = self._sudo(f"pacman -R --noconfirm {safe} 2>&1 || true")
+            cmd = f"pacman -R --noconfirm {safe} 2>&1"
         else:
             return {"success": False, "output": f"Unsupported package manager: {pm}"}
 
-        if self._uninstall_failed(raw) and pm == "apt":
+        raw, rc = self._sudo_status(cmd)
-            forced = self._sudo(f"dpkg --purge --force-all {safe} 2>&1 || true")
+        failed = self._uninstall_failed(raw, rc)
-            if not self._uninstall_failed(forced):
+
-                return {
+        if failed and pm == "apt":
-                    "success": True,
+            forced, forced_rc = self._sudo_status(f"dpkg --purge --force-all {safe} 2>&1")
-                    "output": f"{raw.strip()}\n--- dpkg --purge --force-all ---\n{forced.strip()}",
-                }
             raw = f"{raw.strip()}\n--- dpkg --purge --force-all ---\n{forced.strip()}"
+            failed = self._uninstall_failed(forced, forced_rc)
 
-        return {"success": not self._uninstall_failed(raw), "output": raw.strip()}
+        return {"success": not failed, "output": raw.strip()}
 
-    def _uninstall_failed(self, output: str) -> bool:
+    def _uninstall_failed(self, output: str, rc: int | None = None) -> bool:
-        """Whether the package manager said it did not do the job.
+        """Whether the package manager did not do the job.
 
-        apt prefixes its own errors with ``E: `` at the start of a line, and
+        The exit status decides whenever there is one (netork#267): it is the
-        the commonest of them — ``E: Sub-process /usr/bin/dpkg returned an
+        answer the package manager actually gives, where the output is prose
-        error code (1)`` — contains neither "error:" nor "failed". The keyword
+        that every tool phrases differently. A prerm printing "Failed to stop
-        list alone therefore read a failed removal as a success, which is the
+        …" while the removal completes is a success; a non-zero exit with
-        worst direction for this particular answer to be wrong in.
+        nothing alarming in the output is not.
+
+        Only when the status is unknown (``rc is None``) is the output read,
+        as the best answer left. apt prefixes its own errors with ``E: `` at
+        the start of a line, and the commonest of them — ``E: Sub-process
+        /usr/bin/dpkg returned an error code (1)`` — contains neither "error:"
+        nor "failed". The keyword list alone therefore read a failed removal
+        as a success, which is the worst direction for this particular answer
+        to be wrong in.
 
         Matched at line start rather than anywhere: "note: " ends in "e: ".
         """
+        if rc is not None:
+            return rc != 0
         low = output.lower()
         if any(line.lstrip().startswith("e: ") for line in low.splitlines()):
             return True
@@ -1338,50 +1414,25 @@ class LinuxDriver(OSDriver):
             return {"success": False, "output": "", "error": str(exc)}
 
     # ------------------------------------------------------------------
-    # OSDriver – services (systemd)
+    # OSDriver – services (systemd, through SystemdServicesMixin)
     # ------------------------------------------------------------------
 
     def get_services(self) -> list[ServiceDict]:
-        """Return systemd service units (falls back to service --status-all on SysV)."""
+        """systemd's services in one round trip; ``service --status-all`` without systemd."""
-        out = self._send(
+        try:
-            "systemctl list-units --type=service --all --no-legend --no-pager "
+            return super().get_services()
-            "--plain 2>/dev/null"
+        except SystemdUnavailable:
-        )
-        if not out:
             return self._get_services_sysv()
 
-        services: list[ServiceDict] = []
+    def manage_service(self, name: str, action: str) -> dict[str, Any]:
-        for line in out.splitlines():
+        """Start, stop, restart, enable or disable a systemd service.
-            # ssh.service loaded active running OpenBSD Secure Shell server
-            parts = line.split(None, 4)
-            if len(parts) < 4:
-                continue
-            unit, load, active, sub = parts[0], parts[1], parts[2], parts[3]
-            name = unit.removesuffix(".service")
-            running = active == "active" and sub == "running"
-            enabled_out = self._send(
-                f"systemctl is-enabled {unit} 2>/dev/null"
-            )
-            enabled = enabled_out.strip() == "enabled"
 
-            # Retrieve main PID for running services
+        :raises ValueError: for an unknown action or an invalid name.
-            pid = 0
+        """
-            if running:
+        result = super().manage_service(name, action)
-                pid_out = self._send(
+        if not result["success"] and "password is required" in result["output"]:
-                    f"systemctl show -p MainPID --value {unit} 2>/dev/null"
+            result["output"] = f"{result['output']}\n{_SUDO_PASSWORD_HINT}"
-                )
+        return result
-                try:
-                    pid = int(pid_out.strip())
-                except ValueError:
-                    pid = 0
-
-            services.append({
-                "name": name,
-                "running": running,
-                "enabled": enabled,
-                "pid": pid,
-            })
-        return services
 
     def _get_services_sysv(self) -> list[ServiceDict]:
         out = self._send("service --status-all 2>/dev/null")
@@ -2069,11 +2120,7 @@ class LinuxDriver(OSDriver):
             if not self._sudo_password:
                 return {
                     "success": False,
-                    "output": (
+                    "output": _SUDO_PASSWORD_HINT,
-                        "sudo requires a password on this device but none is configured in "
-                        "netOrk. Please add the sudo password to a Credential Profile assigned "
-                        "to this device, or configure passwordless sudo (NOPASSWD) for this user."
-                    ),
                 }
 
         lines: list[str] = []
@@ -2099,11 +2146,7 @@ class LinuxDriver(OSDriver):
             if not self._sudo_password:
                 return {
                     "success": False,
-                    "output": (
+                    "output": _SUDO_PASSWORD_HINT,
-                        "sudo requires a password on this device but none is configured in netOrk. "
-                        "Please add the sudo password to a Credential Profile assigned to this device, "
-                        "or configure passwordless sudo (NOPASSWD) for this user."
-                    ),
                 }
 
         # 1. Install snmpd if missing

pyproject.toml

@@ -37,7 +37,7 @@ classifiers = [
 ]
 dependencies = [
     "napalm>=4.0",
-    "napalm-device-types>=0.3.0",
+    "napalm-device-types>=2.2.0",
     "netmiko>=4.0.0",
     "paramiko>=5.0.0",  # CVE-2026-44405
 ]

tests/test_linux.py

@@ -984,3 +984,301 @@ class TestUninstallPackage:
 
         assert result["success"] is True
         assert "apk del" in driver._device.send_command.call_args[0][0]
+
+
+# ---------------------------------------------------------------------------
+# uninstall_package – success from the exit status, not from prose (netork#267)
+# ---------------------------------------------------------------------------
+
+
+def _with_rc(output: str, rc: int) -> str:
+    """What the shell prints for a command run through ``_sudo_status``."""
+    return f"{output}\n__NETORK_RC={rc}"
+
+
+class TestSudoStatus:
+    """``_sudo_status`` keeps the exit status that ``|| true`` throws away."""
+
+    def test_returns_output_and_exit_status(self, driver):
+        _mock_send(driver, _with_rc("Removing wazuh-agent ...", 0))
+
+        assert driver._sudo_status("apt-get remove -y wazuh-agent") == (
+            "Removing wazuh-agent ...",
+            0,
+        )
+
+    def test_a_non_zero_exit_status_is_reported(self, driver):
+        _mock_send(driver, _with_rc("E: Unable to locate package nope", 100))
+
+        assert driver._sudo_status("apt-get remove -y nope")[1] == 100
+
+    def test_the_status_is_read_right_after_sudo_returns(self, driver):
+        """``$?`` must be read straight after the sudo pipeline — with an
+        ``|| true`` in between, every command would report 0."""
+        driver._sudo_password = "pw"  # noqa: S105
+        _mock_send(driver, _with_rc("", 0))
+
+        driver._sudo_status("apt-get remove -y x 2>&1")
+
+        sent = driver._device.send_command.call_args[0][0]
+        assert sent.startswith("echo pw | sudo -S")
+        assert sent.endswith("apt-get remove -y x 2>&1; echo __NETORK_RC=$?")
+        assert "|| true" not in sent
+
+    def test_a_missing_marker_means_unknown_not_success(self, driver):
+        """Output cut short before the marker arrived says nothing about the
+        exit status; ``None`` says so instead of guessing 0."""
+        _mock_send(driver, "Removing wazuh-agent ...")
+
+        assert driver._sudo_status("apt-get remove -y wazuh-agent") == (
+            "Removing wazuh-agent ...",
+            None,
+        )
+
+    def test_the_command_echo_is_not_mistaken_for_the_marker(self, driver):
+        """A terminal may echo the command line back; its literal ``$?`` is not
+        a number, and only the marker on a line of its own counts."""
+        _mock_send(
+            driver,
+            "sudo apt-get remove -y x; echo __NETORK_RC=$?\nRemoving x ...\n__NETORK_RC=1",
+        )
+
+        output, rc = driver._sudo_status("apt-get remove -y x")
+
+        assert rc == 1
+        assert "__NETORK_RC=1" not in output
+
+
+class TestUninstallExitStatus:
+    """Whether a removal worked is what the package manager's exit status says.
+
+    Reading it out of human-readable output was guesswork in both directions:
+    apt's commonest failure (``E: Sub-process /usr/bin/dpkg returned an error
+    code (1)``) read as success until #240, and a successful removal whose
+    prerm merely *mentions* a failure read as a failure.
+    """
+
+    def test_a_non_zero_exit_is_a_failure_whatever_the_output_says(self, driver):
+        """Nothing in this output matches a failure keyword; only the exit
+        status knows."""
+        driver._pkg_manager = "dnf"
+        _mock_send(driver, _with_rc("Removing: wazuh-agent", 1))
+
+        result = driver.uninstall_package("wazuh-agent")
+
+        assert result["success"] is False
+
+    def test_a_zero_exit_is_a_success_even_if_the_output_mentions_failure(self, driver):
+        """A prerm that cannot stop an already-dead unit prints "Failed" and
+        still lets the removal complete."""
+        _mock_send(
+            driver,
+            _with_rc(
+                "Removing wazuh-agent (4.14.7-1) ...\n"
+                "Failed to stop wazuh-agent.service: Unit wazuh-agent.service not loaded.",
+                0,
+            ),
+        )
+
+        result = driver.uninstall_package("wazuh-agent")
+
+        assert result["success"] is True
+
+    def test_the_marker_does_not_reach_the_caller(self, driver):
+        _mock_send(driver, _with_rc("Removing wazuh-agent ...", 0))
+
+        result = driver.uninstall_package("wazuh-agent")
+
+        assert result["output"] == "Removing wazuh-agent ..."
+
+    def test_the_uninstall_command_keeps_its_exit_status(self, driver):
+        _mock_send(driver, _with_rc("Removing wazuh-agent ...", 0))
+
+        driver.uninstall_package("wazuh-agent")
+
+        sent = driver._device.send_command.call_args[0][0]
+        assert "|| true" not in sent
+        assert sent.endswith("; echo __NETORK_RC=$?")
+
+    def test_apt_failing_by_exit_status_falls_back_to_dpkg(self, driver):
+        driver._device.send_command.side_effect = [
+            _with_rc("E: Sub-process /usr/bin/dpkg returned an error code (1)", 100),
+            _with_rc("Removing wazuh-agent (4.14.7-1) ...", 0),
+        ]
+
+        result = driver.uninstall_package("wazuh-agent", purge=True)
+
+        assert result["success"] is True
+        second = driver._device.send_command.call_args_list[1][0][0]
+        assert "dpkg --purge --force-all" in second
+        assert "|| true" not in second
+        assert "__NETORK_RC" not in result["output"]
+
+    def test_the_dpkg_fallback_failing_is_a_failure(self, driver):
+        driver._device.send_command.side_effect = [
+            _with_rc("E: Sub-process /usr/bin/dpkg returned an error code (1)", 100),
+            _with_rc("dpkg: error processing package wazuh-agent (--purge):", 1),
+        ]
+
+        result = driver.uninstall_package("wazuh-agent", purge=True)
+
+        assert result["success"] is False
+        assert "dpkg --purge --force-all" in result["output"]
+
+    def test_a_zero_exit_does_not_trigger_the_fallback(self, driver):
+        """Even when the output contains words that used to mean failure: apt
+        exits 0 for a package that is already gone, which is the state the
+        caller asked for."""
+        _mock_send(driver, _with_rc("Package 'x' is not installed, so not removed", 0))
+
+        result = driver.uninstall_package("x", purge=True)
+
+        assert result["success"] is True
+        assert driver._device.send_command.call_count == 1
+
+    def test_without_an_exit_status_the_output_is_read_as_before(self, driver):
+        """If the marker never arrived, the keyword check is still the best
+        answer available — and it errs towards failure on apt's ``E:``."""
+        driver._pkg_manager = "dnf"
+        _mock_send(driver, "E: Sub-process /usr/bin/dpkg returned an error code (1)")
+
+        result = driver.uninstall_package("wazuh-agent")
+
+        assert result["success"] is False
+
+
+# ---------------------------------------------------------------------------
+# get_kernel_facts -- the command and its parse live in napalm-device-types
+# ---------------------------------------------------------------------------
+
+
+def _kernel_wire(report: str) -> str:
+    import base64
+    import gzip
+
+    return "KFACTS_BEGIN\n" + base64.encodebytes(gzip.compress(report.encode())).decode() + "KFACTS_END"
+
+
+def test_get_kernel_facts_carries_the_shared_command_across(driver):
+    from napalm_device_types import KernelFactsMixin
+    from napalm_device_types.kernel import KERNEL_FACTS_COMMAND
+
+    assert isinstance(driver, KernelFactsMixin)
+
+    report = "[release]\n6.1.0-25-amd64\n[loaded]\ntipc\n[available]\nkernel/net/tipc/tipc.ko.xz\n"
+    with patch.object(driver, "_send", return_value=_kernel_wire(report)) as send:
+        facts = driver.get_kernel_facts()
+
+    assert send.call_args.args[0] == KERNEL_FACTS_COMMAND
+    assert facts["release"] == "6.1.0-25-amd64"
+    assert facts["loaded"] == ["tipc"]
+    assert facts["available"] == ["tipc"]
+    assert facts["builtin"] is None
+
+
+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