diff --git a/CHANGELOG.md b/CHANGELOG.md index d93533f..b649378 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,6 +13,17 @@ Format: [Keep a Changelog](https://keepachangelog.com/en/1.0.0/) ## [Unreleased] +### Added +- Added the local-only `audit pr-evidence ` operator path with + strict `PRHeadEvidenceV1` input validation and deterministic + `PRHeadEvidenceVerdictV1` output. Reviews retain actor, state, dismissal, and + `commit_id`; checks and suites retain status, conclusion, and `head_sha`; + required approvals/checks come only from supplied rules; latest-push + approval, stale head evidence, non-success conclusions, missing permissions, + incomplete pagination, missing rules, and malformed input remain explicit. + The fixture-first evaluator does not consult credentials or GitHub, write + files, alter portfolio scoring, or regenerate portfolio truth. + ### Changed - Kept PortfolioTruth `0.11.0` and `GitHubSecurityCoverageReceiptV1` read-compatible while adding normalized diff --git a/README.md b/README.md index 6ef375b..45c056c 100644 --- a/README.md +++ b/README.md @@ -126,6 +126,7 @@ Treat campaign/writeback, GitHub Projects, Notion sync, catalog overrides, score - Project history: [docs/project-history.md](docs/project-history.md) - Contributing: [CONTRIBUTING.md](CONTRIBUTING.md) - Security policy: [SECURITY.md](SECURITY.md) +- PR evidence-to-head binding: [docs/pr-head-evidence.md](docs/pr-head-evidence.md) ## Features @@ -272,6 +273,9 @@ audit run --repos --html # Action Sync — managed campaign preview / writeback audit report --campaign security-review --writeback-target github + +# Local PR evidence binding — no token, network request, or output file +audit pr-evidence tests/fixtures/pr_head_evidence/current.json ``` Normal runs perform a lightweight automatic preflight before fetching repos. By default @@ -288,6 +292,15 @@ groups work into `Needs Re-Approval`, `Ready For Review`, `Approved But Manual`, `Blocked`, and writes `approval-center--.json` plus `.md`. Local approval capture stays separate from writeback apply. +`audit pr-evidence ` is a separate local-only evidence check. It +validates a versioned `PRHeadEvidenceV1` snapshot and emits deterministic +`PRHeadEvidenceVerdictV1` JSON showing whether evidence required by the +supplied rules is current for the supplied PR head, while classifying every +supplied review and check. It does not read GitHub credentials, call GitHub, +write files, affect portfolio scoring, or regenerate portfolio truth. See +[docs/pr-head-evidence.md](docs/pr-head-evidence.md) for the input contract, +coverage requirements, exit codes, and claim ceiling. + Watch mode supports `--watch-strategy adaptive|incremental|full`. `adaptive` is the default and uses the stored baseline contract plus the scheduled full-refresh interval to decide whether each watch cycle should run full or incremental. diff --git a/docs/audit-cli-migration.md b/docs/audit-cli-migration.md index f58e2b1..52a5454 100644 --- a/docs/audit-cli-migration.md +++ b/docs/audit-cli-migration.md @@ -1,6 +1,6 @@ # CLI Migration: flat form → subcommand form -Arc F Sprint 4.3 introduced four subcommands — `run`, `triage`, `report`, and `serve` — +Arc F Sprint 4.3 introduced four workflow subcommands — `run`, `triage`, `report`, and `serve` — to replace the flat `audit --flag` invocation style. The flat form still works and emits a deprecation warning; it will not be removed until a future major version bump. @@ -20,6 +20,12 @@ The mapping is: | `audit report` | Portfolio truth, Excel workbooks, campaigns, writeback, exports | | `audit serve` | Local FastAPI + HTMX web UI | +Specialized read-only gates can also have standalone subcommands. In +particular, `audit pr-evidence ` evaluates a local +`PRHeadEvidenceV1` document without a username, credential lookup, network +request, output directory, or legacy flat-form equivalent. See +[pr-head-evidence.md](pr-head-evidence.md). + ## Flag family mapping ### audit run diff --git a/docs/pr-head-evidence.md b/docs/pr-head-evidence.md new file mode 100644 index 0000000..8c985a0 --- /dev/null +++ b/docs/pr-head-evidence.md @@ -0,0 +1,232 @@ +# Pull Request Evidence-to-Head Binding + +`audit pr-evidence` is a local-only operator command that answers one narrow +question: + +> Is the evidence required by the supplied rules current for the supplied pull +> request head? + +It reads a `PRHeadEvidenceV1` JSON snapshot, writes a deterministic +`PRHeadEvidenceVerdictV1` JSON document to standard output, and performs no +GitHub requests or file writes. + +```bash +audit pr-evidence tests/fixtures/pr_head_evidence/current.json +``` + +The command is intentionally separate from portfolio scoring and generated +portfolio truth. It does not regenerate or modify either contract. + +## Exit codes + +| Code | Meaning | +|---|---| +| `0` | All required evidence is current and successful for the supplied head. | +| `1` | The valid snapshot is stale, blocked, pending, missing, neutral, or skipped. | +| `2` | Coverage is unknown, the snapshot is inaccessible, or the input is malformed. | + +JSON is emitted for every evaluable command result, including malformed input. + +## `PRHeadEvidenceV1` input + +All fields below are required unless they are explicitly nullable. Unknown +fields are rejected so misspellings cannot silently weaken the contract. +Unknown rule *types* are preserved as coverage gaps and force an `unknown` +verdict. + +```json +{ + "schema_version": "PRHeadEvidenceV1", + "pull_request": { + "repository": "owner/repository", + "number": 42, + "author": "pull-request-author", + "head_sha": "40-or-64-character-full-sha", + "base_sha": "40-or-64-character-full-sha" + }, + "latest_reviewable_push": { + "pushed_at": "2026-07-28T10:00:00Z", + "actor": "pusher-login" + }, + "reviews": [ + { + "id": 1001, + "actor": { + "login": "reviewer-login", + "can_count": true + }, + "state": "APPROVED", + "commit_id": "full-reviewed-commit-sha", + "submitted_at": "2026-07-28T10:05:00Z", + "dismissal": { + "dismissed": false, + "dismissed_at": null, + "prior_state": null + } + } + ], + "checks": [ + { + "id": 2001, + "kind": "check_run", + "name": "ci/test", + "head_sha": "full-checked-commit-sha", + "status": "completed", + "conclusion": "success", + "integration_id": 99 + } + ], + "rules": { + "source": "branch_protection", + "availability": "available", + "items": [ + { + "type": "pull_request", + "required_approving_review_count": 1, + "dismiss_stale_reviews": true, + "require_last_push_approval": true, + "require_code_owner_reviews": false + }, + { + "type": "required_status_checks", + "strict_required_status_checks_policy": false, + "checks": [ + { + "context": "ci/test", + "integration_id": 99 + } + ] + } + ] + }, + "coverage": { + "reviews": { + "permission": "granted", + "complete": true, + "truncated": false, + "pages_fetched": 1 + }, + "checks": { + "permission": "granted", + "complete": true, + "truncated": false, + "pages_fetched": 1 + }, + "rules": { + "permission": "granted", + "complete": true, + "truncated": false, + "pages_fetched": 1 + } + }, + "provenance": { + "source": "synthetic_fixture", + "captured_at": "2026-07-28T10:10:00Z", + "freshness": "current" + } +} +``` + +`actor.can_count` is a normalized collector assertion that the reviewer is +eligible to count under the supplied repository policy. Use `null` when the +permission or actor eligibility could not be established; that keeps approval +coverage `unknown`. Actor logins are compared case-insensitively, matching +GitHub identity semantics. An approval from `pull_request.author` never counts, +even if a malformed collector assertion sets `can_count` to `true`. + +`review.submitted_at` may be `null` only for a `PENDING` review, matching +GitHub's response shape for an unsubmitted review. Submitted reviews require a +timezone-qualified timestamp. + +`latest_reviewable_push.pushed_at` and `.actor` are nullable only so a collector +can preserve missing data. If `require_last_push_approval` is true and either is +missing, the verdict is `unknown`. + +`rules.source` is `branch_protection`, `ruleset`, or `combined`. +`rules.availability` is `available`, `missing`, or `inaccessible`. A complete, +available, empty rules array means that the supplied rules snapshot establishes +no approval or status-check requirements. Missing rules never receive invented +defaults. + +The initial evaluator supports normalized pull-request approval counts, +stale-approval dismissal, latest-push approval, and required status-check +contexts. A code-owner requirement is accepted but yields `unknown` because a +head snapshot alone cannot prove every changed path's ownership groups. + +## Binding and outcome are separate + +Every review keeps its original `state`, actor, dismissal fields, and +`commit_id`. Every check keeps its kind, status, conclusion, and `head_sha`. +The verdict adds: + +- `binding_state`: `current` when the evidence SHA matches the supplied head, + otherwise `stale`; +- a review `decision_state`, including `commented` and `dismissed`; +- a check `result_state`, including `success`, `pending`, `neutral`, `skipped`, + `cancelled`, `blocked`, `stale`, and `unknown`. + +GitHub documents that `neutral` and `skipped` can satisfy a required status +check. The verdict exposes that narrow fact as +`github_requirement_satisfied=true`, but its top-level state remains `neutral` +or `skipped`; it is not collapsed into current successful evidence. + +Check suites are SHA-bound and classified, but they do not masquerade as a +required status-check context. GitHub branch rules select checks or commit +statuses, not suite summaries. + +When stale-review dismissal is disabled, GitHub can retain an approval after a +push. The verdict records that as a counted retained approval while keeping its +binding `stale`. If a requirement depends on retained stale approval, the +top-level evidence verdict is `stale`, not green. + +## Coverage rules + +The following conditions force `coverage.state=unknown`: + +- review, check, or rule permission is `missing`, `inaccessible`, or `unknown`; +- pagination is incomplete or truncated; +- protection/ruleset data is missing or inaccessible; +- a rule type is unknown; +- reviewer counting eligibility is unknown and can change the result; +- a code-owner review requirement is present; +- provenance freshness is unknown. + +A stale provenance assertion yields a `stale` verdict. No implicit freshness +window is invented. + +## Supported sources + +The first implementation is fixture-first. A snapshot may be produced by a +synthetic fixture, an operator export, or a separate read-only GitHub collector, +but the collector must normalize the same fields and state completeness +honestly. + +The semantics track GitHub's primary documentation: + +- [REST pull request reviews](https://docs.github.com/en/rest/pulls/reviews) + exposes review `state` and `commit_id`. +- [Protected branches](https://docs.github.com/en/repositories/configuring-branches-and-merges-in-your-repository/managing-protected-branches/about-protected-branches) + defines stale approval dismissal and approval of the most recent reviewable + push. +- [REST check runs](https://docs.github.com/en/rest/checks/runs) and + [REST check suites](https://docs.github.com/en/rest/checks/suites) bind check + evidence to `head_sha`. +- [Troubleshooting required status checks](https://docs.github.com/en/pull-requests/collaborating-with-pull-requests/collaborating-on-repositories-with-code-quality-features/troubleshooting-required-status-checks) + states that required checks apply to the latest commit SHA and describes + successful, neutral, and skipped conclusions. +- [REST repository rules](https://docs.github.com/en/rest/repos/rules) describes + normalized pull-request and required-status-check rule parameters. + +## Claim ceiling + +The command does **not** claim that GitHub would merge the pull request, that +the code is correct, that reviews were effective, that CI is trustworthy, or +that the branch is safe. Its top-level state evaluates required evidence under +the supplied rules. It still classifies every supplied review and check against +the supplied head so historical or non-required evidence cannot masquerade as +current evidence. + +Synthetic fixture triplets live in +`tests/fixtures/pr_head_evidence/semantic_triplets.json`. Direct CLI smoke +snapshots for current, stale, comment-only, dismissed, incomplete, and malformed +cases live beside it. diff --git a/src/app/pr_head_evidence.py b/src/app/pr_head_evidence.py new file mode 100644 index 0000000..41d234f --- /dev/null +++ b/src/app/pr_head_evidence.py @@ -0,0 +1,54 @@ +"""CLI adapter for local PR head-evidence snapshots.""" + +from __future__ import annotations + +import json +from pathlib import Path +from typing import Protocol + +from src.pr_head_evidence import ( + SnapshotValidationError, + evaluate_snapshot, + invalid_snapshot_verdict, + parse_snapshot, +) + + +class PRHeadEvidenceArgs(Protocol): + snapshot: str + + +def run_pr_head_evidence_mode(args: PRHeadEvidenceArgs) -> None: + """Evaluate one local snapshot and emit deterministic JSON to stdout.""" + + path = Path(args.snapshot) + try: + raw_text = path.read_text(encoding="utf-8") + except OSError as exc: + verdict = invalid_snapshot_verdict( + "snapshot_unreadable", + [f"{exc.__class__.__name__}: {exc}"], + ) + print(json.dumps(verdict, indent=2, sort_keys=True)) + raise SystemExit(2) from None + try: + value = json.loads(raw_text) + except json.JSONDecodeError as exc: + verdict = invalid_snapshot_verdict( + "snapshot_invalid_json", + [f"line {exc.lineno}, column {exc.colno}: {exc.msg}"], + ) + print(json.dumps(verdict, indent=2, sort_keys=True)) + raise SystemExit(2) from None + try: + snapshot = parse_snapshot(value) + except SnapshotValidationError as exc: + verdict = invalid_snapshot_verdict("snapshot_malformed", exc.errors) + print(json.dumps(verdict, indent=2, sort_keys=True)) + raise SystemExit(2) from None + verdict = evaluate_snapshot(snapshot) + print(json.dumps(verdict, indent=2, sort_keys=True)) + if verdict["state"] == "unknown": + raise SystemExit(2) + if not verdict["current"]: + raise SystemExit(1) diff --git a/src/cli.py b/src/cli.py index 44091c9..f5d5df1 100644 --- a/src/cli.py +++ b/src/cli.py @@ -77,7 +77,7 @@ - Use --control-center for read-only daily triage - Treat campaigns, writeback, catalog overrides, scorecards overrides, and GitHub Projects as advanced workflows""" -CLI_MODE_EXAMPLES = """Subcommands: run, triage, report, serve +CLI_MODE_EXAMPLES = """Subcommands: run, triage, report, serve, pr-evidence Run `audit run --help`, `audit triage --help`, or `audit report --help` for flags. Subcommand form (preferred): @@ -88,6 +88,7 @@ audit report --portfolio-truth audit report --campaign security-review --writeback-target github audit security-gate --output-dir output + audit pr-evidence path/to/pr-head-evidence.json audit serve [--port 8080] Legacy flat form (deprecated, still supported): @@ -1617,6 +1618,36 @@ def _build_security_gate_subparser(subparsers: argparse._SubParsersAction) -> No ) +def _build_pr_head_evidence_subparser( + subparsers: argparse._SubParsersAction, # type: ignore[type-arg] +) -> None: + """Subcommand: `audit pr-evidence` — bind local evidence to a PR head.""" + p = subparsers.add_parser( + "pr-evidence", + help="Evaluate whether a local PR evidence snapshot is current for its head", + description=( + "Read one PRHeadEvidenceV1 JSON snapshot and emit a deterministic " + "PRHeadEvidenceVerdictV1 JSON verdict. This command performs no " + "GitHub requests and writes no files." + ), + formatter_class=argparse.RawDescriptionHelpFormatter, + ) + p.add_argument("snapshot", help="Path to a local PRHeadEvidenceV1 JSON snapshot") + + +def build_pr_head_evidence_parser() -> argparse.ArgumentParser: + """Build the token-free parser used by the local-only pr-evidence path.""" + parser = argparse.ArgumentParser( + prog="audit pr-evidence", + description=( + "Evaluate a local PRHeadEvidenceV1 snapshot without network access " + "or external writes." + ), + ) + parser.add_argument("snapshot", help="Path to a local PRHeadEvidenceV1 JSON snapshot") + return parser + + def build_subcommand_parser() -> argparse.ArgumentParser: """Return the subcommand-aware parser used by main(). @@ -1647,6 +1678,7 @@ def build_subcommand_parser() -> argparse.ArgumentParser: _build_serve_subparser(subparsers) _build_security_burndown_subparser(subparsers) _build_security_gate_subparser(subparsers) + _build_pr_head_evidence_subparser(subparsers) return parser @@ -1885,7 +1917,15 @@ def _infer_subcommand_from_flags(args: argparse.Namespace) -> str: _KNOWN_SUBCOMMANDS: frozenset[str] = frozenset( - {"run", "triage", "report", "serve", "security-burndown", "security-gate"} + { + "run", + "triage", + "report", + "serve", + "security-burndown", + "security-gate", + "pr-evidence", + } ) @@ -1998,10 +2038,23 @@ def _run_security_gate_mode(args) -> None: run_security_gate_mode(args) +def _run_pr_head_evidence_mode(args) -> None: + from src.app.pr_head_evidence import run_pr_head_evidence_mode + + run_pr_head_evidence_mode(args) + + # ── Main entry point ────────────────────────────────────────────────── def main() -> None: raw_argv = sys.argv[1:] + # This local-only command must not construct the legacy parser because its + # global token default may consult `gh auth token`. + if raw_argv and raw_argv[0] == "pr-evidence": + pr_args = build_pr_head_evidence_parser().parse_args(raw_argv[1:]) + _run_pr_head_evidence_mode(pr_args) + return + # ── Choose parser based on invocation form ─────────────────────── # Subcommand form: audit run|triage|report|serve [args...] # Legacy flat form: audit username [--flags...] (deprecated) diff --git a/src/pr_head_evidence.py b/src/pr_head_evidence.py new file mode 100644 index 0000000..20b8527 --- /dev/null +++ b/src/pr_head_evidence.py @@ -0,0 +1,1309 @@ +"""Bind supplied pull-request evidence to an explicit GitHub head SHA. + +This module is deliberately pure and local. It evaluates a versioned JSON +snapshot without making GitHub requests or writing files. The result is not a +mergeability decision; it only reports whether the supplied review and check +evidence is current for the supplied pull-request head under the supplied +rules. +""" + +from __future__ import annotations + +import re +from dataclasses import dataclass +from datetime import datetime, timezone +from typing import Mapping + +SNAPSHOT_SCHEMA_VERSION = "PRHeadEvidenceV1" +VERDICT_SCHEMA_VERSION = "PRHeadEvidenceVerdictV1" + +_SHA_RE = re.compile(r"^[0-9a-fA-F]{40}(?:[0-9a-fA-F]{24})?$") +_REVIEW_STATES = { + "APPROVED", + "CHANGES_REQUESTED", + "COMMENTED", + "DISMISSED", + "PENDING", +} +_PERMISSIONS = {"granted", "missing", "inaccessible", "unknown"} +_FRESHNESS_STATES = {"current", "stale", "unknown"} +_RULE_AVAILABILITY = {"available", "missing", "inaccessible"} +_RULE_SOURCES = {"branch_protection", "ruleset", "combined"} +_CHECK_KINDS = {"check_run", "check_suite"} +_PENDING_CHECK_STATUSES = { + "queued", + "in_progress", + "requested", + "waiting", + "pending", +} +_SUCCESSFUL_GITHUB_CONCLUSIONS = {"success", "neutral", "skipped"} +_BLOCKING_CHECK_CONCLUSIONS = { + "action_required", + "failure", + "startup_failure", + "timed_out", +} + + +def _id_sort_key(value: str) -> tuple[int, int | str]: + return (0, int(value)) if value.isdigit() else (1, value) + + +def _actor_key(login: str) -> str: + """Return the case-insensitive identity key GitHub uses for logins.""" + + return login.casefold() + + +class SnapshotValidationError(ValueError): + """Raised when a PRHeadEvidenceV1 snapshot is structurally malformed.""" + + def __init__(self, errors: list[str]) -> None: + self.errors = tuple(sorted(set(errors))) + super().__init__("; ".join(self.errors)) + + +@dataclass(frozen=True) +class CoverageSlice: + permission: str + complete: bool + truncated: bool + pages_fetched: int + + +@dataclass(frozen=True) +class Actor: + login: str + can_count: bool | None + + +@dataclass(frozen=True) +class PullRequestIdentity: + repository: str + number: int + author: str + head_sha: str + base_sha: str + + +@dataclass(frozen=True) +class LatestReviewablePush: + pushed_at: datetime | None + actor: str | None + + +@dataclass(frozen=True) +class Review: + review_id: str + actor: Actor + state: str + commit_id: str + submitted_at: datetime | None + dismissed: bool + dismissed_at: datetime | None + dismissed_prior_state: str | None + + +def _review_sort_key(review: Review) -> tuple[bool, datetime, tuple[int, int | str]]: + submitted_at = review.submitted_at or datetime.min.replace(tzinfo=timezone.utc) + return ( + review.submitted_at is None, + submitted_at, + _id_sort_key(review.review_id), + ) + + +@dataclass(frozen=True) +class Check: + check_id: str + kind: str + name: str + head_sha: str + status: str + conclusion: str | None + integration_id: int | None + + +@dataclass(frozen=True) +class PullRequestRule: + required_approving_review_count: int + dismiss_stale_reviews: bool + require_last_push_approval: bool + require_code_owner_reviews: bool + + +@dataclass(frozen=True) +class RequiredCheck: + context: str + integration_id: int | None + + +@dataclass(frozen=True) +class RequiredStatusChecksRule: + strict_required_status_checks_policy: bool + checks: tuple[RequiredCheck, ...] + + +@dataclass(frozen=True) +class UnknownRule: + rule_type: str + + +Rule = PullRequestRule | RequiredStatusChecksRule | UnknownRule + + +@dataclass(frozen=True) +class RulesSnapshot: + source: str + availability: str + items: tuple[Rule, ...] + + +@dataclass(frozen=True) +class Provenance: + source: str + captured_at: datetime + freshness: str + + +@dataclass(frozen=True) +class PRHeadEvidenceSnapshot: + pull_request: PullRequestIdentity + latest_reviewable_push: LatestReviewablePush + reviews: tuple[Review, ...] + checks: tuple[Check, ...] + rules: RulesSnapshot + coverage: Mapping[str, CoverageSlice] + provenance: Provenance + + +def _mapping(value: object, path: str, errors: list[str]) -> Mapping[str, object]: + if isinstance(value, dict): + return value + errors.append(f"{path}: expected object") + return {} + + +def _reject_unknown_keys( + payload: Mapping[str, object], + allowed: set[str], + path: str, + errors: list[str], +) -> None: + for key in sorted(set(payload) - allowed): + errors.append(f"{path}.{key}: unknown field") + + +def _list(value: object, path: str, errors: list[str]) -> list[object]: + if isinstance(value, list): + return value + errors.append(f"{path}: expected array") + return [] + + +def _text(value: object, path: str, errors: list[str]) -> str: + if isinstance(value, str) and value.strip(): + return value.strip() + errors.append(f"{path}: expected non-empty string") + return "" + + +def _optional_text(value: object, path: str, errors: list[str]) -> str | None: + if value is None: + return None + return _text(value, path, errors) + + +def _boolean(value: object, path: str, errors: list[str]) -> bool: + if isinstance(value, bool): + return value + errors.append(f"{path}: expected boolean") + return False + + +def _integer(value: object, path: str, errors: list[str]) -> int: + if isinstance(value, int) and not isinstance(value, bool): + return value + errors.append(f"{path}: expected integer") + return 0 + + +def _optional_integer(value: object, path: str, errors: list[str]) -> int | None: + if value is None: + return None + return _integer(value, path, errors) + + +def _timestamp(value: object, path: str, errors: list[str]) -> datetime: + text = _text(value, path, errors) + if not text: + return datetime.min + try: + parsed = datetime.fromisoformat(text.replace("Z", "+00:00")) + except ValueError: + errors.append(f"{path}: expected ISO-8601 timestamp") + return datetime.min + if parsed.tzinfo is None: + errors.append(f"{path}: timestamp must include timezone") + return parsed + + +def _optional_timestamp( + value: object, + path: str, + errors: list[str], +) -> datetime | None: + if value is None: + return None + return _timestamp(value, path, errors) + + +def _sha(value: object, path: str, errors: list[str]) -> str: + text = _text(value, path, errors).lower() + if text and not _SHA_RE.fullmatch(text): + errors.append(f"{path}: expected full 40- or 64-character hexadecimal SHA") + return text + + +def _choice( + value: object, + choices: set[str], + path: str, + errors: list[str], + *, + normalize_upper: bool = False, +) -> str: + text = _text(value, path, errors) + normalized = text.upper() if normalize_upper else text.lower() + if normalized and normalized not in choices: + errors.append(f"{path}: unsupported value {text!r}") + return normalized + + +def _parse_coverage_slice( + value: object, + path: str, + errors: list[str], +) -> CoverageSlice: + payload = _mapping(value, path, errors) + _reject_unknown_keys( + payload, + {"permission", "complete", "truncated", "pages_fetched"}, + path, + errors, + ) + permission = _choice( + payload.get("permission"), _PERMISSIONS, f"{path}.permission", errors + ) + complete = _boolean(payload.get("complete"), f"{path}.complete", errors) + truncated = _boolean(payload.get("truncated"), f"{path}.truncated", errors) + pages_fetched = _integer( + payload.get("pages_fetched"), f"{path}.pages_fetched", errors + ) + if pages_fetched < 0: + errors.append(f"{path}.pages_fetched: must be non-negative") + if truncated and complete: + errors.append(f"{path}: truncated collection cannot be complete") + if permission != "granted" and complete: + errors.append( + f"{path}: collection without granted permission cannot be complete" + ) + return CoverageSlice(permission, complete, truncated, pages_fetched) + + +def _parse_actor(value: object, path: str, errors: list[str]) -> Actor: + payload = _mapping(value, path, errors) + _reject_unknown_keys(payload, {"login", "can_count"}, path, errors) + login = _text(payload.get("login"), f"{path}.login", errors) + can_count_value = payload.get("can_count") + can_count: bool | None + if can_count_value is None: + can_count = None + else: + can_count = _boolean(can_count_value, f"{path}.can_count", errors) + return Actor(login=login, can_count=can_count) + + +def _parse_review(value: object, index: int, errors: list[str]) -> Review: + path = f"reviews[{index}]" + payload = _mapping(value, path, errors) + _reject_unknown_keys( + payload, + {"id", "actor", "state", "commit_id", "submitted_at", "dismissal"}, + path, + errors, + ) + review_id_value = payload.get("id") + if isinstance(review_id_value, (str, int)) and not isinstance( + review_id_value, bool + ): + review_id = str(review_id_value) + else: + errors.append(f"{path}.id: expected string or integer") + review_id = "" + actor = _parse_actor(payload.get("actor"), f"{path}.actor", errors) + state = _choice( + payload.get("state"), + _REVIEW_STATES, + f"{path}.state", + errors, + normalize_upper=True, + ) + commit_id = _sha(payload.get("commit_id"), f"{path}.commit_id", errors) + submitted_at = _optional_timestamp( + payload.get("submitted_at"), f"{path}.submitted_at", errors + ) + dismissal = _mapping(payload.get("dismissal"), f"{path}.dismissal", errors) + _reject_unknown_keys( + dismissal, + {"dismissed", "dismissed_at", "prior_state"}, + f"{path}.dismissal", + errors, + ) + dismissed = _boolean( + dismissal.get("dismissed"), f"{path}.dismissal.dismissed", errors + ) + dismissed_at = _optional_timestamp( + dismissal.get("dismissed_at"), + f"{path}.dismissal.dismissed_at", + errors, + ) + dismissed_prior_state = _optional_text( + dismissal.get("prior_state"), + f"{path}.dismissal.prior_state", + errors, + ) + if dismissed_prior_state is not None: + dismissed_prior_state = dismissed_prior_state.upper() + if dismissed_prior_state not in {"APPROVED", "CHANGES_REQUESTED", "COMMENTED"}: + errors.append( + f"{path}.dismissal.prior_state: unsupported value {dismissed_prior_state!r}" + ) + if state == "DISMISSED" and not dismissed: + errors.append(f"{path}: DISMISSED state requires dismissal.dismissed=true") + if state != "PENDING" and submitted_at is None: + errors.append(f"{path}: submitted review requires submitted_at") + if state != "DISMISSED" and dismissed: + errors.append(f"{path}: dismissed review must use state DISMISSED") + if dismissed and dismissed_at is None: + errors.append(f"{path}: dismissed review requires dismissal.dismissed_at") + if not dismissed and ( + dismissed_at is not None or dismissed_prior_state is not None + ): + errors.append(f"{path}: active review cannot include dismissal history") + return Review( + review_id=review_id, + actor=actor, + state=state, + commit_id=commit_id, + submitted_at=submitted_at, + dismissed=dismissed, + dismissed_at=dismissed_at, + dismissed_prior_state=dismissed_prior_state, + ) + + +def _parse_check(value: object, index: int, errors: list[str]) -> Check: + path = f"checks[{index}]" + payload = _mapping(value, path, errors) + _reject_unknown_keys( + payload, + { + "id", + "kind", + "name", + "head_sha", + "status", + "conclusion", + "integration_id", + }, + path, + errors, + ) + check_id_value = payload.get("id") + if isinstance(check_id_value, (str, int)) and not isinstance(check_id_value, bool): + check_id = str(check_id_value) + else: + errors.append(f"{path}.id: expected string or integer") + check_id = "" + kind = _choice(payload.get("kind"), _CHECK_KINDS, f"{path}.kind", errors) + name = _text(payload.get("name"), f"{path}.name", errors) + head_sha = _sha(payload.get("head_sha"), f"{path}.head_sha", errors) + status = _text(payload.get("status"), f"{path}.status", errors).lower() + conclusion_value = payload.get("conclusion") + conclusion = ( + None + if conclusion_value is None + else _text(conclusion_value, f"{path}.conclusion", errors).lower() + ) + integration_id = _optional_integer( + payload.get("integration_id"), + f"{path}.integration_id", + errors, + ) + if status == "completed" and conclusion is None: + errors.append(f"{path}: completed check requires conclusion") + if status != "completed" and conclusion is not None: + errors.append(f"{path}: incomplete check must use conclusion=null") + return Check( + check_id=check_id, + kind=kind, + name=name, + head_sha=head_sha, + status=status, + conclusion=conclusion, + integration_id=integration_id, + ) + + +def _parse_rule(value: object, index: int, errors: list[str]) -> Rule: + path = f"rules.items[{index}]" + payload = _mapping(value, path, errors) + rule_type = _text(payload.get("type"), f"{path}.type", errors) + if rule_type == "pull_request": + _reject_unknown_keys( + payload, + { + "type", + "required_approving_review_count", + "dismiss_stale_reviews", + "require_last_push_approval", + "require_code_owner_reviews", + }, + path, + errors, + ) + required_count = _integer( + payload.get("required_approving_review_count"), + f"{path}.required_approving_review_count", + errors, + ) + if required_count < 0 or required_count > 6: + errors.append( + f"{path}.required_approving_review_count: must be between 0 and 6" + ) + return PullRequestRule( + required_approving_review_count=required_count, + dismiss_stale_reviews=_boolean( + payload.get("dismiss_stale_reviews"), + f"{path}.dismiss_stale_reviews", + errors, + ), + require_last_push_approval=_boolean( + payload.get("require_last_push_approval"), + f"{path}.require_last_push_approval", + errors, + ), + require_code_owner_reviews=_boolean( + payload.get("require_code_owner_reviews"), + f"{path}.require_code_owner_reviews", + errors, + ), + ) + if rule_type == "required_status_checks": + _reject_unknown_keys( + payload, + { + "type", + "strict_required_status_checks_policy", + "checks", + }, + path, + errors, + ) + checks_payload = _list(payload.get("checks"), f"{path}.checks", errors) + required_checks: list[RequiredCheck] = [] + for check_index, check_value in enumerate(checks_payload): + check_path = f"{path}.checks[{check_index}]" + check_payload = _mapping(check_value, check_path, errors) + _reject_unknown_keys( + check_payload, + {"context", "integration_id"}, + check_path, + errors, + ) + required_checks.append( + RequiredCheck( + context=_text( + check_payload.get("context"), + f"{check_path}.context", + errors, + ), + integration_id=_optional_integer( + check_payload.get("integration_id"), + f"{check_path}.integration_id", + errors, + ), + ) + ) + return RequiredStatusChecksRule( + strict_required_status_checks_policy=_boolean( + payload.get("strict_required_status_checks_policy"), + f"{path}.strict_required_status_checks_policy", + errors, + ), + checks=tuple(required_checks), + ) + return UnknownRule(rule_type=rule_type or "") + + +def parse_snapshot(value: object) -> PRHeadEvidenceSnapshot: + """Validate and parse a PRHeadEvidenceV1 JSON value.""" + + errors: list[str] = [] + payload = _mapping(value, "$", errors) + _reject_unknown_keys( + payload, + { + "schema_version", + "pull_request", + "latest_reviewable_push", + "reviews", + "checks", + "rules", + "coverage", + "provenance", + }, + "$", + errors, + ) + schema_version = _text(payload.get("schema_version"), "schema_version", errors) + if schema_version and schema_version != SNAPSHOT_SCHEMA_VERSION: + errors.append( + f"schema_version: expected {SNAPSHOT_SCHEMA_VERSION!r}, got {schema_version!r}" + ) + + pr_payload = _mapping(payload.get("pull_request"), "pull_request", errors) + _reject_unknown_keys( + pr_payload, + {"repository", "number", "author", "head_sha", "base_sha"}, + "pull_request", + errors, + ) + pull_request = PullRequestIdentity( + repository=_text( + pr_payload.get("repository"), "pull_request.repository", errors + ), + number=_integer(pr_payload.get("number"), "pull_request.number", errors), + author=_text(pr_payload.get("author"), "pull_request.author", errors), + head_sha=_sha(pr_payload.get("head_sha"), "pull_request.head_sha", errors), + base_sha=_sha(pr_payload.get("base_sha"), "pull_request.base_sha", errors), + ) + if pull_request.number <= 0: + errors.append("pull_request.number: must be positive") + if pull_request.repository and "/" not in pull_request.repository: + errors.append("pull_request.repository: expected owner/name") + + push_payload = _mapping( + payload.get("latest_reviewable_push"), + "latest_reviewable_push", + errors, + ) + _reject_unknown_keys( + push_payload, + {"pushed_at", "actor"}, + "latest_reviewable_push", + errors, + ) + latest_push = LatestReviewablePush( + pushed_at=_optional_timestamp( + push_payload.get("pushed_at"), + "latest_reviewable_push.pushed_at", + errors, + ), + actor=_optional_text( + push_payload.get("actor"), + "latest_reviewable_push.actor", + errors, + ), + ) + + reviews_payload = _list(payload.get("reviews"), "reviews", errors) + reviews = tuple( + _parse_review(review, index, errors) + for index, review in enumerate(reviews_payload) + ) + review_ids = [review.review_id for review in reviews] + if len(set(review_ids)) != len(review_ids): + errors.append("reviews: duplicate id") + checks_payload = _list(payload.get("checks"), "checks", errors) + checks = tuple( + _parse_check(check, index, errors) for index, check in enumerate(checks_payload) + ) + check_ids = [check.check_id for check in checks] + if len(set(check_ids)) != len(check_ids): + errors.append("checks: duplicate id") + + rules_payload = _mapping(payload.get("rules"), "rules", errors) + _reject_unknown_keys( + rules_payload, + {"source", "availability", "items"}, + "rules", + errors, + ) + rule_items_payload = _list(rules_payload.get("items"), "rules.items", errors) + rules = RulesSnapshot( + source=_choice( + rules_payload.get("source"), + _RULE_SOURCES, + "rules.source", + errors, + ), + availability=_choice( + rules_payload.get("availability"), + _RULE_AVAILABILITY, + "rules.availability", + errors, + ), + items=tuple( + _parse_rule(rule, index, errors) + for index, rule in enumerate(rule_items_payload) + ), + ) + + coverage_payload = _mapping(payload.get("coverage"), "coverage", errors) + _reject_unknown_keys( + coverage_payload, + {"reviews", "checks", "rules"}, + "coverage", + errors, + ) + coverage: dict[str, CoverageSlice] = {} + for collection in ("reviews", "checks", "rules"): + coverage[collection] = _parse_coverage_slice( + coverage_payload.get(collection), + f"coverage.{collection}", + errors, + ) + + provenance_payload = _mapping(payload.get("provenance"), "provenance", errors) + _reject_unknown_keys( + provenance_payload, + {"source", "captured_at", "freshness"}, + "provenance", + errors, + ) + provenance = Provenance( + source=_text(provenance_payload.get("source"), "provenance.source", errors), + captured_at=_timestamp( + provenance_payload.get("captured_at"), + "provenance.captured_at", + errors, + ), + freshness=_choice( + provenance_payload.get("freshness"), + _FRESHNESS_STATES, + "provenance.freshness", + errors, + ), + ) + + if errors: + raise SnapshotValidationError(errors) + return PRHeadEvidenceSnapshot( + pull_request=pull_request, + latest_reviewable_push=latest_push, + reviews=reviews, + checks=checks, + rules=rules, + coverage=coverage, + provenance=provenance, + ) + + +def _rule_requirements( + snapshot: PRHeadEvidenceSnapshot, +) -> tuple[int, bool, bool, bool, tuple[RequiredCheck, ...], list[str]]: + required_approvals = 0 + dismiss_stale_reviews = False + require_last_push_approval = False + pull_request_rule_present = False + checks_by_key: dict[tuple[str, int | None], RequiredCheck] = {} + reasons: list[str] = [] + for rule in snapshot.rules.items: + if isinstance(rule, PullRequestRule): + pull_request_rule_present = True + required_approvals = max( + required_approvals, + rule.required_approving_review_count, + ) + dismiss_stale_reviews = dismiss_stale_reviews or rule.dismiss_stale_reviews + require_last_push_approval = ( + require_last_push_approval or rule.require_last_push_approval + ) + if rule.require_code_owner_reviews: + reasons.append("code_owner_review_requirement_not_evaluable") + elif isinstance(rule, RequiredStatusChecksRule): + for check in rule.checks: + checks_by_key[(check.context, check.integration_id)] = check + else: + reasons.append(f"unknown_rule_type:{rule.rule_type}") + required_checks = tuple( + checks_by_key[key] + for key in sorted( + checks_by_key, + key=lambda item: (item[0], -1 if item[1] is None else item[1]), + ) + ) + return ( + required_approvals, + pull_request_rule_present, + dismiss_stale_reviews, + require_last_push_approval, + required_checks, + reasons, + ) + + +def _review_payload( + review: Review, + snapshot: PRHeadEvidenceSnapshot, +) -> dict[str, object]: + binding_state = ( + "current" if review.commit_id == snapshot.pull_request.head_sha else "stale" + ) + if review.state == "DISMISSED": + decision_state = "dismissed" + else: + decision_state = review.state.lower() + is_pull_request_author = _actor_key(review.actor.login) == _actor_key( + snapshot.pull_request.author + ) + if is_pull_request_author: + eligibility_state = "ineligible" + elif review.actor.can_count is None: + eligibility_state = "unknown" + elif review.actor.can_count: + eligibility_state = "eligible" + else: + eligibility_state = "ineligible" + reasons: list[str] = [] + if binding_state == "stale": + reasons.append("review_commit_does_not_match_head") + if decision_state == "commented": + reasons.append("commented_is_not_approval") + if decision_state == "dismissed": + reasons.append("review_dismissed") + if eligibility_state == "unknown": + reasons.append("reviewer_counting_eligibility_unknown") + if eligibility_state == "ineligible": + reasons.append("reviewer_does_not_count") + if is_pull_request_author: + reasons.append("pull_request_author_cannot_approve") + return { + "id": review.review_id, + "actor": review.actor.login, + "actor_can_count": review.actor.can_count, + "review_state": review.state, + "dismissed": review.dismissed, + "dismissed_prior_state": review.dismissed_prior_state, + "commit_id": review.commit_id, + "submitted_at": ( + review.submitted_at.isoformat() if review.submitted_at is not None else None + ), + "binding_state": binding_state, + "decision_state": decision_state, + "eligibility_state": eligibility_state, + "counts_toward_required_approvals": False, + "satisfies_latest_push_approval": False, + "reasons": reasons, + } + + +def _latest_decisive_reviews(reviews: tuple[Review, ...]) -> dict[str, Review]: + latest: dict[str, Review] = {} + for review in sorted(reviews, key=_review_sort_key): + if review.state in {"COMMENTED", "PENDING"}: + continue + latest[_actor_key(review.actor.login)] = review + return latest + + +def _evaluate_reviews( + snapshot: PRHeadEvidenceSnapshot, + *, + required_approvals: int, + pull_request_rule_present: bool, + dismiss_stale_reviews: bool, + require_last_push_approval: bool, +) -> tuple[dict[str, object], list[dict[str, object]], list[str]]: + review_rows = [ + _review_payload(review, snapshot) + for review in sorted(snapshot.reviews, key=_review_sort_key) + ] + rows_by_id = {str(row["id"]): row for row in review_rows} + latest = _latest_decisive_reviews(snapshot.reviews) + counted: list[Review] = [] + current_count = 0 + retained_stale_count = 0 + unknown_eligibility = False + blocking_change_requests: list[str] = [] + + for review in latest.values(): + if _actor_key(review.actor.login) == _actor_key(snapshot.pull_request.author): + continue + if review.actor.can_count is None and review.state in { + "APPROVED", + "CHANGES_REQUESTED", + }: + unknown_eligibility = True + continue + if review.actor.can_count is not True: + continue + if review.state == "CHANGES_REQUESTED" and pull_request_rule_present: + blocking_change_requests.append(review.actor.login) + continue + if review.state != "APPROVED" or review.dismissed: + continue + is_current = review.commit_id == snapshot.pull_request.head_sha + if not is_current and dismiss_stale_reviews: + continue + counted.append(review) + row = rows_by_id[review.review_id] + row["counts_toward_required_approvals"] = True + if is_current: + current_count += 1 + else: + retained_stale_count += 1 + existing_reasons = row["reasons"] + normalized_reasons = ( + [str(reason) for reason in existing_reasons] + if isinstance(existing_reasons, list) + else [] + ) + row["reasons"] = [ + *normalized_reasons, + "approval_retained_by_supplied_rule_but_not_head_bound", + ] + + latest_push_satisfied = not require_last_push_approval + latest_push_approval_actor: str | None = None + latest_push_unknown = False + if require_last_push_approval: + push = snapshot.latest_reviewable_push + if push.actor is None or push.pushed_at is None: + latest_push_unknown = True + else: + candidates = [ + review + for review in counted + if review.commit_id == snapshot.pull_request.head_sha + and review.submitted_at is not None + and review.submitted_at >= push.pushed_at + and _actor_key(review.actor.login) != _actor_key(push.actor) + ] + if candidates: + selected = sorted(candidates, key=_review_sort_key)[-1] + latest_push_satisfied = True + latest_push_approval_actor = selected.actor.login + rows_by_id[selected.review_id]["satisfies_latest_push_approval"] = True + + approval_count_satisfied = len(counted) >= required_approvals + reasons: list[str] = [] + pending_review_count = sum( + 1 for review in snapshot.reviews if review.state == "PENDING" + ) + stale_approval_count = sum( + 1 + for review in snapshot.reviews + if review.state == "APPROVED" + and not review.dismissed + and review.commit_id != snapshot.pull_request.head_sha + ) + + if blocking_change_requests: + state = "blocked" + reasons.append("changes_requested") + elif unknown_eligibility or latest_push_unknown: + state = "unknown" + if unknown_eligibility: + reasons.append("reviewer_counting_eligibility_unknown") + if latest_push_unknown: + reasons.append("latest_push_metadata_missing") + elif not approval_count_satisfied: + if pending_review_count: + state = "pending" + reasons.append("required_approval_pending") + elif stale_approval_count: + state = "stale" + reasons.append("required_approval_only_on_older_sha") + else: + state = "missing" + reasons.append("required_approval_missing") + elif not latest_push_satisfied: + if stale_approval_count or retained_stale_count: + state = "stale" + reasons.append("latest_push_approval_not_current") + else: + state = "missing" + reasons.append("latest_push_approval_missing") + elif required_approvals > current_count: + state = "stale" + reasons.append("approval_requirement_uses_retained_stale_review") + else: + state = "current" + + return ( + { + "state": state, + "required_count": required_approvals, + "counted_actors": sorted( + (review.actor.login for review in counted), + key=_actor_key, + ), + "counted_count": len(counted), + "head_bound_count": current_count, + "retained_stale_count": retained_stale_count, + "dismiss_stale_reviews": dismiss_stale_reviews, + "require_last_push_approval": require_last_push_approval, + "latest_push_approval_actor": latest_push_approval_actor, + "github_requirement_satisfied": ( + approval_count_satisfied + and latest_push_satisfied + and not blocking_change_requests + ), + "reasons": reasons, + }, + review_rows, + reasons, + ) + + +def _check_result_state(check: Check) -> str: + if check.status in _PENDING_CHECK_STATUSES: + return "pending" + if check.status != "completed": + return "unknown" + if check.conclusion == "success": + return "success" + if check.conclusion == "neutral": + return "neutral" + if check.conclusion == "skipped": + return "skipped" + if check.conclusion == "cancelled": + return "cancelled" + if check.conclusion == "stale": + return "stale" + if check.conclusion in _BLOCKING_CHECK_CONCLUSIONS: + return "blocked" + return "unknown" + + +def _check_payload( + check: Check, + snapshot: PRHeadEvidenceSnapshot, +) -> dict[str, object]: + binding_state = ( + "current" if check.head_sha == snapshot.pull_request.head_sha else "stale" + ) + result_state = _check_result_state(check) + reasons: list[str] = [] + if binding_state == "stale": + reasons.append("check_head_does_not_match_pr_head") + if result_state != "success": + reasons.append(f"check_result_{result_state}") + return { + "id": check.check_id, + "kind": check.kind, + "name": check.name, + "head_sha": check.head_sha, + "status": check.status, + "conclusion": check.conclusion, + "integration_id": check.integration_id, + "binding_state": binding_state, + "result_state": result_state, + "required": False, + "github_requirement_satisfied": ( + binding_state == "current" + and result_state in _SUCCESSFUL_GITHUB_CONCLUSIONS + ), + "reasons": reasons, + } + + +def _required_check_matches(required: RequiredCheck, check: Check) -> bool: + if check.kind != "check_run" or check.name != required.context: + return False + return ( + required.integration_id is None + or check.integration_id == required.integration_id + ) + + +def _evaluate_checks( + snapshot: PRHeadEvidenceSnapshot, + required_checks: tuple[RequiredCheck, ...], +) -> tuple[dict[str, object], list[dict[str, object]], list[str]]: + check_rows = [ + _check_payload(check, snapshot) + for check in sorted( + snapshot.checks, + key=lambda item: ( + item.kind, + item.name, + -1 if item.integration_id is None else item.integration_id, + _id_sort_key(item.check_id), + ), + ) + ] + rows_by_id = {str(row["id"]): row for row in check_rows} + requirement_rows: list[dict[str, object]] = [] + + for required in required_checks: + matches = [ + check + for check in snapshot.checks + if _required_check_matches(required, check) + ] + current_matches = [ + check + for check in matches + if check.head_sha == snapshot.pull_request.head_sha + ] + for check in matches: + rows_by_id[check.check_id]["required"] = True + + reasons: list[str] = [] + selected_sha: str | None = None + conclusion: str | None = None + github_satisfied = False + if len(current_matches) > 1: + state = "unknown" + reasons.append("duplicate_current_required_check_evidence") + elif len(current_matches) == 1: + selected = current_matches[0] + selected_sha = selected.head_sha + conclusion = selected.conclusion + state = _check_result_state(selected) + github_satisfied = state in _SUCCESSFUL_GITHUB_CONCLUSIONS + if state != "success": + reasons.append(f"required_check_{state}") + else: + stale_matches = [ + check + for check in matches + if _check_result_state(check) in _SUCCESSFUL_GITHUB_CONCLUSIONS + ] + if stale_matches: + state = "stale" + selected_sha = sorted( + stale_matches, + key=lambda item: ( + item.head_sha, + _id_sort_key(item.check_id), + ), + )[-1].head_sha + reasons.append("required_check_only_on_older_sha") + else: + state = "missing" + reasons.append("required_check_missing") + + requirement_rows.append( + { + "context": required.context, + "integration_id": required.integration_id, + "state": state, + "evidence_sha": selected_sha, + "conclusion": conclusion, + "github_requirement_satisfied": github_satisfied, + "reasons": reasons, + } + ) + + states = {str(row["state"]) for row in requirement_rows} + if "unknown" in states: + overall_state = "unknown" + elif states & {"blocked", "cancelled"}: + overall_state = "blocked" + elif "pending" in states: + overall_state = "pending" + elif "missing" in states: + overall_state = "missing" + elif "stale" in states: + overall_state = "stale" + elif "neutral" in states: + overall_state = "neutral" + elif "skipped" in states: + overall_state = "skipped" + else: + overall_state = "current" + reason_set: set[str] = set() + for row in requirement_rows: + row_reasons = row["reasons"] + if isinstance(row_reasons, list): + reason_set.update(str(reason) for reason in row_reasons) + reasons = sorted(reason_set) + return ( + { + "state": overall_state, + "required": requirement_rows, + "github_requirement_satisfied": all( + bool(row["github_requirement_satisfied"]) for row in requirement_rows + ), + "all_required_checks_successful": all( + row["state"] == "success" for row in requirement_rows + ), + "reasons": reasons, + }, + check_rows, + reasons, + ) + + +def _coverage_reasons(snapshot: PRHeadEvidenceSnapshot) -> list[str]: + reasons: list[str] = [] + for name in ("reviews", "checks", "rules"): + coverage = snapshot.coverage[name] + if coverage.permission != "granted": + reasons.append(f"{name}_permission_{coverage.permission}") + if not coverage.complete: + reasons.append(f"{name}_pagination_incomplete") + if coverage.truncated: + reasons.append(f"{name}_pagination_truncated") + if snapshot.rules.availability != "available": + reasons.append(f"rules_{snapshot.rules.availability}") + return sorted(set(reasons)) + + +def _overall_state( + *, + freshness: str, + coverage_reasons: list[str], + rule_reasons: list[str], + approval_state: str, + check_state: str, +) -> str: + if freshness == "unknown": + return "unknown" + if freshness == "stale": + return "stale" + if coverage_reasons or rule_reasons: + return "unknown" + states = {approval_state, check_state} + if "unknown" in states: + return "unknown" + if "blocked" in states: + return "blocked" + if "pending" in states: + return "pending" + if "missing" in states: + return "missing" + if "stale" in states: + return "stale" + if "neutral" in states: + return "neutral" + if "skipped" in states: + return "skipped" + return "current" + + +def evaluate_snapshot(snapshot: PRHeadEvidenceSnapshot) -> dict[str, object]: + """Return a deterministic PRHeadEvidenceVerdictV1 dictionary.""" + + coverage_reasons = _coverage_reasons(snapshot) + ( + required_approvals, + pull_request_rule_present, + dismiss_stale_reviews, + require_last_push_approval, + required_checks, + rule_reasons, + ) = _rule_requirements(snapshot) + approvals, reviews, approval_reasons = _evaluate_reviews( + snapshot, + required_approvals=required_approvals, + pull_request_rule_present=pull_request_rule_present, + dismiss_stale_reviews=dismiss_stale_reviews, + require_last_push_approval=require_last_push_approval, + ) + checks, check_rows, check_reasons = _evaluate_checks(snapshot, required_checks) + state = _overall_state( + freshness=snapshot.provenance.freshness, + coverage_reasons=coverage_reasons, + rule_reasons=rule_reasons, + approval_state=str(approvals["state"]), + check_state=str(checks["state"]), + ) + reasons = sorted( + set( + coverage_reasons + + rule_reasons + + approval_reasons + + check_reasons + + ( + [f"snapshot_freshness_{snapshot.provenance.freshness}"] + if snapshot.provenance.freshness != "current" + else [] + ) + ) + ) + if state == "current": + claim = "supplied_required_evidence_is_current_for_supplied_head" + else: + claim = ( + "supplied_required_evidence_is_not_all_current_success_for_supplied_head" + ) + return { + "schema_version": VERDICT_SCHEMA_VERSION, + "state": state, + "current": state == "current", + "claim": claim, + "claim_ceiling": ( + "Evaluates only required evidence under the supplied rules, while classifying " + "all supplied evidence against the supplied head; does not determine GitHub " + "mergeability, code quality, reviewer effectiveness, CI correctness, or " + "branch safety." + ), + "repository": snapshot.pull_request.repository, + "pull_request_number": snapshot.pull_request.number, + "head_sha": snapshot.pull_request.head_sha, + "base_sha": snapshot.pull_request.base_sha, + "coverage": { + "state": "complete" + if not coverage_reasons and not rule_reasons + else "unknown", + "reasons": sorted(set(coverage_reasons + rule_reasons)), + }, + "requirements": { + "approvals": approvals, + "checks": checks, + }, + "reviews": reviews, + "checks": check_rows, + "reasons": reasons, + } + + +def invalid_snapshot_verdict( + reason: str, + errors: tuple[str, ...] | list[str], +) -> dict[str, object]: + """Build deterministic machine-readable output for invalid input.""" + + return { + "schema_version": VERDICT_SCHEMA_VERSION, + "state": "unknown", + "current": False, + "claim": "snapshot_not_evaluable", + "claim_ceiling": ( + "No evidence-to-head claim can be made because the supplied snapshot " + "was not a valid PRHeadEvidenceV1 document." + ), + "repository": None, + "pull_request_number": None, + "head_sha": None, + "base_sha": None, + "coverage": { + "state": "unknown", + "reasons": [reason], + }, + "requirements": { + "approvals": None, + "checks": None, + }, + "reviews": [], + "checks": [], + "reasons": [reason], + "validation_errors": sorted(set(errors)), + } diff --git a/tests/fixtures/pr_head_evidence/comment_only.json b/tests/fixtures/pr_head_evidence/comment_only.json new file mode 100644 index 0000000..c667f41 --- /dev/null +++ b/tests/fixtures/pr_head_evidence/comment_only.json @@ -0,0 +1,70 @@ +{ + "schema_version": "PRHeadEvidenceV1", + "pull_request": { + "repository": "octo/example", + "number": 42, + "author": "author", + "head_sha": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "base_sha": "cccccccccccccccccccccccccccccccccccccccc" + }, + "latest_reviewable_push": { + "pushed_at": "2026-07-28T10:00:00Z", + "actor": "bob" + }, + "reviews": [ + { + "id": 1, + "actor": { + "login": "alice", + "can_count": true + }, + "state": "COMMENTED", + "commit_id": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "submitted_at": "2026-07-28T10:05:00Z", + "dismissal": { + "dismissed": false, + "dismissed_at": null, + "prior_state": null + } + } + ], + "checks": [], + "rules": { + "source": "branch_protection", + "availability": "available", + "items": [ + { + "type": "pull_request", + "required_approving_review_count": 1, + "dismiss_stale_reviews": true, + "require_last_push_approval": false, + "require_code_owner_reviews": false + } + ] + }, + "coverage": { + "reviews": { + "permission": "granted", + "complete": true, + "truncated": false, + "pages_fetched": 1 + }, + "checks": { + "permission": "granted", + "complete": true, + "truncated": false, + "pages_fetched": 0 + }, + "rules": { + "permission": "granted", + "complete": true, + "truncated": false, + "pages_fetched": 1 + } + }, + "provenance": { + "source": "synthetic_fixture", + "captured_at": "2026-07-28T10:10:00Z", + "freshness": "current" + } +} diff --git a/tests/fixtures/pr_head_evidence/current.json b/tests/fixtures/pr_head_evidence/current.json new file mode 100644 index 0000000..4fae099 --- /dev/null +++ b/tests/fixtures/pr_head_evidence/current.json @@ -0,0 +1,70 @@ +{ + "schema_version": "PRHeadEvidenceV1", + "pull_request": { + "repository": "octo/example", + "number": 42, + "author": "author", + "head_sha": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "base_sha": "cccccccccccccccccccccccccccccccccccccccc" + }, + "latest_reviewable_push": { + "pushed_at": "2026-07-28T10:00:00Z", + "actor": "bob" + }, + "reviews": [ + { + "id": 1, + "actor": { + "login": "alice", + "can_count": true + }, + "state": "APPROVED", + "commit_id": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "submitted_at": "2026-07-28T10:05:00Z", + "dismissal": { + "dismissed": false, + "dismissed_at": null, + "prior_state": null + } + } + ], + "checks": [], + "rules": { + "source": "branch_protection", + "availability": "available", + "items": [ + { + "type": "pull_request", + "required_approving_review_count": 1, + "dismiss_stale_reviews": true, + "require_last_push_approval": false, + "require_code_owner_reviews": false + } + ] + }, + "coverage": { + "reviews": { + "permission": "granted", + "complete": true, + "truncated": false, + "pages_fetched": 1 + }, + "checks": { + "permission": "granted", + "complete": true, + "truncated": false, + "pages_fetched": 0 + }, + "rules": { + "permission": "granted", + "complete": true, + "truncated": false, + "pages_fetched": 1 + } + }, + "provenance": { + "source": "synthetic_fixture", + "captured_at": "2026-07-28T10:10:00Z", + "freshness": "current" + } +} diff --git a/tests/fixtures/pr_head_evidence/dismissed.json b/tests/fixtures/pr_head_evidence/dismissed.json new file mode 100644 index 0000000..5f24d77 --- /dev/null +++ b/tests/fixtures/pr_head_evidence/dismissed.json @@ -0,0 +1,70 @@ +{ + "schema_version": "PRHeadEvidenceV1", + "pull_request": { + "repository": "octo/example", + "number": 42, + "author": "author", + "head_sha": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "base_sha": "cccccccccccccccccccccccccccccccccccccccc" + }, + "latest_reviewable_push": { + "pushed_at": "2026-07-28T10:00:00Z", + "actor": "bob" + }, + "reviews": [ + { + "id": 1, + "actor": { + "login": "alice", + "can_count": true + }, + "state": "DISMISSED", + "commit_id": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "submitted_at": "2026-07-28T10:05:00Z", + "dismissal": { + "dismissed": true, + "dismissed_at": "2026-07-28T10:06:00Z", + "prior_state": "APPROVED" + } + } + ], + "checks": [], + "rules": { + "source": "branch_protection", + "availability": "available", + "items": [ + { + "type": "pull_request", + "required_approving_review_count": 1, + "dismiss_stale_reviews": true, + "require_last_push_approval": false, + "require_code_owner_reviews": false + } + ] + }, + "coverage": { + "reviews": { + "permission": "granted", + "complete": true, + "truncated": false, + "pages_fetched": 1 + }, + "checks": { + "permission": "granted", + "complete": true, + "truncated": false, + "pages_fetched": 0 + }, + "rules": { + "permission": "granted", + "complete": true, + "truncated": false, + "pages_fetched": 1 + } + }, + "provenance": { + "source": "synthetic_fixture", + "captured_at": "2026-07-28T10:10:00Z", + "freshness": "current" + } +} diff --git a/tests/fixtures/pr_head_evidence/incomplete.json b/tests/fixtures/pr_head_evidence/incomplete.json new file mode 100644 index 0000000..348196c --- /dev/null +++ b/tests/fixtures/pr_head_evidence/incomplete.json @@ -0,0 +1,54 @@ +{ + "schema_version": "PRHeadEvidenceV1", + "pull_request": { + "repository": "octo/example", + "number": 42, + "author": "author", + "head_sha": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "base_sha": "cccccccccccccccccccccccccccccccccccccccc" + }, + "latest_reviewable_push": { + "pushed_at": "2026-07-28T10:00:00Z", + "actor": "bob" + }, + "reviews": [], + "checks": [], + "rules": { + "source": "branch_protection", + "availability": "available", + "items": [ + { + "type": "pull_request", + "required_approving_review_count": 1, + "dismiss_stale_reviews": true, + "require_last_push_approval": false, + "require_code_owner_reviews": false + } + ] + }, + "coverage": { + "reviews": { + "permission": "granted", + "complete": false, + "truncated": true, + "pages_fetched": 1 + }, + "checks": { + "permission": "granted", + "complete": true, + "truncated": false, + "pages_fetched": 0 + }, + "rules": { + "permission": "granted", + "complete": true, + "truncated": false, + "pages_fetched": 1 + } + }, + "provenance": { + "source": "synthetic_fixture", + "captured_at": "2026-07-28T10:10:00Z", + "freshness": "current" + } +} diff --git a/tests/fixtures/pr_head_evidence/malformed.json b/tests/fixtures/pr_head_evidence/malformed.json new file mode 100644 index 0000000..1cd2e09 --- /dev/null +++ b/tests/fixtures/pr_head_evidence/malformed.json @@ -0,0 +1,13 @@ +{ + "schema_version": "PRHeadEvidenceV1", + "pull_request": { + "repository": "octo/example", + "number": 42, + "head_sha": "not-a-full-sha" + }, + "reviews": [ + { + "state": "APPROVED" + } + ] +} diff --git a/tests/fixtures/pr_head_evidence/semantic_triplets.json b/tests/fixtures/pr_head_evidence/semantic_triplets.json new file mode 100644 index 0000000..41572a4 --- /dev/null +++ b/tests/fixtures/pr_head_evidence/semantic_triplets.json @@ -0,0 +1,372 @@ +{ + "fixture_schema_version": "PRHeadEvidenceSemanticTripletsV1", + "base_snapshot": { + "schema_version": "PRHeadEvidenceV1", + "pull_request": { + "repository": "octo/example", + "number": 42, + "author": "author", + "head_sha": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "base_sha": "cccccccccccccccccccccccccccccccccccccccc" + }, + "latest_reviewable_push": { + "pushed_at": "2026-07-28T10:00:00Z", + "actor": "bob" + }, + "reviews": [ + { + "id": 1, + "actor": { + "login": "alice", + "can_count": true + }, + "state": "APPROVED", + "commit_id": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "submitted_at": "2026-07-28T10:05:00Z", + "dismissal": { + "dismissed": false, + "dismissed_at": null, + "prior_state": null + } + } + ], + "checks": [ + { + "id": 10, + "kind": "check_run", + "name": "ci/test", + "head_sha": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "status": "completed", + "conclusion": "success", + "integration_id": 99 + } + ], + "rules": { + "source": "branch_protection", + "availability": "available", + "items": [ + { + "type": "pull_request", + "required_approving_review_count": 1, + "dismiss_stale_reviews": true, + "require_last_push_approval": false, + "require_code_owner_reviews": false + }, + { + "type": "required_status_checks", + "strict_required_status_checks_policy": false, + "checks": [ + { + "context": "ci/test", + "integration_id": 99 + } + ] + } + ] + }, + "coverage": { + "reviews": { + "permission": "granted", + "complete": true, + "truncated": false, + "pages_fetched": 1 + }, + "checks": { + "permission": "granted", + "complete": true, + "truncated": false, + "pages_fetched": 1 + }, + "rules": { + "permission": "granted", + "complete": true, + "truncated": false, + "pages_fetched": 1 + } + }, + "provenance": { + "source": "synthetic_fixture", + "captured_at": "2026-07-28T10:10:00Z", + "freshness": "current" + } + }, + "triplets": [ + { + "name": "stale_review_sha", + "before_patch": {}, + "after_patch": { + "pull_request": { + "head_sha": "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb" + }, + "latest_reviewable_push": { + "pushed_at": "2026-07-28T11:00:00Z" + }, + "checks": [ + { + "id": 10, + "kind": "check_run", + "name": "ci/test", + "head_sha": "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", + "status": "completed", + "conclusion": "success", + "integration_id": 99 + } + ] + }, + "expected": { + "before_state": "current", + "after_state": "stale", + "after_reasons": [ + "required_approval_only_on_older_sha" + ] + } + }, + { + "name": "old_check_sha", + "before_patch": {}, + "after_patch": { + "pull_request": { + "head_sha": "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb" + }, + "latest_reviewable_push": { + "pushed_at": "2026-07-28T11:00:00Z" + }, + "reviews": [ + { + "id": 2, + "actor": { + "login": "alice", + "can_count": true + }, + "state": "APPROVED", + "commit_id": "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", + "submitted_at": "2026-07-28T11:05:00Z", + "dismissal": { + "dismissed": false, + "dismissed_at": null, + "prior_state": null + } + } + ] + }, + "expected": { + "before_state": "current", + "after_state": "stale", + "after_reasons": [ + "required_check_only_on_older_sha" + ] + } + }, + { + "name": "dismissed_approval", + "before_patch": {}, + "after_patch": { + "reviews": [ + { + "id": 1, + "actor": { + "login": "alice", + "can_count": true + }, + "state": "DISMISSED", + "commit_id": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "submitted_at": "2026-07-28T10:05:00Z", + "dismissal": { + "dismissed": true, + "dismissed_at": "2026-07-28T10:06:00Z", + "prior_state": "APPROVED" + } + } + ] + }, + "expected": { + "before_state": "current", + "after_state": "missing", + "after_reasons": [ + "required_approval_missing" + ] + } + }, + { + "name": "comment_vs_approval", + "before_patch": { + "reviews": [ + { + "id": 1, + "actor": { + "login": "alice", + "can_count": true + }, + "state": "COMMENTED", + "commit_id": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "submitted_at": "2026-07-28T10:05:00Z", + "dismissal": { + "dismissed": false, + "dismissed_at": null, + "prior_state": null + } + } + ] + }, + "after_patch": {}, + "expected": { + "before_state": "missing", + "after_state": "current" + } + }, + { + "name": "latest_push_changes", + "before_patch": {}, + "after_patch": { + "latest_reviewable_push": { + "actor": "alice" + }, + "rules": { + "items": [ + { + "type": "pull_request", + "required_approving_review_count": 1, + "dismiss_stale_reviews": true, + "require_last_push_approval": true, + "require_code_owner_reviews": false + }, + { + "type": "required_status_checks", + "strict_required_status_checks_policy": false, + "checks": [ + { + "context": "ci/test", + "integration_id": 99 + } + ] + } + ] + } + }, + "expected": { + "before_state": "current", + "after_state": "missing", + "after_reasons": [ + "latest_push_approval_missing" + ] + } + }, + { + "name": "missing_rules", + "before_patch": {}, + "after_patch": { + "rules": { + "availability": "missing", + "items": [] + }, + "coverage": { + "rules": { + "complete": false, + "pages_fetched": 0 + } + } + }, + "expected": { + "before_state": "current", + "after_state": "unknown", + "after_reasons": [ + "rules_missing" + ] + } + }, + { + "name": "incomplete_pagination", + "before_patch": {}, + "after_patch": { + "coverage": { + "reviews": { + "complete": false, + "truncated": true + } + } + }, + "expected": { + "before_state": "current", + "after_state": "unknown", + "after_reasons": [ + "reviews_pagination_truncated" + ] + } + }, + { + "name": "valid_approval_check_sets", + "before_patch": {}, + "after_patch": { + "reviews": [ + { + "id": 1, + "actor": { + "login": "alice", + "can_count": true + }, + "state": "APPROVED", + "commit_id": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "submitted_at": "2026-07-28T10:05:00Z", + "dismissal": { + "dismissed": false, + "dismissed_at": null, + "prior_state": null + } + }, + { + "id": 2, + "actor": { + "login": "carol", + "can_count": true + }, + "state": "COMMENTED", + "commit_id": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "submitted_at": "2026-07-28T10:06:00Z", + "dismissal": { + "dismissed": false, + "dismissed_at": null, + "prior_state": null + } + } + ], + "checks": [ + { + "id": 10, + "kind": "check_run", + "name": "ci/test", + "head_sha": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "status": "completed", + "conclusion": "success", + "integration_id": 99 + }, + { + "id": 11, + "kind": "check_suite", + "name": "ci-suite", + "head_sha": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "status": "completed", + "conclusion": "success", + "integration_id": 99 + } + ] + }, + "expected": { + "before_state": "current", + "after_state": "current" + } + }, + { + "name": "benign_head_stable_refresh", + "before_patch": {}, + "after_patch": { + "provenance": { + "captured_at": "2026-07-28T10:20:00Z" + } + }, + "expected": { + "before_state": "current", + "after_state": "current" + } + } + ] +} diff --git a/tests/fixtures/pr_head_evidence/stale.json b/tests/fixtures/pr_head_evidence/stale.json new file mode 100644 index 0000000..f9a453d --- /dev/null +++ b/tests/fixtures/pr_head_evidence/stale.json @@ -0,0 +1,70 @@ +{ + "schema_version": "PRHeadEvidenceV1", + "pull_request": { + "repository": "octo/example", + "number": 42, + "author": "author", + "head_sha": "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", + "base_sha": "cccccccccccccccccccccccccccccccccccccccc" + }, + "latest_reviewable_push": { + "pushed_at": "2026-07-28T11:00:00Z", + "actor": "bob" + }, + "reviews": [ + { + "id": 1, + "actor": { + "login": "alice", + "can_count": true + }, + "state": "APPROVED", + "commit_id": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "submitted_at": "2026-07-28T10:05:00Z", + "dismissal": { + "dismissed": false, + "dismissed_at": null, + "prior_state": null + } + } + ], + "checks": [], + "rules": { + "source": "branch_protection", + "availability": "available", + "items": [ + { + "type": "pull_request", + "required_approving_review_count": 1, + "dismiss_stale_reviews": true, + "require_last_push_approval": false, + "require_code_owner_reviews": false + } + ] + }, + "coverage": { + "reviews": { + "permission": "granted", + "complete": true, + "truncated": false, + "pages_fetched": 1 + }, + "checks": { + "permission": "granted", + "complete": true, + "truncated": false, + "pages_fetched": 0 + }, + "rules": { + "permission": "granted", + "complete": true, + "truncated": false, + "pages_fetched": 1 + } + }, + "provenance": { + "source": "synthetic_fixture", + "captured_at": "2026-07-28T11:10:00Z", + "freshness": "current" + } +} diff --git a/tests/test_pr_head_evidence.py b/tests/test_pr_head_evidence.py new file mode 100644 index 0000000..3a4dc5b --- /dev/null +++ b/tests/test_pr_head_evidence.py @@ -0,0 +1,575 @@ +from __future__ import annotations + +import copy +import json +from pathlib import Path +from typing import cast + +import pytest + +from src.pr_head_evidence import ( + SnapshotValidationError, + evaluate_snapshot, + parse_snapshot, +) + +HEAD_A = "a" * 40 +HEAD_B = "b" * 40 +BASE = "c" * 40 + + +def _review( + *, + review_id: int = 1, + actor: str = "alice", + state: str = "APPROVED", + commit_id: str = HEAD_A, + submitted_at: str | None = "2026-07-28T10:05:00Z", + can_count: bool | None = True, +) -> dict[str, object]: + dismissed = state == "DISMISSED" + return { + "id": review_id, + "actor": {"login": actor, "can_count": can_count}, + "state": state, + "commit_id": commit_id, + "submitted_at": submitted_at, + "dismissal": { + "dismissed": dismissed, + "dismissed_at": "2026-07-28T10:06:00Z" if dismissed else None, + "prior_state": "APPROVED" if dismissed else None, + }, + } + + +def _check( + *, + check_id: int = 10, + kind: str = "check_run", + name: str = "ci/test", + head_sha: str = HEAD_A, + status: str = "completed", + conclusion: str | None = "success", + integration_id: int | None = 99, +) -> dict[str, object]: + return { + "id": check_id, + "kind": kind, + "name": name, + "head_sha": head_sha, + "status": status, + "conclusion": conclusion, + "integration_id": integration_id, + } + + +def _snapshot( + *, + head_sha: str = HEAD_A, + reviews: list[dict[str, object]] | None = None, + checks: list[dict[str, object]] | None = None, + required_approvals: int = 1, + dismiss_stale_reviews: bool = True, + require_last_push_approval: bool = False, + rule_items: list[dict[str, object]] | None = None, +) -> dict[str, object]: + if reviews is None: + reviews = [_review(commit_id=head_sha)] + if checks is None: + checks = [_check(head_sha=head_sha)] + if rule_items is None: + rule_items = [ + { + "type": "pull_request", + "required_approving_review_count": required_approvals, + "dismiss_stale_reviews": dismiss_stale_reviews, + "require_last_push_approval": require_last_push_approval, + "require_code_owner_reviews": False, + }, + { + "type": "required_status_checks", + "strict_required_status_checks_policy": False, + "checks": [{"context": "ci/test", "integration_id": 99}], + }, + ] + return { + "schema_version": "PRHeadEvidenceV1", + "pull_request": { + "repository": "octo/example", + "number": 42, + "author": "author", + "head_sha": head_sha, + "base_sha": BASE, + }, + "latest_reviewable_push": { + "pushed_at": "2026-07-28T10:00:00Z", + "actor": "bob", + }, + "reviews": reviews, + "checks": checks, + "rules": { + "source": "branch_protection", + "availability": "available", + "items": rule_items, + }, + "coverage": { + "reviews": { + "permission": "granted", + "complete": True, + "truncated": False, + "pages_fetched": 1, + }, + "checks": { + "permission": "granted", + "complete": True, + "truncated": False, + "pages_fetched": 1, + }, + "rules": { + "permission": "granted", + "complete": True, + "truncated": False, + "pages_fetched": 1, + }, + }, + "provenance": { + "source": "synthetic_fixture", + "captured_at": "2026-07-28T10:10:00Z", + "freshness": "current", + }, + } + + +def _evaluate(value: dict[str, object]) -> dict[str, object]: + return evaluate_snapshot(parse_snapshot(value)) + + +def _requirements(verdict: dict[str, object]) -> dict[str, object]: + return cast(dict[str, object], verdict["requirements"]) + + +def _approval_result(verdict: dict[str, object]) -> dict[str, object]: + return cast(dict[str, object], _requirements(verdict)["approvals"]) + + +def _check_result(verdict: dict[str, object]) -> dict[str, object]: + return cast(dict[str, object], _requirements(verdict)["checks"]) + + +def test_valid_approval_and_check_set_is_current() -> None: + verdict = _evaluate(_snapshot()) + + assert verdict["schema_version"] == "PRHeadEvidenceVerdictV1" + assert verdict["state"] == "current" + assert verdict["current"] is True + assert verdict["head_sha"] == HEAD_A + assert _approval_result(verdict)["head_bound_count"] == 1 + assert _check_result(verdict)["all_required_checks_successful"] is True + + +def test_head_advance_invalidates_only_sha_bound_evidence_and_preserves_history() -> ( + None +): + original_snapshot = _snapshot() + original_verdict = _evaluate(original_snapshot) + advanced = copy.deepcopy(original_snapshot) + cast(dict[str, object], advanced["pull_request"])["head_sha"] = HEAD_B + + advanced_verdict = _evaluate(advanced) + + assert original_verdict["state"] == "current" + assert original_verdict["head_sha"] == HEAD_A + assert advanced_verdict["state"] == "stale" + review_rows = cast(list[dict[str, object]], advanced_verdict["reviews"]) + check_rows = cast(list[dict[str, object]], advanced_verdict["checks"]) + assert review_rows[0]["commit_id"] == HEAD_A + assert review_rows[0]["binding_state"] == "stale" + assert check_rows[0]["head_sha"] == HEAD_A + assert check_rows[0]["binding_state"] == "stale" + assert original_verdict == _evaluate(original_snapshot) + + +def test_old_approval_can_be_retained_by_rule_without_becoming_head_bound() -> None: + snapshot = _snapshot( + head_sha=HEAD_B, + reviews=[_review(commit_id=HEAD_A)], + checks=[_check(head_sha=HEAD_B)], + dismiss_stale_reviews=False, + ) + + verdict = _evaluate(snapshot) + approvals = _approval_result(verdict) + review_rows = cast(list[dict[str, object]], verdict["reviews"]) + + assert verdict["state"] == "stale" + assert approvals["github_requirement_satisfied"] is True + assert approvals["retained_stale_count"] == 1 + assert review_rows[0]["binding_state"] == "stale" + assert review_rows[0]["counts_toward_required_approvals"] is True + assert ( + "approval_retained_by_supplied_rule_but_not_head_bound" + in review_rows[0]["reasons"] + ) + + +def test_commented_review_never_counts_as_approval() -> None: + verdict = _evaluate(_snapshot(reviews=[_review(state="COMMENTED")])) + row = cast(list[dict[str, object]], verdict["reviews"])[0] + + assert verdict["state"] == "missing" + assert row["decision_state"] == "commented" + assert row["counts_toward_required_approvals"] is False + assert "commented_is_not_approval" in row["reasons"] + + +def test_pending_review_accepts_githubs_absent_submitted_at() -> None: + verdict = _evaluate( + _snapshot(reviews=[_review(state="PENDING", submitted_at=None)]) + ) + row = cast(list[dict[str, object]], verdict["reviews"])[0] + + assert verdict["state"] == "pending" + assert row["decision_state"] == "pending" + assert row["submitted_at"] is None + assert row["counts_toward_required_approvals"] is False + + +def test_dismissed_approval_preserves_dismissal_and_does_not_count() -> None: + verdict = _evaluate(_snapshot(reviews=[_review(state="DISMISSED")])) + row = cast(list[dict[str, object]], verdict["reviews"])[0] + + assert verdict["state"] == "missing" + assert row["review_state"] == "DISMISSED" + assert row["dismissed"] is True + assert row["dismissed_prior_state"] == "APPROVED" + assert row["counts_toward_required_approvals"] is False + + +def test_latest_push_approval_requires_other_actor_after_push() -> None: + same_actor = _snapshot( + reviews=[_review(actor="bob")], + require_last_push_approval=True, + ) + missing_verdict = _evaluate(same_actor) + + other_actor = copy.deepcopy(same_actor) + cast(list[dict[str, object]], other_actor["reviews"]).append( + _review( + review_id=2, + actor="carol", + submitted_at="2026-07-28T10:07:00Z", + ) + ) + current_verdict = _evaluate(other_actor) + + assert missing_verdict["state"] == "missing" + assert "latest_push_approval_missing" in missing_verdict["reasons"] + assert current_verdict["state"] == "current" + assert _approval_result(current_verdict)["latest_push_approval_actor"] == "carol" + + +def test_latest_decisive_review_orders_timezone_offsets_by_instant() -> None: + verdict = _evaluate( + _snapshot( + reviews=[ + _review( + review_id=1, + state="APPROVED", + submitted_at="2026-07-28T10:05:00Z", + ), + _review( + review_id=2, + state="CHANGES_REQUESTED", + submitted_at="2026-07-28T09:30:00-01:00", + ), + ] + ) + ) + + assert verdict["state"] == "blocked" + assert "changes_requested" in verdict["reasons"] + assert _approval_result(verdict)["counted_count"] == 0 + + +def test_latest_push_approval_actor_orders_timezone_offsets_by_instant() -> None: + verdict = _evaluate( + _snapshot( + reviews=[ + _review( + review_id=1, + actor="alice", + submitted_at="2026-07-28T10:05:00Z", + ), + _review( + review_id=2, + actor="carol", + submitted_at="2026-07-28T09:30:00-01:00", + ), + ], + require_last_push_approval=True, + ) + ) + + assert verdict["state"] == "current" + assert _approval_result(verdict)["latest_push_approval_actor"] == "carol" + + +def test_actor_logins_are_case_insensitive_for_counts_and_latest_push() -> None: + duplicate_case = _snapshot( + reviews=[ + _review(review_id=1, actor="Alice"), + _review(review_id=2, actor="alice"), + ], + required_approvals=2, + ) + duplicate_verdict = _evaluate(duplicate_case) + + assert duplicate_verdict["state"] == "missing" + assert _approval_result(duplicate_verdict)["counted_count"] == 1 + + same_pusher = _snapshot( + reviews=[_review(actor="Alice")], + require_last_push_approval=True, + ) + cast(dict[str, object], same_pusher["latest_reviewable_push"])["actor"] = "alice" + pusher_verdict = _evaluate(same_pusher) + + assert pusher_verdict["state"] == "missing" + assert _approval_result(pusher_verdict)["latest_push_approval_actor"] is None + + +def test_pull_request_author_cannot_count_own_approval() -> None: + snapshot = _snapshot(reviews=[_review(actor="Alice", can_count=True)]) + cast(dict[str, object], snapshot["pull_request"])["author"] = "alice" + + verdict = _evaluate(snapshot) + row = cast(list[dict[str, object]], verdict["reviews"])[0] + + assert verdict["state"] == "missing" + assert _approval_result(verdict)["counted_count"] == 0 + assert row["eligibility_state"] == "ineligible" + assert row["counts_toward_required_approvals"] is False + assert "pull_request_author_cannot_approve" in row["reasons"] + + +@pytest.mark.parametrize( + ("status", "conclusion", "expected_state", "github_satisfied"), + [ + ("completed", "neutral", "neutral", True), + ("completed", "skipped", "skipped", True), + ("completed", "cancelled", "blocked", False), + ("completed", "failure", "blocked", False), + ("queued", None, "pending", False), + ], +) +def test_non_success_check_states_remain_distinct( + status: str, + conclusion: str | None, + expected_state: str, + github_satisfied: bool, +) -> None: + verdict = _evaluate( + _snapshot(checks=[_check(status=status, conclusion=conclusion)]) + ) + required = cast( + list[dict[str, object]], + _check_result(verdict)["required"], + )[0] + + assert verdict["state"] == expected_state + assert required["state"] in { + "neutral", + "skipped", + "cancelled", + "blocked", + "pending", + } + assert required["github_requirement_satisfied"] is github_satisfied + assert verdict["current"] is False + + +def test_missing_required_check_is_not_green() -> None: + verdict = _evaluate(_snapshot(checks=[])) + required = cast( + list[dict[str, object]], + _check_result(verdict)["required"], + )[0] + + assert verdict["state"] == "missing" + assert required["state"] == "missing" + assert required["evidence_sha"] is None + + +def test_old_check_success_is_stale_not_current_success() -> None: + verdict = _evaluate( + _snapshot( + head_sha=HEAD_B, + reviews=[_review(commit_id=HEAD_B)], + checks=[_check(head_sha=HEAD_A)], + ) + ) + required = cast( + list[dict[str, object]], + _check_result(verdict)["required"], + )[0] + + assert verdict["state"] == "stale" + assert required["state"] == "stale" + assert required["evidence_sha"] == HEAD_A + + +def test_check_suite_is_bound_to_head_but_does_not_masquerade_as_required_check() -> ( + None +): + verdict = _evaluate( + _snapshot( + checks=[ + _check( + kind="check_suite", + name="ci/test", + head_sha=HEAD_A, + ) + ] + ) + ) + row = cast(list[dict[str, object]], verdict["checks"])[0] + + assert verdict["state"] == "missing" + assert row["kind"] == "check_suite" + assert row["binding_state"] == "current" + assert row["required"] is False + + +@pytest.mark.parametrize( + ("collection", "updates", "reason"), + [ + ( + "reviews", + {"complete": False, "truncated": True}, + "reviews_pagination_truncated", + ), + ( + "checks", + {"permission": "inaccessible", "complete": False}, + "checks_permission_inaccessible", + ), + ( + "rules", + {"permission": "missing", "complete": False}, + "rules_permission_missing", + ), + ], +) +def test_incomplete_or_inaccessible_coverage_is_unknown( + collection: str, + updates: dict[str, object], + reason: str, +) -> None: + snapshot = _snapshot() + coverage = cast(dict[str, dict[str, object]], snapshot["coverage"]) + coverage[collection].update(updates) + + verdict = _evaluate(snapshot) + + assert verdict["state"] == "unknown" + assert reason in verdict["reasons"] + + +def test_missing_protection_data_is_unknown_without_inventing_defaults() -> None: + snapshot = _snapshot() + rules = cast(dict[str, object], snapshot["rules"]) + rules["availability"] = "missing" + rules["items"] = [] + coverage = cast(dict[str, dict[str, object]], snapshot["coverage"]) + coverage["rules"].update({"complete": False, "pages_fetched": 0}) + + verdict = _evaluate(snapshot) + + assert verdict["state"] == "unknown" + assert "rules_missing" in verdict["reasons"] + assert _approval_result(verdict)["required_count"] == 0 + + +def test_unknown_rule_type_forces_unknown_coverage() -> None: + snapshot = _snapshot() + cast(dict[str, object], snapshot["rules"])["items"] = [ + {"type": "required_magic_attestation"} + ] + + verdict = _evaluate(snapshot) + + assert verdict["state"] == "unknown" + assert "unknown_rule_type:required_magic_attestation" in verdict["reasons"] + + +def test_malformed_review_commit_id_is_rejected() -> None: + snapshot = _snapshot() + cast(list[dict[str, object]], snapshot["reviews"])[0].pop("commit_id") + + with pytest.raises(SnapshotValidationError) as exc: + parse_snapshot(snapshot) + + assert any("reviews[0].commit_id" in error for error in exc.value.errors) + + +def test_unknown_input_field_is_rejected_instead_of_ignored() -> None: + snapshot = _snapshot() + cast(dict[str, object], snapshot["pull_request"])["headSHa"] = HEAD_A + + with pytest.raises(SnapshotValidationError) as exc: + parse_snapshot(snapshot) + + assert "pull_request.headSHa: unknown field" in exc.value.errors + + +def test_repeat_run_json_is_byte_deterministic() -> None: + snapshot = parse_snapshot(_snapshot()) + + first = json.dumps(evaluate_snapshot(snapshot), indent=2, sort_keys=True) + "\n" + second = json.dumps(evaluate_snapshot(snapshot), indent=2, sort_keys=True) + "\n" + + assert first == second + + +def _deep_merge(target: dict[str, object], patch: dict[str, object]) -> None: + for key, value in patch.items(): + current = target.get(key) + if isinstance(current, dict) and isinstance(value, dict): + _deep_merge( + cast(dict[str, object], current), + cast(dict[str, object], value), + ) + else: + target[key] = copy.deepcopy(value) + + +def test_semantic_fixture_triplets_pin_expected_transitions() -> None: + fixture_path = ( + Path(__file__).parent + / "fixtures" + / "pr_head_evidence" + / "semantic_triplets.json" + ) + fixture_set = json.loads(fixture_path.read_text(encoding="utf-8")) + base = cast(dict[str, object], fixture_set["base_snapshot"]) + triplets = cast(list[dict[str, object]], fixture_set["triplets"]) + + assert len(triplets) == 9 + for triplet in triplets: + before = copy.deepcopy(base) + after = copy.deepcopy(base) + _deep_merge( + before, + cast(dict[str, object], triplet["before_patch"]), + ) + _deep_merge( + after, + cast(dict[str, object], triplet["after_patch"]), + ) + expected = cast(dict[str, object], triplet["expected"]) + before_verdict = _evaluate(before) + after_verdict = _evaluate(after) + assert before_verdict["state"] == expected["before_state"], triplet["name"] + assert after_verdict["state"] == expected["after_state"], triplet["name"] + for reason in cast(list[str], expected.get("after_reasons", [])): + assert reason in after_verdict["reasons"], triplet["name"] diff --git a/tests/test_pr_head_evidence_cli.py b/tests/test_pr_head_evidence_cli.py new file mode 100644 index 0000000..dbf793c --- /dev/null +++ b/tests/test_pr_head_evidence_cli.py @@ -0,0 +1,122 @@ +from __future__ import annotations + +import json +import sys +from pathlib import Path +from types import SimpleNamespace + +import pytest + +import src.cli as cli_module +from src.app.pr_head_evidence import run_pr_head_evidence_mode + +FIXTURES = Path(__file__).parent / "fixtures" / "pr_head_evidence" + + +def _run_fixture(name: str, capsys: pytest.CaptureFixture[str]) -> dict[str, object]: + run_pr_head_evidence_mode(SimpleNamespace(snapshot=str(FIXTURES / name))) + return json.loads(capsys.readouterr().out) + + +def test_pr_evidence_parser_is_local_and_minimal() -> None: + args = cli_module.build_pr_head_evidence_parser().parse_args( + [str(FIXTURES / "current.json")] + ) + + assert args.snapshot.endswith("current.json") + assert not hasattr(args, "token") + assert not hasattr(args, "output_dir") + + +def test_current_fixture_exits_zero_and_emits_json( + capsys: pytest.CaptureFixture[str], +) -> None: + verdict = _run_fixture("current.json", capsys) + + assert verdict["state"] == "current" + assert verdict["current"] is True + + +@pytest.mark.parametrize( + ("fixture", "state"), + [ + ("stale.json", "stale"), + ("comment_only.json", "missing"), + ("dismissed.json", "missing"), + ], +) +def test_non_current_fixture_exits_one( + fixture: str, + state: str, + capsys: pytest.CaptureFixture[str], +) -> None: + with pytest.raises(SystemExit) as exc: + run_pr_head_evidence_mode(SimpleNamespace(snapshot=str(FIXTURES / fixture))) + + verdict = json.loads(capsys.readouterr().out) + assert exc.value.code == 1 + assert verdict["state"] == state + assert verdict["current"] is False + + +def test_incomplete_fixture_exits_two_as_unknown( + capsys: pytest.CaptureFixture[str], +) -> None: + with pytest.raises(SystemExit) as exc: + run_pr_head_evidence_mode( + SimpleNamespace(snapshot=str(FIXTURES / "incomplete.json")) + ) + + verdict = json.loads(capsys.readouterr().out) + assert exc.value.code == 2 + assert verdict["state"] == "unknown" + assert "reviews_pagination_truncated" in verdict["reasons"] + + +def test_malformed_fixture_exits_two_with_versioned_error( + capsys: pytest.CaptureFixture[str], +) -> None: + with pytest.raises(SystemExit) as exc: + run_pr_head_evidence_mode( + SimpleNamespace(snapshot=str(FIXTURES / "malformed.json")) + ) + + verdict = json.loads(capsys.readouterr().out) + assert exc.value.code == 2 + assert verdict["schema_version"] == "PRHeadEvidenceVerdictV1" + assert verdict["state"] == "unknown" + assert verdict["reasons"] == ["snapshot_malformed"] + assert verdict["validation_errors"] + + +def test_main_intercepts_pr_evidence_before_auth_token_lookup( + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], +) -> None: + def _unexpected_auth_lookup() -> str: + raise AssertionError("pr-evidence must not consult GitHub auth") + + monkeypatch.setattr(cli_module, "_gh_auth_token", _unexpected_auth_lookup) + monkeypatch.setattr( + sys, + "argv", + ["audit", "pr-evidence", str(FIXTURES / "current.json")], + ) + + cli_module.main() + + verdict = json.loads(capsys.readouterr().out) + assert verdict["state"] == "current" + + +def test_cli_does_not_modify_snapshot_or_fixture_directory( + capsys: pytest.CaptureFixture[str], +) -> None: + snapshot = FIXTURES / "current.json" + before_bytes = snapshot.read_bytes() + before_names = sorted(path.name for path in FIXTURES.iterdir()) + + _run_fixture("current.json", capsys) + + assert snapshot.read_bytes() == before_bytes + assert sorted(path.name for path in FIXTURES.iterdir()) == before_names