"""Tests for blind PR queue review proofs (Issue #173). Issue #173 requires the reviewer-side workflow to *prove* — not assume — each precondition of a blind queue review: 1. Pinned PR head vs local checkout HEAD (checkout proof). 2. Complete open-PR inventory across all configured repositories. 3. Evidence-strength of validation reporting (exact commands, counts, justified ignores, output actually read). 4. Evidence-backed self-review contamination assessment (never assumed). 5. A final report that distinguishes each proof and can only claim an "A"-grade run when every proof is present; otherwise it downgrades or blocks instead of merging confidently. These are the harness assertions from the issue's Required behavior 7. """ import io import sys import unittest sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent)) from review_proofs import ( # noqa: E402 assess_controller_handoff, assess_inventory_completeness, assess_self_review_contamination, assess_validation_report, build_final_report, resolve_repos_from_user_reference, verify_pinned_head_checkout, ) PINNED = "0fdc8f582026b72a229d59a172c0a63ac4aaeaf9" OTHER = "a4060c5de00f2b1c9e88f4f6f0f3f9a7b2c1d0e9" def _good_checkout(): return verify_pinned_head_checkout( pinned_head_sha=PINNED, local_head_sha=PINNED, pr_base_ref="master", diff_base_ref="prgs/master", ) def _good_inventory(): return assess_inventory_completeness( repo_reports=[ { "repo": "Scaled-Tech-Consulting/Gitea-Tools", "state_filter": "open", "pagination_complete": True, "open_pr_count": 2, }, { "repo": "Scaled-Tech-Consulting/mcp-control-plane", "state_filter": "open", "pagination_complete": True, "open_pr_count": 1, }, ], required_repos=[ "Scaled-Tech-Consulting/Gitea-Tools", "Scaled-Tech-Consulting/mcp-control-plane", ], ) def _good_validation(**overrides): report = { "command": "venv/bin/pytest tests/ --ignore=branches", "output_read": True, "result": "pass", "passed": 421, "failed": 0, "skipped": 13, "ignored_paths": [ { "path": "branches", "justification": ( "review worktrees under branches/ duplicate test module " "names and break collection; they are not part of the " "repository's own suite" ), } ], "canonical_command": "venv/bin/pytest tests/ --ignore=branches", } report.update(overrides) return assess_validation_report(report) def _good_contamination(): return assess_self_review_contamination( reviewer_identity="sysadmin", pr_author="jcwalker3", session_authored=False, evidence_source="session transcript: branch never checked out or pushed here", ) class TestCheckoutProof(unittest.TestCase): """Required behavior 1 + 2: prove HEAD == pinned PR head or stop.""" def test_matching_pinned_and_local_head_is_proven(self): proof = _good_checkout() self.assertTrue(proof["proven"]) self.assertFalse(proof["block"]) def test_fetched_sha_but_unchecked_out_head_blocks(self): # Harness assertion (behavior 7, bullet 1): fetching/pinning a SHA # without checking it out must block review and merge. proof = verify_pinned_head_checkout( pinned_head_sha=PINNED, local_head_sha=OTHER, pr_base_ref="master", diff_base_ref="prgs/master", ) self.assertFalse(proof["proven"]) self.assertTrue(proof["block"]) self.assertTrue(any("HEAD" in r for r in proof["reasons"])) def test_missing_pinned_sha_fails_closed(self): proof = verify_pinned_head_checkout( pinned_head_sha="", local_head_sha=PINNED, pr_base_ref="master", diff_base_ref="master", ) self.assertFalse(proof["proven"]) self.assertTrue(proof["block"]) def test_missing_local_head_fails_closed(self): proof = verify_pinned_head_checkout( pinned_head_sha=PINNED, local_head_sha=None, pr_base_ref="master", diff_base_ref="master", ) self.assertFalse(proof["proven"]) self.assertTrue(proof["block"]) def test_abbreviated_sha_is_not_proof(self): # A prefix match is not proof of the exact pinned head. proof = verify_pinned_head_checkout( pinned_head_sha=PINNED, local_head_sha=PINNED[:12], pr_base_ref="master", diff_base_ref="master", ) self.assertFalse(proof["proven"]) self.assertTrue(proof["block"]) def test_wrong_diff_base_blocks(self): proof = verify_pinned_head_checkout( pinned_head_sha=PINNED, local_head_sha=PINNED, pr_base_ref="master", diff_base_ref="develop", ) self.assertFalse(proof["proven"]) self.assertTrue(proof["block"]) self.assertTrue(any("base" in r.lower() for r in proof["reasons"])) def test_remote_tracking_diff_base_matches_pr_base(self): proof = verify_pinned_head_checkout( pinned_head_sha=PINNED, local_head_sha=PINNED, pr_base_ref="master", diff_base_ref="refs/remotes/prgs/master", ) self.assertTrue(proof["proven"]) class TestInventoryCompleteness(unittest.TestCase): """Required behavior 3: inventory must prove completeness.""" def test_complete_multi_repo_inventory(self): # Harness assertion (behavior 7, bullet 2): complete queue inventory # with multiple PRs across repos. result = _good_inventory() self.assertTrue(result["complete"]) self.assertTrue(result["can_claim_exhaustive"]) self.assertEqual( result["total_open_prs"], { "Scaled-Tech-Consulting/Gitea-Tools": 2, "Scaled-Tech-Consulting/mcp-control-plane": 1, }, ) def test_missing_repo_makes_inventory_incomplete(self): result = assess_inventory_completeness( repo_reports=[ { "repo": "Scaled-Tech-Consulting/Gitea-Tools", "state_filter": "open", "pagination_complete": True, "open_pr_count": 1, } ], required_repos=[ "Scaled-Tech-Consulting/Gitea-Tools", "Scaled-Tech-Consulting/mcp-control-plane", ], ) self.assertFalse(result["complete"]) self.assertFalse(result["can_claim_exhaustive"]) self.assertTrue( any("mcp-control-plane" in r for r in result["reasons"]) ) def test_unfinished_pagination_blocks_exhaustive_claim(self): # Harness assertion (behavior 7, bullet 3): pagination / multi-page # PR listing must be handled or the claim is refused. result = assess_inventory_completeness( repo_reports=[ { "repo": "Scaled-Tech-Consulting/Gitea-Tools", "state_filter": "open", "pagination_complete": False, "open_pr_count": 50, } ], required_repos=["Scaled-Tech-Consulting/Gitea-Tools"], ) self.assertFalse(result["complete"]) self.assertFalse(result["can_claim_exhaustive"]) self.assertTrue(any("pagination" in r.lower() for r in result["reasons"])) def test_missing_state_filter_blocks_exhaustive_claim(self): result = assess_inventory_completeness( repo_reports=[ { "repo": "Scaled-Tech-Consulting/Gitea-Tools", "pagination_complete": True, "open_pr_count": 1, } ], required_repos=["Scaled-Tech-Consulting/Gitea-Tools"], ) self.assertFalse(result["complete"]) def test_missing_count_blocks_exhaustive_claim(self): result = assess_inventory_completeness( repo_reports=[ { "repo": "Scaled-Tech-Consulting/Gitea-Tools", "state_filter": "open", "pagination_complete": True, } ], required_repos=["Scaled-Tech-Consulting/Gitea-Tools"], ) self.assertFalse(result["complete"]) def test_empty_inventory_fails_closed(self): result = assess_inventory_completeness( repo_reports=[], required_repos=["Scaled-Tech-Consulting/Gitea-Tools"], ) self.assertFalse(result["complete"]) self.assertFalse(result["can_claim_exhaustive"]) class TestValidationReporting(unittest.TestCase): """Required behavior 4: exact commands and exact results.""" def test_full_report_is_strong(self): result = _good_validation() self.assertEqual(result["verdict"], "strong") self.assertTrue(result["claimable"]) def test_missing_test_counts_is_weak(self): # Harness assertion (behavior 7, bullet 4): validation result missing # test counts is flagged as weak/incomplete. result = _good_validation(passed=None, failed=None, skipped=None) self.assertEqual(result["verdict"], "weak") self.assertTrue(any("count" in r.lower() for r in result["reasons"])) def test_output_not_read_cannot_claim_result(self): # Harness assertion (behavior 7, bullet 6): a report cannot say # validation passed unless the command output was actually read. result = _good_validation(output_read=False) self.assertEqual(result["verdict"], "invalid") self.assertFalse(result["claimable"]) def test_missing_command_cannot_claim_result(self): result = _good_validation(command="") self.assertEqual(result["verdict"], "invalid") self.assertFalse(result["claimable"]) def test_unjustified_ignored_path_is_weak(self): # Harness assertion (behavior 7, bullet 7): ignored paths in # validation must be explicitly justified. result = _good_validation( ignored_paths=[{"path": "branches", "justification": ""}] ) self.assertEqual(result["verdict"], "weak") self.assertTrue(any("branches" in r for r in result["reasons"])) def test_non_canonical_command_without_note_is_weak(self): result = _good_validation( command="venv/bin/pytest tests/test_prs.py", canonical_command="venv/bin/pytest tests/ --ignore=branches", ) self.assertEqual(result["verdict"], "weak") self.assertTrue(any("canonical" in r.lower() for r in result["reasons"])) def test_non_canonical_command_with_justification_is_strong(self): result = _good_validation( command="venv/bin/pytest tests/test_prs.py", canonical_command="venv/bin/pytest tests/ --ignore=branches", deviation_justification="targeted re-run of the only failing file", ) self.assertEqual(result["verdict"], "strong") class TestSelfReviewContamination(unittest.TestCase): """Required behavior 5: contamination claims need evidence.""" def test_identity_match_is_contaminated(self): result = assess_self_review_contamination( reviewer_identity="jcwalker3", pr_author="jcwalker3", ) self.assertEqual(result["status"], "contaminated") def test_session_authorship_claim_without_evidence_is_unknown(self): # Harness assertion (behavior 7, bullet 5): self-review contamination # requires evidence, not assumption. result = assess_self_review_contamination( reviewer_identity="sysadmin", pr_author="jcwalker3", session_authored=True, evidence_source=None, ) self.assertEqual(result["status"], "unknown") self.assertIn("choose another PR or stop", result["safe_next_action"]) def test_session_authorship_with_evidence_is_contaminated(self): result = assess_self_review_contamination( reviewer_identity="sysadmin", pr_author="jcwalker3", session_authored=True, evidence_source="this session created branch feat/x and opened the PR", ) self.assertEqual(result["status"], "contaminated") def test_clean_requires_evidence_too(self): result = assess_self_review_contamination( reviewer_identity="sysadmin", pr_author="jcwalker3", session_authored=False, evidence_source=None, ) self.assertEqual(result["status"], "unknown") def test_distinct_identities_with_evidence_is_clean(self): result = _good_contamination() self.assertEqual(result["status"], "clean") def test_missing_identities_are_unknown(self): result = assess_self_review_contamination( reviewer_identity=None, pr_author="jcwalker3", ) self.assertEqual(result["status"], "unknown") class TestFinalReport(unittest.TestCase): """Required behavior 6 + acceptance criteria: the report must distinguish each proof, and only a fully proven run earns an "A".""" def _report(self, **overrides): kwargs = { "checkout_proof": _good_checkout(), "inventory": _good_inventory(), "validation": _good_validation(), "contamination": _good_contamination(), "identity_eligible": True, "merge_performed": False, "issue_status_verified": True, } kwargs.update(overrides) return build_final_report(**kwargs) def test_fully_proven_run_is_grade_a(self): report = self._report() self.assertEqual(report["grade"], "A") self.assertEqual(report["violations"], []) # Behavior 6: each distinction is a distinct field. self.assertTrue(report["identity_eligible"]) self.assertTrue(report["pr_author_distinct_from_reviewer"]) self.assertEqual(report["session_contamination"], "clean") self.assertTrue(report["validated_on_pinned_head"]) self.assertFalse(report["merge_performed"]) self.assertTrue(report["issue_status_verified"]) def test_unproven_checkout_downgrades_and_blocks_merge(self): proof = verify_pinned_head_checkout( pinned_head_sha=PINNED, local_head_sha=OTHER, pr_base_ref="master", diff_base_ref="master", ) report = self._report(checkout_proof=proof) self.assertNotEqual(report["grade"], "A") self.assertFalse(report["merge_allowed"]) self.assertFalse(report["validated_on_pinned_head"]) def test_merge_claim_without_checkout_proof_is_a_violation(self): proof = verify_pinned_head_checkout( pinned_head_sha=PINNED, local_head_sha=OTHER, pr_base_ref="master", diff_base_ref="master", ) report = self._report(checkout_proof=proof, merge_performed=True) self.assertEqual(report["grade"], "blocked") self.assertTrue(report["violations"]) def test_unread_validation_output_cannot_be_reported_as_passed(self): report = self._report(validation=_good_validation(output_read=False)) self.assertFalse(report["validation_passed"]) self.assertNotEqual(report["grade"], "A") def test_weak_validation_downgrades(self): report = self._report( validation=_good_validation(passed=None, failed=None, skipped=None) ) self.assertNotEqual(report["grade"], "A") def test_incomplete_inventory_downgrades(self): inventory = assess_inventory_completeness( repo_reports=[], required_repos=["Scaled-Tech-Consulting/Gitea-Tools"], ) report = self._report(inventory=inventory) self.assertNotEqual(report["grade"], "A") def test_unknown_contamination_downgrades_and_blocks_merge(self): contamination = assess_self_review_contamination( reviewer_identity="sysadmin", pr_author="jcwalker3", session_authored=True, evidence_source=None, ) report = self._report(contamination=contamination) self.assertNotEqual(report["grade"], "A") self.assertFalse(report["merge_allowed"]) self.assertEqual(report["session_contamination"], "unknown") def test_merge_by_contaminated_reviewer_is_a_violation(self): contamination = assess_self_review_contamination( reviewer_identity="jcwalker3", pr_author="jcwalker3", ) report = self._report(contamination=contamination, merge_performed=True) self.assertEqual(report["grade"], "blocked") self.assertTrue(report["violations"]) def test_ineligible_identity_downgrades_and_blocks_merge(self): report = self._report(identity_eligible=False) self.assertNotEqual(report["grade"], "A") self.assertFalse(report["merge_allowed"]) class TestStdoutIsolation(unittest.TestCase): """Regression test for #178: tests must not close or corrupt stdout/stderr (prevents need for junitxml workaround in full suite runs and review validation). """ def test_stdout_remains_usable(self): """After typical test activity (mocks, redirects, prints from mains), stdout should be usable.""" # Verify not closed self.assertFalse(getattr(sys.stdout, "closed", False)) # Should be able to write (even if captured by pytest) try: sys.stdout.write("") sys.stdout.flush() except Exception as exc: self.fail(f"stdout write failed after test activity: {exc}") def test_stderr_remains_usable(self): self.assertFalse(getattr(sys.stderr, "closed", False)) try: sys.stderr.write("") sys.stderr.flush() except Exception as exc: self.fail(f"stderr write failed: {exc}") class TestRepoNameDisambiguation(unittest.TestCase): """Harness assertions for repo name disambiguation (new blind-review hardening). "MCP Gitea tool" etc. must resolve to Gitea-Tools, not silently default to mcp-control-plane. "open PRs" or ambiguous must check both. Single-repo zero-result must not hide PRs in the other configured repo. """ CONFIGURED = [ "Scaled-Tech-Consulting/Gitea-Tools", "Scaled-Tech-Consulting/mcp-control-plane", ] def test_gitea_tools_aliases_resolve_to_gitea_tools_only(self): for ref in [ "MCP Gitea tool", "gitea tool", "Gitea-Tools", "gitea mcp tool", "gitea-tools repo", ]: result = resolve_repos_from_user_reference(ref, self.CONFIGURED) self.assertEqual(result, ["Scaled-Tech-Consulting/Gitea-Tools"]) def test_mcp_control_plane_alias_resolves_only_to_it(self): result = resolve_repos_from_user_reference( "mcp-control-plane", self.CONFIGURED ) self.assertEqual(result, ["Scaled-Tech-Consulting/mcp-control-plane"]) def test_ambiguous_or_empty_defaults_to_both(self): self.assertEqual( resolve_repos_from_user_reference("open PRs", self.CONFIGURED), self.CONFIGURED, ) self.assertEqual( resolve_repos_from_user_reference("", self.CONFIGURED), self.CONFIGURED, ) self.assertEqual( resolve_repos_from_user_reference("review the queue", self.CONFIGURED), self.CONFIGURED, ) def test_single_repo_zero_result_does_not_hide_other(self): # Simulate a run that only inventoried mcp because of bad alias resolution. # The inventory must still require Gitea-Tools to claim "no open PRs". mcp_only_reports = [ { "repo": "Scaled-Tech-Consulting/mcp-control-plane", "state_filter": "open", "pagination_complete": True, "open_pr_count": 0, } ] result = assess_inventory_completeness( repo_reports=mcp_only_reports, required_repos=self.CONFIGURED, ) self.assertFalse(result["complete"]) self.assertTrue( any("Gitea-Tools" in r for r in result["reasons"]) ) def test_full_inventory_of_both_is_required_for_exhaustive_claim(self): both_reports = [ { "repo": "Scaled-Tech-Consulting/Gitea-Tools", "state_filter": "open", "pagination_complete": True, "open_pr_count": 1, }, { "repo": "Scaled-Tech-Consulting/mcp-control-plane", "state_filter": "open", "pagination_complete": True, "open_pr_count": 0, }, ] result = assess_inventory_completeness( repo_reports=both_reports, required_repos=self.CONFIGURED ) self.assertTrue(result["complete"]) self.assertTrue(result["can_claim_exhaustive"]) class TestControllerHandoff(unittest.TestCase): """Issue #182: final reports must end with a Controller Handoff.""" BASE_HANDOFF = "\n".join([ "## Controller Handoff", "", "- Task: implement issue #182", "- Repo: Scaled-Tech-Consulting/Gitea-Tools", "- Role: author", "- Identity: jcwalker3 / prgs-author", "- Issue/PR: #182 / PR #999", "- Branch/SHA: feat/x @ 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9", "- Files changed: review_proofs.py", "- Validation: 700 passed, 6 skipped", "- Mutations: one PR opened", "- Current status: PR open", "- Blockers: none", "- Next: review PR #999", "- Safety: no self-review; no self-merge; no secrets", ]) def test_report_without_handoff_is_downgraded(self): result = assess_controller_handoff("long report text, no handoff") self.assertEqual(result["verdict"], "missing") self.assertTrue(result["downgraded"]) def test_wrong_title_is_downgraded(self): text = self.BASE_HANDOFF.replace( "## Controller Handoff", "## Handoff Summary") result = assess_controller_handoff(text) self.assertEqual(result["verdict"], "missing") def test_complete_base_handoff_passes(self): result = assess_controller_handoff( "full report body...\n\n" + self.BASE_HANDOFF) self.assertEqual(result["verdict"], "complete") self.assertFalse(result["downgraded"]) def test_missing_base_fields_are_listed(self): text = "\n".join( line for line in self.BASE_HANDOFF.splitlines() if not line.startswith(("- Mutations:", "- Safety:"))) result = assess_controller_handoff(text) self.assertEqual(result["verdict"], "incomplete") self.assertTrue(result["downgraded"]) self.assertIn("Mutations", result["missing_fields"]) self.assertIn("Safety", result["missing_fields"]) def test_review_role_requires_review_fields(self): result = assess_controller_handoff(self.BASE_HANDOFF, role="review") self.assertEqual(result["verdict"], "incomplete") self.assertIn("Pinned reviewed head", result["missing_fields"]) self.assertIn("Merge result", result["missing_fields"]) complete = self.BASE_HANDOFF + "\n" + "\n".join([ "- Selected PR: #999", "- Reviewer eligibility: passed", "- Pinned reviewed head: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9", "- Review decision: approve", "- Merge result: merged", "- Linked issue status: closed", "- Cleanup status: branch deleted", ]) result = assess_controller_handoff(complete, role="review") self.assertEqual(result["verdict"], "complete") def test_author_role_requires_author_fields(self): complete = self.BASE_HANDOFF + "\n" + "\n".join([ "- Selected issue: #182", "- Claim/comment status: comment-claimed", "- PR number opened: #999", "- No review/merge: confirmed", ]) result = assess_controller_handoff(complete, role="author") self.assertEqual(result["verdict"], "complete") result = assess_controller_handoff(self.BASE_HANDOFF, role="author") self.assertEqual(result["verdict"], "incomplete") self.assertIn("No review/merge confirmation", result["missing_fields"]) def test_author_role_rejects_equivalent_or_multiple_issues(self): # 1. equivalent reference blocked incomplete_eq = self.BASE_HANDOFF + "\n" + "\n".join([ "- Selected issue: Issue #194 / #196 equivalent", "- Claim/comment status: comment-claimed", "- PR number opened: #999", "- No review/merge: confirmed", ]) res = assess_controller_handoff(incomplete_eq, role="author") self.assertEqual(res["verdict"], "incomplete") self.assertIn("Selected issue", res["missing_fields"]) # 2. multiple issues blocked incomplete_multi = self.BASE_HANDOFF + "\n" + "\n".join([ "- Selected issue: #194, #196", "- Claim/comment status: comment-claimed", "- PR number opened: #999", "- No review/merge: confirmed", ]) res = assess_controller_handoff(incomplete_multi, role="author") self.assertEqual(res["verdict"], "incomplete") self.assertIn("Selected issue", res["missing_fields"]) def test_author_role_rejects_fuzzy_pr_number(self): incomplete_pr = self.BASE_HANDOFF + "\n" + "\n".join([ "- Selected issue: #196", "- Claim/comment status: comment-claimed", "- PR number opened: PR #203 / #204 equivalent", "- No review/merge: confirmed", ]) res = assess_controller_handoff(incomplete_pr, role="author") self.assertEqual(res["verdict"], "incomplete") self.assertIn("PR number opened", res["missing_fields"]) def test_inventory_role_requires_inventory_fields(self): complete = self.BASE_HANDOFF + "\n" + "\n".join([ "- Repositories checked: Gitea-Tools, mcp-control-plane", "- Open PR counts: 2 / 0", "- Selected PR or reason: none eligible (self-authored)", "- Inventory completeness: complete, no pagination needed", ]) result = assess_controller_handoff(complete, role="inventory") self.assertEqual(result["verdict"], "complete") def test_skill_doc_declares_handoff_requirement(self): # Doc-contract: SKILL.md must keep requiring the exact section and # naming this validator, or the convention silently rots. skill = ( __import__("pathlib").Path(__file__).resolve().parent.parent / "skills" / "llm-project-workflow" / "SKILL.md" ).read_text(encoding="utf-8") self.assertIn("## Controller Handoff", skill) self.assertIn("assess_controller_handoff", skill) self.assertIn("issue #182", skill) if __name__ == "__main__": unittest.main()