feat(review): add guarded native review publication adapter
This commit is contained in:
@@ -0,0 +1,569 @@
|
||||
import copy
|
||||
import importlib.util
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import time
|
||||
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'
|
||||
SPEC_SCRIPT = ROOT / 'harness/agents/spec/skills/gitea-implementation-spec-review/scripts/review_support.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
|
||||
|
||||
|
||||
support = load(SPEC_SCRIPT, 'spec_support_fixture')
|
||||
|
||||
|
||||
def plan_fixture():
|
||||
target = {'owner': 'acme', 'repository': 'repo', 'issue': 3, 'pull': 7,
|
||||
'head': 'a' * 40, 'base': 'c' * 40}
|
||||
digest = __import__('hashlib').sha256(b'approved contract').hexdigest()
|
||||
marker = ('<!-- gitea-agent:code-review/v1 owner=acme repository=repo issue=3 pr=7 '
|
||||
f'head={target["head"]} contract-sha256={digest} -->')
|
||||
return {'schema': 'gitea-agent.code-review-publication/v1', 'target': target,
|
||||
'contract_sha256': digest, 'reviewer': {'id': 42, 'login': 'gitea-review-agent'},
|
||||
'spec_reviewer': {'id': 5, 'login': 'gitea-spec-agent'}, 'spec_review_id': 9, 'spec_paths': ['src/a.py'],
|
||||
'verdict': 'approve', 'body': marker + '\n- Verdict: `approve`\n검토 완료.\n',
|
||||
'provider_contract': {'supersede_same_author': True, 'retouch_dismissed': True,
|
||||
'unordered_commit_files': True,
|
||||
'evidence': ['saved observed Gitea projection and ordering']}}
|
||||
|
||||
|
||||
def publication_fixture():
|
||||
p = plan_fixture(); t = p['target']
|
||||
spec_result = {'schema': support.SCHEMA,
|
||||
'target': {k: t[k] for k in ['owner', 'repository', 'issue', 'pull', 'head']},
|
||||
'contract_sha256': p['contract_sha256'], 'verdict': 'approve',
|
||||
'criteria': [{'criterion': 'REQ-1', 'contract': '계약',
|
||||
'implementation_paths': ['src/a.py'], 'test_paths': [],
|
||||
'evidence': '직접 검증', 'result': 'satisfied', 'reason': '충족'}],
|
||||
'scope': {'missing': [], 'excess': [], 'spec_conflicts': []}, 'conclusion': '충족'}
|
||||
url = 'https://gitea.jsfamily.quest/acme/repo/pulls/7'
|
||||
old_time = '2026-10-04T00:00:00Z'; new_time = '2026-10-04T00:01:00Z'
|
||||
def review(rid, user, body, head, state):
|
||||
return {'id': rid, 'user': copy.deepcopy(user), 'body': body, 'commit_id': head, 'state': state,
|
||||
'official': True, 'dismissed': False, 'stale': head != t['head'],
|
||||
'submitted_at': old_time, 'updated_at': old_time, 'comments_count': 0,
|
||||
'pull_request_url': url}
|
||||
sr = review(9, p['spec_reviewer'], support.render_review(spec_result, {'src/a.py'}),
|
||||
t['head'], 'APPROVED')
|
||||
previous = review(8, p['reviewer'], 'old review body', 'd' * 40, 'REQUEST_CHANGES')
|
||||
s = {'identity': copy.deepcopy(p['reviewer']), 'spec_user': copy.deepcopy(p['spec_reviewer']),
|
||||
'repository': {'full_name': 'acme/repo', 'default_branch': 'main'},
|
||||
'issue': {'number': 3, 'body': 'approved contract', 'state': 'open',
|
||||
'updated_at': old_time, 'repository': {'full_name': 'acme/repo'}},
|
||||
'pr': {'number': 7, 'html_url': url, 'body': 'Closes #3', 'user': {'id': 99},
|
||||
'state': 'open', 'merged': False, 'updated_at': old_time,
|
||||
'review_comments': 2, 'requested_reviewers': [],
|
||||
'base': {'sha': t['base'], 'ref': 'main', 'repo': {'full_name': 'acme/repo'}},
|
||||
'head': {'sha': t['head'], 'ref': 'feature', 'repo': {'full_name': 'acme/repo'}}},
|
||||
'files': [{'filename': 'src/a.py', 'status': 'modified'}],
|
||||
'commits': [{'sha': t['head'], 'files': [{'filename': 'src/a.py', 'status': 'modified'},
|
||||
{'filename': 'docs/x.md', 'status': 'added'}]}],
|
||||
'reviews': [previous, sr], 'review_comments': {'8': [], '9': []},
|
||||
'issue_comments': [], 'pr_comments': [], 'statuses': [], 'pagination': {}}
|
||||
exact = review(10, p['reviewer'], p['body'], t['head'], 'APPROVED')
|
||||
exact['submitted_at'] = exact['updated_at'] = new_time
|
||||
after = copy.deepcopy(s); after['reviews'].append(exact); after['review_comments']['10'] = []
|
||||
after['pr']['updated_at'] = new_time; after['pr']['review_comments'] += 1
|
||||
for snap in [s, after]:
|
||||
for name in ['files', 'commits', 'reviews', 'issue_comments', 'pr_comments', 'statuses']:
|
||||
snap['pagination'][name] = [{'page': 1, 'count': len(snap[name]),
|
||||
'declared_total': len(snap[name]), 'link': None}]
|
||||
for rid, comments in snap['review_comments'].items():
|
||||
snap['pagination']['review_comments_' + rid] = [{'page': 1, 'count': len(comments),
|
||||
'declared_total': len(comments), 'link': None}]
|
||||
return p, s, after, copy.deepcopy(exact)
|
||||
|
||||
|
||||
class FakeAPI:
|
||||
def __init__(self, plan, before, after, exact):
|
||||
self.plan = plan; self.before = before; self.after = after; self.exact = exact
|
||||
self.created = False; self.posts = 0; self.timeout = False; self.apply = True
|
||||
self.events = []
|
||||
|
||||
def identity(self):
|
||||
self.events.append('identity')
|
||||
return copy.deepcopy(self.plan['reviewer'])
|
||||
|
||||
def snapshot(self, target):
|
||||
self.events.append('snapshot')
|
||||
return copy.deepcopy(self.after if self.created else self.before)
|
||||
|
||||
def create_review(self, target, payload):
|
||||
self.events.append('create'); self.posts += 1
|
||||
if self.apply: self.created = True
|
||||
if self.timeout: raise TimeoutError('secret transport detail must not escape')
|
||||
return copy.deepcopy(self.exact)
|
||||
|
||||
def read_review(self, target, rid):
|
||||
self.events.append('exact')
|
||||
return copy.deepcopy(self.exact)
|
||||
|
||||
|
||||
class FakeLease:
|
||||
def __init__(self): self.valid = True; self.checks = 0
|
||||
def check(self, plan):
|
||||
self.checks += 1
|
||||
if not self.valid: raise support.ValidationError('expired lease')
|
||||
|
||||
|
||||
class NativeReviewTests(unittest.TestCase):
|
||||
def module(self):
|
||||
self.assertTrue(SCRIPT.is_file(), 'shared native review adapter is missing')
|
||||
return load(SCRIPT, 'native_review_test')
|
||||
|
||||
def test_payload_is_exact_bound_native_create(self):
|
||||
m = self.module()
|
||||
p = plan_fixture()
|
||||
self.assertEqual(m.build_payload(p), {'commit_id': p['target']['head'],
|
||||
'body': p['body'], 'event': 'APPROVED', 'comments': []})
|
||||
bad = copy.deepcopy(p); bad['target']['pull'] = True
|
||||
with self.assertRaises(m.ValidationError):
|
||||
m.build_payload(bad)
|
||||
bad = copy.deepcopy(p); bad['body'] = bad['body'].replace('a' * 40, 'd' * 40)
|
||||
with self.assertRaises(m.ValidationError):
|
||||
m.build_payload(bad)
|
||||
def test_verify_exact_target_and_bound_provider_changes(self):
|
||||
m = self.module()
|
||||
self.assertTrue(callable(getattr(m, 'verify_publication', None)), 'readback verifier is missing')
|
||||
p, before, after, exact = publication_fixture()
|
||||
previous = after['reviews'][0]
|
||||
previous.update(official=False, dismissed=True, updated_at=exact['submitted_at'])
|
||||
after['commits'][0]['files'].reverse()
|
||||
result = m.verify_publication(p, before, after, exact)
|
||||
self.assertEqual(result['review_id'], 10)
|
||||
self.assertEqual(result['projected_review_ids'], [8])
|
||||
self.assertTrue(result['commit_file_order_normalized'])
|
||||
for mutation in ['head', 'body', 'unrelated_review', 'commit_file']:
|
||||
bad = copy.deepcopy(after)
|
||||
if mutation == 'head': bad['pr']['head']['sha'] = 'e' * 40
|
||||
elif mutation == 'body': bad['reviews'][-1]['body'] += ' altered'
|
||||
elif mutation == 'unrelated_review': bad['reviews'][1]['body'] += ' altered'
|
||||
else: bad['commits'][0]['files'][0]['status'] = 'removed'
|
||||
with self.subTest(mutation=mutation), self.assertRaises(m.ValidationError):
|
||||
m.verify_publication(p, before, bad, exact)
|
||||
def test_publish_once_and_reconcile_durable_attempt_without_reposting(self):
|
||||
m = self.module()
|
||||
self.assertTrue(callable(getattr(m, 'publish', None)), 'shared publisher is missing')
|
||||
p, before, after, exact = publication_fixture()
|
||||
api = FakeAPI(p, before, after, exact); lease = FakeLease()
|
||||
with tempfile.TemporaryDirectory() as td:
|
||||
result = m.publish(p, before, api, lease, td)
|
||||
self.assertEqual(result['mode'], 'created')
|
||||
self.assertEqual(api.posts, 1)
|
||||
self.assertEqual(m.publish(p, before, api, lease, td)['mode'], 'no_op')
|
||||
self.assertEqual(api.posts, 1)
|
||||
self.assertNotIn('body', json.loads((Path(td) / 'intent.json').read_text()))
|
||||
for applies in [True, False]:
|
||||
api = FakeAPI(p, before, after, exact); api.timeout = True; api.apply = applies
|
||||
with self.subTest(applies=applies), tempfile.TemporaryDirectory() as td:
|
||||
if applies:
|
||||
self.assertEqual(m.publish(p, before, api, lease, td)['mode'], 'reconciled')
|
||||
self.assertEqual(m.publish(p, before, api, lease, td)['mode'], 'no_op')
|
||||
else:
|
||||
for _ in range(2):
|
||||
with self.assertRaises(m.UncertainPublication) as error:
|
||||
m.publish(p, before, api, lease, td)
|
||||
self.assertNotIn('secret transport detail', str(error.exception))
|
||||
self.assertEqual(api.posts, 1)
|
||||
def test_api_adapter_checks_identity_first_and_collects_all_pages(self):
|
||||
m = self.module()
|
||||
self.assertTrue(callable(getattr(m, 'ApiAdapter', None)), 'REST bridge is missing')
|
||||
p, before, after, exact = publication_fixture()
|
||||
calls = []
|
||||
def request(path, payload=None):
|
||||
calls.append((path, payload))
|
||||
if path == '/user': return p['reviewer'], {}, 200
|
||||
if path == '/users/gitea-spec-agent': return p['spec_reviewer'], {}, 200
|
||||
if path == '/repos/acme/repo': return before['repository'], {}, 200
|
||||
if path == '/repos/acme/repo/issues/3': return before['issue'], {}, 200
|
||||
if path == '/repos/acme/repo/pulls/7': return before['pr'], {}, 200
|
||||
from urllib.parse import parse_qs, urlsplit
|
||||
parsed = urlsplit(path); page = int(parse_qs(parsed.query).get('page', ['1'])[0])
|
||||
for name, suffix in [('files', '/pulls/7/files'), ('commits', '/pulls/7/commits'),
|
||||
('reviews', '/pulls/7/reviews'), ('issue_comments', '/issues/3/comments'),
|
||||
('pr_comments', '/issues/7/comments'), ('statuses', '/statuses/' + p['target']['head'])]:
|
||||
if parsed.path == '/repos/acme/repo' + suffix:
|
||||
if name == 'reviews' and payload is not None: return exact, {}, 201
|
||||
return before[name], {'x-total-count': str(len(before[name]))}, 200
|
||||
if parsed.path.endswith('/comments'): return [], {'X-Total-Count': '0'}, 200
|
||||
raise AssertionError('unexpected request')
|
||||
adapter = m.ApiAdapter(request, origin=m.ORIGIN + '/api/v1',
|
||||
credential_source='$HERMES_HOME/.env:GITEA_TOKEN')
|
||||
snapshot = adapter.snapshot(p['target'])
|
||||
m.validate_snapshot(p, snapshot)
|
||||
self.assertEqual(calls[0][0], '/user')
|
||||
self.assertEqual(adapter.create_review(p['target'], m.build_payload(p))['id'], 10)
|
||||
self.assertEqual(calls[-1][1]['comments'], [])
|
||||
with self.assertRaises(m.ValidationError):
|
||||
m.ApiAdapter(request, origin='https://other.test/api/v1', credential_source='unknown')
|
||||
def test_held_lease_checks_real_registered_worktree_and_expiry(self):
|
||||
import fcntl
|
||||
m = self.module()
|
||||
self.assertTrue(callable(getattr(m, 'HeldIssueLease', None)), 'real Issue lease guard is missing')
|
||||
with tempfile.TemporaryDirectory() as td:
|
||||
root = Path(td); policy = root / 'harness/operations/repository-cache.md'
|
||||
policy.parent.mkdir(parents=True); policy.write_text('canonical cache fixture')
|
||||
rs = root / 'repositories'; owner = rs / 'acme'; cache = owner / 'repo'; locks = cache / 'locks'
|
||||
for directory in [rs, owner, cache, locks]: directory.mkdir(mode=0o700)
|
||||
bare = cache / 'repository.git'; seed = root / 'seed'
|
||||
def git(*args):
|
||||
out = subprocess.run(['git', '-c', 'core.hooksPath=/dev/null', *map(str, args)], capture_output=True, text=True,
|
||||
env={'PATH': os.environ['PATH'], 'GIT_CONFIG_GLOBAL': '/dev/null', 'GIT_CONFIG_NOSYSTEM': '1'})
|
||||
self.assertEqual(out.returncode, 0, out.stderr)
|
||||
return out.stdout.strip()
|
||||
git('init', '--bare', bare); git('init', seed)
|
||||
git('-C', seed, 'config', 'user.name', 'Synthetic'); git('-C', seed, 'config', 'user.email', 'fixture@example.test')
|
||||
(seed / 'a.txt').write_text('synthetic')
|
||||
git('-C', seed, 'add', 'a.txt'); git('-C', seed, 'commit', '-m', 'fixture')
|
||||
git('--git-dir=' + str(bare), 'fetch', seed, 'HEAD:refs/heads/main')
|
||||
wt = cache / 'worktrees/issue-3/checkout'; wt.parent.mkdir(parents=True, mode=0o700)
|
||||
git('--git-dir=' + str(bare), 'worktree', 'add', '--detach', '--lock', wt, 'main')
|
||||
head = git('-C', wt, 'rev-parse', 'HEAD')
|
||||
p = plan_fixture(); p['target']['head'] = head; p['body'] = p['body'].replace('a' * 40, head)
|
||||
state = wt.parent / 'state.json'
|
||||
state.write_text(json.dumps({'schema': 'gitea-agent.worktree/v1', 'owner': 'acme', 'repository': 'repo',
|
||||
'issue': 3, 'created_by_role': 'gitea-impl-agent', 'mode': 'detached'}))
|
||||
state.chmod(0o600)
|
||||
path = locks / 'issue-3.lock'; fd = os.open(path, os.O_RDWR | os.O_CREAT, 0o600)
|
||||
try:
|
||||
fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
||||
lease = m.HeldIssueLease(fd, policy, time.monotonic() + 30)
|
||||
lease.check(p)
|
||||
fcntl.flock(fd, fcntl.LOCK_UN)
|
||||
with self.assertRaises(m.ValidationError): lease.check(p)
|
||||
self.assertNotIn('lock:', Path(f'/proc/self/fdinfo/{fd}').read_text())
|
||||
fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
||||
from unittest.mock import patch
|
||||
fakebin = root / 'fakebin'; fakebin.mkdir()
|
||||
marker = root / 'path-git-executed'
|
||||
fakegit = fakebin / 'git'
|
||||
fakegit.write_text(f'#!/bin/sh\n/usr/bin/touch "{marker}"\nexit 1\n'); fakegit.chmod(0o700)
|
||||
with patch.dict(os.environ, {'PATH': str(fakebin)}): lease.check(p)
|
||||
self.assertFalse(marker.exists())
|
||||
lease.deadline_monotonic = time.monotonic() - 1
|
||||
with self.assertRaises(m.ValidationError): lease.check(p)
|
||||
lease.deadline_monotonic = time.monotonic() + 30
|
||||
(wt / 'a.txt').write_text('dirty')
|
||||
with self.assertRaises(m.ValidationError): lease.check(p)
|
||||
finally:
|
||||
os.close(fd)
|
||||
def test_cli_preflight_and_saved_wire_verification_are_read_only(self):
|
||||
m = self.module(); p, before, after, exact = publication_fixture()
|
||||
with tempfile.TemporaryDirectory() as td:
|
||||
root = Path(td)
|
||||
for name, value in [('plan', p), ('before', before), ('after', after), ('exact', exact)]:
|
||||
(root / (name + '.json')).write_text(json.dumps(value, ensure_ascii=False))
|
||||
base = [sys.executable, str(SCRIPT)]
|
||||
env = {'PATH': os.environ['PATH'], 'PYTHONDONTWRITEBYTECODE': '1'}
|
||||
pre = subprocess.run(base + ['preflight', '--plan', str(root / 'plan.json'),
|
||||
'--baseline', str(root / 'before.json')], capture_output=True, text=True, env=env)
|
||||
self.assertEqual(pre.returncode, 0, pre.stderr)
|
||||
self.assertIn('"ready": true', pre.stdout, 'read-only preflight CLI is missing')
|
||||
verified = subprocess.run(base + ['verify', '--plan', str(root / 'plan.json'),
|
||||
'--before', str(root / 'before.json'), '--after', str(root / 'after.json'),
|
||||
'--exact', str(root / 'exact.json')], capture_output=True, text=True, env=env)
|
||||
self.assertEqual(verified.returncode, 0, verified.stderr)
|
||||
self.assertTrue(json.loads(verified.stdout)['read_only'])
|
||||
self.assertEqual(json.loads(verified.stdout)['review_id'], 10)
|
||||
unsupported = subprocess.run(base + ['publish'], capture_output=True, text=True, env=env)
|
||||
self.assertNotEqual(unsupported.returncode, 0)
|
||||
def test_spec_approval_may_reference_unchanged_surrounding_test_paths(self):
|
||||
m = self.module(); p, before, after, exact = publication_fixture()
|
||||
p['spec_paths'] = ['src/a.py', 'tests/existing.py']
|
||||
spec_result = support.parse_review(before['reviews'][1]['body'], {'src/a.py'})
|
||||
spec_result['criteria'][0]['test_paths'] = ['tests/existing.py']
|
||||
before['reviews'][1]['body'] = support.render_review(spec_result, set(p['spec_paths']))
|
||||
try:
|
||||
m.validate_snapshot(p, before)
|
||||
except m.ValidationError:
|
||||
self.fail('spec approval must use its declared scope, not only changed files')
|
||||
def test_malformed_historical_actor_and_missing_exact_review_flags_fail_closed(self):
|
||||
m = self.module(); p, before, after, exact = publication_fixture()
|
||||
bad = copy.deepcopy(before); bad['reviews'][0]['user'] = {'id': True, 'login': 'x'}
|
||||
with self.assertRaises(m.ValidationError):
|
||||
m.validate_snapshot(p, bad)
|
||||
p, before, after, exact = publication_fixture()
|
||||
del exact['official']; del after['reviews'][-1]['official']
|
||||
with self.assertRaises(m.ValidationError):
|
||||
m.verify_publication(p, before, after, exact)
|
||||
def test_invalid_plan_bindings_are_not_normalized(self):
|
||||
m = self.module()
|
||||
mutations = [lambda p: p['target'].__setitem__('issue', True),
|
||||
lambda p: p['target'].__setitem__('head', 'A' * 40),
|
||||
lambda p: p['target'].__setitem__('owner', 'acme/other'),
|
||||
lambda p: p.__setitem__('verdict', 'blocked'),
|
||||
lambda p: p.__setitem__('body', p['body'] + '- Verdict: `approve`\n'),
|
||||
lambda p: p['provider_contract'].__setitem__('evidence', []),
|
||||
lambda p: p.__setitem__('spec_paths', ['../secret']),
|
||||
lambda p: p.__setitem__('spec_paths', ['src/a.py', 'src/a.py']),
|
||||
lambda p: p['reviewer'].__setitem__('id', p['spec_reviewer']['id'])]
|
||||
for i, mutate in enumerate(mutations):
|
||||
with self.subTest(case=i):
|
||||
p = plan_fixture(); mutate(p)
|
||||
with self.assertRaises(m.ValidationError): m.build_payload(p)
|
||||
|
||||
def test_wrong_role_and_expired_lease_prevent_any_post(self):
|
||||
m = self.module()
|
||||
for kind in ['wrong_role', 'expired']:
|
||||
with self.subTest(kind=kind), tempfile.TemporaryDirectory() as td:
|
||||
p, before, after, exact = publication_fixture(); api = FakeAPI(p, before, after, exact); lease = FakeLease()
|
||||
if kind == 'expired': lease.valid = False
|
||||
else: api.identity = lambda: p['spec_reviewer']
|
||||
with self.assertRaises(m.ValidationError): m.publish(p, before, api, lease, td)
|
||||
self.assertEqual(api.posts, 0)
|
||||
self.assertFalse((Path(td) / 'intent.json').exists())
|
||||
|
||||
def test_fresh_head_contract_and_source_linkage_drift_prevent_post(self):
|
||||
m = self.module()
|
||||
changes = [lambda s: s['pr']['head'].__setitem__('sha', 'b' * 40),
|
||||
lambda s: s['issue'].__setitem__('body', 'changed contract'),
|
||||
lambda s: s['pr'].__setitem__('body', 'Closes #4'),
|
||||
lambda s: s['pr']['base'].__setitem__('ref', 'other'),
|
||||
lambda s: s['pr'].__setitem__('merged', True)]
|
||||
for i, change in enumerate(changes):
|
||||
with self.subTest(case=i), tempfile.TemporaryDirectory() as td:
|
||||
p, before, after, exact = publication_fixture(); api = FakeAPI(p, copy.deepcopy(before), after, exact)
|
||||
change(api.before)
|
||||
with self.assertRaises(m.ValidationError): m.publish(p, before, api, FakeLease(), td)
|
||||
self.assertEqual(api.posts, 0)
|
||||
|
||||
def test_pagination_total_duplicate_and_truncation_are_hard_errors(self):
|
||||
m = self.module()
|
||||
changes = [lambda s: s['pagination']['files'][0].__setitem__('declared_total', 2),
|
||||
lambda s: s['pagination']['reviews'][0].__setitem__('link', '<x>; rel="next"'),
|
||||
lambda s: s['pagination']['commits'][0].__setitem__('count', True),
|
||||
lambda s: s['files'].append(copy.deepcopy(s['files'][0])),
|
||||
lambda s: s['review_comments'].pop('8')]
|
||||
for i, change in enumerate(changes):
|
||||
with self.subTest(case=i):
|
||||
p, before, after, exact = publication_fixture(); change(before)
|
||||
with self.assertRaises(m.ValidationError): m.validate_snapshot(p, before)
|
||||
|
||||
def test_spec_approval_body_head_role_and_state_are_bound(self):
|
||||
m = self.module()
|
||||
changes = [lambda s: s['reviews'][1].__setitem__('state', 'REQUEST_CHANGES'),
|
||||
lambda s: s['reviews'][1].__setitem__('commit_id', 'b' * 40),
|
||||
lambda s: s['reviews'][1].__setitem__('dismissed', True),
|
||||
lambda s: s['reviews'][1].__setitem__('body', s['reviews'][1]['body'].replace('충족', '임의 변경', 1)),
|
||||
lambda s: s['reviews'][1]['user'].__setitem__('id', 23)]
|
||||
for i, change in enumerate(changes):
|
||||
with self.subTest(case=i):
|
||||
p, before, after, exact = publication_fixture(); change(before)
|
||||
with self.assertRaises(m.ValidationError): m.validate_snapshot(p, before)
|
||||
|
||||
def test_provider_projection_requires_explicit_bound_evidence(self):
|
||||
m = self.module()
|
||||
for kind in ['disabled', 'wrong_timestamp', 'wrong_author']:
|
||||
with self.subTest(kind=kind):
|
||||
p, before, after, exact = publication_fixture()
|
||||
after['reviews'][0].update(official=False, dismissed=True, updated_at=exact['submitted_at'])
|
||||
if kind == 'disabled': p['provider_contract']['supersede_same_author'] = False
|
||||
elif kind == 'wrong_timestamp': after['reviews'][0]['updated_at'] = '2026-10-04T00:02:00Z'
|
||||
else:
|
||||
before['reviews'][0]['user'] = {'id': 72, 'login': 'other'}
|
||||
after['reviews'][0]['user'] = {'id': 72, 'login': 'other'}
|
||||
with self.assertRaises(m.ValidationError): m.verify_publication(p, before, after, exact)
|
||||
|
||||
def test_order_normalization_does_not_mask_file_contents_or_parent_order(self):
|
||||
m = self.module()
|
||||
p, before, after, exact = publication_fixture()
|
||||
after['commits'][0]['files'].reverse()
|
||||
self.assertTrue(m.verify_publication(p, before, after, exact)['commit_file_order_normalized'])
|
||||
p['provider_contract']['unordered_commit_files'] = False
|
||||
with self.assertRaises(m.ValidationError): m.verify_publication(p, before, after, exact)
|
||||
p, before, after, exact = publication_fixture()
|
||||
after['commits'][0]['files'][0]['status'] = 'removed'
|
||||
with self.assertRaises(m.ValidationError): m.verify_publication(p, before, after, exact)
|
||||
p, before, after, exact = publication_fixture()
|
||||
before['commits'][0]['parents'] = [{'sha': 'e' * 40}, {'sha': 'f' * 40}]
|
||||
after['commits'][0]['parents'] = list(reversed(before['commits'][0]['parents']))
|
||||
with self.assertRaises(m.ValidationError): m.verify_publication(p, before, after, exact)
|
||||
|
||||
def test_historical_head_state_body_and_comments_remain_immutable(self):
|
||||
m = self.module()
|
||||
changes = [lambda s: s['reviews'][0].__setitem__('body', 'altered'),
|
||||
lambda s: s['reviews'][0].__setitem__('commit_id', 'e' * 40),
|
||||
lambda s: s['reviews'][0].__setitem__('state', 'APPROVED')]
|
||||
for i, change in enumerate(changes):
|
||||
with self.subTest(case=i):
|
||||
p, before, after, exact = publication_fixture(); change(after)
|
||||
with self.assertRaises(m.ValidationError): m.verify_publication(p, before, after, exact)
|
||||
p, before, after, exact = publication_fixture()
|
||||
for snap in [before, after]:
|
||||
snap['review_comments']['8'] = [{'id': 71, 'body': 'original'}]
|
||||
snap['reviews'][0]['comments_count'] = 1
|
||||
snap['pagination']['review_comments_8'][0].update(count=1, declared_total=1)
|
||||
after['review_comments']['8'][0]['body'] = 'altered'
|
||||
with self.assertRaises(m.ValidationError): m.verify_publication(p, before, after, exact)
|
||||
|
||||
def test_existing_canonical_review_is_write_free_no_op(self):
|
||||
m = self.module(); p, before, after, exact = publication_fixture()
|
||||
api = FakeAPI(p, before, after, exact); api.created = True
|
||||
with tempfile.TemporaryDirectory() as td:
|
||||
result = m.publish(p, after, api, FakeLease(), td)
|
||||
self.assertEqual(result['mode'], 'no_op'); self.assertFalse(result['wrote'])
|
||||
self.assertEqual(api.posts, 0)
|
||||
self.assertFalse((Path(td) / 'intent.json').exists())
|
||||
|
||||
def test_duplicate_or_conflicting_canonical_review_is_not_reposted(self):
|
||||
m = self.module()
|
||||
for kind in ['body', 'duplicate']:
|
||||
with self.subTest(kind=kind), tempfile.TemporaryDirectory() as td:
|
||||
p, before, after, exact = publication_fixture()
|
||||
if kind == 'body': after['reviews'][-1]['body'] += 'other'
|
||||
else:
|
||||
extra = copy.deepcopy(exact); extra['id'] = 11; after['reviews'].append(extra)
|
||||
after['review_comments']['11'] = []
|
||||
after['pagination']['reviews'][0].update(count=4, declared_total=4)
|
||||
after['pagination']['review_comments_11'] = [{'page': 1, 'count': 0, 'declared_total': 0, 'link': None}]
|
||||
api = FakeAPI(p, before, after, exact); api.created = True
|
||||
with self.assertRaises(m.ValidationError): m.publish(p, before, api, FakeLease(), td)
|
||||
self.assertEqual(api.posts, 0)
|
||||
|
||||
def test_readback_failure_and_changed_journal_plan_never_retry_post(self):
|
||||
m = self.module(); p, before, after, exact = publication_fixture(); api = FakeAPI(p, before, after, exact)
|
||||
api.read_review = lambda target, rid: (_ for _ in ()).throw(TimeoutError('DO_NOT_ECHO_TRANSPORT_SECRET'))
|
||||
with tempfile.TemporaryDirectory() as td:
|
||||
for _ in range(2):
|
||||
with self.assertRaises(Exception) as raised: m.publish(p, before, api, FakeLease(), td)
|
||||
self.assertNotIn('DO_NOT_ECHO_TRANSPORT_SECRET', str(raised.exception))
|
||||
self.assertEqual(api.posts, 1)
|
||||
different = copy.deepcopy(p); different['body'] += 'different'
|
||||
with self.assertRaises(m.ValidationError): m.publish(different, before, api, FakeLease(), td)
|
||||
self.assertEqual(api.posts, 1)
|
||||
|
||||
def test_journal_symlinks_hardlinks_and_changed_baseline_are_rejected(self):
|
||||
m = self.module(); p, before, after, exact = publication_fixture()
|
||||
with tempfile.TemporaryDirectory() as td:
|
||||
root = Path(td); link = root / 'link'; link.symlink_to(root, target_is_directory=True)
|
||||
with self.assertRaises(m.ValidationError): m.Journal(link)
|
||||
api = FakeAPI(p, before, after, exact); api.timeout = True; api.apply = False
|
||||
journal = root / 'journal'
|
||||
with self.assertRaises(m.UncertainPublication): m.publish(p, before, api, FakeLease(), journal)
|
||||
saved = json.loads((journal / 'before.json').read_text()); saved['issue']['body'] = 'changed'
|
||||
(journal / 'before.json').write_text(json.dumps(saved))
|
||||
with self.assertRaises(m.ValidationError): m.publish(p, before, api, FakeLease(), journal)
|
||||
self.assertEqual(api.posts, 1)
|
||||
with m.Journal(root / 'hardlinks') as j:
|
||||
j.write('a.json', {'safe': True}); os.link(j.path / 'a.json', j.path / 'b.json')
|
||||
with self.assertRaises(m.ValidationError): j.read('a.json')
|
||||
|
||||
def test_transport_origin_credential_source_and_exception_text_are_guarded(self):
|
||||
m = self.module()
|
||||
with self.assertRaises(m.ValidationError): m.ApiAdapter(lambda *args: None, origin='https://other/api/v1', credential_source='$HERMES_HOME/.env:GITEA_TOKEN')
|
||||
with self.assertRaises(m.ValidationError): m.ApiAdapter(lambda *args: None, origin=m.ORIGIN + '/api/v1', credential_source='other-profile')
|
||||
def bad(*args): raise TimeoutError('DO_NOT_ECHO_TRANSPORT_SECRET')
|
||||
api = m.ApiAdapter(bad, origin=m.ORIGIN + '/api/v1', credential_source='$HERMES_HOME/.env:GITEA_TOKEN')
|
||||
with self.assertRaises(m.ValidationError) as raised: api.identity()
|
||||
self.assertNotIn('DO_NOT_ECHO_TRANSPORT_SECRET', str(raised.exception))
|
||||
self.assertTrue(raised.exception.__suppress_context__)
|
||||
def test_false_lease_result_does_not_authorize_publication(self):
|
||||
m = self.module(); p, before, after, exact = publication_fixture(); api = FakeAPI(p, before, after, exact)
|
||||
lease = FakeLease(); lease.check = lambda plan: False
|
||||
with tempfile.TemporaryDirectory() as td:
|
||||
with self.assertRaises(m.ValidationError): m.publish(p, before, api, lease, td)
|
||||
self.assertEqual(api.posts, 0)
|
||||
def test_live_rest_bridge_over_loopback_publishes_once_then_no_ops(self):
|
||||
import threading
|
||||
from http.server import BaseHTTPRequestHandler, HTTPServer
|
||||
from urllib.request import Request, urlopen
|
||||
m = self.module(); p, before, after, exact = publication_fixture()
|
||||
after['reviews'][0].update(official=False, dismissed=True, updated_at=exact['submitted_at'])
|
||||
after['commits'][0]['files'].reverse()
|
||||
state = {'posted': False, 'posts': 0}
|
||||
class Handler(BaseHTTPRequestHandler):
|
||||
def log_message(self, *args): pass
|
||||
def respond(self, value, status=200, total=None):
|
||||
body = json.dumps(value, ensure_ascii=False).encode()
|
||||
self.send_response(status); self.send_header('Content-Type', 'application/json')
|
||||
if total is not None: self.send_header('X-Total-Count', str(total))
|
||||
self.send_header('Content-Length', str(len(body))); self.end_headers(); self.wfile.write(body)
|
||||
def do_GET(self):
|
||||
path = self.path.split('?')[0]; s = after if state['posted'] else before
|
||||
if path == '/user': return self.respond(p['reviewer'])
|
||||
if path == '/users/gitea-spec-agent': return self.respond(p['spec_reviewer'])
|
||||
prefix = '/repos/acme/repo'; pull = prefix + '/pulls/7'
|
||||
singles = {prefix: s['repository'], prefix + '/issues/3': s['issue'], pull: s['pr']}
|
||||
if path in singles: return self.respond(singles[path])
|
||||
if path == pull + '/reviews/10': return self.respond(exact)
|
||||
arrays = {pull + '/files': s['files'], pull + '/commits': s['commits'], pull + '/reviews': s['reviews'],
|
||||
prefix + '/issues/3/comments': s['issue_comments'], prefix + '/issues/7/comments': s['pr_comments'],
|
||||
prefix + '/statuses/' + p['target']['head']: s['statuses']}
|
||||
for rid, comments in s['review_comments'].items(): arrays[pull + '/reviews/' + rid + '/comments'] = comments
|
||||
if path in arrays: return self.respond(arrays[path], total=len(arrays[path]))
|
||||
self.respond({}, 404)
|
||||
def do_POST(self):
|
||||
self.server.received = json.loads(self.rfile.read(int(self.headers['Content-Length'])))
|
||||
state['posts'] += 1; state['posted'] = True
|
||||
self.respond(exact, 201)
|
||||
server = HTTPServer(('127.0.0.1', 0), Handler)
|
||||
thread = threading.Thread(target=server.serve_forever, daemon=True); thread.start()
|
||||
def request(path, payload=None):
|
||||
wire = None if payload is None else json.dumps(payload, ensure_ascii=False).encode()
|
||||
req = Request('http://127.0.0.1:' + str(server.server_port) + path, data=wire,
|
||||
headers={'Content-Type': 'application/json'})
|
||||
with urlopen(req, timeout=5) as response:
|
||||
return json.loads(response.read()), dict(response.headers), response.status
|
||||
try:
|
||||
# Controlled local fixture transport, not a real role credential/origin assertion.
|
||||
api = m.ApiAdapter(request, origin=m.ORIGIN + '/api/v1', credential_source='$HERMES_HOME/.env:GITEA_TOKEN')
|
||||
baseline = api.snapshot(p['target'])
|
||||
with tempfile.TemporaryDirectory() as td:
|
||||
result = m.publish(p, baseline, api, FakeLease(), td)
|
||||
self.assertEqual(result['mode'], 'created'); self.assertEqual(result['projected_review_ids'], [8])
|
||||
self.assertTrue(result['commit_file_order_normalized'])
|
||||
second = m.publish(p, baseline, api, FakeLease(), td)
|
||||
self.assertEqual(second['mode'], 'no_op'); self.assertFalse(second['wrote'])
|
||||
self.assertEqual(state['posts'], 1); self.assertEqual(server.received, m.build_payload(p))
|
||||
finally:
|
||||
server.shutdown(); server.server_close(); thread.join(timeout=5)
|
||||
self.assertFalse(thread.is_alive())
|
||||
def test_expired_lease_does_not_create_new_journal_directory(self):
|
||||
m = self.module(); p, before, after, exact = publication_fixture()
|
||||
lease = FakeLease(); lease.valid = False
|
||||
with tempfile.TemporaryDirectory() as td:
|
||||
directory = Path(td) / 'not-created'
|
||||
with self.assertRaises(m.ValidationError): m.publish(p, before, FakeAPI(p, before, after, exact), lease, directory)
|
||||
self.assertFalse(directory.exists())
|
||||
|
||||
def test_journal_intent_fields_are_validated_on_resume(self):
|
||||
m = self.module()
|
||||
for field in ['schema', 'head', 'event', 'body_sha256']:
|
||||
with self.subTest(field=field), tempfile.TemporaryDirectory() as td:
|
||||
p, before, after, exact = publication_fixture(); api = FakeAPI(p, before, after, exact)
|
||||
api.timeout = True; api.apply = False
|
||||
with self.assertRaises(m.UncertainPublication): m.publish(p, before, api, FakeLease(), td)
|
||||
path = Path(td) / 'intent.json'; intent = json.loads(path.read_text()); intent[field] = 'corrupt'
|
||||
path.write_text(json.dumps(intent)); api.created = True
|
||||
with self.assertRaises(m.ValidationError): m.publish(p, before, api, FakeLease(), td)
|
||||
self.assertEqual(api.posts, 1)
|
||||
|
||||
def test_issue_repository_mismatch_is_not_accepted(self):
|
||||
m = self.module(); p, before, after, exact = publication_fixture()
|
||||
before['issue']['repository']['full_name'] = 'other/repo'
|
||||
with self.assertRaises(m.ValidationError): m.validate_snapshot(p, before)
|
||||
def test_fork_head_and_nonofficial_native_states_preserve_accepted_contract(self):
|
||||
m = self.module(); p, before, after, exact = publication_fixture()
|
||||
for snapshot in [before, after]:
|
||||
snapshot['pr']['head']['repo']['full_name'] = 'contributor/repo'
|
||||
snapshot['reviews'][1]['official'] = False
|
||||
after['reviews'][-1]['official'] = exact['official'] = False
|
||||
self.assertTrue(m.verify_publication(p, before, after, exact)['verified'])
|
||||
after['pr']['head']['repo']['full_name'] = 'different/repo'
|
||||
with self.assertRaises(m.ValidationError): m.verify_publication(p, before, after, exact)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user