fix: resolve conflicts for PR #360
This commit is contained in:
@@ -0,0 +1,273 @@
|
||||
"""Tests for composable final-report validator (#327)."""
|
||||
import sys
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
from final_report_validator import ( # noqa: E402
|
||||
FINAL_REPORT_TASK_KINDS,
|
||||
assess_final_report_validator,
|
||||
validator_finding,
|
||||
)
|
||||
|
||||
|
||||
def _review_handoff(**overrides):
|
||||
lines = [
|
||||
"## Controller Handoff",
|
||||
"",
|
||||
"- Task: review PR #203",
|
||||
"- Repo: Scaled-Tech-Consulting/Gitea-Tools",
|
||||
"- Role: reviewer",
|
||||
"- Identity: sysadmin / prgs-reviewer",
|
||||
"- Issue/PR: #182 / PR #203",
|
||||
"- Branch/SHA: feat/x @ 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
|
||||
"- Files changed: review_proofs.py",
|
||||
"- Validation: pytest tests/test_review_proofs.py -q in branches/review-203",
|
||||
"- Mutations: review only",
|
||||
"- Workspace mutations: none",
|
||||
"- File edits by reviewer: none",
|
||||
"- Worktree/index mutations: none",
|
||||
"- Git ref mutations: git fetch prgs master",
|
||||
"- MCP/Gitea mutations: review submitted",
|
||||
"- Current status: PR open",
|
||||
"- Blockers: none",
|
||||
"- Next: await author",
|
||||
"- Safety: no self-review; no self-merge",
|
||||
"- Selected PR: #203",
|
||||
"- Reviewer eligibility: eligible",
|
||||
"- Pinned reviewed head: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
|
||||
"- Worktree path: branches/review-203",
|
||||
"- Worktree dirty: clean",
|
||||
"- Scratch worktree used: yes (branches/review-203)",
|
||||
"- Unrelated local mutations: none",
|
||||
"- Review decision: request_changes",
|
||||
"- Merge result: not attempted",
|
||||
"- Linked issue: #182",
|
||||
"- Linked issue status: open",
|
||||
"- Cleanup status: none",
|
||||
]
|
||||
text = "\n".join(lines)
|
||||
for key, value in overrides.items():
|
||||
text = text.replace(f"- {key}:", f"- {key}: {value}")
|
||||
return text
|
||||
|
||||
|
||||
def _reconcile_handoff(**overrides):
|
||||
lines = [
|
||||
"## Controller Handoff",
|
||||
"",
|
||||
"- Task: reconcile already-landed PR #99",
|
||||
"- Repo: Scaled-Tech-Consulting/Gitea-Tools",
|
||||
"- Role: reconciler",
|
||||
"- Identity: sysadmin / prgs-author",
|
||||
"- Selected PR: #99",
|
||||
"- Eligibility class: ALREADY_LANDED_RECONCILE_REQUIRED",
|
||||
"- Linked issue: #98",
|
||||
"- Linked issue live status: not verified in this session",
|
||||
"- File edits by reconciler: none",
|
||||
"- Git ref mutations: git fetch prgs master",
|
||||
"- Reconciliation mutations: PR comment posted",
|
||||
"- Current status: reconciliation complete",
|
||||
"- Safe next action: none",
|
||||
"- No review/merge confirmation: confirmed",
|
||||
]
|
||||
text = "\n".join(lines)
|
||||
for key, value in overrides.items():
|
||||
if f"- {key}:" in text:
|
||||
text = text.replace(f"- {key}:", f"- {key}: {value}")
|
||||
return text
|
||||
|
||||
|
||||
class TestValidatorFinding(unittest.TestCase):
|
||||
def test_finding_shape(self):
|
||||
finding = validator_finding(
|
||||
"reviewer.test",
|
||||
"block",
|
||||
"Validation",
|
||||
"missing proof",
|
||||
"repair validation proof",
|
||||
)
|
||||
self.assertEqual(finding["rule_id"], "reviewer.test")
|
||||
self.assertEqual(finding["severity"], "block")
|
||||
self.assertEqual(finding["field"], "Validation")
|
||||
|
||||
|
||||
class TestReviewPrRules(unittest.TestCase):
|
||||
def test_clean_reviewer_report_passes(self):
|
||||
result = assess_final_report_validator(
|
||||
_review_handoff(),
|
||||
"review_pr",
|
||||
linked_issue_lock={"issue_number": 182, "pr_number": 203},
|
||||
)
|
||||
self.assertEqual(result["grade"], "A")
|
||||
self.assertFalse(result["blocked"])
|
||||
self.assertEqual(result["findings"], [])
|
||||
|
||||
def test_legacy_workspace_mutations_blocks(self):
|
||||
report = _review_handoff() + "\n- Workspace mutations: none"
|
||||
result = assess_final_report_validator(
|
||||
report,
|
||||
"review_pr",
|
||||
mutations_observed=True,
|
||||
)
|
||||
self.assertEqual(result["grade"], "blocked")
|
||||
rule_ids = [f["rule_id"] for f in result["findings"]]
|
||||
self.assertIn("reviewer.legacy_workspace_mutations", rule_ids)
|
||||
|
||||
def test_vague_mutations_none_blocks(self):
|
||||
report = _review_handoff().replace(
|
||||
"- Mutations: review only",
|
||||
"- Mutations: none",
|
||||
)
|
||||
result = assess_final_report_validator(
|
||||
report,
|
||||
"review_pr",
|
||||
mutations_observed=True,
|
||||
)
|
||||
self.assertTrue(result["blocked"])
|
||||
self.assertTrue(
|
||||
any(f["rule_id"] == "reviewer.vague_mutations_none" for f in result["findings"])
|
||||
)
|
||||
|
||||
def test_bare_pytest_without_cwd_downgrades(self):
|
||||
report = _review_handoff().replace(
|
||||
"- Validation: pytest tests/test_review_proofs.py -q in branches/review-203",
|
||||
"- Validation: pytest passed",
|
||||
)
|
||||
result = assess_final_report_validator(report, "review_pr")
|
||||
self.assertEqual(result["grade"], "downgraded")
|
||||
self.assertTrue(
|
||||
any(f["rule_id"] == "reviewer.validation_command_proof" for f in result["findings"])
|
||||
)
|
||||
|
||||
def test_linked_issue_mismatch_blocks(self):
|
||||
report = _review_handoff().replace("- Linked issue: #182", "- Linked issue: #999")
|
||||
result = assess_final_report_validator(
|
||||
report,
|
||||
"review_pr",
|
||||
linked_issue_lock={"issue_number": 182, "pr_number": 203},
|
||||
)
|
||||
self.assertTrue(result["blocked"])
|
||||
self.assertTrue(
|
||||
any(f["rule_id"] == "reviewer.linked_issue_mismatch" for f in result["findings"])
|
||||
)
|
||||
|
||||
def test_approval_after_full_suite_failure_blocks_without_baseline(self):
|
||||
report = (
|
||||
_review_handoff()
|
||||
.replace("- Review decision: request_changes", "- Review decision: approve")
|
||||
+ "\nSubmitted 'approve' after full suite failed with 3 failed tests."
|
||||
)
|
||||
result = assess_final_report_validator(report, "review_pr")
|
||||
self.assertTrue(result["blocked"])
|
||||
self.assertTrue(
|
||||
any(
|
||||
f["rule_id"] == "reviewer.baseline_on_full_suite_failure"
|
||||
for f in result["findings"]
|
||||
)
|
||||
)
|
||||
|
||||
def test_same_as_master_without_baseline_blocks(self):
|
||||
report = _review_handoff() + "\nFailures are same as master baseline."
|
||||
result = assess_final_report_validator(report, "review_pr")
|
||||
self.assertTrue(result["blocked"])
|
||||
self.assertTrue(
|
||||
any(f["rule_id"] == "reviewer.main_checkout_baseline" for f in result["findings"])
|
||||
)
|
||||
|
||||
def test_already_landed_eligible_wording_blocks(self):
|
||||
report = (
|
||||
_review_handoff()
|
||||
+ "\nPR is already landed and eligible for review next."
|
||||
)
|
||||
result = assess_final_report_validator(report, "review_pr")
|
||||
self.assertTrue(result["blocked"])
|
||||
self.assertTrue(
|
||||
any(
|
||||
f["rule_id"] == "reviewer.already_landed_eligible_wording"
|
||||
for f in result["findings"]
|
||||
)
|
||||
)
|
||||
|
||||
def test_git_fetch_must_not_be_readonly_only(self):
|
||||
report = (
|
||||
_review_handoff()
|
||||
.replace("- Git ref mutations: git fetch prgs master", "- Git ref mutations: none")
|
||||
+ "\n- Read-only diagnostics: git fetch prgs master"
|
||||
)
|
||||
result = assess_final_report_validator(
|
||||
report,
|
||||
"review_pr",
|
||||
action_log=[{"command": "git fetch prgs master", "performed": True}],
|
||||
)
|
||||
self.assertTrue(result["blocked"])
|
||||
self.assertTrue(
|
||||
any(f["rule_id"] == "reviewer.git_fetch_readonly" for f in result["findings"])
|
||||
)
|
||||
|
||||
|
||||
class TestReconciliationRules(unittest.TestCase):
|
||||
def test_clean_reconcile_report_passes(self):
|
||||
result = assess_final_report_validator(
|
||||
_reconcile_handoff(),
|
||||
"reconcile_already_landed",
|
||||
)
|
||||
self.assertEqual(result["grade"], "A")
|
||||
|
||||
def test_stale_author_field_blocks(self):
|
||||
report = _reconcile_handoff() + "\n- PR number opened: #12"
|
||||
result = assess_final_report_validator(report, "reconcile_already_landed")
|
||||
self.assertTrue(result["blocked"])
|
||||
self.assertTrue(
|
||||
any(
|
||||
f["rule_id"] == "reconcile.stale_author_reviewer_fields"
|
||||
for f in result["findings"]
|
||||
)
|
||||
)
|
||||
|
||||
def test_eligible_wording_for_landed_pr_blocks(self):
|
||||
report = _reconcile_handoff() + "\nPR already landed and eligible for merge."
|
||||
result = assess_final_report_validator(report, "reconcile_already_landed")
|
||||
self.assertTrue(result["blocked"])
|
||||
|
||||
def test_linked_issue_status_without_live_proof_downgrades(self):
|
||||
report = _reconcile_handoff().replace(
|
||||
"- Linked issue live status: not verified in this session",
|
||||
"- Linked issue live status: closed",
|
||||
)
|
||||
result = assess_final_report_validator(report, "reconcile_already_landed")
|
||||
self.assertEqual(result["grade"], "downgraded")
|
||||
self.assertTrue(
|
||||
any(f["rule_id"] == "reconcile.linked_issue_live_status" for f in result["findings"])
|
||||
)
|
||||
|
||||
def test_inventory_complete_requires_pagination_proof(self):
|
||||
report = _reconcile_handoff() + "\nInventory complete for open PRs."
|
||||
result = assess_final_report_validator(report, "reconcile_already_landed")
|
||||
self.assertEqual(result["grade"], "downgraded")
|
||||
self.assertTrue(
|
||||
any(f["rule_id"] == "reconcile.inventory_pagination_proof" for f in result["findings"])
|
||||
)
|
||||
|
||||
|
||||
class TestEntryPoint(unittest.TestCase):
|
||||
def test_unknown_task_kind_blocks(self):
|
||||
result = assess_final_report_validator("report", "unknown_mode")
|
||||
self.assertTrue(result["blocked"])
|
||||
self.assertEqual(result["findings"][0]["rule_id"], "shared.unknown_task_kind")
|
||||
|
||||
def test_review_proofs_reexport(self):
|
||||
from review_proofs import assess_final_report_validator as exported
|
||||
|
||||
self.assertTrue(callable(exported))
|
||||
result = exported(_review_handoff(), "review_pr")
|
||||
self.assertIn("grade", result)
|
||||
|
||||
def test_supported_task_kinds_include_review_and_reconcile(self):
|
||||
self.assertIn("review_pr", FINAL_REPORT_TASK_KINDS)
|
||||
self.assertIn("reconcile_already_landed", FINAL_REPORT_TASK_KINDS)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -118,6 +118,12 @@ def test_start_issue_template_references_work_issue_workflow():
|
||||
assert "workflows/work-issue.md" in text
|
||||
|
||||
|
||||
def test_final_report_validator_exported():
|
||||
from review_proofs import assess_final_report_validator
|
||||
|
||||
assert callable(assess_final_report_validator)
|
||||
|
||||
|
||||
def test_create_issue_final_report_verifier_exported():
|
||||
from review_proofs import assess_create_issue_final_report
|
||||
|
||||
@@ -127,4 +133,4 @@ def test_create_issue_final_report_verifier_exported():
|
||||
def test_non_mergeable_skip_verifier_exported():
|
||||
from review_proofs import assess_non_mergeable_skip_proof
|
||||
|
||||
assert callable(assess_non_mergeable_skip_proof)
|
||||
assert callable(assess_non_mergeable_skip_proof)
|
||||
|
||||
@@ -0,0 +1,93 @@
|
||||
"""Doc-contract checks for proof wording enforcement (#330)."""
|
||||
import unittest
|
||||
|
||||
from review_proofs import assess_proof_wording, build_final_report
|
||||
|
||||
|
||||
class TestProofWording(unittest.TestCase):
|
||||
def test_rejects_inventory_complete_without_pagination_proof(self):
|
||||
result = assess_proof_wording(
|
||||
"Open PR inventory is complete because only 3 PRs were returned."
|
||||
)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertTrue(result["block"])
|
||||
|
||||
def test_rejects_default_page_size_assumption(self):
|
||||
result = assess_proof_wording(
|
||||
"Inventory exhaustive: fewer than the default Gitea page size returned."
|
||||
)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertIn("default_page_size_assumption", result["flagged_rules"])
|
||||
|
||||
def test_allows_inventory_complete_with_finality_metadata(self):
|
||||
result = assess_proof_wording(
|
||||
"Inventory complete. pagination_complete: true, is_final_page: true"
|
||||
)
|
||||
self.assertTrue(result["proven"])
|
||||
|
||||
def test_allows_inventory_complete_with_session_evidence(self):
|
||||
result = assess_proof_wording(
|
||||
"Queue inventory complete.",
|
||||
session_evidence={"pagination_complete": True},
|
||||
)
|
||||
self.assertTrue(result["proven"])
|
||||
|
||||
def test_rejects_live_proof_without_session_evidence(self):
|
||||
result = assess_proof_wording("Live proof confirms mergeable state.")
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertIn("live_proof", result["flagged_rules"])
|
||||
|
||||
def test_allows_prior_blocker_reuse_wording(self):
|
||||
result = assess_proof_wording(
|
||||
"Prior blocker reused; head SHA unchanged since blocking review. "
|
||||
"Live proof not claimed for this skip."
|
||||
)
|
||||
self.assertTrue(result["proven"])
|
||||
|
||||
def test_allows_live_proof_with_current_session_evidence(self):
|
||||
result = assess_proof_wording(
|
||||
"Live proof: gitea_view_pr revalidated in the current session."
|
||||
)
|
||||
self.assertTrue(result["proven"])
|
||||
|
||||
def test_rejects_same_as_master_without_baseline_proof(self):
|
||||
result = assess_proof_wording("Diff is same as master.")
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertIn("same_as_master", result["flagged_rules"])
|
||||
|
||||
def test_allows_same_as_master_with_baseline_wording(self):
|
||||
result = assess_proof_wording(
|
||||
"Clean baseline worktree proof: diff base matches master."
|
||||
)
|
||||
self.assertTrue(result["proven"])
|
||||
|
||||
def test_rejects_file_edits_none_without_ledger(self):
|
||||
result = assess_proof_wording("Workspace mutations: file edits none.")
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertIn("no_file_edits", result["flagged_rules"])
|
||||
|
||||
def test_allows_file_edits_none_with_mutation_ledger(self):
|
||||
result = assess_proof_wording(
|
||||
"Mutation ledger: tracked file edits: none",
|
||||
session_evidence={"mutation_ledger_clean": True},
|
||||
)
|
||||
self.assertTrue(result["proven"])
|
||||
|
||||
def test_build_final_report_downgrades_on_proof_wording_violation(self):
|
||||
report = build_final_report(
|
||||
{"proven": True},
|
||||
{"complete": True},
|
||||
{"claimable": True, "verdict": "strong"},
|
||||
{"status": "clean"},
|
||||
True,
|
||||
False,
|
||||
True,
|
||||
report_text="Inventory complete after listing 4 open PRs.",
|
||||
)
|
||||
self.assertEqual(report["grade"], "downgraded")
|
||||
self.assertFalse(report["proof_wording_proven"])
|
||||
self.assertTrue(report["proof_wording_violations"])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,91 @@
|
||||
"""Doc-contract checks for queue-status report verifier (#339)."""
|
||||
import unittest
|
||||
|
||||
from review_proofs import assess_queue_status_report, build_final_report
|
||||
|
||||
BAD_QUEUE_STATUS_REPORT = """
|
||||
## PR Queue Status
|
||||
|
||||
Loaded workflows/review-merge-pr.md. Inventory complete because gitea_list_prs
|
||||
returned 6 open PRs, fewer than the default Gitea page size.
|
||||
|
||||
PR #286 skipped: prior diagnostic worktree simulation showed merge conflicts.
|
||||
Live blocker proof for all skipped PRs.
|
||||
|
||||
All open PRs are conflicted or blocked. Blockers: none.
|
||||
|
||||
## Controller Handoff
|
||||
|
||||
- Task: review queue status
|
||||
- Repo: Scaled-Tech-Consulting/Gitea-Tools
|
||||
- Role: reviewer
|
||||
- Identity: reviewer1 / prgs-reviewer
|
||||
- Selected PR: none
|
||||
- Inventory pagination proof: assumed complete (6 < 50)
|
||||
- Already-landed gate: passed
|
||||
- Author-safety result: passed
|
||||
- Merge preflight: passed
|
||||
- Review worktree used: false
|
||||
- Review worktree dirty before validation: false
|
||||
- Blockers: none
|
||||
- Current status: queue inspected; no PR selected
|
||||
"""
|
||||
|
||||
|
||||
class TestQueueStatusReport(unittest.TestCase):
|
||||
def test_bad_fixture_fails_closed(self):
|
||||
result = assess_queue_status_report(BAD_QUEUE_STATUS_REPORT)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertTrue(result["block"])
|
||||
self.assertTrue(result["queue_status_only"])
|
||||
joined = " ".join(result["violations"]).lower()
|
||||
self.assertIn("already-landed gate", joined)
|
||||
self.assertIn("author-safety", joined)
|
||||
self.assertIn("merge preflight", joined)
|
||||
self.assertIn("prior diagnostic", joined)
|
||||
self.assertIn("blockers", joined)
|
||||
|
||||
def test_passes_with_not_applicable_gates(self):
|
||||
report = """
|
||||
## Controller Handoff
|
||||
- Selected PR: none
|
||||
- Inventory pagination proof: is_final_page: true, has_more: false
|
||||
- Already-landed gate: not applicable
|
||||
- Author-safety result: not run
|
||||
- Merge preflight: not applicable
|
||||
- Review worktree used: false
|
||||
- Review worktree dirty before validation: not applicable
|
||||
- Blockers: all PRs skipped pending reviewer
|
||||
"""
|
||||
result = assess_queue_status_report(report)
|
||||
self.assertTrue(result["proven"])
|
||||
|
||||
def test_rejects_passed_gates_without_selected_pr(self):
|
||||
report = """
|
||||
## Controller Handoff
|
||||
- Selected PR: none
|
||||
- Already-landed gate: passed
|
||||
- Blockers: PR #1 conflicted
|
||||
"""
|
||||
result = assess_queue_status_report(report)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertIn("already-landed gate", " ".join(result["violations"]).lower())
|
||||
|
||||
def test_build_final_report_downgrades_bad_queue_status(self):
|
||||
report = build_final_report(
|
||||
{"proven": True},
|
||||
{"complete": True},
|
||||
{"claimable": True, "verdict": "strong"},
|
||||
{"status": "clean"},
|
||||
True,
|
||||
False,
|
||||
True,
|
||||
report_text=BAD_QUEUE_STATUS_REPORT,
|
||||
)
|
||||
self.assertEqual(report["grade"], "downgraded")
|
||||
self.assertFalse(report["queue_status_report_proven"])
|
||||
self.assertTrue(report["queue_status_violations"])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
+260
-2
@@ -24,6 +24,7 @@ from review_proofs import ( # noqa: E402
|
||||
ISSUE_SELECTION_CONTINUATION_EXPLICIT,
|
||||
ISSUE_SELECTION_REPRESENTED_BY_OPEN_PR,
|
||||
assess_author_pr_report,
|
||||
assess_queue_ordering_proof,
|
||||
assess_reviewer_baseline_validation_proof,
|
||||
assess_capability_evidence,
|
||||
assess_capability_proof,
|
||||
@@ -940,13 +941,18 @@ class TestControllerHandoff(unittest.TestCase):
|
||||
self.assertIn("Safety", result["missing_fields"])
|
||||
|
||||
def test_review_role_requires_review_fields(self):
|
||||
result = assess_controller_handoff(self.BASE_HANDOFF, role="review")
|
||||
# Issue #320: review handoffs must not carry the legacy
|
||||
# 'Workspace mutations' line, so strip it from the shared base.
|
||||
review_base = "\n".join(
|
||||
line for line in self.BASE_HANDOFF.splitlines()
|
||||
if not line.startswith("- Workspace mutations:"))
|
||||
result = assess_controller_handoff(review_base, role="review")
|
||||
self.assertEqual(result["verdict"], "incomplete")
|
||||
self.assertIn("Pinned reviewed head", result["missing_fields"])
|
||||
self.assertIn("Worktree path", result["missing_fields"])
|
||||
self.assertIn("Merge result", result["missing_fields"])
|
||||
|
||||
complete = self.BASE_HANDOFF + "\n" + "\n".join([
|
||||
complete = review_base + "\n" + "\n".join([
|
||||
"- Selected PR: #999",
|
||||
"- Reviewer eligibility: passed",
|
||||
"- Pinned reviewed head: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
|
||||
@@ -958,6 +964,15 @@ class TestControllerHandoff(unittest.TestCase):
|
||||
"- Merge result: merged",
|
||||
"- Linked issue status: closed",
|
||||
"- Cleanup status: branch deleted",
|
||||
"- File edits by reviewer: none",
|
||||
"- Worktree/index mutations: created and removed review worktree",
|
||||
"- Git ref mutations: git fetch prgs",
|
||||
"- MCP/Gitea mutations: none",
|
||||
"- Review mutations: one approve on #999",
|
||||
"- Merge mutations: PR #999 merged",
|
||||
"- Cleanup mutations: removed review worktree",
|
||||
"- External-state mutations: none",
|
||||
"- Read-only diagnostics: git log, git diff",
|
||||
])
|
||||
result = assess_controller_handoff(complete, role="review")
|
||||
self.assertEqual(result["verdict"], "complete")
|
||||
@@ -1080,6 +1095,144 @@ class TestControllerHandoff(unittest.TestCase):
|
||||
self.assertEqual(res2["verdict"], "complete")
|
||||
|
||||
|
||||
class TestReviewHandoffPreciseMutationCategories(unittest.TestCase):
|
||||
"""Issue #320: reviewer handoffs drop legacy 'Workspace mutations'."""
|
||||
|
||||
REVIEW_BASE = "\n".join([
|
||||
"## Controller Handoff",
|
||||
"",
|
||||
"- Task: review PR #999",
|
||||
"- Repo: Scaled-Tech-Consulting/Gitea-Tools",
|
||||
"- Role: reviewer",
|
||||
"- Identity: jcwalker3 / prgs-reviewer",
|
||||
"- Issue/PR: PR #999",
|
||||
"- Branch/SHA: feat/x @ 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
|
||||
"- Files changed: review_proofs.py",
|
||||
"- Validation: 700 passed, 6 skipped",
|
||||
"- Mutations: one review submitted",
|
||||
"- Current status: review complete",
|
||||
"- Blockers: none",
|
||||
"- Next: controller decides",
|
||||
"- Safety: no self-review; no self-merge; no secrets",
|
||||
"- Selected PR: #999",
|
||||
"- Reviewer eligibility: passed",
|
||||
"- Pinned reviewed head: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
|
||||
"- Worktree path: /repo/branches/review-pr-999",
|
||||
"- Worktree dirty: no",
|
||||
"- Scratch worktree used: yes (/repo/branches/review-pr-999)",
|
||||
"- Unrelated local mutations: none",
|
||||
"- Review decision: approve",
|
||||
"- Merge result: none",
|
||||
"- Linked issue status: open",
|
||||
"- Cleanup status: worktree removed",
|
||||
])
|
||||
|
||||
PRECISE_CATEGORIES = "\n".join([
|
||||
"- File edits by reviewer: none",
|
||||
"- Worktree/index mutations: created and removed review worktree",
|
||||
"- Git ref mutations: git fetch prgs",
|
||||
"- MCP/Gitea mutations: none",
|
||||
"- Review mutations: one approve on #999",
|
||||
"- Merge mutations: none",
|
||||
"- Cleanup mutations: removed review worktree",
|
||||
"- External-state mutations: none",
|
||||
"- Read-only diagnostics: git log, git diff",
|
||||
])
|
||||
|
||||
def _complete_handoff(self, **overrides):
|
||||
text = self.REVIEW_BASE + "\n" + self.PRECISE_CATEGORIES
|
||||
for old, new in overrides.items():
|
||||
text = text.replace(old, new)
|
||||
return text
|
||||
|
||||
def test_review_handoff_with_precise_categories_is_complete(self):
|
||||
result = assess_controller_handoff(
|
||||
self._complete_handoff(), role="review")
|
||||
self.assertEqual(result["verdict"], "complete")
|
||||
self.assertFalse(result["downgraded"])
|
||||
|
||||
def test_review_handoff_rejects_legacy_workspace_mutations_none(self):
|
||||
text = self._complete_handoff() + "\n- Workspace mutations: none"
|
||||
result = assess_controller_handoff(text, role="review")
|
||||
self.assertEqual(result["verdict"], "incomplete")
|
||||
self.assertTrue(result["downgraded"])
|
||||
self.assertTrue(any(
|
||||
"workspace mutations" in reason.lower()
|
||||
for reason in result["reasons"]
|
||||
))
|
||||
|
||||
def test_review_handoff_rejects_legacy_workspace_mutations_any_value(self):
|
||||
text = (self._complete_handoff()
|
||||
+ "\n- Workspace mutations: edited review_proofs.py")
|
||||
result = assess_controller_handoff(text, role="review")
|
||||
self.assertEqual(result["verdict"], "incomplete")
|
||||
self.assertTrue(any(
|
||||
"workspace mutations" in reason.lower()
|
||||
for reason in result["reasons"]
|
||||
))
|
||||
|
||||
def test_review_handoff_missing_precise_categories_is_incomplete(self):
|
||||
result = assess_controller_handoff(self.REVIEW_BASE, role="review")
|
||||
self.assertEqual(result["verdict"], "incomplete")
|
||||
self.assertIn("File edits by reviewer", result["missing_fields"])
|
||||
self.assertIn("Worktree/index mutations", result["missing_fields"])
|
||||
self.assertIn("Read-only diagnostics", result["missing_fields"])
|
||||
|
||||
def test_review_handoff_does_not_require_workspace_mutations(self):
|
||||
result = assess_controller_handoff(
|
||||
self._complete_handoff(), role="review")
|
||||
self.assertNotIn("Workspace mutations", result["missing_fields"])
|
||||
|
||||
def test_request_changes_handoff_with_precise_categories_is_complete(self):
|
||||
text = self._complete_handoff(**{
|
||||
"- Review decision: approve": "- Review decision: request-changes",
|
||||
"- Review mutations: one approve on #999":
|
||||
"- Review mutations: one request_changes on #999",
|
||||
})
|
||||
result = assess_controller_handoff(text, role="review")
|
||||
self.assertEqual(result["verdict"], "complete")
|
||||
|
||||
def test_merge_handoff_with_precise_categories_is_complete(self):
|
||||
text = self._complete_handoff(**{
|
||||
"- Merge result: none": "- Merge result: merged",
|
||||
"- Merge mutations: none": "- Merge mutations: PR #999 merged",
|
||||
})
|
||||
result = assess_controller_handoff(text, role="review")
|
||||
self.assertEqual(result["verdict"], "complete")
|
||||
|
||||
def test_fetch_only_handoff_with_precise_categories_is_complete(self):
|
||||
text = self._complete_handoff(**{
|
||||
"- Review decision: approve": "- Review decision: none",
|
||||
"- Review mutations: one approve on #999": "- Review mutations: none",
|
||||
"- Worktree/index mutations: created and removed review worktree":
|
||||
"- Worktree/index mutations: none",
|
||||
"- Cleanup mutations: removed review worktree":
|
||||
"- Cleanup mutations: none",
|
||||
"- Mutations: one review submitted": "- Mutations: fetch only",
|
||||
})
|
||||
result = assess_controller_handoff(text, role="review")
|
||||
self.assertEqual(result["verdict"], "complete")
|
||||
|
||||
def test_blocked_handoff_with_precise_categories_is_complete(self):
|
||||
text = self._complete_handoff(**{
|
||||
"- Blockers: none": "- Blockers: infra_stop (registry unreachable)",
|
||||
"- Review decision: approve": "- Review decision: none",
|
||||
"- Review mutations: one approve on #999": "- Review mutations: none",
|
||||
})
|
||||
result = assess_controller_handoff(text, role="review")
|
||||
self.assertEqual(result["verdict"], "complete")
|
||||
|
||||
def test_author_role_still_requires_workspace_mutations(self):
|
||||
# Non-review roles keep the legacy contract until their own issues
|
||||
# migrate them.
|
||||
author = TestControllerHandoff.BASE_HANDOFF
|
||||
stripped = "\n".join(
|
||||
line for line in author.splitlines()
|
||||
if not line.startswith("- Workspace mutations:"))
|
||||
result = assess_controller_handoff(stripped, role="author")
|
||||
self.assertEqual(result["verdict"], "incomplete")
|
||||
self.assertIn("Workspace mutations", result["missing_fields"])
|
||||
|
||||
|
||||
class TestReviewMutationFinalReport(unittest.TestCase):
|
||||
"""Final reports must list exactly one live review mutation."""
|
||||
@@ -2161,6 +2314,111 @@ class TestCreateIssueFinalReport(unittest.TestCase):
|
||||
self.assertTrue(result["downgraded"])
|
||||
|
||||
|
||||
class TestQueueOrderingProof(unittest.TestCase):
|
||||
"""Issue #321: reviewer queue selection must prove ordering."""
|
||||
|
||||
def test_next_eligible_claim_without_policy_fails(self):
|
||||
result = assess_queue_ordering_proof(
|
||||
report_text="Selected the next eligible PR: PR #286.",
|
||||
)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertTrue(result["claims_selection"])
|
||||
|
||||
def test_next_eligible_claim_with_policy_passes(self):
|
||||
result = assess_queue_ordering_proof(
|
||||
report_text=(
|
||||
"Queue ordering policy: oldest PR number first\n"
|
||||
"Selected the next eligible PR: PR #286."
|
||||
),
|
||||
)
|
||||
self.assertTrue(result["proven"])
|
||||
self.assertEqual(result["policy"], "oldest PR number first")
|
||||
|
||||
def test_newest_first_api_with_oldest_first_selection_passes(self):
|
||||
result = assess_queue_ordering_proof(
|
||||
report_text=(
|
||||
"Queue ordering policy: oldest PR number first\n"
|
||||
"API order was newest-first (PR #291 before PR #286).\n"
|
||||
"Selected the next eligible PR: PR #286."
|
||||
),
|
||||
queue_ordering={
|
||||
"policy": "oldest PR number first",
|
||||
"selected_pr_number": 286,
|
||||
"api_pr_numbers": [291, 286],
|
||||
},
|
||||
)
|
||||
self.assertTrue(result["proven"])
|
||||
|
||||
def test_earlier_actionable_pr_without_skip_reason_fails(self):
|
||||
result = assess_queue_ordering_proof(
|
||||
report_text=(
|
||||
"Queue ordering policy: oldest PR number first\n"
|
||||
"Selected the next eligible PR: PR #300."
|
||||
),
|
||||
queue_ordering={
|
||||
"policy": "oldest PR number first",
|
||||
"selected_pr_number": 300,
|
||||
"api_pr_numbers": [300, 286],
|
||||
},
|
||||
)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertTrue(result["block"])
|
||||
|
||||
def test_priority_override_requires_explanation(self):
|
||||
result = assess_queue_ordering_proof(
|
||||
report_text="Selected the next eligible PR: PR #300.",
|
||||
queue_ordering={
|
||||
"policy": "priority label",
|
||||
"selected_pr_number": 300,
|
||||
"api_pr_numbers": [286, 300],
|
||||
},
|
||||
)
|
||||
self.assertFalse(result["proven"])
|
||||
|
||||
def test_priority_override_with_explanation_passes(self):
|
||||
result = assess_queue_ordering_proof(
|
||||
report_text=(
|
||||
"Queue ordering policy: priority label\n"
|
||||
"Priority override: security label on PR #300.\n"
|
||||
"Selected the next eligible PR: PR #300."
|
||||
),
|
||||
queue_ordering={
|
||||
"policy": "priority label",
|
||||
"selected_pr_number": 300,
|
||||
"api_pr_numbers": [286, 300],
|
||||
"priority_override": "security label on PR #300",
|
||||
},
|
||||
)
|
||||
self.assertTrue(result["proven"])
|
||||
|
||||
def test_build_final_report_downgrades_missing_ordering_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="Selected the next eligible PR: PR #286.",
|
||||
)
|
||||
self.assertNotEqual(report["grade"], "A")
|
||||
self.assertFalse(report["queue_ordering_proven"])
|
||||
|
||||
class TestReviewerBaselineValidationProof(unittest.TestCase):
|
||||
"""Issue #325: baseline validation must use branches/ worktrees."""
|
||||
|
||||
|
||||
@@ -0,0 +1,165 @@
|
||||
"""Tests for workspace-vs-worktree mutation consistency verifier (#313).
|
||||
|
||||
Final reports must not claim ``Workspace mutations: none`` when worktree
|
||||
mutations (reset --hard, checkout, clean, worktree add/remove, ...)
|
||||
occurred, and must report observed worktree / git ref mutations under the
|
||||
precise category fields.
|
||||
"""
|
||||
import sys
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
from review_proofs import assess_workspace_mutation_consistency # noqa: E402
|
||||
|
||||
|
||||
class TestWorkspaceNoneContradiction(unittest.TestCase):
|
||||
def test_reset_hard_with_workspace_none_fails(self):
|
||||
report = (
|
||||
"Controller Handoff\n"
|
||||
"- Workspace mutations: none (no local file edits)\n"
|
||||
)
|
||||
result = assess_workspace_mutation_consistency(
|
||||
report,
|
||||
observed_commands=["git reset --hard FETCH_HEAD"],
|
||||
)
|
||||
self.assertFalse(result["complete"])
|
||||
self.assertTrue(result["downgraded"])
|
||||
self.assertTrue(
|
||||
any("workspace mutations" in r.lower() for r in result["reasons"])
|
||||
)
|
||||
|
||||
def test_checkout_with_workspace_none_fails(self):
|
||||
report = "- Workspace mutations: none\n"
|
||||
result = assess_workspace_mutation_consistency(
|
||||
report,
|
||||
observed_commands=["git checkout feat/some-branch"],
|
||||
)
|
||||
self.assertFalse(result["complete"])
|
||||
|
||||
def test_worktree_add_with_workspace_none_fails(self):
|
||||
report = "- Workspace mutations: none\n"
|
||||
result = assess_workspace_mutation_consistency(
|
||||
report,
|
||||
observed_commands=["git worktree add /tmp/review-pr1 abc123"],
|
||||
)
|
||||
self.assertFalse(result["complete"])
|
||||
|
||||
def test_clean_with_workspace_none_fails(self):
|
||||
report = "- Workspace mutations: none\n"
|
||||
result = assess_workspace_mutation_consistency(
|
||||
report,
|
||||
observed_commands=["git clean -fd"],
|
||||
)
|
||||
self.assertFalse(result["complete"])
|
||||
|
||||
def test_self_reported_worktree_mutation_contradicts_workspace_none(self):
|
||||
# No observed command log; the report itself carries the
|
||||
# contradiction (#313 observed behavior).
|
||||
report = (
|
||||
"- Worktree mutations: git reset --hard FETCH_HEAD\n"
|
||||
"- Workspace mutations: none (no local file edits)\n"
|
||||
)
|
||||
result = assess_workspace_mutation_consistency(report)
|
||||
self.assertFalse(result["complete"])
|
||||
self.assertTrue(
|
||||
any("workspace mutations" in r.lower() for r in result["reasons"])
|
||||
)
|
||||
|
||||
|
||||
class TestPreciseCategoriesPass(unittest.TestCase):
|
||||
def test_file_edits_none_with_worktree_mutation_listed_passes(self):
|
||||
report = (
|
||||
"- File edits by reviewer: none\n"
|
||||
"- Worktree mutations: git reset --hard FETCH_HEAD\n"
|
||||
"- Git ref mutations: git fetch prgs\n"
|
||||
)
|
||||
result = assess_workspace_mutation_consistency(
|
||||
report,
|
||||
observed_commands=[
|
||||
"git fetch prgs",
|
||||
"git reset --hard FETCH_HEAD",
|
||||
],
|
||||
)
|
||||
self.assertTrue(result["complete"])
|
||||
self.assertFalse(result["downgraded"])
|
||||
self.assertEqual(result["reasons"], [])
|
||||
|
||||
def test_noop_report_passes(self):
|
||||
report = (
|
||||
"- File edits by reviewer: none\n"
|
||||
"- Worktree mutations: none\n"
|
||||
"- Git ref mutations: none\n"
|
||||
)
|
||||
result = assess_workspace_mutation_consistency(
|
||||
report, observed_commands=[]
|
||||
)
|
||||
self.assertTrue(result["complete"])
|
||||
self.assertEqual(result["reasons"], [])
|
||||
|
||||
def test_fetch_only_with_ref_mutation_reported_passes(self):
|
||||
# Fetch is a git ref mutation, not a worktree mutation; a
|
||||
# workspace-none claim is not contradicted by fetch alone.
|
||||
report = (
|
||||
"- Workspace mutations: none\n"
|
||||
"- Git ref mutations: git fetch prgs master\n"
|
||||
)
|
||||
result = assess_workspace_mutation_consistency(
|
||||
report,
|
||||
observed_commands=["git fetch prgs master"],
|
||||
)
|
||||
self.assertTrue(result["complete"])
|
||||
|
||||
|
||||
class TestObservedMutationsMustBeReported(unittest.TestCase):
|
||||
def test_reset_without_worktree_mutation_field_fails(self):
|
||||
report = "- File edits by reviewer: none\n"
|
||||
result = assess_workspace_mutation_consistency(
|
||||
report,
|
||||
observed_commands=["git reset --hard FETCH_HEAD"],
|
||||
)
|
||||
self.assertFalse(result["complete"])
|
||||
self.assertTrue(
|
||||
any("worktree mutation" in r.lower() for r in result["reasons"])
|
||||
)
|
||||
|
||||
def test_reset_with_worktree_mutations_none_fails(self):
|
||||
report = (
|
||||
"- File edits by reviewer: none\n"
|
||||
"- Worktree mutations: none\n"
|
||||
)
|
||||
result = assess_workspace_mutation_consistency(
|
||||
report,
|
||||
observed_commands=["git reset --hard FETCH_HEAD"],
|
||||
)
|
||||
self.assertFalse(result["complete"])
|
||||
|
||||
def test_fetch_without_ref_mutation_field_fails(self):
|
||||
report = (
|
||||
"- File edits by reviewer: none\n"
|
||||
"- Worktree mutations: none\n"
|
||||
)
|
||||
result = assess_workspace_mutation_consistency(
|
||||
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"])
|
||||
)
|
||||
|
||||
def test_fetch_with_ref_mutations_none_fails(self):
|
||||
report = (
|
||||
"- Worktree mutations: none\n"
|
||||
"- Git ref mutations: none\n"
|
||||
)
|
||||
result = assess_workspace_mutation_consistency(
|
||||
report,
|
||||
observed_commands=["git fetch prgs master"],
|
||||
)
|
||||
self.assertFalse(result["complete"])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user