This is an automated email from the ASF dual-hosted git repository.
potiuk pushed a commit to branch v3-4-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/v3-4-test by this push:
new b3403ae3412 [v3-4-test] Block merging on fork CI once a PR is moved to
project CI (#76672) (#76981)
b3403ae3412 is described below
commit b3403ae3412fc94db93bfd750dc4e7478b093cf3
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Thu Nov 12 14:08:54 2026 +0000
[v3-4-test] Block merging on fork CI once a PR is moved to project CI
(#76672) (#76981)
The rulesets require the `CI passed` context from any source, and the Fork
CI
monitor posts it from the fork run. Adding `use project ci` did not withdraw
that success, so a pull request could merge on the fork result before
project
CI ran, as happened on #76536.
The revoke is done by the monitor and not by an event-driven workflow,
because
`pull_request_target` is about to be disabled for this repository. The fork
result therefore stays until the next monitor run that completes; the design
document records that window as a limit.
(cherry picked from commit e17cecaabbb08681dddf3fc015307dbcc656a498)
Co-authored-by: Shahar Epstein <[email protected]>
---
contributing-docs/25_maintainer_pr_triage.md | 8 +-
.../doc/ci/08_fork_ci_for_external_contributors.md | 73 +-
.../src/airflow_breeze/utils/fork_ci_messages.py | 3 +
.../src/airflow_breeze/utils/fork_ci_reconcile.py | 363 +++++++++-
dev/breeze/tests/test_fork_ci_reconcile.py | 763 +++++++++++++++++++++
5 files changed, 1192 insertions(+), 18 deletions(-)
diff --git a/contributing-docs/25_maintainer_pr_triage.md
b/contributing-docs/25_maintainer_pr_triage.md
index 333faefdd1f..8ab303197ca 100644
--- a/contributing-docs/25_maintainer_pr_triage.md
+++ b/contributing-docs/25_maintainer_pr_triage.md
@@ -274,7 +274,13 @@ Everything else the skill does (quality criteria, pings,
suspicious-change flags
usual. The `Fork CI / Tests (AMD)` commit status shows the fork CI state and
links to the run in
the fork. To move a single pull request back to project CI, add `use project
ci`; the reconciler
releases it on its next run; removing `fork ci required` by hand is not
enough, the next run
-labels a gated pull request again. Adding the label does not start project CI
by itself, and the
+labels a gated pull request again. The run that releases the pull request sets
its `CI passed`
+status to `pending` until project CI has passed on the head commit, so the
pull request cannot
+merge on the fork result from then on. Until that run, normally within 15
minutes of the label
+and longer when GitHub delays a scheduled run, it still can: wait for the
`pending` before merging
+or enabling auto-merge, or start the monitor by hand (`gh workflow run
fork-ci-monitor.yml -R
+apache/airflow`). After project CI has passed the status turns `success` on
the next monitor run,
+within about 15 minutes. Adding the label does not start project CI by itself,
and the
reconciler deletes the gated `Tests (AMD)` runs, so there is usually nothing
to re-run: close and
reopen the pull request (`gh pr close <n> -R apache/airflow && gh pr reopen
<n> -R apache/airflow`),
or ask the author to push. Pull requests that change CI-defining files are not
drafted
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 16e3981b9fc..0e1ed2cb926 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
@@ -238,7 +238,9 @@ Consequences for the other components:
reconciler removes `fork ci required`, undrafts it if the bot drafted it,
edits the state
comment to say project CI is in use, and posts a `success` status with the
description
"Project CI in use (label `use project ci`)". Project CI then runs on the
next push, or when
- a maintainer closes and reopens the pull request.
+ a maintainer closes and reopens the pull request. From that run on, and
until project CI has
+ passed on the head commit, the pull request cannot merge: its `CI passed`
status is `pending`
+ (see "The `CI passed` check").
- Both override labels on one pull request resolve to project CI and are
reported to Slack as
a conflict.
@@ -933,12 +935,12 @@ suspicious-change flags) is unaffected.
### The `CI passed` check
-The "Require CI on main" ruleset requires a `CI passed` check from GitHub
Actions before a pull
-request can merge or auto-merge. Project CI's `CI passed` job reports it, but
a gated pull
+The "Require CI on main" ruleset requires a `CI passed` check before a pull
request can merge or
+auto-merge; the required check names no integration, so a check run or a
commit status of that
+name from any source satisfies it. Project CI's `CI passed` job reports it,
but a gated pull
request never runs project CI, so the reconciler reports it instead: when the
state of the head
commit is `green`, it posts a `CI passed` commit status with state `success`,
linking the fork
-run. A commit status written with the monitor's `GITHUB_TOKEN` comes from
GitHub Actions, which
-is the source the ruleset accepts.
+run.
- Only `success` is posted. Without the status a gated pull request cannot
merge, which is what a
pending or failing one would say, and it never competes with project CI's
own check.
@@ -946,9 +948,59 @@ is the source the ruleset accepts.
it is not evidence. Once a maintainer adds `use project ci`, project CI
reports the check.
- When the head commit leaves `green`, for example a re-run in the fork that
fails, a `success`
posted earlier is overwritten with `pending` so the pull request cannot
merge on it.
-- A pull request released with `use project ci` keeps any `success` the
reconciler posted for
- its current head commit. A maintainer who wants the project CI result before
merging waits
- for project CI before enabling auto-merge, or pushes a new commit.
+- A pull request carrying `use project ci` cannot merge on the fork result.
GitHub requires both to
+ pass when a check run and a commit status have the same name and that name
is required, so the
+ `CI passed` status and project CI's `CI passed` check run must both succeed.
The run that
+ releases the pull request (see "Override labels") overwrites a fork
`success` on its head commit
+ with `pending` ("Waiting for project CI (use project ci)"). It reads the
labels from the pull
+ request node rather than from the label search, which lags behind a new
label by minutes; a
+ `success` that project CI earned is kept. A labelled pull request that gets
a new head commit can already carry a fork `success`
+ posted for another gated pull request on that commit, while project CI's
check run does not exist
+ until it runs; the pass over labelled pull requests below overwrites that
one.
+- Every monitor run also searches the open pull requests carrying the label,
as they have left the
+ state machine, and keeps the status in line with project CI. It only writes
when the head commit
+ already has a `CI passed` status: the project check run alone decides
otherwise, and a `pending`
+ the monitor created would block the pull request after project CI passed.
The project CI result is
+ the newest `pull_request` run of a project CI workflow file on the head
commit whose job named
+ exactly `CI passed` exists; a run that has no such job and reports `CI
passed (run cancelled)` or
+ `CI passed (project CI not run)` is ignored and does not hide an older run's
result (a re-run
+ keeps its check suite, so one that has both is judged by its `CI passed`
job). While that
+ job has not succeeded, a `success` is overwritten with `pending`; once it
has, the monitor posts
+ `success` ("Project CI passed for this commit"). Adding the label does not
start project CI (see
+ "Override labels"): the pull request stays blocked until a maintainer starts
it and it passes,
+ and then until the next monitor run posts the `success`, which can take
about 15 minutes.
+ Removing the label returns the pull request to the normal flow above:
removing it updates the
+ pull request, so discovery (open pull requests updated within the last 24
hours) labels it
+ `fork ci required` again and the next run reports the fork result, replacing
the waiting
+ `pending` once the fork run is green.
+ A pull request that is not gated after the label is removed (for example the
gate is off) is not
+ labelled again, so discovery posts the project CI `success` for it instead,
when its head commit's
+ `CI passed` status is one the reconciler wrote and is not `success`, and
project CI has passed.
+ Discovery only sees pull requests updated within the last 24 hours: one
whose project CI passes
+ later than that without any other update stays `pending` until it is updated.
+- A commit status belongs to the commit, so the monitor never posts a fork `CI
passed` `success`
+ on a head commit that a pull request carrying `use project ci` has, even for
another pull
+ request on the same commit; the project CI pass owns those. Revocations to
`pending` always go
+ through. When the list of those commits cannot be read, it posts no fork
`success` in that run. The list comes from a search that lags behind a new
label, so
+ before posting a fork `success` the monitor also reads the pull request
itself, and the open
+ pull requests of the same head branch (the listing that finds fork pull
requests), and skips the
+ post when one now carries the label or the head commit is new. A duplicate
pull request from
+ another branch at the same commit is not found this way.
+- The protection is eventual, not atomic. Nothing reacts to the label when it
is added
+ (`pull_request_target` is not used, see "Goals and non-goals"), so a fork
`success` posted
+ before the label stays until the next monitor run that completes: normally
within one cadence
+ (15 minutes), longer when GitHub delays or drops a scheduled run or a run
fails, and one more
+ cycle when the label arrives after a run has taken its snapshot of the pull
request. The pull
+ request can merge on the fork result in that window. A maintainer who adds
the label to a pull
+ request whose fork CI is green waits for the `pending` before merging or
enabling auto-merge,
+ or starts the monitor by hand (`workflow_dispatch`). A strict merge-time
guarantee needs a
+ required check bound to a source, which is out of scope here. Revoking a
`success` is never
+ limited by the per-run budget of commit statuses; posting the project CI
`success` is.
+- Known limitation: the ruleset accepts `CI passed` from any source, so a pull
request that changes
+ workflows can make its own project CI run produce a `CI passed` check run.
This holds for every
+ pull request that runs project CI; reviewers must check changes to CI files
before adding
+ `use project ci`, and again on every push of a labelled pull request, since
the label survives
+ pushes.
- In a gated run of `ci-amd.yml` every job after the gate is skipped, so
project CI's `CI passed`
job must not report success there; it reports under another name when
`build-info` is
skipped, and the reconciler's status is the only `CI passed` the pull
request has.
@@ -1245,8 +1297,9 @@ as a failed check rather than a crash.
## Known limitations
- Drafting, the setup comment and the commit status arrive within one cadence
of the event
- they reflect, not instantly. The cadence can be lowered to 5 minutes without
changing the
- budget analysis, since unchanged pull requests cost no mutations.
+ they reflect, not instantly. The cadence can be lowered, but the per-run
write budgets are
+ sized for four runs an hour against the `GITHUB_TOKEN` limits shared by
every workflow of the
+ repository, so they must be scaled down with it.
- The workflows API lists workflows from the fork's default branch. A fork
whose `main`
predates `ci-amd.yml` reports it missing; the setup comment covers this with
`gh repo sync`.
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 10cca162083..521b4714ef8 100644
--- a/dev/breeze/src/airflow_breeze/utils/fork_ci_messages.py
+++ b/dev/breeze/src/airflow_breeze/utils/fork_ci_messages.py
@@ -45,6 +45,9 @@ JOB_NAME_MAX_LENGTH = 100
CI_PASSED_CONTEXT = "CI passed"
CI_PASSED_DESCRIPTION = "Fork CI passed for this commit"
CI_PASSED_REVOKED_DESCRIPTION = "Fork CI result no longer trusted for this
commit"
+# Replaces a fork `success` on a pull request carrying `use project ci` until
project CI has passed.
+CI_PASSED_WAITING_PROJECT_CI_DESCRIPTION = "Waiting for project CI (use
project ci)"
+CI_PASSED_PROJECT_DESCRIPTION = "Project CI passed for this commit"
# Also required by the merge rulesets, so GitHub lists it as expected and
waiting on every pull
# request from the moment it is pushed. The reconciler reports it, linking the
run: for a gated pull
# request once the fork has a run for the head commit, for any other once
project CI has one.
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 fc316385adf..151f33c0e82 100644
--- a/dev/breeze/src/airflow_breeze/utils/fork_ci_reconcile.py
+++ b/dev/breeze/src/airflow_breeze/utils/fork_ci_reconcile.py
@@ -60,7 +60,9 @@ from airflow_breeze.utils.fork_ci_github import GitHubClient,
GitHubError
from airflow_breeze.utils.fork_ci_messages import (
CI_PASSED_CONTEXT,
CI_PASSED_DESCRIPTION,
+ CI_PASSED_PROJECT_DESCRIPTION,
CI_PASSED_REVOKED_DESCRIPTION,
+ CI_PASSED_WAITING_PROJECT_CI_DESCRIPTION,
CI_RUN_STARTED_CONTEXT,
CI_RUN_STARTED_DESCRIPTION,
CI_RUN_STARTED_PROJECT_DESCRIPTION,
@@ -122,6 +124,7 @@ GATED_RUN_JOBS = (
GATED_RUN_JOB_NAMES = frozenset[str]().union(*GATED_RUN_JOBS)
CHECK_RUN_NEEDED_JOB_NAME = "Check if the run is needed"
PROJECT_CI_NOT_RUN_JOB_NAME = "CI passed (project CI not run)"
+CI_PASSED_CANCELLED_JOB_NAME = "CI passed (run cancelled)"
BOT_LOGINS = frozenset({"github-actions", "github-actions[bot]"})
DISCOVERY_WINDOW = timedelta(hours=24)
PASS_TWO_PAGE_SIZE = 25
@@ -176,6 +179,9 @@ MAX_JOB_STATUSES_PER_RUN = 40
# At most this many, so job statuses keep the rest; the first runs after the
check became required
# catch up on the open pull requests a batch at a time, most recently updated
first.
MAX_CI_RUN_STARTED_PER_RUN = 20
+# `CI passed` successes for pull requests carrying `use project ci`, written
once per new project CI
+# pass, so far fewer than this per run. Revoking a success is never limited.
+MAX_PROJECT_CI_PASSED_PER_RUN = 5
# The GitHub Actions app, whose check suites are the workflow runs of a commit.
GITHUB_ACTIONS_APP_ID = 15368
# A single large fork run can need more writes than one run's budget, which
the next run catches up
@@ -286,6 +292,7 @@ class PullRequestSnapshot:
current_status: tuple[str, str, str] | None = None
status_created_at: datetime | None = None
ci_passed_state: str | None = None
+ ci_passed_description: str | None = None
ci_run_started_state: str | None = None
# GateEvent.ci_run_at of a pilot's pull request created before they
joined; set from its
# `fork ci required` label by `mark_older_pilot_pull_request`.
@@ -591,7 +598,18 @@ def classify(
plan.draft = False
since = marker.since if marker and marker.state == state.value else now
_set_marker_comment(plan, pr, comment(Marker(state.value, pr.head_sha,
since)))
- _set_status(plan, pr, state.value,
f"https://github.com/{repo}/pull/{pr.number}/checks";)
+ checks_url = f"https://github.com/{repo}/pull/{pr.number}/checks";
+ _set_status(plan, pr, state.value, checks_url)
+ # The pull request must not merge on the fork result once a maintainer
moves it to project
+ # CI. Revoked here, in the run that sees the label, rather than only by
+ # ``report_project_ci_passed``, whose search lags behind a new label
by minutes. A success
+ # that project CI earned (a retried release after that pass posted it)
is kept.
+ if (
+ state == ForkState.RELEASED
+ and pr.ci_passed_state == "success"
+ and pr.ci_passed_description != CI_PASSED_PROJECT_DESCRIPTION
+ ):
+ plan.ci_passed = ("pending", checks_url,
CI_PASSED_WAITING_PROJECT_CI_DESCRIPTION)
return plan
if LABEL_USE_PROJECT_CI in labels:
@@ -827,6 +845,10 @@ query($q: String!, $cursor: String) {
author { login }
headRepository { nameWithOwner }
labels(first: 50) { nodes { name } }
+ commits(last: 1) { nodes { commit {
+ oid
+ status { context(name: "%(context)s") { state description } }
+ } } }
} }
}
}
@@ -857,7 +879,7 @@ query($q: String!, $cursor: String) {
oid
status {
context(name: "%(context)s") { state targetUrl description createdAt
}
- ciPassed: context(name: "%(ci_passed)s") { state }
+ ciPassed: context(name: "%(ci_passed)s") { state description }
ciRunStarted: context(name: "%(ci_run_started)s") { state }
}
} } }
@@ -930,6 +952,7 @@ def parse_pull_request(node: dict[str, Any]) ->
PullRequestSnapshot:
current_status = None
status_created_at = None
ci_passed_state = None
+ ci_passed_description = None
ci_run_started_state = None
commits = (node.get("commits") or {}).get("nodes") or []
head_status_unread = bool(commits) and commits[0]["commit"].get("oid") !=
node["headRefOid"]
@@ -937,6 +960,7 @@ def parse_pull_request(node: dict[str, Any]) ->
PullRequestSnapshot:
status = commits[0]["commit"].get("status") or {}
if status.get("ciPassed"):
ci_passed_state = status["ciPassed"]["state"].lower()
+ ci_passed_description = status["ciPassed"].get("description")
if status.get("ciRunStarted"):
ci_run_started_state = status["ciRunStarted"]["state"].lower()
context = status.get("context")
@@ -978,6 +1002,7 @@ def parse_pull_request(node: dict[str, Any]) ->
PullRequestSnapshot:
current_status=current_status,
status_created_at=status_created_at,
ci_passed_state=ci_passed_state,
+ ci_passed_description=ci_passed_description,
ci_run_started_state=ci_run_started_state,
marker=marker,
marker_comment_id=marker_id,
@@ -1102,6 +1127,220 @@ def report_project_ci_runs_started(
return reported
+PROJECT_CI_PASSED_QUERY = """
+query($q: String!, $cursor: String) {
+ search(type: ISSUE, first: 50, after: $cursor, query: $q) {
+ pageInfo { hasNextPage endCursor }
+ nodes { ... on PullRequest {
+ number headRefOid
+ commits(last: 1) { nodes { commit {
+ oid
+ status { context(name: "%(context)s") { state description } }
+ checkSuites(first: 100, filterBy: {appId: %(app_id)d}) { nodes {
+ createdAt status conclusion
+ workflowRun { url event workflow { resourcePath } }
+ checkRuns(first: 10, filterBy: {checkName: "%(context)s"}) { nodes {
name conclusion } }
+ cancelled: checkRuns(first: 1, filterBy: {checkName:
"%(cancelled)s"}) { totalCount }
+ notRun: checkRuns(first: 1, filterBy: {checkName: "%(not_run)s"}) {
totalCount }
+ } }
+ } } }
+ } }
+ }
+}
+"""
+
+
+PROJECT_CI_HEADS_QUERY = """
+query($q: String!, $cursor: String) {
+ search(type: ISSUE, first: 100, after: $cursor, query: $q) {
+ pageInfo { hasNextPage endCursor }
+ nodes { ... on PullRequest { headRefOid } }
+ }
+}
+"""
+
+
+def find_project_ci_passed_url(node: dict[str, Any]) -> str | None:
+ """
+ The latest project CI pull_request run's URL when its `CI passed` job
succeeded on the head commit.
+
+ The job reports under another name when the run was cancelled or skipped;
such a run does not
+ count, nor does it hide an older run's result. A re-run keeps its check
suite, so a suite that
+ also has the exact `CI passed` check run is judged by that one, its latest
attempt.
+ None also when the last commit listed is not the head;
``read_project_ci_passed_url`` reads that.
+ """
+ if not is_head_listed(node):
+ return None
+ suites = [
+ suite
+ for suite in (node["commits"]["nodes"][0]["commit"].get("checkSuites")
or {}).get("nodes") or []
+ if (run := suite.get("workflowRun") or {}).get("event") ==
"pull_request"
+ and _workflow_file((run.get("workflow") or {}).get("resourcePath")) in
PROJECT_CI_WORKFLOW_FILES
+ ]
+ suites = [
+ suite
+ for suite in suites
+ if suite["checkRuns"]["nodes"]
+ or not (
+ suite["cancelled"]["totalCount"]
+ or suite["notRun"]["totalCount"]
+ or (suite["status"], suite["conclusion"]) == ("COMPLETED",
"CANCELLED")
+ )
+ ]
+ latest = max(suites, key=lambda suite: suite["createdAt"], default=None)
+ if latest and any(
+ check["name"] == CI_PASSED_CONTEXT and check["conclusion"] == "SUCCESS"
+ for check in latest["checkRuns"]["nodes"]
+ ):
+ return latest["workflowRun"]["url"]
+ return None
+
+
+def read_project_ci_passed_url(client: GitHubClient, repo: str, head_sha: str)
-> str | None:
+ """``find_project_ci_passed_url`` for a head commit the search did not
list, read from REST."""
+ runs = [
+ run
+ for run in client.rest_get(
+
f"repos/{repo}/actions/runs?event=pull_request&head_sha={head_sha}&per_page=100"
+ ).get("workflow_runs", [])
+ if _workflow_file(run.get("path")) in PROJECT_CI_WORKFLOW_FILES
+ ]
+ if not runs:
+ return None
+
+ def get_check_runs(name: str) -> list[dict[str, Any]]:
+ return client.rest_get(
+
f"repos/{repo}/commits/{head_sha}/check-runs?check_name={quote(name)}&filter=latest&per_page=100"
+ ).get("check_runs", [])
+
+ passed_checks = get_check_runs(CI_PASSED_CONTEXT)
+ skipped_suites = {
+ check["check_suite"]["id"]
+ for name in (CI_PASSED_CANCELLED_JOB_NAME, PROJECT_CI_NOT_RUN_JOB_NAME)
+ for check in get_check_runs(name)
+ } - {check["check_suite"]["id"] for check in passed_checks}
+ passed_suites = {check["check_suite"]["id"] for check in passed_checks}
+ # Cancelled before any job started, the run has no check run at all.
+ skipped_suites |= {
+ run["check_suite_id"]
+ for run in runs
+ if (run["status"], run["conclusion"]) == ("completed", "cancelled")
+ and run["check_suite_id"] not in passed_suites
+ }
+ run = next((run for run in runs if run["check_suite_id"] not in
skipped_suites), None)
+ if run and any(
+ check["name"] == CI_PASSED_CONTEXT
+ and check["conclusion"] == "success"
+ and check["check_suite"]["id"] == run["check_suite_id"]
+ for check in passed_checks
+ ):
+ return run["html_url"]
+ return None
+
+
+def is_unchanged_without_project_ci(client: GitHubClient, repo: str, pr:
PullRequestSnapshot) -> bool:
+ """
+ Whether the pull request, read live, still has the head commit and no `use
project ci`.
+
+ The search that lists labelled pull requests lags behind a new label by
minutes, and a snapshot
+ may predate it, so a fork `success` is only posted after this read. The
open pull requests of the
+ same head branch are read too, since they share the commit status. Listing
by branch is what
+ finds fork pull requests; a duplicate pull request from another branch at
the same commit is not.
+ """
+ live = client.rest_get(f"repos/{repo}/pulls/{pr.number}")
+ if live["head"]["sha"] != pr.head_sha:
+ return False
+ owner = (pr.head_repo or "").split("/", 1)[0]
+ same_branch = client.rest_get(
+
f"repos/{repo}/pulls?state=open&per_page=100&head={quote(f'{owner}:{pr.head_ref}',
safe=':')}"
+ )
+ return all(
+ LABEL_USE_PROJECT_CI not in {label["name"] for label in
pull_request["labels"]}
+ for pull_request in [live, *same_branch]
+ )
+
+
+def find_project_ci_label_heads(client: GitHubClient, repo: str) -> set[str]:
+ """The head commits of the open pull requests carrying `use project ci`."""
+ return {
+ node["headRefOid"]
+ for node in _search(
+ client,
+ PROJECT_CI_HEADS_QUERY,
+ f'repo:{repo} is:pr is:open label:"{LABEL_USE_PROJECT_CI}"',
+ )
+ }
+
+
+def report_project_ci_passed(
+ client: GitHubClient,
+ *,
+ repo: str,
+ attention: list[tuple[int, str]],
+ budget: int = MAX_PROJECT_CI_PASSED_PER_RUN,
+) -> list[int]:
+ """
+ Keep the `CI passed` status of pull requests with `use project ci` in line
with project CI.
+
+ Released pull requests leave the reconciler's pass, but a `success` it
posted from the fork run
+ would still let them merge before project CI has passed. Return the
numbers it wrote for.
+
+ :param budget: the most `success` statuses to write; revoking one is never
limited.
+ :param attention: collects the pull requests whose status could not be
read or written; the
+ others are still handled.
+ """
+ written: list[int] = []
+ successes = 0
+ for node in _search(
+ client,
+ PROJECT_CI_PASSED_QUERY
+ % {
+ "context": CI_PASSED_CONTEXT,
+ "cancelled": CI_PASSED_CANCELLED_JOB_NAME,
+ "not_run": PROJECT_CI_NOT_RUN_JOB_NAME,
+ "app_id": GITHUB_ACTIONS_APP_ID,
+ },
+ f'repo:{repo} is:pr is:open label:"{LABEL_USE_PROJECT_CI}"
sort:updated-desc',
+ ):
+ try:
+ if is_head_listed(node):
+ status = (node["commits"]["nodes"][0]["commit"].get("status")
or {}).get("context") or {}
+ state, description = (status.get("state") or "").lower(),
status.get("description")
+ passed_url = find_project_ci_passed_url(node)
+ else:
+ rest_status = read_ci_passed_status(client, repo,
node["headRefOid"])
+ state, description = rest_status.get("state"),
rest_status.get("description")
+ passed_url = read_project_ci_passed_url(client, repo,
node["headRefOid"])
+ # Both the status and the project check run must pass when they
share a name, so a
+ # status the monitor never posted is not created here.
+ if not state:
+ continue
+ if passed_url:
+ if (state, description) == ("success",
CI_PASSED_PROJECT_DESCRIPTION):
+ continue
+ if successes >= budget:
+ console_print(
+ f"[info]`{CI_PASSED_CONTEXT}` of more pull requests
left for the next run[/]"
+ )
+ continue
+ change = ("success", passed_url, CI_PASSED_PROJECT_DESCRIPTION)
+ elif state == "success":
+ change = (
+ "pending",
+ f"https://github.com/{repo}/pull/{node['number']}/checks",
+ CI_PASSED_WAITING_PROJECT_CI_DESCRIPTION,
+ )
+ else:
+ continue
+ client.set_status(node["number"], node["headRefOid"], *change,
context=CI_PASSED_CONTEXT)
+ written.append(node["number"])
+ successes += change[0] == "success"
+ except Exception as e:
+ console_print(f"[error]#{node['number']}: `{CI_PASSED_CONTEXT}`
not updated: {e}[/]")
+ attention.append((node["number"], f"`{CI_PASSED_CONTEXT}` not
updated: {type(e).__name__}"))
+ return written
+
+
def _search(client: GitHubClient, query: str, search_query: str) ->
Iterable[dict[str, Any]]:
cursor = None
while True:
@@ -1229,11 +1468,14 @@ def discover(
repo: str,
now: datetime,
unknown: list[tuple[int, str]] | None = None,
+ ci_passed_written: list[int] | None = None,
) -> list[int]:
"""
Label newly gated pull requests with `fork ci required` and return their
numbers.
:param unknown: collects the pull requests whose base branch configuration
could not be read.
+ :param ci_passed_written: collects the pull requests that were not gated
and got their `CI passed`
+ `success` from project CI, see ``report_orphaned_ci_passed``.
"""
since = (now - DISCOVERY_WINDOW).strftime("%Y-%m-%dT%H:%M:%SZ")
queries = (
@@ -1241,8 +1483,19 @@ def discover(
f'repo:{repo} is:pr is:open label:"{LABEL_USE_FORK_CI}"
-label:"{LABEL_FORK_CI_REQUIRED}"',
)
labelled: list[int] = []
+
+ def recover_orphaned_status(node: dict[str, Any]) -> None:
+ if ci_passed_written is None or LABEL_USE_PROJECT_CI in _labels(node):
+ return
+ try:
+ report_orphaned_ci_passed(client, repo, node, ci_passed_written)
+ except Exception as e:
+ console_print(f"[error]#{node['number']}: `{CI_PASSED_CONTEXT}`
not updated: {e}[/]")
+ if unknown is not None:
+ unknown.append((node["number"], f"`{CI_PASSED_CONTEXT}` not
updated: {type(e).__name__}"))
+
for search_query in queries:
- for node in _search(client, DISCOVERY_QUERY, search_query):
+ for node in _search(client, DISCOVERY_QUERY % {"context":
CI_PASSED_CONTEXT}, search_query):
if node["number"] in labelled:
continue
try:
@@ -1252,6 +1505,7 @@ def discover(
unknown.append((node["number"], f"{type(e).__name__}:
{e}"))
continue
if config is None:
+ recover_orphaned_status(node)
continue
author = _login(node.get("author"))
created_at = parse_github_datetime(node["createdAt"])
@@ -1278,9 +1532,58 @@ def discover(
if decide(event, config).reason in GATED_REASONS:
client.add_labels(node["number"], [LABEL_FORK_CI_REQUIRED])
labelled.append(node["number"])
+ else:
+ recover_orphaned_status(node)
return labelled
+def read_ci_passed_status(client: GitHubClient, repo: str, head_sha: str) ->
dict[str, Any]:
+ """The ``CI passed`` commit status of a head commit that the search did
not list, from REST."""
+ statuses = client.rest_get(
+ f"repos/{repo}/commits/{head_sha}/status?per_page=100", paginate=True,
items_key="statuses"
+ )
+ return next((status for status in statuses if status["context"] ==
CI_PASSED_CONTEXT), {})
+
+
+def report_orphaned_ci_passed(
+ client: GitHubClient, repo: str, node: dict[str, Any], written: list[int]
+) -> None:
+ """
+ Post the project CI `success` for a pull request that is not gated but has
a stale `CI passed` status.
+
+ A pull request that left the gated flow, for example by losing `use
project ci` or `use fork ci`
+ while the gate is off, keeps the `pending` or fork status the reconciler
wrote, and a status and
+ a check run of one name must both pass. Only a status the reconciler wrote
counts, and only
+ discovery's 24 hour window sees the pull request.
+ """
+ commit = ((node.get("commits") or {}).get("nodes") or
[{}])[0].get("commit") or {}
+ if commit.get("oid") == node["headRefOid"]:
+ status = (commit.get("status") or {}).get("context") or {}
+ else:
+ status = read_ci_passed_status(client, repo, node["headRefOid"])
+ if (
+ (status.get("state") or "").lower() in ("", "success")
+ or status.get("description")
+ not in (
+ CI_PASSED_DESCRIPTION,
+ CI_PASSED_WAITING_PROJECT_CI_DESCRIPTION,
+ CI_PASSED_REVOKED_DESCRIPTION,
+ )
+ or len(written) >= MAX_PROJECT_CI_PASSED_PER_RUN
+ ):
+ return
+ if passed_url := read_project_ci_passed_url(client, repo,
node["headRefOid"]):
+ client.set_status(
+ node["number"],
+ node["headRefOid"],
+ "success",
+ passed_url,
+ CI_PASSED_PROJECT_DESCRIPTION,
+ context=CI_PASSED_CONTEXT,
+ )
+ written.append(node["number"])
+
+
def fetch_fork_observations(
client: GitHubClient, prs: list[PullRequestSnapshot]
) -> dict[int, ForkObservation]:
@@ -1965,6 +2268,7 @@ def read_head_status(client: GitHubClient, repo: str, pr:
PullRequestSnapshot) -
),
status_created_at=parse_github_datetime(status.get("created_at") or
"") if status else None,
ci_passed_state=ci_passed["state"] if ci_passed else None,
+ ci_passed_description=ci_passed.get("description") if ci_passed else
None,
ci_run_started_state=ci_run_started["state"] if ci_run_started else
None,
head_status_unread=False,
)
@@ -2237,7 +2541,10 @@ def run_reconcile(
draft_changes: list[DraftChange] = []
configs = BaseGateConfigs(client, repo=repo,
default_branch=default_branch, default_config=config)
- labelled = discover(client, configs, repo=repo, now=now, unknown=unknown)
+ ci_passed_written: list[int] = []
+ labelled = discover(
+ client, configs, repo=repo, now=now, unknown=unknown,
ci_passed_written=ci_passed_written
+ )
prs = [
parse_pull_request(node)
@@ -2281,8 +2588,26 @@ def run_reconcile(
ci_run_started_note = (
f"reporting `{CI_RUN_STARTED_CONTEXT}` for project CI failed:
{type(e).__name__}"
)
+ # Commit statuses belong to a commit, so a fork `success` posted for any
pull request on one of
+ # these would also let a pull request with `use project ci` merge. None
when they could not be
+ # listed: no fork `success` is then posted at all.
+ ci_passed_notes: list[str] = []
+ project_ci_heads: set[str] | None
+ try:
+ project_ci_heads = find_project_ci_label_heads(client, repo)
+ ci_passed_reserved = min(MAX_PROJECT_CI_PASSED_PER_RUN,
len(project_ci_heads))
+ except Exception as e:
+ console_print(f"[error]Listing the pull requests with
`{LABEL_USE_PROJECT_CI}` failed: {e}[/]")
+ project_ci_heads = None
+ ci_passed_reserved = MAX_PROJECT_CI_PASSED_PER_RUN
+ ci_passed_notes.append(
+ f"listing the pull requests with `{LABEL_USE_PROJECT_CI}` failed:
{type(e).__name__}"
+ )
job_reporter = JobStatusReporter(
- client, repo=repo, budget=max(MAX_JOB_STATUSES_PER_RUN -
ci_run_started_writes, 0), now=now
+ client,
+ repo=repo,
+ budget=max(MAX_JOB_STATUSES_PER_RUN - ci_run_started_writes -
ci_passed_reserved, 0),
+ now=now,
)
previous_job_status_failure_runs =
read_job_status_failure_runs(reported_file)
@@ -2333,6 +2658,18 @@ def run_reconcile(
head_shared_across_bases=pr.head_sha in shared_heads,
find_gate_change=functools.partial(gate_change_finder.find,
pr),
)
+ # A revocation to `pending` always goes through: it is idempotent
and must not depend on
+ # `report_project_ci_passed` succeeding later in this run.
+ if (
+ plan.ci_passed
+ and plan.ci_passed[0] == "success"
+ and (
+ pr.head_sha in (project_ci_heads or ())
+ or project_ci_heads is None
+ or not is_unchanged_without_project_ci(client, repo, pr)
+ )
+ ):
+ plan.ci_passed = None
draft_change = apply_plan(client, pr, plan)
if draft_change is not None:
draft_changes.append(draft_change)
@@ -2350,11 +2687,23 @@ def run_reconcile(
console_print(f"[error]#{listed.number}: {e}[/]")
unknown.append((listed.number, f"{type(e).__name__}: {e}"))
+ # After the loop, so that it reads the labels and statuses as they are
last.
+ try:
+ report_project_ci_passed(
+ client,
+ repo=repo,
+ attention=attention,
+ budget=MAX_PROJECT_CI_PASSED_PER_RUN - len(ci_passed_written),
+ )
+ except Exception as e:
+ console_print(f"[error]Reporting `{CI_PASSED_CONTEXT}` for
`{LABEL_USE_PROJECT_CI}` failed: {e}[/]")
+ ci_passed_notes.append(
+ f"reporting `{CI_PASSED_CONTEXT}` for `{LABEL_USE_PROJECT_CI}`
failed: {type(e).__name__}"
+ )
attention.extend(bypass_finder.ambiguous)
attention.extend(job_reporter.delayed)
notes = [bypass_finder.incomplete] if bypass_finder.incomplete else []
- if ci_run_started_note:
- notes.append(ci_run_started_note)
+ notes.extend(note for note in (ci_run_started_note, *ci_passed_notes) if
note)
previous_deferred_runs = read_run_count(reported_file, "deferred_runs")
deferred_runs = (previous_deferred_runs or 0) + 1 if job_reporter.deferred
else 0
if job_reporter.deferred:
diff --git a/dev/breeze/tests/test_fork_ci_reconcile.py
b/dev/breeze/tests/test_fork_ci_reconcile.py
index a26a3cd21fc..c7b50800756 100644
--- a/dev/breeze/tests/test_fork_ci_reconcile.py
+++ b/dev/breeze/tests/test_fork_ci_reconcile.py
@@ -21,6 +21,7 @@ import dataclasses
import json
from datetime import UTC, datetime, timedelta, timezone
from unittest import mock
+from urllib.parse import quote
import pytest
from click.testing import CliRunner
@@ -30,6 +31,9 @@ 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,
+ CI_PASSED_DESCRIPTION,
+ CI_PASSED_PROJECT_DESCRIPTION,
+ CI_PASSED_WAITING_PROJECT_CI_DESCRIPTION,
CI_RUN_STARTED_CONTEXT,
CI_RUN_STARTED_DESCRIPTION,
CI_RUN_STARTED_PROJECT_DESCRIPTION,
@@ -66,6 +70,8 @@ from airflow_breeze.utils.fork_ci_reconcile import (
fetch_workflow_state,
find_ci_run_since,
find_heads_shared_across_bases,
+ find_project_ci_label_heads,
+ find_project_ci_passed_url,
find_project_ci_run_url,
find_state_comment,
is_cancelled_gate_run,
@@ -80,7 +86,9 @@ from airflow_breeze.utils.fork_ci_reconcile import (
read_ci_run_at,
read_head_status,
read_job_status_failure_runs,
+ read_project_ci_passed_url,
read_project_ci_run_url,
+ report_project_ci_passed,
report_project_ci_runs_started,
run_reconcile,
)
@@ -93,6 +101,7 @@ JOINED = CREATED + timedelta(days=2)
CONFIG = GateConfig(enabled=True, enabled_since=datetime(2026, 10, 1,
tzinfo=UTC), close_after_days=7)
RUN_URL = "https://github.com/someone/airflow/actions/runs/1";
PROJECT_RUN_URL = "https://github.com/apache/airflow/actions/runs/2";
+PROJECT_PASSED_URL = "https://github.com/apache/airflow/actions/runs/5";
GREEN = ForkObservation(suite_status="COMPLETED", suite_conclusion="SUCCESS",
run_url=RUN_URL)
@@ -681,6 +690,39 @@ def test_release_with_use_project_ci():
find_bypass.assert_not_called()
[email protected](
+ ("ci_passed_state", "ci_passed_description", "revoked"),
+ [
+ pytest.param("success", CI_PASSED_DESCRIPTION, True,
id="fork-success-is-revoked"),
+ pytest.param("success", None, True,
id="success-without-description-is-revoked"),
+ pytest.param("success", CI_PASSED_PROJECT_DESCRIPTION, False,
id="project-success-is-kept"),
+ pytest.param("pending", CI_PASSED_WAITING_PROJECT_CI_DESCRIPTION,
False, id="pending-is-kept"),
+ pytest.param(None, None, False, id="no-status-is-not-created"),
+ ],
+)
+def test_release_with_use_project_ci_revokes_a_fork_success(ci_passed_state,
ci_passed_description, revoked):
+ pr = _pr(
+ labels=("fork ci required", "use project ci"),
+ ci_passed_state=ci_passed_state,
+ ci_passed_description=ci_passed_description,
+ )
+
+ plan, _, _ = _classify(pr, GREEN)
+
+ expected = (
+ "pending",
+ "https://github.com/apache/airflow/pull/42/checks";,
+ CI_PASSED_WAITING_PROJECT_CI_DESCRIPTION,
+ )
+ assert plan.ci_passed == (expected if revoked else None)
+
+
+def test_base_branch_without_gate_keeps_the_ci_passed_status():
+ plan, _, _ = _classify(_pr(base_ref="v2-10-test",
ci_passed_state="success"), GREEN, config=None)
+
+ assert plan.ci_passed is None
+
+
def test_base_branch_without_gate_releases():
plan, find_bypass, _ = _classify(_pr(base_ref="v2-10-test"), GREEN,
config=None)
@@ -971,6 +1013,24 @@ def _finder(
return BypassFinder(client, repo=REPO, now=NOW,
base_triggers=BASE_TRIGGERS, is_gated_base=is_gated_base)
[email protected](autouse=True)
+def mock_project_ci_heads():
+ with mock.patch(
+ "airflow_breeze.utils.fork_ci_reconcile.find_project_ci_label_heads",
autospec=True
+ ) as mock_heads:
+ mock_heads.return_value = set()
+ yield mock_heads
+
+
[email protected](autouse=True)
+def mock_report_project_ci_passed():
+ with mock.patch(
+ "airflow_breeze.utils.fork_ci_reconcile.report_project_ci_passed",
autospec=True
+ ) as mock_report:
+ mock_report.return_value = []
+ yield mock_report
+
+
@pytest.fixture
def client():
github = mock.create_autospec(GitHubClient, instance=True)
@@ -1504,6 +1564,36 @@ def test_read_head_status_reads_ci_run_started(client):
assert read_head_status(client, REPO, pr).ci_run_started_state == "success"
+def test_parse_pull_request_reads_the_ci_passed_description():
+ node = _pr_node(
+ 1,
+ commits={
+ "nodes": [
+ {
+ "commit": {
+ "oid": "sha1",
+ "status": {
+ "context": None,
+ "ciPassed": {"state": "SUCCESS", "description":
CI_PASSED_PROJECT_DESCRIPTION},
+ },
+ }
+ }
+ ]
+ },
+ )
+
+ assert parse_pull_request(node).ci_passed_description ==
CI_PASSED_PROJECT_DESCRIPTION
+
+
+def test_read_head_status_reads_the_ci_passed_description(client):
+ client.rest_get.return_value = [
+ {"context": CI_PASSED_CONTEXT, "state": "success", "description":
CI_PASSED_PROJECT_DESCRIPTION}
+ ]
+ pr = dataclasses.replace(parse_pull_request(_pr_node(1)),
head_status_unread=True)
+
+ assert read_head_status(client, REPO, pr).ci_passed_description ==
CI_PASSED_PROJECT_DESCRIPTION
+
+
def _branch_run(created_at: str, head_repo: str = "someone/airflow") -> dict:
return {"created_at": created_at, "head_repository": {"full_name":
head_repo}}
@@ -1611,6 +1701,7 @@ def _run_json(run_id, workflow_file="ci-amd.yml", *,
path=None, conclusion=None)
return {
"html_url": f"https://github.com/apache/airflow/actions/runs/{run_id}";,
"path": path or f".github/workflows/{workflow_file}",
+ "status": "completed" if conclusion else "in_progress",
"conclusion": conclusion,
}
@@ -1805,6 +1896,135 @@ def
test_discover_labels_an_older_pilot_pull_request_after_a_run_since_joining(c
client.add_labels.assert_called_once_with(2, ["fork ci required"])
+def _ci_passed_commit(state, description, oid="sha1"):
+ return {
+ "nodes": [
+ {"commit": {"oid": oid, "status": {"context": {"state": state,
"description": description}}}}
+ ]
+ }
+
+
[email protected](
+ ("enabled", "state", "description", "passed_url", "expected_written"),
+ [
+ pytest.param(
+ False,
+ "PENDING",
+ CI_PASSED_WAITING_PROJECT_CI_DESCRIPTION,
+ PROJECT_PASSED_URL,
+ [1],
+ id="ungated-waiting-project-ci-passed",
+ ),
+ pytest.param(
+ False, "PENDING", CI_PASSED_DESCRIPTION, PROJECT_PASSED_URL, [1],
id="ungated-fork-status"
+ ),
+ pytest.param(
+ False, "PENDING", CI_PASSED_WAITING_PROJECT_CI_DESCRIPTION, None,
[], id="ungated-not-passed"
+ ),
+ pytest.param(False, "SUCCESS", CI_PASSED_DESCRIPTION,
PROJECT_PASSED_URL, [], id="already-success"),
+ pytest.param(False, "PENDING", "someone else", PROJECT_PASSED_URL, [],
id="foreign-status"),
+ pytest.param(
+ True,
+ "PENDING",
+ CI_PASSED_WAITING_PROJECT_CI_DESCRIPTION,
+ PROJECT_PASSED_URL,
+ [],
+ id="gated-is-left-to-the-fork-flow",
+ ),
+ ],
+)
[email protected]("airflow_breeze.utils.fork_ci_reconcile.read_project_ci_passed_url",
autospec=True)
+def
test_discover_posts_project_ci_success_for_an_ungated_pull_request_with_a_stale_status(
+ mock_read, client, enabled, state, description, passed_url,
expected_written
+):
+ config = GateConfig(enabled=enabled, enabled_since=datetime(2026, 10, 1,
tzinfo=UTC))
+ node = {**_pr_node(1), "labels": {"nodes": []}, "commits":
_ci_passed_commit(state, description)}
+ node["headRefOid"] = "sha1"
+ mock_read.return_value = passed_url
+ client.graphql.side_effect = [_search_result(node), _search_result()]
+ configs = BaseGateConfigs(client, repo=REPO, default_branch="main",
default_config=config)
+ written: list[int] = []
+
+ discover(client, configs, repo=REPO, now=NOW, ci_passed_written=written)
+
+ assert written == expected_written
+ if expected_written:
+ client.set_status.assert_called_once_with(
+ 1, "sha1", "success", PROJECT_PASSED_URL,
CI_PASSED_PROJECT_DESCRIPTION, context=CI_PASSED_CONTEXT
+ )
+ else:
+ client.set_status.assert_not_called()
+
+
[email protected]("airflow_breeze.utils.fork_ci_reconcile.read_project_ci_passed_url",
autospec=True)
+def
test_discover_recovers_a_status_on_a_base_branch_without_the_gate(mock_read,
client):
+ configs = mock.create_autospec(BaseGateConfigs, instance=True)
+ configs.get.return_value = None
+ node = {
+ **_pr_node(1),
+ "headRefOid": "sha1",
+ "labels": {"nodes": []},
+ "commits": _ci_passed_commit("PENDING",
CI_PASSED_WAITING_PROJECT_CI_DESCRIPTION),
+ }
+ mock_read.return_value = PROJECT_PASSED_URL
+ client.graphql.side_effect = [_search_result(node), _search_result()]
+ written: list[int] = []
+
+ discover(client, configs, repo=REPO, now=NOW, ci_passed_written=written)
+
+ assert written == [1]
+ client.add_labels.assert_not_called()
+
+
[email protected]("airflow_breeze.utils.fork_ci_reconcile.read_project_ci_passed_url",
autospec=True)
+def
test_discover_reads_the_status_of_a_head_commit_the_search_did_not_list(mock_read,
client):
+ config = GateConfig(enabled=False)
+ node = {
+ **_pr_node(1),
+ "headRefOid": "sha1",
+ "labels": {"nodes": []},
+ "commits": _ci_passed_commit("SUCCESS", CI_PASSED_DESCRIPTION,
oid="older"),
+ }
+ mock_read.return_value = PROJECT_PASSED_URL
+ client.graphql.side_effect = [_search_result(node), _search_result()]
+ client.rest_get.return_value = [
+ {
+ "context": CI_PASSED_CONTEXT,
+ "state": "pending",
+ "description": CI_PASSED_WAITING_PROJECT_CI_DESCRIPTION,
+ }
+ ]
+ configs = BaseGateConfigs(client, repo=REPO, default_branch="main",
default_config=config)
+ written: list[int] = []
+
+ discover(client, configs, repo=REPO, now=NOW, ci_passed_written=written)
+
+ assert written == [1]
+ mock_read.assert_called_once_with(client, REPO, "sha1")
+
+
+def
test_discover_keeps_going_when_the_project_ci_lookup_of_one_pull_request_fails(client):
+ config = GateConfig(enabled=False)
+ nodes = [
+ {
+ **_pr_node(n),
+ "headRefOid": f"sha{n}",
+ "labels": {"nodes": []},
+ "commits": _ci_passed_commit("PENDING",
CI_PASSED_WAITING_PROJECT_CI_DESCRIPTION, f"sha{n}"),
+ }
+ for n in (1, 2)
+ ]
+ client.graphql.side_effect = [_search_result(*nodes), _search_result()]
+ client.rest_get.side_effect = [GitHubError("HTTP 502"), {"workflow_runs":
[]}]
+ configs = BaseGateConfigs(client, repo=REPO, default_branch="main",
default_config=config)
+ unknown: list[tuple[int, str]] = []
+
+ discover(client, configs, repo=REPO, now=NOW, unknown=unknown,
ci_passed_written=[])
+
+ assert unknown == [(1, "`CI passed` not updated: GitHubError")]
+ assert client.rest_get.call_count == 2
+
+
def _contents(files: dict[tuple[str, str], str]):
def rest_get(path, paginate=False, items_key=None):
name, ref = path.removeprefix(f"repos/{REPO}/contents/").split("?ref=")
@@ -3450,3 +3670,546 @@ def
test_run_reconcile_reports_delayed_job_statuses_as_attention_items(
assert _reconcile_with_reporter(client, tmp_path) is True
assert "job statuses of fork run R are still not all posted" in (tmp_path
/ "s.json").read_text()
+
+
+def _passed_node(number=1, *, state=None, description=None, suites=(),
oid=None):
+ node = _project_node(number, status=state, suites=suites, oid=oid)
+ if state:
+
node["commits"]["nodes"][0]["commit"]["status"]["context"]["description"] =
description
+ return node
+
+
+def _passed_suite(
+ run_id,
+ *,
+ conclusion="SUCCESS",
+ created="2026-11-10T10:00:00Z",
+ cancelled=0,
+ not_run=0,
+ running=False,
+ suite_cancelled=False,
+ **kwargs,
+):
+ suite = _suite(run_id, created=created, **kwargs)
+ suite["status"] = "IN_PROGRESS" if running else "COMPLETED"
+ suite["conclusion"] = "CANCELLED" if suite_cancelled else None
+ suite["checkRuns"] = {"nodes": [{"name": "CI passed", "conclusion":
conclusion}] if conclusion else []}
+ suite["cancelled"] = {"totalCount": cancelled}
+ suite["notRun"] = {"totalCount": not_run}
+ return suite
+
+
+WAITING = (
+ "pending",
+ "https://github.com/apache/airflow/pull/1/checks";,
+ CI_PASSED_WAITING_PROJECT_CI_DESCRIPTION,
+)
+PASSED = ("success", PROJECT_PASSED_URL, CI_PASSED_PROJECT_DESCRIPTION)
+
+
[email protected](
+ ("node", "expected"),
+ [
+ pytest.param(
+ _passed_node(state="SUCCESS", description=CI_PASSED_DESCRIPTION),
+ WAITING,
+ id="fork-success-revoked",
+ ),
+ pytest.param(_passed_node(), None, id="no-status-left-alone"),
+ pytest.param(
+ _passed_node(suites=[_passed_suite(5)]), None,
id="no-status-is-not-created-after-project-ci"
+ ),
+ pytest.param(
+ _passed_node(
+ state="SUCCESS", description=CI_PASSED_DESCRIPTION,
suites=[_passed_suite(5, cancelled=1)]
+ ),
+ PASSED,
+ id="rerun-passed-next-to-the-marker-of-its-first-attempt",
+ ),
+ pytest.param(
+ _passed_node(
+ state="SUCCESS",
+ description=CI_PASSED_DESCRIPTION,
+ suites=[
+ _passed_suite(4, created="2026-11-10T09:00:00Z"),
+ _passed_suite(6, conclusion=None, running=True),
+ ],
+ ),
+ WAITING,
+ id="newest-run-still-running-is-not-passed",
+ ),
+ pytest.param(
+ _passed_node(
+ state="SUCCESS",
+ description=CI_PASSED_DESCRIPTION,
+ suites=[
+ _passed_suite(5, created="2026-11-10T09:00:00Z"),
+ _passed_suite(6, conclusion=None, suite_cancelled=True),
+ ],
+ ),
+ ("success", "https://github.com/apache/airflow/actions/runs/5";,
CI_PASSED_PROJECT_DESCRIPTION),
+ id="newer-empty-cancelled-run-does-not-hide-an-older-success",
+ ),
+ pytest.param(
+ _passed_node(
+ state="SUCCESS",
+ description=CI_PASSED_DESCRIPTION,
+ suites=[
+ _passed_suite(5, created="2026-11-10T09:00:00Z"),
+ _passed_suite(6, cancelled=1, conclusion=None),
+ ],
+ ),
+ ("success", "https://github.com/apache/airflow/actions/runs/5";,
CI_PASSED_PROJECT_DESCRIPTION),
+ id="newer-cancelled-run-does-not-hide-an-older-success",
+ ),
+ pytest.param(_passed_node(state="PENDING", description="x"), None,
id="pending-left-alone"),
+ pytest.param(
+ _passed_node(
+ state="SUCCESS", description=CI_PASSED_DESCRIPTION,
suites=[_passed_suite(5, conclusion=None)]
+ ),
+ WAITING,
+ id="project-ci-still-running",
+ ),
+ pytest.param(
+ _passed_node(
+ state="SUCCESS",
+ description=CI_PASSED_DESCRIPTION,
+ suites=[_passed_suite(5, cancelled=1, conclusion=None)],
+ ),
+ WAITING,
+ id="cancelled-run",
+ ),
+ pytest.param(
+ _passed_node(
+ state="SUCCESS",
+ description=CI_PASSED_DESCRIPTION,
+ suites=[_passed_suite(5, not_run=1, conclusion=None)],
+ ),
+ WAITING,
+ id="project-ci-not-run",
+ ),
+ pytest.param(
+ _passed_node(
+ state="SUCCESS",
+ description=CI_PASSED_DESCRIPTION,
+ suites=[_passed_suite(5, event="push"), _passed_suite(6,
path=".github/workflows/other.yml")],
+ ),
+ WAITING,
+ id="not-a-ci-amd-pull-request-run",
+ ),
+ pytest.param(
+ _passed_node(
+ state="SUCCESS",
+ description=CI_PASSED_DESCRIPTION,
+ suites=[
+ _passed_suite(4, created="2026-11-10T09:00:00Z"),
+ _passed_suite(5, conclusion="FAILURE"),
+ ],
+ ),
+ WAITING,
+ id="latest-run-failed",
+ ),
+ pytest.param(
+ _passed_node(state="SUCCESS", description=CI_PASSED_DESCRIPTION,
suites=[_passed_suite(5)]),
+ PASSED,
+ id="project-ci-passed",
+ ),
+ pytest.param(
+ _passed_node(state="PENDING", description="x",
suites=[_passed_suite(5)]),
+ PASSED,
+ id="project-ci-passed-after-pending",
+ ),
+ pytest.param(
+ _passed_node(
+ state="SUCCESS", description=CI_PASSED_PROJECT_DESCRIPTION,
suites=[_passed_suite(5)]
+ ),
+ None,
+ id="project-ci-success-already-posted",
+ ),
+ pytest.param(
+ _passed_node(
+ state="SUCCESS",
+ description=CI_PASSED_PROJECT_DESCRIPTION,
+ suites=[_passed_suite(5, conclusion="FAILURE")],
+ ),
+ WAITING,
+ id="project-ci-rerun-failed",
+ ),
+ ],
+)
+def test_report_project_ci_passed(client, node, expected):
+ client.graphql.return_value = _search_result(node)
+
+ assert report_project_ci_passed(client, repo=REPO, attention=[]) == ([1]
if expected else [])
+
+ if expected is None:
+ client.set_status.assert_not_called()
+ else:
+ client.set_status.assert_called_once_with(1, "sha1", *expected,
context=CI_PASSED_CONTEXT)
+
+
+def test_report_project_ci_passed_limits_successes_but_not_revocations(client):
+ client.graphql.return_value = _search_result(
+ *(_passed_node(n, state="PENDING", description="x",
suites=[_passed_suite(5)]) for n in (1, 2, 3)),
+ _passed_node(4, state="SUCCESS", description=CI_PASSED_DESCRIPTION),
+ )
+
+ assert report_project_ci_passed(client, repo=REPO, attention=[], budget=2)
== [1, 2, 4]
+ assert 'label:"use project ci"' in client.graphql.call_args.args[1]["q"]
+
+
+def test_report_project_ci_passed_keeps_going_after_an_error(client):
+ client.graphql.return_value = _search_result(
+ *(_passed_node(n, state="SUCCESS", description=CI_PASSED_DESCRIPTION)
for n in (1, 2))
+ )
+ client.set_status.side_effect = [GitHubError("gh api failed: HTTP 502"),
None]
+ attention: list[tuple[int, str]] = []
+
+ assert report_project_ci_passed(client, repo=REPO, attention=attention) ==
[2]
+ assert attention == [(1, "`CI passed` not updated: GitHubError")]
+
+
+def _rest_for_project_ci_passed(runs, checks):
+ """``rest_get`` of a commit with these workflow runs and (name,
conclusion, check suite id) checks."""
+
+ def rest_get(path, **_):
+ if "/actions/runs" in path:
+ assert "per_page=100" in path
+ return {"workflow_runs": runs}
+ assert "per_page=100" in path
+ return {
+ "check_runs": [
+ {"name": name, "conclusion": conclusion, "check_suite": {"id":
suite_id}}
+ for name, conclusion, suite_id in checks
+ if f"check_name={quote(name)}&" in path
+ ]
+ }
+
+ return rest_get
+
+
[email protected](
+ ("checks", "expected"),
+ [
+ pytest.param(
+ [("CI passed", "success", 8)],
"https://github.com/apache/airflow/actions/runs/6";, id="passed"
+ ),
+ pytest.param([("CI passed", "failure", 8)], None, id="failed"),
+ pytest.param([("CI passed", "success", 9)], None,
id="check-of-another-run"),
+ pytest.param(
+ [("CI passed", "success", 8), ("CI passed (run cancelled)",
"success", 8)],
+ "https://github.com/apache/airflow/actions/runs/6";,
+ id="rerun-passed-next-to-the-marker-of-its-first-attempt",
+ ),
+ pytest.param([("CI passed", "success", 7)], None,
id="newest-run-still-running-is-not-passed"),
+ pytest.param(
+ [("CI passed", "success", 7), ("CI passed (run cancelled)",
"success", 8)],
+ PROJECT_PASSED_URL,
+ id="newer-cancelled-run-does-not-hide-an-older-success",
+ ),
+ pytest.param(
+ [("CI passed", "success", 7), ("CI passed (project CI not run)",
"success", 8)],
+ PROJECT_PASSED_URL,
+ id="newer-not-run-marker-does-not-hide-an-older-success",
+ ),
+ ],
+)
+def test_read_project_ci_passed_url(client, checks, expected):
+ older = {**_run_json(5), "check_suite_id": 7, "html_url":
PROJECT_PASSED_URL}
+ newer = {**_run_json(6), "check_suite_id": 8}
+ client.rest_get.side_effect = _rest_for_project_ci_passed([newer, older],
checks)
+
+ assert read_project_ci_passed_url(client, REPO, "sha1") == expected
+
+
[email protected](
+ ("newer_status", "newer_conclusion", "expected"),
+ [
+ pytest.param("completed", "cancelled", PROJECT_PASSED_URL,
id="newer-empty-cancelled-run"),
+ pytest.param("in_progress", None, None,
id="newer-run-in-progress-without-checks"),
+ ],
+)
+def test_read_project_ci_passed_url_with_a_newer_run_without_checks(
+ client, newer_status, newer_conclusion, expected
+):
+ older = {**_run_json(5), "check_suite_id": 7, "html_url":
PROJECT_PASSED_URL}
+ newer = {**_run_json(6), "check_suite_id": 8, "status": newer_status,
"conclusion": newer_conclusion}
+ client.rest_get.side_effect = _rest_for_project_ci_passed([newer, older],
[("CI passed", "success", 7)])
+
+ assert read_project_ci_passed_url(client, REPO, "sha1") == expected
+
+
+def test_read_project_ci_passed_url_ignores_other_workflows(client):
+ client.rest_get.return_value = {"workflow_runs": [_run_json(5,
"other.yml")]}
+
+ assert read_project_ci_passed_url(client, REPO, "sha1") is None
+
+
[email protected](
+ ("passed", "expected"),
+ [pytest.param(True, PASSED, id="found-success"), pytest.param(False,
WAITING, id="none-found")],
+)
[email protected]("airflow_breeze.utils.fork_ci_reconcile.read_project_ci_passed_url",
autospec=True)
+def test_report_project_ci_passed_reads_a_head_commit_the_search_did_not_list(
+ mock_read, client, passed, expected
+):
+ mock_read.return_value = PROJECT_PASSED_URL if passed else None
+ client.graphql.return_value = _search_result(
+ _passed_node(state="SUCCESS", description=CI_PASSED_DESCRIPTION,
oid="older")
+ )
+ client.rest_get.return_value = [
+ {"context": CI_PASSED_CONTEXT, "state": "success", "description":
CI_PASSED_DESCRIPTION}
+ ]
+
+ report_project_ci_passed(client, repo=REPO, attention=[])
+
+ client.set_status.assert_called_once_with(1, "sha1", *expected,
context=CI_PASSED_CONTEXT)
+
+
+def test_find_project_ci_label_heads(client):
+ client.graphql.return_value = _search_result({"headRefOid": "a"},
{"headRefOid": "b"})
+
+ assert find_project_ci_label_heads(client, REPO) == {"a", "b"}
+ assert 'label:"use project ci"' in client.graphql.call_args.args[1]["q"]
+
+
+LIVE_PULL_REQUEST = {"head": {"sha": "shared"}, "labels": []}
+
+
+def _run_reconcile_with_green_pull_requests(
+ client,
+ tmp_path,
+ *numbers,
+ live=LIVE_PULL_REQUEST,
+ same_branch=(),
+ ci_passed=("success", RUN_URL, CI_PASSED_DESCRIPTION),
+ state=ForkState.GREEN,
+):
+ """Reconcile gated, fork-green pull requests that all have head commit
``shared``."""
+ nodes = [_pr_node(n, headRefOid="shared") for n in numbers]
+ client.graphql.side_effect = [
+ _search_result(),
+ _search_result(),
+ _search_result(*nodes),
+ {f"pr{index}": {"object": None} for index in range(len(nodes))},
+ ]
+ client.rest_get_pages.return_value = [{"total_count": 0, "workflow_runs":
[]}]
+ client.rest_get.side_effect = lambda path, **_: (
+ live
+ if "/pulls/" in path
+ else same_branch
+ if "/pulls?" in path
+ else []
+ if "/status" in path
+ else {"state": "active"}
+ )
+ plan = Plan(state=state, ci_passed=ci_passed)
+ with (
+ mock.patch("airflow_breeze.utils.fork_ci_reconcile.classify",
autospec=True, return_value=plan),
+ mock.patch("airflow_breeze.utils.fork_ci_reconcile.JobStatusReporter",
autospec=True) as reporter,
+
mock.patch("airflow_breeze.utils.fork_ci_reconcile.report_project_ci_runs_started",
autospec=True),
+ ):
+ reporter.return_value.report.return_value = None
+ reporter.return_value.deferred = []
+ reporter.return_value.delayed = []
+ run_reconcile(
+ client,
+ CONFIG,
+ now=NOW,
+ run_url="https://run";,
+ output_file=tmp_path / "s.json",
+ workflows_dir=tmp_path,
+ )
+
+
[email protected](
+ ("heads", "posted"),
+ [
+ pytest.param({"shared"}, False,
id="head-of-a-pull-request-with-the-label"),
+ pytest.param(set(), True, id="no-label"),
+ pytest.param(GitHubError("HTTP 502"), False, id="heads-unknown"),
+ ],
+)
+def test_run_reconcile_posts_no_fork_success_on_a_head_with_use_project_ci(
+ mock_project_ci_heads, client, tmp_path, heads, posted
+):
+ # Two pull requests on one head commit: statuses belong to the commit, so
one with the label
+ # makes the fork success of the other one count for it.
+ if isinstance(heads, Exception):
+ mock_project_ci_heads.side_effect = heads
+ else:
+ mock_project_ci_heads.return_value = heads
+
+ _run_reconcile_with_green_pull_requests(client, tmp_path, 1, 2)
+
+ posted_calls = [
+ c for c in client.set_status.call_args_list if c.kwargs.get("context")
== CI_PASSED_CONTEXT
+ ]
+ assert bool(posted_calls) is posted
+
+
[email protected](
+ "heads",
+ [
+ pytest.param(set(), id="label-not-yet-in-the-search-index"),
+ pytest.param(GitHubError("HTTP 502"), id="heads-unknown"),
+ pytest.param({"shared"}, id="head-already-in-the-label-search"),
+ ],
+)
+def
test_run_reconcile_revokes_a_fork_success_in_the_run_that_releases_to_project_ci(
+ mock_project_ci_heads, client, tmp_path, heads
+):
+ if isinstance(heads, Exception):
+ mock_project_ci_heads.side_effect = heads
+ else:
+ mock_project_ci_heads.return_value = heads
+ revoke = ("pending", f"https://github.com/{REPO}/pull/1/checks";,
CI_PASSED_WAITING_PROJECT_CI_DESCRIPTION)
+
+ _run_reconcile_with_green_pull_requests(client, tmp_path, 1,
state=ForkState.RELEASED, ci_passed=revoke)
+
+ posted_calls = [
+ c for c in client.set_status.call_args_list if c.kwargs.get("context")
== CI_PASSED_CONTEXT
+ ]
+ assert posted_calls == [mock.call(1, "shared", *revoke,
context=CI_PASSED_CONTEXT)]
+
+
[email protected](
+ "live",
+ [
+ pytest.param(
+ {"head": {"sha": "shared"}, "labels": [{"name": "use project
ci"}]}, id="labelled-since"
+ ),
+ pytest.param({"head": {"sha": "newer"}, "labels": []},
id="head-changed"),
+ ],
+)
+def
test_run_reconcile_reads_the_pull_request_live_before_posting_a_fork_success(client,
tmp_path, live):
+ _run_reconcile_with_green_pull_requests(client, tmp_path, 1, live=live)
+
+ client.set_status.assert_not_called()
+ client.rest_get.assert_any_call(f"repos/{REPO}/pulls/1")
+
+
+def
test_run_reconcile_reads_the_pull_requests_of_the_head_branch_before_posting_a_fork_success(
+ client, tmp_path
+):
+ other = {"head": {"sha": "shared"}, "labels": [{"name": "use project ci"}]}
+
+ _run_reconcile_with_green_pull_requests(client, tmp_path, 1,
same_branch=[other])
+
+ client.set_status.assert_not_called()
+ assert any(
+ "/pulls?state=open" in c.args[0] and "head=" in c.args[0] for c in
client.rest_get.call_args_list
+ )
+
+
+def test_run_reconcile_still_revokes_when_the_labelled_heads_are_unknown(
+ mock_project_ci_heads, client, tmp_path
+):
+ mock_project_ci_heads.side_effect = GitHubError("HTTP 502")
+ pending = ("pending", RUN_URL, "Fork CI result no longer trusted")
+
+ _run_reconcile_with_green_pull_requests(client, tmp_path, 1,
ci_passed=pending)
+
+ client.set_status.assert_called_once_with(1, "shared", *pending,
context=CI_PASSED_CONTEXT)
+
+
+def
test_report_project_ci_passed_revokes_a_pull_request_the_heads_search_missed(
+ mock_project_ci_heads, mock_report_project_ci_passed, client, tmp_path
+):
+ # The heads list was empty at the start of the run, as when the search
lags behind a new label.
+ mock_project_ci_heads.return_value = set()
+ mock_report_project_ci_passed.side_effect = report_project_ci_passed
+ client.graphql.side_effect = [
+ _search_result(),
+ _search_result(),
+ _search_result(),
+ _search_result(_passed_node(state="SUCCESS",
description=CI_PASSED_DESCRIPTION)),
+ ]
+ client.rest_get_pages.return_value = [{"total_count": 0, "workflow_runs":
[]}]
+
+ with
mock.patch("airflow_breeze.utils.fork_ci_reconcile.report_project_ci_runs_started",
autospec=True):
+ run_reconcile(
+ client,
+ CONFIG,
+ now=NOW,
+ run_url="https://run";,
+ output_file=tmp_path / "s.json",
+ workflows_dir=tmp_path,
+ )
+
+ client.set_status.assert_called_once_with(1, "sha1", *WAITING,
context=CI_PASSED_CONTEXT)
+
+
+def test_run_reconcile_still_revokes_a_fork_success_on_a_labelled_head(
+ mock_project_ci_heads, client, tmp_path
+):
+ mock_project_ci_heads.return_value = {"shared"}
+ revoke = ("pending", RUN_URL, "Fork CI result no longer trusted")
+
+ _run_reconcile_with_green_pull_requests(client, tmp_path, 1,
ci_passed=revoke)
+
+ client.set_status.assert_called_once_with(1, "shared", *revoke,
context=CI_PASSED_CONTEXT)
+
+
[email protected]("workflow_file", ["ci-amd.yml", "ci.yml",
"ci-amd-arm.yml"])
+def
test_find_project_ci_passed_url_accepts_the_workflow_files_of_release_branches(workflow_file):
+ node = _passed_node(
+ suites=[_passed_suite(5, path=f".github/workflows/{workflow_file}")],
state="SUCCESS", description="x"
+ )
+
+ assert find_project_ci_passed_url(node) == PROJECT_PASSED_URL
+
+
+def
test_read_project_ci_passed_url_accepts_the_workflow_files_of_release_branches(client):
+ run = {**_run_json(5, "ci.yml"), "check_suite_id": 7}
+ client.rest_get.side_effect = _rest_for_project_ci_passed([run], [("CI
passed", "success", 7)])
+
+ assert read_project_ci_passed_url(client, REPO, "sha1") ==
PROJECT_PASSED_URL
+
+
+def test_run_reconcile_reports_project_ci_passed_after_the_loop(
+ mock_project_ci_heads, mock_report_project_ci_passed, client, tmp_path
+):
+ order = mock.Mock()
+ order.attach_mock(client.set_status, "set_status")
+ order.attach_mock(mock_report_project_ci_passed, "report")
+ mock_project_ci_heads.return_value = {"other"}
+
+ _run_reconcile_with_green_pull_requests(client, tmp_path, 1)
+
+ names = [c[0] for c in order.mock_calls]
+ assert names.index("report") > names.index("set_status")
+ assert mock_report_project_ci_passed.call_args.kwargs["budget"] == 5
+
+
+def
test_run_reconcile_reports_project_ci_passed_failure(mock_report_project_ci_passed,
client, tmp_path):
+ client.graphql.side_effect = [_search_result(), _search_result(),
_search_result()]
+ client.rest_get_pages.return_value = [{"total_count": 0, "workflow_runs":
[]}]
+ mock_report_project_ci_passed.side_effect = GitHubError("gh api failed:
HTTP 502")
+
+ with mock.patch(
+
"airflow_breeze.utils.fork_ci_reconcile.report_project_ci_runs_started",
autospec=True
+ ) as mock_started:
+ mock_started.return_value = []
+ has_alerts = run_reconcile(
+ client,
+ CONFIG,
+ now=NOW,
+ run_url="https://run";,
+ output_file=tmp_path / "s.json",
+ workflows_dir=tmp_path,
+ )
+
+ assert has_alerts is True
+ assert (
+ "reporting `CI passed` for `use project ci` failed: GitHubError" in
(tmp_path / "s.json").read_text()
+ )
+
+
+def
test_report_project_ci_passed_counts_only_the_successes_that_were_written(client):
+ client.graphql.return_value = _search_result(
+ *(_passed_node(n, state="PENDING", description="x",
suites=[_passed_suite(5)]) for n in (1, 2, 3))
+ )
+ client.set_status.side_effect = [GitHubError("HTTP 502"), None, None]
+
+ assert report_project_ci_passed(client, repo=REPO, attention=[], budget=1)
== [2]
