183 lines
8.0 KiB
Python
183 lines
8.0 KiB
Python
"""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()
|