"""Tests for pre-create issue duplicate gate (Issue #207).""" import sys import unittest from unittest.mock import patch sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent)) from issue_duplicate_gate import ( # noqa: E402 VERDICT_AMBIGUOUS, VERDICT_DUPLICATE, VERDICT_NO_DUPLICATE, assess_duplicate_search_proof, assess_pre_create_duplicate, normalize_issue_title, pre_create_issue_duplicate_gate, titles_near_duplicate, ) from mcp_server import gitea_create_issue # noqa: E402 from review_proofs import assess_duplicate_search_proof as proof_assess # noqa: E402 FAKE_AUTH = "Basic dGVzdDp0ZXN0" ISSUE_WRITE_ENV = { "GITEA_ALLOWED_OPERATIONS": "gitea.issue.create", } CANONICAL_TITLE = ( "Add hard queue-target resolution wall before PR inventory " "or empty-queue claims" ) class TestNormalizeAndNearDuplicate(unittest.TestCase): def test_normalize_strips_punctuation_and_case(self): norm = normalize_issue_title(" Hard-Wall: Queue Target! ") self.assertEqual(norm, "hard wall queue target") def test_exact_title_match(self): self.assertTrue(titles_near_duplicate(CANONICAL_TITLE, CANONICAL_TITLE)) def test_punctuation_and_capitalization_only_differs(self): proposed = CANONICAL_TITLE.upper().replace("-", " — ") self.assertTrue(titles_near_duplicate(proposed, CANONICAL_TITLE)) def test_hard_wall_vs_wall_wording(self): self.assertTrue( titles_near_duplicate( "Add hard wall before PR inventory", "Add wall before PR inventory", ) ) class TestPreCreateGate(unittest.TestCase): def test_blocks_exact_duplicate_title(self): existing = [{"number": 201, "title": CANONICAL_TITLE, "state": "open"}] gate = assess_pre_create_duplicate(CANONICAL_TITLE, existing) self.assertEqual(gate["verdict"], VERDICT_DUPLICATE) self.assertFalse(gate["performed"]) self.assertEqual(gate["matches"][0]["number"], 201) def test_final_pre_create_check_blocks_even_if_llm_search_missed(self): """TOCTOU: gate re-queries at create time with the live issue list.""" stale_report = ( "Searched 20 open issues (#194, #196, PR #195); no duplicate found." ) live_existing = [{"number": 201, "title": CANONICAL_TITLE, "state": "open"}] proof = assess_duplicate_search_proof(stale_report, live_existing) self.assertFalse(proof["valid"]) gate = assess_pre_create_duplicate(CANONICAL_TITLE, live_existing) self.assertFalse(gate["performed"]) def test_invalid_proof_when_summary_omits_exact_duplicate(self): report = "Reviewed nearby issues #194, #196, and PR #195; no duplicate." matches = [{"number": 201, "title": CANONICAL_TITLE}] result = assess_duplicate_search_proof(report, matches) self.assertFalse(result["valid"]) self.assertTrue(any("201" in r for r in result["reasons"])) def test_operator_split_override_allowed_with_relationship(self): existing = [{"number": 201, "title": CANONICAL_TITLE, "state": "open"}] gate = pre_create_issue_duplicate_gate( CANONICAL_TITLE, existing, allow_override=True, split_from_issue=201, ) self.assertEqual(gate["verdict"], VERDICT_NO_DUPLICATE) self.assertTrue(gate["performed"]) self.assertTrue(gate.get("override_applied")) def test_ambiguous_when_too_many_near_matches(self): base = "Add hard wall before PR inventory" existing = [ {"number": n, "title": f"{base} variant {n}", "state": "open"} for n in range(1, 6) ] gate = assess_pre_create_duplicate(base, existing) self.assertEqual(gate["verdict"], VERDICT_AMBIGUOUS) self.assertFalse(gate["performed"]) def test_no_duplicate_when_unique_title(self): gate = assess_pre_create_duplicate( "Brand new unique issue title", [{"number": 1, "title": "Unrelated backlog cleanup", "state": "open"}], ) self.assertEqual(gate["verdict"], VERDICT_NO_DUPLICATE) self.assertTrue(gate["performed"]) class TestCreateIssueMCPGate(unittest.TestCase): @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop") @patch("mcp_server.api_request") @patch("mcp_server.api_get_all") @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) def test_blocks_normalized_title_match_without_override( self, _auth, mock_get_all, mock_api, mock_role_check ): mock_role_check.return_value = (True, []) mock_get_all.return_value = [ {"number": 201, "title": CANONICAL_TITLE, "state": "open"}, ] with patch.dict(__import__("os").environ, ISSUE_WRITE_ENV, clear=True): result = gitea_create_issue(title=CANONICAL_TITLE.upper(), remote="prgs") self.assertFalse(result["performed"]) self.assertEqual(result["duplicate_gate"], VERDICT_DUPLICATE) mock_api.assert_not_called() @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop") @patch("mcp_server.api_request") @patch("mcp_server.api_get_all") @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) def test_override_creates_with_split_relationship_in_body( self, _auth, mock_get_all, mock_api, mock_role_check ): mock_role_check.return_value = (True, []) mock_get_all.return_value = [ {"number": 201, "title": CANONICAL_TITLE, "state": "open"}, ] mock_api.return_value = {"number": 210, "html_url": "https://gitea.prgs.cc/issues/210"} with patch.dict(__import__("os").environ, ISSUE_WRITE_ENV, clear=True): result = gitea_create_issue( title=CANONICAL_TITLE, body="Follow-up scope", remote="prgs", allow_duplicate_override=True, split_from_issue=201, ) self.assertEqual(result["number"], 210) payload = mock_api.call_args[0][3] self.assertIn("Operator-approved split from #201.", payload["body"]) self.assertIn("Follow-up scope", payload["body"]) @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop") @patch("mcp_server.api_request") @patch("mcp_server.api_get_all", return_value=[]) @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) def test_creates_when_no_duplicate( self, _auth, _get_all, mock_api, mock_role_check ): mock_role_check.return_value = (True, []) mock_api.return_value = {"number": 1, "html_url": "https://gitea.example.com/issues/1"} with patch.dict(__import__("os").environ, ISSUE_WRITE_ENV, clear=True): result = gitea_create_issue(title="Unique new issue", body="body text") self.assertEqual(result["number"], 1) class TestReviewProofsWrapper(unittest.TestCase): def test_review_proofs_reexports_duplicate_search_validator(self): report = "Checked open issues; nothing similar." matches = [{"number": 200, "title": CANONICAL_TITLE}] result = proof_assess(report, matches) self.assertFalse(result["valid"]) if __name__ == "__main__": unittest.main()