import importlib.util import os import tempfile import unittest from pathlib import Path SCRIPT = Path(__file__).parents[2] / 'agents/impl/skills/gitea-issue-rework/scripts/rework_input_gate.py' class InputGateTests(unittest.TestCase): def load(self): spec = importlib.util.spec_from_file_location('rework_input_gate', SCRIPT) module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module def test_valid_input_schema_is_accepted(self): gate = self.load() value = {'schema':'gitea-agent.rework-input/v1', 'target':{'owner':'acme','repository':'demo','issue':7,'pr':9}, 'head':'a'*40, 'contract_sha256':'b'*64} self.assertEqual(gate.validate_input(value), value) def test_initial_output_is_immutable_and_rejected_before_credential_or_get(self): cli_path = SCRIPT.with_name('rework_readback_cli.py') spec = importlib.util.spec_from_file_location('rework_readback_cli', cli_path) cli = importlib.util.module_from_spec(spec) spec.loader.exec_module(cli) with tempfile.TemporaryDirectory() as root: output = Path(root) / 'decision.json' output.write_bytes(b'old\n') output.chmod(0o600) mode = output.stat().st_mode with self.assertRaisesRegex(ValueError, 'output already exists'): cli.reject_existing_output(str(output)) self.assertEqual(output.read_bytes(), b'old\n') self.assertEqual(output.stat().st_mode, mode) class OrderedDecisionTests(unittest.TestCase): def setUp(self): import copy, hashlib from harness.tests.impl import test_rework_readback as rb from harness.tests.spec import test_implementation_spec_review_support as spec_tests self.rb = rb.ReworkReadbackTests(); self.rb.setUp(); self.addCleanup(self.rb.doCleanups) self.rb.repo['default_branch'] = 'main' self.rb.source['body'] = '\nSynthetic approved source contract.\n' self.inp = {'schema':'gitea-agent.rework-input/v1', 'target':{'owner':'acme','repository':'demo','issue':7,'pr':9}, 'head':'b'*40, 'contract_sha256':hashlib.sha256(self.rb.source['body'].encode()).hexdigest()} self.rows=[] for role, rid, uid in [('gitea-spec-agent',301,33), ('gitea-review-agent',302,34)]: row = copy.deepcopy(self.rb.reviews[0]); row.update(id=rid, commit_id=self.inp['head'], state='APPROVED', user={'id':uid,'login':role}, stale=False, dismissed=False, html_url=self.rb.pr['html_url']+'#pullreview-'+str(rid)) if role == 'gitea-spec-agent': result = spec_tests.result(); result['target'] = dict(self.inp['target']); result['target']['pull'] = result['target'].pop('pr'); result['target']['head']=self.inp['head']; result['contract_sha256']=self.inp['contract_sha256'] row['body']=spec_tests.mod.render_review(result, {'src/a.py','tests/test_a.py'}, declaration={'mode':'none'}) else: row['body']=self.code_body('approve') self.rows.append(row) self.set_rows(self.rows) original = self.rb.request def request(path): from urllib.parse import urlsplit import re if re.fullmatch(r'/repos/acme/demo/pulls/9/reviews/[1-9][0-9]*/comments', urlsplit(path).path): self.rb.calls.append(path); return [], {'X-Total-Count':'0'}, 200 return original(path) self.rb.request=request def code_body(self, verdict): return '\n- Verdict: `'+verdict+'`\nBounded native code-review fixture.\n' def set_rows(self, rows): import copy self.rb.reviews=copy.deepcopy(rows) self.rb.pr['comments']=self.rb.underlying['comments']=self.rb.pr['review_comments']=len(rows) def snapshot(self): return self.rb.adapter().input_snapshot(self.inp) def test_real_spec_renderer_approval_is_accepted(self): receipt=self.snapshot() self.assertEqual(receipt['status'],'no_op',receipt['reason_codes']) def test_fresh_source_contract_digest_is_enforced(self): self.assertEqual(self.snapshot()['status'], 'no_op') self.rb.source['body'] += 'Unexpected contract mutation.\n' with self.assertRaisesRegex(ValueError, 'contract|digest'): self.snapshot() def test_native_state_marker_contradiction_is_blocked(self): import copy rows = copy.deepcopy(self.rows) code = next(row for row in rows if row['user']['login'] == 'gitea-review-agent') code['body'] = self.code_body('request_changes') code['state'] = 'APPROVED' self.set_rows(rows) receipt = self.snapshot() self.assertEqual(receipt['status'], 'blocked') self.assertIn('state_marker_contradiction', receipt['reason_codes']) def test_legacy_output_budget_is_unchanged(self): cli_path = SCRIPT.with_name('rework_readback_cli.py') spec = importlib.util.spec_from_file_location('legacy_budget_probe', cli_path) cli = importlib.util.module_from_spec(spec); spec.loader.exec_module(cli) with tempfile.TemporaryDirectory() as td: output=Path(td)/'legacy.json' try: with cli.OutputFile(output) as writer: writer.write({'data':'x'*1_000_001}) except ValueError as exc: self.fail('legacy output below original 16MB budget must remain accepted: '+str(exc)) self.assertGreater(output.stat().st_size, 1_000_000) def test_mixed_roles_permutations_require_rework_without_raw_mutation(self): import copy rows=copy.deepcopy(self.rows); rows[1].update(state='REQUEST_CHANGES',body=self.code_body('request_changes')) for ordered in (rows, list(reversed(rows))): self.set_rows(ordered); before=copy.deepcopy(self.rb.reviews) receipt=self.snapshot() self.assertEqual(receipt['status'],'rework',receipt['reason_codes']) self.assertEqual(receipt['decision']['selected_review_ids'], {'gitea-spec-agent':301,'gitea-review-agent':302}) self.assertEqual(self.rb.reviews,before) self.assertEqual(receipt['raw_input_snapshot']['reviews']['rows'], before) self.assertEqual([x['id'] for x in receipt['decision']['timeline']], [301,302]) def test_only_same_role_later_approval_supersedes_request(self): import copy rows=copy.deepcopy(self.rows); rows[1].update(state='REQUEST_CHANGES',body=self.code_body('request_changes')) self.set_rows(rows); self.assertEqual(self.snapshot()['status'],'rework') later=copy.deepcopy(self.rows[1]);later.update(id=303,submitted_at='2026-01-02T00:00:00Z',updated_at='2026-01-02T00:00:00Z',html_url=self.rb.pr['html_url']+'#pullreview-303') self.set_rows([later,*rows]); receipt=self.snapshot() self.assertEqual(receipt['status'],'no_op',receipt['reason_codes']) self.assertEqual(receipt['decision']['selected_review_ids']['gitea-review-agent'],303) def test_missing_stale_and_dismissed_current_roles_block(self): import copy self.assertEqual(self.snapshot()['status'],'no_op') for rows in ([self.rows[0]], [self.rows[1]], []): self.set_rows(rows); self.assertEqual(self.snapshot()['status'],'blocked') for field in ('stale','dismissed'): rows=copy.deepcopy(self.rows);rows[1][field]=True;self.set_rows(rows) self.assertEqual(self.snapshot()['status'],'blocked') def test_current_pending_and_manual_comments_never_allow_noop(self): import copy self.assertEqual(self.snapshot()['status'],'no_op') for state in ('PENDING','COMMENT'): rows=copy.deepcopy(self.rows);rows[1]['state']=state;self.set_rows(rows) self.assertEqual(self.snapshot()['status'],'blocked') def test_blocked_spec_comment_dominates_code_request_changes(self): import copy from harness.tests.spec import test_implementation_spec_review_support as spec_tests result=spec_tests.result('blocked_spec');result['target']={'owner':'acme','repository':'demo','issue':7,'pull':9,'head':self.inp['head']};result['contract_sha256']=self.inp['contract_sha256'];result['criteria'][0]['result']='indeterminate';result['scope']['spec_conflicts']=['src/a.py'] rows=copy.deepcopy(self.rows);rows[0].update(state='COMMENT',body=spec_tests.mod.render_review(result,{'src/a.py','tests/test_a.py'},declaration={'mode':'none'}));rows[1].update(state='REQUEST_CHANGES',body=self.code_body('request_changes')) self.set_rows(rows);receipt=self.snapshot();self.assertEqual(receipt['status'],'blocked') self.assertIn('blocked_spec',receipt['reason_codes']) def test_marker_contract_and_target_defects_block_rework(self): import copy self.assertEqual(self.snapshot()['status'],'no_op') for body in ('missing marker',self.code_body('approve').replace('issue=7','issue=8'),self.code_body('approve').replace(self.inp['contract_sha256'],'f'*64)): rows=copy.deepcopy(self.rows);rows[1]['body']=body;self.set_rows(rows) self.assertEqual(self.snapshot()['status'],'blocked') def test_unknown_role_cannot_replace_authoritative_code_review(self): import copy self.assertEqual(self.snapshot()['status'],'no_op') unknown=copy.deepcopy(self.rows[1]);unknown['user']={'id':99,'login':'reader'} self.set_rows([self.rows[0],unknown]);receipt=self.snapshot() self.assertEqual(receipt['status'],'blocked');self.assertIn('unknown_role',receipt['decision']['timeline'][1]['reason_codes']) def test_native_rows_source_marker_and_default_branch_fail_closed(self): import copy self.assertEqual(self.snapshot()['status'],'no_op') for field,bad in [('id',True),('state','UNRECOGNIZED'),('stale','false'),('dismissed',None),('submitted_at','not-a-timestamp')]: rows=copy.deepcopy(self.rows);rows[1][field]=bad;self.set_rows(rows) with self.subTest(field=field),self.assertRaises(ValueError):self.snapshot() self.set_rows(self.rows) original=self.rb.source['body'];self.rb.source['body']=original.replace(' -->',' --> suffix',1) import hashlib self.inp['contract_sha256']=hashlib.sha256(self.rb.source['body'].encode()).hexdigest() with self.assertRaisesRegex(ValueError,'discovery'):self.snapshot() self.rb.source['body']=original;self.inp['contract_sha256']=hashlib.sha256(original.encode()).hexdigest();self.rb.pr['base']['ref']='other' with self.assertRaisesRegex(ValueError,'default branch'):self.snapshot() def test_duplicate_review_ids_and_capture_drift_fail_closed(self): import copy self.assertEqual(self.snapshot()['status'],'no_op') self.set_rows([*self.rows,copy.deepcopy(self.rows[1])]) with self.assertRaisesRegex(ValueError,'duplicate'):self.snapshot() self.set_rows(self.rows);old=self.rb.request;users=[0] def request(path): if path=='/user': users[0]+=1 if users[0]==2:self.rb.source['title']='Changed during second capture' return old(path) self.rb.request=request with self.assertRaisesRegex(ValueError,'concurrent provider drift'):self.snapshot() self.assertEqual(users[0],2) def test_timezone_and_nanosecond_order_is_exact(self): import copy rows=copy.deepcopy(self.rows);rows[1].update(state='REQUEST_CHANGES',body=self.code_body('request_changes'),submitted_at='2026-01-01T00:00:00.000000001Z',updated_at='2026-01-01T00:00:00.000000001Z') later=copy.deepcopy(self.rows[1]);later.update(id=303,submitted_at='2026-01-01T01:00:00.000000002+01:00',updated_at='2026-01-01T01:00:00.000000002+01:00',html_url=self.rb.pr['html_url']+'#pullreview-303') self.set_rows([later,*rows]);receipt=self.snapshot();self.assertEqual(receipt['status'],'no_op');self.assertEqual(receipt['decision']['selected_review_ids']['gitea-review-agent'],303) def test_real_public_cli_runs_gate_without_selected_review(self): import hashlib,json,sys from unittest.mock import patch as mock def load(name,path): spec=importlib.util.spec_from_file_location(name,path);mod=importlib.util.module_from_spec(spec);spec.loader.exec_module(mod);return mod cli=load('gate_cli_test',SCRIPT.with_name('rework_readback_cli.py'));command=load('gate_command_test',SCRIPT.with_name('rework_comment.py')) with tempfile.TemporaryDirectory() as td: root=Path(td);target=root/'target.json';provider=root/'provider.json';output=root/'decision.json' target.write_text(json.dumps(self.inp));provider.write_text(json.dumps(self.rb.contract));target.chmod(0o600);provider.chmod(0o600) args=['rework_comment.py','input-gate','--target',str(target),'--target-sha256',hashlib.sha256(target.read_bytes()).hexdigest(),'--provider-contract',str(provider),'--provider-contract-sha256',hashlib.sha256(provider.read_bytes()).hexdigest(),'--actor-id','42','--output',str(output)] with mock.dict(sys.modules,{'rework_readback_cli':cli}),mock.object(sys,'argv',args),mock.dict(os.environ,{'HERMES_HOME':td}),mock.object(cli,'load_role_token',return_value='PUBLIC'+'-fixture-value') as loader,mock.object(cli,'RestTransport',return_value=self.rb.request): self.assertEqual(command.main(),0);self.assertTrue(loader.called) self.assertEqual(json.loads(output.read_text())['status'],'no_op');self.assertEqual(output.stat().st_mode&0o777,0o600) import copy rows=copy.deepcopy(self.rows);rows[1].update(state='REQUEST_CHANGES',body=self.code_body('request_changes')) self.set_rows(list(reversed(rows)));output=root/'rework-decision.json';args[-1]=str(output) with mock.dict(sys.modules,{'rework_readback_cli':cli}),mock.object(sys,'argv',args),mock.dict(os.environ,{'HERMES_HOME':td}),mock.object(cli,'load_role_token',return_value='PUBLIC'+'-fixture-value'),mock.object(cli,'RestTransport',return_value=self.rb.request): self.assertEqual(command.main(),0) actual=json.loads(output.read_text());self.assertEqual(actual['status'],'rework');self.assertEqual(actual['decision']['rework_review_ids'],[302]) saved=output.read_bytes() with mock.dict(sys.modules,{'rework_readback_cli':cli}),mock.object(sys,'argv',args),mock.object(cli,'load_role_token') as loader,mock.object(cli,'RestTransport') as transport: self.assertEqual(command.main(),2);loader.assert_not_called();transport.assert_not_called() self.assertEqual(output.read_bytes(),saved) def test_initial_output_atomic_publish_preserves_race_created_file(self): from unittest.mock import patch as mock spec=importlib.util.spec_from_file_location('no_clobber_test',SCRIPT.with_name('rework_readback_cli.py'));cli=importlib.util.module_from_spec(spec);spec.loader.exec_module(cli) with tempfile.TemporaryDirectory() as td: output=Path(td)/'decision.json';original_link=cli.os.link with cli.OutputFile(output,initial_only=True) as writer: def race_link(*args,**kwargs): output.write_bytes(b'raced output');output.chmod(0o600);return original_link(*args,**kwargs) with mock.object(cli.os,'link',side_effect=race_link),self.assertRaisesRegex(ValueError,'created during capture'):writer.write({'decision':'bounded'}) self.assertEqual(output.read_bytes(),b'raced output');self.assertEqual(output.stat().st_mode&0o777,0o600) self.assertEqual([x.name for x in Path(td).iterdir()],['decision.json']) def test_initial_output_total_byte_budget_rejects_without_artifact(self): spec=importlib.util.spec_from_file_location('size_gate_test',SCRIPT.with_name('rework_readback_cli.py'));cli=importlib.util.module_from_spec(spec);spec.loader.exec_module(cli) with tempfile.TemporaryDirectory() as td: output=Path(td)/'decision.json' with cli.OutputFile(output,initial_only=True) as writer,self.assertRaisesRegex(ValueError,'output size'):writer.write({'raw':'x'*1_000_001}) self.assertFalse(output.exists()) def test_old_head_state_contradiction_is_history_only(self): import copy self.assertEqual(self.snapshot()['status'],'no_op') old=copy.deepcopy(self.rows[1]);old.update(id=303,commit_id='a'*40, body=self.code_body('request_changes').replace(self.inp['head'],'a'*40), html_url=self.rb.pr['html_url']+'#pullreview-303') self.set_rows([old,*self.rows]);receipt=self.snapshot() self.assertEqual(receipt['status'],'no_op',receipt['reason_codes']) history=next(x for x in receipt['decision']['timeline'] if x['id']==303) self.assertIn('old_head',history['reason_codes']);self.assertFalse(history['eligible']) def test_same_instant_body_conflict_has_no_selected_review(self): import copy self.assertEqual(self.snapshot()['status'],'no_op') other=copy.deepcopy(self.rows[1]);other.update(id=303,body=other['body']+'Different finding text.\n',html_url=self.rb.pr['html_url']+'#pullreview-303') self.set_rows([other,*self.rows]);receipt=self.snapshot() self.assertEqual(receipt['status'],'blocked');self.assertIn('conflicting_same_time_reviews',receipt['reason_codes']) self.assertIsNone(receipt['decision']['selected_review_ids']['gitea-review-agent']) def test_ambiguous_code_marker_or_verdict_lines_block(self): import copy self.assertEqual(self.snapshot()['status'],'no_op') for suffix in ('- Verdict: malformed\n','Quoted marker: '+self.code_body('approve').splitlines()[0]+'\n'): rows=copy.deepcopy(self.rows);rows[1]['body']+=suffix;self.set_rows(rows) with self.subTest(suffix=suffix):self.assertEqual(self.snapshot()['status'],'blocked') def test_invalid_timezone_minute_is_rejected(self): import copy self.assertEqual(self.snapshot()['status'],'no_op') rows=copy.deepcopy(self.rows);rows[1]['submitted_at']='2026-01-01T00:00:00+00:99';self.set_rows(rows) with self.assertRaisesRegex(ValueError,'timestamp|submitted_at'):self.snapshot() def test_rework_targets_exclude_spec_approvals(self): import copy self.assertEqual(self.snapshot()['status'],'no_op') rows=copy.deepcopy(self.rows);rows[1].update(state='REQUEST_CHANGES',body=self.code_body('request_changes')) self.set_rows(rows);receipt=self.snapshot();self.assertEqual(receipt['status'],'rework') self.assertEqual(receipt['decision'].get('rework_review_ids'),[302]) def test_cli_preflight_rejects_bool_target_before_credentials(self): import hashlib,json,sys,copy from unittest.mock import patch as mock spec=importlib.util.spec_from_file_location('cli_preflight_test',SCRIPT.with_name('rework_readback_cli.py'));cli=importlib.util.module_from_spec(spec);spec.loader.exec_module(cli) spec=importlib.util.spec_from_file_location('cmd_preflight_test',SCRIPT.with_name('rework_comment.py'));cmd=importlib.util.module_from_spec(spec);spec.loader.exec_module(cmd) with tempfile.TemporaryDirectory() as td: root=Path(td);target=root/'target.json';provider=root/'provider.json';output=root/'decision.json' bad=copy.deepcopy(self.inp);bad['target']['issue']=True;target.write_text(json.dumps(bad));provider.write_text(json.dumps(self.rb.contract));target.chmod(0o600);provider.chmod(0o600) args=['command','input-gate','--target',str(target),'--target-sha256',hashlib.sha256(target.read_bytes()).hexdigest(),'--provider-contract',str(provider),'--provider-contract-sha256',hashlib.sha256(provider.read_bytes()).hexdigest(),'--actor-id','42','--output',str(output)] with mock.dict(sys.modules,{'rework_readback_cli':cli}),mock.object(sys,'argv',args),mock.object(cli,'load_role_token') as loader,mock.object(cli,'RestTransport') as transport: self.assertEqual(cmd.main(),2);loader.assert_not_called();transport.assert_not_called() self.assertFalse(output.exists()) def test_cli_capture_race_preserves_foreign_output(self): import hashlib,json,sys from unittest.mock import patch as mock spec=importlib.util.spec_from_file_location('cli_race_test',SCRIPT.with_name('rework_readback_cli.py'));cli=importlib.util.module_from_spec(spec);spec.loader.exec_module(cli) spec=importlib.util.spec_from_file_location('cmd_race_test',SCRIPT.with_name('rework_comment.py'));cmd=importlib.util.module_from_spec(spec);spec.loader.exec_module(cmd) with tempfile.TemporaryDirectory() as td: root=Path(td);target=root/'target.json';provider=root/'provider.json';output=root/'decision.json' target.write_text(json.dumps(self.inp));provider.write_text(json.dumps(self.rb.contract));target.chmod(0o600);provider.chmod(0o600) args=['command','input-gate','--target',str(target),'--target-sha256',hashlib.sha256(target.read_bytes()).hexdigest(),'--provider-contract',str(provider),'--provider-contract-sha256',hashlib.sha256(provider.read_bytes()).hexdigest(),'--actor-id','42','--output',str(output)] requests=[] def transport(path): requests.append(path) if len(requests)==1:output.write_bytes(b'created concurrently');output.chmod(0o600) return self.rb.request(path) with mock.dict(sys.modules,{'rework_readback_cli':cli}),mock.object(sys,'argv',args),mock.dict(os.environ,{'HERMES_HOME':td}),mock.object(cli,'load_role_token',return_value='PUBLIC'+'-fixture-value'),mock.object(cli,'RestTransport',return_value=transport): self.assertEqual(cmd.main(),2) self.assertGreater(len(requests),1);self.assertEqual(output.read_bytes(),b'created concurrently');self.assertEqual(output.stat().st_mode&0o777,0o600) if __name__ == '__main__': unittest.main()