sysadmin
e5bf8c9647
Make the preflight bypass test actually exercise purity (not test short-circuit), cover unmerged/open-head fail-closed cleanup, and document post-merge cleanup ownership for prgs-reconciler so authors are not required for that path.
264 lines
9.6 KiB
Python
264 lines
9.6 KiB
Python
"""Integration tests for reconciler merged cleanup (#523)."""
|
|
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 task_capability_map
|
|
from audit_reconciliation_mode import clear_phase
|
|
from task_capability_map import required_role
|
|
|
|
RECONCILER_PROFILE = {
|
|
"profile_name": "prgs-reconciler",
|
|
"context": "prgs",
|
|
"role": "reconciler",
|
|
"username": "sysadmin",
|
|
"base_url": "https://gitea.prgs.cc",
|
|
"allowed_operations": [
|
|
"gitea.read",
|
|
"gitea.pr.close",
|
|
"gitea.pr.comment",
|
|
"gitea.issue.comment",
|
|
],
|
|
"forbidden_operations": [],
|
|
}
|
|
|
|
AUTHOR_PROFILE = {
|
|
"profile_name": "prgs-author",
|
|
"context": "prgs",
|
|
"role": "author",
|
|
"username": "jcwalker3",
|
|
"base_url": "https://gitea.prgs.cc",
|
|
"allowed_operations": [
|
|
"gitea.read",
|
|
"gitea.pr.create",
|
|
"gitea.branch.create",
|
|
"gitea.branch.push",
|
|
"gitea.repo.commit",
|
|
],
|
|
"forbidden_operations": [],
|
|
}
|
|
|
|
CONTROL_ROOT = "/Users/jasonwalker/Development/Gitea-Tools"
|
|
|
|
|
|
class TestReconcilerCleanupIntegration(unittest.TestCase):
|
|
def setUp(self):
|
|
clear_phase()
|
|
mcp_server._preflight_whoami_called = False
|
|
mcp_server._preflight_capability_called = False
|
|
mcp_server._preflight_resolved_role = None
|
|
mcp_server._preflight_resolved_task = None
|
|
mcp_server._preflight_whoami_violation = False
|
|
mcp_server._preflight_capability_violation = False
|
|
mcp_server._preflight_whoami_violation_files = []
|
|
mcp_server._preflight_capability_violation_files = []
|
|
mcp_server._preflight_capability_baseline_porcelain = ""
|
|
self.mock_api = patch("gitea_auth.api_request").start()
|
|
self.mock_auth = patch(
|
|
"mcp_server.get_auth_header", return_value="token test"
|
|
).start()
|
|
|
|
def tearDown(self):
|
|
patch.stopall()
|
|
clear_phase()
|
|
mcp_server._IDENTITY_CACHE.clear()
|
|
|
|
def test_capability_map_routes_merged_cleanup_to_reconciler(self):
|
|
self.assertEqual(required_role("reconcile_merged_cleanups"), "reconciler")
|
|
self.assertEqual(required_role("reconciliation_cleanup"), "reconciler")
|
|
self.assertEqual(
|
|
task_capability_map.required_permission("reconcile_merged_cleanups"),
|
|
"gitea.read",
|
|
)
|
|
|
|
@patch.dict(os.environ, {"GITEA_PROFILE_NAME": "prgs-reconciler"}, clear=True)
|
|
@patch("mcp_server.get_profile", return_value=RECONCILER_PROFILE)
|
|
def test_reconciler_can_run_reconcile_merged_cleanups_directly(self, _profile):
|
|
mcp_server.record_preflight_check("whoami")
|
|
mcp_server.record_preflight_check(
|
|
"capability",
|
|
resolved_role="reconciler",
|
|
resolved_task="reconcile_merged_cleanups",
|
|
)
|
|
|
|
self.mock_api.side_effect = [
|
|
[ # closed pulls page
|
|
{
|
|
"number": 100,
|
|
"title": "merged feature",
|
|
"body": "Closes #100",
|
|
"merged": True,
|
|
"merged_at": "2026-07-06T12:00:00Z",
|
|
"merge_commit_sha": "deadbeef",
|
|
"head": {"ref": "feat/issue-100", "sha": "cafebabe"},
|
|
},
|
|
{
|
|
# closed but NOT merged — must not become a cleanup entry
|
|
"number": 101,
|
|
"title": "abandoned",
|
|
"body": "Closes #101",
|
|
"merged": False,
|
|
"merged_at": None,
|
|
"head": {"ref": "feat/issue-101", "sha": "badcafe"},
|
|
},
|
|
],
|
|
[], # open pulls
|
|
]
|
|
|
|
with patch("mcp_server._remote_branch_exists", return_value=True):
|
|
with patch(
|
|
"mcp_server.merged_cleanup_reconcile.is_head_ancestor_of_ref",
|
|
return_value=True,
|
|
):
|
|
result = mcp_server.gitea_reconcile_merged_cleanups(
|
|
dry_run=True,
|
|
remote="prgs",
|
|
)
|
|
self.assertTrue(result["success"])
|
|
self.assertTrue(result["dry_run"])
|
|
entry_numbers = [e["issue_number"] for e in result["entries"]]
|
|
self.assertEqual(entry_numbers, [100])
|
|
self.assertNotIn(101, entry_numbers)
|
|
|
|
@patch.dict(
|
|
os.environ,
|
|
{
|
|
"GITEA_PROFILE_NAME": "prgs-reconciler",
|
|
# Force preflight purity to evaluate (disable test short-circuit).
|
|
"GITEA_TEST_PORCELAIN": "",
|
|
},
|
|
clear=True,
|
|
)
|
|
@patch("mcp_server.get_profile", return_value=RECONCILER_PROFILE)
|
|
def test_reconciler_role_bypasses_branches_only_mutation_guard(self, _profile):
|
|
mcp_server.record_preflight_check("whoami")
|
|
mcp_server.record_preflight_check(
|
|
"capability",
|
|
resolved_role="reconciler",
|
|
resolved_task="reconcile_merged_cleanups",
|
|
)
|
|
|
|
with patch("mcp_server.PROJECT_ROOT", CONTROL_ROOT):
|
|
with patch(
|
|
"mcp_server.root_checkout_guard.resolve_remote_master_sha",
|
|
return_value="abc123",
|
|
):
|
|
with patch(
|
|
"mcp_server.issue_lock_worktree.read_worktree_git_state",
|
|
return_value={
|
|
"current_branch": "master",
|
|
"head_sha": "abc123",
|
|
"porcelain_status": "",
|
|
},
|
|
):
|
|
try:
|
|
mcp_server.verify_preflight_purity(
|
|
remote="prgs",
|
|
task="reconcile_merged_cleanups",
|
|
)
|
|
except RuntimeError as e:
|
|
self.fail(
|
|
"verify_preflight_purity raised RuntimeError "
|
|
f"unexpectedly for reconciler: {e}"
|
|
)
|
|
|
|
@patch.dict(
|
|
os.environ,
|
|
{"GITEA_PROFILE_NAME": "prgs-author", "GITEA_TEST_PORCELAIN": ""},
|
|
clear=True,
|
|
)
|
|
@patch("mcp_server.get_profile", return_value=AUTHOR_PROFILE)
|
|
def test_author_role_remains_blocked_on_root_checkout(self, _profile):
|
|
mcp_server.record_preflight_check("whoami")
|
|
mcp_server.record_preflight_check(
|
|
"capability",
|
|
resolved_role="author",
|
|
resolved_task="reconcile_merged_cleanups",
|
|
)
|
|
|
|
with patch("mcp_server.PROJECT_ROOT", CONTROL_ROOT):
|
|
with patch(
|
|
"mcp_server.root_checkout_guard.resolve_remote_master_sha",
|
|
return_value="abc123",
|
|
):
|
|
with patch(
|
|
"mcp_server.issue_lock_worktree.read_worktree_git_state",
|
|
return_value={
|
|
"current_branch": "master",
|
|
"head_sha": "abc123",
|
|
"porcelain_status": "",
|
|
},
|
|
):
|
|
with self.assertRaises(RuntimeError) as ctx:
|
|
mcp_server.verify_preflight_purity(
|
|
remote="prgs",
|
|
task="reconcile_merged_cleanups",
|
|
)
|
|
message = str(ctx.exception)
|
|
self.assertTrue(
|
|
"stable control checkout" in message
|
|
or "author mutation blocked" in message
|
|
or "branches/" in message,
|
|
msg=message,
|
|
)
|
|
|
|
@patch.dict(os.environ, {"GITEA_PROFILE_NAME": "prgs-reconciler"}, clear=True)
|
|
@patch("mcp_server.get_profile", return_value=RECONCILER_PROFILE)
|
|
def test_open_unmerged_pr_heads_are_not_safe_cleanup_targets(self, _profile):
|
|
"""AC5: active/unmerged author work must remain blocked from cleanup."""
|
|
mcp_server.record_preflight_check("whoami")
|
|
mcp_server.record_preflight_check(
|
|
"capability",
|
|
resolved_role="reconciler",
|
|
resolved_task="reconcile_merged_cleanups",
|
|
)
|
|
|
|
open_pr = {
|
|
"number": 200,
|
|
"title": "active work",
|
|
"body": "Closes #200",
|
|
"merged": False,
|
|
"state": "open",
|
|
"head": {"ref": "feat/issue-200", "sha": "livehead1"},
|
|
}
|
|
# Same head branch still open on another PR while a closed/merged PR
|
|
# used the same branch name historically — open head must block delete.
|
|
closed_merged = {
|
|
"number": 199,
|
|
"title": "older merge same branch name",
|
|
"body": "Closes #199",
|
|
"merged": True,
|
|
"merged_at": "2026-07-01T12:00:00Z",
|
|
"merge_commit_sha": "deadbeef",
|
|
"head": {"ref": "feat/issue-200", "sha": "oldhead99"},
|
|
}
|
|
self.mock_api.side_effect = [
|
|
[closed_merged], # closed
|
|
[open_pr], # open
|
|
]
|
|
|
|
with patch("mcp_server._remote_branch_exists", return_value=True):
|
|
with patch(
|
|
"mcp_server.merged_cleanup_reconcile.is_head_ancestor_of_ref",
|
|
return_value=True,
|
|
):
|
|
result = mcp_server.gitea_reconcile_merged_cleanups(
|
|
dry_run=True,
|
|
remote="prgs",
|
|
)
|
|
self.assertTrue(result["success"])
|
|
self.assertEqual(len(result["entries"]), 1)
|
|
remote_assessment = result["entries"][0].get("remote_branch") or {}
|
|
self.assertFalse(
|
|
remote_assessment.get("safe_to_delete_remote"),
|
|
msg=f"open head must block remote delete: {remote_assessment}",
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|