refactor!: drop the Docker CLI methods

get_docker_info(), get_docker_outdated(), reconstruct_docker_run() and
the _docker_bin() hook go, with the image-ID helpers. netOrk reads and
handles containers over the Engine API through open_container_engine()
since NetOrk/netork#765 and no longer calls them (phase 9). The
fix_docker_permissions action stays: it is an OS action.

Version 0.3.0.

CHANGELOG.md

@@ -7,6 +7,17 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+## [0.3.0] – 2026-10-08
+
+### Removed
+
+- `get_docker_info()`, `get_docker_outdated()`, `reconstruct_docker_run()` and
+  the `_docker_bin()` hook, with the image-ID helpers they used. netOrk reads
+  and handles containers itself over the Engine API, through
+  `open_container_engine()` (NetOrk/netork#765), and no longer calls them.
+  `run_device_action("fix_docker_permissions")` stays: letting the login user
+  use Docker is an OS action.
+
 ## [0.2.0] – 2026-10-07
 
 ### Added

napalm_linux/linux.py

@@ -54,7 +54,6 @@ from napalm_device_types.models import (
     ApplyUpdatesResultDict,
     CronJobDict,
     DeviceActionResultDict,
-    DockerInfoDict,
     PackageDict,
     ProcessDict,
     ServiceDict,
@@ -204,22 +203,6 @@ 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(
     KernelFactsMixin,
     ListeningSocketsMixin,
@@ -1779,421 +1762,6 @@ class LinuxDriver(
             job["description"] = comment
         return job
 
-    # ------------------------------------------------------------------
-    # 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.
-
-        Uses a single SSH call to collect all Docker data at once, eliminating
-        per-section round-trip overhead.  Labels from ``docker images`` are used
-        directly for the OCI version field — no separate ``docker image inspect``
-        needed.
-
-        Returns a dict with keys:
-          - ``available`` (bool)  — False if docker is not installed/accessible
-          - ``version``   (str)   — Docker Engine version string
-          - ``containers`` (list) — list of container dicts
-          - ``images``    (list)  — list of image dicts
-          - ``volumes``   (list)  — list of volume dicts
-          - ``networks``  (list)  — list of network dicts
-        """
-        import json as _json
-
-        docker = self._docker_bin()
-
-        # Check docker binary first (docker --version doesn't need socket access)
-        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(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(f"{docker} --version 2>/dev/null").strip()
-
-        combined = self._send(
-            "echo '---CONTAINERS---'; "
-            f"{docker} ps -a --format '{{{{json .}}}}' 2>/dev/null; "
-            "echo '---IMAGES---'; "
-            f"{docker} images --format '{{{{json .}}}}' 2>/dev/null; "
-            "echo '---VOLUMES---'; "
-            f"{docker} volume ls --format '{{{{json .}}}}' 2>/dev/null; "
-            "echo '---NETWORKS---'; "
-            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,
-        )
-
-        if "---CONTAINERS---" not in combined:
-            return {"available": False}
-
-        # Split into sections
-        def _section(text: str, marker: str, next_marker: str) -> str:
-            start = text.find(marker)
-            if start == -1:
-                return ""
-            start += len(marker)
-            end = text.find(next_marker, start)
-            return text[start:end] if end != -1 else text[start:]
-
-        raw_containers = _section(combined, "---CONTAINERS---", "---IMAGES---")
-        raw_images     = _section(combined, "---IMAGES---",     "---VOLUMES---")
-        raw_volumes    = _section(combined, "---VOLUMES---",    "---NETWORKS---")
-        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."""
-            if isinstance(raw, dict):
-                return {str(k): str(v) for k, v in raw.items()}
-            if isinstance(raw, str) and raw:
-                result: Dict[str, str] = {}
-                for part in raw.split(","):
-                    if "=" in part:
-                        k, _, v = part.partition("=")
-                        result[k.strip()] = v.strip()
-                return result
-            return {}
-
-        # Containers
-        containers: List[dict[str, Any]] = []
-        for line in raw_containers.splitlines():
-            line = line.strip()
-            if not line:
-                continue
-            try:
-                obj = _json.loads(line)
-                labels = _parse_labels(obj.get("Labels", ""))
-                containers.append({
-                    "id": obj.get("ID", ""),
-                    "name": obj.get("Names", ""),
-                    "image": obj.get("Image", ""),
-                    "image_version": labels.get("org.opencontainers.image.version", ""),
-                    "command": obj.get("Command", ""),
-                    "created": obj.get("CreatedAt", ""),
-                    "status": obj.get("Status", ""),
-                    "ports": obj.get("Ports", ""),
-                    "state": obj.get("State", ""),
-                    "compose_project": labels.get("com.docker.compose.project", ""),
-                    "compose_service": labels.get("com.docker.compose.service", ""),
-                    "compose_file": labels.get("com.docker.compose.project.config_files", ""),
-                })
-            except Exception:
-                pass
-
-        # Images — OCI version comes from Labels, no separate inspect needed
-        images: List[dict[str, Any]] = []
-        for line in raw_images.splitlines():
-            line = line.strip()
-            if not line:
-                continue
-            try:
-                obj = _json.loads(line)
-                labels = _parse_labels(obj.get("Labels", ""))
-                images.append({
-                    "id": obj.get("ID", ""),
-                    "repository": obj.get("Repository", ""),
-                    "tag": obj.get("Tag", ""),
-                    "size": obj.get("Size", ""),
-                    "created": obj.get("CreatedAt", ""),
-                    "version": labels.get("org.opencontainers.image.version", ""),
-                })
-            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():
-            line = line.strip()
-            if not line:
-                continue
-            try:
-                obj = _json.loads(line)
-                volumes.append({
-                    "name": obj.get("Name", ""),
-                    "driver": obj.get("Driver", ""),
-                    "mountpoint": obj.get("Mountpoint", ""),
-                    "scope": obj.get("Scope", ""),
-                })
-            except Exception:
-                pass
-
-        # Networks
-        networks: List[dict[str, Any]] = []
-        for line in raw_networks.splitlines():
-            line = line.strip()
-            if not line:
-                continue
-            try:
-                obj = _json.loads(line)
-                networks.append({
-                    "id": obj.get("ID", ""),
-                    "name": obj.get("Name", ""),
-                    "driver": obj.get("Driver", ""),
-                    "scope": obj.get("Scope", ""),
-                    "ipv6": obj.get("IPv6", ""),
-                    "internal": obj.get("Internal", ""),
-                })
-            except Exception:
-                pass
-
-        return {
-            "available": True,
-            "version": version,
-            "containers": containers,
-            "images": images,
-            "volumes": volumes,
-            "networks": networks,
-            "outdated_images": [],  # populated by separate check_docker_outdated task
-        }
-
-    def get_docker_outdated(self, containers: List[Dict]) -> List[str]:
-        """Check registry for available updates for all container images.
-
-        Runs ``docker buildx imagetools inspect`` (metadata-only, no download)
-        for each unique image referenced by a container.  Intended to be called
-        from a separate Celery task on a long interval (e.g. every 3 hours) so
-        it never blocks the main device poll.
-
-        Returns a list of image references that have a newer digest available.
-        """
-        outdated_images: List[str] = []
-        # Prefer the reference the container was created from. `image` is whatever
-        # `docker ps` displayed, which collapses to a bare image ID once the tag has
-        # moved on — and an image ID is not something a registry can resolve.
-        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"{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:
-                    continue  # locally built or not yet pulled
-                local_digest = local_raw.split("@", 1)[1]
-
-                remote_full = self._send(
-                    f"{self._docker_bin()} buildx imagetools inspect {img_name!r} 2>&1",
-                    read_timeout=30,
-                ).strip()
-                if ("429" in remote_full
-                        or "Too Many Requests" in remote_full
-                        or "toomanyrequests" in remote_full):
-                    logger.warning(
-                        "Docker Hub rate limit hit for %s — run "
-                        "'docker login' on the device to avoid this",
-                        img_name,
-                    )
-                    continue
-                remote_digest = ""
-                for _line in remote_full.splitlines():
-                    _ls = _line.strip()
-                    if _ls.startswith("Digest:"):
-                        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)
-            except Exception as exc:
-                logger.warning("image update check for %s: %s", img_name, exc)
-        return outdated_images
-
-    def reconstruct_docker_run(self, container_id: str) -> dict | None:
-        """Return the information needed to recreate a standalone container.
-
-        Parses ``docker inspect`` JSON and returns a dict with:
-        - ``name``      — container name (without leading slash)
-        - ``image``     — current image reference
-        - ``run_args``  — list of CLI args for ``docker run`` (without image/cmd)
-        - ``cmd``       — command override (may be empty list)
-        - ``entrypoint`` — entrypoint override (may be empty list)
-
-        Returns None if the container does not exist or inspect fails.
-        """
-        import json as _json
-        import shlex as _shlex
-
-        raw = self._send(
-            f"{self._docker_bin()} inspect {_shlex.quote(container_id)} 2>/dev/null",
-            read_timeout=10,
-        ).strip()
-        if not raw:
-            return None
-        try:
-            data = _json.loads(raw)
-        except Exception:
-            return None
-        if not data:
-            return None
-        c = data[0]
-
-        name = c.get("Name", "").lstrip("/")
-        cfg = c.get("Config", {})
-        hcfg = c.get("HostConfig", {})
-        net_settings = c.get("NetworkSettings", {})
-
-        args: List[str] = ["--name", name]
-
-        # Restart policy
-        rp = hcfg.get("RestartPolicy", {})
-        rp_name = rp.get("Name", "no")
-        if rp_name and rp_name != "no":
-            max_retry = rp.get("MaximumRetryCount", 0)
-            if rp_name == "on-failure" and max_retry:
-                args += ["--restart", f"on-failure:{max_retry}"]
-            else:
-                args += ["--restart", rp_name]
-
-        # Hostname
-        hostname = cfg.get("Hostname", "")
-        if hostname and hostname != name[:12]:
-            args += ["--hostname", hostname]
-
-        # Environment (skip vars that look like Docker-injected metadata)
-        _skip_prefixes = ("PATH=", "HOME=", "TERM=", "HOSTNAME=")
-        for env in cfg.get("Env") or []:
-            if not any(env.startswith(p) for p in _skip_prefixes):
-                args += ["-e", env]
-
-        # Volume binds
-        for bind in hcfg.get("Binds") or []:
-            args += ["-v", bind]
-
-        # Port bindings
-        for container_port, host_bindings in (hcfg.get("PortBindings") or {}).items():
-            for hb in (host_bindings or []):
-                host_ip = hb.get("HostIp", "")
-                host_port = hb.get("HostPort", "")
-                if host_ip:
-                    args += ["-p", f"{host_ip}:{host_port}:{container_port}"]
-                else:
-                    args += ["-p", f"{host_port}:{container_port}"]
-
-        # Network mode
-        net_mode = hcfg.get("NetworkMode", "default")
-        if net_mode not in ("default", "bridge"):
-            args += ["--network", net_mode]
-        else:
-            # Check for custom networks from NetworkSettings
-            for net_name in (net_settings.get("Networks") or {}):
-                if net_name not in ("bridge", "host", "none"):
-                    args += ["--network", net_name]
-                    break
-
-        # Privileged
-        if hcfg.get("Privileged"):
-            args.append("--privileged")
-
-        # Cap-add
-        for cap in hcfg.get("CapAdd") or []:
-            args += ["--cap-add", cap]
-
-        # Devices
-        for dev in hcfg.get("Devices") or []:
-            host_p = dev.get("PathOnHost", "")
-            ctr_p = dev.get("PathInContainer", "")
-            perms = dev.get("CgroupPermissions", "rwm")
-            if host_p:
-                args += ["--device", f"{host_p}:{ctr_p}:{perms}"]
-
-        # Extra hosts
-        for eh in hcfg.get("ExtraHosts") or []:
-            args += ["--add-host", eh]
-
-        # DNS
-        for dns in hcfg.get("Dns") or []:
-            args += ["--dns", dns]
-
-        # Labels (skip Docker-internal labels)
-        _skip_label_prefixes = ("com.docker.compose.", "org.opencontainers.")
-        for k, v in (cfg.get("Labels") or {}).items():
-            if not any(k.startswith(p) for p in _skip_label_prefixes):
-                args += ["--label", f"{k}={v}"]
-
-        # Detach always
-        args.append("-d")
-
-        return {
-            "name": name,
-            "image": cfg.get("Image", ""),
-            "run_args": args,
-            "cmd": cfg.get("Cmd") or [],
-            "entrypoint": cfg.get("Entrypoint") or [],
-        }
-
     # ── Device actions ────────────────────────────────────────────────────────
 
     def get_snmp_config(self) -> Optional[SNMPConfigDict]:

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "napalm-linux"
-version = "0.2.0"
+version = "0.3.0"
 description = "NAPALM driver for generic Linux systems via SSH"
 readme = "README.md"
 requires-python = ">=3.9"

tests/test_linux.py

@@ -866,60 +866,18 @@ class TestRunDeviceActionDispatch:
         mock_action.assert_called_once()
 
 
-class TestDockerBinHook:
+class TestNoDockerOfItsOwn:
-    """Where the docker binary lives is device-specific; what to do with it is not.
+    """Containers are read and handled by netOrk over the Engine API, through
+    ``open_container_engine`` (NetOrk/netork#765). The old CLI methods are gone;
+    only the OS action that lets the login user use Docker stays."""
 
-    QTS puts Container Station's docker under /share/<pool>/.qpkg/ and not on
+    def test_the_cli_methods_are_gone(self):
-    PATH. Rather than duplicating the whole Docker surface in the QNAP driver,
+        for name in ("get_docker_info", "get_docker_outdated", "reconstruct_docker_run", "_docker_bin"):
-    the path is a one-method hook here and the logic stays generic. See
+            assert not hasattr(LinuxDriver, name), name
-    docs/ARCHITECTURE.md §4.4, "Generisch vs. treiberspezifisch".
-    """
 
-    def test_defaults_to_docker_on_path(self, driver):
+    def test_the_engine_access_and_the_permission_fix_stay(self):
-        assert driver._docker_bin() == "docker"
+        assert hasattr(LinuxDriver, "open_container_engine")
-
+        assert hasattr(LinuxDriver, "_action_fix_docker_permissions")
-    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}"
-
-
-# ---------------------------------------------------------------------------
-# uninstall_package – purge, and getting out of `install ok unpacked`
-# ---------------------------------------------------------------------------
 
 
 class TestUninstallPackage: