From 3e26df259fc1e6259adcffcdfe134f15204a25d3 Mon Sep 17 00:00:00 2001 From: Vladimir Eremin Date: Sun, 11 Oct 2026 02:18:46 +0300 Subject: [PATCH] fix: check running images with Docker Buildx Compare matching runtime and registry digest types, and distinguish declared Compose builds from registry images. Report unavailable checks explicitly. Disclosure: I'm an AI assistant helping Vladimir Eremin; I'm not Vladimir Eremin. --- actions/sync/packages.py | 244 +++++++++++++++++++++++---- tests/test_sync_packages.py | 318 ++++++++++++++++++++++++++++++++++++ 2 files changed, 529 insertions(+), 33 deletions(-) create mode 100644 tests/test_sync_packages.py diff --git a/actions/sync/packages.py b/actions/sync/packages.py index fa6fe6c..0731e5e 100644 --- a/actions/sync/packages.py +++ b/actions/sync/packages.py @@ -1,12 +1,216 @@ # actions/sync/packages.py -"""Port of sync/packages — apt upgradable + docker image (skopeo) update check; Telegram.""" -import os, re, shutil +"""Check pending APT and running-image updates without pulling images.""" +import json +import os +import re +import shutil from engine.descriptor import Meta, Arg METADATA = Meta(summary="Check pending apt updates + docker image updates; Telegram notify.", args=[Arg("telegram_bot_url", "Telegram Bot URL for notifications")]) TITLE = "Node Security Updates" +MANIFEST_TYPES = {"application/vnd.oci.image.manifest.v1+json", + "application/vnd.docker.distribution.manifest.v2+json"} +INDEX_TYPES = {"application/vnd.oci.image.index.v1+json", + "application/vnd.docker.distribution.manifest.list.v2+json"} + + +def _output(s, command): + result = s.run(command, check=False, capture=True) + if result.returncode: + # Registry errors can contain authentication details; don't relay stderr. + raise ValueError("inspection command failed") + return result.stdout or "" + + +def _digest(value): + if not isinstance(value, str) or not re.fullmatch(r"sha256:[0-9a-f]{64}", value): + raise ValueError("missing or unsupported image digest") + return value + + +def _platform(os_name, architecture, variant=""): + if not isinstance(os_name, str) or not os_name or not isinstance(architecture, str) or not architecture: + raise ValueError("missing image platform") + variant = variant or "" + if architecture == "arm64" and variant == "v8": + variant = "" + return os_name, architecture, variant + + +def _repository(reference): + name = reference.split("@", 1)[0] + # Only a colon in the final path component is a tag separator. + return name.rsplit(":", 1)[0] if ":" in name.rsplit("/", 1)[-1] else name + + +def _remote_digests(s, reference, local, cache, running_manifest=None): + def inspect(ref, raw=False, field="Manifest"): + command = ["docker", "buildx", "imagetools", "inspect"] + command += ["--raw"] if raw else ["--format", "{{json ." + field + "}}"] + key = tuple(command + [ref]) + if key not in cache: + value = json.loads(_output(s, ["timeout", "--signal=TERM", "--kill-after=5s", + "30s", *key])) + if not isinstance(value, dict): + raise ValueError("invalid registry JSON object") + cache[key] = value + return cache[key] + + platform = _platform(local.get("Os"), local.get("Architecture"), local.get("Variant")) + descriptor = inspect(reference) + media_type = descriptor.get("mediaType") + selected = _digest(descriptor.get("digest")) + if media_type in INDEX_TYPES: + manifests = descriptor.get("manifests") + if not isinstance(manifests, list): + raise ValueError("invalid image index") + matches = [] + for item in manifests: + p = item.get("platform", {}) + if _platform(p.get("os"), p.get("architecture"), p.get("variant")) == platform: + if item.get("mediaType") not in MANIFEST_TYPES: + raise ValueError("unsupported nested image index") + matches.append(_digest(item.get("digest"))) + if len(matches) != 1: + raise ValueError("missing or ambiguous matching image platform") + selected = matches[0] + elif media_type not in MANIFEST_TYPES: + raise ValueError("unsupported registry manifest") + + immutable = _repository(reference) + "@" + selected + manifest = inspect(immutable, raw=True) + if manifest.get("schemaVersion") != 2 or manifest.get("mediaType") not in MANIFEST_TYPES: + raise ValueError("invalid platform manifest") + remote_id = _digest(manifest.get("config", {}).get("digest")) + if media_type in MANIFEST_TYPES and selected != running_manifest and remote_id != local["Id"]: + config = inspect(immutable, field="Image") + if _platform(config.get("os"), config.get("architecture"), config.get("variant")) != platform: + raise ValueError("registry image platform changed") + return selected, remote_id + + +def _compose_builds(ctx): + builds = {} + env = dict(os.environ) + env.update({k: v for k, v in ctx.vars().items() if k.startswith("APEX_")}) + for name in sorted(os.listdir(ctx.paths.compositions)): + directory = os.path.join(ctx.paths.compositions, name) + if name.startswith((".", "@")) or not os.path.isfile(os.path.join(directory, "docker-compose.yml")): + continue + command = ["docker", "compose"] + for filename in (".env", "apex.env"): + if os.path.isfile(os.path.join(directory, filename)): + command += ["--env-file", filename] + # Resolved Compose includes secrets: capture privately and never log it. + result = ctx.sys.run(["timeout", "--signal=TERM", "--kill-after=5s", "30s", + *command, "config", "--format", "json"], + cwd=directory, env=env, check=False, capture=True) + if result.returncode: + raise ValueError("Compose inspection failed") + config = json.loads(result.stdout) + project, services = config["name"], config["services"] + if not isinstance(project, str) or not project or not isinstance(services, dict): + raise ValueError("invalid Compose model") + for service, settings in services.items(): + if settings.get("build") is not None: + reference = settings.get("image") + if reference is not None and not isinstance(reference, str): + raise ValueError("invalid Compose image") + builds[(project, service)] = reference + return builds + + +def _images(s, log, ctx=None): + if not shutil.which("docker"): + return "Images: unchecked (Docker unavailable)" + if not shutil.which("timeout"): + return "Images: unchecked (coreutils timeout unavailable)" + try: + ids = _output(s, ["docker", "ps", "--quiet", "--no-trunc"]).split() + if not ids: + return "Images: no running containers" + output = _output(s, ["docker", "container", "inspect", "--format", + '{"Image":{{json .Image}},"Reference":{{json .Config.Image}},' + '"ImageManifestDescriptor":{{json (index . "ImageManifestDescriptor")}},' + '"Labels":{{json .Config.Labels}}}', *ids]) + containers = [json.loads(line) for line in output.splitlines()] + if len(containers) != len(ids): + raise ValueError("incomplete running-container inspection") + if any(not isinstance(c["Reference"], str) for c in containers): + raise ValueError("invalid running-image reference") + except (ValueError, KeyError, TypeError, OSError) as error: + log.warn(f"Images unchecked: {type(error).__name__}") + return "Images: unchecked (container inspection failed)" + + try: + builds = _compose_builds(ctx) if ctx is not None else {} + except (ValueError, KeyError, TypeError, AttributeError, OSError): + log.warn("Images unchecked: Compose build provenance could not be resolved") + return "Images: unchecked (Compose inspection failed)" + + buildx = s.run(["docker", "buildx", "version"], check=False, capture=True).returncode == 0 + counts = dict.fromkeys(("updates", "current", "pinned", "local/untracked", "unchecked"), 0) + cache, images, seen = {}, {}, set() + for container in containers: + reference = container["Reference"] + reason = "" + try: + image_id = _digest(container["Image"]) + descriptor = container.get("ImageManifestDescriptor") + labels = container.get("Labels") or {} + build_key = (labels.get("com.docker.compose.project"), labels.get("com.docker.compose.service")) + declared_build = build_key in builds and builds[build_key] in (None, reference) + key = (reference, image_id, json.dumps(descriptor, sort_keys=True), declared_build) + if key in seen: + continue + seen.add(key) + if not isinstance(reference, str) or not reference or reference.startswith("-"): + raise ValueError("invalid image reference") + if "@" in reference or re.fullmatch(r"(?:sha256:)?[0-9a-f]{64}", reference): + status = "pinned" + elif declared_build: + status = "local/untracked" + else: + if image_id not in images: + data = json.loads(_output(s, ["docker", "image", "inspect", image_id])) + if not isinstance(data, list) or len(data) != 1 or data[0].get("Id") != image_id: + raise ValueError("invalid running-image inspection") + images[image_id] = data[0] + local = images[image_id] + repo_digests = local["RepoDigests"] + if repo_digests is None or repo_digests == []: + status = "local/untracked" + elif not isinstance(repo_digests, list): + raise ValueError("invalid repository digests") + elif not buildx: + raise ValueError("Buildx unavailable") + else: + running_manifest = None + if descriptor is not None: + if descriptor.get("mediaType") not in MANIFEST_TYPES: + raise ValueError("invalid running platform manifest") + running_manifest = _digest(descriptor.get("digest")) + platform = descriptor.get("platform", {}) + os_name, architecture, variant = _platform( + platform.get("os"), platform.get("architecture"), platform.get("variant")) + local = dict(local, Os=os_name, Architecture=architecture, Variant=variant) + elif local.get("Descriptor") is not None: + raise ValueError("containerd running platform manifest unavailable") + remote_manifest, remote_config = _remote_digests( + s, reference, local, cache, running_manifest) + # Classic Docker IDs are configs; containerd IDs may be indices. + current = (remote_manifest == running_manifest if running_manifest + else remote_config == image_id) + status = "current" if current else "updates" + except (ValueError, KeyError, TypeError, AttributeError, OSError) as error: + status, reason = "unchecked", str(error) if isinstance(error, ValueError) else type(error).__name__ + counts[status] += 1 + log.info(f"Image {reference}: {status}" + (f" ({reason})" if reason else "")) + return "Images: " + ", ".join(f"{count} {status}" for status, count in counts.items()) + + def run(ctx, args): log, node, url, s = ctx.log, ctx.node.name, args.telegram_bot_url, ctx.sys try: @@ -31,41 +235,15 @@ def _check(ctx, args, log, node, url, s): check=False, capture=True) upd = [l for l in (cp.stdout or "").splitlines() if l.strip()] packages = (f"*Packages ({len(upd)}):* " + ", ".join(f"`{u}`" for u in upd)) if upd else "*Packages:* _none_" + apt_summary = f"APT: {len(upd)} pending" else: packages = "*Packages:* _Failed to check_" + apt_summary = "APT: failed to check" - if not shutil.which("docker"): - images = "*Images:* _Docker not available_" - elif not shutil.which("skopeo"): - images = "*Images:* _skopeo not available_" - else: - cp = s.run(["bash", "-c", "docker ps --format '{{.Image}}' | sort -u"], check=False, capture=True) - outdated = [] - for image in [l for l in (cp.stdout or "").splitlines() if l.strip()]: - name, sep, tag = image.partition(":") - tag = tag or "latest" - ld = s.run(["bash", "-c", - f"docker image inspect '{image}' --format '{{{{index .RepoDigests 0}}}}' 2>/dev/null | cut -d'@' -f2"], - check=False, capture=True).stdout.strip() - if not ld: - continue - if "/" not in name: - reg = f"docker.io/library/{name}" - elif not re.match(r".*\..*/", name): - reg = f"docker.io/{name}" - else: - reg = name - rd = s.run(["bash", "-c", - f"skopeo inspect 'docker://{reg}:{tag}' 2>/dev/null | " - f"grep -o '\"Digest\"[[:space:]]*:[[:space:]]*\"[^\"]*\"' | cut -d'\"' -f4"], - check=False, capture=True).stdout.strip() - if rd and ld != rd: - disp = name[len("docker.io/"):] if name.startswith("docker.io/") else name - disp = disp[len("library/"):] if disp.startswith("library/") else disp - outdated.append(disp) - images = (f"*Images ({len(outdated)}):* " + ", ".join(f"`{i}`" for i in outdated)) if outdated else "*Images:* _none_" + images = _images(s, log, ctx) + log.info(packages) # A failed send exits 1 (bash: telegram_info's return 1 tripped set -e). - if not ctx.notify.info(TITLE, url, f"*APT:*\n{packages}\n\n*Skopeo:*\n{images}"): + if not ctx.notify.info(TITLE, url, f"{apt_summary}\n{images}\n\n{packages}"): raise SystemExit(1) log.success("Security update check completed.") diff --git a/tests/test_sync_packages.py b/tests/test_sync_packages.py new file mode 100644 index 0000000..81f4b66 --- /dev/null +++ b/tests/test_sync_packages.py @@ -0,0 +1,318 @@ +import json +import subprocess +import tempfile +from pathlib import Path +import unittest +from types import SimpleNamespace +from unittest.mock import Mock, patch + +from actions.sync import packages + + +def digest(char): + return 'sha256:' + char * 64 + + +MANIFEST = 'application/vnd.oci.image.manifest.v1+json' +INDEX = 'application/vnd.oci.image.index.v1+json' + + +class Commands: + def __init__(self): + self.containers = [{'Image': digest('a'), 'Reference': 'alpine:3.21'}] + self.local = {digest('a'): {'Id': digest('a'), 'Os': 'linux', + 'Architecture': 'amd64', 'RepoDigests': ['alpine@' + digest('b')]}} + self.remote = {'alpine:3.21': {'mediaType': MANIFEST, + 'digest': digest('c')}} + self.raw = {'alpine@' + digest('c'): {'schemaVersion': 2, 'mediaType': MANIFEST, + 'config': {'digest': digest('a')}, 'layers': []}} + self.config = {'alpine@' + digest('c'): {'os': 'linux', 'architecture': 'amd64'}} + self.calls = [] + self.fail = set() + self.apt = '' + self.compose = {"name": "apex", "services": {}} + self.compose_options = [] + + def ok(self, cmd): + self.calls.append(cmd) + return tuple(cmd) not in self.fail + + def run(self, cmd, check=False, capture=False, **kwargs): + self.calls.append(cmd) + if cmd[:1] == ['timeout']: + cmd = cmd[4:] + if tuple(cmd) in self.fail: + return subprocess.CompletedProcess(cmd, 1, '', 'registry unavailable') + if cmd[:2] == ['docker', 'compose']: + self.compose_options.append(kwargs) + data = self.compose if isinstance(self.compose, str) else json.dumps(self.compose) + elif cmd[:2] == ['bash', '-c']: + if 'apt-get --just-print' not in cmd[2]: + raise AssertionError('Image references must never reach a shell') + data = self.apt + elif cmd == ['docker', 'ps', '--quiet', '--no-trunc']: + data = '\n'.join(str(i) for i in range(len(self.containers))) + elif cmd[:3] == ['docker', 'container', 'inspect']: + data = '\n'.join(json.dumps(c) for c in self.containers) + elif cmd[:3] == ['docker', 'image', 'inspect']: + data = json.dumps([self.local[cmd[3]]]) + elif cmd == ['docker', 'buildx', 'version']: + data = 'github.com/docker/buildx v0.37.1' + elif cmd[:4] == ['docker', 'buildx', 'imagetools', 'inspect']: + table = self.raw if cmd[4] == '--raw' else ( + self.remote if cmd[5] == '{{json .Manifest}}' else self.config) + value = table[cmd[-1]] + data = value if isinstance(value, str) else json.dumps(value) + else: + raise AssertionError(f'Unexpected command: {cmd}') + return subprocess.CompletedProcess(cmd, 0, data, '') + + +class TestSyncPackages(unittest.TestCase): + def setUp(self): + self.commands = Commands() + self.ctx = SimpleNamespace(sys=self.commands, log=Mock(), node=SimpleNamespace(name='node'), + notify=Mock()) + self.ctx.notify.info.return_value = True + self.tmp = tempfile.TemporaryDirectory() + self.addCleanup(self.tmp.cleanup) + self.ctx.paths = SimpleNamespace(compositions=self.tmp.name) + self.ctx.vars = lambda: {'APEX_NODE_HOST': 'test-node'} + self.args = SimpleNamespace(telegram_bot_url='test-notifier') + + def report(self): + with patch.object(packages.os.path, 'isfile', return_value=True), \ + patch.object(packages.shutil, 'which', return_value='/usr/bin/docker'): + packages.run(self.ctx, self.args) + return self.ctx.notify.info.call_args.args[2] + + def containerd(self, manifest='c'): + local = self.commands.local.pop(digest('a')) + local.update(Id=digest('b'), Descriptor={'mediaType': INDEX, 'digest': digest('b')}) + self.commands.local[digest('b')] = local + self.commands.containers[0].update(Image=digest('b'), ImageManifestDescriptor={ + 'mediaType': MANIFEST, 'digest': digest(manifest), + 'platform': {'os': 'linux', 'architecture': 'amd64'}}) + + def test_containerd_index_id_compares_actual_platform_manifest(self): + self.containerd() + self.index() + self.assertIn('1 current', self.report()) + + def test_containerd_single_manifest_is_current_despite_different_config_digest(self): + self.containerd() + self.assertIn('1 current', self.report()) + + def test_containerd_changed_platform_manifest_is_update(self): + self.containerd('d') + self.index() + self.assertIn('1 updates', self.report()) + + def test_containerd_without_platform_descriptor_is_unchecked(self): + self.containerd() + del self.commands.containers[0]['ImageManifestDescriptor'] + self.assertIn('1 unchecked', self.report()) + + def test_two_containerd_platforms_of_one_index_are_checked_separately(self): + self.containerd() + self.index() + self.commands.containers.append({ + 'Image': digest('b'), 'Reference': 'alpine:3.21', 'ImageManifestDescriptor': { + 'mediaType': MANIFEST, 'digest': digest('d'), + 'platform': {'os': 'linux', 'architecture': 'arm64'}}}) + self.commands.raw['alpine@' + digest('d')] = { + 'schemaVersion': 2, 'mediaType': MANIFEST, 'config': {'digest': digest('f')}} + self.assertIn('2 current', self.report()) + + def test_invalid_container_platform_descriptor_is_unchecked(self): + self.containerd() + self.commands.containers[0]['ImageManifestDescriptor']['digest'] = 'invalid' + self.assertIn('1 unchecked', self.report()) + + def build_service(self, image='alpine:3.21'): + core = Path(self.tmp.name) / 'apex' + core.mkdir() + for filename in ('docker-compose.yml', '.env', 'apex.env'): + (core / filename).touch() + self.commands.compose = {'name': 'apex', 'services': { + 'custom': {'build': {'context': '.'}, 'environment': {'TOKEN': 'private-test-value'}}}} + if image is not None: + self.commands.compose['services']['custom']['image'] = image + self.commands.containers[0]['Labels'] = { + 'com.docker.compose.project': 'apex', 'com.docker.compose.service': 'custom'} + + def test_declared_compose_build_with_synthetic_repo_digest_is_local(self): + self.containerd() + self.build_service() + self.assertIn('1 local/untracked', self.report()) + self.assertFalse(any('imagetools' in c for c in self.commands.calls)) + self.assertIn(['timeout', '--signal=TERM', '--kill-after=5s', '30s', 'docker', + 'compose', '--env-file', '.env', '--env-file', 'apex.env', + 'config', '--format', 'json'], self.commands.calls) + self.assertEqual(self.commands.compose_options[0]['env']['APEX_NODE_HOST'], 'test-node') + self.assertNotIn('private-test-value', str(self.ctx.log.mock_calls)) + + def test_implicit_compose_build_image_uses_service_provenance(self): + self.containerd() + self.build_service(image=None) + self.assertIn('1 local/untracked', self.report()) + + def test_changed_build_declaration_does_not_hide_old_registry_image(self): + self.build_service(image='different:latest') + self.assertIn('1 current', self.report()) + + def test_same_image_name_without_build_service_provenance_is_registry_checked(self): + self.build_service() + self.commands.containers[0]['Labels']['com.docker.compose.service'] = 'other' + self.assertIn('1 current', self.report()) + + def test_failed_compose_resolution_is_unchecked_without_secret_output(self): + self.build_service() + self.commands.compose = 'private-test-value' + self.assertIn('Images: unchecked (Compose inspection failed)', self.report()) + self.assertNotIn('private-test-value', str(self.ctx.log.mock_calls)) + + def test_matching_config_digest_is_current_without_comparing_index_digest(self): + self.assertIn('1 current', self.report()) + self.assertIn(['docker', 'image', 'inspect', digest('a')], self.commands.calls) + + def test_moved_tag_compares_running_image_not_new_local_tag(self): + self.commands.raw['alpine@' + digest('c')]['config']['digest'] = digest('d') + self.assertIn('1 updates', self.report()) + self.assertNotIn(['docker', 'image', 'inspect', 'alpine:3.21'], self.commands.calls) + + def index(self, variants=None): + self.commands.remote['alpine:3.21'] = {'schemaVersion': 2, 'mediaType': INDEX, + 'digest': digest('e'), 'manifests': variants or [ + {'mediaType': MANIFEST, 'digest': digest('d'), + 'platform': {'os': 'linux', 'architecture': 'arm64'}}, + {'mediaType': MANIFEST, 'digest': digest('c'), + 'platform': {'os': 'linux', 'architecture': 'amd64'}}, + {'mediaType': MANIFEST, 'digest': digest('f'), + 'platform': {'os': 'unknown', 'architecture': 'unknown'}}]} + + def test_index_only_other_architecture_changed_is_current(self): + self.index() + self.assertIn('1 current', self.report()) + self.assertFalse(any(c[-1] == 'alpine@' + digest('d') for c in self.commands.calls)) + + def test_index_matching_platform_changed_is_update(self): + self.index() + self.commands.raw['alpine@' + digest('c')]['config']['digest'] = digest('d') + self.assertIn('1 updates', self.report()) + + def test_ambiguous_or_missing_platform_is_unchecked(self): + for platforms in ([{'os': 'linux', 'architecture': 'arm64'}], + [{'os': 'linux', 'architecture': 'amd64'}] * 2): + with self.subTest(platforms=platforms): + self.index([{'mediaType': MANIFEST, 'digest': digest('c'), 'platform': p} + for p in platforms]) + self.assertIn('1 unchecked', self.report()) + + def test_arm64_default_variant_matches_v8(self): + self.commands.local[digest('a')]['Architecture'] = 'arm64' + self.index([{'mediaType': MANIFEST, 'digest': digest('c'), + 'platform': {'os': 'linux', 'architecture': 'arm64', 'variant': 'v8'}}]) + self.assertIn('1 current', self.report()) + + def test_single_manifest_different_platform_is_unchecked(self): + self.commands.raw['alpine@' + digest('c')]['config']['digest'] = digest('d') + self.commands.config['alpine@' + digest('c')]['architecture'] = 'arm64' + self.assertIn('1 unchecked', self.report()) + + def test_registry_port_and_tag_are_preserved(self): + ref = 'registry.example:5000/team/app:stable' + self.commands.containers[0]['Reference'] = ref + self.commands.remote[ref] = self.commands.remote['alpine:3.21'] + self.commands.raw['registry.example:5000/team/app@' + digest('c')] = self.commands.raw['alpine@' + digest('c')] + self.assertIn('1 current', self.report()) + self.assertIn(['timeout', '--signal=TERM', '--kill-after=5s', '30s', + 'docker', 'buildx', 'imagetools', 'inspect', '--raw', + 'registry.example:5000/team/app@' + digest('c')], self.commands.calls) + + def test_registry_port_without_tag_uses_the_whole_repository(self): + ref = 'localhost:5000/team/app' + self.commands.containers[0]['Reference'] = ref + self.commands.remote[ref] = self.commands.remote['alpine:3.21'] + self.commands.raw[ref + '@' + digest('c')] = self.commands.raw['alpine@' + digest('c')] + self.assertIn('1 current', self.report()) + + def test_two_running_versions_of_one_tag_are_both_checked(self): + self.commands.containers.append({'Image': digest('d'), 'Reference': 'alpine:3.21'}) + self.commands.local[digest('d')] = {**self.commands.local[digest('a')], 'Id': digest('d')} + message = self.report() + self.assertIn('1 current', message) + self.assertIn('1 updates', message) + + def test_malformed_local_image_is_unchecked_not_a_local_build(self): + del self.commands.local[digest('a')]['RepoDigests'] + self.assertIn('1 unchecked', self.report()) + + def test_untrusted_reference_is_argv_not_shell_input(self): + self.commands.containers[0]['Reference'] = "registry.example/app:$(touch /tmp/nope)" + self.assertIn('1 unchecked', self.report()) + self.assertTrue(all('touch' not in c[-1] for c in self.commands.calls if c[0] == 'bash')) + + def test_apt_failure_is_explicit_and_images_still_checked(self): + self.commands.fail.add(('sudo', 'apt-get', 'update')) + message = self.report() + self.assertIn('APT: failed to check', message) + self.assertIn('1 current', message) + + def test_pinned_and_local_images_are_explicit_and_not_queried(self): + self.commands.containers = [{'Image': digest('a'), 'Reference': 'alpine@' + digest('b')}, + {'Image': digest('d'), 'Reference': 'project-local:latest'}] + self.commands.local[digest('d')] = {**self.commands.local[digest('a')], + 'Id': digest('d'), 'RepoDigests': []} + message = self.report() + self.assertIn('1 pinned', message) + self.assertIn('1 local/untracked', message) + self.assertFalse(any('imagetools' in c for c in self.commands.calls)) + + def test_registry_timeout_is_bounded_and_reported_unchecked(self): + original = self.commands.run + + def time_out(cmd, **kwargs): + if 'imagetools' in cmd: + self.assertEqual(cmd[:4], ['timeout', '--signal=TERM', '--kill-after=5s', '30s']) + return subprocess.CompletedProcess(cmd, 124, '', '') + return original(cmd, **kwargs) + + self.commands.run = time_out + self.assertIn('1 unchecked', self.report()) + + def test_failed_registry_and_malformed_results_are_unchecked(self): + for value in ('not-json', '{}', '[]', '{"mediaType":"unknown"}'): + with self.subTest(value=value): + self.commands.remote['alpine:3.21'] = value + self.assertIn('1 unchecked', self.report()) + self.commands.fail.add(('docker', 'buildx', 'imagetools', 'inspect', '--format', + '{{json .Manifest}}', 'alpine:3.21')) + self.assertIn('1 unchecked', self.report()) + + def test_bad_config_digest_is_unchecked(self): + self.commands.raw['alpine@' + digest('c')]['config'] = {} + self.assertIn('1 unchecked', self.report()) + + def test_docker_listing_failure_is_not_empty_success(self): + self.commands.fail.add(('docker', 'ps', '--quiet', '--no-trunc')) + self.assertIn('unchecked', self.report()) + + def test_buildx_missing_still_reports_unchecked(self): + self.commands.fail.add(('docker', 'buildx', 'version')) + self.assertIn('1 unchecked', self.report()) + + def test_image_counts_precede_long_package_details_and_notification_failure_exits(self): + self.commands.apt = '\n'.join('package-' + str(i) for i in range(100)) + message = self.report() + self.assertIn('Images:', message[:200]) + self.assertIn('1 current', message[:200]) + self.assertIn('100 pending', message[:200]) + self.assertNotIn('Skopeo', message) + self.ctx.notify.info.return_value = False + with self.assertRaises(SystemExit): + self.report() + + +if __name__ == '__main__': + unittest.main()