"""Tests for atomic per-issue lock store (#438).""" from __future__ import annotations import os import sys import tempfile import threading import unittest from datetime import datetime, timedelta, timezone from unittest.mock import patch sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent)) import issue_lock_store # noqa: E402 def _lease(**overrides): now = datetime.now(timezone.utc) lease = { "operation_type": "author_issue_work", "issue_number": 438, "branch": "feat/issue-438-lock-hardening", "worktree_path": "/tmp/wt", "expires_at": (now + timedelta(hours=4)).isoformat().replace("+00:00", "Z"), "last_heartbeat_at": now.isoformat().replace("+00:00", "Z"), } lease.update(overrides) return lease class TestIssueLockStore(unittest.TestCase): def setUp(self): self.tempdir = tempfile.mkdtemp(prefix="issue-lock-store-") self.legacy = os.path.join(self.tempdir, "legacy.json") self.addCleanup(self._cleanup) def _cleanup(self): for root, _dirs, files in os.walk(self.tempdir, topdown=False): for name in files: try: os.remove(os.path.join(root, name)) except OSError: pass try: os.rmdir(root) except OSError: pass def _acquire(self, issue_number=438, branch="feat/issue-438-lock-hardening", **kwargs): worktree_path = kwargs.pop("worktree_path", "/tmp/wt") return issue_lock_store.acquire_issue_lock( issue_number=issue_number, branch_name=branch, remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", worktree_path=worktree_path, work_lease=_lease(issue_number=issue_number, branch=branch), claimant={"profile": "prgs-author", "username": "jcwalker3"}, lock_dir=self.tempdir, **kwargs, ) def test_acquire_writes_per_issue_lock_and_legacy_pointer(self): with patch.object(issue_lock_store, "LEGACY_LOCK_FILE", self.legacy): result = self._acquire() path = issue_lock_store.issue_lock_path("prgs", "Scaled-Tech-Consulting", "Gitea-Tools", 438, lock_dir=self.tempdir) self.assertTrue(os.path.exists(path)) self.assertTrue(result["acquired"]) self.assertIn("lock_proof", result) with open(self.legacy, encoding="utf-8") as handle: legacy = __import__("json").load(handle) self.assertEqual(legacy["issue_number"], 438) self.assertEqual(legacy["session_id"], result["session_id"]) def test_concurrent_acquire_same_issue_only_one_wins(self): barrier = threading.Barrier(2) results: list[dict | Exception] = [] def worker(): barrier.wait() try: results.append(self._acquire()) 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 isinstance(item, dict)] 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 lock" in failure_text or "lock contention" in failure_text, failures[0], ) def test_distinct_issues_acquire_without_contention(self): first = self._acquire(issue_number=438, branch="feat/issue-438-lock-hardening") second = self._acquire( issue_number=440, branch="feat/issue-440-recovery", worktree_path="/tmp/other-wt", ) self.assertTrue(first["acquired"]) self.assertTrue(second["acquired"]) self.assertNotEqual(first["session_id"], second["session_id"]) def test_stale_lock_requires_explicit_recovery(self): path = issue_lock_store.issue_lock_path("prgs", "Scaled-Tech-Consulting", "Gitea-Tools", 438, lock_dir=self.tempdir) stale = { "lock_version": 1, "issue_number": 438, "branch_name": "feat/issue-438-old", "remote": "prgs", "org": "Scaled-Tech-Consulting", "repo": "Gitea-Tools", "worktree_path": "/tmp/old", "pid": 999999, "session_id": "stale-session", "work_lease": _lease(expires_at="2000-01-01T00:00:00Z"), "last_heartbeat_at": "2000-01-01T00:00:00Z", "lock_path": path, } issue_lock_store.atomic_write_json(path, stale) with self.assertRaises(RuntimeError) as ctx: self._acquire() self.assertIn("Recovery review is required", str(ctx.exception)) recovered = self._acquire(allow_stale_recovery=True) self.assertTrue(recovered["acquired"]) def test_verify_lock_for_mutation_blocks_stale_lock(self): record = { "issue_number": 438, "branch_name": "feat/issue-438-lock-hardening", "worktree_path": "/tmp/wt", "pid": 999999, "work_lease": _lease(expires_at="2000-01-01T00:00:00Z"), "last_heartbeat_at": "2000-01-01T00:00:00Z", } result = issue_lock_store.verify_lock_for_mutation( record, issue_number=438, branch_name="feat/issue-438-lock-hardening", worktree_path="/tmp/wt", ) self.assertTrue(result["block"]) self.assertIn("not live", result["reasons"][0]) def test_list_live_locks_excludes_stale_records(self): path = issue_lock_store.issue_lock_path("prgs", "Scaled-Tech-Consulting", "Gitea-Tools", 438, lock_dir=self.tempdir) issue_lock_store.atomic_write_json( path, { "issue_number": 438, "branch_name": "feat/issue-438-lock-hardening", "remote": "prgs", "org": "Scaled-Tech-Consulting", "repo": "Gitea-Tools", "worktree_path": "/tmp/wt", "pid": os.getpid(), "session_id": "live", "work_lease": _lease(), "last_heartbeat_at": datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"), "lock_path": path, }, ) stale_path = issue_lock_store.issue_lock_path("prgs", "Scaled-Tech-Consulting", "Gitea-Tools", 440, lock_dir=self.tempdir) issue_lock_store.atomic_write_json( stale_path, { "issue_number": 440, "branch_name": "feat/issue-440-recovery", "remote": "prgs", "org": "Scaled-Tech-Consulting", "repo": "Gitea-Tools", "worktree_path": "/tmp/wt", "pid": 999999, "session_id": "stale", "work_lease": _lease(issue_number=440, branch="feat/issue-440-recovery", expires_at="2000-01-01T00:00:00Z"), "last_heartbeat_at": "2000-01-01T00:00:00Z", "lock_path": stale_path, }, ) live = issue_lock_store.list_live_locks(lock_dir=self.tempdir) self.assertEqual([entry["issue_number"] for entry in live], [438]) def test_resolve_lock_for_branch_reads_per_issue_file(self): self._acquire() resolved = issue_lock_store.resolve_lock_for_branch( "feat/issue-438-lock-hardening", lock_dir=self.tempdir, ) self.assertIsNotNone(resolved) self.assertEqual(resolved["issue_number"], 438) if __name__ == "__main__": unittest.main()