Gitea-Tools/tests/test_issue_work_duplicate_gate.py

"""Tests for early duplicate-work detection (#400)."""
import json
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_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 TestMcpDuplicateRecheck(unittest.TestCase):
    def setUp(self):
        self._dir = tempfile.TemporaryDirectory()
        self.lock_path = os.path.join(self._dir.name, "gitea_issue_lock.json")
        self._lock_patch = patch.object(
            mcp_server, "ISSUE_LOCK_FILE", self.lock_path
        )
        self._lock_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"):
        with open(self.lock_path, "w", encoding="utf-8") as fh:
            json.dump({
                "issue_number": issue_number,
                "branch_name": branch,
                "remote": "prgs",
            }, fh)

    @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",
            )
        self.assertFalse(result["success"])
        self.assertIsNone(result.get("number"))
        self.assertIn("duplicate_gate", result)


if __name__ == "__main__":
    unittest.main()