fix: resolve conflicts for PR #384 against latest master
This commit is contained in:
@@ -0,0 +1,151 @@
|
||||
"""Tests for already-landed final-report wording validator (#298).
|
||||
|
||||
An already-landed PR is reconciliation-only, never review/merge
|
||||
eligible, and a head SHA may not be called reviewed unless validation
|
||||
and diff review actually passed. Final reports and controller handoffs
|
||||
must use the reconciliation wording, not the legacy eligible/reviewed
|
||||
fields.
|
||||
"""
|
||||
import sys
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
from review_proofs import assess_already_landed_report_wording # noqa: E402
|
||||
|
||||
|
||||
GOOD_ALREADY_LANDED = (
|
||||
"Controller Handoff\n"
|
||||
"- Oldest open PR requiring action: PR #278\n"
|
||||
"- Eligibility class: ALREADY_LANDED_RECONCILE_REQUIRED\n"
|
||||
"- Candidate head SHA: 2a544b7\n"
|
||||
"- Reviewed head SHA: none\n"
|
||||
"- Review worktree used: false\n"
|
||||
)
|
||||
|
||||
|
||||
class TestAlreadyLandedWording(unittest.TestCase):
|
||||
def test_correct_reconciliation_wording_passes(self):
|
||||
result = assess_already_landed_report_wording(
|
||||
GOOD_ALREADY_LANDED, already_landed=True
|
||||
)
|
||||
self.assertTrue(result["complete"])
|
||||
self.assertFalse(result["downgraded"])
|
||||
self.assertEqual(result["reasons"], [])
|
||||
|
||||
def test_oldest_eligible_wording_fails(self):
|
||||
report = GOOD_ALREADY_LANDED + "- oldest eligible PR: PR #278\n"
|
||||
result = assess_already_landed_report_wording(
|
||||
report, already_landed=True
|
||||
)
|
||||
self.assertFalse(result["complete"])
|
||||
self.assertTrue(
|
||||
any("oldest eligible pr" in r.lower() for r in result["reasons"])
|
||||
)
|
||||
|
||||
def test_next_eligible_wording_fails(self):
|
||||
report = GOOD_ALREADY_LANDED + "- next eligible PR: PR #278\n"
|
||||
result = assess_already_landed_report_wording(
|
||||
report, already_landed=True
|
||||
)
|
||||
self.assertFalse(result["complete"])
|
||||
|
||||
def test_pinned_reviewed_head_fails(self):
|
||||
report = GOOD_ALREADY_LANDED + "- Pinned reviewed head: 2a544b7\n"
|
||||
result = assess_already_landed_report_wording(
|
||||
report, already_landed=True
|
||||
)
|
||||
self.assertFalse(result["complete"])
|
||||
self.assertTrue(
|
||||
any("pinned reviewed head" in r.lower() for r in result["reasons"])
|
||||
)
|
||||
|
||||
def test_scratch_worktree_field_fails(self):
|
||||
report = GOOD_ALREADY_LANDED + "- Scratch worktree used: False\n"
|
||||
result = assess_already_landed_report_wording(
|
||||
report, already_landed=True
|
||||
)
|
||||
self.assertFalse(result["complete"])
|
||||
|
||||
def test_missing_eligibility_class_fails(self):
|
||||
report = (
|
||||
"Controller Handoff\n"
|
||||
"- Oldest open PR requiring action: PR #278\n"
|
||||
"- Candidate head SHA: 2a544b7\n"
|
||||
"- Reviewed head SHA: none\n"
|
||||
"- Review worktree used: false\n"
|
||||
)
|
||||
result = assess_already_landed_report_wording(
|
||||
report, already_landed=True
|
||||
)
|
||||
self.assertFalse(result["complete"])
|
||||
self.assertTrue(
|
||||
any("eligibility class" in r.lower() for r in result["reasons"])
|
||||
)
|
||||
|
||||
def test_populated_reviewed_head_sha_fails(self):
|
||||
report = GOOD_ALREADY_LANDED.replace(
|
||||
"- Reviewed head SHA: none\n",
|
||||
"- Reviewed head SHA: 2a544b7\n",
|
||||
)
|
||||
result = assess_already_landed_report_wording(
|
||||
report, already_landed=True
|
||||
)
|
||||
self.assertFalse(result["complete"])
|
||||
self.assertTrue(
|
||||
any("reviewed head" in r.lower() for r in result["reasons"])
|
||||
)
|
||||
|
||||
|
||||
class TestNonAlreadyLandedReports(unittest.TestCase):
|
||||
def test_normal_reviewed_report_passes(self):
|
||||
report = (
|
||||
"- Selected PR: 281\n"
|
||||
"- Pinned reviewed head: abc1234\n"
|
||||
"- Review decision: approve\n"
|
||||
)
|
||||
result = assess_already_landed_report_wording(
|
||||
report, already_landed=False, review_validated=True
|
||||
)
|
||||
self.assertTrue(result["complete"])
|
||||
|
||||
def test_blocked_infra_report_with_pinned_head_fails(self):
|
||||
report = (
|
||||
"- Selected PR: 281\n"
|
||||
"- Pinned reviewed head: abc1234\n"
|
||||
"- Current status: blocked on infra_stop\n"
|
||||
)
|
||||
result = assess_already_landed_report_wording(
|
||||
report, already_landed=False, review_validated=False
|
||||
)
|
||||
self.assertFalse(result["complete"])
|
||||
self.assertTrue(
|
||||
any("before validation" in r.lower() for r in result["reasons"])
|
||||
)
|
||||
|
||||
def test_validation_failed_report_with_reviewed_sha_fails(self):
|
||||
report = (
|
||||
"- Selected PR: 281\n"
|
||||
"- Reviewed head SHA: abc1234\n"
|
||||
"- Validation: full suite failed\n"
|
||||
)
|
||||
result = assess_already_landed_report_wording(
|
||||
report, already_landed=False, review_validated=False
|
||||
)
|
||||
self.assertFalse(result["complete"])
|
||||
|
||||
def test_validation_failed_report_with_reviewed_none_passes(self):
|
||||
report = (
|
||||
"- Selected PR: 281\n"
|
||||
"- Reviewed head SHA: none\n"
|
||||
"- Validation: full suite failed\n"
|
||||
)
|
||||
result = assess_already_landed_report_wording(
|
||||
report, already_landed=False, review_validated=False
|
||||
)
|
||||
self.assertTrue(result["complete"])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,116 @@
|
||||
"""Tests for issue claim heartbeat leases (#268)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import sys
|
||||
import unittest
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
import issue_claim_heartbeat as ich # noqa: E402
|
||||
|
||||
|
||||
class TestHeartbeatFormatting(unittest.TestCase):
|
||||
def test_format_and_parse_roundtrip(self):
|
||||
body = ich.format_heartbeat_body(
|
||||
kind="claim",
|
||||
issue_number=268,
|
||||
branch="feat/issue-268-heartbeat-leases",
|
||||
phase="claimed",
|
||||
profile="prgs-author",
|
||||
next_action="implement module",
|
||||
)
|
||||
parsed = ich.parse_heartbeat_comment(body)
|
||||
self.assertIsNotNone(parsed)
|
||||
assert parsed is not None
|
||||
self.assertEqual(parsed["issue_number"], 268)
|
||||
self.assertEqual(parsed["branch"], "feat/issue-268-heartbeat-leases")
|
||||
self.assertEqual(parsed["phase"], "claimed")
|
||||
|
||||
|
||||
class TestClaimClassification(unittest.TestCase):
|
||||
NOW = datetime(2026, 7, 7, 12, 0, tzinfo=timezone.utc)
|
||||
|
||||
def _comment(self, *, minutes_ago: int, kind: str = "progress") -> dict:
|
||||
ts = (self.NOW - timedelta(minutes=minutes_ago)).isoformat()
|
||||
body = ich.format_heartbeat_body(
|
||||
kind=kind,
|
||||
issue_number=268,
|
||||
branch="feat/issue-268-heartbeat-leases",
|
||||
phase="implementation",
|
||||
profile="prgs-author",
|
||||
)
|
||||
return {"id": 1, "body": body, "created_at": ts, "updated_at": ts}
|
||||
|
||||
def test_active_claim_within_lease(self):
|
||||
issue = {"number": 268, "title": "t", "labels": [{"name": "status:in-progress"}]}
|
||||
result = ich.classify_issue_claim(
|
||||
issue=issue,
|
||||
comments=[self._comment(minutes_ago=5)],
|
||||
now=self.NOW,
|
||||
)
|
||||
self.assertEqual(result["status"], "active")
|
||||
|
||||
def test_stale_claim_without_branch(self):
|
||||
issue = {"number": 268, "title": "t", "labels": [{"name": "status:in-progress"}]}
|
||||
result = ich.classify_issue_claim(
|
||||
issue=issue,
|
||||
comments=[self._comment(minutes_ago=45)],
|
||||
now=self.NOW,
|
||||
)
|
||||
self.assertEqual(result["status"], "stale")
|
||||
|
||||
def test_reclaimable_after_sixty_minutes_without_branch(self):
|
||||
issue = {"number": 268, "title": "t", "labels": [{"name": "status:in-progress"}]}
|
||||
result = ich.classify_issue_claim(
|
||||
issue=issue,
|
||||
comments=[self._comment(minutes_ago=90)],
|
||||
now=self.NOW,
|
||||
)
|
||||
self.assertEqual(result["status"], "reclaimable")
|
||||
|
||||
def test_awaiting_review_when_open_pr_exists(self):
|
||||
issue = {"number": 268, "title": "t", "labels": [{"name": "status:in-progress"}]}
|
||||
open_prs = [
|
||||
{
|
||||
"number": 400,
|
||||
"title": "feat",
|
||||
"body": "Closes #268",
|
||||
"head": {"ref": "feat/issue-268-heartbeat-leases"},
|
||||
}
|
||||
]
|
||||
result = ich.classify_issue_claim(
|
||||
issue=issue,
|
||||
comments=[self._comment(minutes_ago=120)],
|
||||
open_prs=open_prs,
|
||||
now=self.NOW,
|
||||
)
|
||||
self.assertEqual(result["status"], "awaiting_review")
|
||||
|
||||
def test_phantom_claim_without_heartbeat(self):
|
||||
issue = {"number": 268, "title": "t", "labels": [{"name": "status:in-progress"}]}
|
||||
result = ich.classify_issue_claim(
|
||||
issue=issue,
|
||||
comments=[{"id": 1, "body": "working on it"}],
|
||||
now=self.NOW,
|
||||
)
|
||||
self.assertEqual(result["status"], "phantom")
|
||||
|
||||
|
||||
class TestInventory(unittest.TestCase):
|
||||
def test_build_cleanup_plan_flags_phantom(self):
|
||||
inventory = {
|
||||
"entries": [
|
||||
{"issue_number": 1, "status": "phantom", "reasons": ["no heartbeat"]},
|
||||
{"issue_number": 2, "status": "active", "reasons": []},
|
||||
]
|
||||
}
|
||||
plan = ich.build_cleanup_plan(inventory)
|
||||
self.assertEqual(len(plan), 1)
|
||||
self.assertEqual(plan[0]["issue_number"], 1)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -172,6 +172,8 @@ class TestAllowedIssueWritesProceed(IssueWriteGateBase):
|
||||
return [{"id": 10, "name": "status:in-progress"}]
|
||||
if method == "POST" and "/issues/9/labels" in url:
|
||||
return [{"name": "status:in-progress"}]
|
||||
if method == "POST" and "/issues/9/comments" in url:
|
||||
return {"id": 501}
|
||||
if method == "GET" and url.endswith("/user"):
|
||||
return {"login": "author-user"}
|
||||
return {}
|
||||
|
||||
@@ -152,6 +152,12 @@ def test_merge_simulation_verifier_exported():
|
||||
assert callable(assess_merge_simulation_report)
|
||||
|
||||
|
||||
def test_validation_integrity_verifier_exported():
|
||||
from review_proofs import assess_validation_integrity_report
|
||||
|
||||
assert callable(assess_validation_integrity_report)
|
||||
|
||||
|
||||
def test_prior_blocker_skip_verifier_exported():
|
||||
from review_proofs import assess_prior_blocker_skip_proof
|
||||
|
||||
@@ -170,6 +176,12 @@ def test_validation_worktree_edit_verifier_exported():
|
||||
assert callable(assess_validation_worktree_edit_report)
|
||||
|
||||
|
||||
def test_mutation_categories_verifier_exported():
|
||||
from review_proofs import assess_mutation_categories_report
|
||||
|
||||
assert callable(assess_mutation_categories_report)
|
||||
|
||||
|
||||
def test_worktree_ownership_verifier_exported():
|
||||
from review_proofs import assess_worktree_ownership_report
|
||||
|
||||
|
||||
@@ -321,15 +321,17 @@ class TestMarkIssue(unittest.TestCase):
|
||||
@patch("mcp_server.api_request")
|
||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
||||
def test_start_adds_label(self, _auth, mock_api):
|
||||
# First call: get labels; second call: add label
|
||||
# labels, add label, post claim heartbeat comment
|
||||
mock_api.side_effect = [
|
||||
[{"id": 10, "name": "status:in-progress"}],
|
||||
[{"name": "status:in-progress"}],
|
||||
{"id": 99},
|
||||
]
|
||||
with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True):
|
||||
result = gitea_mark_issue(issue_number=5, action="start")
|
||||
self.assertTrue(result["success"])
|
||||
self.assertIn("claimed", result["message"])
|
||||
self.assertTrue(result.get("heartbeat_posted"))
|
||||
|
||||
@patch("mcp_server.api_request")
|
||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
||||
|
||||
@@ -23,6 +23,8 @@ sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.par
|
||||
from review_proofs import ( # noqa: E402
|
||||
ISSUE_SELECTION_CONTINUATION_EXPLICIT,
|
||||
ISSUE_SELECTION_REPRESENTED_BY_OPEN_PR,
|
||||
assess_already_landed_report_state,
|
||||
assess_already_landed_review_gate,
|
||||
assess_author_pr_report,
|
||||
assess_validation_environment_proof,
|
||||
assess_full_suite_failure_approval_gate,
|
||||
@@ -2639,6 +2641,139 @@ class TestReviewerBaselineValidationProof(unittest.TestCase):
|
||||
self.assertFalse(report["baseline_validation_proven"])
|
||||
|
||||
|
||||
class TestAlreadyLandedReviewGate(unittest.TestCase):
|
||||
"""Issue #292: PR head already on target blocks approval and merge."""
|
||||
|
||||
def _gate(self, **overrides):
|
||||
kwargs = {
|
||||
"pr_number": 278,
|
||||
"candidate_head_sha": PINNED,
|
||||
"target_branch": "master",
|
||||
"target_branch_sha": OTHER,
|
||||
"head_is_ancestor_of_target": True,
|
||||
"live_head_sha": PINNED,
|
||||
}
|
||||
kwargs.update(overrides)
|
||||
return assess_already_landed_review_gate(**kwargs)
|
||||
|
||||
def test_already_landed_blocks_approval_and_merge(self):
|
||||
result = self._gate()
|
||||
self.assertEqual(result["state"], "ALREADY_LANDED_RECONCILE_REQUIRED")
|
||||
self.assertFalse(result["approve_allowed"])
|
||||
self.assertFalse(result["merge_api_allowed"])
|
||||
self.assertFalse(result["request_changes_allowed"])
|
||||
self.assertTrue(result["reconciliation_handoff_required"])
|
||||
|
||||
def test_already_landed_with_real_blocker_allows_request_changes(self):
|
||||
result = self._gate(real_blocker=True)
|
||||
self.assertEqual(result["state"], "ALREADY_LANDED_RECONCILE_REQUIRED")
|
||||
self.assertFalse(result["approve_allowed"])
|
||||
self.assertFalse(result["merge_api_allowed"])
|
||||
self.assertTrue(result["request_changes_allowed"])
|
||||
|
||||
def test_normal_candidate_passes_gate(self):
|
||||
result = self._gate(head_is_ancestor_of_target=False)
|
||||
self.assertEqual(result["state"], "NORMAL_REVIEW_CANDIDATE")
|
||||
self.assertTrue(result["approve_allowed"])
|
||||
self.assertTrue(result["merge_api_allowed"])
|
||||
self.assertFalse(result["reconciliation_handoff_required"])
|
||||
|
||||
def test_non_mergeable_normal_pr_still_passes_this_gate(self):
|
||||
# Mergeability/conflicts are separate gates; ancestry gate only
|
||||
# decides already-landed vs normal candidate.
|
||||
result = self._gate(
|
||||
head_is_ancestor_of_target=False, mergeable=False)
|
||||
self.assertEqual(result["state"], "NORMAL_REVIEW_CANDIDATE")
|
||||
self.assertTrue(result["request_changes_allowed"])
|
||||
|
||||
def test_unchecked_ancestry_blocks_review_mutations(self):
|
||||
result = self._gate(head_is_ancestor_of_target=None)
|
||||
self.assertEqual(result["state"], "GATE_NOT_PROVEN")
|
||||
self.assertFalse(result["approve_allowed"])
|
||||
self.assertFalse(result["merge_api_allowed"])
|
||||
|
||||
def test_changed_head_since_pin_blocks_until_recheck(self):
|
||||
result = self._gate(
|
||||
head_is_ancestor_of_target=False, live_head_sha=OTHER)
|
||||
self.assertEqual(result["state"], "GATE_NOT_PROVEN")
|
||||
self.assertFalse(result["approve_allowed"])
|
||||
self.assertTrue(any(
|
||||
"head" in reason.lower() for reason in result["reasons"]
|
||||
))
|
||||
|
||||
def test_invalid_shas_block_gate(self):
|
||||
result = self._gate(candidate_head_sha="deadbeef")
|
||||
self.assertEqual(result["state"], "GATE_NOT_PROVEN")
|
||||
self.assertFalse(result["approve_allowed"])
|
||||
|
||||
result = self._gate(target_branch_sha=None)
|
||||
self.assertEqual(result["state"], "GATE_NOT_PROVEN")
|
||||
|
||||
def test_already_landed_handoff_lists_required_fields(self):
|
||||
result = self._gate()
|
||||
for field in (
|
||||
"PR number/title",
|
||||
"candidate head SHA",
|
||||
"target branch",
|
||||
"target branch SHA",
|
||||
"ancestor proof",
|
||||
"linked issue status",
|
||||
"recommended reconciliation action",
|
||||
"capability proof for close/comment mutations",
|
||||
):
|
||||
self.assertIn(field, result["required_handoff_fields"])
|
||||
|
||||
|
||||
class TestAlreadyLandedReportState(unittest.TestCase):
|
||||
"""Issue #292: reports cannot say APPROVED after the gate fired."""
|
||||
|
||||
GOOD_REPORT = (
|
||||
"Eligibility class: ALREADY_LANDED_RECONCILE_REQUIRED\n"
|
||||
"Candidate head SHA: " + PINNED + "\n"
|
||||
"Review decision: none\n"
|
||||
"Merge result: none\n"
|
||||
)
|
||||
|
||||
def test_gate_fired_with_reconcile_state_passes(self):
|
||||
result = assess_already_landed_report_state(
|
||||
self.GOOD_REPORT, gate_fired=True)
|
||||
self.assertTrue(result["complete"])
|
||||
self.assertFalse(result["block"])
|
||||
|
||||
def test_gate_fired_with_approved_state_blocks(self):
|
||||
report = self.GOOD_REPORT.replace(
|
||||
"Review decision: none", "Review decision: APPROVED")
|
||||
result = assess_already_landed_report_state(report, gate_fired=True)
|
||||
self.assertTrue(result["block"])
|
||||
|
||||
def test_gate_fired_with_merged_state_blocks(self):
|
||||
report = self.GOOD_REPORT.replace(
|
||||
"Merge result: none", "Merge result: MERGED")
|
||||
result = assess_already_landed_report_state(report, gate_fired=True)
|
||||
self.assertTrue(result["block"])
|
||||
|
||||
def test_gate_fired_with_ready_to_merge_blocks(self):
|
||||
result = assess_already_landed_report_state(
|
||||
self.GOOD_REPORT + "Current status: READY_TO_MERGE\n",
|
||||
gate_fired=True)
|
||||
self.assertTrue(result["block"])
|
||||
|
||||
def test_gate_fired_without_reconcile_state_blocks(self):
|
||||
result = assess_already_landed_report_state(
|
||||
"Current status: review complete.\n", gate_fired=True)
|
||||
self.assertTrue(result["block"])
|
||||
self.assertTrue(any(
|
||||
"already_landed_reconcile_required" in reason.lower()
|
||||
for reason in result["reasons"]
|
||||
))
|
||||
|
||||
def test_gate_not_fired_reports_pass_untouched(self):
|
||||
result = assess_already_landed_report_state(
|
||||
"Review decision: APPROVED\nMerge result: MERGED\n",
|
||||
gate_fired=False)
|
||||
self.assertTrue(result["complete"])
|
||||
|
||||
|
||||
class TestFullSuiteFailureApprovalGate(unittest.TestCase):
|
||||
"""Issue #323: full-suite failure blocks approval without baseline proof."""
|
||||
|
||||
|
||||
@@ -0,0 +1,88 @@
|
||||
"""Tests for precise mutation category verifier (#319)."""
|
||||
import sys
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
from reviewer_mutation_categories import assess_mutation_categories_report # noqa: E402
|
||||
|
||||
|
||||
def _precise_handoff() -> str:
|
||||
return "\n".join([
|
||||
"Controller Handoff",
|
||||
"File edits by reviewer: none",
|
||||
"Worktree/index mutations: git worktree add branches/review-pr1",
|
||||
"Git ref mutations: git fetch prgs master",
|
||||
"MCP/Gitea mutations: gitea_view_pr",
|
||||
"Review mutations: gitea_review_pr request_changes",
|
||||
"Read-only diagnostics: git status, git diff",
|
||||
])
|
||||
|
||||
|
||||
class TestMutationCategories(unittest.TestCase):
|
||||
def test_precise_categories_pass(self):
|
||||
result = assess_mutation_categories_report(_precise_handoff())
|
||||
self.assertTrue(result["proven"], result["reasons"])
|
||||
|
||||
def test_legacy_workspace_mutations_blocks(self):
|
||||
report = "\n".join([
|
||||
"Controller Handoff",
|
||||
"Workspace mutations: none (no local file changes)",
|
||||
"File edits by reviewer: none",
|
||||
"Worktree/index mutations: none",
|
||||
"Git ref mutations: none",
|
||||
])
|
||||
result = assess_mutation_categories_report(report)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertTrue(any("workspace mutations" in r.lower() for r in result["reasons"]))
|
||||
|
||||
def test_workspace_none_with_worktree_command_blocks(self):
|
||||
report = "\n".join([
|
||||
"Controller Handoff",
|
||||
"Workspace mutations: none (no local file changes)",
|
||||
"File edits by reviewer: none",
|
||||
"Worktree/index mutations: none",
|
||||
"Git ref mutations: none",
|
||||
])
|
||||
result = assess_mutation_categories_report(
|
||||
report,
|
||||
observed_commands=["git reset --hard FETCH_HEAD"],
|
||||
)
|
||||
self.assertFalse(result["proven"])
|
||||
|
||||
def test_file_edits_none_with_worktree_mutation_passes(self):
|
||||
report = "\n".join([
|
||||
"Controller Handoff",
|
||||
"File edits by reviewer: none",
|
||||
"Worktree/index mutations: git reset --hard FETCH_HEAD",
|
||||
"Git ref mutations: git fetch prgs",
|
||||
])
|
||||
result = assess_mutation_categories_report(
|
||||
report,
|
||||
observed_commands=["git reset --hard FETCH_HEAD", "git fetch prgs"],
|
||||
)
|
||||
self.assertTrue(result["proven"], result["reasons"])
|
||||
|
||||
def test_missing_required_categories_blocks(self):
|
||||
report = "Controller Handoff\nFile edits by reviewer: none\n"
|
||||
result = assess_mutation_categories_report(report)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertTrue(any("missing" in r.lower() for r in result["reasons"]))
|
||||
|
||||
def test_non_controller_handoff_skips(self):
|
||||
result = assess_mutation_categories_report(
|
||||
"Workspace mutations: none\n",
|
||||
)
|
||||
self.assertTrue(result["proven"])
|
||||
|
||||
|
||||
class TestExport(unittest.TestCase):
|
||||
def test_review_proofs_reexport(self):
|
||||
from review_proofs import assess_mutation_categories_report as exported
|
||||
|
||||
self.assertTrue(callable(exported))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,130 @@
|
||||
"""Tests for PR-head vs diagnostic validation integrity verifier (#316)."""
|
||||
import sys
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
from reviewer_validation_integrity import assess_validation_integrity_report # noqa: E402
|
||||
|
||||
|
||||
def _official_only_report() -> str:
|
||||
return "\n".join([
|
||||
"Official PR-head validation on unmodified worktree.",
|
||||
"Official validation command: pytest tests/ -q",
|
||||
"Official validation result: failed (3 failed)",
|
||||
"Validation integrity status: failed",
|
||||
"Worktree dirty after official validation: clean",
|
||||
"Review decision: request_changes",
|
||||
])
|
||||
|
||||
|
||||
def _diagnostic_report() -> str:
|
||||
return "\n".join([
|
||||
_official_only_report(),
|
||||
"Diagnostic local experiment — not PR-head validation",
|
||||
"Diagnostic edit path: tests/test_example.py",
|
||||
"File edits by reviewer: tests/test_example.py",
|
||||
"Diagnostic command: pytest tests/test_example.py -q",
|
||||
"Diagnostic result: passed after local fix",
|
||||
"Diagnostic results used only for suggested fix; blocker remains PR-head failure.",
|
||||
])
|
||||
|
||||
|
||||
class TestValidationIntegrity(unittest.TestCase):
|
||||
def test_official_only_passes(self):
|
||||
result = assess_validation_integrity_report(
|
||||
_official_only_report(),
|
||||
validation_session={
|
||||
"official_validation_ran": True,
|
||||
"official_result": "fail",
|
||||
"worktree_dirty_after_official": False,
|
||||
},
|
||||
)
|
||||
self.assertTrue(result["proven"], result["reasons"])
|
||||
|
||||
def test_failing_pr_head_with_diagnostic_pass_passes(self):
|
||||
result = assess_validation_integrity_report(
|
||||
_diagnostic_report(),
|
||||
validation_session={
|
||||
"official_validation_ran": True,
|
||||
"official_result": "fail",
|
||||
"diagnostic_validation_ran": True,
|
||||
"diagnostic_result": "pass",
|
||||
"diagnostic_edits": ["tests/test_example.py"],
|
||||
"worktree_dirty_after_official": False,
|
||||
},
|
||||
)
|
||||
self.assertTrue(result["proven"], result["reasons"])
|
||||
|
||||
def test_post_edit_tests_as_official_blocks(self):
|
||||
report = "\n".join([
|
||||
"Official validation result: passed after diagnostic edit",
|
||||
"pytest passed after local fix in validation worktree",
|
||||
])
|
||||
result = assess_validation_integrity_report(
|
||||
report,
|
||||
validation_session={
|
||||
"official_validation_ran": True,
|
||||
"diagnostic_edits": ["tests/test_example.py"],
|
||||
"diagnostic_validation_ran": True,
|
||||
},
|
||||
)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertTrue(result["block"])
|
||||
|
||||
def test_missing_dirty_after_validation_blocks(self):
|
||||
report = _official_only_report().replace(
|
||||
"Worktree dirty after official validation: clean",
|
||||
"",
|
||||
)
|
||||
result = assess_validation_integrity_report(
|
||||
report,
|
||||
validation_session={"official_validation_ran": True, "official_result": "fail"},
|
||||
)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertTrue(any("dirty" in r.lower() for r in result["reasons"]))
|
||||
|
||||
def test_diagnostic_without_label_blocks(self):
|
||||
report = _diagnostic_report().replace(
|
||||
"Diagnostic local experiment — not PR-head validation",
|
||||
"",
|
||||
)
|
||||
result = assess_validation_integrity_report(
|
||||
report,
|
||||
validation_session={
|
||||
"official_validation_ran": True,
|
||||
"diagnostic_edits": ["tests/test_example.py"],
|
||||
"diagnostic_validation_ran": True,
|
||||
},
|
||||
)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertTrue(any("diagnostic" in r.lower() for r in result["reasons"]))
|
||||
|
||||
def test_contaminated_requires_integrity_status(self):
|
||||
report = "Validation worktree was edited before official validation completed."
|
||||
result = assess_validation_integrity_report(
|
||||
report,
|
||||
validation_session={"contaminated": True, "official_validation_ran": True},
|
||||
)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertTrue(any("contaminated" in r.lower() for r in result["reasons"]))
|
||||
|
||||
def test_action_log_edits_trigger_diagnostic_rules(self):
|
||||
result = assess_validation_integrity_report(
|
||||
_official_only_report(),
|
||||
validation_session={"official_validation_ran": True, "official_result": "fail"},
|
||||
action_log=[{"path": "tests/test_example.py", "kind": "file_edit", "performed": True}],
|
||||
)
|
||||
self.assertFalse(result["proven"])
|
||||
|
||||
|
||||
class TestExport(unittest.TestCase):
|
||||
def test_review_proofs_reexport(self):
|
||||
from review_proofs import assess_validation_integrity_report as exported
|
||||
|
||||
self.assertTrue(callable(exported))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user