Gitea-Tools/tests/test_issue_757_bootstrap_guard_agreement.py

"""Regression tests for #757: #274 and #604 must not disagree.

The sanctioned ``create_issue`` bootstrap (#749/#750) was unreachable in
production: the #274 branches-only guard consulted the bootstrap and permitted
a clean canonical control checkout, then the bootstrap-blind #604 anti-stomp
preflight rejected the same checkout as ``wrong_worktree``.

These tests pin the fix:

* one server-derived assessment, interpreted by one shared predicate;
* the waiver is narrow (only the wrong-worktree verdict, only create_issue,
  only from the exact clean canonical control checkout);
* every other guard and rejection reason keeps its fail-closed behaviour;
* eligibility cannot be forged through a public tool signature.
"""

from __future__ import annotations

import inspect
import sys
import unittest
from pathlib import Path
from unittest.mock import patch

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

import anti_stomp_preflight as asp  # noqa: E402
import create_issue_bootstrap as cib  # noqa: E402
import gitea_mcp_server as srv  # noqa: E402

FAKE_AUTH = {"Authorization": "token test-token"}
MASTER_SHA = "a" * 40
STALE_SHA = "b" * 40

current_file_path = Path(__file__).resolve()
if "branches" in current_file_path.parts:
    CONTROL_CHECKOUT_ROOT = str(current_file_path.parents[3])
else:
    CONTROL_CHECKOUT_ROOT = str(current_file_path.parents[1])

BRANCHES_WORKTREE = str(Path(CONTROL_CHECKOUT_ROOT) / "branches" / "issue-1-x")


def proven_bootstrap(
    *,
    workspace=CONTROL_CHECKOUT_ROOT,
    root=CONTROL_CHECKOUT_ROOT,
    task="create_issue",
    branch="master",
    porcelain="",
    head=MASTER_SHA,
    remote_sha=MASTER_SHA,
):
    """Build a real assessment via the production assessor (never hand-rolled)."""
    return cib.assess_create_issue_bootstrap(
        workspace_path=workspace,
        canonical_repo_root=root,
        current_branch=branch,
        head_sha=head,
        porcelain_status=porcelain,
        remote_master_sha=remote_sha,
        task=task,
    )


