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()