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

fix: validate Gitea conversation caches and safe commit messages

Browse Source
This commit is contained in:
gitea-spec-agent
2026-10-13 13:50:10 +00:00
parent a7a13aa4d5
commit a74fb747a5
6 changed files with 227 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
harness/tests/impl/test_rework_readback.py
+165 -4
View File
@@ -185,7 +185,6 @@ class ReworkReadbackTests(unittest.TestCase):
('pr',['state'],'closed'), ('underlying',['number'],10),
('underlying',['pull_request'],True), ('reviews',[0,'commit_id'],'c'*40)]
error = self.load('rework_readback').ReadbackError
# Each load has a distinct exception class; assert ValueError at the public boundary.
self.assertTrue(issubclass(error, ValueError))
for attribute, keys, value in cases:
with self.subTest(attribute=attribute, keys=keys):
@@ -213,7 +212,7 @@ class ReworkReadbackTests(unittest.TestCase):
(first,self.comments[:50],{},200),
(first,[{'id':True}],{'X-Total-Count':'1'},200),
(first,[],{'X-Total-Count':'0','Link':'not a Link'},200),
(first,[],{'X-Total-Count':'0'},True), (first,[],{},403), (first,[],{},302)]
(first,[],{},True), (first,[],{},403), (first,[],{},302)]
for path, rows, headers, status in invalid:
with self.subTest(path=path, headers=headers, status=status):
self.overrides = {path:(rows,headers,status)}
@@ -299,7 +298,6 @@ class ReworkReadbackTests(unittest.TestCase):
baseline = adapter.snapshot(self.expected)
self.add_result()
after = adapter.snapshot(self.expected)
# Prove this exact clean fixture is admitted before adversarial mutation.
adapter.normalized_inputs(self.expected, baseline, after, 'create')
cases = [(['schema'],'forged'), (['actor_id'],True), (['origin'],'https://foreign.invalid/api/v1'),
(['target','owner'],'other'), (['provider','version'],'different'),
@@ -458,7 +456,6 @@ class ReworkReadbackTests(unittest.TestCase):
with self.assertRaises(ValueError): self.adapter().snapshot(self.expected)
def test_native_Gitea_PR_URL_and_omitted_proven_zero_inline_count(self):
# Pinned primary serializers: services/convert/pull.go and modules/structs/pull.go (v1.24.6).
self.pr['url'] = self.pr['html_url']
self.contract['inline']['components'] = ['inline']
self.pr.pop('review_comments')
@@ -543,3 +540,167 @@ class ReworkReadbackTests(unittest.TestCase):
reused = adapter.normalized_inputs(self.expected, after, adapter.snapshot(self.expected), 'reuse')
self.assertEqual(self.legacy.verify_readbacks(self.expected, reused['pr'], reused['comments'],
reused['identity'], action='reuse', baseline_count=reused['baseline_count'])['status'], 'verified')
def _cache_adapter(self):
self.contract['version'] = '1.27.3'
self.overrides['/version'] = ({'version':'1.27.3'}, {}, 200)
self.contract['aggregate']['pr'] = {'field':'comments','components':['top_level','reviews'],
'cache_mode':'gitea-conversation-v1'}
self.contract['aggregate']['pr_issue'] = {'field':'comments','components':['top_level','reviews'],
'cache_mode':'gitea-conversation-v1'}
return self.adapter()
def _stale_baseline(self):
self.pr['comments'] = self.underlying['comments'] = 0
adapter = self._cache_adapter()
return adapter, adapter.snapshot(self.expected)
def test_cache_opt_in_captures_and_validates_bounded_stale_baseline(self):
adapter, baseline = self._stale_baseline()
self.assertEqual(baseline['normalized_counts']['top_level'], 0)
self.assertEqual(baseline['normalized_counts']['reviews'], 1)
self.assertEqual(baseline['raw']['pr']['comments'], 0)
self.assertEqual(baseline['raw']['pr_issue']['comments'], 0)
adapter.validate_snapshot(self.expected, baseline)
self.overrides['/version'] = ({'version':'1.24.6'}, {}, 200)
with self.assertRaisesRegex(ValueError, 'provider version mismatch'):
adapter.snapshot(self.expected)
def test_cache_create_recomputes_post_conversation_count_and_guards_deltas(self):
adapter, baseline = self._stale_baseline()
self.add_result()
self.pr['comments'] = self.underlying['comments'] = 2
after = adapter.snapshot(self.expected)
inputs = adapter.normalized_inputs(self.expected, baseline, after, 'create')
receipt = self.legacy.verify_readbacks(self.expected, inputs['pr'], inputs['comments'],
inputs['identity'], action='create', baseline_count=inputs['baseline_count'])
self.assertEqual(receipt['status'], 'verified')
# Capture a genuine pre-create synchronized baseline, not the post-create rows.
self.comments.clear()
self.pr['comments'] = self.underlying['comments'] = 1
self.pr['updated_at'] = self.underlying['updated_at'] = T0
synchronized = adapter.snapshot(self.expected)
self.add_result()
synchronized_after = adapter.snapshot(self.expected)
adapter.normalized_inputs(self.expected, synchronized, synchronized_after, 'create')
for bad in (1, None, -1, 3):
changed = copy.deepcopy(after)
changed['raw']['pr']['comments'] = bad
changed['raw']['pr_issue']['comments'] = bad
with self.subTest(after_counter=bad), self.assertRaises(ValueError):
adapter.normalized_inputs(self.expected, baseline, changed, 'create')
for mutation in ('old_comment','duplicate','native_review','source_issue'):
changed = copy.deepcopy(after)
if mutation == 'old_comment':
changed['raw']['comments']['rows'] = [{'id':400,'body':'mutated','created_at':T0,
'updated_at':T0,'user':{'id':8,'login':'other'}}] + changed['raw']['comments']['rows']
elif mutation == 'duplicate':
changed['raw']['comments']['rows'].append(copy.deepcopy(changed['raw']['comments']['rows'][0]))
elif mutation == 'native_review': changed['raw']['reviews']['rows'][0]['state'] = 'APPROVED'
else: changed['raw']['source_issue']['title'] = 'drift'
with self.subTest(mutation=mutation), self.assertRaises(ValueError):
adapter.normalized_inputs(self.expected, baseline, changed, 'create')
def test_cache_reuse_allows_same_lag_but_rejects_refresh(self):
self.add_result()
self.pr['comments'] = self.underlying['comments'] = 0
adapter = self._cache_adapter()
baseline = adapter.snapshot(self.expected)
same = adapter.snapshot(self.expected)
inputs = adapter.normalized_inputs(self.expected, baseline, same, 'reuse')
receipt = self.legacy.verify_readbacks(self.expected, inputs['pr'], inputs['comments'],
inputs['identity'], action='reuse', baseline_count=inputs['baseline_count'])
self.assertEqual(receipt['status'], 'verified')
self.pr['comments'] = self.underlying['comments'] = 2
refreshed = adapter.snapshot(self.expected)
with self.assertRaises(ValueError): adapter.normalized_inputs(self.expected, baseline, refreshed, 'reuse')
def test_cache_opt_in_rejects_invalid_mode_version_and_components(self):
module = self.load('rework_readback')
for mode, version, components in (
('unknown','1.27.3',['top_level','reviews']),
(None,'1.27.3',['top_level','reviews']),
(True,'1.27.3',['top_level','reviews']),
('gitea-conversation-v1','1.27.2',['top_level','reviews']),
('gitea-conversation-v1','1.27.3',['reviews']),
('gitea-conversation-v1','1.27.3',['top_level']),
('gitea-conversation-v1','1.27.3',['top_level','reviews','inline'])):
contract = copy.deepcopy(self.contract)
contract['version'] = version
contract['aggregate']['pr'] = {'field':'comments','components':components,'cache_mode':mode}
contract['aggregate']['pr_issue'] = {'field':'comments','components':components,'cache_mode':mode}
with self.subTest(mode=mode,version=version,components=components), self.assertRaises(ValueError):
module.ReworkReadbackAdapter(self.request, origin=ORIGIN+'/api/v1',
credential_source='$HERMES_HOME/.env:GITEA_TOKEN', actor_id=42, provider_contract=contract)
def test_strict_legacy_counter_semantics_are_unchanged_without_cache_mode(self):
self.pr['comments'] = self.underlying['comments'] = 0
with self.assertRaises(ValueError): self.adapter().snapshot(self.expected)
def test_cache_baseline_does_not_count_inline_records_as_conversation(self):
adapter = self._cache_adapter()
self.contract['inline']['components'] = ['reviews']
adapter = self.adapter()
self.reviews[0]['comments_count'] = 1
self.inline = [{'id':801,'body':'inline','created_at':T0,'updated_at':T0,
'user':{'id':33,'login':'gitea-review-agent'}}]
self.pr['comments'] = self.underlying['comments'] = 2
with self.assertRaisesRegex(ValueError, 'cached provider aggregate count exceeds collection'):
adapter.snapshot(self.expected)
def test_cache_baseline_rejects_malformed_missing_and_ahead_counters(self):
adapter = self._cache_adapter()
for scope in (self.pr, self.underlying):
for bad in (None, -1, True, '0', 1.0, 2):
scope['comments'] = bad
with self.subTest(counter=bad), self.assertRaises(ValueError):
adapter.snapshot(self.expected)
scope.pop('comments')
with self.assertRaises(ValueError): adapter.snapshot(self.expected)
scope['comments'] = 1
def test_cache_transition_rejects_mutation_of_a_genuine_existing_comment(self):
old = {'id':400, 'html_url':self.pr['html_url']+'#issuecomment-400',
'body':'preserved previous comment', 'created_at':T0, 'updated_at':T0,
'user':{'id':33,'login':'gitea-review-agent'}}
self.comments.append(old)
adapter, baseline = self._stale_baseline()
self.add_result()
self.pr['comments'] = self.underlying['comments'] = 3
after = adapter.snapshot(self.expected)
adapter.normalized_inputs(self.expected, baseline, after, 'create')
changed = copy.deepcopy(after)
changed['raw']['comments']['rows'][0]['body'] = 'unauthorized mutation'
with self.assertRaisesRegex(ValueError, 'existing comment changed or missing'):
adapter.normalized_inputs(self.expected, baseline, changed, 'create')
def test_cache_opt_in_does_not_relax_an_undeclared_aggregate_scope(self):
self._cache_adapter()
self.contract['aggregate']['pr_issue'].pop('cache_mode')
adapter = self.adapter()
self.pr['comments'] = self.underlying['comments'] = 0
with self.assertRaisesRegex(ValueError, 'provider aggregate count mismatch'):
adapter.snapshot(self.expected)
self.underlying['comments'] = 1
adapter.snapshot(self.expected)
def test_cache_provider_documentation_example_matches_executable_contract(self):
import re
doc = (DIRECTORY.parent/'references/readback.md').read_text()
match = re.search(r'```json\s*(.*?)\s*```', doc, re.S)
if match is None: self.fail('provider documentation requires a JSON example')
example = json.loads(match.group(1))
self._cache_adapter()
self.contract = example
self.adapter().snapshot(self.expected)
def test_shared_commit_message_safety_rule_is_linked_and_amend_is_narrow(self):
common = (DIRECTORY.parents[4]/'operations/repository-cache.md').read_text()
skill = (DIRECTORY.parent/'SKILL.md').read_text()
self.assertIn('## Commit-message reference safety', common)
for clause in ('complete subject and body', 'cross-repository',
'non-default branch', 'explicit user authorization',
'commit is unpublished', 'tree is identical',
'PR body', 'not general permission to amend'):
self.assertIn(clause, common)
self.assertIn('repository-cache.md#commit-message-reference-safety', skill)