class TestSharedPredicate(unittest.TestCase):
    """The one interpretation both guards consume (AC6, AC7)."""

    def _permits(self, assessment, task="create_issue", workspace=None, root=None):
        return cib.bootstrap_permits_control_checkout(
            assessment,
            task=task,
            workspace_path=workspace or CONTROL_CHECKOUT_ROOT,
            canonical_repo_root=root or CONTROL_CHECKOUT_ROOT,
        )

    def test_proven_bootstrap_permits(self):
        self.assertTrue(self._permits(proven_bootstrap()))

    def test_tool_alias_permits(self):
        assessment = proven_bootstrap(task="gitea_create_issue")
        self.assertTrue(self._permits(assessment, task="gitea_create_issue"))

    def test_missing_assessment_fails_closed(self):
        self.assertFalse(self._permits(None))

    def test_malformed_assessment_fails_closed(self):
        for bogus in ("allowed", 1, True, [], ["allowed"], object()):
            with self.subTest(bogus=bogus):
                self.assertFalse(self._permits(bogus))

    def test_refused_assessment_fails_closed(self):
        # Dirty control checkout -> assessor blocks -> predicate must refuse.
        self.assertFalse(
            self._permits(proven_bootstrap(porcelain=" M gitea_mcp_server.py\n"))
        )

    def test_not_applicable_assessment_fails_closed(self):
        # branches/ worktree -> not_applicable -> no waiver.
        self.assertFalse(
            self._permits(proven_bootstrap(workspace=BRANCHES_WORKTREE))
        )

    def test_incomplete_assessment_fails_closed(self):
        incomplete = dict(proven_bootstrap())
        del incomplete["bootstrap_path"]
        self.assertFalse(self._permits(incomplete))

    def test_contradictory_block_and_allow_fails_closed(self):
        contradictory = dict(proven_bootstrap(), block=True)
        self.assertFalse(self._permits(contradictory))

    def test_contradictory_dirty_but_allowed_fails_closed(self):
        contradictory = dict(proven_bootstrap(), dirty_files=["gitea_mcp_server.py"])
        self.assertFalse(self._permits(contradictory))

    def test_contradictory_under_branches_but_allowed_fails_closed(self):
        contradictory = dict(proven_bootstrap(), under_branches=True)
        self.assertFalse(self._permits(contradictory))

    def test_contradictory_reasons_present_fails_closed(self):
        contradictory = dict(proven_bootstrap(), reasons=["something refused"])
        self.assertFalse(self._permits(contradictory))

    def test_truthy_non_true_values_fail_closed(self):
        """Strict identity: no truthy smuggling (1, 'yes') can assert eligibility."""
        for value in (1, "yes", "true", [1]):
            with self.subTest(value=value):
                self.assertFalse(self._permits(dict(proven_bootstrap(), allowed=value)))

    def test_wrong_task_fails_closed(self):
        """An otherwise-proven assessment cannot license a different task."""
        self.assertFalse(self._permits(proven_bootstrap(), task="lock_issue"))
        self.assertFalse(self._permits(proven_bootstrap(), task="create_pr"))
        self.assertFalse(self._permits(proven_bootstrap(), task=None))

    def test_foreign_workspace_binding_fails_closed(self):
        """Assessment must describe the workspace actually being guarded."""
        other = proven_bootstrap(workspace="/other/clone", root="/other/clone")
        self.assertFalse(self._permits(other))

    def test_scope_and_path_tampering_fails_closed(self):
        self.assertFalse(
            self._permits(dict(proven_bootstrap(), task_scope="all_tasks"))
        )
        self.assertFalse(
            self._permits(dict(proven_bootstrap(), bootstrap_path="anything_goes"))
        )


