Compare commits
3
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
9d2c652ae8 | ||
|
|
adc61255b2 | ||
|
|
bde5c5fb20 |
+20
-2
@@ -33,6 +33,7 @@ from __future__ import annotations
|
||||
from typing import Any
|
||||
|
||||
import author_mutation_worktree
|
||||
import create_issue_bootstrap
|
||||
import master_parity_gate
|
||||
import remote_repo_guard
|
||||
import root_checkout_guard
|
||||
@@ -354,6 +355,7 @@ def assess_anti_stomp_preflight(
|
||||
remote_master_sha: str | None = None,
|
||||
check_root_checkout: bool = True,
|
||||
check_worktree: bool = True,
|
||||
create_issue_bootstrap_assessment: dict[str, Any] | None = None,
|
||||
# stale runtime (master parity)
|
||||
startup_head: str | None = None,
|
||||
current_code_head: str | None = None,
|
||||
@@ -584,12 +586,28 @@ def assess_anti_stomp_preflight(
|
||||
project_root=project_root,
|
||||
current_branch=current_branch,
|
||||
)
|
||||
# #757: the #274 guard consults the server-derived create_issue
|
||||
# bootstrap before blocking the canonical control checkout. Route this
|
||||
# guard's decision through the *same* predicate on the *same*
|
||||
# assessment so the two cannot disagree about identical evidence.
|
||||
# Only the wrong-worktree verdict is waived; every other check in this
|
||||
# assessment (root checkout, repo, role, stale runtime, lease, ...) is
|
||||
# evaluated independently and still applies.
|
||||
bootstrap_waived = wt.get("block") and (
|
||||
create_issue_bootstrap.bootstrap_permits_control_checkout(
|
||||
create_issue_bootstrap_assessment,
|
||||
task=task_name,
|
||||
workspace_path=workspace_path,
|
||||
canonical_repo_root=project_root,
|
||||
)
|
||||
)
|
||||
checks["worktree"] = {
|
||||
"block": bool(wt.get("block")),
|
||||
"block": bool(wt.get("block")) and not bootstrap_waived,
|
||||
"reasons": list(wt.get("reasons") or []),
|
||||
"under_branches": wt.get("under_branches"),
|
||||
"create_issue_bootstrap_waived": bool(bootstrap_waived),
|
||||
}
|
||||
if wt.get("block"):
|
||||
if wt.get("block") and not bootstrap_waived:
|
||||
blockers.append(
|
||||
_blocker(
|
||||
BLOCKER_WRONG_WORKTREE,
|
||||
|
||||
+134
-2
@@ -52,6 +52,18 @@ def is_create_issue_task(task: str | None) -> bool:
|
||||
return (task or "").strip() in CREATE_ISSUE_TASKS
|
||||
|
||||
|
||||
def normalize_sha(value: str | None) -> str | None:
|
||||
"""Normalize a Git object id for comparison, or ``None`` when unknown.
|
||||
|
||||
Whitespace and case are the only permitted variation between two spellings
|
||||
of the same commit; anything else is a different commit. Empty and
|
||||
whitespace-only values normalize to ``None`` so an unknown tip can never
|
||||
compare equal to another unknown tip.
|
||||
"""
|
||||
normalized = (value or "").strip().lower()
|
||||
return normalized or None
|
||||
|
||||
|
||||
def assess_create_issue_bootstrap(
|
||||
*,
|
||||
workspace_path: str,
|
||||
@@ -60,6 +72,7 @@ def assess_create_issue_bootstrap(
|
||||
head_sha: str | None = None,
|
||||
porcelain_status: str = "",
|
||||
remote_master_sha: str | None = None,
|
||||
remote_master_sha_error: str | None = None,
|
||||
task: str | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Assess whether create_issue may proceed from the control checkout.
|
||||
@@ -142,8 +155,32 @@ def assess_create_issue_bootstrap(
|
||||
f"is not an accepted base branch ({'/'.join(sorted(BASE_BRANCHES))})"
|
||||
)
|
||||
|
||||
remote_tip = (remote_master_sha or "").strip() or None
|
||||
local_tip = (head_sha or "").strip() or None
|
||||
# #757 AC3/AC4: base-equivalence must be *proven*, never assumed. An
|
||||
# unknown tip on either side is not evidence of agreement, so a missing
|
||||
# local HEAD, an unresolvable live master, or a resolver failure all block.
|
||||
remote_tip = normalize_sha(remote_master_sha)
|
||||
local_tip = normalize_sha(head_sha)
|
||||
resolver_error = (remote_master_sha_error or "").strip() or None
|
||||
|
||||
if not local_tip:
|
||||
reasons.append(
|
||||
"create_issue bootstrap blocked: control checkout HEAD SHA is "
|
||||
"unknown; base equivalence to live master cannot be proven "
|
||||
"(fail closed)"
|
||||
)
|
||||
|
||||
if resolver_error:
|
||||
reasons.append(
|
||||
"create_issue bootstrap blocked: live master tip could not be "
|
||||
f"resolved ({resolver_error}); base equivalence cannot be proven "
|
||||
"(fail closed)"
|
||||
)
|
||||
elif not remote_tip:
|
||||
reasons.append(
|
||||
"create_issue bootstrap blocked: live master tip is unknown; base "
|
||||
"equivalence to live master cannot be proven (fail closed)"
|
||||
)
|
||||
|
||||
if remote_tip and local_tip and remote_tip != local_tip:
|
||||
reasons.append(
|
||||
"create_issue bootstrap blocked: control checkout HEAD does not match "
|
||||
@@ -162,6 +199,8 @@ def assess_create_issue_bootstrap(
|
||||
dirty=dirty,
|
||||
under_branches=False,
|
||||
exact_next_action=EXACT_NEXT_ACTION_BOOTSTRAP,
|
||||
local_head_sha=local_tip,
|
||||
remote_master_sha=remote_tip,
|
||||
)
|
||||
|
||||
return _result(
|
||||
@@ -176,9 +215,92 @@ def assess_create_issue_bootstrap(
|
||||
under_branches=False,
|
||||
exact_next_action=EXACT_NEXT_ACTION_POST_CREATE,
|
||||
bootstrap_path="clean_canonical_control_checkout",
|
||||
local_head_sha=local_tip,
|
||||
remote_master_sha=remote_tip,
|
||||
)
|
||||
|
||||
|
||||
def bootstrap_permits_control_checkout(
|
||||
assessment: Any,
|
||||
*,
|
||||
task: str | None,
|
||||
workspace_path: str | None,
|
||||
canonical_repo_root: str | None,
|
||||
) -> bool:
|
||||
"""Single interpretation of a bootstrap assessment (#757).
|
||||
|
||||
Both author-mutation guards — the #274 branches-only enforcer and the #604
|
||||
anti-stomp preflight — route their "may this workspace mutate" decision
|
||||
through this predicate, so the two can never reach opposite conclusions
|
||||
about identical evidence.
|
||||
|
||||
Fail-closed by construction. Every proof obligation must be present and
|
||||
affirmative in *assessment*, and the assessment must describe the very
|
||||
workspace and canonical root being guarded. A missing, malformed, refused,
|
||||
incomplete, or contradictory assessment returns ``False``, which leaves the
|
||||
caller's ordinary block in force.
|
||||
|
||||
``assessment`` is server-derived only: it is produced by
|
||||
:func:`assess_create_issue_bootstrap` from inspected repository state. It is
|
||||
never accepted from an MCP tool argument, so no caller can assert
|
||||
eligibility it has not proven.
|
||||
"""
|
||||
if not isinstance(assessment, dict):
|
||||
return False
|
||||
if not is_create_issue_task(task):
|
||||
return False
|
||||
|
||||
# Positive proof: the assessment must affirmatively allow, with no
|
||||
# competing refusal or not-applicable disposition recorded alongside it.
|
||||
if assessment.get("allowed") is not True:
|
||||
return False
|
||||
if assessment.get("proven") is not True:
|
||||
return False
|
||||
if assessment.get("block") is not False:
|
||||
return False
|
||||
if assessment.get("not_applicable") is not False:
|
||||
return False
|
||||
if assessment.get("reasons"):
|
||||
return False
|
||||
|
||||
# Scope proof: only the create_issue bootstrap, only via the clean
|
||||
# canonical control checkout path.
|
||||
if assessment.get("task_scope") != "create_issue_only":
|
||||
return False
|
||||
if assessment.get("bootstrap_path") != "clean_canonical_control_checkout":
|
||||
return False
|
||||
|
||||
# State proof: clean, and not a branches/ worktree (those keep #274).
|
||||
if assessment.get("dirty_files"):
|
||||
return False
|
||||
if assessment.get("under_branches") is not False:
|
||||
return False
|
||||
|
||||
# Base-equivalence proof (#757 AC3/AC4): both tips must be recorded,
|
||||
# nonempty, and equal. Re-derived here rather than trusted from the
|
||||
# assessment's own flag, so a hand-built or truncated assessment cannot
|
||||
# assert agreement it never proved.
|
||||
if assessment.get("base_tips_verified") is not True:
|
||||
return False
|
||||
local_tip = normalize_sha(assessment.get("local_head_sha"))
|
||||
remote_tip = normalize_sha(assessment.get("remote_master_sha"))
|
||||
if not local_tip or not remote_tip or local_tip != remote_tip:
|
||||
return False
|
||||
|
||||
# Binding proof: the assessment must describe *this* workspace and root,
|
||||
# and that workspace must be exactly the canonical control checkout.
|
||||
root = os.path.realpath(canonical_repo_root or "")
|
||||
workspace = os.path.realpath(workspace_path or root or ".")
|
||||
if not root or workspace != root:
|
||||
return False
|
||||
if assessment.get("canonical_repo_root") != root:
|
||||
return False
|
||||
if assessment.get("workspace_path") != workspace:
|
||||
return False
|
||||
|
||||
return True
|
||||
|
||||
|
||||
def format_create_issue_bootstrap_error(assessment: dict[str, Any]) -> str:
|
||||
"""RuntimeError / typed-block message for a failed bootstrap assessment."""
|
||||
reasons = "; ".join(
|
||||
@@ -209,7 +331,14 @@ def _result(
|
||||
under_branches: bool,
|
||||
exact_next_action: str | None = None,
|
||||
bootstrap_path: str | None = None,
|
||||
local_head_sha: str | None = None,
|
||||
remote_master_sha: str | None = None,
|
||||
) -> dict[str, Any]:
|
||||
# #757 AC3/AC4: equality is recorded only when BOTH tips are known, so a
|
||||
# consumer can never read agreement out of two missing values.
|
||||
base_tips_verified = bool(
|
||||
local_head_sha and remote_master_sha and local_head_sha == remote_master_sha
|
||||
)
|
||||
return {
|
||||
"not_applicable": not_applicable,
|
||||
"allowed": allowed,
|
||||
@@ -224,4 +353,7 @@ def _result(
|
||||
"exact_next_action": exact_next_action,
|
||||
"bootstrap_path": bootstrap_path,
|
||||
"task_scope": "create_issue_only",
|
||||
"local_head_sha": local_head_sha,
|
||||
"remote_master_sha": remote_master_sha,
|
||||
"base_tips_verified": base_tips_verified,
|
||||
}
|
||||
|
||||
+92
-18
@@ -817,10 +817,62 @@ def _enforce_canonical_repository_root(
|
||||
)
|
||||
|
||||
|
||||
# #757: distinguishes "caller supplied no bootstrap evidence" (compute it) from
|
||||
# "caller supplied a not-applicable/refused assessment" (honour it, fail closed).
|
||||
_BOOTSTRAP_UNSET = object()
|
||||
|
||||
|
||||
def _create_issue_bootstrap_assessment(
|
||||
task: str | None,
|
||||
worktree_path: str | None = None,
|
||||
) -> dict | None:
|
||||
"""#757: the single server-derived create_issue bootstrap assessment.
|
||||
|
||||
Computed once per preflight from inspected repository state, then consumed
|
||||
by BOTH the #274 branches-only guard and the #604 anti-stomp preflight so
|
||||
the two cannot reach opposite conclusions about identical evidence.
|
||||
|
||||
Returns ``None`` when *task* is not a create_issue mutation, which leaves
|
||||
every other author mutation on the ordinary branches-only path. The result
|
||||
is derived from inspected repository state only — never from an MCP tool
|
||||
argument.
|
||||
"""
|
||||
import create_issue_bootstrap as _cib
|
||||
|
||||
if not _cib.is_create_issue_task(task):
|
||||
return None
|
||||
|
||||
ctx = _resolve_namespace_mutation_context(worktree_path)
|
||||
workspace = ctx["workspace_path"]
|
||||
git_state = issue_lock_worktree.read_worktree_git_state(workspace)
|
||||
# #757 AC3/AC4: a resolver failure is not "no constraint" — it is missing
|
||||
# evidence, and must reach the assessor as such so the bootstrap fails
|
||||
# closed instead of proceeding without base-equivalence proof.
|
||||
remote_master_sha_error: str | None = None
|
||||
try:
|
||||
remote_master_sha = root_checkout_guard.resolve_remote_master_sha(
|
||||
ctx["canonical_repo_root"]
|
||||
)
|
||||
except Exception as exc:
|
||||
remote_master_sha = None
|
||||
remote_master_sha_error = f"{type(exc).__name__}: {exc}".strip() or "resolver failed"
|
||||
return _cib.assess_create_issue_bootstrap(
|
||||
workspace_path=workspace,
|
||||
canonical_repo_root=ctx["canonical_repo_root"],
|
||||
current_branch=git_state.get("current_branch"),
|
||||
head_sha=git_state.get("head_sha"),
|
||||
porcelain_status=git_state.get("porcelain_status") or "",
|
||||
remote_master_sha=remote_master_sha,
|
||||
remote_master_sha_error=remote_master_sha_error,
|
||||
task=task,
|
||||
)
|
||||
|
||||
|
||||
def _enforce_branches_only_author_mutation(
|
||||
worktree_path: str | None = None,
|
||||
*,
|
||||
task: str | None = None,
|
||||
bootstrap_assessment: Any = _BOOTSTRAP_UNSET,
|
||||
) -> None:
|
||||
"""#274: author file/branch mutations must run from a branches/ worktree.
|
||||
|
||||
@@ -859,27 +911,29 @@ def _enforce_branches_only_author_mutation(
|
||||
return
|
||||
|
||||
# #749 create_issue bootstrap: narrow phase exemption only.
|
||||
# #757: consume the caller-computed assessment when one was threaded in, so
|
||||
# this guard and the later #604 anti-stomp guard judge identical evidence.
|
||||
# Falling back to computing it here preserves behaviour for callers that
|
||||
# supply none.
|
||||
import create_issue_bootstrap as _cib
|
||||
|
||||
remote_master_sha = None
|
||||
try:
|
||||
remote_master_sha = root_checkout_guard.resolve_remote_master_sha(
|
||||
ctx["canonical_repo_root"]
|
||||
)
|
||||
except Exception:
|
||||
remote_master_sha = None
|
||||
bootstrap = _cib.assess_create_issue_bootstrap(
|
||||
bootstrap = (
|
||||
_create_issue_bootstrap_assessment(task, worktree_path)
|
||||
if bootstrap_assessment is _BOOTSTRAP_UNSET
|
||||
else bootstrap_assessment
|
||||
)
|
||||
if _cib.bootstrap_permits_control_checkout(
|
||||
bootstrap,
|
||||
task=task,
|
||||
workspace_path=workspace,
|
||||
canonical_repo_root=ctx["canonical_repo_root"],
|
||||
current_branch=git_state.get("current_branch"),
|
||||
head_sha=git_state.get("head_sha"),
|
||||
porcelain_status=git_state.get("porcelain_status") or "",
|
||||
remote_master_sha=remote_master_sha,
|
||||
task=task,
|
||||
)
|
||||
if bootstrap.get("allowed"):
|
||||
):
|
||||
return
|
||||
if bootstrap.get("block") and not bootstrap.get("not_applicable"):
|
||||
if (
|
||||
isinstance(bootstrap, dict)
|
||||
and bootstrap.get("block")
|
||||
and not bootstrap.get("not_applicable")
|
||||
):
|
||||
raise RuntimeError(_cib.format_create_issue_bootstrap_error(bootstrap))
|
||||
raise RuntimeError(
|
||||
author_mutation_worktree.format_author_mutation_worktree_error(assessment)
|
||||
@@ -926,6 +980,7 @@ def _run_anti_stomp_preflight(
|
||||
workflow_hash_valid: bool | None = None,
|
||||
workflow_hash_reasons: list[str] | None = None,
|
||||
raise_on_block: bool = True,
|
||||
bootstrap_assessment: Any = _BOOTSTRAP_UNSET,
|
||||
) -> dict | None:
|
||||
"""Shared #604 anti-stomp preflight for MCP mutation entrypoints.
|
||||
|
||||
@@ -1106,6 +1161,14 @@ def _run_anti_stomp_preflight(
|
||||
source_contaminated=source_contaminated,
|
||||
contamination_reasons=contamination_reasons,
|
||||
manual_bypass_attempted=False,
|
||||
# #757: same server-derived bootstrap decision the #274 guard consumed.
|
||||
# Computing it here when unsupplied keeps standalone callers consistent
|
||||
# rather than silently bootstrap-blind.
|
||||
create_issue_bootstrap_assessment=(
|
||||
_create_issue_bootstrap_assessment(task, worktree_path)
|
||||
if bootstrap_assessment is _BOOTSTRAP_UNSET
|
||||
else bootstrap_assessment
|
||||
),
|
||||
)
|
||||
if not assessment.get("block"):
|
||||
return None
|
||||
@@ -1248,7 +1311,12 @@ def verify_preflight_purity(
|
||||
# Historical path: root + branches after purity-order when dirty paths live.
|
||||
_enforce_canonical_repository_root(worktree_path, remote=remote)
|
||||
_enforce_root_checkout_guard(worktree_path)
|
||||
_enforce_branches_only_author_mutation(worktree_path, task=task)
|
||||
# #757: compute the create_issue bootstrap decision ONCE and hand the
|
||||
# same result to both the #274 and #604 guards, so they cannot disagree.
|
||||
bootstrap_assessment = _create_issue_bootstrap_assessment(task, worktree_path)
|
||||
_enforce_branches_only_author_mutation(
|
||||
worktree_path, task=task, bootstrap_assessment=bootstrap_assessment
|
||||
)
|
||||
_enforce_issue_scope_guard(
|
||||
worktree_path,
|
||||
task=task,
|
||||
@@ -1258,6 +1326,7 @@ def verify_preflight_purity(
|
||||
# #604: common anti-stomp preflight after legacy + #683 enforcers.
|
||||
_run_anti_stomp_preflight(
|
||||
task,
|
||||
bootstrap_assessment=bootstrap_assessment,
|
||||
remote=remote,
|
||||
worktree_path=worktree_path,
|
||||
org=org,
|
||||
@@ -1282,7 +1351,11 @@ def verify_preflight_purity(
|
||||
if production_active:
|
||||
_enforce_canonical_repository_root(worktree_path, remote=remote)
|
||||
_enforce_root_checkout_guard(worktree_path)
|
||||
_enforce_branches_only_author_mutation(worktree_path, task=task)
|
||||
# #757: one shared bootstrap decision for both guards (see above).
|
||||
bootstrap_assessment = _create_issue_bootstrap_assessment(task, worktree_path)
|
||||
_enforce_branches_only_author_mutation(
|
||||
worktree_path, task=task, bootstrap_assessment=bootstrap_assessment
|
||||
)
|
||||
_enforce_issue_scope_guard(
|
||||
worktree_path,
|
||||
task=task,
|
||||
@@ -1292,6 +1365,7 @@ def verify_preflight_purity(
|
||||
if force_anti_stomp:
|
||||
_run_anti_stomp_preflight(
|
||||
task,
|
||||
bootstrap_assessment=bootstrap_assessment,
|
||||
remote=remote,
|
||||
worktree_path=worktree_path,
|
||||
org=org,
|
||||
|
||||
@@ -0,0 +1,809 @@
|
||||
"""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()
|
||||
@@ -38,11 +38,17 @@ class TestReconcilerCloseWorkspaceGuard(unittest.TestCase):
|
||||
srv._preflight_capability_violation = False
|
||||
self._orig_in_test = srv._preflight_in_test_mode
|
||||
srv._preflight_in_test_mode = lambda: False
|
||||
self._orig_resolved_task = srv._preflight_resolved_task
|
||||
self._orig_resolved_role = srv._preflight_resolved_role
|
||||
self._env_patch = patch.dict(os.environ, {"GITEA_MCP_DISABLE_PARITY_GATE": "1"}, clear=False)
|
||||
self._env_patch.start()
|
||||
|
||||
def tearDown(self):
|
||||
srv._preflight_in_test_mode = self._orig_in_test
|
||||
# Preflight task/role are module-level; restore so test order cannot
|
||||
# leak a resolved task into sibling cases.
|
||||
srv._preflight_resolved_task = self._orig_resolved_task
|
||||
srv._preflight_resolved_role = self._orig_resolved_role
|
||||
self._env_patch.stop()
|
||||
|
||||
@patch("gitea_mcp_server._auth", return_value=FAKE_AUTH)
|
||||
@@ -81,26 +87,28 @@ class TestReconcilerCloseWorkspaceGuard(unittest.TestCase):
|
||||
"gitea_mcp_server.issue_lock_worktree.read_worktree_git_state",
|
||||
return_value={"current_branch": "master"},
|
||||
)
|
||||
def test_author_create_issue_still_blocked_on_control_checkout(
|
||||
def test_author_non_create_issue_still_blocked_on_control_checkout(
|
||||
self, _git, _get_all, _role, _ns, _prof, _auth
|
||||
):
|
||||
"""Author mutations other than create_issue keep the branches-only rule.
|
||||
|
||||
#749/#750 sanctioned ``create_issue`` from a clean control checkout, and
|
||||
#757 made the #604 anti-stomp guard honour that same decision — so
|
||||
``create_issue`` is no longer a valid probe for this boundary. This case
|
||||
previously asserted create_issue stayed blocked, which only held because
|
||||
the bootstrap-blind #604 guard was overriding #750; that is precisely
|
||||
the defect #757 fixed. ``lock_issue`` is issue-backed and post-ownership,
|
||||
so it still requires a ``branches/`` worktree.
|
||||
"""
|
||||
srv._preflight_resolved_role = "author"
|
||||
srv._preflight_resolved_task = "lock_issue"
|
||||
with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT):
|
||||
try:
|
||||
res = srv.gitea_create_issue(title="Test", body="body")
|
||||
srv.verify_preflight_purity(remote="prgs", task="lock_issue")
|
||||
except RuntimeError as exc:
|
||||
self.assertIn("stable control checkout", str(exc))
|
||||
self.assertIn("control checkout", str(exc).lower())
|
||||
else:
|
||||
# #683 typed blocker at mutation entrypoint
|
||||
self.assertFalse(res.get("success"))
|
||||
blob = " ".join(res.get("reasons") or []) + str(
|
||||
res.get("blocker_kind") or ""
|
||||
)
|
||||
self.assertTrue(
|
||||
"stable control checkout" in blob
|
||||
or "missing_issue_worktree" in blob
|
||||
or "control checkout" in blob.lower()
|
||||
)
|
||||
self.fail("lock_issue must stay blocked on the control checkout")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
Reference in New Issue
Block a user