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 4386cb1f89e Report EC2 Spot interruptions on pull requests and keep
them out of Slack (#75615)
4386cb1f89e is described below
commit 4386cb1f89e67c4d58a87a5e6965cf53ad58dfe3
Author: Jarek Potiuk <[email protected]>
AuthorDate: Sun Oct 25 19:18:29 2026 +0100
Report EC2 Spot interruptions on pull requests and keep them out of Slack
(#75615)
Jobs on the self-hosted RunsOn runners fail with exit code 137 when AWS
reclaims their Spot instance. Nothing in the job's log says why, so the
failure looks like a broken test or an out-of-memory kill, although RunsOn
re-runs the job by itself once the run finishes.
A pull request should say so as soon as it happens, and the canary should
not page anyone on Slack until RunsOn has given up retrying. The Slack
notification also never reported how RunsOn's re-run ended, because it only
ran for the first attempt.
Generated-by: Claude Opus 5
---
.github/workflows/ci-amd.yml | 35 +--
.github/workflows/ci-arm.yml | 35 +--
.github/workflows/spot-interruption-notice.yml | 57 ++++
dev/breeze/doc/ci/05_workflows.md | 43 +++
scripts/ci/spot_interruptions.py | 390 ++++++++++++++++++++++++
scripts/tests/ci/test_spot_interruptions.py | 404 +++++++++++++++++++++++++
6 files changed, 926 insertions(+), 38 deletions(-)
diff --git a/.github/workflows/ci-amd.yml b/.github/workflows/ci-amd.yml
index cd3d5f95eb2..49f08dbd367 100644
--- a/.github/workflows/ci-amd.yml
+++ b/.github/workflows/ci-amd.yml
@@ -1593,37 +1593,33 @@ jobs:
!cancelled() &&
github.repository == 'apache/airflow' &&
github.event_name == 'schedule' &&
- needs.check-run-needed.outputs.run-needed != 'false' &&
- github.run_attempt == 1
+ needs.check-run-needed.outputs.run-needed != 'false'
runs-on: ["ubuntu-slim"]
+ permissions:
+ # The run's jobs, the previous notification state, and the annotations
RunsOn puts on jobs
+ # stopped by an EC2 Spot interruption.
+ actions: read
+ checks: read
+ contents: read
steps:
- name: "Checkout ${{ github.ref }} ( ${{ github.sha }} )"
uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 #
v7.0.1
with:
persist-credentials: false
+ # Skips re-runs started by maintainers, and failures that are only EC2
Spot interruptions
+ # RunsOn will re-run; those leave the notification state alone, and the
re-run reports.
- name: "Get failing jobs"
id: get-failures
shell: bash
- run: |
- FAILED_JOBS=$(gh run view "${{ github.run_id }}" \
- --repo "${{ github.repository }}" \
- --json jobs \
- --jq '[.jobs[] | select(.conclusion == "failure") | .name] | sort
| .[]')
- echo "failed-jobs<<EOF" >> "${GITHUB_OUTPUT}"
- echo "${FAILED_JOBS}" >> "${GITHUB_OUTPUT}"
- echo "EOF" >> "${GITHUB_OUTPUT}"
- # A literal newline would end the double-quoted YAML scalar the
Slack payload
- # interpolates job names into, and the action would reject the whole
payload.
- echo "failed-jobs-text=${FAILED_JOBS//$'\n'/\\n}" >>
"${GITHUB_OUTPUT}"
- if [[ -n "${FAILED_JOBS}" ]]; then
- echo "has-failures=true" >> "${GITHUB_OUTPUT}"
- else
- echo "has-failures=false" >> "${GITHUB_OUTPUT}"
- fi
+ run: python3 scripts/ci/spot_interruptions.py slack
env:
- GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+ GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+ RUN_ID: ${{ github.run_id }}
+ RUN_ATTEMPT: ${{ github.run_attempt }}
+ TRIGGERING_ACTOR: ${{ github.triggering_actor }}
- name: "Determine notification action"
id: notification
+ if: steps.get-failures.outputs.skip != 'true'
shell: bash
run: |
command -v gh >/dev/null 2>&1 || { echo "::error::gh CLI is not
installed"; exit 1; }
@@ -1634,6 +1630,7 @@ jobs:
CURRENT_FAILURES: "${{ steps.get-failures.outputs.failed-jobs }}"
GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
- name: "Upload notification state"
+ if: steps.get-failures.outputs.skip != 'true'
uses: actions/upload-artifact@cf430e030ddbb5b0abf93d22962f4752f3646cd9
# v7.0.2
with:
name: "slack-state-tests-${{ github.ref_name }}-amd"
diff --git a/.github/workflows/ci-arm.yml b/.github/workflows/ci-arm.yml
index f10c1f07716..e6299a05ece 100644
--- a/.github/workflows/ci-arm.yml
+++ b/.github/workflows/ci-arm.yml
@@ -1474,37 +1474,33 @@ jobs:
!cancelled() &&
github.repository == 'apache/airflow' &&
github.event_name == 'schedule' &&
- needs.check-run-needed.outputs.run-needed != 'false' &&
- github.run_attempt == 1
+ needs.check-run-needed.outputs.run-needed != 'false'
runs-on: ["ubuntu-slim"]
+ permissions:
+ # The run's jobs, the previous notification state, and the annotations
RunsOn puts on jobs
+ # stopped by an EC2 Spot interruption.
+ actions: read
+ checks: read
+ contents: read
steps:
- name: "Checkout ${{ github.ref }} ( ${{ github.sha }} )"
uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 #
v7.0.1
with:
persist-credentials: false
+ # Skips re-runs started by maintainers, and failures that are only EC2
Spot interruptions
+ # RunsOn will re-run; those leave the notification state alone, and the
re-run reports.
- name: "Get failing jobs"
id: get-failures
shell: bash
- run: |
- FAILED_JOBS=$(gh run view "${{ github.run_id }}" \
- --repo "${{ github.repository }}" \
- --json jobs \
- --jq '[.jobs[] | select(.conclusion == "failure") | .name] | sort
| .[]')
- echo "failed-jobs<<EOF" >> "${GITHUB_OUTPUT}"
- echo "${FAILED_JOBS}" >> "${GITHUB_OUTPUT}"
- echo "EOF" >> "${GITHUB_OUTPUT}"
- # A literal newline would end the double-quoted YAML scalar the
Slack payload
- # interpolates job names into, and the action would reject the whole
payload.
- echo "failed-jobs-text=${FAILED_JOBS//$'\n'/\\n}" >>
"${GITHUB_OUTPUT}"
- if [[ -n "${FAILED_JOBS}" ]]; then
- echo "has-failures=true" >> "${GITHUB_OUTPUT}"
- else
- echo "has-failures=false" >> "${GITHUB_OUTPUT}"
- fi
+ run: python3 scripts/ci/spot_interruptions.py slack
env:
- GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+ GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+ RUN_ID: ${{ github.run_id }}
+ RUN_ATTEMPT: ${{ github.run_attempt }}
+ TRIGGERING_ACTOR: ${{ github.triggering_actor }}
- name: "Determine notification action"
id: notification
+ if: steps.get-failures.outputs.skip != 'true'
shell: bash
run: |
command -v gh >/dev/null 2>&1 || { echo "::error::gh CLI is not
installed"; exit 1; }
@@ -1515,6 +1511,7 @@ jobs:
CURRENT_FAILURES: "${{ steps.get-failures.outputs.failed-jobs }}"
GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
- name: "Upload notification state"
+ if: steps.get-failures.outputs.skip != 'true'
uses: actions/upload-artifact@cf430e030ddbb5b0abf93d22962f4752f3646cd9
# v7.0.2
with:
name: "slack-state-tests-${{ github.ref_name }}-arm"
diff --git a/.github/workflows/spot-interruption-notice.yml
b/.github/workflows/spot-interruption-notice.yml
new file mode 100644
index 00000000000..df66e33f7d1
--- /dev/null
+++ b/.github/workflows/spot-interruption-notice.yml
@@ -0,0 +1,57 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+---
+# Tells pull requests when jobs of their "Tests (AMD)" run were stopped by an
EC2 Spot interruption
+# of their RunsOn runner, that RunsOn re-runs them, and later how the re-run
ended. It runs on a
+# schedule rather than on the runs' events, so no pull request run ever gets
write access, and a
+# notice appears while the run is still going. It never checks out or runs
pull request code. See
+# scripts/ci/spot_interruptions.py.
+name: "Spot interruption notice"
+on: # yamllint disable-line rule:truthy
+ schedule:
+ - cron: '*/10 * * * *'
+ workflow_dispatch:
+permissions:
+ # The runs and jobs of "Tests (AMD)", and the annotations RunsOn puts on
jobs stopped by an EC2
+ # Spot interruption.
+ actions: read
+ checks: read
+ contents: read
+ # The notice comment on pull requests.
+ pull-requests: write
+concurrency:
+ group: spot-interruption-notice
+ cancel-in-progress: false
+jobs:
+ notice:
+ name: "Report EC2 Spot interruptions on pull requests"
+ if: github.repository == 'apache/airflow'
+ runs-on: ["ubuntu-slim"]
+ timeout-minutes: 10
+ steps:
+ - name: "Checkout main"
+ uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 #
v7.0.1
+ with:
+ ref: main
+ persist-credentials: false
+ sparse-checkout: scripts/ci/spot_interruptions.py
+ sparse-checkout-cone-mode: false
+ - name: "Post or update the notices"
+ env:
+ GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+ run: python3 scripts/ci/spot_interruptions.py pr-comments
diff --git a/dev/breeze/doc/ci/05_workflows.md
b/dev/breeze/doc/ci/05_workflows.md
index eca830bcb7d..7ec7871993c 100644
--- a/dev/breeze/doc/ci/05_workflows.md
+++ b/dev/breeze/doc/ci/05_workflows.md
@@ -451,6 +451,49 @@ unprivileged, which constrains what can go on it:
When adding a job, reach for `ubuntu-slim` if it only shuffles metadata
around, and
`ubuntu-22.04` otherwise.
+### Spot interruptions of self-hosted runners
+
+Runs of `apache/airflow` committers and the canary run on self-hosted
+[RunsOn](https://runs-on.com) runners, which are EC2 Spot instances. When AWS
reclaims
+one, RunsOn fails the job with an error annotation titled "EC2 Spot
interruption", and
+once the run finishes it re-runs the failed jobs, after attempts 1 and 2;
attempt 3 is
+its last. An interrupted job ends with exit code 137 and no test failure in
its log.
+
+- **Pull requests.** `spot-interruption-notice.yml` checks the newest "Tests
(AMD)" run of
+ each pull request every 10 minutes, including runs still in progress, and
keeps one
+ comment on the pull request: which jobs were interrupted and that RunsOn
re-runs them,
+ and, once the re-run finished, whether it passed. It runs on a schedule, not
on the
+ runs' events, so no pull request run gets write access.
+- **Slack.** The `notify-slack` job of a canary does not post when every
failure is a Spot
+ interruption that RunsOn re-runs, and leaves the notification state alone;
it posts for
+ RunsOn's re-run instead, including when the re-run passed. It still posts
for attempt
+ 3, and for any other failure. A re-run started by a maintainer does not post.
+
+Both use `scripts/ci/spot_interruptions.py`, which recognises an interruption
only by the
+annotation on a job that ran on a RunsOn runner: a pull request's run executes
the pull
+request's workflow file, which could add such an annotation, but external
contributors'
+runs use GitHub-hosted runners. A notice names the run it is about, and only
that run's
+re-run reports into it. Completed runs are looked at for 7 days after they
were created,
+so the outcome of a re-run of an older run is not reported.
+
+**API budget of the notice.** Each poll uses the workflow's `GITHUB_TOKEN`,
whose limit for
+`apache/airflow`, a repository of an Enterprise Cloud account, is 15,000
requests an hour,
+shared by every workflow of the repository (it would be 1,000 outside an
enterprise). Measured
+in October 2026, with 120-340 pull request runs of "Tests (AMD)" completing a
day, a poll makes
+about 45-55 requests:
+
+| Requests | Per poll |
+|---|---|
+| Runs in progress | 1 |
+| Runs completed in the last 7 days, listed a day at a time (100 per page) |
~22 |
+| Jobs of each run in progress or changed in the last 2 hours (50 per page;
~140 jobs in a full run) | ~15-27 |
+| Annotations of failed jobs on RunsOn runners, pull request lookups, comments
| ~5, more while runs are interrupted |
+
+That is about 270-330 requests an hour at one poll every 10 minutes, about 2%
of the limit.
+Most of it is the listing of completed runs, which grows with the 7-day
window: shortening
+it to 3 days saves about 12 requests a poll. A day with more than 1,000
completed runs would
+exceed what GitHub lists for one query, and the script then warns that older
runs are missed.
+
## Implementation Details
Here's how the composite workflow system is organized in practice.
diff --git a/scripts/ci/spot_interruptions.py b/scripts/ci/spot_interruptions.py
new file mode 100755
index 00000000000..d2994c67651
--- /dev/null
+++ b/scripts/ci/spot_interruptions.py
@@ -0,0 +1,390 @@
+#!/usr/bin/env python3
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+# /// script
+# requires-python = ">=3.11"
+# ///
+"""
+Tell failures caused by EC2 Spot interruptions of RunsOn runners apart from
real ones.
+
+RunsOn fails a job whose Spot instance AWS reclaimed, marks it with an
annotation titled
+"EC2 Spot interruption", and re-runs the failed jobs of the run once it
finishes, for attempts
+1 and 2. Those failures are not worth anyone's attention until RunsOn gives up.
+
+Modes:
+
+ slack For the Slack notification of a scheduled run. Writes to
GITHUB_OUTPUT:
+ failed-jobs newline-separated failed jobs, spot ones marked
as such
+ failed-jobs-text the same, with escaped newlines, for a Slack
payload
+ has-failures "true" when any job failed
+ skip "true" when nothing is posted and no state is
recorded: for a
+ re-run not started by RunsOn, or when every
failure is a spot
+ interruption that RunsOn will re-run
+ Environment: GITHUB_REPOSITORY, RUN_ID, RUN_ATTEMPT,
TRIGGERING_ACTOR
+
+ pr-comments For the newest CI run of each pull request, in progress or
recently completed,
+ keeps one comment on the pull request: which jobs were stopped
by a spot
+ interruption and whether RunsOn re-runs them, and, once a re-run
finished, how it
+ ended. Run on a schedule, so a pull request run never gets write
access, and a
+ notice appears as soon as a job is interrupted.
+ Environment: GITHUB_REPOSITORY
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import re
+import subprocess
+import sys
+from dataclasses import dataclass
+from datetime import UTC, datetime, timedelta
+from urllib.parse import quote
+
+SPOT_ANNOTATION_TITLE = "EC2 Spot interruption"
+# The annotation is trusted only on a RunsOn runner. A pull request's run
executes the pull
+# request's workflow file, which could add an annotation with that title, but
runs of external
+# contributors use GitHub-hosted runners.
+RUNS_ON_RUNNER_PREFIX = "runs-on--"
+# RunsOn re-runs the failed jobs of attempts 1 and 2; attempt 3 is its last.
+LAST_RETRIED_ATTEMPT = 2
+# Fails whenever any other job failed, so it is no failure of its own when the
others were spot.
+CI_PASSED_JOB = "CI passed"
+# Starts the notice; it records the run it is about, so another run never
reports into it.
+COMMENT_MARKER = "<!-- spot-interruption-notice"
+NOTICE_RUN = re.compile(r"^<!-- spot-interruption-notice run=(\d+) -->")
+# The GitHub App RunsOn re-runs the failed jobs of an interrupted run as.
+RUNS_ON_BOT = "runsonairflow[bot]"
+CI_WORKFLOW = "ci-amd.yml"
+# A completed run is looked at while it changed recently: RunsOn's re-run may
finish long after the
+# run was created, so the creation window is wide and the update window is
what bounds the work.
+CREATED_WINDOW = timedelta(days=7)
+UPDATED_WINDOW = timedelta(hours=2)
+# The author of the notice; a contributor's comment starting with the marker
is not the notice.
+COMMENT_AUTHOR = "github-actions[bot]"
+# A job cancelled while its run was not, e.g. by its timeout, is a failure too.
+FAILED_CONCLUSIONS = ("failure", "cancelled", "timed_out")
+
+
+@dataclass(frozen=True)
+class FailedJob:
+ id: int
+ name: str
+ url: str
+ spot: bool
+ conclusion: str = "failure"
+
+
+def gh_api(*args: str) -> str:
+ return subprocess.run(["gh", "api", *args], check=True,
capture_output=True, text=True).stdout
+
+
+def is_spot_interrupted(repo: str, job_id: int) -> bool:
+ titles = gh_api(
+ "--paginate",
f"repos/{repo}/check-runs/{job_id}/annotations?per_page=100", "--jq",
".[].title"
+ )
+ return SPOT_ANNOTATION_TITLE in titles.splitlines()
+
+
+def get_failed_jobs(repo: str, run_id: str, attempt: int | None = None) ->
list[FailedJob]:
+ """Return the failed jobs of an attempt of the run, by default each job's
latest attempt."""
+ jobs_path = f"attempts/{attempt}/jobs?" if attempt else
"jobs?filter=latest&"
+ # Pages of 100 jobs of a full CI run time out on GitHub's side (HTTP 502).
+ output = gh_api(
+ "--paginate",
+ f"repos/{repo}/actions/runs/{run_id}/{jobs_path}per_page=50",
+ "--jq",
+ ".jobs[] | {id, name, html_url, conclusion, runner_name}",
+ )
+ jobs = [json.loads(line) for line in output.splitlines() if line.strip()]
+ jobs = [job for job in jobs if job["conclusion"] in FAILED_CONCLUSIONS]
+ return [
+ FailedJob(
+ job["id"],
+ job["name"],
+ job["html_url"],
+ # RunsOn fails an interrupted job; one that was cancelled was not
interrupted.
+ job["conclusion"] == "failure"
+ and (job.get("runner_name") or
"").startswith(RUNS_ON_RUNNER_PREFIX)
+ and is_spot_interrupted(repo, job["id"]),
+ job["conclusion"],
+ )
+ for job in jobs
+ ]
+
+
+def find_real_failures(jobs: list[FailedJob]) -> list[FailedJob]:
+ real = [job for job in jobs if not job.spot]
+ if any(job.spot for job in jobs):
+ real = [job for job in real if not job.name.startswith(CI_PASSED_JOB)]
+ return real
+
+
+def should_skip_slack(jobs: list[FailedJob], attempt: int, triggering_actor:
str) -> bool:
+ # A maintainer's re-run is not reported; RunsOn's re-run is, as the
attempt before it was not.
+ if attempt > 1 and triggering_actor != RUNS_ON_BOT:
+ return True
+ spot_only = any(job.spot for job in jobs) and not find_real_failures(jobs)
+ return spot_only and attempt <= LAST_RETRIED_ATTEMPT
+
+
+def format_failed_jobs(jobs: list[FailedJob]) -> str:
+ def describe(job: FailedJob) -> str:
+ if job.spot:
+ return f"{job.name} (spot instance reclaimed)"
+ if job.conclusion != "failure":
+ return f"{job.name} ({job.conclusion.replace('_', ' ')})"
+ return job.name
+
+ return "\n".join(describe(job) for job in sorted(jobs, key=lambda job:
job.name))
+
+
+def format_job_name(name: str) -> str:
+ """Make a job name safe for a code span: no backtick or line break to end
it early."""
+ return re.sub(r"\s+", " ", name.replace("`", "")).strip()
+
+
+def build_comment(
+ jobs: list[FailedJob], *, run_id: int, attempt: int, conclusion: str,
run_url: str
+) -> str | None:
+ """Return the comment for this attempt, or None when it has nothing to
say."""
+ marker = f"{COMMENT_MARKER} run={run_id} -->"
+ spot_jobs = sorted((job for job in jobs if job.spot), key=lambda job:
job.name)
+ if not spot_jobs:
+ if attempt == 1:
+ return None
+ outcome = "passed" if conclusion == "success" else "failed"
+ return (
+ f"{marker}\n"
+ f"The re-run of the jobs stopped by EC2 Spot interruptions
finished: [attempt {attempt}]"
+ f"({run_url}/attempts/{attempt}) {outcome}."
+ )
+ # Job names come from the pull request's workflow file: in a code span
they cannot render
+ # markdown or mention anyone.
+ job_list = "\n".join(f"- [`{format_job_name(job.name)}`]({job.url})" for
job in spot_jobs)
+ if attempt <= LAST_RETRIED_ATTEMPT:
+ what_next = "RunsOn re-runs them automatically when the run finishes;
no action is needed."
+ else:
+ what_next = (
+ "RunsOn does not re-run them again after this attempt. A
maintainer can re-run the failed jobs."
+ )
+ return (
+ f"{marker}\n"
+ f"⚠️ {len(spot_jobs)} job(s) of [attempt
{attempt}]({run_url}/attempts/{attempt}) were stopped "
+ "because AWS reclaimed their EC2 Spot instances. These are not test
failures.\n\n"
+ f"{job_list}\n\n{what_next}"
+ )
+
+
+def find_pull_requests(repo: str, run: dict) -> list[int]:
+ """Return the open pull requests the run tested."""
+ if run["pr_numbers"]:
+ return run["pr_numbers"]
+ # GitHub does not attach pull requests from forks to their runs. Several
pull requests of one
+ # branch, against different base branches, are told apart by the commit
the run tested; when
+ # they share it, the run's pull request is unknown and nothing is posted.
+ output = gh_api(
+ "-X",
+ "GET",
+ f"repos/{repo}/pulls",
+ "-f",
+ f"head={run['head_owner']}:{run['head_branch']}",
+ "-f",
+ "state=open",
+ "--jq",
+ ".[] | {number, sha: .head.sha}",
+ )
+ pulls = [json.loads(line) for line in output.splitlines() if line.strip()]
+ matching = [pull["number"] for pull in pulls if pull["sha"] ==
run["head_sha"]]
+ return matching if len(matching) == 1 else []
+
+
+@dataclass(frozen=True)
+class Comment:
+ id: int
+ body: str
+
+ @property
+ def run_id(self) -> int | None:
+ match = NOTICE_RUN.match(self.body)
+ return int(match.group(1)) if match else None
+
+
+def find_comment(repo: str, pr_number: int) -> Comment | None:
+ output = gh_api(
+ "--paginate",
+ f"repos/{repo}/issues/{pr_number}/comments?per_page=100",
+ "--jq",
+ ".[] | {id, body, author: .user.login}",
+ )
+ comments = [json.loads(line) for line in output.splitlines() if
line.strip()]
+ notices = [
+ Comment(comment["id"], comment["body"])
+ for comment in comments
+ if comment["author"] == COMMENT_AUTHOR and
comment["body"].startswith(COMMENT_MARKER)
+ ]
+ return notices[-1] if notices else None
+
+
+def write_comment(repo: str, pr_number: int, existing: Comment | None, body:
str) -> None:
+ if existing is not None:
+ gh_api("-X", "PATCH", f"repos/{repo}/issues/comments/{existing.id}",
"-f", f"body={body}")
+ else:
+ gh_api("-X", "POST", f"repos/{repo}/issues/{pr_number}/comments",
"-f", f"body={body}")
+
+
+def write_output(name: str, value: str) -> None:
+ if output_file := os.environ.get("GITHUB_OUTPUT"):
+ with open(output_file, "a") as f:
+ f.write(f"{name}<<EOF\n{value}\nEOF\n")
+ print(f"{name}={value}")
+
+
+def run_slack(repo: str, run_id: str, attempt: int) -> None:
+ jobs = get_failed_jobs(repo, run_id)
+ failed_jobs = format_failed_jobs(jobs)
+ write_output("failed-jobs", failed_jobs)
+ # A literal newline would end the double-quoted YAML scalar the Slack
payload interpolates
+ # job names into, and the action would reject the whole payload.
+ write_output("failed-jobs-text", failed_jobs.replace("\n", "\\n"))
+ write_output("has-failures", "true" if jobs else "false")
+ skip = should_skip_slack(jobs, attempt, os.environ.get("TRIGGERING_ACTOR",
""))
+ write_output("skip", "true" if skip else "false")
+
+
+# GitHub returns at most this many runs for a filtered listing, however many
pages are read.
+RUN_LISTING_LIMIT = 1000
+
+
+def list_runs(repo: str, query: str) -> list[dict]:
+ output = gh_api(
+ "--paginate",
+
f"repos/{repo}/actions/workflows/{CI_WORKFLOW}/runs?event=pull_request&per_page=100&{query}",
+ "--jq",
+ ".workflow_runs[] | {id, run_attempt, status, conclusion, html_url,
head_branch, head_sha, "
+ "created_at, updated_at, head_owner: .head_repository.owner.login, "
+ "triggering_actor: .triggering_actor.login, pr_numbers:
[.pull_requests[].number]}",
+ )
+ runs = [json.loads(line) for line in output.splitlines() if line.strip()]
+ if len(runs) >= RUN_LISTING_LIMIT:
+ print(f"Run listing {query} reached GitHub's limit; older runs are
missed.", file=sys.stderr)
+ return runs
+
+
+def list_pull_request_runs(repo: str, now: datetime) -> list[dict]:
+ """Return the newest CI run of each pull request, when it is in progress
or changed recently."""
+ updated_since = (now - UPDATED_WINDOW).strftime("%Y-%m-%dT%H:%M:%SZ")
+ runs = list_runs(repo, "status=in_progress")
+ # A day at a time, so that no listing reaches the limit.
+ for day in range(CREATED_WINDOW.days):
+ end = now - timedelta(days=day)
+ start = end - timedelta(days=1)
+ created = f"{start:%Y-%m-%dT%H:%M:%SZ}..{end:%Y-%m-%dT%H:%M:%SZ}"
+ runs.extend(list_runs(repo,
f"status=completed&created={quote(created)}"))
+ # A run is identified by the pull requests GitHub attaches to it; runs of
pull requests from
+ # forks have none attached, and are told apart by their branch.
+ newest: dict[tuple, dict] = {}
+ for run in runs:
+ keys: list[tuple[str, ...]] = [("pr", str(number)) for number in
run["pr_numbers"]] or [
+ ("branch", run["head_owner"], run["head_branch"])
+ ]
+ for key in keys:
+ if key not in newest or run["created_at"] >
newest[key]["created_at"]:
+ newest[key] = run
+ recent = {
+ run["id"]: run
+ for run in newest.values()
+ if run["status"] != "completed" or run["updated_at"] >= updated_since
+ }
+ return list(recent.values())
+
+
+def update_notice(repo: str, run: dict) -> None:
+ """Post or update the notice for the pull request of a run, when there is
anything to say."""
+ run_id, attempt, completed = run["id"], run["run_attempt"], run["status"]
== "completed"
+ retried_by_runs_on = attempt > 1 and run.get("triggering_actor") ==
RUNS_ON_BOT
+ jobs = get_failed_jobs(repo, str(run_id))
+ has_spot = any(job.spot for job in jobs)
+ # Without interruptions in this attempt, there is only the outcome of a
finished re-run, or the
+ # interruption RunsOn is re-running, to report.
+ if not has_spot and attempt == 1:
+ return
+ if not has_spot and not completed and not retried_by_runs_on:
+ return
+ for pr_number in find_pull_requests(repo, run):
+ update_notice_on(repo, pr_number, run, jobs)
+
+
+def update_notice_on(repo: str, pr_number: int, run: dict, jobs:
list[FailedJob]) -> None:
+ run_id, attempt, completed = run["id"], run["run_attempt"], run["status"]
== "completed"
+ retried_by_runs_on = attempt > 1 and run.get("triggering_actor") ==
RUNS_ON_BOT
+ has_spot = any(job.spot for job in jobs)
+ existing = find_comment(repo, pr_number)
+ notice_of_run = existing is not None and existing.run_id == run_id
+
+ def build(failed: list[FailedJob], for_attempt: int) -> str | None:
+ return build_comment(
+ failed,
+ run_id=run_id,
+ attempt=for_attempt,
+ conclusion=run["conclusion"] or "",
+ run_url=run["html_url"],
+ )
+
+ if has_spot:
+ body = build(jobs, attempt)
+ elif notice_of_run:
+ # The outcome of a re-run is reported only into the notice about that
run.
+ body = build(jobs, attempt) if completed else None
+ elif retried_by_runs_on:
+ # RunsOn started its re-run between two polls, so the interruption is
in the attempt before.
+ previous = get_failed_jobs(repo, str(run_id), attempt - 1)
+ body = build(previous, attempt - 1) if any(job.spot for job in
previous) else None
+ else:
+ body = None
+ if body is None or (existing is not None and existing.body == body):
+ return
+ print(f"#{pr_number}: notice for run {run['html_url']} attempt {attempt}")
+ write_comment(repo, pr_number, existing, body)
+
+
+def run_pr_comments(repo: str, now: datetime) -> bool:
+ """Update the notices; return whether every run could be handled."""
+ ok = True
+ for run in list_pull_request_runs(repo, now):
+ try:
+ update_notice(repo, run)
+ except subprocess.CalledProcessError as e:
+ print(f"Could not handle run {run['html_url']}: {e.stderr or e}",
file=sys.stderr)
+ ok = False
+ return ok
+
+
+def main(argv: list[str]) -> int:
+ if len(argv) != 1 or argv[0] not in ("slack", "pr-comments"):
+ print("Usage: spot_interruptions.py slack|pr-comments",
file=sys.stderr)
+ return 2
+ repo = os.environ["GITHUB_REPOSITORY"]
+ if argv[0] == "slack":
+ run_slack(repo, os.environ["RUN_ID"], int(os.environ["RUN_ATTEMPT"]))
+ elif not run_pr_comments(repo, datetime.now(tz=UTC)):
+ return 1
+ return 0
+
+
+if __name__ == "__main__":
+ sys.exit(main(sys.argv[1:]))
diff --git a/scripts/tests/ci/test_spot_interruptions.py
b/scripts/tests/ci/test_spot_interruptions.py
new file mode 100644
index 00000000000..24d72c84664
--- /dev/null
+++ b/scripts/tests/ci/test_spot_interruptions.py
@@ -0,0 +1,404 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+import importlib.util
+import json
+import subprocess
+import sys
+from datetime import UTC, datetime
+from pathlib import Path
+from unittest import mock
+
+import pytest
+
+MODULE_PATH = Path(__file__).resolve().parents[3] / "scripts" / "ci" /
"spot_interruptions.py"
+RUN_URL = "https://github.com/apache/airflow/actions/runs/7";
+
+
[email protected]
+def spot():
+ module_name = "test_spot_interruptions_module"
+ sys.modules.pop(module_name, None)
+ spec = importlib.util.spec_from_file_location(module_name, MODULE_PATH)
+ assert spec is not None
+ assert spec.loader is not None
+ module = importlib.util.module_from_spec(spec)
+ sys.modules[module_name] = module
+ spec.loader.exec_module(module)
+ return module
+
+
+def _job(spot, name, *, interrupted):
+ """A failed job; `interrupted` is True for a spot interruption, or a
conclusion other than failure."""
+ if isinstance(interrupted, str):
+ return spot.FailedJob(hash(name), name, f"https://job/{name}";,
spot=False, conclusion=interrupted)
+ return spot.FailedJob(hash(name), name, f"https://job/{name}";,
spot=interrupted)
+
+
+def test_get_failed_jobs_marks_spot_interruptions_by_annotation(spot):
+ responses = {
+ "jobs": "\n".join(
+ json.dumps(job)
+ for job in [
+ {"id": 1, "name": "Tests A", "html_url": "https://job/1";,
"conclusion": "failure"},
+ {"id": 2, "name": "Tests B", "html_url": "https://job/2";,
"conclusion": "failure"},
+ {"id": 3, "name": "Tests C", "html_url": "https://job/3";,
"conclusion": "cancelled"},
+ {"id": 4, "name": "Tests D", "html_url": "https://job/4";,
"conclusion": "success"},
+ # A pull request's own workflow could add the annotation on a
GitHub-hosted runner.
+ {"id": 5, "name": "Tests E", "html_url": "https://job/5";,
"conclusion": "failure"},
+ ]
+ ),
+ # Titles of all pages of annotations, one per line.
+ "repos/apache/airflow/check-runs/1/annotations?per_page=100":
"\nProcess completed\nEC2 Spot interruption",
+ "repos/apache/airflow/check-runs/2/annotations?per_page=100": "Process
completed",
+ }
+
+ runners = {1: "runs-on--i-0123--abc", 2: "runs-on--i-0456--def", 5:
"GitHub Actions 1000"}
+ responses["jobs"] = "\n".join(
+ json.dumps({**json.loads(line), "runner_name":
runners.get(json.loads(line)["id"])})
+ for line in responses["jobs"].splitlines()
+ )
+
+ def gh_api(*args):
+ return responses["jobs"] if "/jobs?" in args[1] else responses[args[1]]
+
+ with mock.patch.object(spot, "gh_api", side_effect=gh_api):
+ jobs = spot.get_failed_jobs("apache/airflow", "7")
+
+ # A cancelled job is a failure, but was not interrupted, so its
annotations are not read.
+ assert [(job.name, job.spot, job.conclusion) for job in jobs] == [
+ ("Tests A", True, "failure"),
+ ("Tests B", False, "failure"),
+ ("Tests C", False, "cancelled"),
+ ("Tests E", False, "failure"),
+ ]
+
+
+SPOT_ONLY = [("Tests A", True), ("CI passed", False)]
+BOT = "runsonairflow[bot]"
+
+
[email protected](
+ ("failed", "attempt", "actor", "expected_skip"),
+ [
+ pytest.param(SPOT_ONLY, 1, "github-actions", True,
id="spot-only-retried"),
+ pytest.param(SPOT_ONLY, 2, BOT, True, id="spot-only-retried-again"),
+ pytest.param(SPOT_ONLY, 3, BOT, False, id="runs-on-gave-up"),
+ pytest.param(
+ [("Tests A", True), ("Tests B", False), ("CI passed", False)],
+ 1,
+ "github-actions",
+ False,
+ id="real-failure-too",
+ ),
+ pytest.param([("CI passed", False)], 1, "github-actions", False,
id="ci-passed-alone-is-real"),
+ pytest.param(
+ [("Tests A", True), ("Tests C", "cancelled"), ("CI passed",
False)],
+ 1,
+ "github-actions",
+ False,
+ id="spot-and-timed-out-job",
+ ),
+ pytest.param([], 1, "github-actions", False, id="no-failures"),
+ pytest.param([], 2, BOT, False, id="runs-on-re-run-recovered"),
+ pytest.param([("Tests B", False)], 2, "a-maintainer", True,
id="maintainer-re-run"),
+ ],
+)
+def test_should_skip_slack(spot, failed, attempt, actor, expected_skip):
+ jobs = [_job(spot, name, interrupted=interrupted) for name, interrupted in
failed]
+
+ assert spot.should_skip_slack(jobs, attempt, actor) is expected_skip
+
+
+def test_format_failed_jobs_marks_spot_and_cancelled_jobs(spot):
+ jobs = [
+ _job(spot, "Tests B", interrupted=False),
+ _job(spot, "Tests A", interrupted=True),
+ _job(spot, "Tests C", interrupted="timed_out"),
+ ]
+
+ assert spot.format_failed_jobs(jobs) == "Tests A (spot instance
reclaimed)\nTests B\nTests C (timed out)"
+
+
[email protected](
+ ("failed", "attempt", "conclusion", "expected"),
+ [
+ pytest.param(
+ [("Tests A", True)], 1, "failure", ["attempt 1", "Tests A",
"re-runs them"], id="retried"
+ ),
+ pytest.param([("Tests A", True)], 3, "failure", ["attempt 3", "does
not re-run"], id="gave-up"),
+ pytest.param([], 2, "success", ["attempt 2", "passed"],
id="re-run-passed"),
+ pytest.param([("Tests B", False)], 2, "failure", ["attempt 2",
"failed"], id="re-run-failed"),
+ pytest.param([("Tests B", False)], 1, "failure", None,
id="no-spot-first-attempt"),
+ ],
+)
+def test_build_comment(spot, failed, attempt, conclusion, expected):
+ jobs = [_job(spot, name, interrupted=interrupted) for name, interrupted in
failed]
+
+ body = spot.build_comment(jobs, run_id=7, attempt=attempt,
conclusion=conclusion, run_url=RUN_URL)
+
+ if expected is None:
+ assert body is None
+ else:
+ assert body.startswith("<!-- spot-interruption-notice run=7 -->\n")
+ assert all(text in body for text in expected)
+ assert f"{RUN_URL}/attempts/{attempt}" in body
+
+
+def _run(
+ status="in_progress",
+ attempt=1,
+ conclusion=None,
+ branch="my-change",
+ created="2026-10-25T10:00:00Z",
+ updated="2026-10-25T15:30:00Z",
+):
+ return {
+ "updated_at": updated,
+ "head_sha": "abc123",
+ "pr_numbers": [],
+ "id": 7,
+ "run_attempt": attempt,
+ "status": status,
+ "conclusion": conclusion,
+ "html_url": RUN_URL,
+ "head_branch": branch,
+ "head_owner": "someone",
+ "created_at": created,
+ }
+
+
[email protected](
+ ("run", "failed", "existing_body", "expected_write"),
+ [
+ pytest.param(_run(), [("Tests A", True)], None, "POST",
id="interrupted-while-running"),
+ pytest.param(_run(), [("Tests A", True)], "same", None,
id="notice-unchanged"),
+ pytest.param(_run(), [("Tests B", False)], None, None,
id="running-without-interruptions"),
+ pytest.param(
+ _run("completed", 1, "failure"), [("Tests B", False)], None, None,
id="first-attempt-real-failure"
+ ),
+ pytest.param(_run("in_progress", 2), [], "old", None,
id="re-run-in-progress"),
+ pytest.param(_run("completed", 2, "success"), [], "old", "PATCH",
id="re-run-finished"),
+ pytest.param(_run("completed", 2, "success"), [], None, None,
id="re-run-without-notice"),
+ pytest.param(
+ _run("completed", 2, "failure"),
+ [("Tests B", False)],
+ "other-run",
+ None,
+ id="notice-of-another-run",
+ ),
+ ],
+)
+def test_update_notice(spot, run, failed, existing_body, expected_write):
+ jobs = [_job(spot, name, interrupted=interrupted) for name, interrupted in
failed]
+ expected_body = spot.build_comment(
+ jobs, run_id=7, attempt=run["run_attempt"],
conclusion=run["conclusion"] or "", run_url=RUN_URL
+ )
+ bodies = {
+ "same": expected_body,
+ "old": "<!-- spot-interruption-notice run=7 -->\nold",
+ "other-run": "<!-- spot-interruption-notice run=6 -->\nold",
+ }
+ existing = spot.Comment(99, bodies[existing_body]) if existing_body else
None
+
+ with (
+ mock.patch.object(spot, "get_failed_jobs", autospec=True,
return_value=jobs),
+ mock.patch.object(spot, "find_pull_requests", autospec=True,
return_value=[42]),
+ mock.patch.object(spot, "find_comment", autospec=True,
return_value=existing),
+ mock.patch.object(spot, "gh_api", autospec=True) as mock_gh_api,
+ ):
+ spot.update_notice("apache/airflow", run)
+
+ if expected_write is None:
+ mock_gh_api.assert_not_called()
+ else:
+ path = (
+ "repos/apache/airflow/issues/42/comments"
+ if expected_write == "POST"
+ else "repos/apache/airflow/issues/comments/99"
+ )
+ mock_gh_api.assert_called_once_with("-X", expected_write, path, "-f",
f"body={expected_body}")
+
+
+def
test_list_pull_request_runs_keeps_the_newest_recent_run_per_pull_request(spot):
+ in_progress = [
+ {**_run(created="2026-10-25T12:00:00Z"), "id": 1},
+ {**_run(branch="other"), "id": 2},
+ ]
+ completed = [
+ {**_run("completed", created="2026-10-25T11:00:00Z"), "id": 3},
+ # Created days ago, but RunsOn's re-run just finished: its outcome is
still to report.
+ {
+ **_run("completed", branch="old", created="2026-10-23T10:00:00Z",
updated="2026-10-25T15:00:00Z"),
+ "id": 4,
+ },
+ {**_run("completed", branch="done", updated="2026-10-25T13:00:00Z"),
"id": 5},
+ ]
+
+ def gh_api(*args):
+ if "status=in_progress" in args[1]:
+ runs = in_progress
+ else:
+ runs = completed if "created=2026-10-24T16" in args[1] else []
+ return "\n".join(json.dumps(run) for run in runs)
+
+ with mock.patch.object(spot, "gh_api", side_effect=gh_api) as mock_gh_api:
+ runs = spot.list_pull_request_runs("apache/airflow", datetime(2026,
10, 25, 16, tzinfo=UTC))
+
+ assert sorted((run["head_branch"], run["created_at"]) for run in runs) == [
+ ("my-change", "2026-10-25T12:00:00Z"),
+ ("old", "2026-10-23T10:00:00Z"),
+ ("other", "2026-10-25T10:00:00Z"),
+ ]
+ # A listing per day of the 7, so that none reaches GitHub's limit of 1000
runs.
+ created = [call.args[1].split("created=")[1] for call in
mock_gh_api.call_args_list[1:]]
+ assert created[0] == "2026-10-24T16%3A00%3A00Z..2026-10-25T16%3A00%3A00Z"
+ assert created[-1] == "2026-10-18T16%3A00%3A00Z..2026-10-19T16%3A00%3A00Z"
+ assert len(created) == 7
+
+
+def test_list_runs_warns_when_reaching_the_listing_limit(spot, capsys):
+ runs = "\n".join(json.dumps(_run(branch=str(n))) for n in range(1000))
+
+ with mock.patch.object(spot, "gh_api", autospec=True, return_value=runs):
+ assert len(spot.list_runs("apache/airflow", "status=completed")) ==
1000
+
+ assert "reached GitHub's limit" in capsys.readouterr().err
+
+
+def test_find_comment_ignores_comments_of_others(spot):
+ marker = spot.COMMENT_MARKER
+ comments = [
+ {"id": 1, "body": f"{marker}\nnotice", "author":
"github-actions[bot]"},
+ {"id": 2, "body": f"{marker}\nlook-alike", "author": "someone"},
+ {"id": 3, "body": "unrelated", "author": "github-actions[bot]"},
+ ]
+
+ with mock.patch.object(
+ spot, "gh_api", autospec=True, return_value="\n".join(json.dumps(c)
for c in comments)
+ ):
+ assert spot.find_comment("apache/airflow", 42) == spot.Comment(1,
f"{marker}\nnotice")
+
+
+def test_run_pr_comments_carries_on_past_a_failing_run(spot):
+ runs = [_run(branch="a"), _run(branch="b")]
+ error = subprocess.CalledProcessError(1, ["gh"], stderr="HTTP 502")
+
+ with (
+ mock.patch.object(spot, "list_pull_request_runs", autospec=True,
return_value=runs),
+ mock.patch.object(spot, "update_notice", autospec=True,
side_effect=[error, None]) as mock_update,
+ ):
+ ok = spot.run_pr_comments("apache/airflow", datetime(2026, 10, 25,
tzinfo=UTC))
+
+ assert ok is False
+ assert mock_update.call_count == 2
+
+
+def test_run_slack_writes_outputs(spot, monkeypatch, tmp_path):
+ output = tmp_path / "output"
+ monkeypatch.setenv("GITHUB_OUTPUT", str(output))
+ jobs = [_job(spot, "Tests A", interrupted=True), _job(spot, "CI passed",
interrupted=False)]
+
+ with mock.patch.object(spot, "get_failed_jobs", autospec=True,
return_value=jobs):
+ spot.run_slack("apache/airflow", "7", 1)
+
+ assert output.read_text() == (
+ "failed-jobs<<EOF\nCI passed\nTests A (spot instance reclaimed)\nEOF\n"
+ "failed-jobs-text<<EOF\nCI passed\\nTests A (spot instance
reclaimed)\nEOF\n"
+ "has-failures<<EOF\ntrue\nEOF\n"
+ "skip<<EOF\ntrue\nEOF\n"
+ )
+
+
+def test_build_comment_neutralises_job_names(spot):
+ job = spot.FailedJob(1, "`x` @someone\n\n# [link](https://evil)",
"https://job/1";, spot=True)
+
+ body = spot.build_comment([job], run_id=7, attempt=1,
conclusion="failure", run_url=RUN_URL)
+
+ assert "- [`x @someone # [link](https://evil)`](https://job/1)" in body
+
+
[email protected](
+ ("actor", "previous", "expected_attempt_in_notice"),
+ [
+ pytest.param("runsonairflow[bot]", [("Tests A", True)], 1,
id="interruption-before-runs-on-re-run"),
+ pytest.param("runsonairflow[bot]", [("Tests B", False)], None,
id="runs-on-re-run-of-real-failure"),
+ pytest.param("a-maintainer", [("Tests A", True)], None,
id="maintainer-re-run"),
+ ],
+)
+def test_update_notice_finds_the_interruption_runs_on_is_re_running(
+ spot, actor, previous, expected_attempt_in_notice
+):
+ run = {**_run("in_progress", 2), "triggering_actor": actor}
+ previous_jobs = [_job(spot, name, interrupted=interrupted) for name,
interrupted in previous]
+
+ with (
+ mock.patch.object(spot, "get_failed_jobs", autospec=True,
side_effect=[[], previous_jobs]),
+ mock.patch.object(spot, "find_pull_requests", autospec=True,
return_value=[42]),
+ mock.patch.object(spot, "find_comment", autospec=True,
return_value=None),
+ mock.patch.object(spot, "write_comment", autospec=True) as mock_write,
+ ):
+ spot.update_notice("apache/airflow", run)
+
+ if expected_attempt_in_notice is None:
+ mock_write.assert_not_called()
+ else:
+ body = mock_write.call_args.args[3]
+ assert f"[attempt {expected_attempt_in_notice}]" in body
+ assert "RunsOn re-runs them" in body
+
+
[email protected](
+ ("pr_numbers", "pulls", "expected"),
+ [
+ pytest.param([42], None, [42], id="attached-to-the-run"),
+ pytest.param(
+ [],
+ [{"number": 42, "sha": "abc123"}, {"number": 43, "sha": "def456"}],
+ [42],
+ id="fork-branch-with-two-pull-requests",
+ ),
+ pytest.param(
+ [],
+ [{"number": 42, "sha": "abc123"}, {"number": 43, "sha": "abc123"}],
+ [],
+ id="fork-branch-with-two-pull-requests-of-one-commit",
+ ),
+ ],
+)
+def test_find_pull_requests(spot, pr_numbers, pulls, expected):
+ run = {**_run(), "pr_numbers": pr_numbers}
+ output = "\n".join(json.dumps(pull) for pull in pulls or [])
+
+ with mock.patch.object(spot, "gh_api", autospec=True, return_value=output)
as mock_gh_api:
+ assert spot.find_pull_requests("apache/airflow", run) == expected
+
+ assert mock_gh_api.called is (pulls is not None)
+
+
+def
test_list_pull_request_runs_keeps_the_newest_run_of_each_attached_pull_request(spot):
+ runs = [
+ {**_run(created="2026-10-25T11:00:00Z"), "id": 1, "pr_numbers": [42]},
+ {**_run(created="2026-10-25T12:00:00Z"), "id": 2, "pr_numbers": [43]},
+ {**_run(created="2026-10-25T10:00:00Z"), "id": 3, "pr_numbers": [42]},
+ ]
+
+ with mock.patch.object(spot, "list_runs", autospec=True,
side_effect=[runs] + [[]] * 7):
+ found = spot.list_pull_request_runs("apache/airflow", datetime(2026,
10, 25, 16, tzinfo=UTC))
+
+ assert sorted(run["id"] for run in found) == [1, 2]