class TestAntiStompHonorsBootstrap(unittest.TestCase):
    """#604 consumes the same decision, and waives only wrong_worktree."""

    def _assess(self, *, task="create_issue", bootstrap=None, workspace=None, **kw):
        params = dict(
            task=task,
            profile_role="author",
            required_role="author",
            workspace_path=workspace or CONTROL_CHECKOUT_ROOT,
            project_root=CONTROL_CHECKOUT_ROOT,
            current_branch="master",
            root_head_sha=MASTER_SHA,
            root_porcelain="",
            remote_master_sha=MASTER_SHA,
            check_repo=False,
            create_issue_bootstrap_assessment=bootstrap,
        )
        params.update(kw)
        return asp.assess_anti_stomp_preflight(**params)

    def _blocker_kinds(self, result):
        return {b["kind"] for b in result.get("blockers") or []}

    def test_control_checkout_without_bootstrap_still_blocked(self):
        """Baseline: the exact production failure, unwaived."""
        res = self._assess(bootstrap=None)
        self.assertIn(asp.BLOCKER_WRONG_WORKTREE, self._blocker_kinds(res))

    def test_control_checkout_with_proven_bootstrap_permitted(self):
        """AC1/AC2: the #757 fix — same inputs, bootstrap honoured."""
        res = self._assess(bootstrap=proven_bootstrap())
        self.assertNotIn(asp.BLOCKER_WRONG_WORKTREE, self._blocker_kinds(res))
        self.assertTrue(res["checks"]["worktree"]["create_issue_bootstrap_waived"])
        self.assertFalse(res["checks"]["worktree"]["block"])

    def test_tool_alias_permitted(self):
        res = self._assess(
            task="gitea_create_issue",
            bootstrap=proven_bootstrap(task="gitea_create_issue"),
        )
        self.assertNotIn(asp.BLOCKER_WRONG_WORKTREE, self._blocker_kinds(res))

    def test_non_create_issue_task_never_waived(self):
        """AC5: other author mutations keep the branches-only requirement."""
        for task in ("lock_issue", "create_pr", "commit_files", "mark_issue"):
            with self.subTest(task=task):
                res = self._assess(task=task, bootstrap=proven_bootstrap())
                self.assertIn(asp.BLOCKER_WRONG_WORKTREE, self._blocker_kinds(res))

    def test_dirty_control_checkout_still_blocked(self):
        """AC4: dirty root refuses the bootstrap, so no waiver."""
        dirty = " M gitea_mcp_server.py\n"
        res = self._assess(
            bootstrap=proven_bootstrap(porcelain=dirty), root_porcelain=dirty
        )
        self.assertIn(asp.BLOCKER_WRONG_WORKTREE, self._blocker_kinds(res))

    def test_detached_control_checkout_still_blocked(self):
        res = self._assess(
            bootstrap=proven_bootstrap(branch=""), current_branch=""
        )
        self.assertIn(asp.BLOCKER_WRONG_WORKTREE, self._blocker_kinds(res))

    def test_base_divergence_still_blocked(self):
        res = self._assess(
            bootstrap=proven_bootstrap(head=STALE_SHA), root_head_sha=STALE_SHA
        )
        self.assertIn(asp.BLOCKER_WRONG_WORKTREE, self._blocker_kinds(res))

    def test_waiver_does_not_suppress_stale_runtime(self):
        """AC: waive only wrong_worktree — stale runtime still fails closed."""
        res = self._assess(
            bootstrap=proven_bootstrap(),
            startup_head=STALE_SHA,
            current_code_head=MASTER_SHA,
        )
        self.assertNotIn(asp.BLOCKER_WRONG_WORKTREE, self._blocker_kinds(res))
        self.assertIn(asp.BLOCKER_STALE_RUNTIME, self._blocker_kinds(res))
        self.assertTrue(res["block"])

    def test_waiver_does_not_suppress_wrong_repo(self):
        res = self._assess(
            bootstrap=proven_bootstrap(),
            check_repo=True,
            remote="prgs",
            resolved_org="Scaled-Tech-Consulting",
            resolved_repo="Timesheet",
            # Not both-explicit, so the #530 repo guard actually evaluates the
            # mismatch against the local remote instead of trusting the caller.
            org_explicit=False,
            repo_explicit=False,
            local_remote_url=(
                "https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git"
            ),
        )
        self.assertNotIn(asp.BLOCKER_WRONG_WORKTREE, self._blocker_kinds(res))
        self.assertIn(asp.BLOCKER_WRONG_REPO, self._blocker_kinds(res))

    def test_waiver_does_not_suppress_wrong_role(self):
        res = self._assess(
            bootstrap=proven_bootstrap(),
            profile_role="reviewer",
            required_role="author",
        )
        self.assertIn(asp.BLOCKER_WRONG_ROLE, self._blocker_kinds(res))

    def test_branches_worktree_unaffected(self):
        """AC: ordinary branches/ worktrees keep working, waived or not."""
        for bootstrap in (None, proven_bootstrap(workspace=BRANCHES_WORKTREE)):
            with self.subTest(bootstrap=bool(bootstrap)):
                res = self._assess(
                    workspace=BRANCHES_WORKTREE,
                    bootstrap=bootstrap,
                    current_branch="fix/issue-1-x",
                )
                self.assertNotIn(
                    asp.BLOCKER_WRONG_WORKTREE, self._blocker_kinds(res)
                )

    def test_forged_assessment_rejected(self):
        """AC6: a hand-built 'allowed' dict cannot unlock the waiver."""
        forged = {"allowed": True, "block": False}
        res = self._assess(bootstrap=forged)
        self.assertIn(asp.BLOCKER_WRONG_WORKTREE, self._blocker_kinds(res))


