This is an automated email from the ASF dual-hosted git repository. potiuk pushed a commit to branch cancel-merged-pr-runs in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 3ddaed96b08990e68ea75d6933e1c121b19c3d89 Author: Jarek Potiuk <[email protected]> AuthorDate: Sun Oct 25 19:50:19 2026 +0100 Cancel the workflow runs of pull requests that have already merged Merging a pull request does not stop its pull_request runs, so they keep using the project's GitHub-hosted minutes and self-hosted runners for a result nobody reads once the change is on its target branch. Until now maintainers cancelled such runs by hand, when they noticed them. Generated-by: Claude Opus 5 --- .github/fork-ci-gate.toml | 5 + .github/workflows/fork-ci-monitor.yml | 4 +- dev/breeze/doc/08_ci_tasks.rst | 3 +- .../doc/ci/08_fork_ci_for_external_contributors.md | 33 ++- .../src/airflow_breeze/utils/fork_ci_gate.py | 5 + .../src/airflow_breeze/utils/fork_ci_github.py | 8 + .../src/airflow_breeze/utils/fork_ci_reconcile.py | 8 + .../utils/merged_pull_request_runs.py | 211 ++++++++++++++++ dev/breeze/tests/test_fork_ci_gate.py | 8 + dev/breeze/tests/test_fork_ci_github.py | 17 ++ dev/breeze/tests/test_fork_ci_reconcile.py | 40 ++++ dev/breeze/tests/test_merged_pull_request_runs.py | 266 +++++++++++++++++++++ 12 files changed, 604 insertions(+), 4 deletions(-) diff --git a/.github/fork-ci-gate.toml b/.github/fork-ci-gate.toml index 6fc7a0de1ce..6119f610ef2 100644 --- a/.github/fork-ci-gate.toml +++ b/.github/fork-ci-gate.toml @@ -34,6 +34,11 @@ close_after_days = 7 # status is posted either way. job_statuses = true +# Cancel the unfinished `pull_request` runs of pull requests that have already merged, of any +# author, so they stop using GitHub-hosted minutes and self-hosted runners for a result nobody +# reads. The fork CI monitor does it on its schedule. +cancel_merged_runs = true + # Organisations exempted under AIP-120 ยง2.2. Keys are display names used in comments, # values are GitHub logins. Changes go through a reviewed pull request. [allowlist] diff --git a/.github/workflows/fork-ci-monitor.yml b/.github/workflows/fork-ci-monitor.yml index ec34b9615b0..278787be5af 100644 --- a/.github/workflows/fork-ci-monitor.yml +++ b/.github/workflows/fork-ci-monitor.yml @@ -41,8 +41,8 @@ jobs: name: "Reconcile pull requests gated to fork CI" if: github.repository == 'apache/airflow' permissions: - # Listing the runs and jobs of the gate tampering check, and deleting the runs that only - # gated a pull request to fork CI. + # Listing the runs and jobs of the gate tampering check, deleting the runs that only gated a + # pull request to fork CI, and cancelling the unfinished runs of merged pull requests. actions: write contents: read # Labels, comments and closing of gated pull requests. diff --git a/dev/breeze/doc/08_ci_tasks.rst b/dev/breeze/doc/08_ci_tasks.rst index ca16b4483fe..b1bf0d9f5fb 100644 --- a/dev/breeze/doc/08_ci_tasks.rst +++ b/dev/breeze/doc/08_ci_tasks.rst @@ -188,7 +188,8 @@ Pull requests from external contributors can run their CI in the contributor's f gated pull requests ``fork ci required``, keeps them in draft until the fork run is green, maintains one state comment and a ``Fork CI / Tests (AMD)`` commit status linking to the run in the fork, closes pull requests without fork CI activity for a week, closes gated pull requests that made project CI run -by changing workflows, and writes everything it could not classify to a Slack message. With +by changing workflows, and writes everything it could not classify to a Slack message. It also cancels +the unfinished workflow runs of pull requests, of any author, that have already merged. With ``--dry-run`` it prints the changes instead of applying them. See `Fork CI for external contributors <ci/08_fork_ci_for_external_contributors.md>`_ for the design. 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 799a233e9ca..1f398373944 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 @@ -787,6 +787,36 @@ name within its time limit, and its run is the evidence. A re-run is kept too: d deletes every attempt, and an earlier attempt may have run project CI. Nothing is deleted for a pull request in the `gate_bypassed` state. A dry run only reports the deletions. +### Cancelling runs of merged pull requests + +Merging a pull request does not stop its `pull_request` runs. They keep using `apache/airflow`'s +capacity, GitHub-hosted minutes from the ASF's allocation and the self-hosted runners, for a +result nobody reads once the change is on its target branch. This is not specific to fork CI; the +monitor does it because it already runs every 15 minutes with `actions: write`. + +Each monitor run lists the `pull_request` runs of `apache/airflow` that are `queued`, +`in_progress`, `waiting`, `pending` or `requested`, and looks up the pull requests of their head +branches in one GraphQL query per 25 branches, paging through every pull request of a branch: a +common branch name such as `main` has pull requests from many forks, and an open one with the +run's head must not be missed. A run is cancelled when: + +- a merged pull request has the run's head repository, branch and commit (runs of pull requests + from forks do not list their pull request, so they are matched this way); +- no open pull request has the same head, since a branch can be reused for a new pull request; +- the run was created before the merge. A workflow triggered by the merge itself, such as one + on the `closed` event, is meant to run. + +It applies to the pull requests of every author, not only gated ones. A run is cancelled with +`POST /repos/apache/airflow/actions/runs/<id>/cancel`; one still `in_progress` 30 minutes after the +merge, for example because a step ignores the cancellation, is force-cancelled +(`.../force-cancel`). A run that finished in the meantime (HTTP 409) is not a problem; any other +failure is reported to Slack with the pull request, and a failure of the whole pass is reported +as a note, without affecting the reconciliation of gated pull requests. A dry run only reports +the cancellations. `cancel_merged_runs = false` in `.github/fork-ci-gate.toml` turns it off. + +Only runs of the merged head commit are matched. A run of an earlier commit of the pull request is +normally cancelled by the workflow's concurrency group when the next commit is pushed. + ### Far behind the base branch A fork run tests the pushed commit as it is, while project CI tests it merged with the base @@ -841,7 +871,8 @@ Per run, with ~700 gated pull requests: ~28 GraphQL calls for pass 2, ~14 for fo calls for the recent `pull_request` runs, one REST call per base branch for its workflow files, a jobs call and a compare call only per long `ci-amd.yml` run of a gated pull request, one changed-files call per head commit that turns green, plus one mutation per actual state -change. Job statuses add a jobs call and a status call per pull request with a fork run in +change. Cancelling runs of merged pull requests adds five paginated run listings, one GraphQL +query per 25 head branches with unfinished runs, and one mutation per cancelled run. Job statuses add a jobs call and a status call per pull request with a fork run in progress or a completed run not yet reported, and up to two status mutations per job, at most `MAX_JOB_STATUSES_PER_RUN` per run; see "Write budget". Well inside the `GITHUB_TOKEN` limits at a 15-minute cadence. The reconciler logs its call counts so the diff --git a/dev/breeze/src/airflow_breeze/utils/fork_ci_gate.py b/dev/breeze/src/airflow_breeze/utils/fork_ci_gate.py index 8ce90a34d97..dc05a3b67a5 100644 --- a/dev/breeze/src/airflow_breeze/utils/fork_ci_gate.py +++ b/dev/breeze/src/airflow_breeze/utils/fork_ci_gate.py @@ -129,6 +129,7 @@ class GateConfig: allowlist: Mapping[str, tuple[str, ...]] = field(default_factory=dict) pilot: Mapping[str, datetime] = field(default_factory=dict) job_statuses: bool = True + cancel_merged_runs: bool = True def is_allowlisted(self, login: str) -> bool: wanted = login.casefold() @@ -177,6 +178,9 @@ def parse_config(text: str) -> GateConfig: job_statuses = data.get("job_statuses", True) if not isinstance(job_statuses, bool): raise ValueError(f"job_statuses must be true or false, got {job_statuses!r}") + cancel_merged_runs = data.get("cancel_merged_runs", True) + if not isinstance(cancel_merged_runs, bool): + raise ValueError(f"cancel_merged_runs must be true or false, got {cancel_merged_runs!r}") return GateConfig( enabled=enabled, enabled_since=enabled_since, @@ -184,6 +188,7 @@ def parse_config(text: str) -> GateConfig: allowlist=allowlist, pilot=pilot, job_statuses=job_statuses, + cancel_merged_runs=cancel_merged_runs, ) diff --git a/dev/breeze/src/airflow_breeze/utils/fork_ci_github.py b/dev/breeze/src/airflow_breeze/utils/fork_ci_github.py index 73a44503acb..fe3fdfb651e 100644 --- a/dev/breeze/src/airflow_breeze/utils/fork_ci_github.py +++ b/dev/breeze/src/airflow_breeze/utils/fork_ci_github.py @@ -172,6 +172,14 @@ class GitHubClient: None, ) + def cancel_run(self, pr: int, run_id: int, *, force: bool = False) -> None: + action = "force-cancel" if force else "cancel" + self._mutate( + f"#{pr}: {action} workflow run {run_id}", + ["api", "-X", "POST", f"repos/{self.repo}/actions/runs/{run_id}/{action}"], + None, + ) + def set_status( self, pr: int, 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 6b54f6c7fac..00698b78494 100644 --- a/dev/breeze/src/airflow_breeze/utils/fork_ci_reconcile.py +++ b/dev/breeze/src/airflow_breeze/utils/fork_ci_reconcile.py @@ -74,6 +74,7 @@ from airflow_breeze.utils.fork_ci_messages import ( tamper_close_comment, ungated_comment, ) +from airflow_breeze.utils.merged_pull_request_runs import cancel_merged_pull_request_runs CI_WORKFLOW_FILE = "ci-amd.yml" # The files the gate job of ci-amd.yml reads from a pull request's base branch. @@ -1547,6 +1548,13 @@ def run_reconcile( unknown.append((listed.number, f"{type(e).__name__}: {e}")) notes = [bypass_finder.incomplete] if bypass_finder.incomplete else [] + if config.cancel_merged_runs: + # Not tied to gated pull requests; a failure here must not hide the reconciliation's own. + try: + unknown.extend(cancel_merged_pull_request_runs(client, repo=repo, now=now)) + except Exception as e: + console_print(f"[error]Cancelling runs of merged pull requests failed: {e}[/]") + notes.append(f"cancelling runs of merged pull requests failed: {type(e).__name__}: {e}") if job_reporter.deferred: console_print( f"[info]Job statuses of {len(job_reporter.deferred)} pull requests left for the next run: " diff --git a/dev/breeze/src/airflow_breeze/utils/merged_pull_request_runs.py b/dev/breeze/src/airflow_breeze/utils/merged_pull_request_runs.py new file mode 100644 index 00000000000..7eed315630b --- /dev/null +++ b/dev/breeze/src/airflow_breeze/utils/merged_pull_request_runs.py @@ -0,0 +1,211 @@ +# 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. +""" +Cancels the workflow runs that are still going for pull requests that have already merged. + +Merging a pull request does not stop its `pull_request` runs, which keep using the project's +GitHub-hosted minutes and self-hosted runners for a result nobody reads. The fork CI monitor calls +this on its schedule. See "Cancelling runs of merged pull requests" in +dev/breeze/doc/ci/08_fork_ci_for_external_contributors.md. +""" + +from __future__ import annotations + +from collections.abc import Iterable +from dataclasses import dataclass +from datetime import datetime, timedelta +from typing import TYPE_CHECKING, Any + +from airflow_breeze.utils.console import console_print +from airflow_breeze.utils.fork_ci_gate import parse_github_datetime + +if TYPE_CHECKING: + from airflow_breeze.utils.fork_ci_github import GitHubClient + +# The statuses of a run that has not finished. A run waiting for approval is "completed". +UNFINISHED_RUN_STATUSES = ("queued", "in_progress", "waiting", "pending", "requested") +# A cancelled run whose jobs keep running, for example in a step that ignores the signal, is +# force-cancelled once this long has passed since the merge, which leaves a normal cancel at least +# one monitor run to work. +FORCE_CANCEL_AFTER = timedelta(minutes=30) +BRANCH_BATCH_SIZE = 25 + + +@dataclass(frozen=True) +class UnfinishedRun: + id: int + workflow: str + status: str + head_repo: str + head_branch: str + head_sha: str + created_at: datetime + + +@dataclass(frozen=True) +class PullRequestHead: + number: int + state: str + head_repo: str + head_sha: str + merged_at: datetime | None + + +def list_unfinished_runs(client: GitHubClient, repo: str) -> list[UnfinishedRun]: + runs = [] + for status in UNFINISHED_RUN_STATUSES: + for page in client.rest_get_pages( + f"repos/{repo}/actions/runs?event=pull_request&status={status}&per_page=100" + ): + for run in page["workflow_runs"]: + runs.append( + UnfinishedRun( + id=run["id"], + workflow=run.get("name") or "", + status=run["status"], + head_repo=(run.get("head_repository") or {}).get("full_name") or "", + head_branch=run.get("head_branch") or "", + head_sha=run["head_sha"], + created_at=parse_github_datetime(run["created_at"]), # type: ignore[arg-type] + ) + ) + return runs + + +PULL_REQUEST_FIELDS = ( + "pageInfo { hasNextPage endCursor } " + "nodes { number state mergedAt headRefOid headRepository { nameWithOwner } }" +) +BRANCH_PAGE_QUERY = ( + "query($owner: String!, $name: String!, $branch: String!, $cursor: String) " + "{ repository(owner: $owner, name: $name) { pullRequests(headRefName: $branch, first: 100, " + f"after: $cursor) {{ {PULL_REQUEST_FIELDS} }} }} }}" +) + + +def _parse_pull_requests(connection: dict[str, Any] | None) -> list[PullRequestHead]: + return [ + PullRequestHead( + number=node["number"], + state=node["state"], + head_repo=(node.get("headRepository") or {}).get("nameWithOwner") or "", + head_sha=node["headRefOid"], + merged_at=parse_github_datetime(node.get("mergedAt") or ""), + ) + for node in (connection or {}).get("nodes") or [] + ] + + +def find_pull_requests_by_branch( + client: GitHubClient, repo: str, branches: Iterable[str] +) -> dict[str, list[PullRequestHead]]: + """ + Every pull request of each head branch name, in any state and from any head repository. + + All of them are read, not only the latest: a common branch name such as ``main`` has pull + requests from many forks, and an open one with a run's head must not be missed. + """ + owner, name = repo.split("/", 1) + found: dict[str, list[PullRequestHead]] = {} + unique = sorted(set(branches)) + for start in range(0, len(unique), BRANCH_BATCH_SIZE): + batch = unique[start : start + BRANCH_BATCH_SIZE] + declarations = ", ".join(f"$b{index}: String!" for index in range(len(batch))) + fields = " ".join( + f"b{index}: pullRequests(headRefName: $b{index}, first: 100) {{ {PULL_REQUEST_FIELDS} }}" + for index in range(len(batch)) + ) + data = client.graphql( + f"query($owner: String!, $name: String!, {declarations}) " + f"{{ repository(owner: $owner, name: $name) {{ {fields} }} }}", + {"owner": owner, "name": name, **{f"b{index}": branch for index, branch in enumerate(batch)}}, + )["repository"] + for index, branch in enumerate(batch): + connection = data[f"b{index}"] or {} + found[branch] = _parse_pull_requests(connection) + while (connection.get("pageInfo") or {}).get("hasNextPage"): + connection = client.graphql( + BRANCH_PAGE_QUERY, + { + "owner": owner, + "name": name, + "branch": branch, + "cursor": connection["pageInfo"]["endCursor"], + }, + )["repository"]["pullRequests"] + found[branch].extend(_parse_pull_requests(connection)) + return found + + +def find_merged_pull_request( + run: UnfinishedRun, pull_requests: Iterable[PullRequestHead] +) -> PullRequestHead | None: + """ + The merged pull request a run was started for, if no open pull request has the same head. + + A run is matched by head repository, branch and commit, because runs of pull requests from + forks do not list their pull request. Only runs started before the merge count: a workflow + triggered by the merge itself, such as one on `closed`, is meant to run. + """ + same_head = [ + pr + for pr in pull_requests + if pr.head_repo.casefold() == run.head_repo.casefold() and pr.head_sha == run.head_sha + ] + if any(pr.state == "OPEN" for pr in same_head): + return None + return next( + ( + pr + for pr in same_head + if pr.state == "MERGED" and pr.merged_at is not None and run.created_at < pr.merged_at + ), + None, + ) + + +def cancel_merged_pull_request_runs( + client: GitHubClient, *, repo: str, now: datetime +) -> list[tuple[int, str]]: + """Cancel the unfinished runs of merged pull requests; return ``(pull request, problem)`` for Slack.""" + runs = list_unfinished_runs(client, repo) + if not runs: + return [] + pull_requests = find_pull_requests_by_branch(client, repo, (run.head_branch for run in runs)) + problems: list[tuple[int, str]] = [] + cancelled = 0 + for run in runs: + merged = find_merged_pull_request(run, pull_requests.get(run.head_branch, [])) + if merged is None: + continue + force = ( + run.status == "in_progress" + and merged.merged_at is not None + and now - merged.merged_at >= FORCE_CANCEL_AFTER + ) + try: + client.cancel_run(merged.number, run.id, force=force) + cancelled += 1 + except Exception as e: + # A run that finished meanwhile cannot be cancelled; that is the outcome wanted anyway. + # gh reports the status as "(HTTP 409)"; a bare "409" could be part of the run id. + if "(HTTP 409)" in str(e): + continue + problems.append((merged.number, f"cancelling run {run.id} ({run.workflow}) failed: {e}")) + verb = "Would cancel" if client.dry_run else "Cancelled" + console_print(f"[info]{verb} {cancelled} runs of merged pull requests[/]") + return problems diff --git a/dev/breeze/tests/test_fork_ci_gate.py b/dev/breeze/tests/test_fork_ci_gate.py index b52a0d29a1c..e641e45ca82 100644 --- a/dev/breeze/tests/test_fork_ci_gate.py +++ b/dev/breeze/tests/test_fork_ci_gate.py @@ -321,12 +321,17 @@ def test_parse_config_defaults(): assert config.allowlist == {} assert config.pilot == {} assert config.job_statuses is True + assert config.cancel_merged_runs is True def test_parse_config_turns_job_statuses_off(): assert parse_config("job_statuses = false\n").job_statuses is False +def test_parse_config_turns_cancelling_merged_runs_off(): + assert parse_config("cancel_merged_runs = false\n").cancel_merged_runs is False + + def test_parse_config_lowercases_pilot_logins(): config = parse_config('[pilot]\n"Some-User" = 2026-10-20T00:00:00Z\n') @@ -340,6 +345,9 @@ def test_parse_config_lowercases_pilot_logins(): [ pytest.param("enabled = 1\n", "enabled must be", id="enabled-not-bool"), pytest.param('job_statuses = "no"\n', "job_statuses must be", id="job-statuses-not-bool"), + pytest.param( + "cancel_merged_runs = 0\n", "cancel_merged_runs must be", id="cancel-merged-runs-not-bool" + ), pytest.param( "enabled_since = 2026-01-01T00:00:00\n", "enabled_since must be", id="enabled-since-naive" ), diff --git a/dev/breeze/tests/test_fork_ci_github.py b/dev/breeze/tests/test_fork_ci_github.py index d993add8bb5..09257d0697c 100644 --- a/dev/breeze/tests/test_fork_ci_github.py +++ b/dev/breeze/tests/test_fork_ci_github.py @@ -128,6 +128,7 @@ def test_dry_run_does_not_mutate(mock_run): client.close(1) client.remove_label(1, "fork ci required") client.delete_run(1, 7) + client.cancel_run(1, 7) mock_run.assert_not_called() @@ -170,3 +171,19 @@ def test_set_status_context(mock_run, kwargs, context): GitHubClient().set_status(1, "abc", "success", "https://x";, "desc", **kwargs) assert json.loads(mock_run.call_args.kwargs["input"])["context"] == context + + [email protected](("force", "action"), [(False, "cancel"), (True, "force-cancel")]) [email protected]("airflow_breeze.utils.fork_ci_github.subprocess.run", autospec=True) +def test_cancel_run(mock_run, force, action): + mock_run.return_value = _completed() + + GitHubClient().cancel_run(1, 7, force=force) + + assert mock_run.call_args.args[0] == [ + "gh", + "api", + "-X", + "POST", + f"repos/apache/airflow/actions/runs/7/{action}", + ] diff --git a/dev/breeze/tests/test_fork_ci_reconcile.py b/dev/breeze/tests/test_fork_ci_reconcile.py index 9f3aa61f751..3df786d4a83 100644 --- a/dev/breeze/tests/test_fork_ci_reconcile.py +++ b/dev/breeze/tests/test_fork_ci_reconcile.py @@ -2073,3 +2073,43 @@ def test_job_status_reporter_compares_contexts_case_insensitively(client): assert JobStatusReporter(client, repo=REPO).report(_pr(), COMPLETED_RUN) == "1:1:FAILURE" client.set_status.assert_not_called() + + [email protected]("enabled", [True, False]) [email protected]("airflow_breeze.utils.fork_ci_reconcile.cancel_merged_pull_request_runs", autospec=True) +def test_run_reconcile_cancels_runs_of_merged_pull_requests(mock_cancel, client, tmp_path, enabled): + mock_cancel.return_value = [(10, "cancelling run 7 failed")] + client.graphql.side_effect = [_search_result(), _search_result(), _search_result()] + client.rest_get_pages.return_value = [{"total_count": 0, "workflow_runs": []}] + + has_alerts = run_reconcile( + client, + dataclasses.replace(CONFIG, cancel_merged_runs=enabled), + now=NOW, + run_url="https://run";, + output_file=tmp_path / "s.json", + workflows_dir=tmp_path, + ) + + assert mock_cancel.called is enabled + assert has_alerts is enabled + + [email protected]("airflow_breeze.utils.fork_ci_reconcile.cancel_merged_pull_request_runs", autospec=True) +def test_run_reconcile_reports_a_failed_cancellation_pass(mock_cancel, client, tmp_path): + mock_cancel.side_effect = GitHubError("HTTP 502") + client.graphql.side_effect = [_search_result(), _search_result(), _search_result()] + client.rest_get_pages.return_value = [{"total_count": 0, "workflow_runs": []}] + + assert ( + run_reconcile( + client, + CONFIG, + now=NOW, + run_url="https://run";, + output_file=tmp_path / "s.json", + workflows_dir=tmp_path, + ) + is True + ) + assert "cancelling runs of merged pull requests failed" in (tmp_path / "s.json").read_text() diff --git a/dev/breeze/tests/test_merged_pull_request_runs.py b/dev/breeze/tests/test_merged_pull_request_runs.py new file mode 100644 index 00000000000..5c6223963c0 --- /dev/null +++ b/dev/breeze/tests/test_merged_pull_request_runs.py @@ -0,0 +1,266 @@ +# 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 + +from datetime import UTC, datetime, timedelta +from unittest import mock + +import pytest + +from airflow_breeze.utils.fork_ci_github import GitHubClient, GitHubError +from airflow_breeze.utils.merged_pull_request_runs import ( + BRANCH_BATCH_SIZE, + PullRequestHead, + UnfinishedRun, + cancel_merged_pull_request_runs, + find_merged_pull_request, + find_pull_requests_by_branch, +) + +REPO = "apache/airflow" +NOW = datetime(2026, 10, 25, 19, 0, tzinfo=UTC) +MERGED_AT = NOW - timedelta(minutes=10) +RUN = UnfinishedRun( + id=7, + workflow="Tests (AMD)", + status="in_progress", + head_repo="someone/airflow", + head_branch="fix", + head_sha="aaa", + created_at=MERGED_AT - timedelta(minutes=20), +) +MERGED = PullRequestHead( + number=10, state="MERGED", head_repo="someone/airflow", head_sha="aaa", merged_at=MERGED_AT +) + + [email protected] +def client(): + github = mock.create_autospec(GitHubClient, instance=True) + github.repo = REPO + github.dry_run = False + return github + + [email protected]( + ("pull_requests", "expected"), + [ + pytest.param([MERGED], MERGED, id="merged-pull-request"), + pytest.param( + [PullRequestHead(10, "MERGED", "Someone/Airflow", "aaa", MERGED_AT)], + PullRequestHead(10, "MERGED", "Someone/Airflow", "aaa", MERGED_AT), + id="head-repository-in-another-case", + ), + pytest.param( + [PullRequestHead(11, "OPEN", "someone/airflow", "aaa", None), MERGED], + None, + id="open-pr-same-head", + ), + pytest.param( + [PullRequestHead(10, "MERGED", "other/airflow", "aaa", MERGED_AT)], None, id="other-fork" + ), + pytest.param( + [PullRequestHead(10, "MERGED", "someone/airflow", "bbb", MERGED_AT)], None, id="other-commit" + ), + pytest.param( + [PullRequestHead(10, "MERGED", "someone/airflow", "aaa", RUN.created_at - timedelta(minutes=1))], + None, + id="run-started-after-the-merge", + ), + pytest.param( + [PullRequestHead(10, "CLOSED", "someone/airflow", "aaa", None)], None, id="closed-not-merged" + ), + pytest.param([], None, id="no-pull-request"), + ], +) +def test_find_merged_pull_request(pull_requests, expected): + assert find_merged_pull_request(RUN, pull_requests) == expected + + +def _run(run_id, *, status="in_progress", branch="fix", sha="aaa", created_at=RUN.created_at): + return { + "id": run_id, + "name": "Tests (AMD)", + "status": status, + "head_repository": {"full_name": "someone/airflow"}, + "head_branch": branch, + "head_sha": sha, + "created_at": created_at.strftime("%Y-%m-%dT%H:%M:%SZ"), + } + + +def _pr_node(number, state, sha="aaa", merged_at=MERGED_AT): + return { + "number": number, + "state": state, + "mergedAt": merged_at.strftime("%Y-%m-%dT%H:%M:%SZ") if merged_at else None, + "headRefOid": sha, + "headRepository": {"nameWithOwner": "someone/airflow"}, + } + + +def _runs_by_status(runs): + def pages(path): + status = path.split("status=", 1)[1].split("&", 1)[0] + return [{"workflow_runs": [run for run in runs if run["status"] == status]}] + + return pages + + [email protected]( + ("merged_minutes_ago", "status", "force"), + [ + pytest.param(10, "in_progress", False, id="recently-merged"), + pytest.param(31, "in_progress", True, id="still-running-long-after-the-merge"), + pytest.param(31, "queued", False, id="queued-long-after-the-merge"), + ], +) +def test_cancel_merged_pull_request_runs(client, merged_minutes_ago, status, force): + merged_at = NOW - timedelta(minutes=merged_minutes_ago) + client.rest_get_pages.side_effect = _runs_by_status( + [_run(7, status=status, created_at=merged_at - timedelta(minutes=5))] + ) + client.graphql.return_value = { + "repository": {"b0": {"nodes": [_pr_node(10, "MERGED", merged_at=merged_at)]}} + } + + assert cancel_merged_pull_request_runs(client, repo=REPO, now=NOW) == [] + + client.cancel_run.assert_called_once_with(10, 7, force=force) + + +def test_cancel_merged_pull_request_runs_leaves_open_pull_requests_alone(client): + client.rest_get_pages.side_effect = _runs_by_status([_run(7)]) + client.graphql.return_value = {"repository": {"b0": {"nodes": [_pr_node(10, "OPEN", merged_at=None)]}}} + + cancel_merged_pull_request_runs(client, repo=REPO, now=NOW) + + client.cancel_run.assert_not_called() + + +def test_cancel_merged_pull_request_runs_without_unfinished_runs(client): + client.rest_get_pages.side_effect = _runs_by_status([]) + + assert cancel_merged_pull_request_runs(client, repo=REPO, now=NOW) == [] + client.graphql.assert_not_called() + + [email protected]( + ("error", "problems"), + [ + pytest.param( + GitHubError( + "gh api -X POST repos/apache/airflow/actions/runs/7/cancel failed: gh: done (HTTP 409)" + ), + [], + id="finished-meanwhile", + ), + pytest.param( + GitHubError("gh api repos/apache/airflow/actions/runs/1409/cancel failed: HTTP 500"), + [ + ( + 10, + "cancelling run 7 (Tests (AMD)) failed: gh api repos/apache/airflow/actions/runs/1409/cancel failed: HTTP 500", + ) + ], + id="failed-with-409-in-the-path", + ), + ], +) +def test_cancel_merged_pull_request_runs_reports_failed_cancellations(client, error, problems): + client.rest_get_pages.side_effect = _runs_by_status([_run(7), _run(8)]) + client.graphql.return_value = {"repository": {"b0": {"nodes": [_pr_node(10, "MERGED")]}}} + client.cancel_run.side_effect = [error, None] + + assert cancel_merged_pull_request_runs(client, repo=REPO, now=NOW) == problems + assert client.cancel_run.call_count == 2 + + +def test_find_pull_requests_by_branch_batches_branches(client): + branches = [f"branch-{index:02d}" for index in range(BRANCH_BATCH_SIZE + 1)] + + def graphql(query, variables): + batch = sorted(key for key in variables if key.startswith("b")) + return {"repository": {key: {"nodes": [_pr_node(1, "MERGED")]} for key in batch}} + + client.graphql.side_effect = graphql + + found = find_pull_requests_by_branch(client, REPO, [*branches, branches[0]]) + + assert sorted(found) == branches + assert client.graphql.call_count == 2 + assert client.graphql.call_args_list[0].args[1]["owner"] == "apache" + + +def test_find_pull_requests_by_branch_reads_every_page(client): + first = {"pageInfo": {"hasNextPage": True, "endCursor": "c1"}, "nodes": [_pr_node(1, "MERGED")]} + second = { + "pageInfo": {"hasNextPage": False, "endCursor": None}, + "nodes": [_pr_node(2, "OPEN", merged_at=None)], + } + client.graphql.side_effect = [ + {"repository": {"b0": first}}, + {"repository": {"pullRequests": second}}, + ] + + found = find_pull_requests_by_branch(client, REPO, ["main"]) + + assert [pr.number for pr in found["main"]] == [1, 2] + assert client.graphql.call_args.args[1] == { + "owner": "apache", + "name": "airflow", + "branch": "main", + "cursor": "c1", + } + + +def test_open_pull_request_on_a_later_page_keeps_the_run(client): + client.rest_get_pages.side_effect = _runs_by_status([_run(7)]) + client.graphql.side_effect = [ + { + "repository": { + "b0": { + "pageInfo": {"hasNextPage": True, "endCursor": "c1"}, + "nodes": [_pr_node(10, "MERGED")], + } + } + }, + { + "repository": { + "pullRequests": { + "pageInfo": {"hasNextPage": False}, + "nodes": [_pr_node(11, "OPEN", merged_at=None)], + } + } + }, + ] + + cancel_merged_pull_request_runs(client, repo=REPO, now=NOW) + + client.cancel_run.assert_not_called() + + [email protected](("dry_run", "summary"), [(False, "Cancelled 1 runs"), (True, "Would cancel 1 runs")]) [email protected]("airflow_breeze.utils.merged_pull_request_runs.console_print", autospec=True) +def test_cancel_merged_pull_request_runs_summary_tells_a_dry_run(mock_print, client, dry_run, summary): + client.dry_run = dry_run + client.rest_get_pages.side_effect = _runs_by_status([_run(7)]) + client.graphql.return_value = {"repository": {"b0": {"nodes": [_pr_node(10, "MERGED")]}}} + + cancel_merged_pull_request_runs(client, repo=REPO, now=NOW) + + assert summary in mock_print.call_args.args[0]
