Merge remote-tracking branch 'upstream/logger-buffered-recursion-guard' into integration

This commit is contained in:
J. Nick Koston
2026-06-18 16:12:45 -05:00
664 changed files with 22117 additions and 5672 deletions
+11 -5
View File
@@ -39,8 +39,13 @@ from helpers import BASE_BUS_COMPONENTS, is_validate_only_file
from esphome import yaml_util
from esphome.config_helpers import Extend, Remove
# Path to common bus configs
COMMON_BUS_PATH = Path("tests/test_build_components/common")
# Path to common bus configs (resolved relative to this file, not the CWD)
COMMON_BUS_PATH = (
Path(__file__).resolve().parent.parent
/ "tests"
/ "test_build_components"
/ "common"
)
# Package dependencies - maps packages to the packages they include
# When a component uses a package on the left, it automatically gets
@@ -59,6 +64,7 @@ DIRECT_BUS_TYPES = (
"modbus",
"remote_transmitter",
"remote_receiver",
"i2s_audio",
)
# Signature for components with no bus requirements
@@ -128,7 +134,7 @@ def uses_local_file_references(component_dir: Path) -> bool:
try:
content = common_yaml.read_text()
except Exception: # pylint: disable=broad-exception-caught
except Exception: # noqa: BLE001 # pylint: disable=broad-exception-caught
return False
# Pattern to match $component_dir or ${component_dir} references
@@ -164,7 +170,7 @@ def is_platform_component(component_dir: Path) -> bool:
try:
content = comp_init.read_text()
return "IS_PLATFORM_COMPONENT = True" in content
except Exception: # pylint: disable=broad-exception-caught
except Exception: # noqa: BLE001 # pylint: disable=broad-exception-caught
return False
@@ -222,7 +228,7 @@ def analyze_yaml_file(yaml_file: Path) -> dict[str, Any]:
try:
data = yaml_util.load_yaml(yaml_file)
result["loaded"] = True
except Exception: # pylint: disable=broad-exception-caught
except Exception: # noqa: BLE001 # pylint: disable=broad-exception-caught
return result
# Check for Extend/Remove objects
+1 -1
View File
@@ -392,7 +392,7 @@ def compile_and_get_binary(
if exit_code != 0:
print(f"Error compiling {label} for {', '.join(components)}")
return exit_code, None
except Exception as e:
except Exception as e: # noqa: BLE001
print(f"Error compiling {label} for {', '.join(components)}: {e}")
return EXIT_COMPILE_ERROR, None
+48 -2
View File
@@ -428,6 +428,33 @@ def fix_menu():
menu[S_EXTENDS].append("display_menu_base.MENU_TYPES")
def fix_lvgl_widgets():
# lvgl's `widgets:` is a recursive tree (a widget can contain widgets). The
# dumper has no cycle detection, so — like fix_menu — hoist the inlined
# widget-type enumeration into a named schema and reference it for both the
# top-level list and each widget's own children, instead of expanding it.
if "lvgl" not in output:
return
schemas = output["lvgl"][S_SCHEMAS]
config_vars = schemas["CONFIG_SCHEMA"][S_SCHEMA][S_CONFIG_VARS]
widgets = config_vars.get("widgets")
if not widgets or S_SCHEMA not in widgets or S_CONFIG_VARS not in widgets[S_SCHEMA]:
return
# 1. Hoist the (one-level) widget enumeration into a named schema.
schemas["WIDGET_TYPES"] = {S_TYPE: S_SCHEMA, S_SCHEMA: widgets[S_SCHEMA]}
# 2. Reference it from the top-level widgets: list instead of inlining.
widgets[S_SCHEMA] = {S_EXTENDS: ["lvgl.WIDGET_TYPES"]}
# 3. Let every widget contain child widgets, via the same named ref.
for widget in schemas["WIDGET_TYPES"][S_SCHEMA][S_CONFIG_VARS].values():
if widget.get(S_TYPE) == S_SCHEMA and S_SCHEMA in widget:
widget[S_SCHEMA].setdefault(S_CONFIG_VARS, {})["widgets"] = {
S_TYPE: S_SCHEMA,
"is_list": True,
"key": "Optional",
S_SCHEMA: {S_EXTENDS: ["lvgl.WIDGET_TYPES"]},
}
def get_logger_tags():
pattern = re.compile(r'^static const char \*const TAG = "(\w.*)";', re.MULTILINE)
# tags not in components dir
@@ -740,6 +767,7 @@ def build_schema():
add_logger_tags()
shrink()
fix_menu()
fix_lvgl_widgets()
# aggregate components, so all component info is in same file, otherwise we have dallas.json, dallas.sensor.json, etc.
data = {}
@@ -923,10 +951,24 @@ def convert(schema, config_var, path):
elif schema_type == "enum":
config_var[S_TYPE] = "enum"
config_var["values"] = dict.fromkeys(list(data.keys()))
elif schema_type == "variant_enum":
# Per-variant enum (e.g. psram mode/speed): each value carries the
# list of variants that accept it so clients can filter to the
# user's selected variant. Additive to the plain enum format —
# consumers that ignore the metadata still see every option.
config_var[S_TYPE] = "enum"
config_var["values"] = {
value: {"variants": variants} for value, variants in data.items()
}
elif schema_type == "maybe":
config_var[S_TYPE] = S_SCHEMA
# maybe_simple_value: either a scalar shorthand (mapped to the key in
# data[1]) or the full wrapped schema. The wrapped schema is usually a
# plain Schema (converts to a "schema" config var), but may be something
# else, e.g. a typed_schema (converts to a "typed" config var with
# "types" and no top-level "schema" key). Merge whatever it produced
# rather than assuming a "schema" key is present.
config_var["maybe"] = data[1]
config_var["schema"] = convert_config(data[0], path + "/maybe")["schema"]
config_var.update(convert_config(data[0], path + "/maybe"))
# esphome/on_boot
elif schema_type == "automation":
extra_schema = None
@@ -997,6 +1039,10 @@ def convert(schema, config_var, path):
else:
config_var["use_id_type"] = str(data.base)
config_var[S_TYPE] = "use_id"
elif schema_type == "schema":
# A callable CONFIG_SCHEMA that returned a representative schema
# for extraction (model-driven components); walk it as usual.
convert(data, config_var, path)
else:
raise TypeError("Unknown extracted schema type")
elif config_var.get("key") == "GeneratedID":
+1 -1
View File
@@ -276,7 +276,7 @@ def lint_newline(fname, line, col, content):
return "File contains Windows newline. Please set your editor to Unix newline mode."
@lint_content_check(exclude=["*.svg", ".clang-tidy.hash"])
@lint_content_check(exclude=["*.svg"])
def lint_end_newline(fname, content):
if content and not content.endswith("\n"):
return "File does not end with a newline, please add an empty line at the end of the file."
+214
View File
@@ -0,0 +1,214 @@
#!/usr/bin/env python3
"""Fail when two component test fixtures define the same id with different content.
Component tests are merged and built in groups in CI (see
``script/merge_component_configs.py``). When two components declare the same id
under the same section but with different content, the merge keeps the first and
drops the rest, which can make a cross-reference resolve to an incompatible
entity (this is what broke the i2s_audio speaker tests). That only surfaces when
the two components happen to land in the same group, often in an unrelated PR
long after the duplicate was written.
This script is the complete, batch-independent guard: it scans every component's
``test.<platform>.yaml`` per platform and reports any id that is defined by more
than one component with differing content, so a collision fails the PR that
introduces it and names the exact id and components.
To stay byte-for-byte consistent with what the merge actually does (so the guard
never disagrees with the build), it reuses the merge's own helpers:
* ``prefix_substitutions_in_dict`` -- the merge prefixes every component's
substitution references with the component name before deduplicating, so e.g.
``pin: ${pin}`` in two components becomes ``${a_pin}`` and ``${b_pin}`` and
conflicts. We apply the same prefixing; otherwise a shared id whose only
difference is a substitution looks identical here but conflicts at merge time.
* ``deduplicate_by_id`` -- the actual merge comparison (including the
``INTENTIONALLY_SHARED_IDS`` allowlist for deliberately shared singletons such
as ``sntp_time``). We feed each shared id's prefixed items straight through it
and treat a raised ``ValueError`` as a conflict, so this check and the merge
can never diverge.
``packages:`` are left as opaque ``!include`` objects by the loader -- exactly as
the merge sees them at dedup time -- so package-provided bus ids (``i2c_bus`` ...)
are not compared here, matching the merge, which re-adds those packages once.
"""
from __future__ import annotations
from collections import defaultdict
from collections.abc import Iterator
from dataclasses import dataclass, field
from pathlib import Path
import sys
sys.path.insert(0, str(Path(__file__).parent.parent))
from esphome.core import EsphomeError # noqa: E402
from script.merge_component_configs import ( # noqa: E402
deduplicate_by_id,
load_yaml_file,
prepare_component_body,
)
# Resolved relative to this file (not the CWD) so the scan cannot silently cover
# nothing when run from a different directory.
TESTS_DIR = Path(__file__).resolve().parent.parent / "tests" / "components"
def _collect_ids(
data: object,
path: tuple[str, ...],
out: dict[tuple[tuple[str, ...], object], object],
) -> None:
"""Record (dict_path, id) -> item for id-bearing items in dict-reachable lists.
Keyed by the full dict path (not just the immediate key) so items under
different paths that happen to share a list key name are never compared. Only
lists reached purely through dict keys are recorded: once the merge
concatenates a list, items from different components live in separate elements,
so anything deeper is never compared across components (matching how
``merge_config`` combines bodies). Ids keep their original type so ``5`` and
``"5"`` stay distinct, exactly as ``deduplicate_by_id`` treats them; an
unhashable id (rare) falls back to its ``repr`` so it can still be grouped.
"""
if not isinstance(data, dict):
return
for key, value in data.items():
new_path = path + (key,)
if isinstance(value, list):
for item in value:
if isinstance(item, dict) and "id" in item:
item_id = item["id"]
try:
hash(item_id)
except TypeError:
item_id = repr(item_id)
out[(new_path, item_id)] = item
elif isinstance(value, dict):
_collect_ids(value, new_path, out)
def _discover_platforms() -> set[str]:
platforms: set[str] = set()
for test_file in TESTS_DIR.glob("*/test.*.yaml"):
# test.<platform>.yaml -> platform is the middle dotted part
parts = test_file.name.split(".")
if len(parts) == 3:
platforms.add(parts[1])
return platforms
def _load_components(
platform: str, parse_errors: list[str]
) -> Iterator[tuple[str, object]]:
"""Yield (component, prefixed config) for each component testing this platform.
Each body is prepared with ``prepare_component_body`` (the same helper the
merge uses: it expands component-specific package includes and prefixes
substitutions), so the comparison sees what the build merges. Fixtures that
fail to parse are recorded in ``parse_errors`` so the run can fail rather than
silently skip them.
"""
for comp_dir in sorted(TESTS_DIR.iterdir()):
test_file = comp_dir / f"test.{platform}.yaml"
if not comp_dir.is_dir() or not test_file.exists():
continue
try:
data = load_yaml_file(test_file)
except EsphomeError as err:
parse_errors.append(str(test_file))
print(f"ERROR: could not parse {test_file}: {err}", file=sys.stderr)
continue
yield comp_dir.name, prepare_component_body(data, comp_dir.name, comp_dir)
@dataclass
class ScanResult:
"""Outcome of a scan. A caller cannot observe a clean result while files were
skipped or nothing was scanned -- all three fields are reported together."""
conflicts: list[str] = field(default_factory=list)
parse_errors: list[str] = field(default_factory=list)
components_scanned: int = 0
def scan() -> ScanResult:
"""Scan every component's base test fixture and report cross-component id conflicts.
Only base ``test.<platform>.yaml`` fixtures are scanned because only those are
combined by ``merge_component_configs`` in grouped CI builds; variant
(``test-*.yaml``) fixtures are built individually and never cross-merged.
"""
result = ScanResult()
for platform in sorted(_discover_platforms()):
# (dict_path, id) -> {component: prefixed_item}
groups: dict[tuple[tuple[str, ...], object], dict[str, object]] = defaultdict(
dict
)
for component, data in _load_components(platform, result.parse_errors):
result.components_scanned += 1
collected: dict[tuple[tuple[str, ...], object], object] = {}
_collect_ids(data, (), collected)
for key, item in collected.items():
groups[key][component] = item
for (path, id_), by_component in sorted(
groups.items(), key=lambda kv: (kv[0][0], str(kv[0][1]))
):
if len(by_component) < 2:
continue
# Delegate the decision to the merge's own deduplication so this guard
# can never disagree with what the build does.
try:
deduplicate_by_id({path[-1]: list(by_component.values())})
except ValueError:
result.conflicts.append(
f"[{platform}] id '{id_}' under '{'.'.join(path)}' is defined "
f"differently by: {', '.join(sorted(by_component))}"
)
return result
def main() -> int:
result = scan()
if result.conflicts:
print("Conflicting test component ids found:\n")
for line in result.conflicts:
print(f" - {line}")
print(
"\nGive each component a unique id (e.g. '<component>_<id>'), or add the "
"id to INTENTIONALLY_SHARED_IDS in script/merge_component_configs.py if "
"it is a deliberately shared singleton."
)
if result.parse_errors:
# A fixture we could not parse was never scanned, so the run is not a
# clean pass even if no conflicts were found among the rest.
print(
f"\n{len(result.parse_errors)} test fixture(s) could not be parsed and "
"were not checked:"
)
for path in result.parse_errors:
print(f" - {path}")
if result.components_scanned == 0:
# A scan that covered nothing is a false green -- the whole point of the
# guard is defeated. Fail loudly (wrong working directory or layout change).
print(
f"\nERROR: scanned 0 component test fixtures under {TESTS_DIR}; "
"the guard covered nothing.",
file=sys.stderr,
)
if result.conflicts or result.parse_errors or result.components_scanned == 0:
return 1
print(
f"No conflicting test component ids found "
f"({result.components_scanned} fixtures scanned)."
)
return 0
if __name__ == "__main__":
sys.exit(main())
+55 -3
View File
@@ -35,12 +35,29 @@ def clang_options(idedata):
# extract target architecture from triplet in g++ filename
triplet = Path(idedata["cxx_path"]).name[:-4]
if triplet.startswith("xtensa-"):
# clang doesn't support Xtensa (yet?), so compile in 32-bit mode and pretend we're the Xtensa compiler
# clang has an Xtensa frontend, but only a generic core -- the esp32 IDF
# toolchain headers (xtruntime, xtensa/config) need the GCC core config
# (XCHAL_*) it doesn't ship, so we still compile in 32-bit x86 mode and
# just pretend to be Xtensa. Undefine the host x86 arch macros -m32 sets,
# so libraries with x86 SIMD paths (FastLED's fl/math/simd, simd_x86.hpp)
# fall back to their scalar implementation instead of an incomplete
# host-x86 one, and define the xtensa endianness macro newlib's
# machine/ieeefp.h then needs in their place.
cmd.append("-m32")
cmd.append("-U__i386__")
cmd.append("-U__x86_64__")
cmd.append("-D__XTENSA__")
cmd.append("-D__XTENSA_EL__")
cmd.append("-D_LIBC")
else:
# RISC-V (and other non-Xtensa targets) have a real clang backend, so
# compile for the actual triplet. Espressif's RISC-V GCC -march adds
# vendor extensions (xesploop, xespv) upstream clang doesn't know; those
# are stripped from the copied cxx_flags below.
cmd.append(f"--target={triplet}")
# The GCC build passes flags (e.g. -fno-plt) that clang accepts for some
# targets but not others; don't error on the ones unused for this target.
cmd.append("-Qunused-arguments")
omit_flags = (
"-free",
@@ -52,6 +69,11 @@ def clang_options(idedata):
"-mfix-esp32-psram-cache-issue",
"-mfix-esp32-psram-cache-strategy=memw",
"-fno-tree-switch-conversion",
# GCC-only flags emitted by the native ESP-IDF toolchain build
"-freorder-blocks",
"-fno-jump-tables",
"-fno-shrink-wrap",
"-mno-target-align",
)
if "zephyr" in triplet:
@@ -97,8 +119,28 @@ def clang_options(idedata):
]
)
# copy compiler flags, except those clang doesn't understand.
cmd.extend(flag for flag in idedata["cxx_flags"] if flag not in omit_flags)
# Copy compiler flags, dropping: ones clang doesn't understand; -Werror*
# (clang-tidy enforces .clang-tidy's WarningsAsErrors, and a build -Werror
# would bypass the -clang-diagnostic-* suppressions); and -std= (the native
# ESP-IDF build defaults to gnu++2b, but ESPHome compiles with gnu++20 per
# platformio.ini -- analyzing as C++23 flags code that doesn't build under
# gnu++20). Force gnu++20 to match the real build.
# Strip Espressif's non-standard RISC-V -march extensions (e.g. xesploop,
# xespv); clang rejects the whole arch string otherwise.
def strip_esp_march(flag):
if flag.startswith("-march=") and triplet.startswith("riscv"):
return re.sub(r"_xesp\w+", "", flag)
return flag
cmd.extend(
strip_esp_march(flag)
for flag in idedata["cxx_flags"]
if flag not in omit_flags
and not flag.startswith("-Werror")
and not flag.startswith("-std=")
and not flag.startswith("-mtune=esp")
)
cmd.append("-std=gnu++20")
# defines
cmd.extend(f"-D{define}" for define in idedata["defines"])
@@ -217,6 +259,12 @@ def main():
action="append",
help="only run on files containing value",
)
parser.add_argument(
"-x",
"--exclude-grep",
action="append",
help="skip files containing value",
)
parser.add_argument(
"--split-num", type=int, help="split the files into X jobs.", default=None
)
@@ -251,6 +299,10 @@ def main():
if args.grep:
files = filter_grep(files, args.grep)
if args.exclude_grep:
excluded = set(filter_grep(files, args.exclude_grep))
files = [f for f in files if f not in excluded]
files.sort()
if args.split_num:
Executable → Regular
+32 -169
View File
@@ -1,66 +1,32 @@
#!/usr/bin/env python3
"""Calculate and manage hash for clang-tidy configuration."""
"""Files that affect clang-tidy results, and a content hash over them.
``CLANG_TIDY_GLOBAL_FILES`` (plus ``SDKCONFIG_DEFAULTS_PREFIX``) is the single
source of truth for which files influence clang-tidy output. A change to any of
them can surface warnings in source files a PR didn't touch, so:
* ``script/determine-jobs.py`` runs a full clang-tidy scan when one changes, and
* ``calculate_clang_tidy_hash()`` folds them into the idedata cache key used by
``script/helpers.py`` (a content hash, unlike an mtime check, stays correct
across git checkouts).
"""
from __future__ import annotations
import argparse
import hashlib
from pathlib import Path
import re
import sys
# Add the script directory to path to import helpers
script_dir = Path(__file__).parent
sys.path.insert(0, str(script_dir))
# Root-relative paths whose contents affect clang-tidy results.
CLANG_TIDY_GLOBAL_FILES = (
".clang-tidy",
"platformio.ini",
"requirements_dev.txt",
"esphome/idf_component.yml",
)
def read_file_lines(path: Path) -> list[str]:
"""Read lines from a file."""
with path.open() as f:
return f.readlines()
def parse_requirement_line(line: str) -> tuple[str, str] | None:
"""Parse a requirement line and return (package, original_line) or None.
Handles formats like:
- package==1.2.3
- package==1.2.3 # comment
- package>=1.2.3,<2.0.0
"""
original_line = line.strip()
# Extract the part before any comment for parsing
parse_line = line
if "#" in parse_line:
parse_line = parse_line[: parse_line.index("#")]
parse_line = parse_line.strip()
if not parse_line:
return None
# Use regex to extract package name
# This matches package names followed by version operators
match = re.match(r"^([a-zA-Z0-9_-]+)(==|>=|<=|>|<|!=|~=)(.+)$", parse_line)
if match:
return (match.group(1), original_line) # Return package name and original line
return None
def get_clang_tidy_version_from_requirements(repo_root: Path | None = None) -> str:
"""Get clang-tidy version from requirements_dev.txt"""
repo_root = _ensure_repo_root(repo_root)
requirements_path = repo_root / "requirements_dev.txt"
lines = read_file_lines(requirements_path)
for line in lines:
parsed = parse_requirement_line(line)
if parsed and parsed[0] == "clang-tidy":
# Return the original line (preserves comments)
return parsed[1]
return "clang-tidy version not found"
# sdkconfig.defaults and per-target sdkconfig.defaults.<target> files flip the
# CONFIG flags that decide which variant code paths clang-tidy sees. Matched by
# this prefix at the repo root.
SDKCONFIG_DEFAULTS_PREFIX = "sdkconfig.defaults"
def read_file_bytes(path: Path) -> bytes:
@@ -80,123 +46,20 @@ def _ensure_repo_root(repo_root: Path | None) -> Path:
def calculate_clang_tidy_hash(repo_root: Path | None = None) -> str:
"""Calculate hash of clang-tidy configuration and version"""
"""Calculate a hash of the files that affect clang-tidy results."""
repo_root = _ensure_repo_root(repo_root)
hasher = hashlib.sha256()
# Hash .clang-tidy file
clang_tidy_path = repo_root / ".clang-tidy"
content = read_file_bytes(clang_tidy_path)
hasher.update(content)
for name in CLANG_TIDY_GLOBAL_FILES:
path = repo_root / name
if path.exists():
hasher.update(read_file_bytes(path))
# Hash clang-tidy version from requirements_dev.txt
version = get_clang_tidy_version_from_requirements(repo_root)
hasher.update(version.encode())
# Hash the entire platformio.ini file
platformio_path = repo_root / "platformio.ini"
platformio_content = read_file_bytes(platformio_path)
hasher.update(platformio_content)
# Hash sdkconfig.defaults file
sdkconfig_path = repo_root / "sdkconfig.defaults"
if sdkconfig_path.exists():
sdkconfig_content = read_file_bytes(sdkconfig_path)
hasher.update(sdkconfig_content)
# Hash each sdkconfig.defaults* file. Include the filename so adding or
# renaming a per-target variant is detected, not just content edits.
for path in sorted(repo_root.glob(f"{SDKCONFIG_DEFAULTS_PREFIX}*")):
hasher.update(path.name.encode())
hasher.update(read_file_bytes(path))
return hasher.hexdigest()
def read_stored_hash(repo_root: Path | None = None) -> str | None:
"""Read the stored hash from file"""
repo_root = _ensure_repo_root(repo_root)
hash_file = repo_root / ".clang-tidy.hash"
if hash_file.exists():
lines = read_file_lines(hash_file)
return lines[0].strip() if lines else None
return None
def write_file_content(path: Path, content: str) -> None:
"""Write content to a file."""
with path.open("w") as f:
f.write(content)
def write_hash(hash_value: str, repo_root: Path | None = None) -> None:
"""Write hash to file"""
repo_root = _ensure_repo_root(repo_root)
hash_file = repo_root / ".clang-tidy.hash"
# Strip any trailing newlines to ensure consistent formatting
write_file_content(hash_file, hash_value.strip() + "\n")
def main() -> None:
parser = argparse.ArgumentParser(description="Manage clang-tidy configuration hash")
parser.add_argument(
"--check",
action="store_true",
help="Check if full scan needed (exit 0 if needed)",
)
parser.add_argument("--update", action="store_true", help="Update the hash file")
parser.add_argument(
"--update-if-changed",
action="store_true",
help="Update hash only if configuration changed (for pre-commit)",
)
parser.add_argument(
"--verify", action="store_true", help="Verify hash matches (for CI)"
)
args = parser.parse_args()
current_hash = calculate_clang_tidy_hash()
stored_hash = read_stored_hash()
if args.check:
# Check if hash changed OR if .clang-tidy.hash was updated in this PR
# This is used in CI to determine if a full clang-tidy scan is needed
hash_changed = current_hash != stored_hash
# Lazy import to avoid requiring dependencies that aren't needed for other modes
from helpers import changed_files # noqa: E402
hash_file_updated = ".clang-tidy.hash" in changed_files()
# Exit 0 if full scan needed
sys.exit(0 if (hash_changed or hash_file_updated) else 1)
elif args.verify:
# Verify that hash file is up to date with current configuration
# This is used in pre-commit and CI checks to ensure hash was updated
if current_hash != stored_hash:
print("ERROR: Clang-tidy configuration has changed but hash not updated!")
print(f"Expected: {current_hash}")
print(f"Found: {stored_hash}")
print("\nPlease run: script/clang_tidy_hash.py --update")
sys.exit(1)
print("Hash verification passed")
elif args.update:
write_hash(current_hash)
print(f"Hash updated: {current_hash}")
elif args.update_if_changed:
if current_hash != stored_hash:
write_hash(current_hash)
print(f"Clang-tidy hash updated: {current_hash}")
# Exit 0 so pre-commit can stage the file
sys.exit(0)
else:
print("Clang-tidy hash unchanged")
sys.exit(0)
else:
print(f"Current hash: {current_hash}")
print(f"Stored hash: {stored_hash}")
print(f"Match: {current_hash == stored_hash}")
if __name__ == "__main__":
main()
+166 -127
View File
@@ -55,10 +55,10 @@ from functools import cache
import json
import os
from pathlib import Path
import subprocess
import sys
from typing import Any
from clang_tidy_hash import CLANG_TIDY_GLOBAL_FILES, SDKCONFIG_DEFAULTS_PREFIX
from helpers import (
CPP_FILE_EXTENSIONS,
ESPHOME_TESTS_COMPONENTS_PATH,
@@ -70,6 +70,7 @@ from helpers import (
get_changed_components,
get_component_from_path,
get_component_test_files,
get_component_test_platforms,
get_components_with_dependencies,
get_cpp_changed_components,
get_fixture_to_test_files,
@@ -77,7 +78,6 @@ from helpers import (
get_target_branch,
git_ls_files,
is_validate_only_file,
parse_test_filename,
root_path,
)
from split_components_for_ci import create_intelligent_batches
@@ -169,24 +169,6 @@ MEMORY_IMPACT_FALLBACK_COMPONENT = "api" # Representative component for core ch
MEMORY_IMPACT_FALLBACK_PLATFORM = Platform.ESP32_IDF # Most representative platform
MEMORY_IMPACT_MAX_COMPONENTS = 40 # Max components before results become nonsensical
# Platform-specific components that can only be built on their respective platforms
# These components contain platform-specific code and cannot be cross-compiled
# Regular components (wifi, logger, api, etc.) are cross-platform and not listed here
PLATFORM_SPECIFIC_COMPONENTS = frozenset(
{
"esp32", # ESP32 platform implementation
"esp8266", # ESP8266 platform implementation
"rp2040", # Raspberry Pi Pico / RP2040 platform implementation
"libretiny", # LibreTiny base platform implementation
"bk72xx", # Beken BK72xx platform implementation (uses LibreTiny)
"rtl87xx", # Realtek RTL87xx platform implementation (uses LibreTiny)
"ln882x", # Winner Micro LN882x platform implementation (uses LibreTiny)
"host", # Host platform (for testing on development machine)
"nrf52", # Nordic nRF52 platform implementation (uses Zephyr)
"zephyr", # Zephyr RTOS platform implementation
}
)
# Platform preference order for memory impact analysis
# This order is used when no platform-specific hints are detected from filenames
# Priority rationale:
@@ -298,23 +280,22 @@ def determine_integration_tests(branch: str | None = None) -> tuple[bool, list[s
@cache
def _is_clang_tidy_full_scan() -> bool:
"""Check if clang-tidy configuration changed (requires full scan).
def _is_clang_tidy_full_scan(branch: str | None = None) -> bool:
"""Check if a clang-tidy-relevant config file changed (requires full scan).
A change to a file that affects clang-tidy globally can surface warnings in
source files the PR didn't touch, so the entire codebase must be re-scanned.
Returns:
True if full scan is needed (hash changed), False otherwise.
True if full scan is needed, False otherwise.
"""
try:
result = subprocess.run(
[str(Path(root_path) / "script" / "clang_tidy_hash.py"), "--check"],
capture_output=True,
check=False,
)
# Exit 0 means hash changed (full scan needed)
return result.returncode == 0
except Exception:
# If hash check fails, run full scan to be safe
return True
for file in changed_files(branch):
if file in CLANG_TIDY_GLOBAL_FILES:
return True
# Root-level sdkconfig.defaults and per-target sdkconfig.defaults.<target>
if "/" not in file and file.startswith(SDKCONFIG_DEFAULTS_PREFIX):
return True
return False
def should_run_clang_tidy(branch: str | None = None) -> bool:
@@ -325,13 +306,12 @@ def should_run_clang_tidy(branch: str | None = None) -> bool:
Clang-tidy will run when ANY of the following conditions are met:
1. Clang-tidy configuration changed
- The hash of .clang-tidy configuration file has changed
- The hash includes the .clang-tidy file, clang-tidy version from requirements_dev.txt,
and relevant platformio.ini sections
- When configuration changes, a full scan is needed to ensure all code complies
with the new rules
- Detected by script/clang_tidy_hash.py --check returning exit code 0
1. A clang-tidy-relevant config file changed (full scan needed)
- Any file in CLANG_TIDY_GLOBAL_FILES (.clang-tidy, platformio.ini,
requirements_dev.txt, esphome/idf_component.yml) or a root-level
sdkconfig.defaults* file
- These affect clang-tidy results globally, so all code must be re-checked
to ensure it still complies
2. Any C++ source files changed
- Any file with C++ extensions: .cpp, .h, .hpp, .cc, .cxx, .c, .tcc
@@ -339,27 +319,14 @@ def should_run_clang_tidy(branch: str | None = None) -> bool:
- This ensures all C++ code is checked, including tests, examples, etc.
- Examples: esphome/core/component.cpp, tests/custom/my_component.h
3. The .clang-tidy.hash file itself changed
- This indicates the configuration has been updated and clang-tidy should run
- Ensures that PRs updating the clang-tidy configuration are properly validated
If the hash check fails for any reason, clang-tidy runs as a safety measure to ensure
code quality is maintained.
Args:
branch: Branch to compare against. If None, uses default.
Returns:
True if clang-tidy should run, False otherwise.
"""
# First check if clang-tidy configuration changed (full scan needed)
if _is_clang_tidy_full_scan():
return True
# Check if .clang-tidy.hash file itself was changed
# This handles the case where the hash was properly updated in the PR
files = changed_files(branch)
if ".clang-tidy.hash" in files:
# First check if a clang-tidy-relevant config file changed (full scan needed)
if _is_clang_tidy_full_scan(branch):
return True
return _any_changed_file_endswith(branch, CPP_FILE_EXTENSIONS)
@@ -499,11 +466,11 @@ def should_run_device_builder(branch: str | None = None) -> bool:
return False
# Components tested by the native ESP-IDF compile-test job. This is the
# Components tested by the PlatformIO compile-test job. This is the
# single source of truth: the workflow reads the comma-joined list from the
# `native-idf-components` output of `determine-jobs` and uses it as the
# `TEST_COMPONENTS` env on the `test-native-idf` job.
NATIVE_IDF_TEST_COMPONENTS = frozenset(
# `esp32-platformio-components` output of `determine-jobs` and uses it as the
# `TEST_COMPONENTS` env on the `test-esp32-platformio` job.
ESP32_PLATFORMIO_TEST_COMPONENTS = frozenset(
{
"esp32",
"api",
@@ -523,53 +490,75 @@ NATIVE_IDF_TEST_COMPONENTS = frozenset(
}
)
# Path prefixes whose changes always trigger the native ESP-IDF compile
# test: anything under esphome/espidf/ (the native IDF runner / API /
# framework / component generator).
NATIVE_IDF_TRIGGER_PATH_PREFIXES = ("esphome/espidf/",)
# Path prefixes whose changes always trigger the PlatformIO compile test:
# anything under esphome/platformio/ (the PlatformIO runner / toolchain that
# drives every PlatformIO build). The esp32 platform component is already in
# ESP32_PLATFORMIO_TEST_COMPONENTS, so its changes are covered by the normal
# component-narrowing path.
ESP32_PLATFORMIO_TRIGGER_PATH_PREFIXES = ("esphome/platformio/",)
# Standalone files that, when changed, also trigger the native ESP-IDF
# compile test:
# - esphome/build_gen/espidf.py -- the native IDF build generator
# (other files under build_gen/ target PlatformIO and don't affect
# the native IDF path)
# Standalone files that, when changed, trigger the PlatformIO compile test:
# - esphome/build_gen/platformio.py -- the PlatformIO build generator
# - script/test_build_components.py -- the harness the job invokes
# - .github/workflows/ci.yml -- the job's own definition
NATIVE_IDF_TRIGGER_FILES = frozenset(
ESP32_PLATFORMIO_TRIGGER_FILES = frozenset(
{
"esphome/build_gen/espidf.py",
"esphome/build_gen/platformio.py",
"script/test_build_components.py",
".github/workflows/ci.yml",
}
)
def _native_idf_path_or_file_trigger(files: list[str]) -> bool:
"""Whether any changed file is a native IDF infrastructure / harness trigger."""
def _esp32_platformio_path_or_file_trigger(files: list[str]) -> bool:
"""Whether any changed file is a PlatformIO infrastructure / harness trigger."""
for file in files:
if file in NATIVE_IDF_TRIGGER_FILES:
if file in ESP32_PLATFORMIO_TRIGGER_FILES:
return True
if any(file.startswith(prefix) for prefix in NATIVE_IDF_TRIGGER_PATH_PREFIXES):
if any(
file.startswith(prefix) for prefix in ESP32_PLATFORMIO_TRIGGER_PATH_PREFIXES
):
return True
return False
def native_idf_components_to_test(branch: str | None = None) -> list[str]:
"""Subset of ``NATIVE_IDF_TEST_COMPONENTS`` the job needs to compile.
# ESP-IDF infra: changes under esphome/espidf/ or to the IDF build generator
# affect every esp32 IDF build (now the default toolchain) but aren't
# components, so the component matrix wouldn't otherwise force any esp32
# compile. When they change we fold the `esp32` component into the matrix so
# the default native-IDF build path is still compiled on an infra-only PR.
ESP_IDF_INFRA_TRIGGER_PATH_PREFIXES = ("esphome/espidf/",)
ESP_IDF_INFRA_TRIGGER_FILES = frozenset({"esphome/build_gen/espidf.py"})
The job builds components with the native ESP-IDF toolchain (no
PlatformIO). When only a specific component (or something it depends
on) changed, there's no value in re-building every other unrelated
component in the test list -- the regular ``component-test`` matrix
already covers them via PlatformIO. So we narrow to the intersection
of ``NATIVE_IDF_TEST_COMPONENTS`` and the changed-component dependency
def _esp_idf_infra_changed(files: list[str]) -> bool:
"""Whether any changed file is ESP-IDF build/runner infrastructure."""
for file in files:
if file in ESP_IDF_INFRA_TRIGGER_FILES:
return True
if any(
file.startswith(prefix) for prefix in ESP_IDF_INFRA_TRIGGER_PATH_PREFIXES
):
return True
return False
def esp32_platformio_components_to_test(branch: str | None = None) -> list[str]:
"""Subset of ``ESP32_PLATFORMIO_TEST_COMPONENTS`` the job needs to compile.
The job builds components with the PlatformIO toolchain. When only a
specific component (or something it depends on) changed, there's no
value in re-building every other unrelated component in the test list --
the regular ``component-test`` matrix already covers them via the
default toolchain. So we narrow to the intersection of
``ESP32_PLATFORMIO_TEST_COMPONENTS`` and the changed-component dependency
closure.
Returns the full list (sorted) when we can't safely narrow:
1. Core C++/Python files changed (``esphome/core/*``).
2. Native IDF infrastructure changed (``esphome/espidf/*`` or
``esphome/build_gen/espidf.py``).
2. PlatformIO infrastructure changed (``esphome/platformio/*`` or
``esphome/build_gen/platformio.py``).
3. The test harness or workflow itself changed
(``script/test_build_components.py``, ``.github/workflows/ci.yml``).
@@ -591,31 +580,31 @@ def native_idf_components_to_test(branch: str | None = None) -> list[str]:
"""
files = changed_files(branch)
if core_changed(files) or _native_idf_path_or_file_trigger(files):
return sorted(NATIVE_IDF_TEST_COMPONENTS)
if core_changed(files) or _esp32_platformio_path_or_file_trigger(files):
return sorted(ESP32_PLATFORMIO_TEST_COMPONENTS)
component_files = [f for f in files if filter_component_and_test_files(f)]
changed = get_components_with_dependencies(component_files, True)
return sorted(NATIVE_IDF_TEST_COMPONENTS & set(changed))
return sorted(ESP32_PLATFORMIO_TEST_COMPONENTS & set(changed))
def should_run_native_idf(branch: str | None = None) -> bool:
"""Determine if the `test-native-idf` compile-test job should run.
def should_run_esp32_platformio(branch: str | None = None) -> bool:
"""Determine if the `test-esp32-platformio` compile-test job should run.
Runs whenever ``native_idf_components_to_test()`` returns a non-empty
Runs whenever ``esp32_platformio_components_to_test()`` returns a non-empty
list. Skipping the job on unrelated Python-only PRs avoids ~5 min of
CI per PR (worse on cold caches). The regular ``component-test``
matrix still exercises the same components through PlatformIO when
those components change.
matrix still exercises the same components through the default
toolchain when those components change.
Args:
branch: Branch to compare against. If None, uses default.
Returns:
True if the native ESP-IDF compile test should run, False otherwise.
True if the PlatformIO compile test should run, False otherwise.
"""
return bool(native_idf_components_to_test(branch))
return bool(esp32_platformio_components_to_test(branch))
def determine_cpp_unit_tests(
@@ -1006,23 +995,24 @@ def detect_memory_impact_config(
] = {} # Track which platforms each component supports
for component in sorted(changed_component_set):
# Look for test files on preferred platforms
test_files = get_component_test_files(component, all_variants=True)
if not test_files:
continue
# Check if component has tests for any preferred platform
available_platforms = [
platform
for test_file in test_files
if (platform := parse_test_filename(test_file)[1]) != "all"
and platform in MEMORY_IMPACT_PLATFORM_PREFERENCE
]
# Discover the platforms this component has BASE tests for, using the
# same logic as the build runner (get_component_test_platforms wraps the
# shared get_component_test_files + parse_test_filename helpers). Base
# tests only: the memory impact CI build runs test_build_components.py
# with --base-only, which compiles base test.<platform>.yaml files but
# never variant test-<variant>.<platform>.yaml files. Counting
# variant-only platforms here would let us select a platform the build
# then has nothing to compile for, producing no memory output.
available_platforms = {
Platform(platform)
for platform in get_component_test_platforms(component)
if platform in MEMORY_IMPACT_PLATFORM_PREFERENCE
}
if not available_platforms:
continue
component_platforms_map[component] = set(available_platforms)
component_platforms_map[component] = available_platforms
components_with_tests.append(component)
# If no components have tests, don't run memory impact
@@ -1084,20 +1074,57 @@ def detect_memory_impact_config(
)
platform = _select_platform_by_count(platform_counts)
# Filter out platform-specific components that are incompatible with selected platform
# Platform components (esp32, esp8266, rp2040, etc.) can only build on their own platform
# Other components (wifi, logger, etc.) are cross-platform and can build anywhere
compatible_components = [
component
for component in components_with_tests
if component not in PLATFORM_SPECIFIC_COMPONENTS
or platform in component_platforms_map.get(component, set())
]
# Keep only components that have a base test on the selected platform.
# The merged build runs test_build_components.py -t <platform> --base-only,
# so a component without a base test.<platform>.yaml compiles nothing and
# contributes no memory output. This also covers platform-specific
# components (esp32, esp8266, etc.), which only have tests on their own
# platform. When components don't share a common platform we build the
# largest subset that does, dropping the rest.
def components_supporting(target: Platform) -> list[str]:
return [
component
for component in components_with_tests
if target in component_platforms_map.get(component, set())
]
# If no components are compatible with the selected platform, don't run
compatible_components = components_supporting(platform)
# A platform hint (or no-common-platform fallback) can pick a platform that
# no changed component actually has a base test for, leaving nothing to
# build. In that case fall back to the platform supported by the most
# components. component_platforms_map is non-empty (guarded above) and every
# value is a non-empty platform set (components with no supported platform
# are skipped at discovery), so this always yields a buildable platform with
# at least one compatible component.
if not compatible_components:
platform = _select_platform_by_count(
Counter(
p for platforms in component_platforms_map.values() for p in platforms
)
)
compatible_components = components_supporting(platform)
# Defensive backstop: unreachable given the invariant above, but guards
# against a future regression in platform selection silently passing an
# empty component list to the build.
if not compatible_components:
return {"should_run": "false"}
# Log components dropped because they lack a base test on the selected
# platform so partial-subset builds are visible in CI logs.
dropped_components = [
component
for component in components_with_tests
if component not in compatible_components
]
if dropped_components:
print(
f"Memory impact: Dropping components without a base test on "
f"{platform}: {dropped_components}",
file=sys.stderr,
)
# Debug output
print("Memory impact analysis:", file=sys.stderr)
print(f" Changed components: {sorted(changed_component_set)}", file=sys.stderr)
@@ -1157,8 +1184,8 @@ def main() -> None:
run_python_linters = True
run_import_time = True
run_device_builder = True
native_idf_components = sorted(NATIVE_IDF_TEST_COMPONENTS)
run_native_idf = True
esp32_platformio_components = sorted(ESP32_PLATFORMIO_TEST_COMPONENTS)
run_esp32_platformio = True
else:
integration_run_all, integration_test_files = determine_integration_tests(
args.branch
@@ -1168,8 +1195,8 @@ def main() -> None:
run_python_linters = should_run_python_linters(args.branch)
run_import_time = should_run_import_time(args.branch)
run_device_builder = should_run_device_builder(args.branch)
native_idf_components = native_idf_components_to_test(args.branch)
run_native_idf = bool(native_idf_components)
esp32_platformio_components = esp32_platformio_components_to_test(args.branch)
run_esp32_platformio = bool(esp32_platformio_components)
run_integration, integration_test_buckets = _compute_integration_test_buckets(
integration_run_all, integration_test_files
)
@@ -1223,6 +1250,18 @@ def main() -> None:
if _component_has_tests(component)
]
# ESP-IDF build-gen/runner changed but no component pulled esp32 in: fold the
# `esp32` component into the matrix so the default native-IDF build path is
# still compiled on an infra-only PR. force_all/core already test everything,
# so skip there. Runs grouped (not added to directly-changed).
if (
not is_core_change
and _esp_idf_infra_changed(changed)
and "esp32" not in changed_components_with_tests
and _component_has_tests("esp32")
):
changed_components_with_tests.append("esp32")
# Get directly changed components with tests (for isolated testing)
# These will be tested WITHOUT --testing-mode in CI to enable full validation
# (pin conflicts, etc.) since they contain the actual changes being reviewed
@@ -1256,9 +1295,9 @@ def main() -> None:
# Determine clang-tidy mode based on actual files that will be checked
is_full_scan = False
if run_clang_tidy:
# Full scan needed if: hash changed OR core files changed
# (is_core_change is forced True under --force-all)
is_full_scan = _is_clang_tidy_full_scan() or is_core_change
# Full scan needed if: a clang-tidy-relevant config file changed OR
# core files changed (is_core_change is forced True under --force-all)
is_full_scan = _is_clang_tidy_full_scan(args.branch) or is_core_change
if is_full_scan:
# Full scan checks all files - always use split mode for efficiency
@@ -1340,8 +1379,8 @@ def main() -> None:
"python_linters": run_python_linters,
"import_time": run_import_time,
"device_builder": run_device_builder,
"native_idf": run_native_idf,
"native_idf_components": ",".join(native_idf_components),
"esp32_platformio": run_esp32_platformio,
"esp32_platformio_components": ",".join(esp32_platformio_components),
"changed_components": changed_components,
"changed_components_with_tests": changed_components_with_tests,
"directly_changed_components_with_tests": list(directly_changed_with_tests),
+47 -19
View File
@@ -53,6 +53,7 @@ BASE_BUS_COMPONENTS = {
"canbus",
"remote_transmitter",
"remote_receiver",
"i2s_audio",
}
# Cache version for components graph
@@ -149,6 +150,31 @@ def get_component_test_files(
return files
def get_component_test_platforms(component: str, *, base_only: bool = True) -> set[str]:
"""Return the set of platforms a component has compilable test files for.
Uses the same discovery as ``test_build_components.py`` (``get_component_test_files``
+ ``parse_test_filename``) so callers agree with what the build runner would
actually compile. With ``base_only=True`` (the default, matching the
memory-impact build's ``--base-only``), only base ``test.<platform>.yaml``
files are considered; variant ``test-<variant>.<platform>.yaml`` files are
excluded. The ``"all"`` platform sentinel is excluded.
Args:
component: Component name (e.g. "wifi")
base_only: If True, only consider base test files (default).
Returns:
Set of platform identifiers (e.g. {"esp32-idf", "esp8266-ard"}).
"""
platforms: set[str] = set()
for test_file in get_component_test_files(component, all_variants=not base_only):
platform = parse_test_filename(test_file)[1]
if platform != "all":
platforms.add(platform)
return platforms
def is_validate_only_file(test_file: Path) -> bool:
"""Return True if the given path is a config-only validate file.
@@ -645,26 +671,22 @@ def load_idedata(environment: str) -> dict[str, Any]:
start_time = time.time()
print(f"Loading IDE data for environment '{environment}'...")
platformio_ini = Path(root_path) / "platformio.ini"
# Reuse the clang-tidy input hash as the cache key: it already covers every
# file baked into the generated idedata (platformio.ini, sdkconfig.defaults,
# esphome/idf_component.yml), so this can't drift from that file list. A
# content hash -- unlike an mtime comparison -- stays correct across git
# checkouts, which don't preserve mtimes.
from clang_tidy_hash import calculate_clang_tidy_hash
temp_idedata = Path(temp_folder) / f"idedata-{environment}.json"
changed = False
if (
not platformio_ini.is_file()
or not temp_idedata.is_file()
or platformio_ini.stat().st_mtime >= temp_idedata.stat().st_mtime
):
changed = True
temp_hash = Path(temp_folder) / f"idedata-{environment}.hash"
if "idf" in environment:
# remove full sdkconfig when the defaults have changed so that it is regenerated
default_sdkconfig = Path(root_path) / "sdkconfig.defaults"
temp_sdkconfig = Path(temp_folder) / f"sdkconfig-{environment}"
if not temp_sdkconfig.is_file():
changed = True
elif default_sdkconfig.stat().st_mtime >= temp_sdkconfig.stat().st_mtime:
temp_sdkconfig.unlink()
changed = True
cache_key = calculate_clang_tidy_hash()
changed = (
not temp_idedata.is_file()
or not temp_hash.is_file()
or temp_hash.read_text().strip() != cache_key
)
if not changed:
data = json.loads(temp_idedata.read_text())
@@ -675,7 +697,12 @@ def load_idedata(environment: str) -> dict[str, Any]:
# ensure temp directory exists before running pio, as it writes sdkconfig to it
Path(temp_folder).mkdir(exist_ok=True)
if "nrf" in environment:
platformio_ini = Path(root_path) / "platformio.ini"
if "esp32" in environment:
from esphome.espidf.clang_tidy import load_idedata as idf_load_idedata
data = idf_load_idedata(environment, temp_folder, platformio_ini)
elif "nrf" in environment:
from helpers_zephyr import load_idedata as zephyr_load_idedata
data = zephyr_load_idedata(environment, temp_folder, platformio_ini)
@@ -686,6 +713,7 @@ def load_idedata(environment: str) -> dict[str, Any]:
match = re.search(r'{\s*".*}', stdout.decode("utf-8"))
data = json.loads(match.group())
temp_idedata.write_text(json.dumps(data, indent=2) + "\n")
temp_hash.write_text(cache_key + "\n")
elapsed = time.time() - start_time
print(f"IDE data generated and cached in {elapsed:.2f} seconds")
+97 -69
View File
@@ -161,18 +161,46 @@ def prefix_substitutions_in_dict(
return data
# (section, id) pairs that several components intentionally share. ESPHome
# treats these as a single instance when merged, so duplicates with differing
# content are expected and must not be flagged as accidental collisions. Keyed on
# the section as well as the id so a generic name (e.g. `ldo_id`) is only exempt
# in its intended section -- an accidental collision on the same name elsewhere
# is still caught.
INTENTIONALLY_SHARED_IDS = frozenset(
{
# Several components each declare an `sntp_time` clock; ESPHome merges
# them into one time source.
("time", "sntp_time"),
# esp_ldo and mipi_dsi both configure the channel-3 internal LDO on the
# ESP32-P4; only one LDO per channel may exist, so the shared id lets the
# merge collapse them into a single LDO.
("esp_ldo", "ldo_id"),
}
)
def deduplicate_by_id(data: dict) -> dict:
"""Deduplicate list items with the same ID.
Keeps only the first occurrence of each ID. If items with the same ID
are identical, this silently deduplicates. If they differ, the first
one is kept (ESPHome's validation will catch if this causes issues).
Identical items sharing an ID (e.g. a shared bus from a common package pulled
in by several components) are collapsed to the first occurrence. Two items
that share an ID but differ in content are a real conflict: when merged, the
first silently wins and the others are dropped, which can make a
cross-reference resolve to an incompatible entity. Rather than defer that to
downstream validation (where it surfaces as a confusing, order-dependent
failure in an unrelated build), raise immediately so the offending ID is
named. Ids in ``INTENTIONALLY_SHARED_IDS`` are deliberately shared singletons
and keep their collapse behaviour.
Args:
data: Parsed config dictionary
Returns:
Config with deduplicated lists
Raises:
ValueError: If two items share an ID but have different content.
"""
if not isinstance(data, dict):
return data
@@ -181,16 +209,25 @@ def deduplicate_by_id(data: dict) -> dict:
for key, value in data.items():
if isinstance(value, list):
# Check for items with 'id' field
seen_ids = set()
seen_items: dict[str, Any] = {}
deduped_list = []
for item in value:
if isinstance(item, dict) and "id" in item:
item_id = item["id"]
if item_id not in seen_ids:
seen_ids.add(item_id)
if item_id not in seen_items:
seen_items[item_id] = item
deduped_list.append(item)
# else: skip duplicate ID (keep first occurrence)
elif (key, item_id) in INTENTIONALLY_SHARED_IDS:
# Deliberately shared singleton -> keep first occurrence.
pass
elif item != seen_items[item_id]:
raise ValueError(
f"Conflicting definitions for id '{item_id}' under "
f"'{key}' when merging test configs; give each "
f"component a unique id"
)
# else: identical duplicate (e.g. shared bus package) -> skip
else:
# No ID, just add it
deduped_list.append(item)
@@ -205,6 +242,55 @@ def deduplicate_by_id(data: dict) -> dict:
return result
def prepare_component_body(comp_data: dict, comp_name: str, comp_dir: Path) -> dict:
"""Return a component's test body as it enters the merge.
Expands component-specific package includes inline (common bus packages are
left for the merge to re-add once), applies ESPHome's top-level-substitutions
-override-package-substitutions rule, then prefixes every substitution
reference with the component name. Shared by ``merge_component_configs`` and
the duplicate-id guard (``script/ci_check_duplicate_test_ids.py``) so the
guard compares exactly what the build merges.
"""
# $component_dir resolves to the component's absolute path.
comp_abs_dir = str(comp_dir.absolute())
# Top-level substitutions override package substitutions, so capture them
# before expanding packages can introduce their own.
top_level_subs = (
comp_data["substitutions"].copy()
if isinstance(comp_data.get("substitutions"), dict)
else {}
)
packages_value = comp_data.get("packages")
if isinstance(packages_value, dict):
common_bus_packages = get_common_bus_packages()
for pkg_name, pkg_value in list(packages_value.items()):
if pkg_name in common_bus_packages:
continue
if isinstance(pkg_value, yaml_util.IncludeFile):
pkg_value = pkg_value.load()
if isinstance(pkg_value, dict):
comp_data = merge_config(comp_data, pkg_value)
elif isinstance(packages_value, list):
for pkg_value in packages_value:
if isinstance(pkg_value, yaml_util.IncludeFile):
pkg_value = pkg_value.load()
if isinstance(pkg_value, dict):
comp_data = merge_config(comp_data, pkg_value)
# Common bus packages are re-added once by the caller; drop them here.
comp_data.pop("packages", None)
subs = comp_data.get("substitutions") or {}
subs.update(top_level_subs)
prefixed_subs = {f"{comp_name}_{name}": value for name, value in subs.items()}
prefixed_subs[f"{comp_name}_component_dir"] = comp_abs_dir
comp_data["substitutions"] = prefixed_subs
return prefix_substitutions_in_dict(comp_data, comp_name)
def merge_component_configs(
component_names: list[str],
platform: str,
@@ -266,67 +352,9 @@ def merge_component_configs(
# New package type - add it
all_packages[pkg_name] = pkg_config
# Handle $component_dir by replacing with absolute path
# This allows components that use local file references to be grouped
comp_abs_dir = str(comp_dir.absolute())
# Save top-level substitutions BEFORE expanding packages
# In ESPHome, top-level substitutions override package substitutions
top_level_subs = (
comp_data["substitutions"].copy()
if "substitutions" in comp_data and comp_data["substitutions"] is not None
else {}
)
# Expand packages - but we'll restore substitution priority after
if "packages" in comp_data:
packages_value = comp_data["packages"]
if isinstance(packages_value, dict):
# Dict format - check each package
common_bus_packages = get_common_bus_packages()
for pkg_name, pkg_value in list(packages_value.items()):
if pkg_name in common_bus_packages:
continue
# Resolve deferred !include files before checking type
if isinstance(pkg_value, yaml_util.IncludeFile):
pkg_value = pkg_value.load()
if not isinstance(pkg_value, dict):
continue
# Component-specific package - expand its content into top level
comp_data = merge_config(comp_data, pkg_value)
elif isinstance(packages_value, list):
# List format - expand all package includes
for pkg_value in packages_value:
# Resolve deferred !include files before checking type
if isinstance(pkg_value, yaml_util.IncludeFile):
pkg_value = pkg_value.load()
if not isinstance(pkg_value, dict):
continue
comp_data = merge_config(comp_data, pkg_value)
# Remove all packages (common will be re-added at the end)
del comp_data["packages"]
# Restore top-level substitution priority
# Top-level substitutions override any from packages
if "substitutions" not in comp_data or comp_data["substitutions"] is None:
comp_data["substitutions"] = {}
# Merge: package subs as base, top-level subs override
comp_data["substitutions"].update(top_level_subs)
# Now prefix the final merged substitutions
comp_data["substitutions"] = {
f"{comp_name}_{sub_name}": sub_value
for sub_name, sub_value in comp_data["substitutions"].items()
}
# Add component_dir substitution with absolute path for this component
comp_data["substitutions"][f"{comp_name}_component_dir"] = comp_abs_dir
# Prefix substitution references throughout the config
comp_data = prefix_substitutions_in_dict(comp_data, comp_name)
# Expand component-specific packages and prefix substitutions, exactly as
# the duplicate-id guard does, so both see the same body.
comp_data = prepare_component_body(comp_data, comp_name, comp_dir)
# Use ESPHome's merge_config to merge this component into the result
# merge_config handles list merging with ID-based deduplication automatically
@@ -437,7 +465,7 @@ def main() -> None:
tests_dir=args.tests_dir,
output_file=args.output,
)
except Exception as e:
except Exception as e: # noqa: BLE001
print(f"Error merging configs: {e}", file=sys.stderr)
import traceback
+1 -1
View File
@@ -21,7 +21,7 @@ async def connect_disconnect(client_id: int, iteration: int) -> tuple[int, bool,
await asyncio.wait_for(cli.connect(login=True), timeout=10)
await cli.disconnect()
return iteration, True, ""
except Exception as e:
except Exception as e: # noqa: BLE001
return (
iteration,
False,
+1 -15
View File
@@ -42,6 +42,7 @@ from script.analyze_component_buses import (
from script.helpers import (
get_component_test_files,
is_validate_only_file,
parse_test_filename,
split_conflicting_groups,
)
from script.merge_component_configs import merge_component_configs
@@ -122,21 +123,6 @@ def find_component_tests(
return dict(component_tests)
def parse_test_filename(test_file: Path) -> tuple[str, str]:
"""Parse test filename to extract test name and platform.
Args:
test_file: Path to test file
Returns:
Tuple of (test_name, platform)
"""
parts = test_file.stem.split(".")
if len(parts) == 2:
return parts[0], parts[1] # test, platform
return parts[0], "all"
def get_platform_base_files(base_dir: Path) -> dict[str, list[Path]]:
"""Get all platform base files.
+1 -1
View File
@@ -63,7 +63,7 @@ def test_component_group(
try:
result = subprocess.run(cmd, check=False)
return result.returncode == 0
except Exception as e:
except Exception as e: # noqa: BLE001
print(f"Error running test: {e}")
return False