Gitea-Tools/tests/test_remote_repo_guard.py

"""Regression coverage for the remote/repo mismatch guard (#530).

Bare ``remote=prgs`` historically resolved to ``Scaled-Tech-Consulting/Timesheet``
(the hardcoded ``REMOTES`` default) instead of ``Scaled-Tech-Consulting/Gitea-Tools``,
which is what the local git remote actually points at. That silent mismatch caused
false 404s and risked mutating the wrong repository. The guard fails closed when the
MCP-resolved org/repo disagrees with the local git remote URL and the caller did not
pass explicit ``org``/``repo``.
"""

import os
import unittest
from unittest import mock

import remote_repo_guard


LOCAL_GITEA_TOOLS_URL = "https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git"


class TestAssessRemoteRepoMatch(unittest.TestCase):
    def test_explicit_org_and_repo_skips_guard(self):
        # Caller supplied both org and repo explicitly: never block, even on mismatch.
        assessment = remote_repo_guard.assess_remote_repo_match(
            remote="prgs",
            resolved_org="Scaled-Tech-Consulting",
            resolved_repo="Timesheet",
            local_remote_url=LOCAL_GITEA_TOOLS_URL,
            org_explicit=True,
            repo_explicit=True,
        )
        self.assertTrue(assessment["proven"])
        self.assertFalse(assessment["block"])
        self.assertEqual(assessment["reasons"], [])

    def test_missing_local_remote_url_skips_guard(self):
        # Best-effort lookup: no local remote URL (non-checkout usage) => do not block.
        assessment = remote_repo_guard.assess_remote_repo_match(
            remote="prgs",
            resolved_org="Scaled-Tech-Consulting",
            resolved_repo="Timesheet",
            local_remote_url=None,
            org_explicit=False,
            repo_explicit=False,
        )
        self.assertTrue(assessment["proven"])
        self.assertFalse(assessment["block"])

    def test_matching_repo_is_proven(self):
        assessment = remote_repo_guard.assess_remote_repo_match(
            remote="prgs",
            resolved_org="Scaled-Tech-Consulting",
            resolved_repo="Gitea-Tools",
            local_remote_url=LOCAL_GITEA_TOOLS_URL,
            org_explicit=False,
            repo_explicit=False,
        )
        self.assertTrue(assessment["proven"])
        self.assertFalse(assessment["block"])

    def test_regression_prgs_default_timesheet_vs_local_gitea_tools(self):
        # The exact #530 scenario: default repo Timesheet, local remote Gitea-Tools.
        assessment = remote_repo_guard.assess_remote_repo_match(
            remote="prgs",
            resolved_org="Scaled-Tech-Consulting",
            resolved_repo="Timesheet",
            local_remote_url=LOCAL_GITEA_TOOLS_URL,
            org_explicit=False,
            repo_explicit=False,
        )
        self.assertFalse(assessment["proven"])
        self.assertTrue(assessment["block"])
        self.assertTrue(assessment["reasons"])
        self.assertEqual(assessment["resolved_repo"], "Timesheet")
        self.assertEqual(assessment["resolved_org"], "Scaled-Tech-Consulting")

    def test_only_org_explicit_still_checks_repo(self):
        # Repo left as default => still guarded even if org was explicit.
        assessment = remote_repo_guard.assess_remote_repo_match(
            remote="prgs",
            resolved_org="Scaled-Tech-Consulting",
            resolved_repo="Timesheet",
            local_remote_url=LOCAL_GITEA_TOOLS_URL,
            org_explicit=True,
            repo_explicit=False,
        )
        self.assertTrue(assessment["block"])

    def test_case_insensitive_match(self):
        assessment = remote_repo_guard.assess_remote_repo_match(
            remote="prgs",
            resolved_org="scaled-tech-consulting",
            resolved_repo="gitea-tools",
            local_remote_url=LOCAL_GITEA_TOOLS_URL,
            org_explicit=False,
            repo_explicit=False,
        )
        self.assertTrue(assessment["proven"])

    def test_format_error_mentions_resolved_and_local(self):
        assessment = remote_repo_guard.assess_remote_repo_match(
            remote="prgs",
            resolved_org="Scaled-Tech-Consulting",
            resolved_repo="Timesheet",
            local_remote_url=LOCAL_GITEA_TOOLS_URL,
            org_explicit=False,
            repo_explicit=False,
        )
        message = remote_repo_guard.format_remote_repo_guard_error(assessment)
        self.assertIn("Timesheet", message)
        self.assertIn("Gitea-Tools", message)
        self.assertIn("org=", message)
        self.assertIn("repo=", message)
        # Recovery text must name a tool that actually accepts org/repo (#530 follow-up).
        self.assertIn("mcp_get_control_plane_guide", message)
        self.assertIn("schema does not list", message)

    def test_parse_org_repo_from_https_url(self):
        parsed = remote_repo_guard.parse_org_repo_from_remote_url(LOCAL_GITEA_TOOLS_URL)
        self.assertEqual(parsed, ("Scaled-Tech-Consulting", "Gitea-Tools"))

    def test_parse_org_repo_from_ssh_url(self):
        parsed = remote_repo_guard.parse_org_repo_from_remote_url(
            "[email protected]:Scaled-Tech-Consulting/Gitea-Tools.git"
        )
        self.assertEqual(parsed, ("Scaled-Tech-Consulting", "Gitea-Tools"))

    def test_parse_org_repo_empty(self):
        self.assertIsNone(remote_repo_guard.parse_org_repo_from_remote_url(None))
        self.assertIsNone(remote_repo_guard.parse_org_repo_from_remote_url(""))


