shadowdaoandClaude Sonnet 5.5
f1c772329b
One job on the trusted x86 runner builds amd64 and arm64 (QEMU), pushes each platform by digest, then scripts/assemble-multiarch-index.sh creates :latest. arm64 is continue-on-error: a failed arm64 leg publishes the usual amd64-only tag. No Dockerfile change (both base images publish arm64). Adds workflow_dispatch and an offline test for the assemble script and workflow wiring. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_0185ukXjXyosSkunDsdWz3mb
215 lines
9.6 KiB
Python
215 lines
9.6 KiB
Python
#!/usr/bin/env python3
|
|
"""Offline tests for scripts/assemble-multiarch-index.sh (multi-arch index assembly).
|
|
|
|
The script is the publish-side half of the multi-arch pipeline: it receives the
|
|
amd64 and (optional) arm64 digests that the build steps pushed BY DIGEST and
|
|
turns them into the tag(s). A stub stands in for `docker buildx imagetools`
|
|
(env IMAGETOOLS), records every `create` call and answers `inspect` from a
|
|
table, so no registry or Docker is needed.
|
|
|
|
Held invariants:
|
|
- arm64 failure / cancel / skip -> amd64-only create, exit 0 (tag still published)
|
|
- a STALE arm64 digest is never attached unless this run's arm64 outcome is success
|
|
- bad/missing amd64 digest, or amd64 that is not linux/amd64 -> exit 1, NO create
|
|
- an arm64 digest that does not inspect as arm64 is dropped (amd64-only)
|
|
- every tag x every repo is created; sources are <repo>@<digest> in that repo
|
|
|
|
Running: python3 scripts/test-assemble-multiarch-index.py
|
|
"""
|
|
import json
|
|
import os
|
|
import stat
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
|
|
HERE = os.path.dirname(os.path.abspath(__file__))
|
|
SCRIPT = os.path.join(HERE, 'assemble-multiarch-index.sh')
|
|
A = 'sha256:' + 'a' * 64
|
|
B = 'sha256:' + 'b' * 64
|
|
C = 'sha256:' + 'c' * 64
|
|
|
|
STUB = r'''#!/usr/bin/env python3
|
|
import json, os, sys
|
|
a = sys.argv[1:]
|
|
log = os.environ['STUB_LOG']
|
|
table = json.loads(os.environ['STUB_ARCH'])
|
|
if a[0] == 'inspect':
|
|
ref = [x for x in a[1:] if '@sha256:' in x][0]
|
|
arch = table.get(ref.split('@', 1)[1])
|
|
if arch is None:
|
|
sys.exit(1)
|
|
print(json.dumps({'architecture': arch, 'os': 'linux'}))
|
|
elif a[0] == 'create':
|
|
open(log, 'a').write(json.dumps(a[1:]) + '\n')
|
|
else:
|
|
sys.exit(2)
|
|
'''
|
|
|
|
|
|
class AssembleTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.stub = os.path.join(self.tmp.name, 'stub')
|
|
with open(self.stub, 'w') as f:
|
|
f.write(STUB)
|
|
os.chmod(self.stub, os.stat(self.stub).st_mode | stat.S_IEXEC)
|
|
self.log = os.path.join(self.tmp.name, 'creates.log')
|
|
|
|
def tearDown(self):
|
|
self.tmp.cleanup()
|
|
|
|
def run_script(self, args, arch=None):
|
|
env = dict(os.environ, IMAGETOOLS=self.stub, STUB_LOG=self.log,
|
|
STUB_ARCH=json.dumps(arch if arch is not None else {A: 'amd64', B: 'arm64'}))
|
|
p = subprocess.run(['bash', SCRIPT] + args, env=env, capture_output=True, text=True)
|
|
creates = []
|
|
if os.path.exists(self.log):
|
|
creates = [json.loads(l) for l in open(self.log)]
|
|
return p, creates
|
|
|
|
BASE = ['--repo', 'r.example/o/img', '--tag', 'latest']
|
|
|
|
def test_both_platforms(self):
|
|
p, c = self.run_script(['--amd64', A, '--arm64', B, '--arm64-outcome', 'success'] + self.BASE)
|
|
self.assertEqual(p.returncode, 0, p.stderr)
|
|
self.assertEqual(c, [['-t', 'r.example/o/img:latest', 'r.example/o/img@' + A, 'r.example/o/img@' + B]])
|
|
self.assertIn('platforms=linux/amd64,linux/arm64', p.stdout)
|
|
|
|
def test_arm64_failure_is_amd64_only_even_with_a_digest(self):
|
|
# A stale digest (e.g. left over from an earlier run) must not be attached.
|
|
for outcome in ('failure', 'cancelled', 'skipped', ''):
|
|
if os.path.exists(self.log):
|
|
os.remove(self.log)
|
|
p, c = self.run_script(['--amd64', A, '--arm64', B, '--arm64-outcome', outcome] + self.BASE)
|
|
self.assertEqual(p.returncode, 0, p.stderr)
|
|
self.assertEqual(c, [['-t', 'r.example/o/img:latest', 'r.example/o/img@' + A]], outcome)
|
|
self.assertIn('::warning::', p.stdout)
|
|
self.assertIn('platforms=linux/amd64\n', p.stdout)
|
|
|
|
def test_no_arm64_args_is_amd64_only(self):
|
|
p, c = self.run_script(['--amd64', A] + self.BASE)
|
|
self.assertEqual(p.returncode, 0, p.stderr)
|
|
self.assertEqual(c, [['-t', 'r.example/o/img:latest', 'r.example/o/img@' + A]])
|
|
|
|
def test_success_with_empty_or_malformed_arm64_digest(self):
|
|
for bad in ('', 'sha256:short', 'latest'):
|
|
if os.path.exists(self.log):
|
|
os.remove(self.log)
|
|
p, c = self.run_script(['--amd64', A, '--arm64', bad, '--arm64-outcome', 'success'] + self.BASE)
|
|
self.assertEqual(p.returncode, 0, p.stderr)
|
|
self.assertEqual(len(c[0]), 3, bad) # -t, ref, one source
|
|
|
|
def test_arm64_digest_that_is_not_arm64_is_dropped(self):
|
|
p, c = self.run_script(['--amd64', A, '--arm64', C, '--arm64-outcome', 'success'] + self.BASE,
|
|
arch={A: 'amd64', C: 'amd64'})
|
|
self.assertEqual(p.returncode, 0, p.stderr)
|
|
self.assertEqual(c, [['-t', 'r.example/o/img:latest', 'r.example/o/img@' + A]])
|
|
|
|
def test_arm64_equal_to_amd64_is_dropped(self):
|
|
p, c = self.run_script(['--amd64', A, '--arm64', A, '--arm64-outcome', 'success'] + self.BASE)
|
|
self.assertEqual(p.returncode, 0, p.stderr)
|
|
self.assertEqual(len(c[0]), 3)
|
|
|
|
def test_amd64_problems_publish_nothing(self):
|
|
for amd in ('', 'nope', B): # B inspects as arm64
|
|
p, c = self.run_script(['--amd64', amd, '--arm64', B, '--arm64-outcome', 'success'] + self.BASE)
|
|
self.assertNotEqual(p.returncode, 0, amd)
|
|
self.assertEqual(c, [], amd)
|
|
p, c = self.run_script(['--amd64', C, '--arm64-outcome', 'failure'] + self.BASE, arch={})
|
|
self.assertNotEqual(p.returncode, 0) # not inspectable
|
|
self.assertEqual(c, [])
|
|
|
|
def test_all_tags_in_all_repos(self):
|
|
p, c = self.run_script(['--amd64', A, '--arm64', B, '--arm64-outcome', 'success',
|
|
'--repo', 'gitea.example/o/img', '--repo', 'ghcr.io/o/img',
|
|
'--tag', 'latest', '--tag', '1.2.3', '--tag', 'abc123'])
|
|
self.assertEqual(p.returncode, 0, p.stderr)
|
|
self.assertEqual(len(c), 2)
|
|
self.assertEqual(c[0], ['-t', 'gitea.example/o/img:latest', '-t', 'gitea.example/o/img:1.2.3',
|
|
'-t', 'gitea.example/o/img:abc123',
|
|
'gitea.example/o/img@' + A, 'gitea.example/o/img@' + B])
|
|
self.assertEqual(c[1][-2:], ['ghcr.io/o/img@' + A, 'ghcr.io/o/img@' + B])
|
|
|
|
def test_requires_repo_and_tag_and_rejects_unknown_args(self):
|
|
for args in (['--amd64', A, '--tag', 't'], ['--amd64', A, '--repo', 'r'], ['--bogus', 'x']):
|
|
p, c = self.run_script(args)
|
|
self.assertNotEqual(p.returncode, 0)
|
|
self.assertEqual(c, [])
|
|
|
|
|
|
class WorkflowWiring(unittest.TestCase):
|
|
"""build-push.yaml must keep the properties the assemble script relies on,
|
|
and must not have changed what the amd64 build does."""
|
|
|
|
IMG = 'repo.anhonesthost.net/cloud-hosting-platform/whp-default'
|
|
|
|
def setUp(self):
|
|
self.wf = open(os.path.join(HERE, '..', '.gitea', 'workflows', 'build-push.yaml')).read()
|
|
|
|
def step(self, name):
|
|
parts = self.wf.split(' - name: ')
|
|
for p in parts:
|
|
if p.startswith(name):
|
|
return p
|
|
self.fail('step not found: ' + name)
|
|
|
|
def test_triggers_unchanged_plus_manual_dispatch(self):
|
|
on = self.wf.split('\njobs:')[0]
|
|
self.assertIn('branches:\n - main\n', on)
|
|
self.assertIn('workflow_dispatch:', on)
|
|
self.assertNotIn('schedule:', on)
|
|
self.assertNotIn('pull_request', on)
|
|
|
|
def test_arm64_step_is_non_fatal_and_digest_comes_from_step_output(self):
|
|
arm = self.step('Build arm64')
|
|
self.assertIn('continue-on-error: true', arm)
|
|
self.assertNotIn('continue-on-error', self.step('Build amd64'))
|
|
self.assertIn('steps.arm64.outputs.digest', self.wf)
|
|
self.assertIn('steps.arm64.outcome', self.wf)
|
|
self.assertIn('steps.amd64.outputs.digest', self.wf)
|
|
|
|
def test_each_build_is_one_platform_pushed_by_digest_untagged(self):
|
|
a, b = self.step('Build amd64'), self.step('Build arm64')
|
|
self.assertIn('platforms: linux/amd64\n', a)
|
|
self.assertIn('platforms: linux/arm64\n', b)
|
|
for s in (a, b):
|
|
self.assertIn('push-by-digest=true', s)
|
|
self.assertIn(self.IMG, s)
|
|
self.assertNotIn('tags:', s)
|
|
self.assertNotRegex(s, r'\n\s+push: true')
|
|
self.assertEqual(self.wf.count('push-by-digest=true'), 2)
|
|
|
|
def test_build_args_identical_for_both_platforms_and_keep_base_pin(self):
|
|
def args(s):
|
|
return s.split('build-args: |\n')[1].split('outputs:')[0]
|
|
a, b = args(self.step('Build amd64')), args(self.step('Build arm64'))
|
|
self.assertEqual(a, b)
|
|
self.assertIn('BASE_IMAGE=${{ steps.base.outputs.image }}', a)
|
|
|
|
|
|
def test_assemble_step_uses_script_for_latest_in_the_existing_registry_only(self):
|
|
asm = self.step('Assemble')
|
|
self.assertIn('scripts/assemble-multiarch-index.sh', asm)
|
|
self.assertIn('--repo ' + self.IMG, asm)
|
|
self.assertEqual(asm.count('--repo '), 1) # unchanged: Gitea only, no new registry
|
|
self.assertIn('--tag latest', asm)
|
|
self.assertEqual(asm.count('--tag '), 1)
|
|
|
|
def test_qemu_and_registry_login_and_base_resolution_retained(self):
|
|
for needle in ('docker/setup-qemu-action', 'Login to Gitea', 'secrets.CI_USER',
|
|
'secrets.CI_TOKEN', 'Resolve base image digest', 'Probe package cache'):
|
|
self.assertIn(needle, self.wf)
|
|
self.assertNotIn('ghcr.io', self.wf) # no new registry/secret introduced
|
|
self.assertNotIn('GHCR_TOKEN', self.wf)
|
|
|
|
def test_assemble_step_is_last_and_after_both_builds(self):
|
|
i = [self.wf.index(n) for n in ('Build amd64', 'Build arm64', 'Assemble :latest')]
|
|
self.assertEqual(i, sorted(i))
|
|
self.assertEqual(self.wf.rstrip().split('--tag latest')[-1], '')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main(verbosity=2)
|