This is an automated email from the ASF dual-hosted git repository. potiuk pushed a commit to branch auto-approve-workflow-runs in repository https://gitbox.apache.org/repos/asf/airflow.git
commit b84ff18873a22b76196ad8135873891c1e1cec6d Author: Jarek Potiuk <[email protected]> AuthorDate: Sun Oct 25 18:06:35 2026 +0100 Keep CI-minutes estimates true to the runs they approve A run selects its tests from the labels the PR had when it started, so approving it on the current labels could start a full-tests run whose label was removed while it waited. Calibrating from self-hosted and all-versions runs, and leaving out the jobs that need no CI image, made the estimate describe runs other than those it is used to approve. Generated-by: Claude Opus 5 --- dev/breeze/doc/ci/05_workflows.md | 10 +- .../doc/ci/08_fork_ci_for_external_contributors.md | 4 + .../airflow_breeze/utils/ci_minutes_estimate.py | 52 +++++++--- .../airflow_breeze/utils/workflow_run_approval.py | 83 +++++++++++++-- .../tests/test_calibrate_ci_minutes_estimates.py | 101 ++++++++++++++++++ dev/breeze/tests/test_ci_minutes_estimate.py | 61 +++++++++++ dev/breeze/tests/test_workflow_run_approval.py | 115 +++++++++++++++++++++ dev/calibrate_ci_minutes_estimates.py | 24 ++++- 8 files changed, 425 insertions(+), 25 deletions(-) diff --git a/dev/breeze/doc/ci/05_workflows.md b/dev/breeze/doc/ci/05_workflows.md index df0615ffe4e..c06d4f8a267 100644 --- a/dev/breeze/doc/ci/05_workflows.md +++ b/dev/breeze/doc/ci/05_workflows.md @@ -132,7 +132,15 @@ Only runs created in the last 24 hours (`--max-run-age-hours`) are considered, s approves a backlog of older waiting runs at once; those stay with maintainers. A commit that is the head of several open PRs is never approved automatically, because its runs cannot be told apart by PR, and only runs started after the PR was opened and after its base branch last changed are -approved (retargeting a PR does not start new runs). Other PRs get the +approved (retargeting a PR does not start new runs). A run selects its jobs from the labels the PR +had when it started, while the decision uses the current labels, so for PRs that run project CI a +run started before a label that changes which tests run was added or removed +(`full tests needed`, `all versions`, `default versions only`, `latest versions only`, `canary`, +`upgrade to newer dependencies`, `area:e2e-tests`, `area:kubernetes-tests`) is left for a +maintainer, since it still runs the selection it started with; the next push is decided afresh. +For a PR whose CI runs in the fork the selection does not matter, but the gate falls back to the +labels a run started with when it cannot read the current ones, so a run started before +`use fork ci` or `use project ci` was added or removed is left for a maintainer instead. Other PRs get the `need workflow approval` label and wait for a maintainer. When processing a PR fails (for example on a GitHub API error), the workflow tries to add the label to it and the workflow run fails, so the failure stays visible even if the label could not be added. The label is removed when a later commit 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 080a0554b21..762e920e2d8 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 @@ -367,6 +367,10 @@ applies instead: `github-actions[bot]` with the marker count. A change to `.github/` can remove the gate or add a workflow that does run, so such a run must not start without a review. +A run started before `use fork ci` or `use project ci` was last added or removed is not approved +automatically: the gate falls back to the labels of the run's event when it cannot read the current +ones, and those can still send the run to project CI. The next push is decided afresh. + Both rules use the same list as "Trusting a green fork run", so a gated pull request whose runs are approved automatically is also one whose green fork run is trusted, and one that changes CI is both left waiting for approval and labelled `needs project ci` once its fork run is green. The gate diff --git a/dev/breeze/src/airflow_breeze/utils/ci_minutes_estimate.py b/dev/breeze/src/airflow_breeze/utils/ci_minutes_estimate.py index 65df3d6b208..fd51a2140b5 100644 --- a/dev/breeze/src/airflow_breeze/utils/ci_minutes_estimate.py +++ b/dev/breeze/src/airflow_breeze/utils/ci_minutes_estimate.py @@ -55,6 +55,9 @@ AGENT_FRAMEWORK_ESTIMATE = (11, 5) JAVA_SDK_ESTIMATE = (7, 4) GO_SDK_ESTIMATE = (1, 1) TS_SDK_ESTIMATE = (1, 1) +TS_SDK_DOCS_ESTIMATE = (1, 1) +# The provider mypy job runs whenever provider code changes, also without provider unit tests. +MYPY_PROVIDERS_ESTIMATE = (8, 8) AIRFLOW_CTL_ESTIMATE = (1, 1) # Core tests: about 38 job-minutes for the first test type and 61 for each further one, up to @@ -65,13 +68,14 @@ CORE_ALL_TEST_TYPES_MINUTES = 222 CORE_LONGEST_JOB_MINUTES = {1: 11, 2: 29} CORE_LONGEST_JOB_MINUTES_DEFAULT = 32 -# Provider tests (DB/non-DB, lowest dependencies, compatibility, distributions, mypy) scale with the -# size of the tested providers' test suites: amazon or google cost far more than a small provider. -# Fitted on 224 runs testing 2 to 22 providers: job-minutes ~= 105 + 0.161 per provider test file, and -# the longest provider job ~= 16 + 0.015 per test file, both capped at the cost of testing all providers. -PROVIDERS_BASE_MINUTES = 105 +# Provider tests (DB/non-DB, lowest dependencies, compatibility, distributions) scale with the size of +# the tested providers' test suites: amazon or google cost far more than a small provider. Fitted on +# 224 runs testing 2 to 22 providers, together with the provider mypy job: job-minutes ~= 105 + 0.161 +# per provider test file, and the longest provider job ~= 16 + 0.015 per test file, both capped at the +# cost of testing all providers. The mypy job's median is taken off the base and the all-providers cost. +PROVIDERS_BASE_MINUTES = 105 - MYPY_PROVIDERS_ESTIMATE[0] PROVIDERS_MINUTES_PER_TEST_FILE = 0.161 -PROVIDERS_ALL_MINUTES = 315 +PROVIDERS_ALL_MINUTES = 315 - MYPY_PROVIDERS_ESTIMATE[0] PROVIDERS_LONGEST_JOB_BASE_MINUTES = 16 PROVIDERS_LONGEST_JOB_MINUTES_PER_TEST_FILE = 0.015 PROVIDERS_ALL_LONGEST_JOB_MINUTES = 36 @@ -153,10 +157,34 @@ def estimate_providers_group(provider_test_types: list[str]) -> EstimatedJobGrou ) +# Jobs that need only build-info, so they also run when no CI image is built. The Java SDK group +# includes its docs job. +IMAGE_INDEPENDENT_GROUPS = ( + ("run_java_sdk_tests", "Java SDK tests", JAVA_SDK_ESTIMATE), + ("run_go_sdk_tests", "Go SDK tests", GO_SDK_ESTIMATE), + ("run_ts_sdk_tests", "TypeScript SDK tests", TS_SDK_ESTIMATE), + ("run_ts_sdk_docs", "TypeScript SDK docs", TS_SDK_DOCS_ESTIMATE), + ("run_airflow_ctl_tests", "Airflow CTL tests", AIRFLOW_CTL_ESTIMATE), +) + + +def estimate_image_independent_groups(selective_checks: Any) -> list[EstimatedJobGroup]: + return [ + EstimatedJobGroup(name, *estimate) + for flag, name, estimate in IMAGE_INDEPENDENT_GROUPS + if getattr(selective_checks, flag) + ] + + def estimate_ci_minutes(selective_checks: Any) -> CiMinutesEstimate: """Estimate the job groups, job-minutes and wall-clock time of a ``Tests (AMD)`` PR run.""" if selective_checks.basic_checks_only: - return CiMinutesEstimate([EstimatedJobGroup("Basic checks only", *BASIC_CHECKS_ONLY_ESTIMATE)]) + return CiMinutesEstimate( + [ + EstimatedJobGroup("Basic checks only", *BASIC_CHECKS_ONLY_ESTIMATE), + *estimate_image_independent_groups(selective_checks), + ] + ) groups = [EstimatedJobGroup("CI image, basic tests and static checks", *BASE_ESTIMATE)] if selective_checks.docs_build: groups.append(EstimatedJobGroup("Documentation build", *DOCS_ESTIMATE)) @@ -172,12 +200,15 @@ def estimate_ci_minutes(selective_checks: Any) -> CiMinutesEstimate: ) ) groups.append(EstimatedJobGroup("Integration and system tests", *INTEGRATION_ESTIMATE)) - if not selective_checks.skip_providers_tests: + # Like the provider test jobs of ci-amd.yml, which all need run-unit-tests. + if selective_checks.run_unit_tests and not selective_checks.skip_providers_tests: providers_group = estimate_providers_group( selective_checks._get_providers_test_types_to_run(split_to_individual_providers=True) ) if providers_group: groups.append(providers_group) + if selective_checks.run_mypy_providers: + groups.append(EstimatedJobGroup("Provider mypy checks", *MYPY_PROVIDERS_ESTIMATE)) if selective_checks.full_tests_needed: groups.append(EstimatedJobGroup("Special tests", *SPECIAL_TESTS_ESTIMATE)) optional_groups = ( @@ -187,12 +218,9 @@ def estimate_ci_minutes(selective_checks: Any) -> CiMinutesEstimate: ("run_kustomize_overlays_tests", "Kustomize overlays tests", KUSTOMIZE_ESTIMATE), ("run_task_sdk_tests", "Task SDK tests", TASK_SDK_ESTIMATE), ("run_agent_framework_tests", "Agent framework tests", AGENT_FRAMEWORK_ESTIMATE), - ("run_java_sdk_tests", "Java SDK tests", JAVA_SDK_ESTIMATE), - ("run_go_sdk_tests", "Go SDK tests", GO_SDK_ESTIMATE), - ("run_ts_sdk_tests", "TypeScript SDK tests", TS_SDK_ESTIMATE), - ("run_airflow_ctl_tests", "Airflow CTL tests", AIRFLOW_CTL_ESTIMATE), ) for flag, name, estimate in optional_groups: if getattr(selective_checks, flag): groups.append(EstimatedJobGroup(name, *estimate)) + groups.extend(estimate_image_independent_groups(selective_checks)) return CiMinutesEstimate(groups) diff --git a/dev/breeze/src/airflow_breeze/utils/workflow_run_approval.py b/dev/breeze/src/airflow_breeze/utils/workflow_run_approval.py index f0c4d0d7325..2f7fc1fe34b 100644 --- a/dev/breeze/src/airflow_breeze/utils/workflow_run_approval.py +++ b/dev/breeze/src/airflow_breeze/utils/workflow_run_approval.py @@ -49,6 +49,8 @@ from airflow_breeze.utils.ci_minutes_estimate import CiMinutesEstimate, estimate from airflow_breeze.utils.console import console_print from airflow_breeze.utils.fork_ci_gate import ( GATED_REASONS, + LABEL_USE_FORK_CI, + LABEL_USE_PROJECT_CI, GateConfig, GateEvent, decide, @@ -59,6 +61,16 @@ from airflow_breeze.utils.fork_ci_messages import CI_CHANGE_MARKER, ci_change_ap from airflow_breeze.utils.fork_ci_reconcile import changes_ci_definition from airflow_breeze.utils.path_utils import AIRFLOW_ROOT_PATH from airflow_breeze.utils.run_utils import run_command +from airflow_breeze.utils.selective_checks import ( + ALL_VERSIONS_LABEL, + AREA_E2E_TESTS_LABEL, + AREA_KUBERNETES_TESTS_LABEL, + CANARY_LABEL, + DEFAULT_VERSIONS_ONLY_LABEL, + FULL_TESTS_NEEDED_LABEL, + LATEST_VERSIONS_ONLY_LABEL, + UPGRADE_TO_NEWER_DEPENDENCIES_LABEL, +) NEED_WORKFLOW_APPROVAL_LABEL = "need workflow approval" @@ -104,6 +116,27 @@ NON_SOURCE_FILE_PATTERN = re.compile( GITHUB_API_URL = "https://api.github.com"; +# Labels with which selective checks select a different set of tests or versions. Selective checks +# read the labels of the run's event, so a run started before one was added or removed still runs +# the selection of the labels it started with, which the current labels do not show. +TEST_SELECTION_LABELS = frozenset( + { + ALL_VERSIONS_LABEL, + AREA_E2E_TESTS_LABEL, + AREA_KUBERNETES_TESTS_LABEL, + CANARY_LABEL, + DEFAULT_VERSIONS_ONLY_LABEL, + FULL_TESTS_NEEDED_LABEL, + LATEST_VERSIONS_ONLY_LABEL, + UPGRADE_TO_NEWER_DEPENDENCIES_LABEL, + } +) + +# Labels that move a PR between project CI and fork CI. The gate reads them when it runs, but falls +# back to the labels of the run's event when it cannot, so a run started before one changed could +# still run project CI. +GATE_OVERRIDE_LABELS = frozenset({LABEL_USE_FORK_CI, LABEL_USE_PROJECT_CI}) + # 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" @@ -122,6 +155,8 @@ class ApprovalDecision: estimate: CiMinutesEstimate | None = None # Posted on the PR once, when the runs are left for a maintainer. comment: str | None = None + # The labels the decision depends on: runs started before one of them changed are not approved. + decision_labels: frozenset[str] = TEST_SELECTION_LABELS def count_source_lines(changed_files: list[ChangedFile]) -> int: @@ -190,6 +225,7 @@ def decide_gated_approval(*, changed_files: list[ChangedFile]) -> ApprovalDecisi return ApprovalDecision( approve=True, reasons=["its CI runs in the contributor's fork, so its project CI run only executes the gate"], + decision_labels=GATE_OVERRIDE_LABELS, ) @@ -287,6 +323,7 @@ class GitHubApi: def __init__(self, *, github_repository: str, github_token: str, report_only: bool): self.github_repository = github_repository self.report_only = report_only + self._timelines: dict[int, list[dict[str, Any]]] = {} self._session = requests.Session() self._session.headers.update( { @@ -356,15 +393,29 @@ class GitHubApi: ) response.raise_for_status() + def _get_timeline(self, pr_number: int) -> list[dict[str, Any]]: + if pr_number not in self._timelines: + self._timelines[pr_number] = self._get_paginated(f"issues/{pr_number}/timeline?per_page=100") + return self._timelines[pr_number] + def get_last_base_change(self, pr_number: int) -> datetime | None: - events = self._get_paginated(f"issues/{pr_number}/timeline?per_page=100") changes = [ parse_github_time(event["created_at"]) - for event in events + for event in self._get_timeline(pr_number) if event.get("event") == "base_ref_changed" ] return max(changes, default=None) + def get_last_label_change(self, pr_number: int, labels: frozenset[str]) -> datetime | None: + """When one of ``labels`` was last added to or removed from the PR.""" + changes = [ + parse_github_time(event["created_at"]) + for event in self._get_timeline(pr_number) + if event.get("event") in ("labeled", "unlabeled") + and (event.get("label") or {}).get("name") in labels + ] + return max(changes, default=None) + def remove_label(self, pr_number: int, label: str) -> None: if self.report_only: console_print(f"[info]Report only: would remove label '{label}' from PR #{pr_number}[/]") @@ -543,11 +594,16 @@ def process_waiting_runs(api: GitHubApi, *, created_since: datetime) -> list[Pul decision = decide_for_pull_request(api=api, pr=pr) pr_run_ids = run_ids if decision.approve: - pr_run_ids = find_runs_started_for_current_pr(api=api, pr=pr, runs=head_runs) + pr_run_ids = find_runs_started_for_current_pr( + api=api, pr=pr, runs=head_runs, decision_labels=decision.decision_labels + ) if not pr_run_ids: decision = ApprovalDecision( approve=False, - reasons=["its waiting runs started before the PR was opened or retargeted"], + reasons=[ + "its waiting runs started before the PR was opened, retargeted " + "or changed a label its decision depends on" + ], estimate=decision.estimate, ) pr_run_ids = run_ids @@ -583,7 +639,11 @@ def process_waiting_runs(api: GitHubApi, *, created_since: datetime) -> list[Pul def find_runs_started_for_current_pr( - *, api: GitHubApi, pr: dict[str, Any], runs: list[dict[str, Any]] + *, + api: GitHubApi, + pr: dict[str, Any], + runs: list[dict[str, Any]], + decision_labels: frozenset[str] = TEST_SELECTION_LABELS, ) -> list[int]: """ Return the runs started after the PR was opened and after its base branch last changed. @@ -591,11 +651,18 @@ def find_runs_started_for_current_pr( Retargeting a PR does not start new runs, and a closed PR can leave runs with the same head behind. Such runs were started for another base branch, so the decision about the current PR must not approve them. + + :param decision_labels: the labels the decision depends on. It is made from the current labels, + while a run can act on those of its event: with ``full tests needed`` removed, or + ``default versions only`` added to a run of all versions, the estimate is smaller than the + run, and without ``use fork ci`` in its event a run can fall back to project CI. """ started_after = parse_github_time(pr["created_at"]) - last_base_change = api.get_last_base_change(pr["number"]) - if last_base_change is not None: - started_after = max(started_after, last_base_change) + changes = [ + api.get_last_base_change(pr["number"]), + api.get_last_label_change(pr["number"], decision_labels), + ] + started_after = max([started_after, *(change for change in changes if change is not None)]) return [run["id"] for run in runs if parse_github_time(run["created_at"]) >= started_after] diff --git a/dev/breeze/tests/test_calibrate_ci_minutes_estimates.py b/dev/breeze/tests/test_calibrate_ci_minutes_estimates.py new file mode 100644 index 00000000000..f420a38c3e8 --- /dev/null +++ b/dev/breeze/tests/test_calibrate_ci_minutes_estimates.py @@ -0,0 +1,101 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +from __future__ import annotations + +import importlib.util +import json + +import pytest + +from airflow_breeze.utils.path_utils import AIRFLOW_ROOT_PATH + +# The script is standalone (run with `uv run`), so it is loaded from its path. +SCRIPT_PATH = AIRFLOW_ROOT_PATH / "dev" / "calibrate_ci_minutes_estimates.py" +_spec = importlib.util.spec_from_file_location("calibrate_ci_minutes_estimates", SCRIPT_PATH) +assert _spec is not None +assert _spec.loader is not None +calibrate = importlib.util.module_from_spec(_spec) +_spec.loader.exec_module(calibrate) + +JOBS = [ + { + "name": "Basic tests / Static checks", + "conclusion": "success", + "started_at": "2026-10-22T10:00:00Z", + "completed_at": "2026-10-22T10:06:00Z", + }, + { + "name": "MyPy providers checks", + "conclusion": "success", + "started_at": "2026-10-22T10:00:00Z", + "completed_at": "2026-10-22T10:08:00Z", + }, + { + "name": "TypeScript SDK docs", + "conclusion": "success", + "started_at": "2026-10-22T10:00:00Z", + "completed_at": "2026-10-22T10:01:00Z", + }, +] + + +def _cache_run(cache_dir, run_id, *, runner_type='["ubuntu-22.04"]', all_versions="false"): + (cache_dir / f"{run_id}.jobs.json").write_text(json.dumps(JOBS)) + (cache_dir / f"{run_id}.build-info.log").write_text( + "2026-10-22T10:00:00Z full-tests-needed = false\n" + f"2026-10-22T10:00:00Z runner-type = {runner_type}\n" + f"2026-10-22T10:00:00Z all-versions = {all_versions}\n" + ) + + [email protected]( + ("runner_type", "all_versions", "counted"), + [ + pytest.param('["ubuntu-22.04"]', "false", True, id="public runners, default versions"), + pytest.param( + "runs-on=1/family=m7a/image=ubuntu22-full-x64/volume=150gb/cpu={0}/ram={1}", + "false", + False, + id="self-hosted runners", + ), + pytest.param('["ubuntu-22.04"]', "true", False, id="all versions"), + ], +) +def test_build_rows_counts_only_the_runs_the_estimate_models(tmp_path, runner_type, all_versions, counted): + _cache_run(tmp_path, 1, runner_type=runner_type, all_versions=all_versions) + + rows = calibrate.build_rows([1], tmp_path) + + assert len(rows) == int(counted) + + +def test_build_rows_groups_ts_sdk_docs_separately(tmp_path): + _cache_run(tmp_path, 1) + + (row,) = calibrate.build_rows([1], tmp_path) + + assert row["minutes"]["ts-sdk-docs"] == 1 + assert "ts-sdk" not in row["minutes"] + + +def test_build_rows_groups_provider_mypy_separately(tmp_path): + _cache_run(tmp_path, 1) + + (row,) = calibrate.build_rows([1], tmp_path) + + assert row["minutes"]["mypy-providers"] == 8 + assert "providers" not in row["minutes"] diff --git a/dev/breeze/tests/test_ci_minutes_estimate.py b/dev/breeze/tests/test_ci_minutes_estimate.py index 5396e3e1465..b849ed9f967 100644 --- a/dev/breeze/tests/test_ci_minutes_estimate.py +++ b/dev/breeze/tests/test_ci_minutes_estimate.py @@ -24,9 +24,11 @@ from airflow_breeze.utils.ci_minutes_estimate import ( BASE_ESTIMATE, BASIC_CHECKS_ONLY_ESTIMATE, CORE_ALL_TEST_TYPES_MINUTES, + MYPY_PROVIDERS_ESTIMATE, PROVIDERS_ALL_LONGEST_JOB_MINUTES, PROVIDERS_ALL_MINUTES, RUN_OVERHEAD_MINUTES, + TS_SDK_DOCS_ESTIMATE, estimate_ci_minutes, estimate_core_job_minutes, estimate_providers_group, @@ -47,7 +49,9 @@ OPTIONAL_FLAGS = ( "run_java_sdk_tests", "run_go_sdk_tests", "run_ts_sdk_tests", + "run_ts_sdk_docs", "run_airflow_ctl_tests", + "run_mypy_providers", ) @@ -118,6 +122,32 @@ def test_estimate_basic_checks_only(): assert estimate.wall_clock_minutes == BASIC_CHECKS_ONLY_ESTIMATE[1] [email protected]( + ("flags", "expected_groups"), + [ + pytest.param( + {"run_ts_sdk_docs": True}, ["Basic checks only", "TypeScript SDK docs"], id="ts-sdk docs" + ), + pytest.param( + {"run_java_sdk_tests": True, "run_go_sdk_tests": True, "run_airflow_ctl_tests": True}, + ["Basic checks only", "Java SDK tests", "Go SDK tests", "Airflow CTL tests"], + id="sdk tests", + ), + ], +) +def test_estimate_basic_checks_only_counts_jobs_that_need_no_ci_image(flags, expected_groups): + estimate = estimate_ci_minutes(_selective_checks(basic_checks_only=True, **flags)) + + assert [group.name for group in estimate.job_groups] == expected_groups + + +def test_estimate_counts_ts_sdk_docs_with_the_ci_image(): + estimate = estimate_ci_minutes(_selective_checks(run_ts_sdk_docs=True)) + + assert estimate.job_groups[-1].name == "TypeScript SDK docs" + assert estimate.job_minutes == BASE_ESTIMATE[0] + TS_SDK_DOCS_ESTIMATE[0] + + @mock.patch("airflow_breeze.utils.ci_minutes_estimate.count_provider_test_files", return_value=100) def test_estimate_lists_the_job_groups_that_run(mock_count): estimate = estimate_ci_minutes( @@ -166,3 +196,34 @@ def test_estimate_without_unit_tests_has_only_base_group(): estimate = estimate_ci_minutes(_selective_checks()) assert estimate.job_minutes == BASE_ESTIMATE[0] + + [email protected]( + ("run_unit_tests", "expected_groups"), + [ + pytest.param( + False, ["CI image, basic tests and static checks", "Provider mypy checks"], id="metadata only" + ), + pytest.param( + True, + [ + "CI image, basic tests and static checks", + "Integration and system tests", + "Provider tests (1 providers, 100 test files)", + "Provider mypy checks", + ], + id="with unit tests", + ), + ], +) [email protected]("airflow_breeze.utils.ci_minutes_estimate.count_provider_test_files", return_value=100) +def test_estimate_charges_provider_tests_only_with_unit_tests(mock_count, run_unit_tests, expected_groups): + estimate = estimate_ci_minutes( + _selective_checks( + provider_test_types=["Providers[google]"], run_unit_tests=run_unit_tests, run_mypy_providers=True + ) + ) + + assert [group.name for group in estimate.job_groups] == expected_groups + if not run_unit_tests: + assert estimate.job_minutes == BASE_ESTIMATE[0] + MYPY_PROVIDERS_ESTIMATE[0] diff --git a/dev/breeze/tests/test_workflow_run_approval.py b/dev/breeze/tests/test_workflow_run_approval.py index ef6751a9c59..0877efa90d3 100644 --- a/dev/breeze/tests/test_workflow_run_approval.py +++ b/dev/breeze/tests/test_workflow_run_approval.py @@ -31,10 +31,12 @@ from airflow_breeze.utils.fork_ci_messages import CI_CHANGE_MARKER from airflow_breeze.utils.selective_checks import SelectiveChecks from airflow_breeze.utils.shared_options import set_dry_run from airflow_breeze.utils.workflow_run_approval import ( + GATE_OVERRIDE_LABELS, MAX_AUTO_APPROVED_JOB_MINUTES, MAX_AUTO_APPROVED_SOURCE_LINES, MAX_ERROR_TEXT_LENGTH, NEED_WORKFLOW_APPROVAL_LABEL, + TEST_SELECTION_LABELS, ApprovalDecision, ChangedFile, GitHubApi, @@ -360,6 +362,7 @@ def test_process_waiting_runs_groups_runs_by_pr_commit(mock_decide): api = mock.MagicMock(spec=GitHubApi) api.github_repository = "apache/airflow" api.get_last_base_change.return_value = None + api.get_last_label_change.return_value = None api.get_runs_waiting_for_approval.return_value = [ _run(1, "aaa"), _run(2, "aaa"), @@ -617,6 +620,7 @@ def test_process_waiting_runs_only_approves_runs_started_for_the_current_pr( ] api.get_open_pull_requests.return_value = [_pr(10, "aaa")] api.get_last_base_change.return_value = last_base_change + api.get_last_label_change.return_value = None mock_decide.return_value = ApprovalDecision(approve=True, reasons=["cheap"]) results = process_waiting_runs(api, created_since=CREATED_SINCE) @@ -655,6 +659,7 @@ def test_process_waiting_runs_continues_after_a_pull_request_fails(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"), _run(2, "bbb")] api.get_open_pull_requests.return_value = [_pr(10, "aaa"), _pr(11, "bbb")] mock_decide.side_effect = [error, ApprovalDecision(approve=True, reasons=["cheap"])] @@ -674,6 +679,7 @@ def test_process_waiting_runs_records_a_failure_while_approving_as_not_approved( 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"), _run(2, "aaa")] api.get_open_pull_requests.return_value = [_pr(10, "aaa")] api.approve_run.side_effect = [None, requests.HTTPError("409 Conflict")] @@ -692,6 +698,7 @@ def _api_with_prs(prs, runs): 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 = runs api.get_open_pull_requests.return_value = prs return api @@ -927,3 +934,111 @@ def test_has_bot_comment_only_counts_the_workflow_comments(login, expected): comments = [{"user": {"login": login}, "body": f"{CI_CHANGE_MARKER}\n\ntext"}] with mock.patch.object(api, "_get_paginated", autospec=True, return_value=comments): assert api.has_bot_comment(10, CI_CHANGE_MARKER) is expected + + [email protected]( + ("decision_labels", "approved_runs"), + [ + pytest.param(frozenset({"full tests needed"}), [3], id="runs started before the change"), + pytest.param(frozenset(), [2, 3], id="no change of the decision's labels"), + ], +) [email protected](f"{MODULE}.decide_for_pull_request", autospec=True) +def test_process_waiting_runs_skips_runs_started_before_a_test_selection_label_changed( + mock_decide, decision_labels, approved_runs +): + api = mock.MagicMock(spec=GitHubApi) + api.github_repository = "apache/airflow" + api.get_runs_waiting_for_approval.return_value = [ + _run(2, "aaa", created_at="2026-10-22T10:00:00Z"), + _run(3, "aaa", created_at="2026-10-22T11:00:00Z"), + ] + api.get_open_pull_requests.return_value = [_pr(10, "aaa")] + api.get_last_base_change.return_value = None + api.get_last_label_change.side_effect = lambda number, labels: ( + datetime(2026, 10, 22, 10, 30, tzinfo=UTC) if labels else None + ) + mock_decide.return_value = ApprovalDecision( + approve=True, reasons=["cheap"], decision_labels=decision_labels + ) + + process_waiting_runs(api, created_since=CREATED_SINCE) + + assert api.approve_run.call_args_list == [mock.call(run_id) for run_id in approved_runs] + + +def test_get_last_label_change_ignores_other_labels_and_reads_the_timeline_once(): + api = GitHubApi(github_repository="apache/airflow", github_token="token", report_only=True) + events = [ + {"event": "unlabeled", "created_at": "2026-10-22T09:00:00Z", "label": {"name": "full tests needed"}}, + { + "event": "unlabeled", + "created_at": "2026-10-22T11:00:00Z", + "label": {"name": NEED_WORKFLOW_APPROVAL_LABEL}, + }, + {"event": "labeled", "created_at": "2026-10-22T11:30:00Z", "label": {"name": "area:UI"}}, + { + "event": "unlabeled", + "created_at": "2026-10-22T11:45:00Z", + "label": {"name": "use public runners"}, + }, + {"event": "base_ref_changed", "created_at": "2026-10-22T08:00:00Z"}, + ] + with mock.patch.object(api, "_get_paginated", autospec=True, return_value=events) as get_paginated: + assert api.get_last_label_change(10, TEST_SELECTION_LABELS) == datetime( + 2026, 10, 22, 9, 0, tzinfo=UTC + ) + assert api.get_last_base_change(10) == datetime(2026, 10, 22, 8, 0, tzinfo=UTC) + get_paginated.assert_called_once() + + [email protected](f"{MODULE}.load_config", autospec=True, return_value=GATE_ENABLED) [email protected](f"{MODULE}.get_changed_files", autospec=True, return_value=CORE_CHANGE) [email protected](f"{MODULE}.fetch_merge_commit", autospec=True, return_value="merge") +def test_gated_approval_depends_on_the_gate_override_labels( + mock_fetch, mock_get_changed_files, mock_load_config +): + api = mock.MagicMock(spec=GitHubApi) + api.github_repository = "apache/airflow" + + assert decide_for_pull_request(api=api, pr=_gated_pr()).decision_labels == GATE_OVERRIDE_LABELS + + [email protected]("event", ["labeled", "unlabeled"]) [email protected]("label", sorted(TEST_SELECTION_LABELS)) +def test_get_last_label_change_counts_every_test_selection_label(label, event): + api = GitHubApi(github_repository="apache/airflow", github_token="token", report_only=True) + events = [{"event": event, "created_at": "2026-10-22T09:00:00Z", "label": {"name": label}}] + with mock.patch.object(api, "_get_paginated", autospec=True, return_value=events): + assert api.get_last_label_change(10, TEST_SELECTION_LABELS) == datetime( + 2026, 10, 22, 9, 0, tzinfo=UTC + ) + + [email protected](f"{MODULE}.decide_for_pull_request", autospec=True) +def test_process_waiting_runs_leaves_gated_runs_started_before_use_fork_ci_was_added(mock_decide): + api = GitHubApi(github_repository="apache/airflow", github_token="token", report_only=False) + timeline = [ + {"event": "labeled", "created_at": "2026-10-22T10:30:00Z", "label": {"name": "use fork ci"}}, + {"event": "unlabeled", "created_at": "2026-10-22T10:40:00Z", "label": {"name": "full tests needed"}}, + ] + mock_decide.return_value = ApprovalDecision( + approve=True, reasons=["fork CI"], decision_labels=GATE_OVERRIDE_LABELS + ) + with ( + mock.patch.object( + api, + "get_runs_waiting_for_approval", + autospec=True, + return_value=[ + _run(2, "aaa", created_at="2026-10-22T10:00:00Z"), + _run(3, "aaa", created_at="2026-10-22T11:00:00Z"), + ], + ), + mock.patch.object(api, "get_open_pull_requests", autospec=True, return_value=[_pr(10, "aaa")]), + mock.patch.object(api, "_get_paginated", autospec=True, return_value=timeline), + mock.patch.object(api, "approve_run", autospec=True) as approve_run, + ): + process_waiting_runs(api, created_since=CREATED_SINCE) + + assert approve_run.call_args_list == [mock.call(3)] diff --git a/dev/calibrate_ci_minutes_estimates.py b/dev/calibrate_ci_minutes_estimates.py index c6d3cc14283..eac8583b686 100644 --- a/dev/calibrate_ci_minutes_estimates.py +++ b/dev/calibrate_ci_minutes_estimates.py @@ -23,6 +23,7 @@ Measure the numbers behind ``dev/breeze/src/airflow_breeze/utils/ci_minutes_esti Downloads recent successful ``Tests (AMD)`` PR runs, their job timings and the selective-check outputs their ``Build info`` job printed, then prints the medians and fits to copy into the estimate module. +Only runs on public runners with the default versions are counted, as the estimate models them. Downloads are cached in ``--cache-dir``, so re-runs only fetch new runs. Needs an authenticated ``gh`` CLI. Run from the repository root: @@ -45,16 +46,17 @@ from pathlib import Path REPOSITORY = "apache/airflow" WORKFLOW = "ci-amd.yml" ANSI_ESCAPE = re.compile(r"\x1b\[[0-9;]*m") +# What ci_minutes_estimate.py models: jobs on GitHub's public AMD runners, default versions only. +# Self-hosted (RunsOn) runs and all-versions runs take different times and would skew the medians. +PUBLIC_AMD_RUNNERS = '["ubuntu-22.04"]' # Job-name patterns for each job group of ci_minutes_estimate.py; the first match wins. JOB_GROUPS = [ ("docs", r"^CI image checks / (Build documentation|Publish documentation)"), ("core", r"^(MySQL|Postgres|Sqlite|Non-DB) tests: core|^Low dep tests:core"), ("special", r"^Special tests"), - ( - "providers", - r"^Provider distributions tests|tests: providers|^Low dep tests: providers|^MyPy providers checks", - ), + ("mypy-providers", r"^MyPy providers checks"), + ("providers", r"^Provider distributions tests|tests: providers|^Low dep tests: providers"), ("prod", r"^Build PROD images|^Additional PROD image tests"), ("kubernetes", r"^Kubernetes tests"), ("helm", r"^Helm tests"), @@ -64,6 +66,7 @@ JOB_GROUPS = [ ("agent-framework", r"^Agent framework tests"), ("java-sdk", r"^Java SDK"), ("go-sdk", r"^Go SDK"), + ("ts-sdk-docs", r"^TypeScript SDK docs"), ("ts-sdk", r"^TypeScript SDK"), ("airflow-ctl", r"^Airflow CTL"), ("base", r".*"), @@ -151,6 +154,15 @@ def count_provider_test_files(provider_id: str) -> int: return sum(1 for _ in tests_path.rglob("test_*.py")) +def get_unmodelled_reason(outputs: dict[str, str]) -> str | None: + """Why a run cannot calibrate the estimate, or None when it matches what the estimate models.""" + if outputs.get("runner-type") != PUBLIC_AMD_RUNNERS: + return f"it ran on {outputs.get('runner-type', 'unknown runners')}" + if outputs.get("all-versions") != "false": + return "it ran all Python and database versions" + return None + + def build_rows(run_ids: list[int], cache_dir: Path) -> list[dict]: rows = [] for run_id in run_ids: @@ -161,6 +173,10 @@ def build_rows(run_ids: list[int], cache_dir: Path) -> list[dict]: outputs = read_outputs(log_file) if "full-tests-needed" not in outputs: continue + reason = get_unmodelled_reason(outputs) + if reason is not None: + print(f"skipping run {run_id}: {reason}") + continue minutes: dict[str, float] = defaultdict(float) longest: dict[str, float] = defaultdict(float) for job in json.loads(jobs_file.read_text()):
