Merge branch 'dev' into partition-table-ota

This commit is contained in:
Mat931
2026-04-21 17:07:52 +00:00
committed by GitHub
156 changed files with 4189 additions and 1031 deletions
+76 -3
View File
@@ -8,10 +8,16 @@ from typing import Any
import pytest
from esphome.components.esp32 import VARIANTS
from esphome.components.esp32.const import KEY_ESP32, KEY_SDKCONFIG_OPTIONS
from esphome.components.esp32 import VARIANT_ESP32, VARIANTS
from esphome.components.esp32.const import KEY_ESP32, KEY_SDKCONFIG_OPTIONS, KEY_VARIANT
from esphome.components.esp32.gpio import validate_gpio_pin
import esphome.config_validation as cv
from esphome.const import CONF_ESPHOME, PlatformFramework
from esphome.const import (
CONF_ESPHOME,
CONF_IGNORE_PIN_VALIDATION_ERROR,
CONF_NUMBER,
PlatformFramework,
)
from esphome.core import CORE
from tests.component_tests.types import SetCoreConfigCallable
@@ -149,6 +155,73 @@ def test_execute_from_psram_p4_sdkconfig(
assert "CONFIG_SPIRAM_RODATA" not in sdkconfig
def test_ignore_pin_validation_error_on_clean_pin_warns(
set_core_config: SetCoreConfigCallable,
caplog: pytest.LogCaptureFixture,
) -> None:
"""A pin that passes validation but sets `ignore_pin_validation_error: true`
should log a warning nudging the user to remove the flag, and not raise."""
set_core_config(
PlatformFramework.ESP32_IDF, platform_data={KEY_VARIANT: VARIANT_ESP32}
)
pin = {CONF_NUMBER: 4, CONF_IGNORE_PIN_VALIDATION_ERROR: True}
with caplog.at_level("WARNING"):
result = validate_gpio_pin(pin)
assert result[CONF_NUMBER] == 4
assert "GPIO4 has no validation errors to ignore" in caplog.text
def test_ignore_pin_validation_error_on_dirty_pin_suppresses(
set_core_config: SetCoreConfigCallable,
caplog: pytest.LogCaptureFixture,
) -> None:
"""A pin that fails validation with `ignore_pin_validation_error: true` should
log the suppression warning and not raise (existing behavior)."""
set_core_config(
PlatformFramework.ESP32_IDF, platform_data={KEY_VARIANT: VARIANT_ESP32}
)
# GPIO6 is a flash pin on ESP32 -> pin_validation raises cv.Invalid
pin = {CONF_NUMBER: 6, CONF_IGNORE_PIN_VALIDATION_ERROR: True}
with caplog.at_level("WARNING"):
result = validate_gpio_pin(pin)
assert result[CONF_NUMBER] == 6
assert "Ignoring validation error on pin 6" in caplog.text
def test_dirty_pin_without_ignore_flag_raises(
set_core_config: SetCoreConfigCallable,
) -> None:
"""A pin that fails validation without the ignore flag should still raise."""
set_core_config(
PlatformFramework.ESP32_IDF, platform_data={KEY_VARIANT: VARIANT_ESP32}
)
pin = {CONF_NUMBER: 6, CONF_IGNORE_PIN_VALIDATION_ERROR: False}
with pytest.raises(cv.Invalid, match="flash interface"):
validate_gpio_pin(pin)
def test_clean_pin_without_ignore_flag_does_not_warn(
set_core_config: SetCoreConfigCallable,
caplog: pytest.LogCaptureFixture,
) -> None:
"""A clean pin without the ignore flag should pass silently."""
set_core_config(
PlatformFramework.ESP32_IDF, platform_data={KEY_VARIANT: VARIANT_ESP32}
)
pin = {CONF_NUMBER: 4, CONF_IGNORE_PIN_VALIDATION_ERROR: False}
with caplog.at_level("WARNING"):
result = validate_gpio_pin(pin)
assert result[CONF_NUMBER] == 4
assert "has no validation errors to ignore" not in caplog.text
def test_execute_from_psram_disabled_sdkconfig(
generate_main: Callable[[str | Path], str],
component_config_path: Callable[[str], Path],
@@ -0,0 +1,20 @@
esphome:
name: test
esp32:
board: esp32dev
framework:
type: arduino
spi:
clk_pin: GPIO18
mosi_pin: GPIO23
display:
- platform: ili9xxx
id: tft_display
model: ST7789V
cs_pin: GPIO5
dc_pin: GPIO17
reset_pin: GPIO16
invert_colors: false
@@ -0,0 +1,31 @@
"""Tests for the ili9xxx component."""
from __future__ import annotations
from collections.abc import Callable
from pathlib import Path
def test_ili9xxx_placement_new_uses_model_subclass(
generate_main: Callable[[str | Path], str],
component_config_path: Callable[[str], Path],
) -> None:
"""Regression test for ili9xxx picking the right constructor under placement new.
ili9xxx declares the ID as the base ``ILI9XXXDisplay`` but constructs a
model-specific subclass (e.g. ``ILI9XXXST7789V``) via ``MODELS[...].new()``.
Pvariable must emit placement new for the subclass — otherwise the base
default constructor runs and the panel is left with a null init sequence
and 0x0 dimensions, producing a silent blank screen.
"""
main_cpp = generate_main(component_config_path("ili9xxx_test.yaml"))
# Storage is sized for the subclass so the full object fits.
assert "sizeof(ili9xxx::ILI9XXXST7789V)" in main_cpp
assert "alignas(ili9xxx::ILI9XXXST7789V)" in main_cpp
# Pointer is declared as the base type for polymorphism.
assert "static ili9xxx::ILI9XXXDisplay *const tft_display" in main_cpp
# Placement new runs the subclass constructor — this is the actual regression fix.
assert "new(tft_display) ili9xxx::ILI9XXXST7789V()" in main_cpp
# Base-class default constructor must NOT be used.
assert "new(tft_display) ili9xxx::ILI9XXXDisplay()" not in main_cpp
@@ -7,6 +7,11 @@ import pytest
from esphome import config_validation as cv
from esphome.components.esp32 import KEY_BOARD, VARIANT_ESP32P4
# Importing xl9535 registers its pin schema with pins.PIN_SCHEMA_REGISTRY so that
# models (e.g. SEEED-RETERMINAL-D1001) that reference xl9535-backed pins in their
# defaults can be validated by the mipi_dsi CONFIG_SCHEMA in this test.
import esphome.components.xl9535 # noqa: F401
from esphome.const import (
CONF_DIMENSIONS,
CONF_HEIGHT,
@@ -0,0 +1,185 @@
"""Tests for the _final_validate buffer size calculation in mipi_spi."""
from __future__ import annotations
from typing import Any
import pytest
from esphome.components.display import CONF_SHOW_TEST_CARD
from esphome.components.esp32 import KEY_BOARD, KEY_VARIANT, VARIANT_ESP32
from esphome.components.mipi_spi.display import CONFIG_SCHEMA, FINAL_VALIDATE_SCHEMA
from esphome.const import CONF_BUFFER_SIZE, PlatformFramework
from esphome.types import ConfigType
from tests.component_tests.types import SetCoreConfigCallable
def _validated(config: ConfigType) -> ConfigType:
"""Run the component config schema followed by the final validation."""
config = CONFIG_SCHEMA(config)
FINAL_VALIDATE_SCHEMA(config)
return config
def _custom_config(
width: int,
height: int,
color_depth: str | int | None = None,
**extra: Any,
) -> ConfigType:
"""Build a minimal valid custom-model config with the given dimensions."""
config: ConfigType = {
"model": "custom",
"dc_pin": 18,
"dimensions": {"width": width, "height": height},
"init_sequence": [[0xA0, 0x01]],
}
if color_depth is not None:
config["color_depth"] = color_depth
config.update(extra)
return config
# The auto buffer-size selection inside _final_validate targets ~20 kB of
# pixel buffer. For a buffer of ``depth_bytes * width * height``, it picks the
# smallest integer ``x`` in range(2, 8) such that
# ``min(20000, buffer // 4) / buffer >= 1 / x`` (falling back to ``x = 8``).
# The test cases below cover the full range of possible outcomes (1/4 .. 1/8).
@pytest.mark.parametrize(
("width", "height", "color_depth", "expected"),
[
# 16-bit color depth -- buffer = 2 * width * height
# 128*160*2 = 40960 B -> fraction = 10240/40960 = 0.25 -> x = 4
pytest.param(128, 160, "16bit", 1.0 / 4, id="16bit_tiny"),
# 200*224*2 = 89600 B -> fraction = 20000/89600 ≈ 0.2232 -> x = 5
pytest.param(200, 224, "16bit", 1.0 / 5, id="16bit_small"),
# 240*224*2 = 107520 B -> fraction ≈ 0.1860 -> x = 6
pytest.param(240, 224, "16bit", 1.0 / 6, id="16bit_medium"),
# 200*320*2 = 128000 B -> fraction = 0.15625 -> x = 7
pytest.param(200, 320, "16bit", 1.0 / 7, id="16bit_large"),
# 240*320*2 = 153600 B -> fraction ≈ 0.1302 -> default x = 8
pytest.param(240, 320, "16bit", 1.0 / 8, id="16bit_xlarge"),
# 320*480*2 = 307200 B -> fraction ≈ 0.0651 -> default x = 8
pytest.param(320, 480, "16bit", 1.0 / 8, id="16bit_huge"),
# 8-bit color depth -- buffer = width * height
# 320*240 = 76800 B -> fraction = 19200/76800 = 0.25 -> x = 4
pytest.param(320, 240, "8bit", 1.0 / 4, id="8bit_tiny"),
# 400*224 = 89600 B -> fraction ≈ 0.2232 -> x = 5
pytest.param(400, 224, "8bit", 1.0 / 5, id="8bit_small"),
# 480*224 = 107520 B -> fraction ≈ 0.1860 -> x = 6
pytest.param(480, 224, "8bit", 1.0 / 6, id="8bit_medium"),
# 400*320 = 128000 B -> fraction = 0.15625 -> x = 7
pytest.param(400, 320, "8bit", 1.0 / 7, id="8bit_large"),
# 480*320 = 153600 B -> fraction ≈ 0.1302 -> default x = 8
pytest.param(480, 320, "8bit", 1.0 / 8, id="8bit_xlarge"),
],
)
def test_buffer_size_auto_selected(
width: int,
height: int,
color_depth: str,
expected: float,
set_core_config: SetCoreConfigCallable,
) -> None:
"""Without PSRAM or an explicit buffer_size, a fraction is chosen from the display size.
Without any drawing method and without LVGL, final validation also auto-enables
``show_test_card``, which in turn makes the component require a buffer and therefore
triggers the buffer-size selection path.
"""
set_core_config(
PlatformFramework.ESP32_IDF,
platform_data={KEY_BOARD: "esp32dev", KEY_VARIANT: VARIANT_ESP32},
)
config = _validated(_custom_config(width, height, color_depth))
# Sanity check: final validation should have enabled the test card for us,
# which is what causes the buffer-size calculation to actually run.
assert config.get(CONF_SHOW_TEST_CARD) is True
assert config[CONF_BUFFER_SIZE] == pytest.approx(expected)
@pytest.mark.parametrize(
"buffer_size",
[0.125, 0.25, 0.5, 1.0],
ids=["one_eighth", "one_quarter", "half", "full"],
)
def test_explicit_buffer_size_is_preserved(
buffer_size: float,
set_core_config: SetCoreConfigCallable,
) -> None:
"""An explicitly configured buffer_size is never overridden by final validation."""
set_core_config(
PlatformFramework.ESP32_IDF,
platform_data={KEY_BOARD: "esp32dev", KEY_VARIANT: VARIANT_ESP32},
)
config = _validated(
_custom_config(240, 320, "16bit", buffer_size=buffer_size),
)
assert config[CONF_BUFFER_SIZE] == pytest.approx(buffer_size)
def test_buffer_size_not_set_when_psram_enabled(
set_core_config: SetCoreConfigCallable,
set_component_config,
) -> None:
"""When PSRAM is enabled the auto buffer-size selection is skipped."""
set_core_config(
PlatformFramework.ESP32_IDF,
platform_data={KEY_BOARD: "esp32dev", KEY_VARIANT: VARIANT_ESP32},
)
# Presence of the psram domain in the full config is what _final_validate checks.
set_component_config("psram", True)
config = _validated(_custom_config(240, 320, "16bit"))
assert CONF_BUFFER_SIZE not in config
def test_buffer_size_not_set_when_buffer_not_required(
set_core_config: SetCoreConfigCallable,
set_component_config,
) -> None:
"""With LVGL present and no drawing methods, no buffer fraction is chosen.
LVGL suppresses the automatic show_test_card injection, which means
``requires_buffer`` is False and the early-return branch fires.
"""
set_core_config(
PlatformFramework.ESP32_IDF,
platform_data={KEY_BOARD: "esp32dev", KEY_VARIANT: VARIANT_ESP32},
)
set_component_config("lvgl", [])
config = _validated(_custom_config(240, 320, "16bit"))
assert CONF_BUFFER_SIZE not in config
# And no test card should have been auto-enabled either.
assert not config.get(CONF_SHOW_TEST_CARD)
def test_buffer_size_selected_when_lvgl_with_test_card(
set_core_config: SetCoreConfigCallable,
set_component_config,
) -> None:
"""LVGL present + an explicit drawing method still triggers buffer sizing.
When LVGL is enabled, ``show_test_card`` is not injected automatically,
but users can still request it explicitly -- in that case ``requires_buffer``
is True and the buffer-size heuristic still runs.
"""
set_core_config(
PlatformFramework.ESP32_IDF,
platform_data={KEY_BOARD: "esp32dev", KEY_VARIANT: VARIANT_ESP32},
)
set_component_config("lvgl", [])
# 128x160 @ 16bit -> expected 1/4 (see test_buffer_size_auto_selected).
config = _validated(
_custom_config(128, 160, "16bit", show_test_card=True),
)
assert config[CONF_BUFFER_SIZE] == pytest.approx(1.0 / 4)
@@ -2,18 +2,20 @@
import logging
from pathlib import Path
import re
from unittest.mock import MagicMock, patch
import pytest
from esphome.components.packages import (
CONFIG_SCHEMA,
_substitute_package_definition,
_walk_packages,
do_packages_pass,
is_package_definition,
merge_packages,
)
from esphome.components.substitutions import do_substitution_pass
from esphome.components.substitutions import ContextVars, do_substitution_pass
import esphome.config as config_module
from esphome.config import resolve_extend_remove
from esphome.config_helpers import Extend, Remove
@@ -44,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
from esphome.yaml_util import IncludeFile, add_context, load_yaml
# Test strings
TEST_DEVICE_NAME = "test_device_name"
@@ -1399,3 +1401,85 @@ def test_raw_config_contains_merged_esphome_from_package(tmp_path) -> None:
"CORE.raw_config should contain esphome section after package merge"
)
assert CORE.raw_config[CONF_ESPHOME][CONF_NAME] == TEST_DEVICE_NAME
# ---------------------------------------------------------------------------
# _substitute_package_definition
# ---------------------------------------------------------------------------
def test_substitute_package_definition_local_dict_returned_unchanged() -> None:
"""A plain local config dict is not substituted and is returned as-is."""
pkg = {CONF_WIFI: {CONF_SSID: "test"}}
result = _substitute_package_definition(pkg, ContextVars())
assert result is pkg
def test_substitute_package_definition_string_resolved_with_context() -> None:
"""A string package definition has its variables substituted."""
ctx = ContextVars({"variant": "esp32"})
result = _substitute_package_definition("device-${variant}.yaml", ctx)
assert result == "device-esp32.yaml"
def test_substitute_package_definition_undefined_in_string() -> None:
"""An undefined variable in a package URL string raises cv.Invalid."""
with pytest.raises(cv.Invalid, match="Undefined variable in package definition"):
_substitute_package_definition(
"github://org/repo/${undefined_var}/pkg.yaml", ContextVars()
)
def test_substitute_package_definition_undefined_in_remote_dict_field() -> None:
"""An undefined variable inside a remote-dict field names the offending field."""
with pytest.raises(cv.Invalid) as exc_info:
_substitute_package_definition(
{CONF_URL: "github://${typo}/repo"}, ContextVars()
)
err = str(exc_info.value)
assert "'typo' is undefined" in err
assert CONF_URL in err
def test_substitute_package_definition_undefined_in_remote_dict_non_first_field() -> (
None
):
"""The field path joins correctly for non-first dict fields (e.g. ``ref``)."""
with pytest.raises(cv.Invalid) as exc_info:
_substitute_package_definition(
{
CONF_URL: "github://org/repo",
CONF_REF: "branch-${branch_typo}",
},
ContextVars(),
)
err = str(exc_info.value)
assert "'branch_typo' is undefined" in err
assert CONF_REF in err
def test_substitute_package_definition_includes_source_location(tmp_path: Path) -> None:
"""A package loaded from YAML surfaces file/line/col in the cv.Invalid message.
Line/column are rendered 1-based (matching config.line_info() and editor
line numbering) and point at the offending scalar, not the enclosing dict.
"""
yaml_file = tmp_path / "main.yaml"
yaml_file.write_text(
"packages:\n broken: github://org/repo/${undefined_var}/pkg.yaml\n"
)
config = load_yaml(yaml_file)
package_config = config[CONF_PACKAGES]["broken"]
with pytest.raises(cv.Invalid) as exc_info:
_substitute_package_definition(package_config, ContextVars())
err = str(exc_info.value)
assert "main.yaml" in err
# The offending value lives on line 2 (1-based). Column depends on the YAML
# loader, so we only pin line and check that a 1-based column is present.
match = re.search(r"main\.yaml (\d+):(\d+)", err)
assert match, err
line, col = int(match.group(1)), int(match.group(2))
assert line == 2, f"expected 1-based line 2, got {line} (err={err!r})"
assert col >= 1, f"expected 1-based column ≥ 1, got {col} (err={err!r})"
@@ -0,0 +1,14 @@
esphome:
name: test
host:
text:
- platform: template
name: "Test Text Restore"
id: test_text_restore
optimistic: true
max_length: 10
mode: text
initial_value: "hello"
restore_value: true
@@ -0,0 +1,44 @@
"""Tests for the template text component."""
from __future__ import annotations
from collections.abc import Callable
from pathlib import Path
def test_template_text_saver_uses_placement_new_with_templated_subclass(
generate_main: Callable[[str | Path], str],
component_config_path: Callable[[str], Path],
) -> None:
"""Regression test for template text restore saver using placement new.
When ``restore_value: true``, the saver is its own Pvariable with
placement new: storage is sized for ``TextSaver<MAX_LENGTH>``, the
declared pointer stays at ``TemplateTextSaverBase *`` for polymorphism,
and the templated subclass constructor runs. A regression would either
reintroduce the heap ``new TextSaver<...>()`` expression or size the
storage for the base class and silently skip the subclass ctor.
"""
main_cpp = generate_main(component_config_path("template_text_restore.yaml"))
# Storage is sized and aligned for the templated subclass.
assert "sizeof(template_::TextSaver<10>)" in main_cpp
assert "alignas(template_::TextSaver<10>)" in main_cpp
# Pointer declared as base type for polymorphism.
assert (
"static template_::TemplateTextSaverBase *const test_text_restore_value_saver"
in main_cpp
)
# Placement new runs the templated subclass constructor.
assert "new(test_text_restore_value_saver) template_::TextSaver<10>()" in main_cpp
# Base-class default ctor must NOT be used.
assert (
"new(test_text_restore_value_saver) template_::TemplateTextSaverBase()"
not in main_cpp
)
# No heap `new TextSaver<...>()` left over — the pre-fix pattern.
assert "new template_::TextSaver<" not in main_cpp
# Saver is wired into the text component.
assert (
"test_text_restore->set_value_saver(test_text_restore_value_saver)" in main_cpp
)
+58
View File
@@ -0,0 +1,58 @@
#include <gtest/gtest.h>
#include <cmath>
#include "esphome/core/helpers.h"
namespace esphome {
TEST(HelpersTest, Ilog10PowersOfTen) {
EXPECT_EQ(ilog10(1.0f), 0);
EXPECT_EQ(ilog10(10.0f), 1);
EXPECT_EQ(ilog10(100.0f), 2);
EXPECT_EQ(ilog10(1000.0f), 3);
EXPECT_EQ(ilog10(10000.0f), 4);
EXPECT_EQ(ilog10(100000.0f), 5);
EXPECT_EQ(ilog10(0.1f), -1);
EXPECT_EQ(ilog10(0.001f), -3);
}
TEST(HelpersTest, Ilog10General) {
EXPECT_EQ(ilog10(5.0f), 0);
EXPECT_EQ(ilog10(9.99f), 0);
EXPECT_EQ(ilog10(50.0f), 1);
EXPECT_EQ(ilog10(99.0f), 1);
EXPECT_EQ(ilog10(999.0f), 2);
EXPECT_EQ(ilog10(0.5f), -1);
EXPECT_EQ(ilog10(0.0072f), -3);
EXPECT_EQ(ilog10(120000.0f), 5);
EXPECT_EQ(ilog10(123456.789f), 5);
}
TEST(HelpersTest, Ilog10Negative) {
EXPECT_EQ(ilog10(-1.0f), 0);
EXPECT_EQ(ilog10(-10.0f), 1);
EXPECT_EQ(ilog10(-0.1f), -1);
EXPECT_EQ(ilog10(-123.456f), 2);
}
// Verify that ilog10 + pow10_int produces the same rounding result as log10/pow.
// ilog10 may differ from floor(log10f()) for values not exactly representable in float
// (e.g. 0.01f is 0.00999...), but the full round-trip must match.
TEST(HelpersTest, Ilog10RoundTripMatchesLog10) {
float values[] = {0.0072f, 0.05f, 0.1f, 0.5f, 1.0f, 3.14f, 9.99f, 10.0f, 42.0f, 100.0f,
1234.5f, 9999.0f, 10000.0f, 99999.0f, 120000.0f, 999999.0f, -1.0f, -0.1f, -123.456f, -10000.0f};
for (uint8_t digits = 1; digits <= 6; digits++) {
for (float v : values) {
// New implementation using ilog10 + pow10_int
float factor_new = pow10_int(digits - 1 - ilog10(v));
float result_new = roundf(v * factor_new) / factor_new;
// Reference using log10/pow
double factor_ref = pow(10.0, digits - std::ceil(std::log10(std::fabs(v))));
float result_ref = static_cast<float>(round(v * factor_ref) / factor_ref);
EXPECT_FLOAT_EQ(result_new, result_ref) << "mismatch for value=" << v << " digits=" << (int) digits;
}
}
}
} // namespace esphome
@@ -145,3 +145,19 @@ display:
it.filled_rectangle(0, 0, it.get_width(), it.get_height(), Color::WHITE);
it.circle(it.get_width() / 2, it.get_height() / 2, 30, Color::BLACK);
it.circle(it.get_width() / 2, it.get_height() / 2, 20, Color(255, 0, 0));
- platform: epaper_spi
spi_id: spi_bus
model: goodisplay-gdey042t81-4.2
cs_pin:
allow_other_uses: true
number: GPIO5
dc_pin:
allow_other_uses: true
number: GPIO17
reset_pin:
allow_other_uses: true
number: GPIO16
busy_pin:
allow_other_uses: true
number: GPIO4
@@ -0,0 +1,4 @@
esp32_ble:
use_psram: true
psram:
@@ -1 +1,2 @@
<<: !include common.yaml
<<: !include common_use_psram.yaml
@@ -1,4 +1,5 @@
<<: !include common.yaml
<<: !include common_use_psram.yaml
esp32_ble:
io_capability: keyboard_only
@@ -2,6 +2,7 @@ packages:
ble: !include ../../test_build_components/common/ble/esp32-p4-idf.yaml
<<: !include common.yaml
<<: !include common_use_psram.yaml
esp32_ble:
io_capability: keyboard_only
@@ -0,0 +1,19 @@
ethernet:
type: W5500
clk_pin: 6
mosi_pin: 7
miso_pin: 2
cs_pin: 10
interrupt_pin: 3
reset_pin: 4
clock_speed: 10Mhz
manual_ip:
static_ip: 192.168.178.56
gateway: 192.168.178.1
subnet: 255.255.255.0
domain: .local
mac_address: "02:AA:BB:CC:DD:01"
on_connect:
- logger.log: "Ethernet connected!"
on_disconnect:
- logger.log: "Ethernet disconnected!"
@@ -45,6 +45,11 @@ esphome:
args:
- response->status_code
- body.c_str()
- delay: 1s
- logger.log:
format: "After delay, body still: %s"
args:
- body.c_str()
http_request:
useragent: esphome/tagreader
@@ -0,0 +1,22 @@
#include <gtest/gtest.h>
#include "esphome/components/modbus/modbus_helpers.h"
namespace esphome::modbus::helpers {
TEST(ModbusHelpersTest, PayloadToNumberRejectsOffsetAtEndOfBuffer) {
const std::vector<uint8_t> data{0x12, 0x34};
EXPECT_EQ(payload_to_number(data, SensorValueType::U_WORD, 2, 0xFFFFFFFF), 0);
}
TEST(ModbusHelpersTest, PayloadToNumberRejectsTruncatedMultiRegisterValue) {
const std::vector<uint8_t> data{0x12, 0x34, 0x56};
EXPECT_EQ(payload_to_number(data, SensorValueType::U_DWORD, 0, 0xFFFFFFFF), 0);
}
TEST(ModbusHelpersTest, PayloadToNumberDecodesValidWord) {
const std::vector<uint8_t> data{0x12, 0x34};
EXPECT_EQ(payload_to_number(data, SensorValueType::U_WORD, 0, 0xFFFFFFFF), 0x1234);
}
} // namespace esphome::modbus::helpers
+16
View File
@@ -3,6 +3,22 @@ esphome:
then:
- rtttl.play: 'siren:d=8,o=5,b=100:d,e,d,e,d,e,d,e'
- rtttl.stop
# Test all note features: all notes, denominators (1,2,4,8,16,32), sharp (#), octaves (4-7), dotted (.), note gap (c5,c5), pause (p)
- rtttl.play: 'special:d=4,o=5,b=120:1c4,2d#5,4e6.,8f#7,16g4,32a5,8a#5,4b6,8h5,c5,c5,8p,2c4'
# Different orders of control parameters
- rtttl.play: 'test_odb:o=5,d=8,b=100:c'
- rtttl.play: 'test_bod:b=100,o=5,d=8:c'
- rtttl.play: 'test_bdo:b=100,d=8,o=5:c'
- rtttl.play: 'test_obd:o=5,b=100,d=8:c'
- rtttl.play: 'test_dbo:d=8,b=100,o=5:c'
# Missing parameters (use defaults)
- rtttl.play: 'test_no_d:o=5,b=100:c'
- rtttl.play: 'test_no_o:d=8,b=100:c'
- rtttl.play: 'test_no_b:d=8,o=5:c'
- rtttl.play: 'test_only_d:d=8:c'
- rtttl.play: 'test_only_o:o=5:c'
- rtttl.play: 'test_only_b:b=100:c'
- rtttl.play: 'test_empty::c'
output:
- platform: ${output_platform}
@@ -171,6 +171,7 @@ sensor:
quantile: .9
- round: 1
- round_to_multiple_of: 0.25
- round_to_significant_digits: 3
- skip_initial: 3
- sliding_window_moving_average:
window_size: 15
@@ -0,0 +1,19 @@
import esphome.codegen as cg
import esphome.config_validation as cv
from esphome.const import CONF_ID
CODEOWNERS = ["@esphome/tests"]
wake_test_component_ns = cg.esphome_ns.namespace("wake_test_component")
WakeTestComponent = wake_test_component_ns.class_("WakeTestComponent", cg.Component)
CONFIG_SCHEMA = cv.Schema(
{
cv.GenerateID(): cv.declare_id(WakeTestComponent),
}
).extend(cv.COMPONENT_SCHEMA)
async def to_code(config):
var = cg.new_Pvariable(config[CONF_ID])
await cg.register_component(var, config)
@@ -0,0 +1,19 @@
#include "wake_test_component.h"
#include "esphome/core/application.h"
#include "esphome/core/log.h"
#include <chrono>
#include <thread>
namespace esphome::wake_test_component {
static const char *const TAG = "wake_test_component";
void WakeTestComponent::start_async_wake() {
ESP_LOGI(TAG, "Spawning async wake thread (50ms delay)");
std::thread([] {
std::this_thread::sleep_for(std::chrono::milliseconds(50));
App.wake_loop_threadsafe();
}).detach();
}
} // namespace esphome::wake_test_component
@@ -0,0 +1,27 @@
#pragma once
#include "esphome/core/component.h"
#include <atomic>
namespace esphome::wake_test_component {
class WakeTestComponent : public Component {
public:
void setup() override {}
void loop() override { this->loop_count_.fetch_add(1, std::memory_order_relaxed); }
int get_loop_count() const { return this->loop_count_.load(std::memory_order_relaxed); }
// Spawn a detached thread that sleeps briefly then calls
// App.wake_loop_threadsafe(). Used by the integration test to verify a
// cross-thread wake forces a component-phase iteration even when
// loop_interval_ has been raised high enough to gate it off otherwise.
void start_async_wake();
float get_setup_priority() const override { return setup_priority::DATA; }
protected:
std::atomic<int> loop_count_{0};
};
} // namespace esphome::wake_test_component
@@ -0,0 +1,60 @@
esphome:
name: loop-interval-decouple
on_boot:
priority: -100
then:
- lambda: |-
// Raise loop_interval_ to 500ms. With the decoupling fix the
// component phase should run ~twice per second while the 50ms
// scheduler interval below still fires at its requested cadence.
App.set_loop_interval(500);
# Start measurement after 1s so boot transients settle.
- delay: 1000ms
- lambda: |-
id(loop_at_start) = id(loop_counter)->get_loop_count();
id(sched_at_start) = id(sched_count);
ESP_LOGI("test", "MEASUREMENT_STARTED loop=%d sched=%d",
id(loop_at_start), id(sched_at_start));
# Observe for 2s.
- delay: 2000ms
- lambda: |-
int loop_delta = id(loop_counter)->get_loop_count() - id(loop_at_start);
int sched_delta = id(sched_count) - id(sched_at_start);
ESP_LOGI("test", "MEASUREMENT_DONE loop_delta=%d sched_delta=%d",
loop_delta, sched_delta);
host:
api:
logger:
level: INFO
logs:
loop_test_component: WARN # Silence per-loop log spam
external_components:
- source:
type: local
path: EXTERNAL_COMPONENT_PATH
globals:
- id: sched_count
type: int
initial_value: "0"
- id: loop_at_start
type: int
initial_value: "0"
- id: sched_at_start
type: int
initial_value: "0"
loop_test_component:
components:
- id: loop_counter
name: loop_counter
interval:
# Fast scheduler interval — with the decoupling fix this should fire at
# its requested 50ms cadence regardless of loop_interval_.
- interval: 50ms
then:
- lambda: |-
id(sched_count) += 1;
@@ -0,0 +1,51 @@
esphome:
name: loop-default-not-pulled
on_boot:
priority: -100
then:
# Leave loop_interval_ at its default (16 ms → ~62 Hz). Do NOT call
# set_loop_interval here. The fast scheduler interval below used to
# pull the component phase forward to ~128 Hz via the old
# std::max(next_schedule, delay_time / 2) floor.
# Start measurement after 1s so boot transients settle.
- delay: 1000ms
- lambda: |-
id(loop_at_start) = id(loop_counter)->get_loop_count();
ESP_LOGI("test", "MEASUREMENT_STARTED loop=%d", id(loop_at_start));
# Observe for 2s.
- delay: 2000ms
- lambda: |-
int loop_delta = id(loop_counter)->get_loop_count() - id(loop_at_start);
ESP_LOGI("test", "MEASUREMENT_DONE loop_delta=%d", loop_delta);
host:
api:
logger:
level: INFO
logs:
loop_test_component: WARN # Silence per-loop log spam
external_components:
- source:
type: local
path: EXTERNAL_COMPONENT_PATH
globals:
- id: loop_at_start
type: int
initial_value: "0"
loop_test_component:
components:
- id: loop_counter
name: loop_counter
interval:
# Fast scheduler interval (well under loop_interval_/2 = 8ms). In the
# pre-decoupling code this would have pulled the component phase forward
# to ~128 Hz. After the decoupling fix the component phase stays at
# ~62 Hz regardless.
- interval: 5ms
then:
- lambda: |-
// No-op; the presence of a due scheduler item is what matters.
@@ -0,0 +1,27 @@
esphome:
name: sched-interval-zero
host:
api:
logger:
level: DEBUG
globals:
- id: fire_count
type: int
initial_value: "0"
interval:
# Deliberately configure 0ms — this path goes through the C++
# Scheduler::set_timer_common_ coercion (not the Python cv.update_interval
# path, since interval: doesn't call cv.update_interval — it's an intervals
# component schema, not a PollingComponent's update_interval).
# Expected: scheduler coerces to 1ms at registration, emits ESP_LOGE,
# fires at ~1kHz instead of spinning.
- interval: 0ms
then:
- lambda: |-
id(fire_count) += 1;
if (id(fire_count) == 50) {
ESP_LOGI("test", "ZERO_INTERVAL_50_FIRES_REACHED");
}
@@ -0,0 +1,52 @@
esphome:
name: wake-loop-phase-b
on_boot:
priority: -100
then:
- lambda: |-
// Raise loop_interval_ to 2000ms. Without the wake-request flag,
// a wake_loop_threadsafe() call would only run Phase A (scheduler)
// and leave the component phase gated for ~2s.
App.set_loop_interval(2000);
# Let boot transients settle.
- delay: 1000ms
- lambda: |-
// Snapshot the loop counter, then ask the component to spawn a
// background thread that calls App.wake_loop_threadsafe() after
// ~50ms. With the fix, that wake forces Phase B on the next tick
// and the counter increments well within the 500ms observation
// window below.
id(count_at_start) = id(wake_counter)->get_loop_count();
id(start_time) = millis();
id(wake_counter)->start_async_wake();
ESP_LOGI("test", "WAKE_STARTED count=%d", id(count_at_start));
# Observation window must be much shorter than loop_interval_ (2000ms)
# so a "false pass" isn't possible by simply waiting out the gate.
- delay: 500ms
- lambda: |-
int count_now = id(wake_counter)->get_loop_count();
int delta = count_now - id(count_at_start);
uint32_t elapsed = millis() - id(start_time);
ESP_LOGI("test", "WAKE_RESULT delta=%d elapsed=%u", delta, elapsed);
host:
api:
logger:
level: INFO
external_components:
- source:
type: local
path: EXTERNAL_COMPONENT_PATH
components: [wake_test_component]
globals:
- id: count_at_start
type: int
initial_value: "0"
- id: start_time
type: uint32_t
initial_value: "0"
wake_test_component:
id: wake_counter
@@ -0,0 +1,75 @@
"""Test that loop_interval_ no longer clamps scheduler cadence.
Regression test for the decoupling of Application::loop() component-phase
cadence from scheduler wake timing.
Setup:
- App.set_loop_interval(500) — raised for power-savings style cadence
- Scheduler interval at 50ms — should fire at 50ms regardless of loop_interval_
- Component loop (LoopTestComponent) — should run at 500ms cadence
Before the decoupling fix the old `std::max(next_schedule, delay_time / 2)`
floor clamped the sleep to ~250ms, so the 50ms scheduler only fired ~8 times
per 2s (vs the ~40 expected). After the fix the scheduler fires close to its
requested cadence while the component phase stays gated at loop_interval_.
"""
from __future__ import annotations
import asyncio
import re
import pytest
from .types import APIClientConnectedFactory, RunCompiledFunction
@pytest.mark.asyncio
async def test_loop_interval_decoupling(
yaml_config: str,
run_compiled: RunCompiledFunction,
api_client_connected: APIClientConnectedFactory,
) -> None:
"""Raised loop_interval_ must not clamp scheduler item cadence."""
loop = asyncio.get_running_loop()
measurement_done: asyncio.Future[tuple[int, int]] = loop.create_future()
def on_log_line(line: str) -> None:
match = re.search(r"MEASUREMENT_DONE loop_delta=(\d+) sched_delta=(\d+)", line)
if match and not measurement_done.done():
measurement_done.set_result((int(match.group(1)), int(match.group(2))))
async with (
run_compiled(yaml_config, line_callback=on_log_line),
api_client_connected() as client,
):
device_info = await client.device_info()
assert device_info is not None
assert device_info.name == "loop-interval-decouple"
try:
loop_delta, sched_delta = await asyncio.wait_for(
measurement_done, timeout=10.0
)
except TimeoutError:
pytest.fail("MEASUREMENT_DONE marker never appeared")
# Observation window = 2s, loop_interval_ = 500ms.
# Component phase should fire ~4 times in 2s. The upper bound must be
# less than 8: the pre-decoupling behavior clamped to ~250ms cadence
# giving ~8 loops/2s, so allowing 8 would let the old behavior pass.
# Lower bound 3 (not 2) keeps the test honest: a >30% slowdown from
# the ~4 nominal is not normal CI jitter and should fail.
assert 3 <= loop_delta <= 6, (
f"Component loop should fire ~4 times in 2s at loop_interval=500ms, "
f"got {loop_delta}"
)
# Scheduler interval = 50ms → ~40 fires in 2s. Before the decoupling
# fix this clamped to ~8 fires. Assert >= 20 to catch the old clamped
# behavior with comfortable jitter headroom for slow CI hosts.
assert sched_delta >= 20, (
f"50ms scheduler interval should fire ~40 times in 2s but only "
f"fired {sched_delta}. This indicates loop_interval_ is still "
f"clamping scheduler cadence."
)
@@ -0,0 +1,67 @@
"""Test that a fast scheduler item does not pull the component phase forward.
Regression test for the original ~128 Hz → ~62 Hz bug fixed by decoupling
Application::loop() component-phase cadence from scheduler wake timing.
Setup:
- loop_interval_ left at its default (16 ms → ~62 Hz component phase).
- Scheduler interval at 5 ms (well under the old loop_interval_/2 = 8 ms floor).
Before the decoupling fix the ``std::max(next_schedule, delay_time / 2)`` floor
clamped the sleep to ~8 ms whenever any scheduler item was due sooner than
loop_interval_/2. That pulled the component phase forward to ~128 Hz — twice
what the documented ~62 Hz default promised. After the fix the component
phase stays at ~62 Hz regardless of scheduler activity.
"""
from __future__ import annotations
import asyncio
import re
import pytest
from .types import APIClientConnectedFactory, RunCompiledFunction
@pytest.mark.asyncio
async def test_loop_interval_default_not_pulled_forward(
yaml_config: str,
run_compiled: RunCompiledFunction,
api_client_connected: APIClientConnectedFactory,
) -> None:
"""Fast scheduler item must not pull component phase past default ~62 Hz."""
loop = asyncio.get_running_loop()
measurement_done: asyncio.Future[int] = loop.create_future()
def on_log_line(line: str) -> None:
match = re.search(r"MEASUREMENT_DONE loop_delta=(\d+)", line)
if match and not measurement_done.done():
measurement_done.set_result(int(match.group(1)))
async with (
run_compiled(yaml_config, line_callback=on_log_line),
api_client_connected() as client,
):
device_info = await client.device_info()
assert device_info is not None
assert device_info.name == "loop-default-not-pulled"
try:
loop_delta = await asyncio.wait_for(measurement_done, timeout=10.0)
except TimeoutError:
pytest.fail("MEASUREMENT_DONE marker never appeared")
# Observation window = 2s, loop_interval_ default = 16ms → ~62 Hz →
# ~125 component-phase iterations expected.
# Pre-fix behavior: the 5 ms scheduler interval tripped the old
# delay_time/2 = 8 ms floor, pulling the phase to ~128 Hz → ~256.
# Upper bound 180 is comfortably below the ~256 pre-fix rate but
# above the ~125 nominal with CI jitter.
# Lower bound 80 covers very slow CI hosts without permitting a
# complete regression.
assert 80 <= loop_delta <= 180, (
f"Component loop at default loop_interval_ should fire ~125 times "
f"in 2s (≈62 Hz × 2s); got {loop_delta}. Values >200 indicate the "
f"scheduler is again pulling the component phase forward."
)
+33
View File
@@ -26,6 +26,7 @@ async def test_runtime_stats(
# Track component stats
component_stats_found = set()
main_loop_lines: list[dict[str, str]] = []
# Patterns to match - need to handle ANSI color codes and timestamps
# The log format is: [HH:MM:SS][color codes][I][tag]: message
@@ -34,6 +35,14 @@ async def test_runtime_stats(
component_pattern = re.compile(
r"^\[[^\]]+\].*?\s+([\w.]+):\s+count=(\d+),\s+avg=([\d.]+)ms"
)
# Main loop overhead line emitted by runtime_stats
main_loop_pattern = re.compile(
r"main_loop:\s+iters=(?P<iters>\d+),\s+"
r"active_avg=(?P<active_avg>[\d.]+)ms,\s+"
r"active_max=(?P<active_max>[\d.]+)ms,\s+"
r"active_total=(?P<active_total>[\d.]+)ms,\s+"
r"overhead_total=(?P<overhead_total>[\d.]+)ms"
)
def check_output(line: str) -> None:
"""Check log output for runtime stats messages."""
@@ -54,6 +63,11 @@ async def test_runtime_stats(
component_name = match.group(1)
component_stats_found.add(component_name)
# Check for main_loop overhead line
ml_match = main_loop_pattern.search(line)
if ml_match:
main_loop_lines.append(ml_match.groupdict())
async with (
run_compiled(yaml_config, line_callback=check_output),
api_client_connected() as client,
@@ -86,3 +100,22 @@ async def test_runtime_stats(
assert "template.switch" in component_stats_found, (
f"Expected template.switch stats, found: {component_stats_found}"
)
# Verify the main_loop overhead line is emitted (at least once for
# the period section and once for the total section, per log cycle).
assert len(main_loop_lines) >= 2, (
f"Expected at least 2 main_loop lines, got {len(main_loop_lines)}"
)
for fields in main_loop_lines:
assert int(fields["iters"]) > 0, f"iters should be > 0: {fields}"
assert float(fields["active_total"]) > 0.0, (
f"active_total should be > 0: {fields}"
)
assert float(fields["active_avg"]) >= 0.0, (
f"active_avg should be >= 0: {fields}"
)
# overhead_total is derived and may be 0 if components dominate,
# but the field must still be present and parseable as a float.
assert float(fields["overhead_total"]) >= 0.0, (
f"overhead_total should be >= 0: {fields}"
)
@@ -0,0 +1,67 @@
"""Test that Scheduler::set_timer_common_ coerces interval=0 to 1ms.
Regression test for the scheduler busy-loop when interval=0 was passed
literally. Without the coercion, Scheduler::call() would spin forever
because the item's next_execution == now_64 after re-scheduling, failing
the loop's `> now_64` break condition. The device would fail to yield
back to the main loop and trigger a WDT reset.
With the coercion, interval=0 becomes interval=1 and the scheduler
fires at ~1kHz (bounded by the loop), the main loop continues to run,
and the device stays responsive to API calls.
"""
from __future__ import annotations
import asyncio
import pytest
from .types import APIClientConnectedFactory, RunCompiledFunction
@pytest.mark.asyncio
async def test_scheduler_interval_zero_coerced(
yaml_config: str,
run_compiled: RunCompiledFunction,
api_client_connected: APIClientConnectedFactory,
) -> None:
"""interval=0ms must be coerced to 1ms and not starve the main loop."""
loop = asyncio.get_running_loop()
reached_50: asyncio.Future[None] = loop.create_future()
coerce_warning: asyncio.Future[None] = loop.create_future()
def on_log_line(line: str) -> None:
if "ZERO_INTERVAL_50_FIRES_REACHED" in line and not reached_50.done():
reached_50.set_result(None)
if "would spin main loop" in line and not coerce_warning.done():
coerce_warning.set_result(None)
async with (
run_compiled(yaml_config, line_callback=on_log_line),
api_client_connected() as client,
):
# The API-client connection itself is evidence that the main loop
# is not starved — if set_interval(0) were spinning we could not
# get here at all.
device_info = await client.device_info()
assert device_info is not None
assert device_info.name == "sched-interval-zero"
# Coerce warning must fire at registration
try:
await asyncio.wait_for(coerce_warning, timeout=5.0)
except TimeoutError:
pytest.fail("Expected coerce warning 'would spin main loop' not seen")
# The coerced 1ms interval should fire 50 times quickly — this
# confirms the callback actually runs (not just registered) and the
# scheduler yields back to the main loop each time.
try:
await asyncio.wait_for(reached_50, timeout=5.0)
except TimeoutError:
pytest.fail(
"Coerced interval=0→1ms did not reach 50 fires within 5s, "
"which would indicate either the coercion failed or the "
"main loop is still being starved."
)
+6 -2
View File
@@ -325,9 +325,13 @@ async def test_uart_mock_ld2412_engineering_truncated(
],
)
# Signal when we see Phase 3 recovery values (gate_0_move=50)
# Signal when we see ALL Phase 3 recovery values to avoid race where some
# arrive after the waiter fires but before we index into the lists
recovery_received = collector.add_waiter(
lambda: pytest.approx(50.0) in collector.sensor_states["gate_0_move_energy"]
lambda: (
pytest.approx(50.0) in collector.sensor_states["gate_0_move_energy"]
and pytest.approx(42.0) in collector.sensor_states["light"]
)
)
async with (
@@ -0,0 +1,76 @@
"""Test that wake_loop_threadsafe() forces a component-phase iteration.
Regression test for the wake-request flag added to Application::loop()'s
Phase A / Phase B gate. Background producers (MQTT RX, USB RX, BLE event,
etc.) call App.wake_loop_threadsafe() expecting their component's loop()
to drain queued work; if the component phase stays gated by loop_interval_,
the work waits up to loop_interval_ ms instead of running on the next tick.
Setup:
- App.set_loop_interval(2000) — a wide gate that would clearly mask the bug.
- A test component spawns a detached std::thread that sleeps 50 ms and then
calls App.wake_loop_threadsafe() from a non-main thread.
- The on_boot block snapshots the component's loop counter before/after a
500 ms observation window.
Without the fix, delta=0 (the gate holds Phase B for ~2 s).
With the fix, delta>=1 (the wake forces Phase B within one tick of the wake).
"""
from __future__ import annotations
import asyncio
from pathlib import Path
import re
import pytest
from .types import APIClientConnectedFactory, RunCompiledFunction
@pytest.mark.asyncio
async def test_wake_loop_forces_phase_b(
yaml_config: str,
run_compiled: RunCompiledFunction,
api_client_connected: APIClientConnectedFactory,
) -> None:
"""A wake_loop_threadsafe() call from a background thread must trigger the
component phase within the next tick, even when loop_interval_ is raised
well above the observation window."""
external_components_path = str(
Path(__file__).parent / "fixtures" / "external_components"
)
yaml_config = yaml_config.replace(
"EXTERNAL_COMPONENT_PATH", external_components_path
)
loop = asyncio.get_running_loop()
result: asyncio.Future[tuple[int, int]] = loop.create_future()
def on_log_line(line: str) -> None:
match = re.search(r"WAKE_RESULT delta=(\d+) elapsed=(\d+)", line)
if match and not result.done():
result.set_result((int(match.group(1)), int(match.group(2))))
async with (
run_compiled(yaml_config, line_callback=on_log_line),
api_client_connected() as client,
):
device_info = await client.device_info()
assert device_info is not None
assert device_info.name == "wake-loop-phase-b"
try:
delta, elapsed = await asyncio.wait_for(result, timeout=15.0)
except TimeoutError:
pytest.fail("WAKE_RESULT marker never appeared")
# Without the fix, delta would be 0 — loop_interval_=2000ms held
# Phase B off for the full 500ms observation window. With the fix
# the wake from the background thread (~50ms after start) forces
# Phase B on the next tick, so the counter increments at least once.
assert delta >= 1, (
f"wake_loop_threadsafe() from a background thread should force "
f"Phase B within the next tick; observed delta={delta} after "
f"{elapsed}ms with loop_interval_=2000ms"
)
+156
View File
@@ -1468,3 +1468,159 @@ def test_cache_miss_corrupted_json(
result = helpers.create_components_graph()
# Should handle corruption gracefully and rebuild
assert result == {}
# ---------------------------------------------------------------------------
# parse_component_metadata / split_conflicting_groups
# ---------------------------------------------------------------------------
@pytest.fixture
def fake_components(tmp_path: Path) -> Path:
"""Create a fake esphome/components/ tree and return the repo root.
Component layout (tested against split_conflicting_groups):
alpha -- CONFLICTS_WITH=["beta"]
beta -- CONFLICTS_WITH=["alpha"]
beta_variant -- AUTO_LOAD=["beta"]
gamma -- (no metadata)
one_sided -- CONFLICTS_WITH=["plain"] (plain does not reject back)
plain -- no CONFLICTS_WITH
callable_auto -- AUTO_LOAD is a function (not a list literal) -> ignored
broken -- __init__.py has a SyntaxError
"""
components = tmp_path / "esphome" / "components"
components.mkdir(parents=True)
def write(name: str, body: str) -> None:
(components / name).mkdir()
(components / name / "__init__.py").write_text(body)
write("alpha", 'CONFLICTS_WITH = ["beta"]\n')
write("beta", 'CONFLICTS_WITH = ["alpha"]\n')
write("beta_variant", 'AUTO_LOAD = ["beta"]\n')
write("gamma", "")
write("one_sided", 'CONFLICTS_WITH = ["plain"]\n')
write("plain", "")
write("callable_auto", "def AUTO_LOAD():\n return ['beta']\n")
write("broken", "this is not valid python !!!")
helpers.parse_component_metadata.cache_clear()
return tmp_path
def test_parse_component_metadata_list_literals(
fake_components: Path, monkeypatch: MonkeyPatch
) -> None:
monkeypatch.setattr(helpers, "root_path", str(fake_components))
helpers.parse_component_metadata.cache_clear()
meta = helpers.parse_component_metadata("alpha")
assert meta.conflicts_with == frozenset({"beta"})
assert meta.auto_load == frozenset()
variant = helpers.parse_component_metadata("beta_variant")
assert variant.auto_load == frozenset({"beta"})
assert variant.conflicts_with == frozenset()
def test_parse_component_metadata_missing_empty_and_callable(
fake_components: Path, monkeypatch: MonkeyPatch
) -> None:
monkeypatch.setattr(helpers, "root_path", str(fake_components))
helpers.parse_component_metadata.cache_clear()
# Unknown component -> empty metadata, not an error.
unknown = helpers.parse_component_metadata("does_not_exist")
assert unknown == helpers.ComponentMetadata()
# Empty __init__.py -> empty metadata.
assert helpers.parse_component_metadata("gamma") == helpers.ComponentMetadata()
# Callable AUTO_LOAD cannot be statically evaluated -> empty.
callable_meta = helpers.parse_component_metadata("callable_auto")
assert callable_meta.auto_load == frozenset()
# SyntaxError in __init__.py must not raise.
assert helpers.parse_component_metadata("broken") == helpers.ComponentMetadata()
def test_split_conflicting_groups_splits_direct_conflict(
fake_components: Path, monkeypatch: MonkeyPatch
) -> None:
monkeypatch.setattr(helpers, "root_path", str(fake_components))
helpers.parse_component_metadata.cache_clear()
result = helpers.split_conflicting_groups(
{("esp32", "i2c"): ["alpha", "beta", "gamma"]}
)
# alpha and beta must end up in different buckets; gamma has no conflicts.
buckets = list(result.values())
assert any("alpha" in b for b in buckets)
assert any("beta" in b for b in buckets)
for bucket in buckets:
assert not ({"alpha", "beta"} <= set(bucket))
# Gamma sticks with whichever bucket it landed in first (alpha's).
all_members = {c for b in buckets for c in b}
assert all_members == {"alpha", "beta", "gamma"}
def test_split_conflicting_groups_propagates_through_auto_load(
fake_components: Path, monkeypatch: MonkeyPatch
) -> None:
"""A component that AUTO_LOADs a conflicting one must also be split out."""
monkeypatch.setattr(helpers, "root_path", str(fake_components))
helpers.parse_component_metadata.cache_clear()
result = helpers.split_conflicting_groups(
{("esp32", "i2c"): ["alpha", "beta_variant"]}
)
buckets = list(result.values())
for bucket in buckets:
assert not ({"alpha", "beta_variant"} <= set(bucket))
assert sum(len(b) for b in buckets) == 2
def test_split_conflicting_groups_symmetric_one_sided_declaration(
fake_components: Path, monkeypatch: MonkeyPatch
) -> None:
"""If only one side declares CONFLICTS_WITH, the pair must still be split."""
monkeypatch.setattr(helpers, "root_path", str(fake_components))
helpers.parse_component_metadata.cache_clear()
result = helpers.split_conflicting_groups(
{("esp32", "i2c"): ["one_sided", "plain"]}
)
buckets = list(result.values())
for bucket in buckets:
assert not ({"one_sided", "plain"} <= set(bucket))
def test_split_conflicting_groups_preserves_non_conflicting_group(
fake_components: Path, monkeypatch: MonkeyPatch
) -> None:
monkeypatch.setattr(helpers, "root_path", str(fake_components))
helpers.parse_component_metadata.cache_clear()
original = {("esp32", "i2c"): ["alpha", "gamma", "plain"]}
result = helpers.split_conflicting_groups(original)
# All three are mutually compatible -- the group must not be split.
assert result == original
def test_split_conflicting_groups_preserves_original_signature_for_first_bucket(
fake_components: Path, monkeypatch: MonkeyPatch
) -> None:
"""When a group is split, the first bucket keeps the original signature key."""
monkeypatch.setattr(helpers, "root_path", str(fake_components))
helpers.parse_component_metadata.cache_clear()
result = helpers.split_conflicting_groups({("esp32", "i2c"): ["alpha", "beta"]})
keys = set(result.keys())
assert ("esp32", "i2c") in keys
# One additional bucket with a disambiguated signature.
extra = keys - {("esp32", "i2c")}
assert len(extra) == 1
platform, signature = next(iter(extra))
assert platform == "esp32"
assert signature.startswith("i2c__conflict")
@@ -11,9 +11,9 @@ esp32:
logger:
<<: !include common/base.yaml
wifi:
ssid: !secret wifi_ssid
password: !secret wifi_password
# Plain nested !include — deferred as an IncludeFile until the substitution
# pass. The bundle must force-resolve it to pick up common/wifi.yaml.
wifi: !include common/wifi.yaml
api:
@@ -0,0 +1,2 @@
ssid: !secret wifi_ssid
password: !secret wifi_password
@@ -0,0 +1,5 @@
substitutions:
wifi_password: sub_password
wifi:
ssid: main_ssid
password: sub_password
@@ -0,0 +1,5 @@
substitutions: !include 15-substitutions_inc.yaml
wifi:
ssid: main_ssid
password: $wifi_password
@@ -0,0 +1 @@
wifi_password: sub_password
@@ -0,0 +1,5 @@
substitutions:
wifi_password: sub_password
wifi:
ssid: main_ssid
password: sub_password
@@ -0,0 +1,9 @@
substitutions: !include 15-substitutions_inc.yaml
packages:
wifi_pkg:
wifi:
password: $wifi_password
wifi:
ssid: main_ssid
@@ -0,0 +1,6 @@
substitutions:
subs_file: 15-substitutions_inc
wifi_password: sub_password
wifi:
ssid: main_ssid
password: sub_password
@@ -0,0 +1,8 @@
command_line_substitutions:
subs_file: 15-substitutions_inc
substitutions: !include ${subs_file}.yaml
wifi:
ssid: main_ssid
password: $wifi_password
+149 -1
View File
@@ -5,8 +5,10 @@ from __future__ import annotations
import io
import json
from pathlib import Path
import shutil
import tarfile
from typing import Any
from unittest.mock import patch
import pytest
@@ -20,6 +22,7 @@ from esphome.bundle import (
_add_bytes_to_tar,
_default_target_dir,
_find_used_secret_keys,
_force_load_include_files,
extract_bundle,
is_bundle_path,
prepare_bundle_for_compile,
@@ -485,7 +488,7 @@ def test_read_bundle_manifest_minimal(tmp_path: Path) -> None:
result = read_bundle_manifest(bundle_path)
assert result.esphome_version == "unknown"
assert result.files == []
assert not result.files
assert result.has_secrets is False
@@ -862,6 +865,117 @@ def test_discover_files_skips_missing_directory(tmp_path: Path) -> None:
assert len(files) == 1
def test_discover_files_nested_include(tmp_path: Path) -> None:
"""Nested !include files (e.g. wifi: !include wifi.yaml) are bundled."""
config_dir = _setup_config_dir(tmp_path)
(config_dir / "test.yaml").write_text(
"esphome:\n name: test\nwifi: !include wifi.yaml\n"
)
(config_dir / "wifi.yaml").write_text('ssid: "a"\npassword: "b"\n')
creator = ConfigBundleCreator({})
files = creator.discover_files()
paths = [f.path for f in files]
assert "test.yaml" in paths
assert "wifi.yaml" in paths
def test_discover_files_deeply_nested_include(tmp_path: Path) -> None:
"""Chains of !include (a includes b includes c) are fully resolved."""
config_dir = _setup_config_dir(tmp_path)
(config_dir / "test.yaml").write_text(
"esphome:\n name: test\nwifi: !include level1.yaml\n"
)
(config_dir / "level1.yaml").write_text("nested: !include level2.yaml\n")
(config_dir / "level2.yaml").write_text('value: "leaf"\n')
creator = ConfigBundleCreator({})
files = creator.discover_files()
paths = [f.path for f in files]
assert "level1.yaml" in paths
assert "level2.yaml" in paths
def test_discover_files_nested_include_unresolved_substitution(
tmp_path: Path,
) -> None:
"""!include with substitution vars in path cannot be resolved; skipped gracefully."""
config_dir = _setup_config_dir(tmp_path)
(config_dir / "test.yaml").write_text(
"esphome:\n name: test\nwifi: !include ${platform}.yaml\n"
)
creator = ConfigBundleCreator({})
# Should not raise
files = creator.discover_files()
paths = [f.path for f in files]
assert "test.yaml" in paths
def test_discover_files_nested_include_load_failure(
tmp_path: Path, caplog: pytest.LogCaptureFixture
) -> None:
"""A nested !include pointing at a missing file is logged and skipped."""
config_dir = _setup_config_dir(tmp_path)
(config_dir / "test.yaml").write_text(
"esphome:\n name: test\nwifi: !include missing.yaml\n"
)
creator = ConfigBundleCreator({})
files = creator.discover_files()
paths = [f.path for f in files]
assert "test.yaml" in paths
assert any(
"failed to load !include" in r.message and "missing.yaml" in r.message
for r in caplog.records
)
def test_force_load_skips_duplicate_include_file() -> None:
"""The same IncludeFile referenced twice is only loaded once."""
class _StubInclude:
"""Mimics yaml_util.IncludeFile minimally for _force_load testing."""
def __init__(self) -> None:
self.file = Path("dup.yaml")
self.parent_file = Path("root.yaml")
self.load_calls = 0
def has_unresolved_expressions(self) -> bool:
return False
def load(self) -> dict[str, Any]:
self.load_calls += 1
return {}
stub = _StubInclude()
# Same instance appears twice — second visit must hit the _seen guard.
tree = {"a": stub, "b": [stub]}
with patch("esphome.bundle.yaml_util.IncludeFile", _StubInclude):
_force_load_include_files(tree)
assert stub.load_calls == 1
def test_force_load_handles_cyclic_containers() -> None:
"""Cyclic dict/list references don't cause infinite recursion."""
cyclic_dict: dict[str, Any] = {}
cyclic_dict["self"] = cyclic_dict
cyclic_list: list[Any] = []
cyclic_list.append(cyclic_list)
# Should return without recursing forever
_force_load_include_files(cyclic_dict)
_force_load_include_files(cyclic_list)
def test_discover_files_yaml_reload_failure(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
@@ -1008,6 +1122,40 @@ def test_discover_files_walk_tuple_values(tmp_path: Path) -> None:
assert "a.pem" in paths
# ---------------------------------------------------------------------------
# ConfigBundleCreator - fixture-based end-to-end
# ---------------------------------------------------------------------------
def test_discover_files_fixture_config(fixture_path: Path, tmp_path: Path) -> None:
"""Use the real ``fixtures/bundle/`` tree as an end-to-end reproducer.
The fixture config uses ``wifi: !include common/wifi.yaml`` — a plain
nested !include that is returned as a deferred ``IncludeFile`` and only
resolved during the substitution pass. Before this fix, bundle discovery
never ran substitutions, so ``common/wifi.yaml`` was silently missing
from the bundle.
"""
# Copy the fixture tree into a tmp dir so the test doesn't rely on the
# source repo being writable and so we can set CORE.config_path freely.
src = fixture_path / "bundle"
dst = tmp_path / "bundle"
shutil.copytree(src, dst)
CORE.config_path = dst / "bundle_test.yaml"
creator = ConfigBundleCreator({})
files = creator.discover_files()
paths = {f.path for f in files}
# Root and top-level !secret-referenced files
assert "bundle_test.yaml" in paths
assert "secrets.yaml" in paths
# The nested !include — this is what regressed when IncludeFile became
# deferred (PR #12213).
assert "common/wifi.yaml" in paths
# ---------------------------------------------------------------------------
# ConfigBundleCreator - create_bundle
# ---------------------------------------------------------------------------
@@ -24,6 +24,7 @@ from esphome.const import (
PLATFORM_LN882X,
PLATFORM_RP2040,
PLATFORM_RTL87XX,
SCHEDULER_DONT_RUN,
)
from esphome.core import CORE, HexInt, Lambda
@@ -765,3 +766,30 @@ def test_percentage_validators__raw_number_above_one_without_percent_sign(
config_validation.unbounded_percentage(value)
with pytest.raises(Invalid, match="percent sign"):
config_validation.unbounded_possibly_negative_percentage(value)
def test_update_interval__coerces_zero_to_one_ms(
caplog: pytest.LogCaptureFixture,
) -> None:
"""update_interval: 0ms must be coerced to 1ms (not rejected) because a
literal 0ms schedule causes Scheduler::call() to spin. Coercion keeps
existing configs compiling on upgrade while emitting a user-facing
warning that directs them to set a non-zero value."""
with caplog.at_level("WARNING"):
result = config_validation.update_interval("0ms")
assert result.total_milliseconds == 1
assert "update_interval of 0ms is not supported" in caplog.text
assert "1ms" in caplog.text
def test_update_interval__preserves_nonzero_values() -> None:
"""Non-zero update_interval values must pass through unchanged."""
assert config_validation.update_interval("1ms").total_milliseconds == 1
assert config_validation.update_interval("50ms").total_milliseconds == 50
assert config_validation.update_interval("60s").total_milliseconds == 60000
def test_update_interval__never_passes_through() -> None:
"""update_interval: never must still map to SCHEDULER_DONT_RUN."""
result = config_validation.update_interval("never")
assert result.total_milliseconds == SCHEDULER_DONT_RUN
+85
View File
@@ -14,6 +14,7 @@ from esphome.components.packages import (
do_packages_pass,
merge_packages,
)
from esphome.components.substitutions.jinja import UndefinedError
from esphome.config import resolve_extend_remove
from esphome.config_helpers import Extend, merge_config
import esphome.config_validation as cv
@@ -675,6 +676,90 @@ def test_include_filename_substitution_undefined_var(tmp_path: Path) -> None:
substitutions.do_substitution_pass(config)
def test_raise_first_undefined_logs_extras_at_debug(
caplog: pytest.LogCaptureFixture,
) -> None:
"""Only the first undefined error is raised; extras are logged at debug."""
errors: substitutions.ErrList = [
(UndefinedError("'a' is undefined"), ["url"], None),
(UndefinedError("'b' is undefined"), ["ref"], None),
(UndefinedError("'c' is undefined"), ["path"], None),
]
with (
caplog.at_level(logging.DEBUG, logger="esphome.components.substitutions"),
pytest.raises(cv.Invalid) as exc_info,
):
substitutions.raise_first_undefined(errors, None, "package definition")
# First error is surfaced as the cv.Invalid message.
raised = str(exc_info.value)
assert "'a' is undefined" in raised
assert "'b' is undefined" not in raised
assert "'c' is undefined" not in raised
# Remaining errors are captured via debug logging for troubleshooting.
assert "Additional undefined variables in package definition" in caplog.text
assert "'b' is undefined at 'ref'" in caplog.text
assert "'c' is undefined at 'path'" in caplog.text
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")
def test_do_substitution_pass_included_substitutions_must_be_mapping(
tmp_path: Path,
) -> None:
"""`substitutions: !include list.yaml` where the file holds a list raises cv.Invalid.
Locks in the shape check that runs after the deferred IncludeFile has been
resolved.
"""
parent = tmp_path / "main.yaml"
parent.write_text("")
def loader(path: Path):
return ["not", "a", "mapping"]
include = yaml_util.IncludeFile(parent, "subs.yaml", None, loader)
config = OrderedDict({CONF_SUBSTITUTIONS: include})
with pytest.raises(
cv.Invalid, match="Substitutions must be a key to value mapping"
):
substitutions.do_substitution_pass(config)
def test_do_packages_pass_included_substitutions_must_be_mapping(
tmp_path: Path,
) -> None:
"""`substitutions: !include list.yaml` alongside `packages:` raises cv.Invalid.
Without the shape check, ``UserDict(...)`` would surface a low-level
``TypeError``; the explicit ``cv.Invalid`` points at the substitutions path.
"""
parent = tmp_path / "main.yaml"
parent.write_text("")
def loader(path: Path):
return ["not", "a", "mapping"]
include = yaml_util.IncludeFile(parent, "subs.yaml", None, loader)
config = OrderedDict(
{
CONF_SUBSTITUTIONS: include,
"packages": {"noop": {"wifi": {"ssid": "main"}}},
}
)
with pytest.raises(
cv.Invalid, match="Substitutions must be a key to value mapping"
):
do_packages_pass(config)
def test_resolve_package_undefined_var_in_include_filename(tmp_path: Path) -> None:
"""An undefined substitution in a package include filename raises cv.Invalid.