Complete renamed image source and verify bounded consumer migration readiness

Assistant: codex
Assistant-Model: gpt-6.1-sol
Assistant-Session: 01a126e1-76f8-7643-8cf3-e93336539ade

tests/test_runtime_publication.py

@@ -0,0 +1,53 @@
+import json
+import os
+from pathlib import Path
+import subprocess
+import tempfile
+import unittest
+
+ROOT = Path(__file__).resolve().parents[1]
+
+class PublicationTests(unittest.TestCase):
+    def run_publisher(self, mode='paired', revision='a' * 40, mismatch=False, array=False):
+        with tempfile.TemporaryDirectory() as folder:
+            base=Path(folder); log=base/'commands.jsonl'
+            docker=base/'docker'
+            docker.write_text('''#!/usr/bin/env python3
+import json,os,sys
+with open(os.environ['COMMAND_LOG'],'a') as f:f.write(json.dumps(sys.argv[1:])+'\\n')
+if sys.argv[1:3]==['image','inspect']:
+ digest='sha256:'+('b' if os.environ.get('MISMATCH')=='1' and '/access-engine:' in sys.argv[-1] else 'a')*64
+ payload=[sys.argv[-1].split(':main-')[0]+'@'+digest]
+ print(json.dumps({'invalid':'response'} if os.environ.get('ARRAY')=='1' else payload))
+''');docker.chmod(0o755)
+            result=subprocess.run(['bash',str(ROOT/'tools/publish-runtime-images.sh'),revision,mode,'.'],capture_output=True,text=True,env={**os.environ,'PATH':str(base)+':'+os.environ['PATH'],'COMMAND_LOG':str(log),'MISMATCH':str(int(mismatch)),'ARRAY':str(int(array))})
+            calls=[json.loads(line) for line in log.read_text().splitlines()] if log.exists() else []
+            return result,calls
+
+    def test_paired_builds_once_and_verifies_both_registry_digests(self):
+        result,calls=self.run_publisher();self.assertEqual(result.returncode,0,result.stderr)
+        self.assertEqual(sum(c[0]=='build' for c in calls),1)
+        pushes=[c[1] for c in calls if c[0]=='push']
+        self.assertEqual(set(pushes),{f'forgejo.coulomb.social/coulomb/{name}:{tag}' for name in ('access-engine','flex-auth') for tag in ('latest','main-aaaaaaa')})
+        self.assertEqual(sum(c[:2]==['image','inspect'] for c in calls),2)
+        self.assertEqual(sum(c[0]=='pull' for c in calls),2)
+        self.assertIn('manifest_digest=sha256:',result.stdout)
+        self.assertFalse(any(c[0]=='login' for c in calls))
+
+    def test_normal_push_cannot_publish_new_package(self):
+        result,calls=self.run_publisher('legacy');self.assertEqual(result.returncode,0)
+        self.assertFalse(any('coulomb/access-engine' in arg for call in calls for arg in call))
+
+    def test_different_registry_digest_refuses_success(self):
+        result,calls=self.run_publisher(mismatch=True);self.assertNotEqual(result.returncode,0)
+        self.assertNotIn('verified_repositories=',result.stdout)
+
+    def test_unexpected_inspection_response_refuses_success(self):
+        result,_=self.run_publisher(array=True);self.assertNotEqual(result.returncode,0)
+
+    def test_invalid_mode_or_mutable_revision_stops_before_docker(self):
+        for mode,revision in [('other','a'*40),('paired','main'),('paired','a'*7)]:
+            with self.subTest(mode=mode,revision=revision):
+                result,calls=self.run_publisher(mode,revision);self.assertNotEqual(result.returncode,0);self.assertEqual(calls,[])
+
+if __name__=='__main__':unittest.main()