Merge pull request 'feat: keep reconciliation audits read-only unless cleanup authorized (Closes #419)' (#421) from feat/issue-419-audit-readonly-mode into master
This commit was merged in pull request #421.
This commit is contained in:
@@ -0,0 +1,291 @@
|
||||
"""Tests for audit vs cleanup reconciliation mode (#419)."""
|
||||
import os
|
||||
import sys
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
|
||||
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
|
||||
|
||||
import audit_reconciliation_mode as arm
|
||||
import mcp_server
|
||||
from audit_reconciliation_mode import (
|
||||
AUDIT_FORBIDDEN_TASKS,
|
||||
PHASE_AUDIT,
|
||||
PHASE_CLEANUP,
|
||||
assess_audit_command_allowed,
|
||||
assess_audit_reconciliation_report,
|
||||
authorize_cleanup_phase,
|
||||
check_audit_mutation_allowed,
|
||||
check_cleanup_execution_allowed,
|
||||
classify_cleanup_mutation,
|
||||
clear_phase,
|
||||
enter_audit_phase,
|
||||
)
|
||||
from final_report_validator import assess_final_report_validator
|
||||
from review_proofs import assess_audit_reconciliation_report as proofs_assess
|
||||
from task_capability_map import required_permission, required_role
|
||||
|
||||
DELETE_PROFILE = {
|
||||
"profile_name": "prgs-author-delete",
|
||||
"allowed_operations": ["gitea.read", "gitea.branch.delete"],
|
||||
"forbidden_operations": [],
|
||||
"audit_label": "prgs-author-delete",
|
||||
}
|
||||
|
||||
READ_PROFILE = {
|
||||
"profile_name": "prgs-author",
|
||||
"allowed_operations": ["gitea.read", "gitea.issue.comment"],
|
||||
"forbidden_operations": ["gitea.branch.delete"],
|
||||
"audit_label": "prgs-author",
|
||||
}
|
||||
|
||||
READ_ENV = {
|
||||
"GITEA_MCP_CONFIG": os.path.join(
|
||||
os.path.dirname(__file__), "..", "profiles.json"
|
||||
),
|
||||
"GITEA_MCP_PROFILE": "prgs-author",
|
||||
}
|
||||
|
||||
|
||||
def _authorize_cleanup(**kwargs):
|
||||
defaults = {
|
||||
"operator_approved": True,
|
||||
"delete_capability_proven": True,
|
||||
"safety_proof": {"safe_to_delete_remote": True},
|
||||
"before_after_snapshot": {
|
||||
"before": "remote branch exists",
|
||||
"after": "remote branch absent",
|
||||
},
|
||||
}
|
||||
defaults.update(kwargs)
|
||||
return authorize_cleanup_phase(**defaults)
|
||||
|
||||
|
||||
def _cleanup_report(**overrides):
|
||||
base = (
|
||||
"Task mode: reconcile-landed-pr\n"
|
||||
"Workflow source: workflows/reconcile-landed-pr.md\n"
|
||||
"Audit phase: read-only assessment\n"
|
||||
"Cleanup phase authorized: true\n"
|
||||
"Delete-branch capability proven: true\n"
|
||||
"Branch safe to remove: true\n"
|
||||
"Before/after state snapshot: remote branch feat/x present → absent\n"
|
||||
"External-state mutations: deleted remote branch feat/x\n"
|
||||
"Git ref mutations: none\n"
|
||||
"Cleanup mutations: removed worktree branches/feat-x\n"
|
||||
)
|
||||
for key, value in overrides.items():
|
||||
base = base.replace(key, value)
|
||||
return base
|
||||
|
||||
|
||||
class TestAuditPhaseGates(unittest.TestCase):
|
||||
def setUp(self):
|
||||
clear_phase()
|
||||
enter_audit_phase("reconcile-landed-pr")
|
||||
|
||||
def tearDown(self):
|
||||
clear_phase()
|
||||
|
||||
def test_audit_blocks_delete_branch_task(self):
|
||||
allowed, reasons = check_audit_mutation_allowed("delete_branch")
|
||||
self.assertFalse(allowed)
|
||||
self.assertTrue(reasons)
|
||||
|
||||
def test_audit_blocks_worktree_shell_commands(self):
|
||||
allowed, reasons = assess_audit_command_allowed(
|
||||
"git worktree remove branches/feat-x"
|
||||
)
|
||||
self.assertFalse(allowed)
|
||||
self.assertTrue(reasons)
|
||||
|
||||
def test_audit_blocks_local_branch_delete_command(self):
|
||||
allowed, reasons = assess_audit_command_allowed("git branch -D feat/x")
|
||||
self.assertFalse(allowed)
|
||||
|
||||
def test_audit_allows_read_tasks(self):
|
||||
allowed, _ = check_audit_mutation_allowed("reconcile_landed_pr")
|
||||
self.assertTrue(allowed)
|
||||
|
||||
def test_forbidden_set_covers_issue_and_pr_mutations(self):
|
||||
for task in ("close_pr", "comment_issue", "commit_files", "push_branch"):
|
||||
self.assertIn(task, AUDIT_FORBIDDEN_TASKS)
|
||||
|
||||
|
||||
class TestCleanupAuthorization(unittest.TestCase):
|
||||
def setUp(self):
|
||||
clear_phase()
|
||||
enter_audit_phase("reconcile_merged_cleanups")
|
||||
|
||||
def tearDown(self):
|
||||
clear_phase()
|
||||
|
||||
def test_cleanup_without_approval_blocked(self):
|
||||
result = authorize_cleanup_phase(
|
||||
delete_capability_proven=True,
|
||||
safety_proof={"safe_to_delete_remote": True},
|
||||
before_after_snapshot={"before": "a", "after": "b"},
|
||||
)
|
||||
self.assertFalse(result["authorized"])
|
||||
|
||||
def test_cleanup_without_capability_proof_blocked(self):
|
||||
result = _authorize_cleanup(delete_capability_proven=False)
|
||||
self.assertFalse(result["authorized"])
|
||||
|
||||
def test_cleanup_without_snapshot_blocked(self):
|
||||
result = _authorize_cleanup(before_after_snapshot={"before": "", "after": ""})
|
||||
self.assertFalse(result["authorized"])
|
||||
|
||||
def test_authorized_cleanup_switches_phase(self):
|
||||
result = _authorize_cleanup()
|
||||
self.assertTrue(result["authorized"])
|
||||
self.assertEqual(result["phase"], PHASE_CLEANUP)
|
||||
|
||||
def test_cleanup_execution_allowed_only_after_authorization(self):
|
||||
self.assertFalse(check_cleanup_execution_allowed()[0])
|
||||
_authorize_cleanup()
|
||||
self.assertTrue(check_cleanup_execution_allowed()[0])
|
||||
|
||||
|
||||
class TestReportVerifier(unittest.TestCase):
|
||||
def test_false_no_mutations_after_cleanup_blocked(self):
|
||||
report = (
|
||||
"Task mode: reconcile-landed-pr\n"
|
||||
"No mutations performed.\n"
|
||||
"delete_remote_branch feat/dup\n"
|
||||
)
|
||||
result = assess_audit_reconciliation_report(report)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertIn("no mutations", result["reasons"][0].lower())
|
||||
|
||||
def test_cleanup_without_authorization_fields_blocked(self):
|
||||
report = (
|
||||
"Task mode: reconcile-landed-pr\n"
|
||||
"remove_local_worktree branches/feat-x\n"
|
||||
)
|
||||
result = assess_audit_reconciliation_report(report)
|
||||
self.assertFalse(result["proven"])
|
||||
|
||||
def test_authorized_cleanup_report_passes(self):
|
||||
result = assess_audit_reconciliation_report(_cleanup_report())
|
||||
self.assertTrue(result["proven"])
|
||||
|
||||
def test_mutation_classification_enforced(self):
|
||||
report = (
|
||||
"Task mode: reconcile-landed-pr\n"
|
||||
"Cleanup phase authorized: true\n"
|
||||
"Delete-branch capability proven: true\n"
|
||||
"Branch safe to remove: true\n"
|
||||
"Before/after state snapshot: present\n"
|
||||
"remove_local_worktree branches/feat-x\n"
|
||||
)
|
||||
result = assess_audit_reconciliation_report(report)
|
||||
self.assertFalse(result["proven"])
|
||||
|
||||
def test_proofs_export_matches_module(self):
|
||||
report = "No mutations performed.\ndelete_remote_branch feat/dup"
|
||||
self.assertEqual(
|
||||
proofs_assess(report)["proven"],
|
||||
assess_audit_reconciliation_report(report)["proven"],
|
||||
)
|
||||
|
||||
def test_final_report_validator_includes_boundary_rule(self):
|
||||
report = "No mutations performed.\ndelete_remote_branch feat/dup"
|
||||
result = assess_final_report_validator(
|
||||
report_text=report,
|
||||
task_kind="reconcile_already_landed",
|
||||
)
|
||||
self.assertTrue(result["blocked"])
|
||||
rule_ids = [f["rule_id"] for f in result["findings"]]
|
||||
self.assertIn("reconcile.audit_cleanup_boundary", rule_ids)
|
||||
|
||||
|
||||
class TestMutationClassification(unittest.TestCase):
|
||||
def test_remote_delete_is_external_state(self):
|
||||
self.assertEqual(
|
||||
classify_cleanup_mutation("delete_remote_branch"),
|
||||
"external-state",
|
||||
)
|
||||
|
||||
def test_worktree_remove_is_cleanup(self):
|
||||
self.assertEqual(
|
||||
classify_cleanup_mutation("remove_local_worktree"),
|
||||
"cleanup",
|
||||
)
|
||||
|
||||
|
||||
class TestMcpGates(unittest.TestCase):
|
||||
def setUp(self):
|
||||
clear_phase()
|
||||
enter_audit_phase("reconcile_merged_cleanups")
|
||||
self.mock_api = patch("mcp_server.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()
|
||||
|
||||
@patch.dict(os.environ, READ_ENV, clear=True)
|
||||
@patch("mcp_server.get_profile", return_value=DELETE_PROFILE)
|
||||
def test_delete_branch_blocked_in_audit_phase(self, _profile):
|
||||
mcp_server.record_preflight_check("whoami")
|
||||
mcp_server.record_preflight_check("capability", resolved_role="author")
|
||||
result = mcp_server.gitea_delete_branch(branch="feat/dup", remote="prgs")
|
||||
self.assertFalse(result["success"])
|
||||
self.assertEqual(result["audit_phase"], PHASE_AUDIT)
|
||||
self.mock_api.assert_not_called()
|
||||
|
||||
@patch.dict(os.environ, READ_ENV, clear=True)
|
||||
@patch("mcp_server.get_profile", return_value=DELETE_PROFILE)
|
||||
def test_reconcile_execute_blocked_without_cleanup_auth(self, _profile):
|
||||
with self.assertRaises(ValueError):
|
||||
mcp_server.gitea_reconcile_merged_cleanups(
|
||||
dry_run=False,
|
||||
execute_confirmed=False,
|
||||
remote="prgs",
|
||||
)
|
||||
|
||||
@patch.dict(os.environ, READ_ENV, clear=True)
|
||||
@patch("mcp_server.get_profile", return_value=READ_PROFILE)
|
||||
def test_cleanup_auth_fails_without_delete_capability(self, _profile):
|
||||
result = mcp_server.gitea_authorize_reconciliation_cleanup_phase(
|
||||
operator_approved=True,
|
||||
delete_capability_proven=True,
|
||||
safe_to_delete_remote=True,
|
||||
before_state="exists",
|
||||
after_state="gone",
|
||||
)
|
||||
self.assertFalse(result["authorized"])
|
||||
self.assertFalse(result["delete_capability_verified"])
|
||||
|
||||
@patch.dict(os.environ, READ_ENV, clear=True)
|
||||
@patch("mcp_server.get_profile", return_value=DELETE_PROFILE)
|
||||
def test_delete_branch_allowed_after_cleanup_authorization(self, _profile):
|
||||
mcp_server.gitea_authorize_reconciliation_cleanup_phase(
|
||||
operator_approved=True,
|
||||
delete_capability_proven=True,
|
||||
safe_to_delete_remote=True,
|
||||
before_state="exists",
|
||||
after_state="gone",
|
||||
)
|
||||
mcp_server.record_preflight_check("whoami")
|
||||
mcp_server.record_preflight_check("capability", resolved_role="author")
|
||||
self.mock_api.return_value = {}
|
||||
result = mcp_server.gitea_delete_branch(branch="feat/dup", remote="prgs")
|
||||
self.assertTrue(result["success"])
|
||||
|
||||
|
||||
class TestTaskCapabilityMap(unittest.TestCase):
|
||||
def test_reconciliation_cleanup_maps_delete_permission(self):
|
||||
self.assertEqual(
|
||||
required_permission("reconciliation_cleanup"),
|
||||
"gitea.branch.delete",
|
||||
)
|
||||
self.assertEqual(required_role("reconciliation_cleanup"), "author")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -87,6 +87,14 @@ def test_reconcile_landed_workflow_contract():
|
||||
assert "PARTIAL_RECONCILE_COMMENT_THEN_STOP" in text
|
||||
assert "RECOVERY_HANDOFF_ONLY" in text
|
||||
assert "resolve_partial_reconciliation_plan" in text
|
||||
assert "check_audit_mutation_allowed" in text
|
||||
assert "gitea_authorize_reconciliation_cleanup_phase" in text
|
||||
|
||||
|
||||
def test_audit_reconciliation_verifier_exported():
|
||||
from review_proofs import assess_audit_reconciliation_report
|
||||
|
||||
assert callable(assess_audit_reconciliation_report)
|
||||
|
||||
|
||||
def test_create_issue_workflow_contract():
|
||||
|
||||
Reference in New Issue
Block a user