This is an automated email from the ASF dual-hosted git repository. potiuk pushed a commit to branch auto-approve-workflow-runs in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 7dd7044c7da3d340991c401b484f998fddc44d09 Author: Jarek Potiuk <[email protected]> AuthorDate: Fri Oct 23 12:04:04 2026 +0200 Auto-approve cheap CI runs of external contributors' PRs GitHub holds the workflow runs of fork PRs until a maintainer approves them. With self-hosted runners (AIP-118) the approval gate becomes part of the trust model, so it should cover all external contributors, but approving every push by hand does not scale and makes contributors wait. Runs whose change cannot touch the CI definition and whose CI is cheap are safe to start without a maintainer. Expensive or large changes, and anything that changes how CI or runners are chosen, still get a human look first, which also keeps the load on the shared public runners down. The CI-minutes estimate lets maintainers see what a run will cost before approving it. Generated-by: Claude Opus 5 --- .github/AGENTS.md | 38 ++ .github/workflows/auto-approve-workflow-runs.yml | 65 +++ contributing-docs/05_pull_requests.rst | 34 ++ contributing-docs/25_maintainer_pr_triage.md | 1 + dev/breeze/doc/08_ci_tasks.rst | 17 + dev/breeze/doc/ci/05_workflows.md | 43 ++ .../doc/images/output_ci_approve-workflow-runs.svg | 142 +++++ .../doc/images/output_ci_approve-workflow-runs.txt | 1 + .../src/airflow_breeze/commands/ci_commands.py | 50 ++ .../airflow_breeze/commands/ci_commands_config.py | 17 + .../airflow_breeze/utils/ci_minutes_estimate.py | 198 +++++++ .../airflow_breeze/utils/workflow_run_approval.py | 590 ++++++++++++++++++++ dev/breeze/tests/test_ci_minutes_estimate.py | 168 ++++++ dev/breeze/tests/test_workflow_run_approval.py | 617 +++++++++++++++++++++ dev/calibrate_ci_minutes_estimates.py | 251 +++++++++ 15 files changed, 2232 insertions(+) diff --git a/.github/AGENTS.md b/.github/AGENTS.md new file mode 100644 index 00000000000..f59e68fe317 --- /dev/null +++ b/.github/AGENTS.md @@ -0,0 +1,38 @@ +<!-- SPDX-License-Identifier: Apache-2.0 + https://www.apache.org/licenses/LICENSE-2.0 --> + +<!-- START doctoc generated TOC please keep comment here to allow auto update --> +<!-- DON'T EDIT THIS SECTION, INSTEAD RE-RUN doctoc TO UPDATE --> +**Table of Contents** *generated with [DocToc](https://github.com/thlorenz/doctoc)* + +- [The `.github` folder — Agent Instructions](#the-github-folder--agent-instructions) + - [Keep the CI-minutes estimates in sync with the workflows](#keep-the-ci-minutes-estimates-in-sync-with-the-workflows) + +<!-- END doctoc generated TOC please keep comment here to allow auto update --> + +# The `.github` folder — Agent Instructions + +## Keep the CI-minutes estimates in sync with the workflows + +`breeze ci approve-workflow-runs` (run by `workflows/auto-approve-workflow-runs.yml`) decides whether +to approve an external contributor's CI runs from an estimate of the job-minutes they take. The +numbers live in `dev/breeze/src/airflow_breeze/utils/ci_minutes_estimate.py` and are medians +measured on recent `Tests (AMD)` PR runs on public runners. + +After you change the CI workflows in a way that adds, removes, splits or resizes jobs (new test +groups, a changed matrix, new composite workflows, a move between runner types), refresh the +estimates once the change has run on enough PRs, and in any case about once a month: + +1. Run `uv run --project dev/breeze python dev/calibrate_ci_minutes_estimates.py --since <date>` + from the repository root with an authenticated `gh` CLI. Pick a date that covers at least a + week of runs after the workflow change. +2. Copy the medians and fits it prints into the constants of `ci_minutes_estimate.py`, and update + the "Last calibrated" line next to them. +3. When you add a new job group to the workflows, add a matching group to both + `ci_minutes_estimate.py` (with the selective-check output that turns it on) and the + `JOB_GROUPS` patterns of the calibration script. Unmatched jobs fall into the base group. +4. Run `uv run --project dev/breeze pytest dev/breeze/tests/test_ci_minutes_estimate.py + dev/breeze/tests/test_workflow_run_approval.py -xvs`. + +The estimates cover public runners only. Self-hosted (RunsOn) runners need their own set once there +is enough data from them. diff --git a/.github/workflows/auto-approve-workflow-runs.yml b/.github/workflows/auto-approve-workflow-runs.yml new file mode 100644 index 00000000000..0a692ec3806 --- /dev/null +++ b/.github/workflows/auto-approve-workflow-runs.yml @@ -0,0 +1,65 @@ +# 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. +# +--- +# Approves the waiting workflow runs of external contributors' PRs to main when the change is safe to +# run without a maintainer's look, and labels the others "need workflow approval". The rules are in +# dev/breeze/src/airflow_breeze/utils/workflow_run_approval.py and documented in +# contributing-docs/05_pull_requests.rst. +# +# This workflow runs only trusted code from main. It reads the PR's metadata, changed-file list and +# git objects, and never checks out or runs the PR's code. Approving with GITHUB_TOKEN starts the +# approved runs right away, and workflow_run workflows that follow them still trigger. +name: Auto approve workflow runs +on: # yamllint disable-line rule:truthy + schedule: + - cron: '*/30 * * * *' + workflow_dispatch: + inputs: + report-only: + description: "Only report the decisions and estimates, without approving or labelling" + type: boolean + default: false +permissions: + # Approving workflow runs. + actions: write + # Adding and removing the "need workflow approval" label. + pull-requests: write + contents: read +concurrency: + group: auto-approve-workflow-runs + cancel-in-progress: false +jobs: + approve-workflow-runs: + name: "Approve workflow runs of external contributors" + runs-on: ubuntu-latest + if: github.repository == 'apache/airflow' + timeout-minutes: 20 + steps: + - name: "Checkout main" + uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + persist-credentials: false + ref: main + - name: "Install Breeze" + uses: ./.github/actions/breeze + - name: "Approve workflow runs" + env: + GITHUB_TOKEN: ${{ github.token }} + GITHUB_REPOSITORY: ${{ github.repository }} + REPORT_ONLY: ${{ inputs.report-only && 'true' || 'false' }} + run: breeze ci approve-workflow-runs diff --git a/contributing-docs/05_pull_requests.rst b/contributing-docs/05_pull_requests.rst index c7fee6d2677..b6c5946aee7 100644 --- a/contributing-docs/05_pull_requests.rst +++ b/contributing-docs/05_pull_requests.rst @@ -242,6 +242,40 @@ is the PMC private list — see the ``[email protected]``. That document also covers the further steps the project may take when the same behaviour persists across PRs. +CI workflow runs of external contributors +----------------------------------------- + +GitHub does not start the CI workflows of a PR from a fork until a maintainer approves them, +for the contributors that the repository's approval setting covers. To keep you from waiting +for a maintainer on every push, a scheduled workflow checks the waiting runs every 30 minutes +and approves them automatically when all of these hold: + +* The PR targets ``main``. PRs to release branches, such as ``v3-4-test``, always wait for a + maintainer. +* The PR does not change anything under ``.github/``, ``dev/breeze/`` or ``scripts/ci/``. Those + files define the CI workflows and choose the runners they use. +* The change does not need the full set of tests, does not change dependencies (``pyproject.toml``, + lock files, ``package.json``, ``go.mod``, Gradle build files) and does not add a new provider. +* The CI jobs your change selects are estimated to take at most 310 job-minutes, about a quarter of + a full CI run. Changes to documentation, to the UI, to the Helm chart, to the language SDKs, to + one area of Airflow core or to a small provider usually stay under it. Changes to large providers + such as Amazon or Google, to several providers, or to the Task SDK (which runs the tests of all + providers) usually do not. +* The PR changes at most 100 lines of source code. Tests (including end-to-end tests), + documentation, newsfragments, translations and lock files are not counted, so a fix with + thorough tests still qualifies. Larger changes are usually new features, and a maintainer looks + at them first. + +Each new push is checked again, because approval applies to one commit. Only runs started in the +last 24 hours are considered, so older waiting runs stay with maintainers. If your PR does not +qualify, it gets the ``need workflow approval`` label and a maintainer approves the runs after +looking at the change. That can take a while, so please do not ping maintainers about it; it is +not a judgement of your PR. The label is removed when a later commit is approved automatically. + +Maintainers can see each decision, the CI jobs selective checks expect to run and their estimated +time in the summary of the ``Auto approve workflow runs`` workflow. The rules are described in +more detail in `CI workflows <../dev/breeze/doc/ci/05_workflows.md>`__. + Gen-AI Assisted contributions ----------------------------- diff --git a/contributing-docs/25_maintainer_pr_triage.md b/contributing-docs/25_maintainer_pr_triage.md index e5f5313c5b7..ad6ea684d37 100644 --- a/contributing-docs/25_maintainer_pr_triage.md +++ b/contributing-docs/25_maintainer_pr_triage.md @@ -238,6 +238,7 @@ The triage workflow uses the following labels and states: | Label / State | Meaning | |---|---| | `ready for maintainer review` | PR has passed Stage 1 and is queued for human review. Applied only after the skill has verified that no workflow runs are awaiting approval. | +| `need workflow approval` | The PR's CI workflow runs wait for a maintainer, because the `Auto approve workflow runs` workflow did not approve them: a PR to a branch other than `main`, a change to CI definition files, a full-tests or dependency change, a change whose CI jobs are estimated at more than 310 job-minutes, or a change of more than 100 lines of source code (see [Pull Requests](05_pull_requests.rst)). Approve the runs after looking at the diff. The workflow removes the label w [...] | `closed because of multiple quality violations` | PR was closed because the author has multiple open PRs with quality issues. | | `suspicious changes detected` | PR (and all open PRs by the same author) was closed because the diff contained suspicious patterns (secret exfiltration, malicious CI modifications, etc.). | | `closed because of open PR limit` | PR was closed in the one-time closure that introduced the [open PR limit](32_open_pull_request_limit.rst). Applied by `dev/close_prs_over_open_pr_limit.py`, never by the triage skill. If the author reopens the PR, triage it like any other. | diff --git a/dev/breeze/doc/08_ci_tasks.rst b/dev/breeze/doc/08_ci_tasks.rst index ca16b4483fe..77063630f05 100644 --- a/dev/breeze/doc/08_ci_tasks.rst +++ b/dev/breeze/doc/08_ci_tasks.rst @@ -199,6 +199,23 @@ These are all available flags of ``fork-ci-reconcile`` command: :width: 100% :alt: Breeze ci fork-ci-reconcile +Approving workflow runs of external contributors +------------------------------------------------ + +The ``auto-approve-workflow-runs`` workflow runs ``approve-workflow-runs`` every 30 minutes. It approves +the waiting workflow runs of external contributors' PRs to ``main`` whose CI jobs are estimated at no more +than 310 job-minutes, that change at most 100 lines of source code and that do not touch CI definition files, and labels the other PRs ``need workflow approval``. For every PR it also +estimates the CI jobs selective checks will run and how long they take, and writes them to the job +summary. Use ``--report-only`` to see the decisions without approving runs or changing labels. The rules +are described in `CI workflows <ci/05_workflows.md>`_. + +These are all available flags of ``approve-workflow-runs`` command: + +.. image:: ./images/output_ci_approve-workflow-runs.svg + :target: https://raw.githubusercontent.com/apache/airflow/main/dev/breeze/images/output_ci_approve-workflow-runs.svg + :width: 100% + :alt: Breeze ci approve-workflow-runs + ----- Next step: Follow the `Release management tasks <09_release_management_tasks.rst>`_ guide to learn how diff --git a/dev/breeze/doc/ci/05_workflows.md b/dev/breeze/doc/ci/05_workflows.md index 7ec7871993c..b243f7afd85 100644 --- a/dev/breeze/doc/ci/05_workflows.md +++ b/dev/breeze/doc/ci/05_workflows.md @@ -98,6 +98,49 @@ whose CI runs in the contributor's fork (AIP-120), the gate skips `build-info` a everything after it, so the run finishes in seconds; see [Fork CI for external contributors](08_fork_ci_for_external_contributors.md). +### Approval of pull request runs + +GitHub holds the runs of a fork PR in `action_required` until someone with write access approves +them, for the contributors the repository's approval setting covers. The +`auto-approve-workflow-runs.yml` workflow runs every 30 minutes on `main` and calls +`breeze ci approve-workflow-runs`, which approves the waiting runs of a commit when all of these hold: + +- the PR targets `main`; PRs to release branches (`v3-4-test`, `airflow-ctl/v0-1-test`, ...) run those + branches' own workflows and always wait for a maintainer +- the PR changes nothing under `.github/`, `dev/breeze/` or `scripts/ci/`, which define the workflows + and choose the runners +- the PR changes no dependency manifest or lock file (`pyproject.toml`, `uv.lock`, `package.json`, + `pnpm-lock.yaml`, `go.mod`, Gradle build files, ...) and adds no new provider (selective checks + run from `main`'s tree and do not know it) +- selective checks for the PR's merge commit succeed, and do not need the full set of tests, all Python + and database versions, a canary build or upgraded dependencies (the estimate does not model them) +- the jobs selective checks select are estimated at no more than 310 job-minutes on public runners + (about 23% of a full-tests run, just above the cluster of single core areas and the Helm chart) +- the PR changes at most 100 lines of source code; tests (including e2e tests), docs, newsfragments, + translations, lock files and images are not counted + +Only runs created in the last 24 hours (`--max-run-age-hours`) are considered, so the workflow never +approves a backlog of older waiting runs at once; those stay with maintainers. A commit that is the +head of several open PRs is never approved automatically, because its runs cannot be told apart by +PR, and only runs started after the PR was opened and after its base branch last changed are +approved (retargeting a PR does not start new runs). Other PRs get the +`need workflow approval` label and wait for a maintainer. The label is removed when +a later commit of the PR is approved automatically. + +The workflow uses only `main`'s code. It reads the PR's metadata and fetches its merge commit as git +objects to list the changed files; it never checks out or runs the PR's code. The merge commit must +have been built from the head commit whose runs get approved, so a push after the check cannot slip +into an approved run. Approving with the workflow's `GITHUB_TOKEN` starts the runs right away, and +`workflow_run` workflows that follow them still trigger. + +The estimate comes from `dev/breeze/src/airflow_breeze/utils/ci_minutes_estimate.py`, which turns the +selective-check outputs into job groups with job-minutes and a wall-clock estimate, weighting +providers by the size of their test suites. The job summary lists every waiting PR with its decision, +the expected job groups and the estimated time. The numbers are medians of recent `Tests (AMD)` PR +runs on public runners; refresh them with `dev/calibrate_ci_minutes_estimates.py`, as +`.github/AGENTS.md` describes. Run `breeze ci approve-workflow-runs --report-only` (or dispatch the +workflow with `report-only`) to see the decisions without approving or labelling anything. + ## Fork push run These runs are triggered by a push to any branch of a contributor's fork in which the diff --git a/dev/breeze/doc/images/output_ci_approve-workflow-runs.svg b/dev/breeze/doc/images/output_ci_approve-workflow-runs.svg new file mode 100644 index 00000000000..929558eb134 --- /dev/null +++ b/dev/breeze/doc/images/output_ci_approve-workflow-runs.svg @@ -0,0 +1,142 @@ +<svg class="rich-terminal" viewBox="0 0 1482 538.0" xmlns="http://www.w3.org/2000/svg";> + <!-- Generated with Rich https://www.textualize.io --> + <style> + + @font-face { + font-family: "Fira Code"; + src: local("FiraCode-Regular"), + url("https://cdnjs.cloudflare.com/ajax/libs/firacode/6.2.0/woff2/FiraCode-Regular.woff2";) format("woff2"), + url("https://cdnjs.cloudflare.com/ajax/libs/firacode/6.2.0/woff/FiraCode-Regular.woff";) format("woff"); + font-style: normal; + font-weight: 400; + } + @font-face { + font-family: "Fira Code"; + src: local("FiraCode-Bold"), + url("https://cdnjs.cloudflare.com/ajax/libs/firacode/6.2.0/woff2/FiraCode-Bold.woff2";) format("woff2"), + url("https://cdnjs.cloudflare.com/ajax/libs/firacode/6.2.0/woff/FiraCode-Bold.woff";) format("woff"); + font-style: bold; + font-weight: 700; + } + + .breeze-ci-approve-workflow-runs-matrix { + font-family: Fira Code, monospace; + font-size: 20px; + line-height: 24.4px; + font-variant-east-asian: full-width; + } + + .breeze-ci-approve-workflow-runs-title { + font-size: 18px; + font-weight: bold; + font-family: arial; + } + + .breeze-ci-approve-workflow-runs-r1 { fill: #c5c8c6 } +.breeze-ci-approve-workflow-runs-r2 { fill: #d0b344 } +.breeze-ci-approve-workflow-runs-r3 { fill: #c5c8c6;font-weight: bold } +.breeze-ci-approve-workflow-runs-r4 { fill: #68a0b3;font-weight: bold } +.breeze-ci-approve-workflow-runs-r5 { fill: #868887 } +.breeze-ci-approve-workflow-runs-r6 { fill: #8d7b39 } +.breeze-ci-approve-workflow-runs-r7 { fill: #cc555a } +.breeze-ci-approve-workflow-runs-r8 { fill: #8a4346 } +.breeze-ci-approve-workflow-runs-r9 { fill: #98a84b;font-weight: bold } + </style> + + <defs> + <clipPath id="breeze-ci-approve-workflow-runs-clip-terminal"> + <rect x="0" y="0" width="1463.0" height="487.0" /> + </clipPath> + <clipPath id="breeze-ci-approve-workflow-runs-line-0"> + <rect x="0" y="1.5" width="1464" height="24.65"/> + </clipPath> +<clipPath id="breeze-ci-approve-workflow-runs-line-1"> + <rect x="0" y="25.9" width="1464" height="24.65"/> + </clipPath> +<clipPath id="breeze-ci-approve-workflow-runs-line-2"> + <rect x="0" y="50.3" width="1464" height="24.65"/> + </clipPath> +<clipPath id="breeze-ci-approve-workflow-runs-line-3"> + <rect x="0" y="74.7" width="1464" height="24.65"/> + </clipPath> +<clipPath id="breeze-ci-approve-workflow-runs-line-4"> + <rect x="0" y="99.1" width="1464" height="24.65"/> + </clipPath> +<clipPath id="breeze-ci-approve-workflow-runs-line-5"> + <rect x="0" y="123.5" width="1464" height="24.65"/> + </clipPath> +<clipPath id="breeze-ci-approve-workflow-runs-line-6"> + <rect x="0" y="147.9" width="1464" height="24.65"/> + </clipPath> +<clipPath id="breeze-ci-approve-workflow-runs-line-7"> + <rect x="0" y="172.3" width="1464" height="24.65"/> + </clipPath> +<clipPath id="breeze-ci-approve-workflow-runs-line-8"> + <rect x="0" y="196.7" width="1464" height="24.65"/> + </clipPath> +<clipPath id="breeze-ci-approve-workflow-runs-line-9"> + <rect x="0" y="221.1" width="1464" height="24.65"/> + </clipPath> +<clipPath id="breeze-ci-approve-workflow-runs-line-10"> + <rect x="0" y="245.5" width="1464" height="24.65"/> + </clipPath> +<clipPath id="breeze-ci-approve-workflow-runs-line-11"> + <rect x="0" y="269.9" width="1464" height="24.65"/> + </clipPath> +<clipPath id="breeze-ci-approve-workflow-runs-line-12"> + <rect x="0" y="294.3" width="1464" height="24.65"/> + </clipPath> +<clipPath id="breeze-ci-approve-workflow-runs-line-13"> + <rect x="0" y="318.7" width="1464" height="24.65"/> + </clipPath> +<clipPath id="breeze-ci-approve-workflow-runs-line-14"> + <rect x="0" y="343.1" width="1464" height="24.65"/> + </clipPath> +<clipPath id="breeze-ci-approve-workflow-runs-line-15"> + <rect x="0" y="367.5" width="1464" height="24.65"/> + </clipPath> +<clipPath id="breeze-ci-approve-workflow-runs-line-16"> + <rect x="0" y="391.9" width="1464" height="24.65"/> + </clipPath> +<clipPath id="breeze-ci-approve-workflow-runs-line-17"> + <rect x="0" y="416.3" width="1464" height="24.65"/> + </clipPath> +<clipPath id="breeze-ci-approve-workflow-runs-line-18"> + <rect x="0" y="440.7" width="1464" height="24.65"/> + </clipPath> + </defs> + + <rect fill="#292929" stroke="rgba(255,255,255,0.35)" stroke-width="1" x="1" y="1" width="1480" height="536" rx="8"/><text class="breeze-ci-approve-workflow-runs-title" fill="#c5c8c6" text-anchor="middle" x="740" y="27">Command: ci approve-workflow-runs</text> + <g transform="translate(26,22)"> + <circle cx="0" cy="0" r="7" fill="#ff5f57"/> + <circle cx="22" cy="0" r="7" fill="#febc2e"/> + <circle cx="44" cy="0" r="7" fill="#28c840"/> + </g> + + <g transform="translate(9, 41)" clip-path="url(#breeze-ci-approve-workflow-runs-clip-terminal)"> + + <g class="breeze-ci-approve-workflow-runs-matrix"> + <text class="breeze-ci-approve-workflow-runs-r1" x="1464" y="20" textLength="12.2" clip-path="url(#breeze-ci-approve-workflow-runs-line-0)"> +</text><text class="breeze-ci-approve-workflow-runs-r2" x="12.2" y="44.4" textLength="73.2" clip-path="url(#breeze-ci-approve-workflow-runs-line-1)">Usage:</text><text class="breeze-ci-approve-workflow-runs-r3" x="97.6" y="44.4" textLength="378.2" clip-path="url(#breeze-ci-approve-workflow-runs-line-1)">breeze ci approve-workflow-runs</text><text class="breeze-ci-approve-workflow-runs-r1" x="488" y="44.4" textLength="12.2" clip-path="url(#breeze-ci-approve-workflow-runs-line-1) [...] +</text><text class="breeze-ci-approve-workflow-runs-r1" x="1464" y="68.8" textLength="12.2" clip-path="url(#breeze-ci-approve-workflow-runs-line-2)"> +</text><text class="breeze-ci-approve-workflow-runs-r1" x="12.2" y="93.2" textLength="1427.4" clip-path="url(#breeze-ci-approve-workflow-runs-line-3)">Approve the waiting workflow runs of external contributors' PRs whose changes are safe to run without a maintainer's </text><text class="breeze-ci-approve-workflow-runs-r1" x="1464" y="93.2" textLength="12.2" clip-path="url(#breeze-ci-approv [...] +</text><text class="breeze-ci-approve-workflow-runs-r1" x="12.2" y="117.6" textLength="1232.2" clip-path="url(#breeze-ci-approve-workflow-runs-line-4)">look, and label the others 'need workflow approval'. Used by the auto-approve-workflow-runs workflow.</text><text class="breeze-ci-approve-workflow-runs-r1" x="1464" y="117.6" textLength="12.2" clip-path="url(#breeze-ci-approve-workflow-runs-line-4)"> +</text><text class="breeze-ci-approve-workflow-runs-r1" x="1464" y="142" textLength="12.2" clip-path="url(#breeze-ci-approve-workflow-runs-line-5)"> +</text><text class="breeze-ci-approve-workflow-runs-r5" x="0" y="166.4" textLength="24.4" clip-path="url(#breeze-ci-approve-workflow-runs-line-6)">╭─</text><text class="breeze-ci-approve-workflow-runs-r5" x="24.4" y="166.4" textLength="195.2" clip-path="url(#breeze-ci-approve-workflow-runs-line-6)"> Approval flags </text><text class="breeze-ci-approve-workflow-runs-r5" x="219.6" y="166.4" textLength="1220" clip-path="url(#breeze-ci-approve-workflow-runs-line-6)">────────── [...] +</text><text class="breeze-ci-approve-workflow-runs-r5" x="0" y="190.8" textLength="12.2" clip-path="url(#breeze-ci-approve-workflow-runs-line-7)">│</text><text class="breeze-ci-approve-workflow-runs-r4" x="24.4" y="190.8" textLength="231.8" clip-path="url(#breeze-ci-approve-workflow-runs-line-7)">--max-run-age-hours</text><text class="breeze-ci-approve-workflow-runs-r1" x="280.6" y="190.8" textLength="1159" clip-path="url(#breeze-ci-approve-workflow-runs-line-7)">Only decide o [...] +</text><text class="breeze-ci-approve-workflow-runs-r5" x="0" y="215.2" textLength="12.2" clip-path="url(#breeze-ci-approve-workflow-runs-line-8)">│</text><text class="breeze-ci-approve-workflow-runs-r5" x="280.6" y="215.2" textLength="158.6" clip-path="url(#breeze-ci-approve-workflow-runs-line-8)">[default: 24]</text><text class="breeze-ci-approve-workflow-runs-r6" x="451.4" y="215.2" textLength="244" clip-path="url(#breeze-ci-approve-workflow-runs-line-8)">(INTEGER RANGE  [...] +</text><text class="breeze-ci-approve-workflow-runs-r5" x="0" y="239.6" textLength="12.2" clip-path="url(#breeze-ci-approve-workflow-runs-line-9)">│</text><text class="breeze-ci-approve-workflow-runs-r4" x="24.4" y="239.6" textLength="231.8" clip-path="url(#breeze-ci-approve-workflow-runs-line-9)">--report-only      </text><text class="breeze-ci-approve-workflow-runs-r1" x="280.6" y="239.6" textLength="1012.6" clip-path="url(#breeze-ci-approve-workflow-runs- [...] +</text><text class="breeze-ci-approve-workflow-runs-r5" x="0" y="264" textLength="1464" clip-path="url(#breeze-ci-approve-workflow-runs-line-10)">╰──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╯</text><text class="breeze-ci-approve-workflow-runs-r1" x="1464" y="264" textLength="12.2" clip-path="url(#breeze-ci-approve-workflow-runs-line-10)"> +</text><text class="breeze-ci-approve-workflow-runs-r5" x="0" y="288.4" textLength="24.4" clip-path="url(#breeze-ci-approve-workflow-runs-line-11)">╭─</text><text class="breeze-ci-approve-workflow-runs-r5" x="24.4" y="288.4" textLength="231.8" clip-path="url(#breeze-ci-approve-workflow-runs-line-11)"> GitHub parameters </text><text class="breeze-ci-approve-workflow-runs-r5" x="256.2" y="288.4" textLength="1183.4" clip-path="url(#breeze-ci-approve-workflow-runs-line-11)">── [...] +</text><text class="breeze-ci-approve-workflow-runs-r5" x="0" y="312.8" textLength="12.2" clip-path="url(#breeze-ci-approve-workflow-runs-line-12)">│</text><text class="breeze-ci-approve-workflow-runs-r7" x="24.4" y="312.8" textLength="12.2" clip-path="url(#breeze-ci-approve-workflow-runs-line-12)">*</text><text class="breeze-ci-approve-workflow-runs-r4" x="61" y="312.8" textLength="231.8" clip-path="url(#breeze-ci-approve-workflow-runs-line-12)">--github-token     [...] +</text><text class="breeze-ci-approve-workflow-runs-r5" x="0" y="337.2" textLength="12.2" clip-path="url(#breeze-ci-approve-workflow-runs-line-13)">│</text><text class="breeze-ci-approve-workflow-runs-r4" x="61" y="337.2" textLength="231.8" clip-path="url(#breeze-ci-approve-workflow-runs-line-13)">--github-repository</text><text class="breeze-ci-approve-workflow-runs-r9" x="317.2" y="337.2" textLength="24.4" clip-path="url(#breeze-ci-approve-workflow-runs-line-13)">-g</text><text class=" [...] +</text><text class="breeze-ci-approve-workflow-runs-r5" x="0" y="361.6" textLength="1464" clip-path="url(#breeze-ci-approve-workflow-runs-line-14)">╰──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╯</text><text class="breeze-ci-approve-workflow-runs-r1" x="1464" y="361.6" textLength="12.2" clip-path="url(#breeze-ci-approve-workflow-runs-line-14)"> +</text><text class="breeze-ci-approve-workflow-runs-r5" x="0" y="386" textLength="24.4" clip-path="url(#breeze-ci-approve-workflow-runs-line-15)">╭─</text><text class="breeze-ci-approve-workflow-runs-r5" x="24.4" y="386" textLength="195.2" clip-path="url(#breeze-ci-approve-workflow-runs-line-15)"> Common options </text><text class="breeze-ci-approve-workflow-runs-r5" x="219.6" y="386" textLength="1220" clip-path="url(#breeze-ci-approve-workflow-runs-line-15)">───────────── [...] +</text><text class="breeze-ci-approve-workflow-runs-r5" x="0" y="410.4" textLength="12.2" clip-path="url(#breeze-ci-approve-workflow-runs-line-16)">│</text><text class="breeze-ci-approve-workflow-runs-r4" x="24.4" y="410.4" textLength="109.8" clip-path="url(#breeze-ci-approve-workflow-runs-line-16)">--verbose</text><text class="breeze-ci-approve-workflow-runs-r9" x="158.6" y="410.4" textLength="24.4" clip-path="url(#breeze-ci-approve-workflow-runs-line-16)">-v</text><text class="breeze-c [...] +</text><text class="breeze-ci-approve-workflow-runs-r5" x="0" y="434.8" textLength="12.2" clip-path="url(#breeze-ci-approve-workflow-runs-line-17)">│</text><text class="breeze-ci-approve-workflow-runs-r4" x="24.4" y="434.8" textLength="109.8" clip-path="url(#breeze-ci-approve-workflow-runs-line-17)">--dry-run</text><text class="breeze-ci-approve-workflow-runs-r9" x="158.6" y="434.8" textLength="24.4" clip-path="url(#breeze-ci-approve-workflow-runs-line-17)">-D</text><text class="breeze-c [...] +</text><text class="breeze-ci-approve-workflow-runs-r5" x="0" y="459.2" textLength="12.2" clip-path="url(#breeze-ci-approve-workflow-runs-line-18)">│</text><text class="breeze-ci-approve-workflow-runs-r4" x="24.4" y="459.2" textLength="109.8" clip-path="url(#breeze-ci-approve-workflow-runs-line-18)">--help   </text><text class="breeze-ci-approve-workflow-runs-r9" x="158.6" y="459.2" textLength="24.4" clip-path="url(#breeze-ci-approve-workflow-runs-line-18)">-h</text><text [...] +</text><text class="breeze-ci-approve-workflow-runs-r5" x="0" y="483.6" textLength="1464" clip-path="url(#breeze-ci-approve-workflow-runs-line-19)">╰──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╯</text><text class="breeze-ci-approve-workflow-runs-r1" x="1464" y="483.6" textLength="12.2" clip-path="url(#breeze-ci-approve-workflow-runs-line-19)"> +</text> + </g> + </g> +</svg> diff --git a/dev/breeze/doc/images/output_ci_approve-workflow-runs.txt b/dev/breeze/doc/images/output_ci_approve-workflow-runs.txt new file mode 100644 index 00000000000..f8caf2b0fe0 --- /dev/null +++ b/dev/breeze/doc/images/output_ci_approve-workflow-runs.txt @@ -0,0 +1 @@ +485b9682543dc0dd1e1922e32196b2ee diff --git a/dev/breeze/src/airflow_breeze/commands/ci_commands.py b/dev/breeze/src/airflow_breeze/commands/ci_commands.py index 1f173e5b588..f2c19b0c810 100644 --- a/dev/breeze/src/airflow_breeze/commands/ci_commands.py +++ b/dev/breeze/src/airflow_breeze/commands/ci_commands.py @@ -26,6 +26,7 @@ import subprocess import sys import tempfile from collections.abc import Iterable +from datetime import UTC, datetime, timedelta from io import StringIO from pathlib import Path from typing import TYPE_CHECKING, Any, NamedTuple @@ -1797,3 +1798,52 @@ def audit(fix: bool): run_command(shlex.split(command), check=False) failed = [result for result in report() if result.status == CheckStatus.FAIL] sys.exit(1 if failed else 0) + + +@ci_group.command( + name="approve-workflow-runs", + help="Approve the waiting workflow runs of external contributors' PRs whose changes are safe " + "to run without a maintainer's look, and label the others 'need workflow approval'. " + "Used by the auto-approve-workflow-runs workflow.", +) [email protected]( + "--github-token", + help="GitHub token able to approve workflow runs and label PRs.", + envvar="GITHUB_TOKEN", + required=True, +) [email protected]( + "--max-run-age-hours", + help="Only decide on workflow runs created within this many hours; older waiting runs are left alone.", + type=click.IntRange(min=1), + default=24, + show_default=True, + envvar="MAX_RUN_AGE_HOURS", +) [email protected]( + "--report-only", + is_flag=True, + help="Only report the decisions and estimates, without approving runs or changing labels.", + envvar="REPORT_ONLY", +) +@option_github_repository +@option_verbose +@option_dry_run +def approve_workflow_runs( + github_token: str, max_run_age_hours: int, report_only: bool, github_repository: str +): + from airflow_breeze.utils.shared_options import get_dry_run + from airflow_breeze.utils.workflow_run_approval import ( + GitHubApi, + process_waiting_runs, + write_step_summary, + ) + + api = GitHubApi( + github_repository=github_repository, + github_token=github_token, + report_only=report_only or get_dry_run(), + ) + created_since = datetime.now(UTC) - timedelta(hours=max_run_age_hours) + results = process_waiting_runs(api, created_since=created_since) + write_step_summary(results, github_repository) diff --git a/dev/breeze/src/airflow_breeze/commands/ci_commands_config.py b/dev/breeze/src/airflow_breeze/commands/ci_commands_config.py index 99042fc78a3..07723ef029e 100644 --- a/dev/breeze/src/airflow_breeze/commands/ci_commands_config.py +++ b/dev/breeze/src/airflow_breeze/commands/ci_commands_config.py @@ -28,6 +28,7 @@ CI_COMMANDS: dict[str, str | list[str]] = { "set-milestone", "fork-ci-reconcile", "audit", + "approve-workflow-runs", ], } CI_PARAMETERS: dict[str, list[dict[str, str | list[str]]]] = { @@ -140,4 +141,20 @@ CI_PARAMETERS: dict[str, list[dict[str, str | list[str]]]] = { ], }, ], + "breeze ci approve-workflow-runs": [ + { + "name": "Approval flags", + "options": [ + "--max-run-age-hours", + "--report-only", + ], + }, + { + "name": "GitHub parameters", + "options": [ + "--github-token", + "--github-repository", + ], + }, + ], } diff --git a/dev/breeze/src/airflow_breeze/utils/ci_minutes_estimate.py b/dev/breeze/src/airflow_breeze/utils/ci_minutes_estimate.py new file mode 100644 index 00000000000..65df3d6b208 --- /dev/null +++ b/dev/breeze/src/airflow_breeze/utils/ci_minutes_estimate.py @@ -0,0 +1,198 @@ +# 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. +""" +Estimate the GitHub Actions minutes a ``Tests (AMD)`` PR run takes from its selective-check outputs. + +The numbers are medians measured on recent PR runs on GitHub-hosted (public) runners, which is where +fork PRs run. They are estimates for deciding whether a change's CI run is cheap enough to approve +without a maintainer, not a bill: queueing, retries and cache misses are not modelled. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +from airflow_breeze.global_constants import all_selective_core_test_types +from airflow_breeze.utils.path_utils import AIRFLOW_ROOT_PATH + +# TODO: refresh these estimates periodically with an agent, and always after changing the CI +# workflows (see .github/AGENTS.md). Run ``dev/calibrate_ci_minutes_estimates.py`` and copy the +# medians it prints into the constants below. +# Last calibrated: 2026-10-23, from 310 successful ``Tests (AMD)`` PR runs of 2026-10-13..2026-10-23, +# 224 of which tested a partial set of providers. +# TODO: add a second set of estimates for self-hosted (RunsOn) runners once enough runs on them exist +# to measure. Their machine sizes differ, so the public-runner numbers do not carry over. + +# Each entry: (job-minutes, minutes the group adds to the run's critical path after the CI image). +BASIC_CHECKS_ONLY_ESTIMATE = (6, 3) +BASE_ESTIMATE = (47, 17) +DOCS_ESTIMATE = (15, 10) +# The PROD image waits for constraints, so its tests sit about 26 minutes after the CI image. +PROD_IMAGE_ESTIMATE = (51, 26) +KUBERNETES_ESTIMATE = (84, 29) +HELM_ESTIMATE = (55, 10) +KUSTOMIZE_ESTIMATE = (6, 6) +INTEGRATION_ESTIMATE = (5, 5) +SPECIAL_TESTS_ESTIMATE = (317, 32) +TASK_SDK_ESTIMATE = (6, 6) +AGENT_FRAMEWORK_ESTIMATE = (11, 5) +JAVA_SDK_ESTIMATE = (7, 4) +GO_SDK_ESTIMATE = (1, 1) +TS_SDK_ESTIMATE = (1, 1) +AIRFLOW_CTL_ESTIMATE = (1, 1) + +# Core tests: about 38 job-minutes for the first test type and 61 for each further one, up to +# the cost of all core test types. +CORE_FIRST_TEST_TYPE_MINUTES = 38 +CORE_EXTRA_TEST_TYPE_MINUTES = 61 +CORE_ALL_TEST_TYPES_MINUTES = 222 +CORE_LONGEST_JOB_MINUTES = {1: 11, 2: 29} +CORE_LONGEST_JOB_MINUTES_DEFAULT = 32 + +# Provider tests (DB/non-DB, lowest dependencies, compatibility, distributions, mypy) scale with the +# size of the tested providers' test suites: amazon or google cost far more than a small provider. +# Fitted on 224 runs testing 2 to 22 providers: job-minutes ~= 105 + 0.161 per provider test file, and +# the longest provider job ~= 16 + 0.015 per test file, both capped at the cost of testing all providers. +PROVIDERS_BASE_MINUTES = 105 +PROVIDERS_MINUTES_PER_TEST_FILE = 0.161 +PROVIDERS_ALL_MINUTES = 315 +PROVIDERS_LONGEST_JOB_BASE_MINUTES = 16 +PROVIDERS_LONGEST_JOB_MINUTES_PER_TEST_FILE = 0.015 +PROVIDERS_ALL_LONGEST_JOB_MINUTES = 36 + +# Build info, the CI image build, queueing and the final jobs before and after the parallel part. +RUN_OVERHEAD_MINUTES = 18 + + +@dataclass +class EstimatedJobGroup: + name: str + job_minutes: int + critical_path_minutes: int + + +@dataclass +class CiMinutesEstimate: + job_groups: list[EstimatedJobGroup] + + @property + def job_minutes(self) -> int: + return sum(group.job_minutes for group in self.job_groups) + + @property + def wall_clock_minutes(self) -> int: + if not self.job_groups: + return 0 + longest = max(group.critical_path_minutes for group in self.job_groups) + if self.job_groups[0].name == "Basic checks only": + return longest + return RUN_OVERHEAD_MINUTES + longest + + +def estimate_core_job_minutes(core_test_types: int) -> int: + if core_test_types == 0: + return 0 + return min( + CORE_FIRST_TEST_TYPE_MINUTES + CORE_EXTRA_TEST_TYPE_MINUTES * (core_test_types - 1), + CORE_ALL_TEST_TYPES_MINUTES, + ) + + +def count_provider_test_files(provider_id: str) -> int: + tests_path = AIRFLOW_ROOT_PATH / "providers" / Path(*provider_id.split(".")) / "tests" + return sum(1 for _ in tests_path.rglob("test_*.py")) + + +def find_tested_provider_ids(provider_test_types: list[str]) -> list[str] | None: + """Return the ids of individually tested providers, or None when all providers are tested.""" + provider_ids = [] + for test_type in provider_test_types: + if not test_type.startswith("Providers[") or test_type.startswith("Providers[-"): + return None + provider_ids.extend(test_type[len("Providers[") : -1].split(",")) + return provider_ids + + +def estimate_providers_group(provider_test_types: list[str]) -> EstimatedJobGroup | None: + if not provider_test_types: + return None + provider_ids = find_tested_provider_ids(provider_test_types) + if provider_ids is None: + return EstimatedJobGroup( + "Provider tests (all providers)", PROVIDERS_ALL_MINUTES, PROVIDERS_ALL_LONGEST_JOB_MINUTES + ) + test_files = sum(count_provider_test_files(provider_id) for provider_id in provider_ids) + return EstimatedJobGroup( + f"Provider tests ({len(provider_ids)} providers, {test_files} test files)", + min( + round(PROVIDERS_BASE_MINUTES + PROVIDERS_MINUTES_PER_TEST_FILE * test_files), + PROVIDERS_ALL_MINUTES, + ), + min( + round( + PROVIDERS_LONGEST_JOB_BASE_MINUTES + PROVIDERS_LONGEST_JOB_MINUTES_PER_TEST_FILE * test_files + ), + PROVIDERS_ALL_LONGEST_JOB_MINUTES, + ), + ) + + +def estimate_ci_minutes(selective_checks: Any) -> CiMinutesEstimate: + """Estimate the job groups, job-minutes and wall-clock time of a ``Tests (AMD)`` PR run.""" + if selective_checks.basic_checks_only: + return CiMinutesEstimate([EstimatedJobGroup("Basic checks only", *BASIC_CHECKS_ONLY_ESTIMATE)]) + groups = [EstimatedJobGroup("CI image, basic tests and static checks", *BASE_ESTIMATE)] + if selective_checks.docs_build: + groups.append(EstimatedJobGroup("Documentation build", *DOCS_ESTIMATE)) + if selective_checks.run_unit_tests: + core_test_types = len(selective_checks._get_core_test_types_to_run()) + if core_test_types: + all_types = core_test_types >= len(all_selective_core_test_types()) + groups.append( + EstimatedJobGroup( + f"Core tests ({'all' if all_types else core_test_types} test types)", + estimate_core_job_minutes(core_test_types), + CORE_LONGEST_JOB_MINUTES.get(core_test_types, CORE_LONGEST_JOB_MINUTES_DEFAULT), + ) + ) + groups.append(EstimatedJobGroup("Integration and system tests", *INTEGRATION_ESTIMATE)) + if not selective_checks.skip_providers_tests: + providers_group = estimate_providers_group( + selective_checks._get_providers_test_types_to_run(split_to_individual_providers=True) + ) + if providers_group: + groups.append(providers_group) + if selective_checks.full_tests_needed: + groups.append(EstimatedJobGroup("Special tests", *SPECIAL_TESTS_ESTIMATE)) + optional_groups = ( + ("prod_image_build", "PROD image build and tests", PROD_IMAGE_ESTIMATE), + ("run_kubernetes_tests", "Kubernetes tests", KUBERNETES_ESTIMATE), + ("run_helm_tests", "Helm tests", HELM_ESTIMATE), + ("run_kustomize_overlays_tests", "Kustomize overlays tests", KUSTOMIZE_ESTIMATE), + ("run_task_sdk_tests", "Task SDK tests", TASK_SDK_ESTIMATE), + ("run_agent_framework_tests", "Agent framework tests", AGENT_FRAMEWORK_ESTIMATE), + ("run_java_sdk_tests", "Java SDK tests", JAVA_SDK_ESTIMATE), + ("run_go_sdk_tests", "Go SDK tests", GO_SDK_ESTIMATE), + ("run_ts_sdk_tests", "TypeScript SDK tests", TS_SDK_ESTIMATE), + ("run_airflow_ctl_tests", "Airflow CTL tests", AIRFLOW_CTL_ESTIMATE), + ) + for flag, name, estimate in optional_groups: + if getattr(selective_checks, flag): + groups.append(EstimatedJobGroup(name, *estimate)) + return CiMinutesEstimate(groups) diff --git a/dev/breeze/src/airflow_breeze/utils/workflow_run_approval.py b/dev/breeze/src/airflow_breeze/utils/workflow_run_approval.py new file mode 100644 index 00000000000..1f59b9e948b --- /dev/null +++ b/dev/breeze/src/airflow_breeze/utils/workflow_run_approval.py @@ -0,0 +1,590 @@ +# 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. +""" +Automatic approval of workflow runs from external contributors' fork PRs. + +GitHub holds the workflow runs of a fork PR in ``action_required`` until someone with write access +approves them. This module approves the runs of PRs to ``main`` that do not touch anything that +defines CI or chooses runners, do not need the full set of tests or a dependency upgrade, and whose +jobs, as selected by selective checks, are estimated to take a limited number of job-minutes. +Everything else gets the ``need workflow approval`` label and waits for a maintainer. + +Only API metadata and git objects of the PR are read. The PR's code is never checked out or run: +selective checks come from the trusted ``main`` checkout and only see the list of changed files +and the PR's ``pyproject.toml`` contents through ``git show``. +""" + +from __future__ import annotations + +import os +import re +from dataclasses import dataclass, field +from datetime import datetime +from typing import Any + +import requests +from rich.markup import escape + +from airflow_breeze.global_constants import GithubEvents +from airflow_breeze.utils.ci_minutes_estimate import CiMinutesEstimate, estimate_ci_minutes +from airflow_breeze.utils.console import console_print +from airflow_breeze.utils.path_utils import AIRFLOW_ROOT_PATH +from airflow_breeze.utils.run_utils import run_command + +NEED_WORKFLOW_APPROVAL_LABEL = "need workflow approval" + +# Only PRs to main are approved automatically. PRs to release branches (``v3-4-test``, +# ``airflow-ctl/v0-1-test``, ...) run those branches' own, older workflows and always need a maintainer. +AUTO_APPROVABLE_BASE_BRANCH = "main" + +# Changes here can change which runner a job uses (``runs-on:``, ``runner-type`` computed by +# selective checks, the scripts build-info runs) or what the workflows do, so they always need a +# maintainer's look. +CI_DEFINITION_PATH_PREFIXES = (".github/", "dev/breeze/", "scripts/ci/") + +# Dependency manifests and lock files. Selective checks only switch to dependency-upgrade mode for +# pyproject.toml, but a changed lock file or JS / Go / Java manifest also changes what CI installs. +DEPENDENCY_FILE_PATTERN = re.compile( + r"(^|/)(pyproject\.toml|uv\.lock|requirements[^/]*\.txt|package\.json|pnpm-lock\.yaml" + r"|package-lock\.json|yarn\.lock|go\.mod|go\.sum|build\.gradle(\.kts)?|settings\.gradle(\.kts)?" + r"|gradle\.lockfile|libs\.versions\.toml|pom\.xml" + r"|gradle-wrapper\.(properties|jar)|maven-wrapper\.(properties|jar)|gradlew(\.bat)?|mvnw(\.cmd)?)$" +) + +# About 23% of a full-tests run (~1330 job-minutes on public runners), just above the cluster of +# single core areas and the Helm chart (274-301). Changes to large providers (amazon 364, google 374) +# and anything bigger wait for a maintainer, to keep the load on the shared public runners down. +MAX_AUTO_APPROVED_JOB_MINUTES = 310 + +# Larger changes to non-test code are features or behaviour changes rather than fixes, and get a +# maintainer's look even when their CI is cheap. 100 lines is the 90th percentile of the PRs that +# pass the other rules; above it the distribution thins out into feature-sized changes. +MAX_AUTO_APPROVED_SOURCE_LINES = 100 + +# Files that are not counted as source: tests (including e2e tests), docs, newsfragments, +# translations, lock files and images. +NON_SOURCE_FILE_PATTERN = re.compile( + r"(^|/)(tests?|test_utils|testing|[^/]*e2e[^/]*)/" + r"|(^|/)test_[^/]*$|_test\.(py|go)$|\.test\.(ts|tsx|js)$|\.spec\.(ts|tsx)$|Test\.java$" + r"|\.(md|rst|txt)$|(^|/)newsfragments/|(^|/)docs?/" + r"|/i18n/locales/|uv\.lock$|pnpm-lock\.yaml$|package-lock\.json$|go\.sum$" + r"|\.(svg|png|jpg|snap)$" +) + +GITHUB_API_URL = "https://api.github.com"; + + +@dataclass +class ChangedFile: + path: str + changed_lines: int + + +@dataclass +class ApprovalDecision: + approve: bool + reasons: list[str] = field(default_factory=list) + estimate: CiMinutesEstimate | None = None + + +def count_source_lines(changed_files: list[ChangedFile]) -> int: + return sum(file.changed_lines for file in changed_files if not NON_SOURCE_FILE_PATTERN.search(file.path)) + + +def find_ci_definition_files(files: list[str]) -> list[str]: + return [file for file in files if file.startswith(CI_DEFINITION_PATH_PREFIXES)] + + +def find_dependency_files(files: list[str]) -> list[str]: + return [file for file in files if DEPENDENCY_FILE_PATTERN.search(file)] + + +def find_new_provider_files(files: list[str]) -> list[str]: + """ + Return provider.yaml files that the trusted checkout does not have. + + Selective checks run from main's tree and do not know a provider the PR adds, so they cannot + estimate its tests. + """ + return [ + file + for file in files + if file.startswith("providers/") + and file.endswith("/provider.yaml") + and not (AIRFLOW_ROOT_PATH / file).exists() + ] + + +def decide_approval( + *, + base_branch: str, + changed_files: list[ChangedFile], + selective_checks: Any, +) -> ApprovalDecision: + """Decide whether the workflow runs of a fork PR commit can be approved without a maintainer.""" + try: + full_tests_needed = selective_checks.full_tests_needed + all_versions = selective_checks.all_versions + is_canary_run = selective_checks._is_canary_run() + upgrade_to_newer_dependencies = selective_checks.upgrade_to_newer_dependencies + estimate = estimate_ci_minutes(selective_checks) + except (SystemExit, Exception): + # Selective checks exit when a PR changes a suspended provider without the label allowing it, + # and raise on malformed PR content such as an invalid pyproject.toml. Either way, fail closed. + return ApprovalDecision(approve=False, reasons=["selective checks fail for this change"]) + if base_branch != AUTO_APPROVABLE_BASE_BRANCH: + return ApprovalDecision( + approve=False, + reasons=[f"it targets {base_branch}, not {AUTO_APPROVABLE_BASE_BRANCH}"], + estimate=estimate, + ) + paths = [file.path for file in changed_files] + ci_files = find_ci_definition_files(paths) + if ci_files: + return ApprovalDecision( + approve=False, + reasons=[f"it changes CI definition files: {', '.join(ci_files)}"], + estimate=estimate, + ) + dependency_files = find_dependency_files(paths) + if dependency_files: + return ApprovalDecision( + approve=False, + reasons=[f"it changes dependency files: {', '.join(dependency_files)}"], + estimate=estimate, + ) + new_provider_files = find_new_provider_files(paths) + if new_provider_files: + return ApprovalDecision( + approve=False, + reasons=[f"it adds a new provider: {', '.join(new_provider_files)}"], + estimate=estimate, + ) + if full_tests_needed: + return ApprovalDecision(approve=False, reasons=["it needs the full set of tests"], estimate=estimate) + if is_canary_run: + # The canary label adds special tests and integration jobs that the estimate does not model. + return ApprovalDecision(approve=False, reasons=["it runs as a canary build"], estimate=estimate) + if all_versions: + # The estimate models the default Python and database versions only. + return ApprovalDecision( + approve=False, reasons=["it runs all Python and database versions"], estimate=estimate + ) + if upgrade_to_newer_dependencies: + return ApprovalDecision(approve=False, reasons=["it changes dependencies"], estimate=estimate) + if estimate.job_minutes > MAX_AUTO_APPROVED_JOB_MINUTES: + return ApprovalDecision( + approve=False, + reasons=[ + f"its jobs take an estimated {estimate.job_minutes} job-minutes " + f"(more than {MAX_AUTO_APPROVED_JOB_MINUTES})" + ], + estimate=estimate, + ) + source_lines = count_source_lines(changed_files) + if source_lines > MAX_AUTO_APPROVED_SOURCE_LINES: + return ApprovalDecision( + approve=False, + reasons=[ + f"it changes {source_lines} lines of source code " + f"(more than {MAX_AUTO_APPROVED_SOURCE_LINES}, tests and docs not counted)" + ], + estimate=estimate, + ) + return ApprovalDecision( + approve=True, + reasons=[ + f"its jobs take an estimated {estimate.job_minutes} job-minutes " + f"(at most {MAX_AUTO_APPROVED_JOB_MINUTES}) and it changes {source_lines} lines of source code" + ], + estimate=estimate, + ) + + +def parse_github_time(value: str) -> datetime: + return datetime.fromisoformat(value.replace("Z", "+00:00")) + + +class GitHubApi: + def __init__(self, *, github_repository: str, github_token: str, report_only: bool): + self.github_repository = github_repository + self.report_only = report_only + self._session = requests.Session() + self._session.headers.update( + { + "Accept": "application/vnd.github+json", + "Authorization": f"Bearer {github_token}", + "X-GitHub-Api-Version": "2022-11-28", + } + ) + + def _get_paginated(self, path: str, *, item_key: str | None = None) -> list[dict[str, Any]]: + items: list[dict[str, Any]] = [] + url: str | None = f"{GITHUB_API_URL}/repos/{self.github_repository}/{path}" + while url: + response = self._session.get(url, timeout=60) + response.raise_for_status() + payload = response.json() + items.extend(payload[item_key] if item_key else payload) + url = response.links.get("next", {}).get("url") + return items + + def get_runs_waiting_for_approval(self, created_since: datetime) -> list[dict[str, Any]]: + created = created_since.strftime("%Y-%m-%dT%H:%M:%SZ") + return self._get_paginated( + f"actions/runs?status=action_required&event=pull_request&per_page=100&created=%3E%3D{created}", + item_key="workflow_runs", + ) + + def get_open_pull_requests(self) -> list[dict[str, Any]]: + return self._get_paginated("pulls?state=open&per_page=100") + + def approve_run(self, run_id: int) -> None: + if self.report_only: + console_print(f"[info]Report only: would approve workflow run {run_id}[/]") + return + response = self._session.post( + f"{GITHUB_API_URL}/repos/{self.github_repository}/actions/runs/{run_id}/approve", timeout=60 + ) + response.raise_for_status() + + def add_label(self, pr_number: int, label: str) -> None: + if self.report_only: + console_print(f"[info]Report only: would add label '{label}' to PR #{pr_number}[/]") + return + response = self._session.post( + f"{GITHUB_API_URL}/repos/{self.github_repository}/issues/{pr_number}/labels", + json={"labels": [label]}, + timeout=60, + ) + response.raise_for_status() + + def get_last_base_change(self, pr_number: int) -> datetime | None: + events = self._get_paginated(f"issues/{pr_number}/timeline?per_page=100") + changes = [ + parse_github_time(event["created_at"]) + for event in events + if event.get("event") == "base_ref_changed" + ] + return max(changes, default=None) + + def remove_label(self, pr_number: int, label: str) -> None: + if self.report_only: + console_print(f"[info]Report only: would remove label '{label}' from PR #{pr_number}[/]") + return + response = self._session.delete( + f"{GITHUB_API_URL}/repos/{self.github_repository}/issues/{pr_number}/labels/{label}", + timeout=60, + ) + if response.status_code != 404: + response.raise_for_status() + + +def is_shallow_repository() -> bool: + result = run_command( + ["git", "rev-parse", "--is-shallow-repository"], + check=False, + capture_output=True, + text=True, + dry_run_override=False, + ) + return result.stdout.strip() == "true" + + +def fetch_merge_commit(*, github_repository: str, pr_number: int, head_sha: str) -> str | None: + """ + Fetch the PR's merge commit (what the CI run tests) and return its sha. + + Return None when there is no merge commit or it was not built from ``head_sha``, so the result + is always tied to the commit whose runs get approved. + """ + ref = f"refs/remotes/auto-approve/pull/{pr_number}/merge" + # Only the merge commit and its parents are needed. Limit the depth only in an already shallow + # checkout (CI): in a full clone, --depth would record shallow boundaries and truncate history. + depth = ["--depth=2"] if is_shallow_repository() else [] + result = run_command( + [ + "git", + "fetch", + "--no-tags", + *depth, + f"https://github.com/{github_repository}.git";, + f"+refs/pull/{pr_number}/merge:{ref}", + ], + check=False, + capture_output=True, + text=True, + dry_run_override=False, + ) + if result.returncode != 0: + return None + parents = run_command( + ["git", "rev-list", "--parents", "-n", "1", ref], + check=False, + capture_output=True, + text=True, + dry_run_override=False, + ) + if parents.returncode != 0: + return None + merge_sha, *parent_shas = parents.stdout.split() + if len(parent_shas) != 2 or parent_shas[1] != head_sha: + return None + return merge_sha + + +NUMSTAT_RECORD = re.compile(r"(\d+|-)\t(\d+|-)\t(.+)", re.DOTALL) + + +def get_changed_files(merge_sha: str) -> list[ChangedFile]: + """ + List the files the merge commit changes, with their changed line counts. + + Paths must be literal for the CI-definition and source checks: without ``-z`` git C-quotes unusual + file names, and rename detection reports ``old => new`` instead of both paths. + """ + result = run_command( + [ + "git", + "-c", + "core.quotePath=false", + "diff", + "--no-renames", + "--numstat", + "-z", + f"{merge_sha}^1", + merge_sha, + ], + check=True, + capture_output=True, + text=True, + dry_run_override=False, + ) + changed_files = [] + for record in result.stdout.split("\0"): + if not record: + continue + match = NUMSTAT_RECORD.fullmatch(record) + if match is None: + raise ValueError(f"Unexpected git numstat record: {record!r}") + added, deleted, path = match.groups() + # Binary files report "-" for both counts; count them over the limit so they never pass. + changed_lines = int(added) + int(deleted) if added != "-" else MAX_AUTO_APPROVED_SOURCE_LINES + 1 + changed_files.append(ChangedFile(path=path, changed_lines=changed_lines)) + return changed_files + + +def build_selective_checks( + *, merge_sha: str, changed_files: list[ChangedFile], pr_labels: tuple[str, ...], base_branch: str +): + from airflow_breeze.utils.selective_checks import SelectiveChecks + + # Release-branch PRs select different jobs (no provider tests, for example), so selective checks + # must see the branch the PR targets. + return SelectiveChecks( + files=tuple(file.path for file in changed_files), + default_branch=base_branch, + commit_ref=merge_sha, + pr_labels=pr_labels, + github_event=GithubEvents.PULL_REQUEST, + ) + + +@dataclass +class PullRequestResult: + pr_number: int + head_sha: str + run_ids: list[int] + decision: ApprovalDecision + + +def process_waiting_runs(api: GitHubApi, *, created_since: datetime) -> list[PullRequestResult]: + """ + Decide on the runs created since ``created_since``. + + Older waiting runs are left alone, so the workflow never approves a backlog of stale runs at once. + """ + runs = api.get_runs_waiting_for_approval(created_since) + if not runs: + console_print( + f"[info]No workflow runs created since {created_since:%Y-%m-%d %H:%M} UTC wait for approval.[/]" + ) + return [] + runs_by_head: dict[tuple[str, str, str], list[dict[str, Any]]] = {} + for run in runs: + head_repository = (run.get("head_repository") or {}).get("full_name", "") + if head_repository == api.github_repository: + continue + runs_by_head.setdefault((head_repository, run["head_branch"], run["head_sha"]), []).append(run) + # The runs of fork PRs do not list their pull request, so match them by head repository, branch + # and commit. + pull_requests: dict[tuple[str, str, str], list[dict[str, Any]]] = {} + for pr in api.get_open_pull_requests(): + head_repository = pr["head"]["repo"]["full_name"] if pr["head"].get("repo") else "" + pull_requests.setdefault((head_repository, pr["head"]["ref"], pr["head"]["sha"]), []).append(pr) + results = [] + for (head_repository, head_branch, head_sha), head_runs in runs_by_head.items(): + run_ids = [run["id"] for run in head_runs] + matching_prs = pull_requests.get((head_repository, head_branch, head_sha), []) + if not matching_prs: + console_print( + f"[warning]No open PR has head {escape(head_repository)}:{escape(head_branch)}@{head_sha}; " + f"leaving runs {run_ids} alone[/]" + ) + continue + for pr in matching_prs: + if len(matching_prs) > 1: + # The runs cannot be told apart by PR, so one PR's decision must not approve them all. + decision = ApprovalDecision( + approve=False, reasons=["its head commit is the head of several open PRs"] + ) + else: + decision = decide_for_pull_request(api=api, pr=pr) + if decision.approve: + run_ids = find_runs_started_for_current_pr(api=api, pr=pr, runs=head_runs) + if not run_ids: + decision = ApprovalDecision( + approve=False, + reasons=["its waiting runs started before the PR was opened or retargeted"], + estimate=decision.estimate, + ) + run_ids = [run["id"] for run in head_runs] + results.append( + PullRequestResult( + pr_number=pr["number"], head_sha=head_sha, run_ids=run_ids, decision=decision + ) + ) + apply_decision(api=api, pr=pr, run_ids=run_ids, decision=decision) + return results + + +def find_runs_started_for_current_pr( + *, api: GitHubApi, pr: dict[str, Any], runs: list[dict[str, Any]] +) -> list[int]: + """ + Return the runs started after the PR was opened and after its base branch last changed. + + Retargeting a PR does not start new runs, and a closed PR can leave runs with the same head behind. + Such runs were started for another base branch, so the decision about the current PR must not + approve them. + """ + started_after = parse_github_time(pr["created_at"]) + last_base_change = api.get_last_base_change(pr["number"]) + if last_base_change is not None: + started_after = max(started_after, last_base_change) + return [run["id"] for run in runs if parse_github_time(run["created_at"]) >= started_after] + + +def decide_for_pull_request(*, api: GitHubApi, pr: dict[str, Any]) -> ApprovalDecision: + merge_sha = fetch_merge_commit( + github_repository=api.github_repository, pr_number=pr["number"], head_sha=pr["head"]["sha"] + ) + if merge_sha is None: + return ApprovalDecision( + approve=False, reasons=["its merge commit is missing or was not built from the PR's head commit"] + ) + try: + changed_files = get_changed_files(merge_sha) + except ValueError: + return ApprovalDecision(approve=False, reasons=["its changed files could not be listed"]) + pr_labels = tuple(label["name"] for label in pr.get("labels", [])) + return decide_approval( + base_branch=pr["base"]["ref"], + changed_files=changed_files, + selective_checks=build_selective_checks( + merge_sha=merge_sha, + changed_files=changed_files, + pr_labels=pr_labels, + base_branch=pr["base"]["ref"], + ), + ) + + +def apply_decision( + *, api: GitHubApi, pr: dict[str, Any], run_ids: list[int], decision: ApprovalDecision +) -> None: + pr_number = pr["number"] + has_label = any(label["name"] == NEED_WORKFLOW_APPROVAL_LABEL for label in pr.get("labels", [])) + console_print(f"[info]PR #{pr_number}: {format_estimate(decision.estimate)}[/]") + if decision.approve: + console_print( + f"[success]Approving runs {run_ids} of PR #{pr_number}: {escape('; '.join(decision.reasons))}[/]" + ) + for run_id in run_ids: + api.approve_run(run_id) + if has_label: + api.remove_label(pr_number, NEED_WORKFLOW_APPROVAL_LABEL) + return + # Reasons can contain contributor-chosen file names, which must not be read as Rich markup. + console_print(f"[warning]PR #{pr_number} needs a maintainer: {escape('; '.join(decision.reasons))}[/]") + if not has_label: + api.add_label(pr_number, NEED_WORKFLOW_APPROVAL_LABEL) + + +MARKDOWN_SPECIAL_CHARACTERS = re.compile(r"([\\`*_{}\[\]()<>#+!|~])") + + +def escape_markdown(text: str) -> str: + """Escape Markdown in contributor-controlled text, such as file names, for the job summary.""" + return MARKDOWN_SPECIAL_CHARACTERS.sub(r"\\\1", text) + + +def format_estimate(estimate: CiMinutesEstimate | None) -> str: + if estimate is None: + return "n/a" + return f"~{estimate.job_minutes} job-min, ~{estimate.wall_clock_minutes} min wall clock" + + +def write_step_summary(results: list[PullRequestResult], github_repository: str) -> None: + summary_path = os.environ.get("GITHUB_STEP_SUMMARY") + if not summary_path: + return + lines = [ + "## Workflow run approval", + "", + "Estimates are for `Tests (AMD)` on public runners.", + "", + "| PR | Head | Decision | Estimated CI time | Reasons |", + "|---|---|---|---|---|", + ] + for result in results: + decision = "approved" if result.decision.approve else f"`{NEED_WORKFLOW_APPROVAL_LABEL}`" + lines.append( + f"| [#{result.pr_number}](https://github.com/{github_repository}/pull/{result.pr_number}) " + f"| `{result.head_sha[:10]}` | {decision} | {format_estimate(result.decision.estimate)} " + f"| {escape_markdown('; '.join(result.decision.reasons))} |" + ) + for result in results: + estimate = result.decision.estimate + if estimate is None: + continue + lines.extend( + [ + "", + f"### #{result.pr_number}: expected jobs", + "", + "| Job group | Estimated job-minutes | Critical path (min) |", + "|---|---|---|", + *( + f"| {group.name} | {group.job_minutes} | {group.critical_path_minutes} |" + for group in estimate.job_groups + ), + f"| **Total** | **{estimate.job_minutes}** | **~{estimate.wall_clock_minutes} wall clock** |", + ] + ) + with open(summary_path, "a") as summary_file: + summary_file.write("\n".join(lines) + "\n") diff --git a/dev/breeze/tests/test_ci_minutes_estimate.py b/dev/breeze/tests/test_ci_minutes_estimate.py new file mode 100644 index 00000000000..5396e3e1465 --- /dev/null +++ b/dev/breeze/tests/test_ci_minutes_estimate.py @@ -0,0 +1,168 @@ +# 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 unittest import mock + +import pytest + +from airflow_breeze.utils.ci_minutes_estimate import ( + BASE_ESTIMATE, + BASIC_CHECKS_ONLY_ESTIMATE, + CORE_ALL_TEST_TYPES_MINUTES, + PROVIDERS_ALL_LONGEST_JOB_MINUTES, + PROVIDERS_ALL_MINUTES, + RUN_OVERHEAD_MINUTES, + estimate_ci_minutes, + estimate_core_job_minutes, + estimate_providers_group, + find_tested_provider_ids, +) +from airflow_breeze.utils.selective_checks import SelectiveChecks + +OPTIONAL_FLAGS = ( + "docs_build", + "run_unit_tests", + "full_tests_needed", + "prod_image_build", + "run_kubernetes_tests", + "run_helm_tests", + "run_kustomize_overlays_tests", + "run_task_sdk_tests", + "run_agent_framework_tests", + "run_java_sdk_tests", + "run_go_sdk_tests", + "run_ts_sdk_tests", + "run_airflow_ctl_tests", +) + + +def _selective_checks( + *, core_test_types=(), provider_test_types=(), basic_checks_only=False, **flags +) -> mock.MagicMock: + selective_checks = mock.MagicMock(spec=SelectiveChecks) + selective_checks.basic_checks_only = basic_checks_only + selective_checks.skip_providers_tests = not provider_test_types + for flag in OPTIONAL_FLAGS: + setattr(selective_checks, flag, flags.get(flag, False)) + selective_checks._get_core_test_types_to_run.return_value = list(core_test_types) + selective_checks._get_providers_test_types_to_run.return_value = list(provider_test_types) + return selective_checks + + [email protected]( + ("core_test_types", "expected"), + [(0, 0), (1, 38), (2, 99), (3, 160), (6, CORE_ALL_TEST_TYPES_MINUTES)], +) +def test_estimate_core_job_minutes(core_test_types, expected): + assert estimate_core_job_minutes(core_test_types) == expected + + [email protected]( + ("provider_test_types", "expected"), + [ + (["Providers[amazon]", "Providers[google]"], ["amazon", "google"]), + (["Providers[common.compat,fab]"], ["common.compat", "fab"]), + (["Providers[-amazon,google]", "Providers[amazon]"], None), + (["Providers"], None), + ], +) +def test_find_tested_provider_ids(provider_test_types, expected): + assert find_tested_provider_ids(provider_test_types) == expected + + [email protected]("airflow_breeze.utils.ci_minutes_estimate.count_provider_test_files", autospec=True) +def test_estimate_providers_group_weights_providers_by_test_files(mock_count): + mock_count.side_effect = lambda provider_id: {"fab": 50, "amazon": 500}[provider_id] + + small = estimate_providers_group(["Providers[fab]"]) + large = estimate_providers_group(["Providers[amazon]"]) + + assert small.name == "Provider tests (1 providers, 50 test files)" + assert small.job_minutes < large.job_minutes + assert small.critical_path_minutes < large.critical_path_minutes + + +def test_estimate_providers_group_for_all_providers(): + group = estimate_providers_group(["Providers"]) + + assert (group.job_minutes, group.critical_path_minutes) == ( + PROVIDERS_ALL_MINUTES, + PROVIDERS_ALL_LONGEST_JOB_MINUTES, + ) + + +def test_estimate_providers_group_without_providers(): + assert estimate_providers_group([]) is None + + +def test_estimate_basic_checks_only(): + estimate = estimate_ci_minutes(_selective_checks(basic_checks_only=True)) + + assert [group.name for group in estimate.job_groups] == ["Basic checks only"] + assert estimate.job_minutes == BASIC_CHECKS_ONLY_ESTIMATE[0] + assert estimate.wall_clock_minutes == BASIC_CHECKS_ONLY_ESTIMATE[1] + + [email protected]("airflow_breeze.utils.ci_minutes_estimate.count_provider_test_files", return_value=100) +def test_estimate_lists_the_job_groups_that_run(mock_count): + estimate = estimate_ci_minutes( + _selective_checks( + core_test_types=["Always"], + provider_test_types=["Providers[http]"], + docs_build=True, + run_unit_tests=True, + prod_image_build=True, + run_helm_tests=True, + ) + ) + + assert [group.name for group in estimate.job_groups] == [ + "CI image, basic tests and static checks", + "Documentation build", + "Core tests (1 test types)", + "Integration and system tests", + "Provider tests (1 providers, 100 test files)", + "PROD image build and tests", + "Helm tests", + ] + assert estimate.job_minutes == sum(group.job_minutes for group in estimate.job_groups) + assert estimate.wall_clock_minutes == RUN_OVERHEAD_MINUTES + max( + group.critical_path_minutes for group in estimate.job_groups + ) + + +def test_estimate_full_tests_adds_special_tests_and_all_core_types(): + estimate = estimate_ci_minutes( + _selective_checks( + core_test_types=["API", "Always", "CLI", "Core", "Other", "Serialization"], + provider_test_types=["Providers"], + run_unit_tests=True, + full_tests_needed=True, + ) + ) + + names = [group.name for group in estimate.job_groups] + assert "Core tests (all test types)" in names + assert "Provider tests (all providers)" in names + assert "Special tests" in names + + +def test_estimate_without_unit_tests_has_only_base_group(): + estimate = estimate_ci_minutes(_selective_checks()) + + assert estimate.job_minutes == BASE_ESTIMATE[0] diff --git a/dev/breeze/tests/test_workflow_run_approval.py b/dev/breeze/tests/test_workflow_run_approval.py new file mode 100644 index 00000000000..e472bc8c777 --- /dev/null +++ b/dev/breeze/tests/test_workflow_run_approval.py @@ -0,0 +1,617 @@ +# 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 subprocess +from datetime import UTC, datetime +from unittest import mock + +import pytest + +from airflow_breeze.utils.ci_minutes_estimate import CiMinutesEstimate, EstimatedJobGroup +from airflow_breeze.utils.selective_checks import SelectiveChecks +from airflow_breeze.utils.workflow_run_approval import ( + MAX_AUTO_APPROVED_JOB_MINUTES, + MAX_AUTO_APPROVED_SOURCE_LINES, + NEED_WORKFLOW_APPROVAL_LABEL, + ApprovalDecision, + ChangedFile, + GitHubApi, + PullRequestResult, + apply_decision, + build_selective_checks, + count_source_lines, + decide_approval, + decide_for_pull_request, + escape_markdown, + fetch_merge_commit, + get_changed_files, + process_waiting_runs, + write_step_summary, +) + +MODULE = "airflow_breeze.utils.workflow_run_approval" +CREATED_SINCE = datetime(2026, 10, 22, 8, 0, tzinfo=UTC) +CORE_CHANGE = [ChangedFile("airflow-core/src/airflow/models/dag.py", 10)] +PROVIDER_CHANGE = [ChangedFile(f"providers/http/src/airflow/providers/http/f_{i}.py", 10) for i in range(8)] +LARGE_SOURCE_CHANGE = [ + ChangedFile("airflow-core/src/airflow/models/dag.py", MAX_AUTO_APPROVED_SOURCE_LINES + 1) +] +LARGE_TEST_CHANGE = [ + ChangedFile("airflow-core/src/airflow/models/dag.py", 10), + ChangedFile("airflow-core/tests/unit/models/test_dag.py", 500), +] + + +def _selective_checks( + *, full_tests_needed=False, upgrade_to_newer_dependencies=False, all_versions=False, canary=False +) -> mock.MagicMock: + selective_checks = mock.MagicMock(spec=SelectiveChecks) + selective_checks.full_tests_needed = full_tests_needed + selective_checks.upgrade_to_newer_dependencies = upgrade_to_newer_dependencies + selective_checks.all_versions = all_versions + selective_checks._is_canary_run.return_value = canary + return selective_checks + + +def _estimate(job_minutes: int) -> CiMinutesEstimate: + return CiMinutesEstimate([EstimatedJobGroup("Tests", job_minutes, 20)]) + + [email protected] +def mock_estimate(): + with mock.patch(f"{MODULE}.estimate_ci_minutes", autospec=True) as estimate_ci_minutes: + estimate_ci_minutes.return_value = _estimate(100) + yield estimate_ci_minutes + + [email protected]( + ("changed_files", "selective_checks", "estimated_minutes", "approve", "reason"), + [ + pytest.param(PROVIDER_CHANGE, _selective_checks(), 100, True, "at most", id="few-job-minutes"), + pytest.param( + CORE_CHANGE, + _selective_checks(), + MAX_AUTO_APPROVED_JOB_MINUTES, + True, + "at most", + id="at-the-threshold", + ), + pytest.param( + CORE_CHANGE, + _selective_checks(), + MAX_AUTO_APPROVED_JOB_MINUTES + 1, + False, + "more than", + id="too-many-job-minutes", + ), + pytest.param( + LARGE_SOURCE_CHANGE, + _selective_checks(), + 100, + False, + "lines of source code", + id="too-many-source-lines", + ), + pytest.param(LARGE_TEST_CHANGE, _selective_checks(), 100, True, "at most", id="large-test-change"), + pytest.param( + [ChangedFile(".github/workflows/ci-amd.yml", 1)], + _selective_checks(), + 100, + False, + "CI definition files", + id="workflow-change", + ), + pytest.param( + [ChangedFile("dev/breeze/src/airflow_breeze/utils/selective_checks.py", 1)], + _selective_checks(), + 100, + False, + "CI definition files", + id="selective-checks-change", + ), + pytest.param( + CORE_CHANGE, + _selective_checks(full_tests_needed=True), + 100, + False, + "full set of tests", + id="full-tests", + ), + pytest.param( + [ChangedFile("uv.lock", 4)], + _selective_checks(), + 100, + False, + "dependency files: uv.lock", + id="uv-lock-change", + ), + pytest.param( + [ChangedFile("airflow-core/src/airflow/ui/package.json", 2)], + _selective_checks(), + 100, + False, + "dependency files", + id="package-json-change", + ), + pytest.param( + [ChangedFile("java-sdk/gradle/wrapper/gradle-wrapper.properties", 2)], + _selective_checks(), + 100, + False, + "dependency files", + id="gradle-wrapper-change", + ), + pytest.param( + [ChangedFile("go-sdk/go.mod", 2)], + _selective_checks(), + 100, + False, + "dependency files", + id="go-mod-change", + ), + pytest.param( + [ChangedFile("providers/brand_new/provider.yaml", 20)], + _selective_checks(), + 100, + False, + "adds a new provider", + id="new-provider", + ), + pytest.param( + [ChangedFile("providers/http/provider.yaml", 2)], + _selective_checks(), + 100, + True, + "at most", + id="existing-provider-yaml", + ), + pytest.param( + CORE_CHANGE, + _selective_checks(canary=True), + 100, + False, + "canary build", + id="canary", + ), + pytest.param( + CORE_CHANGE, + _selective_checks(all_versions=True), + 100, + False, + "all Python and database versions", + id="all-versions", + ), + pytest.param( + CORE_CHANGE, + _selective_checks(upgrade_to_newer_dependencies=True), + 100, + False, + "dependencies", + id="dependencies", + ), + ], +) +def test_decide_approval(mock_estimate, changed_files, selective_checks, estimated_minutes, approve, reason): + mock_estimate.return_value = _estimate(estimated_minutes) + + decision = decide_approval( + base_branch="main", + changed_files=changed_files, + selective_checks=selective_checks, + ) + + assert decision.approve is approve + assert reason in "; ".join(decision.reasons) + assert decision.estimate == mock_estimate.return_value + + [email protected]("error", [SystemExit(1), ValueError("Invalid pyproject.toml")]) +def test_decide_approval_when_selective_checks_fail(mock_estimate, error): + mock_estimate.side_effect = error + + decision = decide_approval( + base_branch="main", changed_files=CORE_CHANGE, selective_checks=_selective_checks() + ) + + assert decision == ApprovalDecision(approve=False, reasons=["selective checks fail for this change"]) + + +def _completed(stdout: str = "", returncode: int = 0) -> subprocess.CompletedProcess: + return subprocess.CompletedProcess(args=[], returncode=returncode, stdout=stdout, stderr="") + + [email protected]( + ("fetch_returncode", "parents", "expected"), + [ + pytest.param(0, "merge base head\n", "merge", id="built-from-head"), + pytest.param(0, "merge base other\n", None, id="built-from-other-commit"), + pytest.param(0, "merge head\n", None, id="not-a-merge-commit"), + pytest.param(128, "", None, id="no-merge-ref"), + ], +) [email protected](f"{MODULE}.is_shallow_repository", autospec=True, return_value=True) [email protected](f"{MODULE}.run_command", autospec=True) +def test_fetch_merge_commit(mock_run_command, mock_is_shallow, fetch_returncode, parents, expected): + mock_run_command.side_effect = [_completed(returncode=fetch_returncode), _completed(parents)] + + assert fetch_merge_commit(github_repository="apache/airflow", pr_number=1, head_sha="head") == expected + + [email protected](("shallow", "expect_depth"), [("true", True), ("false", False)]) [email protected](f"{MODULE}.run_command", autospec=True) +def test_fetch_merge_commit_limits_depth_only_in_shallow_checkouts(mock_run_command, shallow, expect_depth): + mock_run_command.side_effect = [_completed(f"{shallow}\n"), _completed(), _completed("merge base head\n")] + + fetch_merge_commit(github_repository="apache/airflow", pr_number=1, head_sha="head") + + fetch_command = mock_run_command.call_args_list[1].args[0] + assert ("--depth=2" in fetch_command) is expect_depth + + [email protected](f"{MODULE}.run_command", autospec=True) +def test_get_changed_files_reads_literal_paths(mock_run_command): + mock_run_command.return_value = _completed( + '3\t2\tREADME.md\x00-\t-\tdata.bin\x001\t0\t.github/workflows/check-\u00e9 "x".yml\x00' + ) + + changed_files = get_changed_files("merge") + + assert changed_files == [ + ChangedFile("README.md", 5), + ChangedFile("data.bin", MAX_AUTO_APPROVED_SOURCE_LINES + 1), + ChangedFile('.github/workflows/check-\u00e9 "x".yml', 1), + ] + command = mock_run_command.call_args.args[0] + assert command[:3] == ["git", "-c", "core.quotePath=false"] + assert {"--no-renames", "--numstat", "-z"} <= set(command) + + [email protected](f"{MODULE}.run_command", autospec=True) +def test_get_changed_files_rejects_unexpected_output(mock_run_command): + mock_run_command.return_value = _completed("README.md => docs/README.md\x00") + + with pytest.raises(ValueError, match="Unexpected git numstat record"): + get_changed_files("merge") + + [email protected](f"{MODULE}.get_changed_files", autospec=True, side_effect=ValueError("bad")) [email protected](f"{MODULE}.fetch_merge_commit", autospec=True, return_value="merge") +def test_decide_for_pull_request_fails_closed_when_files_cannot_be_listed(mock_fetch, mock_get_changed_files): + api = mock.MagicMock(spec=GitHubApi) + api.github_repository = "apache/airflow" + + decision = decide_for_pull_request(api=api, pr=_pr(10, "aaa")) + + assert decision == ApprovalDecision(approve=False, reasons=["its changed files could not be listed"]) + + [email protected]( + ("path", "is_source"), + [ + ("airflow-core/src/airflow/models/dag.py", True), + ("task-sdk/src/airflow/sdk/definitions/dag.py", True), + ("chart/values.yaml", True), + ("providers/amazon/tests/unit/aws/hooks/test_s3.py", False), + ("airflow-e2e-tests/tests/airflow_e2e_tests/test_basic.py", False), + ("java-sdk/e2e/src/main/java/example/Dags.java", False), + ("airflow-core/src/airflow/utils/foo_e2e_helper.py", True), + ("go-sdk/pkg/worker/worker_test.go", False), + ("airflow-core/src/airflow/ui/src/pages/Dag.test.tsx", False), + ("airflow-core/src/airflow/ui/public/i18n/locales/de/common.json", False), + ("airflow-core/newsfragments/12345.significant.rst", False), + ("airflow-core/docs/howto/index.rst", False), + ("uv.lock", False), + ], +) +def test_count_source_lines(path, is_source): + assert count_source_lines([ChangedFile(path, 7)]) == (7 if is_source else 0) + + +def _pr(number: int, head_sha: str, labels=(), head_repo="contributor/airflow", head_ref="fix") -> dict: + return { + "number": number, + "head": {"sha": head_sha, "ref": head_ref, "repo": {"full_name": head_repo}}, + "created_at": "2026-10-22T09:00:00Z", + "labels": [{"name": label} for label in labels], + } + + +def _run( + run_id: int, + head_sha: str, + head_repo="contributor/airflow", + head_branch="fix", + created_at="2026-10-22T10:00:00Z", +) -> dict: + return { + "id": run_id, + "head_sha": head_sha, + "head_branch": head_branch, + "created_at": created_at, + "head_repository": {"full_name": head_repo}, + } + + [email protected](f"{MODULE}.decide_for_pull_request", autospec=True) +def test_process_waiting_runs_groups_runs_by_pr_commit(mock_decide): + api = mock.MagicMock(spec=GitHubApi) + api.github_repository = "apache/airflow" + api.get_last_base_change.return_value = None + api.get_runs_waiting_for_approval.return_value = [ + _run(1, "aaa"), + _run(2, "aaa"), + _run(3, "bbb"), + _run(4, "ccc", head_repo="apache/airflow"), + _run(5, "ddd"), + ] + api.get_open_pull_requests.return_value = [ + _pr(10, "aaa"), + _pr(11, "bbb", labels=[NEED_WORKFLOW_APPROVAL_LABEL]), + ] + mock_decide.side_effect = [ + ApprovalDecision(approve=True, reasons=["localized"]), + ApprovalDecision(approve=True, reasons=["cheap"]), + ] + + results = process_waiting_runs(api, created_since=CREATED_SINCE) + + api.get_runs_waiting_for_approval.assert_called_once_with(CREATED_SINCE) + + assert [(r.pr_number, r.run_ids) for r in results] == [(10, [1, 2]), (11, [3])] + assert api.approve_run.call_args_list == [mock.call(1), mock.call(2), mock.call(3)] + api.remove_label.assert_called_once_with(11, NEED_WORKFLOW_APPROVAL_LABEL) + + [email protected]( + ("labels", "expected_add_label_calls"), + [((), [mock.call(10, NEED_WORKFLOW_APPROVAL_LABEL)]), ((NEED_WORKFLOW_APPROVAL_LABEL,), [])], +) +def test_apply_decision_labels_prs_that_need_a_maintainer(labels, expected_add_label_calls): + api = mock.MagicMock(spec=GitHubApi) + + apply_decision( + api=api, + pr=_pr(10, "aaa", labels=labels), + run_ids=[1], + decision=ApprovalDecision(approve=False, reasons=["it needs the full set of tests"]), + ) + + api.approve_run.assert_not_called() + assert api.add_label.call_args_list == expected_add_label_calls + + [email protected]("method", ["approve_run", "add_label", "remove_label"]) +def test_report_only_api_does_not_change_anything(method): + api = GitHubApi(github_repository="apache/airflow", github_token="token", report_only=True) + with mock.patch.object(api, "_session", autospec=True) as session: + args = (1,) if method == "approve_run" else (1, NEED_WORKFLOW_APPROVAL_LABEL) + getattr(api, method)(*args) + + assert session.method_calls == [] + + +def test_write_step_summary_lists_expected_jobs_and_time(tmp_path, monkeypatch): + summary = tmp_path / "summary.md" + monkeypatch.setenv("GITHUB_STEP_SUMMARY", str(summary)) + estimate = CiMinutesEstimate( + [ + EstimatedJobGroup("CI image, basic tests and static checks", 44, 17), + EstimatedJobGroup("Provider tests (1 providers, 74 test files)", 110, 16), + ] + ) + + write_step_summary( + [ + PullRequestResult( + pr_number=10, + head_sha="abcdef0123456789", + run_ids=[1], + decision=ApprovalDecision(approve=True, reasons=["localized"], estimate=estimate), + ) + ], + "apache/airflow", + ) + + text = summary.read_text() + assert "| [#10](https://github.com/apache/airflow/pull/10) | `abcdef0123` | approved |" in text + assert "~154 job-min, ~35 min wall clock" in text + assert "| Provider tests (1 providers, 74 test files) | 110 | 16 |" in text + + [email protected]( + ("paths", "approve"), + [ + pytest.param( + ["providers/discord/src/airflow/providers/discord/hooks/discord_webhook.py"], + True, + id="small-provider-change", + ), + pytest.param( + ["providers/amazon/src/airflow/providers/amazon/aws/hooks/s3.py"], + False, + id="large-provider-change", + ), + pytest.param(["airflow-core/src/airflow/models/dagrun.py"], True, id="core-change"), + pytest.param(["task-sdk/src/airflow/sdk/definitions/dag.py"], False, id="task-sdk-change"), + ], +) +def test_decide_approval_with_real_selective_checks(paths, approve): + changed_files = [ChangedFile(path, 10) for path in paths] + selective_checks = SelectiveChecks(files=tuple(paths), commit_ref="HEAD", pr_labels=()) + + decision = decide_approval( + base_branch="main", changed_files=changed_files, selective_checks=selective_checks + ) + + assert decision.approve is approve, decision.reasons + assert decision.estimate.job_minutes > 0 + + [email protected](("base_branch", "runs_provider_tests"), [("main", True), ("v3-4-test", False)]) +def test_build_selective_checks_uses_the_pr_base_branch(base_branch, runs_provider_tests): + selective_checks = build_selective_checks( + merge_sha="HEAD", + changed_files=[ChangedFile("providers/http/src/airflow/providers/http/hooks/http.py", 10)], + pr_labels=(), + base_branch=base_branch, + ) + + assert selective_checks.skip_providers_tests is not runs_provider_tests + + [email protected]("base_branch", ["v3-4-test", "airflow-ctl/v0-1-test"]) +def test_decide_approval_never_approves_prs_to_other_branches(mock_estimate, base_branch): + decision = decide_approval( + base_branch=base_branch, changed_files=CORE_CHANGE, selective_checks=_selective_checks() + ) + + assert decision.approve is False + assert decision.reasons == [f"it targets {base_branch}, not main"] + assert decision.estimate == mock_estimate.return_value + + +def test_get_runs_waiting_for_approval_only_lists_runs_created_since(): + api = GitHubApi(github_repository="apache/airflow", github_token="token", report_only=True) + with mock.patch.object(api, "_get_paginated", autospec=True, return_value=[]) as get_paginated: + api.get_runs_waiting_for_approval(CREATED_SINCE) + + assert "&created=%3E%3D2026-10-22T08:00:00Z" in get_paginated.call_args.args[0] + + +def test_process_waiting_runs_without_runs(): + api = mock.MagicMock(spec=GitHubApi) + api.get_runs_waiting_for_approval.return_value = [] + + assert process_waiting_runs(api, created_since=CREATED_SINCE) == [] + api.get_open_pull_requests.assert_not_called() + + [email protected](f"{MODULE}.decide_for_pull_request", autospec=True) +def test_process_waiting_runs_does_not_approve_a_commit_shared_by_several_prs(mock_decide): + api = mock.MagicMock(spec=GitHubApi) + api.github_repository = "apache/airflow" + api.get_runs_waiting_for_approval.return_value = [_run(1, "aaa"), _run(2, "aaa")] + api.get_open_pull_requests.return_value = [_pr(10, "aaa"), _pr(11, "aaa")] + + results = process_waiting_runs(api, created_since=CREATED_SINCE) + + assert [r.pr_number for r in results] == [10, 11] + assert all(not r.decision.approve for r in results) + mock_decide.assert_not_called() + api.approve_run.assert_not_called() + assert api.add_label.call_args_list == [ + mock.call(10, NEED_WORKFLOW_APPROVAL_LABEL), + mock.call(11, NEED_WORKFLOW_APPROVAL_LABEL), + ] + + [email protected]( + ("args", "report_only"), + [([], False), (["--report-only"], True), (["--dry-run"], True)], +) [email protected](f"{MODULE}.write_step_summary", autospec=True) [email protected](f"{MODULE}.process_waiting_runs", autospec=True, return_value=[]) [email protected](f"{MODULE}.GitHubApi", autospec=True) +def test_approve_workflow_runs_command_never_writes_in_dry_run( + mock_api, mock_process, mock_summary, args, report_only +): + from click.testing import CliRunner + + from airflow_breeze.commands.ci_commands import approve_workflow_runs + from airflow_breeze.utils.shared_options import set_dry_run + + try: + result = CliRunner().invoke(approve_workflow_runs, ["--github-token", "token", *args]) + finally: + set_dry_run(False) + + assert result.exit_code == 0, result.output + assert mock_api.call_args.kwargs["report_only"] is report_only + + [email protected](f"{MODULE}.decide_for_pull_request", autospec=True) +def test_process_waiting_runs_requires_the_run_branch_to_match_the_pr(mock_decide): + api = mock.MagicMock(spec=GitHubApi) + api.github_repository = "apache/airflow" + api.get_runs_waiting_for_approval.return_value = [_run(1, "aaa", head_branch="other")] + api.get_open_pull_requests.return_value = [_pr(10, "aaa", head_ref="fix")] + + assert process_waiting_runs(api, created_since=CREATED_SINCE) == [] + mock_decide.assert_not_called() + api.approve_run.assert_not_called() + + +def test_apply_decision_prints_contributor_paths_without_markup(): + api = mock.MagicMock(spec=GitHubApi) + + apply_decision( + api=api, + pr=_pr(10, "aaa"), + run_ids=[1], + decision=ApprovalDecision( + approve=False, reasons=["it changes CI definition files: .github/a[/bogus].txt"] + ), + ) + + api.add_label.assert_called_once_with(10, NEED_WORKFLOW_APPROVAL_LABEL) + + [email protected]( + ("last_base_change", "approved_runs"), + [ + pytest.param(None, [2, 3], id="no-retarget"), + pytest.param(datetime(2026, 10, 22, 10, 30, tzinfo=UTC), [3], id="retargeted-after-first-run"), + pytest.param(datetime(2026, 10, 22, 12, 0, tzinfo=UTC), [], id="retargeted-after-all-runs"), + ], +) [email protected](f"{MODULE}.decide_for_pull_request", autospec=True) +def test_process_waiting_runs_only_approves_runs_started_for_the_current_pr( + mock_decide, last_base_change, approved_runs +): + api = mock.MagicMock(spec=GitHubApi) + api.github_repository = "apache/airflow" + api.get_runs_waiting_for_approval.return_value = [ + _run(1, "aaa", created_at="2026-10-22T08:00:00Z"), + _run(2, "aaa", created_at="2026-10-22T10:00:00Z"), + _run(3, "aaa", created_at="2026-10-22T11:00:00Z"), + ] + api.get_open_pull_requests.return_value = [_pr(10, "aaa")] + api.get_last_base_change.return_value = last_base_change + mock_decide.return_value = ApprovalDecision(approve=True, reasons=["cheap"]) + + results = process_waiting_runs(api, created_since=CREATED_SINCE) + + assert api.approve_run.call_args_list == [mock.call(run_id) for run_id in approved_runs] + assert results[0].decision.approve is bool(approved_runs) + + +def test_get_last_base_change(): + api = GitHubApi(github_repository="apache/airflow", github_token="token", report_only=True) + events = [ + {"event": "labeled", "created_at": "2026-10-22T12:00:00Z"}, + {"event": "base_ref_changed", "created_at": "2026-10-22T09:00:00Z"}, + {"event": "base_ref_changed", "created_at": "2026-10-22T10:00:00Z"}, + ] + with mock.patch.object(api, "_get_paginated", autospec=True, return_value=events): + assert api.get_last_base_change(10) == datetime(2026, 10, 22, 10, 0, tzinfo=UTC) + + +def test_escape_markdown(): + assert escape_markdown("it changes .github/[x](http://evil)|`y`") == ( + "it changes .github/\\[x\\]\\(http://evil\\)\\|\\`y\\`" + ) diff --git a/dev/calibrate_ci_minutes_estimates.py b/dev/calibrate_ci_minutes_estimates.py new file mode 100644 index 00000000000..7572dc9323b --- /dev/null +++ b/dev/calibrate_ci_minutes_estimates.py @@ -0,0 +1,251 @@ +#!/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. +""" +Measure the numbers behind ``dev/breeze/src/airflow_breeze/utils/ci_minutes_estimate.py``. + +Downloads recent successful ``Tests (AMD)`` PR runs, their job timings and the selective-check outputs +their ``Build info`` job printed, then prints the medians and fits to copy into the estimate module. +Downloads are cached in ``--cache-dir``, so re-runs only fetch new runs. + +Needs an authenticated ``gh`` CLI. Run from the repository root: + + uv run --project dev/breeze python dev/calibrate_ci_minutes_estimates.py --since 2026-09-10 +""" + +from __future__ import annotations + +import argparse +import json +import re +import statistics +import subprocess +from collections import defaultdict +from concurrent.futures import ThreadPoolExecutor +from datetime import datetime +from pathlib import Path + +REPOSITORY = "apache/airflow" +WORKFLOW = "ci-amd.yml" +ANSI_ESCAPE = re.compile(r"\x1b\[[0-9;]*m") + +# Job-name patterns for each job group of ci_minutes_estimate.py; the first match wins. +JOB_GROUPS = [ + ("docs", r"^CI image checks / (Build documentation|Publish documentation)"), + ("core", r"^(MySQL|Postgres|Sqlite|Non-DB) tests: core|^Low dep tests:core"), + ("special", r"^Special tests"), + ( + "providers", + r"^Provider distributions tests|tests: providers|^Low dep tests: providers|^MyPy providers checks", + ), + ("prod", r"^Build PROD images|^Additional PROD image tests"), + ("kubernetes", r"^Kubernetes tests"), + ("helm", r"^Helm tests"), + ("kustomize", r"^Kustomize"), + ("integration", r"^Integration and System Tests"), + ("task-sdk", r"^Task SDK tests"), + ("agent-framework", r"^Agent framework tests"), + ("java-sdk", r"^Java SDK"), + ("go-sdk", r"^Go SDK"), + ("ts-sdk", r"^TypeScript SDK"), + ("airflow-ctl", r"^Airflow CTL"), + ("base", r".*"), +] + + +def gh_api(path: str, *extra: str) -> str: + for _ in range(3): + result = subprocess.run(["gh", "api", *extra, path], capture_output=True, text=True, check=False) + if result.returncode == 0: + return result.stdout + raise RuntimeError(f"gh api {path} failed: {result.stderr.strip()}") + + +def list_runs(since: str) -> list[int]: + run_ids: list[int] = [] + for page in range(1, 11): + payload = json.loads( + gh_api( + f"repos/{REPOSITORY}/actions/workflows/{WORKFLOW}/runs?event=pull_request" + f"&status=success&per_page=100&page={page}&created=>={since}" + ) + ) + run_ids.extend(run["id"] for run in payload["workflow_runs"]) + if len(payload["workflow_runs"]) < 100: + break + return run_ids + + +def fetch_run(run_id: int, cache_dir: Path) -> None: + jobs_file = cache_dir / f"{run_id}.jobs.json" + log_file = cache_dir / f"{run_id}.build-info.log" + try: + if not jobs_file.exists(): + jobs = json.loads( + gh_api(f"repos/{REPOSITORY}/actions/runs/{run_id}/jobs?per_page=100", "--paginate", "--slurp") + ) + jobs_file.write_text(json.dumps([job for page in jobs for job in page["jobs"]])) + if not log_file.exists(): + build_info = next(job for job in json.loads(jobs_file.read_text()) if job["name"] == "Build info") + log_file.write_text( + gh_api(f"repos/{REPOSITORY}/actions/jobs/{build_info['id']}/logs", "--allow-escape-sequences") + ) + except (RuntimeError, StopIteration) as error: + print(f"skipping run {run_id}: {error}") + + +def is_incomplete_json(value: str) -> bool: + if not value.startswith(("[", "{")): + return False + try: + json.loads(value) + except json.JSONDecodeError: + return True + return False + + +def read_outputs(log_file: Path) -> dict[str, str]: + # Rich wraps long JSON values over several log lines; glue continuation lines onto the value only + # until it parses, so diagnostics printed after it are not appended. + outputs: dict[str, str] = {} + key = None + for line in log_file.read_text(errors="replace").splitlines(): + text = ANSI_ESCAPE.sub("", line).split(" ", 1)[-1] + match = re.match(r"^([a-z0-9-]+) = (.*)$", text) + if match: + key = match.group(1) + outputs[key] = match.group(2) + elif key and text and is_incomplete_json(outputs[key]): + outputs[key] += text + else: + key = None + return outputs + + +def get_test_types(outputs: dict[str, str], name: str) -> list[str]: + value = outputs.get(name) + if not value or value in ("null", "[]"): + return [] + return sorted({test_type for group in json.loads(value) for test_type in group["test_types"].split()}) + + +def count_provider_test_files(provider_id: str) -> int: + tests_path = Path("providers", *provider_id.split("."), "tests") + return sum(1 for _ in tests_path.rglob("test_*.py")) + + +def build_rows(run_ids: list[int], cache_dir: Path) -> list[dict]: + rows = [] + for run_id in run_ids: + jobs_file = cache_dir / f"{run_id}.jobs.json" + log_file = cache_dir / f"{run_id}.build-info.log" + if not jobs_file.exists() or not log_file.exists(): + continue + outputs = read_outputs(log_file) + if "full-tests-needed" not in outputs: + continue + minutes: dict[str, float] = defaultdict(float) + longest: dict[str, float] = defaultdict(float) + for job in json.loads(jobs_file.read_text()): + if job["conclusion"] == "skipped" or not job["started_at"] or not job["completed_at"]: + continue + duration = ( + datetime.fromisoformat(job["completed_at"]) - datetime.fromisoformat(job["started_at"]) + ).total_seconds() / 60 + group = next(group for group, pattern in JOB_GROUPS if re.search(pattern, job["name"])) + minutes[group] += duration + longest[group] = max(longest[group], duration) + try: + providers = get_test_types(outputs, "individual-providers-test-types-list-as-strings-in-json") + core_types = len(get_test_types(outputs, "core-test-types-list-as-strings-in-json")) + except json.JSONDecodeError: + print(f"skipping run {run_id}: its selective-check outputs could not be parsed") + continue + rows.append( + { + "outputs": outputs, + "minutes": minutes, + "longest": longest, + "core_types": core_types, + "providers": [p[len("Providers[") : -1] for p in providers if p.startswith("Providers[")], + } + ) + return rows + + +def print_report(rows: list[dict]) -> None: + print(f"\nruns analysed: {len(rows)}") + print("\njob group runs median job-minutes median longest job") + for group, _ in JOB_GROUPS: + present = [row for row in rows if row["minutes"].get(group)] + if present: + print( + f"{group:18s}{len(present):5d} {statistics.median(r['minutes'][group] for r in present):18.1f}" + f" {statistics.median(r['longest'][group] for r in present):18.1f}" + ) + print("\ncore test types runs median job-minutes median longest job") + core = [row for row in rows if row["core_types"] and row["minutes"].get("core")] + for count in sorted({row["core_types"] for row in core}): + selected = [row for row in core if row["core_types"] == count] + print( + f"{count:15d}{len(selected):7d} {statistics.median(r['minutes']['core'] for r in selected):18.1f}" + f" {statistics.median(r['longest']['core'] for r in selected):18.1f}" + ) + tested = [row for row in rows if row["providers"] and row["minutes"].get("providers")] + partial = [row for row in tested if len(row["providers"]) < 50] + full = [row for row in tested if len(row["providers"]) >= 50] + if full: + print( + f"\nall providers: {len(full)} runs, median job-minutes " + f"{statistics.median(r['minutes']['providers'] for r in full):.1f}, median longest job " + f"{statistics.median(r['longest']['providers'] for r in full):.1f}" + ) + test_files = [sum(count_provider_test_files(p) for p in row["providers"]) for row in partial] + if len(set(test_files)) < 2: + print("partial provider runs: not enough different provider test-suite sizes to fit") + else: + slope, intercept = statistics.linear_regression( + test_files, [r["minutes"]["providers"] for r in partial] + ) + longest_slope, longest_intercept = statistics.linear_regression( + test_files, [r["longest"]["providers"] for r in partial] + ) + print( + f"partial provider runs: {len(partial)} runs, " + f"{len({tuple(row['providers']) for row in partial})} distinct provider sets" + ) + print(f" job-minutes ~= {intercept:.1f} + {slope:.3f} * provider test files") + print(f" longest job ~= {longest_intercept:.1f} + {longest_slope:.3f} * provider test files") + + +def main() -> None: + parser = argparse.ArgumentParser( + description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter + ) + parser.add_argument("--since", required=True, help="Earliest run creation date (YYYY-MM-DD).") + parser.add_argument("--cache-dir", type=Path, default=Path("files/ci-minutes-cache")) + args = parser.parse_args() + args.cache_dir.mkdir(parents=True, exist_ok=True) + run_ids = list_runs(args.since) + print(f"runs listed: {len(run_ids)}") + with ThreadPoolExecutor(max_workers=4) as executor: + list(executor.map(lambda run_id: fetch_run(run_id, args.cache_dir), run_ids)) + print_report(build_rows(run_ids, args.cache_dir)) + + +if __name__ == "__main__": + main()
