feat: restart the host (reboot_host)

netOrk asks a driver for reboot_host before it offers a restart, and the
Linux driver had none, so a Linux host -- and OpenMediaVault and QTS, which
inherit this driver -- could not be restarted from netOrk at all (netOrk
#637). reboot_host runs /sbin/reboot detached and two seconds late through
_run_privileged: the launcher's exit status comes back before the host goes
down, and closing the session cannot take the restart with it. A refusal
(sudo without a password) raises with what the host said.

napalm_linux/linux.py

@@ -69,6 +69,31 @@ _RC_MARKER = "__NETORK_RC="
 _RC_MARKER_RE = re.compile(rf"^{_RC_MARKER}(\d+)\s*$", re.MULTILINE)
 
 
+#: The end of a command's output when nothing better is known: a line that looks
+#: like a shell prompt.
+_PROMPT_RE = r"[#$\>]\s*$"
+#: A command ending in ``echo __NAME=$?`` reports its exit status on a line of its
+#: own: ``_RC_MARKER`` here, ``__APT_RC=`` and ``__SVC_RC=`` in napalm-device-types.
+_STATUS_ECHO_RE = re.compile(r"echo\s+(__[A-Z_]+=)\$\?")
+
+
+def _expect_for(command: str) -> str:
+    """The pattern that ends *command*'s output.
+
+    netmiko stops reading as soon as the pattern matches what it has read so far.
+    A line the command prints can end in ``#``, ``$`` or ``>`` -- apt's
+    ``<ftpmaster@ubuntu.com>`` after a bad signature -- and was taken for the
+    prompt: half the output came back, and the rest started the next command's
+    (#615). A command that echoes its exit status is read until that marker, with
+    a number, and the prompt line after it. The echoed command line carries a
+    literal ``$?`` and cannot match.
+    """
+    markers = _STATUS_ECHO_RE.findall(command)
+    if not markers:
+        return _PROMPT_RE
+    return re.escape(markers[-1]) + r"\d+\s*\n.*" + _PROMPT_RE
+
+
 def _split_status(raw: str) -> tuple[str, int | None]:
     """``(output, exit_status)`` of a command followed by ``echo {_RC_MARKER}$?``.
 
@@ -95,6 +120,11 @@ _REFRESH = {
 }
 _YUM_SECURITY_COMMAND = "LC_ALL=C yum updateinfo list security -q 2>/dev/null"
 
+#: Restart the host two seconds later, detached from this session: the launcher's
+#: exit status comes back before the host goes down, and closing the session
+#: cannot take the restart with it.
+_REBOOT_DETACHED = "sh -c '(trap \"\" HUP; sleep 2; /sbin/reboot) </dev/null >/dev/null 2>&1 &'"
+
 #: 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. "
@@ -306,7 +336,7 @@ class LinuxDriver(KernelFactsMixin, SystemdServicesMixin, HostStatusMixin, OSDri
             command,
             read_timeout=read_timeout,
             cmd_verify=False,
-            expect_string=r'[#$\>]\s*$',
+            expect_string=_expect_for(command),
         ).strip()
 
     def _sudo(self, command: str, read_timeout: float = 100) -> str:
@@ -364,6 +394,21 @@ class LinuxDriver(KernelFactsMixin, SystemdServicesMixin, HostStatusMixin, OSDri
             return self._sudo(command, read_timeout=timeout)
         return self._send(f"sudo -n {command}", read_timeout=timeout)
 
+    def reboot_host(self) -> None:
+        """Restart the host (``HostRebootMixin``); returns once the restart is under way.
+
+        :raises RuntimeError: when the host refuses -- sudo without a password,
+            no ``reboot`` -- or its answer carried no exit status.
+        """
+        output, status = _split_status(
+            self._run_privileged(f"{_REBOOT_DETACHED}; echo {_RC_MARKER}$?", 30)
+        )
+        if status != 0:
+            reason = output or f"the reboot command exited with status {status}"
+            if "password is required" in output:
+                reason = f"{reason}\n{_SUDO_PASSWORD_HINT}"
+            raise RuntimeError(reason)
+
     def _run_host_status_command(self, command: str) -> str:
         """The transport for ``HostStatusMixin.get_host_status``: read-only, no sudo."""
         return self._send(command, read_timeout=60)

tests/test_linux.py

@@ -1,5 +1,7 @@
 """Unit tests for LinuxDriver – parsing helpers (no real SSH connection needed)."""
 
+import re
+
 import pytest
 from unittest.mock import MagicMock, patch
 from napalm_linux.linux import LinuxDriver, _arm_vendor_from_model
@@ -1050,6 +1052,106 @@ class TestSudoStatus:
         assert "__NETORK_RC=1" not in output
 
 
+class _Channel:
+    """A netmiko connection that hands out its output in chunks and stops where
+    netmiko does: at the first chunk after which ``expect_string`` matches all
+    that was read so far."""
+
+    def __init__(self, chunks):
+        self.chunks = list(chunks)
+        self.patterns: list = []
+
+    def send_command(self, command, *, expect_string, **_kwargs):
+        self.patterns.append(expect_string)
+        output = ""
+        while self.chunks:
+            output += self.chunks.pop(0)
+            if re.search(expect_string, output):
+                return output
+        raise TimeoutError(f"pattern not detected: {expect_string!r}")
+
+
+#: What vault-01 sent on 2026-10-06 while its apt proxy served a corrupted
+#: InRelease: the signature line ends in ">", which looks like a prompt (#615).
+_BADSIG_CHUNKS = [
+    "sudo -n apt-get update -q 2>&1; echo __NETORK_RC=$?\n",
+    "Fehl:2 http://archive.ubuntu.com/ubuntu noble-updates InRelease\n"
+    "  Die folgenden Signaturen waren ungültig: BADSIG 871920D1991BC93C "
+    "Ubuntu Archive Automatic Signing Key (2018) <ftpmaster@ubuntu.com>\n",
+    "W: Fehler beim Holen von http://archive.ubuntu.com/ubuntu/dists/noble-updates/InRelease\n"
+    "E: Das Depot ist nicht signiert.\n__NETORK_RC=100\n",
+    "chris@vault-01:~$ ",
+]
+
+
+class TestReadToTheEnd:
+    """A line the command prints can end in ``>``, ``#`` or ``$`` -- apt's
+    ``<ftpmaster@ubuntu.com>`` after a bad signature. Taken for the prompt, it
+    ended the read while the command still ran, and the rest arrived as the next
+    command's output (#615). A command that echoes its exit status is read until
+    that marker and the prompt after it."""
+
+    def test_a_signature_line_does_not_end_the_refresh(self, driver):
+        driver._root = False
+        driver._device = _Channel(_BADSIG_CHUNKS)
+
+        result = driver.refresh_available_updates()
+
+        assert result["success"] is False
+        assert "E: Das Depot ist nicht signiert." in result["output"]
+
+    def test_the_session_stays_in_step(self, driver):
+        """Everything up to the prompt is consumed, so the next command reads its own output."""
+        driver._root = False
+        driver._device = _Channel(_BADSIG_CHUNKS + ["true\n", "__NETORK_RC=0\nchris@vault-01:~$ "])
+
+        driver.refresh_available_updates()
+        output, status = driver._sudo_status("true")
+
+        assert status == 0
+        assert "BADSIG" not in output
+
+    def test_the_echoed_command_does_not_count_as_the_marker(self, driver):
+        """Its literal ``$?`` is no number."""
+        channel = _Channel(["sudo true; echo __NETORK_RC=$?\n", "__NETORK_RC=0\nchris@vault-01:~$ "])
+        driver._device = channel
+
+        assert driver._sudo_status("true")[1] == 0
+        assert channel.chunks == []
+
+    def test_the_marker_alone_is_not_the_end(self, driver):
+        """The prompt after it has to be read too, or it would start the next output."""
+        channel = _Channel(["out\n__NETORK_RC=0\n", "chris@vault-01:~$ "])
+        driver._device = channel
+
+        driver._sudo_status("true")
+
+        assert channel.chunks == []
+
+    @pytest.mark.parametrize(
+        "command",
+        [
+            "{ LC_ALL=C apt list --upgradable 2>/dev/null; echo __APT_RC=$?; } | cat",
+            "timeout 45 systemctl restart -- cron.service; echo __SVC_RC=$?",
+        ],
+    )
+    def test_every_status_marker_is_waited_for(self, driver, command):
+        marker = re.search(r"echo (__[A-Z_]+=)", command).group(1)
+        channel = _Channel([f"x <a@b>\n", f"{marker}0\nchris@host:~$ "])
+        driver._device = channel
+
+        output = driver._send(command)
+
+        assert f"{marker}0" in output
+
+    def test_a_command_without_a_marker_still_ends_at_the_prompt(self, driver):
+        channel = _Channel(["6.8.0-142-generic\nchris@host:~$ "])
+        driver._device = channel
+
+        assert driver._send("uname -r").startswith("6.8.0-142-generic")
+        assert channel.patterns == [r"[#$\>]\s*$"]
+
+
 class TestUninstallExitStatus:
     """Whether a removal worked is what the package manager's exit status says.
 
@@ -1415,3 +1517,41 @@ class TestTerminalCodes:
         driver._device.send_command.side_effect = ["0", "Hit:1 noble InRelease\n\x1b>__NETORK_RC=0"]
 
         assert driver.refresh_available_updates()["success"] is True
+
+
+class TestRebootHost:
+    """``reboot_host`` (napalm-device-types' ``HostRebootMixin``) restarts the host.
+
+    Without it netOrk could not restart a Linux host at all: its capability check
+    looks for ``reboot_host`` and found nothing (netOrk #637). The restart is
+    detached and a moment late, so the launcher's exit status comes back before
+    the host goes down, and closing the session cannot take it along.
+    """
+
+    def test_the_driver_can_restart_its_host(self):
+        assert callable(getattr(LinuxDriver, "reboot_host", None))
+
+    def test_the_restart_is_detached_and_privileged(self, driver):
+        driver._root = False
+        driver._device.send_command.return_value = "\n__NETORK_RC=0"
+
+        driver.reboot_host()
+
+        sent = driver._device.send_command.call_args[0][0]
+        assert sent.startswith("sudo -n sh -c ")
+        assert "/sbin/reboot" in sent and "trap" in sent and "&" in sent
+        assert sent.endswith("echo __NETORK_RC=$?")
+
+    def test_a_refusal_is_raised_with_what_the_host_said(self, driver):
+        driver._root = False
+        driver._device.send_command.return_value = "sudo: a password is required\n__NETORK_RC=1"
+
+        with pytest.raises(RuntimeError, match="password is required"):
+            driver.reboot_host()
+
+    def test_no_exit_status_is_no_success(self, driver):
+        driver._root = True
+        driver._device.send_command.return_value = "something else"
+
+        with pytest.raises(RuntimeError):
+            driver.reboot_host()