Scaled-Tech-Consulting/Gitea-Tools
2
0
Fork 0
Code Issues 73 Pull Requests 9 Actions Packages Projects Releases Wiki Activity
Files
1158594a200bc7c118b9e3265375cb769f65b83a
Gitea-Tools/tests/test_review_proofs.py
T
sysadmin 1158594a20 feat(continuation): formalize continuation proofs for claim and push (closes #189) 
Extend #188 continuation mode with issue claim status evidence,
force-with-lease push reporting, canonical secret sweep helper, and
composite final-report checks wired into assess_issue_selection_final_report.
2026-07-06 11:54:40 -04:00

1927 lines
75 KiB
Python
Raw Blame History

"""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
ISSUE_SELECTION_CONTINUATION_EXPLICIT,
ISSUE_SELECTION_REPRESENTED_BY_OPEN_PR,
assess_author_pr_report,
assess_capability_evidence,
assess_capability_proof,
assess_contradictory_no_pr_claim,
assess_continuation_mode_report,
assess_force_with_lease_push_report,
canonical_secret_sweep_report,
CANONICAL_SECRET_SWEEP_COMMAND,
assess_controller_handoff,
assess_edited_pr_inventory_coverage,
assess_empty_queue_report,
assess_fresh_issue_selection,
assess_inventory_completeness,
assess_issue_selection_final_report,
assess_reviewer_queue_inventory,
assess_live_state_recheck,
assess_review_mutation_final_report,
assess_role_boundary,
assess_secret_sweep,
assess_self_review_contamination,
assess_sweep_evidence,
assess_validation_report,
build_final_report,
classify_issue_for_selection,
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_review_mutation():
report = (
"Review complete on PR #203.\n"
"Review mutations: one request_changes on #203.\n"
"Review decision: request_changes."
)
lock = {
"live_mutations": [{"pr_number": 203, "action": "request_changes"}],
"correction_authorized": False,
}
return assess_review_mutation_final_report(report, lock)
def _good_worktree(**overrides):
proof = {
"worktree_path": "/repo/branches/review-feat-issue-224",
"porcelain_status": "",
"pr_scope_files": ["docs/wiki/Repositories.md"],
"scratch_used": True,
"scratch_path": "/repo/branches/review-feat-issue-224",
"git_commands": [
"git fetch prgs master feat/issue-224-wiki-proof-refresh",
"git diff prgs/master...prgs/feat/issue-224-wiki-proof-refresh",
],
}
proof.update(overrides)
from reviewer_worktree import assess_reviewer_worktree_proof # noqa: E402
return assess_reviewer_worktree_proof(proof)
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",
"- Workspace mutations: none",
"- 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(),
"review_mutation": _good_review_mutation(),
"controller_handoff": _good_handoff(),
"capability_proof": _good_capability_proof(),
"sweep_proof": _good_secret_sweep(),
"worktree_proof": {
"worktree_path": "/repo/branches/review-feat-issue-224",
"porcelain_status": "",
"pr_scope_files": ["docs/wiki/Repositories.md"],
"scratch_used": True,
"scratch_path": "/repo/branches/review-feat-issue-224",
},
}
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,
"review_mutation": _good_review_mutation(),
}
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",
"- Workspace mutations: none",
"- 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("Worktree path", 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",
"- Worktree path: /repo/branches/review-pr-999",
"- Worktree dirty: no",
"- Scratch worktree used: yes (/repo/branches/review-pr-999)",
"- Unrelated local mutations: none",
"- Review decision: approve",
"- Merge result: merged",
"- Linked issue status: closed",
"- Cleanup status: branch deleted",
])
result = assess_controller_handoff(complete, role="review")
self.assertEqual(result["verdict"], "complete")
def _author_role_fields(self, issue_number=182, pr_number=999):
return [
f"- Selected issue: #{issue_number}",
"- Issue lock proof: lock before diff on feat/x @ master",
"- Claim/comment status: comment-claimed",
f"- PR number opened: #{pr_number}",
"- No review/merge: confirmed",
]
def test_handoff_role_fields_author_includes_issue_lock_proof(self):
from review_proofs import HANDOFF_ROLE_FIELDS
names = [name for name, _ in HANDOFF_ROLE_FIELDS["author"]]
self.assertIn("Issue lock proof", names)
def test_author_role_requires_author_fields(self):
complete = self.BASE_HANDOFF + "\n" + "\n".join(self._author_role_fields())
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("Issue lock proof", result["missing_fields"])
self.assertIn("No review/merge confirmation", result["missing_fields"])
def test_author_role_requires_issue_lock_proof(self):
without_lock = 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(without_lock, role="author")
self.assertEqual(result["verdict"], "incomplete")
self.assertIn("Issue lock proof", 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",
"- Issue lock proof: lock before diff on feat/x @ master",
"- 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",
"- Issue lock proof: lock before diff on feat/x @ master",
"- 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",
"- Issue lock proof: lock before diff on feat/issue-196 @ master",
"- 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",
"- PR inventory trust gate: trusted_nonempty / trusted_empty",
"- Trust gate reasons: none",
"- Trust gate corroborated: true",
"- Inventory profile: prgs-reviewer",
"- 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)
def test_handoff_rejects_none_workspace_mutations_when_local_edits_exist(self):
# 1. Workspace mutations: none is rejected when local_edits is True
incomplete_eq = self.BASE_HANDOFF + "\n" + "\n".join(
self._author_role_fields(issue_number=196, pr_number=203))
res = assess_controller_handoff(incomplete_eq, role="author", local_edits=True)
self.assertEqual(res["verdict"], "incomplete")
self.assertIn("Workspace mutations", res["missing_fields"])
# 2. Workspace mutations: edited files is allowed when local_edits is True
complete_eq = self.BASE_HANDOFF.replace(
"- Workspace mutations: none",
"- Workspace mutations: edited review_proofs.py",
) + "\n" + "\n".join(self._author_role_fields(issue_number=196, pr_number=203))
res2 = assess_controller_handoff(complete_eq, role="author", local_edits=True)
self.assertEqual(res2["verdict"], "complete")
class TestReviewMutationFinalReport(unittest.TestCase):
"""Final reports must list exactly one live review mutation."""
LOCK_ONE = {
"live_mutations": [{"pr_number": 203, "action": "request_changes"}],
"correction_authorized": False,
}
def test_single_mutation_report_complete(self):
report = (
"Review complete on PR #203.\n"
"Review mutations: one request_changes on #203.\n"
"Review decision: request_changes."
)
result = assess_review_mutation_final_report(report, self.LOCK_ONE)
self.assertTrue(result["complete"])
self.assertFalse(result["downgraded"])
def test_missing_mutation_details_downgraded(self):
result = assess_review_mutation_final_report(
"Review complete. No details.", self.LOCK_ONE
)
self.assertFalse(result["complete"])
self.assertTrue(result["downgraded"])
def test_two_mutations_require_correction_explanation(self):
lock = {
"live_mutations": [
{"pr_number": 203, "action": "approve"},
{"pr_number": 203, "action": "request_changes"},
],
"correction_authorized": True,
"correction_reason": "operator approved correcting mistaken approve",
}
incomplete = assess_review_mutation_final_report(
"Submitted approve then request_changes on #203.", lock
)
self.assertFalse(incomplete["complete"])
complete = assess_review_mutation_final_report(
"\n".join([
"Review mutations: approve (mistake), then request_changes (final).",
"Correction flow: operator approved correcting mistaken approve on PR #203.",
]),
lock,
)
self.assertTrue(complete["complete"])
def test_build_final_report_wires_review_mutation_from_report_text(self):
report_text = (
"Review complete on PR #203.\n"
"Review mutations: one request_changes on #203."
)
lock = {
"live_mutations": [{"pr_number": 203, "action": "request_changes"}],
"correction_authorized": False,
}
final = build_final_report(
checkout_proof=_good_checkout(),
inventory=_good_inventory(),
validation=_good_validation(),
contamination=_good_contamination(),
identity_eligible=True,
merge_performed=False,
issue_status_verified=True,
capability_evidence=_good_capability_evidence(),
sweep=_good_sweep(),
live_state=_good_live_state(),
role_boundary=_good_role_boundary(),
report_text=report_text,
review_decision_lock=lock,
)
self.assertTrue(final["review_mutation_complete"])
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 TestAssessReviewerQueueInventory(unittest.TestCase):
"""Issue #196: trust gate wired into canonical queue inventory."""
def _repo_report(self, **overrides):
report = {
"repo": "Scaled-Tech-Consulting/Gitea-Tools",
"state_filter": "open",
"pagination_complete": True,
"open_pr_count": 0,
"list_prs_response": [],
"remote": "prgs",
"authenticated_profile": {
"profile_name": "prgs-reviewer",
"allowed_operations": ["read", "gitea.read"],
},
"local_remote_url": (
"https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git"
),
}
report.update(overrides)
return report
def test_empty_without_corroboration_blocks_empty_queue_claim(self):
result = assess_reviewer_queue_inventory([
self._repo_report(pagination_complete=False),
self._repo_report(
repo="Scaled-Tech-Consulting/mcp-control-plane",
open_pr_count=0,
pagination_complete=False,
),
])
self.assertFalse(result["can_claim_empty_queue"])
self.assertIn("untrusted_empty", str(result["trust_gates"]))
def test_trusted_empty_with_finality_allows_empty_queue_claim(self):
result = assess_reviewer_queue_inventory([
self._repo_report(),
self._repo_report(
repo="Scaled-Tech-Consulting/mcp-control-plane",
local_remote_url=(
"https://gitea.prgs.cc/Scaled-Tech-Consulting/"
"mcp-control-plane.git"
),
),
])
self.assertTrue(result["can_claim_empty_queue"])
self.assertTrue(result["can_claim_exhaustive"])
def test_user_context_indicating_open_prs_blocks_empty_claim(self):
result = assess_reviewer_queue_inventory(
[self._repo_report()],
user_context="please review open PR #195 in the queue",
)
self.assertFalse(result["can_claim_empty_queue"])
self.assertTrue(result["blockers"])
def test_nonempty_inventory_skips_empty_trust_gate_block(self):
result = assess_reviewer_queue_inventory([
self._repo_report(open_pr_count=2),
self._repo_report(
repo="Scaled-Tech-Consulting/mcp-control-plane",
open_pr_count=1,
),
])
self.assertTrue(result["complete"])
self.assertEqual(result["trust_gates"], {})
class TestAssessEmptyQueueReport(unittest.TestCase):
"""Issue #198: empty-queue reports require formal trust-gate proof."""
def _trusted_report(self, **extra):
lines = [
"Queue inventory complete.",
"Repository: Scaled-Tech-Consulting/Gitea-Tools",
"Open PR count: 0",
"pr_inventory_trust_gate.status: trusted_empty",
"pr_inventory_trust_gate.corroborated: true",
"Inventory profile: prgs-reviewer",
"Workflow correctly stops with nothing to review.",
]
lines.extend(extra)
return "\n".join(lines)
def test_non_empty_report_not_claimed(self):
result = assess_empty_queue_report("Reviewed PR #236 and merged.")
self.assertFalse(result["claimed"])
self.assertTrue(result["proven"])
def test_empty_claim_without_trust_gate_blocked(self):
report = (
"Open PR count: 0\n"
"Pagination complete: yes\n"
"Queue cleared."
)
result = assess_empty_queue_report(report)
self.assertTrue(result["claimed"])
self.assertFalse(result["proven"])
self.assertTrue(result["block"])
def test_trusted_empty_report_with_required_fields_passes(self):
result = assess_empty_queue_report(self._trusted_report())
self.assertTrue(result["proven"])
def test_weak_merge_commit_corroboration_blocked(self):
report = (
"Open PR count: 0\n"
"Master latest commit is merge of PR #79 so queue is empty."
)
result = assess_empty_queue_report(report)
self.assertFalse(result["proven"])
self.assertTrue(
any("weak corroboration" in r for r in result["reasons"])
)
def test_author_session_reviewer_queue_wording_blocked(self):
result = assess_empty_queue_report(
self._trusted_report(),
task_role="author",
)
self.assertFalse(result["proven"])
def test_build_final_report_downgrades_weak_empty_queue(self):
final = build_final_report(
checkout_proof=_good_checkout(),
inventory=_good_inventory(),
validation=_good_validation(),
contamination=_good_contamination(),
identity_eligible=True,
merge_performed=False,
issue_status_verified=True,
capability_evidence=_good_capability_evidence(),
sweep=_good_sweep(),
live_state=_good_live_state(),
role_boundary=_good_role_boundary(),
review_mutation=_good_review_mutation(),
controller_handoff=_good_handoff(),
capability_proof=_good_capability_proof(),
sweep_proof=_good_secret_sweep(),
worktree_proof={
"worktree_path": "/repo/branches/review-pr-1",
"porcelain_status": "",
"scratch_used": True,
"scratch_path": "/repo/branches/review-pr-1",
},
report_text="Open PR count: 0. Queue cleared.",
)
self.assertNotEqual(final["grade"], "A")
self.assertFalse(final["empty_queue_trust_gate_proven"])
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(),
"review_mutation": _good_review_mutation(),
"controller_handoff": _good_handoff(),
"capability_proof": _good_capability_proof(),
"sweep_proof": _good_secret_sweep(),
"worktree_proof": {
"worktree_path": "/repo/branches/review-feat-issue-224",
"porcelain_status": "",
"pr_scope_files": ["docs/wiki/Repositories.md"],
"scratch_used": True,
"scratch_path": "/repo/branches/review-feat-issue-224",
},
}
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"])
def test_missing_review_mutation_downgrades(self):
report = self._report(review_mutation=None)
self.assertNotEqual(report["grade"], "A")
self.assertFalse(report["review_mutation_complete"])
def test_incomplete_review_mutation_downgrades(self):
report = self._report(review_mutation={"complete": False, "downgraded": True, "reasons": []})
self.assertNotEqual(report["grade"], "A")
self.assertFalse(report["review_mutation_complete"])
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"])
class TestIssueSelectionContinuation(unittest.TestCase):
"""Issue #188: continuation mode wall for issues with open PRs."""
OLD_SHA = PINNED
NEW_SHA = OTHER
OPEN_PR = [{"number": 187, "head": {"ref": "feat/issue-183-harden-author-run-reporting"}}]
def test_open_pr_issue_excluded_from_fresh_selection(self):
classified = classify_issue_for_selection(
183, open_prs=self.OPEN_PR,
)
self.assertEqual(classified["status"], ISSUE_SELECTION_REPRESENTED_BY_OPEN_PR)
self.assertFalse(classified["selectable_for_fresh_work"])
blocked = assess_fresh_issue_selection([classified])
self.assertTrue(blocked["downgraded"])
def test_explicit_continuation_allows_represented_issue(self):
classified = classify_issue_for_selection(
183,
open_prs=self.OPEN_PR,
operator_continuation_requested=True,
)
self.assertEqual(classified["status"], ISSUE_SELECTION_CONTINUATION_EXPLICIT)
blocked = assess_fresh_issue_selection([classified])
self.assertFalse(blocked["downgraded"])
def test_contradictory_no_pr_claim_downgrades(self):
report = (
"Selected issue #183; no duplicate PR open. "
"Updated PR #187 on branch feat/issue-183-harden-author-run-reporting."
)
result = assess_contradictory_no_pr_claim(
report, edited_pr_numbers=[187], issue_open_pr_map={183: 187},
)
self.assertTrue(result["downgraded"])
def test_edited_pr_must_appear_in_inventory(self):
report = "Open PR inventory: PR #195 only."
result = assess_edited_pr_inventory_coverage(
report,
edited_pr_numbers=[187],
inventoried_pr_numbers=[195],
)
self.assertTrue(result["downgraded"])
def test_continuation_report_requires_old_and_new_head(self):
report = (
"Issue #182 continuation mode. PR #186. "
f"old head {self.OLD_SHA} -> new head {self.NEW_SHA}. "
"PR author: jcwalker3. Branch: feat/issue-182-controller-handoff. "
"Session authored PR: yes. Continuation allowed: operator requested."
)
result = assess_continuation_mode_report(
report,
pr_number=186,
pr_author="jcwalker3",
branch="feat/issue-182-controller-handoff",
old_head_sha=self.OLD_SHA,
new_head_sha=self.NEW_SHA,
session_authored_pr=True,
continuation_allowed_reason="operator requested continuation",
)
self.assertTrue(result["complete"])
def test_issue_selection_final_report_continuation_earns_a(self):
report = "\n".join([
"Issue #182 continuation mode — no new issue claimed.",
f"PR #186 updated: old head {self.OLD_SHA}, "
f"new head {self.NEW_SHA}.",
"PR author: jcwalker3. Branch: feat/issue-182-controller-handoff.",
"Session authored PR: yes.",
"Continuation allowed: operator requested rebase.",
"Open PR inventory included PR #186.",
"## Controller Handoff",
"- Task: continuation",
"- Repo: Scaled-Tech-Consulting/Gitea-Tools",
"- Role: author",
"- Identity: prgs-author",
"- Issue/PR: #182 / PR #186",
"- Branch/SHA: feat/issue-182-controller-handoff",
"- Files changed: review_proofs.py",
"- Validation: tests passed",
"- Mutations: push_branch",
"- Workspace mutations: none",
"- Current status: PR mergeable",
"- Blockers: none",
"- Next: review",
"- Safety: no review/merge",
"- Continuation mode: issue #182 continuation",
"- Existing PR: #186",
"- PR author: jcwalker3",
"- Issue claim status: status:in-progress",
"- Branch: feat/issue-182-controller-handoff",
f"- Old PR head: {self.OLD_SHA}",
f"- New PR head: {self.NEW_SHA}",
"- Session authored PR: yes",
"- Why continuation allowed: operator requested rebase",
])
result = assess_issue_selection_final_report(
report,
mode="continuation",
continuation_proof={
"pr_number": 186,
"pr_author": "jcwalker3",
"issue_number": 182,
"issue_claim_status": "status:in-progress",
"branch": "feat/issue-182-controller-handoff",
"old_head_sha": self.OLD_SHA,
"new_head_sha": self.NEW_SHA,
"session_authored_pr": True,
"continuation_allowed_reason": "operator requested rebase",
},
edited_pr_numbers=[186],
inventoried_pr_numbers=[186],
)
self.assertEqual(result["grade"], "A")
class TestContinuationModeProofs(unittest.TestCase):
"""Issue #189: formal continuation proofs beyond #188 selection wall."""
OLD_SHA = PINNED
NEW_SHA = OTHER
REMOTE_OLD = "601c608c00000000000000000000000000000000"
LEASE_SHA = "45c5cac2bc49dd112766ea218b18a71e9c5f8e99"
PUSHED_SHA = "45c5cac2bc49dd112766ea218b18a71e9c5f8e99"
def test_force_with_lease_requires_lease_evidence(self):
report = (
"Issue #182 continuation. force-with-lease push to "
"feat/issue-182-controller-handoff-enforcement."
)
result = assess_force_with_lease_push_report(
report,
old_remote_head=self.REMOTE_OLD,
expected_lease_head=self.LEASE_SHA,
new_pushed_head=self.PUSHED_SHA,
branch_pushed="feat/issue-182-controller-handoff-enforcement",
used_force_with_lease=True,
)
self.assertTrue(result["downgraded"])
def test_force_with_lease_complete_with_full_evidence(self):
report = (
"Issue #182 continuation with git push --force-with-lease. "
f"Old remote head {self.REMOTE_OLD}. "
f"Lease head {self.LEASE_SHA}. "
f"Pushed head {self.PUSHED_SHA}. "
"Branch feat/issue-182-controller-handoff-enforcement. "
"Push branch only: only the feature branch was pushed."
)
result = assess_force_with_lease_push_report(
report,
old_remote_head=self.REMOTE_OLD,
expected_lease_head=self.LEASE_SHA,
new_pushed_head=self.PUSHED_SHA,
branch_pushed="feat/issue-182-controller-handoff-enforcement",
used_force_with_lease=True,
)
self.assertTrue(result["complete"])
def test_continuation_requires_issue_claim_status(self):
report = (
"Issue #182 continuation mode. PR #186. "
f"old head {self.OLD_SHA} new head {self.NEW_SHA}. "
"PR author jcwalker3. Branch feat/issue-182-test."
)
result = assess_continuation_mode_report(
report,
pr_number=186,
pr_author="jcwalker3",
issue_number=182,
issue_claim_status="status:in-progress",
branch="feat/issue-182-test",
old_head_sha=self.OLD_SHA,
new_head_sha=self.NEW_SHA,
)
self.assertTrue(result["downgraded"])
def test_canonical_secret_sweep_helper(self):
sweep = canonical_secret_sweep_report(clean=True)
self.assertEqual(sweep["method"], CANONICAL_SECRET_SWEEP_COMMAND)
self.assertTrue(sweep["clean"])
def test_continuation_final_report_with_push_and_sweep_earns_a(self):
branch = "feat/issue-182-controller-handoff-enforcement"
report = "\n".join([
"Issue #182 continuation mode — status:in-progress already set.",
f"PR #186 updated via git push --force-with-lease.",
f"Old PR head {self.OLD_SHA}; new PR head {self.NEW_SHA}.",
f"Old remote head {self.REMOTE_OLD}.",
f"Lease head {self.LEASE_SHA}. Pushed head {self.PUSHED_SHA}.",
f"Branch {branch}. Push branch only: only the feature branch.",
"PR author: jcwalker3. Session authored PR: yes.",
"Continuation allowed: operator requested rebase.",
"Open PR inventory included PR #186.",
"## Controller Handoff",
"- Task: continuation",
"- Repo: Scaled-Tech-Consulting/Gitea-Tools",
"- Role: author",
"- Identity: prgs-author",
"- Issue/PR: #182 / PR #186",
f"- Branch/SHA: {branch}",
"- Files changed: review_proofs.py",
"- Validation: tests passed",
"- Mutations: push_branch",
"- Workspace mutations: none",
"- Current status: PR mergeable",
"- Blockers: none",
"- Next: review",
"- Safety: no review/merge",
"- Continuation mode: issue #182 continuation",
"- Existing PR: #186",
"- PR author: jcwalker3",
"- Issue claim status: status:in-progress",
f"- Branch: {branch}",
f"- Old PR head: {self.OLD_SHA}",
f"- New PR head: {self.NEW_SHA}",
"- Session authored PR: yes",
"- Why continuation allowed: operator requested rebase",
])
result = assess_issue_selection_final_report(
report,
mode="continuation",
continuation_proof={
"pr_number": 186,
"pr_author": "jcwalker3",
"issue_number": 182,
"issue_claim_status": "status:in-progress",
"branch": branch,
"old_head_sha": self.OLD_SHA,
"new_head_sha": self.NEW_SHA,
"session_authored_pr": True,
"continuation_allowed_reason": "operator requested rebase",
"push_proof": {
"old_remote_head": self.REMOTE_OLD,
"expected_lease_head": self.LEASE_SHA,
"new_pushed_head": self.PUSHED_SHA,
"branch_pushed": branch,
"used_force_with_lease": True,
},
"secret_sweep": canonical_secret_sweep_report(clean=True),
},
edited_pr_numbers=[186],
inventoried_pr_numbers=[186],
)
self.assertEqual(result["grade"], "A")
if __name__ == "__main__":
unittest.main()
Reference in New Issue View Git Blame Copy Permalink