"""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", "- File edits by reviewer: none", "- Worktree/index mutations: none", "- Git ref mutations: git fetch prgs master", "- MCP/Gitea mutations: review submitted", "- Review mutations: submitted request_changes review", "- Merge mutations: none", "- Cleanup mutations: none", "- External-state mutations: none", "- Read-only diagnostics: issue and PR metadata inspected", "- Current status: PR open", "- Next actor: author", "- Next action: address review feedback", "- Next prompt: Fix PR #203 per reviewer request_changes and push updated head", "- Blockers: none", "- Next: await author", "- Safety: no self-review; no self-merge", "- Selected PR: #203", "- Reviewer eligibility: eligible", "- Pinned reviewed head: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9", "- Reviewed head SHA: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9", "- Final live head SHA before approval: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9", "- Final live head SHA before merge: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9", "- Push occurred during validation: no", "- 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(drop=(), **overrides): lines = [ "## Controller Handoff", "", "- Task: reconcile already-landed PR #99", "- Repo: Scaled-Tech-Consulting/Gitea-Tools", "- Role/profile: reconciler / prgs-reconciler", "- Identity: sysadmin", "- Selected PR: #99", "- PR live state: open", "- Candidate head SHA: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9", "- Target branch: master", "- Target branch SHA: 5e023dc71b0e2b813a0b1eee6b0f42630c47a95c", "- Ancestor proof: candidate head is ancestor of target branch SHA", "- Linked issue: #98", "- Linked issue live status: not verified in this session", "- Eligibility class: ALREADY_LANDED_RECONCILE_REQUIRED", "- Capabilities proven: gitea.read, gitea.pr.comment", "- Missing capabilities: gitea.pr.close", "- PR comments posted: 1 reconciliation comment on PR #99", "- Issue comments posted: none", "- PRs closed: none", "- Issues closed: none", "- File edits by reconciler: none", "- Git ref mutations: git fetch prgs master", "- Worktree mutations: none", "- MCP/Gitea mutations: PR comment posted", "- External-state mutations: none", "- Read-only diagnostics: gitea_view_pr, gitea_view_issue", "- Reconciliation mutations: PR comment posted", "- Current status: reconciliation complete", "- Next actor: reconciler", "- Next action: close PR #99 after capability proof", "- Next prompt: Reconcile already-landed PR #99 with prgs-reconciler profile", "- Blocker: missing gitea.pr.close capability", "- Safe next action: hand off to a profile with gitea.pr.close", "- No review/merge confirmation: confirmed", ] kept = [ line for line in lines if not any(line.startswith(f"- {name}:") for name in drop) ] text = "\n".join(kept) for key, value in overrides.items(): if f"- {key}:" in text: text = text.replace(f"- {key}:", f"- {key}: {value}") return text class TestTwoCommentWorkflowIntegration(unittest.TestCase): def test_tagged_pair_produces_no_two_comment_findings(self): from thread_state_ledger_validator import findings_for_final_report # noqa: E402 from tests.test_thread_state_ledger_validator import ( # noqa: E402 _minimal_handoff, _minimal_ledger, ) report = _minimal_handoff() + "\n\n" + _minimal_ledger() self.assertEqual(findings_for_final_report(report), []) def test_tagged_handoff_without_ledger_blocks_via_validator(self): from thread_state_ledger_validator import findings_for_final_report # noqa: E402 from tests.test_thread_state_ledger_validator import _minimal_handoff # noqa: E402 findings = findings_for_final_report(_minimal_handoff()) self.assertTrue(findings) self.assertTrue( any(f["severity"] == "block" for f in findings) ) 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_already_landed_oldest_eligible_pr_blocks(self): report = ( _review_handoff() + "\nAlready landed.\nOldest eligible PR: PR #278." ) 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_already_landed_gate_blocks_review_terminal_states(self): cases = { "APPROVED": ("- Review decision: request_changes", "- Review decision: APPROVED"), "MERGED": ("- Merge result: not attempted", "- Merge result: MERGED"), "READY_TO_MERGE": ("- Current status: PR open", "- Current status: READY_TO_MERGE"), } for state, (old, new) in cases.items(): with self.subTest(state=state): report = ( _review_handoff() .replace("- Reviewer eligibility: eligible", "- Reviewer eligibility: blocked") .replace(old, new) + "\n- Already-landed gate: fired" + "\n- Ancestor proof: passed" + "\n- Eligibility class: ALREADY_LANDED_RECONCILE_REQUIRED" ) result = assess_final_report_validator(report, "review_pr") self.assertTrue(result["blocked"]) self.assertTrue( any( f["rule_id"] == "reviewer.already_landed_review_state" for f in result["findings"] ) ) def test_target_branch_up_to_date_requires_fetch_and_sha(self): report = ( _review_handoff() .replace("- Git ref mutations: git fetch prgs master", "- Git ref mutations: none") + "\nTarget branch up to date." ) result = assess_final_report_validator(report, "review_pr") self.assertTrue(result["blocked"]) self.assertTrue( any( f["rule_id"] == "reviewer.target_branch_freshness_proof" for f in result["findings"] ) ) def test_target_branch_up_to_date_with_fetch_and_sha_passes(self): report = ( _review_handoff() + "\n- Target branch SHA: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9" + "\nTarget branch up to date." ) result = assess_final_report_validator(report, "review_pr") self.assertEqual(result["grade"], "A") 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") .replace( "- Read-only diagnostics: issue and PR metadata inspected", "- 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 TestCanonicalReconcileSchema(unittest.TestCase): """Issue #307: reconciliation workflows emit only the canonical schema.""" def test_comment_only_reconciliation_passes(self): result = assess_final_report_validator( _reconcile_handoff(), "reconcile_already_landed", ) self.assertEqual(result["grade"], "A") self.assertEqual(result["findings"], []) def test_successful_close_reconciliation_passes(self): report = _reconcile_handoff().replace( "- Capabilities proven: gitea.read, gitea.pr.comment", "- Capabilities proven: gitea.read, gitea.pr.comment, gitea.pr.close", ).replace( "- Missing capabilities: gitea.pr.close", "- Missing capabilities: none", ).replace( "- PRs closed: none", "- PRs closed: #99", ).replace( "- Blocker: missing gitea.pr.close capability", "- Blocker: none", ) result = assess_final_report_validator(report, "reconcile_already_landed") self.assertEqual(result["grade"], "A") def test_blocked_reconciliation_passes(self): report = _reconcile_handoff().replace( "- Capabilities proven: gitea.read, gitea.pr.comment", "- Capabilities proven: gitea.read", ).replace( "- Missing capabilities: gitea.pr.close", "- Missing capabilities: gitea.pr.close, gitea.pr.comment", ).replace( "- PR comments posted: 1 reconciliation comment on PR #99", "- PR comments posted: none", ).replace( "- MCP/Gitea mutations: PR comment posted", "- MCP/Gitea mutations: none", ).replace( "- Reconciliation mutations: PR comment posted", "- Reconciliation mutations: none", ) result = assess_final_report_validator(report, "reconcile_already_landed") self.assertEqual(result["grade"], "A") def test_missing_capabilities_field_required(self): report = _reconcile_handoff(drop=("Missing capabilities",)) result = assess_final_report_validator(report, "reconcile_already_landed") self.assertEqual(result["grade"], "downgraded") self.assertTrue( any( f["rule_id"] == "reconcile.controller_handoff" and "Missing capabilities" in f["reason"] for f in result["findings"] ) ) def test_missing_ancestor_proof_and_candidate_sha_downgrade(self): report = _reconcile_handoff(drop=("Ancestor proof", "Candidate head SHA")) result = assess_final_report_validator(report, "reconcile_already_landed") self.assertEqual(result["grade"], "downgraded") reasons = " ".join(f["reason"] for f in result["findings"]) self.assertIn("Ancestor proof", reasons) self.assertIn("Candidate head SHA", reasons) def test_stale_issue_lock_proof_blocks(self): report = _reconcile_handoff() + "\n- Issue lock proof: locked before diff" result = assess_final_report_validator(report, "reconcile_already_landed") self.assertTrue(result["blocked"]) self.assertTrue( any( f["rule_id"] == "reconcile.stale_author_reviewer_fields" and f["field"] == "issue lock proof" for f in result["findings"] ) ) def test_stale_claim_comment_status_blocks(self): report = _reconcile_handoff() + "\n- Claim/comment status: claimed" result = assess_final_report_validator(report, "reconcile_already_landed") self.assertTrue(result["blocked"]) self.assertTrue( any( f["rule_id"] == "reconcile.stale_author_reviewer_fields" and f["field"] == "claim/comment status" for f in result["findings"] ) ) def test_stale_workspace_mutations_blocks_without_observed_mutations(self): report = _reconcile_handoff() + "\n- Workspace mutations: None" result = assess_final_report_validator(report, "reconcile_already_landed") self.assertTrue(result["blocked"]) self.assertTrue( any( f["rule_id"] == "reconcile.stale_author_reviewer_fields" and f["field"] == "workspace mutations" for f in result["findings"] ) ) def test_pr_number_opened_allowed_when_session_opened_pr(self): report = _reconcile_handoff() + "\n- PR number opened: #12" result = assess_final_report_validator( report, "reconcile_already_landed", session_pr_opened=True, ) self.assertFalse( any( f["rule_id"] == "reconcile.stale_author_reviewer_fields" and f["field"] == "pr number opened" for f in result["findings"] ) ) def test_pr_number_opened_still_blocked_without_session_proof(self): report = _reconcile_handoff() + "\n- PR number opened: #12" result = assess_final_report_validator( report, "reconcile_already_landed", session_pr_opened=False, ) self.assertTrue(result["blocked"]) self.assertTrue( any( f["rule_id"] == "reconcile.stale_author_reviewer_fields" and f["field"] == "pr number opened" for f in result["findings"] ) ) def test_action_log_fetch_requires_git_ref_mutation_entry(self): report = _reconcile_handoff().replace( "- Git ref mutations: git fetch prgs master", "- Git ref mutations: none", ) result = assess_final_report_validator( report, "reconcile_already_landed", action_log=[{"command": "git fetch prgs master", "performed": True}], ) self.assertTrue( any( f["rule_id"] == "reviewer.git_fetch_readonly" for f in result["findings"] ) ) class TestReconcilerCloseProof(unittest.TestCase): """Issue #306: a reconciler PR close must carry its proof fields.""" def _closed_report(self, **kwargs): # A report that claims PR #99 was closed via the reconciler path. report = ( _reconcile_handoff(**kwargs) .replace( "- Capabilities proven: gitea.read, gitea.pr.comment", "- Capabilities proven: gitea.read, gitea.pr.comment, gitea.pr.close", ) .replace("- Missing capabilities: gitea.pr.close", "- Missing capabilities: none") .replace("- PRs closed: none", "- PRs closed: #99") .replace( "- Blocker: missing gitea.pr.close capability", "- Blocker: none" ) ) return report def test_full_proof_close_passes(self): result = assess_final_report_validator( self._closed_report(), "reconcile_already_landed" ) self.assertEqual(result["grade"], "A") self.assertFalse( any(f["rule_id"] == "reconcile.close_proof_fields" for f in result["findings"]) ) def test_close_without_capability_proof_blocks(self): # Claims PRs closed: #99 but never proves gitea.pr.close capability. report = _reconcile_handoff().replace("- PRs closed: none", "- PRs closed: #99") result = assess_final_report_validator(report, "reconcile_already_landed") self.assertTrue(result["blocked"]) self.assertTrue( any(f["rule_id"] == "reconcile.close_proof_fields" for f in result["findings"]) ) def test_close_without_ancestor_proof_blocks(self): report = self._closed_report(drop=("Ancestor proof",)) result = assess_final_report_validator(report, "reconcile_already_landed") self.assertTrue(result["blocked"]) self.assertTrue( any(f["rule_id"] == "reconcile.close_proof_fields" for f in result["findings"]) ) def test_close_without_linked_issue_result_blocks(self): report = self._closed_report(drop=("Linked issue live status", "Issues closed")) result = assess_final_report_validator(report, "reconcile_already_landed") self.assertTrue(result["blocked"]) self.assertTrue( any(f["rule_id"] == "reconcile.close_proof_fields" for f in result["findings"]) ) def test_close_lock_requires_proof_even_if_text_says_none(self): # Session lock proves a PR was closed; the report omits close proof. result = assess_final_report_validator( _reconcile_handoff(), "reconcile_already_landed", reconciler_close_lock={"pr_closed": True}, ) self.assertTrue(result["blocked"]) self.assertTrue( any(f["rule_id"] == "reconcile.close_proof_fields" for f in result["findings"]) ) def test_comment_only_reconcile_needs_no_close_proof(self): result = assess_final_report_validator( _reconcile_handoff(), "reconcile_already_landed" ) self.assertEqual(result["grade"], "A") self.assertFalse( any(f["rule_id"] == "reconcile.close_proof_fields" 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()