Merge remote-tracking branch 'upstream/fix-delay-action-mutable-lambda' into integration

This commit is contained in:
J. Nick Koston
2026-04-17 08:36:08 -05:00
62 changed files with 1208 additions and 208 deletions
@@ -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)
+72
View File
@@ -0,0 +1,72 @@
// Regression tests for ESPTime::is_valid() optional checks.
//
// The RTC components (ds1307, bm8563, pcf85063, pcf8563, rx8130) read date/time
// fields from hardware but do not populate day_of_year. They call
// recalc_timestamp_utc(false) -- which skips day_of_year -- and then is_valid().
// These tests ensure the is_valid() overload can skip day_of_year validation so
// RTCs don't log "Invalid RTC time, not syncing to system clock." for valid times.
#include <gtest/gtest.h>
#include "esphome/core/time.h"
namespace esphome::testing {
// Build an ESPTime that mirrors what the RTC components construct: all fields
// populated from hardware except day_of_year (left zero-initialized).
static ESPTime make_rtc_like_time() {
ESPTime t{};
t.second = 30;
t.minute = 15;
t.hour = 12;
t.day_of_week = 4; // thursday
t.day_of_month = 15;
t.month = 4;
t.year = 2026;
// day_of_year intentionally left at 0 -- RTCs don't compute it.
return t;
}
TEST(ESPTimeIsValid, DefaultRejectsZeroDayOfYear) {
// Default is_valid() checks day_of_year; zero-init is out of range.
ESPTime t = make_rtc_like_time();
EXPECT_FALSE(t.is_valid());
}
TEST(ESPTimeIsValid, SkipDayOfYearAcceptsRTCLikeTime) {
// RTC code path: skip day_of_year validation.
ESPTime t = make_rtc_like_time();
EXPECT_TRUE(t.is_valid(/*check_day_of_week=*/true, /*check_day_of_year=*/false));
}
TEST(ESPTimeIsValid, SkipDayOfYearStillRejectsOutOfRangeFields) {
ESPTime t = make_rtc_like_time();
t.hour = 25;
EXPECT_FALSE(t.is_valid(/*check_day_of_week=*/true, /*check_day_of_year=*/false));
}
TEST(ESPTimeIsValid, SkipDayOfYearStillRejectsYearBefore2019) {
ESPTime t = make_rtc_like_time();
t.year = 2000;
EXPECT_FALSE(t.is_valid(/*check_day_of_week=*/true, /*check_day_of_year=*/false));
}
TEST(ESPTimeIsValid, SkipBothDayChecksAcceptsGPSLikeTime) {
// GPS path (gps_time.cpp) populates neither day_of_week nor day_of_year.
ESPTime t{};
t.second = 30;
t.minute = 15;
t.hour = 12;
t.day_of_month = 15;
t.month = 4;
t.year = 2026;
EXPECT_TRUE(t.is_valid(/*check_day_of_week=*/false, /*check_day_of_year=*/false));
EXPECT_FALSE(t.is_valid()); // default still rejects
}
TEST(ESPTimeIsValid, FullyPopulatedAcceptsWithDefaults) {
ESPTime t = make_rtc_like_time();
t.day_of_year = 105;
EXPECT_TRUE(t.is_valid());
}
} // namespace esphome::testing
@@ -0,0 +1,10 @@
zephyr_ble_server:
on_numeric_comparison_request:
then:
- logger.log:
format: "Compare this passkey with the one on your BLE device: %06d"
args: [passkey]
- ble_server.numeric_comparison_reply:
accept: True
- ble_server.numeric_comparison_reply:
accept: !lambda "return true;"
@@ -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,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."
)
@@ -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
+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