genichin/gitea-agent
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[verified] Fix spec review source-Issue reference readback

Browse Source
This commit is contained in:
gitea-spec-agent
2026-10-14 01:56:46 +00:00
parent dbd381a41b
commit f5227e4a4d
3 changed files with 417 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
harness/tests/spec/test_source_issue_reference_projection.py
+277
View File
@@ -0,0 +1,277 @@
import copy
import json
import subprocess
import sys
import tempfile
import unittest
from harness.tests.spec import test_implementation_spec_review_support as fixture
class SourceIssueReferenceProjectionTests(unittest.TestCase):
def contract(self):
return {'version': {'version': '1.27.3'},
'source_issue_cross_reference_projection': True}
def snapshots(self):
before, after, review, result, paths = fixture.publication_fixture()
url = before['pr']['html_url']
issue_url = url.rsplit('/pulls/', 1)[0] + '/issues/3'
before['issue']['html_url'] = after['issue']['html_url'] = issue_url
pr_issue = {'id': 70, 'number': 7, 'html_url': url,
'repository': {'full_name': 'acme/repo'}, 'user': {'id': 99},
'state': 'open', 'title': 'implementation', 'body': 'Closes #3',
'comments': 2, 'updated_at': before['pr']['updated_at'],
'pull_request': {'merged': False, 'html_url': url}}
before['pr_issue'] = pr_issue
after['pr_issue'] = copy.deepcopy(pr_issue)
after['pr_issue'].update(comments=3, updated_at=review['submitted_at'])
event = after['timeline'][0]
event['html_url'] = url + '#issuecomment-101'
event['issue_url'] = ''
event['assets'] = []
ref_comment = {key: copy.deepcopy(event[key]) for key in
('id', 'html_url', 'pull_request_url', 'issue_url', 'user',
'body', 'assets', 'created_at', 'updated_at')}
reference = {'id': 102, 'type': 'comment_ref', 'user': copy.deepcopy(review['user']),
'body': '', 'html_url': issue_url + '#issuecomment-102',
'issue_url': issue_url, 'pull_request_url': '',
'created_at': '2026-10-03T01:24:01Z',
'updated_at': '2026-10-03T01:24:01Z',
'ref_issue': copy.deepcopy(after['pr_issue']),
'ref_comment': ref_comment, 'ref_action': 'none', 'review_id': 0}
before['issue_timeline'] = []
after['issue_timeline'] = [reference]
after['issue']['updated_at'] = reference['created_at']
return before, after, review, result, paths
def test_exact_reference_allows_only_source_issue_timestamp(self):
before, after, review, result, paths = self.snapshots()
original = copy.deepcopy((before, after))
try:
actual = fixture.verify_fixture(before, after, review, result, paths, self.contract())
except fixture.mod.ValidationError as exc:
self.fail('exact review-bound source reference rejected: ' + str(exc))
self.assertEqual(actual['source_issue_reference_ids'], [102])
self.assertEqual((before, after), original)
after['pr']['updated_at'] = after['issue']['updated_at']
with self.assertRaisesRegex(fixture.mod.ValidationError, 'timestamp projection mismatch'):
fixture.verify_fixture(before, after, review, result, paths, self.contract())
def test_wrong_reference_identity_payload_or_timestamp_is_rejected(self):
baseline = self.snapshots()
fixture.verify_fixture(*baseline, self.contract())
mutations = [('type', 'comment'), ('user.id', 43), ('user.login', 'changed'),
('ref_issue.number', 8), ('ref_issue.repository.full_name', 'other/repo'),
('ref_issue.id', 71), ('ref_comment.id', 100), ('ref_comment.id', True),
('ref_comment.user.id', 43), ('ref_comment.body', 'other review'),
('ref_comment.created_at', '2026-10-03T01:24:01Z'),
('ref_comment.updated_at', '2026-10-03T01:24:01Z'),
('ref_comment.html_url', 'https://other.test/acme/repo/pulls/7#issuecomment-101'),
('ref_comment.pull_request_url', 'https://other.test/acme/repo/pulls/7'),
('ref_comment.assets', [{'id': 9}]), ('ref_comment.issue_url', 'wrong'),
('issue_url', 'https://example.test/acme/repo/issues/4'),
('html_url', 'https://example.test/acme/repo/issues/3#issuecomment-103'),
('pull_request_url', 'https://example.test/acme/repo/pulls/7'),
('ref_action', 'close'), ('review_id', True), ('body', 'unrelated'),
('created_at', '2026-10-03T01:23:59Z'), ('updated_at', '2026-10-03T01:24:02Z')]
for field, value in mutations:
before, after, review, result, paths = copy.deepcopy(baseline)
node = after['issue_timeline'][-1]
parts = field.split('.')
for key in parts[:-1]:
node = node[key]
node[parts[-1]] = value
with self.subTest(field=field, value=value), self.assertRaises(fixture.mod.ValidationError):
fixture.verify_fixture(before, after, review, result, paths, self.contract())
for field in ('ref_issue', 'ref_comment', 'created_at', 'updated_at', 'user'):
before, after, review, result, paths = copy.deepcopy(baseline)
del after['issue_timeline'][-1][field]
with self.subTest(missing=field), self.assertRaises(fixture.mod.ValidationError):
fixture.verify_fixture(before, after, review, result, paths, self.contract())
def test_opt_in_version_and_complete_evidence_are_required(self):
baseline = self.snapshots()
fixture.verify_fixture(*baseline, self.contract())
for contract in (None, {'version': {'version': '1.27.3'}},
{'version': {'version': '1.27.3'}, 'source_issue_cross_reference_projection': False},
{'version': {'version': '1.27.2'}, 'source_issue_cross_reference_projection': True}):
with self.subTest(contract=contract), self.assertRaises(fixture.mod.ValidationError):
fixture.verify_fixture(*baseline, contract)
for value in (1, 'true', None, [], {}):
contract = self.contract(); contract['source_issue_cross_reference_projection'] = value
with self.subTest(value=value), self.assertRaises(fixture.mod.ValidationError):
fixture.verify_fixture(*baseline, contract)
for side in (0, 1):
for field in ('issue_timeline', 'pr_issue'):
args = copy.deepcopy(baseline); del args[side][field]
with self.subTest(side=side, missing=field), self.assertRaises(fixture.mod.ValidationError):
fixture.verify_fixture(*args, self.contract())
def test_no_time_tolerance_or_other_field_relaxation(self):
baseline = self.snapshots()
fixture.verify_fixture(*baseline, self.contract())
for section, field, value in [('issue', 'updated_at', '2026-10-03T01:24:02Z'),
('issue', 'title', 'modified'), ('issue', 'body', 'new contract'),
('pr_issue', 'updated_at', '2026-10-03T01:24:01Z'),
('pr_issue', 'comments', 4), ('pr_issue', 'comments', True),
('pr_issue', 'body', 'modified'), ('pr_issue', 'id', True),
('pr_issue', 'number', 8), ('pr_issue', 'state', 'closed')]:
args = copy.deepcopy(baseline); args[1][section][field] = value
with self.subTest(section=section, field=field), self.assertRaises(fixture.mod.ValidationError):
fixture.verify_fixture(*args, self.contract())
for mode in ('missing', 'duplicate', 'competing'):
args = copy.deepcopy(baseline); events = args[1]['issue_timeline']
if mode == 'missing':
events.clear()
else:
extra = copy.deepcopy(events[-1])
if mode == 'competing':
extra.update(id=103, type='comment')
events.append(extra)
with self.subTest(mode=mode), self.assertRaises(fixture.mod.ValidationError):
fixture.verify_fixture(*args, self.contract())
def test_existing_source_records_preserve_every_field_except_bound_pr_issue_projection(self):
before, after, review, result, paths = self.snapshots()
old = {'id': 90, 'type': 'pull_ref', 'body': '',
'ref_issue': copy.deepcopy(before['pr_issue']), 'user': {'id': 99}}
before['issue_timeline'].append(old)
after['issue_timeline'].insert(0, copy.deepcopy(old))
after['issue_timeline'][0]['ref_issue'] = copy.deepcopy(after['pr_issue'])
baseline = (before, after, review, result, paths)
fixture.verify_fixture(*baseline, self.contract())
for field in ('body', 'user', 'ref_issue', 'missing', 'order'):
args = copy.deepcopy(baseline)
if field == 'missing':
args[1]['issue_timeline'].pop(0)
elif field == 'order':
args[1]['issue_timeline'].reverse()
elif field == 'ref_issue':
args[1]['issue_timeline'][0]['ref_issue']['comments'] = 5
else:
args[1]['issue_timeline'][0][field] = 'tampered'
with self.subTest(field=field), self.assertRaises(fixture.mod.ValidationError):
fixture.verify_fixture(*args, self.contract())
args = copy.deepcopy(baseline)
args[0]['issue_timeline'].append({'id': 89, 'body': 'ordinary'})
args[1]['issue_timeline'].insert(1, {'id': 89, 'body': 'ordinary'})
fixture.verify_fixture(*args, self.contract())
args[1]['issue_timeline'][1]['body'] = 'changed'
with self.assertRaises(fixture.mod.ValidationError):
fixture.verify_fixture(*args, self.contract())
def test_source_reference_timestamp_does_not_apply_to_historical_reviews(self):
before, after, review, result, paths = self.snapshots()
old = copy.deepcopy(review)
old.update(id=10, body='previous', state='REQUEST_CHANGES', commit_id='d' * 40,
updated_at=before['pr']['updated_at'])
before['reviews'] = [old]
after['reviews'].insert(0, copy.deepcopy(old))
after['reviews'][0].update(official=False, dismissed=True, updated_at=review['submitted_at'])
contract = self.contract(); contract['supersession_same_author_only'] = True
fixture.verify_fixture(before, after, review, result, paths, contract)
after['reviews'][0]['updated_at'] = after['issue']['updated_at']
with self.assertRaises(fixture.mod.ValidationError):
fixture.verify_fixture(before, after, review, result, paths, contract)
def test_cli_uses_exact_reference_and_sanitizes_invalid_input(self):
from pathlib import Path
before, after, review, result, paths = self.snapshots()
with tempfile.TemporaryDirectory(dir='/opt/data/cache/scratch') as directory:
root = Path(directory)
data = {'before': before, 'after': after, 'exact-review': review, 'result': result,
'allowed-paths': sorted(paths), 'inline-comments': [], 'provider-contract': self.contract()}
command = [sys.executable, '-B', str(fixture.MODPATH), 'verify-publication',
'--reviewer-id', '42', '--expected-pr-url', before['pr']['html_url']]
for key, value in data.items():
path = root / (key + '.json')
path.write_text(json.dumps(value), encoding='utf-8')
command.extend(['--' + key, str(path)])
actual = subprocess.run(command, capture_output=True, text=True, timeout=15)
self.assertEqual(actual.returncode, 0, actual.stderr)
self.assertEqual(json.loads(actual.stdout)['source_issue_reference_ids'], [102])
after['issue_timeline'][0]['ref_comment']['body'] = 'sensitive-value-do-not-echo'
(root / 'after.json').write_text(json.dumps(after), encoding='utf-8')
rejected = subprocess.run(command, capture_output=True, text=True, timeout=15)
self.assertEqual(rejected.returncode, 2)
self.assertEqual(json.loads(rejected.stderr), {'error': 'invalid input'})
self.assertEqual(rejected.stdout, '')
self.assertNotIn('sensitive-value-do-not-echo', rejected.stdout + rejected.stderr)
def test_collector_rejects_bad_target_or_provider_before_any_read(self):
calls = []
def read_record(endpoint):
calls.append(endpoint)
self.fail('invalid collector request reached read callback')
def fetch_page(endpoint, page):
calls.append((endpoint, page))
self.fail('invalid collector request reached page callback')
for field, value in [('issue', True), ('owner', '../other'), ('repository', 'bad/name'),
('head', 'not-a-full-sha')]:
target = copy.deepcopy(fixture.result()['target']); target[field] = value
with self.subTest(field=field), self.assertRaises(fixture.mod.ValidationError):
fixture.mod.collect_publication_snapshot(read_record, fetch_page, target,
provider_contract=self.contract())
contract = self.contract(); contract['version']['version'] = 'other'
with self.assertRaises(fixture.mod.ValidationError):
fixture.mod.collect_publication_snapshot(read_record, fetch_page, fixture.result()['target'],
provider_contract=contract)
self.assertEqual(calls, [])
def test_collector_legacy_shape_and_complete_source_page_validation(self):
before, after, review, result, paths = self.snapshots()
prefix = '/repos/acme/repo'
records = {prefix + '/issues/3': before['issue'], prefix + '/pulls/7': before['pr'],
prefix + '/issues/7': before['pr_issue']}
calls = []
def fetch_page(endpoint, page):
calls.append((endpoint, page))
return {'items': [], 'total_count': 0, 'has_more': False}
legacy = fixture.mod.collect_publication_snapshot(records.__getitem__, fetch_page, result['target'])
self.assertEqual(set(legacy), {'issue', 'pr', 'issue_comments', 'pr_comments', 'reviews', 'timeline', 'files'})
self.assertNotIn((prefix + '/issues/3/timeline', 1), calls)
for envelope in ({'items': [{'id': 1}, {'id': 1}], 'total_count': 2, 'has_more': False},
{'items': [], 'total_count': 1, 'has_more': True},
{'items': None, 'has_more': False}):
def bad_page(endpoint, page):
if endpoint == prefix + '/issues/3/timeline':
return envelope
return {'items': [], 'total_count': 0, 'has_more': False}
with self.subTest(envelope=envelope), self.assertRaises(fixture.mod.ValidationError):
fixture.mod.collect_publication_snapshot(records.__getitem__, bad_page, result['target'],
provider_contract=self.contract())
def test_collector_reads_source_timeline_and_pr_issue_through_readonly_callbacks(self):
before, after, review, result, paths = self.snapshots()
mod = fixture.mod
self.assertTrue(callable(getattr(mod, 'collect_publication_snapshot', None)),
'canonical read-only publication snapshot collector missing')
prefix = '/repos/acme/repo'
endpoints = {'issue': '/issues/3', 'pr': '/pulls/7', 'pr_issue': '/issues/7',
'issue_comments': '/issues/3/comments', 'pr_comments': '/issues/7/comments',
'reviews': '/pulls/7/reviews', 'timeline': '/issues/7/timeline',
'issue_timeline': '/issues/3/timeline', 'files': '/pulls/7/files'}
captured = []
for snapshot in (before, after):
records = {prefix + endpoints[k]: v for k, v in snapshot.items()}
calls = []
def read_record(endpoint):
calls.append(('GET', endpoint))
return copy.deepcopy(records[endpoint])
def fetch_page(endpoint, page):
calls.append(('GET-page', endpoint, page))
items = records[endpoint]
return {'items': copy.deepcopy(items), 'total_count': len(items), 'has_more': False}
out = mod.collect_publication_snapshot(read_record, fetch_page, result['target'],
provider_contract=self.contract())
self.assertEqual(out, snapshot)
self.assertIn(('GET', prefix + '/issues/7'), calls)
self.assertIn(('GET-page', prefix + '/issues/3/timeline', 1), calls)
self.assertTrue(all(call[0].startswith('GET') for call in calls))
captured.append(out)
self.assertEqual(fixture.verify_fixture(*captured, review, result, paths,
self.contract())['source_issue_reference_ids'], [102])
if __name__ == '__main__':
unittest.main()