Fix selective tests checks when system tests are modified

When providers system tests were modified in provider tests were
modified, selective checks had two bugs that compounded led to provider tests
being skipped.

1) Modification of system tests lead to "all providers affected"
   condition (wrongly) - because "TESTS" were checked as root path
   before "SYSTEM TESTS" - and since system tests were subfolder
   of tests, the code that checked if system tests path belongs
   to provider never found it.

2) "All Providers affected" in such case (when it was not caused by
   another condition lead to "skip-provider-tests" set to true due
   to missing "else" in one of the conditions.

This PR fixes both cases and simplifies the conditions so that they are
easier to understand and modify. Those conditions will be further
simplified when we separate providers to separate projects #44511
because there we will not have to check separately several sub-folders
of airflow, but for now it's good enough.

Some of those conditions are simplified as well because we are in
Python 3.9+ and "is_relative_to" is now available in Path.

dev/breeze/src/airflow_breeze/utils/selective_checks.py

@@ -326,30 +326,28 @@ def __hash__(self):
 
 def find_provider_affected(changed_file: str, include_docs: bool) -> str | None:
     file_path = AIRFLOW_SOURCES_ROOT / changed_file
-    # is_relative_to is only available in Python 3.9 - we should simplify this check when we are Python 3.9+
-    for provider_root in (TESTS_PROVIDERS_ROOT, SYSTEM_TESTS_PROVIDERS_ROOT, AIRFLOW_PROVIDERS_NS_PACKAGE):
-        try:
-            file_path.relative_to(provider_root)
-            relative_base_path = provider_root
+    # Check providers in SRC/SYSTEM_TESTS/TESTS/(optionally) DOCS
+    for provider_root in (SYSTEM_TESTS_PROVIDERS_ROOT, TESTS_PROVIDERS_ROOT, AIRFLOW_PROVIDERS_NS_PACKAGE):
+        if file_path.is_relative_to(provider_root):
+            provider_base_path = provider_root
             break
-        except ValueError:
-            pass
     else:
-        if include_docs:
-            try:
-                relative_path = file_path.relative_to(DOCS_DIR)
-                if relative_path.parts[0].startswith("apache-airflow-providers-"):
-                    return relative_path.parts[0].replace("apache-airflow-providers-", "").replace("-", ".")
-            except ValueError:
-                pass
+        if include_docs and file_path.is_relative_to(DOCS_DIR):
+            relative_path = file_path.relative_to(DOCS_DIR)
+            if relative_path.parts[0].startswith("apache-airflow-providers-"):
+                return relative_path.parts[0].replace("apache-airflow-providers-", "").replace("-", ".")
         return None
 
+    # Find if the path under src/system tests/tests belongs to provider or is a common code across
+    # multiple providers
     for parent_dir_path in file_path.parents:
-        if parent_dir_path == relative_base_path:
+        if parent_dir_path == provider_base_path:
+            # We have not found any provider specific path up to the root of the provider base folder
             break
-        relative_path = parent_dir_path.relative_to(relative_base_path)
+        relative_path = parent_dir_path.relative_to(provider_base_path)
+        # check if this path belongs to a specific provider
         if (AIRFLOW_PROVIDERS_NS_PACKAGE / relative_path / "provider.yaml").exists():
-            return str(parent_dir_path.relative_to(relative_base_path)).replace(os.sep, ".")
+            return str(parent_dir_path.relative_to(provider_base_path)).replace(os.sep, ".")
     # If we got here it means that some "common" files were modified. so we need to test all Providers
     return "Providers"
 
@@ -852,14 +850,13 @@ def _get_core_test_types_to_run(self) -> list[str]:
             )
         # sort according to predefined order
         sorted_candidate_test_types = sorted(candidate_test_types)
-        get_console().print("[warning]Selected test type candidates to run:[/]")
+        get_console().print("[warning]Selected core test type candidates to run:[/]")
         get_console().print(sorted_candidate_test_types)
         return sorted_candidate_test_types
 
     def _get_providers_test_types_to_run(self, split_to_individual_providers: bool = False) -> list[str]:
         if self._default_branch != "main":
             return []
-        affected_providers: AllProvidersSentinel | list[str] | None = ALL_PROVIDERS_SENTINEL
         if self.full_tests_needed:
             if split_to_individual_providers:
                 return list(providers_test_type())
@@ -885,18 +882,20 @@ def _get_providers_test_types_to_run(self, split_to_individual_providers: bool =
                     include_docs=False,
                 )
         candidate_test_types: set[str] = set()
