Gitea-Tools/tests/test_issue_duplicate_gate.py

"""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"
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"},
        ]
        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"}
        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"}
        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()