"""Integration regression tests for coordinator-backed initial delivery.""" import importlib.util import sys import unittest from pathlib import Path ROOT = Path(__file__).resolve().parents[3] SCRIPT = ROOT / "harness/agents/impl/skills/gitea-issue-implementation/scripts/implementation_delivery.py" sys.path.insert(0, str(ROOT)) spec = importlib.util.spec_from_file_location("implementation_delivery", SCRIPT) delivery = importlib.util.module_from_spec(spec) spec.loader.exec_module(delivery) from harness.lib.impl_operation import Approval, Coordinator, Intent HEAD = "a" * 40 class Adapter: def __init__(self): self.events=[]; self.number=7; self.fail_pr=False; self.uncertain=False; self.received=[] def push_operation(self, intent): self.received.append(intent) return self.push(intent.payload['head']) def create_or_reuse_pr_operation(self, intent): self.received.append(intent) return self.create_or_reuse_pr(intent.payload['head']) def push(self, head): self.events.append("push"); return True def read_remote_head(self): self.events.append("read-head"); return HEAD def create_or_reuse_pr(self, head): self.events.append("create-pr") if self.fail_pr: raise RuntimeError("failed") if self.uncertain: raise RuntimeError("uncertain ack") return self.number def read_pr(self, number): self.events.append("read-pr"); return {"number":number,"head":HEAD,"state":"open"} def preflight_operation(self, intent): self.events.append("preflight:"+intent.operation); return {"fresh":True,"scope":intent.target} def verify_operation(self, intent, before, observed): self.events.append("verify:"+intent.operation); return True def find_intended_pr(self, head): self.events.append("find-pr"); return None if self.fail_pr else {"number":self.number} def make_intents(): push=Intent("v1","push","org/repo",42,HEAD,{"head":HEAD}) pr=Intent("v1","pr.create","org/repo",42,HEAD,{"head":HEAD,"body":"approved body","base":"main"}) return push, Approval("v1",push.digest,"approved"), pr, Approval("v1",pr.digest,"approved") class IntegrationTests(unittest.TestCase): def invoke(self, adapter, intents=None, approvals=None, recorder=None): p,pa,r,ra=make_intents(); intents=intents or (p,r); approvals=approvals or (pa,ra) return delivery.deliver_approved_implementation(adapter,head=HEAD,declaration={"mode":"none"},push_intent=intents[0],push_approval=approvals[0],pr_intent=intents[1],pr_approval=approvals[1],coordinator=Coordinator(),on_attempt=recorder or (lambda entry: None)) def test_full_composed_delivery(self): a=Adapter(); result=self.invoke(a) self.assertEqual(result["status"],"verified"); self.assertEqual(result["artifact"]["pr"]["number"],7) self.assertEqual(a.events.count("push"),1); self.assertEqual(a.events.count("create-pr"),1) def test_uncertain_pr_acknowledgement_reconciles_without_second_write(self): a = Adapter() a.uncertain = True result = self.invoke(a) self.assertEqual(result["status"], "verified") self.assertEqual(a.events.count("create-pr"), 1) self.assertEqual(a.events.count("find-pr"), 1) def test_adoption_rejection_never_promotes_existing_remote_state(self): a = Adapter() old = delivery._delivery_adoption.require_adoption delivery._delivery_adoption.require_adoption = lambda *args, **kwargs: (_ for _ in ()).throw(ValueError("rejected")) try: result = self.invoke(a) finally: delivery._delivery_adoption.require_adoption = old self.assertEqual(result["status"], "partial") self.assertNotIn("push", a.events) self.assertNotIn("create-pr", a.events) def test_approved_payload_head_mismatch_rejected_before_any_write(self): a = Adapter() push, push_approval, pr, pr_approval = make_intents() for operation in ('push', 'pr.create'): candidate = Intent('v1', operation, 'org/repo', 42, HEAD, {'head':'f'*40,'body':'approved'}) candidate_approval = Approval('v1',candidate.digest,'approved') intents = (candidate,pr) if operation == 'push' else (push,candidate) approvals = (candidate_approval,pr_approval) if operation == 'push' else (push_approval,candidate_approval) with self.subTest(operation=operation),self.assertRaises(ValueError): self.invoke(a,intents=intents,approvals=approvals) self.assertNotIn('push',a.events) self.assertNotIn('create-pr',a.events) def test_mutation_callbacks_receive_entire_approved_intent(self): a = Adapter() push, push_approval, pr, pr_approval = make_intents() self.assertEqual(self.invoke(a)['status'],'verified') self.assertEqual([item.as_dict() for item in a.received],[push.as_dict(),pr.as_dict()]) def test_new_adapter_contract_does_not_require_legacy_mutators(self): a = Adapter() a.push = None a.create_or_reuse_pr = None a.push_operation = lambda intent: a.events.append('push') a.create_or_reuse_pr_operation = lambda intent: a.events.append('create-pr') or 7 self.assertEqual(self.invoke(a)['status'],'verified') self.assertEqual(a.events.count('push'),1) self.assertEqual(a.events.count('create-pr'),1) def test_missing_approval_no_writes(self): a=Adapter(); p,pa,r,ra=make_intents(); ra=Approval("v1","0"*64,"approved") with self.assertRaises(ValueError): self.invoke(a,approvals=(pa,ra)) self.assertNotIn("push",a.events) def test_preflight_failure_no_mutation(self): a=Adapter(); a.preflight_operation=lambda intent:{"fresh":False} result=self.invoke(a); self.assertEqual(result["status"],"partial"); self.assertNotIn("push",a.events) def test_pr_failure_preserves_verified_push(self): a=Adapter(); a.fail_pr=True; result=self.invoke(a) self.assertEqual(result["status"],"partial"); self.assertEqual(result["push"].status,"verified"); self.assertIsNone(result["artifact"]) def test_mutations_each_require_fresh_adoption(self): a=Adapter(); calls=[]; old=delivery._delivery_adoption.require_adoption delivery._delivery_adoption.require_adoption=lambda *args,**kwargs:calls.append(kwargs["expected_stage"]) try: self.invoke(a) finally: delivery._delivery_adoption.require_adoption=old self.assertEqual(len(calls),4) if __name__ == "__main__": unittest.main()