diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index beedf7e..58f78d1 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -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 diff --git a/pr_sync_status.py b/pr_sync_status.py index 546c42c..4d0e691 100644 --- a/pr_sync_status.py +++ b/pr_sync_status.py @@ -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. diff --git a/tests/test_issue_751_checks_assessor.py b/tests/test_issue_751_checks_assessor.py new file mode 100644 index 0000000..36ce374 --- /dev/null +++ b/tests/test_issue_751_checks_assessor.py @@ -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()