Gitea-Tools/tests/test_preflight_workspace_repo_forwarding.py

"""Regression matrix: session-bound mutation preflight resolves the workspace repo.

Root-cause follow-up to #733. ``_resolve`` (the actual API-targeting resolver)
already prefers the workspace-aligned git remote over the ``REMOTES`` default
(bare ``prgs`` → ``Scaled-Tech-Consulting/Timesheet``). The shared #604
anti-stomp preflight (``_run_anti_stomp_preflight``) did **not** mirror that: for
every mutation task whose call site omitted ``org``/``repo`` it filled the
remote-wide ``REMOTES`` default, so the #530 repo guard false-positived
``wrong_repo`` against a ``Gitea-Tools`` checkout and blocked native
issue/PR/review/merge/lock/cleanup mutations.

The fix makes ``_run_anti_stomp_preflight`` prefer the trusted, git-remote-
derived workspace identity for omitted coordinates on **session-bound** mutation
tasks (marking those filled sides explicit, exactly like ``_resolve``), while
``delete_branch`` keeps its explicit cross-repository *target* contract (#733)
and continues to fail closed on omitted coordinates.
"""

import os
import sys
import unittest
from unittest.mock import patch

sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))

import mcp_server
import remote_repo_guard

# Workspace is bound to Gitea-Tools even though the prgs REMOTES default repo is
# Timesheet (the exact cross-repo scenario).
LOCAL_GITEA_TOOLS_URL = "https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git"

RECONCILER = {
    "profile_name": "prgs-reconciler",
    "role": "reconciler",
    "allowed_operations": [
        "gitea.read", "gitea.issue.comment", "gitea.pr.comment",
        "gitea.pr.close", "gitea.branch.delete",
    ],
    "forbidden_operations": ["gitea.pr.create", "gitea.pr.merge"],
    "audit_label": "prgs-reconciler",
}

# Session-bound mutation tasks (every MUTATION_TASK except the explicit-target
# delete_branch) must auto-resolve the workspace repository for omitted coords.
SESSION_BOUND_TASKS = [
    "create_issue",
    "comment_issue",
    "close_issue",
    "mark_issue",
    "lock_issue",
    "set_issue_labels",
    "create_label",
    "create_pr",
    "close_pr",
    "edit_pr",
    "commit_files",
    "cleanup_merged_pr_branch",
    "cleanup_stale_claims",
    "reconcile_merged_cleanups",
    "reconcile_already_landed_pr",
    "reconcile_close_superseded_pr",
    "post_heartbeat",
    "review_pr",
    "submit_pr_review",
    "merge_pr",
]


class _AntiStompResolutionBase(unittest.TestCase):
    def setUp(self):
        self._remotes = patch.dict(mcp_server.REMOTES, {
            "prgs": {
                "host": "gitea.prgs.cc",
                "org": "Scaled-Tech-Consulting",
                "repo": "Timesheet",
            },
        })
        self._remotes.start()

    def tearDown(self):
        patch.stopall()

    def _capture_resolution(self, task, org, repo, *, local_url=LOCAL_GITEA_TOOLS_URL):
        """Drive the live anti-stomp runner and capture the org/repo it resolved
        and passed to the pure assessor."""
        passthrough = {
            "allowed": True, "block": False, "blockers": [], "reasons": [],
            "exact_next_action": "proceed", "blocker_kind": None, "checks": {},
        }
        with patch.dict(
            os.environ, {"GITEA_TEST_FORCE_ANTI_STOMP": "1"}, clear=False
        ), patch(
            "mcp_server._local_git_remote_url", return_value=local_url
        ), patch(
            "mcp_server.get_profile", return_value=RECONCILER
        ), patch.object(
            mcp_server.anti_stomp_preflight,
            "assess_anti_stomp_preflight",
            return_value=passthrough,
        ) as m:
            mcp_server._run_anti_stomp_preflight(
                task, remote="prgs", org=org, repo=repo
            )
        self.assertTrue(m.called)
        return m.call_args.kwargs


class TestSessionBoundResolution(_AntiStompResolutionBase):
    def test_omitted_coords_resolve_workspace_repo_and_mark_explicit(self):
        for task in SESSION_BOUND_TASKS:
            with self.subTest(task=task):
                kw = self._capture_resolution(task, None, None)
                self.assertEqual(
                    kw["resolved_org"], "Scaled-Tech-Consulting", task
                )
                self.assertEqual(kw["resolved_repo"], "Gitea-Tools", task)
                # Workspace-filled sides are intentional alignment → explicit.
                self.assertTrue(kw["org_explicit"], task)
                self.assertTrue(kw["repo_explicit"], task)

    def test_resolved_workspace_repo_passes_the_530_guard(self):
        kw = self._capture_resolution("create_pr", None, None)
        assessment = remote_repo_guard.assess_remote_repo_match(
            remote="prgs",
            resolved_org=kw["resolved_org"],
            resolved_repo=kw["resolved_repo"],
            local_remote_url=LOCAL_GITEA_TOOLS_URL,
            org_explicit=kw["org_explicit"],
            repo_explicit=kw["repo_explicit"],
        )
        self.assertFalse(assessment["block"], assessment)

    def test_explicit_caller_coords_are_preserved(self):
        kw = self._capture_resolution(
            "create_pr", "Scaled-Tech-Consulting", "Gitea-Tools"
        )
        self.assertEqual(kw["resolved_org"], "Scaled-Tech-Consulting")
        self.assertEqual(kw["resolved_repo"], "Gitea-Tools")
        self.assertTrue(kw["org_explicit"])
        self.assertTrue(kw["repo_explicit"])

    def test_no_workspace_identity_falls_back_to_remotes_default(self):
        # No derivable local remote → best-effort REMOTES default preserved,
        # not marked explicit (so the guard still corroborates when able).
        kw = self._capture_resolution("create_pr", None, None, local_url=None)
        self.assertEqual(kw["resolved_repo"], "Timesheet")
        self.assertFalse(kw["org_explicit"])
        self.assertFalse(kw["repo_explicit"])


class TestDeleteBranchContractPreserved(_AntiStompResolutionBase):
    """delete_branch keeps its explicit-target contract (#733): omitted coords
    still resolve the remote-wide default and are NOT marked explicit, so the
    guard fails closed against a mismatched workspace."""

    def test_omitted_coords_still_remote_default_not_explicit(self):
        kw = self._capture_resolution("delete_branch", None, None)
        self.assertEqual(kw["resolved_repo"], "Timesheet")
        self.assertFalse(kw["org_explicit"])
        self.assertFalse(kw["repo_explicit"])
        assessment = remote_repo_guard.assess_remote_repo_match(
            remote="prgs",
            resolved_org=kw["resolved_org"],
            resolved_repo=kw["resolved_repo"],
            local_remote_url=LOCAL_GITEA_TOOLS_URL,
            org_explicit=kw["org_explicit"],
            repo_explicit=kw["repo_explicit"],
        )
        self.assertTrue(assessment["block"])

    def test_explicit_coords_still_forwarded(self):
        kw = self._capture_resolution(
            "delete_branch", "Scaled-Tech-Consulting", "Gitea-Tools"
        )
        self.assertEqual(kw["resolved_repo"], "Gitea-Tools")
        self.assertTrue(kw["org_explicit"])
        self.assertTrue(kw["repo_explicit"])


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