Files
computer-playbook/tests/unit/python/cli/contributing/mirror/test_sync.py
T
kevinveenbirkenbachandClaude Opus 5.5 79e88c5ea4 fix(mirror): refresh a mirrored tag that upstream re-pushed
Fork CI run 41147980556 pgAdmin#0 failed in sys-ctl-hlth-runtime with
"stalwart_webmail: PHP 8.4.26 is below the security floor 8.4.27".
The deploy ran the fork's GHCR mirror of roundcube/roundcubemail:1.7.5-
apache, copied on 10-19 with PHP 8.4.26, while upstream re-pushed the
same tag on 10-27 with PHP 8.4.27. Branch CI runs the sync with
--only-missing, which skipped the tag because it existed ("destination
exists, skipping") without comparing digests, and the full refresh only
runs on the fork's default branch, which still pins 1.7.4.

- utils/docker/registry.py: manifest_digest() reads the
  Docker-Content-Digest of a reference with a HEAD request, which does
  not count against Docker Hub's pull allowance.
- GHCRProvider.is_stale() reports a mirrored tag as stale only when both
  registries answer with a digest and the two differ.
- Under --only-missing an existing tag is re-copied when it is stale; a
  failed re-copy is reported on stderr with the image and keeps the
  existing copy, so one refresh cannot block every deploy.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-31 09:50:14 +01:00

122 lines
3.9 KiB
Python

from __future__ import annotations
import subprocess
import unittest
from unittest.mock import patch
import cli.contributing.mirror.sync.__main__ as sync_main
from cli.contributing.mirror.model import ImageRef
class TestMirrorSync(unittest.TestCase):
IMAGE = ImageRef(
role="web-app-nextcloud",
service="app",
name="nextcloud",
version="31.0.0",
source="docker.io/library/nextcloud:31.0.0",
registry="docker.io",
)
def _run_only_missing(self, *, stale: bool, mirror_error=None):
with (
patch(
"cli.contributing.mirror.sync.__main__.iter_role_images",
return_value=[self.IMAGE],
),
patch(
"cli.contributing.mirror.sync.__main__.iter_runner_images",
return_value=[],
),
patch.object(
sync_main.GHCRProvider,
"image_base",
return_value="ghcr.io/acme/myrepo/mirror/docker.io/nextcloud",
),
patch.object(
sync_main.GHCRProvider, "tag_exists", return_value=True
) as mock_tag_exists,
patch.object(
sync_main.GHCRProvider, "is_stale", return_value=stale
) as mock_is_stale,
patch.object(
sync_main.GHCRProvider, "mirror", side_effect=mirror_error
) as mock_mirror,
patch(
"sys.argv",
[
"mirror-sync",
"--ghcr-namespace",
"acme",
"--ghcr-repository",
"myrepo",
"--only-missing",
],
),
):
result = sync_main.main()
mock_tag_exists.assert_called_once_with(self.IMAGE)
mock_is_stale.assert_called_once_with(self.IMAGE)
return result, mock_mirror
def test_only_missing_skips_current_destination(self) -> None:
result, mock_mirror = self._run_only_missing(stale=False)
self.assertEqual(result, 0)
mock_mirror.assert_not_called()
def test_only_missing_refreshes_a_stale_destination(self) -> None:
result, mock_mirror = self._run_only_missing(stale=True)
self.assertEqual(result, 0)
mock_mirror.assert_called_once_with(self.IMAGE)
def test_failed_refresh_keeps_the_existing_copy_without_failing(self) -> None:
error = subprocess.CalledProcessError(1, ["skopeo", "copy"])
result, mock_mirror = self._run_only_missing(stale=True, mirror_error=error)
self.assertEqual(result, 0)
mock_mirror.assert_called_once_with(self.IMAGE)
def test_runner_images_are_mirrored_with_the_role_images(self) -> None:
image = ImageRef(
role="runner",
service="coredns",
name="coredns/coredns",
version="1.14.4",
source="docker.io/coredns/coredns:1.14.4",
registry="docker.io",
source_file="compose.yml",
)
with (
patch(
"cli.contributing.mirror.sync.__main__.iter_role_images",
return_value=[],
),
patch(
"cli.contributing.mirror.sync.__main__.iter_runner_images",
return_value=[image],
),
patch.object(sync_main.GHCRProvider, "mirror") as mock_mirror,
patch(
"sys.argv",
[
"mirror-sync",
"--ghcr-namespace",
"acme",
"--ghcr-repository",
"myrepo",
],
),
):
result = sync_main.main()
self.assertEqual(result, 0)
mock_mirror.assert_called_once_with(image)
if __name__ == "__main__": # pragma: no cover
unittest.main()