Gitea-Tools/tests/test_issue_941_scope_guard_bootstrap_wiring.py

"""Regression: author bootstrap scope reaches workflow_scope_guard (#941).

PR #926 (#892) made ``bootstrap_permits_control_checkout`` accept
``task_scope=author_issue_bootstrap`` and wired that canonical decision into
the #274 branches-only enforcer and the #604 anti-stomp preflight. A third
enforcement path was left unwired.

``workflow_scope_guard.assess_root_source_mutation`` kept its own copy of the
clean-root author decision, gated on ``create_issue_bootstrap.is_create_issue_task``
— a task-name allowlist that never contained ``bootstrap_author_issue_worktree``.
So the real call path

    gitea_bootstrap_author_issue_worktree
      -> verify_preflight_purity
        -> _enforce_issue_scope_guard
          -> workflow_scope_guard.assess_production_mutation_guards

raised ProductionGuardError(missing_issue_worktree) before
``assess_author_issue_bootstrap`` was ever consulted.

These tests drive the real enforcer, not the authorization helper in
isolation. A helper-only test cannot observe this defect: #892's own predicate
tests all passed while the live bootstrap stayed blocked.
"""

from __future__ import annotations

import os
import subprocess
import tempfile
import unittest
from unittest import mock

import author_issue_bootstrap as aib
import create_issue_bootstrap as cib
import workflow_scope_guard

BOOTSTRAP_TASK = "bootstrap_author_issue_worktree"
BOOTSTRAP_TOOL = "gitea_bootstrap_author_issue_worktree"


def _make_control_repo(tmp: str) -> tuple[str, str]:
    """Create a clean control checkout on master and return (path, head)."""
    repo = os.path.join(tmp, "repo")
    os.makedirs(os.path.join(repo, "branches"))
    subprocess.check_call(
        ["git", "init", "-b", "master", repo],
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
    )
    subprocess.check_call(
        [
            "git", "-C", repo,
            "-c", "user.email=t@t", "-c", "user.name=t",
            "commit", "--allow-empty", "-m", "init",
        ],
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
    )
    head = subprocess.check_output(
        ["git", "-C", repo, "rev-parse", "HEAD"], text=True
    ).strip()
    return repo, head


def _assessment(
    repo: str,
    head: str,
    *,
    task: str = BOOTSTRAP_TASK,
    porcelain: str = "",
    remote: str | None = None,
) -> dict:
    return aib.assess_author_issue_bootstrap(
        workspace_path=repo,
        canonical_repo_root=repo,
        current_branch="master",
        head_sha=head,
        porcelain_status=porcelain,
        remote_master_sha=head if remote is None else remote,
        task=task,
    )


class _ControlCheckoutHarness(unittest.TestCase):
    """Drive the real server guard against a temporary clean control checkout."""

    def setUp(self):
        self._tmp = tempfile.TemporaryDirectory()
        self.addCleanup(self._tmp.cleanup)
        self.repo, self.head = _make_control_repo(self._tmp.name)

        # #683 force-on: production guards must execute under pytest.
        patcher = mock.patch.dict(
            os.environ,
            {workflow_scope_guard.FORCE_PRODUCTION_GUARDS_ENV: "1"},
        )
        patcher.start()
        self.addCleanup(patcher.stop)

    def _enforce(
        self,
        task: str,
        *,
        porcelain: str = "",
        assessment: object = "auto",
        role_kind: str = "author",
    ):
        """Call the real _enforce_issue_scope_guard for *task*."""
        import gitea_mcp_server as srv

        if assessment == "auto":
            assessment = _assessment(
                self.repo, self.head, task=task, porcelain=porcelain
            )

        ctx = {
            "workspace_path": self.repo,
            "canonical_repo_root": self.repo,
            "workspace_role_kind": role_kind,
            "workspace_binding_source": "process_project_root",
        }
        git_state = {
            "current_branch": "master",
            "head_sha": self.head,
            "porcelain_status": porcelain,
        }

        with mock.patch.object(
            srv, "_resolve_namespace_mutation_context", return_value=ctx
        ), mock.patch.object(
            srv.issue_lock_worktree,
            "read_worktree_git_state",
            return_value=git_state,
        ), mock.patch.object(
            srv,
            "_session_issue_lock_snapshot",
            return_value={
                "locked_issue_number": None,
                "lock_branch_name": None,
                "worktrees_match": False,
            },
        ), mock.patch.object(
            srv, "_actual_profile_role", return_value=role_kind
        ), mock.patch.object(
            srv, "_effective_workspace_role", return_value=role_kind
        ), mock.patch.object(
            srv, "_create_issue_bootstrap_assessment", return_value=assessment
        ):
            srv._enforce_issue_scope_guard(None, task=task)


