sysadminandClaude Opus 4.8
e956040b6b
Update lock-issue and duplicate-recheck tests to mock branch listing, use GITEA_ISSUE_LOCK_DIR session binding, and satisfy create_pr provenance/worktree guards after the keyed persistent lock store landed. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
272 lines
9.7 KiB
Python
272 lines
9.7 KiB
Python
"""Tests for early duplicate-work detection (#400)."""
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
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 (
|
|
OUTCOME_DUPLICATE_BRANCH_PREVENTED,
|
|
OUTCOME_DUPLICATE_COMMIT_PREVENTED,
|
|
OUTCOME_DUPLICATE_PR_PREVENTED,
|
|
OUTCOME_DUPLICATE_WORK_NOT_PREVENTED,
|
|
PHASE_COMMIT,
|
|
PHASE_CREATE_PR,
|
|
PHASE_LOCK,
|
|
assess_work_issue_duplicate_gate,
|
|
assess_work_issue_duplicate_report,
|
|
)
|
|
|
|
|
|
class TestDuplicateGateAssessment(unittest.TestCase):
|
|
def test_clear_issue_passes(self):
|
|
result = assess_work_issue_duplicate_gate(
|
|
400,
|
|
open_prs=[],
|
|
branch_names=["feat/other-issue-99"],
|
|
claim_entry={"status": "not_claimed"},
|
|
locked_branch="feat/issue-400-duplicate-work-preflight",
|
|
phase=PHASE_LOCK,
|
|
)
|
|
self.assertFalse(result["block"])
|
|
self.assertEqual(result["outcome"], OUTCOME_DUPLICATE_WORK_NOT_PREVENTED)
|
|
|
|
def test_open_pr_blocks(self):
|
|
prs = [{
|
|
"number": 397,
|
|
"title": "feat: handoff",
|
|
"body": "Closes #395",
|
|
"head": {"ref": "feat/issue-395-proof-backed-review-handoff"},
|
|
}]
|
|
result = assess_work_issue_duplicate_gate(
|
|
395,
|
|
open_prs=prs,
|
|
branch_names=[],
|
|
phase=PHASE_LOCK,
|
|
)
|
|
self.assertTrue(result["block"])
|
|
self.assertEqual(result["outcome"], OUTCOME_DUPLICATE_PR_PREVENTED)
|
|
|
|
def test_conflicting_remote_branch_blocks(self):
|
|
result = assess_work_issue_duplicate_gate(
|
|
395,
|
|
open_prs=[],
|
|
branch_names=["feat/issue-395-proof-backed-handoff-claims"],
|
|
locked_branch="feat/issue-395-new-attempt",
|
|
phase=PHASE_LOCK,
|
|
)
|
|
self.assertTrue(result["block"])
|
|
self.assertEqual(result["outcome"], OUTCOME_DUPLICATE_BRANCH_PREVENTED)
|
|
|
|
def test_active_claim_on_other_branch_blocks(self):
|
|
result = assess_work_issue_duplicate_gate(
|
|
398,
|
|
open_prs=[],
|
|
branch_names=[],
|
|
claim_entry={
|
|
"status": "active",
|
|
"latest_heartbeat": {
|
|
"branch": "feat/issue-398-validation-cwd-proof",
|
|
},
|
|
},
|
|
locked_branch="feat/issue-398-other-branch",
|
|
phase=PHASE_LOCK,
|
|
)
|
|
self.assertTrue(result["block"])
|
|
|
|
def test_commit_phase_maps_to_commit_outcome(self):
|
|
prs = [{
|
|
"number": 411,
|
|
"title": "x",
|
|
"body": "Closes #398",
|
|
"head": {"ref": "feat/issue-398-validation-cwd-proof"},
|
|
}]
|
|
result = assess_work_issue_duplicate_gate(
|
|
398,
|
|
open_prs=prs,
|
|
branch_names=[],
|
|
locked_branch="feat/issue-398-alt",
|
|
phase=PHASE_COMMIT,
|
|
)
|
|
self.assertTrue(result["block"])
|
|
self.assertEqual(result["outcome"], OUTCOME_DUPLICATE_COMMIT_PREVENTED)
|
|
|
|
def test_stale_claim_does_not_block_by_status_alone(self):
|
|
result = assess_work_issue_duplicate_gate(
|
|
400,
|
|
open_prs=[],
|
|
branch_names=[],
|
|
claim_entry={"status": "reclaimable", "reasons": ["stale"]},
|
|
locked_branch="feat/issue-400-duplicate-work-preflight",
|
|
phase=PHASE_LOCK,
|
|
)
|
|
self.assertFalse(result["block"])
|
|
|
|
|
|
class TestDuplicateReportOutcome(unittest.TestCase):
|
|
def test_requires_exactly_one_outcome(self):
|
|
bad = assess_work_issue_duplicate_report("work finished")
|
|
self.assertFalse(bad["complete"])
|
|
|
|
good = assess_work_issue_duplicate_report(
|
|
"Duplicate work not prevented for issue #400."
|
|
)
|
|
self.assertTrue(good["complete"])
|
|
self.assertEqual(good["outcome"], OUTCOME_DUPLICATE_WORK_NOT_PREVENTED)
|
|
|
|
|
|
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, _api):
|
|
seen = {}
|
|
|
|
def fetcher(h, o, r, auth, issue_number):
|
|
seen["issue_number"] = issue_number
|
|
return [], [], {"status": "not_claimed"}
|
|
|
|
with patch(
|
|
"mcp_server.issue_duplicate_context_fetcher",
|
|
side_effect=fetcher,
|
|
), patch(
|
|
"mcp_server.issue_lock_worktree.read_worktree_git_state",
|
|
return_value={
|
|
"current_branch": "master",
|
|
"porcelain_status": "",
|
|
"base_equivalent": True,
|
|
},
|
|
):
|
|
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._env_patch = patch.dict(
|
|
os.environ,
|
|
{"GITEA_ISSUE_LOCK_DIR": self._dir.name},
|
|
clear=False,
|
|
)
|
|
self._env_patch.start()
|
|
self._remotes = patch.dict(mcp_server.REMOTES, {
|
|
"prgs": {"host": "gitea.example.com", "org": "Example-Org",
|
|
"repo": "Example-Repo"},
|
|
})
|
|
self._remotes.start()
|
|
mcp_server._IDENTITY_CACHE.clear()
|
|
|
|
def tearDown(self):
|
|
patch.stopall()
|
|
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,
|
|
"branch": branch,
|
|
"claimant": {"username": "test-user", "profile": "test-author"},
|
|
"expires_at": "2999-01-01T00:00:00Z",
|
|
}
|
|
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={
|
|
"profile_name": "test-author",
|
|
"allowed_operations": ["gitea.read", "gitea.repo.commit"],
|
|
"forbidden_operations": [],
|
|
"audit_label": "test-author",
|
|
})
|
|
@patch("mcp_server.get_auth_header", return_value="token x")
|
|
def test_commit_files_blocked_on_recheck(self, _auth, _profile, mock_gate):
|
|
self._write_lock()
|
|
mock_gate.return_value = {
|
|
"block": True,
|
|
"reasons": ["open PR #412 already covers issue #400"],
|
|
"outcome": OUTCOME_DUPLICATE_COMMIT_PREVENTED,
|
|
"safe_next_action": "stop",
|
|
}
|
|
mcp_server.record_preflight_check("whoami")
|
|
mcp_server.record_preflight_check("capability", resolved_role="author")
|
|
with patch(
|
|
"mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
|
|
return_value=(True, []),
|
|
):
|
|
result = mcp_server.gitea_commit_files(
|
|
files=[{
|
|
"operation": "create",
|
|
"path": "a.txt",
|
|
"content_plain": "hi",
|
|
}],
|
|
message="test",
|
|
remote="prgs",
|
|
)
|
|
self.assertFalse(result["success"])
|
|
self.assertIn("duplicate_gate", result)
|
|
|
|
@patch("mcp_server._assess_issue_duplicate_gate")
|
|
@patch("mcp_server.get_profile", return_value={
|
|
"profile_name": "test-author",
|
|
"allowed_operations": ["gitea.read", "gitea.pr.create"],
|
|
"forbidden_operations": [],
|
|
"audit_label": "test-author",
|
|
})
|
|
@patch("mcp_server.get_auth_header", return_value="token x")
|
|
def test_create_pr_returns_handoff_on_duplicate(self, _auth, _profile, mock_gate):
|
|
self._write_lock()
|
|
mock_gate.return_value = {
|
|
"block": True,
|
|
"reasons": ["open PR #412 already covers issue #400"],
|
|
"outcome": OUTCOME_DUPLICATE_PR_PREVENTED,
|
|
"safe_next_action": "reconciliation handoff",
|
|
}
|
|
mcp_server.record_preflight_check("whoami")
|
|
mcp_server.record_preflight_check("capability", resolved_role="author")
|
|
with patch(
|
|
"mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
|
|
return_value=(True, []),
|
|
):
|
|
result = mcp_server.gitea_create_pr(
|
|
title="feat: x (Closes #400)",
|
|
head="feat/issue-400-x",
|
|
base="master",
|
|
body="Closes #400",
|
|
remote="prgs",
|
|
worktree_path=os.path.realpath(os.getcwd()),
|
|
)
|
|
self.assertFalse(result["success"])
|
|
self.assertIsNone(result.get("number"))
|
|
self.assertIn("duplicate_gate", result)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main() |