dry

script/analyze_component_buses.py

@@ -34,6 +34,8 @@ from typing import Any
 # Add esphome to path
 sys.path.insert(0, str(Path(__file__).parent.parent))
 
+from helpers import BASE_BUS_COMPONENTS
+
 from esphome import yaml_util
 from esphome.config_helpers import Extend, Remove
 
@@ -67,18 +69,6 @@ NO_BUSES_SIGNATURE = "no_buses"
 # Isolated components have unique signatures and cannot be merged with others
 ISOLATED_SIGNATURE_PREFIX = "isolated_"
 
-# Base bus components - these ARE the bus implementations and should not
-# be flagged as needing migration since they are the platform/base components
-BASE_BUS_COMPONENTS = {
-    "i2c",
-    "spi",
-    "uart",
-    "modbus",
-    "canbus",
-    "remote_transmitter",
-    "remote_receiver",
-}
-
 # Components that must be tested in isolation (not grouped or batched with others)
 # These have known build issues that prevent grouping
 # NOTE: This should be kept in sync with both test_build_components and split_components_for_ci.py

script/determine-jobs.py

@@ -38,6 +38,7 @@ Options:
 from __future__ import annotations
 
 import argparse
+from collections import Counter
 from enum import StrEnum
 from functools import cache
 import json
