This is an automated email from the ASF dual-hosted git repository.
shahar1 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 11d25a587ac Run leaking provider tests with providers changed after
them in CI (#74123)
11d25a587ac is described below
commit 11d25a587ac57204170726d5f10fe9caadfb9746
Author: Shahar Epstein <[email protected]>
AuthorDate: Sat Oct 3 15:29:11 2026 +0300
Run leaking provider tests with providers changed after them in CI (#74123)
A PR that only changes a provider runs that provider's tests, with its
direct dependents, in a pytest process of its own. On canary, the same
tests run inside the shared Providers[-amazon,celery,google,standard]
process, after every provider that sorts before them. cncf.kubernetes
tests reload airflow.executors.executor_loader and leave its state
behind, so a test that mocks ExecutorLoader through the module path can
pass in its PR and fail on every canary build and release-candidate
check afterwards. PR #73832 did exactly that with an edge3 test.
Selective checks now add the providers listed in
PROVIDERS_WITH_PROCESS_GLOBAL_TEST_SIDE_EFFECTS to the Providers[...]
test type whenever a provider whose own files changed sorts after them,
so the PR runs those tests in the order canary does. Dependents pulled
in only for coverage do not widen the selection, and the CI job
description keeps naming the selected providers.
---
dev/breeze/doc/ci/04_selective_checks.md | 12 +++
.../src/airflow_breeze/utils/selective_checks.py | 74 +++++++++++++--
dev/breeze/tests/test_selective_checks.py | 102 ++++++++++++++++++++-
3 files changed, 176 insertions(+), 12 deletions(-)
diff --git a/dev/breeze/doc/ci/04_selective_checks.md
b/dev/breeze/doc/ci/04_selective_checks.md
index 2381870a2da..f4f0ce8badf 100644
--- a/dev/breeze/doc/ci/04_selective_checks.md
+++ b/dev/breeze/doc/ci/04_selective_checks.md
@@ -224,6 +224,14 @@ When unit tests run, selective checks narrow *which* test
types execute, separat
selects the test files that import it (directly or through other helpers) as
if those files had
changed. Suspended providers are excluded (and
a PR that touches one fails unless it carries the `allow suspended provider
changes` label).
+ Providers whose DB tests leave process-global state behind
(`PROVIDERS_WITH_PROCESS_GLOBAL_TEST_SIDE_EFFECTS`,
+ currently `cncf.kubernetes`) are added to the `Providers[...]` test type
when a provider whose own files
+ changed sorts after them: on canary the DB tests of
`Providers[-amazon,celery,google,standard]` run in
+ one pytest process with provider test folders in sorted order, so the
changed provider's tests run after
+ the leaked state is in place. Running them together in the PR surfaces a
test that depends on clean
+ state in the PR instead of after merge; a new leak added on the
`cncf.kubernetes` side is not caught
+ this way. Dependents pulled in only for coverage do not trigger this. The
individually-listed test types
+ run each provider on its own and are not widened, and the job description
names the selected providers.
The same matched-file approach drives the **prek hook skip list**
(`skip_prek_hooks`): each mypy /
compile / lint hook is skipped when nothing in its area changed. See
@@ -451,6 +459,10 @@ together using `pytest-xdist` (pytest-xdist distributes
the tests among parallel
of affected providers (but not recursively - only direct dependencies are
added)
* if there are any changes to "common" provider code not belonging to any
provider (usually system tests
or tests), then tests for all Providers are run
+ * if a provider whose own files changed sorts after a provider with
process-global DB-test side effects
+ (`PROVIDERS_WITH_PROCESS_GLOBAL_TEST_SIDE_EFFECTS`, currently
`cncf.kubernetes`), the side-effect
+ provider is added to the same `Providers[...]` test type, because the two
share one pytest process on
+ canary and the changed tests must pass after it
* `Java SDK E2E tests` (the `java_sdk` mode of the deployed-stack tests,
exposed as the
`run-java-sdk-e2e-tests` output) run when the Java SDK sources (`java-sdk/`,
excluding `.md`), the
Java test-fixture bundle (`airflow-e2e-tests/java-test-bundle/`), the Java
e2e suite or its Docker
diff --git a/dev/breeze/src/airflow_breeze/utils/selective_checks.py
b/dev/breeze/src/airflow_breeze/utils/selective_checks.py
index d8241e0040a..5d4f268cc59 100644
--- a/dev/breeze/src/airflow_breeze/utils/selective_checks.py
+++ b/dev/breeze/src/airflow_breeze/utils/selective_checks.py
@@ -96,6 +96,19 @@ ALLOW_PROVIDER_DEPENDENCY_BUMP_LABEL = "allow provider
dependency bump"
SKIP_COMMON_COMPAT_CHECK_LABEL = "skip common compat check"
AREA_E2E_TESTS_LABEL = "area:e2e-tests"
AREA_KUBERNETES_TESTS_LABEL = "area:kubernetes-tests"
+
+# Providers split into their own test type, see _extract_long_provider_tests.
Every other provider runs
+# in one shared ``Providers[-amazon,celery,google,standard]`` test type on
canary builds.
+LONG_RUNNING_TEST_PROVIDERS = ["amazon", "celery", "google", "standard"]
+# Providers whose DB tests leave process-global core state behind
(``importlib.reload()`` of
+# ``airflow.executors.executor_loader``, which swaps the ``ExecutorLoader``
class object and refills its
+# module-level caches). DB tests of one test type run in a single pytest
process, with provider test
+# folders in sorted order, so every provider sorting after one of these
inherits that state on canary.
+# A PR that changes such a later provider runs these providers in the same
test type too, otherwise
+# isolation failures in the changed tests surface only after merge. Non-DB
tests share one xdist pool
+# across all test types, so no deterministic order exists there to reproduce.
+PROVIDERS_WITH_PROCESS_GLOBAL_TEST_SIDE_EFFECTS = ["cncf.kubernetes"]
+
ALL_CI_SELECTIVE_TEST_TYPES = "API Always CLI Core Other Serialization"
ALL_PROVIDERS_SELECTIVE_TEST_TYPES = (
@@ -820,11 +833,20 @@ def _split_list(input_list, n) -> list[list[str]]:
]
+def _strip_test_side_effect_providers(test_type: str) -> str:
+ """Drop the test side-effect providers so the description names the
providers selected for the change."""
+ if not test_type.startswith("Providers[") or
test_type.startswith("Providers[-"):
+ return test_type
+ providers =
test_type.removeprefix("Providers[").removesuffix("]").split(",")
+ selected = [p for p in providers if p not in
PROVIDERS_WITH_PROCESS_GLOBAL_TEST_SIDE_EFFECTS]
+ return ",".join(selected or providers)
+
+
def _get_test_type_description(provider_test_types: list[str]) -> str:
if not provider_test_types:
return ""
- first_provider = provider_test_types[0]
- last_provider = provider_test_types[-1]
+ first_provider = _strip_test_side_effect_providers(provider_test_types[0])
+ last_provider = _strip_test_side_effect_providers(provider_test_types[-1])
if first_provider.startswith("Providers["):
first_provider = first_provider.replace("Providers[", "").replace("]",
"")
if last_provider.startswith("Providers["):
@@ -1563,12 +1585,51 @@ class SelectiveChecks:
for provider in providers_to_test:
candidate_test_types.add(f"Providers[{provider}]")
else:
+ providers_to_test =
self._add_providers_sharing_test_process_state(
+ providers_to_test,
changed_providers=self._find_changed_providers(), suspended=suspended
+ )
candidate_test_types.add(f"Providers[{','.join(sorted(providers_to_test))}]")
sorted_candidate_test_types = sorted(candidate_test_types)
console_print("[warning]Selected providers test type candidates to
run:[/]")
console_print(sorted_candidate_test_types)
return sorted_candidate_test_types
+ def _find_changed_providers(self) -> set[str]:
+ """Providers whose own files changed, without their upstream and
downstream dependents."""
+ return {
+ provider
+ for changed_file in self._files
+ if (provider := find_provider_affected(changed_file,
include_docs=False))
+ not in (None, "Providers")
+ }
+
+ @staticmethod
+ def _add_providers_sharing_test_process_state(
+ providers_to_test: list[str], *, changed_providers: set[str],
suspended: set[str]
+ ) -> list[str]:
+ """
+ Add the providers from PROVIDERS_WITH_PROCESS_GLOBAL_TEST_SIDE_EFFECTS
that precede a changed one.
+
+ Only providers whose own files changed count: a PR can make a
provider's tests order-sensitive only
+ by changing that provider, and dependents pulled in for coverage keep
their canary behaviour.
+ """
+
+ def get_test_folder(provider_id: str) -> str:
+ return provider_id.replace(".", "/")
+
+ changed_shared_process_providers = [
+ p for p in providers_to_test if p in changed_providers and p not
in LONG_RUNNING_TEST_PROVIDERS
+ ]
+ if not changed_shared_process_providers:
+ return providers_to_test
+ last_folder = max(get_test_folder(p) for p in
changed_shared_process_providers)
+ leaking_providers = [
+ p
+ for p in PROVIDERS_WITH_PROCESS_GLOBAL_TEST_SIDE_EFFECTS
+ if get_test_folder(p) < last_folder and p not in providers_to_test
and p not in suspended
+ ]
+ return sorted([*providers_to_test, *leaking_providers])
+
@staticmethod
def _extract_long_provider_tests(current_test_types: set[str]):
"""
@@ -1584,20 +1645,19 @@ class SelectiveChecks:
:param current_test_types: The set of test types to run
"""
- long_tests = ["amazon", "celery", "google", "standard"]
for original_test_type in tuple(current_test_types):
if original_test_type == "Providers":
current_test_types.remove(original_test_type)
- for long_test in long_tests:
+ for long_test in LONG_RUNNING_TEST_PROVIDERS:
current_test_types.add(f"Providers[{long_test}]")
- current_test_types.add(f"Providers[-{','.join(long_tests)}]")
+
current_test_types.add(f"Providers[-{','.join(LONG_RUNNING_TEST_PROVIDERS)}]")
elif original_test_type.startswith("Providers["):
provider_tests_to_run = (
original_test_type.replace("Providers[", "").replace("]",
"").split(",")
)
- if any(long_test in provider_tests_to_run for long_test in
long_tests):
+ if any(long_test in provider_tests_to_run for long_test in
LONG_RUNNING_TEST_PROVIDERS):
current_test_types.remove(original_test_type)
- for long_test in long_tests:
+ for long_test in LONG_RUNNING_TEST_PROVIDERS:
if long_test in provider_tests_to_run:
current_test_types.add(f"Providers[{long_test}]")
provider_tests_to_run.remove(long_test)
diff --git a/dev/breeze/tests/test_selective_checks.py
b/dev/breeze/tests/test_selective_checks.py
index a115a2637a2..8c7d74079a7 100644
--- a/dev/breeze/tests/test_selective_checks.py
+++ b/dev/breeze/tests/test_selective_checks.py
@@ -45,9 +45,12 @@ from airflow_breeze.utils.path_utils import AIRFLOW_ROOT_PATH
from airflow_breeze.utils.provider_dependencies import
get_provider_dependencies
from airflow_breeze.utils.selective_checks import (
ALL_CI_SELECTIVE_TEST_TYPES,
+ LONG_RUNNING_TEST_PROVIDERS,
+ PROVIDERS_WITH_PROCESS_GLOBAL_TEST_SIDE_EFFECTS,
SelectiveChecks,
_find_test_helper_importers,
_get_test_list_as_json,
+ _get_test_type_description,
_imports_module,
_split_list,
)
@@ -544,7 +547,7 @@ def assert_outputs_are_printed(expected_outputs: dict[str,
str], stderr: str):
{
"description": "amazon...google",
"test_types": "Providers[amazon] "
-
"Providers[common.compat,common.sql,fab,microsoft.azure,openlineage,pgvector,postgres]
"
+
"Providers[cncf.kubernetes,common.compat,common.sql,fab,microsoft.azure,openlineage,pgvector,postgres]
"
"Providers[google]",
}
]
@@ -964,7 +967,7 @@ def assert_outputs_are_printed(expected_outputs: dict[str,
str], stderr: str):
{
"description": "amazon...google",
"test_types": "Providers[amazon] "
-
"Providers[common.compat,common.sql,microsoft.azure,openlineage,pgvector,postgres]
"
+
"Providers[cncf.kubernetes,common.compat,common.sql,microsoft.azure,openlineage,pgvector,postgres]
"
"Providers[google]",
}
]
@@ -1004,7 +1007,7 @@ def assert_outputs_are_printed(expected_outputs:
dict[str, str], stderr: str):
[
{
"description": "amazon...google",
- "test_types": "Providers[amazon]
Providers[apache.livy,atlassian.jira,common.compat,dbt.cloud,dingding,discord,http,informatica,pagerduty]
Providers[google]",
+ "test_types": "Providers[amazon]
Providers[apache.livy,atlassian.jira,cncf.kubernetes,common.compat,dbt.cloud,dingding,discord,http,informatica,pagerduty]
Providers[google]",
}
]
),
@@ -1283,7 +1286,7 @@ def assert_outputs_are_printed(expected_outputs:
dict[str, str], stderr: str):
[
{
"description": "common.compat,common.io,openl",
- "test_types":
"Providers[common.compat,common.io,openlineage]",
+ "test_types":
"Providers[cncf.kubernetes,common.compat,common.io,openlineage]",
}
]
),
@@ -1385,7 +1388,12 @@ def assert_outputs_are_printed(expected_outputs:
dict[str, str], stderr: str):
"full-tests-needed": "false",
"run-unit-tests": "true",
"providers-test-types-list-as-strings-in-json": json.dumps(
- [{"description": "common.compat,fab", "test_types":
"Providers[common.compat,fab]"}]
+ [
+ {
+ "description": "common.compat,fab",
+ "test_types":
"Providers[cncf.kubernetes,common.compat,fab]",
+ }
+ ]
),
},
id="Only the tests importing a test helper should run when it
changes",
@@ -4483,3 +4491,87 @@ def
test_suspended_provider_check_ignores_test_helper_importers(
default_branch="main",
)
assert selective_checks.selected_providers_list_as_string is None
+
+
[email protected](
+ ("providers_to_test", "changed_providers", "suspended", "expected"),
+ [
+ pytest.param(
+ ["common.compat", "edge3"],
+ {"edge3"},
+ set(),
+ ["cncf.kubernetes", "common.compat", "edge3"],
+ id="changed provider runs after the leaking one in the shared
canary process",
+ ),
+ pytest.param(
+ ["apache.beam", "common.compat"],
+ {"apache.beam"},
+ set(),
+ ["apache.beam", "common.compat"],
+ id="only a dependent runs after the leaking one",
+ ),
+ pytest.param(
+ ["apache.beam"],
+ {"apache.beam"},
+ set(),
+ ["apache.beam"],
+ id="changed provider runs before the leaking one",
+ ),
+ pytest.param(
+ ["cncf.kubernetes", "edge3"],
+ {"edge3"},
+ set(),
+ ["cncf.kubernetes", "edge3"],
+ id="leaking provider already selected",
+ ),
+ pytest.param(
+ ["amazon", "common.compat"],
+ {"amazon"},
+ set(),
+ ["amazon", "common.compat"],
+ id="long running providers run in their own process",
+ ),
+ pytest.param(
+ ["amazon", "edge3"],
+ {"amazon", "edge3"},
+ set(),
+ ["amazon", "cncf.kubernetes", "edge3"],
+ id="only shared-process providers decide",
+ ),
+ pytest.param(
+ ["edge3"],
+ {"edge3"},
+ {"cncf.kubernetes"},
+ ["edge3"],
+ id="suspended leaking provider is not added",
+ ),
+ ],
+)
+def test_add_providers_sharing_test_process_state(providers_to_test,
changed_providers, suspended, expected):
+ assert (
+ SelectiveChecks._add_providers_sharing_test_process_state(
+ providers_to_test, changed_providers=changed_providers,
suspended=suspended
+ )
+ == expected
+ )
+
+
[email protected](
+ ("test_types", "expected"),
+ [
+ pytest.param(
+ ["Providers[cncf.kubernetes,common.compat,edge3]"],
"common.compat,edge3", id="stripped"
+ ),
+ pytest.param(["Providers[cncf.kubernetes]"], "cncf.kubernetes",
id="only the leaking provider"),
+ pytest.param(["Providers[-amazon,google]"], "-amazon,google",
id="excluded"),
+ pytest.param(["Providers[amazon]",
"Providers[cncf.kubernetes,edge3]"], "amazon...edge3", id="range"),
+ ],
+)
+def test_get_test_type_description_names_selected_providers(test_types,
expected):
+ assert _get_test_type_description(test_types) == expected
+
+
+def test_providers_with_test_side_effects_are_shared_process_providers():
+ for provider in PROVIDERS_WITH_PROCESS_GLOBAL_TEST_SIDE_EFFECTS:
+ assert provider in get_available_distributions()
+ assert provider not in LONG_RUNNING_TEST_PROVIDERS
