import hashlib, json, subprocess, sys, tempfile, unittest from pathlib import Path from harness.tests._support.paths import ROOT SCRIPT = ROOT / "harness/agents/impl/skills/gitea-issue-rework/scripts/rework_comment.py" class ReworkCommentTests(unittest.TestCase): def setUp(self): self.t = tempfile.TemporaryDirectory() self.d = Path(self.t.name) self.data = {'owner': 'acme', 'repository': 'widget', 'issue': 7, 'pr': 12, 'review': 99, 'reviewed_head': 'a' * 40, 'final_head': 'b' * 40, 'contract_sha256': 'c' * 64, 'result': 'partial', 'findings': [{'id': 'F-1', 'status': 'resolved', 'classification': 'valid_fix', 'changes': 'src/main.py', 'verification': 'pytest -q (0), pass', 'reason': 'bounded fix'}], 'local_checks': 'pytest -q (0), pass', 'runtime_browser': '해당 없음·미검증', 'commit_status': 'success', 'actions_runs': '없음'} self.write('input.json', self.data) self.invoke('render', '--no-delegated-evidence', '--input', self.p('input.json'), '--comment', self.p('comment.md'), '--expected', self.p('expected.json')) self.expected = json.loads(self.d.joinpath('expected.json').read_text()) def tearDown(self): self.t.cleanup() def p(self, n): return str(self.d / n) def write(self, n, x): self.d.joinpath(n).write_text(json.dumps(x, ensure_ascii=False) if not isinstance(x, str) else x) def invoke(self, *args): return subprocess.run([sys.executable, str(SCRIPT), *args], text=True, capture_output=True) def read(self, n): return json.loads(self.d.joinpath(n).read_text()) def readback(self, comments=None, head=None): self.write('pr.json', {'object_kind': 'pull_request', 'owner': 'acme', 'repository': 'widget', 'number': 12, 'state': 'open', 'head': {'sha': head or self.data['final_head']}, 'html_url': 'https://gitea/acme/widget/pulls/12', 'metadata_binding_verified': True}) self.write('identity.json', {'id': 42, 'login': 'bot', 'role': 'gitea-impl-agent', 'authenticated': True, 'stable_account_verified': True}) items = comments if comments is not None else [{'id': 501, 'html_url': 'https://gitea/x/501', 'created_at': '2026-01-01T00:00:00Z', 'updated_at': '2026-01-01T00:00:00Z', 'body': self.expected['body'], 'user': {'id': 42, 'login': 'bot'}}] self.write('comments.json', {'object_kind': 'pull_request_top_level_comments', 'owner': 'acme', 'repository': 'widget', 'pr': 12, 'complete': True, 'endpoint_binding_verified': True, 'source_issue_unchanged': True, 'review_collection_unchanged': True, 'pr_metadata_unchanged': True, 'page_count': 1, 'total_count': len(items), 'provider_counts': {'top_level': len(items), 'reviews': 2, 'inline': 3, 'aggregate_comments': len(items), 'aggregate_components': ['top_level']}, 'comments': items}) def verify(self, action='create', baseline=0): return self.invoke('verify', '--expected', self.p('expected.json'), '--pr-readback', self.p('pr.json'), '--comments-readback', self.p('comments.json'), '--identity-readback', self.p('identity.json'), '--action', action, '--baseline-count', str(baseline), '--output', self.p('receipt.json')) def test_render_digest_and_correlation(self): self.assertEqual(hashlib.sha256(self.expected['marker_free_body'].encode()).hexdigest(), self.expected['result_sha256']) parts = ['acme', 'widget', '7', '12', '99', 'a' * 40, 'b' * 40, 'c' * 64] self.assertEqual(hashlib.sha256('\x00'.join(parts).encode()).hexdigest(), self.expected['correlation']) self.assertTrue(self.d.joinpath('comment.md').read_text().endswith('-->\n')) def test_no_comment_and_marker_body_mismatch(self): self.readback([]) self.assertNotEqual(self.verify().returncode, 0) self.readback([{'id': 1, 'body': self.expected['body'].replace('correlation=', 'correlation=x'), 'user': {'id': 42, 'login': 'bot'}}]) self.assertNotEqual(self.verify().returncode, 0) def test_wrong_head_and_author(self): self.readback(head='d' * 40) self.assertNotEqual(self.verify().returncode, 0) self.readback() c = self.read('comments.json') c['comments'][0]['user'] = {'id': 43, 'login': 'bot'} self.write('comments.json', c) self.assertNotEqual(self.verify().returncode, 0) def test_digest_and_correlation_tampering_duplicate(self): self.expected['result_sha256'] = '0' * 64 self.write('expected.json', self.expected) self.readback() self.assertNotEqual(self.verify().returncode, 0) self.setUp_render_again() self.expected['correlation'] = '0' * 64 self.write('expected.json', self.expected) self.readback() self.assertNotEqual(self.verify().returncode, 0) self.setUp_render_again() self.readback() c = self.read('comments.json') c['comments'].append(dict(c['comments'][0], id=502)) c['total_count'] = 2 self.write('comments.json', c) self.assertNotEqual(self.verify('reuse', 1).returncode, 0) def setUp_render_again(self): self.invoke('render', '--no-delegated-evidence', '--input', self.p('input.json'), '--comment', self.p('comment.md'), '--expected', self.p('expected.json')) self.expected = self.read('expected.json') def test_create_reuse_cardinality(self): self.readback() self.assertEqual(self.verify('create', 0).returncode, 0) self.assertEqual(self.read('receipt.json')['status'], 'verified') original = self.read('comments.json')['comments'][0] self.readback([original, {'id': 2, 'html_url': 'u', 'created_at': 't', 'updated_at': 't', 'body': 'other', 'user': {'id': 42, 'login': 'bot'}}]) self.assertEqual(self.verify('reuse', 2).returncode, 0) def test_exact_success_and_receipt_preserved(self): self.readback() self.assertEqual(self.verify().returncode, 0) old = self.d.joinpath('receipt.json').read_bytes() self.readback(head='e' * 40) self.assertNotEqual(self.verify().returncode, 0) self.assertEqual(old, self.d.joinpath('receipt.json').read_bytes()) def test_unresolved_delivery_and_incomplete_or_wrong_scope_readback_fail(self): bad = dict(self.data) bad['result'] = 'rework_delivered' bad['findings'] = [dict(self.data['findings'][0], status='unresolved')] self.write('bad-input.json', bad) self.assertNotEqual(self.invoke('render', '--no-delegated-evidence', '--input', self.p('bad-input.json'), '--comment', self.p('bad.md'), '--expected', self.p('bad.json')).returncode, 0) self.readback() comments = self.read('comments.json') comments['complete'] = False self.write('comments.json', comments) self.assertNotEqual(self.verify().returncode, 0) self.readback() pr = self.read('pr.json') pr['repository'] = 'other' self.write('pr.json', pr) self.assertNotEqual(self.verify().returncode, 0) def test_impl_role_and_provider_aggregate_binding_are_required(self): self.readback() identity = self.read('identity.json') identity['role'] = 'other' self.write('identity.json', identity) self.assertNotEqual(self.verify().returncode, 0) self.readback() comments = self.read('comments.json') comments['provider_counts']['aggregate_comments'] += 1 self.write('comments.json', comments) self.assertNotEqual(self.verify().returncode, 0) def test_render_visible_identifier_lines_do_not_emit_issue_references(self): body = self.expected['marker_free_body'] self.assertIn('- Source Issue: 7', body) self.assertIn('- Pull Request: 12', body) self.assertIn('- Source Review: 99', body) self.assertNotIn('- Source Issue: #7', body) self.assertNotIn('- Pull Request: #12', body) self.assertNotIn('- Source Review: #99', body) def test_render_rejects_reference_hazards_in_every_safe_human_field(self): hazards = ('#123', 'otherrepo#123', '\\\\#123', '\\#123', 'review #99') fields = ('changes', 'verification', 'reason', 'local_checks', 'runtime_browser') for field in fields: for snippet in hazards: with self.subTest(field=field, snippet=snippet): data = json.loads(json.dumps(self.data)) if field in ('changes', 'verification', 'reason'): data['findings'][0][field] = 'unsafe payload ' + snippet else: data[field] = 'unsafe payload ' + snippet self.write('hazard.json', data) result = self.invoke('render', '--no-delegated-evidence', '--input', self.p('hazard.json'), '--comment', self.p('hazard.md'), '--expected', self.p('hazard-expected.json')) self.assertNotEqual(result.returncode, 0) def test_offline_verifier_rejects_reference_in_expected_body(self): self.expected['marker_free_body'] += '\n- unsafe: otherrepo#123' before_hash = self.expected['result_sha256'] self.expected['result_sha256'] = hashlib.sha256(self.expected['marker_free_body'].encode()).hexdigest() self.expected['marker'] = self.expected['marker'].replace(before_hash, self.expected['result_sha256']) self.expected['body'] = self.expected['marker_free_body'] + self.expected['marker'] self.write('expected.json', self.expected) self.readback() self.assertNotEqual(self.verify().returncode, 0) def _reseal(self, value, body): value = json.loads(json.dumps(value)) old = value['result_sha256'] value['marker_free_body'] = body value['result_sha256'] = hashlib.sha256(body.encode()).hexdigest() value['marker'] = value['marker'].replace(old, value['result_sha256']) value['body'] = body + value['marker'] return value def _legacy_expected(self): old = json.loads(json.dumps(self.expected)) body = old['marker_free_body'] body = body.replace('- Source Issue: #7', '- Source Issue: 7').replace('- Pull Request: #12', '- Pull Request: 12').replace('- Source Review: #99', '- Source Review: 99') body = body.replace('- Source Issue: 7', '- Source Issue: #7').replace('- Pull Request: 12', '- Pull Request: #12').replace('- Source Review: 99', '- Source Review: #99') old['marker_free_body'] = body old['result_sha256'] = hashlib.sha256(body.encode()).hexdigest() old['marker'] = old['marker'].replace(self.expected['result_sha256'], old['result_sha256']) old['body'] = body + old['marker'] return old def test_prepare_correction_returns_bounded_verified_legacy_migration_plan(self): import importlib.util spec = importlib.util.spec_from_file_location('rework_comment_correction_test', SCRIPT) module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) self.assertTrue(callable(getattr(module, 'prepare_correction', None)), 'approved correction API missing') plan = module.prepare_correction(self._legacy_expected(), self.expected, comment_id=501, actor_id=42) self.assertEqual(plan['schema'], 'gitea-rework-correction-plan/v1') self.assertEqual(plan['target'], self.expected['target']) self.assertEqual(plan['comment_id'], 501) self.assertEqual(plan['actor_id'], 42) self.assertEqual(plan['old_body'], self._legacy_expected()['body']) self.assertEqual(plan['new_body'], self.expected['body']) self.assertEqual(plan['old_body_sha256'], hashlib.sha256(plan['old_body'].encode()).hexdigest()) self.assertEqual(plan['new_body_sha256'], hashlib.sha256(plan['new_body'].encode()).hexdigest()) self.assertIs(plan['prior_handoff_verified'], False) self.assertTrue(plan['immutable_source_required']) def test_prepare_correction_rejects_tampered_or_unbounded_changes(self): import importlib.util spec = importlib.util.spec_from_file_location('rework_comment_correction_reject_test', SCRIPT) module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) self.assertTrue(callable(getattr(module, 'prepare_correction', None)), 'approved correction API missing') old = self._legacy_expected() cases = [] bad = json.loads(json.dumps(old)) bad['result_sha256'] = '0' * 64 cases.append((bad, self.expected)) bad = json.loads(json.dumps(old)) bad['body'] = bad['body'].replace('#7', '#8', 1) cases.append((bad, self.expected)) bad = json.loads(json.dumps(old)) bad['marker_free_body'] = bad['marker_free_body'].replace('bounded fix', 'unsafe old text #123') cases.append((bad, self.expected)) bad = json.loads(json.dumps(self.expected)) bad['marker_free_body'] = bad['marker_free_body'].replace('bounded fix', 'different extra text') cases.append((old, bad)) bad = self._reseal(self.expected, self.expected['marker_free_body'].replace('bounded fix', 'different extra text')) cases.append((old, bad)) unsafe_old = self._reseal(old, old['marker_free_body'].replace('bounded fix', 'same unsafe #123')) unsafe_new = self._reseal(self.expected, self.expected['marker_free_body'].replace('bounded fix', 'same unsafe #123')) cases.append((unsafe_old, unsafe_new)) bad = json.loads(json.dumps(self.expected)) bad['head'] = 'd' * 40 cases.append((old, bad)) bad = json.loads(json.dumps(self.expected)) bad['correlation'] = 'e' * 64 cases.append((old, bad)) for previous, new in cases: with self.subTest(previous=previous['result_sha256'], new=new['correlation']): with self.assertRaises(module.Bad): module.prepare_correction(previous, new, comment_id=501, actor_id=42) def test_prepare_correction_cli_emits_plan_not_approval_or_receipt(self): self.write('previous.json', self._legacy_expected()) r = self.invoke('prepare-correction', '--previous-expected', self.p('previous.json'), '--expected', self.p('expected.json'), '--comment-id', '501', '--actor-id', '42', '--output', self.p('plan.json')) self.assertEqual(r.returncode, 0, r.stderr) plan = self.read('plan.json') self.assertEqual(plan['schema'], 'gitea-rework-correction-plan/v1') self.assertIs(plan['approval_required'], True) self.assertIs(plan['prior_handoff_verified'], False) self.assertNotEqual(plan.get('status'), 'verified') self.assertEqual(self.d.joinpath('plan.json').stat().st_mode & 511, 384) def test_prepare_correction_cli_refuses_input_alias_and_preserves_file(self): self.write('previous.json', self._legacy_expected()) before = self.d.joinpath('expected.json').read_bytes() r = self.invoke('prepare-correction', '--previous-expected', self.p('previous.json'), '--expected', self.p('expected.json'), '--comment-id', '501', '--actor-id', '42', '--output', self.p('expected.json')) self.assertNotEqual(r.returncode, 0) self.assertEqual(self.d.joinpath('expected.json').read_bytes(), before) def test_different_body_same_correlation_is_not_default_reuse(self): old = self._legacy_expected() self.readback([{'id': 501, 'html_url': 'https://gitea/x/501', 'created_at': '2026-01-01T00:00:00Z', 'updated_at': '2026-01-01T00:00:00Z', 'body': old['body'], 'user': {'id': 42, 'login': 'bot'}}]) self.assertNotEqual(self.verify('reuse', 1).returncode, 0)