Merge pull request 'feat: say where a pending update comes from, whether it is a security fix, and whether the host needs a reboot' (#6) from feat/update-origin-host-status into main

README.md

@@ -79,6 +79,17 @@ 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.
 
+`HostStatusMixin` (`get_host_status`) is mixed in the same way: whether a Linux host needs
+a reboot to finish an update (`/var/run/reboot-required`, `needs-restarting -r`, or a newer
+kernel of the running flavour installed) and whether it patches itself (unattended-upgrades,
+dnf-automatic). `package_updates` holds the shared apt and dnf parsers: apt's suites become
+an update's `origin`, a `-security` suite makes it a security update, and dnf's security
+advisories do the same.
+
+**Update readers raise when they cannot read.** `get_available_updates` returns an empty
+list only when nothing is pending; netOrk keeps "pending since" per package, and an empty
+list for "don't know" would reset it.
+
 `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

napalm_device_types/__init__.py

@@ -38,6 +38,7 @@ instead of being restated on every role that happens to need it:
 * :class:`~napalm_device_types.dhcp.DhcpServerMixin`
 * :class:`~napalm_device_types.firewall_rules.FirewallRuleMixin`
 * :class:`~napalm_device_types.health_metrics.HealthMetricsMixin`
+* :class:`~napalm_device_types.host_status.HostStatusMixin`
 * :class:`~napalm_device_types.host_reboot.HostRebootMixin`
 * :class:`~napalm_device_types.interface_filter.InterfaceFilterMixin`
 * :class:`~napalm_device_types.kernel.KernelFactsMixin`
@@ -74,7 +75,15 @@ from napalm_device_types.packages import PackageManagementMixin
 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.host_status import HOST_STATUS_COMMAND, HostStatusMixin, parse_host_status
+from napalm_device_types.package_updates import (
+    APT_UPGRADABLE_COMMAND,
+    DNF_SECURITY_COMMAND,
+    parse_apt_upgradable,
+    parse_dnf_security,
+)
 from napalm_device_types.services import ServiceControlMixin
+from napalm_device_types.terminal import strip_terminal_codes
 from napalm_device_types.systemd import (
     SYSTEMD_SERVICES_COMMAND,
     SystemdServicesMixin,
@@ -95,6 +104,8 @@ __all__ = [
     "FirewallDriver",
     "FirewallRuleMixin",
     "HealthMetricsMixin",
+    "HOST_STATUS_COMMAND",
+    "HostStatusMixin",
     "HostRebootMixin",
     "HypervisorDriver",
     "InterfaceFilterMixin",
@@ -103,6 +114,12 @@ __all__ = [
     "NatVpnMixin",
     "OSDriver",
     "PackageManagementMixin",
+    "APT_UPGRADABLE_COMMAND",
+    "DNF_SECURITY_COMMAND",
+    "parse_apt_upgradable",
+    "parse_dnf_security",
+    "parse_host_status",
+    "strip_terminal_codes",
     "KernelFactsMixin",
     "KERNEL_FACTS_COMMAND",
     "parse_kernel_facts",

napalm_device_types/host_status.py

@@ -0,0 +1,177 @@
+# -*- coding: utf-8 -*-
+"""Host status: does the host need a reboot, and does it patch itself?
+
+A patch run that installed a new kernel or libc has not closed anything until
+the host restarts, so "reboot required" is part of being patched. Whether the
+host installs updates on its own (unattended-upgrades, dnf-automatic) decides
+how far netOrk's maintenance window reaches. Both are read the same way on every
+Linux host, so the command and its parse live here once and a driver only
+carries the command across.
+
+**Reboot required** is any of:
+
+- ``/var/run/reboot-required`` exists. Ubuntu always writes it; Debian does when
+  update-notifier or unattended-upgrades is installed.
+- ``needs-restarting -r`` exits 1 (dnf-utils).
+- A kernel newer than the running one is installed, of the same flavour. A
+  Raspberry Pi carries ``rpi-v8`` and ``rpi-2712`` builds side by side, and only
+  the running one's counts.
+
+It is ``None`` when none of these could be read, for example in a container
+without a ``/lib/modules`` of its own.
+
+**Auto updates** is apt's ``APT::Periodic::Unattended-Upgrade`` (set, not "0",
+and ``apt-daily-upgrade.timer`` not disabled) or an enabled dnf-automatic timer.
+It is ``None`` on a host with neither apt nor dnf-automatic.
+"""
+
+from __future__ import annotations
+
+import re
+from typing import Dict, List, Optional, Tuple, TYPE_CHECKING
+
+from napalm_device_types.models import HostStatusDict
+from napalm_device_types.terminal import strip_terminal_codes
+
+_BEGIN = "HSTAT_BEGIN"
+_END = "HSTAT_END"
+_REBOOT_FILE = "/var/run/reboot-required"
+_APT_TIMER = "apt-daily-upgrade.timer"
+_DNF_TIMERS = ("dnf-automatic.timer", "dnf-automatic-install.timer")
+
+#: One line, POSIX ``sh``, read-only, no privileges. The frame markers are
+#: printed in two halves so that an echoing transport does not show them early.
+#: Each timer is asked on its own: older systemd prints nothing for an unknown
+#: unit, which would shift a combined answer. Run through a pipe, so nothing in
+#: it sees a terminal and colours its output.
+HOST_STATUS_COMMAND = (
+    "{ printf '%s%s\\n' HSTAT_ BEGIN; "
+    f"[ -f {_REBOOT_FILE} ] && echo '[reboot-required]'; "
+    "if command -v needs-restarting >/dev/null 2>&1; then echo '[needs-restarting]'; "
+    "needs-restarting -r >/dev/null 2>&1; echo $?; fi; "
+    "echo '[kernel]'; uname -r; echo '[modules]'; ls -1 /lib/modules 2>/dev/null; "
+    "if command -v apt-config >/dev/null 2>&1; then echo '[apt-config]'; "
+    "apt-config dump 2>/dev/null | grep '^APT::Periodic::Unattended-Upgrade '; fi; "
+    f"echo '[timers]'; for u in {_APT_TIMER} {' '.join(_DNF_TIMERS)}; do "
+    'printf \'%s %s\\n\' "$u" "$(systemctl is-enabled "$u" 2>/dev/null)"; done; '
+    "printf '%s%s\\n' HSTAT_ END; } 2>/dev/null | cat"
+)
+
+_PERIODIC = re.compile(r'^APT::Periodic::Unattended-Upgrade\s+"([^"]*)"')
+_OFF_STATES = frozenset({"disabled", "masked"})
+
+
+def _sections(output: str) -> Dict[str, List[str]]:
+    lines = [line.strip() for line in strip_terminal_codes(output).splitlines()]
+    try:
+        start = lines.index(_BEGIN)
+        end = lines.index(_END, start)
+    except ValueError:
+        raise ValueError("no intact host status report in the output") from None
+    sections: Dict[str, List[str]] = {}
+    current: List[str] = []
+    for line in lines[start + 1 : end]:
+        if line.startswith("[") and line.endswith("]"):
+            current = sections.setdefault(line[1:-1], [])
+        elif line:
+            current.append(line)
+    return sections
+
+
+def _version_key(version: str) -> Tuple[object, ...]:
+    """Natural order: 6.8.0-142 after 6.8.0-87, 7.0.14 after 7.0.2."""
+    return tuple(int(part) if part.isdigit() else part for part in re.split(r"(\d+)", version))
+
+
+def kernel_reboot_pending(running: str, installed: List[str]) -> Optional[str]:
+    """The newest installed kernel of the running flavour, if it is newer than the
+    running one; otherwise None.
+
+    The flavour is what follows the last ``-`` (``generic``, ``amd64``, ``pve``,
+    ``v8``); a kernel of another flavour is never a reason to reboot.
+    """
+    flavour = running.rsplit("-", 1)[-1]
+    same = [k for k in installed if k.rsplit("-", 1)[-1] == flavour]
+    if not same:
+        return None
+    newest = max(same, key=lambda k: _version_key(k.rsplit("-", 1)[0]))
+    if _version_key(newest.rsplit("-", 1)[0]) > _version_key(running.rsplit("-", 1)[0]):
+        return newest
+    return None
+
+
+def _reboot(sections: Dict[str, List[str]]) -> Tuple[Optional[bool], Optional[str]]:
+    if "reboot-required" in sections:
+        return True, f"{_REBOOT_FILE} is present"
+    needs = sections.get("needs-restarting")
+    if needs and needs[0] == "1":
+        return True, "needs-restarting -r reports a reboot"
+    running = (sections.get("kernel") or [""])[0]
+    modules = sections.get("modules") or []
+    newer = kernel_reboot_pending(running, modules) if running and modules else None
+    if newer:
+        return True, f"kernel {newer} installed, {running} running"
+    if needs or modules:
+        return False, None
+    return None, None
+
+
+def _timer_states(sections: Dict[str, List[str]]) -> Dict[str, str]:
+    states: Dict[str, str] = {}
+    for line in sections.get("timers") or []:
+        unit, _, state = line.partition(" ")
+        states[unit] = state.strip()
+    return states
+
+
+def _auto_updates(sections: Dict[str, List[str]]) -> Optional[bool]:
+    timers = _timer_states(sections)
+    if any(timers.get(t) == "enabled" for t in _DNF_TIMERS):
+        return True
+    if "apt-config" not in sections:
+        return None
+    match = next(filter(None, (_PERIODIC.match(line) for line in sections["apt-config"])), None)
+    switched_on = match is not None and match.group(1) not in ("", "0")
+    return switched_on and timers.get(_APT_TIMER) not in _OFF_STATES
+
+
+def parse_host_status(output: str) -> HostStatusDict:
+    """Parse what :data:`HOST_STATUS_COMMAND` printed.
+
+    :raises ValueError: when the output carries no intact report.
+    """
+    sections = _sections(output)
+    required, reason = _reboot(sections)
+    return {
+        "reboot_required": required,
+        "reboot_reason": reason,
+        "auto_updates": _auto_updates(sections),
+    }
+
+
+class HostStatusMixin:
+    """Adds :meth:`get_host_status` to a driver that can run a command on a Linux host.
+
+    The template form, like :class:`~napalm_device_types.kernel.KernelFactsMixin`:
+    the reading and its parse are the same everywhere, and a driver supplies only
+    :meth:`_run_host_status_command`. Mixed in by the drivers that can, so
+    ``hasattr(driver, "get_host_status")`` stays a truthful answer.
+    """
+
+    if TYPE_CHECKING:  # pragma: no cover - declared for type checkers only
+
+        def _run_host_status_command(self, command: str) -> str:
+            """Run *command* on the host with ``sh`` and return what it printed."""
+            ...
+
+    def get_host_status(self) -> HostStatusDict:
+        """
+        Returns whether the host needs a reboot and whether it patches itself.
+
+        * reboot_required (bool or None)
+        * reboot_reason (string or None)
+        * auto_updates (bool or None)
+
+        :raises ValueError: if the host's output carried no intact report.
+        """
+        return parse_host_status(self._run_host_status_command(HOST_STATUS_COMMAND))

napalm_device_types/models.py

@@ -60,11 +60,30 @@ class ServiceDict(TypedDict):
 
 
 class UpdateDict(TypedDict):
-    """A software package that has a newer version available in the package repository."""
+    """A software package that has a newer version available in the package repository.
+
+    ``origin`` and ``security`` are optional: a reader that cannot tell leaves
+    them out, and netOrk treats a missing ``security`` as unknown.
+    """
 
     name: str
     current_version: str
     new_version: str
+    #: Where the new version comes from, e.g. apt's suites "noble-updates,noble-security".
+    origin: NotRequired[Optional[str]]
+    #: True for a security update, False for a known other one, None when unknown.
+    security: NotRequired[Optional[bool]]
+
+
+class HostStatusDict(TypedDict):
+    """What a host says about its own patch state (``HostStatusMixin.get_host_status``)."""
+
+    #: True when the host needs a reboot to finish an update, None when it cannot tell.
+    reboot_required: Optional[bool]
+    #: Why, e.g. "kernel 6.8.0-142-generic installed, 6.8.0-139-generic running".
+    reboot_reason: Optional[str]
+    #: True when the host installs updates on its own (unattended-upgrades, dnf-automatic).
+    auto_updates: Optional[bool]
 
 
 # ---------------------------------------------------------------------------

napalm_device_types/package_updates.py

@@ -0,0 +1,111 @@
+# -*- coding: utf-8 -*-
+"""Pending package updates: which package, from where, and whether it is a security fix.
+
+A patch deadline -- "security updates within 14 days" -- needs to know which
+pending update is a security update. apt says so in the suite a candidate comes
+from (``noble-security``, ``stable-security``), dnf in its update advisories.
+Reading that is the same for every driver whose host runs apt or dnf, so the
+parsers live here once and a driver only carries the command across.
+
+apt: the suites a candidate comes from are its ``origin``; any suite ending in
+``-security`` makes it a security update. A security fix that a later
+``-updates`` build superseded shows only ``-updates`` and counts as not
+security -- netOrk's CVE matching is what catches those.
+"""
+
+from __future__ import annotations
+
+import re
+from typing import Dict, List, Set
+
+from napalm_device_types.models import UpdateDict
+from napalm_device_types.terminal import strip_terminal_codes
+
+#: Read-only, no root needed, in a fixed language so the parse holds, and
+#: through a pipe: without a terminal apt draws no progress and no terminal
+#: codes, one of which (``ESC >``) a screen-scraping transport took for a shell
+#: prompt and stopped reading at. Its exit status is printed inside the group,
+#: so it is apt's, not cat's.
+APT_UPGRADABLE_COMMAND = "{ LC_ALL=C apt list --upgradable 2>/dev/null; echo __APT_RC=$?; } | cat"
+
+#: Read-only. Lists the packages that a pending security advisory covers.
+DNF_SECURITY_COMMAND = "LC_ALL=C dnf updateinfo list --security --quiet 2>/dev/null"
+
+# openssl/noble-updates,noble-security 3.0.13-0ubuntu3.6 amd64 [upgradable from: 3.0.13-0ubuntu3.5]
+_APT_LINE = re.compile(r"^(\S+)/(\S+)\s+(\S+)\s+\S+\s+\[upgradable from:\s+(\S+)\]")
+_SECURITY_SUITE = "-security"
+_APT_STATUS = re.compile(r"^__APT_RC=(\d+)\s*$", re.MULTILINE)
+
+
+def _joined_lines(output: str) -> List[str]:
+    """Lines as apt printed them: a terminal wraps long ones, and the
+    continuation starts with a space."""
+    lines: List[str] = []
+    for line in output.splitlines():
+        if line.startswith(" ") and lines:
+            lines[-1] += line.strip()
+        else:
+            lines.append(line)
+    return lines
+
+
+def _apt_listing(output: str) -> str:
+    """The listing without its exit status, or ``ValueError`` when apt failed or
+    the output was cut short -- "could not read" must never look like "nothing
+    pending"."""
+    text = strip_terminal_codes(output)
+    statuses = _APT_STATUS.findall(text)
+    if not statuses:
+        raise ValueError("apt list --upgradable reported no exit status; the output was cut short")
+    if statuses[-1] != "0":
+        raise ValueError(f"apt list --upgradable failed with exit status {statuses[-1]}")
+    return _APT_STATUS.sub("", text)
+
+
+def parse_apt_upgradable(output: str) -> List[UpdateDict]:
+    """Parse :data:`APT_UPGRADABLE_COMMAND`'s output, one entry per package.
+
+    A package listed for several architectures (``libc6`` for amd64 and i386)
+    is one entry; it counts as a security update if any of its lines does.
+
+    :raises ValueError: when apt failed or its exit status never arrived.
+    """
+    by_name: Dict[str, UpdateDict] = {}
+    for line in _joined_lines(_apt_listing(output)):
+        match = _APT_LINE.match(line)
+        if not match:
+            continue
+        name, listed, new_version, current_version = match.groups()
+        suites = list(dict.fromkeys(listed.split(",")))  # apt may list a suite twice
+        security = any(suite.endswith(_SECURITY_SUITE) for suite in suites)
+        seen = by_name.get(name)
+        if seen is not None:
+            seen["security"] = bool(seen.get("security")) or security
+            continue
+        by_name[name] = {
+            "name": name,
+            "current_version": current_version,
+            "new_version": new_version,
+            "origin": ",".join(suites),
+            "security": security,
+        }
+    return list(by_name.values())
+
+
+def nevra_name(nevra: str) -> str:
+    """The package name of an RPM ``name-[epoch:]version-release.arch``."""
+    without_arch = nevra.rsplit(".", 1)[0]
+    return without_arch.rsplit("-", 2)[0]
+
+
+def parse_dnf_security(output: str) -> Set[str]:
+    """The names of the packages a pending security advisory covers.
+
+    Parses :data:`DNF_SECURITY_COMMAND`'s ``ADVISORY  SEVERITY/Sec.  NEVRA`` lines.
+    """
+    names: Set[str] = set()
+    for line in output.splitlines():
+        parts = line.split()
+        if len(parts) >= 3 and parts[1].endswith("/Sec."):
+            names.add(nevra_name(parts[-1]))
+    return names

napalm_device_types/systemd.py

@@ -35,6 +35,7 @@ from typing import Any, Dict, List, Set, Tuple, TYPE_CHECKING
 
 from napalm_device_types.models import ServiceDict
 from napalm_device_types.services import ServiceControlMixin
+from napalm_device_types.terminal import strip_terminal_codes
 
 _BEGIN = "SVC_BEGIN"
 _END = "SVC_END"
@@ -88,9 +89,6 @@ _RC_RE = re.compile(rf"^{_RC_MARKER}(\d+)\s*$", re.MULTILINE)
 _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"})
@@ -139,7 +137,7 @@ def parse_action_result(output: str) -> Dict[str, Any]:
     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)
+    output = strip_terminal_codes(output)
     statuses = _RC_RE.findall(output)
     text = _RC_RE.sub("", output).strip()
     if not statuses:
@@ -154,7 +152,7 @@ def parse_action_result(output: str) -> Dict[str, Any]:
 
 
 def _frame(output: str) -> List[str]:
-    lines = [line.strip() for line in _ANSI_RE.sub("", output).splitlines()]
+    lines = [line.strip() for line in strip_terminal_codes(output).splitlines()]
     try:
         start = lines.index(_BEGIN)
         end = lines.index(_END, start)

napalm_device_types/terminal.py

@@ -0,0 +1,23 @@
+# -*- coding: utf-8 -*-
+"""What a pseudo-terminal adds to a command's output, taken out again.
+
+A screen-scraping transport (netmiko) gives the remote command a terminal. Tools
+then colour their output and draw progress: systemctl colours its errors, apt
+switches the keypad mode with ``ESC =`` / ``ESC >``. Every parser in this package
+reads the text without them.
+"""
+
+from __future__ import annotations
+
+import re
+
+#: CSI sequences (colours, cursor), OSC sequences (window titles) and the
+#: two-character escapes (``ESC =``, ``ESC >``, ``ESC (B``).
+_TERMINAL_CODES = re.compile(
+    r"\x1b(?:\[[0-?]*[ -/]*[@-~]|\][^\x07\x1b]*(?:\x07|\x1b\\)|\([0-9A-Za-z]|[=>78DEHMNOc])"
+)
+
+
+def strip_terminal_codes(text: str) -> str:
+    """*text* without terminal escape sequences."""
+    return _TERMINAL_CODES.sub("", text)

napalm_device_types/updates.py

@@ -13,7 +13,7 @@ this class in can never shadow a working implementation from a sibling base.
 
 from __future__ import annotations
 
-from typing import List, TYPE_CHECKING
+from typing import Any, Dict, List, TYPE_CHECKING
 
 from napalm_device_types.models import ApplyUpdatesResultDict, UpdateDict
 
@@ -30,6 +30,14 @@ class UpdateMixin:
             * name (string) - package name
             * current_version (string) - currently installed version
             * new_version (string) - version available in the repository
+            * origin (string, optional) - where it comes from, e.g. apt's suites
+            * security (bool or None, optional) - a security update; leave it
+              out or None when the source does not say
+
+            **An empty list means nothing is pending.** A reader that cannot
+            read -- no package index yet, an API that did not answer -- raises
+            instead: netOrk keeps "pending since" per package, and an empty
+            list for "don't know" would reset every one of those clocks.
 
             Example::
 
@@ -43,6 +51,16 @@ class UpdateMixin:
             """
             ...
 
+        def refresh_available_updates(self) -> Dict[str, Any]:
+            """
+            Refreshes the host's package index, so that :meth:`get_available_updates`
+            reports what the repositories offer now (``apt-get update``,
+            ``dnf makecache``, ``opkg update``, a firmware check). Installs nothing.
+
+            :returns: ``{"success": bool, "output": str}``
+            """
+            ...
+
         def apply_updates(self, packages: List[str]) -> ApplyUpdatesResultDict:
             """
             Upgrades the given packages to the newest available version.

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "napalm-device-types"
-version = "2.2.0"
+version = "2.3.0"
 description = "Abstract device-type base classes for NAPALM drivers"
 readme = "README.md"
 requires-python = ">=3.10"

tests/test_host_status.py

@@ -0,0 +1,200 @@
+"""Host status: does the host need a reboot, and does it patch itself?
+
+A patch run that installed a new kernel has not closed anything until the host
+boots it, so "reboot required" is part of being patched. Whether the host
+installs updates on its own (unattended-upgrades, dnf-automatic) is what netOrk
+shows next to the window it governs. Both are read the same way on every Linux
+host, so the command and its parse live here once (netOrk MVP 5).
+
+The fixtures are the real states of six hosts on netOrk's test server.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from napalm_device_types import OSDriver
+from napalm_device_types.host_status import (
+    HOST_STATUS_COMMAND,
+    HostStatusMixin,
+    kernel_reboot_pending,
+    parse_host_status,
+)
+
+
+def _wire(
+    *,
+    reboot_file=False,
+    running="6.8.0-142-generic",
+    modules=("6.8.0-139-generic", "6.8.0-142-generic"),
+    needs_restarting=None,
+    periodic=None,
+    timer="enabled",
+    dnf_timers=("not-found", "not-found"),
+):
+    lines = ["HSTAT_BEGIN"]
+    if reboot_file:
+        lines.append("[reboot-required]")
+    if needs_restarting is not None:
+        lines += ["[needs-restarting]", str(needs_restarting)]
+    lines += ["[kernel]", running, "[modules]", *modules]
+    if periodic is not None:
+        lines += ["[apt-config]", *periodic]
+    lines += [
+        "[timers]",
+        f"apt-daily-upgrade.timer {timer}",
+        f"dnf-automatic.timer {dnf_timers[0]}",
+        f"dnf-automatic-install.timer {dnf_timers[1]}",
+        "HSTAT_END",
+    ]
+    return "\n".join(lines) + "\n"
+
+
+UNATTENDED = ['APT::Periodic::Update-Package-Lists "1";', 'APT::Periodic::Unattended-Upgrade "1";']
+
+
+class TestRebootRequired:
+    def test_the_reboot_required_file_says_so(self):
+        status = parse_host_status(_wire(reboot_file=True))
+
+        assert status["reboot_required"] is True
+        assert "reboot-required" in status["reboot_reason"]
+
+    def test_a_newer_installed_kernel_than_the_running_one(self):
+        """z2m-garden: running 6.8.0-139, 6.8.0-142 installed."""
+        status = parse_host_status(
+            _wire(
+                running="6.8.0-139-generic",
+                modules=("6.8.0-87-generic", "6.8.0-139-generic", "6.8.0-142-generic"),
+            )
+        )
+
+        assert status["reboot_required"] is True
+        assert "6.8.0-142-generic" in status["reboot_reason"]
+
+    def test_the_newest_kernel_running_needs_none(self):
+        """vault-01: 6.8.0-142 running and newest; 6.8.0-94 sorts below it."""
+        status = parse_host_status(
+            _wire(running="6.8.0-142-generic", modules=("6.8.0-94-generic", "6.8.0-142-generic"))
+        )
+
+        assert status["reboot_required"] is False
+        assert status["reboot_reason"] is None
+
+    def test_needs_restarting_exit_1_means_reboot(self):
+        assert parse_host_status(_wire(needs_restarting=1))["reboot_required"] is True
+
+    def test_needs_restarting_exit_0_does_not(self):
+        assert parse_host_status(_wire(needs_restarting=0))["reboot_required"] is False
+
+    def test_no_kernel_information_is_unknown(self):
+        """A container has no /lib/modules of its own."""
+        status = parse_host_status(_wire(modules=()))
+
+        assert status["reboot_required"] is None
+
+
+class TestKernelRebootPending:
+    @pytest.mark.parametrize(
+        ("running", "installed", "newer"),
+        [
+            # Raspberry Pi: two flavours side by side; only the running one counts.
+            (
+                "6.18.33+rpt-rpi-v8",
+                [
+                    "6.12.75+rpt-rpi-2712",
+                    "6.12.75+rpt-rpi-v8",
+                    "6.18.33+rpt-rpi-2712",
+                    "6.18.33+rpt-rpi-v8",
+                ],
+                None,
+            ),
+            # Debian (OMV): 7.1.8 installed while 7.1.3 runs.
+            (
+                "7.1.3+deb13-amd64",
+                ["6.12.57+deb13-amd64", "7.1.3+deb13-amd64", "7.1.8+deb13-amd64"],
+                "7.1.8+deb13-amd64",
+            ),
+            # Proxmox: 7.0.14-19 is newer than 7.0.2-6, numerically.
+            ("7.0.14-19-pve", ["7.0.14-19-pve", "7.0.2-6-pve"], None),
+            # Arch: the running kernel's modules were replaced by the upgrade.
+            ("6.10.5-arch1-1", ["6.10.9-arch1-1"], "6.10.9-arch1-1"),
+        ],
+    )
+    def test_the_newer_kernel_of_the_running_flavour(self, running, installed, newer):
+        assert kernel_reboot_pending(running, installed) == newer
+
+
+class TestAutoUpdates:
+    def test_unattended_upgrades_switched_on(self):
+        assert parse_host_status(_wire(periodic=UNATTENDED))["auto_updates"] is True
+
+    def test_apt_without_the_setting_does_not_patch_itself(self):
+        """Proxmox and Raspberry Pi OS: apt-config answers, the setting is absent."""
+        assert parse_host_status(_wire(periodic=[]))["auto_updates"] is False
+
+    def test_switched_off_by_zero(self):
+        off = ['APT::Periodic::Unattended-Upgrade "0";']
+
+        assert parse_host_status(_wire(periodic=off))["auto_updates"] is False
+
+    def test_a_disabled_timer_stops_it_even_when_configured(self):
+        status = parse_host_status(_wire(periodic=UNATTENDED, timer="disabled"))
+
+        assert status["auto_updates"] is False
+
+    def test_dnf_automatic(self):
+        status = parse_host_status(_wire(dnf_timers=("enabled", "not-found")))
+
+        assert status["auto_updates"] is True
+
+    def test_neither_apt_nor_dnf_is_unknown(self):
+        assert parse_host_status(_wire())["auto_updates"] is None
+
+
+class TestTheReport:
+    def test_a_cut_short_report_raises(self):
+        with pytest.raises(ValueError):
+            parse_host_status(_wire().replace("HSTAT_END\n", ""))
+
+    def test_the_frame_is_not_in_the_command_itself(self):
+        assert "HSTAT_BEGIN" not in HOST_STATUS_COMMAND
+        assert "HSTAT_END" not in HOST_STATUS_COMMAND
+
+    def test_it_runs_without_a_terminal(self):
+        """No colour codes from ls, nothing a screen scraper could take for a prompt."""
+        assert HOST_STATUS_COMMAND.rstrip().endswith("| cat")
+
+    def test_terminal_codes_are_dropped(self):
+        status = parse_host_status(
+            "\x1b[0m" + _wire(reboot_file=True).replace("HSTAT_END", "\x1b>HSTAT_END")
+        )
+
+        assert status["reboot_required"] is True
+
+    def test_it_changes_nothing(self):
+        for word in ("rm ", "apt-get ", "dnf install", "systemctl start", "reboot"):
+            assert word not in HOST_STATUS_COMMAND.replace("reboot-required", "")
+
+
+class _Driver(HostStatusMixin):
+    def __init__(self, reply: str) -> None:
+        self.reply = reply
+        self.commands: list = []
+
+    def _run_host_status_command(self, command: str) -> str:
+        self.commands.append(command)
+        return self.reply
+
+
+class TestHostStatusMixin:
+    def test_a_driver_supplies_only_the_transport(self):
+        driver = _Driver(_wire(reboot_file=True, periodic=UNATTENDED))
+
+        status = driver.get_host_status()
+
+        assert driver.commands == [HOST_STATUS_COMMAND]
+        assert (status["reboot_required"], status["auto_updates"]) == (True, True)
+
+    def test_not_every_os_driver_has_it(self):
+        assert not hasattr(OSDriver, "get_host_status")

tests/test_package_updates.py

@@ -0,0 +1,149 @@
+"""Pending updates: which package, from where, and whether it closes a security hole.
+
+A patch deadline ("security updates within 14 days") needs to know which pending
+update is a security update. apt says so in the suite a candidate comes from
+(``noble-security``, ``stable-security``); dnf says so in its update advisories.
+Reading that is the same for every driver whose host runs apt or dnf, so the
+parsers live here once (netOrk MVP 5, #556).
+
+Fixture lines are from real hosts (Ubuntu 24.04, Debian 13 / OMV, Proxmox VE 9).
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from napalm_device_types.package_updates import (
+    APT_UPGRADABLE_COMMAND,
+    nevra_name,
+    parse_apt_upgradable,
+    parse_dnf_security,
+)
+
+UBUNTU = """\
+docker-compose-plugin/noble 5.6.0-1~ubuntu.24.04~noble amd64 [upgradable from: 5.5.1-1~ubuntu.24.04~noble]
+openssl/noble-updates,noble-security 3.0.13-0ubuntu3.6 amd64 [upgradable from: 3.0.13-0ubuntu3.5]
+__APT_RC=0
+"""
+
+DEBIAN = """\
+linux-image-amd64/stable-backports 7.1.13-1~bpo13+1 amd64 [upgradable from: 7.1.8-1~bpo13+1]
+libssl3t64/stable-security 3.5.1-1+deb13u2 amd64 [upgradable from: 3.5.1-1+deb13u1]
+__APT_RC=0
+"""
+
+
+def _by_name(updates):
+    return {u["name"]: u for u in updates}
+
+
+class TestAptUpgradable:
+    def test_each_line_is_a_package_with_both_versions(self):
+        update = _by_name(parse_apt_upgradable(UBUNTU))["docker-compose-plugin"]
+
+        assert update["current_version"] == "5.5.1-1~ubuntu.24.04~noble"
+        assert update["new_version"] == "5.6.0-1~ubuntu.24.04~noble"
+
+    def test_the_suites_are_its_origin(self):
+        updates = _by_name(parse_apt_upgradable(UBUNTU))
+
+        assert updates["openssl"]["origin"] == "noble-updates,noble-security"
+        assert updates["docker-compose-plugin"]["origin"] == "noble"
+
+    def test_a_suite_apt_lists_twice_is_named_once(self):
+        line = (
+            "fonts-opensymbol/noble-updates,noble-updates,noble-security,noble-security "
+            "4:102.12+LibO24.2.7-0ubuntu0.24.04.7 all [upgradable from: 4:102.12+LibO24.2.7-0ubuntu0.24.04.6]\n"
+            "__APT_RC=0\n"
+        )
+
+        assert parse_apt_upgradable(line)[0]["origin"] == "noble-updates,noble-security"
+
+    @pytest.mark.parametrize(
+        ("output", "name", "security"),
+        [
+            (UBUNTU, "openssl", True),
+            (UBUNTU, "docker-compose-plugin", False),
+            (DEBIAN, "libssl3t64", True),
+            (DEBIAN, "linux-image-amd64", False),
+        ],
+    )
+    def test_a_security_suite_makes_it_a_security_update(self, output, name, security):
+        assert _by_name(parse_apt_upgradable(output))[name]["security"] is security
+
+    def test_a_line_the_terminal_wrapped_is_joined(self):
+        wrapped = (
+            "openssl/noble-updates,noble-security 3.0.13-0ubuntu3.6 amd64 [upgradable fro\n"
+            " m: 3.0.13-0ubuntu3.5]\n__APT_RC=0\n"
+        )
+
+        assert _by_name(parse_apt_upgradable(wrapped))["openssl"]["current_version"] == (
+            "3.0.13-0ubuntu3.5"
+        )
+
+    def test_one_package_for_several_architectures_is_one_entry(self):
+        multiarch = (
+            "libc6/noble-updates 2.39-0ubuntu8.5 amd64 [upgradable from: 2.39-0ubuntu8.4]\n"
+            "libc6/noble-updates,noble-security 2.39-0ubuntu8.5 i386 [upgradable from: 2.39-0ubuntu8.4]\n"
+            "__APT_RC=0\n"
+        )
+
+        updates = parse_apt_upgradable(multiarch)
+
+        assert [u["name"] for u in updates] == ["libc6"]
+        assert updates[0]["security"] is True
+
+    def test_noise_is_ignored(self):
+        noisy = "Listing... Done\nWARNING: apt does not have a stable CLI interface.\n" + UBUNTU
+
+        assert len(parse_apt_upgradable(noisy)) == 2
+
+    def test_nothing_pending_is_an_empty_list(self):
+        assert parse_apt_upgradable("__APT_RC=0\n") == []
+
+    def test_a_list_without_its_exit_status_raises(self):
+        """Cut short: a transport stopped reading early, so nothing can be concluded."""
+        with pytest.raises(ValueError):
+            parse_apt_upgradable(UBUNTU.replace("__APT_RC=0\n", ""))
+
+    def test_a_failed_apt_raises(self):
+        with pytest.raises(ValueError, match="100"):
+            parse_apt_upgradable("E: Could not get lock\n__APT_RC=100\n")
+
+    def test_terminal_codes_around_the_status_are_dropped(self):
+        """What a pseudo-terminal left on a Raspberry Pi OS host."""
+        raw = "Listing... 0%\n\x1b[?1h\x1b=\n" + UBUNTU.replace("__APT_RC=0", "\x1b>__APT_RC=0")
+
+        assert len(parse_apt_upgradable(raw)) == 2
+
+    def test_the_command_reads_without_root_or_a_terminal(self):
+        """Through a pipe apt draws no progress and no terminal codes; one of
+        those, ESC >, ended a screen-scraping read at a false prompt."""
+        assert "LC_ALL=C apt list --upgradable" in APT_UPGRADABLE_COMMAND
+        assert APT_UPGRADABLE_COMMAND.rstrip().endswith("| cat")
+        assert "sudo" not in APT_UPGRADABLE_COMMAND
+
+
+class TestDnfSecurity:
+    ADVISORIES = """\
+FEDORA-2024-1a2b3c4d5e Important/Sec. openssl-libs-1:3.1.4-2.fc40.x86_64
+FEDORA-2024-1a2b3c4d5e Important/Sec. openssl-1:3.1.4-2.fc40.x86_64
+RLSA-2024:1234         Moderate/Sec.  kernel-core-5.14.0-427.13.1.el9_4.x86_64
+"""
+
+    def test_the_names_of_packages_with_a_security_advisory(self):
+        assert parse_dnf_security(self.ADVISORIES) == {"openssl-libs", "openssl", "kernel-core"}
+
+    def test_nothing_is_an_empty_set(self):
+        assert parse_dnf_security("") == set()
+
+    @pytest.mark.parametrize(
+        ("nevra", "name"),
+        [
+            ("openssl-libs-1:3.1.4-2.fc40.x86_64", "openssl-libs"),
+            ("kernel-core-5.14.0-427.13.1.el9_4.x86_64", "kernel-core"),
+            ("python3-dnf-4.14.0-9.el9.noarch", "python3-dnf"),
+        ],
+    )
+    def test_the_name_of_a_nevra(self, nevra, name):
+        assert nevra_name(nevra) == name