From 4f67932e0d73f5254a6e2591e587baf92e27c469 Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Mon, 3 Aug 2026 20:48:11 -0500 Subject: [PATCH] [substitutions][core] Expand templated !include paths to on-disk candidates during bundle discovery (#17647) --- esphome/components/substitutions/__init__.py | 54 +++- esphome/expression.py | 6 +- esphome/yaml_util.py | 174 +++++++++++-- tests/unit_tests/test_bundle.py | 59 ++++- tests/unit_tests/test_substitutions.py | 133 ++++++++++ tests/unit_tests/test_yaml_util.py | 245 ++++++++++++++++++- 6 files changed, 648 insertions(+), 23 deletions(-) diff --git a/esphome/components/substitutions/__init__.py b/esphome/components/substitutions/__init__.py index ea79054c88..b4fcf36c9e 100644 --- a/esphome/components/substitutions/__init__.py +++ b/esphome/components/substitutions/__init__.py @@ -1,5 +1,7 @@ from collections import ChainMap +from itertools import product import logging +import re from typing import Any import esphome @@ -7,6 +9,7 @@ from esphome import core from esphome.config_helpers import Extend, Remove, merge_config, merge_dicts_ordered import esphome.config_validation as cv from esphome.const import CONF_SUBSTITUTIONS, VALID_SUBSTITUTIONS_CHARACTERS +from esphome.expression import JINJA_PROG from esphome.types import ConfigType from esphome.util import OrderedDict from esphome.yaml_util import ( @@ -27,6 +30,14 @@ _LOGGER = logging.getLogger(__name__) ContextVars = ChainMap[str, Any] ErrList = list[tuple[UndefinedError, DocumentPath, Any]] +# Candidate-pattern shaping for include_candidate_patterns. +_ADJACENT_WILDCARDS_RE = re.compile(r"\*+") +# Dots are included so a variant like `../*` counts as fully dynamic too; +# it would otherwise glob everything in the parent directory. +_WILDCARDS_ONLY_RE = re.compile(r"[*./\\]+") +_GLOB_META_RE = re.compile(r"[?\[]") +_STRING_LITERAL_RE = re.compile(r"'([^']*)'|\"([^\"]*)\"") + # Module-level instance is safe: context_vars is passed per-call, and context_trace # is stack-saved/restored within expand(). Not thread-safe — only use from one thread. jinja = Jinja() @@ -360,9 +371,7 @@ def resolve_include( ) substituted = filename != original_str if substituted: - include = IncludeFile( - include.parent_file, filename, include.vars, include.yaml_loader - ) + include = include.with_file(filename) try: return include.load() except esphome.core.EsphomeError as err: @@ -374,6 +383,45 @@ def resolve_include( ) from err +def include_candidate_patterns(value: str) -> list[str]: + """Expand a substitution/Jinja-templated path into glob-style candidate patterns. + + Mirrors the two phases of :func:`_expand_substitutions` without variable + values: ``$var`` / ``${var}`` references become ``*`` and each remaining + Jinja expression contributes one pattern per quoted string literal it + holds (``*`` when it holds none), so every conditional branch is a + candidate — deliberately over-inclusive. Emitted wildcard patterns are + glob-safe: adjacent wildcards collapse (no recursive ``**``), ``[`` / + ``?`` from the filename text are escaped, and variants reduced to + nothing but wildcards, dots and separators are dropped so a fully + dynamic filename never expands to "everything in the directory", + including via a ``../*`` parent traversal. + """ + # Replacing $var / ${var} first also keeps JINJA_PROG's first-} span + # matching correct for references nested inside string literals, the + # same ordering _expand_substitutions relies on. + value = cv.VARIABLE_PROG.sub("*", value) + options = [ + [a or b for a, b in _STRING_LITERAL_RE.findall(expr)] or ["*"] + for expr in JINJA_PROG.findall(value) + ] + + variants: list[str] = [] + for combination in product(*options): + replacements = iter(combination) + spliced = JINJA_PROG.sub(lambda _, _next=replacements: next(_next), value) + variants.append(_ADJACENT_WILDCARDS_RE.sub("*", spliced)) + + patterns: list[str] = [] + for variant in dict.fromkeys(variants): + if not variant or _WILDCARDS_ONLY_RE.fullmatch(variant): + continue + if "*" in variant: + variant = _GLOB_META_RE.sub(r"[\g<0>]", variant) + patterns.append(variant) + return patterns + + def _substitute_include( include: IncludeFile, path: DocumentPath, diff --git a/esphome/expression.py b/esphome/expression.py index d425d822a4..13da3b6a06 100644 --- a/esphome/expression.py +++ b/esphome/expression.py @@ -1,4 +1,4 @@ -"""Helpers for detecting substitution variables and Jinja expressions.""" +"""Helpers for detecting and matching substitution variables and Jinja expressions.""" import re @@ -8,7 +8,7 @@ SUBSTITUTION_VARIABLE_PROG = re.compile( rf"\$([{VALID_SUBSTITUTIONS_CHARACTERS}]+|\{{[{VALID_SUBSTITUTIONS_CHARACTERS}]*\}})" ) -_JINJA_RE = re.compile( +JINJA_PROG = re.compile( r"<%.+?%>" # Block: <% ... %> r"|\$\{[^}]+\}", # Braced: ${ ... } flags=re.MULTILINE, @@ -17,7 +17,7 @@ _JINJA_RE = re.compile( def has_jinja(value: str) -> bool: """Check if a string contains Jinja expressions.""" - return _JINJA_RE.search(value) is not None + return JINJA_PROG.search(value) is not None def has_substitution_or_expression(value: str) -> bool: diff --git a/esphome/yaml_util.py b/esphome/yaml_util.py index c2db9b97ed..833d6f1dbf 100644 --- a/esphome/yaml_util.py +++ b/esphome/yaml_util.py @@ -14,6 +14,7 @@ from pathlib import Path from typing import Any import uuid +from voluptuous import Invalid import yaml from yaml import SafeLoader as PurePythonLoader import yaml.constructor @@ -253,8 +254,6 @@ class IncludeFile: if self._content is not _UNSET: return self._content if self.has_unresolved_expressions(): - from esphome.config_validation import Invalid - raise Invalid( f"Cannot load include with unresolved substitutions: {self.file}" ) @@ -266,12 +265,133 @@ class IncludeFile: """Check if the filename contains substitution variables or Jinja expressions.""" return has_substitution_or_expression(str(self.file)) + def with_file(self, file: Path | str) -> IncludeFile: + """Clone this include with *file* as the filename.""" + return IncludeFile(self.parent_file, file, self.vars, self.yaml_loader) + + +def _is_visible_path(rel: Path) -> bool: + """Report whether no component of *rel* is hidden (``..`` stays valid).""" + return all(part == ".." or _is_file_valid(part) for part in rel.parts) + + +def _glob_include_candidates(parent_dir: Path, pattern: str) -> list[Path]: + """ + Expand a candidate glob under *parent_dir*, keeping hidden files out. + + An un-globbable pattern (absolute, or one the filesystem rejects) is + skipped instead of crashing discovery. + """ + try: + found_paths = parent_dir.glob(pattern) + return [ + rel + for found in found_paths + if _is_visible_path(rel := found.relative_to(parent_dir)) + ] + except (NotImplementedError, ValueError) as err: + _LOGGER.debug("Cannot glob include pattern %r: %s", pattern, err) + return [] + except OSError as err: + _LOGGER.warning("I/O error globbing include pattern %r: %s", pattern, err) + return [] + + +def _candidate_include_paths(include: IncludeFile) -> list[Path]: + """Enumerate resolved files an expression-templated ``!include`` could select. + + Wildcard patterns from ``substitutions.include_candidate_patterns`` glob + under the including file's directory with hidden files excluded (like + ``!include_dir_*``); literal branch patterns are tried verbatim. Matches + still carrying expression markers or pointing back at the including file + are skipped. + """ + # Deferred import — the substitutions component imports this module. + from esphome.components.substitutions import include_candidate_patterns + + parent_dir = include.parent_file.parent + parent_resolved = include.parent_file.resolve() + candidates: list[Path] = [] + for pattern in include_candidate_patterns(str(include.file)): + if "*" in pattern: + matches = sorted(_glob_include_candidates(parent_dir, pattern)) + else: + matches = [Path(pattern)] + for match in matches: + if has_substitution_or_expression(str(match)): + continue + candidate = parent_dir / match + if not candidate.is_file(): + continue + resolved = candidate.resolve() + if resolved == parent_resolved: + continue + candidates.append(resolved) + return candidates + + +def _load_include_candidates( + include: IncludeFile, + *, + warn_on_unresolved: bool, + seen: set[int], + expanded_paths: set[Path], + keepalive: list[Any], +) -> None: + """Load every filesystem candidate for an unresolved ``IncludeFile``.""" + log = _LOGGER.warning if warn_on_unresolved else _LOGGER.debug + candidates = _candidate_include_paths(include) + if not candidates: + log( + "Cannot resolve !include %s (referenced from %s) with substitutions in path", + include.file, + include.parent_file, + ) + return + _LOGGER.debug( + "Expanding !include %s (referenced from %s) to %d candidate file(s)", + include.file, + include.parent_file, + len(candidates), + ) + for candidate in candidates: + if candidate in expanded_paths: + continue + expanded_paths.add(candidate) + try: + loaded = include.with_file(candidate).load() + except (EsphomeError, Invalid) as err: + # Unlike an unresolved pattern (expected during the discovery + # re-parse), a matched on-disk candidate that fails to load is a + # genuine user error; warn in every mode. The file itself is + # still tracked (the load listener fires before parsing), only + # its nested includes go undiscovered. + _LOGGER.warning( + "Failed to load candidate %s for !include %s: %s", + candidate, + include.file, + err, + ) + continue + # The throwaway IncludeFile is this tree's only owner; keep the tree + # alive so ids recorded in ``seen`` stay unique for the traversal. + keepalive.append(loaded) + force_load_include_files( + loaded, + warn_on_unresolved=warn_on_unresolved, + _seen=seen, + _expanded_paths=expanded_paths, + _keepalive=keepalive, + ) + def force_load_include_files( obj: Any, *, warn_on_unresolved: bool = True, _seen: set[int] | None = None, + _expanded_paths: set[Path] | None = None, + _keepalive: list[Any] | None = None, ) -> None: """Recursively resolve any deferred ``IncludeFile`` instances in a YAML tree. @@ -282,29 +402,41 @@ def force_load_include_files( loader fires and records every reachable file. ``IncludeFile`` instances whose path contains unresolved substitution - variables cannot be loaded. By default a warning is logged for each one; - pass ``warn_on_unresolved=False`` (used by discovery paths that run on a - fresh re-parse where substitutions haven't been applied yet) to demote it - to a debug log. + variables or Jinja expressions are expanded against the filesystem and + every existing candidate file is loaded, so bundles ship all branches the + expression could select. By default a warning is logged when no candidate + exists; pass ``warn_on_unresolved=False`` (used by discovery paths that + run on a fresh re-parse where substitutions haven't been applied yet) to + demote it to a debug log. """ if _seen is None: _seen = set() + if _expanded_paths is None: + _expanded_paths = set() + if _keepalive is None: + # ``_seen`` tracks ids, which is only safe while every traversed + # object stays alive; candidate trees are otherwise freed between + # loop iterations and CPython recycles their addresses, making a + # fresh tree look already seen. Discovery is a one-shot operation, + # so holding the parsed trees costs nothing. + _keepalive = [] if isinstance(obj, IncludeFile): if id(obj) in _seen: return _seen.add(id(obj)) if obj.has_unresolved_expressions(): - log = _LOGGER.warning if warn_on_unresolved else _LOGGER.debug - log( - "Cannot resolve !include %s (referenced from %s) with substitutions in path", - obj.file, - obj.parent_file, + _load_include_candidates( + obj, + warn_on_unresolved=warn_on_unresolved, + seen=_seen, + expanded_paths=_expanded_paths, + keepalive=_keepalive, ) return try: loaded = obj.load() - except EsphomeError as err: + except (EsphomeError, Invalid) as err: _LOGGER.warning( "Failed to load !include %s (referenced from %s): %s", obj.file, @@ -313,7 +445,11 @@ def force_load_include_files( ) return force_load_include_files( - loaded, warn_on_unresolved=warn_on_unresolved, _seen=_seen + loaded, + warn_on_unresolved=warn_on_unresolved, + _seen=_seen, + _expanded_paths=_expanded_paths, + _keepalive=_keepalive, ) elif isinstance(obj, dict): if id(obj) in _seen: @@ -321,7 +457,11 @@ def force_load_include_files( _seen.add(id(obj)) for value in obj.values(): force_load_include_files( - value, warn_on_unresolved=warn_on_unresolved, _seen=_seen + value, + warn_on_unresolved=warn_on_unresolved, + _seen=_seen, + _expanded_paths=_expanded_paths, + _keepalive=_keepalive, ) elif isinstance(obj, (list, tuple)): if id(obj) in _seen: @@ -329,7 +469,11 @@ def force_load_include_files( _seen.add(id(obj)) for item in obj: force_load_include_files( - item, warn_on_unresolved=warn_on_unresolved, _seen=_seen + item, + warn_on_unresolved=warn_on_unresolved, + _seen=_seen, + _expanded_paths=_expanded_paths, + _keepalive=_keepalive, ) diff --git a/tests/unit_tests/test_bundle.py b/tests/unit_tests/test_bundle.py index f0abcc74c6..5c71b72d86 100644 --- a/tests/unit_tests/test_bundle.py +++ b/tests/unit_tests/test_bundle.py @@ -1248,7 +1248,8 @@ def test_discover_files_deeply_nested_include(tmp_path: Path) -> None: def test_discover_files_nested_include_unresolved_substitution( tmp_path: Path, ) -> None: - """!include with substitution vars in path cannot be resolved; skipped gracefully.""" + """!include with substitution vars in path but no candidate files on disk + (the glob's only match is the config itself) is skipped gracefully.""" config_dir = _setup_config_dir(tmp_path) (config_dir / "test.yaml").write_text( "esphome:\n name: test\nwifi: !include ${platform}.yaml\n" @@ -1262,6 +1263,62 @@ def test_discover_files_nested_include_unresolved_substitution( assert "test.yaml" in paths +def test_discover_files_bundles_all_include_candidates(tmp_path: Path) -> None: + """The issue-17650 layout: templated package includes chain through a glob + candidate into a Jinja conditional whose ``../`` branch is bundled.""" + config_dir = _setup_config_dir( + tmp_path, + files={ + "includes/esp-basics.yaml": ( + "packages:\n" + " - !include boards/${board}.yaml\n" + " - !include keys/${system_name}.yaml\n" + ), + "includes/boards/wemos-d1-mini.yaml": ( + 'packages:\n - !include ${ "NO BT.yaml" if bt else "../empty.yaml" }\n' + ), + "includes/keys/device-a.yaml": "api:\n", + "includes/keys/device-b.yaml": "api:\n", + "includes/empty.yaml": "{}\n", + }, + ) + (config_dir / "test.yaml").write_text( + "esphome:\n name: test\npackages:\n - !include includes/esp-basics.yaml\n" + ) + + creator = ConfigBundleCreator({}) + files = creator.discover_files() + + paths = [f.path for f in files] + assert "includes/esp-basics.yaml" in paths + assert "includes/boards/wemos-d1-mini.yaml" in paths + assert "includes/keys/device-a.yaml" in paths + assert "includes/keys/device-b.yaml" in paths + assert "includes/empty.yaml" in paths + + +def test_discover_files_candidate_outside_config_dir_skipped( + tmp_path: Path, caplog: pytest.LogCaptureFixture +) -> None: + """A candidate branch resolving above the config dir is not bundled.""" + config_dir = _setup_config_dir(tmp_path) + (tmp_path / "outside.yaml").write_text("api:\n") + (config_dir / "test.yaml").write_text( + "esphome:\n name: test\n" + 'wifi: !include ${ "a.yaml" if x else "../outside.yaml" }\n' + ) + + creator = ConfigBundleCreator({}) + files = creator.discover_files() + + paths = [f.path for f in files] + assert not any("outside" in p for p in paths) + assert any( + "outside config directory" in r.message and "outside.yaml" in r.message + for r in caplog.records + ) + + def test_discover_files_nested_include_load_failure( tmp_path: Path, caplog: pytest.LogCaptureFixture ) -> None: diff --git a/tests/unit_tests/test_substitutions.py b/tests/unit_tests/test_substitutions.py index baaa99f2a7..bcaf3fb354 100644 --- a/tests/unit_tests/test_substitutions.py +++ b/tests/unit_tests/test_substitutions.py @@ -1,3 +1,5 @@ +from collections import ChainMap +from fnmatch import fnmatchcase import logging from pathlib import Path from typing import Any @@ -961,3 +963,134 @@ def test_remote_package_scalar_yaml_raises_helpful_error( msg = str(exc_info.value) assert "mapping at the top level" in msg assert "file1.yaml" in msg + + +@pytest.mark.parametrize( + ("value", "expected"), + [ + pytest.param("wifi.yaml", ["wifi.yaml"], id="literal_passthrough"), + pytest.param( + "keys/${system_name}.yaml", ["keys/*.yaml"], id="embedded_substitution" + ), + pytest.param( + "network/${eth_model}/config.yaml", + ["network/*/config.yaml"], + id="directory_substitution", + ), + pytest.param( + "device-$platform.yaml", ["device-*.yaml"], id="unbraced_substitution" + ), + pytest.param("${a}${b}.yaml", ["*.yaml"], id="adjacent_wildcards_collapse"), + pytest.param( + '${ "a.yaml" if x else "../empty.yaml" }', + ["a.yaml", "../empty.yaml"], + id="conditional_literals", + ), + pytest.param( + 'pre-${ "a" if c else "b" }.yaml', + ["pre-a.yaml", "pre-b.yaml"], + id="conditional_spliced", + ), + pytest.param( + '${ "x.yaml" if a else ("y.yaml" if b else "z.yaml") }', + ["x.yaml", "y.yaml", "z.yaml"], + id="nested_conditional", + ), + pytest.param( + '${ "same.yaml" if x else "same.yaml" }', + ["same.yaml"], + id="duplicate_literals_dedupe", + ), + pytest.param('${ "a.yaml" if x }', ["a.yaml"], id="conditional_no_else"), + pytest.param( + '${ "NO BLUETOOTH SUPPORT ON ESP8266.yaml"' + ' if enable_bluetooth_proxy else "../empty.yaml" }', + ["NO BLUETOOTH SUPPORT ON ESP8266.yaml", "../empty.yaml"], + id="issue_17650_verbatim", + ), + pytest.param( + '${ "" if x else "b.yaml" }', ["b.yaml"], id="empty_literal_dropped" + ), + pytest.param( + "keys\\${system_name}.yaml", + ["keys\\*.yaml"], + id="backslash_separator", + ), + pytest.param( + '${ "it\'s.yaml" if x else "b.yaml" }', + ["it's.yaml", "b.yaml"], + id="apostrophe_in_literal", + ), + pytest.param( + '${ "a-${x}.yaml" if c else "b.yaml" }', + ["a-*.yaml", "b.yaml"], + id="substitution_inside_literal", + ), + pytest.param("sensor [${x}].yaml", ["sensor [[]*].yaml"], id="bracket_escaped"), + pytest.param( + "config?${x}.yaml", ["config[?]*.yaml"], id="question_mark_escaped" + ), + pytest.param( + "../${x}/config.yaml", ["../*/config.yaml"], id="ascending_directory" + ), + pytest.param("${file}", [], id="bare_variable_dropped"), + pytest.param("../${file}", [], id="ascending_bare_variable_dropped"), + pytest.param( + '${ name ~ ".yaml" }', [".yaml"], id="dynamic_concat_extracts_literal" + ), + pytest.param("${ if }", [], id="no_literal_expression_dropped"), + pytest.param( + "<% if x %>a.yaml<% endif %>", ["*a.yaml*"], id="block_statement_globs" + ), + ], +) +def test_include_candidate_patterns(value: str, expected: list[str]) -> None: + """Templated include paths expand to glob patterns and branch literals.""" + assert substitutions.include_candidate_patterns(value) == expected + + +@pytest.mark.parametrize( + ("template", "variables"), + [ + pytest.param( + "keys/${system_name}.yaml", {"system_name": "esp-buero"}, id="embedded" + ), + pytest.param("device-$platform.yaml", {"platform": "esp32"}, id="unbraced"), + pytest.param( + "network/${eth_model}/config.yaml", {"eth_model": "eth01"}, id="directory" + ), + pytest.param( + '${ "NO BT.yaml" if bt else "../empty.yaml" }', + {"bt": True}, + id="conditional_true", + ), + pytest.param( + '${ "NO BT.yaml" if bt else "../empty.yaml" }', + {"bt": False}, + id="conditional_false", + ), + pytest.param('pre-${ "a" if c else "b" }.yaml', {"c": True}, id="spliced"), + pytest.param("${a}${b}.yaml", {"a": "x", "b": "y"}, id="adjacent"), + pytest.param("sensor [${x}].yaml", {"x": "a"}, id="bracket"), + ], +) +def test_include_candidate_patterns_cover_real_expansion( + template: str, variables: dict[str, Any] +) -> None: + """ + Lockstep pin against the real substitution machinery. + + include_candidate_patterns mirrors _expand_substitutions without + variable values (the evaluator returns the one selected branch, so it + cannot enumerate candidates itself); this asserts every filename the + real pass resolves is covered by a candidate pattern, so a change to + reference syntax or expansion order breaks here instead of silently + dropping files from bundles. + """ + resolved = str( + substitutions._expand_substitutions( + template, [], ChainMap(variables), True, None + ) + ) + patterns = substitutions.include_candidate_patterns(template) + assert any(fnmatchcase(resolved, p) or resolved == p for p in patterns) diff --git a/tests/unit_tests/test_yaml_util.py b/tests/unit_tests/test_yaml_util.py index 5c38fce105..7a08ad2eb4 100644 --- a/tests/unit_tests/test_yaml_util.py +++ b/tests/unit_tests/test_yaml_util.py @@ -1003,8 +1003,10 @@ class _StubInclude: load_result: object = None, raise_on_load: EsphomeError | None = None, ) -> None: + # Default parent lives in a nonexistent directory so unresolved + # stubs never glob real files during candidate expansion. self.file = Path(file) - self.parent_file = parent_file or Path("/tmp/parent.yaml") + self.parent_file = parent_file or Path("/nonexistent/parent.yaml") self._unresolved = unresolved self._load_result = load_result if load_result is not None else {} self._raise = raise_on_load @@ -1182,6 +1184,247 @@ def test_discover_user_yaml_files_deduplicates(tmp_path: Path) -> None: assert discovered.files.count(wifi_resolved) == 1 +def test_discover_user_yaml_files_expands_directory_substitution( + tmp_path: Path, +) -> None: + """A substitution spanning a directory segment globs across directories.""" + _write(tmp_path, "network/eth01/config.yaml", "ethernet:\n") + _write(tmp_path, "network/eth02/config.yaml", "ethernet:\n") + discovered = discover_user_yaml_files( + _write_entry_including(tmp_path, "network/${eth_model}/config.yaml") + ) + resolved = set(discovered.files) + assert (tmp_path / "network/eth01/config.yaml").resolve() in resolved + assert (tmp_path / "network/eth02/config.yaml").resolve() in resolved + + +def test_discover_user_yaml_files_loads_both_branches_of_issue_conditional( + tmp_path: Path, +) -> None: + """Both branch files of the issue-17650 conditional load when present, + including the filename with spaces.""" + _write(tmp_path, "empty.yaml", "{}\n") + _write(tmp_path, "boards/NO BLUETOOTH SUPPORT ON ESP8266.yaml", "api:\n") + _write( + tmp_path, + "boards/esp8266.yaml", + "packages:\n" + ' - !include ${ "NO BLUETOOTH SUPPORT ON ESP8266.yaml"' + ' if enable_bluetooth_proxy else "../empty.yaml" }\n', + ) + discovered = discover_user_yaml_files( + _write_entry_including(tmp_path, "boards/esp8266.yaml") + ) + resolved = set(discovered.files) + assert (tmp_path / "boards/NO BLUETOOTH SUPPORT ON ESP8266.yaml").resolve() in ( + resolved + ) + assert (tmp_path / "empty.yaml").resolve() in resolved + + +def test_discover_user_yaml_files_glob_matches_bracket_filenames( + tmp_path: Path, +) -> None: + """Glob metacharacters in the literal filename text stay literal.""" + _write(tmp_path, "sensor [a].yaml", "api:\n") + discovered = discover_user_yaml_files( + _write_entry_including(tmp_path, "sensor [${x}].yaml") + ) + assert "sensor [a].yaml" in {p.name for p in discovered.files} + + +def test_discover_user_yaml_files_ascending_glob(tmp_path: Path) -> None: + """A templated include reaching into a sibling directory via ``..`` globs.""" + _write(tmp_path, "shared/common.yaml", "api:\n") + _write(tmp_path, "nodes/dev.yaml", "p: !include ../shared/${x}.yaml\n") + discovered = discover_user_yaml_files( + _write_entry_including(tmp_path, "nodes/dev.yaml") + ) + assert (tmp_path / "shared/common.yaml").resolve() in discovered.files + + +def test_discover_user_yaml_files_mapping_include_with_vars(tmp_path: Path) -> None: + """The mapping !include form (file + vars) expands a templated filename.""" + _write(tmp_path, "keys/a.yaml", "pin: ${num}\n") + entry = _write( + tmp_path, + "entry.yaml", + "wifi: !include\n file: keys/${n}.yaml\n vars:\n num: 4\n", + ) + discovered = discover_user_yaml_files(entry) + assert (tmp_path / "keys/a.yaml").resolve() in discovered.files + + +def test_discover_user_yaml_files_absolute_templated_include_skipped( + tmp_path: Path, caplog: pytest.LogCaptureFixture +) -> None: + """An absolute templated include is skipped gracefully instead of crashing.""" + shared = tmp_path / "shared" + _write(tmp_path, "shared/common.yaml", "api:\n") + with caplog.at_level("DEBUG", logger="esphome.yaml_util"): + discovered = discover_user_yaml_files( + _write_entry_including(tmp_path, f"{shared}/${{x}}.yaml") + ) + assert (shared / "common.yaml").resolve() not in discovered.files + assert any("Cannot glob include pattern" in r.message for r in caplog.records) + + +def test_discover_user_yaml_files_glob_skips_dollar_named_files( + tmp_path: Path, +) -> None: + """An on-disk filename containing ``$`` can't load; the glob skips it.""" + _write(tmp_path, "keys/a.yaml", "api:\n") + _write(tmp_path, "keys/b$roken.yaml", "api:\n") + discovered = discover_user_yaml_files( + _write_entry_including(tmp_path, "keys/${n}.yaml") + ) + names = {p.name for p in discovered.files} + assert "a.yaml" in names + assert "b$roken.yaml" not in names + + +def test_discover_user_yaml_files_glob_error_skips_include( + tmp_path: Path, caplog: pytest.LogCaptureFixture +) -> None: + """A filesystem error during candidate globbing warns and skips the include.""" + entry = _write_entry_including(tmp_path, "keys/${n}.yaml") + with ( + patch.object(Path, "glob", side_effect=OSError("boom")), + caplog.at_level("DEBUG", logger="esphome.yaml_util"), + ): + discovered = discover_user_yaml_files(entry) + assert [p.name for p in discovered.files] == ["entry.yaml"] + matching = [ + r.levelname + for r in caplog.records + if "I/O error globbing include pattern" in r.message + ] + assert matching == ["WARNING"] + + +def test_force_load_candidate_failure_warns_by_default( + tmp_path: Path, caplog: pytest.LogCaptureFixture +) -> None: + """A broken candidate logs at WARNING outside the discovery re-parse.""" + _write(tmp_path, "keys/bad.yaml", "esphome: [unterminated\n") + entry = _write_entry_including(tmp_path, "keys/${n}.yaml") + with caplog.at_level("DEBUG", logger="esphome.yaml_util"): + force_load_include_files(yaml_util.load_yaml(entry)) + matching = [ + r.levelname for r in caplog.records if "Failed to load candidate" in r.message + ] + assert matching == ["WARNING"] + + +def test_discover_user_yaml_files_glob_skips_hidden_files(tmp_path: Path) -> None: + """Candidate globs exclude hidden files, matching ``!include_dir_*``.""" + _write(tmp_path, "keys/device-a.yaml", "api:\n") + _write(tmp_path, "keys/.hidden.yaml", "api:\n") + discovered = discover_user_yaml_files( + _write_entry_including(tmp_path, "keys/${name}.yaml") + ) + names = {p.name for p in discovered.files} + assert "device-a.yaml" in names + assert ".hidden.yaml" not in names + + +def test_discover_user_yaml_files_bare_expression_not_expanded( + tmp_path: Path, caplog: pytest.LogCaptureFixture +) -> None: + """A fully dynamic filename never globs the whole directory.""" + _write(tmp_path, "sibling.yaml", "api:\n") + with caplog.at_level("DEBUG", logger="esphome.yaml_util"): + discovered = discover_user_yaml_files( + _write_entry_including(tmp_path, "${file}") + ) + assert (tmp_path / "sibling.yaml").resolve() not in discovered.files + assert any( + "Cannot resolve !include" in r.message and r.levelname == "DEBUG" + for r in caplog.records + ) + + +def test_discover_user_yaml_files_self_glob_match_skipped( + tmp_path: Path, caplog: pytest.LogCaptureFixture +) -> None: + """A glob whose only match is the including file itself claims nothing.""" + entry = _write_entry_including(tmp_path, "${platform}.yaml") + with caplog.at_level("DEBUG", logger="esphome.yaml_util"): + discovered = discover_user_yaml_files(entry) + assert [p.name for p in discovered.files] == ["entry.yaml"] + assert any("Cannot resolve !include" in r.message for r in caplog.records) + + +def test_discover_user_yaml_files_candidate_cycle_terminates(tmp_path: Path) -> None: + """Mutually glob-matching includes expand finitely and capture both files.""" + _write(tmp_path, "sub/a.yaml", "p: !include ${x}.yaml\n") + _write(tmp_path, "sub/b.yaml", "p: !include ${y}.yaml\n") + entry = _write(tmp_path, "entry.yaml", "wifi: !include sub/a.yaml\n") + discovered = discover_user_yaml_files(entry) + names = {p.name for p in discovered.files} + assert names == {"entry.yaml", "a.yaml", "b.yaml"} + + +def test_discover_user_yaml_files_many_candidates_keep_nested_includes( + tmp_path: Path, +) -> None: + """Every candidate's nested includes are discovered. + + Regression test: the id()-based cycle guard is only safe while every + traversed tree stays alive. Candidate trees used to be freed between + loop iterations, so CPython recycled their addresses and later + candidates' fresh trees were skipped as already seen, silently dropping + their nested includes. Needs several candidates to manifest; two were + not enough to trigger the reuse.""" + count = 12 + for i in range(count): + _write( + tmp_path, f"keys/k{i}.yaml", f"sensor{i}: !include ../nested/n{i}.yaml\n" + ) + _write(tmp_path, f"nested/n{i}.yaml", f"api{i}: true\n") + discovered = discover_user_yaml_files( + _write_entry_including(tmp_path, "keys/${x}.yaml") + ) + names = {p.name for p in discovered.files} + expected = {f"n{i}.yaml" for i in range(count)} + expected |= {f"k{i}.yaml" for i in range(count)} + expected.add("entry.yaml") + assert names == expected + + +def test_discover_user_yaml_files_bad_candidate_still_tracked( + tmp_path: Path, caplog: pytest.LogCaptureFixture +) -> None: + """A matched candidate that fails to parse warns even during discovery, + stays tracked (the load listener fires before parsing), and doesn't block + other candidates.""" + _write(tmp_path, "keys/good.yaml", "api:\n") + _write(tmp_path, "keys/bad.yaml", "esphome: [unterminated\n") + with caplog.at_level("DEBUG", logger="esphome.yaml_util"): + discovered = discover_user_yaml_files( + _write_entry_including(tmp_path, "keys/${name}.yaml") + ) + resolved = set(discovered.files) + assert (tmp_path / "keys/good.yaml").resolve() in resolved + assert (tmp_path / "keys/bad.yaml").resolve() in resolved + matching = [ + r.levelname for r in caplog.records if "Failed to load candidate" in r.message + ] + assert matching == ["WARNING"] + + +def test_discover_user_yaml_files_tolerates_templated_top_level_include( + tmp_path: Path, +) -> None: + """A literal include whose entire content is a templated ``!include`` is + tracked and skipped instead of aborting discovery.""" + _write(tmp_path, "wrapper.yaml", "!include ${x}_settings.yaml\n") + discovered = discover_user_yaml_files( + _write_entry_including(tmp_path, "wrapper.yaml") + ) + assert (tmp_path / "wrapper.yaml").resolve() in discovered.files + + def test_track_yaml_loads_records_resolved_paths(tmp_path: Path) -> None: """`track_yaml_loads` is the building block — sanity-check it resolves symlinks so callers can dedupe by identity."""