class TestGuardAgreement(unittest.TestCase):
    """AC7: the regression that would have caught the #757 defect.

    For identical evidence, the #274 guard and the #604 guard must return the
    same wrong-worktree verdict across the full workspace-state matrix.
    """

    MATRIX = [
        (
            "clean control + create_issue",
            CONTROL_CHECKOUT_ROOT, "create_issue", "master", "", MASTER_SHA,
        ),
        (
            "clean control + alias",
            CONTROL_CHECKOUT_ROOT, "gitea_create_issue", "master", "", MASTER_SHA,
        ),
        (
            "clean control + lock_issue",
            CONTROL_CHECKOUT_ROOT, "lock_issue", "master", "", MASTER_SHA,
        ),
        (
            "clean control + create_pr",
            CONTROL_CHECKOUT_ROOT, "create_pr", "master", "", MASTER_SHA,
        ),
        (
            "dirty control + create_issue",
            CONTROL_CHECKOUT_ROOT, "create_issue", "master",
            " M gitea_mcp_server.py\n", MASTER_SHA,
        ),
        (
            "detached control + create_issue",
            CONTROL_CHECKOUT_ROOT, "create_issue", "", "", MASTER_SHA,
        ),
        (
            "non-base control + create_issue",
            CONTROL_CHECKOUT_ROOT, "create_issue", "feat/x", "", MASTER_SHA,
        ),
        (
            "diverged control + create_issue",
            CONTROL_CHECKOUT_ROOT, "create_issue", "master", "", STALE_SHA,
        ),
        (
            "branches wt + create_issue",
            BRANCHES_WORKTREE, "create_issue", "fix/issue-1-x", "", MASTER_SHA,
        ),
        (
            "branches wt + lock_issue",
            BRANCHES_WORKTREE, "lock_issue", "fix/issue-1-x", "", MASTER_SHA,
        ),
    ]

    def _guard_274_blocks(self, *, workspace, task, branch, porcelain, head, bootstrap):
        git_state = {
            "current_branch": branch,
            "head_sha": head,
            "porcelain_status": porcelain,
        }
        ctx = {
            "workspace_path": workspace,
            "canonical_repo_root": CONTROL_CHECKOUT_ROOT,
        }
        with patch.object(srv, "_effective_workspace_role", return_value="author"), \
             patch.object(srv, "_actual_profile_role", return_value="author"), \
             patch.object(
                 srv, "_resolve_namespace_mutation_context", return_value=ctx
             ), \
             patch.object(
                 srv.issue_lock_worktree,
                 "read_worktree_git_state",
                 return_value=git_state,
             ):
            try:
                srv._enforce_branches_only_author_mutation(
                    workspace, task=task, bootstrap_assessment=bootstrap
                )
                return False
            except RuntimeError:
                return True

    def _guard_604_blocks(self, *, workspace, task, branch, porcelain, head, bootstrap):
        res = asp.assess_anti_stomp_preflight(
            task=task,
            profile_role="author",
            required_role="author",
            workspace_path=workspace,
            project_root=CONTROL_CHECKOUT_ROOT,
            current_branch=branch,
            root_head_sha=head,
            root_porcelain=porcelain,
            remote_master_sha=MASTER_SHA,
            check_repo=False,
            create_issue_bootstrap_assessment=bootstrap,
        )
        return asp.BLOCKER_WRONG_WORKTREE in {
            b["kind"] for b in res.get("blockers") or []
        }

    def test_guards_agree_across_matrix(self):
        for label, workspace, task, branch, porcelain, head in self.MATRIX:
            with self.subTest(case=label):
                # ONE server-derived assessment, exactly as production computes it.
                bootstrap = cib.assess_create_issue_bootstrap(
                    workspace_path=workspace,
                    canonical_repo_root=CONTROL_CHECKOUT_ROOT,
                    current_branch=branch,
                    head_sha=head,
                    porcelain_status=porcelain,
                    remote_master_sha=MASTER_SHA,
                    task=task,
                )
                kwargs = dict(
                    workspace=workspace,
                    task=task,
                    branch=branch,
                    porcelain=porcelain,
                    head=head,
                    bootstrap=bootstrap,
                )
                blocked_274 = self._guard_274_blocks(**kwargs)
                blocked_604 = self._guard_604_blocks(**kwargs)
                self.assertEqual(
                    blocked_274,
                    blocked_604,
                    f"{label}: #274 blocked={blocked_274} "
                    f"but #604 blocked={blocked_604}",
                )

    def test_clean_control_create_issue_permitted_by_both(self):
        """The specific case that was broken: both guards must permit."""
        kwargs = dict(
            workspace=CONTROL_CHECKOUT_ROOT,
            task="create_issue",
            branch="master",
            porcelain="",
            head=MASTER_SHA,
            bootstrap=proven_bootstrap(),
        )
        self.assertFalse(self._guard_274_blocks(**kwargs))
        self.assertFalse(self._guard_604_blocks(**kwargs))


