244 lines
11 KiB
Python
244 lines
11 KiB
Python
"""Tests for issue #970: safely resolve worktree bindings whose paths are missing.
|
|
|
|
Covers all acceptance criteria for #970:
|
|
- missing path reporting with repo/branch/issue/session/lease/host correlation
|
|
- fail closed on unavailable host or mount failure
|
|
- fail closed on moved worktree paths
|
|
- fail closed on live lease or active session/process
|
|
- atomic pre-mutation re-validation preventing recreation races
|
|
- targeted retirement preserving unrelated worktrees & metadata
|
|
- safe and idempotent repeated execution
|
|
- worktrees dimension resolution post-cleanup
|
|
- resolution of the two historical post-PR #968 records (#795 / #635)
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import shutil
|
|
import tempfile
|
|
import unittest
|
|
|
|
import control_plane_db as cpd
|
|
import lease_lifecycle
|
|
import missing_worktree_reconcile as mwr
|
|
|
|
|
|
class TestMissingWorktreeReconcile(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp_dir = tempfile.mkdtemp()
|
|
self.db_path = os.path.join(self.temp_dir, "control_plane.sqlite3")
|
|
self.db = cpd.ControlPlaneDB(self.db_path)
|
|
self.project_root = os.path.join(self.temp_dir, "repo")
|
|
os.makedirs(os.path.join(self.project_root, "branches"), exist_ok=True)
|
|
# Initialize git repo in project_root for git rev-parse checks
|
|
os.system(f"git -C '{self.project_root}' init -q")
|
|
|
|
def tearDown(self):
|
|
shutil.rmtree(self.temp_dir, ignore_errors=True)
|
|
|
|
def test_audit_missing_worktree_bindings_reports_associations(self):
|
|
res = self.db.assign_and_lease(
|
|
remote="prgs",
|
|
org="Scaled-Tech-Consulting",
|
|
repo="Gitea-Tools",
|
|
kind="pr",
|
|
number=795,
|
|
expected_head_sha="head795",
|
|
role="reviewer",
|
|
session_id="reviewer-pr795-test-session",
|
|
worktree_path="/tmp/nonexistent_wt_path_795",
|
|
)
|
|
lease_id = res.lease_id
|
|
self.assertIsNotNone(lease_id)
|
|
# Release the lease (dead/released)
|
|
self.db.release_lease_recorded(lease_id, session_id="reviewer-pr795-test-session")
|
|
|
|
audit = mwr.audit_missing_worktree_bindings(
|
|
self.db,
|
|
project_root=self.project_root,
|
|
remote="prgs",
|
|
org="Scaled-Tech-Consulting",
|
|
repo="Gitea-Tools",
|
|
)
|
|
|
|
self.assertTrue(audit["host_mount_healthy"])
|
|
self.assertEqual(audit["missing_count"], 1)
|
|
self.assertEqual(audit["confirmed_stale_count"], 1)
|
|
|
|
candidate = audit["confirmed_stale_candidates"][0]
|
|
self.assertEqual(candidate["lease_id"], lease_id)
|
|
self.assertEqual(candidate["work_number"], 795)
|
|
self.assertEqual(candidate["session_id"], "reviewer-pr795-test-session")
|
|
self.assertEqual(candidate["classification"], mwr.CLASS_CONFIRMED_STALE_DELETED)
|
|
self.assertTrue(candidate["retire_eligible"])
|
|
|
|
def test_unavailable_host_mount_blocks_retirement(self):
|
|
# Point to a non-existent project_root
|
|
bad_root = os.path.join(self.temp_dir, "nonexistent_repo_dir")
|
|
audit = mwr.audit_missing_worktree_bindings(
|
|
self.db,
|
|
project_root=bad_root,
|
|
remote="prgs",
|
|
)
|
|
self.assertFalse(audit["host_mount_healthy"])
|
|
|
|
res = self.db.assign_and_lease(
|
|
remote="prgs", org="org", repo="repo", kind="issue", number=100,
|
|
role="author", session_id="sess-100", worktree_path="/tmp/missing_100",
|
|
)
|
|
self.db.release_lease_recorded(res.lease_id, session_id="sess-100")
|
|
|
|
audit2 = mwr.audit_missing_worktree_bindings(self.db, project_root=bad_root, remote="prgs")
|
|
self.assertEqual(audit2["unavailable_host_count"], 1)
|
|
self.assertEqual(audit2["confirmed_stale_count"], 0)
|
|
self.assertFalse(audit2["missing_bindings"][0]["retire_eligible"])
|
|
|
|
def test_live_lease_and_live_session_block_retirement(self):
|
|
res = self.db.assign_and_lease(
|
|
remote="prgs", org="org", repo="repo", kind="issue", number=200,
|
|
role="author", session_id="sess-live-200", worktree_path="/tmp/missing_live_200",
|
|
owner_pid=os.getpid(),
|
|
)
|
|
|
|
audit = mwr.audit_missing_worktree_bindings(self.db, project_root=self.project_root, remote="prgs")
|
|
self.assertEqual(audit["live_protected_count"], 1)
|
|
self.assertEqual(audit["confirmed_stale_count"], 0)
|
|
|
|
# Attempting resolution on live protected binding is refused
|
|
c = audit["missing_bindings"][0]
|
|
res_mut = mwr.resolve_missing_worktree_binding(self.db, binding=c, dry_run=False, operator_authorized=True, project_root=self.project_root)
|
|
self.assertFalse(res_mut["success"])
|
|
self.assertEqual(res_mut["reason"], "revalidation_failed")
|
|
|
|
def test_moved_worktree_path_blocks_retirement(self):
|
|
# Create a worktree under branches/ to simulate moved path
|
|
moved_dir = os.path.join(self.project_root, "branches", "issue-635-project-registry-api")
|
|
os.makedirs(moved_dir, exist_ok=True)
|
|
|
|
res = self.db.assign_and_lease(
|
|
remote="prgs", org="org", repo="repo", kind="issue", number=635,
|
|
role="author", session_id="sess-635", worktree_path="/tmp/old_missing_path_635",
|
|
)
|
|
self.db.release_lease_recorded(res.lease_id, session_id="sess-635")
|
|
|
|
cls_info = mwr.classify_worktree_binding(
|
|
recorded_path="/tmp/old_missing_path_635",
|
|
branch="issue-635-project-registry-api",
|
|
project_root=self.project_root,
|
|
)
|
|
self.assertEqual(cls_info["classification"], mwr.CLASS_MOVED_WORKTREE)
|
|
self.assertFalse(cls_info["retire_eligible"])
|
|
self.assertEqual(os.path.realpath(cls_info["moved_to_path"]), os.path.realpath(moved_dir))
|
|
|
|
def test_recreation_race_revalidation_blocks_mutation(self):
|
|
path = os.path.join(self.temp_dir, "recreated_wt")
|
|
res = self.db.assign_and_lease(
|
|
remote="prgs", org="org", repo="repo", kind="issue", number=300,
|
|
role="author", session_id="sess-300", worktree_path=path,
|
|
)
|
|
self.db.release_lease_recorded(res.lease_id, session_id="sess-300")
|
|
|
|
# Initial audit when path is missing
|
|
audit = mwr.audit_missing_worktree_bindings(self.db, project_root=self.project_root, remote="prgs")
|
|
candidate = audit["confirmed_stale_candidates"][0]
|
|
|
|
# Simulate concurrent recreation of worktree path
|
|
os.makedirs(path, exist_ok=True)
|
|
|
|
# Pre-mutation revalidation detects recreation and blocks
|
|
res_mut = mwr.resolve_missing_worktree_binding(
|
|
self.db,
|
|
binding=candidate,
|
|
dry_run=False,
|
|
operator_authorized=True,
|
|
project_root=self.project_root,
|
|
)
|
|
self.assertFalse(res_mut["success"])
|
|
self.assertEqual(res_mut["reason"], "revalidation_failed")
|
|
|
|
def test_authorized_cleanup_retires_stale_binding_and_is_idempotent(self):
|
|
# Create 2 leases: 1 stale missing, 1 valid existing worktree
|
|
valid_path = os.path.join(self.project_root, "branches", "valid_wt")
|
|
os.makedirs(valid_path, exist_ok=True)
|
|
|
|
stale_path = "/tmp/stale_missing_400"
|
|
|
|
res1 = self.db.assign_and_lease(
|
|
remote="prgs", org="org", repo="repo", kind="issue", number=401,
|
|
role="author", session_id="sess-401", worktree_path=valid_path,
|
|
)
|
|
|
|
res2 = self.db.assign_and_lease(
|
|
remote="prgs", org="org", repo="repo", kind="issue", number=402,
|
|
role="author", session_id="sess-402", worktree_path=stale_path,
|
|
)
|
|
self.db.release_lease_recorded(res2.lease_id, session_id="sess-402")
|
|
|
|
# Dry-run reconciliation
|
|
dry = mwr.reconcile_missing_worktree_bindings(
|
|
self.db, project_root=self.project_root, remote="prgs", dry_run=True, operator_authorized=True
|
|
)
|
|
self.assertTrue(dry["success"])
|
|
self.assertTrue(dry["dry_run"])
|
|
self.assertEqual(dry["stale_candidates_count"], 1)
|
|
|
|
# Apply reconciliation
|
|
applied = mwr.reconcile_missing_worktree_bindings(
|
|
self.db, project_root=self.project_root, remote="prgs", dry_run=False, operator_authorized=True
|
|
)
|
|
self.assertTrue(applied["success"])
|
|
self.assertFalse(applied["dry_run"])
|
|
self.assertTrue(applied["worktrees_dimension_resolved"])
|
|
self.assertEqual(applied["after_audit"]["missing_count"], 0)
|
|
|
|
# Verify DB: lease 2 worktree_path is cleared, lease 1 worktree_path remains intact
|
|
l1 = self.db.get_lease_workflow_state(res1.lease_id)
|
|
self.assertEqual(l1["lease"]["worktree_path"], valid_path)
|
|
|
|
l2 = self.db.get_lease_workflow_state(res2.lease_id)
|
|
self.assertEqual(l2["lease"]["worktree_path"], "")
|
|
prov = json.loads(l2["lease"]["provenance_json"])
|
|
self.assertTrue(prov.get("worktree_path_retired"))
|
|
self.assertEqual(prov.get("retired_worktree_path"), stale_path)
|
|
|
|
# Re-run (idempotency check)
|
|
rerun = mwr.reconcile_missing_worktree_bindings(
|
|
self.db, project_root=self.project_root, remote="prgs", dry_run=False, operator_authorized=True
|
|
)
|
|
self.assertTrue(rerun["success"])
|
|
self.assertTrue(rerun["worktrees_dimension_resolved"])
|
|
self.assertEqual(rerun["stale_candidates_count"], 0)
|
|
|
|
def test_historical_post_pr968_records_resolution(self):
|
|
# Create historical records matching #970 description
|
|
res_795 = self.db.assign_and_lease(
|
|
remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", kind="pr", number=795,
|
|
expected_head_sha="head795hist", role="reviewer", session_id="reviewer-pr795-hist",
|
|
worktree_path="/Users/jasonwalker/Development/Gitea-Tools/branches/review-feat-issue-628-autonomous-handoffs-orchestration",
|
|
)
|
|
self.db.release_lease_recorded(res_795.lease_id, session_id="reviewer-pr795-hist")
|
|
|
|
res_635 = self.db.assign_and_lease(
|
|
remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", kind="issue", number=635,
|
|
role="author", session_id="author-issue635-hist",
|
|
worktree_path="/Users/jasonwalker/Development/Gitea-Tools/branches/issue-635-project-registry-api",
|
|
)
|
|
self.db.release_lease_recorded(res_635.lease_id, session_id="author-issue635-hist")
|
|
|
|
# Audit finds both historical records
|
|
audit = mwr.audit_missing_worktree_bindings(self.db, project_root=self.project_root, remote="prgs")
|
|
self.assertEqual(audit["confirmed_stale_count"], 2)
|
|
|
|
# Reconcile safely resolves both
|
|
rec = mwr.reconcile_missing_worktree_bindings(
|
|
self.db, project_root=self.project_root, remote="prgs", dry_run=False, operator_authorized=True
|
|
)
|
|
self.assertTrue(rec["worktrees_dimension_resolved"])
|
|
self.assertEqual(rec["after_audit"]["missing_count"], 0)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|