fix(registry): verify publish tags with docker auth

src/aio_fleet/cli.py

@@ -1117,29 +1117,32 @@ def cmd_registry_publish(args: argparse.Namespace) -> int:
     if args.dry_run:
         print(" ".join(shlex.quote(part) for part in command))
         return 0
+    tags = compute_registry_tags(repo, sha=sha, component=args.component)
     print(f"{repo.name}:{args.component}: registry=publishing", flush=True)
     try:
         with _registry_publish_environment(repo) as publish_env:
             result = _run_streaming(command, cwd=repo.path, env=publish_env)
+            if result.returncode != 0:
+                return result.returncode
+            verification_failures = verify_registry_tags(tags.all_tags, env=publish_env)
     except RuntimeError as exc:
         print(str(exc), file=sys.stderr)
         return 1
-    if result.returncode != 0:
-        return result.returncode
-    return cmd_registry_verify(
-        argparse.Namespace(
-            manifest=args.manifest,
-            all=False,
-            repo=args.repo,
-            repo_path=args.repo_path,
-            sha=sha,
-            component=args.component,
-            include_manual=True,
-            dry_run=False,
-            format="text",
-            verbose=True,
+    label = repo.name if args.component == "aio" else f"{repo.name}:{args.component}"
+    state = "failed" if verification_failures else "ok"
+    print(f"{label}: registry={state}")
+    for tag in [*tags.dockerhub, *tags.ghcr]:
+        print(f"- {tag}")
+    if verification_failures:
+        print(
+            "\n".join(
+                f"{repo.name}:{args.component}: {failure}"
+                for failure in verification_failures
+            ),
+            file=sys.stderr,
         )
-    )
+        return 1
+    return 0
 
 
 @contextmanager

src/aio_fleet/registry.py

@@ -8,6 +8,7 @@
 import urllib.error
 import urllib.parse
 import urllib.request
+from collections.abc import Mapping
 from dataclasses import dataclass
 
 from aio_fleet.manifest import RepoConfig
@@ -61,7 +62,9 @@ def compute_registry_tags(
     )
 
 
-def verify_registry_tags(tags: list[str]) -> list[str]:
+def verify_registry_tags(
+    tags: list[str], *, env: Mapping[str, str] | None = None
+) -> list[str]:
     docker = shutil.which("docker")
     if docker is None:
         return ["docker CLI is required to verify registry tags"]
@@ -70,19 +73,22 @@ def verify_registry_tags(tags: list[str]) -> list[str]:
         failure = (
             _verify_dockerhub_tag(tag)
             if _is_dockerhub_tag(tag)
-            else _verify_with_docker_imagetools(docker, tag)
+            else _verify_with_docker_imagetools(docker, tag, env=env)
         )
         if failure:
             failures.append(failure)
     return failures
 
 
-def _verify_with_docker_imagetools(docker: str, tag: str) -> str | None:
+def _verify_with_docker_imagetools(
+    docker: str, tag: str, *, env: Mapping[str, str] | None = None
+) -> str | None:
     result = subprocess.run(  # nosec B603
         [docker, "buildx", "imagetools", "inspect", tag],
         check=False,
         text=True,
         capture_output=True,
+        env=env,
     )
     if result.returncode == 0:
         return None

tests/test_cli.py

@@ -705,13 +705,12 @@ def fake_run(
         seen["publish_env"] = env
         return SimpleNamespace(returncode=0, stdout="", stderr="")
 
-    def fake_registry_verify(args: Namespace) -> int:
-        seen.setdefault("verify_calls", []).append(args)
-        return 0
+    def fake_verify(tags: list[str], *, env=None) -> list[str]:
+        seen.setdefault("verify_calls", []).append({"tags": tags, "env": env})
+        return []
 
     monkeypatch.setattr(cli, "_run_streaming", fake_run)
-    monkeypatch.setattr(cli, "verify_registry_tags", lambda _tags: ["tag missing"])
-    monkeypatch.setattr(cli, "cmd_registry_verify", fake_registry_verify)
+    monkeypatch.setattr(cli, "verify_registry_tags", fake_verify)
 
     result = cmd_registry_publish(
         Namespace(
@@ -729,8 +728,10 @@ def fake_registry_verify(args: Namespace) -> int:
     assert seen["publish_env"] is None  # nosec B101
     verify_calls = seen["verify_calls"]
     assert len(verify_calls) == 1  # nosec B101
-    assert verify_calls[0].repo_path == str(repo_path)  # nosec B101
-    assert verify_calls[0].sha == "a" * 40  # nosec B101
+    assert verify_calls[0]["env"] is None  # nosec B101
+    assert any(
+        "sha-" + "a" * 40 in tag for tag in verify_calls[0]["tags"]
+    )  # nosec B101
     captured = capsys.readouterr()
     assert "example-aio:aio: registry=publishing" in captured.out  # nosec B101
     assert "preflight" not in captured.err  # nosec B101
@@ -755,8 +756,7 @@ def fake_run(
         return SimpleNamespace(returncode=0, stdout="", stderr="")
 
     monkeypatch.setattr(cli, "_run_streaming", fake_run)
-    monkeypatch.setattr(cli, "verify_registry_tags", lambda _tags: [])
-    monkeypatch.setattr(cli, "cmd_registry_verify", lambda _args: 0)
+    monkeypatch.setattr(cli, "verify_registry_tags", lambda _tags, **_kwargs: [])
 
     result = cmd_registry_publish(
         Namespace(
@@ -818,8 +818,19 @@ def fake_publish(command: list[str], cwd: Path | None = None, env=None):
     monkeypatch.setenv("GITHUB_TOKEN", "github-token")
     monkeypatch.setattr(cli.subprocess, "run", fake_docker)
     monkeypatch.setattr(cli, "_run_streaming", fake_publish)
-    monkeypatch.setattr(cli, "verify_registry_tags", lambda _tags: ["tag missing"])
-    monkeypatch.setattr(cli, "cmd_registry_verify", lambda _args: 0)
+
+    def fake_verify(_tags: list[str], *, env=None) -> list[str]:
+        seen["verify_env"] = env
+        assert isinstance(env, dict)  # nosec B101
+        assert "BUILDX_BUILDER" in env  # nosec B101
+        assert "DOCKER_CONFIG" in env  # nosec B101
+        assert "DOCKERHUB_TOKEN" not in env  # nosec B101
+        assert "AIO_FLEET_GHCR_TOKEN" not in env  # nosec B101
+        assert "GH_TOKEN" not in env  # nosec B101
+        assert "GITHUB_TOKEN" not in env  # nosec B101
+        return []
+
+    monkeypatch.setattr(cli, "verify_registry_tags", fake_verify)
 
     result = cmd_registry_publish(
         Namespace(
@@ -858,6 +869,7 @@ def fake_publish(command: list[str], cwd: Path | None = None, env=None):
         ["docker", "buildx", "rm"],
     ]
     assert seen["publish_cwd"] == repo_path.resolve()  # nosec B101
+    assert seen["verify_env"] == seen["publish_env"]  # nosec B101
 
 
 def test_registry_verify_all_skips_manual_template_publish(
@@ -889,7 +901,7 @@ def test_registry_verify_all_skips_manual_template_publish(
 
     monkeypatch.setattr(cli, "_git_head", lambda _path: "a" * 40)
 
-    def fake_verify(tags: list[str]) -> list[str]:
+    def fake_verify(tags: list[str], **_kwargs) -> list[str]:
         verified.extend(tags)
         return []
 

tests/test_poll.py

@@ -120,7 +120,11 @@ def test_publish_required_accepts_runtime_and_release_commits(
     repo = load_manifest(manifest_path).repo("example-aio")
 
     for path in ("Dockerfile", "rootfs/etc/services.d/web/run", "CHANGELOG.md"):
-        monkeypatch.setattr(poll, "_commit_changed_paths", lambda _repo, _sha: [path])
+        monkeypatch.setattr(
+            poll,
+            "_commit_changed_paths",
+            lambda _repo, _sha, changed_path=path: [changed_path],
+        )
 
         assert (
             poll.publish_required(repo, sha="a" * 40, event="push") is True

tests/test_registry.py

@@ -218,16 +218,22 @@ def fake_urlopen(url: str, timeout: int):
 
 def test_ghcr_verification_uses_docker_imagetools(monkeypatch) -> None:
     seen_commands: list[list[str]] = []
+    inspect_env = {"DOCKER_CONFIG": "/workspace/aio-fleet-docker"}
+    seen_envs: list[dict[str, str] | None] = []
     monkeypatch.setattr(registry.shutil, "which", lambda _name: "docker")
 
-    def fake_run(command: list[str], **_kwargs):
+    def fake_run(command: list[str], **kwargs):
         seen_commands.append(command)
+        seen_envs.append(kwargs.get("env"))
         return SimpleNamespace(returncode=0, stdout="", stderr="")
 
     monkeypatch.setattr(registry.subprocess, "run", fake_run)
 
     assert (
-        registry.verify_registry_tags(["ghcr.io/jsonbored/sure-aio:latest"]) == []
+        registry.verify_registry_tags(
+            ["ghcr.io/jsonbored/sure-aio:latest"], env=inspect_env
+        )
+        == []
     )  # nosec B101
     assert seen_commands == [  # nosec B101
         [
@@ -238,3 +244,4 @@ def fake_run(command: list[str], **_kwargs):
             "ghcr.io/jsonbored/sure-aio:latest",
         ]
     ]
+    assert seen_envs == [inspect_env]  # nosec B101