diff --git a/README.md b/README.md index 12cbb7d..09949af 100644 --- a/README.md +++ b/README.md @@ -79,6 +79,15 @@ everywhere, so the command and its parse are concrete here and a driver supplies `_run_kernel_facts_command`. `OSDriver` does not carry it — a Windows host is an OS driver too, and `hasattr(driver, "get_kernel_facts")` has to stay truthful. +`SystemdServicesMixin` (`get_services`, `manage_service`) is mixed in the same way, by +the drivers whose host runs systemd. Listing the services, checking a unit name and +reading an action's exit status are the same on every such host, so they are concrete +here, and a driver supplies only `_run_service_command(command, *, privileged, timeout)` +— how a command reaches its host and how it gains root there. The listing is one round +trip (`list-unit-files` plus one `systemctl show` over every loaded unit) instead of an +`is-enabled` and a `show` per unit. A host without systemd raises `SystemdUnavailable`, +a `NotImplementedError`, so a driver can fall back to another init system. + A function class may use the **template form** — public method concrete, the device-specific part a `_hook` declared under `if TYPE_CHECKING` — *when the base genuinely does work* on the result: normalising, sorting, validating, or orchestrating diff --git a/napalm_device_types/__init__.py b/napalm_device_types/__init__.py index 37b1686..ecabd65 100644 --- a/napalm_device_types/__init__.py +++ b/napalm_device_types/__init__.py @@ -46,6 +46,7 @@ instead of being restated on every role that happens to need it: * :class:`~napalm_device_types.packages.PackageManagementMixin` * :class:`~napalm_device_types.ping_sweep.PingSweepMixin` * :class:`~napalm_device_types.services.ServiceControlMixin` +* :class:`~napalm_device_types.systemd.SystemdServicesMixin` * :class:`~napalm_device_types.updates.UpdateMixin` Introspection -- :func:`~napalm_device_types.roles.roles_of`, @@ -74,6 +75,12 @@ from napalm_device_types.phone import PhoneDriver from napalm_device_types.ping_sweep import PingSweepMixin, driver_supports_ping from napalm_device_types.roles import primary_role_of, role_keys_of, roles_of from napalm_device_types.services import ServiceControlMixin +from napalm_device_types.systemd import ( + SYSTEMD_SERVICES_COMMAND, + SystemdServicesMixin, + SystemdUnavailable, + parse_systemd_services, +) from napalm_device_types.updates import UpdateMixin from napalm_device_types.residential_gateway import ResidentialGatewayDriver from napalm_device_types.storage import StorageDriver @@ -106,6 +113,10 @@ __all__ = [ "ServiceControlMixin", "StorageDriver", "SwitchDriver", + "SYSTEMD_SERVICES_COMMAND", + "SystemdServicesMixin", + "SystemdUnavailable", + "parse_systemd_services", "UpdateMixin", "add_lag_interfaces", "driver_supports_ping", diff --git a/napalm_device_types/systemd.py b/napalm_device_types/systemd.py new file mode 100644 index 0000000..3733c32 --- /dev/null +++ b/napalm_device_types/systemd.py @@ -0,0 +1,331 @@ +# -*- coding: utf-8 -*- +"""systemd services: listing them in one round trip, and starting and stopping them. + +What systemd reports about its services, and how one is started or stopped, is +the same on every host that runs it. 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 only carries a command across: SSH, an API's exec endpoint, whatever it +has. + +**Listing.** One command prints the installed unit files and, for every loaded +service unit, what ``systemctl show`` knows about it -- state, boot state and +main PID together, instead of asking ``systemctl is-enabled`` and ``systemctl +show`` once per unit (two hundred round trips on an ordinary Linux host). The +report is framed, and a report whose end is missing raises: a list cut short +must never read as services that went away. + +**What counts as enabled.** A unit file state of ``enabled`` or +``enabled-runtime``. ``static`` does not: such a unit starts only when +something else pulls it in, and calling it enabled made every one of them look +like a service of the host. The state is read from ``UnitFileState``, never +from a column of ``list-unit-files``, whose second column has been followed by +a preset column since systemd 245. + +**Starting and stopping.** ``systemctl`` runs bounded by ``timeout`` and never +asks for a password, and its exit status is printed after it. The marker also +keeps the output from ever being empty, which a transport that retries on an +empty answer would otherwise take as a reason to run the action twice. +""" + +from __future__ import annotations + +import re +from shlex import quote +from typing import Any, Dict, List, Set, Tuple, TYPE_CHECKING + +from napalm_device_types.models import ServiceDict +from napalm_device_types.services import ServiceControlMixin + +_BEGIN = "SVC_BEGIN" +_END = "SVC_END" +_NO_SYSTEMD = "no-systemd" +_SUFFIX = ".service" + +#: What ``systemctl show`` prints per unit. It prints them in its own order. +_PROPERTIES = "Id,Names,LoadState,ActiveState,SubState,UnitFileState,MainPID" + +#: Picks the units whose file state is ``generated`` out of ``systemctl show``'s +#: output, whatever order it prints the properties in. +_GENERATED_AWK = ( + 'awk -F= \'NF<2{id="";g=0;next} $1=="Id"{id=$2} ' + '$1=="UnitFileState"{g=($2=="generated")} id!=""&&g{print id;id="";g=0}\'' +) + +#: One line, POSIX ``sh``, read-only. The frame markers are printed in two +#: halves so that a transport which echoes the command does not show them early. +#: ``xargs -0`` passes escaped names such as ``foo\x2dbar.service`` unchanged. +#: A generated unit -- the wrapper systemd makes for a SysV script -- has no unit +#: file whose state says whether it starts at boot; ``systemctl is-enabled`` +#: asks the script's rc links instead, for those few units only. +SYSTEMD_SERVICES_COMMAND = ( + "printf '%s%s\\n' SVC_ BEGIN; " + "[ -d /run/systemd/system ] || echo '[no-systemd]'; " + "echo '[files]'; systemctl list-unit-files --type=service --no-legend --no-pager 2>/dev/null; " + "echo '[units]'; s=$(systemctl list-units --type=service --all --no-legend --no-pager --plain " + "2>/dev/null | awk '{print $1}' | tr '\\n' '\\0' | xargs -0 -r systemctl show --no-pager " + f"-p {_PROPERTIES} -- 2>/dev/null); printf '%s\\n' \"$s\"; " + f"echo '[generated]'; printf '%s\\n' \"$s\" | {_GENERATED_AWK} | while read -r u; do " + 'printf \'%s %s\\n\' "$u" "$(systemctl is-enabled -- "$u" 2>/dev/null)"; done; ' + "printf '%s%s\\n' SVC_ END" +) + +#: The lifecycle actions :meth:`SystemdServicesMixin.manage_service` accepts. +SERVICE_ACTIONS = ("start", "stop", "restart", "enable", "disable") + +#: Seconds an action may run on the host before ``timeout`` stops waiting for +#: it. systemd itself carries on with the job. +ACTION_TIMEOUT = 45 + +#: What a transport should allow for one command: the action's own bound plus +#: the round trip around it. +_TRANSPORT_TIMEOUT = ACTION_TIMEOUT + 15 + +_TIMED_OUT = 124 # timeout(1)'s exit status when the time ran out +_RC_MARKER = "__SVC_RC=" +_RC_RE = re.compile(rf"^{_RC_MARKER}(\d+)\s*$", re.MULTILINE) + +#: The characters systemd allows in a unit name, with ``\xHH`` for any other byte. +_UNIT_RE = re.compile(r"(?:[A-Za-z0-9_.:@-]|\\x[0-9A-Fa-f]{2})+") +_MAX_UNIT_LENGTH = 255 + +#: Terminal colour codes, which systemctl adds when a transport gives it a terminal. +_ANSI_RE = re.compile(r"\x1b\[[0-9;?]*[A-Za-z]") + +_ENABLED = frozenset({"enabled", "enabled-runtime"}) +#: Unit file states of a service that is installed but need not be loaded. +_INSTALLED = frozenset({"enabled", "enabled-runtime", "disabled", "indirect"}) + + +class SystemdUnavailable(NotImplementedError): + """The host does not run systemd; a driver may fall back to another init system.""" + + +def unit_name(name: str) -> str: + """*name* as a service unit's name without ``.service``, or ``ValueError``. + + Accepts template instances (``wg-quick@wg0``), dots (``snapd.apparmor``), + colons and systemd's ``\\xHH`` escapes. Refuses a bare template + (``getty@``), a leading ``-`` that a command would read as an option, and + anything a shell would read. + """ + base = name[: -len(_SUFFIX)] if name.endswith(_SUFFIX) else name + if ( + not _UNIT_RE.fullmatch(base) + or base.startswith("-") + or base.endswith("@") + or len(base) + len(_SUFFIX) > _MAX_UNIT_LENGTH + ): + raise ValueError(f"Invalid service name: {name!r}") + return base + + +def service_action_command(name: str, action: str) -> str: + """The shell command that applies *action* to the service *name*. + + :raises ValueError: for an unknown action or an invalid name. + """ + if action not in SERVICE_ACTIONS: + raise ValueError(f"Invalid action {action!r}; use one of {', '.join(SERVICE_ACTIONS)}") + unit = quote(unit_name(name) + _SUFFIX) + return ( + f"timeout {ACTION_TIMEOUT} systemctl --no-ask-password {action} -- {unit} 2>&1; " + f"echo {_RC_MARKER}$?" + ) + + +def parse_action_result(output: str) -> Dict[str, Any]: + """``{"success", "output"}`` from what :func:`service_action_command` printed. + + Only the exit status decides. A job still running when ``timeout`` gave up + is not reported as done, and output without a status is no success. + """ + output = _ANSI_RE.sub("", output) + statuses = _RC_RE.findall(output) + text = _RC_RE.sub("", output).strip() + if not statuses: + return {"success": False, "output": text or "No exit status came back from the host."} + status = int(statuses[-1]) + if status == 0: + return {"success": True, "output": text} + if status == _TIMED_OUT: + note = f"Still running after {ACTION_TIMEOUT} s; systemd carries on with the job." + return {"success": False, "output": f"{text}\n{note}".strip()} + return {"success": False, "output": text or f"systemctl exited with status {status}."} + + +def _frame(output: str) -> List[str]: + lines = [line.strip() for line in _ANSI_RE.sub("", output).splitlines()] + try: + start = lines.index(_BEGIN) + end = lines.index(_END, start) + except ValueError: + raise ValueError("no intact systemd service report in the output") from None + return lines[start + 1 : end] + + +def _sections(lines: List[str]) -> Dict[str, List[str]]: + sections: Dict[str, List[str]] = {} + current: List[str] = [] + for line in lines: + if line.startswith("[") and line.endswith("]"): + current = sections.setdefault(line[1:-1], []) + else: + current.append(line) + return sections + + +def _unit_blocks(lines: List[str]) -> List[Dict[str, str]]: + """``systemctl show``'s output, one dict per unit. + + Units are separated by a blank line -- except where ``xargs`` split the + list over two runs and the blocks meet, so a key seen twice starts the next + unit as well. + """ + blocks: List[Dict[str, str]] = [] + current: Dict[str, str] = {} + for line in lines: + key, sep, value = line.partition("=") + if not sep or key in current: + if current: + blocks.append(current) + current = {} + if sep: + current[key] = value + if current: + blocks.append(current) + return blocks + + +def _base(unit: str) -> str: + return unit[: -len(_SUFFIX)] + + +def _main_pid(block: Dict[str, str]) -> int: + try: + return int(block.get("MainPID") or 0) + except ValueError: + return 0 + + +def _loaded(blocks: List[Dict[str, str]]) -> Tuple[Dict[str, ServiceDict], Set[str]]: + """The loaded services, and every name they go by (aliases included).""" + services: Dict[str, ServiceDict] = {} + names: Set[str] = set() + for block in blocks: + unit = block.get("Id", "") + if not unit.endswith(_SUFFIX) or block.get("LoadState") == "not-found": + continue + names.update(block.get("Names", unit).split()) + running = block.get("ActiveState") == "active" and block.get("SubState") == "running" + services[_base(unit)] = { + "name": _base(unit), + "running": running, + "enabled": block.get("UnitFileState") in _ENABLED, + "pid": _main_pid(block) if running else 0, + } + return services, names + + +def _installed(lines: List[str], known: Set[str]) -> Dict[str, ServiceDict]: + """Installed services that are not loaded: neither running nor starting now. + + Templates, static units and aliases are left out -- the last also when an + older systemd lists an alias as ``enabled``, which is why every name a + loaded unit goes by is skipped. + """ + services: Dict[str, ServiceDict] = {} + for line in lines: + parts = line.split() + if len(parts) < 2: + continue + unit, state = parts[0], parts[1] + if ( + not unit.endswith(_SUFFIX) + or unit.endswith("@" + _SUFFIX) + or unit in known + or state not in _INSTALLED + ): + continue + services[_base(unit)] = { + "name": _base(unit), + "running": False, + "enabled": state in _ENABLED, + "pid": 0, + } + return services + + +def _apply_generated(services: Dict[str, ServiceDict], lines: List[str]) -> None: + """Take a generated unit's boot state from ``is-enabled``'s answer.""" + for line in lines: + parts = line.split() + if len(parts) == 2 and parts[0].endswith(_SUFFIX) and _base(parts[0]) in services: + services[_base(parts[0])]["enabled"] = parts[1] in _ENABLED + + +def parse_systemd_services(output: str) -> List[ServiceDict]: + """Parse what :data:`SYSTEMD_SERVICES_COMMAND` printed, sorted by name. + + Lists every loaded service unit but those that are not found, and every + installed one that is not loaded. + + :raises SystemdUnavailable: when the host does not run systemd. + :raises ValueError: when the output carries no intact report. + """ + sections = _sections(_frame(output)) + if _NO_SYSTEMD in sections: + raise SystemdUnavailable("the host does not run systemd") + loaded, known = _loaded(_unit_blocks(sections.get("units", []))) + _apply_generated(loaded, sections.get("generated", [])) + merged = {**_installed(sections.get("files", []), known), **loaded} + return [merged[name] for name in sorted(merged)] + + +class SystemdServicesMixin(ServiceControlMixin): + """Implements :class:`ServiceControlMixin` for a driver whose host runs systemd. + + The template form (README, "Function classes"): the command, the parse, + the check of the name and the reading of the exit status are the same + everywhere, so they are concrete here, and a driver supplies only + :meth:`_run_service_command` -- how a command reaches its host, and how it + gains root there when it needs to. + """ + + if TYPE_CHECKING: # pragma: no cover - declared for type checkers only + + def _run_service_command(self, command: str, *, privileged: bool, timeout: int) -> str: + """Run *command* with ``sh`` on the host and return what it printed. + + *privileged* commands change the system and need root; *timeout* + is how long the transport should wait for the output, in seconds. + """ + ... + + def get_services(self) -> List[ServiceDict]: + """ + Returns the services systemd knows, in one round trip. + + * name (string) - the unit name without ``.service`` + * running (bool) - active and running + * enabled (bool) - the unit file is enabled + * pid (int) - the main process; 0 when not running + + :raises SystemdUnavailable: if the host does not run systemd. + :raises ValueError: if the host's output carried no intact report. + """ + output = self._run_service_command( + SYSTEMD_SERVICES_COMMAND, privileged=False, timeout=_TRANSPORT_TIMEOUT + ) + return parse_systemd_services(output) + + def manage_service(self, name: str, action: str) -> Dict[str, Any]: + """ + Applies *action* (start, stop, restart, enable, disable) to the service *name*. + + :returns: ``{"success": bool, "output": str}`` + :raises ValueError: for an unknown action or an invalid name, before + anything is sent. + """ + command = service_action_command(name, action) + output = self._run_service_command(command, privileged=True, timeout=_TRANSPORT_TIMEOUT) + return parse_action_result(output) diff --git a/pyproject.toml b/pyproject.toml index 9ad75c2..21062d9 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "napalm-device-types" -version = "2.1.0" +version = "2.2.0" description = "Abstract device-type base classes for NAPALM drivers" readme = "README.md" requires-python = ">=3.10" diff --git a/tests/test_systemd_services.py b/tests/test_systemd_services.py new file mode 100644 index 0000000..9423baf --- /dev/null +++ b/tests/test_systemd_services.py @@ -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"))