github/esphome
2
0
Fork 0
mirror of https://github.com/esphome/esphome.git synced 2026-09-03 19:46:02 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'upstream/dev' into integration

Browse Source
This commit is contained in:
J. Nick Koston
2026-04-22 05:41:38 +02:00
parent bfa5cf697f bb81c91d0c
commit d9cb0a8f2d
13 changed files with 712 additions and 133 deletions
Download Patch File Download Diff File Expand all files Collapse all files
esphome/components/bl0906/bl0906.cpp
+47 -20
View File
@@ -20,58 +20,77 @@ constexpr uint8_t bl0906_checksum(const uint8_t address, const DataPacket *data)
}
void BL0906::loop() {
if (this->current_channel_ == UINT8_MAX) {
return;
}
while (this->available())
this->flush();
if (this->current_channel_ == 0) {
if (this->current_stage_ == STAGE_IDLE) {
// Woken up between cycles to drain the action queue. Go back to sleep.
this->handle_actions_();
this->disable_loop();
return;
}
if (this->current_stage_ == STAGE_TEMP) {
// Temperature
this->read_data_(BL0906_TEMPERATURE, BL0906_TREF, this->temperature_sensor_);
} else if (this->current_channel_ == 1) {
} else if (this->current_stage_ == STAGE_CHANNEL_1) {
this->read_data_(BL0906_I_1_RMS, BL0906_IREF, this->current_1_sensor_);
this->read_data_(BL0906_WATT_1, BL0906_PREF, this->power_1_sensor_);
this->read_data_(BL0906_CF_1_CNT, BL0906_EREF, this->energy_1_sensor_);
} else if (this->current_channel_ == 2) {
} else if (this->current_stage_ == STAGE_CHANNEL_2) {
this->read_data_(BL0906_I_2_RMS, BL0906_IREF, this->current_2_sensor_);
this->read_data_(BL0906_WATT_2, BL0906_PREF, this->power_2_sensor_);
this->read_data_(BL0906_CF_2_CNT, BL0906_EREF, this->energy_2_sensor_);
} else if (this->current_channel_ == 3) {
} else if (this->current_stage_ == STAGE_CHANNEL_3) {
this->read_data_(BL0906_I_3_RMS, BL0906_IREF, this->current_3_sensor_);
this->read_data_(BL0906_WATT_3, BL0906_PREF, this->power_3_sensor_);
this->read_data_(BL0906_CF_3_CNT, BL0906_EREF, this->energy_3_sensor_);
} else if (this->current_channel_ == 4) {
} else if (this->current_stage_ == STAGE_CHANNEL_4) {
this->read_data_(BL0906_I_4_RMS, BL0906_IREF, this->current_4_sensor_);
this->read_data_(BL0906_WATT_4, BL0906_PREF, this->power_4_sensor_);
this->read_data_(BL0906_CF_4_CNT, BL0906_EREF, this->energy_4_sensor_);
} else if (this->current_channel_ == 5) {
} else if (this->current_stage_ == STAGE_CHANNEL_5) {
this->read_data_(BL0906_I_5_RMS, BL0906_IREF, this->current_5_sensor_);
this->read_data_(BL0906_WATT_5, BL0906_PREF, this->power_5_sensor_);
this->read_data_(BL0906_CF_5_CNT, BL0906_EREF, this->energy_5_sensor_);
} else if (this->current_channel_ == 6) {
} else if (this->current_stage_ == STAGE_CHANNEL_6) {
this->read_data_(BL0906_I_6_RMS, BL0906_IREF, this->current_6_sensor_);
this->read_data_(BL0906_WATT_6, BL0906_PREF, this->power_6_sensor_);
this->read_data_(BL0906_CF_6_CNT, BL0906_EREF, this->energy_6_sensor_);
} else if (this->current_channel_ == UINT8_MAX - 2) {
} else if (this->current_stage_ == STAGE_FREQ) {
// Frequency
this->read_data_(BL0906_FREQUENCY, BL0906_FREF, frequency_sensor_);
this->read_data_(BL0906_FREQUENCY, BL0906_FREF, this->frequency_sensor_);
// Voltage
this->read_data_(BL0906_V_RMS, BL0906_UREF, voltage_sensor_);
} else if (this->current_channel_ == UINT8_MAX - 1) {
this->read_data_(BL0906_V_RMS, BL0906_UREF, this->voltage_sensor_);
} else if (this->current_stage_ == STAGE_POWER) {
// Total power
this->read_data_(BL0906_WATT_SUM, BL0906_WATT, this->total_power_sensor_);
// Total Energy
this->read_data_(BL0906_CF_SUM_CNT, BL0906_CF, this->total_energy_sensor_);
} else {
this->current_channel_ = UINT8_MAX - 2; // Go to frequency and voltage
return;
}
this->current_channel_++;
this->advance_stage_();
this->handle_actions_();
}
void BL0906::advance_stage_() {
switch (this->current_stage_) {
case STAGE_CHANNEL_6:
this->current_stage_ = STAGE_FREQ;
break;
case STAGE_FREQ:
this->current_stage_ = STAGE_POWER;
break;
case STAGE_POWER:
// Cycle complete; sleep until the next update().
this->current_stage_ = STAGE_IDLE;
this->disable_loop();
break;
default:
this->current_stage_ = static_cast<BL0906Stage>(this->current_stage_ + 1);
break;
}
}
void BL0906::setup() {
while (this->available())
this->flush();
@@ -85,12 +104,20 @@ void BL0906::setup() {
this->bias_correction_(BL0906_RMSOS_6, 0.01200, 0); // Calibration current_6
this->write_array(USR_WRPROT_ONLYREAD, sizeof(USR_WRPROT_ONLYREAD));
// Loop stays idle until the first update() or enqueued action.
this->disable_loop();
}
void BL0906::update() { this->current_channel_ = 0; }
void BL0906::update() {
this->current_stage_ = STAGE_TEMP;
this->enable_loop();
}
size_t BL0906::enqueue_action_(ActionCallbackFuncPtr function) {
this->action_queue_.push_back(function);
// Ensure the queue is serviced even if the read cycle has already completed.
this->enable_loop();
return this->action_queue_.size();
}
esphome/components/bl0906/bl0906.h
+18 -1
View File
@@ -12,6 +12,22 @@
namespace esphome {
namespace bl0906 {
// Stage values for the read state machine. After STAGE_CHANNEL_6 the state machine
// jumps to the two sentinel stages below, then to STAGE_IDLE which marks the cycle
// as complete and disables the loop.
enum BL0906Stage : uint8_t {
STAGE_TEMP = 0, // chip temperature
STAGE_CHANNEL_1 = 1, // per-phase current + power + energy
STAGE_CHANNEL_2 = 2,
STAGE_CHANNEL_3 = 3,
STAGE_CHANNEL_4 = 4,
STAGE_CHANNEL_5 = 5,
STAGE_CHANNEL_6 = 6,
STAGE_FREQ = UINT8_MAX - 2, // frequency + voltage
STAGE_POWER = UINT8_MAX - 1, // total power + total energy
STAGE_IDLE = UINT8_MAX, // cycle complete
};
struct DataPacket { // NOLINT(altera-struct-pack-align)
uint8_t l{0};
uint8_t m{0};
@@ -79,7 +95,8 @@ class BL0906 : public PollingComponent, public uart::UARTDevice {
void bias_correction_(uint8_t address, float measurements, float correction);
uint8_t current_channel_{0};
BL0906Stage current_stage_{STAGE_IDLE};
void advance_stage_();
size_t enqueue_action_(ActionCallbackFuncPtr function);
void handle_actions_();
esphome/components/esp32/__init__.py
+75 -20
View File
@@ -128,23 +128,30 @@ ASSERTION_LEVELS = {
SIGNING_SCHEMES = {
"rsa3072": "CONFIG_SECURE_SIGNED_APPS_RSA_SCHEME",
"ecdsa256": "CONFIG_SECURE_SIGNED_APPS_ECDSA_V2_SCHEME",
"ecdsa_v1": "CONFIG_SECURE_SIGNED_APPS_ECDSA_SCHEME",
}
# Chip variants that only support one signing scheme for Secure Boot V2.
# Chip variants that only support one V2 signing scheme.
# Based on SOC_SECURE_BOOT_V2_RSA / SOC_SECURE_BOOT_V2_ECC in soc_caps.h.
# Variants not listed in either set support both RSA and ECDSA
# Variants not listed in either set support both RSA and ECDSA V2
# (e.g. C5, C6, H2, P4). New variants should be added to the
# appropriate set if they only support one scheme.
SIGNED_OTA_RSA_ONLY_VARIANTS = {
VARIANT_ESP32,
# Note: VARIANT_ESP32 is not listed here because it supports V2 RSA only
# when minimum_chip_revision >= 3.0, which requires special handling.
SIGNED_OTA_V2_RSA_ONLY_VARIANTS = {
VARIANT_ESP32S2,
VARIANT_ESP32S3,
VARIANT_ESP32C3,
}
SIGNED_OTA_ECC_ONLY_VARIANTS = {
SIGNED_OTA_V2_ECC_ONLY_VARIANTS = {
VARIANT_ESP32C2,
VARIANT_ESP32C61,
}
# V1 ECDSA (Secure Boot V1) is only supported on the original ESP32.
# Based on SOC_SECURE_BOOT_V1 in soc_caps.h.
SIGNED_OTA_V1_ECDSA_VARIANTS = {
VARIANT_ESP32,
}
COMPILER_OPTIMIZATIONS = {
"DEBUG": "CONFIG_COMPILER_OPTIMIZATION_DEBUG",
@@ -991,25 +998,73 @@ def final_validate(config):
if signed_ota := advanced.get(CONF_SIGNED_OTA_VERIFICATION):
scheme = signed_ota[CONF_SIGNING_SCHEME]
variant = config[CONF_VARIANT]
scheme_variant_conflicts = {
"ecdsa256": (SIGNED_OTA_RSA_ONLY_VARIANTS, "rsa3072"),
"rsa3072": (SIGNED_OTA_ECC_ONLY_VARIANTS, "ecdsa256"),
}
if (conflict := scheme_variant_conflicts.get(scheme)) and variant in conflict[
0
]:
min_rev = advanced.get(CONF_MINIMUM_CHIP_REVISION)
scheme_path = [
CONF_FRAMEWORK,
CONF_ADVANCED,
CONF_SIGNED_OTA_VERIFICATION,
CONF_SIGNING_SCHEME,
]
# V1 ECDSA is only available on the original ESP32
if scheme == "ecdsa_v1" and variant not in SIGNED_OTA_V1_ECDSA_VARIANTS:
errs.append(
cv.Invalid(
f"Signing scheme '{scheme}' is not supported on "
f"{VARIANT_FRIENDLY[variant]}. Use '{conflict[1]}' instead.",
path=[
CONF_FRAMEWORK,
CONF_ADVANCED,
CONF_SIGNED_OTA_VERIFICATION,
CONF_SIGNING_SCHEME,
],
f"Signing scheme 'ecdsa_v1' is only supported on "
f"{VARIANT_FRIENDLY[VARIANT_ESP32]}. "
f"Use 'rsa3072' or 'ecdsa256' instead.",
path=scheme_path,
)
)
elif variant == VARIANT_ESP32:
# On ESP32, V2 RSA requires minimum_chip_revision >= 3.0
# Note: string comparison works here because cv.one_of constrains
# min_rev to known ESP32_CHIP_REVISIONS values ("0.0".."3.1").
if scheme == "rsa3072" and (min_rev is None or min_rev < "3.0"):
errs.append(
cv.Invalid(
f"Signing scheme 'rsa3072' on {VARIANT_FRIENDLY[variant]} "
f"requires minimum_chip_revision: '3.0' or higher "
f"(Secure Boot V2 RSA needs chip revision 3.0+). "
f"For older chip revisions, use 'ecdsa_v1' instead.",
path=scheme_path,
)
)
# ESP32 does not support V2 ECDSA (no SOC_SECURE_BOOT_V2_ECC)
elif scheme == "ecdsa256":
errs.append(
cv.Invalid(
f"Signing scheme 'ecdsa256' is not supported on "
f"{VARIANT_FRIENDLY[variant]}. Use 'rsa3072' (with "
f"minimum_chip_revision: '3.0') or 'ecdsa_v1' instead.",
path=scheme_path,
)
)
# V1 on rev 3.0+ -- suggest V2 RSA for stronger security
elif scheme == "ecdsa_v1" and min_rev is not None and min_rev >= "3.0":
_LOGGER.info(
"Using Secure Boot V1 ECDSA on %s rev %s. "
"Consider using 'rsa3072' (Secure Boot V2 RSA) for "
"stronger security on chip revision 3.0+.",
VARIANT_FRIENDLY[variant],
min_rev,
)
else:
# Non-ESP32 variants: check V2 scheme-variant compatibility
scheme_variant_conflicts = {
"ecdsa256": (SIGNED_OTA_V2_RSA_ONLY_VARIANTS, "rsa3072"),
"rsa3072": (SIGNED_OTA_V2_ECC_ONLY_VARIANTS, "ecdsa256"),
}
if (
conflict := scheme_variant_conflicts.get(scheme)
) and variant in conflict[0]:
errs.append(
cv.Invalid(
f"Signing scheme '{scheme}' is not supported on "
f"{VARIANT_FRIENDLY[variant]}. Use '{conflict[1]}' instead.",
path=scheme_path,
)
)
if CONF_OTA not in full_config:
_LOGGER.warning(
"Signed OTA verification is enabled but no OTA component is configured. "
esphome/components/esp32/post_build.py.script
+120 -3
View File
@@ -5,6 +5,7 @@ import json # noqa: E402
import os # noqa: E402
import pathlib # noqa: E402
import shutil # noqa: E402
import subprocess # noqa: E402
from glob import glob # noqa: E402
@@ -25,6 +26,114 @@ def _parse_sdkconfig(sdkconfig_path):
return options
def _generate_v1_verification_key(env):
"""Generate the V1 ECDSA verification key binary and assembly source file.
Secure Boot V1 embeds the public verification key directly in the app binary
as a compiled object (via a .S assembly file). The ESP-IDF CMake build generates
these files via custom commands, but PlatformIO's SCons bridge does not execute
them. This function replicates that logic:
1. Extracts the raw public key from the PEM signing key using espsecure.
2. Generates the .S assembly source that embeds the key bytes.
"""
build_dir = pathlib.Path(env.subst("$BUILD_DIR"))
project_dir = pathlib.Path(env.subst("$PROJECT_DIR"))
pioenv = env.subst("$PIOENV")
sdkconfig = _parse_sdkconfig(project_dir / f"sdkconfig.{pioenv}")
if sdkconfig.get("CONFIG_SECURE_SIGNED_APPS_ECDSA_SCHEME") != "y":
return
bin_path = build_dir / "signature_verification_key.bin"
asm_path = build_dir / "signature_verification_key.bin.S"
# Determine the source of the verification key
if sdkconfig.get("CONFIG_SECURE_BOOT_BUILD_SIGNED_BINARIES") == "y":
# Extract public key from the signing key
signing_key = sdkconfig.get("CONFIG_SECURE_BOOT_SIGNING_KEY")
if not signing_key:
return
signing_key_path = pathlib.Path(signing_key)
if not signing_key_path.exists():
print(f"Error: V1 ECDSA signing key not found: {signing_key_path}")
env.Exit(1)
return
if not bin_path.exists() or bin_path.stat().st_mtime < signing_key_path.stat().st_mtime:
python_exe = env.subst("$PYTHONEXE")
result = subprocess.run(
[python_exe, "-m", "espsecure", "extract_public_key",
"--keyfile", str(signing_key_path), str(bin_path)],
capture_output=True, text=True,
)
if result.returncode != 0:
print(f"Error extracting V1 verification key: {result.stderr}")
env.Exit(1)
return
print(f"Extracted V1 ECDSA verification key from {signing_key_path.name}")
else:
# User-provided verification key -- should already be a raw binary file
verification_key = sdkconfig.get("CONFIG_SECURE_BOOT_VERIFICATION_KEY")
if not verification_key:
return
verification_key_path = pathlib.Path(verification_key)
if not verification_key_path.exists():
print(f"Error: Verification key not found: {verification_key_path}")
env.Exit(1)
return
shutil.copyfile(str(verification_key_path), str(bin_path))
if not bin_path.exists():
return
# Generate the .S assembly file from the binary key data.
# Replicates ESP-IDF's data_file_embed_asm.cmake with RENAME_TO=signature_verification_key_bin.
# The file is needed in both the app build dir and the bootloader build dir, since
# the bootloader also embeds the verification key when CONFIG_SECURE_SIGNED_ON_BOOT_NO_SECURE_BOOT
# is enabled. PlatformIO's SCons bridge does not execute the CMake custom commands that
# normally generate these files.
data = bin_path.read_bytes()
varname = "signature_verification_key_bin"
lines = []
lines.append(f"/* Data converted from {bin_path.name} */")
lines.append(".data")
lines.append("#if !defined (__APPLE__) && !defined (__linux__)")
lines.append(".section .rodata.embedded")
lines.append("#endif")
lines.append(f"\n.global {varname}")
lines.append(f"{varname}:")
lines.append(f"\n.global _binary_{varname}_start")
lines.append(f"_binary_{varname}_start: /* for objcopy compatibility */")
# Format binary data as .byte lines (16 bytes per line)
for i in range(0, len(data), 16):
chunk = data[i:i + 16]
hex_bytes = ", ".join(f"0x{b:02x}" for b in chunk)
lines.append(f".byte {hex_bytes}")
lines.append(f"\n.global _binary_{varname}_end")
lines.append(f"_binary_{varname}_end: /* for objcopy compatibility */")
lines.append(f"\n.global {varname}_length")
lines.append(f"{varname}_length:")
lines.append(f".long {len(data)}")
lines.append("")
lines.append('#if defined (__linux__)')
lines.append('.section .note.GNU-stack,"",@progbits')
lines.append("#endif")
asm_content = "\n".join(lines) + "\n"
# Write to app build dir and bootloader build dir
asm_path.write_text(asm_content)
bootloader_dir = build_dir / "bootloader"
if bootloader_dir.is_dir():
bootloader_bin = bootloader_dir / "signature_verification_key.bin"
bootloader_asm = bootloader_dir / "signature_verification_key.bin.S"
shutil.copyfile(str(bin_path), str(bootloader_bin))
bootloader_asm.write_text(asm_content)
def sign_firmware(source, target, env):
"""
Sign the firmware binary using espsecure.py if signed OTA verification is enabled.
@@ -55,9 +164,12 @@ def sign_firmware(source, target, env):
env.Exit(1)
return
# ESPHome only exposes RSA3072 and ECDSA256 (both Secure Boot V2 schemes),
# so the espsecure signature version is always 2.
sign_version = "2"
# Determine espsecure signature version from the signing scheme:
# V1 ECDSA (Secure Boot V1) uses --version 1, V2 RSA/ECDSA use --version 2.
if sdkconfig.get("CONFIG_SECURE_SIGNED_APPS_ECDSA_SCHEME") == "y":
sign_version = "1"
else:
sign_version = "2"
firmware_name = os.path.basename(env.subst("$PROGNAME")) + ".bin"
firmware_path = build_dir / firmware_name
@@ -217,6 +329,11 @@ def esp32_copy_ota_bin(source, target, env):
print(f"Copied firmware to {new_file_name}")
# Generate V1 ECDSA verification key files before build starts.
# Workaround for PlatformIO not executing CMake custom commands that extract
# the public key and generate the .S assembly file for Secure Boot V1.
_generate_v1_verification_key(env) # noqa: F821
# Run signing first, then merge, then ota copy
env.AddPostAction("$BUILD_DIR/${PROGNAME}.bin", sign_firmware) # noqa: F821
env.AddPostAction("$BUILD_DIR/${PROGNAME}.bin", merge_factory_bin) # noqa: F821
esphome/components/packages/__init__.py
+45 -19
View File
@@ -42,6 +42,11 @@ DOMAIN = CONF_PACKAGES
# Guard against infinite include chains (e.g. A includes B includes A).
MAX_INCLUDE_DEPTH = 20
PackageCallback = Callable[
[dict | str | yaml_util.IncludeFile, ContextVars | None, yaml_util.DocumentPath],
dict,
]
def is_remote_package(package_config: dict) -> bool:
"""Returns True if the package_config is a remote package definition."""
@@ -281,8 +286,9 @@ def _process_remote_package(config: dict, skip_update: bool = False) -> dict:
def _walk_package_dict(
packages: dict,
callback: Callable[[dict, ContextVars | None], dict],
callback: PackageCallback,
context: ContextVars | None,
path: yaml_util.DocumentPath,
) -> cv.Invalid | None:
"""Iterate a packages dict in reverse priority order, invoking callback on each entry.
@@ -291,7 +297,9 @@ def _walk_package_dict(
for package_name, package_config in reversed(packages.items()):
with cv.prepend_path(package_name):
try:
packages[package_name] = callback(package_config, context)
packages[package_name] = callback(
package_config, context, path + [package_name]
)
except cv.Invalid as err:
return err
return None
@@ -299,20 +307,22 @@ def _walk_package_dict(
def _walk_package_list(
packages: list,
callback: Callable[[dict, ContextVars | None], dict],
callback: PackageCallback,
context: ContextVars | None,
path: yaml_util.DocumentPath,
) -> None:
"""Iterate a packages list in reverse priority order, invoking callback on each entry."""
for idx in reversed(range(len(packages))):
with cv.prepend_path(idx):
packages[idx] = callback(packages[idx], context)
packages[idx] = callback(packages[idx], context, path + [idx])
def _walk_packages(
config: dict,
callback: Callable[[dict, ContextVars | None], dict],
callback: PackageCallback,
context: ContextVars | None = None,
validate_deprecated: bool = True,
path: yaml_util.DocumentPath | None = None,
) -> dict:
"""Walks the packages structure in priority order, invoking ``callback`` on each package definition found.
@@ -323,19 +333,24 @@ def _walk_packages(
if CONF_PACKAGES not in config:
return config
packages = config[CONF_PACKAGES]
packages_path = (path or []) + [CONF_PACKAGES]
with cv.prepend_path(CONF_PACKAGES):
if isinstance(packages, yaml_util.IncludeFile):
# If the packages key is an IncludeFile, resolve it first before processing.
packages, _ = resolve_include(packages, [], context, strict_undefined=False)
packages = resolve_include(
packages, packages_path, context, strict_undefined=False
)
if not isinstance(packages, (dict, list)):
raise cv.Invalid(
f"Packages must be a key to value mapping or list, got {type(packages)} instead"
)
if not isinstance(packages, dict):
_walk_package_list(packages, callback, context)
elif (result := _walk_package_dict(packages, callback, context)) is not None:
_walk_package_list(packages, callback, context, packages_path)
elif (
result := _walk_package_dict(packages, callback, context, packages_path)
) is not None:
if not validate_deprecated or any(
is_package_definition(v) for v in packages.values()
):
@@ -344,14 +359,18 @@ def _walk_packages(
# This block can be removed once the single-package
# deprecation period (2026.7.0) is over.
config[CONF_PACKAGES] = [packages]
return _walk_packages(deprecate_single_package(config), callback, context)
return _walk_packages(
deprecate_single_package(config), callback, context, path=path
)
config[CONF_PACKAGES] = packages
return config
def _substitute_package_definition(
package_config: dict | str, context_vars: ContextVars | None
package_config: dict | str,
context_vars: ContextVars | None,
path: yaml_util.DocumentPath | None = None,
) -> dict | str:
"""Substitute variables in a package definition string or remote package dict.
@@ -369,12 +388,12 @@ def _substitute_package_definition(
errors: ErrList = []
package_config = substitute(
item=package_config,
path=[],
path=path or [],
parent_context=context_vars or ContextVars(),
strict_undefined=False,
errors=errors,
)
raise_first_undefined(errors, package_config, "package definition")
raise_first_undefined(errors, "package definition")
return package_config
@@ -432,6 +451,7 @@ class _PackageProcessor:
self,
package_config: dict | str | yaml_util.IncludeFile,
context_vars: ContextVars | None,
path: yaml_util.DocumentPath,
) -> dict:
"""Resolve a package definition to a concrete ``dict`` and fetch remote packages.
@@ -454,15 +474,15 @@ class _PackageProcessor:
"""
for _ in range(MAX_INCLUDE_DEPTH):
if isinstance(package_config, yaml_util.IncludeFile):
package_config, _ = resolve_include(
package_config = resolve_include(
package_config,
[],
path,
context_vars or ContextVars(),
strict_undefined=False,
)
package_config = _substitute_package_definition(
package_config, context_vars
package_config, context_vars, path
)
package_config = PACKAGE_SCHEMA(package_config)
if isinstance(package_config, dict):
@@ -483,13 +503,16 @@ class _PackageProcessor:
_update_substitutions_context(self.parent_context, subs)
def process_package(
self, package_config: dict | str, context_vars: ContextVars | None
self,
package_config: dict | str,
context_vars: ContextVars | None,
path: yaml_util.DocumentPath,
) -> dict:
"""Resolve a single package and recurse into any nested packages."""
from_remote = isinstance(package_config, dict) and is_remote_package(
package_config
)
package_config = self.resolve_package(package_config, context_vars)
package_config = self.resolve_package(package_config, context_vars, path)
self.collect_substitutions(package_config)
if CONF_PACKAGES not in package_config:
@@ -509,6 +532,7 @@ class _PackageProcessor:
self.process_package,
context_vars,
validate_deprecated=not from_remote,
path=path,
)
@@ -565,11 +589,13 @@ def merge_packages(config: dict) -> dict:
merge_list: list[dict] = []
def process_package_callback(
package_config: dict, context: ContextVars | None
package_config: dict,
context: ContextVars | None,
path: yaml_util.DocumentPath | None = None,
) -> dict:
"""This will be called for each package found in the config."""
merge_list.append(package_config)
return _walk_packages(package_config, process_package_callback)
return _walk_packages(package_config, process_package_callback, path=path)
_walk_packages(config, process_package_callback, validate_deprecated=False)
# Merge all packages into the main config:
esphome/components/substitutions/__init__.py
+29 -54
View File
@@ -11,9 +11,11 @@ from esphome.types import ConfigType
from esphome.util import OrderedDict
from esphome.yaml_util import (
ConfigContext,
DocumentPath,
ESPHomeDataBase,
ESPLiteralValue,
IncludeFile,
format_path,
make_data_base,
)
@@ -23,8 +25,8 @@ CODEOWNERS = ["@esphome/core"]
_LOGGER = logging.getLogger(__name__)
ContextVars = ChainMap[str, Any]
SubstitutionPath = list[int | str]
ErrList = list[tuple[UndefinedError, SubstitutionPath, Any]]
ErrList = list[tuple[UndefinedError, DocumentPath, Any]]
# 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()
@@ -32,16 +34,13 @@ jinja = Jinja()
def raise_first_undefined(
errors: ErrList,
source: Any,
context_label: str,
) -> None:
"""If *errors* is non-empty, raise ``cv.Invalid`` for the first undefined variable.
The raised error names the missing variable, the path walked into *source*
(for nested dicts, e.g. ``url`` or ``ref``), and the YAML source location
when *source* carries one. Only the first error is surfaced; the user will
re-run after fixing it and any remaining undefined variables will be
reported then.
The raised error names the missing variable and its location in the include
stack. Only the first error is surfaced; the user will re-run after fixing it
and any remaining undefined variables will be reported then.
``context_label`` is the noun describing where the undefined variable
appeared (e.g. ``"package definition"``).
@@ -57,26 +56,8 @@ def raise_first_undefined(
for e, p_path, _ in errors[1:]
)
_LOGGER.debug("Additional undefined variables in %s: %s", context_label, extras)
# Prefer the location of the offending scalar (e.g. the `url:` value) over
# the enclosing package-definition dict so the message points at the exact
# line/column that carries the undefined variable.
location_node = (
err_value
if isinstance(err_value, ESPHomeDataBase) and err_value.esp_range is not None
else source
)
location = ""
if (
isinstance(location_node, ESPHomeDataBase)
and location_node.esp_range is not None
):
mark = location_node.esp_range.start_mark
# DocumentLocation.line/column are 0-based (from the YAML Mark). Render
# as 1-based to match config.line_info() and editor line numbering.
location = f" (in {mark.document} {mark.line + 1}:{mark.column + 1})"
field = f" at '{'->'.join(str(p) for p in err_path)}'" if err_path else ""
raise cv.Invalid(
f"Undefined variable in {context_label}{field}: {err.message}{location}"
f"Undefined variable in {context_label}: {err.message}\n{format_path(err_path, err_value)}"
)
@@ -145,7 +126,7 @@ def _resolve_var(name: str, context_vars: ContextVars) -> Any:
def _handle_undefined(
err: UndefinedError,
path: SubstitutionPath,
path: DocumentPath,
value: Any,
strict_undefined: bool,
errors: ErrList | None,
@@ -163,7 +144,7 @@ def _handle_undefined(
def _expand_substitutions(
value: str,
path: SubstitutionPath,
path: DocumentPath,
context_vars: ContextVars,
strict_undefined: bool,
errors: ErrList | None,
@@ -236,7 +217,7 @@ def _expand_substitutions(
f"\nEvaluation stack: (most recent evaluation last)"
f"\n{err.stack_trace_str()}"
f"\nRelevant context:\n{err.context_trace_str()}"
f"\nSee {'->'.join(str(x) for x in path)}",
f"\n{format_path(path, orig_value)}",
path,
) from err
else:
@@ -345,15 +326,13 @@ def push_context(
def resolve_include(
include: IncludeFile,
path: list[int | str],
path: DocumentPath,
context_vars: ContextVars,
strict_undefined: bool = True,
errors: ErrList | None = None,
) -> tuple[Any, str]:
) -> Any:
"""Resolve an include, substituting the filename if needed.
Returns the loaded content and the resolved filename.
Note: no path-traversal validation is performed on the resolved filename.
A substitution that resolves to an absolute path will bypass the parent
directory (Path.__truediv__ ignores the left operand for absolute paths).
@@ -361,44 +340,44 @@ def resolve_include(
values (including command-line substitutions), so path restrictions are
an explicit non-goal here.
"""
original = str(include.file)
original = include.file
original_str = str(original)
filename = str(
_expand_substitutions(
original, path + ["file"], context_vars, strict_undefined, errors
original_str, path + ["file"], context_vars, strict_undefined, errors
)
)
if filename != original:
substituted = filename != original_str
if substituted:
include = IncludeFile(
include.parent_file, filename, include.vars, include.yaml_loader
)
try:
return include.load(), filename
return include.load()
except esphome.core.EsphomeError as err:
resolved = f" (expanded from '{original}')" if substituted else ""
raise cv.Invalid(
f"Error including file '{filename}': {err}",
f"Error including file '{filename}'{resolved}: {err}"
f"\n{format_path(path, original)}",
path + [f"<{filename}>"],
) from err
def _substitute_include(
include: IncludeFile,
path: list[int | str],
path: DocumentPath,
context_vars: ContextVars,
strict_undefined: bool,
errors: ErrList | None,
) -> Any:
"""Resolve an include and substitute its content."""
content, filename = resolve_include(
include, path, context_vars, strict_undefined, errors
)
return substitute(
content, path + [f"<{filename}>"], context_vars, strict_undefined, errors
)
content = resolve_include(include, path, context_vars, strict_undefined, errors)
return substitute(content, path, context_vars, strict_undefined, errors)
def substitute(
item: Any,
path: SubstitutionPath,
path: DocumentPath,
parent_context: ContextVars,
strict_undefined: bool,
errors: ErrList | None = None,
@@ -451,16 +430,12 @@ def _warn_unresolved_variables(errors: ErrList) -> None:
for err, path, expression in errors:
if "password" in path:
continue
location: str = "->".join(str(x) for x in path)
if isinstance(expression, ESPHomeDataBase) and expression.esp_range is not None:
location += f" in {str(expression.esp_range.start_mark)}"
_LOGGER.warning(
"The string '%s' looks like an expression,"
" but could not resolve all the variables: %s (see %s)",
" but could not resolve all the variables: %s\n%s",
expression,
err.message,
location,
format_path(path, expression),
)
@@ -479,7 +454,7 @@ def resolve_substitutions_block(
# Single-shot resolution — matches ``_walk_packages`` for the
# ``packages: !include`` entry point. Chained includes (an include that
# itself loads another ``!include`` at the top level) are not supported.
substitutions, _ = resolve_include(
substitutions = resolve_include(
substitutions,
[],
ContextVars(command_line_substitutions or {}),
esphome/yaml_util.py
+119
View File
@@ -48,6 +48,8 @@ _SECRET_VALUES = {}
# Not thread-safe — config processing is single-threaded today.
_load_listeners: list[Callable[[Path], None]] = []
DocumentPath = list[str | int]
@contextmanager
def track_yaml_loads() -> Generator[list[Path]]:
@@ -679,6 +681,123 @@ def is_secret(value):
return None
def _path_doc(item: Any) -> str | None:
"""Return the source document name if *item* carries location info."""
if isinstance(item, ESPHomeDataBase) and (r := item.esp_range) is not None:
return r.start_mark.document
return None
def _fmt_mark(loc: Any) -> str:
"""Render a DocumentLocation as a 1-based 'file line:col' string."""
return f"{loc.document} {loc.line + 1}:{loc.column + 1}"
def _obj_loc(obj: Any) -> str:
"""Return formatted source location for *obj*, or '' if it has none."""
if isinstance(obj, ESPHomeDataBase) and (r := obj.esp_range) is not None:
return _fmt_mark(r.start_mark)
return ""
def _fmt_segment(seg: list) -> str:
"""Format a path segment, rendering integers as [n] subscripts."""
parts: list[str] = []
for item in seg:
if isinstance(item, int):
if parts:
parts[-1] = f"{parts[-1]}[{item}]"
else:
parts.append(f"[{item}]")
else:
parts.append(str(item))
return "->".join(parts)
def _split_into_frames(
path: DocumentPath,
) -> list[tuple[list, str]]:
"""Group *path* into per-file frames at include boundaries.
A "frame" is the slice of the path that belongs to one source document.
Each path item is either:
* a **located key** — has an ``ESPHomeDataBase`` source mark; this is
what tells us which document owns the surrounding keys.
* an **integer** — a list subscript; always attaches to the open frame
(renders as ``foo[3]`` on the previous name).
* an **unlocated string** — a key with no source mark (e.g. constants
like ``CONF_PACKAGES``); it describes the parent of the *next* file,
so it migrates to the next frame when the document changes.
Returns a list of ``(items, "file line:col")`` tuples in walk order
(outermost frame first).
"""
frames: list[tuple[list, str]] = []
open_frame: list = []
next_frame_keys: list = [] # unlocated strings buffered for the next frame
open_doc: str | None = None
open_loc = ""
for item in path:
doc = _path_doc(item)
if doc is None:
# Ints subscript the open frame's last name; everything else
# (strings, or leading ints with no open frame) is buffered for
# the next frame.
if isinstance(item, int) and open_doc is not None:
open_frame.append(item)
else:
next_frame_keys.append(item)
continue
if open_doc is not None and doc != open_doc:
# Crossed an include boundary: close the open frame.
frames.append((open_frame, open_loc))
open_frame = []
open_frame.extend(next_frame_keys)
next_frame_keys.clear()
open_frame.append(item)
open_doc = doc
open_loc = _fmt_mark(item.esp_range.start_mark)
if open_doc is not None:
# Trailing buffered keys belong to the innermost (last) frame.
open_frame.extend(next_frame_keys)
frames.append((open_frame, open_loc))
return frames
def format_path(path: DocumentPath, current_obj: Any) -> str:
"""Build a human-readable include stack from a config path.
Each YAML key in *path* that carries an ``ESPHomeDataBase`` ``esp_range``
reveals which file it came from. When the source document changes between
consecutive such keys, that is an include boundary. The path is split
into per-file frames and formatted innermost-first, e.g.::
In: packages->roam in common/package/wifi.yaml 26:10
Included from packages->net in common/hardware.yaml 44:2
Included from packages->device in my_project.yaml 11:2
The innermost ``In:`` line uses the location from *current_obj* when
available (the value that triggered the error) for extra precision.
"""
frames = _split_into_frames(path)
obj_loc = _obj_loc(current_obj)
if not frames:
# No source info anywhere in the path: render as a flat path,
# using current_obj's location if it happens to have one.
suffix = f" in {obj_loc}" if obj_loc else ""
return f"In: {_fmt_segment(path)}{suffix}"
inner_seg, inner_loc = frames[-1]
lines = [f"In: {_fmt_segment(inner_seg)} in {obj_loc or inner_loc}"]
for seg, loc in reversed(frames[:-1]):
lines.append(f" Included from {_fmt_segment(seg)} in {loc}")
return "\n".join(lines)
class ESPHomeDumper(yaml.SafeDumper):
def represent_mapping(self, tag, mapping, flow_style=None):
value = []
requirements.txt
+3 -3
View File
@@ -6,13 +6,13 @@ colorama==0.4.6
icmplib==3.0.4
tornado==6.5.5
tzlocal==5.3.1 # from time
tzdata>=2021.1 # from time
tzdata>=2026.1 # from time
pyserial==3.5
platformio==6.1.19
esptool==5.2.0
click==8.3.2
esphome-dashboard==20260408.1
aioesphomeapi==44.16.1
aioesphomeapi==44.18.0
zeroconf==0.148.0
puremagic==1.30
ruamel.yaml==0.19.1 # dashboard_import
@@ -27,7 +27,7 @@ smpclient==6.0.0
requests==2.33.1
# esp-idf >= 5.0 requires this
pyparsing >= 3.0
pyparsing >= 3.3.2
# For autocompletion
argcomplete>=2.0.0
tests/component_tests/packages/test_packages.py
+16 -8
View File
@@ -46,7 +46,7 @@ from esphome.const import (
)
from esphome.core import CORE
from esphome.util import OrderedDict
from esphome.yaml_util import IncludeFile, add_context, load_yaml
from esphome.yaml_util import DocumentPath, IncludeFile, add_context, load_yaml
# Test strings
TEST_DEVICE_NAME = "test_device_name"
@@ -1113,7 +1113,7 @@ def test_packages_include_file_resolves_to_list(mock_resolve_include) -> None:
"""When packages: is an IncludeFile that resolves to a list, it is processed correctly."""
include_file = MagicMock(spec=IncludeFile)
package_content = {CONF_WIFI: {CONF_SSID: TEST_PACKAGE_WIFI_SSID}}
mock_resolve_include.return_value = ([package_content], None)
mock_resolve_include.return_value = [package_content]
config = {CONF_PACKAGES: include_file}
result = do_packages_pass(config)
@@ -1127,7 +1127,7 @@ def test_packages_include_file_resolves_to_dict(mock_resolve_include) -> None:
"""When packages: is an IncludeFile that resolves to a dict, it is processed correctly."""
include_file = MagicMock(spec=IncludeFile)
package_content = {CONF_WIFI: {CONF_SSID: TEST_PACKAGE_WIFI_SSID}}
mock_resolve_include.return_value = ({"network": package_content}, None)
mock_resolve_include.return_value = {"network": package_content}
config = {CONF_PACKAGES: include_file}
result = do_packages_pass(config)
@@ -1142,7 +1142,7 @@ def test_packages_include_file_resolves_to_invalid_type_raises(
) -> None:
"""When packages: is an IncludeFile that resolves to an invalid type, cv.Invalid is raised."""
include_file = MagicMock(spec=IncludeFile)
mock_resolve_include.return_value = ("not_a_dict_or_list", None)
mock_resolve_include.return_value = "not_a_dict_or_list"
config = {CONF_PACKAGES: include_file}
with pytest.raises(
@@ -1215,7 +1215,9 @@ def test_named_dict_with_include_files_no_false_deprecation_warning(
call_count = 0
def failing_callback(package_config: dict, context: object) -> dict:
def failing_callback(
package_config: dict, context: object, path: DocumentPath | None = None
) -> dict:
nonlocal call_count
call_count += 1
if call_count == 1:
@@ -1251,7 +1253,9 @@ def test_validate_deprecated_false_raises_directly(
call_count = 0
def failing_callback(package_config: dict, context: object) -> dict:
def failing_callback(
package_config: dict, context: object, path: DocumentPath | None = None
) -> dict:
nonlocal call_count
call_count += 1
if call_count == 1:
@@ -1283,7 +1287,9 @@ def test_error_on_first_declared_package_still_detected() -> None:
call_count = 0
def fail_on_last(package_config: dict, context: object) -> dict:
def fail_on_last(
package_config: dict, context: object, path: DocumentPath | None = None
) -> dict:
nonlocal call_count
call_count += 1
# Reverse iteration: third_pkg (1), second_pkg (2), first_pkg (3)
@@ -1312,7 +1318,9 @@ def test_deprecated_single_package_fallback_still_works(
attempt = 0
def fail_then_succeed(package_config: dict, context: object) -> dict:
def fail_then_succeed(
package_config: dict, context: object, path: DocumentPath | None = None
) -> dict:
nonlocal attempt
attempt += 1
if attempt == 1:
tests/components/esp32/dummy_signing_key_v1_ecdsa.pem
+7
View File
@@ -0,0 +1,7 @@
*** DO NOT USE THIS KEY...EVER ***
-----BEGIN EC PRIVATE KEY-----
MHcCAQEEIEZIp96p7Z7QN6vxOFE5FdRNm535vW81Ax07KnGxVjiMoAoGCCqGSM49
AwEHoUQDQgAEK+fBQDn1Q+r5lGwcDoMUgeg2Aq16LLrLUz7xWI6mS0PUClzolDIo
eaV/Pfjl7zAvkbQQsZq3rTNnr1eGAk5P+A==
-----END EC PRIVATE KEY-----
*** DO NOT USE THIS KEY...EVER ***
tests/components/esp32/test-signed_ota_v1.esp32-idf.yaml
+10
View File
@@ -0,0 +1,10 @@
esp32:
variant: esp32
framework:
type: esp-idf
advanced:
signed_ota_verification:
signing_key: ../../components/esp32/dummy_signing_key_v1_ecdsa.pem
signing_scheme: ecdsa_v1
<<: !include common.yaml
tests/unit_tests/test_substitutions.py
+43 -4
View File
@@ -659,7 +659,7 @@ def test_resolve_package_max_depth_exceeded(tmp_path: Path) -> None:
cv.Invalid,
match=f"Maximum include nesting depth \\({MAX_INCLUDE_DEPTH}\\) exceeded",
):
processor.resolve_package(package_config, substitutions.ContextVars())
processor.resolve_package(package_config, substitutions.ContextVars(), [])
def test_include_filename_substitution_undefined_var(tmp_path: Path) -> None:
@@ -690,7 +690,7 @@ def test_raise_first_undefined_logs_extras_at_debug(
caplog.at_level(logging.DEBUG, logger="esphome.components.substitutions"),
pytest.raises(cv.Invalid) as exc_info,
):
substitutions.raise_first_undefined(errors, None, "package definition")
substitutions.raise_first_undefined(errors, "package definition")
# First error is surfaced as the cv.Invalid message.
raised = str(exc_info.value)
@@ -706,7 +706,7 @@ def test_raise_first_undefined_logs_extras_at_debug(
def test_raise_first_undefined_noop_on_empty() -> None:
"""An empty errors list is a no-op — no exception, no log."""
substitutions.raise_first_undefined([], None, "package definition")
substitutions.raise_first_undefined([], "package definition")
def test_do_substitution_pass_included_substitutions_must_be_mapping(
@@ -778,4 +778,43 @@ def test_resolve_package_undefined_var_in_include_filename(tmp_path: Path) -> No
)
processor = _PackageProcessor({}, None, False)
with pytest.raises(cv.Invalid, match="unresolved substitutions"):
processor.resolve_package(package_config, substitutions.ContextVars())
processor.resolve_package(package_config, substitutions.ContextVars(), [])
def test_resolve_include_error_shows_expanded_from_when_substituted(
tmp_path: Path,
) -> None:
"""When a substituted filename fails to load, the error includes '(expanded from ...)'."""
parent = tmp_path / "main.yaml"
parent.write_text("")
def failing_loader(_path: Path) -> None:
raise EsphomeError("File not found")
include = yaml_util.IncludeFile(parent, "${device}.yaml", None, failing_loader)
context = substitutions.ContextVars({"device": "my_device"})
with pytest.raises(cv.Invalid) as exc_info:
substitutions.resolve_include(include, [], context)
msg = str(exc_info.value)
assert "my_device.yaml" in msg
assert "expanded from '${device}.yaml'" in msg
def test_resolve_include_error_no_expanded_from_for_literal_filename(
tmp_path: Path,
) -> None:
"""When a literal filename fails to load, the error has no 'expanded from' clause."""
parent = tmp_path / "main.yaml"
parent.write_text("")
def failing_loader(_path: Path) -> None:
raise EsphomeError("File not found")
include = yaml_util.IncludeFile(parent, "literal.yaml", None, failing_loader)
with pytest.raises(cv.Invalid) as exc_info:
substitutions.resolve_include(include, [], substitutions.ContextVars())
assert "expanded from" not in str(exc_info.value)
tests/unit_tests/test_yaml_util.py
+180 -1
View File
@@ -9,8 +9,9 @@ from esphome import core, yaml_util
from esphome.components import substitutions
from esphome.config_helpers import Extend, Remove
import esphome.config_validation as cv
from esphome.core import EsphomeError
from esphome.core import DocumentLocation, DocumentRange, EsphomeError
from esphome.util import OrderedDict
from esphome.yaml_util import ESPHomeDataBase, format_path, make_data_base
@pytest.fixture(autouse=True)
@@ -712,3 +713,181 @@ def test_yaml_merge_chain_include_depth_exceeded() -> None:
yaml_text = "base:\n <<: !include loop.yaml\n"
with pytest.raises(EsphomeError, match="Maximum include chain depth"):
yaml_util.parse_yaml(parent, io.StringIO(yaml_text), self_referencing_loader)
def _located(value, doc: str, line: int, col: int):
"""Return *value* wrapped with a fake ESPHomeDataBase source location."""
loc = DocumentLocation(doc, line, col)
obj = make_data_base(value)
if isinstance(obj, ESPHomeDataBase):
obj._esp_range = DocumentRange(loc, loc)
return obj
def test_format_path_no_location_info_returns_flat_path():
"""Plain path items with no esp_range produce a simple flat 'In:' line."""
result = format_path(["wifi", "ssid"], None)
assert result == "In: wifi->ssid"
def test_format_path_no_location_info_current_obj_adds_file():
"""When path has no location but current_obj does, its location is shown."""
obj = _located("${var}", "main.yaml", 5, 10)
result = format_path(["wifi", "ssid"], obj)
assert result == "In: wifi->ssid in main.yaml 6:11"
def test_format_path_single_frame_no_include_boundary():
"""All located keys from the same document → single 'In:' line, no 'Included from'."""
path = ["packages", _located("pkg1", "root.yaml", 5, 2)]
result = format_path(path, None)
assert result.startswith("In: packages->pkg1 in root.yaml 6:3")
assert "Included from" not in result
def test_format_path_two_frames_shows_included_from():
"""Keys from two different documents produce 'In:' + one 'Included from' line."""
path = [
"packages",
_located("device", "root.yaml", 10, 2),
"packages",
_located("inner", "hardware.yaml", 3, 2),
]
result = format_path(path, None)
assert "In: packages->inner in hardware.yaml 4:3" in result
assert "Included from packages->device in root.yaml 11:3" in result
def test_format_path_three_frames_full_include_stack():
"""Three document levels produce two 'Included from' lines in correct order."""
path = [
"packages",
_located("device", "root.yaml", 10, 2),
"packages",
_located("_wifi_", "hardware.yaml", 43, 2),
"packages",
_located("_roam_", "wifi.yaml", 25, 2),
]
result = format_path(path, None)
lines = result.splitlines()
assert lines[0].startswith("In: packages->_roam_ in wifi.yaml")
assert lines[1].startswith(" Included from packages->_wifi_ in hardware.yaml")
assert lines[2].startswith(" Included from packages->device in root.yaml")
def test_format_path_current_obj_overrides_innermost_location():
"""current_obj's esp_range replaces the key's column for the 'In:' line."""
path = ["packages", _located("pkg1", "root.yaml", 5, 2)]
# Value (the expression) sits at column 10, not column 2 like the key
value = _located("${undefined}", "root.yaml", 5, 10)
result = format_path(path, value)
assert "6:11" in result
assert "6:3" not in result
def test_format_path_empty_path_with_no_location():
"""Empty path with no location info returns 'In: '."""
result = format_path([], None)
assert result == "In: "
def test_format_path_integer_path_items_formatted_as_subscript():
"""Integer indices are rendered as [n] subscripts in the flat fallback."""
result = format_path(["packages", 0], None)
assert result == "In: packages[0]"
def test_format_path_integer_list_index_attached_to_previous_frame():
"""A list index between two include boundaries attaches to the outer frame."""
path = [
"packages",
_located("packages", "main.yaml", 5, 0),
0,
_located("packages", "level1.yaml", 2, 0),
0,
_located("esphome", "level2.yaml", 0, 0),
_located("name", "level2.yaml", 1, 8),
]
result = format_path(path, None)
lines = result.splitlines()
assert lines[0].startswith("In: esphome->name in level2.yaml")
assert "packages[0]" in lines[1] and "level1.yaml" in lines[1]
assert "packages[0]" in lines[2] and "main.yaml" in lines[2]
def test_format_path_trailing_unlocated_string_after_located_key():
"""Plain string keys after the last located key must still appear in output."""
path = [_located("packages", "main.yaml", 5, 0), "sub", "key"]
result = format_path(path, None)
assert result == "In: packages->sub->key in main.yaml 6:1"
def test_format_path_trailing_unlocated_int_attaches_to_current_frame():
"""Trailing ints attach to the open frame's last key (subscript), strings
buffer until end-of-path and then flush behind."""
path = [_located("packages", "main.yaml", 5, 0), 0, "sub"]
result = format_path(path, None)
# Int attaches to 'packages' as [0] subscript; trailing 'sub' is flushed
# at end and appears after.
assert result == "In: packages[0]->sub in main.yaml 6:1"
def test_format_path_only_trailing_unlocated_strings_are_preserved():
"""Trailing pending items must not be silently dropped after the last frame."""
path = [
_located("packages", "main.yaml", 5, 0),
_located("inner", "hardware.yaml", 3, 0),
"tail1",
"tail2",
]
result = format_path(path, None)
lines = result.splitlines()
assert lines[0] == "In: inner->tail1->tail2 in hardware.yaml 4:1"
assert lines[1] == " Included from packages in main.yaml 6:1"
def test_format_path_leading_int_with_no_current_doc_goes_to_pending():
"""An int before any located key is buffered and shown in the first frame."""
path = [0, _located("name", "main.yaml", 1, 0)]
result = format_path(path, None)
# Leading ints have no preceding name to subscript onto, so they render
# as bare [n] in the formatted segment.
assert result == "In: [0]->name in main.yaml 2:1"
def test_format_path_only_unlocated_int_returns_flat_fallback():
"""Path with only an int and no location info renders via the flat fallback."""
result = format_path([0], None)
assert result == "In: [0]"
def test_format_path_current_obj_in_different_doc_than_innermost_frame():
"""current_obj's location is preferred even when its document differs from the frame's."""
path = [_located("packages", "root.yaml", 1, 0)]
value = _located("${var}", "other.yaml", 9, 4)
result = format_path(path, value)
# Innermost line uses current_obj's mark (other.yaml 10:5), not the key's.
assert result == "In: packages in other.yaml 10:5"
def test_format_path_current_obj_without_location_falls_back_to_key():
"""An ESPHomeDataBase current_obj with no esp_range falls back to the key's location."""
class _NoRange(ESPHomeDataBase, str):
pass
obj = _NoRange.__new__(_NoRange, "value")
str.__init__(obj)
# No _esp_range set on this instance.
assert obj.esp_range is None
path = [_located("packages", "main.yaml", 5, 2)]
result = format_path(path, obj)
assert result == "In: packages in main.yaml 6:3"
def test_format_path_empty_path_with_located_current_obj():
"""An empty path with a located current_obj still surfaces the location."""
obj = _located("${var}", "main.yaml", 0, 0)
result = format_path([], obj)
assert result == "In: in main.yaml 1:1"