class TestResolveServerWiring(unittest.TestCase):
    """The guard is wired into gitea_mcp_server._resolve, so every read/lookup/
    mutation tool that resolves a target fails closed on a repo mismatch.

    Under pytest the guard is bypassed unless GITEA_FORCE_REMOTE_REPO_CHECK is set,
    so these tests set the force flag and patch the local remote lookup + REMOTES.
    """

    def setUp(self):
        import gitea_mcp_server

        self.server = gitea_mcp_server

        force = mock.patch.dict(
            os.environ, {"GITEA_FORCE_REMOTE_REPO_CHECK": "1"}, clear=False
        )
        force.start()
        self.addCleanup(force.stop)

        url_patch = mock.patch.object(
            gitea_mcp_server,
            "_local_git_remote_url",
            return_value=LOCAL_GITEA_TOOLS_URL,
        )
        url_patch.start()
        self.addCleanup(url_patch.stop)

    def _set_prgs_default_repo(self, repo):
        original = dict(self.server.REMOTES["prgs"])
        self.addCleanup(self.server.REMOTES.__setitem__, "prgs", original)
        self.server.REMOTES["prgs"] = {**original, "repo": repo}

    def test_resolve_blocks_on_default_repo_mismatch(self):
        self._set_prgs_default_repo("Timesheet")
        with self.assertRaises(RuntimeError) as ctx:
            self.server._resolve("prgs", None, None, None)
        self.assertIn("Timesheet", str(ctx.exception))
        self.assertIn("Gitea-Tools", str(ctx.exception))

    def test_resolve_allows_explicit_org_and_repo(self):
        self._set_prgs_default_repo("Timesheet")
        # Explicit org/repo is authoritative even if it does not match local remote.
        host, org, repo = self.server._resolve(
            "prgs", None, "Scaled-Tech-Consulting", "Timesheet"
        )
        self.assertEqual((org, repo), ("Scaled-Tech-Consulting", "Timesheet"))

    def test_resolve_allows_matching_default(self):
        self._set_prgs_default_repo("Gitea-Tools")
        host, org, repo = self.server._resolve("prgs", None, None, None)
        self.assertEqual((org, repo), ("Scaled-Tech-Consulting", "Gitea-Tools"))

    def test_lookup_tool_cannot_silently_query_different_repo(self):
        # gitea_view_issue resolves the target via _resolve first, so a bare
        # remote=prgs pointing at the wrong default repo fails closed before any
        # API call is made.
        self._set_prgs_default_repo("Timesheet")
        with self.assertRaises(RuntimeError):
            self.server.gitea_view_issue(issue_number=1, remote="prgs")

    def test_control_plane_guide_accepts_explicit_gitea_tools(self):
        """Guide schema accepts org/repo and resolves Gitea-Tools, not Timesheet."""
        self._set_prgs_default_repo("Timesheet")
        with mock.patch.object(
            self.server, "api_request", return_value={"login": "guide-bot"}
        ), mock.patch.object(
            self.server, "get_auth_header", return_value="Basic dGVzdA=="
        ), mock.patch.dict(
            os.environ,
            {
                "GITEA_PROFILE_NAME": "author-test",
                "GITEA_ALLOWED_OPERATIONS": "gitea.read",
                "GITEA_FORBIDDEN_OPERATIONS": "gitea.pr.merge",
            },
            clear=False,
        ):
            guide = self.server.mcp_get_control_plane_guide(
                remote="prgs",
                org="Scaled-Tech-Consulting",
                repo="Gitea-Tools",
            )
        self.assertTrue(guide["read_only"])
        self.assertEqual(
            guide["resolved_repository"],
            {"org": "Scaled-Tech-Consulting", "repo": "Gitea-Tools"},
        )
        self.assertNotEqual(guide["resolved_repository"]["repo"], "Timesheet")

    def test_control_plane_guide_bare_remote_aligns_to_local_gitea_tools(self):
        """Bare remote=prgs in a Gitea-Tools worktree must not stay on Timesheet."""
        self._set_prgs_default_repo("Timesheet")
        with mock.patch.object(
            self.server, "api_request", return_value={"login": "guide-bot"}
        ), mock.patch.object(
            self.server, "get_auth_header", return_value="Basic dGVzdA=="
        ), mock.patch.dict(
            os.environ,
            {
                "GITEA_PROFILE_NAME": "author-test",
                "GITEA_ALLOWED_OPERATIONS": "gitea.read",
                "GITEA_FORBIDDEN_OPERATIONS": "gitea.pr.merge",
            },
            clear=False,
        ):
            guide = self.server.mcp_get_control_plane_guide(remote="prgs")
        self.assertEqual(guide["resolved_repository"]["repo"], "Gitea-Tools")
        self.assertEqual(
            guide["resolved_repository"]["org"], "Scaled-Tech-Consulting"
        )

    def test_control_plane_guide_wrong_default_fails_closed_without_local(self):
        """Without local remote context, mismatched defaults still fail closed."""
        self._set_prgs_default_repo("Timesheet")
        with mock.patch.object(
            self.server, "_local_git_remote_url", return_value=None
        ):
            # No local URL → preference cannot recover; bare resolve uses Timesheet
            # and guard is skipped only because local URL is missing (best-effort).
            host, org, repo = self.server._resolve_control_plane_guide_target(
                "prgs", None, None, None
            )
        self.assertEqual(repo, "Timesheet")

    def test_control_plane_guide_schema_recovery_matches_parameters(self):
        """Remediation must only recommend parameters the guide actually accepts."""
        import inspect

        sig = inspect.signature(self.server.mcp_get_control_plane_guide)
        self.assertIn("org", sig.parameters)
        self.assertIn("repo", sig.parameters)
        self.assertIn("remote", sig.parameters)
        remediation = remote_repo_guard.REMEDIATION
        self.assertIn("mcp_get_control_plane_guide", remediation)
        self.assertIn("org=", remediation)
        self.assertIn("repo=", remediation)


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