sysadmin
780ef0b1be
Make active leases queryable workflow state with canonical adopt, release, expire/reclaim, and abandon operations on the #613 control-plane DB substrate. - lease_lifecycle.py: policy, proof gates, safe_next_action vocabulary - control_plane_db: schema v3 provenance columns + list/inspect/adopt/abandon - MCP tools: gitea_*_workflow_lease(s) for list/inspect/adopt/release/expire/abandon/reclaim - issue_lock_store: sanctioned reclaim of expired author locks (dead pid / missing wt) - Tests cover lifecycle, foreign steal refusal, allocator compatibility, merger handoff
300 lines
11 KiB
Python
300 lines
11 KiB
Python
"""Unit tests for keyed issue-lock storage (#443) and flock hardening (#438)."""
|
|
import json
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
import threading
|
|
import unittest
|
|
from datetime import datetime, timedelta, timezone
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
|
|
|
import issue_lock_store as ils # noqa: E402
|
|
|
|
|
|
def _lease(expires_at: str) -> dict:
|
|
return {
|
|
"operation_type": ils.AUTHOR_ISSUE_WORK_LEASE,
|
|
"expires_at": expires_at,
|
|
"created_at": "2026-01-01T00:00:00Z",
|
|
"last_heartbeat_at": "2026-01-01T00:00:00Z",
|
|
}
|
|
|
|
|
|
def _lock_record(**overrides) -> dict:
|
|
record = {
|
|
"issue_number": 420,
|
|
"branch_name": "feat/issue-420-server-code-parity",
|
|
"remote": "prgs",
|
|
"org": "Scaled-Tech-Consulting",
|
|
"repo": "Gitea-Tools",
|
|
"worktree_path": "/tmp/wt-420",
|
|
"work_lease": _lease("2999-01-01T00:00:00Z"),
|
|
}
|
|
record.update(overrides)
|
|
return record
|
|
|
|
|
|
class TestIssueLockStore(unittest.TestCase):
|
|
def setUp(self):
|
|
self._dir = tempfile.TemporaryDirectory()
|
|
self.lock_dir = self._dir.name
|
|
self._env = mock.patch.dict(os.environ, {"GITEA_ISSUE_LOCK_DIR": self.lock_dir})
|
|
self._env.start()
|
|
|
|
def tearDown(self):
|
|
self._env.stop()
|
|
self._dir.cleanup()
|
|
|
|
def test_concurrent_repo_locks_do_not_overwrite(self):
|
|
lock_a = _lock_record(
|
|
issue_number=108,
|
|
branch_name="feat/issue-108-root-menu",
|
|
repo="mcp-control-plane",
|
|
worktree_path="/tmp/wt-108",
|
|
)
|
|
lock_b = _lock_record(
|
|
issue_number=420,
|
|
branch_name="feat/issue-420-server-code-parity",
|
|
repo="Gitea-Tools",
|
|
worktree_path="/tmp/wt-420",
|
|
)
|
|
path_a = ils.bind_session_lock(lock_a)
|
|
with mock.patch("os.getpid", return_value=9999):
|
|
path_b = ils.bind_session_lock(lock_b)
|
|
|
|
self.assertNotEqual(path_a, path_b)
|
|
self.assertTrue(os.path.exists(path_a))
|
|
self.assertTrue(os.path.exists(path_b))
|
|
stored_a = ils.read_lock_file(path_a)
|
|
stored_b = ils.read_lock_file(path_b)
|
|
self.assertEqual(stored_a["issue_number"], 108)
|
|
self.assertEqual(stored_b["issue_number"], 420)
|
|
|
|
def test_concurrent_issue_locks_same_repo_do_not_overwrite(self):
|
|
lock_a = _lock_record(issue_number=427, branch_name="feat/issue-427-a")
|
|
lock_b = _lock_record(issue_number=428, branch_name="feat/issue-428-b")
|
|
path_a = ils.bind_session_lock(lock_a)
|
|
with mock.patch("os.getpid", return_value=4242):
|
|
path_b = ils.bind_session_lock(lock_b)
|
|
|
|
self.assertNotEqual(path_a, path_b)
|
|
self.assertEqual(ils.read_lock_file(path_a)["issue_number"], 427)
|
|
self.assertEqual(ils.read_lock_file(path_b)["issue_number"], 428)
|
|
|
|
def test_foreign_live_lease_blocks_overwrite(self):
|
|
existing = _lock_record(
|
|
branch_name="feat/issue-420-other",
|
|
worktree_path="/tmp/other",
|
|
work_lease=_lease("2999-01-01T00:00:00Z"),
|
|
)
|
|
path = ils.lock_file_path(
|
|
remote="prgs",
|
|
org="Scaled-Tech-Consulting",
|
|
repo="Gitea-Tools",
|
|
issue_number=420,
|
|
)
|
|
ils.save_lock_file(path, existing)
|
|
|
|
incoming = _lock_record(worktree_path="/tmp/mine")
|
|
block = ils.assess_foreign_lock_overwrite(existing, incoming)
|
|
self.assertIn("live foreign issue lock", block or "")
|
|
|
|
def test_expired_lease_allows_takeover_with_conflict_check(self):
|
|
# #601: expired + dead pid / missing worktree → sanctioned reclaim (no block).
|
|
existing = _lock_record(
|
|
branch_name="feat/issue-420-other",
|
|
worktree_path="/tmp/other-does-not-exist-420",
|
|
session_pid=1,
|
|
work_lease=_lease("2000-01-01T00:00:00Z"),
|
|
)
|
|
incoming = _lock_record(worktree_path="/tmp/mine")
|
|
self.assertIsNone(ils.assess_foreign_lock_overwrite(existing, incoming))
|
|
with mock.patch.object(ils, "is_process_alive", return_value=False):
|
|
block = ils.assess_same_issue_lease_conflict(
|
|
existing,
|
|
issue_number=420,
|
|
branch_name="feat/issue-420-server-code-parity",
|
|
worktree_path="/tmp/mine",
|
|
)
|
|
self.assertIsNone(block)
|
|
|
|
# Expired but still-live owner pid AND present worktree → fail closed.
|
|
present_wt = self.lock_dir # exists
|
|
sticky = _lock_record(
|
|
branch_name="feat/issue-420-other",
|
|
worktree_path=present_wt,
|
|
session_pid=os.getpid(),
|
|
work_lease=_lease("2000-01-01T00:00:00Z"),
|
|
)
|
|
with mock.patch.object(ils, "is_process_alive", return_value=True):
|
|
blocked = ils.assess_same_issue_lease_conflict(
|
|
sticky,
|
|
issue_number=420,
|
|
branch_name="feat/issue-420-server-code-parity",
|
|
worktree_path="/tmp/mine",
|
|
)
|
|
self.assertIn("Recovery review is required", blocked or "")
|
|
|
|
def test_same_owner_lease_conflict_allows_refresh(self):
|
|
worktree = "/tmp/wt-420"
|
|
existing = _lock_record(worktree_path=worktree)
|
|
block = ils.assess_same_issue_lease_conflict(
|
|
existing,
|
|
issue_number=420,
|
|
branch_name="feat/issue-420-server-code-parity",
|
|
worktree_path=worktree,
|
|
)
|
|
self.assertIsNone(block)
|
|
|
|
def test_find_lock_for_branch_after_restart(self):
|
|
record = _lock_record()
|
|
path = ils.lock_file_path(
|
|
remote="prgs",
|
|
org="Scaled-Tech-Consulting",
|
|
repo="Gitea-Tools",
|
|
issue_number=420,
|
|
)
|
|
ils.save_lock_file(path, record)
|
|
|
|
with mock.patch("os.getpid", return_value=5555):
|
|
self.assertIsNone(ils.read_session_issue_lock())
|
|
|
|
found = ils.find_lock_for_branch(
|
|
remote="prgs",
|
|
org="Scaled-Tech-Consulting",
|
|
repo="Gitea-Tools",
|
|
branch_name="feat/issue-420-server-code-parity",
|
|
)
|
|
self.assertEqual(found["issue_number"], 420)
|
|
|
|
def test_has_active_issue_lock_scans_keyed_store(self):
|
|
ils.bind_session_lock(_lock_record())
|
|
self.assertTrue(
|
|
ils.has_active_issue_lock("feat/issue-420-server-code-parity")
|
|
)
|
|
self.assertFalse(ils.has_active_issue_lock("feat/issue-999-other"))
|
|
|
|
def test_approved_stacked_base_survives_round_trip(self):
|
|
# #484: the approved stacked base recorded on the lock must persist so
|
|
# gitea_create_pr can validate the non-master base at PR time.
|
|
record = _lock_record(
|
|
issue_number=482,
|
|
branch_name="feat/issue-482-skip-stale-request-changes-pr",
|
|
approved_stacked_base={
|
|
"branch": "feat/issue-478-mcp-menu-shell",
|
|
"pr_number": 479,
|
|
"verified_open": True,
|
|
},
|
|
)
|
|
path = ils.lock_file_path(
|
|
remote="prgs",
|
|
org="Scaled-Tech-Consulting",
|
|
repo="Gitea-Tools",
|
|
issue_number=482,
|
|
)
|
|
ils.save_lock_file(path, record)
|
|
stored = ils.read_lock_file(path)
|
|
self.assertEqual(stored["approved_stacked_base"]["branch"], "feat/issue-478-mcp-menu-shell")
|
|
self.assertEqual(stored["approved_stacked_base"]["pr_number"], 479)
|
|
self.assertTrue(stored["approved_stacked_base"]["verified_open"])
|
|
|
|
def test_atomic_write_preserves_unrelated_lock(self):
|
|
path_a = ils.lock_file_path(
|
|
remote="prgs",
|
|
org="Scaled-Tech-Consulting",
|
|
repo="Gitea-Tools",
|
|
issue_number=108,
|
|
)
|
|
ils.save_lock_file(path_a, _lock_record(issue_number=108, repo="mcp-control-plane"))
|
|
path_b = ils.lock_file_path(
|
|
remote="prgs",
|
|
org="Scaled-Tech-Consulting",
|
|
repo="Gitea-Tools",
|
|
issue_number=420,
|
|
)
|
|
ils.save_lock_file(path_b, _lock_record())
|
|
|
|
self.assertTrue(os.path.exists(path_a))
|
|
self.assertTrue(os.path.exists(path_b))
|
|
self.assertEqual(ils.read_lock_file(path_a)["issue_number"], 108)
|
|
|
|
def test_concurrent_bind_same_issue_only_one_wins(self):
|
|
barrier = threading.Barrier(2)
|
|
results: list[str | Exception] = []
|
|
|
|
def worker():
|
|
barrier.wait()
|
|
try:
|
|
ils.bind_session_lock(
|
|
_lock_record(worktree_path=f"/tmp/wt-{threading.get_ident()}")
|
|
)
|
|
results.append("ok")
|
|
except Exception as exc: # noqa: BLE001
|
|
results.append(exc)
|
|
|
|
threads = [threading.Thread(target=worker) for _ in range(2)]
|
|
for thread in threads:
|
|
thread.start()
|
|
for thread in threads:
|
|
thread.join()
|
|
|
|
successes = [item for item in results if item == "ok"]
|
|
failures = [item for item in results if isinstance(item, Exception)]
|
|
self.assertEqual(len(successes), 1)
|
|
self.assertEqual(len(failures), 1)
|
|
failure_text = str(failures[0]).lower()
|
|
self.assertTrue(
|
|
"active" in failure_text or "lock contention" in failure_text,
|
|
failures[0],
|
|
)
|
|
|
|
def test_verify_lock_for_mutation_blocks_stale_lock(self):
|
|
record = _lock_record(
|
|
work_lease=_lease("2000-01-01T00:00:00Z"),
|
|
)
|
|
record["pid"] = 999999
|
|
record["session_pid"] = 999999
|
|
result = ils.verify_lock_for_mutation(
|
|
record,
|
|
issue_number=420,
|
|
branch_name="feat/issue-420-server-code-parity",
|
|
worktree_path="/tmp/wt-420",
|
|
)
|
|
self.assertTrue(result["block"])
|
|
self.assertIn("not live", result["reasons"][0])
|
|
|
|
def test_list_live_locks_excludes_stale_records(self):
|
|
live_path = ils.lock_file_path(
|
|
remote="prgs",
|
|
org="Scaled-Tech-Consulting",
|
|
repo="Gitea-Tools",
|
|
issue_number=420,
|
|
)
|
|
ils.save_lock_file(
|
|
live_path,
|
|
_lock_record(worktree_path="/tmp/wt-420"),
|
|
)
|
|
stale_path = ils.lock_file_path(
|
|
remote="prgs",
|
|
org="Scaled-Tech-Consulting",
|
|
repo="Gitea-Tools",
|
|
issue_number=440,
|
|
)
|
|
ils.save_lock_file(
|
|
stale_path,
|
|
_lock_record(
|
|
issue_number=440,
|
|
branch_name="feat/issue-440-recovery",
|
|
work_lease=_lease("2000-01-01T00:00:00Z"),
|
|
worktree_path="/tmp/wt-440",
|
|
),
|
|
)
|
|
live = ils.list_live_locks(lock_dir=self.lock_dir)
|
|
self.assertEqual([entry["issue_number"] for entry in live], [420])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main() |