class TestNoCallerForgeableEligibility(unittest.TestCase):
    """AC6: eligibility is never reachable through a public tool signature."""

    def test_create_issue_tool_exposes_no_bootstrap_argument(self):
        params = set(inspect.signature(srv.gitea_create_issue).parameters)
        for forbidden in (
            "bootstrap",
            "bootstrap_assessment",
            "create_issue_bootstrap",
            "create_issue_bootstrap_assessment",
            "allow_control_checkout",
            "bootstrap_allowed",
        ):
            self.assertNotIn(forbidden, params)

    def test_no_mcp_tool_exposes_bootstrap_argument(self):
        """No public gitea_* tool may take bootstrap evidence from the caller."""
        for name in dir(srv):
            if not name.startswith("gitea_"):
                continue
            fn = getattr(srv, name)
            if not callable(fn):
                continue
            try:
                params = set(inspect.signature(fn).parameters)
            except (TypeError, ValueError):
                continue
            leaked = {p for p in params if "bootstrap" in p.lower()}
            self.assertFalse(leaked, f"{name} exposes bootstrap args: {leaked}")

    def test_internal_helper_yields_nothing_for_other_tasks(self):
        self.assertTrue(hasattr(srv, "_create_issue_bootstrap_assessment"))
        self.assertIsNone(
            srv._create_issue_bootstrap_assessment("lock_issue"),
            "non-create_issue tasks must yield no bootstrap evidence",
        )


