Merge branch 'prgs/master' into feat/issue-389-review-workflow-load-proof
This commit is contained in:
@@ -3888,7 +3888,7 @@ class TestPreflightVerification(unittest.TestCase):
|
||||
os.environ["GITEA_TEST_PORCELAIN"] = " M reviewer_edit.py\n"
|
||||
with self.assertRaises(RuntimeError) as ctx:
|
||||
mcp_server.verify_preflight_purity()
|
||||
self.assertIn("Reviewer profile is forbidden from modifying tracked workspace files", str(ctx.exception))
|
||||
self.assertIn("forbidden from modifying tracked workspace files", str(ctx.exception))
|
||||
self.assertIn("reviewer_edit.py", str(ctx.exception))
|
||||
|
||||
# Foreign pre-existing dirty state does not block when unchanged.
|
||||
@@ -3954,7 +3954,8 @@ class TestPreflightVerification(unittest.TestCase):
|
||||
with self.assertRaises(RuntimeError) as ctx:
|
||||
mcp_server.verify_preflight_purity(worktree_path=worktree)
|
||||
msg = str(ctx.exception)
|
||||
self.assertIn("active task workspace root", msg)
|
||||
self.assertIn("inspected git root", msg)
|
||||
self.assertIn("dirty files: task_file.py", msg)
|
||||
self.assertIn("dirty scope:", msg)
|
||||
self.assertIn("resolved workspace", msg)
|
||||
self.assertIn(worktree, msg)
|
||||
self.assertIn("worktree_path argument", msg)
|
||||
self.assertIn("task_file.py", msg)
|
||||
self.assertIn("author namespace", msg)
|
||||
|
||||
@@ -0,0 +1,240 @@
|
||||
"""Tests for namespace-scoped MCP workspace binding (#510)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import sys
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest import mock
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
import gitea_mcp_server as srv # noqa: E402
|
||||
import namespace_workspace_binding as nwb # noqa: E402
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parent.parent
|
||||
if REPO_ROOT.parent.name == "branches":
|
||||
CONTROL_ROOT = str(REPO_ROOT.parent.parent)
|
||||
else:
|
||||
CONTROL_ROOT = str(REPO_ROOT)
|
||||
|
||||
AUTHOR_DIRTY = f"{CONTROL_ROOT}/branches/mcp-author-worktree"
|
||||
MERGER_CLEAN = f"{CONTROL_ROOT}/branches/merge-pr487-submit"
|
||||
REVIEWER_CLEAN = f"{CONTROL_ROOT}/branches/review-pr487-submit"
|
||||
RECONCILER_CLEAN = f"{CONTROL_ROOT}/branches/reconcile-pr487"
|
||||
MCP_PROCESS_ROOT = CONTROL_ROOT
|
||||
|
||||
|
||||
class TestNamespaceWorkspaceModule(unittest.TestCase):
|
||||
def test_author_env_ignored_for_merger_namespace(self):
|
||||
workspace, source = nwb.resolve_namespace_workspace(
|
||||
role_kind="merger",
|
||||
worktree_path=None,
|
||||
process_project_root=MCP_PROCESS_ROOT,
|
||||
env={
|
||||
nwb.AUTHOR_WORKTREE_ENV: AUTHOR_DIRTY,
|
||||
nwb.MERGER_WORKTREE_ENV: MERGER_CLEAN,
|
||||
},
|
||||
profile_name="gitea-merger",
|
||||
)
|
||||
self.assertEqual(workspace, os.path.realpath(MERGER_CLEAN))
|
||||
self.assertEqual(source, f"{nwb.MERGER_WORKTREE_ENV} environment variable")
|
||||
|
||||
def test_author_env_ignored_for_reviewer_namespace(self):
|
||||
workspace, source = nwb.resolve_namespace_workspace(
|
||||
role_kind="reviewer",
|
||||
worktree_path=None,
|
||||
process_project_root=MCP_PROCESS_ROOT,
|
||||
env={
|
||||
nwb.AUTHOR_WORKTREE_ENV: AUTHOR_DIRTY,
|
||||
nwb.REVIEWER_WORKTREE_ENV: REVIEWER_CLEAN,
|
||||
},
|
||||
)
|
||||
self.assertEqual(workspace, os.path.realpath(REVIEWER_CLEAN))
|
||||
self.assertEqual(source, f"{nwb.REVIEWER_WORKTREE_ENV} environment variable")
|
||||
|
||||
def test_author_env_ignored_for_reconciler_namespace(self):
|
||||
workspace, source = nwb.resolve_namespace_workspace(
|
||||
role_kind="reconciler",
|
||||
worktree_path=None,
|
||||
process_project_root=MCP_PROCESS_ROOT,
|
||||
env={
|
||||
nwb.AUTHOR_WORKTREE_ENV: AUTHOR_DIRTY,
|
||||
nwb.RECONCILER_WORKTREE_ENV: RECONCILER_CLEAN,
|
||||
},
|
||||
)
|
||||
self.assertEqual(workspace, os.path.realpath(RECONCILER_CLEAN))
|
||||
self.assertEqual(source, f"{nwb.RECONCILER_WORKTREE_ENV} environment variable")
|
||||
|
||||
def test_merger_profile_maps_to_merger_namespace(self):
|
||||
role = nwb.normalize_role_kind("reviewer", profile_name="gitea-merger")
|
||||
self.assertEqual(role, "merger")
|
||||
|
||||
def test_error_message_includes_path_and_binding_source(self):
|
||||
msg = nwb.format_namespace_workspace_binding_error(
|
||||
role_kind="merger",
|
||||
workspace_path=AUTHOR_DIRTY,
|
||||
binding_source=f"{nwb.AUTHOR_WORKTREE_ENV} environment variable",
|
||||
dirty_files=["gitea_mcp_server.py"],
|
||||
ignored_bindings=[f"{nwb.AUTHOR_WORKTREE_ENV}={AUTHOR_DIRTY} (ignored for merger namespace)"],
|
||||
)
|
||||
self.assertIn(AUTHOR_DIRTY, msg)
|
||||
self.assertIn("via", msg.lower())
|
||||
self.assertIn(nwb.AUTHOR_WORKTREE_ENV, msg)
|
||||
self.assertIn("Do not clean or reset foreign role worktrees", msg)
|
||||
|
||||
|
||||
class TestNamespaceWorkspaceIntegration(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self._saved = {
|
||||
"whoami_called": srv._preflight_whoami_called,
|
||||
"capability_called": srv._preflight_capability_called,
|
||||
"resolved_role": srv._preflight_resolved_role,
|
||||
"whoami_violation": srv._preflight_whoami_violation,
|
||||
"capability_violation": srv._preflight_capability_violation,
|
||||
"in_test": srv._preflight_in_test_mode,
|
||||
}
|
||||
srv._preflight_whoami_called = True
|
||||
srv._preflight_capability_called = True
|
||||
srv._preflight_whoami_violation = False
|
||||
srv._preflight_capability_violation = False
|
||||
srv._preflight_in_test_mode = lambda: False
|
||||
self._env_patch = mock.patch.dict(os.environ, {"GITEA_TEST_PORCELAIN": ""}, clear=False)
|
||||
self._env_patch.start()
|
||||
|
||||
def tearDown(self):
|
||||
srv._preflight_whoami_called = self._saved["whoami_called"]
|
||||
srv._preflight_capability_called = self._saved["capability_called"]
|
||||
srv._preflight_resolved_role = self._saved["resolved_role"]
|
||||
srv._preflight_whoami_violation = self._saved["whoami_violation"]
|
||||
srv._preflight_capability_violation = self._saved["capability_violation"]
|
||||
srv._preflight_in_test_mode = self._saved["in_test"]
|
||||
self._env_patch.stop()
|
||||
for key in (
|
||||
nwb.AUTHOR_WORKTREE_ENV,
|
||||
nwb.MERGER_WORKTREE_ENV,
|
||||
nwb.REVIEWER_WORKTREE_ENV,
|
||||
nwb.RECONCILER_WORKTREE_ENV,
|
||||
nwb.ACTIVE_WORKTREE_ENV,
|
||||
):
|
||||
os.environ.pop(key, None)
|
||||
|
||||
def _merger_profile(self):
|
||||
return {
|
||||
"profile_name": "gitea-merger",
|
||||
"allowed_operations": ["gitea.pr.merge", "gitea.read"],
|
||||
"forbidden_operations": ["gitea.pr.create", "gitea.branch.push"],
|
||||
}
|
||||
|
||||
def _reviewer_profile(self):
|
||||
return {
|
||||
"profile_name": "prgs-reviewer",
|
||||
"allowed_operations": ["gitea.pr.approve", "gitea.pr.review", "gitea.read"],
|
||||
"forbidden_operations": ["gitea.pr.create", "gitea.branch.push"],
|
||||
}
|
||||
|
||||
def _reconciler_profile(self):
|
||||
return {
|
||||
"profile_name": "prgs-reconciler",
|
||||
"allowed_operations": ["gitea.pr.close", "gitea.read"],
|
||||
"forbidden_operations": [
|
||||
"gitea.pr.approve",
|
||||
"gitea.pr.merge",
|
||||
"gitea.pr.create",
|
||||
],
|
||||
}
|
||||
|
||||
def _author_profile(self):
|
||||
return {
|
||||
"profile_name": "prgs-author",
|
||||
"allowed_operations": ["gitea.pr.create", "gitea.branch.push", "gitea.read"],
|
||||
"forbidden_operations": ["gitea.pr.merge", "gitea.pr.approve"],
|
||||
}
|
||||
|
||||
@mock.patch("subprocess.run")
|
||||
@mock.patch("os.path.isdir", return_value=True)
|
||||
@mock.patch("os.path.exists", return_value=True)
|
||||
def test_dirty_author_worktree_does_not_block_merger_with_clean_workspace(
|
||||
self, _exists, _isdir, mock_run
|
||||
):
|
||||
mock_run.return_value = MagicMock(returncode=0, stdout=f"{CONTROL_ROOT}/.git\n")
|
||||
os.environ[nwb.AUTHOR_WORKTREE_ENV] = AUTHOR_DIRTY
|
||||
os.environ[nwb.MERGER_WORKTREE_ENV] = MERGER_CLEAN
|
||||
srv._preflight_resolved_role = "reviewer"
|
||||
with mock.patch.object(srv, "PROJECT_ROOT", MCP_PROCESS_ROOT):
|
||||
with mock.patch("gitea_mcp_server.get_profile", return_value=self._merger_profile()):
|
||||
srv.verify_preflight_purity("prgs", worktree_path=MERGER_CLEAN)
|
||||
|
||||
@mock.patch("subprocess.run")
|
||||
@mock.patch("os.path.isdir", return_value=True)
|
||||
@mock.patch("os.path.exists", return_value=True)
|
||||
def test_dirty_author_worktree_does_not_block_reviewer_with_clean_workspace(
|
||||
self, _exists, _isdir, mock_run
|
||||
):
|
||||
mock_run.return_value = MagicMock(returncode=0, stdout=f"{CONTROL_ROOT}/.git\n")
|
||||
os.environ[nwb.AUTHOR_WORKTREE_ENV] = AUTHOR_DIRTY
|
||||
os.environ[nwb.REVIEWER_WORKTREE_ENV] = REVIEWER_CLEAN
|
||||
srv._preflight_resolved_role = "reviewer"
|
||||
with mock.patch.object(srv, "PROJECT_ROOT", MCP_PROCESS_ROOT):
|
||||
with mock.patch("gitea_mcp_server.get_profile", return_value=self._reviewer_profile()):
|
||||
srv.verify_preflight_purity("prgs", worktree_path=REVIEWER_CLEAN)
|
||||
|
||||
@mock.patch("subprocess.run")
|
||||
@mock.patch("os.path.isdir", return_value=True)
|
||||
@mock.patch("os.path.exists", return_value=True)
|
||||
def test_dirty_author_worktree_does_not_block_reconciler_with_clean_workspace(
|
||||
self, _exists, _isdir, mock_run
|
||||
):
|
||||
mock_run.return_value = MagicMock(returncode=0, stdout=f"{CONTROL_ROOT}/.git\n")
|
||||
os.environ[nwb.AUTHOR_WORKTREE_ENV] = AUTHOR_DIRTY
|
||||
os.environ[nwb.RECONCILER_WORKTREE_ENV] = RECONCILER_CLEAN
|
||||
srv._preflight_resolved_role = "reconciler"
|
||||
with mock.patch.object(srv, "PROJECT_ROOT", MCP_PROCESS_ROOT):
|
||||
with mock.patch("gitea_mcp_server.get_profile", return_value=self._reconciler_profile()):
|
||||
srv.verify_preflight_purity("prgs", worktree_path=RECONCILER_CLEAN)
|
||||
|
||||
@mock.patch("subprocess.run")
|
||||
@mock.patch("os.path.isdir", return_value=True)
|
||||
@mock.patch("os.path.exists", return_value=True)
|
||||
def test_dirty_active_task_workspace_still_blocks_mutations(
|
||||
self, _exists, _isdir, mock_run
|
||||
):
|
||||
mock_run.return_value = MagicMock(returncode=0, stdout=f"{CONTROL_ROOT}/.git\n")
|
||||
os.environ[nwb.MERGER_WORKTREE_ENV] = MERGER_CLEAN
|
||||
srv._preflight_resolved_role = "reviewer"
|
||||
dirty = " M namespace_workspace_binding.py\n"
|
||||
with mock.patch.object(srv, "PROJECT_ROOT", MCP_PROCESS_ROOT):
|
||||
with mock.patch("gitea_mcp_server.get_profile", return_value=self._merger_profile()):
|
||||
with mock.patch("gitea_mcp_server._get_workspace_porcelain", return_value=dirty):
|
||||
with self.assertRaises(RuntimeError) as ctx:
|
||||
srv.verify_preflight_purity("prgs", worktree_path=MERGER_CLEAN)
|
||||
self.assertIn(MERGER_CLEAN, str(ctx.exception))
|
||||
self.assertIn("binding", str(ctx.exception).lower())
|
||||
|
||||
def test_root_workspace_mutation_still_blocked_for_author(self):
|
||||
srv._preflight_resolved_role = "author"
|
||||
with mock.patch.object(srv, "PROJECT_ROOT", CONTROL_ROOT):
|
||||
with mock.patch("gitea_mcp_server.get_profile", return_value=self._author_profile()):
|
||||
with mock.patch(
|
||||
"gitea_mcp_server.issue_lock_worktree.read_worktree_git_state",
|
||||
return_value={"current_branch": "master"},
|
||||
):
|
||||
with self.assertRaises(RuntimeError) as ctx:
|
||||
srv.verify_preflight_purity("prgs")
|
||||
self.assertIn("stable control checkout", str(ctx.exception))
|
||||
|
||||
@mock.patch("subprocess.run")
|
||||
@mock.patch("os.path.isdir", return_value=True)
|
||||
@mock.patch("os.path.exists", return_value=True)
|
||||
def test_pr487_style_merge_binds_clean_merger_workspace(
|
||||
self, _exists, _isdir, mock_run
|
||||
):
|
||||
mock_run.return_value = MagicMock(returncode=0, stdout=f"{CONTROL_ROOT}/.git\n")
|
||||
os.environ[nwb.AUTHOR_WORKTREE_ENV] = AUTHOR_DIRTY
|
||||
srv._preflight_resolved_role = "reviewer"
|
||||
with mock.patch.object(srv, "PROJECT_ROOT", MCP_PROCESS_ROOT):
|
||||
with mock.patch("gitea_mcp_server.get_profile", return_value=self._merger_profile()):
|
||||
resolved = srv._verify_role_mutation_workspace("prgs")
|
||||
self.assertEqual(resolved, os.path.realpath(MCP_PROCESS_ROOT))
|
||||
@@ -0,0 +1,169 @@
|
||||
"""Tests for post-merge cleanup proof enforcement (#402)."""
|
||||
import sys
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
from final_report_validator import assess_final_report_validator # noqa: E402
|
||||
from post_merge_cleanup_proof import ( # noqa: E402
|
||||
CLEANUP_SKIPPED,
|
||||
assess_post_merge_cleanup_proof,
|
||||
)
|
||||
|
||||
MERGE_SHA = "a" * 40
|
||||
HEAD_BRANCH = "feat/issue-274-branches-only-worktrees"
|
||||
WORKTREE = "branches/review-pr374-conflicts"
|
||||
|
||||
|
||||
def _full_remote_cleanup_report(**overrides):
|
||||
fields = {
|
||||
"Task": "review PR #374",
|
||||
"Merge result": "merged",
|
||||
"Merge commit SHA": MERGE_SHA,
|
||||
"Cleanup status": "remote branch deleted",
|
||||
"Delete-branch capability resolved": "gitea.branch.delete allowed via delete_branch task",
|
||||
"Merged PR head branch": HEAD_BRANCH,
|
||||
"Deleted branch": HEAD_BRANCH,
|
||||
"Branch protection": "none",
|
||||
"Open PR inventory proof": "no other open PR references branch (inventory complete)",
|
||||
"Active heartbeat/claim/lease": "none",
|
||||
"Cleanup mutations": "gitea_delete_branch on remote head branch",
|
||||
}
|
||||
fields.update(overrides)
|
||||
lines = ["## Controller Handoff", ""]
|
||||
lines.extend(f"- {key}: {value}" for key, value in fields.items())
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
def _full_worktree_cleanup_report(**overrides):
|
||||
fields = {
|
||||
"Task": "review PR #374",
|
||||
"Merge result": "merged",
|
||||
"Cleanup status": "local worktree removed",
|
||||
"Removed worktree path": WORKTREE,
|
||||
"Pre-removal tracked state": "clean",
|
||||
"Pre-removal untracked state": "clean",
|
||||
"Git worktree list after removal": "only main checkout listed",
|
||||
"Cleanup mutations": "git worktree remove on session-owned review worktree",
|
||||
}
|
||||
fields.update(overrides)
|
||||
lines = ["## Controller Handoff", ""]
|
||||
lines.extend(f"- {key}: {value}" for key, value in fields.items())
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
class TestPostMergeCleanupProof(unittest.TestCase):
|
||||
def test_no_cleanup_claim_passes(self):
|
||||
report = "## Controller Handoff\n- Task: review PR #1\n- Merge result: merged"
|
||||
result = assess_post_merge_cleanup_proof(report)
|
||||
self.assertFalse(result["block"])
|
||||
|
||||
def test_missing_capability_proof_blocked(self):
|
||||
report = _full_remote_cleanup_report(
|
||||
**{"Delete-branch capability resolved": "delete succeeded"}
|
||||
)
|
||||
result = assess_post_merge_cleanup_proof(report)
|
||||
self.assertTrue(result["block"])
|
||||
self.assertTrue(any("capability" in r.lower() for r in result["reasons"]))
|
||||
|
||||
def test_unmerged_pr_blocked(self):
|
||||
report = _full_remote_cleanup_report(**{"Merge result": "not merged"})
|
||||
result = assess_post_merge_cleanup_proof(report)
|
||||
self.assertTrue(result["block"])
|
||||
self.assertTrue(any("merge result" in r.lower() for r in result["reasons"]))
|
||||
|
||||
def test_wrong_branch_blocked(self):
|
||||
report = _full_remote_cleanup_report(
|
||||
**{"Deleted branch": "feat/other-branch"}
|
||||
)
|
||||
report += "\nDeleted branch does not match merged PR head branch"
|
||||
result = assess_post_merge_cleanup_proof(report)
|
||||
self.assertTrue(result["block"])
|
||||
|
||||
def test_protected_branch_blocked(self):
|
||||
report = _full_remote_cleanup_report(**{"Branch protection": "enabled"})
|
||||
result = assess_post_merge_cleanup_proof(report)
|
||||
self.assertTrue(result["block"])
|
||||
|
||||
def test_open_pr_reference_blocked(self):
|
||||
report = _full_remote_cleanup_report(
|
||||
**{"Open PR inventory proof": "PR #999 still references branch"}
|
||||
)
|
||||
result = assess_post_merge_cleanup_proof(report)
|
||||
self.assertTrue(result["block"])
|
||||
|
||||
def test_active_claim_blocked(self):
|
||||
report = _full_remote_cleanup_report(
|
||||
**{"Active heartbeat/claim/lease": "status:in-progress on linked issue"}
|
||||
)
|
||||
result = assess_post_merge_cleanup_proof(report)
|
||||
self.assertTrue(result["block"])
|
||||
|
||||
def test_dirty_worktree_blocked(self):
|
||||
report = _full_worktree_cleanup_report(
|
||||
**{"Pre-removal tracked state": "dirty"}
|
||||
)
|
||||
result = assess_post_merge_cleanup_proof(report)
|
||||
self.assertTrue(result["block"])
|
||||
self.assertTrue(any("tracked" in r.lower() for r in result["reasons"]))
|
||||
|
||||
def test_foreign_worktree_blocked(self):
|
||||
report = _full_worktree_cleanup_report(
|
||||
**{"Removed worktree path": "/tmp/foreign-worktree"}
|
||||
)
|
||||
result = assess_post_merge_cleanup_proof(report)
|
||||
self.assertTrue(result["block"])
|
||||
self.assertTrue(any("branches/" in r.lower() for r in result["reasons"]))
|
||||
|
||||
def test_cleanup_skipped_with_blocker_passes(self):
|
||||
report = "\n".join([
|
||||
"## Controller Handoff",
|
||||
"- Cleanup outcome: CLEANUP_SKIPPED",
|
||||
"- Cleanup blocker: gitea.branch.delete capability not resolved in active profile",
|
||||
])
|
||||
result = assess_post_merge_cleanup_proof(report)
|
||||
self.assertFalse(result["block"])
|
||||
self.assertEqual(result["outcome"], CLEANUP_SKIPPED)
|
||||
|
||||
def test_cleanup_skipped_without_blocker_blocked(self):
|
||||
report = "## Controller Handoff\n- Cleanup outcome: CLEANUP_SKIPPED"
|
||||
result = assess_post_merge_cleanup_proof(report)
|
||||
self.assertTrue(result["block"])
|
||||
|
||||
def test_full_remote_cleanup_passes(self):
|
||||
result = assess_post_merge_cleanup_proof(_full_remote_cleanup_report())
|
||||
self.assertFalse(result["block"])
|
||||
self.assertTrue(result["remote_delete_claimed"])
|
||||
|
||||
def test_full_worktree_cleanup_passes(self):
|
||||
result = assess_post_merge_cleanup_proof(_full_worktree_cleanup_report())
|
||||
self.assertFalse(result["block"])
|
||||
self.assertTrue(result["worktree_remove_claimed"])
|
||||
|
||||
def test_validator_integration_blocks_unproven_delete(self):
|
||||
report = (
|
||||
"## Controller Handoff\n"
|
||||
"- Cleanup mutations: gitea_delete_branch deleted remote branch\n"
|
||||
)
|
||||
result = assess_final_report_validator(report, task_kind="review_pr")
|
||||
self.assertTrue(result["blocked"])
|
||||
rule_ids = [f["rule_id"] for f in result["findings"]]
|
||||
self.assertIn("reviewer.post_merge_cleanup_proof", rule_ids)
|
||||
|
||||
def test_validator_integration_allows_skipped(self):
|
||||
report = "\n".join([
|
||||
"## Controller Handoff",
|
||||
"- Cleanup outcome: CLEANUP_SKIPPED",
|
||||
"- Cleanup blocker: branch still referenced by open PR #414",
|
||||
])
|
||||
result = assess_final_report_validator(report, task_kind="review_pr")
|
||||
cleanup_findings = [
|
||||
f for f in result["findings"]
|
||||
if f["rule_id"] == "reviewer.post_merge_cleanup_proof"
|
||||
]
|
||||
self.assertEqual(cleanup_findings, [])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user