"""Contract tests for branch-bound repair leases (16 tests).""" from __future__ import annotations import tempfile import unittest from datetime import datetime, timedelta, timezone from pathlib import Path from branch_repair_lease import ( MAX_LEASE_SECONDS, TASK_ID, GitProbeProtocol, LeaseError, LeaseStore, RepairLeaseManager, normalize_path, path_is_protected, ) class FakeGit(GitProbeProtocol): def __init__(self) -> None: self.heads = {"fix/lease": "a" * 40, "candidate": "a" * 40} self.parents: dict[str, str] = {} self.paths: dict[str, tuple[str, ...]] = {} def branch_head(self, branch: str) -> str: return self.heads[branch] def parent(self, commit: str) -> str: return self.parents[commit] def is_ancestor(self, older: str, newer: str) -> bool: current = newer seen: set[str] = set() while current not in seen: if current == older: return True seen.add(current) current = self.parents.get(current, "") if not current: return False return False def changed_paths(self, commit: str) -> tuple[str, ...]: return self.paths[commit] def commits_between(self, older: str, newer: str) -> tuple[str, ...]: result: list[str] = [] current = newer while current != older: result.append(current) current = self.parents[current] return tuple(reversed(result)) class LeaseTests(unittest.TestCase): def setUp(self) -> None: self.now = datetime(2026, 8, 6, 7, 0, tzinfo=timezone.utc) self.git = FakeGit() self.temp = tempfile.TemporaryDirectory() self.manager = RepairLeaseManager(LeaseStore(Path(self.temp.name) / "leases"), self.git, clock=lambda: self.now) self.paths = ("active-runtime/app/branch_repair_lease.py",) def tearDown(self) -> None: self.temp.cleanup() def issue(self): return self.manager.issue( task_id=TASK_ID, owner_identity="aiman", branch="fix/lease", description="repair", allowed_paths=self.paths, ttl_seconds=3600, ) def test_01_issue_persists_exact_head(self) -> None: lease = self.issue() self.assertEqual(lease.current_head_oid, "a" * 40) self.assertEqual(self.manager.store.load(lease.lease_id).state, "ACTIVE") def test_02_main_branch_rejected(self) -> None: self.git.heads["main"] = "a" * 40 with self.assertRaises(LeaseError): self.manager.issue(task_id=TASK_ID, owner_identity="aiman", branch="main", description="x", allowed_paths=self.paths) def test_03_ttl_maximum_enforced(self) -> None: with self.assertRaises(LeaseError): self.manager.issue(task_id=TASK_ID, owner_identity="aiman", branch="fix/lease", description="x", allowed_paths=self.paths, ttl_seconds=MAX_LEASE_SECONDS + 1) def test_04_attach_requires_exact_head(self) -> None: lease = self.issue() attached = self.manager.attach_existing_branch(lease.lease_id, owner_identity="aiman", expected_head_oid="a" * 40) self.assertTrue(attached.attached) def test_05_attach_stale_head_rejected(self) -> None: lease = self.issue() with self.assertRaises(LeaseError): self.manager.attach_existing_branch(lease.lease_id, owner_identity="aiman", expected_head_oid="b" * 40) def test_06_commit_requires_attach(self) -> None: lease = self.issue() with self.assertRaises(LeaseError): self.manager.authorize_commit(lease.lease_id, owner_identity="aiman", expected_head_oid="a" * 40, paths=self.paths) def test_07_prohibited_path_rejected(self) -> None: self.assertTrue(path_is_protected("VERSION")) lease = self.issue() self.manager.attach_existing_branch(lease.lease_id, owner_identity="aiman", expected_head_oid="a" * 40) with self.assertRaises(LeaseError): self.manager.authorize_commit(lease.lease_id, owner_identity="aiman", expected_head_oid="a" * 40, paths=("VERSION",)) def test_08_secret_content_rejected(self) -> None: lease = self.issue() self.manager.attach_existing_branch(lease.lease_id, owner_identity="aiman", expected_head_oid="a" * 40) with self.assertRaises(LeaseError): self.manager.authorize_commit(lease.lease_id, owner_identity="aiman", expected_head_oid="a" * 40, paths=self.paths, content={self.paths[0]: "tok" + "en=" + "x" * 24}) def test_09_first_commit_is_fast_forward(self) -> None: lease = self.issue() self.manager.attach_existing_branch(lease.lease_id, owner_identity="aiman", expected_head_oid="a" * 40) b = "b" * 40 self.git.parents[b] = "a" * 40 self.git.paths[b] = self.paths self.git.heads["fix/lease"] = b updated = self.manager.record_commit(lease.lease_id, owner_identity="aiman", expected_head_oid="a" * 40, new_head_oid=b, paths=self.paths) self.assertEqual(updated.current_head_oid, b) def test_10_multiple_commits_allowed(self) -> None: self.test_09_first_commit_is_fast_forward() lease_id = next(Path(self.temp.name, "leases").glob("lease-*.json")).stem b = "b" * 40 c = "c" * 40 self.git.parents[b] = "a" * 40; self.git.parents[c] = b self.git.paths[b] = self.paths; self.git.paths[c] = self.paths self.git.heads["fix/lease"] = c updated = self.manager.record_commit(lease_id, owner_identity="aiman", expected_head_oid=b, new_head_oid=c, paths=self.paths) self.assertEqual(updated.revision, 3) def test_11_stale_head_rejected(self) -> None: lease = self.issue() self.manager.attach_existing_branch(lease.lease_id, owner_identity="aiman", expected_head_oid="a" * 40) self.git.heads["fix/lease"] = "b" * 40 with self.assertRaises(LeaseError): self.manager.authorize_commit(lease.lease_id, owner_identity="aiman", expected_head_oid="a" * 40, paths=self.paths) def test_12_recover_descendant(self) -> None: lease = self.issue() self.manager.attach_existing_branch(lease.lease_id, owner_identity="aiman", expected_head_oid="a" * 40) b = "b" * 40; self.git.parents[b] = "a" * 40; self.git.paths[b] = self.paths; self.git.heads["fix/lease"] = b recovered = self.manager.recover(lease.lease_id, owner_identity="aiman") self.assertEqual(recovered.current_head_oid, b) def test_13_recover_protected_change_fails_closed(self) -> None: lease = self.issue(); self.manager.attach_existing_branch(lease.lease_id, owner_identity="aiman", expected_head_oid="a" * 40) b = "b" * 40; self.git.parents[b] = "a" * 40; self.git.paths[b] = ("VERSION",); self.git.heads["fix/lease"] = b with self.assertRaises(LeaseError): self.manager.recover(lease.lease_id, owner_identity="aiman") def test_14_restart_recovers_persisted_lease(self) -> None: lease = self.issue(); self.manager.attach_existing_branch(lease.lease_id, owner_identity="aiman", expected_head_oid="a" * 40) restarted = RepairLeaseManager(self.manager.store, self.git, clock=lambda: self.now) self.assertTrue(restarted.store.load(lease.lease_id).attached) def test_15_renew_respects_absolute_cap(self) -> None: lease = self.issue() with self.assertRaises(LeaseError): self.manager.renew(lease.lease_id, owner_identity="aiman", ttl_seconds=MAX_LEASE_SECONDS + 1) def test_16_complete_and_revoke_are_terminal(self) -> None: completed = self.issue() self.assertEqual(self.manager.complete(completed.lease_id, owner_identity="aiman").state, "COMPLETED") with self.assertRaises(LeaseError): self.manager.recover(completed.lease_id, owner_identity="aiman") revoked = self.issue() self.assertEqual(self.manager.revoke(revoked.lease_id, owner_identity="aiman").state, "REVOKED") if __name__ == "__main__": unittest.main()