sysadminandClaude Opus 4.8
0cbd1fc801
Normal author work is unchanged: gitea_lock_issue still requires the worktree
to be base-equivalent to master/main/dev, and gitea_create_pr still targets a
normal base branch. A *stacked* PR (based on another unmerged PR's branch) is a
new explicit, proof-backed path — it never bypasses the issue lock.
- stacked_pr_support.py: pure policy. Approve a non-master base only when it is
declared AND owned by a live OPEN PR whose number matches; reject arbitrary,
mismatched, or stale (merged/closed) branches; require the PR body to document
the stack (base branch, stacked-on PR, merge ordering).
- issue_lock_worktree.py: read_worktree_git_state / _find_matching_base_ref gain
an opt-in extra_bases arg so an approved stacked base can anchor
base-equivalence in addition to master/main/dev (empty by default = unchanged).
- gitea_mcp_server.py:
- gitea_lock_issue gains stacked_base_branch + stacked_base_pr. When declared,
it verifies the open PR owns the branch, anchors base-equivalence to it, and
records approved_stacked_base = {branch, pr_number, verified_open} on the lock.
- gitea_create_pr validates a non-master base against the lock's approved
stacked base, re-checks the dependency PR is still open, and enforces the
stacked-PR body fields. Master-based path untouched.
- Docs: llm-workflow-runbooks.md and work-issue.md document when stacked PRs are
allowed and the required PR-body wording.
Tests: test_stacked_pr_support.py (18 policy cases), stacked base-equivalence in
test_issue_lock_worktree.py, approved_stacked_base persistence round-trip in
test_issue_lock_store.py. Existing lock/create_pr regressions still pass.
The #482 -> #479/#478 case motivates this: create_pr previously could not open a
stacked PR because the lock required a master-equivalent worktree.
Closes #484.
Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
280 lines
9.8 KiB
Python
280 lines
9.8 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):
|
|
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_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() |