class TestNativeCreateIssueEndToEnd(unittest.TestCase):
    """AC8: the production handler, with the #604 gate LIVE (not patched out).

    The pre-existing #749 e2e test patched ``_run_anti_stomp_preflight`` to a
    no-op, which is exactly why this defect reached production. These tests
    leave it running.
    """

    def setUp(self):
        srv._preflight_whoami_called = True
        srv._preflight_capability_called = True
        srv._preflight_resolved_role = "author"
        srv._preflight_resolved_task = "create_issue"
        srv._preflight_whoami_violation = False
        srv._preflight_capability_violation = False
        self._orig_in_test = srv._preflight_in_test_mode
        srv._preflight_in_test_mode = lambda: False

    def tearDown(self):
        srv._preflight_in_test_mode = self._orig_in_test
        srv._preflight_resolved_task = None

    def _git_state(self, branch="master", head=MASTER_SHA, porcelain=""):
        return {
            "current_branch": branch,
            "head_sha": head,
            "porcelain_status": porcelain,
        }

    def _run_create_issue(
        self,
        *,
        git_state,
        remote_sha=MASTER_SHA,
        parity=None,
        title="Bootstrap issue from clean control",
    ):
        """Invoke the native handler with the anti-stomp gate live."""
        parity = parity or {
            "startup_head": MASTER_SHA,
            "current_head": MASTER_SHA,
        }
        with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT), \
             patch.object(srv, "_auth", return_value=FAKE_AUTH), \
             patch.object(srv, "_profile_permission_block", return_value=None), \
             patch.object(srv, "_namespace_mutation_block", return_value=None), \
             patch.object(
                 srv.role_session_router,
                 "check_author_mutation_after_reviewer_stop",
                 return_value=(True, []),
             ), \
             patch.object(srv, "api_get_all", return_value=[]), \
             patch.object(srv, "api_request") as mock_api, \
             patch.object(
                 srv.root_checkout_guard,
                 "resolve_remote_master_sha",
                 return_value=remote_sha,
             ), \
             patch.object(srv, "_current_master_parity", return_value=parity), \
             patch.object(
                 srv,
                 "get_profile",
                 return_value={
                     "profile_name": "prgs-author",
                     "allowed_operations": [
                         "gitea.issue.create",
                         "gitea.issue.comment",
                         "gitea.pr.create",
                         "gitea.read",
                     ],
                 },
             ), \
             patch.object(srv, "_actual_profile_role", return_value="author"), \
             patch.object(srv, "_effective_workspace_role", return_value="author"), \
             patch.object(
                 srv.issue_lock_worktree,
                 "read_worktree_git_state",
                 return_value=git_state,
             ), \
             patch.object(
                 srv,
                 "_get_workspace_porcelain",
                 return_value=git_state["porcelain_status"],
             ), \
             patch.object(srv, "_enforce_root_checkout_guard"):
            mock_api.return_value = {
                "number": 99,
                "html_url": "https://gitea.example.com/issues/99",
            }
            try:
                result = srv.gitea_create_issue(
                    title=title,
                    body="Body text for content gate.",
                )
            except RuntimeError as exc:
                return {"raised": str(exc)}, mock_api
            return result, mock_api

    def test_clean_control_checkout_reaches_api(self):
        """The exact production failure that blocked filing #757 and #758."""
        res, mock_api = self._run_create_issue(git_state=self._git_state())
        self.assertNotIn("raised", res, f"guard still blocks: {res.get('raised')}")
        self.assertEqual(res.get("number"), 99)
        mock_api.assert_called_once()

    def test_dirty_control_checkout_blocked(self):
        dirty = " M gitea_mcp_server.py\n"
        res, mock_api = self._run_create_issue(
            git_state=self._git_state(porcelain=dirty)
        )
        self.assertFalse(isinstance(res, dict) and res.get("number") == 99)
        mock_api.assert_not_called()

    def test_detached_control_checkout_blocked(self):
        res, mock_api = self._run_create_issue(git_state=self._git_state(branch=""))
        self.assertFalse(isinstance(res, dict) and res.get("number") == 99)
        mock_api.assert_not_called()

    def test_base_mismatch_blocked(self):
        res, mock_api = self._run_create_issue(
            git_state=self._git_state(head=STALE_SHA)
        )
        self.assertFalse(isinstance(res, dict) and res.get("number") == 99)
        mock_api.assert_not_called()

    def test_stale_runtime_blocked(self):
        res, mock_api = self._run_create_issue(
            git_state=self._git_state(),
            parity={"startup_head": STALE_SHA, "current_head": MASTER_SHA},
        )
        self.assertFalse(isinstance(res, dict) and res.get("number") == 99)
        mock_api.assert_not_called()

    def test_non_create_issue_mutation_still_blocked_from_control(self):
        """AC5 through the real preflight, with anti-stomp live."""
        srv._preflight_resolved_task = "lock_issue"
        with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT), \
             patch.object(srv, "_actual_profile_role", return_value="author"), \
             patch.object(srv, "_effective_workspace_role", return_value="author"), \
             patch.object(
                 srv.issue_lock_worktree,
                 "read_worktree_git_state",
                 return_value=self._git_state(),
             ), \
             patch.object(srv, "_get_workspace_porcelain", return_value=""), \
             patch.object(
                 srv.root_checkout_guard,
                 "resolve_remote_master_sha",
                 return_value=MASTER_SHA,
             ), \
             patch.object(srv, "_enforce_root_checkout_guard"):
            with self.assertRaises(RuntimeError) as ctx:
                srv.verify_preflight_purity(remote="prgs", task="lock_issue")
        self.assertIn("control checkout", str(ctx.exception))


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