@@ -48,11 +49,13 @@ import sys
 from typing import Any
 
 from helpers import (
+    BASE_BUS_COMPONENTS,
     CPP_FILE_EXTENSIONS,
-    ESPHOME_COMPONENTS_PATH,
     PYTHON_FILE_EXTENSIONS,
     changed_files,
     get_all_dependencies,
+    get_component_from_path,
+    get_component_test_files,
     get_components_from_integration_fixtures,
     parse_test_filename,
     root_path,
@@ -142,11 +145,8 @@ def should_run_integration_tests(branch: str | None = None) -> bool:
 
     # Check if any required components changed
     for file in files:
-        if file.startswith(ESPHOME_COMPONENTS_PATH):
+        component = get_component_from_path(file)
-            parts = file.split("/")
+        if component and component in all_required_components:
-            if len(parts) >= 3:
-                component = parts[2]
-                if component in all_required_components:
             return True
 
     return False
@@ -261,10 +261,7 @@ def _component_has_tests(component: str) -> bool:
     Returns:
         True if the component has test YAML files
     """
-    tests_dir = Path(root_path) / "tests" / "components" / component
+    return bool(get_component_test_files(component))
-    if not tests_dir.exists():
-        return False
-    return any(tests_dir.glob("test.*.yaml"))
 
 
 def detect_memory_impact_config(
@@ -291,16 +288,14 @@ def detect_memory_impact_config(
     files = changed_files(branch)
 
     # Find all changed components (excluding core and base bus components)
-    changed_component_set = set()
+    changed_component_set: set[str] = set()
     has_core_changes = False
 
     for file in files:
-        if file.startswith(ESPHOME_COMPONENTS_PATH):
+        component = get_component_from_path(file)
-            parts = file.split("/")
+        if component:
-            if len(parts) >= 3:
-                component = parts[2]
             # Skip base bus components as they're used across many builds
-                if component not in ["i2c", "spi", "uart", "modbus", "canbus"]:
+            if component not in BASE_BUS_COMPONENTS:
                 changed_component_set.add(component)
         elif file.startswith("esphome/"):
             # Core ESPHome files changed (not component-specific)
@@ -321,25 +316,24 @@ def detect_memory_impact_config(
         return {"should_run": "false"}
 
     # Find components that have tests and collect their supported platforms
-    components_with_tests = []
+    components_with_tests: list[str] = []
-    component_platforms_map = {}  # Track which platforms each component supports
+    component_platforms_map: dict[
+        str, set[Platform]
+    ] = {}  # Track which platforms each component supports
 
     for component in sorted(changed_component_set):
-        tests_dir = Path(root_path) / "tests" / "components" / component
-        if not tests_dir.exists():
-            continue
-
         # Look for test files on preferred platforms
-        test_files = list(tests_dir.glob("test.*.yaml"))
+        test_files = get_component_test_files(component)
         if not test_files:
             continue
 
         # Check if component has tests for any preferred platform
-        available_platforms = []
+        available_platforms = [
-        for test_file in test_files:
+            platform
-            _, platform = parse_test_filename(test_file)
+            for test_file in test_files
-            if platform != "all" and platform in MEMORY_IMPACT_PLATFORM_PREFERENCE:
+            if (platform := parse_test_filename(test_file)[1]) != "all"
-                available_platforms.append(platform)
+            and platform in MEMORY_IMPACT_PLATFORM_PREFERENCE
+        ]
 
         if not available_platforms:
             continue
@@ -367,10 +361,10 @@ def detect_memory_impact_config(
     else:
         # No common platform - pick the most commonly supported platform
         # This allows testing components individually even if they can't be merged
-        platform_counts = {}
+        # Count how many components support each platform
-        for platforms in component_platforms_map.values():
+        platform_counts = Counter(
-            for p in platforms:
+            p for platforms in component_platforms_map.values() for p in platforms
-                platform_counts[p] = platform_counts.get(p, 0) + 1
+        )
         # Pick the platform supported by most components, preferring earlier in MEMORY_IMPACT_PLATFORM_PREFERENCE
         platform = max(
             platform_counts.keys(),

script/helpers.py

@@ -29,6 +29,18 @@ YAML_FILE_EXTENSIONS = (".yaml", ".yml")
 # Component path prefix
 ESPHOME_COMPONENTS_PATH = "esphome/components/"
 
+# Base bus components - these ARE the bus implementations and should not
+# be flagged as needing migration since they are the platform/base components
+BASE_BUS_COMPONENTS = {
+    "i2c",
+    "spi",
+    "uart",
+    "modbus",
+    "canbus",
+    "remote_transmitter",
+    "remote_receiver",
+}
+
 
 def parse_list_components_output(output: str) -> list[str]:
     """Parse the output from list-components.py script.
@@ -63,6 +75,48 @@ def parse_test_filename(test_file: Path) -> tuple[str, str]:
     return parts[0], "all"
 
 
+def get_component_from_path(file_path: str) -> str | None:
+    """Extract component name from a file path.
+
+    Args:
+        file_path: Path to a file (e.g., "esphome/components/wifi/wifi.cpp")
+
+    Returns:
+        Component name if path is in components directory, None otherwise
+    """
+    if not file_path.startswith(ESPHOME_COMPONENTS_PATH):
+        return None
+    parts = file_path.split("/")
+    if len(parts) >= 3:
+        return parts[2]
+    return None
+
+
+def get_component_test_files(
+    component: str, *, all_variants: bool = False
+) -> list[Path]:
+    """Get test files for a component.
+
+    Args:
+        component: Component name (e.g., "wifi")
+        all_variants: If True, returns all test files including variants (test-*.yaml).
+                     If False, returns only base test files (test.*.yaml).
+                     Default is False.
+
+    Returns:
+        List of test file paths for the component, or empty list if none exist
+    """
+    tests_dir = Path(root_path) / "tests" / "components" / component
+    if not tests_dir.exists():
+        return []
+
+    if all_variants:
+        # Match both test.*.yaml and test-*.yaml patterns
+        return list(tests_dir.glob("test[.-]*.yaml"))
+    # Match only test.*.yaml (base tests)
+    return list(tests_dir.glob("test.*.yaml"))
+
+
 def styled(color: str | tuple[str, ...], msg: str, reset: bool = True) -> str:
     prefix = "".join(color) if isinstance(color, tuple) else color
     suffix = colorama.Style.RESET_ALL if reset else ""
@@ -331,10 +385,8 @@ def _filter_changed_ci(files: list[str]) -> list[str]:
     # because changes in one file can affect other files in the same component.
     filtered_files = []
     for f in files:
-        if f.startswith(ESPHOME_COMPONENTS_PATH):
+        component = get_component_from_path(f)
-            # Check if file belongs to any of the changed components
+        if component and component in component_set:
-            parts = f.split("/")
-            if len(parts) >= 3 and parts[2] in component_set:
             filtered_files.append(f)
 
     return filtered_files

script/list-components.py

@@ -4,7 +4,7 @@ from collections.abc import Callable
 from pathlib import Path
 import sys
 
-from helpers import changed_files, git_ls_files
+from helpers import changed_files, get_component_from_path, git_ls_files
 
 from esphome.const import (
     KEY_CORE,
@@ -30,10 +30,8 @@ def get_all_component_files() -> list[str]:
 def extract_component_names_array_from_files_array(files):
     components = []
     for file in files:
-        file_parts = file.split("/")
+        component_name = get_component_from_path(file)
-        if len(file_parts) >= 4:
+        if component_name and component_name not in components:
-            component_name = file_parts[2]
-            if component_name not in components:
             components.append(component_name)
     return components
 

script/split_components_for_ci.py

@@ -28,6 +28,7 @@ from script.analyze_component_buses import (
     create_grouping_signature,
     merge_compatible_bus_groups,
 )
+from script.helpers import get_component_test_files
 
 # Weighting for batch creation
 # Isolated components can't be grouped/merged, so they count as 10x
@@ -45,17 +46,12 @@ def has_test_files(component_name: str, tests_dir: Path) -> bool:
 
     Args:
         component_name: Name of the component
-        tests_dir: Path to tests/components directory
+        tests_dir: Path to tests/components directory (unused, kept for compatibility)
 
     Returns:
         True if the component has test.*.yaml files
     """
-    component_dir = tests_dir / component_name
+    return bool(get_component_test_files(component_name))
-    if not component_dir.exists() or not component_dir.is_dir():
-        return False
-
-    # Check for test.*.yaml files
-    return any(component_dir.glob("test.*.yaml"))
 
 
 def create_intelligent_batches(

script/test_build_components.py

@@ -39,6 +39,7 @@ from script.analyze_component_buses import (
     merge_compatible_bus_groups,
     uses_local_file_references,
 )
+from script.helpers import get_component_test_files
 from script.merge_component_configs import merge_component_configs
 
 
@@ -100,10 +101,10 @@ def find_component_tests(
         if not comp_dir.is_dir():
             continue
 
-        # Find test files - either base only (test.*.yaml) or all (test[.-]*.yaml)
+        # Get test files using helper function
-        pattern = "test.*.yaml" if base_only else "test[.-]*.yaml"
+        test_files = get_component_test_files(comp_dir.name, all_variants=not base_only)
-        for test_file in comp_dir.glob(pattern):
+        if test_files:
-            component_tests[comp_dir.name].append(test_file)
+            component_tests[comp_dir.name] = test_files
 
     return dict(component_tests)