sysadmin
8d8d2d8f81
Session-bound author mutations (create_issue, lock_issue, create_pr, etc.) called verify_preflight_purity without org/repo, so anti-stomp filled the remote-wide REMOTES default (prgs → Timesheet) and false-positived wrong_repo against a Gitea-Tools checkout even when callers passed explicit coordinates. Mirror _resolve: prefer workspace-derived identity for omitted coordinates on session-bound tasks; keep delete_branch's explicit-target contract (#733). Also forward org/repo at author mutation call sites. Add cross-repo regression coverage. Do not change REMOTES["prgs"]["repo"].
281 lines
11 KiB
Python
281 lines
11 KiB
Python
"""Regression matrix: session-bound mutation preflight resolves the workspace repo.
|
|
|
|
Root-cause follow-up to #733. ``_resolve`` (the actual API-targeting resolver)
|
|
already prefers the workspace-aligned git remote over the ``REMOTES`` default
|
|
(bare ``prgs`` → ``Scaled-Tech-Consulting/Timesheet``). The shared #604
|
|
anti-stomp preflight (``_run_anti_stomp_preflight``) did **not** mirror that: for
|
|
every mutation task whose call site omitted ``org``/``repo`` it filled the
|
|
remote-wide ``REMOTES`` default, so the #530 repo guard false-positived
|
|
``wrong_repo`` against a ``Gitea-Tools`` checkout and blocked native
|
|
issue/PR/review/merge/lock/cleanup mutations.
|
|
|
|
The fix makes ``_run_anti_stomp_preflight`` prefer the trusted, git-remote-
|
|
derived workspace identity for omitted coordinates on **session-bound** mutation
|
|
tasks (marking those filled sides explicit, exactly like ``_resolve``), while
|
|
``delete_branch`` keeps its explicit cross-repository *target* contract (#733)
|
|
and continues to fail closed on omitted coordinates.
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
|
|
|
|
import mcp_server
|
|
import remote_repo_guard
|
|
|
|
# Workspace is bound to Gitea-Tools even though the prgs REMOTES default repo is
|
|
# Timesheet (the exact cross-repo scenario).
|
|
LOCAL_GITEA_TOOLS_URL = "https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git"
|
|
|
|
RECONCILER = {
|
|
"profile_name": "prgs-reconciler",
|
|
"role": "reconciler",
|
|
"allowed_operations": [
|
|
"gitea.read", "gitea.issue.comment", "gitea.pr.comment",
|
|
"gitea.pr.close", "gitea.branch.delete",
|
|
],
|
|
"forbidden_operations": ["gitea.pr.create", "gitea.pr.merge"],
|
|
"audit_label": "prgs-reconciler",
|
|
}
|
|
|
|
# Session-bound mutation tasks (every MUTATION_TASK except the explicit-target
|
|
# delete_branch) must auto-resolve the workspace repository for omitted coords.
|
|
SESSION_BOUND_TASKS = [
|
|
"create_issue",
|
|
"comment_issue",
|
|
"close_issue",
|
|
"mark_issue",
|
|
"lock_issue",
|
|
"set_issue_labels",
|
|
"create_label",
|
|
"create_pr",
|
|
"close_pr",
|
|
"edit_pr",
|
|
"commit_files",
|
|
"cleanup_merged_pr_branch",
|
|
"cleanup_stale_claims",
|
|
"reconcile_merged_cleanups",
|
|
"reconcile_already_landed_pr",
|
|
"reconcile_close_superseded_pr",
|
|
"post_heartbeat",
|
|
"review_pr",
|
|
"submit_pr_review",
|
|
"merge_pr",
|
|
]
|
|
|
|
|
|
class _AntiStompResolutionBase(unittest.TestCase):
|
|
def setUp(self):
|
|
self._remotes = patch.dict(mcp_server.REMOTES, {
|
|
"prgs": {
|
|
"host": "gitea.prgs.cc",
|
|
"org": "Scaled-Tech-Consulting",
|
|
"repo": "Timesheet",
|
|
},
|
|
})
|
|
self._remotes.start()
|
|
|
|
def tearDown(self):
|
|
patch.stopall()
|
|
|
|
def _capture_resolution(self, task, org, repo, *, local_url=LOCAL_GITEA_TOOLS_URL):
|
|
"""Drive the live anti-stomp runner and capture the org/repo it resolved
|
|
and passed to the pure assessor."""
|
|
passthrough = {
|
|
"allowed": True, "block": False, "blockers": [], "reasons": [],
|
|
"exact_next_action": "proceed", "blocker_kind": None, "checks": {},
|
|
}
|
|
with patch.dict(
|
|
os.environ, {"GITEA_TEST_FORCE_ANTI_STOMP": "1"}, clear=False
|
|
), patch(
|
|
"mcp_server._local_git_remote_url", return_value=local_url
|
|
), patch(
|
|
"mcp_server.get_profile", return_value=RECONCILER
|
|
), patch.object(
|
|
mcp_server.anti_stomp_preflight,
|
|
"assess_anti_stomp_preflight",
|
|
return_value=passthrough,
|
|
) as m:
|
|
mcp_server._run_anti_stomp_preflight(
|
|
task, remote="prgs", org=org, repo=repo
|
|
)
|
|
self.assertTrue(m.called)
|
|
return m.call_args.kwargs
|
|
|
|
|
|
class TestSessionBoundResolution(_AntiStompResolutionBase):
|
|
def test_omitted_coords_resolve_workspace_repo_and_mark_explicit(self):
|
|
for task in SESSION_BOUND_TASKS:
|
|
with self.subTest(task=task):
|
|
kw = self._capture_resolution(task, None, None)
|
|
self.assertEqual(
|
|
kw["resolved_org"], "Scaled-Tech-Consulting", task
|
|
)
|
|
self.assertEqual(kw["resolved_repo"], "Gitea-Tools", task)
|
|
# Workspace-filled sides are intentional alignment → explicit.
|
|
self.assertTrue(kw["org_explicit"], task)
|
|
self.assertTrue(kw["repo_explicit"], task)
|
|
|
|
def test_resolved_workspace_repo_passes_the_530_guard(self):
|
|
kw = self._capture_resolution("create_pr", None, None)
|
|
assessment = remote_repo_guard.assess_remote_repo_match(
|
|
remote="prgs",
|
|
resolved_org=kw["resolved_org"],
|
|
resolved_repo=kw["resolved_repo"],
|
|
local_remote_url=LOCAL_GITEA_TOOLS_URL,
|
|
org_explicit=kw["org_explicit"],
|
|
repo_explicit=kw["repo_explicit"],
|
|
)
|
|
self.assertFalse(assessment["block"], assessment)
|
|
|
|
def test_explicit_caller_coords_are_preserved(self):
|
|
kw = self._capture_resolution(
|
|
"create_pr", "Scaled-Tech-Consulting", "Gitea-Tools"
|
|
)
|
|
self.assertEqual(kw["resolved_org"], "Scaled-Tech-Consulting")
|
|
self.assertEqual(kw["resolved_repo"], "Gitea-Tools")
|
|
self.assertTrue(kw["org_explicit"])
|
|
self.assertTrue(kw["repo_explicit"])
|
|
|
|
def test_no_workspace_identity_falls_back_to_remotes_default(self):
|
|
# No derivable local remote → best-effort REMOTES default preserved,
|
|
# not marked explicit (so the guard still corroborates when able).
|
|
kw = self._capture_resolution("create_pr", None, None, local_url=None)
|
|
self.assertEqual(kw["resolved_repo"], "Timesheet")
|
|
self.assertFalse(kw["org_explicit"])
|
|
self.assertFalse(kw["repo_explicit"])
|
|
|
|
|
|
class TestDeleteBranchContractPreserved(_AntiStompResolutionBase):
|
|
"""delete_branch keeps its explicit-target contract (#733): omitted coords
|
|
still resolve the remote-wide default and are NOT marked explicit, so the
|
|
guard fails closed against a mismatched workspace."""
|
|
|
|
def test_omitted_coords_still_remote_default_not_explicit(self):
|
|
kw = self._capture_resolution("delete_branch", None, None)
|
|
self.assertEqual(kw["resolved_repo"], "Timesheet")
|
|
self.assertFalse(kw["org_explicit"])
|
|
self.assertFalse(kw["repo_explicit"])
|
|
assessment = remote_repo_guard.assess_remote_repo_match(
|
|
remote="prgs",
|
|
resolved_org=kw["resolved_org"],
|
|
resolved_repo=kw["resolved_repo"],
|
|
local_remote_url=LOCAL_GITEA_TOOLS_URL,
|
|
org_explicit=kw["org_explicit"],
|
|
repo_explicit=kw["repo_explicit"],
|
|
)
|
|
self.assertTrue(assessment["block"])
|
|
|
|
def test_explicit_coords_still_forwarded(self):
|
|
kw = self._capture_resolution(
|
|
"delete_branch", "Scaled-Tech-Consulting", "Gitea-Tools"
|
|
)
|
|
self.assertEqual(kw["resolved_repo"], "Gitea-Tools")
|
|
self.assertTrue(kw["org_explicit"])
|
|
self.assertTrue(kw["repo_explicit"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|
|
|
|
|
|
class TestAuthorCallSiteForwardsOrgRepo(_AntiStompResolutionBase):
|
|
"""#735: author mutation entrypoints must pass org/repo into preflight."""
|
|
|
|
def _assert_call_forwards(self, fn_name: str, *args, **kwargs):
|
|
import inspect
|
|
fn = getattr(mcp_server, fn_name)
|
|
# Force preflight path to raise with the kwargs it received so we can
|
|
# prove org/repo were forwarded without performing the full mutation.
|
|
captured = {}
|
|
|
|
def _capture(*a, **kw):
|
|
captured.update(kw)
|
|
captured["_args"] = a
|
|
raise RuntimeError("capture-only")
|
|
|
|
with patch.object(mcp_server, "verify_preflight_purity", side_effect=_capture), \
|
|
patch.object(mcp_server, "get_profile", return_value={
|
|
"profile_name": "prgs-author",
|
|
"role": "author",
|
|
"allowed_operations": [
|
|
"gitea.read", "gitea.issue.create", "gitea.issue.comment",
|
|
"gitea.pr.create", "gitea.repo.commit", "gitea.branch.push",
|
|
],
|
|
"forbidden_operations": [],
|
|
"audit_label": "prgs-author",
|
|
}), patch.object(
|
|
mcp_server, "_resolve",
|
|
return_value=("gitea.prgs.cc", "Scaled-Tech-Consulting", "Gitea-Tools"),
|
|
), patch.object(
|
|
mcp_server, "_auth", return_value="token fake",
|
|
), patch.object(
|
|
mcp_server.role_session_router,
|
|
"check_author_mutation_after_reviewer_stop",
|
|
return_value=(True, []),
|
|
), patch.object(
|
|
mcp_server, "_namespace_mutation_block", return_value=None
|
|
), patch.object(
|
|
mcp_server, "_profile_permission_block", return_value=None
|
|
), patch.object(
|
|
mcp_server, "_production_guard_block_from_exc", return_value=None
|
|
):
|
|
try:
|
|
fn(*args, **kwargs)
|
|
except RuntimeError as exc:
|
|
if "capture-only" not in str(exc):
|
|
# Some tools re-raise; still require capture when preflight ran.
|
|
if not captured:
|
|
raise
|
|
self.assertTrue(captured, f"{fn_name} never called verify_preflight_purity")
|
|
self.assertEqual(captured.get("org"), "Scaled-Tech-Consulting", fn_name)
|
|
self.assertEqual(captured.get("repo"), "Gitea-Tools", fn_name)
|
|
|
|
def test_create_issue_forwards_explicit_org_repo(self):
|
|
self._assert_call_forwards(
|
|
"gitea_create_issue",
|
|
"title",
|
|
body="body with acceptance criteria for gate",
|
|
remote="prgs",
|
|
org="Scaled-Tech-Consulting",
|
|
repo="Gitea-Tools",
|
|
worktree_path="/tmp/fake-wt",
|
|
)
|
|
|
|
def test_create_pr_forwards_explicit_org_repo(self):
|
|
self._assert_call_forwards(
|
|
"gitea_create_pr",
|
|
title="t",
|
|
head="fix/issue-735-x",
|
|
base="master",
|
|
body="Closes #735",
|
|
remote="prgs",
|
|
org="Scaled-Tech-Consulting",
|
|
repo="Gitea-Tools",
|
|
worktree_path="/tmp/fake-wt",
|
|
)
|
|
|
|
def test_mark_issue_forwards_explicit_org_repo(self):
|
|
self._assert_call_forwards(
|
|
"gitea_mark_issue",
|
|
issue_number=735,
|
|
action="start",
|
|
remote="prgs",
|
|
org="Scaled-Tech-Consulting",
|
|
repo="Gitea-Tools",
|
|
worktree_path="/tmp/fake-wt",
|
|
)
|
|
|
|
def test_commit_files_forwards_explicit_org_repo(self):
|
|
self._assert_call_forwards(
|
|
"gitea_commit_files",
|
|
files=[{"operation": "update", "path": "x", "content_plain": "y"}],
|
|
message="m",
|
|
remote="prgs",
|
|
org="Scaled-Tech-Consulting",
|
|
repo="Gitea-Tools",
|
|
)
|