fix: resolve conflicts for PR #373 against latest master
This commit is contained in:
@@ -62,7 +62,24 @@ class TestPreflightWarnings(unittest.TestCase):
|
||||
self.assertIn("_encode_commit_payload.py", status["preflight_warnings"][0])
|
||||
|
||||
|
||||
# Issue-write tools are profile-gated (#69); gitea_lock_issue requires
|
||||
# gitea.issue.comment (see task_capability_map), so the gate must be
|
||||
# seeded exactly like tests/test_mcp_server.py::TestIssueLocking (#359).
|
||||
ISSUE_WRITE_ENV = {
|
||||
"GITEA_ALLOWED_OPERATIONS": (
|
||||
"gitea.issue.create,gitea.issue.close,gitea.issue.comment"
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
class TestIssueLockArtifactWarning(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self._env_patcher = patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True)
|
||||
self._env_patcher.start()
|
||||
|
||||
def tearDown(self):
|
||||
self._env_patcher.stop()
|
||||
|
||||
@patch("mcp_server.api_get_all", return_value=[])
|
||||
@patch("mcp_server._auth", return_value="token x")
|
||||
@patch("mcp_server._resolve", return_value=("h", "o", "r"))
|
||||
@@ -72,6 +89,9 @@ class TestIssueLockArtifactWarning(unittest.TestCase):
|
||||
mock_state.return_value = {
|
||||
"current_branch": "master",
|
||||
"porcelain_status": "?? _emit_payload.py\n",
|
||||
"base_equivalent": True,
|
||||
"inspected_git_root": "/scratch/wt",
|
||||
"base_branch": "origin/master",
|
||||
}
|
||||
result = mcp_server.gitea_lock_issue(
|
||||
issue_number=261,
|
||||
|
||||
@@ -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,118 @@
|
||||
"""Tests for linked-issue consistency verifier (#314).
|
||||
|
||||
Reviewer reports must verify the linked issue live for the selected PR;
|
||||
stale issue numbers from a previous PR must not leak into the final
|
||||
report, and linked-issue status must never be claimed without live proof.
|
||||
"""
|
||||
import sys
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
from review_proofs import assess_linked_issue_consistency # noqa: E402
|
||||
|
||||
|
||||
class TestCorrectLinkedIssue(unittest.TestCase):
|
||||
def test_matching_linked_issue_passes(self):
|
||||
report = (
|
||||
"Selected PR: 280\n"
|
||||
"Linked issue status: Issue #275 open, closes on merge\n"
|
||||
)
|
||||
result = assess_linked_issue_consistency(
|
||||
report, selected_pr=280, linked_issues=[275]
|
||||
)
|
||||
self.assertTrue(result["complete"])
|
||||
self.assertFalse(result["downgraded"])
|
||||
self.assertEqual(result["reasons"], [])
|
||||
|
||||
def test_multiple_linked_issues_all_reported_passes(self):
|
||||
report = (
|
||||
"Linked issue status: Issue #275 and Issue #276 both open\n"
|
||||
)
|
||||
result = assess_linked_issue_consistency(
|
||||
report, selected_pr=280, linked_issues=[275, 276]
|
||||
)
|
||||
self.assertTrue(result["complete"])
|
||||
|
||||
def test_multiple_linked_issues_one_missing_fails(self):
|
||||
report = "Linked issue status: Issue #275 open\n"
|
||||
result = assess_linked_issue_consistency(
|
||||
report, selected_pr=280, linked_issues=[275, 276]
|
||||
)
|
||||
self.assertFalse(result["complete"])
|
||||
self.assertTrue(
|
||||
any("276" in r for r in result["reasons"])
|
||||
)
|
||||
|
||||
|
||||
class TestStaleIssueLeak(unittest.TestCase):
|
||||
def test_stale_issue_in_status_field_fails(self):
|
||||
# #314 observed behavior: PR #280 links Issue #275 but the report
|
||||
# carries Issue #260 from the previous PR.
|
||||
report = "Linked issue status: Issue #260 open\n"
|
||||
result = assess_linked_issue_consistency(
|
||||
report, selected_pr=280, linked_issues=[275]
|
||||
)
|
||||
self.assertFalse(result["complete"])
|
||||
self.assertTrue(any("260" in r for r in result["reasons"]))
|
||||
|
||||
def test_stale_issue_in_diagnostics_fails(self):
|
||||
report = (
|
||||
"Linked issue status: Issue #275 open\n"
|
||||
"Read-only diagnostics: viewed Issue #260 status\n"
|
||||
)
|
||||
result = assess_linked_issue_consistency(
|
||||
report, selected_pr=280, linked_issues=[275]
|
||||
)
|
||||
self.assertFalse(result["complete"])
|
||||
self.assertTrue(
|
||||
any("stale" in r.lower() and "260" in r for r in result["reasons"])
|
||||
)
|
||||
|
||||
def test_selected_pr_number_mention_is_not_stale(self):
|
||||
# "PR #280" mentions must not be confused with issue mentions.
|
||||
report = (
|
||||
"Selected PR: 280\n"
|
||||
"Validation: ran tests at PR #280 head\n"
|
||||
"Linked issue status: Issue #275 open\n"
|
||||
)
|
||||
result = assess_linked_issue_consistency(
|
||||
report, selected_pr=280, linked_issues=[275]
|
||||
)
|
||||
self.assertTrue(result["complete"])
|
||||
|
||||
|
||||
class TestMissingProof(unittest.TestCase):
|
||||
def test_status_claim_without_live_proof_fails(self):
|
||||
report = "Linked issue status: Issue #275 open\n"
|
||||
result = assess_linked_issue_consistency(
|
||||
report, selected_pr=280, linked_issues=None
|
||||
)
|
||||
self.assertFalse(result["complete"])
|
||||
self.assertTrue(
|
||||
any("live proof" in r.lower() for r in result["reasons"])
|
||||
)
|
||||
|
||||
def test_unverified_wording_without_proof_passes(self):
|
||||
report = (
|
||||
"Linked issue status: not verified in this session\n"
|
||||
)
|
||||
result = assess_linked_issue_consistency(
|
||||
report, selected_pr=280, linked_issues=None
|
||||
)
|
||||
self.assertTrue(result["complete"])
|
||||
|
||||
def test_missing_field_fails(self):
|
||||
report = "Selected PR: 280\n"
|
||||
result = assess_linked_issue_consistency(
|
||||
report, selected_pr=280, linked_issues=[275]
|
||||
)
|
||||
self.assertFalse(result["complete"])
|
||||
self.assertTrue(
|
||||
any("linked issue status" in r.lower() for r in result["reasons"])
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -136,7 +136,43 @@ def test_create_issue_final_report_verifier_exported():
|
||||
assert callable(assess_create_issue_final_report)
|
||||
|
||||
|
||||
def test_merge_simulation_verifier_exported():
|
||||
from review_proofs import assess_merge_simulation_report
|
||||
|
||||
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
|
||||
|
||||
assert callable(assess_prior_blocker_skip_proof)
|
||||
|
||||
|
||||
def test_non_mergeable_skip_verifier_exported():
|
||||
from review_proofs import assess_non_mergeable_skip_proof
|
||||
|
||||
assert callable(assess_non_mergeable_skip_proof)
|
||||
|
||||
|
||||
def test_validation_worktree_edit_verifier_exported():
|
||||
from review_proofs import assess_validation_worktree_edit_report
|
||||
|
||||
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
|
||||
|
||||
assert callable(assess_worktree_ownership_report)
|
||||
+406
-3
@@ -23,7 +23,11 @@ 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,
|
||||
assess_queue_ordering_proof,
|
||||
assess_reviewer_baseline_validation_proof,
|
||||
assess_capability_evidence,
|
||||
@@ -2314,10 +2318,108 @@ class TestCreateIssueFinalReport(unittest.TestCase):
|
||||
self.assertTrue(result["downgraded"])
|
||||
|
||||
|
||||
class TestQueueOrderingProof(unittest.TestCase):
|
||||
"""Issue #321: reviewer queue selection must prove ordering."""
|
||||
class TestValidationEnvironmentProof(unittest.TestCase):
|
||||
"""Issue #311: exact validation command and execution environment."""
|
||||
|
||||
ROOT = "/repo/Gitea-Tools/branches/review-pr-280"
|
||||
|
||||
def test_venv_pytest_with_full_report_passes(self):
|
||||
result = assess_validation_environment_proof(
|
||||
report_text=(
|
||||
"Validation command: venv/bin/pytest tests/ --ignore=branches\n"
|
||||
f"Working directory: {self.ROOT}\n"
|
||||
"Result: 1014 passed, 6 skipped"
|
||||
),
|
||||
)
|
||||
self.assertTrue(result["proven"])
|
||||
|
||||
def test_bare_pytest_without_path_proof_fails(self):
|
||||
result = assess_validation_environment_proof(
|
||||
report_text=(
|
||||
"Validation command: pytest\n"
|
||||
f"Working directory: {self.ROOT}\n"
|
||||
"1014 passed, 6 skipped"
|
||||
),
|
||||
)
|
||||
self.assertFalse(result["proven"])
|
||||
|
||||
def test_bare_pytest_with_path_and_version_proof_passes(self):
|
||||
result = assess_validation_environment_proof(
|
||||
report_text=(
|
||||
"Validation command: pytest tests/\n"
|
||||
f"Working directory: {self.ROOT}\n"
|
||||
"which pytest: /repo/Gitea-Tools/venv/bin/pytest\n"
|
||||
"pytest --version: pytest 8.3.4\n"
|
||||
"Resolves to the project venv: yes\n"
|
||||
"1014 passed, 6 skipped"
|
||||
),
|
||||
)
|
||||
self.assertTrue(result["proven"])
|
||||
|
||||
def test_vague_pytest_summary_without_command_fails(self):
|
||||
result = assess_validation_environment_proof(
|
||||
report_text=(
|
||||
f"Working directory: {self.ROOT}\n"
|
||||
"pytest executed inside the worktree: 1014 passed, 6 skipped"
|
||||
),
|
||||
)
|
||||
self.assertFalse(result["proven"])
|
||||
|
||||
def test_structured_environment_proof_passes(self):
|
||||
result = assess_validation_environment_proof(
|
||||
validation_environment={
|
||||
"command": "pytest tests/",
|
||||
"working_directory": self.ROOT,
|
||||
"which_pytest": "/repo/Gitea-Tools/venv/bin/pytest",
|
||||
"pytest_version": "pytest 8.3.4",
|
||||
"venv_resolved": True,
|
||||
"passed": 1014,
|
||||
"skipped": 6,
|
||||
},
|
||||
)
|
||||
self.assertTrue(result["proven"])
|
||||
|
||||
def test_missing_working_directory_fails(self):
|
||||
result = assess_validation_environment_proof(
|
||||
report_text=(
|
||||
"Validation command: venv/bin/pytest tests/\n"
|
||||
"1014 passed, 6 skipped"
|
||||
),
|
||||
)
|
||||
self.assertFalse(result["proven"])
|
||||
|
||||
def test_build_final_report_downgrades_missing_environment_proof(self):
|
||||
report = build_final_report(
|
||||
checkout_proof=_good_checkout(),
|
||||
inventory=_good_inventory(),
|
||||
validation=_good_validation(),
|
||||
contamination=_good_contamination(),
|
||||
identity_eligible=True,
|
||||
merge_performed=False,
|
||||
issue_status_verified=True,
|
||||
capability_evidence=_good_capability_evidence(),
|
||||
sweep=_good_sweep(),
|
||||
live_state=_good_live_state(),
|
||||
role_boundary=_good_role_boundary(),
|
||||
review_mutation=_good_review_mutation(),
|
||||
controller_handoff=_good_handoff(),
|
||||
capability_proof=_good_capability_proof(),
|
||||
sweep_proof=_good_secret_sweep(),
|
||||
worktree_proof={
|
||||
"worktree_path": "/repo/branches/review-feat-issue-224",
|
||||
"porcelain_status": "",
|
||||
"pr_scope_files": ["docs/wiki/Repositories.md"],
|
||||
"scratch_used": True,
|
||||
},
|
||||
report_text=(
|
||||
f"Working directory: {self.ROOT}\n"
|
||||
"pytest executed: 1014 passed, 6 skipped"
|
||||
),
|
||||
)
|
||||
self.assertNotEqual(report["grade"], "A")
|
||||
self.assertFalse(report["validation_environment_proven"])
|
||||
|
||||
|
||||
def test_next_eligible_claim_without_policy_fails(self):
|
||||
result = assess_queue_ordering_proof(
|
||||
report_text="Selected the next eligible PR: PR #286.",
|
||||
)
|
||||
@@ -2539,5 +2641,306 @@ 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."""
|
||||
|
||||
ROOT = "/repo/Gitea-Tools"
|
||||
|
||||
def _good_proof(self, **overrides):
|
||||
proof = {
|
||||
"worktree_path": f"{self.ROOT}/branches/baseline-master-pr280",
|
||||
"baseline_target_sha": PINNED,
|
||||
"pr_head_sha": OTHER,
|
||||
"baseline_failures": ["tests/test_foo.py::test_bar"],
|
||||
"pr_failures": ["tests/test_foo.py::test_bar"],
|
||||
"failure_signatures_match": True,
|
||||
"clean_before": True,
|
||||
"clean_after": True,
|
||||
"pr_suite_command": "venv/bin/pytest tests/",
|
||||
"baseline_suite_command": "venv/bin/pytest tests/",
|
||||
"new_tests_passed": True,
|
||||
"unrelated_explanation": (
|
||||
"failures touch agent_temp_artifacts only; PR changes "
|
||||
"review_proofs handoff fields"
|
||||
),
|
||||
}
|
||||
proof.update(overrides)
|
||||
return proof
|
||||
|
||||
def _good_report(self):
|
||||
return (
|
||||
"Full suite failed on PR worktree and on baseline.\n"
|
||||
"- Validation command: venv/bin/pytest tests/\n"
|
||||
f"- Working directory: {self.ROOT}/branches/review-pr-280\n"
|
||||
"Failures are pre-existing on master.\n"
|
||||
)
|
||||
|
||||
def test_full_suite_pass_allows_approval_without_baseline(self):
|
||||
result = assess_full_suite_failure_approval_gate(
|
||||
review_decision="approve",
|
||||
full_suite_passed=True,
|
||||
project_root=self.ROOT,
|
||||
)
|
||||
self.assertTrue(result["approve_allowed"])
|
||||
self.assertFalse(result["block"])
|
||||
self.assertEqual(result["default_action"], "proceed")
|
||||
|
||||
def test_full_suite_failure_defaults_to_request_changes(self):
|
||||
result = assess_full_suite_failure_approval_gate(
|
||||
review_decision="request_changes",
|
||||
full_suite_passed=False,
|
||||
project_root=self.ROOT,
|
||||
)
|
||||
self.assertFalse(result["block"])
|
||||
self.assertEqual(result["default_action"], "request_changes")
|
||||
|
||||
def test_approval_with_failure_and_no_proof_blocks(self):
|
||||
result = assess_full_suite_failure_approval_gate(
|
||||
review_decision="approve",
|
||||
full_suite_passed=False,
|
||||
project_root=self.ROOT,
|
||||
)
|
||||
self.assertFalse(result["approve_allowed"])
|
||||
self.assertTrue(result["block"])
|
||||
self.assertEqual(result["default_action"], "request_changes")
|
||||
|
||||
def test_approval_with_matching_baseline_proof_allowed(self):
|
||||
result = assess_full_suite_failure_approval_gate(
|
||||
review_decision="approve",
|
||||
full_suite_passed=False,
|
||||
baseline_proof=self._good_proof(),
|
||||
report_text=self._good_report(),
|
||||
project_root=self.ROOT,
|
||||
)
|
||||
self.assertTrue(result["approve_allowed"])
|
||||
self.assertFalse(result["block"])
|
||||
|
||||
def test_mismatched_failure_signatures_block_approval(self):
|
||||
result = assess_full_suite_failure_approval_gate(
|
||||
review_decision="approve",
|
||||
full_suite_passed=False,
|
||||
baseline_proof=self._good_proof(
|
||||
failure_signatures_match=False,
|
||||
pr_failures=["tests/test_other.py::test_other"],
|
||||
),
|
||||
report_text=self._good_report(),
|
||||
project_root=self.ROOT,
|
||||
)
|
||||
self.assertFalse(result["approve_allowed"])
|
||||
self.assertTrue(result["block"])
|
||||
|
||||
def test_main_checkout_baseline_blocks_approval(self):
|
||||
result = assess_full_suite_failure_approval_gate(
|
||||
review_decision="approve",
|
||||
full_suite_passed=False,
|
||||
baseline_proof=self._good_proof(worktree_path=self.ROOT),
|
||||
report_text=self._good_report(),
|
||||
project_root=self.ROOT,
|
||||
)
|
||||
self.assertFalse(result["approve_allowed"])
|
||||
self.assertTrue(result["block"])
|
||||
|
||||
def test_missing_new_tests_proof_blocks_approval(self):
|
||||
result = assess_full_suite_failure_approval_gate(
|
||||
review_decision="approve",
|
||||
full_suite_passed=False,
|
||||
baseline_proof=self._good_proof(new_tests_passed=None),
|
||||
report_text=self._good_report(),
|
||||
project_root=self.ROOT,
|
||||
)
|
||||
self.assertFalse(result["approve_allowed"])
|
||||
self.assertTrue(any(
|
||||
"new" in reason.lower() and "test" in reason.lower()
|
||||
for reason in result["reasons"]
|
||||
))
|
||||
|
||||
def test_missing_unrelated_explanation_blocks_approval(self):
|
||||
result = assess_full_suite_failure_approval_gate(
|
||||
review_decision="approve",
|
||||
full_suite_passed=False,
|
||||
baseline_proof=self._good_proof(unrelated_explanation=""),
|
||||
report_text=self._good_report(),
|
||||
project_root=self.ROOT,
|
||||
)
|
||||
self.assertFalse(result["approve_allowed"])
|
||||
|
||||
def test_missing_suite_commands_block_approval(self):
|
||||
result = assess_full_suite_failure_approval_gate(
|
||||
review_decision="approve",
|
||||
full_suite_passed=False,
|
||||
baseline_proof=self._good_proof(
|
||||
pr_suite_command="", baseline_suite_command=""),
|
||||
report_text=self._good_report(),
|
||||
project_root=self.ROOT,
|
||||
)
|
||||
self.assertFalse(result["approve_allowed"])
|
||||
self.assertTrue(any(
|
||||
"command" in reason.lower() for reason in result["reasons"]
|
||||
))
|
||||
|
||||
def test_unknown_suite_result_blocks_approval(self):
|
||||
result = assess_full_suite_failure_approval_gate(
|
||||
review_decision="approve",
|
||||
full_suite_passed=None,
|
||||
project_root=self.ROOT,
|
||||
)
|
||||
self.assertFalse(result["approve_allowed"])
|
||||
self.assertTrue(result["block"])
|
||||
|
||||
def test_vague_same_as_master_claim_without_proof_blocks(self):
|
||||
result = assess_full_suite_failure_approval_gate(
|
||||
review_decision="approve",
|
||||
full_suite_passed=False,
|
||||
report_text="Validation: failures same as master; approving.",
|
||||
project_root=self.ROOT,
|
||||
)
|
||||
self.assertFalse(result["approve_allowed"])
|
||||
self.assertTrue(result["block"])
|
||||
|
||||
def test_non_approve_decision_never_blocks(self):
|
||||
result = assess_full_suite_failure_approval_gate(
|
||||
review_decision="comment",
|
||||
full_suite_passed=None,
|
||||
project_root=self.ROOT,
|
||||
)
|
||||
self.assertFalse(result["block"])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
@@ -0,0 +1,139 @@
|
||||
"""Tests for prior-blocker skip proof verifier (#318)."""
|
||||
import sys
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
from reviewer_blocker_skip import assess_prior_blocker_skip_proof # noqa: E402
|
||||
|
||||
HEAD_CURRENT = "0fdc8f582026b72a229d59a172c0a63ac4aaeaf9"
|
||||
HEAD_BLOCKER = "1111111111111111111111111111111111111111"
|
||||
|
||||
|
||||
def _good_blocker_skip_report(pr_number: int) -> str:
|
||||
return "\n".join([
|
||||
f"Skipped PR #{pr_number} due to prior REQUEST_CHANGES.",
|
||||
f"- PR number: #{pr_number}",
|
||||
f"- Current head SHA: {HEAD_CURRENT}",
|
||||
"- Blocking review decision: request_changes",
|
||||
f"- Blocking review head SHA: {HEAD_CURRENT}",
|
||||
"- Head unchanged since blocker",
|
||||
"- Reason remains blocked: unresolved review feedback at same head",
|
||||
"- Blocker revalidated live in this session via gitea_get_pr_review_feedback",
|
||||
])
|
||||
|
||||
|
||||
class TestPriorBlockerSkipProof(unittest.TestCase):
|
||||
def test_no_skips_passes(self):
|
||||
result = assess_prior_blocker_skip_proof(
|
||||
"Selected PR #281 for review.",
|
||||
skipped_prs=[],
|
||||
)
|
||||
self.assertTrue(result["proven"])
|
||||
|
||||
def test_blocked_unchanged_with_live_proof_passes(self):
|
||||
report = _good_blocker_skip_report(280)
|
||||
result = assess_prior_blocker_skip_proof(
|
||||
report,
|
||||
skipped_prs=[{
|
||||
"pr_number": 280,
|
||||
"head_sha": HEAD_CURRENT,
|
||||
"blocking_decision": "request_changes",
|
||||
"blocking_review_head_sha": HEAD_CURRENT,
|
||||
"head_changed_since_blocker": False,
|
||||
"blocker_verified": True,
|
||||
}],
|
||||
)
|
||||
self.assertTrue(result["proven"])
|
||||
|
||||
def test_head_changed_after_blocker_blocks_skip(self):
|
||||
result = assess_prior_blocker_skip_proof(
|
||||
_good_blocker_skip_report(280),
|
||||
skipped_prs=[{
|
||||
"pr_number": 280,
|
||||
"head_sha": HEAD_CURRENT,
|
||||
"blocking_decision": "request_changes",
|
||||
"blocking_review_head_sha": HEAD_BLOCKER,
|
||||
"head_changed_since_blocker": True,
|
||||
"blocker_verified": True,
|
||||
}],
|
||||
)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertTrue(any("head changed" in r for r in result["reasons"]))
|
||||
|
||||
def test_missing_live_proof_blocks(self):
|
||||
report = _good_blocker_skip_report(280).replace(
|
||||
"- Blocker revalidated live in this session via gitea_get_pr_review_feedback",
|
||||
"",
|
||||
)
|
||||
result = assess_prior_blocker_skip_proof(
|
||||
report,
|
||||
skipped_prs=[{
|
||||
"pr_number": 280,
|
||||
"head_sha": HEAD_CURRENT,
|
||||
"blocking_decision": "request_changes",
|
||||
"blocking_review_head_sha": HEAD_CURRENT,
|
||||
"head_changed_since_blocker": False,
|
||||
"blocker_verified": True,
|
||||
}],
|
||||
)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertTrue(any("live blocker proof" in r for r in result["reasons"]))
|
||||
|
||||
def test_blocker_unverified_classification_passes(self):
|
||||
report = "\n".join([
|
||||
"Skipped PR #280.",
|
||||
"Classification: BLOCKER_STATUS_UNVERIFIED",
|
||||
"Review feedback could not be fetched live in this session.",
|
||||
])
|
||||
result = assess_prior_blocker_skip_proof(
|
||||
report,
|
||||
skipped_prs=[{
|
||||
"pr_number": 280,
|
||||
"blocking_decision": "request_changes",
|
||||
"blocker_verified": False,
|
||||
}],
|
||||
)
|
||||
self.assertTrue(result["proven"])
|
||||
|
||||
def test_unverified_without_classification_blocks(self):
|
||||
result = assess_prior_blocker_skip_proof(
|
||||
"Skipped PR #280 based on memory from last session.",
|
||||
skipped_prs=[{
|
||||
"pr_number": 280,
|
||||
"blocking_decision": "request_changes",
|
||||
"blocker_verified": False,
|
||||
}],
|
||||
)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertTrue(any("BLOCKER_STATUS_UNVERIFIED" in r for r in result["reasons"]))
|
||||
|
||||
def test_missing_blocking_head_sha_blocks(self):
|
||||
report = _good_blocker_skip_report(280).replace(
|
||||
f"- Blocking review head SHA: {HEAD_CURRENT}",
|
||||
"",
|
||||
)
|
||||
result = assess_prior_blocker_skip_proof(
|
||||
report,
|
||||
skipped_prs=[{
|
||||
"pr_number": 280,
|
||||
"head_sha": HEAD_CURRENT,
|
||||
"blocking_decision": "request_changes",
|
||||
"blocking_review_head_sha": HEAD_CURRENT,
|
||||
"head_changed_since_blocker": False,
|
||||
"blocker_verified": True,
|
||||
}],
|
||||
)
|
||||
self.assertFalse(result["proven"])
|
||||
|
||||
|
||||
class TestExport(unittest.TestCase):
|
||||
def test_review_proofs_reexport(self):
|
||||
from review_proofs import assess_prior_blocker_skip_proof as exported
|
||||
|
||||
self.assertTrue(callable(exported))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,135 @@
|
||||
"""Tests for merge-simulation worktree mutation verifier (#317)."""
|
||||
import sys
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
from reviewer_merge_simulation import ( # noqa: E402
|
||||
assess_merge_simulation_report,
|
||||
merge_simulation_commands,
|
||||
)
|
||||
|
||||
|
||||
def _good_simulation_report(**overrides) -> str:
|
||||
lines = [
|
||||
"Worktree/index mutations: merge simulation in branches/review-pr281",
|
||||
"Worktree path: branches/review-pr281",
|
||||
"Pre-simulation clean status: clean (git status --porcelain empty)",
|
||||
"Merge result: conflicts in review_proofs.py",
|
||||
"Abort command: git merge --abort",
|
||||
"Post-abort clean status: clean after abort (git status --porcelain empty)",
|
||||
]
|
||||
text = "\n".join(lines)
|
||||
for key, value in overrides.items():
|
||||
text = text.replace(f"{key}:", f"{key}: {value}")
|
||||
return text
|
||||
|
||||
|
||||
class TestMergeSimulationDetection(unittest.TestCase):
|
||||
def test_detects_no_commit_merge(self):
|
||||
commands = merge_simulation_commands([
|
||||
"git merge prgs/master --no-commit --no-ff",
|
||||
"git status",
|
||||
])
|
||||
self.assertEqual(commands, ["git merge prgs/master --no-commit --no-ff"])
|
||||
|
||||
def test_detects_merge_abort(self):
|
||||
commands = merge_simulation_commands([
|
||||
"git merge prgs/master --no-commit --no-ff",
|
||||
"git merge --abort",
|
||||
])
|
||||
self.assertIn("git merge --abort", commands)
|
||||
|
||||
|
||||
class TestMergeSimulationReport(unittest.TestCase):
|
||||
def test_no_simulation_passes(self):
|
||||
result = assess_merge_simulation_report(
|
||||
"Review complete. Worktree/index mutations: none",
|
||||
command_log=["git status --porcelain", "git diff prgs/master...HEAD"],
|
||||
)
|
||||
self.assertTrue(result["proven"])
|
||||
|
||||
def test_documented_simulation_passes(self):
|
||||
report = _good_simulation_report()
|
||||
result = assess_merge_simulation_report(
|
||||
report,
|
||||
command_log=[
|
||||
"git merge prgs/master --no-commit --no-ff",
|
||||
"git merge --abort",
|
||||
],
|
||||
)
|
||||
self.assertTrue(result["proven"], result["reasons"])
|
||||
|
||||
def test_readonly_classification_blocks(self):
|
||||
report = "\n".join([
|
||||
"Read-only diagnostics: git merge prgs/master --no-commit --no-ff, git status",
|
||||
"Worktree/index mutations: none",
|
||||
])
|
||||
result = assess_merge_simulation_report(
|
||||
report,
|
||||
command_log=["git merge prgs/master --no-commit --no-ff"],
|
||||
)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertTrue(result["block"])
|
||||
self.assertTrue(any("read-only" in r.lower() for r in result["reasons"]))
|
||||
|
||||
def test_missing_worktree_field_blocks(self):
|
||||
report = "Merge simulation ran but only noted in prose."
|
||||
result = assess_merge_simulation_report(
|
||||
report,
|
||||
command_log=["git merge prgs/master --no-commit --no-ff"],
|
||||
)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertTrue(any("worktree/index mutations" in r.lower() for r in result["reasons"]))
|
||||
|
||||
def test_conflict_simulation_requires_merge_result(self):
|
||||
report = _good_simulation_report().replace(
|
||||
"Merge result: conflicts in review_proofs.py",
|
||||
"",
|
||||
)
|
||||
result = assess_merge_simulation_report(
|
||||
report,
|
||||
command_log=["git merge prgs/master --no-commit --no-ff"],
|
||||
)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertTrue(any("merge result" in r.lower() for r in result["reasons"]))
|
||||
|
||||
def test_aborted_simulation_requires_post_clean_proof(self):
|
||||
report = _good_simulation_report().replace(
|
||||
"Post-abort clean status: clean after abort (git status --porcelain empty)",
|
||||
"",
|
||||
)
|
||||
result = assess_merge_simulation_report(
|
||||
report,
|
||||
command_log=[
|
||||
"git merge prgs/master --no-commit --no-ff",
|
||||
"git merge --abort",
|
||||
],
|
||||
)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertTrue(any("post-abort" in r.lower() for r in result["reasons"]))
|
||||
|
||||
def test_successful_simulation_without_abort_omits_abort_fields(self):
|
||||
report = "\n".join([
|
||||
"Worktree/index mutations: merge simulation completed cleanly",
|
||||
"Worktree path: branches/review-pr281",
|
||||
"Pre-simulation clean status: clean before simulation",
|
||||
"Merge result: clean fast-forward simulation",
|
||||
])
|
||||
result = assess_merge_simulation_report(
|
||||
report,
|
||||
command_log=["git merge prgs/master --no-commit --no-ff"],
|
||||
)
|
||||
self.assertTrue(result["proven"], result["reasons"])
|
||||
|
||||
|
||||
class TestExport(unittest.TestCase):
|
||||
def test_review_proofs_reexport(self):
|
||||
from review_proofs import assess_merge_simulation_report as exported
|
||||
|
||||
self.assertTrue(callable(exported))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -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()
|
||||
@@ -0,0 +1,134 @@
|
||||
"""Tests for PR validation worktree no-edit verifier (#315)."""
|
||||
import sys
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
from reviewer_validation_worktree_edits import ( # noqa: E402
|
||||
assess_validation_worktree_edit_report,
|
||||
)
|
||||
|
||||
|
||||
def _read_only_report() -> str:
|
||||
return "\n".join([
|
||||
"Validation worktree path: branches/review-pr280-feat-issue-275-claim",
|
||||
"Official PR-head validation on unmodified worktree.",
|
||||
"Official validation result: failed (1 failed)",
|
||||
"File edits by reviewer: none",
|
||||
"Review decision: request_changes",
|
||||
])
|
||||
|
||||
|
||||
def _diagnostic_report() -> str:
|
||||
return "\n".join([
|
||||
"Validation worktree path: branches/review-pr280-feat-issue-275-claim",
|
||||
"Official PR-head validation on unmodified worktree.",
|
||||
"Official validation result: failed (1 failed)",
|
||||
"Review decision: request_changes",
|
||||
"Diagnostic local experiment — not PR-head validation",
|
||||
"Diagnostic scratch worktree path: branches/diagnostic-pr280-fix-test",
|
||||
"File edits by reviewer: tests/test_agent_temp_artifacts.py (diagnostic only)",
|
||||
"Diagnostic result: passed after local fix",
|
||||
])
|
||||
|
||||
|
||||
class TestValidationWorktreeEdits(unittest.TestCase):
|
||||
def test_read_only_review_passes(self):
|
||||
result = assess_validation_worktree_edit_report(
|
||||
_read_only_report(),
|
||||
validation_session={
|
||||
"validation_worktree_path": (
|
||||
"branches/review-pr280-feat-issue-275-claim"
|
||||
),
|
||||
},
|
||||
)
|
||||
self.assertTrue(result["proven"], result["reasons"])
|
||||
|
||||
def test_validation_worktree_edit_blocks(self):
|
||||
result = assess_validation_worktree_edit_report(
|
||||
_read_only_report(),
|
||||
validation_session={
|
||||
"validation_worktree_path": (
|
||||
"branches/review-pr280-feat-issue-275-claim"
|
||||
),
|
||||
"validation_worktree_edits": ["tests/test_agent_temp_artifacts.py"],
|
||||
},
|
||||
)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertTrue(result["block"])
|
||||
self.assertTrue(any("validation worktree" in r.lower() for r in result["reasons"]))
|
||||
|
||||
def test_file_edits_none_contradiction_blocks(self):
|
||||
result = assess_validation_worktree_edit_report(
|
||||
_read_only_report(),
|
||||
validation_session={
|
||||
"validation_worktree_path": (
|
||||
"branches/review-pr280-feat-issue-275-claim"
|
||||
),
|
||||
},
|
||||
action_log=[
|
||||
{
|
||||
"path": "tests/test_agent_temp_artifacts.py",
|
||||
"kind": "file_edit",
|
||||
"performed": True,
|
||||
"worktree_path": "branches/review-pr280-feat-issue-275-claim",
|
||||
}
|
||||
],
|
||||
)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertTrue(any("file edits" in r.lower() for r in result["reasons"]))
|
||||
|
||||
def test_diagnostic_scratch_with_reporting_passes(self):
|
||||
result = assess_validation_worktree_edit_report(
|
||||
_diagnostic_report(),
|
||||
validation_session={
|
||||
"validation_worktree_path": (
|
||||
"branches/review-pr280-feat-issue-275-claim"
|
||||
),
|
||||
"diagnostic_worktree_path": "branches/diagnostic-pr280-fix-test",
|
||||
"diagnostic_edits": ["tests/test_agent_temp_artifacts.py"],
|
||||
"diagnostic_experiment": True,
|
||||
},
|
||||
)
|
||||
self.assertTrue(result["proven"], result["reasons"])
|
||||
|
||||
def test_diagnostic_without_scratch_worktree_blocks(self):
|
||||
result = assess_validation_worktree_edit_report(
|
||||
"File edits by reviewer: tests/test_example.py",
|
||||
validation_session={
|
||||
"diagnostic_edits": ["tests/test_example.py"],
|
||||
"diagnostic_experiment": True,
|
||||
},
|
||||
)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertTrue(any("diagnostic" in r.lower() for r in result["reasons"]))
|
||||
|
||||
def test_post_edit_official_validation_blocks(self):
|
||||
report = "\n".join([
|
||||
"Validation worktree path: branches/review-pr280-feat-issue-275-claim",
|
||||
"Official validation result: passed after local edit",
|
||||
"File edits by reviewer: tests/test_agent_temp_artifacts.py",
|
||||
])
|
||||
result = assess_validation_worktree_edit_report(
|
||||
report,
|
||||
validation_session={
|
||||
"validation_worktree_path": (
|
||||
"branches/review-pr280-feat-issue-275-claim"
|
||||
),
|
||||
"validation_worktree_edits": ["tests/test_agent_temp_artifacts.py"],
|
||||
"official_validation_after_edit": True,
|
||||
},
|
||||
)
|
||||
self.assertFalse(result["proven"])
|
||||
|
||||
|
||||
class TestExport(unittest.TestCase):
|
||||
def test_review_proofs_reexport(self):
|
||||
from review_proofs import assess_validation_worktree_edit_report as exported
|
||||
|
||||
self.assertTrue(callable(exported))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,137 @@
|
||||
"""Tests for reviewer worktree ownership verifier (#312)."""
|
||||
import sys
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
from reviewer_worktree_ownership import assess_worktree_ownership_report # noqa: E402
|
||||
|
||||
|
||||
def _fresh_review_report() -> str:
|
||||
return "\n".join([
|
||||
"Review worktree path: branches/review-pr280-feat-issue-275-claim",
|
||||
"Worktree is inside branches/ and not the main checkout.",
|
||||
"Tracked state: clean; untracked state: clean.",
|
||||
"Official PR-head validation on session-owned worktree.",
|
||||
"Worktree mutations: none",
|
||||
])
|
||||
|
||||
|
||||
def _safe_reuse_report() -> str:
|
||||
return "\n".join([
|
||||
"Safe-reuse proof for existing review worktree.",
|
||||
"Review worktree path: branches/docs-issue-260-forbid-commit-fallbacks",
|
||||
"Worktree inside branches/; not the main checkout.",
|
||||
"Not owned by another active task/session.",
|
||||
"Clean tracked state; clean untracked state.",
|
||||
"Branch/head before reset: feat/issue-260-example",
|
||||
"Reset target SHA: abc123def4567890abcdef1234567890abcdef12",
|
||||
"Project policy allows reuse/reset for this clean review worktree.",
|
||||
"Worktree mutations: git reset --hard FETCH_HEAD",
|
||||
"Destructive reset operations: git reset --hard FETCH_HEAD",
|
||||
])
|
||||
|
||||
|
||||
class TestWorktreeOwnership(unittest.TestCase):
|
||||
def test_fresh_session_owned_passes(self):
|
||||
result = assess_worktree_ownership_report(
|
||||
_fresh_review_report(),
|
||||
ownership_session={
|
||||
"validation_ran": True,
|
||||
"worktree_path": "branches/review-pr280-feat-issue-275-claim",
|
||||
},
|
||||
)
|
||||
self.assertTrue(result["proven"], result["reasons"])
|
||||
self.assertTrue(result["session_owned"])
|
||||
|
||||
def test_safe_reuse_with_reset_passes(self):
|
||||
result = assess_worktree_ownership_report(
|
||||
_safe_reuse_report(),
|
||||
ownership_session={
|
||||
"safe_reuse": True,
|
||||
"worktree_path": "branches/docs-issue-260-forbid-commit-fallbacks",
|
||||
},
|
||||
command_log=["git reset --hard FETCH_HEAD"],
|
||||
)
|
||||
self.assertTrue(result["proven"], result["reasons"])
|
||||
|
||||
def test_destructive_reset_without_proof_blocks(self):
|
||||
report = "\n".join([
|
||||
"Review worktree path: branches/docs-issue-260-forbid-commit-fallbacks",
|
||||
"Worktree mutations: git reset --hard FETCH_HEAD",
|
||||
])
|
||||
result = assess_worktree_ownership_report(
|
||||
report,
|
||||
ownership_session={
|
||||
"validation_ran": True,
|
||||
"worktree_path": "branches/docs-issue-260-forbid-commit-fallbacks",
|
||||
},
|
||||
command_log=["git reset --hard FETCH_HEAD"],
|
||||
)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertTrue(result["block"])
|
||||
|
||||
def test_reused_dirty_tracked_blocks(self):
|
||||
result = assess_worktree_ownership_report(
|
||||
_fresh_review_report(),
|
||||
ownership_session={
|
||||
"validation_ran": True,
|
||||
"worktree_path": "branches/review-pr280-feat-issue-275-claim",
|
||||
"dirty_tracked": True,
|
||||
},
|
||||
)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertTrue(any("tracked" in r.lower() for r in result["reasons"]))
|
||||
|
||||
def test_safe_reuse_dirty_untracked_blocks(self):
|
||||
result = assess_worktree_ownership_report(
|
||||
_safe_reuse_report(),
|
||||
ownership_session={
|
||||
"safe_reuse": True,
|
||||
"dirty_untracked": True,
|
||||
"worktree_path": "branches/docs-issue-260-forbid-commit-fallbacks",
|
||||
},
|
||||
command_log=["git reset --hard FETCH_HEAD"],
|
||||
)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertTrue(any("untracked" in r.lower() for r in result["reasons"]))
|
||||
|
||||
def test_workspace_none_with_reset_blocks(self):
|
||||
report = "\n".join([
|
||||
"Review worktree path: branches/review-pr280-feat-issue-275-claim",
|
||||
"Workspace mutations: none",
|
||||
"Worktree mutations: git reset --hard FETCH_HEAD",
|
||||
])
|
||||
result = assess_worktree_ownership_report(
|
||||
report,
|
||||
ownership_session={
|
||||
"session_owned": True,
|
||||
"worktree_path": "branches/review-pr280-feat-issue-275-claim",
|
||||
},
|
||||
command_log=["git reset --hard FETCH_HEAD"],
|
||||
)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertTrue(any("workspace mutations" in r.lower() for r in result["reasons"]))
|
||||
|
||||
def test_main_checkout_blocks(self):
|
||||
result = assess_worktree_ownership_report(
|
||||
"Review worktree path: /Users/dev/Gitea-Tools",
|
||||
ownership_session={
|
||||
"validation_ran": True,
|
||||
"main_checkout": True,
|
||||
"worktree_path": "/Users/dev/Gitea-Tools",
|
||||
},
|
||||
)
|
||||
self.assertFalse(result["proven"])
|
||||
|
||||
|
||||
class TestExport(unittest.TestCase):
|
||||
def test_review_proofs_reexport(self):
|
||||
from review_proofs import assess_worktree_ownership_report as exported
|
||||
|
||||
self.assertTrue(callable(exported))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user