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 = ('') 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'}, declaration={'mode': 'none'}), 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') m = load(SCRIPT, 'native_review_test_' + str(id(self))) # Every test gets a private registry root; production has no test-only # root override, so isolate through the journal module seam. private_root = tempfile.TemporaryDirectory() self.addCleanup(private_root.cleanup) root = Path(private_root.name) / 'native-review-journals' m._journal.root_for = lambda anchor, root=root: root return m @staticmethod def canonical(m, plan): return m.canonical_journal_path(plan) 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, self.canonical(m, p), declaration={'mode': 'none'}) self.assertEqual(result['mode'], 'created') self.assertEqual(api.posts, 1) self.assertEqual(m.publish(p, before, api, lease, self.canonical(m, p), declaration={'mode': 'none'})['mode'], 'no_op') self.assertEqual(api.posts, 1) self.assertNotIn('body', json.loads((self.canonical(m, p) / 'intent.json').read_text())) for applies in [True, False]: m = self.module(); p, before, after, exact = publication_fixture() 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, self.canonical(m, p), declaration={'mode': 'none'})['mode'], 'reconciled') self.assertEqual(m.publish(p, before, api, lease, self.canonical(m, p), declaration={'mode': 'none'})['mode'], 'no_op') else: for _ in range(2): with self.assertRaises(m.UncertainPublication) as error: m.publish(p, before, api, lease, self.canonical(m, p), declaration={'mode': 'none'}) self.assertNotIn('secret transport detail', str(error.exception)) self.assertEqual(api.posts, 1) def test_empty_review_comments_no_header_contract_is_explicit(self): text = (ROOT / 'harness/agents/review/skills/gitea-code-review/references/native-review.md').read_text() self.assertIn('### Empty review-comments readback without a count header', text) section = text.split('### Empty review-comments readback without a count header', 1)[1].split('\n### ', 1)[0] for rule in ['X-Total-Count', 'comments_count=0', 'JSON `[]`', 'declared_total=None', 'rel=next', 'blind retry']: self.assertIn(rule, section) def test_api_adapter_empty_comments_without_header_verifies_publication(self): m = self.module() p, before, after, exact = publication_fixture() def collect(source, comments=None, headers=None): calls = [] def request(path, payload=None): self.assertIsNone(payload, 'readback regression must be GET-only') calls.append(path) direct = {'/user': p['reviewer'], '/users/gitea-spec-agent': p['spec_reviewer'], '/repos/acme/repo': source['repository'], '/repos/acme/repo/issues/3': source['issue'], '/repos/acme/repo/pulls/7': source['pr'], '/repos/acme/repo/pulls/7/reviews/10': exact} if path in direct: return direct[path], {}, 200 from urllib.parse import urlsplit endpoint = urlsplit(path).path 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 endpoint == '/repos/acme/repo' + suffix: return copy.deepcopy(source[name]), {'X-Total-Count': str(len(source[name]))}, 200 if '/reviews/' in endpoint and endpoint.endswith('/comments'): return ([] if comments is None else comments), ({} if headers is None else headers), 200 raise AssertionError('unexpected synthetic readback endpoint') api = m.ApiAdapter(request, origin=m.ORIGIN + '/api/v1', credential_source='$HERMES_HOME/.env:GITEA_TOKEN') snapshot = api.snapshot(p['target']) self.assertEqual(calls[0], '/user') self.assertTrue(any('/reviews/' in path and '/comments?' in path for path in calls)) return api, snapshot _, initial = collect(before) api, final = collect(after) m.validate_snapshot(p, initial) m.validate_snapshot(p, final) authoritative = api.read_review(p['target'], exact['id']) m.verify_publication(p, initial, final, authoritative) for snapshot in [initial, final]: for rid in snapshot['review_comments']: self.assertEqual(snapshot['review_comments'][rid], []) self.assertEqual(snapshot['pagination']['review_comments_' + rid], [{'page': 1, 'count': 0, 'declared_total': None, 'link': None}]) # The admitted control above distinguishes intentional rejection from # a broken fixture. Missing headers never waive the review count or # the complete-collection and newly-created no-inline-comment checks. for count in [1, None, True]: with self.subTest(count=count): drift = copy.deepcopy(final) drift['reviews'][-1]['comments_count'] = count with self.assertRaisesRegex(m.ValidationError, 'review comment cardinality mismatch'): m.validate_snapshot(p, drift) _, unexpected = collect(after, comments=[{'id': 999}]) for review in unexpected['reviews']: review['comments_count'] = 1 m.validate_snapshot(p, unexpected) exact_with_inline = copy.deepcopy(authoritative) exact_with_inline['comments_count'] = 1 with self.assertRaisesRegex(m.ValidationError, 'exact review binding mismatch'): m.verify_publication(p, initial, unexpected, exact_with_inline) with self.assertRaisesRegex(m.ValidationError, 'truncated pages'): collect(after, headers={'X-Total-Count': '1'}) truncated = copy.deepcopy(final) truncated['pagination']['review_comments_' + str(exact['id'])][0]['link'] = '; rel=next' with self.assertRaisesRegex(m.ValidationError, 'truncated pagination'): m.validate_snapshot(p, truncated) with self.assertRaisesRegex(m.ValidationError, 'truncated pagination'): m.check_pages([{'id': i + 1} for i in range(50)], [{'page': 1, 'count': 50, 'declared_total': None, 'link': None}]) 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) canonical_bytes = state.read_bytes(); canonical_stat = state.stat() lease.check(p) self.assertEqual(state.read_bytes(), canonical_bytes) self.assertEqual((state.stat().st_mode, state.stat().st_mtime_ns, state.stat().st_ino), (canonical_stat.st_mode, canonical_stat.st_mtime_ns, canonical_stat.st_ino)) # The bounded compatibility admission is read-only: mutate only # this scratch fixture from canonical producer state to legacy impl. state.write_text(json.dumps({'schema': 'gitea-agent.worktree/v1', 'owner': 'acme', 'repository': 'repo', 'issue': 3, 'created_by_role': 'impl', 'mode': 'detached'})) state.chmod(0o600) legacy_bytes = state.read_bytes(); legacy_stat = state.stat() try: lease.check(p) except m.ValidationError as exc: self.fail(f'valid legacy impl registration must be admitted: {exc}') after_stat = state.stat() self.assertEqual(state.read_bytes(), legacy_bytes) self.assertEqual((after_stat.st_mode, after_stat.st_mtime_ns, after_stat.st_ino), (legacy_stat.st_mode, legacy_stat.st_mtime_ns, legacy_stat.st_ino)) for bad_role in [None, True, False, 1, [], {}, '', ' impl', 'impl ', 'Impl', 'IMPL', 'gitea-impl-agent ', 'review', 'gitea-review-agent', 'gitea-spec-agent', 'gitea-release-agent', 'other']: state.write_text(json.dumps({'schema': 'gitea-agent.worktree/v1', 'owner': 'acme', 'repository': 'repo', 'issue': 3, 'created_by_role': bad_role, 'mode': 'detached'})) state.chmod(0o600) with self.subTest(bad_role=repr(bad_role)), self.assertRaisesRegex(m.ValidationError, 'worktree registration identity mismatch'): lease.check(p) missing_role = json.loads(legacy_bytes); missing_role.pop('created_by_role') state.write_text(json.dumps(missing_role)); state.chmod(0o600) with self.assertRaisesRegex(m.ValidationError, 'worktree registration identity mismatch'): lease.check(p) bool_issue = json.loads(legacy_bytes); bool_issue['issue'] = True state.write_text(json.dumps(bool_issue)); state.chmod(0o600) with self.assertRaisesRegex(m.ValidationError, 'worktree registration identity mismatch'): lease.check(p) state.write_text(json.dumps({'schema': 'gitea-agent.worktree/v1', 'owner': 'other', 'repository': 'repo', 'issue': 3, 'created_by_role': 'impl', 'mode': 'detached'})); state.chmod(0o600) with self.assertRaisesRegex(m.ValidationError, 'worktree registration identity mismatch'): lease.check(p) state.write_text(json.dumps({'schema': 'gitea-agent.worktree/v1', 'owner': 'acme', 'repository': 'other', 'issue': 3, 'created_by_role': 'impl', 'mode': 'detached'})); state.chmod(0o600) with self.assertRaisesRegex(m.ValidationError, 'worktree registration identity mismatch'): lease.check(p) state.write_text(json.dumps({'schema': 'gitea-agent.worktree/v1', 'owner': 'acme', 'repository': 'repo', 'issue': 4, 'created_by_role': 'impl', 'mode': 'detached'})); state.chmod(0o600) with self.assertRaisesRegex(m.ValidationError, 'worktree registration identity mismatch'): lease.check(p) state.write_text(json.dumps({'schema': 'wrong/v1', 'owner': 'acme', 'repository': 'repo', 'issue': 3, 'created_by_role': 'impl', 'mode': 'detached'})); state.chmod(0o600) with self.assertRaisesRegex(m.ValidationError, 'worktree registration identity mismatch'): lease.check(p) state.write_text(json.dumps({'schema': 'gitea-agent.worktree/v1', 'owner': 'acme', 'repository': 'repo', 'issue': 3, 'created_by_role': 'impl', 'mode': 'attached'})); state.chmod(0o600) with self.assertRaisesRegex(m.ValidationError, 'worktree registration identity mismatch'): lease.check(p) state.write_text(json.dumps({'schema': 'gitea-agent.worktree/v1', 'owner': 'acme', 'repository': 'repo', 'issue': 3, 'created_by_role': 'impl', 'mode': 'detached'})); state.chmod(0o600) wrong_head = copy.deepcopy(p); wrong_head['target']['head'] = 'b' * 40 wrong_head['body'] = wrong_head['body'].replace(head, 'b' * 40) with self.assertRaisesRegex(m.ValidationError, 'worktree HEAD mismatch'): lease.check(wrong_head) 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) # Every failed source validation must close all transient FDs. before_fds = len(list((Path('/proc/self/fd')).iterdir())) for _ in range(20): with self.assertRaises(m.ValidationError): lease.check(p) after_fds = len(list((Path('/proc/self/fd')).iterdir())) self.assertLessEqual(after_fds, before_fds + 1) (wt / 'a.txt').unlink() for directory in [wt.parent.parent, wt.parent, wt]: directory.chmod(0o770) with self.subTest(directory=directory), self.assertRaises(m.ValidationError): lease.check(p) directory.chmod(0o700) 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']), declaration={'mode': 'none'}) 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, self.canonical(m, p), declaration={'mode': 'none'}) self.assertEqual(api.posts, 0) self.assertFalse((self.canonical(m, p) / '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(), self.canonical(m, p), declaration={'mode': 'none'}) 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', '; 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_normal_supersession_is_allowed_without_projection_opt_in(self): for already_dismissed in (False, True): with self.subTest(already_dismissed=already_dismissed): m = self.module() p, before, after, exact = publication_fixture() p['provider_contract'].update(supersede_same_author=False, retouch_dismissed=False, unordered_commit_files=False, evidence=[]) if already_dismissed: before['reviews'][0].update(official=False, dismissed=True) after['reviews'][0].update(official=False, dismissed=True, updated_at=exact['submitted_at']) saved_plan = copy.deepcopy(p) saved_before = copy.deepcopy(before) try: result = m.verify_publication(p, before, after, exact) except m.ValidationError as exc: self.fail(f'normal same-author replacement was rejected: {exc}') self.assertTrue(result['verified']) self.assertEqual(result['projected_review_ids'], [8]) self.assertEqual(p, saved_plan) self.assertEqual(before, saved_before) def test_legacy_replacement_flags_do_not_require_external_evidence(self): m = self.module() p, before, after, exact = publication_fixture() p['provider_contract'].update(unordered_commit_files=False, evidence=[]) after['reviews'][0].update(official=False, dismissed=True, updated_at=exact['submitted_at']) try: result = m.verify_publication(p, before, after, exact) except m.ValidationError as exc: self.fail(f'compatibility-only replacement flags required an opt-in: {exc}') self.assertTrue(result['verified']) self.assertEqual(m.build_payload(p)['body'], p['body']) # Unordered file comparisons remain a separate, evidence-backed exception. p['provider_contract']['unordered_commit_files'] = True with self.assertRaisesRegex(m.ValidationError, 'provider projection requires evidence'): m.validate_plan(p) def test_normal_replacement_does_not_allow_historical_content_mutation(self): m = self.module() for field, value in [('body', 'altered'), ('commit_id', 'e' * 40), ('state', 'APPROVED'), ('user', {'id': 72, 'login': 'other'}), ('submitted_at', '2026-10-04T00:03:00Z'), ('official', True), ('dismissed', False), ('updated_at', '2026-10-04T00:03:00Z')]: with self.subTest(field=field): p, before, after, exact = publication_fixture() p['provider_contract'].update(supersede_same_author=False, retouch_dismissed=False, unordered_commit_files=False, evidence=[]) after['reviews'][0].update(official=False, dismissed=True, updated_at=exact['submitted_at']) after['reviews'][0][field] = value with self.assertRaises(m.ValidationError): m.verify_publication(p, before, after, exact) p, before, after, exact = publication_fixture() p['provider_contract'].update(supersede_same_author=False, retouch_dismissed=False, unordered_commit_files=False, evidence=[]) after['reviews'][0].update(official=False, dismissed=True, updated_at=exact['submitted_at']) 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.assertRaisesRegex(m.ValidationError, 'historical review comments changed'): m.verify_publication(p, before, after, exact) def test_normal_replacement_reconciles_original_intent_without_second_post(self): m = self.module() p, before, after, exact = publication_fixture() p['provider_contract'].update(supersede_same_author=False, retouch_dismissed=False, unordered_commit_files=False, evidence=[]) after['reviews'][0].update(official=False, dismissed=True, updated_at=exact['submitted_at']) api = FakeAPI(p, before, after, exact) original_read = api.read_review api.read_review = lambda target, rid: (_ for _ in ()).throw(TimeoutError('unavailable')) with self.assertRaises(m.ValidationError): m.publish(p, before, api, FakeLease(), self.canonical(m, p), declaration={'mode': 'none'}) self.assertEqual(api.posts, 1) journal = self.canonical(m, p) intent = (journal / 'intent.json').read_bytes() self.assertFalse((journal / 'verified.json').exists()) api.read_review = original_read result = m.publish(p, before, api, FakeLease(), journal, declaration={'mode': 'none'}) self.assertTrue(result['verified']) self.assertFalse(result['wrote']) self.assertEqual(result['projected_review_ids'], [8]) self.assertEqual(api.posts, 1) self.assertEqual((journal / 'intent.json').read_bytes(), intent) self.assertEqual(json.loads(intent)['fingerprint'], __import__('hashlib').sha256(m.canonical(p)).hexdigest()) self.assertTrue(json.loads((journal / 'verified.json').read_text())['verified']) def test_provider_projection_requires_explicit_bound_evidence(self): m = self.module() for kind in ['wrong_initial_flags', '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 == 'wrong_initial_flags': before['reviews'][0].update(official=True, dismissed=True) 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(), self.canonical(m, p), declaration={'mode': 'none'}) self.assertEqual(result['mode'], 'no_op'); self.assertFalse(result['wrote']) self.assertEqual(api.posts, 0) self.assertFalse((self.canonical(m, p) / '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(), self.canonical(m, p), declaration={'mode': 'none'}) 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(), self.canonical(m, p), declaration={'mode': 'none'}) 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, declaration={'mode': 'none'}) 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 = self.canonical(m, p) with self.assertRaises(m.UncertainPublication): m.publish(p, before, api, FakeLease(), journal, declaration={'mode': 'none'}) 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, declaration={'mode': 'none'}) self.assertEqual(api.posts, 1) with m.Journal(root / 'hardlinks') as j: j.write('before.json', {'safe': True}); os.link(j.path / 'before.json', j.path / 'b.json') with self.assertRaises(m.ValidationError): j.read('before.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, self.canonical(m, p), declaration={'mode': 'none'}) 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(), self.canonical(m, p), declaration={'mode': 'none'}) 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(), self.canonical(m, p), declaration={'mode': 'none'}) 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_broker_issue_lease_accepts_authenticated_matching_status(self): m = self.module(); p, before, after, exact = publication_fixture() now = time.monotonic() lease = m.BrokerIssueLease('/trusted/handle.json', '/trusted/policy.md', now + 30) handle = {'owner': 'acme', 'repository': 'repo', 'issue': 3, 'role': 'gitea-review-agent', 'pid': 1234, 'deadline': now + 20, 'token': 'not-returned', 'socket': '/trusted/broker.sock'} status = {'status': 'held', 'owner': 'acme', 'repository': 'repo', 'issue': 3, 'role': 'gitea-review-agent', 'pid': 1234, 'deadline': handle['deadline']} from unittest.mock import patch with patch.object(m._broker, 'read_handle_metadata', return_value={k: v for k, v in handle.items() if k != 'token'}), \ patch.object(m._broker, 'request_handle', return_value=status), \ patch.object(m, '_validate_registered_source') as source: self.assertIsNone(lease.check(p)) source.assert_called_once_with(p, '/trusted/policy.md', lease.deadline_monotonic) def test_broker_issue_lease_rejects_wrong_identity_status_or_deadline(self): m = self.module(); p, before, after, exact = publication_fixture(); now = time.monotonic() lease = m.BrokerIssueLease('/trusted/handle.json', '/trusted/policy.md', now + 30) handle = {'pid': 1234, 'deadline': now + 20, 'token': 'hidden', 'socket': '/trusted/broker.sock'} cases = [ {'status': 'held', 'owner': 'acme', 'repository': 'repo', 'issue': 3, 'role': 'gitea-spec-agent', 'pid': 1234, 'deadline': handle['deadline']}, {'status': 'held', 'owner': 'other', 'repository': 'repo', 'issue': 3, 'role': 'gitea-review-agent', 'pid': 1234, 'deadline': handle['deadline']}, {'status': 'released', 'owner': 'acme', 'repository': 'repo', 'issue': 3, 'role': 'gitea-review-agent', 'pid': 1234, 'deadline': handle['deadline']}, {'status': 'held', 'owner': 'acme', 'repository': 'repo', 'issue': 3, 'role': 'gitea-review-agent', 'pid': 9999, 'deadline': handle['deadline']}, {'status': 'held', 'owner': 'acme', 'repository': 'repo', 'issue': 3, 'role': 'gitea-review-agent', 'pid': 1234, 'deadline': now + 40}, ] from unittest.mock import patch for status in cases: with self.subTest(status=status): with patch.object(m._broker, 'read_handle_metadata', return_value={k: v for k, v in handle.items() if k != 'token'}), \ patch.object(m._broker, 'request_handle', return_value=status), \ patch.object(m, '_validate_registered_source') as source: with self.assertRaises(m.ValidationError): lease.check(p) source.assert_not_called() def test_journal_intent_fields_are_validated_on_resume(self): for field in ['schema', 'head', 'event', 'body_sha256']: m = self.module() 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.ValidationError): m.publish(p, before, api, FakeLease(), self.canonical(m, p), declaration={'mode': 'none'}) path = self.canonical(m, p) / '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(), self.canonical(m, p), declaration={'mode': 'none'}) 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()