-        if affected_providers and not isinstance(affected_providers, AllProvidersSentinel):
+        if isinstance(affected_providers, AllProvidersSentinel):
+            if split_to_individual_providers:
+                for provider in get_available_packages():
+                    candidate_test_types.add(f"Providers[{provider}]")
+            else:
+                candidate_test_types.add("Providers")
+        elif affected_providers:
             if split_to_individual_providers:
                 for provider in affected_providers:
                     candidate_test_types.add(f"Providers[{provider}]")
             else:
                 candidate_test_types.add(f"Providers[{','.join(sorted(affected_providers))}]")
-        elif split_to_individual_providers:
-            candidate_test_types.discard("Providers")
-            for provider in get_available_packages():
-                candidate_test_types.add(f"Providers[{provider}]")
         sorted_candidate_test_types = sorted(candidate_test_types)
-        get_console().print("[warning]Selected test type candidates to run:[/]")
+        get_console().print("[warning]Selected providers test type candidates to run:[/]")
         get_console().print(sorted_candidate_test_types)
         return sorted_candidate_test_types
 
@@ -1442,7 +1441,7 @@ def is_committer_build(self):
         return self._github_actor in COMMITTERS
 
     def _find_all_providers_affected(self, include_docs: bool) -> list[str] | AllProvidersSentinel | None:
-        all_providers: set[str] = set()
+        affected_providers: set[str] = set()
 
         all_providers_affected = False
         suspended_providers: set[str] = set()
@@ -1454,11 +1453,11 @@ def _find_all_providers_affected(self, include_docs: bool) -> list[str] | AllPro
                 if provider not in DEPENDENCIES:
                     suspended_providers.add(provider)
                 else:
-                    all_providers.add(provider)
+                    affected_providers.add(provider)
         if self.needs_api_tests:
-            all_providers.add("fab")
+            affected_providers.add("fab")
         if self.needs_ol_tests:
-            all_providers.add("openlineage")
+            affected_providers.add("openlineage")
         if all_providers_affected:
             return ALL_PROVIDERS_SENTINEL
         if suspended_providers:
@@ -1490,14 +1489,14 @@ def _find_all_providers_affected(self, include_docs: bool) -> list[str] | AllPro
                 get_console().print(
                     "[info]This PR had `allow suspended provider changes` label set so it will continue"
                 )
-        if not all_providers:
+        if not affected_providers:
             return None
 
-        for provider in list(all_providers):
-            all_providers.update(
+        for provider in list(affected_providers):
+            affected_providers.update(
                 get_related_providers(provider, upstream_dependencies=True, downstream_dependencies=True)
             )
-        return sorted(all_providers)
+        return sorted(affected_providers)
 
     def _is_canary_run(self):
         return (

dev/breeze/tests/test_selective_checks.py

@@ -369,6 +369,36 @@ def assert_outputs_are_printed(expected_outputs: dict[str, str], stderr: str):
                 id="Selected Providers and docs should run",
             )
         ),
+        (
+            pytest.param(
+                ("providers/tests/system/apache/beam/file.py",),
+                {
+                    "selected-providers-list-as-string": "apache.beam common.compat google",
+                    "all-python-versions": "['3.9']",
+                    "all-python-versions-list-as-string": "3.9",
+                    "python-versions": "['3.9']",
+                    "python-versions-list-as-string": "3.9",
+                    "ci-image-build": "true",
+                    "prod-image-build": "false",
+                    "needs-helm-tests": "false",
+                    "run-tests": "true",
+                    "run-amazon-tests": "false",
+                    "docs-build": "false",
+                    "skip-pre-commits": (
+                        "identity,lint-helm-chart,mypy-airflow,mypy-dev,mypy-docs"
+                        ",mypy-providers,mypy-task-sdk,ts-compile-format-lint-ui,ts-compile-format-lint-www"
+                    ),
+                    "run-kubernetes-tests": "false",
+                    "upgrade-to-newer-dependencies": "false",
+                    "core-test-types-list-as-string": "Always",
+                    "providers-test-types-list-as-string": "Providers[apache.beam,common.compat] Providers[google]",
+                    "individual-providers-test-types-list-as-string": "Providers[apache.beam] Providers[common.compat] Providers[google]",
+                    "needs-mypy": "true",
+                    "mypy-checks": "['mypy-providers']",
+                },
+                id="Selected Providers and docs should run when system tests are modified",
+            )
+        ),
         (
             pytest.param(
                 ("docs/file.rst",),