Gitea-Tools/tests/test_issue_lock_store.py

"""Unit tests for keyed issue-lock storage (#443)."""
import json
import os
import sys
import tempfile
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):
        existing = _lock_record(
            branch_name="feat/issue-420-other",
            worktree_path="/tmp/other",
            work_lease=_lease("2000-01-01T00:00:00Z"),
        )
        incoming = _lock_record(worktree_path="/tmp/mine")
        self.assertIsNone(ils.assess_foreign_lock_overwrite(existing, incoming))
        block = ils.assess_same_issue_lease_conflict(
            existing,
            issue_number=420,
            branch_name="feat/issue-420-server-code-parity",
            worktree_path="/tmp/mine",
        )
        self.assertIn("Recovery review is required", block 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_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)


if __name__ == "__main__":
    unittest.main()