diff --git a/.github/fork-ci-gate.toml b/.github/fork-ci-gate.toml index 6fc7a0de1ce3c..f7a2a608fdcc9 100644 --- a/.github/fork-ci-gate.toml +++ b/.github/fork-ci-gate.toml @@ -16,6 +16,11 @@ # under the License. # Fork CI gate (AIP-120). See dev/breeze/doc/ci/08_fork_ci_for_external_contributors.md +# +# This file is part of the gate. A pull request whose CI runs in the contributor's fork may only +# add, change or remove its author's own [pilot] entry here; the fork CI monitor closes it if it +# changes anything else, because that change would decide where its own CI runs. See +# "Gate protection" in the document above. # Master switch. When false the gate job lets every pull request run project CI, except # pull requests labelled `use fork ci` and pull requests of [pilot] participants. Push diff --git a/.github/workflows/auto-approve-workflow-runs.yml b/.github/workflows/auto-approve-workflow-runs.yml index 61d86be7a64fb..c6885dccb2b50 100644 --- a/.github/workflows/auto-approve-workflow-runs.yml +++ b/.github/workflows/auto-approve-workflow-runs.yml @@ -19,7 +19,8 @@ # 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". A PR whose CI runs # in the contributor's fork (AIP-120) is approved unless it changes CI definition files, and then gets -# a comment saying why its runs wait. The rules are in +# a comment saying why its runs wait. PRs from the repositories of trusted stakeholders (Amazon, Google, +# Astronomer) are always approved. The rules are in # dev/breeze/src/airflow_breeze/utils/workflow_run_approval.py and documented in # contributing-docs/05_pull_requests.rst. # diff --git a/.github/workflows/basic-tests.yml b/.github/workflows/basic-tests.yml index 944315d8e8350..9d3a9046e5695 100644 --- a/.github/workflows/basic-tests.yml +++ b/.github/workflows/basic-tests.yml @@ -332,6 +332,17 @@ jobs: GITHUB_TOKEN: ${{ github.token }} SKIP: ${{ inputs.skip-prek-hooks }} COLUMNS: "202" + - name: "Fail if static checks modified tracked files" + # prek does not report files that a nested-project hook modifies outside its own project, + # so check the tree independently. + shell: bash + run: | + if ! git diff --quiet; then + echo "::error::Static checks passed but modified tracked files. Commit the changes below." + git diff --stat + git diff | head -n 500 || true + exit 1 + fi test-git-clone-on-windows: timeout-minutes: 5 diff --git a/.github/workflows/ci-amd.yml b/.github/workflows/ci-amd.yml index e56377e053bdb..f709ee36fe278 100644 --- a/.github/workflows/ci-amd.yml +++ b/.github/workflows/ci-amd.yml @@ -164,7 +164,9 @@ jobs: gate: # Fork CI gate (AIP-120): decides whether project CI runs for this event. Configuration and # code come from the base branch so a pull request cannot change its own eligibility. See - # dev/breeze/doc/ci/08_fork_ci_for_external_contributors.md. + # dev/breeze/doc/ci/08_fork_ci_for_external_contributors.md. A pull request whose CI runs + # in the contributor's fork and changes this job, or how build-info depends on it, is closed + # by the fork CI monitor ("Gate protection" in that document). name: "Fork CI gate" runs-on: ["ubuntu-slim"] timeout-minutes: 5 diff --git a/.github/workflows/ci-image-checks.yml b/.github/workflows/ci-image-checks.yml index cca6c1826b47c..ce2e8c41b33f5 100644 --- a/.github/workflows/ci-image-checks.yml +++ b/.github/workflows/ci-image-checks.yml @@ -162,6 +162,17 @@ jobs: PR_NUMBER: ${{ inputs.newsfragment-pr-number }} DEFAULT_BRANCH: ${{ inputs.branch }} RUFF_FORMAT: "github" + - name: "Fail if static checks modified tracked files" + # prek does not report files that a nested-project hook modifies outside its own project + # (e.g. `breeze k8s setup-env` relocking uv.lock), so check the tree independently. + shell: bash + run: | + if ! git diff --quiet; then + echo "::error::Static checks passed but modified tracked files. Commit the changes below." + git diff --stat + git diff | head -n 500 || true + exit 1 + fi - name: "Show prek log on failure" run: cat ~/.cache/prek/prek.log || true if: failure() diff --git a/.github/workflows/finalize-tests.yml b/.github/workflows/finalize-tests.yml index b664e49cf06ed..c0a04fafe5ff6 100644 --- a/.github/workflows/finalize-tests.yml +++ b/.github/workflows/finalize-tests.yml @@ -111,10 +111,12 @@ jobs: - name: Checkout ${{ steps.constraints-branch.outputs.branch }} uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 with: + # Forks do not carry the constraints branches - diff against the canonical ones + repository: apache/airflow path: "constraints" ref: ${{ steps.constraints-branch.outputs.branch }} - persist-credentials: true - fetch-depth: 0 + persist-credentials: false + fetch-depth: 1 - name: "Download constraints from the constraints generated by build CI image" uses: actions/download-artifact@9000827ccba6bdab643e8b6fd33ac0654aef8333 # v8.0.2 with: diff --git a/.github/workflows/fork-ci-monitor.yml b/.github/workflows/fork-ci-monitor.yml index ec34b9615b028..e95becc9e273f 100644 --- a/.github/workflows/fork-ci-monitor.yml +++ b/.github/workflows/fork-ci-monitor.yml @@ -21,6 +21,9 @@ # those pull requests, and deletes their project CI runs that only skipped every job. It never checks # out or runs pull request code. See # dev/breeze/doc/ci/08_fork_ci_for_external_contributors.md. +# +# Part of the fork CI gate: a pull request whose CI runs in the contributor's fork and changes +# this file is closed by this monitor ("Gate protection" in the document above). name: "Fork CI monitor" on: # yamllint disable-line rule:truthy schedule: diff --git a/contributing-docs/05_pull_requests.rst b/contributing-docs/05_pull_requests.rst index 405c55da48a23..f3e32816dd73b 100644 --- a/contributing-docs/05_pull_requests.rst +++ b/contributing-docs/05_pull_requests.rst @@ -275,6 +275,10 @@ automatically unless the PR changes files that define CI: anything under ``.gith ``Dockerfile*``, or a ``.pre-commit-config.yaml`` in any directory. Then the runs wait for a maintainer and a comment on the PR explains why. CI in your fork runs either way. +The runs of PRs opened from the repositories of trusted stakeholders (``astronomer/airflow``, +``aws-mwaa/upstream-to-airflow`` and ``VladaZakharova/airflow``) are always approved, so none of +the rules above apply to them. + 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 diff --git a/contributing-docs/18_contribution_workflow.rst b/contributing-docs/18_contribution_workflow.rst index 89a73df97128c..8b647dd16f179 100644 --- a/contributing-docs/18_contribution_workflow.rst +++ b/contributing-docs/18_contribution_workflow.rst @@ -147,15 +147,17 @@ Set up your checkout and fork once: 2. If fork CI applies to you, enable the ``Tests (AMD)`` workflow in your fork: today that means you joined the pilot or a maintainer labelled your pull request ``use fork ci``; once fork CI is switched on for everyone, it means you are not a committer, a collaborator or a member of an - exempted organisation. GitHub disables workflows in new forks. Otherwise keep it disabled, so - your pushes do not run CI twice: + exempted organisation. GitHub disables workflows in new forks. Sync your fork's ``main`` with + ``apache/airflow`` first, so it has the current version of the workflow: .. code-block:: bash + gh repo sync /airflow --branch main gh workflow enable ci-amd.yml -R /airflow - If the workflow is missing, your fork's ``main`` is older than it; sync it first with - ``gh repo sync /airflow --branch main``. + Keep the workflow enabled for as long as you have open pull requests. If it gets disabled, from + your fork's Actions tab or by GitHub after 60 days without activity in your fork, the fork CI + monitor comments on your pull requests with the commands to turn it back on. 3. Check everything at once. ``breeze ci audit`` prints one row per check and the command that fixes anything missing; ``--fix`` runs those commands after asking: @@ -184,7 +186,10 @@ at 00:00 UTC: "your-login" = YYYY-MM-DDT00:00:00Z Every pull request you open after that instant uses fork CI; your earlier pull requests keep project -CI. To leave the pilot, open a pull request that removes the line. A pilot entry gates your pull +CI. To leave the pilot, open a pull request that removes the line. Your own ``[pilot]`` line is +the only part of the fork CI gate a pull request using fork CI may change: one that changes the +gate's code, the rest of its configuration or the ``gate`` job of ``ci-amd.yml`` is closed with an +explanation. A pilot entry gates your pull requests whatever your role, so a committer in the pilot also enables the workflow; ``breeze ci audit`` tells you which applies to you. The design is described in `Fork CI for external contributors <../dev/breeze/doc/ci/08_fork_ci_for_external_contributors.md>`__. diff --git a/contributing-docs/25_maintainer_pr_triage.md b/contributing-docs/25_maintainer_pr_triage.md index e3f6909194c94..e340f98275e17 100644 --- a/contributing-docs/25_maintainer_pr_triage.md +++ b/contributing-docs/25_maintainer_pr_triage.md @@ -250,7 +250,7 @@ The triage workflow uses the following labels and states: ## Pull requests gated to fork CI Pull requests labelled `fork ci required` run their CI in the contributor's fork. The scheduled -fork CI reconciler (`breeze ci fork-ci-reconcile`, run by `fork-ci-monitor.yml`) owns four things +fork CI reconciler (`breeze ci fork-ci-reconcile`, run by `fork-ci-monitor.yml`) owns five things for them, and the triage skill must treat those pull requests as out of scope for each: - **Draft state.** The reconciler keeps the pull request in draft until the fork run for its @@ -261,6 +261,10 @@ for them, and the triage skill must treat those pull requests as out of scope fo - **`needs project ci`.** The reconciler adds it when the fork run is green but the pull request changes CI-defining files, and removes it once the state is anything else. These pull requests stay in draft and wait for a maintainer to review the CI change and add `use project ci`. +- **Closing for changing the gate.** The reconciler closes a pull request that changes the fork + CI gate (its code, configuration other than the author's own `[pilot]` entry, the `gate` job + of `ci-amd.yml` or the monitor), and reports it on Slack. Add `use project ci` to accept such + a change. - **Closing after inactivity.** The reconciler closes a pull request whose fork CI has not run for `close_after_days`. diff --git a/dev/breeze/doc/08_ci_tasks.rst b/dev/breeze/doc/08_ci_tasks.rst index d54eadbaa694d..7147b6150967e 100644 --- a/dev/breeze/doc/08_ci_tasks.rst +++ b/dev/breeze/doc/08_ci_tasks.rst @@ -168,8 +168,8 @@ Pull requests from external contributors can run their CI in the contributor's f ``breeze ci audit`` checks, read-only, whether your checkout and fork are set up for that: ``gh`` installed and authenticated, the ``upstream`` remote pointing at ``apache/airflow`` and ``origin`` at your fork, whether you are a committer, collaborator, member of an exempted organisation, pilot participant or -external contributor, whether the ``Tests (AMD)`` workflow in your fork is enabled (or disabled, for those -whose pull requests run project CI), how far your fork's ``main`` is behind, and the branch you are on. +external contributor, whether the ``Tests (AMD)`` workflow in your fork is enabled (when your pull requests +use fork CI), how far your fork's ``main`` is behind, and the branch you are on. Each failing check prints the command that fixes it; ``--fix`` runs those commands after asking. These are all available flags of ``audit`` command: @@ -207,7 +207,8 @@ the waiting workflow runs of external contributors' PRs to ``main`` whose CI job 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``. PRs whose CI runs in the contributor's fork (AIP-120) only run the fork CI gate in ``apache/airflow``, so their runs are approved whatever their size unless they change CI definition files; such a PR also gets a comment -explaining that its CI change needs a maintainer. For every PR it also +explaining that its CI change needs a maintainer. The runs of PRs from the repositories of trusted +stakeholders (Amazon, Google, Astronomer) are always approved. 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 `_. diff --git a/dev/breeze/doc/ci/05_workflows.md b/dev/breeze/doc/ci/05_workflows.md index 9352647116dce..a98d3a1586144 100644 --- a/dev/breeze/doc/ci/05_workflows.md +++ b/dev/breeze/doc/ci/05_workflows.md @@ -128,6 +128,15 @@ those CI-defining files. Such a PR gets the `need workflow approval` label and o the author that the CI change needs a maintainer; see [Approving gated runs](08_fork_ci_for_external_contributors.md#approving-gated-runs). +The runs of PRs whose head repository belongs to a trusted stakeholder (`astronomer/airflow` for +Astronomer, `aws-mwaa/upstream-to-airflow` for Amazon, `VladaZakharova/airflow` for Google; listed in +`TRUSTED_STAKEHOLDER_REPOSITORIES`) are approved whatever the PR changes or targets: only people with +write access to those repositories can push the PR's commits. The repositories are matched by their +numeric id, so a repository that later takes over one of these names is not trusted. None of the +rules above apply to them, and of the rules below only the 24-hour window does: their runs are +approved even when several open PRs share the head commit, or when the runs started before the PR +was opened, retargeted or relabelled. + 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 diff --git a/dev/breeze/doc/ci/08_fork_ci_for_external_contributors.md b/dev/breeze/doc/ci/08_fork_ci_for_external_contributors.md index ab33c09aec374..10a96244cfee4 100644 --- a/dev/breeze/doc/ci/08_fork_ci_for_external_contributors.md +++ b/dev/breeze/doc/ci/08_fork_ci_for_external_contributors.md @@ -490,16 +490,19 @@ that matters. It is matched by file, not by the workflow's display name, which a change, and only `push` runs count because only they execute the workflow from the head commit itself; a `pull_request` run inside the fork, into the fork's own `main`, runs a merge the contributor controls. For pull requests -whose recorded state is `setup_required` (or that have no recorded state yet), one REST call -per pull request reads `GET /repos//actions/workflows/ci-amd.yml` to distinguish -"workflows disabled" from "enabled but not yet pushed". This is bounded by the number of -pull requests in that state, not by the total. +with no fork run for the head commit, one REST call per pull request reads +`GET /repos//actions/workflows/ci-amd.yml` to distinguish "workflows disabled" from +"enabled but not yet pushed". It is read on every run, including for pull requests already +recorded as `awaiting_push`: a workflow can be disabled after it was enabled, by the +contributor or by GitHub after 60 days without activity in the fork, and the pull request +would otherwise keep waiting for a push that never runs CI. This is bounded by the number of +pull requests without a fork run for their head commit, not by the total. ### State machine | Observed fork state | Draft | `ready for maintainer review` | Commit status | Comment | |---|---|---|---|---| -| **setup_required**: workflow state is `disabled_fork`, or the fork has no `ci-amd.yml` | draft | remove | `error` (red), link to the fork's Actions page for `ci-amd.yml` | create or update: how to enable, with the exact `gh workflow enable ci-amd.yml -R ` line, and `breeze ci audit` | +| **setup_required**: workflow state is `disabled_fork`, `disabled_manually` or `disabled_inactivity`, or the fork has no `ci-amd.yml` | draft | remove | `error` (red), link to the fork's Actions page for `ci-amd.yml` | create or update: how to enable, with the exact `gh repo sync --branch main` and `gh workflow enable ci-amd.yml -R ` lines, and `breeze ci audit`; on transition from any other state but `closed` (the workflow stopped running after it worked): also post a **new** comment saying why it stopped | | **awaiting_push**: enabled, no run for the head SHA | draft | remove | `pending` (yellow), link to the fork's Actions page | on transition from setup_required: post a **new** comment asking to push again (an edit would not notify) | | **running**: run in progress | draft | remove | `pending` (yellow), link to the run | update state and run link | | **failed**: conclusion is failure, cancelled or timed out | draft | remove | `failure` (red), link to the run | update with the run link | @@ -507,6 +510,7 @@ pull requests in that state, not by the total. | **green_unverified**: conclusion success, but the pull request changes CI-defining files (see "Trusting a green fork run") | draft | remove | `pending` (yellow), link to the run | update: a maintainer must run project CI; labelled `needs project ci` | | setup_required or awaiting_push for `close_after_days` | unchanged | unchanged | unchanged | close with a new comment; reopening restarts the clock | | **ungated**: the base branch has no gate, for example after retargeting to an older release branch | undraft\* | unchanged | `success` (green), link to the pull request's checks | update: project CI is in use; `fork ci required` is removed | +| **gate_modified**: the pull request changes the fork CI gate (see "Gate protection") | draft | remove | `error` (red), link to the pull request's changed files | post a **new** comment explaining why, then close; checked after gate tampering, before the fork state | | **gate_bypassed**: a recorded project CI run of the pull request's branch bypassed the gate (see "Gate tampering") | draft | remove | `error` (red), link to the project CI run that should not have happened | post a **new** comment explaining why, then close; takes precedence over every row above | \* only if the last convert-to-draft event on the pull request was by `github-actions[bot]`; @@ -542,6 +546,10 @@ Rules that sit under the table: closing comment or close failed, and the next run finishes the closure. The same holds for a `gate_bypassed` marker, where only a reopen by someone other than the author counts, so the closure is finished even after the offending run has left the tampering check's window. +- **New comments go first.** A nudge, a "fork CI stopped running" notice, a closing explanation + or a tampering notice is posted before the state comment records the new state, so when + posting fails the next run sees the old state and posts it again; a repeat is preferred to a + lost notice. - **`fork ci required` goes last.** When a run removes the label (on release, or for an ungated base branch), it does so after every other change, so a failure on the way leaves the label in place and the next run retries the rest. @@ -569,7 +577,10 @@ The decision is read back from the `Fork CI / Tests (AMD)` commit status on the which only `apache/airflow`'s own tokens can write, so it is made once per commit and cannot be forged through a comment. The changed files depend on the base branch as well, so a decision recorded before the pull request was last retargeted (a `BaseRefChangedEvent` in the timeline -newer than the status) is made again, and the status is written again to date it. +newer than the status) is made again, and the status is written again to date it. A commit status +belongs to the commit, so it cannot hold one decision per base: when another gated pull +request with a different base has the same head commit, both are `green_unverified` and wait +for a maintainer, reported to Slack once. This is a guard against the cheapest way to fake a green run, not a guarantee. Any change can weaken the tests it ships with (a `conftest.py` that skips tests, for example), exactly as it @@ -605,28 +616,36 @@ ignored, because they ran legitimately while the label was on. So are runs creat most recent reopen of the pull request by someone other than its author, which is how a maintainer brings back a pull request closed for tampering; reopening one whose gate is still missing triggers a new run, which is caught again. A reopen by the author does not count, so -closing and reopening one's own pull request cannot hide a tampered run. A run counts as a -bypass when both signals hold: +closing and reopening one's own pull request cannot hide a tampered run. Runs are listed per +fork branch, because GitHub records no pull request on runs from forks. A run made while the +pull request targeted a base branch without the gate, known from its retargets in the +timeline, is project CI by design and ignored; its changes are compared with that base. When +another pull request from the same branch was open within the window, a matching run could be +that one's, so it is reported to Slack as an attention item rather than closing the pull +request. A run counts as a bypass when both signals hold: 1. **The run did project CI work.** Either the default branch's version of the run's workflow is not triggered by the run's event, which covers an injected workflow and a trigger the pull request added to an existing one (for example `pull_request` on `ci-arm.yml`, which - has no gate), or it is a `ci-amd.yml` run that has run for more than 5 minutes, finished or - still running, and whose jobs (`GET .../actions/runs//jobs`) show more than the gate: a - started job that was not skipped, other than `Fork CI gate` and the jobs that run in every - gated run (`Check if the run is needed`, which has no steps for pull requests, and the - `CI passed` job, named `CI passed (project CI not run)` when project CI was skipped and - `CI passed` when a newer push cancelled the run before `Build info` started), a missing or - duplicated gate job, one of those jobs appearing twice (either `CI passed` name counts as - the same job), or one of them running for more than 5 minutes. The last two cases catch - project CI work done in jobs renamed to one of their names. A job counts as started - only when a runner picked it up (a non-zero `runner_id`, or steps): a queued job reports its - creation as `started_at`, and a run cancelled by a newer push marks its never-started jobs - `completed` and `cancelled`, so a gate waiting for a runner would otherwise look like project - CI work. A running run is judged as it goes, so one - tampered to outlast the 6-hour window is still caught. A gate-only run finishes in about a - minute, so jobs are listed only for the rare long runs. The triggers are read from the - monitor's own checkout of `main`, so no API call is needed; a pull request into a release + has no gate), and one of its jobs started and was not skipped (judged as for `ci-amd.yml` + below when it is that workflow, and counted as work when the jobs cannot be listed), so a + fork-only workflow the pull request adds and guards off in `apache/airflow` does not count; + or it is a `ci-amd.yml` run that has run for more than 5 minutes, finished or still running, + and whose jobs (`GET .../actions/runs//jobs`) show more than the gate: a started job that + was not skipped, other than `Fork CI gate` and the jobs that run in every gated run + (`Check if the run is needed`, which has no steps for pull requests, and the `CI passed` job, + named `CI passed (project CI not run)` when project CI was skipped and, when a newer push + cancelled the run before `Build info` started, `CI passed (run cancelled)` -- or `CI passed` + from a `ci-amd.yml` that does not use that name yet), a missing or duplicated gate job, one + of those jobs appearing twice (every `CI passed` name counts as the same job), or one of them + running for more than 5 minutes. The last two cases catch project CI work done in jobs + renamed to one of their names. A job counts as started only when a runner picked it up (a + non-zero `runner_id`, or steps): a queued job reports its creation as `started_at`, and a run + cancelled by a newer push marks its never-started jobs `completed` and `cancelled`, so a gate + waiting for a runner would otherwise look like project CI work. A running run is judged as it + goes, so one tampered to outlast the 6-hour window is still caught. A gate-only run finishes + in about a minute, so jobs are listed only for the rare long runs. The triggers are read from + the monitor's own checkout of `main`, so no API call is needed; a pull request into a release branch is judged by `main`'s workflows. 2. **The run's own commit changes workflow files.** The compare of the base branch with that commit (`GET /repos/apache/airflow/compare/...:`, first page only: @@ -665,6 +684,46 @@ signal 1 no longer holds and the pull request re-enters the fork state machine. ci` releases it like any other gated pull request; a maintainer who wants to keep an external workflow change has to leave that label on. +### Gate protection + +The gate decides where a pull request's CI runs, so a pull request whose CI runs in a fork must +not change it: the change would decide where its own CI runs, and a maintainer reviewing it later +would be reviewing a pull request that already escaped the gate. For every pull request the +tampering check applies to (`external` and `pilot`), the reconciler lists the changed files once +per head commit, counting a renamed file under both paths, and closes the pull request +with a comment explaining why when it changes any of: + +- `dev/breeze/src/airflow_breeze/utils/fork_ci_gate.py`, the reconciler modules + `fork_ci_reconcile.py`, `fork_ci_github.py` and `fork_ci_messages.py`, or + `.github/workflows/fork-ci-monitor.yml`; +- `.github/fork-ci-gate.toml`, other than adding, changing or removing the author's own + `[pilot]` entry, which is how contributors join and leave the pilot (the configuration is + parsed from the head commit and compared with its version at the pull request's merge base, + so comments and later changes on the base branch do not count); +- the `gate` job of `.github/workflows/ci-amd.yml`, the `needs` and `if` that make + `build-info` depend on it, the set of jobs that do not depend on the gate (directly or through + other jobs), or the workflow-level `on`, `defaults`, `env`, `permissions` and `concurrency` (the head and merge-base versions are parsed and those parts + compared, so other changes to the workflow are left to the green-trust check; a head version + that does not parse as a workflow counts as a change). + +A changed-file listing at GitHub's 3000-file cap may leave out a gate file, so such a pull +request is closed as one whose gate change cannot be checked. + +The state comment's marker records `"gate_checked": true` with the head commit only when this +check ran for it, so a marker written before the check existed, or while `use fork ci` exempted +the pull request, does not count, and a retarget since the state began makes it check again. +The changed files are listed for the pull request as it is when listed, so the head commit is +read again afterwards: if it moved, nothing is recorded and the next run checks the new one. + +The state is `gate_modified`: the pull request is drafted, loses `ready for maintainer review` +and `needs project ci`, gets an `error` status linking its changed files, and the closure is +reported to Slack as an attention item. Gate tampering is checked first and wins. As for gate +tampering, only a reopen by someone other than the author counts: an open pull request whose +marker says `gate_modified` and that was not reopened since is closed again, and a maintainer's +reopen without restoring the gate closes it again on the next run. A maintainer who wants the +change adds `use project ci`, which releases the pull request before any of these checks. Each +of these files carries a comment saying so. + ### State comment One comment per pull request, created on first contact and edited afterwards. It starts with @@ -872,7 +931,7 @@ Unknown means any of: that outlasted its retries. - A check suite for `Tests (AMD)`, in the fork or in `apache/airflow`, with a status or conclusion outside the modelled set. -- A workflow state outside `active` / `disabled_fork` / `disabled_manually` / not found. +- A workflow state outside `active` / `disabled_fork` / `disabled_manually` / `disabled_inactivity` / not found. - A mutation that returned an error (label, comment, close, status). - A head repository that no longer exists. - A state comment whose marker cannot be parsed. @@ -949,18 +1008,17 @@ prints one row: status, finding, and the command that fixes it. | 2 | Remote naming | a remote named `upstream` points at `apache/airflow` and `origin` does not | `git remote rename`/`add` per `contributing-docs/10_working_with_git.rst` | | 3 | Fork exists | `origin` is a fork of `apache/airflow` owned by the authenticated user | `gh repo fork apache/airflow --remote=false` then add `origin`, unless check 2's fix already adds or repoints it | | 4 | Who you are | classifies the user in the gate's order: pilot participant (`[pilot]`, which gates whatever the role), committer (public `apache` membership, `GET /orgs/apache/public_members/`, since the gate cannot see private membership), collaborator (`.asf.yaml`), exempted (`[allowlist]`), external; an external user whose `apache` membership is private is told to make it public or ask for `[allowlist]`; when the membership call fails for any reason but a 404, the role is reported as not checked and checks 2, 3 and 5 stay informational | informational | -| 5 | Fork workflows | external or pilot participant: `ci-amd.yml` state on the fork is `active`; committer/collaborator/exempted: informational, with a hint when it is `active` | `gh workflow enable ci-amd.yml -R `, or `gh workflow disable ...` shown but not run by `--fix` | +| 5 | Fork workflows | external or pilot participant: `ci-amd.yml` state on the fork is `active`; committer/collaborator/exempted: always OK, whatever the state | `gh workflow enable ci-amd.yml -R ` | | 6 | Fork `main` freshness | fork `main` is not behind `upstream/main` by more than 200 commits (`MAX_FORK_MAIN_BEHIND`) | `gh repo sync --branch main` | | 7 | Current branch | not `main`, has an upstream tracking `origin` | informational | The role is determined before checks 2 and 3, though printed as check 4. For a committer, collaborator or exempted contributor, whose pull requests run project CI, checks 2 and 3 are informational: working from an `apache/airflow` clone without a fork is fine for them, and -`--fix` only runs the fixes of failed checks. The same goes for check 5: an enabled fork -workflow runs every push twice and attaches a second status, so they are told they can disable -it, but it is not a failure, because a committer trying fork CI with the `use fork ci` label -needs it enabled. Forks of a public repository cannot be made private on GitHub, so disabling -the workflow is the only recommendation available. +`--fix` only runs the fixes of failed checks. Check 5 passes for them whether the fork workflow +is enabled or not. The audit never suggests disabling it: a committer trying fork CI with the +`use fork ci` label needs it enabled, and a workflow disabled while a pull request waits for its +fork CI leaves that pull request stuck until the workflow is enabled again. A failed API call is reported as "could not check" rather than as a finding: a failed fork lookup is not "not a fork of apache/airflow", and a failed workflow lookup is not "missing". @@ -1089,7 +1147,8 @@ as a failed check rather than a crash. CI-defining files keeps the pull request a draft, decided once per commit) and the gate tampering check: a tampered run on an earlier commit is caught after a clean push, each signal alone does not close, short runs and gate-only jobs do not count, an in-progress - non-gate job, a removed, duplicated or long-running gate job and an injected workflow do, + non-gate job, a removed, duplicated or long-running gate job and an injected workflow with a + job that ran do, an added workflow whose jobs all skipped does not, runs before the last `use project ci` removal or the last reopen are ignored, an uncheckable 300-file change set is reported, `label_fork_ci` pull requests are exempt and `pilot` pull requests are checked. Discovery labels a pilot participant's diff --git a/dev/breeze/src/airflow_breeze/commands/kubernetes_commands.py b/dev/breeze/src/airflow_breeze/commands/kubernetes_commands.py index 772eab0ee04f1..c980fcd1d1360 100644 --- a/dev/breeze/src/airflow_breeze/commands/kubernetes_commands.py +++ b/dev/breeze/src/airflow_breeze/commands/kubernetes_commands.py @@ -2035,7 +2035,7 @@ def _run_tests( f"[info]You can deploy airflow with {executor} by running:[/]\nbreeze k8s configure-cluster\nbreeze k8s deploy-airflow --multi-namespace-mode --executor {executor}" ) return 1, f"Tests {kubectl_cluster_name}" - pytest_cmd = ["uv", "run", "pytest"] + pytest_cmd = ["uv", "run", "--frozen", "pytest"] the_tests: list[str] = ["tests"] ordered_unique_args = dict.fromkeys(chain(pytest_cmd, the_tests, test_args)) command_to_run = " ".join(quote(arg) for arg in ordered_unique_args) diff --git a/dev/breeze/src/airflow_breeze/commands/kubernetes_kustomize_commands.py b/dev/breeze/src/airflow_breeze/commands/kubernetes_kustomize_commands.py index d7aa3b2486229..fe1da3f0689ac 100644 --- a/dev/breeze/src/airflow_breeze/commands/kubernetes_kustomize_commands.py +++ b/dev/breeze/src/airflow_breeze/commands/kubernetes_kustomize_commands.py @@ -488,7 +488,7 @@ def _run_overlay_pytest( env["OVERLAY_UNDER_TEST"] = overlay_name env["OVERLAY_NAMESPACE"] = namespace env["OVERLAY_RELEASE_NAME"] = release_name - pytest_cmd = ["uv", "run", "pytest", str(test_file.relative_to(CHART_PATH)), "-xvs"] + pytest_cmd = ["uv", "run", "--frozen", "pytest", str(test_file.relative_to(CHART_PATH)), "-xvs"] console_print(f"[info]Running behavioural tests: {' '.join(pytest_cmd)}") result = run_command(pytest_cmd, env=env, check=False, cwd=CHART_PATH.as_posix()) return result.returncode diff --git a/dev/breeze/src/airflow_breeze/commands/verify_commands.py b/dev/breeze/src/airflow_breeze/commands/verify_commands.py index 0dc8555a8ccea..7233e202defb4 100644 --- a/dev/breeze/src/airflow_breeze/commands/verify_commands.py +++ b/dev/breeze/src/airflow_breeze/commands/verify_commands.py @@ -42,7 +42,15 @@ def _run_git(*args: str) -> CompletedProcess: - return run_command(["git", *args], capture_output=True, text=True, check=False, cwd=AIRFLOW_ROOT_PATH) + # Read-only queries: dry-run would skip them and return empty output. + return run_command( + ["git", *args], + capture_output=True, + text=True, + check=False, + cwd=AIRFLOW_ROOT_PATH, + dry_run_override=False, + ) def find_default_base_ref() -> str | None: diff --git a/dev/breeze/src/airflow_breeze/utils/ci_audit.py b/dev/breeze/src/airflow_breeze/utils/ci_audit.py index 3b27ef8eb15ed..94b0cf7ce9d7b 100644 --- a/dev/breeze/src/airflow_breeze/utils/ci_audit.py +++ b/dev/breeze/src/airflow_breeze/utils/ci_audit.py @@ -327,18 +327,6 @@ def run_audit(*, run: Runner, config: GateConfig, collaborators: frozenset[str]) enable, ) ) - elif required is False and state == "active": - # Not a failure: a committer trying fork CI with the `use fork ci` label needs it on. - results.append( - CheckResult( - 5, - "fork workflows", - CheckStatus.INFO, - f"`Tests (AMD)` is enabled in {fork}, but your pull requests run project CI, so each " - "push runs CI twice; disable it unless you use the `use fork ci` label", - (f"gh workflow disable {CI_WORKFLOW_FILE} -R {fork}",), - ) - ) elif required is None: results.append( CheckResult( diff --git a/dev/breeze/src/airflow_breeze/utils/fork_ci_gate.py b/dev/breeze/src/airflow_breeze/utils/fork_ci_gate.py index 10fcc9a106a86..33c5dd9754300 100644 --- a/dev/breeze/src/airflow_breeze/utils/fork_ci_gate.py +++ b/dev/breeze/src/airflow_breeze/utils/fork_ci_gate.py @@ -22,6 +22,11 @@ The gate job in ``ci-amd.yml`` runs this file with the runner's ``python3`` from a sparse checkout, without installing breeze, so it must only import from the standard library. + +This file is part of the fork CI gate. A pull request whose CI runs in the contributor's fork +must not change it, because the change would decide where its own CI runs: the fork CI +monitor closes such a pull request with a comment explaining why. See "Gate protection" in +the document above. """ from __future__ import annotations diff --git a/dev/breeze/src/airflow_breeze/utils/fork_ci_github.py b/dev/breeze/src/airflow_breeze/utils/fork_ci_github.py index df00485a2b156..77a7ef9101180 100644 --- a/dev/breeze/src/airflow_breeze/utils/fork_ci_github.py +++ b/dev/breeze/src/airflow_breeze/utils/fork_ci_github.py @@ -14,7 +14,13 @@ # KIND, either express or implied. See the License for the # specific language governing permissions and limitations # under the License. -"""GitHub access for the fork CI reconciler (AIP-120); the only module of it that calls ``gh``.""" +""" +GitHub access for the fork CI reconciler (AIP-120); the only module of it that calls ``gh``. + +Part of the fork CI gate: a pull request whose CI runs in the contributor's fork and changes +this file is closed by the fork CI monitor. See "Gate protection" in +dev/breeze/doc/ci/08_fork_ci_for_external_contributors.md. +""" from __future__ import annotations @@ -33,8 +39,11 @@ RATE_LIMIT_MARKERS = ("rate limit", "secondary rate", "abuse detection") MAX_ATTEMPTS = 4 # GitHub answers with these while overloaded, and GraphQL searches that take too long end in HTTP 504; -# the same call usually succeeds shortly after. -SERVER_ERROR = re.compile(r"\bHTTP 5\d\d\b") +# the same call usually succeeds shortly after. A connection dropped mid-request is as transient, and as +# unclear about whether the request took effect. +SERVER_ERROR = re.compile(r"\bHTTP 5\d\d\b|connection reset by peer|unexpected EOF") +# Network failures before the request reached GitHub, so repeating it is safe for any call. +CONNECTION_ERROR = re.compile(r"TLS handshake timeout|dial tcp") SERVER_ERROR_WAIT_SECONDS = 10 # A hung `gh` would otherwise hold the job until its timeout-minutes. GH_TIMEOUT_SECONDS = 120 @@ -64,7 +73,8 @@ def _gh( graphql`` exits non-zero when some fields of the response errored, with the data GitHub could resolve still on stdout. :param retry_server_errors: repeat a call that failed with a server error. Off for a call that - may have taken effect anyway and must not be repeated, such as creating a comment. + may have taken effect anyway and must not be repeated, such as creating a comment. A call + that failed to connect is repeated either way. """ payload = json.dumps(body) if body is not None else None rate_limited = False @@ -84,10 +94,21 @@ def _gh( if result.returncode == 0: return json.loads(result.stdout) if result.stdout.strip() else None rate_limited = any(marker in result.stderr.lower() for marker in RATE_LIMIT_MARKERS) - if rate_limited or (retry_server_errors and SERVER_ERROR.search(result.stderr)): + connection_error = bool(CONNECTION_ERROR.search(result.stderr)) + if ( + rate_limited + or connection_error + or (retry_server_errors and SERVER_ERROR.search(result.stderr)) + ): if attempt == MAX_ATTEMPTS - 1: break - reason = "rate limit" if rate_limited else "server error" + reason = ( + "rate limit" + if rate_limited + else "connection error" + if connection_error + else "server error" + ) wait = (30 if rate_limited else SERVER_ERROR_WAIT_SECONDS) * 2**attempt console_print(f"[warning]GitHub {reason}, retrying in {wait}s[/]") time.sleep(wait) diff --git a/dev/breeze/src/airflow_breeze/utils/fork_ci_messages.py b/dev/breeze/src/airflow_breeze/utils/fork_ci_messages.py index 48208ae90a1e4..d19162d448f64 100644 --- a/dev/breeze/src/airflow_breeze/utils/fork_ci_messages.py +++ b/dev/breeze/src/airflow_breeze/utils/fork_ci_messages.py @@ -20,6 +20,10 @@ All wording lives here so it can be reviewed in one place. Texts must say that CI is free for the contributor and finite for the project, must not characterise the recipient, and must give the exact command to run. + +Part of the fork CI gate: a pull request whose CI runs in the contributor's fork and changes +this file is closed by the fork CI monitor. See "Gate protection" in +dev/breeze/doc/ci/08_fork_ci_for_external_contributors.md. """ from __future__ import annotations @@ -80,6 +84,9 @@ class Marker: behind: int | None = None # The completed fork run (``::``) whose jobs are all reported. jobs: str | None = None + # Set only by a reconciler that ran the gate protection check for this head commit, so a + # marker written before that check existed, or while it did not apply, never counts as one. + gate_checked: bool = False def render_marker(marker: Marker) -> str: @@ -88,6 +95,8 @@ def render_marker(marker: Marker) -> str: data["behind"] = marker.behind if marker.jobs is not None: data["jobs"] = marker.jobs + if marker.gate_checked: + data["gate_checked"] = True return f"{MARKER_PREFIX}{json.dumps(data, sort_keys=True)} -->" @@ -104,6 +113,7 @@ def parse_marker(body: str) -> Marker | None: since=datetime.fromisoformat(data["since"]), behind=data.get("behind"), jobs=data.get("jobs"), + gate_checked=data.get("gate_checked") is True, ) except (json.JSONDecodeError, KeyError, TypeError, ValueError) as e: raise ValueError(f"Malformed fork CI marker: {match.group(0)!r}") from e @@ -116,11 +126,12 @@ def fork_actions_url(fork: str) -> str: def _setup_instructions(fork: str) -> str: return ( "To get CI for this pull request:\n\n" - "1. Enable the workflow in your fork (once per fork):\n\n" + "1. Sync your fork's `main` with apache/airflow, so that it has the current version of the " + "workflow:\n\n" + f" ```bash\n gh repo sync {fork} --branch main\n ```\n\n" + "2. Enable the workflow in your fork:\n\n" f" ```bash\n gh workflow enable ci-amd.yml -R {fork}\n ```\n\n" - " If your fork's `main` is older than the workflow, sync it first with " - f"`gh repo sync {fork} --branch main`.\n" - "2. Push your branch again. Every push then runs CI in your fork.\n\n" + "3. Push your branch again. Every push then runs CI in your fork.\n\n" "`breeze ci audit` checks your whole local and fork setup and prints the command that fixes " f"anything missing. See [the contribution workflow]({CONTRIBUTING_URL})." ) @@ -144,6 +155,7 @@ def _setup_instructions(fork: str) -> str: "comment below for how to continue.", "gate_bypassed": "**Fork CI: closed.** Project CI ran for this pull request although its CI runs in " "the contributor's fork; see the comment below.", + "gate_modified": "**Fork CI: closed.** This pull request changes the fork CI gate; see the comment below.", } @@ -180,6 +192,20 @@ def released_comment(marker: Marker) -> str: ) +def gate_modified_close_comment(*, change: str | None) -> str: + return ( + f"Closing this pull request because it changes the fork CI gate{f': {change}' if change else ''}.\n\n" + "The gate decides whether a pull request runs its CI in apache/airflow or in the contributor's " + "fork. It is made of the `gate` job of `.github/workflows/ci-amd.yml`, " + "`dev/breeze/src/airflow_breeze/utils/fork_ci_gate.py`, `.github/fork-ci-gate.toml` and the fork CI " + "monitor, and a pull request whose CI runs in a fork must not change it, because the change would " + "decide where its own CI runs. The one exception is adding, changing or removing your own `[pilot]` " + "entry in `.github/fork-ci-gate.toml`.\n\n" + "If the change is wanted, a maintainer can reopen the pull request and add the `use project ci` " + "label." + FOOTER + ) + + def ungated_comment(marker: Marker, *, base_ref: str) -> str: return ( f"{render_marker(marker)}\n\n" @@ -195,6 +221,23 @@ def nudge_comment(*, fork: str) -> str: ) +_DISABLED_REASONS = { + "disabled_manually": "was disabled in your fork, from its Actions tab or with `gh workflow disable`", + "disabled_inactivity": "was disabled by GitHub because your fork had no activity for 60 days", + "disabled_fork": "is disabled in your fork", + "missing": "is missing from your fork, for example because its `main` was reset to an older commit", +} + + +def workflow_disabled_comment(*, fork: str, workflow_state: str | None) -> str: + reason = _DISABLED_REASONS.get(workflow_state or "missing", "is not enabled in your fork") + return ( + f"**Fork CI stopped running.** The `Tests (AMD)` workflow {reason}, so pushes to this " + "branch no longer run CI, and this pull request stays a draft until they do.\n\n" + f"{_setup_instructions(fork)}" + FOOTER + ) + + def inactivity_close_comment(*, days: int, fork: str) -> str: return ( f"Closing this pull request because CI has not run in your fork for {days} days. {WHY_FORK_CI}\n\n" @@ -251,6 +294,7 @@ def ci_change_approval_comment(*, ci_files: list[str]) -> str: "gate_bypassed": "Project CI ran for a gated pull request - see Details", "released": "Project CI in use (label `use project ci`) - see the checks", "ungated": "Project CI in use (base branch without fork CI) - see the checks", + "gate_modified": "Pull request changes the fork CI gate - see the comment", } # (commit status state, description) of a fork job, by its status and conclusion. A queued job is @@ -290,6 +334,7 @@ def job_status_context(job_name: str) -> str: "gate_bypassed": "error", "released": "success", "ungated": "success", + "gate_modified": "error", } diff --git a/dev/breeze/src/airflow_breeze/utils/fork_ci_reconcile.py b/dev/breeze/src/airflow_breeze/utils/fork_ci_reconcile.py index 6b54f6c7fac26..79b5035ac7c3c 100644 --- a/dev/breeze/src/airflow_breeze/utils/fork_ci_reconcile.py +++ b/dev/breeze/src/airflow_breeze/utils/fork_ci_reconcile.py @@ -20,6 +20,9 @@ ``classify`` is a pure function from what GitHub recorded about one pull request to the mutations that bring it to the desired state; ``run_reconcile`` gathers that data and applies the result. See "Reconciler" in dev/breeze/doc/ci/08_fork_ci_for_external_contributors.md. + +Part of the fork CI gate: a pull request whose CI runs in the contributor's fork and changes +this file is closed by the fork CI monitor, see "Gate protection" in the same document. """ from __future__ import annotations @@ -64,6 +67,7 @@ STATUS_STATES, Marker, fork_actions_url, + gate_modified_close_comment, inactivity_close_comment, job_status_context, nudge_comment, @@ -73,12 +77,26 @@ status_description, tamper_close_comment, ungated_comment, + workflow_disabled_comment, ) CI_WORKFLOW_FILE = "ci-amd.yml" # The files the gate job of ci-amd.yml reads from a pull request's base branch. GATE_CONFIG_PATH = ".github/fork-ci-gate.toml" GATE_SCRIPT_PATH = "dev/breeze/src/airflow_breeze/utils/fork_ci_gate.py" +CI_WORKFLOW_PATH = f".github/workflows/{CI_WORKFLOW_FILE}" +# Files of the fork CI gate that a pull request whose CI runs in a fork must not change at all. +# The gate configuration and the gate job in ci-amd.yml are checked separately, since the first +# may carry the author's own pilot entry and the second shares its file with the CI pipeline. +GATE_PROTECTED_PATHS = frozenset( + { + GATE_SCRIPT_PATH, + "dev/breeze/src/airflow_breeze/utils/fork_ci_github.py", + "dev/breeze/src/airflow_breeze/utils/fork_ci_messages.py", + "dev/breeze/src/airflow_breeze/utils/fork_ci_reconcile.py", + ".github/workflows/fork-ci-monitor.yml", + } +) GATE_JOB_NAME = "Fork CI gate" # Jobs that also run, without needing the gate, in a run the gate left to fork CI: main's # check-run-needed (no steps outside schedule and push runs) and the "CI passed" gate job of the @@ -144,7 +162,7 @@ RUNNING_STATUSES = frozenset({"QUEUED", "IN_PROGRESS", "PENDING", "WAITING", "REQUESTED"}) FAILED_CONCLUSIONS = frozenset({"FAILURE", "CANCELLED", "TIMED_OUT", "STARTUP_FAILURE", "ACTION_REQUIRED"}) -SETUP_WORKFLOW_STATES = frozenset({"disabled_fork", "disabled_manually", "missing"}) +SETUP_WORKFLOW_STATES = frozenset({"disabled_fork", "disabled_inactivity", "disabled_manually", "missing"}) class ForkState(str, Enum): @@ -158,6 +176,7 @@ class ForkState(str, Enum): CLOSED = "closed" RELEASED = "released" UNGATED = "ungated" + GATE_MODIFIED = "gate_modified" CLOSABLE_STATES = frozenset({ForkState.SETUP_REQUIRED, ForkState.AWAITING_PUSH}) @@ -211,6 +230,8 @@ class PullRequestSnapshot: last_reopened_at: datetime | None = None last_reopened_by_anyone_at: datetime | None = None last_base_change_at: datetime | None = None + # (changed at, previous base, new base) for each retarget, oldest first. + base_changes: tuple[tuple[datetime, str, str], ...] = () current_status: tuple[str, str, str] | None = None status_created_at: datetime | None = None ci_passed_state: str | None = None @@ -284,8 +305,6 @@ def observe_fork_state(fork: ForkObservation, marker: Marker | None) -> tuple[Fo return ForkState.FAILED, None return None, f"unexpected fork check suite conclusion {fork.suite_conclusion!r}" if fork.workflow_state is None: - if marker is not None and marker.state == ForkState.AWAITING_PUSH.value: - return ForkState.AWAITING_PUSH, None return None, "no fork run and the fork workflow state was not read" if fork.workflow_state in SETUP_WORKFLOW_STATES: return ForkState.SETUP_REQUIRED, None @@ -294,10 +313,6 @@ def observe_fork_state(fork: ForkObservation, marker: Marker | None) -> tuple[Fo return None, f"unexpected fork workflow state {fork.workflow_state!r}" -def needs_workflow_state(marker: Marker | None) -> bool: - return marker is None or marker.state != ForkState.AWAITING_PUSH.value - - def has_job_started(job: dict[str, Any]) -> bool: # A job that never reached a runner -- queued, or cancelled while queued -- has no steps and a # `runner_id` of null or 0, but its `status` can be "completed" and its `started_at` set. @@ -452,6 +467,8 @@ def classify( changes_ci_files: Callable[[], bool | None], count_behind: Callable[[], int | None] = lambda: None, jobs_reported: str | None = None, + head_shared_across_bases: bool = False, + find_gate_change: Callable[[], str | None] = lambda: None, ) -> Plan: """ Decide the mutations for one gated pull request. @@ -468,6 +485,11 @@ def classify( unknown; called at the same time as ``changes_ci_files``. :param jobs_reported: the completed fork run whose jobs are all reported as statuses, recorded in the marker so that later runs do not read its jobs again. + :param head_shared_across_bases: another gated pull request has the same head commit and a + different base, so their changed files differ but they share one commit status. + :param find_gate_change: describes how the pull request changes the fork CI gate, or returns + None; only called for pull requests that the gate tampering check applies to, once per + head commit. """ plan = Plan() labels = set(pr.labels) @@ -530,6 +552,7 @@ def release(state: ForkState, comment: Callable[[Marker], str]) -> Plan: reopened_since_closing = { ForkState.CLOSED.value: pr.last_reopened_by_anyone_at, ForkState.GATE_BYPASSED.value: pr.last_reopened_at, + ForkState.GATE_MODIFIED.value: pr.last_reopened_at, } if marker and marker.state in reopened_since_closing: reopened = reopened_since_closing[marker.state] @@ -538,6 +561,8 @@ def release(state: ForkState, comment: Callable[[Marker], str]) -> Plan: _set_label(plan, labels, LABEL_NEEDS_PROJECT_CI, present=False) if plan.state == ForkState.CLOSED: comment = inactivity_close_comment(days=config.close_after_days, fork=pr.head_repo) + elif plan.state == ForkState.GATE_MODIFIED: + comment = gate_modified_close_comment(change=None) else: # The status written with the marker links the run that bypassed the gate. bypass_url = ( @@ -550,6 +575,44 @@ def release(state: ForkState, comment: Callable[[Marker], str]) -> Plan: plan.close = True return plan + # A pull request whose CI runs in a fork must not change the gate that sent it there. Checked + # once per head commit: the marker records a head commit as checked only when the check ran + # for it (not while `use fork ci` exempted it), and a retarget since the state began changes + # what the pull request changes. A + # maintainer's reopen of one closed for it leaves the marker at gate_modified, so reopening + # without restoring the gate closes it again; `use project ci` is the way to accept the change. + gate_checked = reason in TAMPER_CHECKED_REASONS + already_checked = ( + marker is not None + and marker.gate_checked + and marker.sha == pr.head_sha + and (pr.last_base_change_at is None or pr.last_base_change_at < marker.since) + ) + if gate_checked and not already_checked: + change = find_gate_change() + if change is not None: + plan.state = ForkState.GATE_MODIFIED + if not pr.is_draft: + plan.draft = True + _set_label(plan, labels, LABEL_READY_FOR_REVIEW, present=False) + _set_label(plan, labels, LABEL_NEEDS_PROJECT_CI, present=False) + _set_status( + plan, pr, ForkState.GATE_MODIFIED.value, f"https://github.com/{repo}/pull/{pr.number}/files" + ) + _set_marker_comment( + plan, + pr, + state_comment( + Marker(ForkState.GATE_MODIFIED.value, pr.head_sha, now), fork=pr.head_repo, run_url=None + ), + ) + plan.new_comments.append(gate_modified_close_comment(change=change)) + plan.close = True + plan.attention.append( + f"closed for changing the fork CI gate (author @{pr.author_login}): {change}" + ) + return plan + state, unknown_reason = observe_fork_state(fork, marker) if state is None: plan.unknown.append(unknown_reason or "unknown fork state") @@ -560,14 +623,22 @@ def release(state: ForkState, comment: Callable[[Marker], str]) -> Plan: # The decision is read back from the commit status on the head commit, which only # apache/airflow's own tokens can write, never from a comment. The changed files depend # on the base too, so a status written before the pull request was retargeted is stale. - # The status is shared by every pull request with this head commit; scoping it to one is - # tracked at https://github.com/apache/airflow/issues/74907 previous = status_state(pr.current_status) decided = previous in (ForkState.GREEN.value, ForkState.GREEN_UNVERIFIED.value) stale = pr.last_base_change_at is not None and ( pr.status_created_at is None or pr.status_created_at <= pr.last_base_change_at ) - if decided and not stale: + if head_shared_across_bases: + # One commit status cannot hold a decision per base, and trusting the one written for + # the other pull request could skip the check; both wait for a maintainer instead. + unverified = True + if previous != ForkState.GREEN_UNVERIFIED.value: + plan.attention.append( + "fork CI is green, but another gated pull request to a different base has the " + "same head commit, so its CI-defining changes cannot be told apart; review and add " + "`use project ci` to verify" + ) + elif decided and not stale: unverified = previous == ForkState.GREEN_UNVERIFIED.value else: # Writing the status again, even unchanged, records when this decision was made. @@ -583,10 +654,18 @@ def release(state: ForkState, comment: Callable[[Marker], str]) -> Plan: plan.state = state since = marker.since if marker and marker.state == state.value else now - # The marker records awaiting_push before the nudge is posted, so a failed post is not retried; - # tracked at https://github.com/apache/airflow/issues/74908 if marker and marker.state == ForkState.SETUP_REQUIRED.value and state == ForkState.AWAITING_PUSH: plan.new_comments.append(nudge_comment(fork=pr.head_repo)) + # A new comment, because the edited state comment would not notify the author that fork CI they + # had working stopped running. + if ( + marker + and marker.state not in (ForkState.SETUP_REQUIRED.value, ForkState.CLOSED.value) + and state == ForkState.SETUP_REQUIRED + ): + plan.new_comments.append( + workflow_disabled_comment(fork=pr.head_repo, workflow_state=fork.workflow_state) + ) if state in CLOSABLE_STATES and now - since >= timedelta(days=config.close_after_days): plan.state = ForkState.CLOSED @@ -594,7 +673,11 @@ def release(state: ForkState, comment: Callable[[Marker], str]) -> Plan: _set_marker_comment( plan, pr, - state_comment(Marker(ForkState.CLOSED.value, pr.head_sha, now), fork=pr.head_repo, run_url=None), + state_comment( + Marker(ForkState.CLOSED.value, pr.head_sha, now, gate_checked=gate_checked), + fork=pr.head_repo, + run_url=None, + ), ) plan.new_comments.append(inactivity_close_comment(days=config.close_after_days, fork=pr.head_repo)) plan.close = True @@ -623,7 +706,14 @@ def release(state: ForkState, comment: Callable[[Marker], str]) -> Plan: plan, pr, state_comment( - Marker(state.value, pr.head_sha, since, behind=behind if green else None, jobs=jobs_reported), + Marker( + state.value, + pr.head_sha, + since, + behind=behind if green else None, + jobs=jobs_reported, + gate_checked=gate_checked, + ), fork=pr.head_repo, run_url=fork.run_url, base_ref=pr.base_ref, @@ -668,7 +758,7 @@ def release(state: ForkState, comment: Callable[[Marker], str]) -> Plan: ... on ConvertToDraftEvent { createdAt actor { login } } ... on UnlabeledEvent { createdAt label { name } } ... on ReopenedEvent { createdAt actor { login } } - ... on BaseRefChangedEvent { createdAt } + ... on BaseRefChangedEvent { createdAt previousRefName currentRefName } } } commits(last: 1) { nodes { commit { @@ -724,22 +814,25 @@ def parse_pull_request(node: dict[str, Any]) -> PullRequestSnapshot: last_reopened = None last_reopened_by_anyone = None last_base_change = None + base_changes = [] for item in (node.get("timelineItems") or {}).get("nodes") or []: if item["__typename"] == "ConvertToDraftEvent": last_draft_actor = _login(item.get("actor")) elif item["__typename"] == "BaseRefChangedEvent": last_base_change = parse_github_datetime(item["createdAt"]) + base_changes.append( + (last_base_change, item.get("previousRefName") or "", item.get("currentRefName") or "") + ) elif item["__typename"] == "ReopenedEvent": last_reopened_by_anyone = parse_github_datetime(item["createdAt"]) # Only a reopen by someone else counts for the gate tampering check: an author closing # and reopening their own pull request must not hide a tampered run from it. if _login(item.get("actor")).casefold() != _login(node.get("author")).casefold(): last_reopened = last_reopened_by_anyone - elif ( - item["__typename"] == "UnlabeledEvent" - and (item.get("label") or {}).get("name") == LABEL_USE_PROJECT_CI - ): - last_unlabel = parse_github_datetime(item["createdAt"]) + elif item["__typename"] == "UnlabeledEvent": + label = (item.get("label") or {}).get("name") + if label == LABEL_USE_PROJECT_CI: + last_unlabel = parse_github_datetime(item["createdAt"]) current_status = None status_created_at = None @@ -785,6 +878,7 @@ def parse_pull_request(node: dict[str, Any]) -> PullRequestSnapshot: last_reopened_at=last_reopened, last_reopened_by_anyone_at=last_reopened_by_anyone, last_base_change_at=last_base_change, + base_changes=tuple(base_changes), current_status=current_status, status_created_at=status_created_at, ci_passed_state=ci_passed_state, @@ -811,6 +905,17 @@ def _is_not_found(error: GitHubError) -> bool: return "404" in str(error) or "Not Found" in str(error) +def read_file(client: GitHubClient, repo: str, path: str, ref: str) -> str | None: + """Return a file's content at a ref, or None when it does not exist there.""" + try: + content = client.rest_get(f"repos/{repo}/contents/{path}?ref={quote(ref, safe='')}") + except GitHubError as e: + if _is_not_found(e): + return None + raise + return base64.b64decode(content["content"]).decode() + + class BaseGateConfigs: """ The gate configuration of each base branch, as the gate job of a pull request against it reads it. @@ -825,15 +930,7 @@ def __init__(self, client: GitHubClient, *, repo: str, default_branch: str, defa self._configs: dict[str, GateConfig | None] = {default_branch: default_config} def _read(self, path: str, base_ref: str) -> str | None: - try: - content = self._client.rest_get( - f"repos/{self._repo}/contents/{path}?ref={quote(base_ref, safe='')}" - ) - except GitHubError as e: - if _is_not_found(e): - return None - raise - return base64.b64decode(content["content"]).decode() + return read_file(self._client, self._repo, path, base_ref) def get(self, base_ref: str) -> GateConfig | None: """Return the base branch's configuration, or None when the base branch has no gate.""" @@ -1096,6 +1193,15 @@ def load_workflow_triggers(workflows_dir: Path) -> dict[str, frozenset[str]]: return triggers +def base_at(pr: PullRequestSnapshot, when: datetime) -> str: + """The base branch the pull request had at ``when``, from its retargets.""" + base = pr.base_ref + for changed_at, previous, _ in reversed(pr.base_changes): + if when < changed_at: + base = previous + return base + + class BypassFinder: """ Finds project CI runs that bypassed the gate, from the runs apache/airflow recorded. @@ -1106,17 +1212,28 @@ class BypassFinder: signals of the design must hold for a run: it did project CI work, and its own commit changes `.github/workflows/`. A run did project CI work when the base branch's version of its workflow is not triggered by the run's event (an injected workflow, or a trigger the pull - request added to an existing one, such as `pull_request` on ci-arm.yml), or when it is a long - ci-amd.yml run whose jobs show more than the gate. + request added to an existing one, such as `pull_request` on ci-arm.yml) and a job of it ran, + or when it is a long ci-amd.yml run whose jobs show more than the gate. """ def __init__( - self, client: GitHubClient, *, repo: str, now: datetime, base_triggers: dict[str, frozenset[str]] + self, + client: GitHubClient, + *, + repo: str, + now: datetime, + base_triggers: dict[str, frozenset[str]], + is_gated_base: Callable[[str], bool] = lambda base_ref: True, ): + """:param is_gated_base: whether a base branch has the gate; runs against one without it are project CI by design.""" self._client = client self._repo = repo self._now = now self._base_triggers = base_triggers + self._is_gated_base = is_gated_base + # Pull requests whose matching run could belong to another pull request from the same + # branch; reported instead of closed. + self.ambiguous: list[tuple[int, str]] = [] since = (now - TAMPER_WINDOW).strftime("%Y-%m-%dT%H:%M:%SZ") self._runs_by_branch: dict[tuple[str, str], list[WorkflowRun]] = {} incomplete = [] @@ -1153,9 +1270,19 @@ def _get_jobs(self, run: WorkflowRun) -> list[dict[str, Any]]: def _did_run_project_ci(self, run: WorkflowRun) -> bool: if run.id not in self._did_project_ci: if run.event not in self._base_triggers.get(run.workflow_file, frozenset()): - # Counts even when every job skipped; tracked at - # https://github.com/apache/airflow/issues/74905 - result = True + # A workflow or trigger the base branch does not have, for example a fork-only + # workflow the pull request adds, did no work in apache/airflow if its jobs all + # skipped there. Jobs that cannot be listed count as work. + try: + jobs = self._get_jobs(run) + except GitHubError: + result = True + else: + result = ( + jobs_show_project_ci(jobs, self._now) + if run.workflow_file == CI_WORKFLOW_FILE + else any(has_job_started(job) and job.get("conclusion") != "skipped" for job in jobs) + ) elif run.workflow_file != CI_WORKFLOW_FILE or not run.is_long_running(self._now): result = False else: @@ -1202,11 +1329,26 @@ def _commit_changes_workflows(self, run: WorkflowRun, base_ref: str) -> bool | N self._changes_workflows[key] = None if len(files) >= COMPARE_FILES_CAP else False return self._changes_workflows[key] + def _other_pull_requests(self, pr: PullRequestSnapshot) -> list[int]: + """Other pull requests from the same branch that were open within the tampering window.""" + owner = pr.head_repo.split("/", 1)[0] # type: ignore[union-attr] + pulls = self._client.rest_get( + f"repos/{self._repo}/pulls?head={quote(owner, safe='')}:{quote(pr.head_ref, safe='')}" + "&state=all&per_page=100", + paginate=True, + ) + window_start = self._now - TAMPER_WINDOW + return [ + pull["number"] + for pull in pulls + if pull["number"] != pr.number + and (pull.get("closed_at") is None or parse_github_datetime(pull["closed_at"]) >= window_start) + ] + def find(self, pr: PullRequestSnapshot) -> str | None: if not pr.head_repo: return None - # Matching by fork branch alone also picks up runs of another pull request from that branch - # or from before a retarget; tracked at https://github.com/apache/airflow/issues/74906 + # Runs are listed per fork branch: GitHub records no pull request on runs from forks. runs = self._runs_by_branch.get((pr.head_repo.casefold(), pr.head_ref), []) # Runs before `use project ci` was removed were legitimate. Runs before a maintainer # reopened the pull request were already acted on; reopening without restoring the gate @@ -1217,15 +1359,161 @@ def find(self, pr: PullRequestSnapshot) -> str | None: for run in sorted(runs, key=lambda r: r.created_at, reverse=True): if not_before is not None and run.created_at <= not_before: continue + base_ref = base_at(pr, run.created_at) + # Project CI against a base branch without the gate is what its gate job decides. + if not self._is_gated_base(base_ref): + continue if not self._did_run_project_ci(run): continue # A change set too large to list fails closed: padding a tampered commit past the # compare API's cap must not hide the workflow change. - if self._commit_changes_workflows(run, pr.base_ref) is not False: + if self._commit_changes_workflows(run, base_ref) is not False: + others = self._other_pull_requests(pr) + if others: + self.ambiguous.append( + ( + pr.number, + f"project CI run {run.html_url} may have bypassed the gate, but it could also " + f"belong to {', '.join(f'#{n}' for n in others)} from the same branch; not closed, " + "check it", + ) + ) + return None return run.html_url return None +def get_gate_wiring(workflow_text: str | None) -> Any: + """ + The parts of a ci-amd.yml that make up the gate: its job, how build-info depends on it, the + jobs that do not depend on it (a job freed from the gate runs whatever it decides), and the + workflow-level triggers and settings that every job, the gate included, runs with. + + A file that is not a workflow mapping yields an object equal to nothing, so a pull request + breaking the workflow counts as changing the gate rather than escaping the check. + """ + if workflow_text is None: + return None + try: + workflow = yaml.safe_load(workflow_text) + except yaml.YAMLError: + return object() + jobs = workflow.get("jobs") if isinstance(workflow, dict) else None + if not isinstance(jobs, dict): + return object() + build_info = jobs.get("build-info") + build_info = build_info if isinstance(build_info, dict) else {} + return ( + jobs.get("gate"), + build_info.get("needs"), + build_info.get("if"), + find_jobs_outside_gate(jobs), + workflow.get("on", workflow.get(True)), + workflow.get("defaults"), + workflow.get("env"), + workflow.get("permissions"), + workflow.get("concurrency"), + ) + + +def find_jobs_outside_gate(jobs: dict[str, Any]) -> frozenset[str]: + """Jobs that do not depend on the gate job, directly or through other jobs.""" + + def get_needs(job_id: str) -> list[str]: + job = jobs.get(job_id) + needs = job.get("needs") if isinstance(job, dict) else None + if isinstance(needs, str): + return [needs] + return [need for need in needs if isinstance(need, str)] if isinstance(needs, list) else [] + + def depends_on_gate(job_id: str, seen: frozenset[str] = frozenset()) -> bool: + # `seen` stops a dependency cycle, which GitHub rejects anyway. + return any( + need == "gate" or (need not in seen and depends_on_gate(need, seen | {job_id})) + for need in get_needs(job_id) + ) + + return frozenset(job_id for job_id in jobs if job_id != "gate" and not depends_on_gate(job_id)) + + +def is_own_pilot_change(base_text: str | None, head_text: str | None, author_login: str) -> bool: + """Whether a gate configuration differs from the base only in the author's own pilot entry.""" + if base_text is None or head_text is None: + return False + try: + base = parse_config(base_text) + head = parse_config(head_text) + except (ValueError, TypeError, AttributeError): + # Valid TOML of the wrong shape, such as `pilot = []`, is no permitted change either. + return False + base_pilot = {login.casefold(): joined for login, joined in base.pilot.items()} + head_pilot = {login.casefold(): joined for login, joined in head.pilot.items()} + changed = { + login + for login in base_pilot.keys() | head_pilot.keys() + if base_pilot.get(login) != head_pilot.get(login) + } + return replace(head, pilot=base.pilot) == base and changed <= {author_login.casefold()} + + +class HeadMovedError(RuntimeError): + """The pull request's head moved while it was checked; the next run checks the new head.""" + + +class GateChangeFinder: + """Finds how a pull request changes the fork CI gate, from its changed files.""" + + def __init__(self, client: GitHubClient, *, repo: str): + self._client = client + self._repo = repo + + def _merge_base(self, pr: PullRequestSnapshot) -> str: + # The pull request's changes are against the merge base, not the base branch's tip, which + # may have moved on: comparing with the tip would count others' later changes as its own. + owner = (pr.head_repo or "").split("/", 1)[0] + compare = self._client.rest_get( + f"repos/{self._repo}/compare/{quote(pr.base_ref, safe='')}...{owner}:{pr.head_sha}?per_page=1" + ) + return compare["merge_base_commit"]["sha"] + + def find(self, pr: PullRequestSnapshot) -> str | None: + files = self._client.rest_get( + f"repos/{self._repo}/pulls/{pr.number}/files?per_page=100", paginate=True + ) + # The listing is of the pull request as it is now, while the head commit recorded as checked + # is the one read at the start of the run: a push in between would mark a commit checked + # by another one's files. + current_head = self._client.rest_get(f"repos/{self._repo}/pulls/{pr.number}")["head"]["sha"] + if current_head != pr.head_sha: + raise HeadMovedError( + f"head moved from {pr.head_sha[:8]} to {current_head[:8]} during the gate check" + ) + # A listing at GitHub's cap may leave out a gate file, so it is not trusted. + if len(files) >= PULL_REQUEST_FILES_CAP: + return f"it changes {len(files)} or more files, more than GitHub lists, so the gate cannot be checked" + paths = set(changed_paths(files)) + for path in sorted(GATE_PROTECTED_PATHS & paths): + return f"it changes `{path}`" + if not paths & {GATE_CONFIG_PATH, CI_WORKFLOW_PATH}: + return None + head_repo = pr.head_repo or "" + merge_base = self._merge_base(pr) + if GATE_CONFIG_PATH in paths and not is_own_pilot_change( + read_file(self._client, self._repo, GATE_CONFIG_PATH, merge_base), + read_file(self._client, head_repo, GATE_CONFIG_PATH, pr.head_sha), + pr.author_login, + ): + return f"it changes `{GATE_CONFIG_PATH}` beyond the author's own `[pilot]` entry" + if CI_WORKFLOW_PATH in paths and get_gate_wiring( + read_file(self._client, self._repo, CI_WORKFLOW_PATH, merge_base) + ) != get_gate_wiring(read_file(self._client, head_repo, CI_WORKFLOW_PATH, pr.head_sha)): + return ( + f"it changes the `gate` job of `{CI_WORKFLOW_PATH}`, which jobs depend on it, or the " + "workflow-level triggers or settings it runs with" + ) + return None + + COMMENTS_QUERY = """ query($owner: String!, $name: String!, $number: Int!, $cursor: String) { repository(owner: $owner, name: $name) { pullRequest(number: $number) { @@ -1313,6 +1601,14 @@ def count_commits_behind(client: GitHubClient, repo: str, pr: PullRequestSnapsho return compare.get("behind_by") +def find_heads_shared_across_bases(prs: Iterable[PullRequestSnapshot]) -> set[str]: + """Head commits of more than one pull request whose bases differ.""" + bases_by_head: dict[str, set[str]] = {} + for pr in prs: + bases_by_head.setdefault(pr.head_sha, set()).add(pr.base_ref) + return {head for head, bases in bases_by_head.items() if len(bases) > 1} + + @dataclass(frozen=True) class DraftChange: """A draft state change, applied by the monitor's draft-state job rather than the reconciler.""" @@ -1341,6 +1637,11 @@ def apply_plan(client: GitHubClient, pr: PullRequestSnapshot, plan: Plan) -> Dra for label in plan.remove_labels: if label != LABEL_FORK_CI_REQUIRED: client.remove_label(pr.number, label) + # New comments go before the state comment records the new state: the state recorded there is + # what the next run compares with, so a failure on the way posts them again rather than never. + # They carry no marker, so the state comment stays the earliest bot comment with one. + for comment in plan.new_comments: + client.create_comment(pr.number, comment) if plan.marker_comment is not None: if pr.marker_comment_id is not None: client.update_comment(pr.number, pr.marker_comment_id, plan.marker_comment) @@ -1350,8 +1651,6 @@ def apply_plan(client: GitHubClient, pr: PullRequestSnapshot, plan: Plan) -> Dra client.set_status(pr.number, pr.head_sha, *plan.status) if plan.ci_passed is not None: client.set_status(pr.number, pr.head_sha, *plan.ci_passed, context=CI_PASSED_CONTEXT) - for comment in plan.new_comments: - client.create_comment(pr.number, comment) if plan.close: client.close(pr.number) # Last, because the label is what brings the pull request back to the next run: when an @@ -1490,23 +1789,34 @@ def run_reconcile( ) ] observations = fetch_fork_observations(client, prs) + + def is_gated_base(base_ref: str) -> bool: + # A base branch whose configuration cannot be read is treated as gated. + try: + return configs.get(base_ref) is not None + except (GitHubError, ValueError): + return True + bypass_finder = BypassFinder( - client, repo=repo, now=now, base_triggers=load_workflow_triggers(workflows_dir) + client, + repo=repo, + now=now, + base_triggers=load_workflow_triggers(workflows_dir), + is_gated_base=is_gated_base, ) job_reporter = JobStatusReporter(client, repo=repo) + shared_heads = find_heads_shared_across_bases(prs) + gate_change_finder = GateChangeFinder(client, repo=repo) for listed in prs: try: pr = listed if listed.comments_cursor is None else find_state_comment(client, repo, listed) if pr.head_status_unread: pr = read_head_status(client, repo, pr) fork = observations.get(pr.number, ForkObservation()) - if ( - pr.head_repo - and fork.suite_status is None - and fork.error is None - and needs_workflow_state(pr.marker) - ): + # Read on every run, not only until the workflow is first seen enabled: it can be + # disabled later, and a pull request waiting for a push would then wait forever. + if pr.head_repo and fork.suite_status is None and fork.error is None: fork = ForkObservation(workflow_state=fetch_workflow_state(client, pr.head_repo)) jobs_reported = None if config.job_statuses: @@ -1530,6 +1840,8 @@ def run_reconcile( ), count_behind=functools.partial(count_commits_behind, client, repo, pr), jobs_reported=jobs_reported, + head_shared_across_bases=pr.head_sha in shared_heads, + find_gate_change=functools.partial(gate_change_finder.find, pr), ) draft_change = apply_plan(client, pr, plan) if draft_change is not None: @@ -1542,10 +1854,13 @@ def run_reconcile( client.delete_run(pr.number, run.id) unknown.extend((pr.number, reason) for reason in plan.unknown) attention.extend((pr.number, item) for item in plan.attention) + except HeadMovedError as e: + console_print(f"[info]#{listed.number}: {e}; checked again on the next run[/]") except Exception as e: console_print(f"[error]#{listed.number}: {e}[/]") unknown.append((listed.number, f"{type(e).__name__}: {e}")) + attention.extend(bypass_finder.ambiguous) notes = [bypass_finder.incomplete] if bypass_finder.incomplete else [] if job_reporter.deferred: console_print( diff --git a/dev/breeze/src/airflow_breeze/utils/kubernetes_utils.py b/dev/breeze/src/airflow_breeze/utils/kubernetes_utils.py index d6e9c343b9b1b..55b1a9195bd4e 100644 --- a/dev/breeze/src/airflow_breeze/utils/kubernetes_utils.py +++ b/dev/breeze/src/airflow_breeze/utils/kubernetes_utils.py @@ -291,7 +291,8 @@ def make_sure_kubernetes_tools_are_installed(): def sync_virtualenv(force_venv_setup: bool) -> RunCommandResult: - uv_sync_cmd = ["uv", "sync"] + # kubernetes-tests is a workspace member: a plain `uv sync` would re-resolve and rewrite the root uv.lock. + uv_sync_cmd = ["uv", "sync", "--frozen"] if force_venv_setup: uv_sync_cmd.append("--reinstall") if os.environ.get("CI", "false") == "true": diff --git a/dev/breeze/src/airflow_breeze/utils/workflow_run_approval.py b/dev/breeze/src/airflow_breeze/utils/workflow_run_approval.py index 9592fef64cc00..107fee1e4d849 100644 --- a/dev/breeze/src/airflow_breeze/utils/workflow_run_approval.py +++ b/dev/breeze/src/airflow_breeze/utils/workflow_run_approval.py @@ -27,6 +27,9 @@ CI run only executes the fork CI gate, so it is approved whatever its size, unless it changes files that define CI, in which case it waits for a maintainer and the author gets a comment saying why. +The runs of PRs from the repositories of trusted stakeholders (Amazon, Google, Astronomer) are +always approved. + 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``. @@ -125,6 +128,17 @@ } ) +# Repositories of stakeholders that maintain large parts of Airflow, with the stakeholder's display +# name. Only people with write access to such a repository can push the commits of its PRs, so their +# runs are always approved, whatever the PR changes or targets. Keyed by repository id rather than +# name: after a rename or deletion someone else could register the name. Find the id with +# ``gh api repos// --jq .id``. +TRUSTED_STAKEHOLDER_REPOSITORIES: dict[int, tuple[str, str]] = { + 65323404: ("astronomer/airflow", "Astronomer"), + 484591340: ("aws-mwaa/upstream-to-airflow", "Amazon"), + 733432432: ("VladaZakharova/airflow", "Google"), +} + # Read from the trusted main checkout, which is also the base branch whose gate job decides the # PRs that are approved automatically. FORK_CI_GATE_CONFIG_PATH = AIRFLOW_ROOT_PATH / ".github" / "fork-ci-gate.toml" @@ -186,6 +200,20 @@ def is_gated_to_fork_ci(*, pr: dict[str, Any], github_repository: str, config: G return decide(event, config).reason in GATED_REASONS +def find_trusted_stakeholder(run: dict[str, Any]) -> str | None: + repository_id = (run.get("head_repository") or {}).get("id") + trusted = TRUSTED_STAKEHOLDER_REPOSITORIES.get(repository_id) if isinstance(repository_id, int) else None + return trusted[1] if trusted else None + + +def decide_trusted_approval(*, stakeholder: str) -> ApprovalDecision: + return ApprovalDecision( + approve=True, + reasons=[f"it comes from a repository of {stakeholder}, a trusted stakeholder"], + decision_labels=frozenset(), + ) + + def find_fork_ci_definition_files(files: list[str]) -> list[str]: """The files the fork CI reconciler counts as defining CI, so both decide alike.""" return [file for file in files if changes_ci_definition([file], file_count=1)] @@ -570,11 +598,17 @@ def process_waiting_runs(api: GitHubApi, *, created_since: datetime) -> list[Pul f"leaving runs {run_ids} alone[/]" ) continue + # The runs of a group share their head repository. + stakeholder = find_trusted_stakeholder(head_runs[0]) + # Approving a run again fails, and the PRs that share a trusted head share its runs. + approved_run_ids: set[int] = set() for pr in matching_prs: failed = False decision: ApprovalDecision | None = None try: - if len(matching_prs) > 1: + if stakeholder is not None: + decision = decide_trusted_approval(stakeholder=stakeholder) + elif 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"] @@ -582,7 +616,8 @@ def process_waiting_runs(api: GitHubApi, *, created_since: datetime) -> list[Pul else: decision = decide_for_pull_request(api=api, pr=pr) pr_run_ids = run_ids - if decision.approve: + # A trusted repository's runs do not depend on the base branch or labels they started with. + if decision.approve and stakeholder is None: pr_run_ids = find_runs_started_for_current_pr( api=api, pr=pr, runs=head_runs, decision_labels=decision.decision_labels ) @@ -596,7 +631,11 @@ def process_waiting_runs(api: GitHubApi, *, created_since: datetime) -> list[Pul estimate=decision.estimate, ) pr_run_ids = run_ids + if decision.approve: + pr_run_ids = [run_id for run_id in pr_run_ids if run_id not in approved_run_ids] apply_decision(api=api, pr=pr, run_ids=pr_run_ids, decision=decision) + if decision.approve: + approved_run_ids.update(pr_run_ids) except (requests.RequestException, subprocess.CalledProcessError) as e: # One failing PR must not stop the others; the failure is reported through ``failed``. failed = True diff --git a/dev/breeze/tests/test_ci_audit.py b/dev/breeze/tests/test_ci_audit.py index 2dbc0d8a3a548..9697a67acd04d 100644 --- a/dev/breeze/tests/test_ci_audit.py +++ b/dev/breeze/tests/test_ci_audit.py @@ -260,8 +260,8 @@ def test_fork_problems(repo, message): frozenset(), (0, "active"), Role.COMMITTER, - CheckStatus.INFO, - ("gh workflow disable ci-amd.yml -R someone/airflow",), + CheckStatus.OK, + (), id="committer-enabled-is-not-a-failure", ), pytest.param( diff --git a/dev/breeze/tests/test_fork_ci_github.py b/dev/breeze/tests/test_fork_ci_github.py index 1bd85bcb74926..9f6505e295458 100644 --- a/dev/breeze/tests/test_fork_ci_github.py +++ b/dev/breeze/tests/test_fork_ci_github.py @@ -53,7 +53,17 @@ def test_gives_up_after_the_last_attempt_without_waiting(mock_run, mock_sleep): assert [c.args[0] for c in mock_sleep.call_args_list] == [30, 60, 120] -@pytest.mark.parametrize("stderr", ["gh: HTTP 504", "gh: Server Error (HTTP 502)"]) +@pytest.mark.parametrize( + "stderr", + [ + "gh: HTTP 504", + "gh: Server Error (HTTP 502)", + 'Post "https://api.github.com/graphql": read tcp 10.1.0.4:51234->140.82.112.6:443: read: connection reset by peer', + 'Post "https://api.github.com/graphql": unexpected EOF', + 'Post "https://api.github.com/graphql": net/http: TLS handshake timeout', + 'Post "https://api.github.com/graphql": dial tcp 140.82.112.6:443: i/o timeout', + ], +) @mock.patch("airflow_breeze.utils.fork_ci_github.time.sleep", autospec=True) @mock.patch("airflow_breeze.utils.fork_ci_github.subprocess.run", autospec=True) def test_retries_on_server_errors(mock_run, mock_sleep, stderr): @@ -90,6 +100,23 @@ def test_does_not_repeat_creating_a_comment_after_a_server_error(mock_run, mock_ mock_sleep.assert_not_called() +@pytest.mark.parametrize( + "stderr", + [ + 'Post "https://api.github.com/repos/x/y/issues/10/comments": net/http: TLS handshake timeout', + 'Post "https://api.github.com/repos/x/y/issues/10/comments": dial tcp: lookup api.github.com: no such host', + ], +) +@mock.patch("airflow_breeze.utils.fork_ci_github.time.sleep", autospec=True) +@mock.patch("airflow_breeze.utils.fork_ci_github.subprocess.run", autospec=True) +def test_repeats_creating_a_comment_that_did_not_connect(mock_run, mock_sleep, stderr): + mock_run.side_effect = [_completed(1, stderr=stderr), _completed(0, stdout="{}")] + + GitHubClient().create_comment(10, "body") + assert mock_run.call_count == 2 + mock_sleep.assert_called_once_with(10) + + @mock.patch( "airflow_breeze.utils.fork_ci_github.subprocess.run", autospec=True, diff --git a/dev/breeze/tests/test_fork_ci_messages.py b/dev/breeze/tests/test_fork_ci_messages.py index 407065ca4a049..f7c205c0973c2 100644 --- a/dev/breeze/tests/test_fork_ci_messages.py +++ b/dev/breeze/tests/test_fork_ci_messages.py @@ -34,6 +34,7 @@ MAX_LISTED_CI_FILES, Marker, ci_change_approval_comment, + gate_modified_close_comment, inactivity_close_comment, job_status_context, nudge_comment, @@ -42,6 +43,8 @@ render_marker, state_comment, tamper_close_comment, + ungated_comment, + workflow_disabled_comment, ) MARKER = Marker(state="awaiting_push", sha="abc123", since=datetime(2026, 10, 15, 10, 0, tzinfo=UTC)) @@ -53,6 +56,7 @@ pytest.param(MARKER, id="without-behind"), pytest.param(replace(MARKER, behind=12), id="behind"), pytest.param(replace(MARKER, jobs="123:SUCCESS"), id="jobs"), + pytest.param(replace(MARKER, gate_checked=True), id="gate-checked"), ], ) def test_marker_round_trip(marker): @@ -114,6 +118,11 @@ def test_parse_marker_malformed(body): pytest.param(nudge_comment(fork="u/airflow"), id="nudge"), pytest.param(inactivity_close_comment(days=7, fork="u/airflow"), id="inactivity"), pytest.param(tamper_close_comment(run_url="https://x"), id="tamper"), + pytest.param( + workflow_disabled_comment(fork="u/airflow", workflow_state="disabled_manually"), id="disabled" + ), + pytest.param(gate_modified_close_comment(change="it changes `x`"), id="gate-modified"), + pytest.param(ungated_comment(MARKER, base_ref="v2-10-test"), id="ungated"), ], ) def test_every_comment_ends_with_footer(text): @@ -125,9 +134,11 @@ def test_every_comment_ends_with_footer(text): [ pytest.param(state_comment(Marker("setup_required", "a", MARKER.since), fork="u/fork", run_url=None)), pytest.param(inactivity_close_comment(days=7, fork="u/fork")), + pytest.param(workflow_disabled_comment(fork="u/fork", workflow_state="disabled_manually")), ], ) def test_setup_texts_give_the_exact_command(text): + assert "gh repo sync u/fork --branch main" in text assert "gh workflow enable ci-amd.yml -R u/fork" in text diff --git a/dev/breeze/tests/test_fork_ci_reconcile.py b/dev/breeze/tests/test_fork_ci_reconcile.py index 9f3aa61f751d0..fc4294a1f83a7 100644 --- a/dev/breeze/tests/test_fork_ci_reconcile.py +++ b/dev/breeze/tests/test_fork_ci_reconcile.py @@ -44,6 +44,8 @@ DraftChange, ForkObservation, ForkState, + GateChangeFinder, + HeadMovedError, JobStatusReporter, Plan, PullRequestSnapshot, @@ -57,8 +59,10 @@ drop_reported, fetch_fork_observations, fetch_workflow_state, + find_heads_shared_across_bases, find_state_comment, is_gated_skip_run, + is_own_pilot_change, jobs_show_project_ci, load_workflow_triggers, parse_pull_request, @@ -96,11 +100,21 @@ def _pr(**overrides) -> PullRequestSnapshot: return PullRequestSnapshot(**values) -def _marker(state: str, *, days_ago: float = 0, sha: str = "abc123") -> Marker: - return Marker(state=state, sha=sha, since=NOW - timedelta(days=days_ago)) +def _marker(state: str, *, days_ago: float = 0, sha: str = "abc123", gate_checked: bool = False) -> Marker: + return Marker(state=state, sha=sha, since=NOW - timedelta(days=days_ago), gate_checked=gate_checked) -def _classify(pr, fork=None, *, bypass_run=None, changes_ci_files=False, config=CONFIG, behind=None): +def _classify( + pr, + fork=None, + *, + bypass_run=None, + changes_ci_files=False, + config=CONFIG, + behind=None, + head_shared_across_bases=False, + gate_change=None, +): find_bypass = mock.Mock(return_value=bypass_run) changes = mock.Mock(return_value=changes_ci_files) plan = classify( @@ -112,6 +126,8 @@ def _classify(pr, fork=None, *, bypass_run=None, changes_ci_files=False, config= find_bypass_run=find_bypass, changes_ci_files=changes, count_behind=mock.Mock(return_value=behind), + head_shared_across_bases=head_shared_across_bases, + find_gate_change=lambda: gate_change, ) return plan, find_bypass, changes @@ -141,12 +157,19 @@ def _classify(pr, fork=None, *, bypass_run=None, changes_ci_files=False, config= id="enabled-no-run", ), pytest.param( - ForkObservation(), + ForkObservation(workflow_state="active"), _marker("awaiting_push"), ForkState.AWAITING_PUSH, "pending", id="still-awaiting", ), + pytest.param( + ForkObservation(workflow_state="disabled_inactivity"), + None, + ForkState.SETUP_REQUIRED, + "error", + id="workflow-disabled-for-inactivity", + ), pytest.param( ForkObservation(suite_status="IN_PROGRESS", run_url=RUN_URL), None, @@ -395,6 +418,28 @@ def test_green_trust_is_decided_again_after_retargeting(status_created_at, lists assert (plan.status == GREEN_STATUS) is reposts_status +def test_green_is_not_trusted_for_a_head_shared_across_bases(): + plan, _, changes = _classify( + _pr(current_status=GREEN_STATUS), GREEN, changes_ci_files=False, head_shared_across_bases=True + ) + + assert plan.state == ForkState.GREEN_UNVERIFIED + assert plan.add_labels == ["needs project ci"] + assert len(plan.attention) == 1 + changes.assert_not_called() + + +def test_find_heads_shared_across_bases(): + prs = [ + _pr(number=1, head_sha="a", base_ref="main"), + _pr(number=2, head_sha="a", base_ref="v3-1-test"), + _pr(number=3, head_sha="b", base_ref="main"), + _pr(number=4, head_sha="b", base_ref="main"), + ] + + assert find_heads_shared_across_bases(prs) == {"a"} + + @pytest.mark.parametrize( ("files", "expected"), [ @@ -451,6 +496,37 @@ def test_nudge_on_transition_from_setup_required(): assert "Push this branch again" in plan.new_comments[0] +@pytest.mark.parametrize( + ("marker", "workflow_state", "expected_reason"), + [ + pytest.param( + _marker("awaiting_push"), "disabled_manually", "was disabled in your fork", id="awaiting-push" + ), + pytest.param(_marker("green"), "disabled_inactivity", "no activity for 60 days", id="green"), + pytest.param(_marker("failed"), "missing", "is missing from your fork", id="failed"), + ], +) +def test_comment_when_workflow_disabled_after_it_worked(marker, workflow_state, expected_reason): + plan, _, _ = _classify(_pr(marker=marker), ForkObservation(workflow_state=workflow_state)) + + assert plan.state == ForkState.SETUP_REQUIRED + assert len(plan.new_comments) == 1 + assert "Fork CI stopped running" in plan.new_comments[0] + assert expected_reason in plan.new_comments[0] + assert "gh workflow enable ci-amd.yml -R someone/airflow" in plan.new_comments[0] + + +@pytest.mark.parametrize( + "marker", + [pytest.param(None, id="first-run"), pytest.param(_marker("setup_required"), id="still-disabled")], +) +def test_no_disabled_comment_when_workflow_never_worked(marker): + plan, _, _ = _classify(_pr(marker=marker), ForkObservation(workflow_state="disabled_fork")) + + assert plan.state == ForkState.SETUP_REQUIRED + assert plan.new_comments == [] + + @pytest.mark.parametrize( ("marker", "should_close"), [ @@ -519,7 +595,7 @@ def test_unfinished_tampering_closure_is_retried(pr_overrides, expected_link): def test_since_is_kept_while_state_is_unchanged(): - marker = Marker("setup_required", "abc123", NOW - timedelta(days=2)) + marker = Marker("setup_required", "abc123", NOW - timedelta(days=2), gate_checked=True) plan, _, _ = _classify(_pr(marker=marker), ForkObservation(workflow_state="disabled_fork")) @@ -822,6 +898,8 @@ def _finder( jobs=(), compare_files=(".github/workflows/ci-amd.yml",), total_count=None, + other_pulls=(), + is_gated_base=lambda base_ref: True, ): def rest_get_pages(path): event = path.split("event=", 1)[1].split("&", 1)[0] @@ -835,11 +913,13 @@ def rest_get(path, paginate=False, items_key=None): if "/compare/" in path: assert not paginate, "the compare API returns files only on its first page" return {"files": [{"filename": name} for name in compare_files]} + if "/pulls?head=" in path: + return [{"number": 42, "closed_at": None}, *other_pulls] raise AssertionError(path) client.rest_get_pages.side_effect = rest_get_pages client.rest_get.side_effect = rest_get - return BypassFinder(client, repo=REPO, now=NOW, base_triggers=BASE_TRIGGERS) + return BypassFinder(client, repo=REPO, now=NOW, base_triggers=BASE_TRIGGERS, is_gated_base=is_gated_base) @pytest.fixture @@ -942,7 +1022,7 @@ def test_finder_ignores_legitimate_runs(client, runs, jobs, compare_files, pr_ov ], ) def test_finder_catches_triggers_the_base_branch_does_not_have(client, run): - finder = _finder(client, [run]) + finder = _finder(client, [run], jobs=[_job("Tests", "success")]) assert finder.find(_pr()) == "https://github.com/apache/airflow/actions/runs/9" @@ -975,11 +1055,44 @@ def test_finder_fails_closed_on_a_change_set_it_cannot_check(client): assert finder.find(_pr()) == "https://github.com/apache/airflow/actions/runs/7" +def test_finder_ignores_runs_made_while_the_base_had_no_gate(client): + finder = _finder( + client, [_run(7)], jobs=TAMPERED_JOBS, is_gated_base=lambda base_ref: base_ref != "v2-10-test" + ) + pr = _pr(base_changes=((NOW - timedelta(minutes=30), "v2-10-test", "main"),)) + + assert finder.find(pr) is None + + +@pytest.mark.parametrize( + "other_pulls", + [ + pytest.param([{"number": 43, "closed_at": None}], id="open"), + pytest.param( + [{"number": 43, "closed_at": (NOW - timedelta(hours=2)).isoformat()}], id="closed-recently" + ), + ], +) +def test_finder_reports_a_run_another_pull_request_may_own(client, other_pulls): + finder = _finder(client, [_run(7)], jobs=TAMPERED_JOBS, other_pulls=other_pulls) + + assert finder.find(_pr()) is None + assert finder.ambiguous == [ + ( + 42, + "project CI run https://github.com/apache/airflow/actions/runs/7 may have bypassed the gate, " + "but it could also belong to #43 from the same branch; not closed, check it", + ) + ] + + def test_finder_catches_a_workflow_moved_away(client): finder = _finder(client, [_run(7)], jobs=TAMPERED_JOBS) client.rest_get.side_effect = lambda path, paginate=False, items_key=None: ( TAMPERED_JOBS if path.endswith("jobs?filter=latest&per_page=100") + else [{"number": 42, "closed_at": None}] + if "/pulls?head=" in path else {"files": [{"filename": "docs/x.yml", "previous_filename": ".github/workflows/x.yml"}]} ) @@ -1115,6 +1228,11 @@ def test_parse_pull_request_reads_timeline_status_and_marker(): "createdAt": "2026-11-03T00:00:00Z", "label": {"name": "use project ci"}, }, + { + "__typename": "UnlabeledEvent", + "createdAt": "2026-11-03T12:00:00Z", + "label": {"name": "use fork ci"}, + }, { "__typename": "ConvertToDraftEvent", "createdAt": "2026-11-04T00:00:00Z", @@ -1130,7 +1248,12 @@ def test_parse_pull_request_reads_timeline_status_and_marker(): "createdAt": "2026-11-06T00:00:00Z", "actor": {"login": "someone"}, }, - {"__typename": "BaseRefChangedEvent", "createdAt": "2026-11-07T00:00:00Z"}, + { + "__typename": "BaseRefChangedEvent", + "createdAt": "2026-11-07T00:00:00Z", + "previousRefName": "v3-1-test", + "currentRefName": "main", + }, ] }, commits={ @@ -1171,6 +1294,7 @@ def test_parse_pull_request_reads_timeline_status_and_marker(): assert pr.last_reopened_at == datetime(2026, 11, 5, tzinfo=UTC) assert pr.last_reopened_by_anyone_at == datetime(2026, 11, 6, tzinfo=UTC) assert pr.last_base_change_at == datetime(2026, 11, 7, tzinfo=UTC) + assert pr.base_changes == ((datetime(2026, 11, 7, tzinfo=UTC), "v3-1-test", "main"),) assert pr.current_status == ("pending", "u", "d") assert pr.status_created_at == datetime(2026, 11, 8, tzinfo=UTC) assert pr.ci_passed_state == "success" @@ -1472,6 +1596,10 @@ def rest_get(path, paginate=False, items_key=None): if len(workflow_state_calls) == 1: raise GitHubError("gh api failed: HTTP 500") return {"state": "disabled_fork"} + if path.endswith("/files?per_page=100"): + return [] + if path.startswith(f"repos/{REPO}/pulls/"): + return {"head": {"sha": f"sha{path.rsplit('/', 1)[1]}"}} raise AssertionError(path) client.rest_get.side_effect = rest_get @@ -1501,6 +1629,44 @@ def rest_get(path, paginate=False, items_key=None): assert client.set_status.call_args.args[:3] == (2, "sha2", "error") +def test_run_reconcile_rereads_workflow_state_while_awaiting_push(client, tmp_path): + marker_comment = { + "databaseId": 5, + "body": render_marker(_marker("awaiting_push", sha="sha1")), + "author": {"login": "github-actions"}, + } + client.graphql.side_effect = [ + _search_result(), + _search_result(), + _search_result(_pr_node(1, comments={"nodes": [marker_comment]})), + {"pr0": {"object": None}}, + ] + client.rest_get_pages.return_value = [{"total_count": 0, "workflow_runs": []}] + + def rest_get(path, paginate=False, items_key=None): + if path == "repos/someone/my-airflow/actions/workflows/ci-amd.yml": + return {"state": "disabled_manually"} + if path.endswith("/files?per_page=100"): + return [] + if path.startswith(f"repos/{REPO}/pulls/"): + return {"head": {"sha": "sha1"}} + raise AssertionError(path) + + client.rest_get.side_effect = rest_get + + run_reconcile( + client, + CONFIG, + now=NOW, + run_url="https://run", + output_file=tmp_path / "s.json", + workflows_dir=tmp_path, + ) + + assert client.set_status.call_args.args[:3] == (1, "sha1", "error") + assert "Fork CI stopped running" in client.create_comment.call_args.args[1] + + @pytest.mark.parametrize( ("state", "deletes"), [ @@ -2073,3 +2239,371 @@ def test_job_status_reporter_compares_contexts_case_insensitively(client): assert JobStatusReporter(client, repo=REPO).report(_pr(), COMPLETED_RUN) == "1:1:FAILURE" client.set_status.assert_not_called() + + +@pytest.mark.parametrize( + ("marker", "workflow_state", "expected_text"), + [ + pytest.param(_marker("setup_required"), "active", "Push this branch again", id="nudge"), + pytest.param( + _marker("awaiting_push"), "disabled_manually", "Fork CI stopped running", id="workflow-disabled" + ), + ], +) +def test_new_comment_is_posted_before_the_state_comment_records_the_state( + client, marker, workflow_state, expected_text +): + pr = _pr(marker=marker, marker_comment_id=5) + plan, _, _ = _classify(pr, ForkObservation(workflow_state=workflow_state)) + client.create_comment.side_effect = GitHubError("gh api failed: HTTP 500") + + with pytest.raises(GitHubError): + apply_plan(client, pr, plan) + + assert expected_text in client.create_comment.call_args.args[1] + client.update_comment.assert_not_called() + + +@pytest.mark.parametrize( + "jobs", + [ + pytest.param([_job("Fork only"), _job("Also fork only")], id="all-skipped"), + pytest.param([_job("Fork only", "cancelled", started=False)], id="cancelled-before-a-runner-took-it"), + ], +) +def test_finder_ignores_an_added_workflow_whose_jobs_did_not_run(client, jobs): + run = _run(9, path=".github/workflows/fork-only.yml", minutes=1) + finder = _finder(client, [run], jobs=jobs) + + assert finder.find(_pr()) is None + + +@mock.patch("airflow_breeze.utils.fork_ci_reconcile.BypassFinder", autospec=True) +def test_run_reconcile_reports_ambiguous_runs(mock_finder, client, tmp_path): + output = tmp_path / "slack-message.json" + client.graphql.side_effect = [_search_result(), _search_result(), _search_result()] + mock_finder.return_value.incomplete = None + mock_finder.return_value.ambiguous = [(42, "could also belong to #43")] + + has_alerts = run_reconcile( + client, CONFIG, now=NOW, run_url="https://run", output_file=output, workflows_dir=tmp_path + ) + + assert has_alerts is True + assert "|#42>: could also belong to #43" in output.read_text() + + +@pytest.mark.parametrize( + ("base_ref", "rest_get", "expected"), + [ + pytest.param("main", None, True, id="default-branch"), + pytest.param("v2-10-test", _contents({}), False, id="no-gate"), + pytest.param("v3-1-test", GitHubError("gh api failed: HTTP 500"), True, id="unreadable"), + pytest.param( + "v3-1-test", _contents({(GATE_FILES[1], "v3-1-test"): "# gate\n"}), True, id="half-gate" + ), + ], +) +@mock.patch("airflow_breeze.utils.fork_ci_reconcile.BypassFinder", autospec=True) +def test_run_reconcile_tells_the_finder_which_bases_are_gated( + mock_finder, client, tmp_path, base_ref, rest_get, expected +): + client.graphql.side_effect = [_search_result(), _search_result(), _search_result()] + mock_finder.return_value.incomplete = None + mock_finder.return_value.ambiguous = [] + run_reconcile( + client, + CONFIG, + now=NOW, + run_url="https://run", + output_file=tmp_path / "s.json", + workflows_dir=tmp_path, + ) + client.rest_get.side_effect = rest_get + + assert mock_finder.call_args.kwargs["is_gated_base"](base_ref) is expected + + +def test_a_change_to_the_gate_closes(): + pr = _pr(is_draft=False, labels=("fork ci required", "ready for maintainer review")) + + plan, _, _ = _classify(pr, GREEN, gate_change="it changes `x`") + + assert plan.state == ForkState.GATE_MODIFIED + assert plan.close is True + assert plan.draft is True + assert plan.remove_labels == ["ready for maintainer review"] + assert plan.status == ( + "error", + "https://github.com/apache/airflow/pull/42/files", + "Pull request changes the fork CI gate - see the comment", + ) + assert "changes the fork CI gate: it changes `x`" in plan.new_comments[0] + assert '"state": "gate_modified"' in plan.marker_comment + assert len(plan.attention) == 1 + + +@pytest.mark.parametrize( + ("pr_overrides", "checked"), + [ + pytest.param({}, True, id="new-pull-request"), + pytest.param({"marker": _marker("running", sha="older")}, True, id="new-head-commit"), + pytest.param( + {"marker": _marker("running", gate_checked=True)}, False, id="head-commit-already-checked" + ), + pytest.param({"marker": _marker("running")}, True, id="marker-from-before-the-check"), + pytest.param( + {"marker": _marker("gate_modified", days_ago=1), "last_reopened_at": NOW}, + True, + id="reopened-by-maintainer", + ), + pytest.param( + {"author_association": "MEMBER", "labels": ("fork ci required", "use fork ci")}, + False, + id="committer-on-fork-ci", + ), + pytest.param({"marker": _marker("released")}, True, id="released-without-a-check"), + pytest.param({"marker": _marker("ungated")}, True, id="ungated-without-a-check"), + pytest.param( + { + "marker": _marker("running", days_ago=1, gate_checked=True), + "last_base_change_at": NOW - timedelta(hours=1), + }, + True, + id="retargeted-since", + ), + ], +) +def test_gate_change_is_checked_once_per_head_commit(pr_overrides, checked): + find = mock.Mock(return_value=None) + classify( + _pr(**pr_overrides), + GREEN, + config=CONFIG, + now=NOW, + repo=REPO, + find_bypass_run=lambda: None, + changes_ci_files=lambda: False, + find_gate_change=find, + ) + + assert find.called is checked + + +def test_unfinished_gate_change_closure_is_retried(): + plan, _, _ = _classify(_pr(marker=_marker("gate_modified", days_ago=1)), GREEN, gate_change=None) + + assert plan.state == ForkState.GATE_MODIFIED + assert plan.close is True + assert "Closing this pull request because it changes the fork CI gate." in plan.new_comments[0] + + +PILOT_BASE = "enabled = true\n[pilot]\nother = 2026-10-01T00:00:00Z\n" + + +@pytest.mark.parametrize( + ("head_text", "expected"), + [ + pytest.param( + "enabled = true\n[pilot]\nother = 2026-10-01T00:00:00Z\nSomeone = 2026-12-01T00:00:00Z\n", + True, + id="adds-own-entry", + ), + pytest.param("enabled = true\n", False, id="removes-another-entry"), + pytest.param( + "enabled = false\n[pilot]\nother = 2026-10-01T00:00:00Z\nsomeone = 2026-12-01T00:00:00Z\n", + False, + id="also-switches-the-gate-off", + ), + pytest.param("enabled = [", False, id="unparsable"), + pytest.param("pilot = []\n", False, id="wrong-shape"), + pytest.param(None, False, id="deleted"), + ], +) +def test_is_own_pilot_change(head_text, expected): + assert is_own_pilot_change(PILOT_BASE, head_text, "someone") is expected + + +CI_AMD = """ +on: [push, pull_request] +jobs: + gate: + steps: [{run: gate}] + build-info: + needs: [gate] + if: needs.gate.outputs.project-ci == 'true' + tests: + needs: [build-info] + steps: [{run: tests}] +""" + + +GATE_JOB_CHANGE = ( + "it changes the `gate` job of `.github/workflows/ci-amd.yml`, which jobs depend on it, " + "or the workflow-level triggers or settings it runs with" +) + + +def _gate_change_finder(client, changed, head_files, base_files=None, head_sha="abc123"): + base_files = { + ".github/workflows/ci-amd.yml": CI_AMD, + ".github/fork-ci-gate.toml": PILOT_BASE, + **(base_files or {}), + } + + def rest_get(path, paginate=False, items_key=None): + if path.endswith("/files?per_page=100"): + return changed + if "/compare/" in path: + return {"merge_base_commit": {"sha": "merge-base"}} + if path == f"repos/{REPO}/pulls/42": + return {"head": {"sha": head_sha}} + repo_name, rest = path.removeprefix("repos/").split("/contents/") + name, ref = rest.split("?ref=") + files = base_files if (repo_name, ref) == (REPO, "merge-base") else head_files + if name not in files: + raise GitHubError("gh api failed: HTTP 404: Not Found") + return {"content": base64.b64encode(files[name].encode()).decode()} + + client.rest_get.side_effect = rest_get + return GateChangeFinder(client, repo=REPO) + + +@pytest.mark.parametrize( + ("changed", "head_files", "expected"), + [ + pytest.param( + [{"filename": "dev/breeze/src/airflow_breeze/utils/fork_ci_gate.py"}], + {}, + "it changes `dev/breeze/src/airflow_breeze/utils/fork_ci_gate.py`", + id="gate-code", + ), + pytest.param( + [{"filename": "docs/x.yml", "previous_filename": ".github/workflows/fork-ci-monitor.yml"}], + {}, + "it changes `.github/workflows/fork-ci-monitor.yml`", + id="monitor-moved-away", + ), + pytest.param( + [{"filename": ".github/fork-ci-gate.toml"}], + {".github/fork-ci-gate.toml": "enabled = false\n"}, + "it changes `.github/fork-ci-gate.toml` beyond the author's own `[pilot]` entry", + id="gate-config", + ), + pytest.param( + [{"filename": ".github/workflows/ci-amd.yml"}], + {".github/workflows/ci-amd.yml": CI_AMD.replace("[gate]", "[]")}, + GATE_JOB_CHANGE, + id="gate-unwired", + ), + pytest.param( + [{"filename": ".github/fork-ci-gate.toml"}, {"filename": ".github/workflows/ci-amd.yml"}], + { + ".github/fork-ci-gate.toml": ( + "enabled = true\n[pilot]\nother = 2026-10-01T00:00:00Z\nsomeone = 2026-12-01T00:00:00Z\n" + ), + ".github/workflows/ci-amd.yml": CI_AMD.replace("run: tests", "run: more-tests"), + }, + None, + id="own-pilot-entry-and-other-ci-change", + ), + pytest.param( + [{"filename": ".github/workflows/ci-amd.yml"}], + {".github/workflows/ci-amd.yml": "jobs: [broken"}, + GATE_JOB_CHANGE, + id="workflow-unparsable", + ), + pytest.param( + [{"filename": ".github/workflows/ci-amd.yml"}], + {".github/workflows/ci-amd.yml": "jobs: [one, two]\n"}, + GATE_JOB_CHANGE, + id="jobs-not-a-mapping", + ), + pytest.param( + [{"filename": ".github/workflows/ci-amd.yml"}], + {".github/workflows/ci-amd.yml": "defaults:\n run:\n working-directory: /tmp\n" + CI_AMD}, + GATE_JOB_CHANGE, + id="workflow-defaults", + ), + pytest.param( + [{"filename": ".github/workflows/ci-amd.yml"}], + {".github/workflows/ci-amd.yml": CI_AMD.replace("needs: [build-info]", "needs: []")}, + GATE_JOB_CHANGE, + id="job-freed-from-the-gate", + ), + pytest.param( + [{"filename": ".github/workflows/ci-amd.yml"}], + { + ".github/workflows/ci-amd.yml": CI_AMD.replace( + "[push, pull_request]", "[push, pull_request, issue_comment]" + ) + }, + GATE_JOB_CHANGE, + id="trigger-added", + ), + pytest.param( + [{"filename": f"airflow-core/f{i}.py"} for i in range(3000)], + {}, + "it changes 3000 or more files, more than GitHub lists, so the gate cannot be checked", + id="listing-at-cap", + ), + ], +) +def test_gate_change_finder(client, changed, head_files, expected): + finder = _gate_change_finder(client, changed, head_files) + + assert finder.find(_pr()) == expected + + +@pytest.mark.parametrize( + ("pr_overrides", "gate_checked"), + [ + pytest.param({}, True, id="checked"), + pytest.param( + {"author_association": "MEMBER", "labels": ("fork ci required", "use fork ci")}, + False, + id="exempt-with-use-fork-ci", + ), + ], +) +def test_marker_records_whether_the_gate_was_checked(pr_overrides, gate_checked): + plan, _, _ = _classify(_pr(**pr_overrides), ForkObservation(suite_status="IN_PROGRESS", run_url=RUN_URL)) + + assert ('"gate_checked": true' in plan.marker_comment) is gate_checked + + +def test_gate_change_finder_aborts_when_the_head_moved(client): + finder = _gate_change_finder(client, [], {}, head_sha="newer") + + with pytest.raises(HeadMovedError, match="head moved from abc123 to newer"): + finder.find(_pr()) + + +def test_run_reconcile_retries_a_pull_request_whose_head_moved(client, tmp_path): + output = tmp_path / "slack-message.json" + client.graphql.side_effect = [ + _search_result(), + _search_result(), + _search_result(_pr_node(1)), + {"pr0": {"object": None}}, + ] + client.rest_get_pages.return_value = [{"total_count": 0, "workflow_runs": []}] + + def rest_get(path, paginate=False, items_key=None): + if path.endswith("/actions/workflows/ci-amd.yml"): + return {"state": "active"} + if path == f"repos/{REPO}/pulls/1/files?per_page=100": + return [] + if path == f"repos/{REPO}/pulls/1": + return {"head": {"sha": "moved"}} + raise AssertionError(path) + + client.rest_get.side_effect = rest_get + + has_alerts = run_reconcile( + client, CONFIG, now=NOW, run_url="https://run", output_file=output, workflows_dir=tmp_path + ) + + assert has_alerts is False + client.set_status.assert_not_called() + client.create_comment.assert_not_called() diff --git a/dev/breeze/tests/test_kubernetes_commands.py b/dev/breeze/tests/test_kubernetes_commands.py index 1a50e2f627a50..971d7d0df58a2 100644 --- a/dev/breeze/tests/test_kubernetes_commands.py +++ b/dev/breeze/tests/test_kubernetes_commands.py @@ -125,3 +125,20 @@ def test_commands_pass_the_github_repository_from_the_environment( assert result.exit_code == 0, result.output mock_helper.assert_called_once() assert mock_helper.call_args.kwargs["github_repository"] == FORK_REPOSITORY + + +@mock.patch.object(kubernetes_commands, "run_command", autospec=True) +@mock.patch.object(kubernetes_commands, "get_kubectl_cluster_name", autospec=True, return_value="airflow") +@mock.patch.object(kubernetes_commands, "get_k8s_env", autospec=True, return_value={"SHELL": "/bin/bash"}) +def test_run_tests_does_not_relock(mock_get_k8s_env, mock_get_cluster_name, mock_run_command): + mock_run_command.return_value = mock.Mock(returncode=0) + + kubernetes_commands._run_tests( + python="3.11", + kubernetes_version="v1.31.12", + output=None, + executor="LocalExecutor", + test_args=[], + ) + + assert mock_run_command.call_args.args[0][-1].startswith("uv run --frozen pytest ") diff --git a/dev/breeze/tests/test_kubernetes_kustomize_commands.py b/dev/breeze/tests/test_kubernetes_kustomize_commands.py index 0c6d48deea3a7..e58ac56b1fb27 100644 --- a/dev/breeze/tests/test_kubernetes_kustomize_commands.py +++ b/dev/breeze/tests/test_kubernetes_kustomize_commands.py @@ -16,8 +16,11 @@ # under the License. from __future__ import annotations +from unittest import mock + import pytest +from airflow_breeze.commands import kubernetes_kustomize_commands from airflow_breeze.commands.kubernetes_kustomize_commands import ( ALLOWED_OVERLAY_IMAGES, _discover_overlay_images, @@ -76,3 +79,20 @@ def test_allowed_images_yield_no_findings(self): ) def test_flags_images_not_on_allow_list(self, images, expected): assert _find_disallowed_overlay_images(images) == expected + + +@mock.patch.object(kubernetes_kustomize_commands, "run_command", autospec=True) +@mock.patch.object(kubernetes_kustomize_commands, "get_k8s_env", autospec=True, return_value={}) +def test_run_overlay_pytest_does_not_relock(mock_get_k8s_env, mock_run_command): + mock_run_command.return_value = mock.Mock(returncode=0) + + kubernetes_kustomize_commands._run_overlay_pytest( + overlay_name="kerberos", + namespace="airflow", + release_name="airflow", + python="3.11", + kubernetes_version="v1.31.12", + executor="LocalExecutor", + ) + + assert mock_run_command.call_args.args[0][:4] == ["uv", "run", "--frozen", "pytest"] diff --git a/dev/breeze/tests/test_kubernetes_utils.py b/dev/breeze/tests/test_kubernetes_utils.py index 88dee2d715fa0..d0ce39f84d212 100644 --- a/dev/breeze/tests/test_kubernetes_utils.py +++ b/dev/breeze/tests/test_kubernetes_utils.py @@ -27,6 +27,7 @@ SCRIPTS_CI_KUBERNETES_PATH, _render_kind_cluster_config, get_kubernetes_port_numbers, + sync_virtualenv, ) # Rendered config of a cluster created before the switch to single-node clusters — @@ -84,3 +85,9 @@ def test_get_kubernetes_port_numbers_raises_when_forwarded_port_mapping_is_missi def test_nodeport_service_exposes_forwarded_node_port(): node_port_service = yaml.safe_load((SCRIPTS_CI_KUBERNETES_PATH / "nodeport.yaml").read_text()) assert [port["nodePort"] for port in node_port_service["spec"]["ports"]] == [FORWARDED_NODE_PORT] + + +@patch("airflow_breeze.utils.kubernetes_utils.run_command", autospec=True) +def test_sync_virtualenv_does_not_relock(mock_run_command): + sync_virtualenv(force_venv_setup=False) + assert mock_run_command.call_args.args[0][:3] == ["uv", "sync", "--frozen"] diff --git a/dev/breeze/tests/test_verify_commands.py b/dev/breeze/tests/test_verify_commands.py index 0aebf5f219b6e..005f996208673 100644 --- a/dev/breeze/tests/test_verify_commands.py +++ b/dev/breeze/tests/test_verify_commands.py @@ -32,6 +32,7 @@ has_merged_commits_missing_from, verify, ) +from airflow_breeze.utils.shared_options import set_dry_run ANSI = re.compile(r"\x1b\[[0-9;]*m") @@ -360,3 +361,17 @@ def test_merged_commits_missing_from_the_base_are_warned_about(mock_files, mock_ assert warning in " ".join(ANSI.sub("", stream).split()) if as_json: assert json.loads(result.stdout)["base_ref"] == "upstream/main" + + +@patch( + "airflow_breeze.utils.run_utils.subprocess.run", + autospec=True, + side_effect=[ + CompletedProcess(args=[], returncode=0, stdout="abc123\n", stderr=""), + CompletedProcess(args=[], returncode=0, stdout="file.py\n", stderr=""), + CompletedProcess(args=[], returncode=0, stdout="", stderr=""), + ], +) +def test_changed_files_are_computed_in_dry_run(mock_subprocess_run): + set_dry_run(True) + assert get_changed_files_against("main") == ("file.py",) diff --git a/dev/breeze/tests/test_workflow_run_approval.py b/dev/breeze/tests/test_workflow_run_approval.py index 2b46aa1980204..d82a5aabe93b2 100644 --- a/dev/breeze/tests/test_workflow_run_approval.py +++ b/dev/breeze/tests/test_workflow_run_approval.py @@ -363,13 +363,14 @@ def _run( head_repo="contributor/airflow", head_branch="fix", created_at="2026-10-22T10:00:00Z", + head_repo_id=1, ) -> dict: return { "id": run_id, "head_sha": head_sha, "head_branch": head_branch, "created_at": created_at, - "head_repository": {"full_name": head_repo}, + "head_repository": {"full_name": head_repo, "id": head_repo_id}, } @@ -1058,3 +1059,60 @@ def test_process_waiting_runs_approves_gated_runs_started_before_use_fork_ci_was process_waiting_runs(api, created_since=CREATED_SINCE) assert approve_run.call_args_list == [mock.call(2), mock.call(3)] + + +@pytest.mark.parametrize( + ("head_repo", "head_repo_id", "stakeholder"), + [ + ("astronomer/airflow", 65323404, "Astronomer"), + ("aws-mwaa/upstream-to-airflow", 484591340, "Amazon"), + ("VladaZakharova/airflow", 733432432, "Google"), + ], +) +@mock.patch(f"{MODULE}.decide_for_pull_request", autospec=True) +def test_process_waiting_runs_always_approves_trusted_stakeholder_repositories( + mock_decide, head_repo, head_repo_id, stakeholder +): + api = mock.MagicMock(spec=GitHubApi) + api.github_repository = "apache/airflow" + api.get_runs_waiting_for_approval.return_value = [ + _run(1, "aaa", head_repo=head_repo, head_repo_id=head_repo_id, created_at="2026-10-22T08:30:00Z"), + _run(2, "aaa", head_repo=head_repo, head_repo_id=head_repo_id), + ] + # Several PRs share the head and a run started before the PRs were opened. + api.get_open_pull_requests.return_value = [ + _pr(10, "aaa", head_repo=head_repo, labels=[NEED_WORKFLOW_APPROVAL_LABEL]), + _pr(11, "aaa", head_repo=head_repo, labels=[NEED_WORKFLOW_APPROVAL_LABEL]), + ] + + results = process_waiting_runs(api, created_since=CREATED_SINCE) + + mock_decide.assert_not_called() + assert [(r.pr_number, r.run_ids, r.decision.approve) for r in results] == [ + (10, [1, 2], True), + (11, [], True), + ] + assert stakeholder in results[0].decision.reasons[0] + assert api.approve_run.call_args_list == [mock.call(1), mock.call(2)] + api.add_label.assert_not_called() + assert api.remove_label.call_args_list == [ + mock.call(10, NEED_WORKFLOW_APPROVAL_LABEL), + mock.call(11, NEED_WORKFLOW_APPROVAL_LABEL), + ] + + +@mock.patch(f"{MODULE}.decide_for_pull_request", autospec=True) +def test_process_waiting_runs_does_not_trust_a_repository_by_its_name(mock_decide): + api = mock.MagicMock(spec=GitHubApi) + api.github_repository = "apache/airflow" + api.get_last_base_change.return_value = None + api.get_last_label_change.return_value = None + api.get_runs_waiting_for_approval.return_value = [_run(1, "aaa", head_repo="astronomer/airflow")] + api.get_open_pull_requests.return_value = [_pr(10, "aaa", head_repo="astronomer/airflow")] + mock_decide.return_value = ApprovalDecision(approve=False, reasons=["too big"]) + + results = process_waiting_runs(api, created_since=CREATED_SINCE) + + mock_decide.assert_called_once() + assert not results[0].decision.approve + api.approve_run.assert_not_called() diff --git a/scripts/ci/prek/check_ci_workflows_in_sync.py b/scripts/ci/prek/check_ci_workflows_in_sync.py index 047c3f0dd7c89..7043131d24271 100755 --- a/scripts/ci/prek/check_ci_workflows_in_sync.py +++ b/scripts/ci/prek/check_ci_workflows_in_sync.py @@ -161,7 +161,9 @@ AMD_ONLY_GATE_BLOCK = """ gate: # Fork CI gate (AIP-120): decides whether project CI runs for this event. Configuration and # code come from the base branch so a pull request cannot change its own eligibility. See - # dev/breeze/doc/ci/08_fork_ci_for_external_contributors.md. + # dev/breeze/doc/ci/08_fork_ci_for_external_contributors.md. A pull request whose CI runs + # in the contributor's fork and changes this job, or how build-info depends on it, is closed + # by the fork CI monitor ("Gate protection" in that document). name: "Fork CI gate" runs-on: ["ubuntu-slim"] timeout-minutes: 5 diff --git a/scripts/in_container/run_generate_constraints.py b/scripts/in_container/run_generate_constraints.py index daf0b43c2bfa5..5ca30fc59f143 100755 --- a/scripts/in_container/run_generate_constraints.py +++ b/scripts/in_container/run_generate_constraints.py @@ -20,6 +20,7 @@ import ast import json import os +import re import sys import tomllib from concurrent.futures import ThreadPoolExecutor @@ -233,21 +234,49 @@ def freeze_distributions_to_file( console.print(f"[green]Constraints generated to file: {file.name}. Wrote {count_lines} lines") +def _github_headers(accept: str) -> dict[str, str]: + headers = {"Accept": accept} + if os.environ.get("GITHUB_TOKEN"): + headers["Authorization"] = f"Bearer {os.environ.get('GITHUB_TOKEN')}" + else: + console.print("[bright_blue]No GITHUB_TOKEN - using non-authenticated request.") + return headers + + +def _constraints_branch_has_other_python_files(config_params: ConfigParams) -> bool: + """Tell a first publication for a new Python from a baseline that should exist but is missing.""" + listing_url = ( + f"https://api.github.com/repos/{config_params.constraints_github_repository}/contents/" + f"?ref={config_params.default_constraints_branch}" + ) + r = requests.get(listing_url, timeout=60, headers=_github_headers("application/vnd.github.v3+json")) + r.raise_for_status() + other_python_file = re.compile(rf"{re.escape(config_params.airflow_constraints_mode)}-3\.\d+\.txt") + return any( + other_python_file.fullmatch(entry["name"]) + and entry["name"] != f"{config_params.airflow_constraints_mode}-{config_params.python}.txt" + for entry in r.json() + ) + + def download_latest_constraint_file(config_params: ConfigParams): constraints_url = ( "https://api.github.com/repos/" f"{config_params.constraints_github_repository}/contents/" f"{config_params.airflow_constraints_mode}-{config_params.python}.txt?ref={config_params.default_constraints_branch}" ) - # download the latest constraints file - # download using requests - headers = {"Accept": "application/vnd.github.v3.raw"} - if os.environ.get("GITHUB_TOKEN"): - headers["Authorization"] = f"Bearer {os.environ.get('GITHUB_TOKEN')}" - else: - console.print("[bright_blue]No GITHUB_TOKEN - using non-authenticated request.") console.print(f"[bright_blue]Downloading constraints file from {constraints_url}") - r = requests.get(constraints_url, timeout=60, headers=headers) + r = requests.get(constraints_url, timeout=60, headers=_github_headers("application/vnd.github.v3.raw")) + if r.status_code == 404 and _constraints_branch_has_other_python_files(config_params): + # The first generation for a new Python has no file to compare with: diff against an empty + # baseline. A 404 on a branch that carries no file of this mode at all is still an error. + console.print( + f"[yellow]No {config_params.airflow_constraints_mode} file for Python {config_params.python} on " + f"{config_params.default_constraints_branch}, but other Pythons have one - first publication, " + "using an empty baseline." + ) + config_params.latest_constraints_file.write_text("") + return r.raise_for_status() with config_params.latest_constraints_file.open("w") as constraints_file: constraints_file.write(r.text) diff --git a/scripts/tests/in_container/test_run_generate_constraints.py b/scripts/tests/in_container/test_run_generate_constraints.py index e8cc482acc4b1..f01653a422d36 100644 --- a/scripts/tests/in_container/test_run_generate_constraints.py +++ b/scripts/tests/in_container/test_run_generate_constraints.py @@ -272,3 +272,72 @@ def fake_lookup(distribution, python_version, allow_pre_releases): assert seen_for_distributions == ["3.14"] assert seen_for_lookup == [("3.14", False)] + + +class TestDownloadLatestConstraintFile: + @staticmethod + def _download_config(tmp_path: Path, mode: str = "constraints-source-providers") -> SimpleNamespace: + return SimpleNamespace( + latest_constraints_file=tmp_path / "original.txt", + airflow_constraints_mode=mode, + constraints_github_repository="apache/airflow", + default_constraints_branch="constraints-main", + python="3.15", + ) + + @staticmethod + def _response(status_code: int = 200, text: str = "", payload: list | None = None) -> SimpleNamespace: + def raise_for_status(): + if status_code >= 400: + raise RuntimeError(f"HTTP {status_code}") + + return SimpleNamespace( + status_code=status_code, + text=text, + json=lambda: payload or [], + raise_for_status=raise_for_status, + ) + + def test_writes_the_downloaded_file(self, tmp_path, monkeypatch): + config = self._download_config(tmp_path) + monkeypatch.setattr(m.requests, "get", lambda *args, **kwargs: self._response(text="a==1\n")) + + m.download_latest_constraint_file(config) + + assert config.latest_constraints_file.read_text() == "a==1\n" + + def test_missing_file_for_a_new_python_bootstraps_an_empty_baseline(self, tmp_path, monkeypatch): + config = self._download_config(tmp_path) + listing = [ + {"name": "constraints-source-providers-3.14.txt"}, + {"name": "constraints-3.15.txt"}, + {"name": "README.md"}, + ] + responses = iter([self._response(404), self._response(payload=listing)]) + monkeypatch.setattr(m.requests, "get", lambda *args, **kwargs: next(responses)) + + m.download_latest_constraint_file(config) + + assert config.latest_constraints_file.read_text() == "" + + def test_missing_file_is_an_error_when_no_other_python_has_one(self, tmp_path, monkeypatch): + config = self._download_config(tmp_path) + listing = [ + {"name": "constraints-3.14.txt"}, + {"name": "constraints-no-providers-3.14.txt"}, + {"name": "constraints-source-providers-3.15.txt"}, + ] + responses = iter([self._response(404), self._response(payload=listing)]) + monkeypatch.setattr(m.requests, "get", lambda *args, **kwargs: next(responses)) + + with pytest.raises(RuntimeError, match="HTTP 404"): + m.download_latest_constraint_file(config) + + assert not config.latest_constraints_file.exists() + + def test_other_http_errors_are_not_treated_as_a_first_publication(self, tmp_path, monkeypatch): + config = self._download_config(tmp_path) + monkeypatch.setattr(m.requests, "get", lambda *args, **kwargs: self._response(500)) + + with pytest.raises(RuntimeError, match="HTTP 500"): + m.download_latest_constraint_file(config)