Merge pull request 'fix(mcp): derive checks requirement from live branch protection (Closes #751)' (#752) from fix/issue-751-checks-assessor into master
This commit was merged in pull request #752.
This commit is contained in:
+184
-52
@@ -15367,46 +15367,151 @@ def _count_commits_behind(
|
||||
return None
|
||||
|
||||
|
||||
def _matching_protection_rule(protections: Any, branch: str) -> dict | None:
|
||||
"""Select the branch-protection rule governing ``branch`` (#751)."""
|
||||
if not isinstance(protections, list):
|
||||
return None
|
||||
for rule in protections:
|
||||
if not isinstance(rule, dict):
|
||||
continue
|
||||
name = (rule.get("branch_name") or rule.get("rule_name") or "").strip()
|
||||
# Exact match or glob-ish contains for common patterns.
|
||||
if name == branch or name in ("*", f"{branch}"):
|
||||
return rule
|
||||
# Fallback: any protection that mentions the base branch.
|
||||
for rule in protections:
|
||||
if not isinstance(rule, dict):
|
||||
continue
|
||||
name = (rule.get("branch_name") or rule.get("rule_name") or "").strip()
|
||||
if name and (branch in name or name.endswith(branch)):
|
||||
return rule
|
||||
return None
|
||||
|
||||
|
||||
def _branch_protection_policy(
|
||||
base_url: str,
|
||||
auth: dict,
|
||||
*,
|
||||
base_branch: str,
|
||||
) -> dict:
|
||||
"""Read the live branch-protection policy for ``base_branch`` (#751).
|
||||
|
||||
Exposes both the current-base rule (``block_on_outdated_branch``) and the
|
||||
status-check requirement (``enable_status_check`` / ``status_check_contexts``)
|
||||
from the same payload, so the checks assessor can tell "no checks required"
|
||||
apart from "required checks pending".
|
||||
|
||||
``determinable`` is False only when the policy genuinely could not be read
|
||||
(missing branch or API failure) — never merely because no rule exists. A
|
||||
successful read that finds no protection for the branch is authoritative
|
||||
evidence that status checks are not required.
|
||||
"""
|
||||
policy: dict[str, Any] = {
|
||||
"determinable": False,
|
||||
"protection_found": False,
|
||||
"requires_current_base": None,
|
||||
"checks_enabled": None,
|
||||
"required_contexts": [],
|
||||
"base_branch": (base_branch or "").strip() or None,
|
||||
}
|
||||
branch = (base_branch or "").strip()
|
||||
if not branch:
|
||||
return policy
|
||||
|
||||
def _apply(rule: dict) -> None:
|
||||
policy["protection_found"] = True
|
||||
if "block_on_outdated_branch" in rule:
|
||||
policy["requires_current_base"] = bool(
|
||||
rule.get("block_on_outdated_branch")
|
||||
)
|
||||
if "enable_status_check" in rule:
|
||||
policy["checks_enabled"] = bool(rule.get("enable_status_check"))
|
||||
contexts = rule.get("status_check_contexts")
|
||||
if isinstance(contexts, list):
|
||||
policy["required_contexts"] = [
|
||||
str(ctx).strip() for ctx in contexts if str(ctx or "").strip()
|
||||
]
|
||||
|
||||
try:
|
||||
protections = api_request(
|
||||
"GET", f"{base_url}/branch_protections", auth
|
||||
)
|
||||
rule = _matching_protection_rule(protections, branch)
|
||||
if rule is not None:
|
||||
_apply(rule)
|
||||
if not policy["protection_found"]:
|
||||
# Branch payload may embed effective protection.
|
||||
br = api_request("GET", f"{base_url}/branches/{branch}", auth) or {}
|
||||
prot = br.get("protection") or br.get("effective_branch_protection") or {}
|
||||
if isinstance(prot, dict) and prot:
|
||||
_apply(prot)
|
||||
policy["determinable"] = True
|
||||
except Exception:
|
||||
# Genuine read failure — leave determinable False so callers fail closed.
|
||||
return policy
|
||||
|
||||
if not policy["protection_found"]:
|
||||
# Authoritative absence: no protection governs the branch, so no status
|
||||
# check is required by policy.
|
||||
policy["checks_enabled"] = False
|
||||
elif policy["checks_enabled"] is None:
|
||||
# Protection exists but omits the status-check field entirely: Gitea
|
||||
# only enforces contexts when the flag is set, so absence means off.
|
||||
policy["checks_enabled"] = False
|
||||
|
||||
return policy
|
||||
|
||||
|
||||
def _branch_protection_requires_current_base(
|
||||
base_url: str,
|
||||
auth: dict,
|
||||
*,
|
||||
base_branch: str,
|
||||
) -> bool | None:
|
||||
"""Read Gitea branch protection ``block_on_outdated_branch`` when present."""
|
||||
branch = (base_branch or "").strip()
|
||||
if not branch:
|
||||
return None
|
||||
"""Read Gitea branch protection ``block_on_outdated_branch`` when present.
|
||||
|
||||
Thin accessor over :func:`_branch_protection_policy`; return semantics are
|
||||
unchanged (``None`` when the rule is absent or unreadable).
|
||||
"""
|
||||
policy = _branch_protection_policy(base_url, auth, base_branch=base_branch)
|
||||
return policy.get("requires_current_base")
|
||||
|
||||
|
||||
def _commit_checks_snapshot(
|
||||
base_url: str,
|
||||
auth: dict,
|
||||
*,
|
||||
sha: str,
|
||||
) -> dict:
|
||||
"""Read the combined commit status *and its context collection* (#751).
|
||||
|
||||
The combined ``state`` alone is not decisive: Gitea reports ``pending`` for
|
||||
a commit with an empty status-context collection, which is indistinguishable
|
||||
from executing CI unless the collection itself is inspected.
|
||||
"""
|
||||
snapshot: dict[str, Any] = {
|
||||
"determinable": False,
|
||||
"combined_state": None,
|
||||
"statuses": [],
|
||||
}
|
||||
head = (sha or "").strip()
|
||||
if not head:
|
||||
return snapshot
|
||||
try:
|
||||
# Prefer the named protection rule matching the base branch.
|
||||
protections = api_request(
|
||||
"GET", f"{base_url}/branch_protections", auth
|
||||
) or []
|
||||
if isinstance(protections, list):
|
||||
for rule in protections:
|
||||
if not isinstance(rule, dict):
|
||||
continue
|
||||
name = (rule.get("branch_name") or rule.get("rule_name") or "").strip()
|
||||
# Exact match or glob-ish contains for common patterns.
|
||||
if name == branch or name in ("*", f"{branch}"):
|
||||
if "block_on_outdated_branch" in rule:
|
||||
return bool(rule.get("block_on_outdated_branch"))
|
||||
# Fallback: any protection that mentions the base branch.
|
||||
for rule in protections:
|
||||
if not isinstance(rule, dict):
|
||||
continue
|
||||
name = (rule.get("branch_name") or rule.get("rule_name") or "").strip()
|
||||
if branch in name or name.endswith(branch):
|
||||
if "block_on_outdated_branch" in rule:
|
||||
return bool(rule.get("block_on_outdated_branch"))
|
||||
# Branch payload may embed effective protection.
|
||||
br = api_request("GET", f"{base_url}/branches/{branch}", auth) or {}
|
||||
prot = br.get("protection") or br.get("effective_branch_protection") or {}
|
||||
if isinstance(prot, dict) and "block_on_outdated_branch" in prot:
|
||||
return bool(prot.get("block_on_outdated_branch"))
|
||||
payload = api_request(
|
||||
"GET", f"{base_url}/commits/{head}/status", auth
|
||||
)
|
||||
except Exception:
|
||||
return None
|
||||
return None
|
||||
return snapshot
|
||||
if payload is None:
|
||||
return snapshot
|
||||
snapshot["determinable"] = True
|
||||
if not isinstance(payload, dict):
|
||||
return snapshot
|
||||
snapshot["combined_state"] = (payload.get("state") or "").strip().lower() or None
|
||||
statuses = payload.get("statuses")
|
||||
snapshot["statuses"] = statuses if isinstance(statuses, list) else []
|
||||
return snapshot
|
||||
|
||||
|
||||
def _prove_author_ownership_for_pr(
|
||||
@@ -15651,11 +15756,14 @@ def gitea_assess_pr_sync_status(
|
||||
# Fail closed on unknown behind-count only when SHAs differ.
|
||||
commits_behind = 0 if pr_head_sha == base_head_sha else None
|
||||
|
||||
# Single live read of the base-branch protection policy; it carries both
|
||||
# the current-base rule and the status-check requirement (#751).
|
||||
protection_policy = _branch_protection_policy(
|
||||
base, auth, base_branch=base_branch
|
||||
)
|
||||
if branch_protection_requires_current_base is None:
|
||||
branch_protection_requires_current_base = (
|
||||
_branch_protection_requires_current_base(
|
||||
base, auth, base_branch=base_branch
|
||||
)
|
||||
branch_protection_requires_current_base = protection_policy.get(
|
||||
"requires_current_base"
|
||||
)
|
||||
# When protection cannot be read, fail closed by requiring current base
|
||||
# whenever the PR is behind (safer for protected repos). Callers may pass
|
||||
@@ -15722,23 +15830,30 @@ def gitea_assess_pr_sync_status(
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# ── Live checks derivation (#751) ────────────────────────────────────
|
||||
# Classify from the actual status-context collection plus the live
|
||||
# protection policy. The combined ``state`` is never treated as proof that
|
||||
# CI is executing, because Gitea reports ``pending`` for an empty
|
||||
# collection. ``checks_required`` is always derived from live evidence and
|
||||
# is deliberately not a caller-supplied input, so no session can declare
|
||||
# checks optional without proof.
|
||||
checks_snapshot = _commit_checks_snapshot(base, auth, sha=pr_head_sha or "")
|
||||
checks_classification = pr_sync_status.classify_commit_checks(
|
||||
combined_state=checks_snapshot.get("combined_state"),
|
||||
statuses=checks_snapshot.get("statuses"),
|
||||
checks_enabled=protection_policy.get("checks_enabled"),
|
||||
required_contexts=protection_policy.get("required_contexts"),
|
||||
policy_determinable=bool(protection_policy.get("determinable")),
|
||||
status_determinable=bool(checks_snapshot.get("determinable")),
|
||||
)
|
||||
checks_required = bool(checks_classification.get("checks_required", True))
|
||||
caller_supplied_checks_status = checks_status
|
||||
if checks_status is None:
|
||||
checks_status = "unknown"
|
||||
# Combined status on PR head when Actions/status API is available.
|
||||
if pr_head_sha:
|
||||
try:
|
||||
st = api_request(
|
||||
"GET",
|
||||
f"{base}/commits/{pr_head_sha}/status",
|
||||
auth,
|
||||
) or {}
|
||||
state = (st.get("state") or "").strip().lower()
|
||||
if state:
|
||||
checks_status = state
|
||||
elif not st:
|
||||
checks_status = "none"
|
||||
except Exception:
|
||||
checks_status = "unknown"
|
||||
checks_status = checks_classification.get("checks_status")
|
||||
elif checks_classification.get("checks_status") != pr_sync_status.CHECKS_UNKNOWN:
|
||||
# Live evidence outranks a caller-supplied value; the override only
|
||||
# applies when live status could not be classified at all.
|
||||
checks_status = checks_classification.get("checks_status")
|
||||
|
||||
assessment = pr_sync_status.assess_pr_sync_status(
|
||||
host=h,
|
||||
@@ -15759,9 +15874,26 @@ def gitea_assess_pr_sync_status(
|
||||
active_reviewer_lease=active_reviewer_lease,
|
||||
active_merger_lease=active_merger_lease,
|
||||
prepared_verdict_head_sha=prepared_verdict_head_sha,
|
||||
checks_required=checks_required,
|
||||
)
|
||||
assessment["remote"] = remote if remote in REMOTES else None
|
||||
assessment["base_branch"] = base_branch
|
||||
# Evidence for the checks decision (#751) — no secrets, read-only.
|
||||
assessment["checks_evidence"] = {
|
||||
"combined_state": checks_classification.get("combined_state"),
|
||||
"context_count": checks_classification.get("context_count"),
|
||||
"observed_contexts": checks_classification.get("observed_contexts"),
|
||||
"required_contexts": checks_classification.get("required_contexts"),
|
||||
"missing_required_contexts": checks_classification.get(
|
||||
"missing_required_contexts"
|
||||
),
|
||||
"policy_determinable": checks_classification.get("policy_determinable"),
|
||||
"status_determinable": checks_classification.get("status_determinable"),
|
||||
"protection_found": protection_policy.get("protection_found"),
|
||||
"checks_enabled": protection_policy.get("checks_enabled"),
|
||||
"caller_supplied_checks_status": caller_supplied_checks_status,
|
||||
"reasons": checks_classification.get("reasons"),
|
||||
}
|
||||
assessment["success"] = True
|
||||
assessment["performed"] = False
|
||||
return assessment
|
||||
|
||||
+213
-12
@@ -41,6 +41,186 @@ _DENIED_UPDATE_ROLES = frozenset({"reviewer", "merger", "reconciler", "mixed", "
|
||||
UPDATE_STYLE_MERGE = "merge"
|
||||
_FORBIDDEN_UPDATE_STYLES = frozenset({"rebase", "rebase-merge", "squash", "force"})
|
||||
|
||||
# ── Commit check classifications (#751) ──────────────────────────────────
|
||||
# Gitea's *combined* commit status reports ``state: pending`` both when a real
|
||||
# check is executing and when the status-context collection is empty. Reading
|
||||
# ``state`` alone therefore cannot distinguish "CI is running" from "no CI
|
||||
# exists", which permanently blocks a merge-ready PR that no check will ever
|
||||
# report on. These classifications are derived from the actual context
|
||||
# collection plus the live branch-protection policy.
|
||||
CHECKS_SUCCESS = "success"
|
||||
CHECKS_FAILURE = "failure"
|
||||
CHECKS_PENDING = "pending"
|
||||
CHECKS_NONE = "none" # configured/produced nothing
|
||||
CHECKS_NOT_REQUIRED = "not_required" # protection does not require checks
|
||||
CHECKS_MISSING_REQUIRED = "missing_required" # required contexts have no result
|
||||
CHECKS_UNKNOWN = "unknown" # indeterminable — fail closed
|
||||
|
||||
# Values that permit merge_now when checks are required.
|
||||
_CHECKS_OK = frozenset({"success", "passed", "ok", "skipped", "not_required"})
|
||||
|
||||
# Raw per-context state vocabularies reported by Gitea.
|
||||
_CTX_SUCCESS = frozenset({"success", "passed", "ok"})
|
||||
_CTX_FAILURE = frozenset({"failure", "failed", "error", "cancelled", "canceled"})
|
||||
_CTX_PENDING = frozenset({"pending", "running", "queued", "expected"})
|
||||
_CTX_SKIPPED = frozenset({"skipped", "neutral"})
|
||||
|
||||
|
||||
def _normalize_context_rows(statuses: Any) -> list[dict[str, str]]:
|
||||
"""Reduce a raw status collection to newest-wins ``{context, state}`` rows.
|
||||
|
||||
Gitea returns the status collection newest-first, so the first row seen for
|
||||
a context wins. Rows without a usable state are discarded rather than being
|
||||
silently treated as passing.
|
||||
"""
|
||||
rows: list[dict[str, str]] = []
|
||||
seen: set[str] = set()
|
||||
if not isinstance(statuses, list):
|
||||
return rows
|
||||
for raw in statuses:
|
||||
if not isinstance(raw, dict):
|
||||
continue
|
||||
context = (raw.get("context") or raw.get("name") or "").strip()
|
||||
state = (raw.get("status") or raw.get("state") or "").strip().lower()
|
||||
if not state:
|
||||
continue
|
||||
key = context or f"__unnamed__{len(rows)}"
|
||||
if key in seen:
|
||||
continue
|
||||
seen.add(key)
|
||||
rows.append({"context": context, "state": state})
|
||||
return rows
|
||||
|
||||
|
||||
def _aggregate_context_states(rows: list[dict[str, str]]) -> str:
|
||||
"""Fail-closed aggregate: failure > pending > unknown-state > success."""
|
||||
states = {row["state"] for row in rows}
|
||||
if states & _CTX_FAILURE:
|
||||
return CHECKS_FAILURE
|
||||
if states & _CTX_PENDING:
|
||||
return CHECKS_PENDING
|
||||
unresolved = states - _CTX_SUCCESS - _CTX_SKIPPED
|
||||
if unresolved:
|
||||
# An unrecognized context state must never read as success.
|
||||
return CHECKS_UNKNOWN
|
||||
return CHECKS_SUCCESS
|
||||
|
||||
|
||||
def classify_commit_checks(
|
||||
*,
|
||||
combined_state: str | None = None,
|
||||
statuses: Any = None,
|
||||
checks_enabled: bool | None = None,
|
||||
required_contexts: Any = None,
|
||||
policy_determinable: bool = True,
|
||||
status_determinable: bool = True,
|
||||
) -> dict[str, Any]:
|
||||
"""Classify head checks from live evidence (#751).
|
||||
|
||||
``combined_state`` is deliberately **not** authoritative: it is recorded for
|
||||
observability but never used to infer that CI is executing. The context
|
||||
collection and the live protection policy decide.
|
||||
|
||||
Returns ``checks_status`` (one of the ``CHECKS_*`` values), the derived
|
||||
``checks_required`` flag, and structured ``reasons``.
|
||||
"""
|
||||
reasons: list[str] = []
|
||||
rows = _normalize_context_rows(statuses)
|
||||
required = [
|
||||
str(ctx).strip()
|
||||
for ctx in (required_contexts or [])
|
||||
if str(ctx or "").strip()
|
||||
]
|
||||
observed_combined = (combined_state or "").strip().lower() or None
|
||||
|
||||
result: dict[str, Any] = {
|
||||
"checks_status": CHECKS_UNKNOWN,
|
||||
"checks_required": True,
|
||||
"combined_state": observed_combined,
|
||||
"context_count": len(rows),
|
||||
"observed_contexts": [row["context"] for row in rows],
|
||||
"required_contexts": required,
|
||||
"missing_required_contexts": [],
|
||||
"policy_determinable": bool(policy_determinable),
|
||||
"status_determinable": bool(status_determinable),
|
||||
"reasons": reasons,
|
||||
}
|
||||
|
||||
# Policy unreadable → never assume checks are optional.
|
||||
if not policy_determinable:
|
||||
reasons.append(
|
||||
"branch-protection check policy could not be read; cannot prove "
|
||||
"whether status checks are required (fail closed)"
|
||||
)
|
||||
return result
|
||||
|
||||
if checks_enabled is False:
|
||||
result["checks_required"] = False
|
||||
result["checks_status"] = CHECKS_NOT_REQUIRED
|
||||
reasons.append(
|
||||
"live branch protection does not require status checks for the base "
|
||||
"branch; head status contexts do not gate merge"
|
||||
)
|
||||
return result
|
||||
|
||||
if checks_enabled is None:
|
||||
reasons.append(
|
||||
"branch-protection status-check requirement is indeterminate "
|
||||
"(fail closed)"
|
||||
)
|
||||
return result
|
||||
|
||||
# Checks are required from here on.
|
||||
if not status_determinable:
|
||||
reasons.append(
|
||||
"head commit status collection could not be read while branch "
|
||||
"protection requires status checks (fail closed)"
|
||||
)
|
||||
return result
|
||||
|
||||
if required:
|
||||
by_context = {row["context"]: row["state"] for row in rows if row["context"]}
|
||||
missing = [ctx for ctx in required if ctx not in by_context]
|
||||
if missing:
|
||||
result["missing_required_contexts"] = missing
|
||||
result["checks_status"] = CHECKS_MISSING_REQUIRED
|
||||
reasons.append(
|
||||
"branch protection requires status context(s) "
|
||||
f"{', '.join(missing)} but no matching status result exists at "
|
||||
"the head commit (fail closed)"
|
||||
)
|
||||
return result
|
||||
matched = [
|
||||
{"context": ctx, "state": by_context[ctx]} for ctx in required
|
||||
]
|
||||
result["checks_status"] = _aggregate_context_states(matched)
|
||||
reasons.append(
|
||||
f"evaluated {len(matched)} required status context(s) from live "
|
||||
"branch protection; unrelated contexts were ignored"
|
||||
)
|
||||
return result
|
||||
|
||||
# Status checks enabled with no specific required contexts configured.
|
||||
if not rows:
|
||||
result["checks_status"] = CHECKS_NONE
|
||||
reasons.append(
|
||||
"branch protection enables status checks but no status context was "
|
||||
"produced for the head commit"
|
||||
)
|
||||
if observed_combined in _CTX_PENDING:
|
||||
reasons.append(
|
||||
f"combined commit state '{observed_combined}' does not indicate "
|
||||
"executing CI because the status-context collection is empty"
|
||||
)
|
||||
return result
|
||||
|
||||
result["checks_status"] = _aggregate_context_states(rows)
|
||||
reasons.append(
|
||||
f"aggregated {len(rows)} reported status context(s); branch protection "
|
||||
"configures no explicit required-context list"
|
||||
)
|
||||
return result
|
||||
|
||||
|
||||
def _normalize_sha(value: str | None) -> str | None:
|
||||
text = (value or "").strip().lower()
|
||||
@@ -120,6 +300,7 @@ def assess_pr_sync_status(
|
||||
"branch_protection_requires_current_base": requires_current,
|
||||
"approval_at_current_head": approval_ok if approval_at_current_head is not None else None,
|
||||
"checks_status": checks,
|
||||
"checks_required": bool(checks_required),
|
||||
"active_locks_and_leases": {
|
||||
"author_lock": bool(active_author_lock) if active_author_lock is not None else None,
|
||||
"reviewer_lease": bool(active_reviewer_lease) if active_reviewer_lease is not None else None,
|
||||
@@ -258,21 +439,41 @@ def assess_pr_sync_status(
|
||||
result["recommended_next_action"] = ACTION_BLOCKED
|
||||
return result
|
||||
|
||||
# ── Checks gate for merge_now ────────────────────────────────────────
|
||||
if checks_required and checks not in ("success", "passed", "ok", "none", "skipped", "not_required"):
|
||||
if checks in ("pending", "running", "queued"):
|
||||
# ── Checks gate for merge_now (#751) ─────────────────────────────────
|
||||
# ``checks_required`` is derived from the live branch-protection policy by
|
||||
# the production caller. When protection does not require status checks,
|
||||
# head contexts cannot gate the merge and this whole gate is skipped.
|
||||
if not checks_required:
|
||||
reasons.append(
|
||||
"live branch protection does not require status checks; head check "
|
||||
f"state ({checks}) does not gate merge"
|
||||
)
|
||||
elif checks not in _CHECKS_OK:
|
||||
if checks in _CTX_PENDING:
|
||||
reasons.append(f"required checks are not finished (status={checks})")
|
||||
result["recommended_next_action"] = ACTION_BLOCKED
|
||||
return result
|
||||
if checks in ("failure", "failed", "error", "cancelled"):
|
||||
elif checks in _CTX_FAILURE:
|
||||
reasons.append(f"required checks failed (status={checks})")
|
||||
result["recommended_next_action"] = ACTION_BLOCKED
|
||||
return result
|
||||
# unknown — fail closed when checks_required
|
||||
if checks == "unknown":
|
||||
elif checks == CHECKS_MISSING_REQUIRED:
|
||||
reasons.append(
|
||||
"branch protection configures required status context(s) but no "
|
||||
"matching status result exists at the current head (fail closed)"
|
||||
)
|
||||
elif checks == CHECKS_NONE:
|
||||
reasons.append(
|
||||
"branch protection requires status checks but no status context "
|
||||
"was produced for the current head (fail closed); an empty "
|
||||
"status collection is not executing CI"
|
||||
)
|
||||
elif checks == CHECKS_UNKNOWN:
|
||||
reasons.append("checks status unknown (fail closed)")
|
||||
result["recommended_next_action"] = ACTION_BLOCKED
|
||||
return result
|
||||
else:
|
||||
# Unrecognized vocabulary must never fall through to merge_now.
|
||||
reasons.append(
|
||||
f"unrecognized checks status '{checks}' cannot prove required "
|
||||
"checks passed (fail closed)"
|
||||
)
|
||||
result["recommended_next_action"] = ACTION_BLOCKED
|
||||
return result
|
||||
|
||||
# ── Ready to merge without update ────────────────────────────────────
|
||||
# Includes: current with approval; outdated when update is NOT required.
|
||||
|
||||
@@ -0,0 +1,602 @@
|
||||
"""Regression coverage for the PR checks assessor defect (#751).
|
||||
|
||||
Gitea's *combined* commit status reports ``state: pending`` both when a check is
|
||||
executing and when the status-context collection is empty. The assessor read
|
||||
``state`` alone and defaulted ``checks_required`` to ``True``, so a PR whose head
|
||||
had no status contexts — and never would — was routed to ``blocked`` forever.
|
||||
|
||||
These tests pin the corrected semantics end to end: live branch protection
|
||||
decides whether checks are required, and the actual context collection decides
|
||||
what the checks say.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import sys
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
|
||||
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
|
||||
|
||||
import pr_sync_status # noqa: E402
|
||||
from pr_sync_status import ( # noqa: E402
|
||||
ACTION_BLOCKED,
|
||||
ACTION_MERGE_NOW,
|
||||
ACTION_UPDATE_BRANCH_BY_MERGE,
|
||||
CHECKS_FAILURE,
|
||||
CHECKS_MISSING_REQUIRED,
|
||||
CHECKS_NONE,
|
||||
CHECKS_NOT_REQUIRED,
|
||||
CHECKS_PENDING,
|
||||
CHECKS_SUCCESS,
|
||||
CHECKS_UNKNOWN,
|
||||
assess_pr_sync_status,
|
||||
classify_commit_checks,
|
||||
)
|
||||
|
||||
|
||||
def _sha(prefix: str) -> str:
|
||||
return (prefix + "0" * 40)[:40]
|
||||
|
||||
|
||||
PR_HEAD = _sha("aaaaaaaa")
|
||||
BASE_HEAD = _sha("bbbbbbbb")
|
||||
|
||||
|
||||
def _base_kwargs(**overrides):
|
||||
data = {
|
||||
"host": "gitea.prgs.cc",
|
||||
"org": "Scaled-Tech-Consulting",
|
||||
"repo": "Gitea-Tools",
|
||||
"pr_number": 751,
|
||||
"pr_state": "open",
|
||||
"source_branch": "fix/issue-751-checks-assessor",
|
||||
"pr_head_sha": PR_HEAD,
|
||||
"base_head_sha": BASE_HEAD,
|
||||
"commits_behind": 0,
|
||||
"mergeable": True,
|
||||
"has_conflicts": False,
|
||||
"branch_protection_requires_current_base": False,
|
||||
"approval_at_current_head": True,
|
||||
"checks_status": "success",
|
||||
"active_author_lock": True,
|
||||
"active_reviewer_lease": False,
|
||||
"active_merger_lease": False,
|
||||
}
|
||||
data.update(overrides)
|
||||
return data
|
||||
|
||||
|
||||
def _reasons(result) -> str:
|
||||
return " | ".join(result["reasons"]).lower()
|
||||
|
||||
|
||||
class TestClassifyCommitChecks(unittest.TestCase):
|
||||
"""Pure classification from live evidence."""
|
||||
|
||||
def test_empty_collection_with_combined_pending_is_not_executing_ci(self):
|
||||
# The exact PR #750 failure mode at the classification layer.
|
||||
result = classify_commit_checks(
|
||||
combined_state="pending",
|
||||
statuses=[],
|
||||
checks_enabled=True,
|
||||
required_contexts=[],
|
||||
)
|
||||
self.assertNotEqual(result["checks_status"], CHECKS_PENDING)
|
||||
self.assertEqual(result["checks_status"], CHECKS_NONE)
|
||||
self.assertEqual(result["context_count"], 0)
|
||||
joined = " ".join(result["reasons"]).lower()
|
||||
self.assertIn("empty", joined)
|
||||
self.assertIn("does not indicate", joined)
|
||||
|
||||
def test_protection_disables_status_checks(self):
|
||||
result = classify_commit_checks(
|
||||
combined_state="pending",
|
||||
statuses=[],
|
||||
checks_enabled=False,
|
||||
required_contexts=[],
|
||||
)
|
||||
self.assertFalse(result["checks_required"])
|
||||
self.assertEqual(result["checks_status"], CHECKS_NOT_REQUIRED)
|
||||
|
||||
def test_no_required_contexts_aggregates_reported_contexts(self):
|
||||
result = classify_commit_checks(
|
||||
combined_state="success",
|
||||
statuses=[{"context": "build", "status": "success"}],
|
||||
checks_enabled=True,
|
||||
required_contexts=[],
|
||||
)
|
||||
self.assertTrue(result["checks_required"])
|
||||
self.assertEqual(result["checks_status"], CHECKS_SUCCESS)
|
||||
|
||||
def test_required_checks_pending(self):
|
||||
result = classify_commit_checks(
|
||||
combined_state="pending",
|
||||
statuses=[{"context": "build", "status": "pending"}],
|
||||
checks_enabled=True,
|
||||
required_contexts=["build"],
|
||||
)
|
||||
self.assertEqual(result["checks_status"], CHECKS_PENDING)
|
||||
|
||||
def test_required_checks_failed(self):
|
||||
result = classify_commit_checks(
|
||||
combined_state="failure",
|
||||
statuses=[{"context": "build", "status": "failure"}],
|
||||
checks_enabled=True,
|
||||
required_contexts=["build"],
|
||||
)
|
||||
self.assertEqual(result["checks_status"], CHECKS_FAILURE)
|
||||
|
||||
def test_required_checks_successful(self):
|
||||
result = classify_commit_checks(
|
||||
combined_state="success",
|
||||
statuses=[{"context": "build", "status": "success"}],
|
||||
checks_enabled=True,
|
||||
required_contexts=["build"],
|
||||
)
|
||||
self.assertEqual(result["checks_status"], CHECKS_SUCCESS)
|
||||
self.assertTrue(result["checks_required"])
|
||||
|
||||
def test_required_context_configured_with_no_matching_result(self):
|
||||
result = classify_commit_checks(
|
||||
combined_state="success",
|
||||
statuses=[{"context": "lint", "status": "success"}],
|
||||
checks_enabled=True,
|
||||
required_contexts=["build"],
|
||||
)
|
||||
self.assertEqual(result["checks_status"], CHECKS_MISSING_REQUIRED)
|
||||
self.assertEqual(result["missing_required_contexts"], ["build"])
|
||||
|
||||
def test_mixed_required_and_unrelated_contexts_ignores_unrelated(self):
|
||||
# An unrelated failing context must not fail a satisfied required set.
|
||||
result = classify_commit_checks(
|
||||
combined_state="failure",
|
||||
statuses=[
|
||||
{"context": "build", "status": "success"},
|
||||
{"context": "optional-scan", "status": "failure"},
|
||||
],
|
||||
checks_enabled=True,
|
||||
required_contexts=["build"],
|
||||
)
|
||||
self.assertEqual(result["checks_status"], CHECKS_SUCCESS)
|
||||
# ...and a failing *required* context still fails despite passing extras.
|
||||
failing = classify_commit_checks(
|
||||
combined_state="success",
|
||||
statuses=[
|
||||
{"context": "build", "status": "failure"},
|
||||
{"context": "optional-scan", "status": "success"},
|
||||
],
|
||||
checks_enabled=True,
|
||||
required_contexts=["build"],
|
||||
)
|
||||
self.assertEqual(failing["checks_status"], CHECKS_FAILURE)
|
||||
|
||||
def test_policy_unreadable_fails_closed(self):
|
||||
result = classify_commit_checks(
|
||||
combined_state=None,
|
||||
statuses=[],
|
||||
checks_enabled=None,
|
||||
required_contexts=[],
|
||||
policy_determinable=False,
|
||||
)
|
||||
self.assertTrue(result["checks_required"])
|
||||
self.assertEqual(result["checks_status"], CHECKS_UNKNOWN)
|
||||
self.assertIn("fail closed", " ".join(result["reasons"]).lower())
|
||||
|
||||
def test_malformed_policy_indeterminate_flag_fails_closed(self):
|
||||
result = classify_commit_checks(
|
||||
combined_state="success",
|
||||
statuses=[{"context": "build", "status": "success"}],
|
||||
checks_enabled=None,
|
||||
required_contexts=[],
|
||||
policy_determinable=True,
|
||||
)
|
||||
self.assertTrue(result["checks_required"])
|
||||
self.assertEqual(result["checks_status"], CHECKS_UNKNOWN)
|
||||
|
||||
def test_status_collection_unreadable_fails_closed_when_required(self):
|
||||
result = classify_commit_checks(
|
||||
checks_enabled=True,
|
||||
required_contexts=["build"],
|
||||
status_determinable=False,
|
||||
)
|
||||
self.assertTrue(result["checks_required"])
|
||||
self.assertEqual(result["checks_status"], CHECKS_UNKNOWN)
|
||||
|
||||
def test_unrecognized_context_state_never_reads_as_success(self):
|
||||
result = classify_commit_checks(
|
||||
combined_state="success",
|
||||
statuses=[{"context": "build", "status": "banana"}],
|
||||
checks_enabled=True,
|
||||
required_contexts=["build"],
|
||||
)
|
||||
self.assertEqual(result["checks_status"], CHECKS_UNKNOWN)
|
||||
|
||||
def test_newest_wins_per_context(self):
|
||||
# Gitea returns newest-first; the stale failure must not win.
|
||||
result = classify_commit_checks(
|
||||
combined_state="success",
|
||||
statuses=[
|
||||
{"context": "build", "status": "success"},
|
||||
{"context": "build", "status": "failure"},
|
||||
],
|
||||
checks_enabled=True,
|
||||
required_contexts=["build"],
|
||||
)
|
||||
self.assertEqual(result["checks_status"], CHECKS_SUCCESS)
|
||||
|
||||
def test_malformed_status_rows_are_discarded_not_treated_as_passing(self):
|
||||
result = classify_commit_checks(
|
||||
combined_state="pending",
|
||||
statuses=[{"context": "build"}, "not-a-dict", None],
|
||||
checks_enabled=True,
|
||||
required_contexts=["build"],
|
||||
)
|
||||
self.assertEqual(result["checks_status"], CHECKS_MISSING_REQUIRED)
|
||||
|
||||
|
||||
class TestChecksGateSemantics(unittest.TestCase):
|
||||
"""``assess_pr_sync_status`` routing and blocker reasons."""
|
||||
|
||||
def test_not_required_allows_merge_now_with_empty_checks(self):
|
||||
result = assess_pr_sync_status(
|
||||
**_base_kwargs(checks_status=CHECKS_NOT_REQUIRED),
|
||||
checks_required=False,
|
||||
)
|
||||
self.assertEqual(result["recommended_next_action"], ACTION_MERGE_NOW)
|
||||
self.assertTrue(result["approval_valid_for_merge"])
|
||||
self.assertFalse(result["checks_required"])
|
||||
self.assertIn("does not require status checks", _reasons(result))
|
||||
|
||||
def test_none_blocks_when_checks_required(self):
|
||||
result = assess_pr_sync_status(
|
||||
**_base_kwargs(checks_status=CHECKS_NONE),
|
||||
checks_required=True,
|
||||
)
|
||||
self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED)
|
||||
self.assertIn("no status context", _reasons(result))
|
||||
self.assertIn("not executing ci", _reasons(result))
|
||||
|
||||
def test_missing_required_blocks(self):
|
||||
result = assess_pr_sync_status(
|
||||
**_base_kwargs(checks_status=CHECKS_MISSING_REQUIRED),
|
||||
checks_required=True,
|
||||
)
|
||||
self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED)
|
||||
self.assertIn("no matching status result", _reasons(result))
|
||||
|
||||
def test_required_pending_blocks(self):
|
||||
result = assess_pr_sync_status(
|
||||
**_base_kwargs(checks_status=CHECKS_PENDING), checks_required=True
|
||||
)
|
||||
self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED)
|
||||
self.assertIn("not finished", _reasons(result))
|
||||
|
||||
def test_required_failure_blocks(self):
|
||||
result = assess_pr_sync_status(
|
||||
**_base_kwargs(checks_status=CHECKS_FAILURE), checks_required=True
|
||||
)
|
||||
self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED)
|
||||
self.assertIn("failed", _reasons(result))
|
||||
|
||||
def test_required_success_merges(self):
|
||||
result = assess_pr_sync_status(
|
||||
**_base_kwargs(checks_status=CHECKS_SUCCESS), checks_required=True
|
||||
)
|
||||
self.assertEqual(result["recommended_next_action"], ACTION_MERGE_NOW)
|
||||
|
||||
def test_unknown_blocks_when_required(self):
|
||||
result = assess_pr_sync_status(
|
||||
**_base_kwargs(checks_status=CHECKS_UNKNOWN), checks_required=True
|
||||
)
|
||||
self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED)
|
||||
self.assertIn("unknown", _reasons(result))
|
||||
|
||||
def test_unrecognized_status_does_not_fall_through_to_merge(self):
|
||||
# Regression: the previous gate only handled a fixed vocabulary and let
|
||||
# anything else reach merge_now.
|
||||
result = assess_pr_sync_status(
|
||||
**_base_kwargs(checks_status="totally-unexpected"),
|
||||
checks_required=True,
|
||||
)
|
||||
self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED)
|
||||
self.assertIn("unrecognized checks status", _reasons(result))
|
||||
|
||||
def test_checks_gate_does_not_bypass_approval_requirement(self):
|
||||
result = assess_pr_sync_status(
|
||||
**_base_kwargs(
|
||||
checks_status=CHECKS_NOT_REQUIRED, approval_at_current_head=False
|
||||
),
|
||||
checks_required=False,
|
||||
)
|
||||
self.assertNotEqual(result["recommended_next_action"], ACTION_MERGE_NOW)
|
||||
self.assertFalse(result["approval_valid_for_merge"])
|
||||
|
||||
def test_checks_gate_does_not_bypass_current_base_protection(self):
|
||||
# Existing current-base behavior stays intact (#727).
|
||||
result = assess_pr_sync_status(
|
||||
**_base_kwargs(
|
||||
checks_status=CHECKS_NOT_REQUIRED,
|
||||
commits_behind=3,
|
||||
branch_protection_requires_current_base=True,
|
||||
),
|
||||
checks_required=False,
|
||||
)
|
||||
self.assertEqual(
|
||||
result["recommended_next_action"], ACTION_UPDATE_BRANCH_BY_MERGE
|
||||
)
|
||||
|
||||
def test_checks_gate_does_not_bypass_conflicts(self):
|
||||
result = assess_pr_sync_status(
|
||||
**_base_kwargs(checks_status=CHECKS_NOT_REQUIRED, mergeable=False),
|
||||
checks_required=False,
|
||||
)
|
||||
self.assertNotEqual(result["recommended_next_action"], ACTION_MERGE_NOW)
|
||||
|
||||
|
||||
class TestBranchProtectionPolicy(unittest.TestCase):
|
||||
"""Live protection reader derives the status-check requirement."""
|
||||
|
||||
def setUp(self):
|
||||
import gitea_mcp_server as gms
|
||||
|
||||
self.gms = gms
|
||||
|
||||
def _policy(self, responses):
|
||||
def fake_api_request(method, url, auth, *args, **kwargs):
|
||||
for fragment, payload in responses.items():
|
||||
if fragment in url:
|
||||
if isinstance(payload, Exception):
|
||||
raise payload
|
||||
return payload
|
||||
return None
|
||||
|
||||
with patch.object(self.gms, "api_request", side_effect=fake_api_request):
|
||||
return self.gms._branch_protection_policy(
|
||||
"https://example/api/v1/repos/o/r", {"h": "1"}, base_branch="master"
|
||||
)
|
||||
|
||||
def test_status_checks_enabled_with_contexts(self):
|
||||
policy = self._policy({
|
||||
"branch_protections": [{
|
||||
"branch_name": "master",
|
||||
"block_on_outdated_branch": True,
|
||||
"enable_status_check": True,
|
||||
"status_check_contexts": ["ci/build", " "],
|
||||
}],
|
||||
})
|
||||
self.assertTrue(policy["determinable"])
|
||||
self.assertTrue(policy["checks_enabled"])
|
||||
self.assertTrue(policy["requires_current_base"])
|
||||
self.assertEqual(policy["required_contexts"], ["ci/build"])
|
||||
|
||||
def test_status_checks_disabled(self):
|
||||
policy = self._policy({
|
||||
"branch_protections": [{
|
||||
"branch_name": "master",
|
||||
"enable_status_check": False,
|
||||
}],
|
||||
})
|
||||
self.assertTrue(policy["determinable"])
|
||||
self.assertFalse(policy["checks_enabled"])
|
||||
|
||||
def test_no_protection_rule_means_checks_not_required(self):
|
||||
# PR #750's repository shape: protection list readable but empty.
|
||||
policy = self._policy({"branch_protections": [], "branches/master": {}})
|
||||
self.assertTrue(policy["determinable"])
|
||||
self.assertFalse(policy["protection_found"])
|
||||
self.assertFalse(policy["checks_enabled"])
|
||||
self.assertIsNone(policy["requires_current_base"])
|
||||
|
||||
def test_protection_without_status_field_means_not_enabled(self):
|
||||
policy = self._policy({
|
||||
"branch_protections": [{
|
||||
"branch_name": "master",
|
||||
"block_on_outdated_branch": True,
|
||||
}],
|
||||
})
|
||||
self.assertTrue(policy["determinable"])
|
||||
self.assertFalse(policy["checks_enabled"])
|
||||
self.assertTrue(policy["requires_current_base"])
|
||||
|
||||
def test_api_failure_is_not_determinable(self):
|
||||
policy = self._policy({
|
||||
"branch_protections": RuntimeError("boom"),
|
||||
})
|
||||
self.assertFalse(policy["determinable"])
|
||||
self.assertIsNone(policy["checks_enabled"])
|
||||
|
||||
def test_missing_branch_is_not_determinable(self):
|
||||
with patch.object(self.gms, "api_request", return_value=[]):
|
||||
policy = self.gms._branch_protection_policy(
|
||||
"https://example/api/v1/repos/o/r", {}, base_branch=" "
|
||||
)
|
||||
self.assertFalse(policy["determinable"])
|
||||
|
||||
def test_requires_current_base_accessor_preserved(self):
|
||||
def fake(method, url, auth, *a, **k):
|
||||
if "branch_protections" in url:
|
||||
return [{"branch_name": "master",
|
||||
"block_on_outdated_branch": True}]
|
||||
return None
|
||||
|
||||
with patch.object(self.gms, "api_request", side_effect=fake):
|
||||
value = self.gms._branch_protection_requires_current_base(
|
||||
"https://example/api/v1/repos/o/r", {}, base_branch="master"
|
||||
)
|
||||
self.assertTrue(value)
|
||||
|
||||
|
||||
class TestCommitChecksSnapshot(unittest.TestCase):
|
||||
def setUp(self):
|
||||
import gitea_mcp_server as gms
|
||||
|
||||
self.gms = gms
|
||||
|
||||
def test_reads_state_and_context_collection(self):
|
||||
payload = {"state": "pending", "statuses": [{"context": "b",
|
||||
"status": "pending"}]}
|
||||
with patch.object(self.gms, "api_request", return_value=payload):
|
||||
snap = self.gms._commit_checks_snapshot(
|
||||
"https://example/api/v1/repos/o/r", {}, sha=PR_HEAD
|
||||
)
|
||||
self.assertTrue(snap["determinable"])
|
||||
self.assertEqual(snap["combined_state"], "pending")
|
||||
self.assertEqual(len(snap["statuses"]), 1)
|
||||
|
||||
def test_empty_collection_recorded_as_determinable(self):
|
||||
with patch.object(
|
||||
self.gms, "api_request", return_value={"state": "pending", "statuses": []}
|
||||
):
|
||||
snap = self.gms._commit_checks_snapshot(
|
||||
"https://example/api/v1/repos/o/r", {}, sha=PR_HEAD
|
||||
)
|
||||
self.assertTrue(snap["determinable"])
|
||||
self.assertEqual(snap["statuses"], [])
|
||||
|
||||
def test_api_failure_is_not_determinable(self):
|
||||
with patch.object(self.gms, "api_request", side_effect=RuntimeError("x")):
|
||||
snap = self.gms._commit_checks_snapshot(
|
||||
"https://example/api/v1/repos/o/r", {}, sha=PR_HEAD
|
||||
)
|
||||
self.assertFalse(snap["determinable"])
|
||||
|
||||
|
||||
class TestMcpWrapperForwarding(unittest.TestCase):
|
||||
"""The production MCP path must reach the derived ``checks_required``."""
|
||||
|
||||
def setUp(self):
|
||||
import gitea_mcp_server as gms
|
||||
|
||||
self.gms = gms
|
||||
self.base = (
|
||||
"https://gitea.prgs.cc/api/v1/repos/Scaled-Tech-Consulting/Gitea-Tools"
|
||||
)
|
||||
|
||||
def _patches(self, fake_api_request, spy):
|
||||
return [
|
||||
patch.object(self.gms, "_profile_operation_gate", return_value=None),
|
||||
patch.object(self.gms, "_resolve", return_value=(
|
||||
"gitea.prgs.cc", "Scaled-Tech-Consulting", "Gitea-Tools")),
|
||||
patch.object(self.gms, "_auth", return_value={"Authorization": "x"}),
|
||||
patch.object(self.gms, "repo_api_url", return_value=self.base),
|
||||
patch.object(self.gms, "api_request", side_effect=fake_api_request),
|
||||
patch.object(self.gms, "gitea_get_pr_review_feedback", return_value={
|
||||
"success": True, "approval_at_current_head": True}),
|
||||
patch.object(self.gms, "_prove_author_ownership_for_pr", return_value={
|
||||
"has_author_lock": True}),
|
||||
patch.object(pr_sync_status, "assess_pr_sync_status", side_effect=spy),
|
||||
]
|
||||
|
||||
def _run(self, *, protections, status_payload, pr_number=750,
|
||||
caller_checks_status=None):
|
||||
captured = {}
|
||||
real_assess = pr_sync_status.assess_pr_sync_status
|
||||
|
||||
def spy(**kwargs):
|
||||
captured.update(kwargs)
|
||||
return real_assess(**kwargs)
|
||||
|
||||
def fake_api_request(method, url, auth, *args, **kwargs):
|
||||
if f"/pulls/{pr_number}" in url:
|
||||
return {
|
||||
"number": pr_number,
|
||||
"state": "open",
|
||||
"title": "t",
|
||||
"body": "b",
|
||||
"mergeable": True,
|
||||
"head": {"sha": PR_HEAD, "ref": "fix/x"},
|
||||
"base": {"sha": BASE_HEAD, "ref": "master"},
|
||||
}
|
||||
if "/branch_protections" in url:
|
||||
if isinstance(protections, Exception):
|
||||
raise protections
|
||||
return protections
|
||||
if "/branches/master" in url:
|
||||
return {"commit": {"id": BASE_HEAD}}
|
||||
if "/status" in url:
|
||||
if isinstance(status_payload, Exception):
|
||||
raise status_payload
|
||||
return status_payload
|
||||
if "/compare/" in url:
|
||||
return {"total_commits": 0}
|
||||
if "/comments" in url:
|
||||
return []
|
||||
return None
|
||||
|
||||
stack = self._patches(fake_api_request, spy)
|
||||
for p in stack:
|
||||
p.start()
|
||||
try:
|
||||
result = self.gms.gitea_assess_pr_sync_status(
|
||||
pr_number=pr_number, remote="prgs",
|
||||
checks_status=caller_checks_status,
|
||||
)
|
||||
finally:
|
||||
for p in reversed(stack):
|
||||
p.stop()
|
||||
return result, captured
|
||||
|
||||
def test_derived_checks_required_is_forwarded(self):
|
||||
result, captured = self._run(
|
||||
protections=[{"branch_name": "master", "enable_status_check": True,
|
||||
"status_check_contexts": ["ci/build"]}],
|
||||
status_payload={"state": "pending", "statuses": [
|
||||
{"context": "ci/build", "status": "pending"}]},
|
||||
)
|
||||
self.assertIn("checks_required", captured)
|
||||
self.assertTrue(captured["checks_required"])
|
||||
self.assertEqual(captured["checks_status"], CHECKS_PENDING)
|
||||
self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED)
|
||||
|
||||
def test_pr750_empty_status_with_no_protection_is_merge_ready(self):
|
||||
# The exact reported failure: combined pending, zero contexts, and no
|
||||
# branch protection requiring checks.
|
||||
result, captured = self._run(
|
||||
protections=[],
|
||||
status_payload={"state": "pending", "statuses": []},
|
||||
)
|
||||
self.assertFalse(captured["checks_required"])
|
||||
self.assertEqual(captured["checks_status"], CHECKS_NOT_REQUIRED)
|
||||
self.assertNotEqual(result["checks_status"], CHECKS_PENDING)
|
||||
self.assertEqual(result["recommended_next_action"], ACTION_MERGE_NOW)
|
||||
|
||||
def test_protection_read_failure_blocks(self):
|
||||
result, captured = self._run(
|
||||
protections=RuntimeError("protection unavailable"),
|
||||
status_payload={"state": "success", "statuses": []},
|
||||
)
|
||||
self.assertTrue(captured["checks_required"])
|
||||
self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED)
|
||||
|
||||
def test_caller_supplied_status_cannot_mask_live_required_failure(self):
|
||||
# A caller-declared "success" must not override live evidence.
|
||||
result, captured = self._run(
|
||||
protections=[{"branch_name": "master", "enable_status_check": True,
|
||||
"status_check_contexts": ["ci/build"]}],
|
||||
status_payload={"state": "failure", "statuses": [
|
||||
{"context": "ci/build", "status": "failure"}]},
|
||||
caller_checks_status="success",
|
||||
)
|
||||
self.assertEqual(captured["checks_status"], CHECKS_FAILURE)
|
||||
self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED)
|
||||
self.assertEqual(
|
||||
result["checks_evidence"]["caller_supplied_checks_status"], "success"
|
||||
)
|
||||
|
||||
def test_checks_evidence_is_reported_without_secrets(self):
|
||||
result, _ = self._run(
|
||||
protections=[],
|
||||
status_payload={"state": "pending", "statuses": []},
|
||||
)
|
||||
evidence = result["checks_evidence"]
|
||||
self.assertEqual(evidence["context_count"], 0)
|
||||
self.assertEqual(evidence["combined_state"], "pending")
|
||||
self.assertFalse(evidence["protection_found"])
|
||||
blob = repr(result).lower()
|
||||
self.assertNotIn("authorization", blob)
|
||||
self.assertNotIn("token", blob)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user