"""Exercise real shell control flow against a stateful registry/daemon double.""" import json import os from pathlib import Path import shutil import subprocess import tempfile import unittest ROOT = Path(__file__).resolve().parents[2] SHA = 'a' * 40 REGISTRY = '192.168.2.28:3000/wanjakrah/agiledancers-' DOCKER = r'''#!/usr/bin/env python3 import json, os, pathlib, sys p = pathlib.Path(os.environ['FAKE_STATE']) s = json.loads(p.read_text()) a = sys.argv[1:] s['calls'].append(a) p.write_text(json.dumps(s)) if a[0] == 'info': print('amd64' if 'Architecture' in ' '.join(a) else 'linux') elif a[0] == 'context': print('unix:///tmp/test-docker.sock') elif a[0] == 'login': sys.stdin.read() elif a[:2] == ['manifest', 'inspect']: if os.environ.get('FAKE_MANIFEST_ERROR'): print('unauthorized: authentication required', file=sys.stderr); sys.exit(1) if a[-1] not in s['images']: print('manifest unknown', file=sys.stderr); sys.exit(1) elif a[0] == 'push': if os.environ.get('FAKE_PUSH_FAILURE', '') in a[-1] and os.environ.get('FAKE_PUSH_FAILURE'): sys.exit(1) s['images'][a[-1]] = os.environ['CI_COMMIT']; p.write_text(json.dumps(s)) elif a[0] == 'pull': if os.environ.get('FAKE_PULL_FAILURE', '') in a[-1] and os.environ.get('FAKE_PULL_FAILURE'): sys.exit(1) elif a[:2] == ['image', 'inspect']: if 'RepoDigests' in ' '.join(a): print(a[-1].rsplit(':', 1)[0] + '@sha256:' + 'b' * 64) else: print(s['images'].get(a[-1], 'missing')) elif a[:2] == ['container', 'inspect'] or a[:2] == ['network', 'inspect']: sys.exit(1) ''' class PublicationTest(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory() self.addCleanup(self.temp.cleanup) self.root = Path(self.temp.name) shutil.copytree(ROOT / 'ci', self.root / 'ci', ignore=shutil.ignore_patterns('artifacts', '__pycache__')) (self.root / 'ci/main-head.py').write_text("import os; print(os.environ.get('FAKE_MAIN_HEAD', os.environ['CI_COMMIT']))\n") self.bin = self.root / 'bin'; self.bin.mkdir() (self.bin / 'docker').write_text(DOCKER); (self.bin / 'docker').chmod(0o755) self.state = self.root / 'state.json' self.state.write_text(json.dumps({'images': {}, 'calls': []})) artifacts = self.root / 'ci/artifacts'; artifacts.mkdir() (artifacts / 'verified-commit').write_text(SHA + '\n') self.env = dict(os.environ, PATH=str(self.bin) + ':' + os.environ['PATH'], CI_COMMIT=SHA, CI_REF='refs/heads/main', CI_REPOSITORY='wanjakrah/agiledancers', CI_RUN_ID='test', CI_RUN_ATTEMPT='1', TARGET_PLATFORM='linux/amd64', GITEA_SERVER_URL='http://192.168.2.28:3000', REGISTRY_USER='test', REGISTRY_TOKEN='not-a-secret', CI_PUBLISH_LOCK_DIR=str(self.root / 'lock'), FAKE_STATE=str(self.state)) def run_script(self, script='publish.sh', **extra): return subprocess.run(['bash', str(self.root / 'ci' / script)], cwd=self.root, env=dict(self.env, **extra), capture_output=True, text=True, timeout=20) def calls(self, command): return [a for a in json.loads(self.state.read_text())['calls'] if a[0] == command] def test_feature_branch_does_not_contact_registry(self): result = self.run_script(CI_REF='refs/heads/feature/test') self.assertEqual(0, result.returncode, result.stderr) self.assertEqual([], self.calls('login')) self.assertEqual([], self.calls('push')) def test_unverified_commit_cannot_publish(self): (self.root / 'ci/artifacts/verified-commit').unlink() self.assertNotEqual(0, self.run_script().returncode) self.assertEqual([], self.calls('push')) def test_publish_pull_and_rerun_preserves_commit_tags(self): result = self.run_script() self.assertEqual(0, result.returncode, result.stdout + result.stderr) self.assertEqual(4, len(self.calls('push'))) result = self.run_script() self.assertEqual(0, result.returncode, result.stdout + result.stderr) self.assertEqual(6, len(self.calls('push'))) # only main aliases on rerun self.assertEqual(8, len(self.calls('pull'))) def test_partial_publication_is_completed_without_overwriting_backend(self): self.assertNotEqual(0, self.run_script(FAKE_PUSH_FAILURE='frontend:' + SHA).returncode) self.assertFalse(any(a[-1].endswith(':main') for a in self.calls('push'))) result = self.run_script() self.assertEqual(0, result.returncode, result.stdout + result.stderr) backend_pushes = [a for a in self.calls('push') if a[-1] == REGISTRY + 'backend:' + SHA] self.assertEqual(1, len(backend_pushes)) def test_registry_auth_failure_is_not_treated_as_missing_image(self): self.assertNotEqual(0, self.run_script(FAKE_MANIFEST_ERROR='1').returncode) self.assertEqual([], self.calls('push')) def test_pull_failure_blocks_main_aliases(self): self.assertNotEqual(0, self.run_script(FAKE_PULL_FAILURE='backend:' + SHA).returncode) self.assertFalse(any(a[-1].endswith(':main') for a in self.calls('push'))) def test_old_commit_does_not_move_main(self): result = self.run_script(FAKE_MAIN_HEAD='c' * 40) self.assertEqual(0, result.returncode, result.stdout + result.stderr) self.assertEqual(2, len(self.calls('push'))) self.assertTrue(all(a[-1].endswith(':' + SHA) for a in self.calls('push'))) def test_architecture_mismatch_fails_before_publication(self): self.assertNotEqual(0, self.run_script(TARGET_PLATFORM='linux/arm64').returncode) self.assertEqual([], self.calls('push')) def test_failed_verification_removes_stale_success_marker_and_never_builds_images(self): for command, content in [('java', 'exit 0'), ('node', 'exit 0'), ('mvn', 'exit 42'), ('git', 'echo ' + SHA)]: path = self.bin / command; path.write_text('#!/bin/sh\n' + content + '\n'); path.chmod(0o755) result = self.run_script('verify.sh') self.assertEqual(42, result.returncode, result.stdout + result.stderr) self.assertFalse((self.root / 'ci/artifacts/verified-commit').exists()) self.assertEqual([], self.calls('build')) self.assertEqual([], self.calls('push')) if __name__ == '__main__': unittest.main()