class TestRealPathBootstrapReachesGuard(_ControlCheckoutHarness):
    """The defect and its fix, observed through the real enforcer."""

    def test_bootstrap_task_passes_scope_guard_from_clean_control(self):
        # Pre-fix this raises ProductionGuardError(missing_issue_worktree)
        # because the guard consulted a task-name allowlist instead of the
        # canonical authorization decision.
        self._enforce(BOOTSTRAP_TASK)

    def test_bootstrap_tool_alias_passes_scope_guard(self):
        self._enforce(BOOTSTRAP_TOOL)

    def test_guard_consults_canonical_predicate(self):
        """The guard must reach bootstrap_permits_control_checkout, not a name list."""
        real = cib.bootstrap_permits_control_checkout
        seen: list[str | None] = []

        def _spy(assessment, *, task, workspace_path, canonical_repo_root):
            seen.append(task)
            return real(
                assessment,
                task=task,
                workspace_path=workspace_path,
                canonical_repo_root=canonical_repo_root,
            )

        with mock.patch.object(
            cib, "bootstrap_permits_control_checkout", side_effect=_spy
        ):
            self._enforce(BOOTSTRAP_TASK)

        self.assertIn(
            BOOTSTRAP_TASK,
            seen,
            "workflow_scope_guard did not consult the canonical bootstrap "
            "authorization decision",
        )


class TestFailClosedOnBadEvidence(_ControlCheckoutHarness):
    """Missing, malformed, or mismatched scope evidence must still block."""

    def _assert_blocked(self, **kwargs):
        with self.assertRaises(workflow_scope_guard.ProductionGuardError):
            self._enforce(BOOTSTRAP_TASK, **kwargs)

    def test_missing_assessment_fails_closed(self):
        self._assert_blocked(assessment=None)

    def test_malformed_assessment_fails_closed(self):
        self._assert_blocked(assessment={"allowed": True})

    def test_non_dict_assessment_fails_closed(self):
        self._assert_blocked(assessment="allowed")

    def test_wrong_task_scope_fails_closed(self):
        bad = dict(_assessment(self.repo, self.head))
        bad["task_scope"] = "create_issue_only"
        self._assert_blocked(assessment=bad)

    def test_nonempty_reasons_fail_closed(self):
        bad = dict(_assessment(self.repo, self.head), reasons=["note"])
        self._assert_blocked(assessment=bad)

    def test_mismatched_base_tips_fail_closed(self):
        bad = dict(_assessment(self.repo, self.head))
        bad["remote_master_sha"] = "b" * 40
        self._assert_blocked(assessment=bad)

    def test_unverified_base_tips_fail_closed(self):
        bad = dict(_assessment(self.repo, self.head), base_tips_verified=False)
        self._assert_blocked(assessment=bad)

    def test_mismatched_workspace_binding_fails_closed(self):
        bad = dict(_assessment(self.repo, self.head))
        bad["workspace_path"] = os.path.join(self.repo, "elsewhere")
        self._assert_blocked(assessment=bad)

    def test_mismatched_repo_root_binding_fails_closed(self):
        bad = dict(_assessment(self.repo, self.head))
        bad["canonical_repo_root"] = os.path.join(self.repo, "other-root")
        self._assert_blocked(assessment=bad)

    def test_blocked_assessment_fails_closed(self):
        bad = dict(_assessment(self.repo, self.head), block=True, allowed=False)
        self._assert_blocked(assessment=bad)


