Scaled-Tech-Consulting/Gitea-Tools
2
0
Fork 0
Code Issues 73 Pull Requests 9 Actions Packages Projects Releases Wiki Activity

Merge pull request 'Add fail-closed proofs for blind PR queue review workflow (Issue #173)' (#176) from feat/issue-173-reviewer-workflow-proofs into master

Browse Source
This commit was merged in pull request #176.
This commit is contained in:
sysadmin
2026-07-05 14:08:58 -05:00
parent ae7a390495 ccc5ef79df
commit 64dc334a92
4 changed files with 880 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
review_proofs.py
+363
View File
@@ -0,0 +1,363 @@
"""Fail-closed proof helpers for the blind PR queue review workflow (#173).
A blind queue review must *prove* its preconditions instead of asserting
them. Each helper here evaluates one proof from issue #173's required
behaviors and returns a plain dict verdict; ``build_final_report`` combines
them and only awards an "A" grade when every proof is present. Missing
evidence always downgrades or blocks — never upgrades.
These helpers are pure (no network, no git calls): the workflow gathers the
raw facts (Gitea PR head SHA, ``git rev-parse HEAD``, listing pagination
state, pytest output, whoami results) and passes them in, so the same logic
is usable from prompts, harness assertions, and tests. The MCP-level gates
(mcp-control-plane #68/#76) remain authoritative for mutations; nothing
here weakens or replaces them.
"""
import re
_FULL_SHA = re.compile(r"^[0-9a-f]{40}$")
SAFE_NEXT_ACTION_UNKNOWN_CONTAMINATION = (
"evidence missing: report contamination as unknown and "
"choose another PR or stop"
)
def _normalize_ref(ref):
"""Normalize a git ref for base comparison.
Strips ``refs/heads/`` and ``refs/remotes/<remote>/`` prefixes and a
plain ``<remote>/`` prefix so ``master``, ``prgs/master`` and
``refs/remotes/prgs/master`` all compare equal. Returns '' for empty
input (which then fails closed in the caller).
"""
ref = (ref or "").strip()
if ref.startswith("refs/heads/"):
ref = ref[len("refs/heads/"):]
elif ref.startswith("refs/remotes/"):
ref = ref[len("refs/remotes/"):]
ref = ref.split("/", 1)[1] if "/" in ref else ref
elif "/" in ref:
# Remote-tracking shorthand like 'prgs/master'.
ref = ref.split("/", 1)[1]
return ref
def verify_pinned_head_checkout(pinned_head_sha, local_head_sha,
pr_base_ref, diff_base_ref):
"""Required behaviors 1–2: prove HEAD == pinned PR head, correct base.
Both SHAs must be full 40-hex and identical — an abbreviated or prefix
match is not proof. The diff base must normalize to the PR base branch.
Returns {'proven', 'block', 'reasons', 'pinned_head_sha',
'local_head_sha'}; ``block`` is True whenever the proof fails, meaning
review and merge must stop before proceeding.
"""
reasons = []
pinned = (pinned_head_sha or "").strip().lower()
local = (local_head_sha or "").strip().lower()
if not pinned:
reasons.append("pinned PR head SHA missing; fail closed")
elif not _FULL_SHA.match(pinned):
reasons.append("pinned PR head SHA is not a full 40-hex SHA; fail closed")
if not local:
reasons.append("local checkout HEAD SHA missing; fail closed")
elif not _FULL_SHA.match(local):
reasons.append(
"local checkout HEAD SHA is not a full 40-hex SHA "
"(abbreviated SHAs are not proof); fail closed"
)
if pinned and local and _FULL_SHA.match(pinned) and _FULL_SHA.match(local):
if pinned != local:
reasons.append(
"local checkout HEAD does not match the pinned PR head SHA; "
"stop before review/merge"
)
base = _normalize_ref(pr_base_ref)
diff_base = _normalize_ref(diff_base_ref)
if not base:
reasons.append("PR base ref missing; fail closed")
if not diff_base:
reasons.append("diff base ref missing; fail closed")
if base and diff_base and base != diff_base:
reasons.append(
f"diff base '{diff_base}' is not the PR base branch '{base}'"
)
proven = not reasons
return {
"proven": proven,
"block": not proven,
"reasons": reasons,
"pinned_head_sha": pinned or None,
"local_head_sha": local or None,
}
def assess_inventory_completeness(repo_reports, required_repos):
"""Required behavior 3: the queue inventory must prove completeness.
*repo_reports* is a list of dicts, one per repository listed:
``{'repo', 'state_filter', 'pagination_complete', 'open_pr_count'}``.
The inventory is complete only when every required repository has a
report that states its filter, proves pagination was handled, and gives
a total open-PR count. ``can_claim_exhaustive`` — permission to say
"these are the only open PRs" — is granted only for a complete
inventory.
"""
reasons = []
by_repo = {}
for report in repo_reports or []:
name = (report.get("repo") or "").strip()
if name:
by_repo[name] = report
totals = {}
for repo in required_repos:
report = by_repo.get(repo)
if report is None:
reasons.append(f"repository '{repo}' was not inventoried")
continue
if not (report.get("state_filter") or "").strip():
reasons.append(f"repository '{repo}': PR state filter not stated")
if report.get("pagination_complete") is not True:
reasons.append(
f"repository '{repo}': pagination not proven complete; "
"a truncated listing cannot support an exhaustive claim"
)
count = report.get("open_pr_count")
if not isinstance(count, int) or count < 0:
reasons.append(
f"repository '{repo}': total open PR count not reported"
)
else:
totals[repo] = count
complete = not reasons and bool(required_repos)
if not required_repos:
reasons.append("no required repositories configured; fail closed")
return {
"complete": complete,
"can_claim_exhaustive": complete,
"reasons": reasons,
"total_open_prs": totals,
}
def assess_validation_report(report):
"""Required behavior 4: validation reporting needs exact evidence.
*report* keys: ``command``, ``output_read``, ``result`` ('pass'/'fail'),
``passed``/``failed``/``skipped`` counts, ``ignored_paths`` (each with a
``justification``), ``canonical_command``, ``deviation_justification``.
Verdicts:
- 'invalid' — the result may not be claimed at all (no command stated,
or the command output was never read).
- 'weak' — claimable but downgraded (missing counts, unjustified
ignored paths, unexplained deviation from the canonical command).
- 'strong' — full evidence.
"""
reasons = []
command = (report.get("command") or "").strip()
if not command:
return {
"verdict": "invalid",
"claimable": False,
"reasons": ["no validation command stated; result cannot be claimed"],
}
if report.get("output_read") is not True:
return {
"verdict": "invalid",
"claimable": False,
"reasons": [
"command output was not read to completion; a validation "
"result cannot be claimed"
],
}
counts = [report.get("passed"), report.get("failed"), report.get("skipped")]
if any(not isinstance(c, int) for c in counts):
reasons.append(
"test counts (passed/failed/skipped) missing; report is incomplete"
)
for ignored in report.get("ignored_paths") or []:
path = ignored.get("path") or "<unnamed>"
if not (ignored.get("justification") or "").strip():
reasons.append(
f"ignored path '{path}' has no justification for why it is "
"safe to ignore"
)
canonical = (report.get("canonical_command") or "").strip()
if canonical and command != canonical:
if not (report.get("deviation_justification") or "").strip():
reasons.append(
"command differs from the repository's canonical validation "
"command and the deviation is not justified"
)
verdict = "strong" if not reasons else "weak"
return {"verdict": verdict, "claimable": True, "reasons": reasons}
def assess_self_review_contamination(reviewer_identity, pr_author,
session_authored=None,
evidence_source=None):
"""Required behavior 5: contamination is evidence-backed, never assumed.
Distinguishes the two cases the issue calls out:
- reviewer identity == PR author → contaminated (the identities
themselves are the evidence);
- same-session authorship → contaminated only with an explicit
``evidence_source``; a bare claim (either way) yields 'unknown'.
'unknown' is a stop signal: choose another PR or stop — it is never
silently treated as clean or as contaminated.
"""
reviewer = (reviewer_identity or "").strip()
author = (pr_author or "").strip()
evidence = (evidence_source or "").strip()
if not reviewer or not author:
return {
"status": "unknown",
"reasons": ["reviewer identity or PR author unknown; fail closed"],
"safe_next_action": SAFE_NEXT_ACTION_UNKNOWN_CONTAMINATION,
}
if reviewer == author:
return {
"status": "contaminated",
"reasons": [
f"authenticated reviewer '{reviewer}' is the PR author; "
"self-review is blocked"
],
"safe_next_action": "choose another PR or stop",
}
if session_authored is True:
if evidence:
return {
"status": "contaminated",
"reasons": [
"this session authored/touched the PR branch "
f"(evidence: {evidence})"
],
"safe_next_action": "choose another PR or stop",
}
return {
"status": "unknown",
"reasons": [
"same-session authorship was claimed without an evidence "
"source; contamination may not be declared by assumption"
],
"safe_next_action": SAFE_NEXT_ACTION_UNKNOWN_CONTAMINATION,
}
if session_authored is False and evidence:
return {
"status": "clean",
"reasons": [
f"reviewer '{reviewer}' differs from author '{author}' and "
f"session non-authorship is evidenced ({evidence})"
],
"safe_next_action": "proceed",
}
return {
"status": "unknown",
"reasons": [
"session authorship was not established with evidence; "
"contamination status is unknown"
],
"safe_next_action": SAFE_NEXT_ACTION_UNKNOWN_CONTAMINATION,
}
def build_final_report(checkout_proof, inventory, validation, contamination,
identity_eligible, merge_performed,
issue_status_verified):
"""Required behavior 6 + acceptance criteria: one report, distinct proofs.
Combines the individual proof verdicts into the final-report fields the
issue requires, computes ``merge_allowed`` from the proofs, and grades
the run:
- 'A' — every proof present (merge itself is optional).
- 'downgraded' — one or more proofs missing/weak; do not merge.
- 'blocked' — a *violation*: a merge was claimed although the proofs
did not allow one.
"""
contamination_status = contamination.get("status", "unknown")
checkout_proven = bool(checkout_proof.get("proven"))
validation_claimable = bool(validation.get("claimable"))
validation_strong = validation.get("verdict") == "strong"
downgrade_reasons = []
if not identity_eligible:
downgrade_reasons.append("identity/profile not eligible for review")
if not inventory.get("complete"):
downgrade_reasons.append("open-PR inventory completeness not proven")
downgrade_reasons.extend(inventory.get("reasons", []))
if not checkout_proven:
downgrade_reasons.append("pinned-head checkout not proven")
downgrade_reasons.extend(checkout_proof.get("reasons", []))
if not validation_strong:
downgrade_reasons.append(
f"validation evidence is {validation.get('verdict', 'missing')}"
)
downgrade_reasons.extend(validation.get("reasons", []))
if contamination_status != "clean":
downgrade_reasons.append(
f"session contamination status is '{contamination_status}'"
)
if not issue_status_verified:
downgrade_reasons.append("linked issue status not verified")
merge_allowed = (
identity_eligible
and checkout_proven
and contamination_status == "clean"
and validation_claimable
and validation.get("verdict") != "invalid"
)
violations = []
if merge_performed and not merge_allowed:
violations.append(
"merge was performed/claimed although the proofs did not allow "
"one; this run is blocked, not graded"
)
if violations:
grade = "blocked"
elif downgrade_reasons:
grade = "downgraded"
else:
grade = "A"
return {
"grade": grade,
"violations": violations,
"downgrade_reasons": downgrade_reasons,
"identity_eligible": bool(identity_eligible),
"pr_author_distinct_from_reviewer":
contamination_status in ("clean",),
"session_contamination": contamination_status,
"inventory_complete": bool(inventory.get("complete")),
"validated_on_pinned_head": checkout_proven and validation_claimable,
"validation_passed":
validation_claimable and validation.get("verdict") != "invalid",
"validation_verdict": validation.get("verdict"),
"merge_allowed": merge_allowed,
"merge_performed": bool(merge_performed),
"issue_status_verified": bool(issue_status_verified),
}
skills/llm-project-workflow/SKILL.md
+31 -4
View File
@@ -167,11 +167,38 @@ Worktree folder = branch with `/` replaced by `-`
1. Use a separate review worktree (`scripts/worktree-review <branch>`), detached.
2. Verify your authenticated identity.
3. Verify the PR author — **you must not be the author.**
3. Verify the PR author — **you must not be the author.** Self-review
contamination must be *evidence-backed* (#173): state the authenticated
reviewer identity, the PR author identity, whether this session
authored/touched the PR branch, and the evidence source for any
"same-session author" claim. If the evidence is missing, report the
status as **unknown** — never declare contamination by assumption — and
choose another PR or stop (`review_proofs.assess_self_review_contamination`).
4. Verify the worktree is clean.
5. Inspect the full diff; confirm scope matches the linked issue; flag unrelated files.
6. Run the tests.
7. **Do not merge if checks fail. Do not merge if the reviewer is the author.**
5. **Checkout proof (#173):** before reviewing or validating, prove and
state: the selected PR head SHA from Gitea (pinned), the local checkout
SHA (`git rev-parse HEAD`), that `HEAD ==` the pinned PR head SHA, and
that the diff base is the PR base branch. If `HEAD` does not match the
pinned head, **stop before review/merge**
(`review_proofs.verify_pinned_head_checkout`).
6. **Inventory proof (#173):** a blind queue review must prove listing
completeness before claiming "only PRs found": both configured
repositories checked, open-PR filters stated, pagination handled or
explicitly not needed, and the total open PR count per repo reported
(`review_proofs.assess_inventory_completeness`).
7. Inspect the full diff; confirm scope matches the linked issue; flag unrelated files.
8. Run the tests. Validation reporting must include the exact command and
exact results: pass/fail, counts of tests passed/skipped/failed, any
ignored paths and why they are safe to ignore, and whether the command
differs from the repository's canonical validation command. Only claim a
validation result after the command has completed and its output has
been read (`review_proofs.assess_validation_report`).
9. **Do not merge if checks fail. Do not merge if the reviewer is the author.**
10. The final report must distinguish (`review_proofs.build_final_report`):
identity eligible; PR author different from reviewer; session
contamination absent (with evidence); validation performed on the pinned
head; merge performed; issue status verified. If any proof is missing,
stop or downgrade the result instead of merging confidently.
## G. Merge / cleanup workflow
skills/llm-project-workflow/templates/review-pr.md
+19 -4
View File
@@ -20,13 +20,28 @@ Steps:
*If the current identity does not match the required role (or is the PR author), STOP. Relaunch/switch to the correct profile first.*
2. Verify your authenticated identity (whoami) and the active profile.
3. Fetch the PR facts: PR author, head SHA, state (must be open), base branch.
Pin the head SHA in your notes; every later step validates THAT SHA.
4. If authenticated user == PR author → STOP (no self-review).
4. scripts/worktree-review <pr-head-branch> # detached, branches/review-*
Session-contamination claims must be evidence-backed (#173): if you
cannot evidence whether this session authored/touched the PR branch,
report contamination as UNKNOWN (not contaminated, not clean) and choose
another PR or stop.
5. scripts/worktree-review <pr-head-branch> # detached, branches/review-*
cd branches/review-<pr-head-branch-slug>
5. Confirm the worktree is clean. Inspect the FULL diff; confirm scope matches
6. Checkout proof (#173) — prove and state, before any diff review or
validation:
- pinned PR head SHA (from Gitea)
- local checkout SHA (git rev-parse HEAD)
- HEAD == pinned PR head SHA
- diff base == the PR base branch
If HEAD does not match the pinned head → STOP before review/merge.
7. Confirm the worktree is clean. Inspect the FULL diff; confirm scope matches
issue #<n>; flag any unrelated files, secrets, or formatting churn. Check that the PR body correctly uses Gitea-closing keywords (`Closes #N` or `Fixes #N`) instead of non-closing ones (`Implements #N`, `Refs #N`).
6. Run the test suite; note results.
7. Post the review verdict: approve only if scope is clean and checks pass;
8. Run the test suite; report the exact command and exact results — pass/fail
plus passed/skipped/failed counts, any ignored paths and why they are safe
to ignore, and whether the command differs from the repository's canonical
validation command. Only claim a result after the output has been read.
9. Post the review verdict: approve only if scope is clean and checks pass;
otherwise request changes with specifics. Never merge from this review step.
Include a "Review Metadata" block (attribution only — docs/llm-agent-sha.md):
tests/test_review_proofs.py
+467
View File
@@ -0,0 +1,467 @@
"""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 sys
import unittest
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
from review_proofs import ( # noqa: E402
assess_inventory_completeness,
assess_self_review_contamination,
assess_validation_report,
build_final_report,
verify_pinned_head_checkout,
)
PINNED = "0fdc8f582026b72a229d59a172c0a63ac4aaeaf9"
OTHER = "a4060c5de00f2b1c9e88f4f6f0f3f9a7b2c1d0e9"
def _good_checkout():
return verify_pinned_head_checkout(
pinned_head_sha=PINNED,
local_head_sha=PINNED,
pr_base_ref="master",
diff_base_ref="prgs/master",
)
def _good_inventory():
return assess_inventory_completeness(
repo_reports=[
{
"repo": "Scaled-Tech-Consulting/Gitea-Tools",
"state_filter": "open",
"pagination_complete": True,
"open_pr_count": 2,
},
{
"repo": "Scaled-Tech-Consulting/mcp-control-plane",
"state_filter": "open",
"pagination_complete": True,
"open_pr_count": 1,
},
],
required_repos=[
"Scaled-Tech-Consulting/Gitea-Tools",
"Scaled-Tech-Consulting/mcp-control-plane",
],
)
def _good_validation(**overrides):
report = {
"command": "venv/bin/pytest tests/ --ignore=branches",
"output_read": True,
"result": "pass",
"passed": 421,
"failed": 0,
"skipped": 13,
"ignored_paths": [
{
"path": "branches",
"justification": (
"review worktrees under branches/ duplicate test module "
"names and break collection; they are not part of the "
"repository's own suite"
),
}
],
"canonical_command": "venv/bin/pytest tests/ --ignore=branches",
}
report.update(overrides)
return assess_validation_report(report)
def _good_contamination():
return assess_self_review_contamination(
reviewer_identity="sysadmin",
pr_author="jcwalker3",
session_authored=False,
evidence_source="session transcript: branch never checked out or pushed here",
)
class TestCheckoutProof(unittest.TestCase):
"""Required behavior 1 + 2: prove HEAD == pinned PR head or stop."""
def test_matching_pinned_and_local_head_is_proven(self):
proof = _good_checkout()
self.assertTrue(proof["proven"])
self.assertFalse(proof["block"])
def test_fetched_sha_but_unchecked_out_head_blocks(self):
# Harness assertion (behavior 7, bullet 1): fetching/pinning a SHA
# without checking it out must block review and merge.
proof = verify_pinned_head_checkout(
pinned_head_sha=PINNED,
local_head_sha=OTHER,
pr_base_ref="master",
diff_base_ref="prgs/master",
)
self.assertFalse(proof["proven"])
self.assertTrue(proof["block"])
self.assertTrue(any("HEAD" in r for r in proof["reasons"]))
def test_missing_pinned_sha_fails_closed(self):
proof = verify_pinned_head_checkout(
pinned_head_sha="",
local_head_sha=PINNED,
pr_base_ref="master",
diff_base_ref="master",
)
self.assertFalse(proof["proven"])
self.assertTrue(proof["block"])
def test_missing_local_head_fails_closed(self):
proof = verify_pinned_head_checkout(
pinned_head_sha=PINNED,
local_head_sha=None,
pr_base_ref="master",
diff_base_ref="master",
)
self.assertFalse(proof["proven"])
self.assertTrue(proof["block"])
def test_abbreviated_sha_is_not_proof(self):
# A prefix match is not proof of the exact pinned head.
proof = verify_pinned_head_checkout(
pinned_head_sha=PINNED,
local_head_sha=PINNED[:12],
pr_base_ref="master",
diff_base_ref="master",
)
self.assertFalse(proof["proven"])
self.assertTrue(proof["block"])
def test_wrong_diff_base_blocks(self):
proof = verify_pinned_head_checkout(
pinned_head_sha=PINNED,
local_head_sha=PINNED,
pr_base_ref="master",
diff_base_ref="develop",
)
self.assertFalse(proof["proven"])
self.assertTrue(proof["block"])
self.assertTrue(any("base" in r.lower() for r in proof["reasons"]))
def test_remote_tracking_diff_base_matches_pr_base(self):
proof = verify_pinned_head_checkout(
pinned_head_sha=PINNED,
local_head_sha=PINNED,
pr_base_ref="master",
diff_base_ref="refs/remotes/prgs/master",
)
self.assertTrue(proof["proven"])
class TestInventoryCompleteness(unittest.TestCase):
"""Required behavior 3: inventory must prove completeness."""
def test_complete_multi_repo_inventory(self):
# Harness assertion (behavior 7, bullet 2): complete queue inventory
# with multiple PRs across repos.
result = _good_inventory()
self.assertTrue(result["complete"])
self.assertTrue(result["can_claim_exhaustive"])
self.assertEqual(
result["total_open_prs"],
{
"Scaled-Tech-Consulting/Gitea-Tools": 2,
"Scaled-Tech-Consulting/mcp-control-plane": 1,
},
)
def test_missing_repo_makes_inventory_incomplete(self):
result = assess_inventory_completeness(
repo_reports=[
{
"repo": "Scaled-Tech-Consulting/Gitea-Tools",
"state_filter": "open",
"pagination_complete": True,
"open_pr_count": 1,
}
],
required_repos=[
"Scaled-Tech-Consulting/Gitea-Tools",
"Scaled-Tech-Consulting/mcp-control-plane",
],
)
self.assertFalse(result["complete"])
self.assertFalse(result["can_claim_exhaustive"])
self.assertTrue(
any("mcp-control-plane" in r for r in result["reasons"])
)
def test_unfinished_pagination_blocks_exhaustive_claim(self):
# Harness assertion (behavior 7, bullet 3): pagination / multi-page
# PR listing must be handled or the claim is refused.
result = assess_inventory_completeness(
repo_reports=[
{
"repo": "Scaled-Tech-Consulting/Gitea-Tools",
"state_filter": "open",
"pagination_complete": False,
"open_pr_count": 50,
}
],
required_repos=["Scaled-Tech-Consulting/Gitea-Tools"],
)
self.assertFalse(result["complete"])
self.assertFalse(result["can_claim_exhaustive"])
self.assertTrue(any("pagination" in r.lower() for r in result["reasons"]))
def test_missing_state_filter_blocks_exhaustive_claim(self):
result = assess_inventory_completeness(
repo_reports=[
{
"repo": "Scaled-Tech-Consulting/Gitea-Tools",
"pagination_complete": True,
"open_pr_count": 1,
}
],
required_repos=["Scaled-Tech-Consulting/Gitea-Tools"],
)
self.assertFalse(result["complete"])
def test_missing_count_blocks_exhaustive_claim(self):
result = assess_inventory_completeness(
repo_reports=[
{
"repo": "Scaled-Tech-Consulting/Gitea-Tools",
"state_filter": "open",
"pagination_complete": True,
}
],
required_repos=["Scaled-Tech-Consulting/Gitea-Tools"],
)
self.assertFalse(result["complete"])
def test_empty_inventory_fails_closed(self):
result = assess_inventory_completeness(
repo_reports=[],
required_repos=["Scaled-Tech-Consulting/Gitea-Tools"],
)
self.assertFalse(result["complete"])
self.assertFalse(result["can_claim_exhaustive"])
class TestValidationReporting(unittest.TestCase):
"""Required behavior 4: exact commands and exact results."""
def test_full_report_is_strong(self):
result = _good_validation()
self.assertEqual(result["verdict"], "strong")
self.assertTrue(result["claimable"])
def test_missing_test_counts_is_weak(self):
# Harness assertion (behavior 7, bullet 4): validation result missing
# test counts is flagged as weak/incomplete.
result = _good_validation(passed=None, failed=None, skipped=None)
self.assertEqual(result["verdict"], "weak")
self.assertTrue(any("count" in r.lower() for r in result["reasons"]))
def test_output_not_read_cannot_claim_result(self):
# Harness assertion (behavior 7, bullet 6): a report cannot say
# validation passed unless the command output was actually read.
result = _good_validation(output_read=False)
self.assertEqual(result["verdict"], "invalid")
self.assertFalse(result["claimable"])
def test_missing_command_cannot_claim_result(self):
result = _good_validation(command="")
self.assertEqual(result["verdict"], "invalid")
self.assertFalse(result["claimable"])
def test_unjustified_ignored_path_is_weak(self):
# Harness assertion (behavior 7, bullet 7): ignored paths in
# validation must be explicitly justified.
result = _good_validation(
ignored_paths=[{"path": "branches", "justification": ""}]
)
self.assertEqual(result["verdict"], "weak")
self.assertTrue(any("branches" in r for r in result["reasons"]))
def test_non_canonical_command_without_note_is_weak(self):
result = _good_validation(
command="venv/bin/pytest tests/test_prs.py",
canonical_command="venv/bin/pytest tests/ --ignore=branches",
)
self.assertEqual(result["verdict"], "weak")
self.assertTrue(any("canonical" in r.lower() for r in result["reasons"]))
def test_non_canonical_command_with_justification_is_strong(self):
result = _good_validation(
command="venv/bin/pytest tests/test_prs.py",
canonical_command="venv/bin/pytest tests/ --ignore=branches",
deviation_justification="targeted re-run of the only failing file",
)
self.assertEqual(result["verdict"], "strong")
class TestSelfReviewContamination(unittest.TestCase):
"""Required behavior 5: contamination claims need evidence."""
def test_identity_match_is_contaminated(self):
result = assess_self_review_contamination(
reviewer_identity="jcwalker3",
pr_author="jcwalker3",
)
self.assertEqual(result["status"], "contaminated")
def test_session_authorship_claim_without_evidence_is_unknown(self):
# Harness assertion (behavior 7, bullet 5): self-review contamination
# requires evidence, not assumption.
result = assess_self_review_contamination(
reviewer_identity="sysadmin",
pr_author="jcwalker3",
session_authored=True,
evidence_source=None,
)
self.assertEqual(result["status"], "unknown")
self.assertIn("choose another PR or stop", result["safe_next_action"])
def test_session_authorship_with_evidence_is_contaminated(self):
result = assess_self_review_contamination(
reviewer_identity="sysadmin",
pr_author="jcwalker3",
session_authored=True,
evidence_source="this session created branch feat/x and opened the PR",
)
self.assertEqual(result["status"], "contaminated")
def test_clean_requires_evidence_too(self):
result = assess_self_review_contamination(
reviewer_identity="sysadmin",
pr_author="jcwalker3",
session_authored=False,
evidence_source=None,
)
self.assertEqual(result["status"], "unknown")
def test_distinct_identities_with_evidence_is_clean(self):
result = _good_contamination()
self.assertEqual(result["status"], "clean")
def test_missing_identities_are_unknown(self):
result = assess_self_review_contamination(
reviewer_identity=None,
pr_author="jcwalker3",
)
self.assertEqual(result["status"], "unknown")
class TestFinalReport(unittest.TestCase):
"""Required behavior 6 + acceptance criteria: the report must
distinguish each proof, and only a fully proven run earns an "A"."""
def _report(self, **overrides):
kwargs = {
"checkout_proof": _good_checkout(),
"inventory": _good_inventory(),
"validation": _good_validation(),
"contamination": _good_contamination(),
"identity_eligible": True,
"merge_performed": False,
"issue_status_verified": True,
}
kwargs.update(overrides)
return build_final_report(**kwargs)
def test_fully_proven_run_is_grade_a(self):
report = self._report()
self.assertEqual(report["grade"], "A")
self.assertEqual(report["violations"], [])
# Behavior 6: each distinction is a distinct field.
self.assertTrue(report["identity_eligible"])
self.assertTrue(report["pr_author_distinct_from_reviewer"])
self.assertEqual(report["session_contamination"], "clean")
self.assertTrue(report["validated_on_pinned_head"])
self.assertFalse(report["merge_performed"])
self.assertTrue(report["issue_status_verified"])
def test_unproven_checkout_downgrades_and_blocks_merge(self):
proof = verify_pinned_head_checkout(
pinned_head_sha=PINNED,
local_head_sha=OTHER,
pr_base_ref="master",
diff_base_ref="master",
)
report = self._report(checkout_proof=proof)
self.assertNotEqual(report["grade"], "A")
self.assertFalse(report["merge_allowed"])
self.assertFalse(report["validated_on_pinned_head"])
def test_merge_claim_without_checkout_proof_is_a_violation(self):
proof = verify_pinned_head_checkout(
pinned_head_sha=PINNED,
local_head_sha=OTHER,
pr_base_ref="master",
diff_base_ref="master",
)
report = self._report(checkout_proof=proof, merge_performed=True)
self.assertEqual(report["grade"], "blocked")
self.assertTrue(report["violations"])
def test_unread_validation_output_cannot_be_reported_as_passed(self):
report = self._report(validation=_good_validation(output_read=False))
self.assertFalse(report["validation_passed"])
self.assertNotEqual(report["grade"], "A")
def test_weak_validation_downgrades(self):
report = self._report(
validation=_good_validation(passed=None, failed=None, skipped=None)
)
self.assertNotEqual(report["grade"], "A")
def test_incomplete_inventory_downgrades(self):
inventory = assess_inventory_completeness(
repo_reports=[],
required_repos=["Scaled-Tech-Consulting/Gitea-Tools"],
)
report = self._report(inventory=inventory)
self.assertNotEqual(report["grade"], "A")
def test_unknown_contamination_downgrades_and_blocks_merge(self):
contamination = assess_self_review_contamination(
reviewer_identity="sysadmin",
pr_author="jcwalker3",
session_authored=True,
evidence_source=None,
)
report = self._report(contamination=contamination)
self.assertNotEqual(report["grade"], "A")
self.assertFalse(report["merge_allowed"])
self.assertEqual(report["session_contamination"], "unknown")
def test_merge_by_contaminated_reviewer_is_a_violation(self):
contamination = assess_self_review_contamination(
reviewer_identity="jcwalker3",
pr_author="jcwalker3",
)
report = self._report(contamination=contamination, merge_performed=True)
self.assertEqual(report["grade"], "blocked")
self.assertTrue(report["violations"])
def test_ineligible_identity_downgrades_and_blocks_merge(self):
report = self._report(identity_eligible=False)
self.assertNotEqual(report["grade"], "A")
self.assertFalse(report["merge_allowed"])
if __name__ == "__main__":
unittest.main()