NAPALM/napalm-linux
1
0
Fork 0
Code Issues 9 Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request 'refactor!: drop the Docker CLI methods' (#20) from chore/drop-legacy-docker into master
CI / test (3.10) (push) Successful in 37s
Details
CI / test (3.11) (push) Successful in 36s
Details
CI / test (3.12) (push) Successful in 38s
Details

Browse Source
This commit was merged in pull request #20.
This commit is contained in:
christianmanivong
2026-10-08 06:43:46 +00:00
parent fe29b507b0 3333d8e8f5
commit 3a76b6d47f
4 changed files with 22 additions and 485 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CHANGELOG.md
+11
View File
@@ -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
-432
View File
@@ -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
+1 -1
View File
@@ -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
+10 -52
View File
@@ -866,60 +866,18 @@ class TestRunDeviceActionDispatch:
mock_action.assert_called_once()
class TestDockerBinHook:
"""Where the docker binary lives is device-specific; what to do with it is not.
class TestNoDockerOfItsOwn:
"""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
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_the_cli_methods_are_gone(self):
for name in ("get_docker_info", "get_docker_outdated", "reconstruct_docker_run", "_docker_bin"):
assert not hasattr(LinuxDriver, name), name
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}"
# ---------------------------------------------------------------------------
# uninstall_package – purge, and getting out of `install ok unpacked`
# ---------------------------------------------------------------------------
def test_the_engine_access_and_the_permission_fix_stay(self):
assert hasattr(LinuxDriver, "open_container_engine")
assert hasattr(LinuxDriver, "_action_fix_docker_permissions")
class TestUninstallPackage: