fix: resolve conflicts for PR #382
Merge prgs/master; keep already-landed oldest-eligible-PR test from #295 alongside master already-landed gate and target-branch freshness tests.
This commit is contained in:
+131
-2
@@ -87,15 +87,35 @@ _ALREADY_LANDED_RE = re.compile(
|
||||
r"already[- ]landed|ancestor proof\s*:\s*passed|eligibility class\s*:\s*already_landed",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_ALREADY_LANDED_GATE_FIRED_RE = re.compile(
|
||||
r"\b(?:fired|passed|true|yes|already_landed_reconcile_required)\b",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_ELIGIBLE_REVIEW_RE = re.compile(
|
||||
r"eligible for (?:review|merge)|ready for (?:review|merge)|"
|
||||
r"(?:next|oldest)\s+eligible\s+pr",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_APPROVED_STATE_RE = re.compile(r"\bapproved?\b", re.IGNORECASE)
|
||||
_MERGED_STATE_RE = re.compile(r"\bmerged\b", re.IGNORECASE)
|
||||
_READY_TO_MERGE_STATE_RE = re.compile(r"\bready_to_merge\b|\bready to merge\b", re.IGNORECASE)
|
||||
_NEGATED_STATE_RE = re.compile(
|
||||
r"\b(?:not|no|none|n/a|never)\b|request_changes|not attempted",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_REVIEWED_LANDED_RE = re.compile(
|
||||
r"(?:reviewed|approved).{0,40}already[- ]landed|already[- ]landed.{0,40}(?:reviewed|eligible)",
|
||||
re.IGNORECASE | re.DOTALL,
|
||||
)
|
||||
_TARGET_BRANCH_UP_TO_DATE_RE = re.compile(
|
||||
r"\btarget branch (?:is )?up[- ]to[- ]date\b",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_TARGET_BRANCH_SHA_RE = re.compile(
|
||||
r"target branch sha\s*:\s*[0-9a-f]{40}",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_FULL_SHA_RE = re.compile(r"\b[0-9a-f]{40}\b", re.IGNORECASE)
|
||||
_RECONCILE_STALE_FIELDS = (
|
||||
"pr number opened",
|
||||
"pinned reviewed head",
|
||||
@@ -191,6 +211,54 @@ def _handoff_fields(report_text: str) -> dict[str, str]:
|
||||
return fields
|
||||
|
||||
|
||||
def _already_landed_gate_fired(report_text: str) -> bool:
|
||||
text = report_text or ""
|
||||
fields = _handoff_fields(text)
|
||||
|
||||
gate_value = fields.get("already-landed gate", "")
|
||||
if gate_value and _ALREADY_LANDED_GATE_FIRED_RE.search(gate_value):
|
||||
return True
|
||||
|
||||
ancestor_value = fields.get("ancestor proof", "")
|
||||
if re.search(r"\bpassed\b", ancestor_value, re.IGNORECASE):
|
||||
return True
|
||||
|
||||
eligibility_value = fields.get("eligibility class", "")
|
||||
if "already_landed" in eligibility_value.lower():
|
||||
return True
|
||||
|
||||
if re.search(
|
||||
r"\bpr\b.{0,60}\balready[- ]landed\b|\balready[- ]landed\b.{0,60}\bpr\b",
|
||||
text,
|
||||
re.IGNORECASE | re.DOTALL,
|
||||
):
|
||||
return True
|
||||
|
||||
return bool(_ALREADY_LANDED_RE.search(text))
|
||||
|
||||
|
||||
def _positive_review_state_terms(fields: dict[str, str]) -> list[str]:
|
||||
terms: list[str] = []
|
||||
for key, value in fields.items():
|
||||
lowered_key = key.lower()
|
||||
lowered_value = value.lower()
|
||||
if _NEGATED_STATE_RE.search(value):
|
||||
continue
|
||||
if lowered_key == "review decision" and _APPROVED_STATE_RE.search(value):
|
||||
terms.append("APPROVED")
|
||||
elif lowered_key == "merge result" and _MERGED_STATE_RE.search(value):
|
||||
terms.append("MERGED")
|
||||
elif _READY_TO_MERGE_STATE_RE.search(value):
|
||||
terms.append("READY_TO_MERGE")
|
||||
elif lowered_value in {"approved", "approve"}:
|
||||
terms.append("APPROVED")
|
||||
elif lowered_value == "merged":
|
||||
terms.append("MERGED")
|
||||
elif lowered_value == "ready_to_merge":
|
||||
terms.append("READY_TO_MERGE")
|
||||
return sorted(set(terms))
|
||||
|
||||
|
||||
def _rule_shared_controller_handoff(
|
||||
report_text: str,
|
||||
task_kind: str,
|
||||
@@ -493,7 +561,7 @@ def _rule_reviewer_main_checkout_path(report_text: str) -> list[dict[str, str]]:
|
||||
|
||||
def _rule_reviewer_already_landed_eligible(report_text: str) -> list[dict[str, str]]:
|
||||
text = report_text or ""
|
||||
if not _ALREADY_LANDED_RE.search(text):
|
||||
if not _already_landed_gate_fired(text):
|
||||
return []
|
||||
if not _ELIGIBLE_REVIEW_RE.search(text):
|
||||
return []
|
||||
@@ -508,6 +576,65 @@ def _rule_reviewer_already_landed_eligible(report_text: str) -> list[dict[str, s
|
||||
]
|
||||
|
||||
|
||||
def _rule_reviewer_already_landed_state(report_text: str) -> list[dict[str, str]]:
|
||||
text = report_text or ""
|
||||
if not _already_landed_gate_fired(text):
|
||||
return []
|
||||
|
||||
terms = _positive_review_state_terms(_handoff_fields(text))
|
||||
if not terms:
|
||||
return []
|
||||
|
||||
return [
|
||||
validator_finding(
|
||||
"reviewer.already_landed_review_state",
|
||||
"block",
|
||||
"Review decision",
|
||||
"already-landed gate fired but final report claims "
|
||||
f"{', '.join(terms)} state",
|
||||
"stop normal review/merge; classify as ALREADY_LANDED_RECONCILE_REQUIRED",
|
||||
)
|
||||
]
|
||||
|
||||
|
||||
def _rule_reviewer_target_branch_freshness(
|
||||
report_text: str,
|
||||
*,
|
||||
action_log: list[dict] | None = None,
|
||||
) -> list[dict[str, str]]:
|
||||
text = report_text or ""
|
||||
if not _TARGET_BRANCH_UP_TO_DATE_RE.search(text):
|
||||
return []
|
||||
|
||||
fields = _handoff_fields(text)
|
||||
fetch_reported = bool(_GIT_FETCH_RE.search(text)) or any(
|
||||
_GIT_FETCH_RE.search(str(e.get("command") or e.get("action") or ""))
|
||||
for e in (action_log or [])
|
||||
)
|
||||
target_sha_reported = bool(_TARGET_BRANCH_SHA_RE.search(text)) or any(
|
||||
"target branch" in key and "sha" in key and _FULL_SHA_RE.search(value)
|
||||
for key, value in fields.items()
|
||||
)
|
||||
if fetch_reported and target_sha_reported:
|
||||
return []
|
||||
|
||||
missing = []
|
||||
if not fetch_reported:
|
||||
missing.append("fresh git fetch")
|
||||
if not target_sha_reported:
|
||||
missing.append("target branch SHA")
|
||||
return [
|
||||
validator_finding(
|
||||
"reviewer.target_branch_freshness_proof",
|
||||
"block",
|
||||
"Target branch SHA",
|
||||
"target branch up-to-date claim lacks " + " and ".join(missing),
|
||||
"fetch the target branch and report the exact target branch SHA before "
|
||||
"claiming it is up to date",
|
||||
)
|
||||
]
|
||||
|
||||
|
||||
def _rule_reconcile_controller_handoff(report_text: str) -> list[dict[str, str]]:
|
||||
from review_proofs import _handoff_section_lines
|
||||
|
||||
@@ -707,6 +834,8 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
|
||||
_rule_reviewer_main_checkout_baseline,
|
||||
_rule_reviewer_main_checkout_path,
|
||||
_rule_reviewer_already_landed_eligible,
|
||||
_rule_reviewer_already_landed_state,
|
||||
_rule_reviewer_target_branch_freshness,
|
||||
_rule_reviewer_mutation_ledger,
|
||||
_rule_reviewer_review_mutation,
|
||||
],
|
||||
@@ -876,4 +1005,4 @@ def assess_final_report_validator(
|
||||
"safe_next_action": _primary_safe_next_action(findings),
|
||||
"complete": grade == "A",
|
||||
"handoff_heading": HANDOFF_HEADING,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1710,6 +1710,11 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
|
||||
if report_text
|
||||
else {"proven": True, "block": False, "reasons": [], "violations": []}
|
||||
)
|
||||
reconcile_linked_issue = (
|
||||
assess_reconcile_linked_issue_report(report_text)
|
||||
if report_text and _RECONCILE_LINKED_ISSUE_HINT.search(report_text)
|
||||
else {"proven": True, "block": False, "reasons": []}
|
||||
)
|
||||
already_landed_handoff = (
|
||||
assess_already_landed_handoff_report(report_text)
|
||||
if report_text
|
||||
@@ -1892,6 +1897,11 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
|
||||
"queue-status report violates loaded workflow proof gates (#339)"
|
||||
)
|
||||
downgrade_reasons.extend(queue_status_report.get("reasons", []))
|
||||
if not reconcile_linked_issue.get("proven"):
|
||||
downgrade_reasons.append(
|
||||
"reconciliation linked issue status lacks live fetch proof (#300)"
|
||||
)
|
||||
downgrade_reasons.extend(reconcile_linked_issue.get("reasons", []))
|
||||
if not pr_queue_cleanup_report.get("proven"):
|
||||
downgrade_reasons.append(
|
||||
"PR-only cleanup report violates cleanup-mode proof gates (#390)"
|
||||
@@ -2005,6 +2015,10 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
|
||||
"queue_status_violations": list(
|
||||
queue_status_report.get("violations") or []
|
||||
),
|
||||
"reconcile_linked_issue_proven": bool(reconcile_linked_issue.get("proven")),
|
||||
"reconcile_linked_issue_violations": list(
|
||||
reconcile_linked_issue.get("reasons") or []
|
||||
),
|
||||
"pr_queue_cleanup_report_proven": bool(
|
||||
pr_queue_cleanup_report.get("proven")
|
||||
),
|
||||
@@ -4909,6 +4923,11 @@ _QUEUE_STATUS_REPORT_HINT = re.compile(
|
||||
re.I,
|
||||
)
|
||||
|
||||
_RECONCILE_LINKED_ISSUE_HINT = re.compile(
|
||||
r"already[- ]landed|reconcil|linked issue(?:\s+live)?\s+status",
|
||||
re.I,
|
||||
)
|
||||
|
||||
_PR_QUEUE_CLEANUP_REPORT_HINT = re.compile(
|
||||
r"workflows/pr-queue-cleanup\.md|task:\s*pr-queue-cleanup|"
|
||||
r"pr[- ]only queue cleanup",
|
||||
@@ -4922,6 +4941,7 @@ def assess_pr_queue_cleanup_report(report_text: str | None) -> dict:
|
||||
|
||||
return _assess(report_text or "")
|
||||
|
||||
|
||||
_GATE_PASSED_VALUE = re.compile(r"\bpassed\b", re.I)
|
||||
|
||||
_NOT_APPLICABLE_VALUE = re.compile(
|
||||
@@ -5345,6 +5365,13 @@ def assess_validation_worktree_edit_report(report_text, **kwargs):
|
||||
return _assess(report_text, **kwargs)
|
||||
|
||||
|
||||
def assess_reconcile_linked_issue_report(report_text, **kwargs):
|
||||
"""#300: prove linked issue status was fetched live in reconciliation handoffs."""
|
||||
from reviewer_reconcile_linked_issue import assess_reconcile_linked_issue_report as _assess
|
||||
|
||||
return _assess(report_text, **kwargs)
|
||||
|
||||
|
||||
def assess_already_landed_handoff_report(report_text, **kwargs):
|
||||
"""#299: reject stale fields after the already-landed gate fires."""
|
||||
from reviewer_already_landed_handoff import assess_already_landed_handoff_report as _assess
|
||||
|
||||
@@ -0,0 +1,191 @@
|
||||
"""Reconciliation linked-issue live proof verifier (#300)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from typing import Any
|
||||
|
||||
_HANDOFF_SECTION_RE = re.compile(r"^##\s*Controller Handoff\s*$", re.I | re.M)
|
||||
|
||||
_LINKED_ISSUE_FIELD_RE = re.compile(
|
||||
r"linked issue\s*:\s*(.+)$",
|
||||
re.I | re.M,
|
||||
)
|
||||
_LINKED_ISSUE_STATUS_RE = re.compile(
|
||||
r"linked issue(?:\s+live)?\s+status\s*:\s*(.+)$",
|
||||
re.I | re.M,
|
||||
)
|
||||
|
||||
_STATUS_CLAIM_RE = re.compile(
|
||||
r"\b(?:issue\s+#?\d+\s*\()?(open|closed|resolved)\b|"
|
||||
r"issue\s+(?:is\s+)?(open|closed|resolved)\b|"
|
||||
r"linked issue.*\b(open|closed|resolved)\b",
|
||||
re.I,
|
||||
)
|
||||
_NOT_VERIFIED_RE = re.compile(r"not verified in this session", re.I)
|
||||
|
||||
_LIVE_FETCH_PROOF_RE = re.compile(
|
||||
r"gitea_view_issue|issue fetched live|live issue fetch|"
|
||||
r"fetched linked issue.*(?:current|this) session|"
|
||||
r"linked issue (?:fetch|proof).*(?:current|this) session|"
|
||||
r"live linked issue proof",
|
||||
re.I,
|
||||
)
|
||||
|
||||
_ISSUE_ACTION_RECOMMEND_RE = re.compile(
|
||||
r"(?:recommend(?:ed)?|should|must)\s+(?:close|reopen|comment on)\s+"
|
||||
r"(?:the\s+)?(?:linked\s+)?issue|"
|
||||
r"(?:close|reopen|comment on)\s+linked issue\s+#?\d+",
|
||||
re.I,
|
||||
)
|
||||
_CAPABILITY_PROOF_RE = re.compile(
|
||||
r"capability proof|exact capability|gitea_close_issue|"
|
||||
r"gitea_mark_issue|gitea_create_issue_comment|gitea_edit_issue",
|
||||
re.I,
|
||||
)
|
||||
|
||||
_NO_LINKED_ISSUE_VALUES = frozenset({
|
||||
"",
|
||||
"none",
|
||||
"n/a",
|
||||
"not applicable",
|
||||
"no linked issue",
|
||||
"unknown",
|
||||
})
|
||||
|
||||
|
||||
def _handoff_field_map(report_text: str) -> dict[str, str]:
|
||||
text = report_text or ""
|
||||
match = _HANDOFF_SECTION_RE.search(text)
|
||||
if not match:
|
||||
return {}
|
||||
fields: dict[str, str] = {}
|
||||
for line in text[match.end() :].splitlines():
|
||||
stripped = line.strip().lstrip("-*").strip()
|
||||
if ":" not in stripped:
|
||||
continue
|
||||
key, value = stripped.split(":", 1)
|
||||
fields[key.strip().lower()] = value.strip()
|
||||
return fields
|
||||
|
||||
|
||||
def _extract_linked_issue_number(text: str, fields: dict[str, str]) -> int | None:
|
||||
linked = fields.get("linked issue", "")
|
||||
if linked.lower() in _NO_LINKED_ISSUE_VALUES:
|
||||
return None
|
||||
match = re.search(r"#?(\d+)", linked)
|
||||
if match:
|
||||
return int(match.group(1))
|
||||
field_match = _LINKED_ISSUE_FIELD_RE.search(text)
|
||||
if field_match:
|
||||
num = re.search(r"#?(\d+)", field_match.group(1))
|
||||
if num:
|
||||
return int(num.group(1))
|
||||
return None
|
||||
|
||||
|
||||
def _status_value(text: str, fields: dict[str, str]) -> str:
|
||||
for key in ("linked issue live status", "linked issue status"):
|
||||
if fields.get(key):
|
||||
return fields[key]
|
||||
match = _LINKED_ISSUE_STATUS_RE.search(text)
|
||||
return match.group(1).strip() if match else ""
|
||||
|
||||
|
||||
def _live_proof_present(text: str, session: dict, issue_number: int | None) -> bool:
|
||||
if session.get("live_fetched") or session.get("verified_live"):
|
||||
return True
|
||||
if session.get("issue_fetch_proof"):
|
||||
return True
|
||||
if _LIVE_FETCH_PROOF_RE.search(text):
|
||||
return True
|
||||
if issue_number is not None:
|
||||
pattern = re.compile(
|
||||
rf"gitea_view_issue.*#?{issue_number}|"
|
||||
rf"issue\s+#?{issue_number}.*(?:fetched|viewed).*(?:current|this) session",
|
||||
re.I,
|
||||
)
|
||||
if pattern.search(text):
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def assess_reconcile_linked_issue_report(
|
||||
report_text: str,
|
||||
*,
|
||||
reconcile_session: dict | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Validate linked issue status claims in reconciliation handoffs (#300)."""
|
||||
text = report_text or ""
|
||||
session = dict(reconcile_session or {})
|
||||
fields = _handoff_field_map(text)
|
||||
reasons: list[str] = []
|
||||
|
||||
issue_number = session.get("linked_issue_number")
|
||||
if issue_number is None:
|
||||
issue_number = _extract_linked_issue_number(text, fields)
|
||||
|
||||
if issue_number is None:
|
||||
status = _status_value(text, fields)
|
||||
if status and status.lower() not in _NO_LINKED_ISSUE_VALUES:
|
||||
if _STATUS_CLAIM_RE.search(status):
|
||||
reasons.append(
|
||||
"linked issue status reported without identifying the linked issue (#300)"
|
||||
)
|
||||
if not reasons:
|
||||
return {
|
||||
"proven": True,
|
||||
"block": False,
|
||||
"reasons": [],
|
||||
"linked_issue_number": None,
|
||||
"safe_next_action": "proceed",
|
||||
}
|
||||
|
||||
status = _status_value(text, fields)
|
||||
status_lower = status.lower()
|
||||
|
||||
if session.get("issue_missing"):
|
||||
if status_lower and "not verified" not in status_lower:
|
||||
reasons.append(
|
||||
"missing linked issue must be reported as "
|
||||
"'not verified in this session' (#300)"
|
||||
)
|
||||
elif status:
|
||||
if _NOT_VERIFIED_RE.search(status):
|
||||
pass
|
||||
elif _STATUS_CLAIM_RE.search(status):
|
||||
if not _live_proof_present(text, session, issue_number):
|
||||
reasons.append(
|
||||
"linked issue open/closed/resolved status requires live "
|
||||
"gitea_view_issue proof in the current session (#300)"
|
||||
)
|
||||
elif status_lower not in _NO_LINKED_ISSUE_VALUES:
|
||||
if not _live_proof_present(text, session, issue_number):
|
||||
reasons.append(
|
||||
"linked issue status claim requires live fetch proof or "
|
||||
"'not verified in this session' (#300)"
|
||||
)
|
||||
|
||||
if _ISSUE_ACTION_RECOMMEND_RE.search(text):
|
||||
if not _CAPABILITY_PROOF_RE.search(text) and not session.get(
|
||||
"issue_action_capability_proven"
|
||||
):
|
||||
reasons.append(
|
||||
"recommended linked issue close/reopen/comment requires "
|
||||
"exact capability proof (#300)"
|
||||
)
|
||||
|
||||
proven = not reasons
|
||||
return {
|
||||
"proven": proven,
|
||||
"block": not proven,
|
||||
"reasons": list(dict.fromkeys(reasons)),
|
||||
"linked_issue_number": issue_number,
|
||||
"live_proof_present": _live_proof_present(text, session, issue_number),
|
||||
"safe_next_action": (
|
||||
"fetch linked issue with gitea_view_issue in this session or "
|
||||
"report 'Linked issue status: not verified in this session'"
|
||||
if reasons
|
||||
else "proceed"
|
||||
),
|
||||
}
|
||||
@@ -25,11 +25,15 @@ def _review_handoff(**overrides):
|
||||
"- Files changed: review_proofs.py",
|
||||
"- Validation: pytest tests/test_review_proofs.py -q in branches/review-203",
|
||||
"- Mutations: review only",
|
||||
"- Workspace mutations: none",
|
||||
"- File edits by reviewer: none",
|
||||
"- Worktree/index mutations: none",
|
||||
"- Git ref mutations: git fetch prgs master",
|
||||
"- MCP/Gitea mutations: review submitted",
|
||||
"- Review mutations: submitted request_changes review",
|
||||
"- Merge mutations: none",
|
||||
"- Cleanup mutations: none",
|
||||
"- External-state mutations: none",
|
||||
"- Read-only diagnostics: issue and PR metadata inspected",
|
||||
"- Current status: PR open",
|
||||
"- Blockers: none",
|
||||
"- Next: await author",
|
||||
@@ -204,11 +208,62 @@ class TestReviewPrRules(unittest.TestCase):
|
||||
)
|
||||
)
|
||||
|
||||
def test_already_landed_gate_blocks_review_terminal_states(self):
|
||||
cases = {
|
||||
"APPROVED": ("- Review decision: request_changes", "- Review decision: APPROVED"),
|
||||
"MERGED": ("- Merge result: not attempted", "- Merge result: MERGED"),
|
||||
"READY_TO_MERGE": ("- Current status: PR open", "- Current status: READY_TO_MERGE"),
|
||||
}
|
||||
for state, (old, new) in cases.items():
|
||||
with self.subTest(state=state):
|
||||
report = (
|
||||
_review_handoff()
|
||||
.replace("- Reviewer eligibility: eligible", "- Reviewer eligibility: blocked")
|
||||
.replace(old, new)
|
||||
+ "\n- Already-landed gate: fired"
|
||||
+ "\n- Ancestor proof: passed"
|
||||
+ "\n- Eligibility class: ALREADY_LANDED_RECONCILE_REQUIRED"
|
||||
)
|
||||
result = assess_final_report_validator(report, "review_pr")
|
||||
self.assertTrue(result["blocked"])
|
||||
self.assertTrue(
|
||||
any(
|
||||
f["rule_id"] == "reviewer.already_landed_review_state"
|
||||
for f in result["findings"]
|
||||
)
|
||||
)
|
||||
|
||||
def test_target_branch_up_to_date_requires_fetch_and_sha(self):
|
||||
report = (
|
||||
_review_handoff()
|
||||
.replace("- Git ref mutations: git fetch prgs master", "- Git ref mutations: none")
|
||||
+ "\nTarget branch up to date."
|
||||
)
|
||||
result = assess_final_report_validator(report, "review_pr")
|
||||
self.assertTrue(result["blocked"])
|
||||
self.assertTrue(
|
||||
any(
|
||||
f["rule_id"] == "reviewer.target_branch_freshness_proof"
|
||||
for f in result["findings"]
|
||||
)
|
||||
)
|
||||
|
||||
def test_target_branch_up_to_date_with_fetch_and_sha_passes(self):
|
||||
report = (
|
||||
_review_handoff()
|
||||
+ "\n- Target branch SHA: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9"
|
||||
+ "\nTarget branch up to date."
|
||||
)
|
||||
result = assess_final_report_validator(report, "review_pr")
|
||||
self.assertEqual(result["grade"], "A")
|
||||
def test_git_fetch_must_not_be_readonly_only(self):
|
||||
report = (
|
||||
_review_handoff()
|
||||
.replace("- Git ref mutations: git fetch prgs master", "- Git ref mutations: none")
|
||||
+ "\n- Read-only diagnostics: git fetch prgs master"
|
||||
.replace(
|
||||
"- Read-only diagnostics: issue and PR metadata inspected",
|
||||
"- Read-only diagnostics: git fetch prgs master",
|
||||
)
|
||||
)
|
||||
result = assess_final_report_validator(
|
||||
report,
|
||||
@@ -284,4 +339,4 @@ class TestEntryPoint(unittest.TestCase):
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
unittest.main()
|
||||
|
||||
@@ -201,6 +201,12 @@ def test_validation_worktree_edit_verifier_exported():
|
||||
assert callable(assess_validation_worktree_edit_report)
|
||||
|
||||
|
||||
def test_reconcile_linked_issue_verifier_exported():
|
||||
from review_proofs import assess_reconcile_linked_issue_report
|
||||
|
||||
assert callable(assess_reconcile_linked_issue_report)
|
||||
|
||||
|
||||
def test_already_landed_handoff_verifier_exported():
|
||||
from review_proofs import assess_already_landed_handoff_report
|
||||
|
||||
|
||||
@@ -0,0 +1,117 @@
|
||||
"""Tests for reconciliation linked-issue live proof verifier (#300)."""
|
||||
import sys
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
from reviewer_reconcile_linked_issue import assess_reconcile_linked_issue_report # noqa: E402
|
||||
|
||||
|
||||
def _good_handoff() -> str:
|
||||
return "\n".join([
|
||||
"## PR reconciliation summary",
|
||||
"Fetched linked issue #263 live with gitea_view_issue in this session.",
|
||||
"",
|
||||
"## Controller Handoff",
|
||||
"- Selected PR: #278",
|
||||
"- Eligibility class: ALREADY_LANDED_RECONCILE_REQUIRED",
|
||||
"- Linked issue: #263",
|
||||
"- Linked issue live status: Issue #263 (open)",
|
||||
"- Safe next action: reconcile open PR",
|
||||
])
|
||||
|
||||
|
||||
class TestReconcileLinkedIssueLiveProof(unittest.TestCase):
|
||||
def test_live_fetch_with_open_status_passes(self):
|
||||
result = assess_reconcile_linked_issue_report(
|
||||
_good_handoff(),
|
||||
reconcile_session={"live_fetched": True, "linked_issue_number": 263},
|
||||
)
|
||||
self.assertTrue(result["proven"], result["reasons"])
|
||||
self.assertEqual(result["linked_issue_number"], 263)
|
||||
|
||||
def test_no_linked_issue_passes(self):
|
||||
report = "\n".join([
|
||||
"## Controller Handoff",
|
||||
"- Selected PR: #99",
|
||||
"- Linked issue: none",
|
||||
"- Linked issue live status: not applicable",
|
||||
])
|
||||
result = assess_reconcile_linked_issue_report(report)
|
||||
self.assertTrue(result["proven"])
|
||||
self.assertIsNone(result["linked_issue_number"])
|
||||
|
||||
def test_closed_status_without_live_proof_blocks(self):
|
||||
report = "\n".join([
|
||||
"## Controller Handoff",
|
||||
"- Linked issue: #263",
|
||||
"- Linked issue live status: Issue #263 (closed)",
|
||||
])
|
||||
result = assess_reconcile_linked_issue_report(report)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertTrue(result["block"])
|
||||
self.assertTrue(any("live" in r.lower() for r in result["reasons"]))
|
||||
|
||||
def test_not_verified_wording_passes_without_live_proof(self):
|
||||
report = "\n".join([
|
||||
"## Controller Handoff",
|
||||
"- Linked issue: #263",
|
||||
"- Linked issue live status: not verified in this session",
|
||||
])
|
||||
result = assess_reconcile_linked_issue_report(report)
|
||||
self.assertTrue(result["proven"], result["reasons"])
|
||||
|
||||
def test_missing_issue_requires_not_verified(self):
|
||||
report = "\n".join([
|
||||
"## Controller Handoff",
|
||||
"- Linked issue: #999",
|
||||
"- Linked issue live status: Issue #999 (open)",
|
||||
])
|
||||
result = assess_reconcile_linked_issue_report(
|
||||
report,
|
||||
reconcile_session={"issue_missing": True, "linked_issue_number": 999},
|
||||
)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertTrue(any("not verified" in r.lower() for r in result["reasons"]))
|
||||
|
||||
def test_text_live_proof_without_session_lock_passes(self):
|
||||
report = _good_handoff()
|
||||
result = assess_reconcile_linked_issue_report(report)
|
||||
self.assertTrue(result["proven"], result["reasons"])
|
||||
self.assertTrue(result["live_proof_present"])
|
||||
|
||||
def test_issue_action_recommendation_requires_capability_proof(self):
|
||||
report = "\n".join([
|
||||
_good_handoff(),
|
||||
"Recommended: close linked issue #263 after PR reconciliation.",
|
||||
])
|
||||
result = assess_reconcile_linked_issue_report(
|
||||
report,
|
||||
reconcile_session={"live_fetched": True},
|
||||
)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertTrue(any("capability" in r.lower() for r in result["reasons"]))
|
||||
|
||||
def test_issue_action_with_capability_proof_passes(self):
|
||||
report = "\n".join([
|
||||
_good_handoff(),
|
||||
"Recommended: close linked issue #263.",
|
||||
"Capability proof: gitea_close_issue allowed for prgs-author.",
|
||||
])
|
||||
result = assess_reconcile_linked_issue_report(
|
||||
report,
|
||||
reconcile_session={"live_fetched": True},
|
||||
)
|
||||
self.assertTrue(result["proven"], result["reasons"])
|
||||
|
||||
|
||||
class TestExport(unittest.TestCase):
|
||||
def test_review_proofs_reexport(self):
|
||||
from review_proofs import assess_reconcile_linked_issue_report as exported
|
||||
|
||||
self.assertTrue(callable(exported))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user