221 lines
14 KiB
Python
221 lines
14 KiB
Python
import hashlib
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
import stat
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from harness.tests._support.paths import ROOT
|
|
SCRIPT = ROOT / 'harness/agents/review/skills/gitea-code-review/scripts/native_review.py'
|
|
FIXTURES = ROOT / 'harness/tests/_support/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_inventory_rejects_unknown_artifact_before_import(self):
|
|
m = load(SCRIPT, 'legacy_inventory_contract')
|
|
f = load(FIXTURES, '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(FIXTURES, '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(FIXTURES, '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)
|
|
|
|
class LegacyJournalProcessContractTests(unittest.TestCase):
|
|
def test_valid_legacy_registration_is_read_only_and_preserves_source_bytes(self):
|
|
p, before, after, exact = self.plan_data(); source = self._legacy_files(p, before)
|
|
original = {name: (source / name).read_bytes() for name in ['before.json', 'intent.json', 'verified.json']}
|
|
modes = {name: stat.S_IMODE((source / name).stat().st_mode) for name in original}
|
|
api = self.fixtures.FakeAPI(p, before, after, exact)
|
|
result = self.m.register_legacy_journal(p, source, expected_legacy_sha256=self._legacy_digest(source),
|
|
api=api, lease=self.fixtures.FakeLease(), declaration=self.declaration())
|
|
self.assertTrue(result['read_only']); self.assertEqual(api.posts, 0)
|
|
for name, data in original.items():
|
|
self.assertEqual((source / name).read_bytes(), data)
|
|
self.assertEqual(stat.S_IMODE((source / name).stat().st_mode), modes[name])
|
|
target = self.m.canonical_journal_path(p)
|
|
self.assertEqual((target / 'before.json').read_bytes(), original['before.json'])
|
|
self.assertFalse((target / 'import.pending.json').exists())
|
|
def test_legacy_source_mutation_digest_mismatch_rejects_without_post_or_target_intent(self):
|
|
p, before, after, exact = self.plan_data(); source = self._legacy_files(p, before)
|
|
digest = self._legacy_digest(source)
|
|
mutated = json.loads((source / 'before.json').read_text()); mutated['issue']['body'] = 'changed'
|
|
(source / 'before.json').write_text(json.dumps(mutated, separators=(',', ':')))
|
|
api = self.fixtures.FakeAPI(p, before, after, exact)
|
|
with self.assertRaises(self.m.ValidationError):
|
|
self.m.register_legacy_journal(p, source, expected_legacy_sha256=digest, api=api,
|
|
lease=self.fixtures.FakeLease(), declaration=self.declaration())
|
|
self.assertEqual(api.posts, 0)
|
|
self.assertFalse((self.m.canonical_journal_path(p) / 'intent.json').exists())
|
|
|
|
def test_legacy_extra_intent_fields_union_rejects_without_post_or_target_intent(self):
|
|
p, before, after, exact = self.plan_data(); source = self._legacy_files(p, before)
|
|
intent = json.loads((source / 'intent.json').read_text()); intent['unexpected'] = True
|
|
(source / 'intent.json').write_text(json.dumps(intent, separators=(',', ':')))
|
|
api = self.fixtures.FakeAPI(p, before, after, exact)
|
|
with self.assertRaises(self.m.ValidationError):
|
|
self.m.register_legacy_journal(p, source, expected_legacy_sha256=self._legacy_digest(source), api=api,
|
|
lease=self.fixtures.FakeLease(), declaration=self.declaration())
|
|
self.assertEqual(api.posts, 0)
|
|
self.assertFalse((self.m.canonical_journal_path(p) / 'intent.json').exists())
|
|
|
|
m = load(SCRIPT, 'legacy_contract_publisher')
|
|
f = load(FIXTURES, '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_symlink_hardlink_and_duplicate_json_are_rejected(self):
|
|
p, before, after, exact = self.plan_data()
|
|
source = self._legacy_files(p, before)
|
|
(source / 'before.json').unlink(); (source / 'before.json').symlink_to(Path(self.tmp.name) / 'elsewhere')
|
|
with self.assertRaises(self.m.ValidationError):
|
|
self.m.register_legacy_journal(p, source, expected_legacy_sha256='0' * 64, api=self.fixtures.FakeAPI(p, before, after, exact),
|
|
lease=self.fixtures.FakeLease(), declaration=self.declaration())
|
|
source = self._legacy_files(p, before)
|
|
os.link(source / 'before.json', source / 'duplicate.json')
|
|
with self.assertRaises(self.m.ValidationError):
|
|
self.m.register_legacy_journal(p, source, expected_legacy_sha256=self._legacy_digest(source), api=self.fixtures.FakeAPI(p, before, after, exact),
|
|
lease=self.fixtures.FakeLease(), declaration=self.declaration())
|
|
def test_registration_rejects_existing_canonical_binding_without_post(self):
|
|
p, before, after, exact = self.plan_data(); source = self._legacy_files(p, before)
|
|
target = self.m.canonical_journal_path(p)
|
|
target.parent.mkdir(parents=True, exist_ok=True); target.mkdir(mode=0o700)
|
|
(target / 'intent.json').write_text('{}'); (target / 'intent.json').chmod(0o600)
|
|
api = self.fixtures.FakeAPI(p, before, after, exact)
|
|
with self.assertRaises(self.m.ValidationError):
|
|
self.m.register_legacy_journal(p, source, expected_legacy_sha256=self._legacy_digest(source), api=api,
|
|
lease=self.fixtures.FakeLease(), declaration=self.declaration())
|
|
self.assertEqual(api.posts, 0)
|
|
def setUp(self):
|
|
self.m = load(SCRIPT, 'native_review_process_' + str(id(self)))
|
|
self.fixtures = load(FIXTURES, 'native_review_process_fixture_' + str(id(self)))
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.tmp.cleanup)
|
|
self.root = Path(self.tmp.name) / 'native-review-journals'
|
|
self.m._journal.root_for = lambda anchor, root=self.root: root
|
|
def plan_data(self):
|
|
p, before, after, exact = self.fixtures.publication_fixture()
|
|
return p, before, after, exact
|
|
def declaration(self):
|
|
return {'mode': 'none'}
|
|
|
|
def _legacy_files(self, p, before, *, verified=True):
|
|
directory = Path(tempfile.mkdtemp(prefix='legacy-', dir=self.tmp.name))
|
|
directory.chmod(0o700)
|
|
canonical = self.m.canonical
|
|
intent = {'schema': 'gitea-agent.native-review-intent/v1',
|
|
'fingerprint': hashlib.sha256(self.m.canonical(p)).hexdigest(),
|
|
'before_sha256': hashlib.sha256(self.m.canonical(before)).hexdigest(),
|
|
'head': p['target']['head'], 'event': 'APPROVED',
|
|
'body_sha256': hashlib.sha256(p['body'].encode()).hexdigest()}
|
|
with self.m.Journal(directory) as journal:
|
|
journal.write('before.json', before)
|
|
journal.write('intent.json', intent)
|
|
if verified:
|
|
journal.write('verified.json', {'verified': True})
|
|
return directory
|
|
def _legacy_digest(self, directory):
|
|
helper = getattr(self.m, 'legacy_journal_digest', None)
|
|
self.assertTrue(callable(helper), 'public legacy journal digest helper is required')
|
|
return helper(directory)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|