[verified] Canonicalize native review publication journals

harness/tests/review/test_legacy_journal_contract.py

@@ -0,0 +1,126 @@
+import hashlib
+import importlib.util
+import json
+import stat
+import tempfile
+import unittest
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[3]
+SCRIPT = ROOT / 'harness/agents/review/skills/gitea-code-review/scripts/native_review.py'
+FIXTURE = ROOT / 'harness/tests/review/test_native_review.py'
+
+
+def load(path, name):
+    spec = importlib.util.spec_from_file_location(name, path)
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+class LegacyJournalContractTests(unittest.TestCase):
+    def test_legacy_intent_extra_fields_are_not_registered(self):
+        m = load(SCRIPT, 'legacy_contract_publisher')
+        f = load(FIXTURE, 'legacy_contract_fixture')
+        plan, before, after, exact = f.publication_fixture()
+        with tempfile.TemporaryDirectory(dir='/opt/data/cache/scratch') as td:
+            m._journal.root_for = lambda anchor: Path(td) / 'native-review-journals'
+            source = Path(td) / 'legacy'
+            payload = m.build_payload(plan)
+            intent = {'schema': 'gitea-agent.native-review-intent/v1',
+                      'fingerprint': hashlib.sha256(m.canonical(plan)).hexdigest(),
+                      'before_sha256': hashlib.sha256(m.canonical(before)).hexdigest(),
+                      'head': plan['target']['head'], 'event': payload['event'],
+                      'body_sha256': hashlib.sha256(payload['body'].encode()).hexdigest(),
+                      'unexpected': 'cannot silently import this'}
+            with m.Journal(source) as journal:
+                journal.write('before.json', before)
+                journal.write('intent.json', intent)
+            digest = m.legacy_journal_digest(source)
+            api = f.FakeAPI(plan, before, after, exact)
+            with self.assertRaises(m.ValidationError):
+                m.register_legacy_journal(plan, source, expected_legacy_sha256=digest,
+                                          api=api, lease=f.FakeLease(), declaration={'mode': 'none'})
+            self.assertEqual(api.posts, 0)
+            self.assertFalse((m.canonical_journal_path(plan) / 'intent.json').exists())
+
+    def test_legacy_inventory_rejects_unknown_artifact_before_import(self):
+        m = load(SCRIPT, 'legacy_inventory_contract')
+        f = load(FIXTURE, 'legacy_inventory_fixture')
+        plan, before, after, exact = f.publication_fixture()
+        with tempfile.TemporaryDirectory(dir='/opt/data/cache/scratch') as td:
+            m._journal.root_for = lambda anchor: Path(td) / 'native-review-journals'
+            source = Path(td) / 'legacy'; source.mkdir(mode=0o700)
+            with m.Journal(source) as journal:
+                journal.write('before.json', before)
+                journal.write('intent.json', m._expected_intent(plan, before))
+            extra = source / 'unknown.json'; extra.write_bytes(b'{"unknown":true}')
+            extra.chmod(0o600)
+            with self.assertRaises(m.ValidationError):
+                m.legacy_journal_digest(source)
+
+    def test_import_preserves_raw_bytes_and_archives_verified_separately(self):
+        m = load(SCRIPT, 'legacy_raw_import_contract')
+        f = load(FIXTURE, 'legacy_raw_import_fixture')
+        plan, before, after, exact = f.publication_fixture()
+        with tempfile.TemporaryDirectory(dir='/opt/data/cache/scratch') as td:
+            m._journal.root_for = lambda anchor: Path(td) / 'native-review-journals'
+            source = Path(td) / 'legacy'; source.mkdir(mode=0o700)
+            intent = m._expected_intent(plan, before)
+            before_raw = (json.dumps(before, sort_keys=True, indent=1) + '\n').encode()
+            intent_raw = (json.dumps(intent, sort_keys=True, indent=1) + '\n').encode()
+            verified_raw = b'{\n  "verified": true,\n  "legacy": "exact"\n}\n'
+            for name, raw in [('before.json', before_raw), ('intent.json', intent_raw),
+                              ('verified.json', verified_raw)]:
+                path = source / name; path.write_bytes(raw); path.chmod(0o600)
+            with m.Journal(source):
+                pass
+            original = {name: (source / name).read_bytes() for name in
+                        ['before.json', 'intent.json', 'verified.json']}
+            digest = m.legacy_journal_digest(source)
+            result = m.register_legacy_journal(
+                plan, source, expected_legacy_sha256=digest,
+                api=f.FakeAPI(plan, before, after, exact), lease=f.FakeLease(),
+                declaration={'mode': 'none'})
+            self.assertTrue(result['read_only'])
+            target = m.canonical_journal_path(plan)
+            self.assertEqual((target / 'before.json').read_bytes(), original['before.json'])
+            self.assertEqual((target / 'intent.json').read_bytes(), original['intent.json'])
+            self.assertEqual((target / 'legacy-verified.json').read_bytes(), original['verified.json'])
+            self.assertFalse((target / 'verified.json').exists())
+            self.assertEqual(stat.S_IMODE((target / 'legacy-verified.json').stat().st_mode), 0o600)
+            api = f.FakeAPI(plan, before, after, exact); api.created = True
+            for _ in range(2):
+                result = m.publish(plan, before, api, f.FakeLease(), declaration={'mode': 'none'})
+                self.assertFalse(result['wrote'])
+            self.assertEqual(api.posts, 0)
+            self.assertTrue((target / 'verified.json').exists())
+            self.assertEqual((target / 'legacy-verified.json').read_bytes(), verified_raw)
+            (target / 'legacy-verified.json').write_bytes(b'{"verified": false}')
+            with self.assertRaisesRegex(m.ValidationError, 'integrity mismatch'):
+                m.publish(plan, before, api, f.FakeLease(), declaration={'mode': 'none'})
+            self.assertEqual(api.posts, 0)
+
+
+
+    def test_receipt_cannot_omit_imported_verified_archive(self):
+        m = load(SCRIPT, 'legacy_receipt_contract'); f = load(FIXTURE, 'legacy_receipt_fixture')
+        plan, before, after, exact = f.publication_fixture()
+        with tempfile.TemporaryDirectory(dir='/opt/data/cache/scratch') as td:
+            m._journal.root_for = lambda anchor: Path(td) / 'native-review-journals'
+            source = Path(td) / 'legacy'
+            with m.Journal(source) as journal:
+                journal.write('before.json', before)
+                journal.write('intent.json', m._expected_intent(plan, before))
+                journal.write('verified.json', {'verified': True})
+            m.register_legacy_journal(plan, source, expected_legacy_sha256=m.legacy_journal_digest(source),
+                api=f.FakeAPI(plan, before, after, exact), lease=f.FakeLease(), declaration={'mode': 'none'})
+            target = m.canonical_journal_path(plan)
+            receipt = json.loads((target / 'registration.json').read_text())
+            receipt['imported_files'].pop()
+            with m.Journal(target) as journal:
+                journal.write('registration.json', receipt)
+                with self.assertRaises(m.ValidationError): m._check_registration(journal, plan)
+
+if __name__ == '__main__':
+    unittest.main()