fix: resolve conflicts for PR #385
Merge prgs/master and keep work-issue author routing (#139) alongside reconciler close task routing (#309).
This commit is contained in:
@@ -78,6 +78,12 @@ def test_reconcile_landed_workflow_contract():
|
||||
assert "canonical: true" in text
|
||||
assert "Do not review or merge normal PRs" in text
|
||||
assert "Already-landed proof" in text
|
||||
# Issue #302: partial reconciliation policy must stay documented.
|
||||
assert "## 12A. Partial reconciliation policy (#302)" in text
|
||||
assert "comment-then-stop" in text
|
||||
assert "PARTIAL_RECONCILE_COMMENT_THEN_STOP" in text
|
||||
assert "RECOVERY_HANDOFF_ONLY" in text
|
||||
assert "resolve_partial_reconciliation_plan" in text
|
||||
|
||||
|
||||
def test_create_issue_workflow_contract():
|
||||
@@ -174,6 +180,18 @@ def test_validation_worktree_edit_verifier_exported():
|
||||
assert callable(assess_validation_worktree_edit_report)
|
||||
|
||||
|
||||
def test_inventory_worktree_verifier_exported():
|
||||
from review_proofs import assess_inventory_worktree_report
|
||||
|
||||
assert callable(assess_inventory_worktree_report)
|
||||
|
||||
|
||||
def test_infra_stop_handoff_verifier_exported():
|
||||
from review_proofs import assess_infra_stop_handoff_report
|
||||
|
||||
assert callable(assess_infra_stop_handoff_report)
|
||||
|
||||
|
||||
def test_mutation_categories_verifier_exported():
|
||||
from review_proofs import assess_mutation_categories_report
|
||||
|
||||
|
||||
@@ -0,0 +1,121 @@
|
||||
"""Tests for native MCP preference gate and shell circuit breaker (#270)."""
|
||||
import sys
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
import native_mcp_preference as nmp # noqa: E402
|
||||
|
||||
|
||||
class TestShellHealthCircuitBreaker(unittest.TestCase):
|
||||
def setUp(self):
|
||||
nmp.clear_shell_health_for_tests()
|
||||
|
||||
def test_two_spawn_failures_hard_stop(self):
|
||||
nmp.record_shell_spawn_outcome(exit_code=-1, stdout="", stderr="")
|
||||
status = nmp.record_shell_spawn_outcome(exit_code=-1, stdout="", stderr="")
|
||||
self.assertTrue(status["hard_stopped"])
|
||||
self.assertFalse(status["shell_use_allowed"])
|
||||
self.assertEqual(status["consecutive_spawn_failures"], 2)
|
||||
|
||||
def test_success_resets_breaker(self):
|
||||
nmp.record_shell_spawn_outcome(exit_code=-1, stdout="", stderr="")
|
||||
status = nmp.record_shell_spawn_outcome(exit_code=0, stdout="ok", stderr="")
|
||||
self.assertFalse(status["hard_stopped"])
|
||||
self.assertEqual(status["consecutive_spawn_failures"], 0)
|
||||
|
||||
|
||||
class TestNativeMcpPreferenceGate(unittest.TestCase):
|
||||
def setUp(self):
|
||||
nmp.clear_shell_health_for_tests()
|
||||
|
||||
def test_mcp_available_blocks_local_script(self):
|
||||
result = nmp.assess_gitea_operation_path(
|
||||
task="create_pr",
|
||||
command_or_detail="python create_pr.py --remote prgs --title T --head h",
|
||||
mcp_available=True,
|
||||
mcp_tool_visible=True,
|
||||
)
|
||||
self.assertTrue(result["block"])
|
||||
self.assertEqual(result["path_kind"], "shell_script")
|
||||
|
||||
def test_mcp_native_path_allowed(self):
|
||||
result = nmp.assess_gitea_operation_path(
|
||||
task="comment_issue",
|
||||
path_kind="mcp_native",
|
||||
mcp_available=True,
|
||||
mcp_tool_visible=True,
|
||||
)
|
||||
self.assertFalse(result["block"])
|
||||
self.assertTrue(result["allowed"])
|
||||
|
||||
def test_mcp_unavailable_requires_terminal_report(self):
|
||||
result = nmp.assess_gitea_operation_path(
|
||||
task="merge_pr",
|
||||
path_kind="shell_script",
|
||||
mcp_available=False,
|
||||
mcp_tool_visible=False,
|
||||
)
|
||||
self.assertTrue(result["block"])
|
||||
self.assertTrue(
|
||||
any("terminal recovery report" in r for r in result["reasons"])
|
||||
)
|
||||
self.assertIn(nmp.TERMINAL_REPORT_HEADING, result["terminal_report"])
|
||||
|
||||
def test_recovery_fallback_allowed_with_proof(self):
|
||||
result = nmp.assess_gitea_operation_path(
|
||||
task="merge_pr",
|
||||
path_kind="shell_script",
|
||||
command_or_detail="Recovery mode: MCP tools unavailable in this session.",
|
||||
mcp_available=False,
|
||||
recovery_mode=True,
|
||||
recovery_proof_complete=True,
|
||||
)
|
||||
self.assertFalse(result["block"])
|
||||
|
||||
def test_cli_auth_divergence_blocked(self):
|
||||
result = nmp.assess_gitea_operation_path(
|
||||
task="create_pr",
|
||||
command_or_detail="GITEA_MCP_PROFILE=prgs-reviewer python create_pr.py",
|
||||
mcp_available=True,
|
||||
active_profile="prgs-author",
|
||||
)
|
||||
self.assertTrue(result["block"])
|
||||
self.assertTrue(any("CLI auth divergence" in r for r in result["reasons"]))
|
||||
|
||||
def test_unsafe_helper_fallback_denied(self):
|
||||
result = nmp.assess_gitea_operation_path(
|
||||
task="commit_files",
|
||||
command_or_detail="python _encode_payload.py",
|
||||
mcp_available=True,
|
||||
mcp_tool_visible=True,
|
||||
)
|
||||
self.assertTrue(result["block"])
|
||||
self.assertEqual(result["path_kind"], "unsafe_helper")
|
||||
|
||||
def test_reviewer_mcp_server_touch_blocked(self):
|
||||
result = nmp.assess_gitea_operation_path(
|
||||
task="review_pr",
|
||||
command_or_detail="pkill -f gitea_mcp_server.py",
|
||||
mcp_available=True,
|
||||
role="reviewer",
|
||||
active_profile="prgs-reviewer",
|
||||
)
|
||||
self.assertTrue(result["block"])
|
||||
self.assertEqual(result["path_kind"], "mcp_server_touch")
|
||||
|
||||
def test_hard_stopped_shell_blocks_non_mcp_path(self):
|
||||
nmp.record_shell_spawn_outcome(exit_code=-1, stdout="", stderr="")
|
||||
nmp.record_shell_spawn_outcome(exit_code=-1, stdout="", stderr="")
|
||||
result = nmp.assess_gitea_operation_path(
|
||||
task="lock_issue",
|
||||
path_kind="shell_script",
|
||||
mcp_available=True,
|
||||
)
|
||||
self.assertTrue(result["block"])
|
||||
self.assertTrue(any("circuit breaker" in r for r in result["reasons"]))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,188 @@
|
||||
"""Issue #309: dedicated reconciler capability to close already-landed PRs.
|
||||
|
||||
The reconciler path must prove exact ``gitea.pr.close`` capability, must
|
||||
never imply review/approve/request-changes/merge capability, and may close
|
||||
a PR only after full already-landed proof.
|
||||
"""
|
||||
|
||||
import sys
|
||||
import unittest
|
||||
|
||||
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
|
||||
|
||||
import role_session_router # noqa: E402
|
||||
import task_capability_map # noqa: E402
|
||||
from review_proofs import assess_reconciler_close_gate # noqa: E402
|
||||
|
||||
PINNED = "0fdc8f582026b72a229d59a172c0a63ac4aaeaf9"
|
||||
OTHER = "a4060c5de00f2b1c9e88f4f6f0f3f9a7b2c1d0e9"
|
||||
|
||||
|
||||
class TestReconcilerCapabilityMap(unittest.TestCase):
|
||||
def test_reconcile_close_pr_task_maps_to_pr_close(self):
|
||||
self.assertEqual(
|
||||
task_capability_map.required_permission(
|
||||
"reconcile_close_landed_pr"),
|
||||
"gitea.pr.close",
|
||||
)
|
||||
self.assertEqual(
|
||||
task_capability_map.required_role("reconcile_close_landed_pr"),
|
||||
"reconciler",
|
||||
)
|
||||
|
||||
def test_reconcile_close_issue_task_maps_to_issue_close(self):
|
||||
self.assertEqual(
|
||||
task_capability_map.required_permission(
|
||||
"reconcile_close_landed_issue"),
|
||||
"gitea.issue.close",
|
||||
)
|
||||
self.assertEqual(
|
||||
task_capability_map.required_role(
|
||||
"reconcile_close_landed_issue"),
|
||||
"reconciler",
|
||||
)
|
||||
|
||||
def test_reconciler_tasks_do_not_grant_review_permissions(self):
|
||||
for task in ("reconcile_close_landed_pr",
|
||||
"reconcile_close_landed_issue"):
|
||||
permission = task_capability_map.required_permission(task)
|
||||
self.assertNotIn("review", permission)
|
||||
self.assertNotIn("approve", permission)
|
||||
self.assertNotIn("merge", permission)
|
||||
self.assertNotIn("request_changes", permission)
|
||||
|
||||
|
||||
class TestReconcilerRouting(unittest.TestCase):
|
||||
def test_reconciler_task_requires_reconciler_role(self):
|
||||
self.assertEqual(
|
||||
role_session_router.required_role_for_task(
|
||||
"reconcile_close_landed_pr"),
|
||||
"reconciler",
|
||||
)
|
||||
|
||||
def test_reconciler_task_blocked_in_author_session(self):
|
||||
result = role_session_router.route_task_session(
|
||||
"reconcile_close_landed_pr",
|
||||
active_profile="prgs-author",
|
||||
active_role_kind="author",
|
||||
allowed_in_current_session=False,
|
||||
)
|
||||
self.assertEqual(result["route_result"], "wrong_role_stop")
|
||||
self.assertFalse(result["downstream_allowed"])
|
||||
|
||||
def test_reconciler_task_allowed_in_matching_session(self):
|
||||
result = role_session_router.route_task_session(
|
||||
"reconcile_close_landed_pr",
|
||||
active_profile="prgs-reconciler",
|
||||
active_role_kind="reconciler",
|
||||
allowed_in_current_session=True,
|
||||
)
|
||||
self.assertEqual(result["route_result"], "allowed_current_session")
|
||||
self.assertTrue(result["downstream_allowed"])
|
||||
|
||||
|
||||
class TestReconcilerCloseGate(unittest.TestCase):
|
||||
def _gate(self, **overrides):
|
||||
kwargs = {
|
||||
"pr_number": 278,
|
||||
"pr_state": "open",
|
||||
"candidate_head_sha": PINNED,
|
||||
"live_head_sha": PINNED,
|
||||
"target_branch": "master",
|
||||
"target_branch_sha": OTHER,
|
||||
"head_is_ancestor_of_target": True,
|
||||
"close_pr_capability": True,
|
||||
"close_issue_capability": False,
|
||||
"linked_issue_state": "closed",
|
||||
}
|
||||
kwargs.update(overrides)
|
||||
return assess_reconciler_close_gate(**kwargs)
|
||||
|
||||
def test_already_landed_open_pr_close_allowed(self):
|
||||
result = self._gate()
|
||||
self.assertEqual(result["outcome"], "CLOSE_ALLOWED")
|
||||
self.assertTrue(result["pr_close_allowed"])
|
||||
|
||||
def test_not_landed_pr_cannot_be_closed(self):
|
||||
result = self._gate(head_is_ancestor_of_target=False)
|
||||
self.assertEqual(result["outcome"], "NOT_LANDED_CLOSE_BLOCKED")
|
||||
self.assertFalse(result["pr_close_allowed"])
|
||||
|
||||
def test_unchecked_ancestry_blocks_close(self):
|
||||
result = self._gate(head_is_ancestor_of_target=None)
|
||||
self.assertEqual(result["outcome"], "GATE_NOT_PROVEN")
|
||||
self.assertFalse(result["pr_close_allowed"])
|
||||
|
||||
def test_stale_target_branch_sha_blocks_close(self):
|
||||
result = self._gate(target_branch_sha=None)
|
||||
self.assertEqual(result["outcome"], "GATE_NOT_PROVEN")
|
||||
self.assertFalse(result["pr_close_allowed"])
|
||||
|
||||
def test_closed_pr_state_blocks_close(self):
|
||||
result = self._gate(pr_state="closed")
|
||||
self.assertEqual(result["outcome"], "GATE_NOT_PROVEN")
|
||||
self.assertFalse(result["pr_close_allowed"])
|
||||
|
||||
def test_changed_head_blocks_close(self):
|
||||
result = self._gate(live_head_sha=OTHER)
|
||||
self.assertEqual(result["outcome"], "GATE_NOT_PROVEN")
|
||||
self.assertFalse(result["pr_close_allowed"])
|
||||
|
||||
def test_missing_close_capability_produces_recovery_handoff(self):
|
||||
result = self._gate(close_pr_capability=False)
|
||||
self.assertEqual(result["outcome"], "RECOVERY_HANDOFF_REQUIRED")
|
||||
self.assertFalse(result["pr_close_allowed"])
|
||||
self.assertTrue(any(
|
||||
"gitea.pr.close" in reason for reason in result["reasons"]
|
||||
))
|
||||
|
||||
def test_linked_issue_already_closed_skips_issue_close(self):
|
||||
result = self._gate(linked_issue_state="closed",
|
||||
close_issue_capability=True)
|
||||
self.assertFalse(result["issue_close_allowed"])
|
||||
|
||||
def test_linked_open_issue_close_requires_capability(self):
|
||||
allowed = self._gate(
|
||||
linked_issue_state="open",
|
||||
issue_resolved_by_landing=True,
|
||||
close_issue_capability=True,
|
||||
)
|
||||
self.assertTrue(allowed["issue_close_allowed"])
|
||||
|
||||
blocked = self._gate(
|
||||
linked_issue_state="open",
|
||||
issue_resolved_by_landing=True,
|
||||
close_issue_capability=False,
|
||||
)
|
||||
self.assertFalse(blocked["issue_close_allowed"])
|
||||
|
||||
def test_gate_never_allows_review_mutations(self):
|
||||
for result in (
|
||||
self._gate(),
|
||||
self._gate(head_is_ancestor_of_target=False),
|
||||
self._gate(close_pr_capability=False),
|
||||
):
|
||||
self.assertFalse(result["review_allowed"])
|
||||
self.assertFalse(result["approve_allowed"])
|
||||
self.assertFalse(result["request_changes_allowed"])
|
||||
self.assertFalse(result["merge_allowed"])
|
||||
|
||||
def test_report_fields_listed_for_close(self):
|
||||
result = self._gate()
|
||||
for field in (
|
||||
"identity/profile",
|
||||
"close capability proof",
|
||||
"PR live state",
|
||||
"candidate head SHA",
|
||||
"target branch SHA",
|
||||
"ancestor proof",
|
||||
"linked issue status",
|
||||
"PR close result",
|
||||
"issue close result",
|
||||
"no review/merge confirmation",
|
||||
):
|
||||
self.assertIn(field, result["required_report_fields"])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,162 @@
|
||||
"""Tests for the dedicated reconciliation controller-handoff schema (#303).
|
||||
|
||||
Already-landed PR reconciliation runs must emit the reconciliation
|
||||
schema, not stale author/reviewer handoff fields; observed git ref
|
||||
mutations (fetch) must be reported, and legacy fields fail validation.
|
||||
"""
|
||||
import sys
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
from review_proofs import assess_reconciliation_handoff # noqa: E402
|
||||
|
||||
|
||||
def _good_handoff(**overrides):
|
||||
fields = {
|
||||
"Task": "Reconcile already-landed open PRs",
|
||||
"Repo": "prgs/Scaled-Tech-Consulting/Gitea-Tools",
|
||||
"Role/profile": "prgs-reconciler",
|
||||
"Identity": "jcwalker3",
|
||||
"Selected PR": "278",
|
||||
"PR live state": "open",
|
||||
"Candidate head SHA": "2a544b7",
|
||||
"Target branch": "master",
|
||||
"Target branch SHA": "fa0cb12",
|
||||
"Ancestor proof": "candidate head is ancestor of target branch SHA",
|
||||
"Linked issue": "263",
|
||||
"Linked issue live status": "open (fetched live this session)",
|
||||
"Eligibility class": "ALREADY_LANDED_RECONCILE_REQUIRED",
|
||||
"Capabilities proven": "gitea.pr.comment",
|
||||
"Missing capabilities": "gitea.pr.close",
|
||||
"PR comments posted": "1 (PR #278 reconciliation notice)",
|
||||
"Issue comments posted": "none",
|
||||
"PRs closed": "none",
|
||||
"Issues closed": "none",
|
||||
"Git ref mutations": "git fetch prgs master",
|
||||
"Worktree mutations": "none",
|
||||
"MCP/Gitea mutations": "PR comment on #278",
|
||||
"External-state mutations": "none",
|
||||
"Read-only diagnostics": "gitea_view_pr 278, gitea_view_issue 263",
|
||||
"Blocker": "PR close capability missing",
|
||||
"Safe next action": "operator closes PR #278 or grants close capability",
|
||||
"No review/merge confirmation": "no review or merge mutations performed",
|
||||
}
|
||||
fields.update(overrides)
|
||||
lines = ["Controller Handoff"]
|
||||
lines += [f"- {k}: {v}" for k, v in fields.items() if v is not None]
|
||||
return "\n".join(lines) + "\n"
|
||||
|
||||
|
||||
class TestReconciliationSchemaPasses(unittest.TestCase):
|
||||
def test_blocked_reconciliation_passes(self):
|
||||
result = assess_reconciliation_handoff(
|
||||
_good_handoff(),
|
||||
observed_commands=["git fetch prgs master"],
|
||||
)
|
||||
self.assertTrue(result["complete"])
|
||||
self.assertEqual(result["reasons"], [])
|
||||
|
||||
def test_successful_pr_close_passes(self):
|
||||
report = _good_handoff(**{
|
||||
"Missing capabilities": "none",
|
||||
"PRs closed": "PR #278",
|
||||
"Blocker": "none",
|
||||
"Safe next action": "none; reconciliation complete",
|
||||
})
|
||||
result = assess_reconciliation_handoff(
|
||||
report, observed_commands=["git fetch prgs master"]
|
||||
)
|
||||
self.assertTrue(result["complete"])
|
||||
|
||||
def test_comment_only_reconciliation_passes(self):
|
||||
report = _good_handoff(**{
|
||||
"Git ref mutations": "none",
|
||||
})
|
||||
result = assess_reconciliation_handoff(report, observed_commands=[])
|
||||
self.assertTrue(result["complete"])
|
||||
|
||||
def test_noop_already_closed_passes(self):
|
||||
report = _good_handoff(**{
|
||||
"PR live state": "closed",
|
||||
"PR comments posted": "none",
|
||||
"MCP/Gitea mutations": "none",
|
||||
"Git ref mutations": "none",
|
||||
"Blocker": "none",
|
||||
"Safe next action": "none; PR already closed",
|
||||
})
|
||||
result = assess_reconciliation_handoff(report, observed_commands=[])
|
||||
self.assertTrue(result["complete"])
|
||||
|
||||
|
||||
class TestSchemaViolationsFail(unittest.TestCase):
|
||||
def test_missing_linked_issue_proof_fails(self):
|
||||
report = _good_handoff(**{"Linked issue live status": None})
|
||||
result = assess_reconciliation_handoff(report, observed_commands=[])
|
||||
self.assertFalse(result["complete"])
|
||||
self.assertTrue(
|
||||
any("linked issue live status" in r.lower()
|
||||
for r in result["reasons"])
|
||||
)
|
||||
|
||||
def test_wrong_eligibility_class_fails(self):
|
||||
report = _good_handoff(**{"Eligibility class": "REVIEW_ELIGIBLE"})
|
||||
result = assess_reconciliation_handoff(report, observed_commands=[])
|
||||
self.assertFalse(result["complete"])
|
||||
|
||||
def test_missing_handoff_section_fails(self):
|
||||
result = assess_reconciliation_handoff(
|
||||
"no handoff here", observed_commands=[]
|
||||
)
|
||||
self.assertFalse(result["complete"])
|
||||
|
||||
|
||||
class TestStaleFieldsFail(unittest.TestCase):
|
||||
def test_pr_number_opened_fails(self):
|
||||
report = _good_handoff() + "- PR number opened: 278\n"
|
||||
result = assess_reconciliation_handoff(report, observed_commands=[])
|
||||
self.assertFalse(result["complete"])
|
||||
self.assertTrue(
|
||||
any("pr number opened" in r.lower() for r in result["reasons"])
|
||||
)
|
||||
|
||||
def test_pinned_reviewed_head_fails(self):
|
||||
report = _good_handoff() + "- Pinned reviewed head: 2a544b7\n"
|
||||
result = assess_reconciliation_handoff(report, observed_commands=[])
|
||||
self.assertFalse(result["complete"])
|
||||
|
||||
def test_scratch_worktree_used_fails(self):
|
||||
report = _good_handoff() + "- Scratch worktree used: False\n"
|
||||
result = assess_reconciliation_handoff(report, observed_commands=[])
|
||||
self.assertFalse(result["complete"])
|
||||
|
||||
def test_workspace_mutations_fails(self):
|
||||
report = _good_handoff() + "- Workspace mutations: None\n"
|
||||
result = assess_reconciliation_handoff(report, observed_commands=[])
|
||||
self.assertFalse(result["complete"])
|
||||
|
||||
def test_author_lock_fields_fail(self):
|
||||
report = (
|
||||
_good_handoff()
|
||||
+ "- Issue lock proof: locked before diff\n"
|
||||
+ "- Claim/comment status: claimed\n"
|
||||
)
|
||||
result = assess_reconciliation_handoff(report, observed_commands=[])
|
||||
self.assertFalse(result["complete"])
|
||||
|
||||
|
||||
class TestObservedFetchMustBeReported(unittest.TestCase):
|
||||
def test_fetch_with_ref_mutations_none_fails(self):
|
||||
report = _good_handoff(**{"Git ref mutations": "none"})
|
||||
result = assess_reconciliation_handoff(
|
||||
report, observed_commands=["git fetch prgs master"]
|
||||
)
|
||||
self.assertFalse(result["complete"])
|
||||
self.assertTrue(
|
||||
any("git ref mutation" in r.lower() for r in result["reasons"])
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -26,6 +26,8 @@ from review_proofs import ( # noqa: E402
|
||||
assess_already_landed_report_state,
|
||||
assess_already_landed_review_gate,
|
||||
assess_author_pr_report,
|
||||
assess_partial_reconciliation_report,
|
||||
resolve_partial_reconciliation_plan,
|
||||
assess_validation_environment_proof,
|
||||
assess_full_suite_failure_approval_gate,
|
||||
assess_queue_ordering_proof,
|
||||
@@ -2690,6 +2692,133 @@ class TestReviewerBaselineValidationProof(unittest.TestCase):
|
||||
self.assertFalse(report["baseline_validation_proven"])
|
||||
|
||||
|
||||
class TestPartialReconciliationPlan(unittest.TestCase):
|
||||
"""Issue #302: policy when PR-close capability is missing (Option B)."""
|
||||
|
||||
def _plan(self, **overrides):
|
||||
kwargs = {
|
||||
"ancestor_proven": True,
|
||||
"capabilities": {
|
||||
"close_pr": False,
|
||||
"comment_pr": True,
|
||||
"close_issue": True,
|
||||
"comment_issue": True,
|
||||
},
|
||||
}
|
||||
kwargs.update(overrides)
|
||||
return resolve_partial_reconciliation_plan(**kwargs)
|
||||
|
||||
def test_close_capability_present_allows_full_reconcile(self):
|
||||
plan = self._plan(capabilities={
|
||||
"close_pr": True, "comment_pr": True,
|
||||
"close_issue": True, "comment_issue": True,
|
||||
})
|
||||
self.assertEqual(plan["outcome"], "FULL_RECONCILE_CLOSE_ALLOWED")
|
||||
self.assertIn("close_pr", plan["allowed_mutations"])
|
||||
|
||||
def test_close_missing_with_comment_posts_comment_then_stops(self):
|
||||
plan = self._plan()
|
||||
self.assertEqual(
|
||||
plan["outcome"], "PARTIAL_RECONCILE_COMMENT_THEN_STOP")
|
||||
self.assertEqual(plan["allowed_mutations"], ("comment_pr",))
|
||||
for field in (
|
||||
"PR head SHA",
|
||||
"target branch SHA",
|
||||
"ancestor proof",
|
||||
"linked issue status",
|
||||
"required missing capability",
|
||||
):
|
||||
self.assertIn(field, plan["required_comment_fields"])
|
||||
|
||||
def test_comment_capability_missing_produces_handoff_only(self):
|
||||
plan = self._plan(capabilities={
|
||||
"close_pr": False, "comment_pr": False,
|
||||
"close_issue": True, "comment_issue": True,
|
||||
})
|
||||
self.assertEqual(plan["outcome"], "RECOVERY_HANDOFF_ONLY")
|
||||
self.assertEqual(plan["allowed_mutations"], ())
|
||||
|
||||
def test_all_capabilities_missing_produces_handoff_only(self):
|
||||
plan = self._plan(capabilities={
|
||||
"close_pr": False, "comment_pr": False,
|
||||
"close_issue": False, "comment_issue": False,
|
||||
})
|
||||
self.assertEqual(plan["outcome"], "RECOVERY_HANDOFF_ONLY")
|
||||
self.assertEqual(plan["allowed_mutations"], ())
|
||||
|
||||
def test_unproven_ancestry_blocks_all_mutations(self):
|
||||
plan = self._plan(ancestor_proven=False, capabilities={
|
||||
"close_pr": True, "comment_pr": True,
|
||||
"close_issue": True, "comment_issue": True,
|
||||
})
|
||||
self.assertEqual(plan["outcome"], "GATE_NOT_PROVEN")
|
||||
self.assertEqual(plan["allowed_mutations"], ())
|
||||
|
||||
def test_missing_capability_dict_fails_closed(self):
|
||||
plan = self._plan(capabilities=None)
|
||||
self.assertEqual(plan["outcome"], "RECOVERY_HANDOFF_ONLY")
|
||||
self.assertEqual(plan["allowed_mutations"], ())
|
||||
|
||||
|
||||
class TestPartialReconciliationReport(unittest.TestCase):
|
||||
"""Issue #302: reports must explain why comments were or were not posted."""
|
||||
|
||||
def _plan(self, outcome):
|
||||
capabilities = {
|
||||
"FULL_RECONCILE_CLOSE_ALLOWED": {
|
||||
"close_pr": True, "comment_pr": True,
|
||||
"close_issue": True, "comment_issue": True,
|
||||
},
|
||||
"PARTIAL_RECONCILE_COMMENT_THEN_STOP": {
|
||||
"close_pr": False, "comment_pr": True,
|
||||
"close_issue": True, "comment_issue": True,
|
||||
},
|
||||
"RECOVERY_HANDOFF_ONLY": {
|
||||
"close_pr": False, "comment_pr": False,
|
||||
"close_issue": False, "comment_issue": False,
|
||||
},
|
||||
}[outcome]
|
||||
return resolve_partial_reconciliation_plan(
|
||||
ancestor_proven=True, capabilities=capabilities)
|
||||
|
||||
def test_partial_report_requires_comment_and_missing_capability(self):
|
||||
plan = self._plan("PARTIAL_RECONCILE_COMMENT_THEN_STOP")
|
||||
good = (
|
||||
"Reconciliation comment posted on PR #278 with ancestor proof. "
|
||||
"PR close skipped: close capability gitea.pr.close missing; "
|
||||
"stopped for authorized close."
|
||||
)
|
||||
result = assess_partial_reconciliation_report(good, plan=plan)
|
||||
self.assertTrue(result["complete"])
|
||||
|
||||
bad = "Stopped. Nothing done."
|
||||
result = assess_partial_reconciliation_report(bad, plan=plan)
|
||||
self.assertTrue(result["block"])
|
||||
|
||||
def test_handoff_only_report_requires_no_comment_explanation(self):
|
||||
plan = self._plan("RECOVERY_HANDOFF_ONLY")
|
||||
good = (
|
||||
"No reconciliation comment posted: comment capability missing. "
|
||||
"No Gitea mutation performed; recovery handoff produced."
|
||||
)
|
||||
result = assess_partial_reconciliation_report(good, plan=plan)
|
||||
self.assertTrue(result["complete"])
|
||||
|
||||
bad = "Recovery handoff produced."
|
||||
result = assess_partial_reconciliation_report(bad, plan=plan)
|
||||
self.assertTrue(result["block"])
|
||||
|
||||
def test_full_reconcile_report_requires_close_result(self):
|
||||
plan = self._plan("FULL_RECONCILE_CLOSE_ALLOWED")
|
||||
good = "PR close result: closed PR #278 after ancestor proof."
|
||||
result = assess_partial_reconciliation_report(good, plan=plan)
|
||||
self.assertTrue(result["complete"])
|
||||
|
||||
bad = "Reconciliation done."
|
||||
result = assess_partial_reconciliation_report(bad, plan=plan)
|
||||
self.assertTrue(result["block"])
|
||||
|
||||
|
||||
class TestAlreadyLandedReviewGate(unittest.TestCase):
|
||||
"""Issue #292: PR head already on target blocks approval and merge."""
|
||||
|
||||
|
||||
@@ -0,0 +1,106 @@
|
||||
"""Tests for infra-stop repair handoff verifier (#289)."""
|
||||
import sys
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
from capability_stop_terminal import TERMINAL_REPORT_HEADING
|
||||
from reviewer_infra_stop_handoff import assess_infra_stop_handoff_report # noqa: E402
|
||||
|
||||
|
||||
def _repair_handoff() -> str:
|
||||
return "\n".join([
|
||||
TERMINAL_REPORT_HEADING,
|
||||
"Repair handoff for infra_stop blocked reviewer workflow.",
|
||||
"CONTROL-CHECKOUT REPAIR MODE",
|
||||
"MCP process root: /Users/dev/Gitea-Tools",
|
||||
"Inspected git root: /Users/dev/Gitea-Tools",
|
||||
"Conflict marker path: gitea_mcp_server.py",
|
||||
"Merge/rebase control path: none",
|
||||
"Safe next repair action: resolve conflict markers and restart MCP",
|
||||
"infra_stop: true",
|
||||
])
|
||||
|
||||
|
||||
class TestInfraStopHandoff(unittest.TestCase):
|
||||
def test_repair_handoff_passes(self):
|
||||
result = assess_infra_stop_handoff_report(
|
||||
_repair_handoff(),
|
||||
stop_session={
|
||||
"infra_stop": True,
|
||||
"infra_stop_assessment": {
|
||||
"infra_stop": True,
|
||||
"conflict_file": "gitea_mcp_server.py",
|
||||
},
|
||||
},
|
||||
)
|
||||
self.assertTrue(result["proven"], result["reasons"])
|
||||
|
||||
def test_next_pr_selection_blocks(self):
|
||||
report = "\n".join([
|
||||
TERMINAL_REPORT_HEADING,
|
||||
"Next eligible PR to review: PR #276",
|
||||
"Pinned review head SHA: abc123",
|
||||
])
|
||||
result = assess_infra_stop_handoff_report(
|
||||
report,
|
||||
stop_session={"infra_stop": True, "require_infra_diagnostics": False},
|
||||
)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertTrue(any("queue" in r.lower() for r in result["reasons"]))
|
||||
|
||||
def test_missing_repair_handoff_blocks(self):
|
||||
result = assess_infra_stop_handoff_report(
|
||||
"Validation result: pass\nReview summary for PR #276",
|
||||
stop_session={"infra_stop": True},
|
||||
)
|
||||
self.assertFalse(result["proven"])
|
||||
|
||||
def test_main_checkout_diagnostics_without_label_blocks(self):
|
||||
report = "\n".join([
|
||||
TERMINAL_REPORT_HEADING,
|
||||
"Repair handoff",
|
||||
"Ran git status in main checkout for MCP diagnostics.",
|
||||
])
|
||||
result = assess_infra_stop_handoff_report(
|
||||
report,
|
||||
stop_session={
|
||||
"infra_stop": True,
|
||||
"main_checkout_diagnostics": True,
|
||||
"require_infra_diagnostics": False,
|
||||
},
|
||||
)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertTrue(any("control-checkout" in r.lower() for r in result["reasons"]))
|
||||
|
||||
def test_background_scheduling_blocks(self):
|
||||
report = "\n".join([
|
||||
TERMINAL_REPORT_HEADING,
|
||||
"Repair handoff",
|
||||
"Used schedule/manage_task while waiting for MCP recovery.",
|
||||
])
|
||||
result = assess_infra_stop_handoff_report(
|
||||
report,
|
||||
stop_session={"infra_stop": True, "require_infra_diagnostics": False},
|
||||
)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertTrue(any("schedule" in r.lower() for r in result["reasons"]))
|
||||
|
||||
def test_non_infra_stop_skips(self):
|
||||
result = assess_infra_stop_handoff_report(
|
||||
"Selected PR #1 for review",
|
||||
stop_session={"infra_stop": False},
|
||||
)
|
||||
self.assertTrue(result["proven"])
|
||||
|
||||
|
||||
class TestExport(unittest.TestCase):
|
||||
def test_review_proofs_reexport(self):
|
||||
from review_proofs import assess_infra_stop_handoff_report as exported
|
||||
|
||||
self.assertTrue(callable(exported))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,126 @@
|
||||
"""Tests for reviewer inventory/worktree verifier (#293)."""
|
||||
import sys
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
from reviewer_inventory_worktree import assess_inventory_worktree_report # noqa: E402
|
||||
|
||||
|
||||
def _good_handoff() -> str:
|
||||
return "\n".join([
|
||||
"## Controller Handoff",
|
||||
"- Selected PR: 280",
|
||||
"- Inventory pagination proof: is_final_page: true, has_more: false, total_count: 6",
|
||||
"- Review worktree used: true",
|
||||
"- Review worktree path: branches/review-pr280-feat-issue-275-claim",
|
||||
"- Review worktree inside branches: true",
|
||||
"- Review worktree HEAD state: detached",
|
||||
"- Main checkout branch: master",
|
||||
"- Current status: reviewed PR #280",
|
||||
])
|
||||
|
||||
|
||||
class TestInventoryPaginationProof(unittest.TestCase):
|
||||
def test_no_next_page_proof_passes(self):
|
||||
result = assess_inventory_worktree_report(_good_handoff())
|
||||
self.assertTrue(result["proven"], result["reasons"])
|
||||
self.assertTrue(result["pagination_proven"])
|
||||
|
||||
def test_rejects_partial_first_page_eligibility_claim(self):
|
||||
report = "\n".join([
|
||||
"Oldest eligible PR is #280.",
|
||||
"gitea_list_prs returned 10 open PRs on the first page.",
|
||||
"## Controller Handoff",
|
||||
"- Inventory pagination proof: first page only",
|
||||
])
|
||||
result = assess_inventory_worktree_report(report)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertTrue(result["block"])
|
||||
self.assertTrue(any("eligible" in r.lower() for r in result["reasons"]))
|
||||
|
||||
def test_rejects_default_page_size_assumption(self):
|
||||
report = (
|
||||
"Inventory exhaustive because fewer than the default Gitea page size "
|
||||
"were returned."
|
||||
)
|
||||
result = assess_inventory_worktree_report(report)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertTrue(any("page size" in r.lower() for r in result["reasons"]))
|
||||
|
||||
def test_rejects_exactly_full_first_page_without_finality(self):
|
||||
report = "\n".join([
|
||||
"Next eligible PR: #290 based on complete inventory.",
|
||||
"gitea_list_prs returned 50 open PRs.",
|
||||
"## Controller Handoff",
|
||||
"- Inventory pagination proof: returned 50 open PRs",
|
||||
])
|
||||
result = assess_inventory_worktree_report(
|
||||
report,
|
||||
inventory_session={"requested_page_size": 50, "returned_page_size": 50},
|
||||
)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertTrue(any("full first page" in r.lower() for r in result["reasons"]))
|
||||
|
||||
def test_allows_exact_page_with_session_pagination_complete(self):
|
||||
report = "Oldest eligible PR: #12 after queue inventory."
|
||||
result = assess_inventory_worktree_report(
|
||||
report,
|
||||
inventory_session={"pagination_complete": True},
|
||||
)
|
||||
self.assertTrue(result["proven"], result["reasons"])
|
||||
|
||||
|
||||
class TestWorktreeConsistency(unittest.TestCase):
|
||||
def test_branches_worktree_requires_review_worktree_used_true(self):
|
||||
report = "\n".join([
|
||||
"## Controller Handoff",
|
||||
"- Inventory pagination proof: is_final_page: true",
|
||||
"- Review worktree used: false",
|
||||
"- Review worktree path: branches/review-pr280-feat",
|
||||
"- Scratch worktree used: false",
|
||||
])
|
||||
result = assess_inventory_worktree_report(report)
|
||||
self.assertFalse(result["proven"])
|
||||
joined = " ".join(result["reasons"]).lower()
|
||||
self.assertIn("review worktree used", joined)
|
||||
self.assertIn("scratch worktree", joined)
|
||||
|
||||
def test_rejects_contradictory_detached_and_branch_wording(self):
|
||||
report = "\n".join([
|
||||
"Checkout: Detached HEAD / Branch master",
|
||||
"## Controller Handoff",
|
||||
"- Inventory pagination proof: is_final_page: true",
|
||||
"- Review worktree used: true",
|
||||
"- Review worktree path: branches/review-pr280-feat",
|
||||
"- Review worktree HEAD state: detached",
|
||||
"- Main checkout branch: master",
|
||||
])
|
||||
result = assess_inventory_worktree_report(report)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertTrue(
|
||||
any("contradictory" in r.lower() for r in result["reasons"])
|
||||
)
|
||||
|
||||
def test_requires_head_state_when_review_worktree_used(self):
|
||||
report = "\n".join([
|
||||
"## Controller Handoff",
|
||||
"- Inventory pagination proof: pagination_complete: true",
|
||||
"- Review worktree used: true",
|
||||
"- Review worktree path: branches/review-pr280-feat",
|
||||
])
|
||||
result = assess_inventory_worktree_report(report)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertTrue(any("head state" in r.lower() for r in result["reasons"]))
|
||||
|
||||
|
||||
class TestExport(unittest.TestCase):
|
||||
def test_review_proofs_reexport(self):
|
||||
from review_proofs import assess_inventory_worktree_report as exported
|
||||
|
||||
self.assertTrue(callable(exported))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user