Measure a review against a severity threshold like ci review --fail-on

README.md

@@ -20,6 +20,29 @@ with ProoflineClient(os.environ["PROOFLINE_TOKEN"]) as client:
 
 When no token is passed, the clients read `PROOFLINE_TOKEN` from the environment, keep the token in memory, and redact it from diagnostic output. `AsyncProoflineClient` exposes the same methods as coroutines that you call with `await`. The [SDK guide](https://proofline.sh/docs/sdks/) and [API reference](https://proofline.sh/docs/api/) cover every operation.
 
+## Failing a job on open findings
+
+`ReviewOutcome.gate(fail_on)` measures a review against a `WARNING` or `SERIOUS` threshold with the rules of `proofline ci review --fail-on`, so a Python pipeline can fail the same way the CLI does.
+
+```python
+import sys
+
+outcome = client.wait_for_review(organization_id, review_id)
+result = outcome.gate(fail_on="SERIOUS")
+if result != "NOT_REACHED":
+    print(f"Proofline: {result}, {outcome.review_url}")
+    sys.exit({"REACHED": 3, "NOT_COMPLETE": 2, "UNAVAILABLE": 4}[result])
+```
+
+- `REACHED`: the completed review has an open finding at or above the threshold.
+- `NOT_REACHED`: the completed review has none.
+- `NOT_COMPLETE`: the review is incomplete, failed, or stopped, whatever it found.
+- `UNAVAILABLE`: the wait timed out, or the completed review lacks the server's open-finding counts, so the threshold cannot be measured.
+
+The server counts open findings from their current decisions, so dismissed findings and unexpired risk acceptances do not count. A code-only finding counts while it is not `RESOLVED` and its complete CODE assessment is of the review's own commits. A severity the SDK does not know reaches every threshold. The exit codes above are the CLI's.
+
+The threshold is the caller's, as with `--fail-on`, and applies to any review. An organization that opts in to a merge gate instead gets the server's frozen `ALLOW` or `BLOCK` decision for a pull request head in `outcome.review.merge_gate`, which is null otherwise.
+
 ## Review exports
 
 `export_reviews` reads one CSV or NDJSON page over the bounded transport. Export tokens require `analytics:read`.

proofline/outcomes.py

@@ -3,15 +3,20 @@
 from __future__ import annotations
 
 from dataclasses import dataclass, field
-from typing import Literal
+from typing import Literal, cast
 from uuid import UUID
 
 from ._generated.models.customer_code_assessment import CustomerCodeAssessment
 from ._generated.models.customer_review_detail import CustomerReviewDetail
+from ._generated.models.open_items_view import OpenItemsView
 from ._generated.types import Unset
 from ._transport import ProoflineError
 
 OutcomeStatus = Literal["COMPLETED", "INCOMPLETE", "FAILED", "STOPPED", "TIMED_OUT"]
+SeverityThreshold = Literal["WARNING", "SERIOUS"]
+GateResult = Literal["REACHED", "NOT_REACHED", "NOT_COMPLETE", "UNAVAILABLE"]
+# Least severe first.
+SEVERITIES: tuple[SeverityThreshold, ...] = ("WARNING", "SERIOUS")
 COMPLETE_CHECKS = frozenset(("READY", "BLOCKED", "NOT_APPLICABLE"))
 COMPLETE_LIFECYCLES = frozenset(("GATING", "SHADOW", "QUARANTINED", "RETIRED"))
 _KNOWN_ASSESSMENTS = frozenset(
@@ -47,6 +52,57 @@ class ReviewOutcome:
     gaps: tuple[AssessmentGap, ...] = field(default=(), repr=False)
     reason: str | None = None
 
+    def gate(self, fail_on: SeverityThreshold) -> GateResult:
+        """Measure the review against a severity threshold as `proofline ci review --fail-on`.
+
+        Completeness decides first. A review that is incomplete, failed or stopped is
+        NOT_COMPLETE whatever it found, and a wait that timed out is UNAVAILABLE. A
+        completed review REACHES the threshold when the server counts an open finding at
+        or above it, or when a complete CODE assessment of the review's own commits holds
+        an unresolved finding at or above it. Dismissed findings and unexpired risk
+        acceptances are not open. A severity this SDK does not know reaches every
+        threshold. Without the server's open-finding counts the threshold cannot be
+        measured, which is UNAVAILABLE and never NOT_REACHED.
+        """
+        if fail_on not in SEVERITIES:
+            raise ValueError("fail_on must be WARNING or SERIOUS.")
+        if self.status == "TIMED_OUT":
+            return "UNAVAILABLE"
+        if self.status != "COMPLETED":
+            return "NOT_COMPLETE"
+        if self.review is None:
+            return "UNAVAILABLE"
+        counts = self.review.open_items
+        if not isinstance(counts, OpenItemsView):
+            return "UNAVAILABLE"
+        count = counts.serious_findings if fail_on == "SERIOUS" else counts.open_findings
+        if type(count) is not int or count < 0:
+            return "UNAVAILABLE"
+        # Complete environments complete a review even beside a partial CODE assessment,
+        # whose findings are not its whole result. CODE findings have no recorded
+        # decisions, so their lifecycle and severity decide.
+        code = _recorded_code(self.review)
+        findings = (
+            code.findings
+            if code is not None
+            and code.status == "COMPLETED"
+            and not text(code.incompleteness_kind)
+            else []
+        )
+        if count > 0 or any(
+            finding.lifecycle_state != "RESOLVED" and _reaches(finding.severity, fail_on)
+            for finding in findings
+        ):
+            return "REACHED"
+        return "NOT_REACHED"
+
+
+def _reaches(severity: object, threshold: SeverityThreshold) -> bool:
+    # An unknown severity, which a newer server can send, is never read as below a threshold.
+    if severity not in SEVERITIES:
+        return True
+    return SEVERITIES.index(cast(SeverityThreshold, severity)) >= SEVERITIES.index(threshold)
+
 
 def text(value: object) -> str | None:
     return value if isinstance(value, str) and value else None
@@ -68,6 +124,11 @@ def _code_assessment(review: CustomerReviewDetail) -> CustomerCodeAssessment | N
     # A CODE result cannot replace a missing ENV result or its check receipts.
     if review.assessment_count != 0 or review.assessments:
         return None
+    return _recorded_code(review)
+
+
+def _recorded_code(review: CustomerReviewDetail) -> CustomerCodeAssessment | None:
+    """The CODE record of this review's own commits with published findings, or None."""
     code = review.code_assessment
     if (
         not isinstance(code, CustomerCodeAssessment)

tests/test_gate.py

@@ -0,0 +1,177 @@
+"""The severity gate matches `proofline ci review --fail-on`."""
+
+from __future__ import annotations
+
+import inspect
+from typing import Any
+
+import pytest
+from pydantic import SecretStr
+
+from proofline import AsyncProoflineClient, CustomerReviewDetail, ProoflineClient
+from proofline.outcomes import ReviewOutcome, outcome
+
+from .fixtures import ORG, REVIEW, TOKEN, assessment, code_assessment, finding, review
+from .http_server import Reply, Server
+
+URL = "https://proofline.sh/orgs/by-id/org/reviews/review"
+
+
+def counts(open_findings: Any = 0, serious_findings: Any = 0) -> dict[str, Any]:
+    return {
+        "open_findings": open_findings,
+        "serious_findings": serious_findings,
+        "accepted_risks": 0,
+        "not_assessed": [],
+    }
+
+
+def code_finding(severity: str, lifecycle: str = "NEW") -> dict[str, Any]:
+    return {**finding(), "severity": severity, "lifecycle_state": lifecycle}
+
+
+def read(body: dict[str, Any], forced: Any = None) -> ReviewOutcome:
+    result = outcome(CustomerReviewDetail.from_dict(body), URL, forced)
+    assert result is not None
+    return result
+
+
+def env_review(open_items: Any, status: str = "COMPLETED", **extra: Any) -> dict[str, Any]:
+    return {**review(status), "open_items": open_items, **extra}
+
+
+def code_only(code: dict[str, Any], open_items: Any = None, **extra: Any) -> dict[str, Any]:
+    if open_items is None:
+        open_items = counts()
+    return {
+        **review(),
+        "assessment_count": 0,
+        "assessments": [],
+        "code_assessment": code,
+        "open_items": open_items,
+        **extra,
+    }
+
+
+def test_completeness_decides_before_the_open_finding_counts() -> None:
+    serious = counts(1, 1)
+    incomplete = {**review(), "assessments": [assessment("INCOMPLETE")], "open_items": serious}
+    assert read(env_review(serious)).gate("SERIOUS") == "REACHED"
+    assert read(incomplete).status == "INCOMPLETE"
+    assert read(incomplete).gate("SERIOUS") == "NOT_COMPLETE"
+    assert read(env_review(serious, "FAILED")).gate("SERIOUS") == "NOT_COMPLETE"
+    assert read(env_review(serious, "STOPPED")).gate("SERIOUS") == "NOT_COMPLETE"
+    assert read(env_review(serious), "TIMED_OUT").gate("SERIOUS") == "UNAVAILABLE"
+    assert ReviewOutcome("TIMED_OUT", URL).gate("WARNING") == "UNAVAILABLE"
+    # A gap decides before the missing counts do.
+    incomplete["open_items"] = None
+    assert read(incomplete).gate("SERIOUS") == "NOT_COMPLETE"
+
+
+@pytest.mark.parametrize(
+    ("open_items", "fail_on", "result"),
+    [
+        (counts(1, 1), "WARNING", "REACHED"),
+        (counts(1, 0), "WARNING", "REACHED"),
+        (counts(1, 0), "SERIOUS", "NOT_REACHED"),
+        (counts(0, 0), "WARNING", "NOT_REACHED"),
+        (counts(0, 0), "SERIOUS", "NOT_REACHED"),
+        # A threshold that cannot be measured is never NOT_REACHED.
+        (None, "WARNING", "UNAVAILABLE"),
+        (None, "SERIOUS", "UNAVAILABLE"),
+        ({"open_findings": 0}, "WARNING", "UNAVAILABLE"),
+        (counts("0", 0), "WARNING", "UNAVAILABLE"),
+        (counts(0, -1), "SERIOUS", "UNAVAILABLE"),
+        (counts(False, 0), "WARNING", "UNAVAILABLE"),
+    ],
+)
+def test_a_completed_review_is_measured_by_the_server_open_finding_counts(
+    open_items: Any, fail_on: Any, result: str
+) -> None:
+    assert read(env_review(open_items)).gate(fail_on) == result
+
+
+def test_missing_open_items_field_is_unavailable() -> None:
+    body = review()
+    assert "open_items" not in body
+    assert read(body).status == "COMPLETED"
+    assert read(body).gate("WARNING") == "UNAVAILABLE"
+
+
+@pytest.mark.parametrize(
+    ("severity", "lifecycle", "fail_on", "result"),
+    [
+        ("SERIOUS", "NEW", "SERIOUS", "REACHED"),
+        ("SERIOUS", "REINTRODUCED", "SERIOUS", "REACHED"),
+        ("SERIOUS", "UNCHANGED", "WARNING", "REACHED"),
+        ("WARNING", "NEW", "WARNING", "REACHED"),
+        ("WARNING", "NEW", "SERIOUS", "NOT_REACHED"),
+        ("SERIOUS", "RESOLVED", "SERIOUS", "NOT_REACHED"),
+        # A severity a newer server sends is never read as below a threshold.
+        ("CRITICAL", "NEW", "SERIOUS", "REACHED"),
+    ],
+)
+def test_complete_code_findings_reach_by_lifecycle_and_severity(
+    severity: str, lifecycle: str, fail_on: Any, result: str
+) -> None:
+    code = {**code_assessment(), "findings": [code_finding(severity, lifecycle)]}
+    assert read(env_review(counts(), code_assessment=code)).gate(fail_on) == result
+    assert read(code_only(code)).gate(fail_on) == result
+
+
+@pytest.mark.parametrize(
+    "code",
+    [
+        {**code_assessment("INCOMPLETE")},
+        {**code_assessment(), "incompleteness_kind": "MODEL_BUDGET_EXHAUSTED"},
+        {**code_assessment(), "head_commit_sha": "c" * 40},
+        {**code_assessment(), "base_commit_sha": "c" * 40},
+        {
+            **code_assessment(),
+            "findings": [
+                {**code_finding("WARNING"), "risk_domain": "[withheld: environment value]"}
+            ],
+        },
+    ],
+)
+def test_unusable_code_records_supply_no_findings_and_never_complete_a_code_only_review(
+    code: dict[str, Any],
+) -> None:
+    code = {**code, "findings": [*code["findings"], code_finding("SERIOUS")]}
+    recorded = {"head_commit_sha": "A" * 40, "base_commit_sha": "B" * 40}
+    # Complete environments complete a review even beside an unusable CODE record.
+    beside = env_review(counts(), code_assessment=code, **recorded)
+    assert read(beside).status == "COMPLETED"
+    assert read(beside).gate("SERIOUS") == "NOT_REACHED"
+    assert read(code_only(code, **recorded)).gate("SERIOUS") == "NOT_COMPLETE"
+
+
+def test_code_commits_are_compared_only_where_the_review_recorded_them() -> None:
+    code = {**code_assessment(), "findings": [code_finding("SERIOUS")]}
+    assert read(code_only(code)).gate("SERIOUS") == "REACHED"
+    assert read(code_only(code, head_commit_sha="A" * 40)).gate("SERIOUS") == "REACHED"
+    assert read(code_only(code, head_commit_sha=None)).gate("SERIOUS") == "REACHED"
+
+
+def test_unknown_threshold_is_refused() -> None:
+    threshold: Any = "HIGH"
+    with pytest.raises(ValueError):
+        read(env_review(counts())).gate(threshold)
+
+
+@pytest.mark.parametrize("mode", ["sync", "async"])
+async def test_wait_for_review_result_measures_the_threshold(mode: str) -> None:
+    body = env_review(counts(2, 1))
+    with Server(lambda request, index: Reply(body=body)) as server:
+        if mode == "sync":
+            with ProoflineClient(SecretStr(TOKEN), base_url=server.url) as client:
+                result = client.wait_for_review(ORG, REVIEW, timeout=5)
+        else:
+            async with AsyncProoflineClient(SecretStr(TOKEN), base_url=server.url) as client:
+                waited = client.wait_for_review(ORG, REVIEW, timeout=5)
+                assert inspect.isawaitable(waited)
+                result = await waited
+    assert result.status == "COMPLETED"
+    assert result.gate("SERIOUS") == "REACHED"
+    assert result.gate(fail_on="WARNING") == "REACHED"
+    assert len(server.requests) == 1