import importlib.util import sys from pathlib import Path from tempfile import TemporaryDirectory import unittest ROOT = Path(__file__).resolve().parents[3] PATH = ROOT / "harness/agents/impl/skills/gitea-issue-implementation/scripts/implementation_delivery.py" INTEGRATION = ROOT / "harness/tests/common/test_delegation_consumers_integration.py" def load(path, name): spec = importlib.util.spec_from_file_location(name, path) module = importlib.util.module_from_spec(spec) sys.modules[name] = module spec.loader.exec_module(module) return module class RecordingAdapter: def __init__(self, head, *, remote_head=None, pr=None, mutate=None): self.head = head self.remote_head = remote_head if remote_head is not None else head self.pr = pr if pr is not None else {"number": 17, "head": head, "state": "open"} self.mutate = mutate self.calls = [] def push(self, head): self.calls.append(("push", head)) if self.mutate: self.mutate("push") def read_remote_head(self): self.calls.append(("read_remote_head",)) if self.mutate: self.mutate("read_remote_head") return self.remote_head def create_or_reuse_pr(self, head): self.calls.append(("create_or_reuse_pr", head)) if self.mutate: self.mutate("create_or_reuse_pr") return self.pr["number"] def read_pr(self, number): self.calls.append(("read_pr", number)) return self.pr class ImplementationDeliveryTests(unittest.TestCase): @classmethod def setUpClass(cls): cls.delivery = load(PATH, "implementation_delivery_under_test") cls.evidence = load(INTEGRATION, "implementation_delivery_evidence") def setUp(self): self.tmp = TemporaryDirectory(dir="/opt/data/cache/scratch") self.ev = self.evidence.RealEvidence(Path(self.tmp.name)) self.ev.set_stage("implementation-delivery") def tearDown(self): self.tmp.cleanup() def test_real_git_delivery_pushes_once_creates_once_and_returns_verified_readback(self): adapter = RecordingAdapter(self.ev.revision) result = self.delivery.deliver_implementation( adapter, head=self.ev.revision, declaration=self.ev.declaration() ) self.assertEqual(result, { "remote_head": self.ev.revision, "pr": {"number": 17, "head": self.ev.revision, "state": "open"}, }) self.assertEqual( [call[0] for call in adapter.calls], ["push", "read_remote_head", "create_or_reuse_pr", "read_pr"], ) def test_explicit_none_is_checked_and_exact_readback_is_required(self): adapter = RecordingAdapter(self.ev.revision) result = self.delivery.deliver_implementation( adapter, head=self.ev.revision, declaration={"mode": "none"} ) self.assertEqual(result["remote_head"], self.ev.revision) self.assertEqual(result["pr"]["state"], "open") def test_missing_or_malformed_declaration_has_no_effects(self): for declaration in (None, {"mode": "bad"}): adapter = RecordingAdapter(self.ev.revision) with self.subTest(declaration=declaration), self.assertRaises(self.delivery.AdoptionError): self.delivery.deliver_implementation(adapter, head=self.ev.revision, declaration=declaration) self.assertEqual(adapter.calls, []) def test_remote_drift_prevents_pr_creation(self): adapter = RecordingAdapter(self.ev.revision, remote_head="0" * 40) with self.assertRaises(self.delivery.AdoptionError): self.delivery.deliver_implementation(adapter, head=self.ev.revision, declaration=self.ev.declaration()) self.assertEqual([call[0] for call in adapter.calls], ["push", "read_remote_head"]) def test_evidence_mutation_during_remote_readback_prevents_pr_creation(self): def mutate(phase): if phase == "read_remote_head": self.ev.bundle.joinpath("src/a.py").write_text("VALUE = 2\n") adapter = RecordingAdapter(self.ev.revision, mutate=mutate) with self.assertRaises(self.delivery.AdoptionError): self.delivery.deliver_implementation(adapter, head=self.ev.revision, declaration=self.ev.declaration()) self.assertEqual([call[0] for call in adapter.calls], ["push", "read_remote_head"]) def test_missing_adapter_capability_prevents_push(self): class Incomplete: def read_remote_head(self): return self.ev.revision with self.assertRaises(self.delivery.AdoptionError): self.delivery.deliver_implementation(Incomplete(), head=self.ev.revision, declaration={"mode": "none"}) def test_pr_readback_mismatch_does_not_complete(self): for pr in ( {"number": 17, "head": "0" * 40, "state": "open"}, {"number": 17, "head": self.ev.revision, "state": "closed"}, {"number": True, "head": self.ev.revision, "state": "open"}, ): adapter = RecordingAdapter(self.ev.revision, pr=pr) with self.subTest(pr=pr), self.assertRaises(self.delivery.AdoptionError): self.delivery.deliver_implementation(adapter, head=self.ev.revision, declaration=self.ev.declaration()) def test_workflow_can_load_from_external_cwd_without_pythonpath(self): import os import subprocess code = "import importlib.util,json; s=importlib.util.spec_from_file_location('delivery'," + repr(str(PATH)) + "); m=importlib.util.module_from_spec(s); s.loader.exec_module(m); head='a'*40; A=type('A',(),{'push':lambda self,h:None,'read_remote_head':lambda self:head,'create_or_reuse_pr':lambda self,h:17,'read_pr':lambda self,n:{'number':n,'head':head,'state':'open'}}); print(json.dumps(m.deliver_implementation(A(),head=head,declaration={'mode':'none'})))" env = os.environ.copy(); env.pop('PYTHONPATH', None) proc = subprocess.run([sys.executable, '-c', code], cwd=self.tmp.name, env=env, capture_output=True, text=True, timeout=10) self.assertEqual(proc.returncode, 0, 'canonical delivery must load independently of cwd') self.assertIn('remote_head', proc.stdout) def test_uncertain_write_is_not_retried(self): class FailingAdapter(RecordingAdapter): def create_or_reuse_pr(self, head): self.calls.append(("create_or_reuse_pr", head)) raise RuntimeError("transport uncertain") adapter = FailingAdapter(self.ev.revision) with self.assertRaises(RuntimeError): self.delivery.deliver_implementation(adapter, head=self.ev.revision, declaration=self.ev.declaration()) self.assertEqual([call[0] for call in adapter.calls], ["push", "read_remote_head", "create_or_reuse_pr"]) if __name__ == "__main__": unittest.main()