sysadmin
cbaed49761
- Add 'mark_issue' to TASK_MAP in resolve_task_capability for exact mutation proof (closes gap on unknown treated as allowed). - Add assess_controller_handoff and integrate into build_final_report (downgrades missing handoff). - Add tests for handoff and update existing. - Update SKILL.md and review-pr.md to mandate exact 'Controller Handoff' section, exact sweep evidence, PR head SHA in reports. - Author profile used throughout; no review/merge. Refs #183
1024 lines
39 KiB
Python
1024 lines
39 KiB
Python
"""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_proof,
|
|
assess_controller_handoff,
|
|
assess_inventory_completeness,
|
|
assess_role_boundary,
|
|
assess_secret_sweep,
|
|
assess_self_review_contamination,
|
|
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,
|
|
}
|
|
)
|
|
|
|
|
|
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,
|
|
"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 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()
|