Merge branch 'dev' into inline-micros-esp32

This commit is contained in:
J. Nick Koston
2026-04-29 05:14:57 -05:00
committed by GitHub
24 changed files with 1068 additions and 124 deletions
+13 -6
View File
@@ -793,8 +793,11 @@ class MemoryAnalyzer:
"""Scan ESPHome source object files to map extern "C" symbols to components.
When no linker map file is available, this uses ``nm`` to scan ``.o`` files
under ``src/esphome/`` and build a symbol-to-component mapping. This catches
``extern "C"`` functions and other symbols that lack C++ namespace prefixes.
under ``src/`` (including ``src/main.cpp.o`` and everything beneath
``src/esphome/``) and build a symbol-to-component mapping. This catches
``extern "C"`` functions, the ESPHome-generated ``setup()``/``loop()``
entry points in ``main.cpp``, and other symbols that lack C++ namespace
prefixes.
Skips scanning if ``_source_symbol_map`` was already populated by
``_parse_map_file()``.
@@ -806,12 +809,12 @@ class MemoryAnalyzer:
if obj_dir is None:
return
# Find ESPHome source object files
esphome_src_dir = obj_dir / "src" / "esphome"
if not esphome_src_dir.is_dir():
# Scan all ESPHome-owned source object files: src/main.cpp.o and src/esphome/...
src_dir = obj_dir / "src"
if not src_dir.is_dir():
return
obj_files = sorted(esphome_src_dir.rglob("*.o"))
obj_files = sorted(src_dir.rglob("*.o"))
if not obj_files:
return
@@ -1064,6 +1067,10 @@ class MemoryAnalyzer:
if component_name in self.external_components:
return f"{_COMPONENT_PREFIX_EXTERNAL}{component_name}"
# ESPHome-generated entry point: src/main.cpp.o (contains setup()/loop())
if len(parts) >= 2 and parts[-2:] == ("src", "main.cpp.o"):
return _COMPONENT_CORE
# ESPHome core: src/esphome/core/... or src/esphome/...
if "core" in parts and "esphome" in parts:
return _COMPONENT_CORE
+5 -13
View File
@@ -1,4 +1,5 @@
from dataclasses import dataclass, field
from functools import partial
import hashlib
import logging
from pathlib import Path
@@ -19,7 +20,7 @@ from esphome.const import (
)
from esphome.core import CORE, ID, HexInt
from esphome.cpp_generator import MockObj
from esphome.external_files import download_content
from esphome.external_files import download_web_files_in_config
from esphome.types import ConfigType
_LOGGER = logging.getLogger(__name__)
@@ -63,15 +64,6 @@ def _compute_local_file_path(value: ConfigType) -> Path:
return base_dir / key
def _download_web_file(value: ConfigType) -> ConfigType:
url = value[CONF_URL]
path = _compute_local_file_path(value)
download_content(url, path)
_LOGGER.debug("download_web_file: path=%s", path)
return value
def _file_schema(value: ConfigType | str) -> ConfigType:
if isinstance(value, str):
return _validate_file_shorthand(value)
@@ -142,11 +134,10 @@ LOCAL_SCHEMA = cv.Schema(
}
)
WEB_SCHEMA = cv.All(
WEB_SCHEMA = cv.Schema(
{
cv.Required(CONF_URL): cv.url,
},
_download_web_file,
}
)
@@ -209,6 +200,7 @@ def _validate_supported_local_file(config: list[ConfigType]) -> list[ConfigType]
CONFIG_SCHEMA = cv.All(
cv.only_on_esp32,
cv.ensure_list(MEDIA_FILE_TYPE_SCHEMA),
partial(download_web_files_in_config, path_for=_compute_local_file_path),
_validate_supported_local_file,
)
@@ -375,12 +375,10 @@ void FeedbackCover::start_direction_(CoverOperation dir) {
// check if we have a wait time
if (this->direction_change_waittime_.has_value() && dir != COVER_OPERATION_IDLE &&
this->current_operation != COVER_OPERATION_IDLE && dir != this->current_operation) {
const uint32_t waittime = *this->direction_change_waittime_;
ESP_LOGD(TAG, "'%s' - Reversing direction.", this->name_.c_str());
this->start_direction_(COVER_OPERATION_IDLE);
this->set_timeout(DIRECTION_CHANGE_TIMEOUT_ID, *this->direction_change_waittime_,
[this, dir]() { this->start_direction_(dir); });
this->set_timeout(DIRECTION_CHANGE_TIMEOUT_ID, waittime, [this, dir]() { this->start_direction_(dir); });
} else {
this->set_current_operation_(dir, true);
this->prev_command_trigger_ = trig;
+64 -50
View File
@@ -8,84 +8,93 @@ namespace esphome::light {
enum class LimitMode { CLAMP, DO_NOTHING };
template<typename... Ts> class ToggleAction : public Action<Ts...> {
template<bool HasTransitionLength, typename... Ts> class ToggleAction : public Action<Ts...> {
public:
explicit ToggleAction(LightState *state) : state_(state) {}
TEMPLATABLE_VALUE(uint32_t, transition_length)
template<typename V> void set_transition_length(V value) requires(HasTransitionLength) {
this->transition_length_ = value;
}
void play(const Ts &...x) override {
auto call = this->state_->toggle();
call.set_transition_length(this->transition_length_.optional_value(x...));
if constexpr (HasTransitionLength) {
call.set_transition_length(this->transition_length_.optional_value(x...));
}
call.perform();
}
protected:
LightState *state_;
struct NoTransition {};
[[no_unique_address]] std::conditional_t<HasTransitionLength, TemplatableFn<uint32_t, Ts...>, NoTransition>
transition_length_{};
};
template<typename... Ts> class LightControlAction : public Action<Ts...> {
// Unique Empty<Tag> per field so [[no_unique_address]] is guaranteed to coalesce.
namespace light_control_detail {
template<int Tag> struct Empty {};
} // namespace light_control_detail
// X-macro: (type, field_name, bit_index). Order and bit values must match
// the FIELDS table in automation.py.
#define LIGHT_CONTROL_FIELDS(X) \
X(ColorMode, color_mode, 0) \
X(bool, state, 1) \
X(uint32_t, transition_length, 2) \
X(uint32_t, flash_length, 3) \
X(float, brightness, 4) \
X(float, color_brightness, 5) \
X(float, red, 6) \
X(float, green, 7) \
X(float, blue, 8) \
X(float, white, 9) \
X(float, color_temperature, 10) \
X(float, cold_white, 11) \
X(float, warm_white, 12) \
X(uint32_t, effect, 13)
template<uint16_t Fields, typename... Ts> class LightControlAction : public Action<Ts...> {
public:
explicit LightControlAction(LightState *parent) : parent_(parent) {}
TEMPLATABLE_VALUE(ColorMode, color_mode)
TEMPLATABLE_VALUE(bool, state)
TEMPLATABLE_VALUE(uint32_t, transition_length)
TEMPLATABLE_VALUE(uint32_t, flash_length)
TEMPLATABLE_VALUE(float, brightness)
TEMPLATABLE_VALUE(float, color_brightness)
TEMPLATABLE_VALUE(float, red)
TEMPLATABLE_VALUE(float, green)
TEMPLATABLE_VALUE(float, blue)
TEMPLATABLE_VALUE(float, white)
TEMPLATABLE_VALUE(float, color_temperature)
TEMPLATABLE_VALUE(float, cold_white)
TEMPLATABLE_VALUE(float, warm_white)
TEMPLATABLE_VALUE(uint32_t, effect)
#define LIGHT_FIELD_SETTER_(type, name, idx) \
template<typename V> void set_##name(V value) requires((Fields & (1 << (idx))) != 0) { this->name##_ = value; }
#define LIGHT_FIELD_APPLY_(type, name, idx) \
if constexpr ((Fields & (1 << (idx))) != 0) \
call.set_##name(this->name##_.value(x...));
#define LIGHT_FIELD_DECL_(type, name, idx) \
[[no_unique_address]] std::conditional_t<(Fields & (1 << (idx))) != 0, TemplatableFn<type, Ts...>, \
light_control_detail::Empty<(idx)>> \
name##_{};
LIGHT_CONTROL_FIELDS(LIGHT_FIELD_SETTER_)
void play(const Ts &...x) override {
auto call = this->parent_->make_call();
if (this->color_mode_.has_value())
call.set_color_mode(this->color_mode_.value(x...));
if (this->state_.has_value())
call.set_state(this->state_.value(x...));
if (this->transition_length_.has_value())
call.set_transition_length(this->transition_length_.value(x...));
if (this->flash_length_.has_value())
call.set_flash_length(this->flash_length_.value(x...));
if (this->brightness_.has_value())
call.set_brightness(this->brightness_.value(x...));
if (this->color_brightness_.has_value())
call.set_color_brightness(this->color_brightness_.value(x...));
if (this->red_.has_value())
call.set_red(this->red_.value(x...));
if (this->green_.has_value())
call.set_green(this->green_.value(x...));
if (this->blue_.has_value())
call.set_blue(this->blue_.value(x...));
if (this->white_.has_value())
call.set_white(this->white_.value(x...));
if (this->color_temperature_.has_value())
call.set_color_temperature(this->color_temperature_.value(x...));
if (this->cold_white_.has_value())
call.set_cold_white(this->cold_white_.value(x...));
if (this->warm_white_.has_value())
call.set_warm_white(this->warm_white_.value(x...));
if (this->effect_.has_value())
call.set_effect(this->effect_.value(x...));
LIGHT_CONTROL_FIELDS(LIGHT_FIELD_APPLY_)
call.perform();
}
protected:
LightState *parent_;
};
LIGHT_CONTROL_FIELDS(LIGHT_FIELD_DECL_)
template<typename... Ts> class DimRelativeAction : public Action<Ts...> {
#undef LIGHT_FIELD_DECL_
#undef LIGHT_FIELD_APPLY_
#undef LIGHT_FIELD_SETTER_
};
#undef LIGHT_CONTROL_FIELDS
template<bool HasTransitionLength, typename... Ts> class DimRelativeAction : public Action<Ts...> {
public:
explicit DimRelativeAction(LightState *parent) : parent_(parent) {}
TEMPLATABLE_VALUE(float, relative_brightness)
TEMPLATABLE_VALUE(uint32_t, transition_length)
template<typename V> void set_transition_length(V value) requires(HasTransitionLength) {
this->transition_length_ = value;
}
void play(const Ts &...x) override {
auto call = this->parent_->make_call();
@@ -99,7 +108,9 @@ template<typename... Ts> class DimRelativeAction : public Action<Ts...> {
call.set_state(new_brightness != 0.0f);
call.set_brightness(new_brightness);
call.set_transition_length(this->transition_length_.optional_value(x...));
if constexpr (HasTransitionLength) {
call.set_transition_length(this->transition_length_.optional_value(x...));
}
call.perform();
}
@@ -115,6 +126,9 @@ template<typename... Ts> class DimRelativeAction : public Action<Ts...> {
float min_brightness_{0.0};
float max_brightness_{1.0};
LimitMode limit_mode_{LimitMode::CLAMP};
struct NoTransition {};
[[no_unique_address]] std::conditional_t<HasTransitionLength, TemplatableFn<uint32_t, Ts...>, NoTransition>
transition_length_{};
};
template<typename... Ts> class LightIsOnCondition : public Condition<Ts...> {
+21 -7
View File
@@ -60,8 +60,10 @@ from .types import (
)
async def light_toggle_to_code(config, action_id, template_arg, args):
paren = await cg.get_variable(config[CONF_ID])
var = cg.new_Pvariable(action_id, template_arg, paren)
if CONF_TRANSITION_LENGTH in config:
has_transition_length = CONF_TRANSITION_LENGTH in config
toggle_template_arg = cg.TemplateArguments(has_transition_length, *template_arg)
var = cg.new_Pvariable(action_id, toggle_template_arg, paren)
if has_transition_length:
template_ = await cg.templatable(
config[CONF_TRANSITION_LENGTH], args, cg.uint32
)
@@ -178,9 +180,9 @@ def _resolve_effect_index(config: ConfigType) -> int:
)
async def light_control_to_code(config, action_id, template_arg, args):
paren = await cg.get_variable(config[CONF_ID])
var = cg.new_Pvariable(action_id, template_arg, paren)
# (config_key, setter_name, c++ type)
# Order/bits must match LIGHT_CONTROL_FIELDS in automation.h.
# EFFECT has special handling below; setter=None skips the generic loop.
FIELDS = (
(CONF_COLOR_MODE, "set_color_mode", ColorMode),
(CONF_STATE, "set_state", cg.bool_),
@@ -195,9 +197,19 @@ async def light_control_to_code(config, action_id, template_arg, args):
(CONF_COLOR_TEMPERATURE, "set_color_temperature", cg.float_),
(CONF_COLD_WHITE, "set_cold_white", cg.float_),
(CONF_WARM_WHITE, "set_warm_white", cg.float_),
(CONF_EFFECT, None, cg.uint32),
)
# Bitmask is passed as uint16_t in C++ — must stay within 16 bits.
assert len(FIELDS) <= 16, "LightControlAction Fields bitmask exceeds uint16_t"
field_mask = sum(1 << i for i, (k, _, _) in enumerate(FIELDS) if k in config)
control_template_arg = cg.TemplateArguments(
cg.RawExpression(f"static_cast<uint16_t>({field_mask})"), *template_arg
)
var = cg.new_Pvariable(action_id, control_template_arg, paren)
for conf_key, setter, type_ in FIELDS:
if conf_key in config:
if conf_key in config and setter is not None:
template_ = await cg.templatable(config[conf_key], args, type_)
cg.add(getattr(var, setter)(template_))
@@ -261,10 +273,12 @@ LIGHT_DIM_RELATIVE_ACTION_SCHEMA = cv.Schema(
)
async def light_dim_relative_to_code(config, action_id, template_arg, args):
paren = await cg.get_variable(config[CONF_ID])
var = cg.new_Pvariable(action_id, template_arg, paren)
has_transition_length = CONF_TRANSITION_LENGTH in config
dim_template_arg = cg.TemplateArguments(has_transition_length, *template_arg)
var = cg.new_Pvariable(action_id, dim_template_arg, paren)
templ = await cg.templatable(config[CONF_RELATIVE_BRIGHTNESS], args, cg.float_)
cg.add(var.set_relative_brightness(templ))
if CONF_TRANSITION_LENGTH in config:
if has_transition_length:
templ = await cg.templatable(config[CONF_TRANSITION_LENGTH], args, cg.uint32)
cg.add(var.set_transition_length(templ))
if conf := config.get(CONF_BRIGHTNESS_LIMITS):
@@ -1,5 +1,6 @@
"""Speaker Media Player Setup."""
from functools import partial
import hashlib
import logging
from pathlib import Path
@@ -32,7 +33,7 @@ from esphome.const import (
CONF_URL,
)
from esphome.core import CORE, HexInt
from esphome.external_files import download_content
from esphome.external_files import download_web_files_in_config
_LOGGER = logging.getLogger(__name__)
@@ -92,15 +93,6 @@ def _compute_local_file_path(value: dict) -> Path:
return base_dir / key
def _download_web_file(value):
url = value[CONF_URL]
path = _compute_local_file_path(value)
download_content(url, path)
_LOGGER.debug("download_web_file: path=%s", path)
return value
_PURPOSE_MAP = {
"MEDIA": media_player.MEDIA_PLAYER_FORMAT_PURPOSE_ENUM["default"],
"ANNOUNCEMENT": media_player.MEDIA_PLAYER_FORMAT_PURPOSE_ENUM["announcement"],
@@ -229,11 +221,10 @@ LOCAL_SCHEMA = cv.Schema(
}
)
WEB_SCHEMA = cv.All(
WEB_SCHEMA = cv.Schema(
{
cv.Required(CONF_URL): cv.url,
},
_download_web_file,
}
)
@@ -285,7 +276,12 @@ CONFIG_SCHEMA = cv.All(
),
# Remove before 2026.10.0
cv.Optional(CONF_CODEC_SUPPORT_ENABLED): cv.Any(cv.boolean, cv.string),
cv.Optional(CONF_FILES): cv.ensure_list(MEDIA_FILE_TYPE_SCHEMA),
cv.Optional(CONF_FILES): cv.All(
cv.ensure_list(MEDIA_FILE_TYPE_SCHEMA),
partial(
download_web_files_in_config, path_for=_compute_local_file_path
),
),
cv.Optional(CONF_TASK_STACK_IN_PSRAM): cv.All(
cv.boolean, cv.requires_component(psram.DOMAIN)
),
+16
View File
@@ -73,6 +73,7 @@ NO_WIFI_VARIANTS = [const.VARIANT_ESP32H2, const.VARIANT_ESP32P4]
CONF_SAVE = "save"
CONF_BAND_MODE = "band_mode"
CONF_MIN_AUTH_MODE = "min_auth_mode"
CONF_PHY_MODE = "phy_mode"
CONF_POST_CONNECT_ROAMING = "post_connect_roaming"
# Maximum number of WiFi networks that can be configured
@@ -112,6 +113,14 @@ WIFI_MIN_AUTH_MODES = {
"WPA3": WifiMinAuthMode.WIFI_MIN_AUTH_MODE_WPA3,
}
VALIDATE_WIFI_MIN_AUTH_MODE = cv.enum(WIFI_MIN_AUTH_MODES, upper=True)
WiFi8266PhyMode = wifi_ns.enum("WiFi8266PhyMode")
WIFI_8266_PHY_MODES = {
"AUTO": WiFi8266PhyMode.WIFI_8266_PHY_MODE_AUTO,
"11B": WiFi8266PhyMode.WIFI_8266_PHY_MODE_11B,
"11G": WiFi8266PhyMode.WIFI_8266_PHY_MODE_11G,
"11N": WiFi8266PhyMode.WIFI_8266_PHY_MODE_11N,
}
WiFiConnectedCondition = wifi_ns.class_("WiFiConnectedCondition", Condition)
WiFiEnabledCondition = wifi_ns.class_("WiFiEnabledCondition", Condition)
WiFiAPActiveCondition = wifi_ns.class_("WiFiAPActiveCondition", Condition)
@@ -406,6 +415,10 @@ CONFIG_SCHEMA = cv.All(
cv.only_on_esp32,
only_on_variant(supported=[const.VARIANT_ESP32C5]),
),
cv.Optional(CONF_PHY_MODE): cv.All(
cv.enum(WIFI_8266_PHY_MODES, upper=True),
cv.only_on_esp8266,
),
cv.Optional(CONF_PASSIVE_SCAN, default=False): cv.boolean,
cv.Optional(CONF_ENABLE_ON_BOOT, default=True): cv.boolean,
cv.Optional(CONF_POST_CONNECT_ROAMING, default=True): cv.boolean,
@@ -569,6 +582,9 @@ async def to_code(config):
if CORE.is_esp8266:
cg.add_library("ESP8266WiFi", None)
if CONF_PHY_MODE in config:
cg.add_define("USE_WIFI_PHY_MODE")
cg.add(var.set_phy_mode(config[CONF_PHY_MODE]))
elif CORE.is_rp2040:
cg.add_library("WiFi", None)
@@ -309,6 +309,18 @@ bool CompactString::operator==(const StringRef &other) const {
/// └──────────────────────────────────────────────────────────────────────┘
#if ESPHOME_LOG_LEVEL >= ESPHOME_LOG_LEVEL_INFO
#ifdef USE_WIFI_PHY_MODE
// Use if-chain instead of switch to avoid jump table in RODATA (wastes RAM on ESP8266)
static const LogString *phy_mode_to_log_string(WiFi8266PhyMode mode) {
if (mode == WIFI_8266_PHY_MODE_11B)
return LOG_STR("11B");
if (mode == WIFI_8266_PHY_MODE_11G)
return LOG_STR("11G");
if (mode == WIFI_8266_PHY_MODE_11N)
return LOG_STR("11N");
return LOG_STR("Auto");
}
#endif
// Use if-chain instead of switch to avoid jump table in RODATA (wastes RAM on ESP8266)
static const LogString *retry_phase_to_log_string(WiFiRetryPhase phase) {
if (phase == WiFiRetryPhase::INITIAL_CONNECT)
@@ -1535,6 +1547,9 @@ void WiFiComponent::dump_config() {
break;
}
ESP_LOGCONFIG(TAG, " Band Mode: %s", band_mode_s);
#endif
#ifdef USE_WIFI_PHY_MODE
ESP_LOGCONFIG(TAG, " PHY Mode: %s", LOG_STR_ARG(phy_mode_to_log_string(this->phy_mode_)));
#endif
if (this->is_connected()) {
this->print_connect_params_();
+20
View File
@@ -345,6 +345,17 @@ enum WifiMinAuthMode : uint8_t {
WIFI_MIN_AUTH_MODE_WPA3,
};
#ifdef USE_WIFI_PHY_MODE
// Values 1-3 match ESP8266 SDK phy_mode_t (PHY_MODE_11B=1, PHY_MODE_11G=2, PHY_MODE_11N=3).
// AUTO leaves the SDK at its default (no wifi_set_phy_mode() call).
enum WiFi8266PhyMode : uint8_t {
WIFI_8266_PHY_MODE_AUTO = 0,
WIFI_8266_PHY_MODE_11B = 1,
WIFI_8266_PHY_MODE_11G = 2,
WIFI_8266_PHY_MODE_11N = 3,
};
#endif
#ifdef USE_ESP32
struct IDFWiFiEvent;
#endif
@@ -455,6 +466,9 @@ class WiFiComponent final : public Component {
#if defined(USE_ESP32) && defined(SOC_WIFI_SUPPORT_5G)
void set_band_mode(wifi_band_mode_t band_mode) { this->band_mode_ = band_mode; }
#endif
#ifdef USE_WIFI_PHY_MODE
void set_phy_mode(WiFi8266PhyMode phy_mode) { this->phy_mode_ = phy_mode; }
#endif
void set_passive_scan(bool passive);
@@ -672,6 +686,9 @@ class WiFiComponent final : public Component {
bool wifi_apply_power_save_();
#if defined(USE_ESP32) && defined(SOC_WIFI_SUPPORT_5G)
bool wifi_apply_band_mode_();
#endif
#ifdef USE_WIFI_PHY_MODE
bool wifi_apply_phy_mode_();
#endif
bool wifi_sta_ip_config_(const optional<ManualIP> &manual_ip);
bool wifi_apply_hostname_();
@@ -810,6 +827,9 @@ class WiFiComponent final : public Component {
WiFiPowerSaveMode power_save_{WIFI_POWER_SAVE_NONE};
#if defined(USE_ESP32) && defined(SOC_WIFI_SUPPORT_5G)
wifi_band_mode_t band_mode_{WIFI_BAND_MODE_AUTO};
#endif
#ifdef USE_WIFI_PHY_MODE
WiFi8266PhyMode phy_mode_{WIFI_8266_PHY_MODE_AUTO};
#endif
WifiMinAuthMode min_auth_mode_{WIFI_MIN_AUTH_MODE_WPA2};
WiFiRetryPhase retry_phase_{WiFiRetryPhase::INITIAL_CONNECT};
@@ -621,10 +621,25 @@ bool WiFiComponent::wifi_sta_pre_setup_() {
ESP_LOGV(TAG, "Disabling Auto-Connect failed");
}
#ifdef USE_WIFI_PHY_MODE
if (!this->wifi_apply_phy_mode_()) {
ESP_LOGV(TAG, "Setting PHY Mode failed");
}
#endif
delay(10);
return true;
}
#ifdef USE_WIFI_PHY_MODE
bool WiFiComponent::wifi_apply_phy_mode_() {
if (this->phy_mode_ == WIFI_8266_PHY_MODE_AUTO)
return true;
// Values of WiFi8266PhyMode are aligned with the SDK's phy_mode_t enum.
return wifi_set_phy_mode(static_cast<phy_mode_t>(this->phy_mode_));
}
#endif
void WiFiComponent::wifi_pre_setup_() {
wifi_set_event_handler_cb(&WiFiComponent::wifi_event_callback);
+1
View File
@@ -297,6 +297,7 @@
#define USE_CAPTIVE_PORTAL_GZIP
#define USE_WIFI_11KV_SUPPORT
#define USE_WIFI_FAST_CONNECT
#define USE_WIFI_PHY_MODE
#define USE_WIFI_IP_STATE_LISTENERS
#define USE_WIFI_SCAN_RESULTS_LISTENERS
#define USE_WIFI_CONNECT_STATE_LISTENERS
+94 -4
View File
@@ -1,5 +1,7 @@
from __future__ import annotations
from collections.abc import Callable, Iterable
from concurrent.futures import ThreadPoolExecutor
import contextlib
from datetime import UTC, datetime
import logging
@@ -9,9 +11,10 @@ from pathlib import Path
import requests
import esphome.config_validation as cv
from esphome.const import __version__
from esphome.const import CONF_FILE, CONF_TYPE, CONF_URL, __version__
from esphome.core import CORE, EsphomeError, TimePeriodSeconds
from esphome.helpers import write_file
from esphome.types import ConfigType
_LOGGER = logging.getLogger(__name__)
CODEOWNERS = ["@landonr"]
@@ -85,7 +88,9 @@ def _write_etag(local_file_path: Path, etag: str | None) -> None:
)
def has_remote_file_changed(url: str, local_file_path: Path) -> bool:
def has_remote_file_changed(
url: str, local_file_path: Path, timeout: int = NETWORK_TIMEOUT
) -> bool:
if local_file_path.exists():
_LOGGER.debug("has_remote_file_changed: File exists at %s", local_file_path)
try:
@@ -101,7 +106,7 @@ def has_remote_file_changed(url: str, local_file_path: Path) -> bool:
if etag := _read_etag(local_file_path):
headers[IF_NONE_MATCH] = etag
response = requests.head(
url, headers=headers, timeout=NETWORK_TIMEOUT, allow_redirects=True
url, headers=headers, timeout=timeout, allow_redirects=True
)
_LOGGER.debug(
@@ -153,7 +158,7 @@ def download_content(url: str, path: Path, timeout: int = NETWORK_TIMEOUT) -> by
if CORE.skip_external_update and path.exists():
_LOGGER.debug("Skipping update for %s (refresh disabled)", url)
return path.read_bytes()
if not has_remote_file_changed(url, path):
if not has_remote_file_changed(url, path, timeout):
_LOGGER.debug("Remote file has not changed %s", url)
return path.read_bytes()
@@ -184,3 +189,88 @@ def download_content(url: str, path: Path, timeout: int = NETWORK_TIMEOUT) -> by
write_file(path, data)
_write_etag(path, req.headers.get(ETAG))
return data
# Cap concurrent connections so a config with hundreds of remote files doesn't
# open hundreds of sockets at once. 8 matches the requests connection-pool
# default and the per-host connection limit browsers use, which keeps us
# polite to the upstream host while still cutting wall time roughly 8x for
# typical configs (a couple dozen files).
DEFAULT_DOWNLOAD_WORKERS = 8
def download_content_many(
items: Iterable[tuple[str, Path]],
timeout: int = NETWORK_TIMEOUT,
max_workers: int = DEFAULT_DOWNLOAD_WORKERS,
) -> None:
"""Run `download_content` for each (url, path) pair concurrently.
Wall time drops from `sum(latency)` to roughly `max(latency)` for cached
files where the HEAD round-trip dominates. All workers run to
completion before this returns; every `cv.Invalid` raised by a worker
is collected and surfaced together as `cv.MultipleInvalid` so the user
sees every broken file in a single validation pass instead of fixing
them one round-trip at a time.
Items are de-duplicated by `path` -- two callers asking for the same
cache file (e.g. the same URL referenced twice in a config) would
otherwise race on `download_content`'s non-atomic write. When the
same `path` appears more than once, the last URL wins (standard dict
comprehension semantics); in practice duplicate paths only arise when
the URL is duplicated, so the choice doesn't matter.
"""
seen: dict[Path, str] = {path: url for url, path in items}
if not seen:
return
if len(seen) == 1:
path, url = next(iter(seen.items()))
download_content(url, path, timeout)
return
def _download_one(path_url: tuple[Path, str]) -> None:
# `seen` stores entries as (path, url) so the dict can dedupe by
# path; flip them back to download_content's (url, path) order.
path, url = path_url
download_content(url, path, timeout)
workers = max(1, min(max_workers, len(seen)))
errors: list[cv.Invalid] = []
with ThreadPoolExecutor(max_workers=workers) as ex:
futures = [ex.submit(_download_one, item) for item in seen.items()]
for future in futures:
try:
future.result()
except cv.Invalid as e:
errors.append(e)
if not errors:
return
if len(errors) == 1:
raise errors[0]
raise cv.MultipleInvalid(errors)
# Each component that uses external_files defines its own local
# `TYPE_WEB = "web"`; the string is repeated here rather than imported
# because there is no canonical `TYPE_WEB` in `esphome.const` to share.
WEB_TYPE = "web"
def download_web_files_in_config(
config: list[ConfigType],
path_for: Callable[[ConfigType], Path],
) -> list[ConfigType]:
"""Voluptuous-friendly validator that downloads any web-sourced files in
`config` in parallel.
Each entry is expected to contain a `file` key whose value is a dict
that may be `{type: "web", url: ...}`; `path_for(file_dict)` returns
the cache path for that file. Returns `config` unchanged so it can be
slotted directly into a `cv.All(...)` chain.
"""
download_content_many(
(conf_file[CONF_URL], path_for(conf_file))
for entry in config
if (conf_file := entry.get(CONF_FILE, {})).get(CONF_TYPE) == WEB_TYPE
)
return config
+54 -25
View File
@@ -57,6 +57,59 @@ def hash_components(components: list[str]) -> str:
return hashlib.sha256(key.encode()).hexdigest()[:16]
def populate_dependency_config(
config: dict,
component_names: list[str],
*,
get_component_fn: Callable[[str], object | None] = get_component,
register_platform_fn: Callable[[str], None] | None = None,
) -> None:
"""Populate ``config`` with empty entries for transitive dependencies.
For every name in ``component_names``:
* ``domain.platform`` form (e.g. ``sensor.gpio``) appends
``{platform: <name>}`` to ``config[domain]``, creating the list if needed.
* Bare components are looked up via ``get_component_fn``. Platform
components (``IS_PLATFORM_COMPONENT``) and ``MULTI_CONF`` components are
initialised as ``[]`` so the sibling ``domain.platform`` branch can
``append`` into them. Everything else is populated by running the
component's schema with ``{}`` so defaults exist; if the schema requires
explicit input, an empty ``{}`` is used as a fallback.
Platform components must always be a list here even when no
``domain.platform`` entry follows, because the ``domain.platform`` branch
does ``config.setdefault(domain, []).append(...)`` and would crash on a
leftover dict.
"""
if register_platform_fn is None:
register_platform_fn = CORE.testing_ensure_platform_registered
for component_name in component_names:
if "." in component_name:
domain, component = component_name.split(".", maxsplit=1)
domain_list = config.setdefault(domain, [])
register_platform_fn(domain)
domain_list.append({CONF_PLATFORM: component})
continue
# Skip "core" — it's a pseudo-component handled by the build
# system, not a real loadable component (get_component returns None)
component = get_component_fn(component_name)
if component is None:
continue
if component.multi_conf or component.is_platform_component:
config.setdefault(component_name, [])
elif component_name not in config:
schema = component.config_schema
try:
config[component_name] = schema({}) if schema is not None else {}
except Exception: # noqa: BLE001
# Schema requires explicit input we can't synthesize; fall
# back to an empty mapping so subscripting at least returns
# KeyError on missing keys rather than crashing on the
# wrong type.
config[component_name] = {}
def filter_components_with_files(components: list[str], tests_dir: Path) -> list[str]:
"""Filter out components that do not have .cpp or .h files in the tests dir.
@@ -316,31 +369,7 @@ def compile_and_get_binary(
# Add remaining components and dependencies to the configuration after
# validation, so their source files are included in the build.
for component_name in components_with_dependencies:
if "." in component_name:
domain, component = component_name.split(".", maxsplit=1)
domain_list = config.setdefault(domain, [])
CORE.testing_ensure_platform_registered(domain)
domain_list.append({CONF_PLATFORM: component})
# Skip "core" — it's a pseudo-component handled by the build
# system, not a real loadable component (get_component returns None)
elif (component := get_component(component_name)) is not None:
# MULTI_CONF components store their config as a list of dicts,
# everything else stores a single dict. Run the component's
# schema with {} so defaults get populated -- code paths like
# socket.FILTER_SOURCE_FILES expect a fully-populated mapping.
if component.multi_conf:
config.setdefault(component_name, [])
elif component_name not in config:
schema = component.config_schema
try:
config[component_name] = schema({}) if schema is not None else {}
except Exception: # noqa: BLE001
# Schema requires explicit input we can't synthesize; fall
# back to an empty mapping so subscripting at least returns
# KeyError on missing keys rather than crashing on the
# wrong type.
config[component_name] = {}
populate_dependency_config(config, components_with_dependencies)
# Register platforms from the extra config (benchmark.yaml) so
# USE_SENSOR, USE_LIGHT, etc. defines are emitted without needing
+56
View File
@@ -29,3 +29,59 @@ climate:
heat_action:
- switch.turn_on: climate_heater_switch
- switch.turn_off: climate_cooler_switch
# Thermostat-based climate so climate.control: action variants get build
# coverage (bang_bang doesn't support fan modes, presets, etc.). Climate
# has no template platform, so thermostat is the right vehicle.
- platform: thermostat
id: climate_test_thermostat
name: Test Thermostat
sensor: climate_temperature_sensor
min_idle_time: 30s
min_heating_off_time: 300s
min_heating_run_time: 300s
min_cooling_off_time: 300s
min_cooling_run_time: 300s
heat_action:
- logger.log: heating
idle_action:
- logger.log: idle
cool_action:
- logger.log: cooling
auto_mode:
- logger.log: auto
heat_cool_mode:
- logger.log: heat_cool
preset:
- name: Default
default_target_temperature_low: 18°C
default_target_temperature_high: 22°C
button:
# Exercise the climate.control: action so ControlAction templates get
# build coverage. Various field combinations are tested.
- platform: template
name: "Climate Control Mode"
on_press:
- climate.control:
id: climate_test_thermostat
mode: HEAT
- platform: template
name: "Climate Control Mode And Temps"
on_press:
- climate.control:
id: climate_test_thermostat
mode: HEAT_COOL
target_temperature_low: 19.0°C
target_temperature_high: 23.0°C
- platform: template
name: "Climate Control Lambda Temp"
on_press:
- climate.control:
id: climate_test_thermostat
target_temperature_high: !lambda "return 21.5;"
- platform: template
name: "Climate Control Off"
on_press:
- climate.control:
id: climate_test_thermostat
mode: "OFF"
+4
View File
@@ -108,6 +108,10 @@ esphome:
relative_brightness: 5%
brightness_limits:
max_brightness: 90%
- light.dim_relative:
id: test_monochromatic_light
relative_brightness: -5%
transition_length: 250ms
- light.turn_on:
id: test_addressable_transition
brightness: 50%
@@ -293,6 +293,60 @@ cover:
cover.is_closed: template_cover_with_triggers
then:
logger.log: Cover is closed
# Exercise cover.control / cover.template.publish action variants so they
# get build coverage in CI (and so memory-impact analysis on PRs that
# touch ControlAction / CoverPublishAction sees real instances).
- platform: template
name: "Template Cover Actions"
id: template_cover_actions
has_position: true
optimistic: true
open_action:
# CONF_STATE alias for the position bit
- cover.template.publish:
id: template_cover_actions
state: OPEN
- cover.template.publish:
id: template_cover_actions
position: 1.0
- cover.template.publish:
id: template_cover_actions
current_operation: IDLE
close_action:
- cover.template.publish:
id: template_cover_actions
position: 0.0
tilt: 0.0
stop_action:
- cover.template.publish:
id: template_cover_actions
current_operation: IDLE
tilt_action:
- lambda: |-
id(template_cover_actions).tilt = tilt;
id(template_cover_actions).publish_state();
on_idle:
# position only
- cover.control:
id: template_cover_actions
position: 50%
# tilt only
- cover.control:
id: template_cover_actions
tilt: 75%
# position + tilt
- cover.control:
id: template_cover_actions
position: 25%
tilt: 30%
# stop
- cover.control:
id: template_cover_actions
stop: true
# CONF_STATE alias for position
- cover.control:
id: template_cover_actions
state: OPEN
number:
- platform: template
@@ -1,6 +1,7 @@
wifi:
min_auth_mode: WPA2
post_connect_roaming: true
phy_mode: 11G
packages:
- !include common.yaml
@@ -0,0 +1,60 @@
esphome:
name: light-dim-relative-action-test
host:
api:
logger:
level: DEBUG
output:
- platform: template
id: test_out
type: float
write_action:
- lambda: ""
light:
- platform: monochromatic
name: "Test Light"
id: test_light
output: test_out
default_transition_length: 0s
button:
# Set up: turn on at 50% brightness
- platform: template
id: btn_setup
name: "Setup"
on_press:
- light.turn_on:
id: test_light
brightness: 50%
# Test 1: dim_relative without transition_length (HasTransitionLength=false)
- platform: template
id: btn_dim_up
name: "Dim Up"
on_press:
- light.dim_relative:
id: test_light
relative_brightness: 25%
# Test 2: dim_relative with transition_length (HasTransitionLength=true)
- platform: template
id: btn_dim_down
name: "Dim Down"
on_press:
- light.dim_relative:
id: test_light
relative_brightness: -10%
transition_length: 0s
# Test 3: dim_relative with brightness limits
- platform: template
id: btn_dim_clamp
name: "Dim Clamp"
on_press:
- light.dim_relative:
id: test_light
relative_brightness: 50%
brightness_limits:
max_brightness: 80%
@@ -0,0 +1,37 @@
esphome:
name: light-toggle-action-test
host:
api:
logger:
level: DEBUG
output:
- platform: template
id: test_out
type: float
write_action:
- lambda: ""
light:
- platform: monochromatic
name: "Test Light"
id: test_light
output: test_out
default_transition_length: 0s
button:
# Test 1: light.toggle without transition_length (HasTransitionLength=false)
- platform: template
id: btn_toggle
name: "Toggle"
on_press:
- light.toggle: test_light
# Test 2: light.toggle with transition_length (HasTransitionLength=true)
- platform: template
id: btn_toggle_with_trans
name: "Toggle With Trans"
on_press:
- light.toggle:
id: test_light
transition_length: 0s
@@ -0,0 +1,72 @@
"""Integration test for light::DimRelativeAction.
Tests both DimRelativeAction<HasTransitionLength=false> and
DimRelativeAction<HasTransitionLength=true> instantiations.
"""
from __future__ import annotations
import asyncio
from aioesphomeapi import ButtonInfo, EntityState, LightInfo, LightState
import pytest
from .state_utils import InitialStateHelper, require_entity
from .types import APIClientConnectedFactory, RunCompiledFunction
@pytest.mark.asyncio
async def test_light_dim_relative_action(
yaml_config: str,
run_compiled: RunCompiledFunction,
api_client_connected: APIClientConnectedFactory,
) -> None:
"""Test light.dim_relative with and without transition_length."""
loop = asyncio.get_running_loop()
async with run_compiled(yaml_config), api_client_connected() as client:
light_state_future: asyncio.Future[LightState] | None = None
def on_state(state: EntityState) -> None:
if (
isinstance(state, LightState)
and light_state_future is not None
and not light_state_future.done()
):
light_state_future.set_result(state)
async def wait_for_light_state(timeout: float = 5.0) -> LightState:
nonlocal light_state_future
light_state_future = loop.create_future()
try:
return await asyncio.wait_for(light_state_future, timeout)
finally:
light_state_future = None
entities, _ = await client.list_entities_services()
initial_state_helper = InitialStateHelper(entities)
client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
await initial_state_helper.wait_for_initial_states()
require_entity(entities, "test_light", LightInfo)
async def press_and_wait(name: str) -> LightState:
btn = require_entity(entities, name.lower().replace(" ", "_"), ButtonInfo)
client.button_command(btn.key)
return await wait_for_light_state()
# Setup: turn on at 50%
state = await press_and_wait("Setup")
assert state.state is True
assert state.brightness == pytest.approx(0.5, abs=0.05)
# Test 1: dim_relative without transition_length: 50% + 25% = 75%
state = await press_and_wait("Dim Up")
assert state.brightness == pytest.approx(0.75, abs=0.05)
# Test 2: dim_relative with transition_length: 75% - 10% = 65%
state = await press_and_wait("Dim Down")
assert state.brightness == pytest.approx(0.65, abs=0.05)
# Test 3: dim_relative with max_brightness limit: 65% + 50% clamped to 80%
state = await press_and_wait("Dim Clamp")
assert state.brightness == pytest.approx(0.80, abs=0.05)
@@ -0,0 +1,67 @@
"""Integration test for light::ToggleAction.
Tests both ToggleAction<HasTransitionLength=false> and
ToggleAction<HasTransitionLength=true> instantiations.
"""
from __future__ import annotations
import asyncio
from aioesphomeapi import ButtonInfo, EntityState, LightInfo, LightState
import pytest
from .state_utils import InitialStateHelper, require_entity
from .types import APIClientConnectedFactory, RunCompiledFunction
@pytest.mark.asyncio
async def test_light_toggle_action(
yaml_config: str,
run_compiled: RunCompiledFunction,
api_client_connected: APIClientConnectedFactory,
) -> None:
"""Test light.toggle with and without transition_length."""
loop = asyncio.get_running_loop()
async with run_compiled(yaml_config), api_client_connected() as client:
light_state_future: asyncio.Future[LightState] | None = None
def on_state(state: EntityState) -> None:
if (
isinstance(state, LightState)
and light_state_future is not None
and not light_state_future.done()
):
light_state_future.set_result(state)
async def wait_for_light_state(timeout: float = 5.0) -> LightState:
nonlocal light_state_future
light_state_future = loop.create_future()
try:
return await asyncio.wait_for(light_state_future, timeout)
finally:
light_state_future = None
entities, _ = await client.list_entities_services()
initial_state_helper = InitialStateHelper(entities)
client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
await initial_state_helper.wait_for_initial_states()
require_entity(entities, "test_light", LightInfo)
async def press_and_wait(name: str) -> LightState:
btn = require_entity(entities, name.lower().replace(" ", "_"), ButtonInfo)
client.button_command(btn.key)
return await wait_for_light_state()
# Test 1: toggle without transition_length flips off->on
state = await press_and_wait("Toggle")
assert state.state is True
# Test 2: toggle with transition_length flips on->off
state = await press_and_wait("Toggle With Trans")
assert state.state is False
# Test 3: toggle without transition_length flips off->on again
state = await press_and_wait("Toggle")
assert state.state is True
+158
View File
@@ -258,3 +258,161 @@ def test_load_wraps_platform_component(tmp_path: Path) -> None:
assert key == "bthome.sensor"
assert isinstance(installed, ComponentManifestOverride)
assert installed.to_code is None
# ---------------------------------------------------------------------------
# populate_dependency_config
# ---------------------------------------------------------------------------
def _make_component_stub(
*,
multi_conf: bool = False,
is_platform_component: bool = False,
config_schema=None,
) -> MagicMock:
stub = MagicMock()
stub.multi_conf = multi_conf
stub.is_platform_component = is_platform_component
stub.config_schema = config_schema
return stub
def test_populate_platform_component_listed_alone_uses_list() -> None:
"""Regression: a platform component (sensor) with no `sensor.x` siblings
must land as `[]` in config. Previously it was populated as a dict via
`schema({})`, which then crashed the sibling `domain.platform` branch
when later dependencies tried `config.setdefault('sensor', []).append(...)`.
"""
sensor = _make_component_stub(is_platform_component=True)
config: dict = {}
build_helpers.populate_dependency_config(
config,
["sensor"],
get_component_fn=lambda name: sensor if name == "sensor" else None,
register_platform_fn=lambda _: None,
)
assert config["sensor"] == []
def test_populate_platform_component_then_platform_entry() -> None:
"""When `sensor` is processed before `sensor.gpio` (sorted order),
the bare-component branch must leave `config['sensor']` as a list so
the platform-entry branch can append into it.
"""
sensor = _make_component_stub(is_platform_component=True)
gpio = _make_component_stub() # the bare `gpio` component
components: dict[str, object] = {"sensor": sensor, "gpio": gpio}
config: dict = {}
build_helpers.populate_dependency_config(
config,
["gpio", "sensor", "sensor.gpio"],
get_component_fn=components.get,
register_platform_fn=lambda _: None,
)
assert config["sensor"] == [{"platform": "gpio"}]
def test_populate_multi_conf_component_uses_list() -> None:
multi = _make_component_stub(multi_conf=True)
config: dict = {}
build_helpers.populate_dependency_config(
config,
["multi"],
get_component_fn=lambda name: multi if name == "multi" else None,
register_platform_fn=lambda _: None,
)
assert config["multi"] == []
def test_populate_plain_component_uses_schema_defaults() -> None:
schema = MagicMock(return_value={"default_key": 42})
plain = _make_component_stub(config_schema=schema)
config: dict = {}
build_helpers.populate_dependency_config(
config,
["plain"],
get_component_fn=lambda name: plain if name == "plain" else None,
register_platform_fn=lambda _: None,
)
schema.assert_called_once_with({})
assert config["plain"] == {"default_key": 42}
def test_populate_plain_component_falls_back_when_schema_raises() -> None:
def picky_schema(_):
raise ValueError("required field missing")
plain = _make_component_stub(config_schema=picky_schema)
config: dict = {}
build_helpers.populate_dependency_config(
config,
["plain"],
get_component_fn=lambda name: plain if name == "plain" else None,
register_platform_fn=lambda _: None,
)
assert config["plain"] == {}
def test_populate_skips_unresolvable_pseudo_components() -> None:
"""`core` and other names that get_component returns None for are skipped
silently without inserting anything into the config.
"""
config: dict = {}
build_helpers.populate_dependency_config(
config,
["core"],
get_component_fn=lambda _: None,
register_platform_fn=lambda _: None,
)
assert config == {}
def test_populate_preserves_existing_plain_component_config() -> None:
"""If a plain component already has a config entry (e.g. from the user's
YAML), the schema-defaults branch must not overwrite it.
"""
schema = MagicMock()
plain = _make_component_stub(config_schema=schema)
config: dict = {"plain": {"user_key": "set_by_user"}}
build_helpers.populate_dependency_config(
config,
["plain"],
get_component_fn=lambda name: plain if name == "plain" else None,
register_platform_fn=lambda _: None,
)
schema.assert_not_called()
assert config["plain"] == {"user_key": "set_by_user"}
def test_populate_registers_platform_for_platform_entry() -> None:
"""Each `domain.platform` entry triggers register_platform_fn(domain) so
USE_<DOMAIN> defines get emitted later in the build pipeline.
"""
registered: list[str] = []
config: dict = {}
build_helpers.populate_dependency_config(
config,
["sensor.gpio", "binary_sensor.gpio"],
get_component_fn=lambda _: None,
register_platform_fn=registered.append,
)
assert registered == ["sensor", "binary_sensor"]
assert config["sensor"] == [{"platform": "gpio"}]
assert config["binary_sensor"] == [{"platform": "gpio"}]
@@ -0,0 +1,43 @@
"""Tests for source-file-to-component attribution in memory analyzer."""
from unittest.mock import patch
from esphome.analyze_memory import MemoryAnalyzer
def _make_analyzer(external_components: set[str] | None = None) -> MemoryAnalyzer:
"""Create a MemoryAnalyzer with mocked dependencies."""
with patch.object(MemoryAnalyzer, "__init__", lambda self, *a, **kw: None):
analyzer = MemoryAnalyzer.__new__(MemoryAnalyzer)
analyzer.external_components = external_components or set()
analyzer._lib_hash_to_name = {}
return analyzer
def test_source_file_to_component_main_cpp_relative() -> None:
"""ESPHome-generated src/main.cpp.o (nm path form) attributes to core."""
analyzer = _make_analyzer()
assert analyzer._source_file_to_component("src/main.cpp.o") == "[esphome]core"
def test_source_file_to_component_main_cpp_pioenvs_path() -> None:
"""Linker map paths like .pioenvs/<env>/src/main.cpp.o attribute to core."""
analyzer = _make_analyzer()
result = analyzer._source_file_to_component(".pioenvs/drivewaygate/src/main.cpp.o")
assert result == "[esphome]core"
def test_source_file_to_component_esphome_core() -> None:
"""Sources under src/esphome/core/ attribute to core."""
analyzer = _make_analyzer()
result = analyzer._source_file_to_component("src/esphome/core/application.cpp.o")
assert result == "[esphome]core"
def test_source_file_to_component_known_component() -> None:
"""Known ESPHome components attribute to their component name."""
analyzer = _make_analyzer()
result = analyzer._source_file_to_component(
"src/esphome/components/wifi/wifi_component.cpp.o"
)
assert result == "[esphome]wifi"
+186 -1
View File
@@ -9,7 +9,7 @@ import pytest
import requests
from esphome import external_files
from esphome.config_validation import Invalid
from esphome.config_validation import Invalid, MultipleInvalid
from esphome.core import CORE, EsphomeError, TimePeriod
@@ -60,6 +60,24 @@ def mock_write_file() -> MagicMock:
yield m
@pytest.fixture
def mock_download_content() -> MagicMock:
"""Patch `external_files.download_content` for tests that exercise the
parallel batch helper without doing real I/O.
"""
with patch("esphome.external_files.download_content") as m:
yield m
@pytest.fixture
def mock_download_content_many() -> MagicMock:
"""Patch `external_files.download_content_many` for tests that exercise
the URL-collection helper without dispatching to the thread pool.
"""
with patch("esphome.external_files.download_content_many") as m:
yield m
def test_compute_local_file_dir(setup_core: Path) -> None:
"""Test compute_local_file_dir creates and returns correct path."""
domain = "font"
@@ -494,6 +512,173 @@ def test_download_content_skip_external_update_downloads_when_missing(
assert test_file.read_bytes() == new_content
def test_download_content_many_empty_is_noop(
mock_download_content: MagicMock, setup_core: Path
) -> None:
"""Empty input shouldn't spin up a thread pool or call download_content."""
external_files.download_content_many([])
mock_download_content.assert_not_called()
def test_download_content_many_single_item_avoids_pool(
mock_download_content: MagicMock, setup_core: Path
) -> None:
"""A single item should be downloaded inline (no thread pool overhead)."""
item = ("https://example.com/file.txt", setup_core / "f.txt")
external_files.download_content_many([item])
mock_download_content.assert_called_once_with(
item[0], item[1], external_files.NETWORK_TIMEOUT
)
def test_download_content_many_runs_in_parallel(
mock_download_content: MagicMock, setup_core: Path
) -> None:
"""Multiple items should run concurrently — total wall time ≈ max latency."""
import threading
barrier = threading.Barrier(3)
def slow_download(url: str, path: Path, timeout: int) -> bytes:
# If calls were serial this would deadlock (third caller never arrives
# while the first is blocked at the barrier).
barrier.wait(timeout=2.0)
return b""
mock_download_content.side_effect = slow_download
items = [
("https://example.com/a", setup_core / "a"),
("https://example.com/b", setup_core / "b"),
("https://example.com/c", setup_core / "c"),
]
external_files.download_content_many(items, max_workers=4)
assert mock_download_content.call_count == 3
def test_download_content_many_propagates_single_error(
mock_download_content: MagicMock, setup_core: Path
) -> None:
"""A single failing worker should raise its `Invalid` directly, not wrap
it in a `MultipleInvalid` that the caller would have to unpack.
"""
def fake_download(url: str, path: Path, timeout: int) -> bytes:
if url.endswith("bad"):
raise Invalid(f"could not download {url}")
return b""
mock_download_content.side_effect = fake_download
items = [
("https://example.com/ok", setup_core / "ok"),
("https://example.com/bad", setup_core / "bad"),
]
with pytest.raises(Invalid, match="could not download") as exc_info:
external_files.download_content_many(items)
assert not isinstance(exc_info.value, MultipleInvalid)
def test_download_content_many_aggregates_multiple_errors(
mock_download_content: MagicMock, setup_core: Path
) -> None:
"""Every failing worker should be reported in a single MultipleInvalid so
the user sees all broken URLs in one validation pass instead of fixing
them one network round-trip at a time.
"""
def fake_download(url: str, path: Path, timeout: int) -> bytes:
if url.endswith("ok"):
return b""
raise Invalid(f"could not download {url}")
mock_download_content.side_effect = fake_download
items = [
("https://example.com/ok", setup_core / "ok"),
("https://example.com/bad1", setup_core / "bad1"),
("https://example.com/bad2", setup_core / "bad2"),
]
with pytest.raises(MultipleInvalid) as exc_info:
external_files.download_content_many(items)
messages = {str(e) for e in exc_info.value.errors}
assert messages == {
"could not download https://example.com/bad1",
"could not download https://example.com/bad2",
}
def test_download_content_many_dedupes_by_path(
mock_download_content: MagicMock, setup_core: Path
) -> None:
"""Two items pointing at the same cache path must collapse to one
download -- otherwise concurrent writes race on the same file. Which
URL wins doesn't matter (in practice duplicate paths only arise when
the URL is duplicated), so we only assert the call count and path.
"""
path = setup_core / "shared"
items = [
("https://example.com/a", path),
("https://example.com/b", path),
("https://example.com/a", path),
]
external_files.download_content_many(items)
assert mock_download_content.call_count == 1
args, _ = mock_download_content.call_args
assert args[1] == path
def test_download_content_many_clamps_invalid_max_workers(
mock_download_content: MagicMock, setup_core: Path
) -> None:
"""`max_workers <= 0` must not raise from ThreadPoolExecutor; it should
be clamped up to at least 1 worker.
"""
items = [
("https://example.com/a", setup_core / "a"),
("https://example.com/b", setup_core / "b"),
]
external_files.download_content_many(items, max_workers=0)
assert mock_download_content.call_count == 2
def test_download_web_files_in_config_filters_and_dispatches(
mock_download_content_many: MagicMock, setup_core: Path
) -> None:
"""Only `file.type == "web"` entries should be forwarded to
download_content_many, and the unmodified config should be returned so
the helper can sit in a `cv.All(...)` chain.
"""
def path_for(file_dict: dict) -> Path:
return setup_core / file_dict["url"].rsplit("/", 1)[-1]
config = [
{"file": {"type": "web", "url": "https://example.com/a"}},
{"file": {"type": "local", "path": "/tmp/b"}},
{"file": {"type": "web", "url": "https://example.com/c"}},
{}, # no `file` key at all
]
result = external_files.download_web_files_in_config(config, path_for)
assert result is config
mock_download_content_many.assert_called_once()
assert list(mock_download_content_many.call_args[0][0]) == [
("https://example.com/a", setup_core / "a"),
("https://example.com/c", setup_core / "c"),
]
def test_download_web_files_in_config_no_web_entries(
mock_download_content_many: MagicMock, setup_core: Path
) -> None:
"""A config with no web entries should still call through to
download_content_many (which is itself a no-op for empty input) so the
behavior stays consistent.
"""
config = [{"file": {"type": "local", "path": "/tmp/a"}}]
external_files.download_web_files_in_config(config, lambda _: setup_core / "x")
mock_download_content_many.assert_called_once()
assert list(mock_download_content_many.call_args[0][0]) == []
def test_download_content_saves_etag(
mock_has_remote_file_changed: MagicMock,
mock_requests_get: MagicMock,