This is an automated email from the ASF dual-hosted git repository.
potiuk pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 2f7a7ea4be9 Show each fork CI job's status on gated pull requests
(#75616)
2f7a7ea4be9 is described below
commit 2f7a7ea4be9c5c48e45d2d45d14595f2e2e594d3
Author: Jarek Potiuk <[email protected]>
AuthorDate: Sun Oct 25 19:37:20 2026 +0100
Show each fork CI job's status on gated pull requests (#75616)
A gated pull request showed one aggregate fork CI status, so a maintainer
or contributor had to open the fork run to see which jobs were running,
which failed and which passed. Reporting every job as its own check, each
linking to the job in the fork, puts that on the pull request itself,
where project CI shows its jobs.
Generated-by: Claude Opus 5
---
.github/fork-ci-gate.toml | 5 +
dev/AGENTS.md | 17 +
.../doc/ci/08_fork_ci_for_external_contributors.md | 87 ++++-
.../src/airflow_breeze/utils/fork_ci_gate.py | 5 +
.../src/airflow_breeze/utils/fork_ci_messages.py | 36 ++
.../src/airflow_breeze/utils/fork_ci_reconcile.py | 159 ++++++++-
dev/breeze/tests/test_fork_ci_gate.py | 6 +
dev/breeze/tests/test_fork_ci_messages.py | 22 +-
dev/breeze/tests/test_fork_ci_reconcile.py | 362 ++++++++++++++++++++-
9 files changed, 690 insertions(+), 9 deletions(-)
diff --git a/.github/fork-ci-gate.toml b/.github/fork-ci-gate.toml
index c80593d221c..6fc7a0de1ce 100644
--- a/.github/fork-ci-gate.toml
+++ b/.github/fork-ci-gate.toml
@@ -29,6 +29,11 @@ enabled_since = 2026-01-01T00:00:00Z
# Days without any fork CI activity after which a gated pull request is closed.
close_after_days = 7
+# Report each job of a gated pull request's fork run as its own commit status.
Set to false when
+# the fork CI monitor exceeds GitHub's rate limits (see dev/AGENTS.md); the
`Fork CI / Tests (AMD)`
+# status is posted either way.
+job_statuses = true
+
# Organisations exempted under AIP-120 §2.2. Keys are display names used in
comments,
# values are GitHub logins. Changes go through a reviewed pull request.
[allowlist]
diff --git a/dev/AGENTS.md b/dev/AGENTS.md
index 947e0ac9e8e..313c7860183 100644
--- a/dev/AGENTS.md
+++ b/dev/AGENTS.md
@@ -7,6 +7,7 @@
- [dev/ directory guidelines](#dev-directory-guidelines)
- [Scripts](#scripts)
+ - [Fork CI job statuses and GitHub's rate
limits](#fork-ci-job-statuses-and-githubs-rate-limits)
<!-- END doctoc generated TOC please keep comment here to allow auto update -->
@@ -42,3 +43,19 @@ uv run dev/my_script.py [args...]
```
Document `uv run` (not `python`) as the invocation method in READMEs and
instructions.
+
+## Fork CI job statuses and GitHub's rate limits
+
+The fork CI monitor (`breeze ci fork-ci-reconcile`, run by
`.github/workflows/fork-ci-monitor.yml`)
+reports every job of a gated pull request's fork run as its own commit status.
Those writes are the
+largest share of what the monitor does with the repository's `GITHUB_TOKEN`,
whose rate limits every
+workflow of `apache/airflow` shares (see "Write budget" in
+`dev/breeze/doc/ci/08_fork_ci_for_external_contributors.md`).
+
+If the budget is exceeded, turn the job statuses off: set `job_statuses =
false` in
+`.github/fork-ci-gate.toml`. Signs that it is exceeded are the monitor's Slack
note that job
+statuses "did not fit in the budget" persisting across runs, rate-limit
retries ("GitHub rate limit,
+retrying") or "still rate limited" errors in the monitor's run log or its
Slack messages, or other
+workflows of the repository failing on secondary rate limits. The aggregate
`Fork CI / Tests (AMD)`
+status and `CI passed` are posted either way, so turning job statuses off
loses detail, not gating.
+Lowering `MAX_JOB_STATUSES_PER_RUN` in `fork_ci_reconcile.py` is the softer
alternative.
diff --git a/dev/breeze/doc/ci/08_fork_ci_for_external_contributors.md
b/dev/breeze/doc/ci/08_fork_ci_for_external_contributors.md
index 1b2bd742f90..799a233e9ca 100644
--- a/dev/breeze/doc/ci/08_fork_ci_for_external_contributors.md
+++ b/dev/breeze/doc/ci/08_fork_ci_for_external_contributors.md
@@ -661,6 +661,87 @@ Commit statuses were chosen over check runs because a
check run created with the
token is attached to the creating workflow's check suite and its details link
then opens the
monitor run rather than `details_url`. A status honours `target_url`.
+### Job statuses
+
+Next to the aggregate status, each job of the fork run is reported as its own
commit status, so
+the pull request shows which jobs run, failed or passed, each linking to the
job in the fork.
+The run is the one `Fork CI / Tests (AMD)` reports (its `databaseId` is
fetched with the check
+suites); its jobs come from `GET
/repos/<fork>/actions/runs/<id>/jobs?filter=latest`.
+
+- The context is `Fork CI job / <job name>`, with non-printable characters
replaced and the name
+ shortened to 100 characters. Job names come from the workflow of the
contributor's commit, so
+ the prefix is what keeps a job from posting `Fork CI / Tests (AMD)`, from
which the reconciler
+ reads its decisions back, or `CI passed`, which the merge ruleset requires.
Jobs that share a
+ context, which GitHub compares case-insensitively, are reported as the worst
of them.
+- A queued or running job is `pending` ("Running in the fork"), a successful
one `success`, and a
+ failed, timed-out, cancelled or unstartable one `failure`. Skipped and
neutral jobs, and the jobs
+ every run has (`Fork CI gate`, `Check if the run is needed`, `Platform: AMD`
and `CI passed`), are
+ not reported. Queued and running share one status so that starting a job
costs no write.
+- The desired statuses are compared with the head commit's current ones
+ (`GET /repos/apache/airflow/commits/<sha>/status`), and only differences are
posted, failures
+ first. GraphQL has no mutation for commit statuses (`createCheckRun` exists,
but see above for
+ why check runs do not fit), so each one is a REST call, and they count
towards GitHub's limits on
+ content-creating requests: a reconciler run posts at most
`MAX_JOB_STATUSES_PER_RUN` (60) of
+ them, each write charged as it is made. What does not fit is posted by the
next run, and once
+ the budget is spent no further pull request's jobs or statuses are read in
that run. A run that
+ defers job statuses adds a note to the Slack message, with fixed wording so
it is posted once
+ while deferrals persist; the run log lists the pull requests.
+- Commit statuses cannot be deleted and have no neutral state. When a run has
completed, a job
+ context posted for an earlier run of the same commit that the completed run
does not have, or
+ skipped, is overwritten with `success` and the description "Not in the
latest fork run - see
+ Fork CI / Tests (AMD)", so it no longer shows as running or failed.
+- Once every job of a completed run is reported, the run is recorded in the
marker as
+ `jobs: "<run id>:<attempt>:<conclusion>"` (`runAttempt` is fetched with the
check suites), so
+ later runs do not read its jobs again. A run in progress is read on every
run, and a re-run in
+ the fork is read again because its attempt changes, even when it ends with
the same conclusion.
+- Reporting job statuses is caught on its own: when it fails, the failure is
listed with the
+ unknown items for Slack, and the pull request is still classified, so for
example a `CI passed`
+ that no longer holds is still revoked.
+
+The job statuses are informational. Merging still depends on `CI passed`, and
the state machine
+on `Fork CI / Tests (AMD)`; a job status never changes either.
+
+`job_statuses = false` in `.github/fork-ci-gate.toml` turns them off; the
reconciler then posts
+only `Fork CI / Tests (AMD)` and `CI passed`, as before. Statuses already
posted stay on their
+commits. Turn them off when the fork CI monitor's runs hit GitHub's rate
limits (rate-limit
+retries in the run log, or "still rate limited" errors in Slack);
`dev/AGENTS.md` records this for
+agents.
+
+#### Write budget
+
+Measured on 2026-10-25 from `apache/airflow`'s own `Tests (AMD)` pull request
runs (the same
+workflow runs in the forks; `GET
/repos/apache/airflow/actions/runs/<id>/jobs?filter=latest`,
+jobs whose conclusion is not `skipped`, minus the four jobs that are not
reported):
+
+| Pull request | Run | Started jobs | Reported job statuses |
+|---|---|---|---|
+| Breeze test fix |
[40138784120](https://github.com/apache/airflow/actions/runs/40138784120) | 11
| 7 |
+| Documentation fix |
[40169750323](https://github.com/apache/airflow/actions/runs/40169750323) | 16
| 12 |
+| Google provider change |
[40158907296](https://github.com/apache/airflow/actions/runs/40158907296) | 43
| 39 |
+| CI change (full tests) |
[40207881565](https://github.com/apache/airflow/actions/runs/40207881565) | 131
| 127 |
+
+`apache/airflow` had 1898 `Tests (AMD)` pull request runs from 2026-10-18 to
2026-10-24, about 270 a
+day or 11 an hour. Each reported job costs at most two writes (running, then
its result) and one
+when it starts and finishes between two reconciler runs; a fork run of 30-60
minutes spans two to
+four of them, so most jobs cost two.
+
+The limits that matter are GitHub's documented ones for the `GITHUB_TOKEN` of
`apache/airflow`,
+shared by every workflow of the repository: 1000 REST requests an hour, and 80
content-creating
+requests a minute and 500 an hour. Assuming about 35 reported jobs for an
average push:
+
+- **Contributor pilot**, a few gated pushes a day: about 70 writes a push,
under 20 an hour, under
+ 5% of the content-creating limit.
+- **Gate enabled**, assuming about half of the pull request runs come from
gated pull requests
+ (about 5-6 pushes an hour): about 400 writes an hour without a cap, 80% of
the content-creating
+ limit and more at peaks. The cap of 60 writes a run, four runs an hour,
holds job statuses to at
+ most 240 an hour: 48% of the content-creating limit and 24% of the REST
limit, plus two reads per
+ pull request with a run in progress. Job statuses then lag behind at busy
times, the aggregate
+ status does not.
+
+These are estimates: the share of gated runs, the jobs per push and the pushes
per pull request
+are assumptions, not measurements. Watch the monitor's call counts in its run
log after enabling
+the gate, and turn `job_statuses` off if they approach the limits.
+
### Relationship to the triage skill
The `pr-management-triage` skill described in
`contributing-docs/25_maintainer_pr_triage.md`
@@ -760,7 +841,9 @@ Per run, with ~700 gated pull requests: ~28 GraphQL calls
for pass 2, ~14 for fo
calls for the recent `pull_request` runs, one REST call per base branch for
its workflow
files, a jobs call and a compare call only per long `ci-amd.yml` run of a
gated pull request,
one changed-files call per head commit that turns green, plus one mutation per
actual state
-change. Well inside
+change. Job statuses add a jobs call and a status call per pull request with a
fork run in
+progress or a completed run not yet reported, and up to two status mutations
per job, at most
+`MAX_JOB_STATUSES_PER_RUN` per run; see "Write budget". Well inside
the `GITHUB_TOKEN` limits at a 15-minute cadence. The reconciler logs its call
counts so the
budget stays observable.
@@ -906,6 +989,8 @@ as a failed check rather than a crash.
workflow files on the base branch.
- A green fork run is trusted only for pull requests that do not change what
fork CI runs;
see "Trusting a green fork run".
+- Job statuses carry names from the contributor's workflow under their own
`Fork CI job / <name>`
+ prefix, so they cannot post the reconciler's own contexts or `CI passed`;
see "Job statuses".
- The reconciler runs on `main` with `pull-requests: write` and never checks
out or executes
pull request content. Its inputs are API data; its comment templates are
static.
- Fork runs get the fork's `GITHUB_TOKEN` only. Nothing in project CI trusts a
fork run's
diff --git a/dev/breeze/src/airflow_breeze/utils/fork_ci_gate.py
b/dev/breeze/src/airflow_breeze/utils/fork_ci_gate.py
index 80c3632ce77..8ce90a34d97 100644
--- a/dev/breeze/src/airflow_breeze/utils/fork_ci_gate.py
+++ b/dev/breeze/src/airflow_breeze/utils/fork_ci_gate.py
@@ -128,6 +128,7 @@ class GateConfig:
close_after_days: int = DEFAULT_CLOSE_AFTER_DAYS
allowlist: Mapping[str, tuple[str, ...]] = field(default_factory=dict)
pilot: Mapping[str, datetime] = field(default_factory=dict)
+ job_statuses: bool = True
def is_allowlisted(self, login: str) -> bool:
wanted = login.casefold()
@@ -173,12 +174,16 @@ def parse_config(text: str) -> GateConfig:
login.casefold(): _require_aware_datetime(joined_at, f"pilot entry
{login!r}")
for login, joined_at in data.get("pilot", {}).items()
}
+ job_statuses = data.get("job_statuses", True)
+ if not isinstance(job_statuses, bool):
+ raise ValueError(f"job_statuses must be true or false, got
{job_statuses!r}")
return GateConfig(
enabled=enabled,
enabled_since=enabled_since,
close_after_days=close_after_days,
allowlist=allowlist,
pilot=pilot,
+ job_statuses=job_statuses,
)
diff --git a/dev/breeze/src/airflow_breeze/utils/fork_ci_messages.py
b/dev/breeze/src/airflow_breeze/utils/fork_ci_messages.py
index d28e5f54283..e00048df167 100644
--- a/dev/breeze/src/airflow_breeze/utils/fork_ci_messages.py
+++ b/dev/breeze/src/airflow_breeze/utils/fork_ci_messages.py
@@ -30,6 +30,10 @@ from dataclasses import dataclass
from datetime import datetime
STATUS_CONTEXT = "Fork CI / Tests (AMD)"
+# Each job of the fork run is reported under this prefix. Job names come from
the contributor's
+# workflow, so the prefix keeps them from posting STATUS_CONTEXT or
CI_PASSED_CONTEXT.
+JOB_STATUS_CONTEXT_PREFIX = "Fork CI job / "
+JOB_NAME_MAX_LENGTH = 100
# The check the "Require CI on main" ruleset requires before a pull request
can (auto-)merge. Project
# CI's gate job reports it; for a gated pull request, which never runs project
CI, the reconciler
# reports it from the fork run.
@@ -74,12 +78,16 @@ class Marker:
since: datetime
# Commits the head commit was behind its base branch when its fork CI
turned green.
behind: int | None = None
+ # The completed fork run (``<run id>:<attempt>:<conclusion>``) whose jobs
are all reported.
+ jobs: str | None = None
def render_marker(marker: Marker) -> str:
data: dict[str, object] = {"state": marker.state, "sha": marker.sha,
"since": marker.since.isoformat()}
if marker.behind is not None:
data["behind"] = marker.behind
+ if marker.jobs is not None:
+ data["jobs"] = marker.jobs
return f"{MARKER_PREFIX}{json.dumps(data, sort_keys=True)} -->"
@@ -95,6 +103,7 @@ def parse_marker(body: str) -> Marker | None:
sha=data["sha"],
since=datetime.fromisoformat(data["since"]),
behind=data.get("behind"),
+ jobs=data.get("jobs"),
)
except (json.JSONDecodeError, KeyError, TypeError, ValueError) as e:
raise ValueError(f"Malformed fork CI marker: {match.group(0)!r}") from
e
@@ -217,6 +226,33 @@ _STATUS_DESCRIPTIONS = {
"ungated": "Project CI in use (base branch without fork CI) - see the
checks",
}
+# (commit status state, description) of a fork job, by its status and
conclusion. A queued job is
+# reported like a running one: each change is one more write, and commit
statuses can only be posted
+# one at a time (GraphQL has no mutation for them).
+JOB_STATUSES = {
+ "running": ("pending", "Running in the fork - see Details"),
+ "success": ("success", "Passed in the fork - see Details"),
+ "failure": ("failure", "Failed in the fork - see Details"),
+ "timed_out": ("failure", "Timed out in the fork - see Details"),
+ "startup_failure": ("failure", "Failed to start in the fork - see
Details"),
+ "cancelled": ("failure", "Cancelled in the fork - see Details"),
+}
+
+
+# Commit statuses cannot be deleted and have no neutral state. A job reported
for an earlier run that
+# the latest completed run does not have is overwritten with this, so it no
longer shows as running
+# or failed; `Fork CI / Tests (AMD)` is what tells the result.
+STALE_JOB_STATUS = ("success", "Not in the latest fork run - see Fork CI /
Tests (AMD)")
+
+
+def job_status_context(job_name: str) -> str:
+ # A line break or control character would make an unreadable check name.
+ name = "".join(char if char.isprintable() else " " for char in
job_name).strip()
+ if len(name) > JOB_NAME_MAX_LENGTH:
+ name = f"{name[: JOB_NAME_MAX_LENGTH - 1]}…"
+ return f"{JOB_STATUS_CONTEXT_PREFIX}{name}"
+
+
STATUS_STATES = {
"setup_required": "error",
"awaiting_push": "pending",
diff --git a/dev/breeze/src/airflow_breeze/utils/fork_ci_reconcile.py
b/dev/breeze/src/airflow_breeze/utils/fork_ci_reconcile.py
index 41ffaab6f26..6b54f6c7fac 100644
--- a/dev/breeze/src/airflow_breeze/utils/fork_ci_reconcile.py
+++ b/dev/breeze/src/airflow_breeze/utils/fork_ci_reconcile.py
@@ -57,11 +57,15 @@ from airflow_breeze.utils.fork_ci_messages import (
CI_PASSED_CONTEXT,
CI_PASSED_DESCRIPTION,
CI_PASSED_REVOKED_DESCRIPTION,
+ JOB_STATUS_CONTEXT_PREFIX,
+ JOB_STATUSES,
+ STALE_JOB_STATUS,
STATUS_CONTEXT,
STATUS_STATES,
Marker,
fork_actions_url,
inactivity_close_comment,
+ job_status_context,
nudge_comment,
parse_marker,
released_comment,
@@ -122,6 +126,22 @@ SLACK_ITEM_LIMIT = 500
# can change what they run. pull_request_target and issue_comment use the base
branch's files.
MERGE_REF_EVENTS = ("pull_request", "pull_request_review",
"pull_request_review_comment")
+# Jobs of every fork run whose outcome the `Fork CI / Tests (AMD)` status
already shows.
+UNREPORTED_JOB_NAMES = GATED_RUN_JOB_NAMES | {GATE_JOB_NAME, "Platform: AMD"}
+RUNNING_JOB_STATUSES = frozenset({"queued", "waiting", "pending", "requested",
"in_progress"})
+JOB_STATE_RANKS = {"failure": 0, "pending": 1, "success": 2}
+# Commit statuses count towards GitHub's limit of 500 content-creating
requests per hour, shared by
+# every workflow of the repository. At four runs an hour this keeps job
statuses to at most about
+# half of it; the rest follow on the next run, failures first. See "Job
statuses" in the design doc.
+MAX_JOB_STATUSES_PER_RUN = 60
+# Fixed text, so the note is posted to Slack once while deferrals persist; the
run log lists the pull
+# requests.
+JOB_STATUSES_DEFERRED_NOTE = (
+ f"job statuses did not fit in the budget of {MAX_JOB_STATUSES_PER_RUN}
writes per run and are "
+ "posted by later runs; if this persists, set `job_statuses = false` in
.github/fork-ci-gate.toml "
+ "(see dev/AGENTS.md)"
+)
+
RUNNING_STATUSES = frozenset({"QUEUED", "IN_PROGRESS", "PENDING", "WAITING",
"REQUESTED"})
FAILED_CONCLUSIONS = frozenset({"FAILURE", "CANCELLED", "TIMED_OUT",
"STARTUP_FAILURE", "ACTION_REQUIRED"})
SETUP_WORKFLOW_STATES = frozenset({"disabled_fork", "disabled_manually",
"missing"})
@@ -212,6 +232,8 @@ class ForkObservation:
suite_status: str | None = None
suite_conclusion: str | None = None
run_url: str | None = None
+ run_id: int | None = None
+ run_attempt: int | None = None
workflow_state: str | None = None
error: str | None = None
@@ -429,6 +451,7 @@ def classify(
find_bypass_run: Callable[[], str | None],
changes_ci_files: Callable[[], bool | None],
count_behind: Callable[[], int | None] = lambda: None,
+ jobs_reported: str | None = None,
) -> Plan:
"""
Decide the mutations for one gated pull request.
@@ -443,6 +466,8 @@ def classify(
commit the answer may not belong to.
:param count_behind: how many commits the head commit is behind the base
branch, None when
unknown; called at the same time as ``changes_ci_files``.
+ :param jobs_reported: the completed fork run whose jobs are all reported
as statuses, recorded
+ in the marker so that later runs do not read its jobs again.
"""
plan = Plan()
labels = set(pr.labels)
@@ -598,7 +623,7 @@ def classify(
plan,
pr,
state_comment(
- Marker(state.value, pr.head_sha, since, behind=behind if green
else None),
+ Marker(state.value, pr.head_sha, since, behind=behind if green
else None, jobs=jobs_reported),
fork=pr.head_repo,
run_url=fork.run_url,
base_ref=pr.base_ref,
@@ -888,7 +913,7 @@ def fetch_fork_observations(
fields.append(
f"pr{index}: repository(owner: $o{index}, name: $n{index}) {{
object(oid: $s{index}) {{ "
"... on Commit { checkSuites(first: 30) { nodes { createdAt
status conclusion "
- "workflowRun { url event workflow { resourcePath } } } } } } }"
+ "workflowRun { databaseId runAttempt url event workflow {
resourcePath } } } } } } }"
)
variables.update({f"o{index}": owner, f"n{index}": name,
f"s{index}": pr.head_sha})
query = f"query({', '.join(declarations)}) {{ {' '.join(fields)} }}"
@@ -916,10 +941,113 @@ def fetch_fork_observations(
suite_status=latest["status"],
suite_conclusion=latest["conclusion"],
run_url=latest["workflowRun"]["url"],
+ run_id=latest["workflowRun"].get("databaseId"),
+ run_attempt=latest["workflowRun"].get("runAttempt"),
)
return observations
+def build_job_statuses(jobs: Iterable[dict[str, Any]]) -> dict[str, tuple[str,
str, str]]:
+ """The ``(state, target URL, description)`` to report for each queued or
started fork job, by context."""
+ statuses: dict[str, tuple[str, str, str]] = {}
+ # GitHub compares contexts case-insensitively, so differently capitalised
names are one context.
+ contexts: dict[str, str] = {}
+ for job in jobs:
+ if job["name"] in UNREPORTED_JOB_NAMES:
+ continue
+ if job.get("status") in RUNNING_JOB_STATUSES:
+ outcome = "running"
+ else:
+ outcome = job.get("conclusion") or ""
+ if outcome not in JOB_STATUSES:
+ # Skipped and neutral jobs did no work worth a check on the pull
request.
+ continue
+ state, description = JOB_STATUSES[outcome]
+ context = contexts.setdefault(
+ job_status_context(job["name"]).casefold(),
job_status_context(job["name"])
+ )
+ # Jobs can share a name, or two names can be the same once shortened;
the worst one counts.
+ reported = statuses.get(context)
+ if reported is None or JOB_STATE_RANKS[state] <
JOB_STATE_RANKS[reported[0]]:
+ statuses[context] = (state, job["html_url"], description)
+ return statuses
+
+
+class JobStatusReporter:
+ """Reports each job of a pull request's fork run as a commit status,
within a per-run budget."""
+
+ def __init__(self, client: GitHubClient, *, repo: str, budget: int =
MAX_JOB_STATUSES_PER_RUN):
+ self._client = client
+ self._repo = repo
+ self.remaining = budget
+ self.deferred: list[int] = []
+
+ def report(self, pr: PullRequestSnapshot, fork: ForkObservation) -> str |
None:
+ """
+ Post the job statuses that changed; return the completed run's key
once all are reported.
+
+ The key is recorded in the marker, so a completed run's jobs are read
once. A run in
+ progress, or one whose statuses did not all fit in the budget, is read
again next time. The
+ key holds the run attempt, so a re-run in the fork is read again even
when it ends with the
+ same conclusion.
+ """
+ if fork.run_id is None or not pr.head_repo:
+ return None
+ key = (
+ f"{fork.run_id}:{fork.run_attempt}:{fork.suite_conclusion}"
+ if fork.suite_status == "COMPLETED"
+ else None
+ )
+ recorded = pr.marker.jobs if pr.marker is not None and pr.marker.sha
== pr.head_sha else None
+ if key is not None and key == recorded:
+ return key
+ # Without writes left, the reads are spent for nothing: a backlog of
hundreds of pull requests
+ # would otherwise cost two reads each on every run.
+ if self.remaining == 0:
+ self.deferred.append(pr.number)
+ return None
+ jobs = self._client.rest_get(
+
f"repos/{pr.head_repo}/actions/runs/{fork.run_id}/jobs?filter=latest&per_page=100",
+ paginate=True,
+ items_key="jobs",
+ )
+ # By the case-folded context, as GitHub compares them.
+ current = {
+ status["context"].casefold(): (
+ status["context"],
+ (status["state"], status.get("target_url") or "",
status.get("description") or ""),
+ )
+ for status in self._client.rest_get(
+
f"repos/{self._repo}/commits/{pr.head_sha}/status?per_page=100",
+ paginate=True,
+ items_key="statuses",
+ )
+ }
+ desired = build_job_statuses(jobs)
+ if key is not None:
+ # Jobs of an earlier run of this commit that the completed run did
not run, or skipped.
+ wanted = {context.casefold() for context in desired}
+ for folded, (context, (_, target_url, _)) in current.items():
+ if context.startswith(JOB_STATUS_CONTEXT_PREFIX) and folded
not in wanted:
+ desired[context] = (STALE_JOB_STATUS[0], target_url,
STALE_JOB_STATUS[1])
+ changes = sorted(
+ (
+ (context, status)
+ for context, status in desired.items()
+ if current.get(context.casefold(), ("", None))[1] != status
+ ),
+ key=lambda change: (JOB_STATE_RANKS[change[1][0]], change[0]),
+ )
+ for context, (state, target_url, description) in changes:
+ if self.remaining == 0:
+ self.deferred.append(pr.number)
+ return None
+ self._client.set_status(pr.number, pr.head_sha, state, target_url,
description, context=context)
+ # Charged per write, so writes made before a later one fails still
count.
+ self.remaining -= 1
+ return key
+
+
def fetch_workflow_state(client: GitHubClient, fork: str) -> str:
try:
return
client.rest_get(f"repos/{fork}/actions/workflows/{CI_WORKFLOW_FILE}")["state"]
@@ -1135,8 +1263,14 @@ def find_state_comment(client: GitHubClient, repo: str,
pr: PullRequestSnapshot)
def read_head_status(client: GitHubClient, repo: str, pr: PullRequestSnapshot)
-> PullRequestSnapshot:
"""Read the reconciler's statuses from the head commit, for a snapshot
that could not."""
- combined =
client.rest_get(f"repos/{repo}/commits/{pr.head_sha}/status?per_page=100")
- statuses = {status["context"]: status for status in
combined.get("statuses") or []}
+ # Every page: with job statuses a commit can carry more than 100 contexts,
and a missed `CI passed`
+ # success would not be revoked.
+ statuses = {
+ status["context"]: status
+ for status in client.rest_get(
+ f"repos/{repo}/commits/{pr.head_sha}/status?per_page=100",
paginate=True, items_key="statuses"
+ )
+ }
status = statuses.get(STATUS_CONTEXT)
ci_passed = statuses.get(CI_PASSED_CONTEXT)
return replace(
@@ -1359,6 +1493,7 @@ def run_reconcile(
bypass_finder = BypassFinder(
client, repo=repo, now=now,
base_triggers=load_workflow_triggers(workflows_dir)
)
+ job_reporter = JobStatusReporter(client, repo=repo)
for listed in prs:
try:
@@ -1373,6 +1508,15 @@ def run_reconcile(
and needs_workflow_state(pr.marker)
):
fork =
ForkObservation(workflow_state=fetch_workflow_state(client, pr.head_repo))
+ jobs_reported = None
+ if config.job_statuses:
+ # Job statuses are informational; failing to post them must
not keep the pull request
+ # from being reconciled, such as a `CI passed` that no longer
holds being revoked.
+ try:
+ jobs_reported = job_reporter.report(pr, fork)
+ except Exception as e:
+ console_print(f"[warning]#{pr.number}: reporting job
statuses failed: {e}[/]")
+ unknown.append((pr.number, f"reporting job statuses
failed: {type(e).__name__}: {e}"))
plan = classify(
pr,
@@ -1385,6 +1529,7 @@ def run_reconcile(
pull_request_changes_ci_files, client, repo, pr.number,
pr.head_sha
),
count_behind=functools.partial(count_commits_behind, client,
repo, pr),
+ jobs_reported=jobs_reported,
)
draft_change = apply_plan(client, pr, plan)
if draft_change is not None:
@@ -1402,6 +1547,12 @@ def run_reconcile(
unknown.append((listed.number, f"{type(e).__name__}: {e}"))
notes = [bypass_finder.incomplete] if bypass_finder.incomplete else []
+ if job_reporter.deferred:
+ console_print(
+ f"[info]Job statuses of {len(job_reporter.deferred)} pull requests
left for the next run: "
+ f"{job_reporter.deferred}[/]"
+ )
+ notes.append(JOB_STATUSES_DEFERRED_NOTE)
# The full list for the run log; Slack may show only part of a very long
one.
for number, reason in [*unknown, *attention]:
console_print(f"[warning]#{number}: {reason}[/]")
diff --git a/dev/breeze/tests/test_fork_ci_gate.py
b/dev/breeze/tests/test_fork_ci_gate.py
index 186dd395673..b52a0d29a1c 100644
--- a/dev/breeze/tests/test_fork_ci_gate.py
+++ b/dev/breeze/tests/test_fork_ci_gate.py
@@ -320,6 +320,11 @@ def test_parse_config_defaults():
assert config.close_after_days == 7
assert config.allowlist == {}
assert config.pilot == {}
+ assert config.job_statuses is True
+
+
+def test_parse_config_turns_job_statuses_off():
+ assert parse_config("job_statuses = false\n").job_statuses is False
def test_parse_config_lowercases_pilot_logins():
@@ -334,6 +339,7 @@ def test_parse_config_lowercases_pilot_logins():
("text", "message"),
[
pytest.param("enabled = 1\n", "enabled must be",
id="enabled-not-bool"),
+ pytest.param('job_statuses = "no"\n', "job_statuses must be",
id="job-statuses-not-bool"),
pytest.param(
"enabled_since = 2026-01-01T00:00:00\n", "enabled_since must be",
id="enabled-since-naive"
),
diff --git a/dev/breeze/tests/test_fork_ci_messages.py
b/dev/breeze/tests/test_fork_ci_messages.py
index ab21a0c277e..659c22d8ff0 100644
--- a/dev/breeze/tests/test_fork_ci_messages.py
+++ b/dev/breeze/tests/test_fork_ci_messages.py
@@ -28,8 +28,11 @@ from airflow_breeze.utils.fork_ci_messages import (
CI_PASSED_DESCRIPTION,
CI_PASSED_REVOKED_DESCRIPTION,
FOOTER,
+ JOB_NAME_MAX_LENGTH,
+ JOB_STATUS_CONTEXT_PREFIX,
Marker,
inactivity_close_comment,
+ job_status_context,
nudge_comment,
parse_marker,
released_comment,
@@ -43,7 +46,11 @@ MARKER = Marker(state="awaiting_push", sha="abc123",
since=datetime(2026, 10, 15
@pytest.mark.parametrize(
"marker",
- [pytest.param(MARKER, id="without-behind"), pytest.param(replace(MARKER,
behind=12), id="behind")],
+ [
+ pytest.param(MARKER, id="without-behind"),
+ pytest.param(replace(MARKER, behind=12), id="behind"),
+ pytest.param(replace(MARKER, jobs="123:SUCCESS"), id="jobs"),
+ ],
)
def test_marker_round_trip(marker):
assert parse_marker(f"text before\n{render_marker(marker)}\ntext after")
== marker
@@ -51,6 +58,7 @@ def test_marker_round_trip(marker):
def test_marker_without_behind_keeps_its_rendering():
assert "behind" not in render_marker(MARKER)
+ assert "jobs" not in render_marker(MARKER)
@pytest.mark.parametrize(
@@ -123,3 +131,15 @@ def test_setup_texts_give_the_exact_command(text):
def test_status_descriptions_fit_github_limit():
descriptions = [*_STATUS_DESCRIPTIONS.values(), CI_PASSED_DESCRIPTION,
CI_PASSED_REVOKED_DESCRIPTION]
assert all(len(description) < 140 for description in descriptions)
+
+
[email protected](
+ ("name", "expected"),
+ [
+ ("Static checks", "Static checks"),
+ ("a\nb\tc", "a b c"),
+ ("x" * (JOB_NAME_MAX_LENGTH + 5), "x" * (JOB_NAME_MAX_LENGTH - 1) +
"…"),
+ ],
+)
+def test_job_status_context(name, expected):
+ assert job_status_context(name) == f"{JOB_STATUS_CONTEXT_PREFIX}{expected}"
diff --git a/dev/breeze/tests/test_fork_ci_reconcile.py
b/dev/breeze/tests/test_fork_ci_reconcile.py
index ec8828ac9b7..9f3aa61f751 100644
--- a/dev/breeze/tests/test_fork_ci_reconcile.py
+++ b/dev/breeze/tests/test_fork_ci_reconcile.py
@@ -30,20 +30,25 @@ from airflow_breeze.utils.fork_ci_gate import GateConfig
from airflow_breeze.utils.fork_ci_github import GitHubClient, GitHubError
from airflow_breeze.utils.fork_ci_messages import (
CI_PASSED_CONTEXT,
+ JOB_STATUS_CONTEXT_PREFIX,
+ STALE_JOB_STATUS,
STATUS_CONTEXT,
Marker,
parse_marker,
render_marker,
)
from airflow_breeze.utils.fork_ci_reconcile import (
+ JOB_STATUSES_DEFERRED_NOTE,
BaseGateConfigs,
BypassFinder,
DraftChange,
ForkObservation,
ForkState,
+ JobStatusReporter,
Plan,
PullRequestSnapshot,
apply_plan,
+ build_job_statuses,
build_slack_payload,
changes_ci_definition,
classify,
@@ -1226,15 +1231,26 @@ def
test_parse_pull_request_leaves_statuses_of_another_commit_unread():
id="both-statuses",
),
pytest.param([], None, None, id="no-statuses"),
+ pytest.param(
+ [
+ *({"context": f"{JOB_STATUS_CONTEXT_PREFIX}job {i}", "state":
"success"} for i in range(150)),
+ {"context": CI_PASSED_CONTEXT, "state": "success"},
+ ],
+ None,
+ "success",
+ id="ci-passed-after-many-job-statuses",
+ ),
],
)
def test_read_head_status(client, statuses, expected_status,
expected_ci_passed):
- client.rest_get.return_value = {"statuses": statuses}
+ client.rest_get.return_value = statuses
pr = dataclasses.replace(parse_pull_request(_pr_node(1)),
head_status_unread=True)
read = read_head_status(client, REPO, pr)
-
client.rest_get.assert_called_once_with(f"repos/{REPO}/commits/sha1/status?per_page=100")
+ client.rest_get.assert_called_once_with(
+ f"repos/{REPO}/commits/sha1/status?per_page=100", paginate=True,
items_key="statuses"
+ )
assert read.current_status == expected_status
assert read.ci_passed_state == expected_ci_passed
assert read.head_status_unread is False
@@ -1332,6 +1348,8 @@ def
test_fork_observations_use_the_fork_name_and_latest_suite(client):
"status": "COMPLETED",
"conclusion": "SUCCESS",
"workflowRun": {
+ "databaseId": 77,
+ "runAttempt": 2,
"url": "new",
"event": "push",
"workflow": {"resourcePath": "/x/ci-amd.yml"},
@@ -1366,7 +1384,7 @@ def
test_fork_observations_use_the_fork_name_and_latest_suite(client):
observations = fetch_fork_observations(client, [pr])
assert observations[1] == ForkObservation(
- suite_status="COMPLETED", suite_conclusion="SUCCESS", run_url="new"
+ suite_status="COMPLETED", suite_conclusion="SUCCESS", run_url="new",
run_id=77, run_attempt=2
)
assert client.graphql.call_args.args[1] == {"o0": "someone", "n0":
"my-airflow", "s0": "sha1"}
@@ -1717,3 +1735,341 @@ def test_count_commits_behind(client, side_effect,
expected):
assert count_commits_behind(client, REPO, _pr(base_ref="v3-1-test")) ==
expected
assert client.rest_get.call_args.args[0] ==
f"repos/{REPO}/compare/v3-1-test...someone:abc123?per_page=1"
+
+
+def _fork_job(name, status="completed", conclusion="success", url=None):
+ return {
+ "name": name,
+ "status": status,
+ "conclusion": conclusion,
+ "html_url": url or f"https://job/{name}";,
+ }
+
+
+def test_build_job_statuses_reports_each_started_or_queued_job():
+ jobs = [
+ _fork_job("Static checks"),
+ _fork_job("Postgres tests: core / A", conclusion="failure"),
+ _fork_job("Helm tests", status="in_progress", conclusion=None),
+ _fork_job("Kubernetes tests", status="queued", conclusion=None),
+ _fork_job("Special tests", conclusion="cancelled"),
+ _fork_job("Java SDK tests", conclusion="skipped"),
+ _fork_job("Fork CI gate"),
+ _fork_job("CI passed"),
+ _fork_job("Check if the run is needed"),
+ ]
+
+ statuses = build_job_statuses(jobs)
+
+ assert {
+ context.removeprefix(JOB_STATUS_CONTEXT_PREFIX): state for context,
(state, _, _) in statuses.items()
+ } == {
+ "Static checks": "success",
+ "Postgres tests: core / A": "failure",
+ "Helm tests": "pending",
+ "Kubernetes tests": "pending",
+ "Special tests": "failure",
+ }
+ assert statuses[f"{JOB_STATUS_CONTEXT_PREFIX}Static checks"][1] ==
"https://job/Static checks"
+
+
+def test_build_job_statuses_reports_the_worst_of_jobs_sharing_a_name():
+ jobs = [
+ _fork_job("Tests", url="ok"),
+ _fork_job("Tests", conclusion="failure", url="bad"),
+ _fork_job("Tests", url="ok2"),
+ ]
+
+ assert build_job_statuses(jobs)[f"{JOB_STATUS_CONTEXT_PREFIX}Tests"][:2]
== ("failure", "bad")
+
+
[email protected]("name", ["Tests (AMD)", "CI passed (fork)", "x\nCI
passed"])
+def test_build_job_statuses_cannot_post_the_reconciler_contexts(name):
+ (context,) = build_job_statuses([_fork_job(name)])
+
+ assert context.startswith(JOB_STATUS_CONTEXT_PREFIX)
+ assert context not in (STATUS_CONTEXT, CI_PASSED_CONTEXT)
+ assert "\n" not in context
+
+
+RUNNING_RUN = ForkObservation(suite_status="IN_PROGRESS", run_url=RUN_URL,
run_id=1)
+COMPLETED_RUN = ForkObservation(
+ suite_status="COMPLETED", suite_conclusion="FAILURE", run_url=RUN_URL,
run_id=1, run_attempt=1
+)
+
+
+def _reporter_client(client, jobs, current=()):
+ def rest_get(path, paginate=False, items_key=None):
+ if path.startswith("repos/someone/airflow/actions/runs/1/jobs"):
+ return jobs
+ if path.startswith(f"repos/{REPO}/commits/abc123/status"):
+ return list(current)
+ raise AssertionError(path)
+
+ client.rest_get.side_effect = rest_get
+ return client
+
+
+def test_job_status_reporter_posts_changed_statuses_failures_first(client):
+ already = {
+ "context": f"{JOB_STATUS_CONTEXT_PREFIX}Static checks",
+ "state": "success",
+ "target_url": "https://job/Static checks",
+ "description": "Passed in the fork - see Details",
+ }
+ _reporter_client(
+ client,
+ [_fork_job("Static checks"), _fork_job("Helm tests"),
_fork_job("Unit", conclusion="failure")],
+ [already],
+ )
+
+ key = JobStatusReporter(client, repo=REPO).report(_pr(), COMPLETED_RUN)
+
+ assert key == "1:1:FAILURE"
+ assert [(c.args[2], c.kwargs["context"]) for c in
client.set_status.call_args_list] == [
+ ("failure", f"{JOB_STATUS_CONTEXT_PREFIX}Unit"),
+ ("success", f"{JOB_STATUS_CONTEXT_PREFIX}Helm tests"),
+ ]
+
+
+def
test_job_status_reporter_leaves_what_does_not_fit_in_the_budget_for_the_next_run(client):
+ _reporter_client(client, [_fork_job("A", conclusion="failure"),
_fork_job("B"), _fork_job("C")])
+ reporter = JobStatusReporter(client, repo=REPO, budget=2)
+
+ assert reporter.report(_pr(), COMPLETED_RUN) is None
+ assert [c.kwargs["context"] for c in client.set_status.call_args_list] == [
+ f"{JOB_STATUS_CONTEXT_PREFIX}A",
+ f"{JOB_STATUS_CONTEXT_PREFIX}B",
+ ]
+ assert reporter.remaining == 0
+ assert reporter.deferred == [42]
+
+
+def test_job_status_reporter_does_not_record_a_run_in_progress(client):
+ _reporter_client(client, [_fork_job("A", status="in_progress",
conclusion=None)])
+
+ assert JobStatusReporter(client, repo=REPO).report(_pr(), RUNNING_RUN) is
None
+ client.set_status.assert_called_once()
+
+
[email protected](
+ ("marker_jobs", "marker_sha", "reads"),
+ [
+ pytest.param("1:1:FAILURE", "abc123", False, id="already-reported"),
+ pytest.param("1:1:SUCCESS", "abc123", True,
id="re-run-with-another-conclusion"),
+ pytest.param("1:2:FAILURE", "abc123", True,
id="another-attempt-with-the-same-conclusion"),
+ pytest.param("1:1:FAILURE", "other", True,
id="marker-of-another-commit"),
+ ],
+)
+def test_job_status_reporter_reads_a_completed_run_once(client, marker_jobs,
marker_sha, reads):
+ _reporter_client(client, [])
+ pr = _pr(marker=dataclasses.replace(_marker("failed", sha=marker_sha),
jobs=marker_jobs))
+
+ assert JobStatusReporter(client, repo=REPO).report(pr, COMPLETED_RUN) ==
"1:1:FAILURE"
+ assert client.rest_get.called is reads
+
+
[email protected](
+ "fork",
+ [ForkObservation(), ForkObservation(error="fork not readable"),
ForkObservation(workflow_state="active")],
+)
+def test_job_status_reporter_without_a_fork_run(client, fork):
+ assert JobStatusReporter(client, repo=REPO).report(_pr(), fork) is None
+ client.rest_get.assert_not_called()
+
+
+def test_classify_records_the_reported_run_in_the_marker():
+ plan = classify(
+ _pr(marker=_marker("running")),
+ dataclasses.replace(COMPLETED_RUN),
+ config=CONFIG,
+ now=NOW,
+ repo=REPO,
+ find_bypass_run=mock.Mock(return_value=None),
+ changes_ci_files=mock.Mock(return_value=False),
+ jobs_reported="1:1:FAILURE",
+ )
+
+ assert parse_marker(plan.marker_comment).jobs == "1:1:FAILURE"
+
+
[email protected]("airflow_breeze.utils.fork_ci_reconcile.JobStatusReporter",
autospec=True)
[email protected]("airflow_breeze.utils.fork_ci_reconcile.classify", autospec=True)
+def test_run_reconcile_reports_job_statuses_before_classifying(
+ mock_classify, mock_reporter, client, tmp_path
+):
+ mock_classify.return_value = Plan(state=ForkState.RUNNING)
+ mock_reporter.return_value.report.return_value = "7:SUCCESS"
+ mock_reporter.return_value.deferred = []
+ client.graphql.side_effect = [
+ _search_result(),
+ _search_result(),
+ _search_result(_pr_node(1, headRepository={"nameWithOwner":
"someone/airflow"})),
+ {"pr0": {"object": None}},
+ ]
+ client.rest_get_pages.return_value = [{"total_count": 0, "workflow_runs":
[]}]
+ client.rest_get.return_value = {"state": "active"}
+
+ run_reconcile(
+ client,
+ CONFIG,
+ now=NOW,
+ run_url="https://run";,
+ output_file=tmp_path / "s.json",
+ workflows_dir=tmp_path,
+ )
+
+ (pr, fork), _ = mock_reporter.return_value.report.call_args
+ assert pr.number == 1
+ assert mock_classify.call_args.kwargs["jobs_reported"] == "7:SUCCESS"
+
+
+def test_build_job_statuses_reports_a_queued_job_like_a_running_one():
+ queued = build_job_statuses([_fork_job("A", status="queued",
conclusion=None, url="u")])
+ running = build_job_statuses([_fork_job("A", status="in_progress",
conclusion=None, url="u")])
+
+ assert queued == running
+
+
+def _stale(state):
+ return {
+ "context": f"{JOB_STATUS_CONTEXT_PREFIX}Gone",
+ "state": state,
+ "target_url": "https://job/Gone";,
+ "description": "Running in the fork - see Details",
+ }
+
+
[email protected](
+ ("fork", "resolved"),
+ [pytest.param(COMPLETED_RUN, True, id="completed"),
pytest.param(RUNNING_RUN, False, id="running")],
+)
+def
test_job_status_reporter_resolves_jobs_missing_from_a_completed_run(client,
fork, resolved):
+ _reporter_client(client, [], [_stale("pending")])
+
+ JobStatusReporter(client, repo=REPO).report(_pr(), fork)
+
+ expected = [
+ mock.call(
+ 42,
+ "abc123",
+ STALE_JOB_STATUS[0],
+ "https://job/Gone";,
+ STALE_JOB_STATUS[1],
+ context=f"{JOB_STATUS_CONTEXT_PREFIX}Gone",
+ )
+ ]
+ assert client.set_status.call_args_list == (expected if resolved else [])
+
+
+def test_job_status_reporter_leaves_other_contexts_alone(client):
+ other = {"context": STATUS_CONTEXT, "state": "pending", "target_url": "u",
"description": "d"}
+ _reporter_client(client, [], [other])
+
+ JobStatusReporter(client, repo=REPO).report(_pr(), COMPLETED_RUN)
+
+ client.set_status.assert_not_called()
+
+
+def
test_job_status_reporter_charges_each_write_even_when_a_later_one_fails(client):
+ _reporter_client(client, [_fork_job("A", conclusion="failure"),
_fork_job("B"), _fork_job("C")])
+ client.set_status.side_effect = [None, GitHubError("HTTP 500")]
+ reporter = JobStatusReporter(client, repo=REPO, budget=10)
+
+ with pytest.raises(GitHubError):
+ reporter.report(_pr(), COMPLETED_RUN)
+
+ assert reporter.remaining == 9
+
+
+def _reconcile_with_reporter(client, tmp_path, config=CONFIG):
+ client.graphql.side_effect = [
+ _search_result(),
+ _search_result(),
+ _search_result(_pr_node(1, headRepository={"nameWithOwner":
"someone/airflow"})),
+ {"pr0": {"object": None}},
+ ]
+ client.rest_get_pages.return_value = [{"total_count": 0, "workflow_runs":
[]}]
+ client.rest_get.return_value = {"state": "active"}
+ return run_reconcile(
+ client,
+ config,
+ now=NOW,
+ run_url="https://run";,
+ output_file=tmp_path / "s.json",
+ workflows_dir=tmp_path,
+ )
+
+
[email protected]("airflow_breeze.utils.fork_ci_reconcile.JobStatusReporter",
autospec=True)
[email protected]("airflow_breeze.utils.fork_ci_reconcile.classify", autospec=True)
+def test_run_reconcile_classifies_when_reporting_job_statuses_fails(
+ mock_classify, mock_reporter, client, tmp_path
+):
+ mock_classify.return_value = Plan(state=ForkState.RUNNING)
+ mock_reporter.return_value.report.side_effect = GitHubError("HTTP 502")
+ mock_reporter.return_value.deferred = []
+
+ assert _reconcile_with_reporter(client, tmp_path) is True
+
+ assert mock_classify.call_args.kwargs["jobs_reported"] is None
+ assert "reporting job statuses failed" in (tmp_path / "s.json").read_text()
+
+
[email protected]("airflow_breeze.utils.fork_ci_reconcile.JobStatusReporter",
autospec=True)
[email protected]("airflow_breeze.utils.fork_ci_reconcile.classify", autospec=True)
+def test_run_reconcile_without_job_statuses(mock_classify, mock_reporter,
client, tmp_path):
+ mock_classify.return_value = Plan(state=ForkState.RUNNING)
+ mock_reporter.return_value.deferred = []
+
+ _reconcile_with_reporter(client, tmp_path,
config=dataclasses.replace(CONFIG, job_statuses=False))
+
+ mock_reporter.return_value.report.assert_not_called()
+ assert mock_classify.call_args.kwargs["jobs_reported"] is None
+
+
+def test_job_status_reporter_does_not_read_without_writes_left(client):
+ reporter = JobStatusReporter(client, repo=REPO, budget=0)
+
+ assert reporter.report(_pr(), COMPLETED_RUN) is None
+ client.rest_get.assert_not_called()
+ assert reporter.deferred == [42]
+
+
[email protected](("deferred", "has_alerts"), [([1], True), ([],
False)])
[email protected]("airflow_breeze.utils.fork_ci_reconcile.JobStatusReporter",
autospec=True)
[email protected]("airflow_breeze.utils.fork_ci_reconcile.classify", autospec=True)
+def test_run_reconcile_reports_deferred_job_statuses_to_slack(
+ mock_classify, mock_reporter, client, tmp_path, deferred, has_alerts
+):
+ mock_classify.return_value = Plan(state=ForkState.RUNNING)
+ mock_reporter.return_value.report.return_value = None
+ mock_reporter.return_value.deferred = deferred
+
+ assert _reconcile_with_reporter(client, tmp_path) is has_alerts
+
+ if has_alerts:
+ assert JOB_STATUSES_DEFERRED_NOTE in (tmp_path / "s.json").read_text()
+
+
+def test_build_job_statuses_treats_contexts_case_insensitively():
+ statuses = build_job_statuses(
+ [_fork_job("Tests", conclusion="failure", url="bad"),
_fork_job("tests", url="ok")]
+ )
+
+ assert statuses == {
+ f"{JOB_STATUS_CONTEXT_PREFIX}Tests": ("failure", "bad", "Failed in the
fork - see Details")
+ }
+
+
+def test_job_status_reporter_compares_contexts_case_insensitively(client):
+ posted = {
+ "context": f"{JOB_STATUS_CONTEXT_PREFIX}TESTS",
+ "state": "failure",
+ "target_url": "https://job/Tests";,
+ "description": "Failed in the fork - see Details",
+ }
+ _reporter_client(client, [_fork_job("Tests", conclusion="failure")],
[posted])
+
+ assert JobStatusReporter(client, repo=REPO).report(_pr(), COMPLETED_RUN)
== "1:1:FAILURE"
+ client.set_status.assert_not_called()
