[ci] Only run integration tests for changed components (#14776)

script/helpers.py

@@ -700,37 +700,141 @@ def get_all_dependencies(
     return all_components
 
 
+def _extract_components_from_yaml(config: dict) -> set[str]:
+    """Extract component names from a parsed YAML config.
+
+    Args:
+        config: Parsed YAML configuration dictionary
+
+    Returns:
+        Set of component names found in the config
+    """
+    components: set[str] = set()
+
+    # Add all top-level component keys (skip YAML anchor keys starting with '.')
+    components.update(k for k in config if isinstance(k, str) and not k.startswith("."))
+
+    # Add platform values from list entries (e.g., sensor -> platform: template adds "template")
+    for value in config.values():
+        if isinstance(value, list):
+            components.update(
+                item["platform"]
+                for item in value
+                if isinstance(item, dict) and "platform" in item
+            )
+
+    return components
+
+
 def get_components_from_integration_fixtures() -> set[str]:
     """Extract all components used in integration test fixtures.
 
     Returns:
         Set of component names used in integration test fixtures
     """
+    return {
+        comp
+        for components in get_components_per_integration_fixture().values()
+        for comp in components
+    }
+
+
+@cache
+def get_components_per_integration_fixture() -> dict[str, set[str]]:
+    """Extract components used in each integration test fixture.
+
+    Returns:
+        Dictionary mapping fixture name (stem) to set of component names
+    """
     from esphome import yaml_util
 
-    components: set[str] = set()
+    result: dict[str, set[str]] = {}
     fixtures_dir = Path(__file__).parent.parent / "tests" / "integration" / "fixtures"
 
     for yaml_file in fixtures_dir.glob("*.yaml"):
-        config: dict[str, any] | None = yaml_util.load_yaml(yaml_file)
+        config: dict[str, Any] | None = yaml_util.load_yaml(yaml_file)
         if not config:
             continue
 
-        # Add all top-level component keys (skip YAML anchor keys starting with '.')
-        components.update(
-            k for k in config if isinstance(k, str) and not k.startswith(".")
-        )
+        result[yaml_file.stem] = _extract_components_from_yaml(config)
 
-        # Add platform components (e.g., output.template)
-        for value in config.values():
-            if not isinstance(value, list):
-                continue
+    return result
 
-            for item in value:
-                if isinstance(item, dict) and "platform" in item:
-                    components.add(item["platform"])
 
-    return components
+_TEST_FUNC_RE = re.compile(r"async def (test_\w+)")
+
+
+@cache
+def get_fixture_to_test_files() -> dict[str, frozenset[str]]:
+    """Map integration test fixture names to the test files that use them.
+
+    Returns:
+        Dictionary mapping fixture name to frozenset of test file paths
+        (relative to repo root)
+    """
+    integration_dir = Path(__file__).parent.parent / "tests" / "integration"
+    result: dict[str, set[str]] = {}
+
+    for test_file in integration_dir.glob("test_*.py"):
+        content = test_file.read_text(encoding="utf-8")
+        rel_path = test_file.relative_to(Path(__file__).parent.parent).as_posix()
+        for func in _TEST_FUNC_RE.findall(content):
+            base_name = func.replace("test_", "").partition("[")[0]
+            result.setdefault(base_name, set()).add(rel_path)
+
+    return {k: frozenset(v) for k, v in result.items()}
+
+
+@cache
+def _get_component_to_integration_test_files() -> dict[str, frozenset[str]]:
+    """Build index mapping each component to the test files that depend on it.
+
+    Resolves full dependency trees once per fixture, then inverts the mapping
+    so lookups are O(1) per component.
+
+    Returns:
+        Dictionary mapping component name to frozenset of test file paths
+    """
+    fixture_components = get_components_per_integration_fixture()
+    fixture_to_test_files = get_fixture_to_test_files()
+
+    result: dict[str, set[str]] = {}
+    for fixture_name, components in fixture_components.items():
+        test_files = fixture_to_test_files.get(fixture_name)
+        if not test_files:
+            continue
+        # Get full dependency tree for this fixture's components
+        all_deps = get_all_dependencies(components)
+        for dep in all_deps:
+            result.setdefault(dep, set()).update(test_files)
+
+    return {k: frozenset(v) for k, v in result.items()}
+
+
+def get_integration_test_files_for_components(
+    changed_components: set[str],
+) -> list[str]:
+    """Get integration test file paths that use any of the given components.
+
+    Uses a precomputed component → test files index for O(C) lookup
+    where C is the number of changed components.
+
+    Args:
+        changed_components: Set of component names that have changed
+
+    Returns:
+        Sorted list of test file paths relative to repo root
+        (e.g., ["tests/integration/test_api.py", ...])
+    """
+    component_to_tests = _get_component_to_integration_test_files()
+
+    return sorted(
+        {
+            test_file
+            for component in changed_components
+            for test_file in component_to_tests.get(component, ())
+        }
+    )
 
 
 def filter_component_and_test_files(file_path: str) -> bool: