fix(docker): keep a resolvable image reference when the tag has moved

`docker ps` reports a bare image ID instead of the tag as soon as that tag
points at a newer image — which is exactly what a pull without a recreate
does. That ID went straight into `docker buildx imagetools inspect`, which
cannot resolve an image ID, so the lookup failed and the image was silently
dropped from the outdated list. The check was blind in precisely the state
that means an update is waiting.

get_docker_info() now also reads `.Config.Image`, the reference the container
was created from, which never degrades. It compares each running container's
image ID against the ID its own tag currently resolves to, and reports
`image_ref`, `running_image_id`, `restart_pending` and `pending_version`.

get_docker_outdated() prefers `image_ref`, skips bare IDs with a log line
instead of asking the registry about them, and no longer swallows a failed
registry lookup — silence there was indistinguishable from "up to date".

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

napalm_linux/linux.py

@@ -116,6 +116,22 @@ def _arm_vendor_from_model(model: str) -> str:
     return " ".join(brand)
 
 
+#: A bare image ID (``d626d04934cd``), not a registry reference. ``docker ps``
+#: falls back to this whenever the tag a container was created from has since
+#: been moved to a newer image — i.e. exactly after a pull without a recreate.
+_IMAGE_ID_RE = re.compile(r"^(sha256:)?[0-9a-f]{12,64}$")
+
+
+def _looks_like_image_id(ref: str) -> bool:
+    """True if *ref* is an image ID rather than something a registry can resolve."""
+    return bool(_IMAGE_ID_RE.match(ref.strip()))
+
+
+def _short_image_id(raw: str) -> str:
+    """Normalise ``sha256:<64hex>`` and ``<12hex>`` to a comparable 12-char form."""
+    return raw.strip().removeprefix("sha256:")[:12]
+
+
 class LinuxDriver(OSDriver):
     """NAPALM driver for generic Linux systems.
 
@@ -126,6 +142,9 @@ class LinuxDriver(OSDriver):
     TYPE_LABEL = "Linux"
     VENDOR = "Linux"
     DRIVER_NAME = "linux"
+    # A general-purpose host runs through a full init sequence; NAS derivatives
+    # (OpenMediaVault, QNAP) inherit this and are, if anything, slower.
+    REBOOT_SETTLE_SECONDS = 90
     SNMP_FINGERPRINT = [
         FingerprintRule("linux", weight=5.0),
     ]
@@ -381,6 +400,10 @@ class LinuxDriver(OSDriver):
         iface_out = self._send("ip -o link show | awk -F': ' '{print $2}' | cut -d@ -f1")
         interface_list = [i.strip() for i in iface_out.splitlines() if i.strip() and i.strip() != "lo"]
 
+        # Currently-booted kernel release, distinct from an installed-but-not-yet-
+        # booted newer kernel (used for kernel CVE relevance).
+        running_kernel = self._send("uname -r").strip()
+
         return {
             "hostname": hostname,
             "fqdn": fqdn,
@@ -390,6 +413,7 @@ class LinuxDriver(OSDriver):
             "os_version": os_version,
             "uptime": uptime_secs,
             "interface_list": interface_list,
+            "running_kernel": running_kernel,
         }
 
     def _parse_uptime(self) -> int:
@@ -810,17 +834,20 @@ class LinuxDriver(OSDriver):
 
     def _get_packages_apt(self) -> list[PackageDict]:
         out = self._send(
-            "dpkg-query -W -f='${Package}\\t${Version}\\t${Installed-Size}\\t${binary:Summary}\\n' 2>/dev/null"
+            "dpkg-query -W -f='${Package}\\t${Version}\\t${Installed-Size}"
+            "\\t${source:Package}\\t${binary:Summary}\\n' 2>/dev/null"
         )
         packages: list[PackageDict] = []
         for line in out.splitlines():
-            parts = line.split("\t", 3)
+            parts = line.split("\t", 4)
             if len(parts) < 2:
                 continue
             name = parts[0].strip()
             version = parts[1].strip()
             size = int(parts[2].strip()) * 1024 if len(parts) > 2 and parts[2].strip().isdigit() else 0
-            description = parts[3].strip() if len(parts) > 3 else ""
+            # Debian source package (e.g. openssh-server → openssh) for OSV matching.
+            source_package = parts[3].strip() if len(parts) > 3 and parts[3].strip() else name
+            description = parts[4].strip() if len(parts) > 4 else ""
             packages.append({
                 "name": name,
                 "version": version,
@@ -828,6 +855,7 @@ class LinuxDriver(OSDriver):
                 "description": description,
                 "size": size,
                 "source": "apt",
+                "source_package": source_package,
             })
         return packages
 
@@ -1019,7 +1047,7 @@ class LinuxDriver(OSDriver):
         success = not any(kw in low for kw in ("error:", "failed", "not found", "is not installed", "no packages"))
         return {"success": success, "output": raw.strip()}
 
-    def get_pending_updates(self) -> list[UpdateDict]:
+    def get_available_updates(self) -> list[UpdateDict]:
         if self._pkg_manager == "apt":
             return self._get_updates_apt()
         if self._pkg_manager in ("dnf", "yum"):
@@ -1032,10 +1060,6 @@ class LinuxDriver(OSDriver):
             f"Package manager '{self._pkg_manager}' is not supported"
         )
 
-    def get_available_updates(self) -> list[UpdateDict]:
-        """Alias for get_pending_updates(); called by the netork API backend."""
-        return self.get_pending_updates()
-
     def get_device_warnings(self) -> List[dict[str, Any]]:
         """Return warning dicts for issues detected on this device.
 
