Merge pull request 'fix: tell unattended-upgrade from Ubuntu's idle shutdown helper' (#27) from fix/unattended-upgrade-shutdown into main

napalm_device_types/host_status.py

@@ -39,7 +39,8 @@ anything until ``dpkg --configure -a`` has run, while ``apt.systemd.daily``
 exits quietly every day (NetOrk/netork#738). While dpkg, apt, aptitude or
 unattended-upgrade is running, packages are halfway on purpose -- also between
 two of apt's dpkg calls -- so the answer is then ``None``, as on a host without
-dpkg.
+dpkg. unattended-upgrade is told from Ubuntu's idle shutdown helper by its
+command line (:data:`BUSY_CHECK`).
 """
 
 from __future__ import annotations
@@ -57,9 +58,16 @@ _APT_TIMER = "apt-daily-upgrade.timer"
 _DNF_TIMERS = ("dnf-automatic.timer", "dnf-automatic-install.timer")
 _UPGRADE_STAMP = "/var/lib/apt/periodic/upgrade-stamp"
 _DPKG_JOURNAL = "/var/lib/dpkg/updates"
-#: The processes that leave packages halfway while they work (``comm``, at most
-#: 15 characters: unattended-upgrade shows as ``unattended-upgr``).
-_PACKAGE_MANAGERS = "dpkg|apt|apt-get|aptitude|unattended-upgr"
+#: Prints ``[dpkg-busy]`` while a process that leaves packages halfway is working:
+#: dpkg, apt, apt-get or aptitude by name, unattended-upgrade by its command line.
+#: Its name is cut to 15 characters, ``unattended-upgr``, the same as that of
+#: ``unattended-upgrade-shutdown --wait-for-signal``, which Ubuntu keeps running all
+#: the time and which installs nothing.
+BUSY_CHECK = (
+    "{ ps -e -o comm= 2>/dev/null | grep -qxE 'dpkg|apt|apt-get|aptitude' || "
+    "ps -e -o args= 2>/dev/null | grep -qE '(^|[ /])unattended-upgrade( |$)'; } "
+    "&& echo '[dpkg-busy]'; "
+)
 
 #: 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.
@@ -79,7 +87,7 @@ HOST_STATUS_COMMAND = (
     "apt-config dump 2>/dev/null | grep '^APT::Periodic::Unattended-Upgrade '; "
     f"echo '[upgrade-stamp]'; stat -c %Y {_UPGRADE_STAMP} 2>/dev/null; fi; "
     "if command -v dpkg-query >/dev/null 2>&1; then "
-    f"ps -e -o comm= 2>/dev/null | grep -qxE '{_PACKAGE_MANAGERS}' && echo '[dpkg-busy]'; "
+    f"{BUSY_CHECK}"
     f"echo '[dpkg-journal]'; ls -1 {_DPKG_JOURNAL} 2>/dev/null | head -n 20; "
     "echo '[dpkg-audit]'; dpkg-query -W -f='${db:Status-Abbrev} ${Package}\\n' 2>/dev/null "
     "| awk 'substr($0, 2, 1) ~ /[HUFWt]/ || substr($0, 3, 1) == \"R\"' | head -n 50; fi; "

pyproject.toml

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

tests/test_host_status.py

@@ -11,10 +11,14 @@ The fixtures are the real states of six hosts on netOrk's test server.
 
 from __future__ import annotations
 
+import re
+import subprocess
+
 import pytest
 
 from napalm_device_types import OSDriver
 from napalm_device_types.host_status import (
+    BUSY_CHECK,
     HOST_STATUS_COMMAND,
     HostStatusMixin,
     kernel_reboot_pending,
@@ -375,3 +379,52 @@ class TestAutoUpdatesLastSuccess:
     def test_the_command_reads_the_upgrade_stamp_only(self):
         assert "/var/lib/apt/periodic/upgrade-stamp" in HOST_STATUS_COMMAND
         assert "update-success-stamp" not in HOST_STATUS_COMMAND
+
+
+class TestBusyCheck:
+    """Whether a package manager is working right now, run through ``sh`` with a
+    stand-in ``ps``. On Ubuntu, unattended-upgrades keeps
+    ``unattended-upgrade-shutdown --wait-for-signal`` running all the time; its
+    process name is cut to 15 characters, ``unattended-upgr``, the same as a real
+    run's. Taken for a run, it made every such host's dpkg state unknown -- aris
+    among them (NetOrk/netork#738)."""
+
+    UBUNTU_IDLE = [
+        ("systemd", "/sbin/init"),
+        ("unattended-upgr", "/usr/bin/python3 /usr/share/unattended-upgrades/unattended-upgrade-shutdown --wait-for-signal"),
+        ("sshd", "sshd: netork [priv]"),
+    ]
+
+    @staticmethod
+    def _busy(tmp_path, processes) -> bool:
+        comm = "\n".join(c for c, _ in processes)
+        args = "\n".join(a for _, a in processes)
+        ps = tmp_path / "ps"
+        ps.write_text(
+            "#!/bin/sh\n"
+            f"case \"$*\" in *comm=*) printf '%s\\n' '{comm}';; *) printf '%s\\n' '{args}';; esac\n"
+        )
+        ps.chmod(0o755)
+        env = {"PATH": f"{tmp_path}:/usr/bin:/bin"}
+        out = subprocess.run(["sh", "-c", BUSY_CHECK], capture_output=True, text=True, env=env)
+        return "[dpkg-busy]" in out.stdout
+
+    def test_the_shutdown_helper_of_an_idle_ubuntu_host_is_no_run(self, tmp_path):
+        assert self._busy(tmp_path, self.UBUNTU_IDLE) is False
+
+    def test_unattended_upgrade_running(self, tmp_path):
+        run = ("unattended-upgr", "/usr/bin/python3 /usr/bin/unattended-upgrade")
+
+        assert self._busy(tmp_path, [*self.UBUNTU_IDLE, run]) is True
+
+    @pytest.mark.parametrize("name", ["dpkg", "apt", "apt-get", "aptitude"])
+    def test_dpkg_and_apt(self, tmp_path, name):
+        assert self._busy(tmp_path, [*self.UBUNTU_IDLE, (name, f"/usr/bin/{name} upgrade")]) is True
+
+    def test_it_is_part_of_the_command(self):
+        assert BUSY_CHECK in HOST_STATUS_COMMAND
+
+    def test_the_command_does_not_match_itself(self):
+        """Run over an exec channel, the command is the command line of the shell
+        that runs it, and ``ps -o args`` lists it."""
+        assert not re.search(r"(^|[ /])unattended-upgrade( |$)", HOST_STATUS_COMMAND)