Scaled-Tech-Consulting/Gitea-Tools
2
0
Fork 0
Code Issues 77 Pull Requests 10 Actions Packages Projects Releases Wiki Activity

Merge pull request 'feat: keyed issue-lock store replaces global lock file (Closes #443)' (#465) from feat/issue-443-lock-store into master

Browse Source
This commit was merged in pull request #465.
This commit is contained in:
sysadmin
2026-07-07 16:57:54 -05:00
parent 5dfc00319f e956040b6b
commit 54a75153a9
13 changed files with 1187 additions and 185 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_agent_temp_artifacts.py
+6 -3
View File
@@ -74,21 +74,24 @@ ISSUE_WRITE_ENV = {
class TestIssueLockArtifactWarning(unittest.TestCase):
def setUp(self):
self._env_patcher = patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True)
self._lock_dir = tempfile.TemporaryDirectory()
env = {**ISSUE_WRITE_ENV, "GITEA_ISSUE_LOCK_DIR": self._lock_dir.name}
self._env_patcher = patch.dict(os.environ, env, clear=True)
self._env_patcher.start()
def tearDown(self):
self._env_patcher.stop()
self._lock_dir.cleanup()
@patch(
"mcp_server.issue_duplicate_context_fetcher",
return_value=([], [], {"status": "not_claimed"}),
)
@patch("mcp_server.api_get_all", return_value=[])
@patch("mcp_server._auth", return_value="token x")
@patch("mcp_server._resolve", return_value=("h", "o", "r"))
@patch("mcp_server.ISSUE_LOCK_FILE", new_callable=lambda: tempfile.mktemp())
@patch("issue_lock_worktree.read_worktree_git_state")
def test_lock_success_includes_artifact_warning(self, mock_state, _lock_file, *_mocks):
def test_lock_success_includes_artifact_warning(self, mock_state, *_mocks):
mock_state.return_value = {
"current_branch": "master",
"porcelain_status": "?? _emit_payload.py\n",
tests/test_commit_payloads.py
+7 -5
View File
@@ -66,9 +66,12 @@ class TestCommitPayloads(unittest.TestCase):
)
self.locked_worktree_path = os.path.realpath(self.locked_worktree_dir.name)
self.lock_file_path = "/tmp/gitea_issue_lock.json"
import issue_lock_store
import issue_lock_provenance
self._lock_dir = tempfile.TemporaryDirectory()
os.environ["GITEA_ISSUE_LOCK_DIR"] = self._lock_dir.name
work_lease = {
"operation_type": "author_issue_work",
"issue_number": 263,
@@ -89,8 +92,7 @@ class TestCommitPayloads(unittest.TestCase):
claimant=work_lease.get("claimant"),
),
}
with open(self.lock_file_path, "w", encoding="utf-8") as fh:
fh.write(json.dumps(self.lock_data))
self.lock_file_path = issue_lock_store.bind_session_lock(self.lock_data)
# Reset preflight status to bypass/pass verification in tests
self.orig_whoami_called = mcp_server._preflight_whoami_called
@@ -114,8 +116,7 @@ class TestCommitPayloads(unittest.TestCase):
self._dir.cleanup()
self.locked_worktree_dir.cleanup()
if os.path.exists(self.lock_file_path):
os.remove(self.lock_file_path)
self._lock_dir.cleanup()
def _env(self, profile: str) -> dict:
return {
@@ -124,6 +125,7 @@ class TestCommitPayloads(unittest.TestCase):
"GITEA_TOKEN_AUTHOR": "author-pass",
"GITEA_TEST_PORCELAIN": "",
"GITEA_AUTHOR_WORKTREE": self.locked_worktree_path,
"GITEA_ISSUE_LOCK_DIR": self._lock_dir.name,
}
@patch("mcp_server.api_request")
tests/test_issue_lock_adoption.py
+68
View File
@@ -0,0 +1,68 @@
"""Unit tests for own-branch lock adoption decision (#442 / #443)."""
import sys
import unittest
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from issue_lock_adoption import ( # noqa: E402
ADOPT,
BLOCK_COMPETING,
NO_MATCH,
assess_own_branch_adoption,
build_adoption_proof,
)
REQ = "feat/issue-420-server-code-parity"
class TestAssessOwnBranchAdoption(unittest.TestCase):
def test_exact_own_branch_is_adopted(self):
result = assess_own_branch_adoption(
issue_number=420,
requested_branch=REQ,
existing_branches=[{"name": REQ, "commit_sha": "934688a"}],
)
self.assertEqual(result["outcome"], ADOPT)
self.assertTrue(result["adopt"])
def test_different_branch_same_issue_blocks(self):
result = assess_own_branch_adoption(
issue_number=420,
requested_branch=REQ,
existing_branches=[{"name": "feat/issue-420-other-work"}],
)
self.assertEqual(result["outcome"], BLOCK_COMPETING)
self.assertTrue(result["block"])
def test_no_matching_branch_is_normal_path(self):
result = assess_own_branch_adoption(
issue_number=420,
requested_branch=REQ,
existing_branches=[{"name": "feat/issue-999-unrelated"}],
)
self.assertEqual(result["outcome"], NO_MATCH)
class TestBuildAdoptionProof(unittest.TestCase):
def test_proof_has_required_fields(self):
assessment = assess_own_branch_adoption(
issue_number=420,
requested_branch=REQ,
existing_branches=[{"name": REQ, "commit_sha": "934688a"}],
)
proof = build_adoption_proof(
issue_number=420,
branch_name=REQ,
assessment=assessment,
open_pr_checked=True,
competing_lock_checked=True,
lock_file_path="/tmp/example-lock.json",
lock_file_status="written",
)
self.assertEqual(proof["branch_head_commit"], "934688a")
self.assertTrue(proof["no_existing_pr_proof"])
if __name__ == "__main__":
unittest.main()
tests/test_issue_lock_store.py
+181
View File
@@ -0,0 +1,181 @@
"""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()
tests/test_issue_work_duplicate_gate.py
+34 -26
View File
@@ -1,5 +1,4 @@
"""Tests for early duplicate-work detection (#400)."""
import json
import os
import sys
import tempfile
@@ -10,6 +9,7 @@ from unittest.mock import patch
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
import issue_lock_provenance
import issue_lock_store
import issue_work_duplicate_gate as dup_gate
import mcp_server
from issue_work_duplicate_gate import (
@@ -123,8 +123,9 @@ class TestDuplicateReportOutcome(unittest.TestCase):
class TestInjectableDuplicateFetcher(unittest.TestCase):
@patch("mcp_server.api_get_all", return_value=[])
@patch("mcp_server.get_auth_header", return_value="token x")
def test_lock_issue_uses_injected_fetcher(self, _auth):
def test_lock_issue_uses_injected_fetcher(self, _auth, _api):
seen = {}
def fetcher(h, o, r, auth, issue_number):
@@ -141,26 +142,29 @@ class TestInjectableDuplicateFetcher(unittest.TestCase):
"porcelain_status": "",
"base_equivalent": True,
},
), patch.dict(os.environ, {
"GITEA_ALLOWED_OPERATIONS": "gitea.issue.comment",
}, clear=True):
with patch.object(mcp_server, "ISSUE_LOCK_FILE", tempfile.mktemp()):
mcp_server.gitea_lock_issue(
issue_number=400,
branch_name="feat/issue-400-duplicate-work-preflight",
remote="prgs",
)
):
with tempfile.TemporaryDirectory() as lock_dir:
with patch.dict(os.environ, {
"GITEA_ALLOWED_OPERATIONS": "gitea.issue.comment",
"GITEA_ISSUE_LOCK_DIR": lock_dir,
}, clear=True):
mcp_server.gitea_lock_issue(
issue_number=400,
branch_name="feat/issue-400-duplicate-work-preflight",
remote="prgs",
)
self.assertEqual(seen["issue_number"], 400)
class TestMcpDuplicateRecheck(unittest.TestCase):
def setUp(self):
self._dir = tempfile.TemporaryDirectory()
self.lock_path = os.path.join(self._dir.name, "gitea_issue_lock.json")
self._lock_patch = patch.object(
mcp_server, "ISSUE_LOCK_FILE", self.lock_path
self._env_patch = patch.dict(
os.environ,
{"GITEA_ISSUE_LOCK_DIR": self._dir.name},
clear=False,
)
self._lock_patch.start()
self._env_patch.start()
self._remotes = patch.dict(mcp_server.REMOTES, {
"prgs": {"host": "gitea.example.com", "org": "Example-Org",
"repo": "Example-Repo"},
@@ -173,6 +177,7 @@ class TestMcpDuplicateRecheck(unittest.TestCase):
self._dir.cleanup()
def _write_lock(self, issue_number=400, branch="feat/issue-400-x"):
worktree_path = os.path.realpath(os.getcwd())
work_lease = {
"operation_type": "author_issue_work",
"issue_number": issue_number,
@@ -180,17 +185,19 @@ class TestMcpDuplicateRecheck(unittest.TestCase):
"claimant": {"username": "test-user", "profile": "test-author"},
"expires_at": "2999-01-01T00:00:00Z",
}
with open(self.lock_path, "w", encoding="utf-8") as fh:
json.dump({
"issue_number": issue_number,
"branch_name": branch,
"remote": "prgs",
"work_lease": work_lease,
"lock_provenance": issue_lock_provenance.build_sanctioned_lock_provenance(
tool="gitea_lock_issue",
claimant=work_lease.get("claimant"),
),
}, fh)
issue_lock_store.bind_session_lock({
"issue_number": issue_number,
"branch_name": branch,
"remote": "prgs",
"org": "Example-Org",
"repo": "Example-Repo",
"worktree_path": worktree_path,
"work_lease": work_lease,
"lock_provenance": issue_lock_provenance.build_sanctioned_lock_provenance(
tool="gitea_lock_issue",
claimant=work_lease.get("claimant"),
),
})
@patch("mcp_server._assess_issue_duplicate_gate")
@patch("mcp_server.get_profile", return_value={
@@ -254,6 +261,7 @@ class TestMcpDuplicateRecheck(unittest.TestCase):
base="master",
body="Closes #400",
remote="prgs",
worktree_path=os.path.realpath(os.getcwd()),
)
self.assertFalse(result["success"])
self.assertIsNone(result.get("number"))
tests/test_mcp_server.py
+209 -83
View File
@@ -6,6 +6,7 @@ the MCP protocol) with mocked API responses.
import json
import os
import sys
import tempfile
import unittest
from unittest.mock import patch, MagicMock
@@ -45,6 +46,7 @@ from gitea_auth import get_profile # noqa: E402
import gitea_config # noqa: E402
import mcp_server
import issue_lock_store
FAKE_AUTH = "Basic dGVzdDp0ZXN0"
@@ -174,6 +176,7 @@ def _sample_issue_lock(issue_number=123, branch_name="feat/x", **overrides):
"remote": "dadeschools",
"org": "Scaled-Tech-Consulting",
"repo": "Gitea-Tools",
"worktree_path": "/tmp/test-worktree",
"work_lease": work_lease,
"lock_provenance": issue_lock_provenance.build_sanctioned_lock_provenance(
tool="gitea_lock_issue",
@@ -189,6 +192,17 @@ def _clear_duplicate_context_fetcher(*_args, **_kwargs):
return [], [], {"status": "not_claimed"}
def _bind_test_lock(**overrides) -> str:
remote = overrides.get("remote", "dadeschools")
record = _sample_issue_lock(**overrides)
if remote in mcp_server.REMOTES:
profile = mcp_server.REMOTES[remote]
record.setdefault("org", profile["org"])
record.setdefault("repo", profile["repo"])
record["remote"] = remote
return issue_lock_store.bind_session_lock(record)
# ---------------------------------------------------------------------------
# Create Issue
# ---------------------------------------------------------------------------
@@ -251,18 +265,21 @@ class TestCreatePR(unittest.TestCase):
return_value=(True, []))
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
@patch("os.path.exists", return_value=True)
@patch("builtins.open")
def test_creates_pr(self, mock_open, mock_exists, _auth, mock_api, _role, _dup_fetcher):
lock_json = json.dumps(_sample_issue_lock(issue_number=123, branch_name="feat/x"))
mock_open.return_value.__enter__.return_value.read.return_value = lock_json
def test_creates_pr(self, _auth, mock_api, _role, _dup_fetcher):
worktree = os.path.realpath(os.getcwd())
mock_api.return_value = {"number": 3, "html_url": "https://example.com/pulls/3"}
with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
result = gitea_create_pr(title="feat: X Closes #123", head="feat/x", base="main")
with tempfile.TemporaryDirectory() as lock_dir:
env = {**CREATE_PR_ENV, "GITEA_ISSUE_LOCK_DIR": lock_dir}
with patch.dict(os.environ, env, clear=True):
_bind_test_lock(issue_number=123, branch_name="feat/x", worktree_path=worktree)
result = gitea_create_pr(
title="feat: X Closes #123",
head="feat/x",
base="main",
worktree_path=worktree,
)
self.assertEqual(result["number"], 3)
self.assertNotIn("url", result)
mock_exists.assert_called_with(ISSUE_LOCK_FILE)
mock_open.assert_called_with(ISSUE_LOCK_FILE, "r", encoding="utf-8")
payload = mock_api.call_args[0][3]
self.assertEqual(payload["head"], "feat/x")
self.assertEqual(payload["base"], "main")
@@ -276,30 +293,42 @@ class TestCreatePR(unittest.TestCase):
return_value=(True, []))
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
@patch("os.path.exists", return_value=True)
@patch("builtins.open")
def test_create_pr_reveal_opt_in_includes_url(self, mock_open, mock_exists, _auth, mock_api, _role, _dup_fetcher):
lock_json = json.dumps(_sample_issue_lock(issue_number=123, branch_name="feat/x"))
mock_open.return_value.__enter__.return_value.read.return_value = lock_json
def test_create_pr_reveal_opt_in_includes_url(self, _auth, mock_api, _role, _dup_fetcher):
worktree = os.path.realpath(os.getcwd())
mock_api.return_value = {"number": 3, "html_url": "https://example.com/pulls/3"}
env = {**CREATE_PR_ENV, "GITEA_MCP_REVEAL_ENDPOINTS": "1"}
with patch.dict(os.environ, env, clear=True):
result = gitea_create_pr(title="feat: X Closes #123", head="feat/x", base="main")
with tempfile.TemporaryDirectory() as lock_dir:
env = {**CREATE_PR_ENV, "GITEA_ISSUE_LOCK_DIR": lock_dir, "GITEA_MCP_REVEAL_ENDPOINTS": "1"}
with patch.dict(os.environ, env, clear=True):
_bind_test_lock(issue_number=123, branch_name="feat/x", worktree_path=worktree)
result = gitea_create_pr(
title="feat: X Closes #123",
head="feat/x",
base="main",
worktree_path=worktree,
)
self.assertIn("pulls/3", result["url"])
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
return_value=(True, []))
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
@patch("os.path.exists", return_value=True)
@patch("builtins.open")
def test_create_pr_locked_issue_mismatch_fails(self, mock_open, mock_exists, _auth, _role):
lock_json = json.dumps(_sample_issue_lock(issue_number=123, branch_name="feat/x"))
mock_open.return_value.__enter__.return_value.read.return_value = lock_json
with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
with self.assertRaises(ValueError) as ctx:
gitea_create_pr(title="feat: X Closes #999", head="feat/x", base="main")
def test_create_pr_locked_issue_mismatch_fails(self, _auth, _role):
worktree = os.path.realpath(os.getcwd())
with tempfile.TemporaryDirectory() as lock_dir:
env = {**CREATE_PR_ENV, "GITEA_ISSUE_LOCK_DIR": lock_dir}
with patch.dict(os.environ, env, clear=True):
_bind_test_lock(
issue_number=123,
branch_name="feat/x",
worktree_path=worktree,
)
with self.assertRaises(ValueError) as ctx:
gitea_create_pr(
title="feat: X Closes #999",
head="feat/x",
base="main",
worktree_path=worktree,
)
self.assertIn("Closes #123", str(ctx.exception))
mock_open.assert_called_with(ISSUE_LOCK_FILE, "r", encoding="utf-8")
# ---------------------------------------------------------------------------
@@ -3184,7 +3213,12 @@ class TestIssueLocking(unittest.TestCase):
"""Test issue locking and PR gating constraints."""
def setUp(self):
self._env_patcher = patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True)
self._lock_dir = tempfile.TemporaryDirectory()
env = {
**ISSUE_WRITE_ENV,
"GITEA_ISSUE_LOCK_DIR": self._lock_dir.name,
}
self._env_patcher = patch.dict(os.environ, env, clear=True)
self._env_patcher.start()
self._dup_fetcher_patcher = patch(
"mcp_server.issue_duplicate_context_fetcher",
@@ -3195,15 +3229,21 @@ class TestIssueLocking(unittest.TestCase):
def tearDown(self):
self._dup_fetcher_patcher.stop()
self._env_patcher.stop()
if os.path.exists(ISSUE_LOCK_FILE):
os.remove(ISSUE_LOCK_FILE)
self._lock_dir.cleanup()
def _create_pr_env(self) -> dict:
return {
**CREATE_PR_ENV,
"GITEA_ISSUE_LOCK_DIR": self._lock_dir.name,
}
@patch(
"mcp_server.issue_lock_worktree.read_worktree_git_state",
return_value=_clean_master_git_state_for_lock(),
)
@patch("mcp_server.api_get_all", return_value=[])
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_lock_issue_success(self, _auth, _git_state):
def test_lock_issue_success(self, _auth, _api, _git_state):
res = gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
self.assertTrue(res["success"])
self.assertEqual(res["work_lease"]["operation_type"], "author_issue_work")
@@ -3213,9 +3253,8 @@ class TestIssueLocking(unittest.TestCase):
self.assertIn("expires_at", res["work_lease"])
self.assertIn("last_heartbeat_at", res["work_lease"])
self.assertEqual(res["work_lease"]["claimant"]["profile"], "gitea-default")
self.assertTrue(os.path.exists(ISSUE_LOCK_FILE))
with open(ISSUE_LOCK_FILE, encoding="utf-8") as f:
lock = json.load(f)
self.assertIn("lock_file_path", res)
lock = issue_lock_store.read_lock_file(res["lock_file_path"])
self.assertIn("worktree_path", lock)
self.assertIn("work_lease", lock)
@@ -3271,32 +3310,81 @@ class TestIssueLocking(unittest.TestCase):
gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
self.assertIn("remote branch(es) already match issue pattern", str(ctx.exception))
def test_lock_issue_blocks_active_same_operation_lease(self):
with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
json.dump({
@patch(
"mcp_server.issue_lock_worktree.read_worktree_git_state",
return_value=_clean_master_git_state_for_lock(),
)
@patch("mcp_server.api_get_all")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_lock_issue_adopts_exact_own_branch(self, _auth, mock_api, _git_state):
branch = "feat/issue-196-mutations"
self.mock_dup_fetcher.return_value = ([], [branch], {"status": "not_claimed"})
mock_api.return_value = [{"name": branch, "commit": {"id": "abc123"}}]
res = gitea_lock_issue(issue_number=196, branch_name=branch, remote="prgs")
self.assertTrue(res["success"])
self.assertIn("adoption", res)
self.assertEqual(res["adoption"]["branch_head_commit"], "abc123")
@patch(
"mcp_server.issue_lock_worktree.read_worktree_git_state",
return_value=_clean_master_git_state_for_lock(),
)
@patch("mcp_server.api_get_all", return_value=[])
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_lock_issue_blocks_active_same_operation_lease(self, _auth, _api, _git_state):
prgs_repo = mcp_server.REMOTES["prgs"]["repo"]
issue_lock_store.save_lock_file(
issue_lock_store.lock_file_path(
remote="prgs",
org="Scaled-Tech-Consulting",
repo=prgs_repo,
issue_number=196,
),
{
"issue_number": 196,
"branch_name": "feat/issue-196-other-work",
"remote": "prgs",
"org": "Scaled-Tech-Consulting",
"repo": prgs_repo,
"worktree_path": "/tmp/other-worktree",
"work_lease": {
"operation_type": "author_issue_work",
"expires_at": "2999-01-01T00:00:00Z",
},
}, f)
},
)
with self.assertRaises(RuntimeError) as ctx:
gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
self.assertIn("already has an active author_issue_work lease", str(ctx.exception))
def test_lock_issue_blocks_expired_same_operation_lease_for_recovery(self):
with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
json.dump({
@patch(
"mcp_server.issue_lock_worktree.read_worktree_git_state",
return_value=_clean_master_git_state_for_lock(),
)
@patch("mcp_server.api_get_all", return_value=[])
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_lock_issue_blocks_expired_same_operation_lease_for_recovery(self, _auth, _api, _git_state):
prgs_repo = mcp_server.REMOTES["prgs"]["repo"]
issue_lock_store.save_lock_file(
issue_lock_store.lock_file_path(
remote="prgs",
org="Scaled-Tech-Consulting",
repo=prgs_repo,
issue_number=196,
),
{
"issue_number": 196,
"branch_name": "feat/issue-196-other-work",
"remote": "prgs",
"org": "Scaled-Tech-Consulting",
"repo": prgs_repo,
"worktree_path": "/tmp/other-worktree",
"work_lease": {
"operation_type": "author_issue_work",
"expires_at": "2000-01-01T00:00:00Z",
},
}, f)
},
)
with self.assertRaises(RuntimeError) as ctx:
gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
self.assertIn("Recovery review is required before takeover", str(ctx.exception))
@@ -3363,9 +3451,7 @@ class TestIssueLocking(unittest.TestCase):
return_value=(True, []))
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_create_pr_missing_lock_fails(self, _auth, _role):
if os.path.exists(ISSUE_LOCK_FILE):
os.remove(ISSUE_LOCK_FILE)
with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
with patch.dict(os.environ, self._create_pr_env(), clear=True):
with self.assertRaises(RuntimeError) as ctx:
gitea_create_pr(title="feat: X Closes #196", head="feat/issue-196-mutations", remote="prgs")
self.assertIn("Issue lock is missing", str(ctx.exception))
@@ -3374,37 +3460,64 @@ class TestIssueLocking(unittest.TestCase):
return_value=(True, []))
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_create_pr_branch_mismatch_fails(self, _auth, _role):
with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
json.dump(_sample_issue_lock(
issue_number=196, branch_name="feat/issue-196-mutations"), f)
with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
worktree = os.path.realpath(os.getcwd())
_bind_test_lock(
issue_number=196,
branch_name="feat/issue-196-mutations",
remote="prgs",
worktree_path=worktree,
)
with patch.dict(os.environ, self._create_pr_env(), clear=True):
with self.assertRaises(ValueError) as ctx:
gitea_create_pr(title="feat: X Closes #196", head="feat/issue-196-different", remote="prgs")
gitea_create_pr(
title="feat: X Closes #196",
head="feat/issue-196-different",
remote="prgs",
worktree_path=worktree,
)
self.assertIn("does not match locked branch", str(ctx.exception))
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
return_value=(True, []))
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_create_pr_forbidden_terms_fails(self, _auth, _role):
with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
json.dump(_sample_issue_lock(
issue_number=196, branch_name="feat/issue-196-mutations"), f)
with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
worktree = os.path.realpath(os.getcwd())
_bind_test_lock(
issue_number=196,
branch_name="feat/issue-196-mutations",
remote="prgs",
worktree_path=worktree,
)
with patch.dict(os.environ, self._create_pr_env(), clear=True):
for term in ("equivalent to #196", "related to #196", "same as #196"):
with self.assertRaises(ValueError) as ctx:
gitea_create_pr(title=f"feat: X {term}", head="feat/issue-196-mutations", remote="prgs")
gitea_create_pr(
title=f"feat: X {term}",
head="feat/issue-196-mutations",
remote="prgs",
worktree_path=worktree,
)
self.assertIn("contains forbidden term", str(ctx.exception))
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
return_value=(True, []))
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_create_pr_missing_closes_ref_fails(self, _auth, _role):
with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
json.dump(_sample_issue_lock(
issue_number=196, branch_name="feat/issue-196-mutations"), f)
with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
worktree = os.path.realpath(os.getcwd())
_bind_test_lock(
issue_number=196,
branch_name="feat/issue-196-mutations",
remote="prgs",
worktree_path=worktree,
)
with patch.dict(os.environ, self._create_pr_env(), clear=True):
with self.assertRaises(ValueError) as ctx:
gitea_create_pr(title="feat: X refs #196", head="feat/issue-196-mutations", remote="prgs")
gitea_create_pr(
title="feat: X refs #196",
head="feat/issue-196-mutations",
remote="prgs",
worktree_path=worktree,
)
self.assertIn("must contain 'Closes #196' or 'Fixes #196' exactly", str(ctx.exception))
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
@@ -3412,13 +3525,13 @@ class TestIssueLocking(unittest.TestCase):
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_create_pr_worktree_mismatch_fails(self, _auth, _role):
scratch = os.path.realpath("/tmp/gitea-tools-author-scratch/issue-249-pr")
with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
json.dump(_sample_issue_lock(
issue_number=249,
branch_name="feat/issue-249-issue-lock-scratch-worktree",
worktree_path=scratch,
), f)
with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
_bind_test_lock(
issue_number=249,
branch_name="feat/issue-249-issue-lock-scratch-worktree",
worktree_path=scratch,
remote="prgs",
)
with patch.dict(os.environ, self._create_pr_env(), clear=True):
with self.assertRaises(ValueError) as ctx:
gitea_create_pr(
title="feat: lock scratch worktree Closes #249",
@@ -3432,22 +3545,35 @@ class TestIssueLocking(unittest.TestCase):
return_value=(True, []))
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_create_pr_manual_lock_seed_blocked(self, _auth, _role):
with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
json.dump(
worktree = os.path.realpath(os.getcwd())
with tempfile.TemporaryDirectory() as lock_dir:
env = {**self._create_pr_env(), "GITEA_ISSUE_LOCK_DIR": lock_dir}
with patch.dict(os.environ, env, clear=True):
issue_lock_store.save_lock_file(
issue_lock_store.lock_file_path(
remote="prgs",
org="Scaled-Tech-Consulting",
repo=mcp_server.REMOTES["prgs"]["repo"],
issue_number=447,
lock_dir=lock_dir,
),
_sample_issue_lock(
issue_number=447,
branch_name="feat/issue-447-lock-provenance",
remote="prgs",
org="Scaled-Tech-Consulting",
repo=mcp_server.REMOTES["prgs"]["repo"],
worktree_path=worktree,
lock_provenance=None,
),
f,
)
with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
with self.assertRaises(RuntimeError) as ctx:
gitea_create_pr(
title="feat: lock provenance Closes #447",
head="feat/issue-447-lock-provenance",
remote="prgs",
)
with self.assertRaises(RuntimeError) as ctx:
gitea_create_pr(
title="feat: lock provenance Closes #447",
head="feat/issue-447-lock-provenance",
remote="prgs",
worktree_path=worktree,
)
self.assertIn("lock provenance", str(ctx.exception).lower())
@patch("mcp_server.api_request")
@@ -3457,13 +3583,13 @@ class TestIssueLocking(unittest.TestCase):
def test_create_pr_honors_scratch_worktree_lock(self, _auth, _role, mock_api):
scratch = os.path.realpath("/tmp/gitea-tools-author-scratch/issue-249-e2e")
mock_api.return_value = {"number": 250, "html_url": "https://example/pr/250"}
with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
json.dump(_sample_issue_lock(
issue_number=249,
branch_name="feat/issue-249-issue-lock-scratch-worktree",
worktree_path=scratch,
), f)
with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
_bind_test_lock(
issue_number=249,
branch_name="feat/issue-249-issue-lock-scratch-worktree",
worktree_path=scratch,
remote="prgs",
)
with patch.dict(os.environ, self._create_pr_env(), clear=True):
res = gitea_create_pr(
title="feat: issue-lock scratch worktree Closes #249",
head="feat/issue-249-issue-lock-scratch-worktree",
tests/test_worktrees.py
+26 -9
View File
@@ -20,33 +20,50 @@ def run(script, *args):
branch = arg
break
lock_file = Path("/tmp/gitea_issue_lock.json")
created_lock = False
lock_dir_ctx = None
extra_env = os.environ.copy()
if script == "worktree-start" and branch:
import re
import json
import tempfile
import issue_lock_store
m = re.search(r"issue-(\d+)", branch)
if not m:
m = re.search(r"pr-(\d+)", branch)
issue_num = int(m.group(1)) if m else 999
lock_file.write_text(json.dumps({
lock_dir_ctx = tempfile.TemporaryDirectory()
extra_env["GITEA_ISSUE_LOCK_DIR"] = lock_dir_ctx.name
record = {
"issue_number": issue_num,
"branch_name": branch,
"remote": "prgs",
"org": "Scaled-Tech-Consulting",
"repo": "Gitea-Tools"
}), encoding="utf-8")
created_lock = True
"repo": "Gitea-Tools",
"worktree_path": "/tmp/test-worktree",
"work_lease": {
"operation_type": "author_issue_work",
"expires_at": "2999-01-01T00:00:00Z",
},
}
path = issue_lock_store.lock_file_path(
remote="prgs",
org="Scaled-Tech-Consulting",
repo="Gitea-Tools",
issue_number=issue_num,
lock_dir=lock_dir_ctx.name,
)
issue_lock_store.save_lock_file(path, record)
try:
proc = subprocess.run(
["bash", str(SCRIPTS / script), *args],
capture_output=True, text=True, cwd=str(REPO),
env=extra_env,
)
return proc.returncode, proc.stdout, proc.stderr
finally:
if created_lock and lock_file.exists():
lock_file.unlink()
if lock_dir_ctx is not None:
lock_dir_ctx.cleanup()
class TestWorktreeStart(unittest.TestCase):