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

fix: resolve conflicts for issue 330 (master refresh)

Browse Source
This commit is contained in:
sysadmin
2026-07-07 05:21:43 -04:00
parent 94b022b24d 897d256f98
commit e9fdb356dd
7 changed files with 905 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
review_proofs.py
+365 -3
View File
@@ -1231,7 +1231,8 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
role_boundary=None, review_mutation=None,
report_text=None, review_decision_lock=None,
controller_handoff=None, capability_proof=None,
sweep_proof=None, worktree_proof=None):
sweep_proof=None, worktree_proof=None,
baseline_validation=None, project_root=None):
"""Required behavior 6 + acceptance criteria: one report, distinct proofs.
Combines the individual proof verdicts into the final-report fields the
@@ -1258,11 +1259,28 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
)
empty_queue_report = assess_empty_queue_report(report_text)
queue_status_report = (
assess_queue_status_report(report_text)
if report_text and _QUEUE_STATUS_REPORT_HINT.search(report_text)
else {"proven": True, "block": False, "reasons": [], "violations": []}
)
proof_wording = (
assess_proof_wording(report_text)
if report_text
else {"proven": True, "block": False, "reasons": [], "violations": []}
)
if baseline_validation is None and report_text:
baseline_validation = assess_reviewer_baseline_validation_proof(
report_text=report_text,
project_root=project_root,
)
elif baseline_validation is None:
baseline_validation = {
"proven": True,
"block": False,
"reasons": [],
"violations": [],
}
contamination_status = contamination.get("status", "unknown")
checkout_proven = bool(checkout_proof.get("proven"))
@@ -1397,6 +1415,16 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
"unsupported proof wording in final report (#330)"
)
downgrade_reasons.extend(proof_wording.get("reasons", []))
if not queue_status_report.get("proven"):
downgrade_reasons.append(
"queue-status report violates loaded workflow proof gates (#339)"
)
downgrade_reasons.extend(queue_status_report.get("reasons", []))
if not baseline_validation.get("proven"):
downgrade_reasons.append(
"reviewer baseline validation proof missing or failed (#325)"
)
downgrade_reasons.extend(baseline_validation.get("reasons", []))
merge_allowed = (
identity_eligible
@@ -1408,9 +1436,11 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
# #179: no merge without a proven final live-state recheck.
and live_state_proven
and worktree_proven
and baseline_validation.get("proven")
)
violations = []
violations.extend(baseline_validation.get("violations", []))
if merge_performed and not merge_allowed:
violations.append(
"merge was performed/claimed although the proofs did not allow "
@@ -1464,6 +1494,14 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
"empty_queue_trust_gate_status": empty_queue_report.get("status"),
"proof_wording_proven": bool(proof_wording.get("proven")),
"proof_wording_violations": list(proof_wording.get("violations") or []),
"queue_status_report_proven": bool(queue_status_report.get("proven")),
"queue_status_violations": list(
queue_status_report.get("violations") or []
),
"baseline_validation_proven": bool(baseline_validation.get("proven")),
"baseline_validation_violations": list(
baseline_validation.get("violations") or []
),
}
@@ -3032,6 +3070,180 @@ def build_review_mutation_proof(run_log: list[dict]) -> dict:
}
# ---------------------------------------------------------------------------
# Reviewer baseline validation (#325)
# ---------------------------------------------------------------------------
_TEST_VALIDATION_COMMAND_RE = re.compile(
r"(?:^|\s)(?:venv/)?(?:bin/)?(?:python\s+-m\s+)?"
r"(?:pytest|make\s+test|npm\s+test|cargo\s+test|go\s+test\b)",
re.IGNORECASE,
)
_PREEXISTING_FAILURE_CLAIM_RE = re.compile(
r"(?:\bsame\s+as\s+master\b|"
r"\bpre-?existing(?:\s+(?:on\s+)?master)?\b|"
r"\bfailures?\s+(?:are|is)\s+pre-?existing\b|"
r"\bmaster\s+also\s+fails?\b|"
r"\bfull-?suite\s+failures?\s+(?:are|is)\s+pre-?existing\b)",
re.IGNORECASE,
)
_REPORT_VALIDATION_CWD_RE = re.compile(
r"(?:working\s+directory|cwd|directory)\s*:\s*(\S+)",
re.IGNORECASE,
)
def _normalize_path(path: str) -> str:
return (path or "").replace("\\", "/").rstrip("/")
def _path_under_branches(path: str, project_root: str | None = None) -> bool:
"""True when *path* is inside the project's ``branches/`` directory."""
normalized = _normalize_path(path)
if not normalized:
return False
if "/branches/" in f"{normalized}/":
return True
if normalized.endswith("/branches"):
return True
if project_root:
root = _normalize_path(project_root)
if normalized.startswith(f"{root}/"):
rel = normalized[len(root) + 1 :]
return rel == "branches" or rel.startswith("branches/")
return False
def _is_test_validation_command(command: str) -> bool:
return bool(_TEST_VALIDATION_COMMAND_RE.search((command or "").strip()))
def _is_full_sha(value: str | None) -> bool:
return bool(value and _FULL_SHA.match((value or "").strip()))
def assess_reviewer_baseline_validation_proof(
*,
validation_runs: list[dict] | None = None,
baseline_proof: dict | None = None,
report_text: str | None = None,
project_root: str | None = None,
) -> dict:
"""#325: reviewer validation and baseline comparison must use ``branches/``.
*validation_runs* entries: ``command``, ``working_directory`` (or ``cwd``),
optional ``project_root``.
*baseline_proof* keys when claiming pre-existing master failures:
``worktree_path``, ``baseline_target_sha``, ``pr_head_sha``,
``baseline_failures``, ``pr_failures``, ``failure_signatures_match``,
``clean_before``, ``clean_after``.
"""
reasons: list[str] = []
violations: list[str] = []
text = report_text or ""
runs = list(validation_runs or [])
pending_command = None
for raw_line in text.splitlines():
line = raw_line.strip().lstrip("-*").strip()
lower = line.lower()
if lower.startswith("validation command:"):
pending_command = line.split(":", 1)[1].strip()
continue
cwd_match = _REPORT_VALIDATION_CWD_RE.search(line)
if cwd_match:
cwd = cwd_match.group(1).strip().rstrip(",.;")
command = pending_command or line
if _is_test_validation_command(command):
runs.append(
{
"command": command,
"working_directory": cwd,
"project_root": project_root,
}
)
pending_command = None
for run in runs:
command = (run.get("command") or "").strip()
if not command or not _is_test_validation_command(command):
continue
cwd = (run.get("working_directory") or run.get("cwd") or "").strip()
root = run.get("project_root") or project_root
if not cwd:
reasons.append(
"test validation command stated without working directory; "
"cannot prove branches-only execution (#325)"
)
continue
if not _path_under_branches(cwd, root):
violations.append(
f"test validation ran outside branches/ worktree: cwd={cwd!r}"
)
reasons.append(
"reviewer workflow must not run test suites in the main "
f"checkout; cwd {cwd!r} is not under branches/ (#325)"
)
claims_preexisting = bool(_PREEXISTING_FAILURE_CLAIM_RE.search(text))
if claims_preexisting:
proof = baseline_proof or {}
worktree = (proof.get("worktree_path") or "").strip()
if not worktree or not _path_under_branches(worktree, project_root):
reasons.append(
"pre-existing master failure claimed without a baseline "
"worktree path under branches/ (#325)"
)
if not _is_full_sha(proof.get("baseline_target_sha")):
reasons.append(
"pre-existing master failure claimed without "
"baseline_target_sha proof (#325)"
)
if not _is_full_sha(proof.get("pr_head_sha")):
reasons.append(
"pre-existing master failure claimed without pr_head_sha "
"proof (#325)"
)
baseline_failures = proof.get("baseline_failures")
pr_failures = proof.get("pr_failures")
if baseline_failures is None or pr_failures is None:
reasons.append(
"pre-existing master failure claimed without baseline and "
"PR failure listings (#325)"
)
if proof.get("failure_signatures_match") is not True:
reasons.append(
"pre-existing master failure claimed without proven matching "
"failure signatures (#325)"
)
if proof.get("clean_before") is not True:
reasons.append(
"baseline worktree clean-before validation not proven (#325)"
)
if proof.get("clean_after") is not True:
reasons.append(
"baseline worktree clean-after validation not proven (#325)"
)
proven = not reasons and not violations
return {
"proven": proven,
"block": bool(violations),
"claims_preexisting": claims_preexisting,
"reasons": reasons,
"violations": violations,
"safe_next_action": (
"create a clean baseline worktree under branches/, e.g. "
"branches/baseline-master-pr<N>, and rerun validation there"
if not proven
else "proceed"
),
}
# ---------------------------------------------------------------------------
# Identity disclosure (#305)
# ---------------------------------------------------------------------------
@@ -3113,9 +3325,22 @@ def assess_email_disclosure(
}
_QUEUE_STATUS_REPORT_HINT = re.compile(
r"queue[- ]status|selected pr:\s*none|no pr selected|"
r"queue[- ]status[- ]only",
re.I,
)
_GATE_PASSED_VALUE = re.compile(r"\bpassed\b", re.I)
_NOT_APPLICABLE_VALUE = re.compile(
r"\b(?:none|not applicable|n/?a|not run|not verified|—|-)\b",
re.I,
)
_PRIOR_PROOF_LABEL = re.compile(
r"prior (?:blocker|proof|request[- ]changes|feedback)|"
r"head sha unchanged since|prior blocker reused",
r"head sha unchanged since|prior blocker reused|labeled as prior",
re.I,
)
@@ -3123,7 +3348,8 @@ _PAGINATION_FINALITY_EVIDENCE = re.compile(
r"pagination_complete\s*:\s*true|inventory_complete\s*:\s*true|"
r"is_final_page\s*:\s*true|pages_fetched|has_more\s*:\s*false|"
r"no next page|final[- ]page|pr_inventory_trust_gate\.status|"
r"total_count\s*:|pagination.*(?:final|complete)",
r"total_count\s*:|pagination.*(?:final|complete)|"
r"inventory pagination proof:",
re.I,
)
@@ -3186,6 +3412,48 @@ _PROOF_WORDING_RULES: tuple[dict, ...] = (
},
)
_QUEUE_STATUS_GATES_NO_PR = (
"already-landed gate",
"author-safety result",
"merge preflight",
)
_REVIEW_WORKTREE_DETAIL_FIELDS = (
"review worktree dirty before validation",
"review worktree dirty after validation",
"review worktree head state",
)
def _controller_handoff_field_map(report_text: str | None) -> dict[str, str]:
"""Parse ``- Field: value`` lines from the Controller Handoff section."""
section = _handoff_section_lines(report_text)
if section is None:
return {}
fields: dict[str, str] = {}
for line in section:
stripped = line.strip().lstrip("-*").strip()
if ":" not in stripped:
continue
key, value = stripped.split(":", 1)
fields[key.strip().lower()] = value.strip()
return fields
def _queue_status_only_run(fields: dict[str, str], report_text: str) -> bool:
selected = fields.get("selected pr", "")
if selected and _NOT_APPLICABLE_VALUE.search(selected):
return True
if re.search(r"\bnone\b", selected, re.I):
return True
if re.search(
r"no pr selected|queue[- ]status[- ]only|selected pr:\s*none",
report_text or "",
re.I,
):
return True
return not selected
def assess_proof_wording(
report_text: str | None,
@@ -3237,6 +3505,93 @@ def assess_proof_wording(
}
def assess_queue_status_report(
report_text: str | None,
*,
session_evidence: dict | None = None,
) -> dict:
"""Issue #339: reject contradictory reviewer queue-status-only reports."""
text = report_text or ""
fields = _controller_handoff_field_map(text)
violations: list[str] = []
proof = assess_proof_wording(text, session_evidence=session_evidence)
violations.extend(proof.get("violations", []))
if re.search(r"prior diagnostic", text, re.I) and not _PRIOR_PROOF_LABEL.search(
text
):
violations.append(
"prior diagnostic worktree simulation used as current conflict proof"
)
queue_only = _queue_status_only_run(fields, text)
if queue_only:
for gate in _QUEUE_STATUS_GATES_NO_PR:
value = fields.get(gate, "")
if value and _GATE_PASSED_VALUE.search(value):
if not _NOT_APPLICABLE_VALUE.search(value):
violations.append(
f"{gate} cannot be 'passed' when no PR is selected (#339)"
)
worktree_used = fields.get("review worktree used", "")
if worktree_used and re.search(r"\bfalse\b", worktree_used, re.I):
for detail_field in _REVIEW_WORKTREE_DETAIL_FIELDS:
detail_value = fields.get(detail_field, "")
if (
detail_value
and not _NOT_APPLICABLE_VALUE.search(detail_value)
):
violations.append(
f"{detail_field} must be 'not applicable' or 'none' "
"when no review worktree was created"
)
blockers = fields.get("blockers", "")
if blockers and re.search(r"\bnone\b", blockers, re.I):
if re.search(
r"all (?:open )?prs? (?:are |is )?(?:conflicted|blocked|unverified)|"
r"every (?:open )?pr (?:is )?(?:conflicted|blocked|unverified)",
text,
re.I,
):
violations.append(
"Blockers: none contradicts report stating all PRs are "
"conflicted, blocked, or unverified"
)
if re.search(r"skipped (?:pr|#)|earlier pr.*skipped", text, re.I):
has_skip_proof = bool(
re.search(
r"current[- ]session|prior blocker reused|conflict proof:|"
r"gitea_view_pr|review[- ]feedback proof|unchanged head",
text,
re.I,
)
or (session_evidence or {}).get("skip_proof_per_pr")
)
if not has_skip_proof:
violations.append(
"skipped PRs listed without current-session or labeled prior proof"
)
if re.search(r"workflows/review-merge-pr\.md", text, re.I) and violations:
violations.append(
"canonical workflow cited but mandatory queue-status proof gates violated"
)
deduped = list(dict.fromkeys(violations))
proven = not deduped
return {
"proven": proven,
"block": not proven,
"queue_status_only": queue_only,
"violations": deduped,
"reasons": deduped,
}
# ---------------------------------------------------------------------------
# Git ref mutations (#297)
# ---------------------------------------------------------------------------
@@ -3351,3 +3706,10 @@ def assess_git_ref_mutation_report(report_text, *, command_log=None):
"ref_mutating_commands": ref_commands,
"fetch_targets": fetch_targets,
}
def assess_non_mergeable_skip_proof(report_text, **kwargs):
"""#322: require conflict proof when skipping non-mergeable PRs."""
from reviewer_mergeability_skip import assess_non_mergeable_skip_proof as _assess
return _assess(report_text, **kwargs)
reviewer_mergeability_skip.py
+159
View File
@@ -0,0 +1,159 @@
"""Non-mergeable PR skip conflict-proof verifier for reviewer reports (#322)."""
from __future__ import annotations
import re
from typing import Any
_FULL_SHA = re.compile(r"\b[0-9a-f]{40}\b", re.IGNORECASE)
_SHORT_SHA = re.compile(r"\b[0-9a-f]{7,39}\b", re.IGNORECASE)
_PR_NUMBER_RE = re.compile(r"(?:\bPR\s*#?|#)(\d+)\b", re.IGNORECASE)
_MERGEABILITY_FALSE_RE = re.compile(
r"(?:mergeable\s*:\s*false|not mergeable|non[- ]mergeable|mergeability\s*:\s*false|"
r"mergeability result\s*:\s*false)",
re.IGNORECASE,
)
_CONFLICT_PROOF_RE = re.compile(
r"(?:merge-tree|git merge-tree|gitea_view_pr|gitea_check_pr_eligibility|"
r"conflict proof|mergeability tool|merge simulation|conflicting files?)",
re.IGNORECASE,
)
_CONFLICTING_FILES_RE = re.compile(
r"(?:conflicting files?|conflict files?)\s*:\s*(.+)$",
re.IGNORECASE | re.MULTILINE,
)
_HEAD_CHANGED_RE = re.compile(
r"(?:head (?:changed|unchanged)|head sha (?:changed|unchanged)|"
r"head changed since|head unchanged since)",
re.IGNORECASE,
)
_MERGEABILITY_UNVERIFIED_RE = re.compile(
r"\bMERGEABILITY_UNVERIFIED\b",
)
def _pr_section(text: str, pr_number: int) -> str:
"""Return report lines likely describing one skipped PR."""
lines = (text or "").splitlines()
chunks: list[str] = []
capture = False
token = f"#{pr_number}"
for line in lines:
lower = line.lower()
if token in lower or f"pr {pr_number}" in lower or f"pr#{pr_number}" in lower.replace(" ", ""):
capture = True
chunks.append(line)
continue
if capture:
if _PR_NUMBER_RE.search(line) and token not in line.lower():
break
if line.strip() == "" and len(chunks) > 3:
break
chunks.append(line)
if chunks:
return "\n".join(chunks)
return text or ""
def _has_head_sha(text: str, head_sha: str | None) -> bool:
if not head_sha:
return bool(_FULL_SHA.search(text) or _SHORT_SHA.search(text))
head = head_sha.strip().lower()
if head in text.lower():
return True
if len(head) >= 7 and head[:7] in text.lower():
return True
return bool(_FULL_SHA.search(text))
def assess_non_mergeable_skip_proof(
report_text: str,
*,
skipped_prs: list[dict] | None = None,
) -> dict[str, Any]:
"""Validate skipped non-mergeable PR documentation in reviewer reports (#322)."""
text = report_text or ""
reasons: list[str] = []
assessments: list[dict[str, Any]] = []
for entry in skipped_prs or []:
pr_number = entry.get("pr_number")
if pr_number is None:
reasons.append("skipped PR entry missing pr_number")
continue
pr_number = int(pr_number)
section = _pr_section(text, pr_number)
mergeable = entry.get("mergeable")
verified = entry.get("mergeability_verified")
classification = (entry.get("classification") or "").strip().upper()
head_sha = (entry.get("head_sha") or "").strip() or None
item_reasons: list[str] = []
if f"#{pr_number}" not in text.lower() and f"pr {pr_number}" not in text.lower():
item_reasons.append(f"skipped PR #{pr_number} not documented in final report")
if mergeable is False or verified is False:
if not _MERGEABILITY_UNVERIFIED_RE.search(section) and verified is False:
if "MERGEABILITY_UNVERIFIED" not in classification:
item_reasons.append(
f"PR #{pr_number} conflict proof unavailable; report must classify "
"MERGEABILITY_UNVERIFIED"
)
elif verified is not False:
if not _MERGEABILITY_FALSE_RE.search(section):
item_reasons.append(
f"PR #{pr_number} skip missing mergeability:false proof"
)
if not _has_head_sha(section, head_sha):
item_reasons.append(
f"PR #{pr_number} skip missing current head SHA"
)
if not _CONFLICT_PROOF_RE.search(section):
item_reasons.append(
f"PR #{pr_number} skip missing conflict proof command/tool"
)
if entry.get("head_changed_since_prior_blocker") is not None:
if not _HEAD_CHANGED_RE.search(section):
item_reasons.append(
f"PR #{pr_number} skip missing head-changed-since-blocker proof"
)
if (
entry.get("conflicting_files")
and not _CONFLICTING_FILES_RE.search(section)
and not any(
path.lower() in section.lower()
for path in (entry.get("conflicting_files") or [])
)
):
item_reasons.append(
f"PR #{pr_number} has conflicting files in session proof but "
"report omits file-level conflict proof"
)
proven = not item_reasons
assessments.append({
"pr_number": pr_number,
"proven": proven,
"classification": classification or (
"MERGEABILITY_UNVERIFIED" if verified is False else "NON_MERGEABLE_SKIPPED"
),
"reasons": item_reasons,
})
reasons.extend(item_reasons)
proven = not reasons
return {
"proven": proven,
"block": not proven,
"downgraded": False,
"assessments": assessments,
"reasons": reasons,
"safe_next_action": (
"document each skipped non-mergeable PR with head SHA, mergeability result, "
"conflict proof command, conflicting files, and head-change status"
if reasons
else "proceed"
),
}
skills/llm-project-workflow/schemas/review-merge-final-report.md
+12
View File
@@ -77,6 +77,18 @@ When eligibility class is `ALREADY_LANDED_RECONCILE_REQUIRED`:
Identity format: `username / profile` (not personal email unless required — #305).
### Queue-status-only runs (no selected PR)
When the run inventories the queue but selects no PR for review:
- Selected PR: `none`
- Already-landed gate, Author-safety result, Merge preflight: `not applicable` or `not run` — never `passed`
- Review worktree detail fields: `not applicable` or `none` when Review worktree used is `false`
- Blockers must not be `none` if the narrative says all open PRs are conflicted, blocked, or unverified
- Inventory pagination proof must cite final-page metadata, not default page-size assumptions
Verifier: `review_proofs.assess_queue_status_report()`.
Narrative final report and controller handoff must agree on eligibility class,
candidate/reviewed head SHA, mutation state, worktree usage, review decision,
terminal review mutation, merge result, and linked issue status.
tests/test_llm_workflow_split.py
+7 -1
View File
@@ -121,4 +121,10 @@ def test_start_issue_template_references_work_issue_workflow():
def test_create_issue_final_report_verifier_exported():
from review_proofs import assess_create_issue_final_report
assert callable(assess_create_issue_final_report)
assert callable(assess_create_issue_final_report)
def test_non_mergeable_skip_verifier_exported():
from review_proofs import assess_non_mergeable_skip_proof
assert callable(assess_non_mergeable_skip_proof)
tests/test_queue_status_report.py
+91
View File
@@ -0,0 +1,91 @@
"""Doc-contract checks for queue-status report verifier (#339)."""
import unittest
from review_proofs import assess_queue_status_report, build_final_report
BAD_QUEUE_STATUS_REPORT = """
## PR Queue Status
Loaded workflows/review-merge-pr.md. Inventory complete because gitea_list_prs
returned 6 open PRs, fewer than the default Gitea page size.
PR #286 skipped: prior diagnostic worktree simulation showed merge conflicts.
Live blocker proof for all skipped PRs.
All open PRs are conflicted or blocked. Blockers: none.
## Controller Handoff
- Task: review queue status
- Repo: Scaled-Tech-Consulting/Gitea-Tools
- Role: reviewer
- Identity: reviewer1 / prgs-reviewer
- Selected PR: none
- Inventory pagination proof: assumed complete (6 < 50)
- Already-landed gate: passed
- Author-safety result: passed
- Merge preflight: passed
- Review worktree used: false
- Review worktree dirty before validation: false
- Blockers: none
- Current status: queue inspected; no PR selected
"""
class TestQueueStatusReport(unittest.TestCase):
def test_bad_fixture_fails_closed(self):
result = assess_queue_status_report(BAD_QUEUE_STATUS_REPORT)
self.assertFalse(result["proven"])
self.assertTrue(result["block"])
self.assertTrue(result["queue_status_only"])
joined = " ".join(result["violations"]).lower()
self.assertIn("already-landed gate", joined)
self.assertIn("author-safety", joined)
self.assertIn("merge preflight", joined)
self.assertIn("prior diagnostic", joined)
self.assertIn("blockers", joined)
def test_passes_with_not_applicable_gates(self):
report = """
## Controller Handoff
- Selected PR: none
- Inventory pagination proof: is_final_page: true, has_more: false
- Already-landed gate: not applicable
- Author-safety result: not run
- Merge preflight: not applicable
- Review worktree used: false
- Review worktree dirty before validation: not applicable
- Blockers: all PRs skipped pending reviewer
"""
result = assess_queue_status_report(report)
self.assertTrue(result["proven"])
def test_rejects_passed_gates_without_selected_pr(self):
report = """
## Controller Handoff
- Selected PR: none
- Already-landed gate: passed
- Blockers: PR #1 conflicted
"""
result = assess_queue_status_report(report)
self.assertFalse(result["proven"])
self.assertIn("already-landed gate", " ".join(result["violations"]).lower())
def test_build_final_report_downgrades_bad_queue_status(self):
report = build_final_report(
{"proven": True},
{"complete": True},
{"claimable": True, "verdict": "strong"},
{"status": "clean"},
True,
False,
True,
report_text=BAD_QUEUE_STATUS_REPORT,
)
self.assertEqual(report["grade"], "downgraded")
self.assertFalse(report["queue_status_report_proven"])
self.assertTrue(report["queue_status_violations"])
if __name__ == "__main__":
unittest.main()
tests/test_review_proofs.py
+121
View File
@@ -24,6 +24,7 @@ from review_proofs import ( # noqa: E402
ISSUE_SELECTION_CONTINUATION_EXPLICIT,
ISSUE_SELECTION_REPRESENTED_BY_OPEN_PR,
assess_author_pr_report,
assess_reviewer_baseline_validation_proof,
assess_capability_evidence,
assess_capability_proof,
assess_contradictory_no_pr_claim,
@@ -2160,5 +2161,125 @@ class TestCreateIssueFinalReport(unittest.TestCase):
self.assertTrue(result["downgraded"])
class TestReviewerBaselineValidationProof(unittest.TestCase):
"""Issue #325: baseline validation must use branches/ worktrees."""
ROOT = "/repo/Gitea-Tools"
def _good_baseline_proof(self, **overrides):
proof = {
"worktree_path": f"{self.ROOT}/branches/baseline-master-pr280",
"baseline_target_sha": PINNED,
"pr_head_sha": OTHER,
"baseline_failures": ["tests/test_foo.py::test_bar"],
"pr_failures": ["tests/test_foo.py::test_bar"],
"failure_signatures_match": True,
"clean_before": True,
"clean_after": True,
}
proof.update(overrides)
return proof
def test_main_checkout_test_run_blocks(self):
result = assess_reviewer_baseline_validation_proof(
validation_runs=[
{
"command": "venv/bin/pytest tests/",
"working_directory": self.ROOT,
"project_root": self.ROOT,
}
],
project_root=self.ROOT,
)
self.assertFalse(result["proven"])
self.assertTrue(result["block"])
self.assertTrue(result["violations"])
def test_branches_worktree_test_run_passes(self):
result = assess_reviewer_baseline_validation_proof(
validation_runs=[
{
"command": "venv/bin/pytest tests/",
"working_directory": f"{self.ROOT}/branches/review-pr-280",
"project_root": self.ROOT,
}
],
project_root=self.ROOT,
)
self.assertTrue(result["proven"])
def test_preexisting_claim_without_baseline_proof_fails(self):
result = assess_reviewer_baseline_validation_proof(
report_text=(
"Validation: full-suite failures are pre-existing on master."
),
project_root=self.ROOT,
)
self.assertFalse(result["proven"])
self.assertTrue(result["claims_preexisting"])
def test_preexisting_claim_with_complete_baseline_proof_passes(self):
result = assess_reviewer_baseline_validation_proof(
report_text="Failures are pre-existing on master.",
baseline_proof=self._good_baseline_proof(),
project_root=self.ROOT,
)
self.assertTrue(result["proven"])
def test_mismatched_failures_block_preexisting_claim(self):
result = assess_reviewer_baseline_validation_proof(
report_text="Same as master.",
baseline_proof=self._good_baseline_proof(
failure_signatures_match=False,
pr_failures=["tests/test_other.py::test_other"],
),
project_root=self.ROOT,
)
self.assertFalse(result["proven"])
def test_report_parsed_main_checkout_cwd_blocks(self):
result = assess_reviewer_baseline_validation_proof(
report_text=(
"- Validation command: venv/bin/pytest tests/\n"
f"- Working directory: {self.ROOT}\n"
),
project_root=self.ROOT,
)
self.assertFalse(result["proven"])
self.assertTrue(result["block"])
def test_build_final_report_downgrades_main_checkout_validation(self):
report = 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-feat-issue-224",
"porcelain_status": "",
"pr_scope_files": ["docs/wiki/Repositories.md"],
"scratch_used": True,
},
report_text=(
"- Validation command: venv/bin/pytest tests/\n"
f"- Working directory: {self.ROOT}\n"
),
project_root=self.ROOT,
)
self.assertNotEqual(report["grade"], "A")
self.assertFalse(report["baseline_validation_proven"])
if __name__ == "__main__":
unittest.main()
tests/test_reviewer_mergeability_skip.py
+150
View File
@@ -0,0 +1,150 @@
"""Tests for non-mergeable skip conflict-proof verifier (#322)."""
import sys
import unittest
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from reviewer_mergeability_skip import assess_non_mergeable_skip_proof # noqa: E402
HEAD_A = "0fdc8f582026b72a229d59a172c0a63ac4aaeaf9"
HEAD_B = "1111111111111111111111111111111111111111"
def _good_skip_report(pr_number: int, head: str, files: str = "review_proofs.py") -> str:
return "\n".join([
f"Skipped PR #{pr_number} (non-mergeable).",
f"- PR number: #{pr_number}",
f"- Current head SHA: {head}",
"- Mergeability result: false",
"- Conflict proof command: git merge-tree $(git merge-base prgs/master prgs/feat) prgs/master prgs/feat",
f"- Conflicting files: {files}",
"- Head unchanged since prior blocker",
"- Blocking category: merge conflict",
])
class TestNonMergeableSkipProof(unittest.TestCase):
def test_no_skips_passes(self):
report = "Selected PR #203 for review. No earlier PRs skipped."
result = assess_non_mergeable_skip_proof(report, skipped_prs=[])
self.assertTrue(result["proven"])
self.assertFalse(result["block"])
def test_documented_non_mergeable_skip_passes(self):
report = _good_skip_report(282, HEAD_A)
result = assess_non_mergeable_skip_proof(
report,
skipped_prs=[{
"pr_number": 282,
"mergeable": False,
"head_sha": HEAD_A,
"mergeability_verified": True,
"conflicting_files": ["review_proofs.py"],
"head_changed_since_prior_blocker": False,
}],
)
self.assertTrue(result["proven"])
def test_skip_without_conflict_proof_blocks(self):
report = "\n".join([
"Skipped PR #284 because it was not mergeable.",
f"Head SHA: {HEAD_B}",
])
result = assess_non_mergeable_skip_proof(
report,
skipped_prs=[{
"pr_number": 284,
"mergeable": False,
"head_sha": HEAD_B,
"mergeability_verified": True,
}],
)
self.assertFalse(result["proven"])
self.assertTrue(result["block"])
self.assertTrue(any("conflict proof" in r for r in result["reasons"]))
def test_missing_head_sha_blocks(self):
report = _good_skip_report(282, HEAD_A).replace(HEAD_A, "")
result = assess_non_mergeable_skip_proof(
report,
skipped_prs=[{
"pr_number": 282,
"mergeable": False,
"head_sha": HEAD_A,
"mergeability_verified": True,
}],
)
self.assertFalse(result["proven"])
def test_mergeability_unverified_classification_passes(self):
report = "\n".join([
"Skipped PR #282.",
"Classification: MERGEABILITY_UNVERIFIED",
"Conflict proof could not be retrieved in this session.",
])
result = assess_non_mergeable_skip_proof(
report,
skipped_prs=[{
"pr_number": 282,
"mergeable": False,
"mergeability_verified": False,
}],
)
self.assertTrue(result["proven"])
def test_unverified_without_classification_blocks(self):
report = "Skipped PR #282 due to conflicts."
result = assess_non_mergeable_skip_proof(
report,
skipped_prs=[{
"pr_number": 282,
"mergeable": False,
"mergeability_verified": False,
}],
)
self.assertFalse(result["proven"])
self.assertTrue(any("MERGEABILITY_UNVERIFIED" in r for r in result["reasons"]))
def test_non_mergeable_without_file_details_still_passes_with_command(self):
report = _good_skip_report(284, HEAD_B, files="not available")
result = assess_non_mergeable_skip_proof(
report,
skipped_prs=[{
"pr_number": 284,
"mergeable": False,
"head_sha": HEAD_B,
"mergeability_verified": True,
"conflicting_files": [],
}],
)
self.assertTrue(result["proven"])
def test_stale_head_requires_head_change_field(self):
report = _good_skip_report(282, HEAD_A).replace(
"- Head unchanged since prior blocker",
"",
)
result = assess_non_mergeable_skip_proof(
report,
skipped_prs=[{
"pr_number": 282,
"mergeable": False,
"head_sha": HEAD_A,
"mergeability_verified": True,
"head_changed_since_prior_blocker": True,
}],
)
self.assertFalse(result["proven"])
self.assertTrue(any("head-changed" in r for r in result["reasons"]))
class TestExport(unittest.TestCase):
def test_review_proofs_reexport(self):
from review_proofs import assess_non_mergeable_skip_proof as exported
self.assertTrue(callable(exported))
if __name__ == "__main__":
unittest.main()