class TestOrdinaryControlCheckoutMutationStillForbidden(_ControlCheckoutHarness):
    """The waiver must not leak to ordinary author work."""

    def test_ordinary_author_task_still_blocked(self):
        with self.assertRaises(workflow_scope_guard.ProductionGuardError):
            self._enforce("commit_files", assessment=None)

    def test_lock_issue_still_blocked_from_control(self):
        with self.assertRaises(workflow_scope_guard.ProductionGuardError):
            self._enforce("lock_issue", assessment=None)

    def test_bootstrap_assessment_cannot_license_other_task(self):
        # Cross-task smuggling: valid bootstrap evidence must not waive a
        # different author mutation.
        good = _assessment(self.repo, self.head)
        with self.assertRaises(workflow_scope_guard.ProductionGuardError):
            self._enforce("commit_files", assessment=good)

    def test_dirty_control_checkout_still_blocked_for_bootstrap(self):
        with self.assertRaises(workflow_scope_guard.ProductionGuardError):
            self._enforce(BOOTSTRAP_TASK, porcelain=" M gitea_mcp_server.py\n")


class TestCreateIssueBehaviorUnchanged(_ControlCheckoutHarness):
    """#749 create_issue keeps its own sanctioned path."""

    def test_create_issue_still_allowed_from_clean_control(self):
        self._enforce("create_issue", assessment=None)

    def test_create_issue_tool_alias_still_allowed(self):
        self._enforce("gitea_create_issue", assessment=None)

    def test_create_issue_blocked_when_control_dirty(self):
        with self.assertRaises(workflow_scope_guard.ProductionGuardError):
            self._enforce(
                "create_issue",
                porcelain=" M gitea_mcp_server.py\n",
                assessment=None,
            )


class TestGuardUnitLevelWiring(unittest.TestCase):
    """assess_root_source_mutation itself must accept and honour the evidence."""

    def setUp(self):
        self._tmp = tempfile.TemporaryDirectory()
        self.addCleanup(self._tmp.cleanup)
        self.repo, self.head = _make_control_repo(self._tmp.name)
        patcher = mock.patch.dict(
            os.environ,
            {workflow_scope_guard.FORCE_PRODUCTION_GUARDS_ENV: "1"},
        )
        patcher.start()
        self.addCleanup(patcher.stop)

    def _assess(self, *, task=BOOTSTRAP_TASK, bootstrap_assessment="auto"):
        if bootstrap_assessment == "auto":
            bootstrap_assessment = _assessment(self.repo, self.head, task=task)
        return workflow_scope_guard.assess_root_source_mutation(
            workspace_path=self.repo,
            canonical_repo_root=self.repo,
            porcelain_status="",
            current_branch="master",
            role_kind="author",
            mutation_task=task,
            bootstrap_assessment=bootstrap_assessment,
        )

    def test_valid_evidence_unblocks(self):
        result = self._assess()
        self.assertFalse(result["block"])
        self.assertIsNone(result["blocker_kind"])

    def test_absent_evidence_blocks(self):
        result = self._assess(bootstrap_assessment=None)
        self.assertTrue(result["block"])
        self.assertEqual(
            result["blocker_kind"], workflow_scope_guard.BLOCKER_MISSING_WORKTREE
        )

    def test_reconciler_exemption_preserved(self):
        result = workflow_scope_guard.assess_root_source_mutation(
            workspace_path=self.repo,
            canonical_repo_root=self.repo,
            porcelain_status="",
            current_branch="master",
            role_kind="reconciler",
            mutation_task=BOOTSTRAP_TASK,
        )
        self.assertFalse(result["block"])

    def test_signature_accepts_evidence_without_it_being_required(self):
        # Callers that supply no evidence keep the pre-existing behaviour.
        result = workflow_scope_guard.assess_root_source_mutation(
            workspace_path=self.repo,
            canonical_repo_root=self.repo,
            porcelain_status="",
            current_branch="master",
            role_kind="author",
            mutation_task="create_issue",
        )
        self.assertFalse(result["block"])


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