"""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)) class TestBaseEquivalenceProofRequired(unittest.TestCase): """#757 AC3/AC4: an unknown tip is not evidence of agreement. The original fix compared SHAs only inside ``if remote_tip and local_tip and remote_tip != local_tip``, so a missing local HEAD, an unresolvable live master, or a resolver failure all fell through and *granted* the bootstrap exemption. Base equivalence must be proven, and anything less must fail closed. """ def _permits(self, assessment): return cib.bootstrap_permits_control_checkout( assessment, task="create_issue", workspace_path=CONTROL_CHECKOUT_ROOT, canonical_repo_root=CONTROL_CHECKOUT_ROOT, ) def _assert_fails_closed(self, assessment, *, expect_reason): self.assertTrue(assessment["block"], "assessor must block") self.assertFalse(assessment["allowed"]) self.assertFalse(assessment["proven"]) self.assertFalse(assessment["base_tips_verified"]) self.assertFalse(self._permits(assessment), "predicate must refuse") joined = " ".join(assessment["reasons"]).lower() self.assertIn(expect_reason, joined) def test_missing_local_head_fails_closed(self): self._assert_fails_closed( proven_bootstrap(head=None), expect_reason="control checkout head sha is unknown", ) def test_missing_remote_master_fails_closed(self): self._assert_fails_closed( proven_bootstrap(remote_sha=None), expect_reason="live master tip is unknown", ) def test_both_tips_missing_fails_closed(self): assessment = proven_bootstrap(head=None, remote_sha=None) self._assert_fails_closed( assessment, expect_reason="control checkout head sha is unknown" ) self.assertIn("live master tip is unknown", " ".join(assessment["reasons"])) def test_empty_and_whitespace_tips_fail_closed(self): for head, remote in (("", MASTER_SHA), (MASTER_SHA, ""), (" ", " ")): with self.subTest(head=repr(head), remote=repr(remote)): assessment = proven_bootstrap(head=head, remote_sha=remote) self.assertTrue(assessment["block"]) self.assertFalse(self._permits(assessment)) def test_mismatched_tips_fail_closed(self): self._assert_fails_closed( proven_bootstrap(head=STALE_SHA), expect_reason="does not match", ) def test_resolver_failure_fails_closed(self): assessment = cib.assess_create_issue_bootstrap( workspace_path=CONTROL_CHECKOUT_ROOT, canonical_repo_root=CONTROL_CHECKOUT_ROOT, current_branch="master", head_sha=MASTER_SHA, porcelain_status="", remote_master_sha=None, remote_master_sha_error="TimeoutError: remote unreachable", task="create_issue", ) self._assert_fails_closed( assessment, expect_reason="could not be resolved" ) # The operator must be able to see *why*, not just that it blocked. self.assertIn("remote unreachable", " ".join(assessment["reasons"])) def test_proven_bootstrap_records_both_normalized_shas(self): assessment = proven_bootstrap() self.assertEqual(assessment["local_head_sha"], MASTER_SHA) self.assertEqual(assessment["remote_master_sha"], MASTER_SHA) self.assertTrue(assessment["base_tips_verified"]) self.assertTrue(self._permits(assessment)) def test_tips_are_normalized_before_comparison(self): """Case and surrounding whitespace are not a different commit.""" assessment = proven_bootstrap( head=f" {MASTER_SHA.upper()} ", remote_sha=MASTER_SHA ) self.assertTrue(assessment["allowed"]) self.assertEqual(assessment["local_head_sha"], MASTER_SHA) self.assertTrue(self._permits(assessment)) def test_predicate_rejects_assessment_with_tips_stripped(self): """A recorded proof that is later removed cannot still permit.""" for field in ("local_head_sha", "remote_master_sha"): with self.subTest(field=field): assessment = dict(proven_bootstrap()) assessment[field] = None self.assertFalse(self._permits(assessment)) def test_predicate_rejects_forged_verified_flag(self): """base_tips_verified is re-derived, never trusted on its own.""" assessment = dict(proven_bootstrap()) assessment["local_head_sha"] = MASTER_SHA assessment["remote_master_sha"] = STALE_SHA assessment["base_tips_verified"] = True self.assertFalse(self._permits(assessment)) def test_predicate_rejects_missing_verified_flag(self): assessment = dict(proven_bootstrap()) assessment.pop("base_tips_verified") self.assertFalse(self._permits(assessment)) class TestServerAssessmentFailsClosedOnResolverError(unittest.TestCase): """AC3/AC4 at the single server-derived computation site.""" def setUp(self): # verify_preflight_purity short-circuits under pytest; the production # path only runs with test mode disabled (same setup the #757 e2e uses). srv._preflight_whoami_called = True srv._preflight_capability_called = True srv._preflight_resolved_role = "author" 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): return { "current_branch": "master", "head_sha": MASTER_SHA, "porcelain_status": "", } def test_resolver_exception_produces_blocking_assessment(self): """A raising resolver must not become a silent, permissive None.""" with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT), \ patch.object( srv.issue_lock_worktree, "read_worktree_git_state", return_value=self._git_state(), ), \ patch.object( srv.root_checkout_guard, "resolve_remote_master_sha", side_effect=TimeoutError("remote unreachable"), ): assessment = srv._create_issue_bootstrap_assessment("create_issue") self.assertIsNotNone(assessment) self.assertTrue(assessment["block"]) self.assertFalse(assessment["allowed"]) self.assertFalse(assessment["base_tips_verified"]) self.assertIsNone(assessment["remote_master_sha"]) self.assertIn("could not be resolved", " ".join(assessment["reasons"])) self.assertFalse( cib.bootstrap_permits_control_checkout( assessment, task="create_issue", workspace_path=CONTROL_CHECKOUT_ROOT, canonical_repo_root=CONTROL_CHECKOUT_ROOT, ) ) def test_resolver_exception_blocks_real_preflight(self): """Production path: verify_preflight_purity must fail closed.""" srv._preflight_resolved_task = "create_issue" try: 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", side_effect=TimeoutError("remote unreachable"), ), \ patch.object(srv, "_enforce_root_checkout_guard"): with self.assertRaises(RuntimeError): srv.verify_preflight_purity(remote="prgs", task="create_issue") finally: srv._preflight_resolved_task = None if __name__ == "__main__": unittest.main()