"""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_author_pr_report, assess_capability_evidence, assess_capability_proof, assess_controller_handoff, assess_inventory_completeness, assess_live_state_recheck, assess_role_boundary, assess_secret_sweep, assess_self_review_contamination, assess_sweep_evidence, assess_validation_report, build_final_report, pr_inventory_trust_gate, 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", ) def _good_role_boundary(): return assess_role_boundary( { "task_role": "reviewer", "task_kind": "blind_pr_queue_review", "reviewer_namespace_used": True, "author_namespace_used": False, "author_mutations": [], "review_mutations": [], "operator_authorized_author_work": False, "scratch_evidence_claimed": False, } ) def _good_capability_evidence(): return assess_capability_evidence([ { "task": "review_pr", "allowed": True, "evidence_source": ( "gitea_resolve_task_capability(review_pr) output: " "allowed_in_current_session=true, profile prgs-reviewer" ), }, { "task": "merge_pr", "allowed": True, "evidence_source": ( "gitea_resolve_task_capability(merge_pr) output: " "allowed_in_current_session=true, profile prgs-reviewer" ), }, ]) def _good_sweep(**overrides): sweep = { "command": ( "git diff prgs/master...HEAD | grep -inE " "'password|token|secret|api[_-]?key|authorization|bearer|https?://'" ), "scope": "full PR diff against prgs/master", "clean": True, } sweep.update(overrides) return assess_sweep_evidence(sweep) def _good_live_state(**overrides): recheck = { "pr_state": "open", "pinned_head_sha": PINNED, "live_head_sha": PINNED, "pinned_base_ref": "master", "live_base_ref": "master", "blocking_change_requests": False, } recheck.update(overrides) return assess_live_state_recheck(recheck) def _good_role_boundary_179(**overrides): kwargs = { "task_role": "reviewer", "namespaces_used": ["gitea-reviewer"], "justification": None, } kwargs.update(overrides) return assess_role_boundary(**kwargs) GOOD_HANDOFF = "\n".join([ "## Controller Handoff", "", "- Task: review PR #999", "- Repo: Scaled-Tech-Consulting/Gitea-Tools", "- Role: reviewer", "- Identity: sysadmin / prgs-reviewer", "- Issue/PR: #182 / PR #999", "- Branch/SHA: feat/x @ 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9", "- Files changed: review_proofs.py", "- Validation: 700 passed, 6 skipped", "- Mutations: review only", "- Current status: PR open", "- Blockers: none", "- Next: merge if approved", "- Safety: no self-review; no self-merge; no secrets", ]) def _good_handoff(): return assess_controller_handoff(GOOD_HANDOFF) def _good_capability_proof(): return assess_capability_proof({ "comment_issue": { "requested_task": "comment_issue", "required_operation_permission": "gitea.issue.comment", "allowed_in_current_session": True, } }) def _good_secret_sweep(**overrides): sweep = { "method": "git diff prgs/master...HEAD | grep -iE 'token|secret'", "scope": "full PR diff relative to prgs/master", "clean": True, } sweep.update(overrides) return assess_secret_sweep(sweep) 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 TestRoleBoundary(unittest.TestCase): """Issue #175: reviewer queue tasks must not pivot into author work.""" def test_reviewer_queue_without_author_mutations_is_clean(self): result = _good_role_boundary() self.assertEqual(result["status"], "clean") self.assertEqual(result["violations"], []) def test_reviewer_queue_author_mutation_without_authorization_violates(self): result = assess_role_boundary( { "task_role": "reviewer", "task_kind": "blind_pr_queue_review", "reviewer_namespace_used": True, "author_namespace_used": True, "author_mutations": ["claim issue #171", "push branch"], "operator_authorized_author_work": False, "mixed_namespace_justification": ( "author namespace was used for implementation" ), } ) self.assertEqual(result["status"], "violation") self.assertTrue(any("pivot" in r for r in result["violations"])) def test_mixed_namespace_use_without_justification_is_warning(self): result = assess_role_boundary( { "task_role": "reviewer", "task_kind": "blind_pr_queue_review", "reviewer_namespace_used": True, "author_namespace_used": True, "author_mutations": [], } ) self.assertEqual(result["status"], "warning") self.assertTrue( any("mixed" in r.lower() for r in result["reasons"]) ) def test_author_task_cannot_perform_review_mutations(self): result = assess_role_boundary( { "task_role": "author", "reviewer_namespace_used": False, "author_namespace_used": True, "review_mutations": ["approve PR"], } ) self.assertEqual(result["status"], "violation") self.assertTrue( any("reviewer-only" in r for r in result["violations"]) ) def test_scratch_only_notes_are_not_durable_evidence(self): result = assess_role_boundary( { "task_role": "reviewer", "task_kind": "blind_pr_queue_review", "reviewer_namespace_used": True, "scratch_evidence_claimed": True, "scratch_evidence_durable": False, } ) self.assertEqual(result["status"], "warning") self.assertTrue(any("scratch-only" in r for r in result["reasons"])) 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, "capability_evidence": _good_capability_evidence(), "sweep": _good_sweep(), "live_state": _good_live_state(), "role_boundary": _good_role_boundary(), "controller_handoff": _good_handoff(), "capability_proof": _good_capability_proof(), "sweep_proof": _good_secret_sweep(), } 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.assertEqual(report["role_boundary"], "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"]) def test_missing_role_boundary_downgrades_and_blocks_merge(self): kwargs = { "checkout_proof": _good_checkout(), "inventory": _good_inventory(), "validation": _good_validation(), "contamination": _good_contamination(), "identity_eligible": True, "merge_performed": False, "issue_status_verified": True, } report = build_final_report(**kwargs) self.assertNotEqual(report["grade"], "A") self.assertFalse(report["merge_allowed"]) self.assertEqual(report["role_boundary"], "warning") def test_role_boundary_violation_blocks_report(self): boundary = assess_role_boundary( { "task_role": "reviewer", "task_kind": "blind_pr_queue_review", "reviewer_namespace_used": True, "author_namespace_used": True, "author_mutations": ["create PR"], "operator_authorized_author_work": False, "mixed_namespace_justification": "implementation pivot", } ) report = self._report(role_boundary=boundary) self.assertEqual(report["grade"], "blocked") self.assertFalse(report["merge_allowed"]) def test_failed_capability_proof_downgrades(self): report = self._report( capability_proof={ "proven": False, "reasons": ["task mark_issue not allowed"], } ) self.assertNotEqual(report["grade"], "A") self.assertTrue( any("capability" in r for r in report["downgrade_reasons"]) ) def test_failed_sweep_proof_downgrades(self): report = self._report( sweep_proof={ "proven": False, "reasons": ["method missing"], } ) self.assertNotEqual(report["grade"], "A") self.assertTrue(any("sweep" in r for r in report["downgrade_reasons"])) 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_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) class TestPRInventoryTrustGate(unittest.TestCase): """Issue #194: unit tests for the PR inventory trust gate.""" def setUp(self): self.profile = { "profile_name": "prgs-reviewer", "allowed_operations": ["read", "gitea.read", "gitea.pr.approve"], } self.local_url = "https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git" def test_trusted_nonempty(self): res = pr_inventory_trust_gate([{"number": 1}]) self.assertEqual(res["status"], "trusted_nonempty") self.assertFalse(res["corroborated"]) def test_inventory_error_none_or_not_list(self): self.assertEqual(pr_inventory_trust_gate(None)["status"], "inventory_error") self.assertEqual(pr_inventory_trust_gate("not a list")["status"], "inventory_error") def test_untrusted_empty_no_pagination_or_corroboration(self): res = pr_inventory_trust_gate( [], remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", state="open", authenticated_profile=self.profile, local_remote_url=self.local_url, user_context=None, corroboration_open_pr_counter=None, has_finality_metadata=False ) self.assertEqual(res["status"], "untrusted_empty") self.assertIn("pagination finality not proven and open_pr_counter corroboration is missing or non-zero", res["reasons"]) def test_untrusted_empty_profile_permission_mismatch(self): bad_profile = {"profile_name": "prgs-bad", "allowed_operations": ["write"]} res = pr_inventory_trust_gate( [], remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", state="open", authenticated_profile=bad_profile, local_remote_url=self.local_url, user_context=None, corroboration_open_pr_counter=0, has_finality_metadata=False ) self.assertEqual(res["status"], "untrusted_empty") self.assertIn("authenticated profile lacks read permissions", res["reasons"]) def test_untrusted_empty_remote_url_mismatch(self): bad_url = "https://gitea.prgs.cc/other-org/other-repo.git" res = pr_inventory_trust_gate( [], remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", state="open", authenticated_profile=self.profile, local_remote_url=bad_url, user_context=None, corroboration_open_pr_counter=0, has_finality_metadata=False ) self.assertEqual(res["status"], "untrusted_empty") self.assertIn("local remote URL does not match target repository 'Scaled-Tech-Consulting/Gitea-Tools'", res["reasons"]) def test_untrusted_empty_user_context_indicates_prs(self): res = pr_inventory_trust_gate( [], remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", state="open", authenticated_profile=self.profile, local_remote_url=self.local_url, user_context="please check open PR #181", corroboration_open_pr_counter=0, has_finality_metadata=False ) self.assertEqual(res["status"], "untrusted_empty") self.assertTrue(any("user context indicates open PRs should exist" in r for r in res["reasons"])) def test_trusted_empty_with_corroboration(self): res = pr_inventory_trust_gate( [], remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", state="open", authenticated_profile=self.profile, local_remote_url=self.local_url, user_context=None, corroboration_open_pr_counter=0, has_finality_metadata=False ) self.assertEqual(res["status"], "trusted_empty") self.assertTrue(res["corroborated"]) def test_trusted_empty_with_finality_metadata(self): res = pr_inventory_trust_gate( [], remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", state="open", authenticated_profile=self.profile, local_remote_url=self.local_url, user_context=None, corroboration_open_pr_counter=None, has_finality_metadata=True ) self.assertEqual(res["status"], "trusted_empty") self.assertTrue(res["corroborated"]) class TestCapabilityEvidence(unittest.TestCase): """#179 gap 1: capability claims need exact evidence.""" def test_evidence_backed_claims_are_proven(self): result = _good_capability_evidence() self.assertTrue(result["proven"]) self.assertEqual(result["reasons"], []) def test_claim_without_evidence_source_is_not_proven(self): result = assess_capability_evidence([ {"task": "review_pr", "allowed": True, "evidence_source": ""}, ]) self.assertFalse(result["proven"]) self.assertTrue(any("evidence" in r.lower() for r in result["reasons"])) def test_no_claims_at_all_fails_closed(self): result = assess_capability_evidence([]) self.assertFalse(result["proven"]) def test_disallowed_task_is_not_proven(self): result = assess_capability_evidence([ { "task": "merge_pr", "allowed": False, "evidence_source": "gitea_resolve_task_capability output", }, ]) self.assertFalse(result["proven"]) class TestSweepEvidence(unittest.TestCase): """#179 gap 2: secret/provenance sweep must be exact.""" def test_exact_sweep_is_proven(self): result = _good_sweep() self.assertEqual(result["verdict"], "exact") self.assertTrue(result["proven"]) def test_vague_sweep_without_command_is_downgraded(self): result = _good_sweep(command="") self.assertEqual(result["verdict"], "vague") self.assertFalse(result["proven"]) def test_sweep_without_scope_is_downgraded(self): result = _good_sweep(scope="") self.assertEqual(result["verdict"], "vague") self.assertFalse(result["proven"]) def test_missing_sweep_fails_closed(self): result = assess_sweep_evidence(None) self.assertEqual(result["verdict"], "missing") self.assertFalse(result["proven"]) def test_unstated_result_is_downgraded(self): result = _good_sweep(clean=None) self.assertFalse(result["proven"]) class TestLiveStateRecheck(unittest.TestCase): """#179 gap 3: explicit pre-mutation live-state recheck.""" def test_clean_recheck_is_proven(self): result = _good_live_state() self.assertTrue(result["proven"]) self.assertFalse(result["block"]) def test_missing_recheck_fails_closed(self): result = assess_live_state_recheck(None) self.assertFalse(result["proven"]) self.assertTrue(result["block"]) def test_closed_pr_blocks(self): result = _good_live_state(pr_state="closed") self.assertFalse(result["proven"]) self.assertTrue(result["block"]) def test_moved_head_blocks(self): result = _good_live_state(live_head_sha=OTHER) self.assertFalse(result["proven"]) self.assertTrue(any("head" in r.lower() for r in result["reasons"])) def test_changed_base_blocks(self): result = _good_live_state(live_base_ref="develop") self.assertFalse(result["proven"]) def test_unresolved_blocking_reviews_block(self): result = _good_live_state(blocking_change_requests=True) self.assertFalse(result["proven"]) def test_unchecked_blocking_state_fails_closed(self): result = _good_live_state(blocking_change_requests=None) self.assertFalse(result["proven"]) class TestRoleBoundary179(unittest.TestCase): """#179 gap 4: reviewer flows avoid unjustified author-namespace use.""" def test_native_namespace_only_is_clean(self): result = _good_role_boundary_179() self.assertTrue(result["proven"]) def test_foreign_namespace_without_justification_is_downgraded(self): result = _good_role_boundary_179( namespaces_used=["gitea-reviewer", "gitea-author"] ) self.assertFalse(result["proven"]) self.assertTrue(any("justif" in r.lower() for r in result["reasons"])) def test_foreign_namespace_with_justification_is_clean(self): result = _good_role_boundary_179( namespaces_used=["gitea-reviewer", "gitea-author"], justification=( "author namespace read-only whoami used to evidence " "self-review contamination status" ), ) self.assertTrue(result["proven"]) def test_unreported_namespaces_fail_closed(self): result = _good_role_boundary_179(namespaces_used=None) self.assertFalse(result["proven"]) class TestFinalReport179Bar(unittest.TestCase): """#179 acceptance adds capability, sweep, live-state, and role proofs.""" 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, "capability_evidence": _good_capability_evidence(), "sweep": _good_sweep(), "live_state": _good_live_state(), "role_boundary": _good_role_boundary(), "controller_handoff": _good_handoff(), "capability_proof": _good_capability_proof(), "sweep_proof": _good_secret_sweep(), } kwargs.update(overrides) return build_final_report(**kwargs) def test_all_179_proofs_present_is_grade_a(self): report = self._report() self.assertEqual(report["grade"], "A") self.assertTrue(report["capability_evidence_proven"]) self.assertEqual(report["sweep_verdict"], "exact") self.assertTrue(report["live_state_recheck_proven"]) self.assertTrue(report["role_boundary_clean"]) def test_missing_capability_evidence_downgrades(self): report = self._report(capability_evidence=None) self.assertNotEqual(report["grade"], "A") self.assertFalse(report["capability_evidence_proven"]) def test_unevidenced_capability_claim_downgrades(self): report = self._report( capability_evidence=assess_capability_evidence([ {"task": "review_pr", "allowed": True, "evidence_source": ""}, ]) ) self.assertNotEqual(report["grade"], "A") def test_vague_sweep_downgrades(self): report = self._report(sweep=_good_sweep(command="")) self.assertNotEqual(report["grade"], "A") self.assertEqual(report["sweep_verdict"], "vague") def test_missing_sweep_downgrades(self): report = self._report(sweep=None) self.assertNotEqual(report["grade"], "A") def test_missing_live_state_recheck_downgrades_and_blocks_merge(self): report = self._report(live_state=None) self.assertNotEqual(report["grade"], "A") self.assertFalse(report["merge_allowed"]) self.assertFalse(report["live_state_recheck_proven"]) def test_stale_live_state_blocks_merge(self): report = self._report(live_state=_good_live_state(live_head_sha=OTHER)) self.assertNotEqual(report["grade"], "A") self.assertFalse(report["merge_allowed"]) def test_merge_claim_without_live_recheck_is_a_violation(self): report = self._report(live_state=None, merge_performed=True) self.assertEqual(report["grade"], "blocked") self.assertTrue(report["violations"]) def test_unjustified_author_namespace_downgrades(self): report = self._report( role_boundary=_good_role_boundary_179( namespaces_used=["gitea-reviewer", "gitea-author"] ) ) self.assertNotEqual(report["grade"], "A") self.assertFalse(report["role_boundary_clean"]) def test_positive_baseline_from_173_still_holds(self): report = self._report() self.assertTrue(report["inventory_complete"]) self.assertTrue(report["validated_on_pinned_head"]) class TestAuthorReporting(unittest.TestCase): """Harness assertions for author reporting and capability/sweep proofs.""" def test_good_capability_proof_passes(self): result = _good_capability_proof() self.assertTrue(result["proven"]) def test_missing_capability_proof_fails_closed(self): result = assess_capability_proof({}) self.assertFalse(result["proven"]) self.assertTrue(any("fail closed" in r for r in result["reasons"])) def test_unresolved_capability_proof_fails_closed(self): result = assess_capability_proof({ "mark_issue": { "requested_task": "unknown_task", "required_operation_permission": None, "allowed_in_current_session": None, } }) self.assertFalse(result["proven"]) self.assertTrue( any("could not be resolved" in r for r in result["reasons"]) ) def test_good_secret_sweep_passes(self): result = _good_secret_sweep() self.assertTrue(result["proven"]) self.assertEqual(result["verdict"], "strong") def test_vague_secret_sweep_without_method_is_weak(self): result = _good_secret_sweep(method="") self.assertFalse(result["proven"]) self.assertEqual(result["verdict"], "weak") self.assertTrue( any("method" in r or "scan" in r for r in result["reasons"]) ) def test_unconfirmed_secret_sweep_is_weak(self): result = _good_secret_sweep(clean=False) self.assertFalse(result["proven"]) self.assertEqual(result["verdict"], "weak") def test_good_author_pr_report_passes(self): result = assess_author_pr_report({ "pr_number": 185, "branch": "feat/issue-184-repo-name-disambiguation", "head_sha": "e2bccbafeeb93124ba068bfb06058d5aa7467cae", }) self.assertTrue(result["complete"]) def test_author_pr_report_without_head_sha_is_incomplete(self): result = assess_author_pr_report({ "pr_number": 185, "branch": "feat/issue-184-repo-name-disambiguation", "head_sha": "", }) self.assertFalse(result["complete"]) self.assertTrue(any("head SHA" in r for r in result["reasons"])) def test_author_pr_report_rejects_short_sha(self): result = assess_author_pr_report({ "pr_number": 185, "branch": "feat/issue-184-repo-name-disambiguation", "head_sha": "e2bccba", }) self.assertFalse(result["complete"]) if __name__ == "__main__": unittest.main()