@@ -1052,8 +1076,6 @@ class LinuxDriver(OSDriver):
         if updates:
             warnings.append({
                 "code": "updates_available",
-                "severity": "warning",
-                "action": None,
                 "meta": {
                     "count": len(updates),
                     "packages": [u.get("name", "") for u in updates],
@@ -1065,8 +1087,6 @@ class LinuxDriver(OSDriver):
                 if self._apt_proxy_url not in current:
                     warnings.append({
                         "code": "apt_proxy_missing",
-                        "severity": "warning",
-                        "action": "fix_apt_proxy",
                         "meta": {"expected_url": self._apt_proxy_url},
                     })
             except Exception as exc:
@@ -1484,6 +1504,16 @@ class LinuxDriver(OSDriver):
     # Docker
     # ------------------------------------------------------------------
 
+    def _docker_bin(self) -> str:
+        """Path to the docker binary.
+
+        A hook rather than a literal because the Docker *logic* is the same
+        everywhere while the *location* is not: QTS ships Container Station's
+        docker under /share/<pool>/.qpkg/ and never puts it on PATH. Subclasses
+        override this one method instead of reimplementing the surface.
+        """
+        return "docker"
+
     def get_docker_info(self) -> DockerInfoDict:
         """Return information about the local Docker environment.
 
@@ -1502,26 +1532,31 @@ class LinuxDriver(OSDriver):
         """
         import json as _json
 
+        docker = self._docker_bin()
+
         # Check docker binary first (docker --version doesn't need socket access)
-        if not self._send("command -v docker 2>/dev/null").strip():
+        if not self._send(f"command -v {docker} 2>/dev/null").strip():
             return {"available": False}
 
         # Verify socket access — docker ps is cheaper and fails immediately on permission errors
-        ps_check = self._send("docker ps 2>&1")
+        ps_check = self._send(f"{docker} ps 2>&1")
         if "permission denied" in ps_check.lower() or "cannot connect" in ps_check.lower():
             return {"available": False, "permission_denied": True}
 
-        version = self._send("docker --version 2>/dev/null").strip()
+        version = self._send(f"{docker} --version 2>/dev/null").strip()
 
         combined = self._send(
             "echo '---CONTAINERS---'; "
-            "docker ps -a --format '{{json .}}' 2>/dev/null; "
+            f"{docker} ps -a --format '{{{{json .}}}}' 2>/dev/null; "
             "echo '---IMAGES---'; "
-            "docker images --format '{{json .}}' 2>/dev/null; "
+            f"{docker} images --format '{{{{json .}}}}' 2>/dev/null; "
             "echo '---VOLUMES---'; "
-            "docker volume ls --format '{{json .}}' 2>/dev/null; "
+            f"{docker} volume ls --format '{{{{json .}}}}' 2>/dev/null; "
             "echo '---NETWORKS---'; "
-            "docker network ls --format '{{json .}}' 2>/dev/null",
+            f"{docker} network ls --format '{{{{json .}}}}' 2>/dev/null; "
+            "echo '---CONFIGIMAGES---'; "
+            f"{docker} ps -aq 2>/dev/null | xargs -r {docker} inspect "
+            f"--format '{{{{.Id}}}}|{{{{.Config.Image}}}}|{{{{.Image}}}}' 2>/dev/null",
             read_timeout=60,
         )
 
@@ -1540,7 +1575,8 @@ class LinuxDriver(OSDriver):
         raw_containers = _section(combined, "---CONTAINERS---", "---IMAGES---")
         raw_images     = _section(combined, "---IMAGES---",     "---VOLUMES---")
         raw_volumes    = _section(combined, "---VOLUMES---",    "---NETWORKS---")
-        raw_networks   = _section(combined, "---NETWORKS---",   "\x00")  # sentinel
+        raw_networks   = _section(combined, "---NETWORKS---",   "---CONFIGIMAGES---")
+        raw_cfgimages  = _section(combined, "---CONFIGIMAGES---", "\x00")  # sentinel
 
         def _parse_labels(raw: Any) -> Dict[str, str]:
             """Parse Docker labels — may be a dict (JSON map) or comma-sep string."""
@@ -1601,6 +1637,44 @@ class LinuxDriver(OSDriver):
             except Exception:
                 pass
 
+        # Stable image reference + restart-pending detection.
+        #
+        # ``docker ps`` only reports a usable tag while that tag still resolves to
+        # the running image. Pull a newer image without recreating the container and
+        # it degrades to a bare image ID — useless as a registry reference, and the
+        # very state in which an update is waiting. ``.Config.Image`` is the
+        # reference the container was created from and never degrades.
+        cfg_by_cid: dict[str, tuple] = {}
+        for line in raw_cfgimages.splitlines():
+            parts = line.strip().split("|")
+            if len(parts) != 3 or not parts[0]:
+                continue
+            cid, cfg_ref, run_id = parts
+            cfg_by_cid[cid[:12]] = (cfg_ref.strip(), run_id.strip())
+
+        tag_index: dict[str, tuple] = {}
+        for im in images:
+            repo, tag = im.get("repository", ""), im.get("tag", "")
+            if not repo or not tag or "<none>" in (repo, tag):
+                continue
+            tag_index[f"{repo}:{tag}"] = (_short_image_id(im.get("id", "")), im.get("version", ""))
+
+        for c in containers:
+            cfg_ref, run_id = cfg_by_cid.get(c.get("id", "")[:12], ("", ""))
+            if not cfg_ref:
+                continue
+            c["image_ref"] = cfg_ref
+            c["running_image_id"] = _short_image_id(run_id)
+            c["restart_pending"] = False
+            c["pending_version"] = ""
+            # A stopped container is not "pending a restart" in any useful sense.
+            if c.get("state") != "running":
+                continue
+            tag_id, tag_version = tag_index.get(cfg_ref, ("", ""))
+            if tag_id and c["running_image_id"] and tag_id != c["running_image_id"]:
+                c["restart_pending"] = True
+                c["pending_version"] = tag_version
+
         # Volumes
         volumes: List[dict[str, Any]] = []
         for line in raw_volumes.splitlines():
@@ -1658,15 +1732,28 @@ class LinuxDriver(OSDriver):
         Returns a list of image references that have a newer digest available.
         """
         outdated_images: List[str] = []
-        candidate_images: List[str] = list({
+        # Prefer the reference the container was created from. `image` is whatever
-            c["image"] for c in containers
+        # `docker ps` displayed, which collapses to a bare image ID once the tag has
-            if c.get("image")
+        # moved on — and an image ID is not something a registry can resolve.
-            and "@sha256:" not in c.get("image", "")  # skip digest-pinned
+        candidate_images: List[str] = []
-        })
+        for c in containers:
+            ref = (c.get("image_ref") or c.get("image") or "").strip()
+            if not ref or "@sha256:" in ref:  # skip digest-pinned
+                continue
+            if _looks_like_image_id(ref):
+                logger.warning(
+                    "container %s reports image ID %r instead of a tag — cannot ask the "
+                    "registry about it; skipping update check",
+                    c.get("name", "?"), ref,
+                )
+                continue
+            if ref not in candidate_images:
+                candidate_images.append(ref)
         for img_name in candidate_images:
             try:
                 local_raw = self._send(
-                    f"docker inspect {img_name!r} --format '{{{{index .RepoDigests 0}}}}' 2>/dev/null",
+                    f"{self._docker_bin()} inspect {img_name!r} "
+                    f"--format '{{{{index .RepoDigests 0}}}}' 2>/dev/null",
                     read_timeout=5,
                 ).strip()
                 if not local_raw or "@" not in local_raw:
@@ -1674,7 +1761,7 @@ class LinuxDriver(OSDriver):
                 local_digest = local_raw.split("@", 1)[1]
 
                 remote_full = self._send(
-                    f"docker buildx imagetools inspect {img_name!r} 2>&1",
+                    f"{self._docker_bin()} buildx imagetools inspect {img_name!r} 2>&1",
                     read_timeout=30,
                 ).strip()
                 if ("429" in remote_full
@@ -1693,6 +1780,11 @@ class LinuxDriver(OSDriver):
                         remote_digest = _ls[7:].strip()
                         break
                 if not remote_digest or not remote_digest.startswith("sha256:"):
+                    # Silence here is indistinguishable from "up to date" — say so.
+                    logger.warning(
+                        "no digest returned for %s; skipping update check. Registry said: %s",
+                        img_name, remote_full[:200].replace("\n", " ") or "(nothing)",
+                    )
                     continue
                 if local_digest != remote_digest:
                     outdated_images.append(img_name)
@@ -1716,7 +1808,7 @@ class LinuxDriver(OSDriver):
         import shlex as _shlex
 
         raw = self._send(
-            f"docker inspect {_shlex.quote(container_id)} 2>/dev/null",
+            f"{self._docker_bin()} inspect {_shlex.quote(container_id)} 2>/dev/null",
             read_timeout=10,
         ).strip()
         if not raw:
@@ -1884,8 +1976,48 @@ class LinuxDriver(OSDriver):
             return self._action_fix_snmp()
         if action == "fix_apt_proxy":
             return self._action_fix_apt_proxy()
+        if action == "apt_update_upgrade":
+            return self._action_apt_update_upgrade()
         raise NotImplementedError(f"Unknown action: {action!r}")
 
+    def _action_apt_update_upgrade(self) -> DeviceActionResultDict:
+        """Refresh the apt cache and fully upgrade all packages (apt-based systems only).
+
+        Uses full-upgrade (not plain upgrade) — plain "apt-get upgrade" refuses
+        to install/remove packages even when required to satisfy a newer
+        version's dependencies, silently leaving those updates pending.
+        """
+        if self._pkg_manager != "apt":
+            return {
+                "success": True,
+                "output": f"Skipped — package manager is {self._pkg_manager!r}, not apt.",
+            }
+
+        sudo_check = self._send("sudo -n true 2>&1 || echo __SUDO_NEEDS_PW__")
+        if "__SUDO_NEEDS_PW__" in sudo_check or "password is required" in sudo_check.lower():
+            if not self._sudo_password:
+                return {
+                    "success": False,
+                    "output": (
+                        "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] = []
+        try:
+            out = self._sudo("apt-get update -y 2>&1", read_timeout=90)
+            lines.append(f"[update] {out.strip()[-300:]}")
+            out = self._sudo(
+                "DEBIAN_FRONTEND=noninteractive apt-get full-upgrade -y 2>&1", read_timeout=240
+            )
+            lines.append(f"[upgrade] {out.strip()[-300:]}")
+            return {"success": True, "output": "\n".join(lines)}
+        except Exception as exc:
+            lines.append(f"[error] {exc}")
+            return {"success": False, "output": "\n".join(lines)}
+
     def _action_fix_snmp(self) -> DeviceActionResultDict:
         """Install, configure and start snmpd with community 'public'."""
         lines: list[str] = []
@@ -1908,6 +2040,16 @@ class LinuxDriver(OSDriver):
         if not pkg_mgr:
             return {"success": False, "output": "Package manager not detected — cannot install snmpd."}
 
+        # Refresh the package index first — a freshly provisioned (or simply
+        # long-untouched) system's cache can be stale/empty, which makes the
+        # install below fail outright rather than just being slow.
+        if pkg_mgr == "apt":
+            try:
+                update_out = self._sudo("apt-get update -y 2>&1", read_timeout=90)
+                lines.append(f"[update] {update_out.strip()[-200:]}")
+            except Exception as exc:
+                lines.append(f"[warn] apt-get update failed: {exc}")
+
         # Install both snmpd (daemon) and snmp (client tools incl. snmpget for probing)
         install_cmd: dict[str, str] = {
             "apt":    "DEBIAN_FRONTEND=noninteractive apt-get install -y snmpd snmp 2>&1",
@@ -1918,8 +2060,12 @@ class LinuxDriver(OSDriver):
         }
         cmd = install_cmd.get(pkg_mgr)
         if cmd:
-            out = self._sudo(cmd, read_timeout=120)
+            try:
-            lines.append(f"[install] {out.strip()[-200:]}")
+                out = self._sudo(cmd, read_timeout=120)
+                lines.append(f"[install] {out.strip()[-200:]}")
+            except Exception as exc:
+                lines.append(f"[error] install failed: {exc}")
+                return {"success": False, "output": "\n".join(lines)}
 
         # 2. Determine the IP netOrk is connecting from by checking the established SSH connection
         netork_ip = ""

tests/test_linux.py

@@ -22,6 +22,11 @@ def driver():
     d._secret = "pass"  # noqa: S105
     d._forced_pkg_manager = None
     d._pkg_manager = "apt"
+    # Set by __init__, which this fixture bypasses via __new__. Without it every
+    # call through _sudo() raises AttributeError, which the callers' broad
+    # `except Exception` turns into a plain {"success": False} -- so the tests
+    # failed for a reason that had nothing to do with what they were testing.
+    d._sudo_password = None
     d.netmiko_optional_args = {}
     d._device = MagicMock()
     return d
@@ -166,7 +171,7 @@ def test_get_packages_apt(driver):
 
 
 # ---------------------------------------------------------------------------
-# get_pending_updates (apt)
+# get_available_updates (apt)
 # ---------------------------------------------------------------------------
 
 
@@ -177,10 +182,10 @@ APT_UPGRADABLE = (
 )
 
 
-def test_get_pending_updates_apt(driver):
+def test_get_available_updates_apt(driver):
     driver._pkg_manager = "apt"
-    with patch.object(driver, "_send", side_effect=["", APT_UPGRADABLE]):
+    with patch.object(driver, "_send", return_value=APT_UPGRADABLE):
-        updates = driver.get_pending_updates()
+        updates = driver.get_available_updates()
     assert len(updates) == 2
     assert updates[0]["name"] == "openssh-server"
     assert updates[0]["current_version"] == "1:9.2p1-2+deb12u1"
@@ -312,7 +317,10 @@ def test_apply_updates_apt_all_packages(driver):
     driver._pkg_manager = "apt"
     sent_commands = []
 
-    def capture_send(cmd):
+    def capture_send(cmd, **kwargs):
+        # _sudo() passes read_timeout as a keyword; without **kwargs this raises
+        # TypeError, which the caller's `except Exception` reports as a failed
+        # upgrade rather than a broken test double.
         sent_commands.append(cmd)
         return APT_UPGRADE_SUCCESS
 
@@ -661,3 +669,195 @@ def test_get_facts_fallback_vendor_when_dmi_empty(driver):
          patch.object(driver, "_send", side_effect=["host", "host.local", "Alpine Linux 3.19", "eth0"]):
         facts = driver.get_facts()
     assert facts["vendor"] == "Linux"  # fallback to VENDOR class attribute
+
+
+# ---------------------------------------------------------------------------
+# _action_fix_snmp / _action_apt_update_upgrade
+# ---------------------------------------------------------------------------
+
+
+def _fix_snmp_send_side_effect(command: str, read_timeout: float = 100) -> str:
+    """Canned responses covering every _send() call _action_fix_snmp makes."""
+    if command.startswith("sudo -n true"):
+        return ""  # passwordless sudo works
+    if command.startswith("command -v apt"):
+        return "/usr/bin/apt"
+    if command.startswith("command -v"):
+        return ""  # ufw/iptables not found, other pkg managers not found
+    if command.startswith("ss -tnp"):
+        return ""  # no netork_ip detected — skips firewall step
+    if command.startswith("cat /etc/snmp/snmpd.conf"):
+        return "agentAddress udp:161\nrocommunity public\n"
+    if command.startswith("snmpget"):
+        return "STRING: Linux test"
+    return ""
+
+
+class TestActionFixSnmp:
+    def test_runs_apt_get_update_before_install(self, driver):
+        driver._pkg_manager = "apt"
+        driver._sudo_password = None
+        sudo_calls: list[str] = []
+
+        def _sudo_side_effect(command: str, read_timeout: float = 100) -> str:
+            sudo_calls.append(command)
+            return ""
+
+        with (
+            patch.object(driver, "_send", side_effect=_fix_snmp_send_side_effect),
+            patch.object(driver, "_sudo", side_effect=_sudo_side_effect),
+        ):
+            driver._action_fix_snmp()
+
+        update_idx = next(i for i, c in enumerate(sudo_calls) if "apt-get update" in c)
+        install_idx = next(i for i, c in enumerate(sudo_calls) if "apt-get install" in c)
+        assert update_idx < install_idx
+
+    def test_install_exception_returns_failure_instead_of_raising(self, driver):
+        driver._pkg_manager = "apt"
+        driver._sudo_password = None
+
+        def _sudo_side_effect(command: str, read_timeout: float = 100) -> str:
+            if "apt-get install" in command:
+                raise TimeoutError("connection timed out")
+            return ""
+
+        with (
+            patch.object(driver, "_send", side_effect=_fix_snmp_send_side_effect),
+            patch.object(driver, "_sudo", side_effect=_sudo_side_effect),
+        ):
+            result = driver._action_fix_snmp()
+
+        assert result["success"] is False
+        assert "install failed" in result["output"]
+
+    def test_apt_get_update_failure_is_non_fatal(self, driver):
+        """apt-get update failing (e.g. transient network issue) must not
+        abort the whole action — install is still attempted."""
+        driver._pkg_manager = "apt"
+        driver._sudo_password = None
+
+        def _sudo_side_effect(command: str, read_timeout: float = 100) -> str:
+            if "apt-get update" in command:
+                raise TimeoutError("network unreachable")
+            return ""
+
+        with (
+            patch.object(driver, "_send", side_effect=_fix_snmp_send_side_effect),
+            patch.object(driver, "_sudo", side_effect=_sudo_side_effect),
+        ):
+            result = driver._action_fix_snmp()
+
+        assert "apt-get update failed" in result["output"]
+
+
+class TestActionAptUpdateUpgrade:
+    def test_skips_when_not_apt(self, driver):
+        driver._pkg_manager = "dnf"
+        result = driver._action_apt_update_upgrade()
+        assert result["success"] is True
+        assert "Skipped" in result["output"]
+
+    def test_fails_when_sudo_needs_password_and_none_configured(self, driver):
+        driver._pkg_manager = "apt"
+        driver._sudo_password = None
+        with patch.object(driver, "_send", return_value="__SUDO_NEEDS_PW__"):
+            result = driver._action_apt_update_upgrade()
+        assert result["success"] is False
+        assert "sudo requires a password" in result["output"]
+
+    def test_runs_update_then_upgrade_successfully(self, driver):
+        driver._pkg_manager = "apt"
+        driver._sudo_password = "secret"  # noqa: S105
+        with (
+            patch.object(driver, "_send", return_value=""),
+            patch.object(
+                driver,
+                "_sudo",
+                side_effect=["Reading package lists... Done", "0 upgraded, 0 newly installed"],
+            ) as mock_sudo,
+        ):
+            result = driver._action_apt_update_upgrade()
+
+        assert result["success"] is True
+        assert "[update]" in result["output"]
+        assert "[upgrade]" in result["output"]
+        update_call, upgrade_call = mock_sudo.call_args_list
+        assert "apt-get update" in update_call.args[0]
+        # full-upgrade (not plain upgrade) — plain upgrade silently holds back
+        # packages whose newer version needs to install/remove dependencies.
+        assert "apt-get full-upgrade" in upgrade_call.args[0]
+
+    def test_exception_during_upgrade_returns_failure(self, driver):
+        driver._pkg_manager = "apt"
+        driver._sudo_password = "secret"  # noqa: S105
+        with (
+            patch.object(driver, "_send", return_value=""),
+            patch.object(
+                driver, "_sudo", side_effect=["update ok", TimeoutError("connection lost")]
+            ),
+        ):
+            result = driver._action_apt_update_upgrade()
+
+        assert result["success"] is False
+        assert "[error]" in result["output"]
+
+
+class TestRunDeviceActionDispatch:
+    def test_apt_update_upgrade_action_dispatches(self, driver):
+        with patch.object(
+            driver, "_action_apt_update_upgrade", return_value={"success": True, "output": ""}
+        ) as mock_action:
+            driver.run_device_action("apt_update_upgrade")
+        mock_action.assert_called_once()
+
+
+class TestDockerBinHook:
+    """Where the docker binary lives is device-specific; what to do with it is not.
+
+    QTS puts Container Station's docker under /share/<pool>/.qpkg/ and not on
+    PATH. Rather than duplicating the whole Docker surface in the QNAP driver,
+    the path is a one-method hook here and the logic stays generic. See
+    docs/ARCHITECTURE.md §4.4, "Generisch vs. treiberspezifisch".
+    """
+
+    def test_defaults_to_docker_on_path(self, driver):
+        assert driver._docker_bin() == "docker"
+
+    def test_detection_uses_the_hook(self, driver):
+        """A subclass pointing elsewhere must not be probed for a PATH docker."""
+        sent = []
+
+        def _record(cmd, **kwargs):
+            sent.append(cmd)
+            return ""
+
+        with patch.object(driver, "_docker_bin", return_value="/opt/cs/docker"):
+            with patch.object(driver, "_send", side_effect=_record):
+                result = driver.get_docker_info()
+
+        assert result == {"available": False}
+        assert any("/opt/cs/docker" in cmd for cmd in sent)
+        assert not any("command -v docker " in cmd for cmd in sent)
+
+    def test_all_docker_subcommands_use_the_hook(self, driver):
+        """Half-converted call sites are the failure mode here: detection would
+        find the binary and the actual queries would still miss it."""
+        sent = []
+
+        def _record(cmd, **kwargs):
+            sent.append(cmd)
+            if "command -v" in cmd:
+                return "/opt/cs/docker"
+            if "---CONTAINERS---" in cmd:
+                return "---CONTAINERS---\n---IMAGES---\n---VOLUMES---\n---NETWORKS---\n"
+            return ""
+
+        with patch.object(driver, "_docker_bin", return_value="/opt/cs/docker"):
+            with patch.object(driver, "_send", side_effect=_record):
+                driver.get_docker_info()
+
+        docker_cmds = [c for c in sent if "docker" in c]
+        assert docker_cmds
+        for cmd in docker_cmds:
+            assert "/opt/cs/docker" in cmd, f"unconverted call site: {cmd}"