"""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()