"""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()