Merge remote-tracking branch 'upstream/fix-slow-serial-log-test' into integration

# Conflicts:
#	.github/workflows/ci.yml
#	esphome/components/esp32_ble_server/ble_server.cpp
#	esphome/components/mqtt/mqtt_backend_esp32.cpp
#	script/cpp_benchmark.py
#	script/determine-jobs.py
This commit is contained in:
J. Nick Koston
2026-03-17 14:30:31 -10:00
38 changed files with 902 additions and 195 deletions
+3 -1
View File
@@ -316,7 +316,9 @@ jobs:
needs:
- common
- determine-jobs
if: github.event_name == 'pull_request' && needs.determine-jobs.outputs.benchmarks == 'true'
if: >-
(github.event_name == 'push' && github.ref_name == 'dev') ||
(github.event_name == 'pull_request' && needs.determine-jobs.outputs.benchmarks == 'true')
steps:
- name: Check out code from GitHub
uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
+1 -1
View File
@@ -1,4 +1,4 @@
# ESPHome [![Discord Chat](https://img.shields.io/discord/429907082951524364.svg)](https://discord.gg/KhAMKrd) [![GitHub release](https://img.shields.io/github/release/esphome/esphome.svg)](https://GitHub.com/esphome/esphome/releases/)
# ESPHome [![Discord Chat](https://img.shields.io/discord/429907082951524364.svg)](https://discord.gg/KhAMKrd) [![GitHub release](https://img.shields.io/github/release/esphome/esphome.svg)](https://GitHub.com/esphome/esphome/releases/) [![CodSpeed](https://img.shields.io/endpoint?url=https://codspeed.io/badge.json)](https://codspeed.io/esphome/esphome)
<a href="https://esphome.io/">
<picture>
+6 -9
View File
@@ -24,7 +24,7 @@ void EE895Component::setup() {
this->read(serial_number, 20);
crc16_check = (serial_number[19] << 8) + serial_number[18];
if (crc16_check != calc_crc16_(serial_number, 19)) {
if (crc16_check != calc_crc16_(serial_number, 18)) {
this->error_code_ = CRC_CHECK_FAILED;
this->mark_failed();
return;
@@ -84,7 +84,7 @@ void EE895Component::write_command_(uint16_t addr, uint16_t reg_cnt) {
address[2] = addr & 0xFF;
address[3] = (reg_cnt >> 8) & 0xFF;
address[4] = reg_cnt & 0xFF;
crc16 = calc_crc16_(address, 6);
crc16 = calc_crc16_(address, 5);
address[5] = crc16 & 0xFF;
address[6] = (crc16 >> 8) & 0xFF;
this->write(address, 7);
@@ -95,7 +95,7 @@ float EE895Component::read_float_() {
uint8_t i2c_response[8];
this->read(i2c_response, 8);
crc16_check = (i2c_response[7] << 8) + i2c_response[6];
if (crc16_check != calc_crc16_(i2c_response, 7)) {
if (crc16_check != calc_crc16_(i2c_response, 6)) {
this->error_code_ = CRC_CHECK_FAILED;
this->status_set_warning();
return 0;
@@ -107,12 +107,9 @@ float EE895Component::read_float_() {
}
uint16_t EE895Component::calc_crc16_(const uint8_t buf[], uint8_t len) {
uint8_t crc_check_buf[22];
for (int i = 0; i < len; i++) {
crc_check_buf[i + 1] = buf[i];
}
crc_check_buf[0] = this->address_;
return crc16(crc_check_buf, len);
uint8_t addr = this->address_;
uint16_t crc = crc16(&addr, 1);
return crc16(buf, len, crc);
}
} // namespace ee895
} // namespace esphome
@@ -16,13 +16,9 @@ BLECharacteristic::~BLECharacteristic() {
for (auto *descriptor : this->descriptors_) {
delete descriptor; // NOLINT(cppcoreguidelines-owning-memory)
}
vSemaphoreDelete(this->set_value_lock_);
}
BLECharacteristic::BLECharacteristic(const ESPBTUUID uuid, uint32_t properties) : uuid_(uuid) {
this->set_value_lock_ = xSemaphoreCreateBinary();
xSemaphoreGive(this->set_value_lock_);
this->properties_ = (esp_gatt_char_prop_t) 0;
this->set_broadcast_property((properties & PROPERTY_BROADCAST) != 0);
@@ -35,11 +31,7 @@ BLECharacteristic::BLECharacteristic(const ESPBTUUID uuid, uint32_t properties)
void BLECharacteristic::set_value(ByteBuffer buffer) { this->set_value(buffer.get_data()); }
void BLECharacteristic::set_value(std::vector<uint8_t> &&buffer) {
xSemaphoreTake(this->set_value_lock_, 0L);
this->value_ = std::move(buffer);
xSemaphoreGive(this->set_value_lock_);
}
void BLECharacteristic::set_value(std::vector<uint8_t> &&buffer) { this->value_ = std::move(buffer); }
void BLECharacteristic::set_value(std::initializer_list<uint8_t> data) {
this->set_value(std::vector<uint8_t>(data)); // Delegate to move overload
@@ -16,8 +16,6 @@
#include <esp_gattc_api.h>
#include <esp_gatts_api.h>
#include <esp_bt_defs.h>
#include <freertos/FreeRTOS.h>
#include <freertos/semphr.h>
namespace esphome {
namespace esp32_ble_server {
@@ -84,8 +82,6 @@ class BLECharacteristic {
uint16_t value_read_offset_{0};
std::vector<uint8_t> value_;
SemaphoreHandle_t set_value_lock_;
std::vector<BLEDescriptor *> descriptors_;
struct ClientNotificationEntry {
@@ -39,11 +39,11 @@ void BLEServer::loop() {
// Start all services that are pending to start
if (!this->services_to_start_.empty()) {
size_t write_idx = 0;
for (auto service : this->services_to_start_) {
for (auto *service : this->services_to_start_) {
if (service->is_created()) {
service->start(); // Needs to be called once per characteristic in the service
}
// Keep services still pending (drop those that are starting or running)
// Remove services that have started or are starting
if (!service->is_starting() && !service->is_running()) {
this->services_to_start_[write_idx++] = service;
}
+31 -45
View File
@@ -7,50 +7,36 @@ namespace hdc2010 {
static const char *const TAG = "hdc2010";
static const uint8_t HDC2010_ADDRESS = 0x40; // 0b1000000 or 0b1000001 from datasheet
static const uint8_t HDC2010_CMD_CONFIGURATION_MEASUREMENT = 0x8F;
static const uint8_t HDC2010_CMD_START_MEASUREMENT = 0xF9;
static const uint8_t HDC2010_CMD_TEMPERATURE_LOW = 0x00;
static const uint8_t HDC2010_CMD_TEMPERATURE_HIGH = 0x01;
static const uint8_t HDC2010_CMD_HUMIDITY_LOW = 0x02;
static const uint8_t HDC2010_CMD_HUMIDITY_HIGH = 0x03;
static const uint8_t CONFIG = 0x0E;
static const uint8_t MEASUREMENT_CONFIG = 0x0F;
// Register addresses
static constexpr uint8_t REG_TEMPERATURE_LOW = 0x00;
static constexpr uint8_t REG_TEMPERATURE_HIGH = 0x01;
static constexpr uint8_t REG_HUMIDITY_LOW = 0x02;
static constexpr uint8_t REG_HUMIDITY_HIGH = 0x03;
static constexpr uint8_t REG_RESET_DRDY_INT_CONF = 0x0E;
static constexpr uint8_t REG_MEASUREMENT_CONF = 0x0F;
// REG_MEASUREMENT_CONF (0x0F) bit masks
static constexpr uint8_t MEAS_TRIG = 0x01; // Bit 0: measurement trigger
static constexpr uint8_t MEAS_CONF_MASK = 0x06; // Bits 2:1: measurement mode
static constexpr uint8_t HRES_MASK = 0x30; // Bits 5:4: humidity resolution
static constexpr uint8_t TRES_MASK = 0xC0; // Bits 7:6: temperature resolution
// REG_RESET_DRDY_INT_CONF (0x0E) bit masks
static constexpr uint8_t AMM_MASK = 0x70; // Bits 6:4: auto measurement mode
void HDC2010Component::setup() {
ESP_LOGCONFIG(TAG, "Running setup");
const uint8_t data[2] = {
0b00000000, // resolution 14bit for both humidity and temperature
0b00000000 // reserved
};
if (!this->write_bytes(HDC2010_CMD_CONFIGURATION_MEASUREMENT, data, 2)) {
ESP_LOGW(TAG, "Initial config instruction error");
this->status_set_warning();
return;
}
// Set measurement mode to temperature and humidity
// Set 14-bit resolution for both sensors and measurement mode to temp + humidity
uint8_t config_contents;
this->read_register(MEASUREMENT_CONFIG, &config_contents, 1);
config_contents = (config_contents & 0xF9); // Always set to TEMP_AND_HUMID mode
this->write_bytes(MEASUREMENT_CONFIG, &config_contents, 1);
this->read_register(REG_MEASUREMENT_CONF, &config_contents, 1);
config_contents &= ~(TRES_MASK | HRES_MASK | MEAS_CONF_MASK); // 14-bit temp, 14-bit humidity, temp+humidity mode
this->write_bytes(REG_MEASUREMENT_CONF, &config_contents, 1);
// Set rate to manual
this->read_register(CONFIG, &config_contents, 1);
config_contents &= 0x8F;
this->write_bytes(CONFIG, &config_contents, 1);
// Set temperature resolution to 14bit
this->read_register(CONFIG, &config_contents, 1);
config_contents &= 0x3F;
this->write_bytes(CONFIG, &config_contents, 1);
// Set humidity resolution to 14bit
this->read_register(CONFIG, &config_contents, 1);
config_contents &= 0xCF;
this->write_bytes(CONFIG, &config_contents, 1);
// Set auto measurement rate to manual (on-demand via MEAS_TRIG)
this->read_register(REG_RESET_DRDY_INT_CONF, &config_contents, 1);
config_contents &= ~AMM_MASK;
this->write_bytes(REG_RESET_DRDY_INT_CONF, &config_contents, 1);
}
void HDC2010Component::dump_config() {
@@ -67,9 +53,9 @@ void HDC2010Component::dump_config() {
void HDC2010Component::update() {
// Trigger measurement
uint8_t config_contents;
this->read_register(CONFIG, &config_contents, 1);
config_contents |= 0x01;
this->write_bytes(MEASUREMENT_CONFIG, &config_contents, 1);
this->read_register(REG_MEASUREMENT_CONF, &config_contents, 1);
config_contents |= MEAS_TRIG;
this->write_bytes(REG_MEASUREMENT_CONF, &config_contents, 1);
// 1ms delay after triggering the sample
set_timeout(1, [this]() {
@@ -90,8 +76,8 @@ void HDC2010Component::update() {
float HDC2010Component::read_temp() {
uint8_t byte[2];
this->read_register(HDC2010_CMD_TEMPERATURE_LOW, &byte[0], 1);
this->read_register(HDC2010_CMD_TEMPERATURE_HIGH, &byte[1], 1);
this->read_register(REG_TEMPERATURE_LOW, &byte[0], 1);
this->read_register(REG_TEMPERATURE_HIGH, &byte[1], 1);
uint16_t temp = encode_uint16(byte[1], byte[0]);
return (float) temp * 0.0025177f - 40.0f;
@@ -100,8 +86,8 @@ float HDC2010Component::read_temp() {
float HDC2010Component::read_humidity() {
uint8_t byte[2];
this->read_register(HDC2010_CMD_HUMIDITY_LOW, &byte[0], 1);
this->read_register(HDC2010_CMD_HUMIDITY_HIGH, &byte[1], 1);
this->read_register(REG_HUMIDITY_LOW, &byte[0], 1);
this->read_register(REG_HUMIDITY_HIGH, &byte[1], 1);
uint16_t humidity = encode_uint16(byte[1], byte[0]);
return (float) humidity * 0.001525879f;
@@ -84,7 +84,7 @@ void MQTTBackendESP32::loop() {
// handle only 1 message per loop iteration
Event *event = this->mqtt_event_queue_.pop();
if (event != nullptr) {
mqtt_event_handler_(*event);
this->mqtt_event_handler_(*event);
this->mqtt_event_pool_.release(event);
}
+3 -1
View File
@@ -406,7 +406,9 @@ QUANTILE_SCHEMA = cv.All(
cv.Optional(CONF_WINDOW_SIZE, default=5): cv.int_range(min=1, max=65535),
cv.Optional(CONF_SEND_EVERY, default=5): cv.int_range(min=1, max=65535),
cv.Optional(CONF_SEND_FIRST_AT, default=1): cv.int_range(min=1, max=65535),
cv.Optional(CONF_QUANTILE, default=0.9): cv.zero_to_one_float,
cv.Optional(CONF_QUANTILE, default=0.9): cv.float_range(
min=0, min_included=False, max=1
),
}
),
validate_send_first_at,
+10 -16
View File
@@ -26,16 +26,13 @@ void USBCDCACMInstance::queue_line_state_event(bool dtr, bool rts) {
event->data.line_state.dtr = dtr;
event->data.line_state.rts = rts;
if (!this->event_queue_.push(event)) {
ESP_LOGW(TAG, "Event queue full, line state event dropped (itf=%d)", this->itf_);
// Return event to pool since we couldn't queue it
this->event_pool_.release(event);
} else {
// Wake main loop immediately to process event
// Push always succeeds: pool is sized to queue capacity (SIZE-1), so if
// allocate() returned non-null, the queue cannot be full.
this->event_queue_.push(event);
#if defined(USE_SOCKET_SELECT_SUPPORT) && defined(USE_WAKE_LOOP_THREADSAFE)
App.wake_loop_threadsafe();
App.wake_loop_threadsafe();
#endif
}
}
void USBCDCACMInstance::queue_line_coding_event(uint32_t bit_rate, uint8_t stop_bits, uint8_t parity,
@@ -53,16 +50,13 @@ void USBCDCACMInstance::queue_line_coding_event(uint32_t bit_rate, uint8_t stop_
event->data.line_coding.parity = parity;
event->data.line_coding.data_bits = data_bits;
if (!this->event_queue_.push(event)) {
ESP_LOGW(TAG, "Event queue full, line coding event dropped (itf=%d)", this->itf_);
// Return event to pool since we couldn't queue it
this->event_pool_.release(event);
} else {
// Wake main loop immediately to process event
// Push always succeeds: pool is sized to queue capacity (SIZE-1), so if
// allocate() returned non-null, the queue cannot be full.
this->event_queue_.push(event);
#if defined(USE_SOCKET_SELECT_SUPPORT) && defined(USE_WAKE_LOOP_THREADSAFE)
App.wake_loop_threadsafe();
App.wake_loop_threadsafe();
#endif
}
}
void USBCDCACMInstance::process_events_() {
+5 -1
View File
@@ -102,7 +102,11 @@ class USBCDCACMInstance : public uart::UARTComponent, public Parented<USBCDCACMC
LineStateCallback line_state_callback_{nullptr};
// Lock-free queue and event pool for cross-task event passing
EventPool<CDCEvent, EVENT_QUEUE_SIZE> event_pool_;
// Pool sized to queue capacity (SIZE-1) because LockFreeQueue<T,N> is a ring
// buffer that holds N-1 elements. This guarantees allocate() returns nullptr
// before push() can fail, preventing both a pool slot leak and an SPSC
// violation on the pool's internal free list.
EventPool<CDCEvent, EVENT_QUEUE_SIZE - 1> event_pool_;
LockFreeQueue<CDCEvent, EVENT_QUEUE_SIZE> event_queue_;
};
+4 -1
View File
@@ -144,7 +144,10 @@ class USBClient : public Component {
// Lock-free event queue and pool for USB task to main loop communication
// Must be public for access from static callbacks
LockFreeQueue<UsbEvent, USB_EVENT_QUEUE_SIZE> event_queue;
EventPool<UsbEvent, USB_EVENT_QUEUE_SIZE> event_pool;
// Pool sized to queue capacity (SIZE-1) because LockFreeQueue<T,N> is a ring
// buffer that holds N-1 elements. This guarantees allocate() returns nullptr
// before push() can fail, preventing a pool slot leak.
EventPool<UsbEvent, USB_EVENT_QUEUE_SIZE - 1> event_pool;
protected:
// Process USB events from the queue. Returns true if any work was done.
@@ -193,7 +193,8 @@ static void client_event_cb(const usb_host_client_event_msg_t *event_msg, void *
return;
}
// Push to lock-free queue (always succeeds since pool size == queue size)
// Push always succeeds: pool is sized to queue capacity (SIZE-1), so if
// allocate() returned non-null, the queue cannot be full.
client->event_queue.push(event);
// Re-enable component loop to process the queued event
+5 -6
View File
@@ -160,11 +160,9 @@ void USBUartChannel::write_array(const uint8_t *data, size_t len) {
size_t chunk_len = std::min(len, UsbOutputChunk::MAX_CHUNK_SIZE);
memcpy(chunk->data, data, chunk_len);
chunk->length = static_cast<uint8_t>(chunk_len);
if (!this->output_queue_.push(chunk)) {
this->output_pool_.release(chunk);
ESP_LOGE(TAG, "Output queue full - lost %zu bytes", len);
break;
}
// Push always succeeds: pool is sized to queue capacity (SIZE-1), so if
// allocate() returned non-null, the queue cannot be full.
this->output_queue_.push(chunk);
data += chunk_len;
len -= chunk_len;
}
@@ -320,7 +318,8 @@ void USBUartComponent::start_input(USBUartChannel *channel) {
chunk->channel = channel;
// Push to lock-free queue for main loop processing
// Push always succeeds because pool size == queue size
// Push always succeeds: pool is sized to queue capacity (SIZE-1), so if
// allocate() returned non-null, the queue cannot be full.
this->usb_data_queue_.push(chunk);
// Re-enable component loop to process the queued data
+6 -2
View File
@@ -158,7 +158,10 @@ class USBUartChannel : public uart::UARTComponent, public Parented<USBUartCompon
// Larger structures first (8+ bytes)
RingBuffer input_buffer_;
LockFreeQueue<UsbOutputChunk, USB_OUTPUT_CHUNK_COUNT> output_queue_;
EventPool<UsbOutputChunk, USB_OUTPUT_CHUNK_COUNT> output_pool_;
// Pool sized to queue capacity (SIZE-1) because LockFreeQueue<T,N> is a ring
// buffer that holds N-1 elements. This guarantees allocate() returns nullptr
// before push() can fail, preventing a pool slot leak.
EventPool<UsbOutputChunk, USB_OUTPUT_CHUNK_COUNT - 1> output_pool_;
std::function<void()> rx_callback_{};
CdcEps cdc_dev_{};
StringRef debug_prefix_{};
@@ -190,7 +193,8 @@ class USBUartComponent : public usb_host::USBClient {
// Lock-free data transfer from USB task to main loop
static constexpr int USB_DATA_QUEUE_SIZE = 32;
LockFreeQueue<UsbDataChunk, USB_DATA_QUEUE_SIZE> usb_data_queue_;
EventPool<UsbDataChunk, USB_DATA_QUEUE_SIZE> chunk_pool_;
// Pool sized to queue capacity (SIZE-1) — see USBUartChannel::output_pool_ comment.
EventPool<UsbDataChunk, USB_DATA_QUEUE_SIZE - 1> chunk_pool_;
protected:
std::vector<USBUartChannel *> channels_{};
-6
View File
@@ -615,10 +615,6 @@ class EsphomeCore:
self.address_cache: AddressCache | None = None
# Cached config hash (computed lazily)
self._config_hash: int | None = None
# True if compiling for C++ unit tests
self.cpp_testing = False
# Allowlist of components whose to_code should run during C++ testing
self.cpp_testing_codegen: set[str] = set()
def reset(self):
from esphome.pins import PIN_SCHEMA_REGISTRY
@@ -648,8 +644,6 @@ class EsphomeCore:
self.current_component = None
self.address_cache = None
self._config_hash = None
self.cpp_testing = False
self.cpp_testing_codegen = set()
PIN_SCHEMA_REGISTRY.reset()
@contextmanager
+10 -5
View File
@@ -71,11 +71,6 @@ class ComponentManifest:
@property
def to_code(self) -> Callable[[Any], None] | None:
if CORE.cpp_testing:
# During C++ testing, only run to_code for allowlisted components
name = self.module.__package__.rsplit(".", 1)[-1]
if name not in CORE.cpp_testing_codegen:
return None
return getattr(self.module, "to_code", None)
@property
@@ -243,3 +238,13 @@ def get_platform(domain: str, platform: str) -> ComponentManifest | None:
_COMPONENT_CACHE: dict[str, ComponentManifest] = {}
CORE_COMPONENTS_PATH = (Path(__file__).parent / "components").resolve()
_COMPONENT_CACHE["esphome"] = ComponentManifest(esphome.core.config)
def _replace_component_manifest(domain: str, manifest: ComponentManifest) -> None:
"""Replace the cached manifest for a component.
This is an intentionally-supported hook for the C++ test infrastructure
to install ``ComponentManifestOverride`` wrappers. Normal application
code should never call this.
"""
_COMPONENT_CACHE[domain] = manifest
@@ -2,21 +2,29 @@
from __future__ import annotations
from collections.abc import Callable
import hashlib
import importlib.util
import os
from pathlib import Path
import subprocess
import sys
from helpers import get_all_dependencies
from helpers import get_all_dependencies, root_path as _root_path
import yaml
# Ensure the repo root is on sys.path so that ``tests.testing_helpers`` and
# override ``__init__.py`` modules can ``from tests.testing_helpers import ...``.
if _root_path not in sys.path:
sys.path.insert(0, _root_path)
from esphome.__main__ import command_compile, parse_args
from esphome.config import validate_config
from esphome.const import CONF_PLATFORM
from esphome.core import CORE
from esphome.loader import get_component
from esphome.loader import get_component, get_platform
from esphome.platformio_api import get_idedata
from tests.testing_helpers import ComponentManifestOverride, set_testing_manifest
# This must coincide with the version in /platformio.ini
PLATFORMIO_GOOGLE_TEST_LIB = "google/googletest@^1.15.2"
@@ -32,19 +40,6 @@ ESPHOME_KEY = "esphome"
HOST_KEY = "host"
LOGGER_KEY = "logger"
# Base config keys that are always present and must not be fully overridden
# by component benchmark.yaml files. esphome: allows sub-key merging.
BASE_CONFIG_KEYS = frozenset({ESPHOME_KEY, HOST_KEY, LOGGER_KEY})
# Shared build flag — enables timezone code paths for testing/benchmarking.
USE_TIME_TIMEZONE_FLAG = "-DUSE_TIME_TIMEZONE"
# Components whose to_code should always run during C++ test/benchmark builds.
# These are the minimal infrastructure components needed for host compilation.
# Note: "core" is the esphome core config module (esphome/core/config.py),
# which registers under package name "core" not "esphome".
BASE_CODEGEN_COMPONENTS = {"core", "host", "logger"}
# Exit codes
EXIT_OK = 0
EXIT_SKIPPED = 1
@@ -110,6 +105,8 @@ def get_platform_components(components: list[str], tests_dir: Path) -> list[str]
if not domain_dir.is_dir():
continue
domain = domain_dir.name
if domain.startswith("__"):
continue
domain_module = get_component(domain)
if domain_module is None or not domain_module.is_platform_component:
raise ValueError(
@@ -130,7 +127,8 @@ def load_component_yaml_configs(components: list[str], tests_dir: Path) -> dict:
The ``esphome:`` key is special: its sub-keys are merged into the
existing esphome config (e.g. to add ``areas:`` or ``devices:``).
Other base config keys (``host:``, ``logger:``) are not overridable.
Keys already present in the base config (e.g. ``host:``, ``logger:``)
are protected by ``setdefault`` in the caller.
Args:
components: List of component directory names
@@ -139,11 +137,6 @@ def load_component_yaml_configs(components: list[str], tests_dir: Path) -> dict:
Returns:
Merged dict of component configs to add to the base config
"""
# Note: components are processed in sorted order. For conflicting keys
# (e.g. two benchmark.yaml files both declaring sensor:), the first
# component alphabetically wins via setdefault(). This is fine for now
# with a single benchmark component (api) but would need a real merge
# strategy if multiple components declare overlapping configs.
merged: dict = {}
for component in components:
yaml_path = tests_dir / component / BENCHMARK_YAML_FILENAME
@@ -153,9 +146,6 @@ def load_component_yaml_configs(components: list[str], tests_dir: Path) -> dict:
component_config = yaml.safe_load(f)
if component_config and isinstance(component_config, dict):
for key, value in component_config.items():
if key in BASE_CONFIG_KEYS - {ESPHOME_KEY}:
# host: and logger: are not overridable
continue
if key == ESPHOME_KEY and isinstance(value, dict):
# Merge esphome sub-keys rather than replacing
esphome_extra = merged.setdefault(ESPHOME_KEY, {})
@@ -198,11 +188,78 @@ def create_host_config(
}
def _wrap_manifest(
comp_name: str,
) -> ComponentManifestOverride | None:
"""Wrap a component manifest in a ComponentManifestOverride with to_code suppressed.
If the manifest is already wrapped or not found, returns None.
Otherwise returns the newly created override after installing it.
"""
if "." in comp_name:
domain, component = comp_name.split(".", maxsplit=1)
manifest = get_platform(domain, component)
cache_key = f"{component}.{domain}"
else:
manifest = get_component(comp_name)
cache_key = comp_name
if manifest is None or isinstance(manifest, ComponentManifestOverride):
return None
override = ComponentManifestOverride(manifest)
override.to_code = None # suppress by default
set_testing_manifest(cache_key, override)
return override
def load_test_manifest_overrides(
components: list[str],
tests_dir: Path,
) -> None:
"""Apply per-component manifest overrides from test ``__init__.py`` files.
For every component, wraps its manifest and suppresses ``to_code``.
If the component's test directory contains an ``__init__.py`` that
defines ``override_manifest(manifest)``, it is called to customise
the override (e.g. ``manifest.enable_codegen()``).
"""
for comp_name in components:
override = _wrap_manifest(comp_name)
if override is None:
continue
if "." in comp_name:
domain, component = comp_name.split(".", maxsplit=1)
cache_key = f"{component}.{domain}"
test_init = tests_dir / component / domain / "__init__.py"
else:
cache_key = comp_name
test_init = tests_dir / comp_name / "__init__.py"
if not test_init.is_file():
continue
spec = importlib.util.spec_from_file_location(
f"_test_manifest_override.{cache_key}", test_init
)
if spec is None or spec.loader is None:
continue
mod = importlib.util.module_from_spec(spec)
spec.loader.exec_module(mod)
override_fn = getattr(mod, "override_manifest", None)
if override_fn is not None:
override_fn(override)
# Type alias for manifest override loaders
ManifestOverrideLoader = Callable[[list[str]], None]
def compile_and_get_binary(
config: dict,
components: list[str],
codegen_components: set[str],
tests_dir: Path,
manifest_override_loader: ManifestOverrideLoader,
label: str = "build",
) -> tuple[int, str | None]:
"""Compile an ESPHome configuration and return the binary path.
@@ -210,8 +267,8 @@ def compile_and_get_binary(
Args:
config: ESPHome configuration dict (already created via create_host_config)
components: List of components to include in the build
codegen_components: Set of component names whose to_code should run
tests_dir: Base directory for test files (used as config_path base)
manifest_override_loader: Callback to apply manifest overrides for components
label: Label for log messages (e.g. "unit tests", "benchmarks")
Returns:
@@ -238,18 +295,21 @@ def compile_and_get_binary(
else:
config.setdefault(key, value)
# Apply manifest overrides before dependency resolution so that any
# dependency additions made by override_manifest() are picked up.
manifest_override_loader(components)
# Obtain possible dependencies BEFORE validate_config, because
# get_all_dependencies calls CORE.reset() which clears build_path.
# Always include 'time' because USE_TIME_TIMEZONE is defined as a build flag,
# which causes core/time.h to include components/time/posix_tz.h.
components_with_dependencies: list[str] = sorted(
get_all_dependencies(set(components) | {"time"}, cpp_testing=True)
get_all_dependencies(set(components))
)
# Apply overrides for any transitively discovered dependencies.
manifest_override_loader(components_with_dependencies)
CORE.config_path = tests_dir / "dummy.yaml"
CORE.dashboard = None
CORE.cpp_testing = True
CORE.cpp_testing_codegen = codegen_components
# Validate config will expand the above with defaults:
config = validate_config(config, {})
@@ -305,7 +365,7 @@ def compile_and_get_binary(
def build_and_run(
selected_components: list[str],
tests_dir: Path,
codegen_components: set[str],
manifest_override_loader: ManifestOverrideLoader,
config_prefix: str,
friendly_name: str,
libraries: str | list[str],
@@ -324,7 +384,7 @@ def build_and_run(
Args:
selected_components: Components to include (directory names in tests_dir)
tests_dir: Directory containing test/benchmark files
codegen_components: Components whose to_code should run
manifest_override_loader: Callback to apply manifest overrides for components
config_prefix: Prefix for the config name (e.g. "cpptests", "cppbench")
friendly_name: Human-readable name for the config
libraries: PlatformIO library specification(s)
@@ -382,7 +442,7 @@ def build_and_run(
)
exit_code, program_path = compile_and_get_binary(
config, components, codegen_components, tests_dir, label
config, components, tests_dir, manifest_override_loader, label
)
if exit_code != EXIT_OK or program_path is None:
+7 -11
View File
@@ -2,18 +2,18 @@
"""Build and run C++ benchmarks for ESPHome components using Google Benchmark."""
import argparse
from functools import partial
import json
import os
from pathlib import Path
import sys
from helpers import root_path
from test_helpers import (
BASE_CODEGEN_COMPONENTS,
from build_helpers import (
PLATFORMIO_GOOGLE_BENCHMARK_LIB,
USE_TIME_TIMEZONE_FLAG,
build_and_run,
load_test_manifest_overrides,
)
from helpers import root_path
# Path to /tests/benchmarks/components
BENCHMARKS_DIR: Path = Path(root_path) / "tests" / "benchmarks" / "components"
@@ -21,11 +21,6 @@ BENCHMARKS_DIR: Path = Path(root_path) / "tests" / "benchmarks" / "components"
# Path to /tests/benchmarks/core (always included, not a component)
CORE_BENCHMARKS_DIR: Path = Path(root_path) / "tests" / "benchmarks" / "core"
# Additional codegen components beyond the base set.
# json is needed because its to_code adds the ArduinoJson library
# (auto-loaded by api, but cpp_testing suppresses to_code unless listed).
BENCHMARK_CODEGEN_COMPONENTS = BASE_CODEGEN_COMPONENTS | {"json"}
PLATFORMIO_OPTIONS = {
"build_unflags": [
"-Os", # remove default size-opt
@@ -33,7 +28,6 @@ PLATFORMIO_OPTIONS = {
"build_flags": [
"-O2", # optimize for speed (CodSpeed recommends RelWithDebInfo)
"-g", # debug symbols for profiling
USE_TIME_TIMEZONE_FLAG,
"-DUSE_BENCHMARK", # disable WarnIfComponentBlockingGuard in finish()
],
# Use deep+ LDF mode to ensure PlatformIO detects the benchmark
@@ -73,7 +67,9 @@ def run_benchmarks(selected_components: list[str], build_only: bool = False) ->
return build_and_run(
selected_components=selected_components,
tests_dir=BENCHMARKS_DIR,
codegen_components=BENCHMARK_CODEGEN_COMPONENTS,
manifest_override_loader=partial(
load_test_manifest_overrides, tests_dir=BENCHMARKS_DIR
),
config_prefix="cppbench",
friendly_name="CPP Benchmarks",
libraries=benchmark_lib,
+7 -6
View File
@@ -1,15 +1,15 @@
#!/usr/bin/env python3
import argparse
from functools import partial
from pathlib import Path
import sys
from helpers import get_all_components, root_path
from test_helpers import (
BASE_CODEGEN_COMPONENTS,
from build_helpers import (
PLATFORMIO_GOOGLE_TEST_LIB,
USE_TIME_TIMEZONE_FLAG,
build_and_run,
load_test_manifest_overrides,
)
from helpers import get_all_components, root_path
# Path to /tests/components
COMPONENTS_TESTS_DIR: Path = Path(root_path) / "tests" / "components"
@@ -21,7 +21,6 @@ PLATFORMIO_OPTIONS = {
],
"build_flags": [
"-Og", # optimize for debug
USE_TIME_TIMEZONE_FLAG,
"-DESPHOME_DEBUG", # enable debug assertions
# Enable the address and undefined behavior sanitizers
"-fsanitize=address",
@@ -39,7 +38,9 @@ def run_tests(selected_components: list[str]) -> int:
return build_and_run(
selected_components=selected_components,
tests_dir=COMPONENTS_TESTS_DIR,
codegen_components=BASE_CODEGEN_COMPONENTS,
manifest_override_loader=partial(
load_test_manifest_overrides, tests_dir=COMPONENTS_TESTS_DIR
),
config_prefix="cpptests",
friendly_name="CPP Unit Tests",
libraries=PLATFORMIO_GOOGLE_TEST_LIB,
+2 -2
View File
@@ -388,7 +388,7 @@ BENCHMARKS_COMPONENTS_PATH = "tests/benchmarks/components"
BENCHMARK_INFRASTRUCTURE_FILES = frozenset(
{
"script/cpp_benchmark.py",
"script/test_helpers.py",
"script/build_helpers.py",
"script/setup_codspeed_lib.py",
}
)
@@ -402,7 +402,7 @@ def should_run_benchmarks(branch: str | None = None) -> bool:
1. Core C++ files changed (esphome/core/*)
2. A directly changed component has benchmark files (no dependency expansion)
3. Benchmark infrastructure changed (tests/benchmarks/*, script/cpp_benchmark.py,
script/test_helpers.py, script/setup_codspeed_lib.py)
script/build_helpers.py, script/setup_codspeed_lib.py)
Unlike unit tests, benchmarks do NOT expand to dependent components.
Changing ``sensor`` does not trigger ``api`` benchmarks just because
+1 -4
View File
@@ -627,14 +627,12 @@ def get_usable_cpu_count() -> int:
def get_all_dependencies(
component_names: set[str], cpp_testing: bool = False
component_names: set[str],
) -> set[str]:
"""Get all dependencies for a set of components.
Args:
component_names: Set of component names to get dependencies for
cpp_testing: If True, set CORE.cpp_testing so AUTO_LOAD callables that
conditionally include testing-only dependencies work correctly
Returns:
Set of all components including dependencies and auto-loaded components
@@ -652,7 +650,6 @@ def get_all_dependencies(
# Reset CORE to ensure clean state
CORE.reset()
CORE.cpp_testing = cpp_testing
# Set up fake config path for component loading
root = Path(__file__).parent.parent
@@ -0,0 +1,7 @@
from tests.testing_helpers import ComponentManifestOverride
def override_manifest(manifest: ComponentManifestOverride) -> None:
# core (esphome/core/config.py) must run its to_code during builds
# because it bootstraps the fundamental application infrastructure.
manifest.enable_codegen()
@@ -0,0 +1,7 @@
from tests.testing_helpers import ComponentManifestOverride
def override_manifest(manifest: ComponentManifestOverride) -> None:
# host must run its to_code during builds because it sets up
# the host platform target execution environment.
manifest.enable_codegen()
@@ -0,0 +1,7 @@
from tests.testing_helpers import ComponentManifestOverride
def override_manifest(manifest: ComponentManifestOverride) -> None:
# json must run its to_code during benchmark builds because it
# adds the ArduinoJson library dependency needed by the API component.
manifest.enable_codegen()
@@ -0,0 +1,7 @@
from tests.testing_helpers import ComponentManifestOverride
def override_manifest(manifest: ComponentManifestOverride) -> None:
# logger must run its to_code during builds because it configures
# the logging subsystem used by ESP_LOG* macros.
manifest.enable_codegen()
@@ -0,0 +1,9 @@
import esphome.codegen as cg
from tests.testing_helpers import ComponentManifestOverride
def override_manifest(manifest: ComponentManifestOverride) -> None:
async def to_code(config):
cg.add_build_flag("-DUSE_TIME_TIMEZONE")
manifest.to_code = to_code
+56 -4
View File
@@ -14,11 +14,63 @@ include the relevant `.cpp` and `.h` test files there.
### Override component code generation for testing
When generating code for testing, ESPHome won't invoke the component's `to_code` function, since most components do not
need to generate configuration code for testing.
During C++ test builds, `to_code` is suppressed for every component by default — most components do not
need to generate configuration code for a unit test binary.
If you do need to generate code to for example configure compilation flags or add libraries,
add the component name to the `CPP_TESTING_CODEGEN_COMPONENTS` allowlist in `script/cpp_unit_test.py`.
#### Manifest overrides
If your component needs to customise code generation behavior for testing — for example to re-enable
`to_code`, supply a lightweight stub, add a test-only dependency, or change any other manifest attribute —
create an `__init__.py` in your component's test directory and define `override_manifest`:
**Top-level component** (`tests/components/<component>/__init__.py`):
```python
from tests.testing_helpers import ComponentManifestOverride
def override_manifest(manifest: ComponentManifestOverride) -> None:
# Re-enable the component's own to_code (needed when the component must
# emit C++ setup code that the test binary depends on at link time).
manifest.enable_codegen()
```
Or supply a lightweight stub instead of the real `to_code`:
```python
from tests.testing_helpers import ComponentManifestOverride
def override_manifest(manifest: ComponentManifestOverride) -> None:
async def to_code_testing(config):
# Only emit what the C++ tests actually need
pass
manifest.to_code = to_code_testing
manifest.dependencies = manifest.dependencies + ["some_test_only_dep"]
```
**Platform component** (`tests/components/<component>/<domain>/__init__.py`,
e.g. `tests/components/my_sensor/sensor/__init__.py`):
```python
from tests.testing_helpers import ComponentManifestOverride
def override_manifest(manifest: ComponentManifestOverride) -> None:
manifest.enable_codegen()
```
`override_manifest` receives a `ComponentManifestOverride` that wraps the real manifest.
Attribute assignments store an override; reads fall back to the real manifest when no
override is present.
Key methods:
| Method | Effect |
|---|---|
| `manifest.enable_codegen()` | Remove the `to_code` suppression, re-enabling code generation |
| `manifest.restore()` | Clear **all** overrides, reverting every attribute to the original |
The function is called after `to_code` has already been set to `None`, so calling
`enable_codegen()` is a deliberate opt-in.
## Running component unit tests
+8
View File
@@ -0,0 +1,8 @@
from tests.testing_helpers import ComponentManifestOverride
def override_manifest(manifest: ComponentManifestOverride) -> None:
# core (esphome/core/config.py) must run its to_code during C++ test builds
# because it bootstraps the fundamental application infrastructure that all
# components depend on (component registration, event loop, etc.).
manifest.enable_codegen()
+7
View File
@@ -0,0 +1,7 @@
from tests.testing_helpers import ComponentManifestOverride
def override_manifest(manifest: ComponentManifestOverride) -> None:
# host must run its to_code during C++ test builds because it sets up the
# host platform target, which is the execution environment for all unit tests.
manifest.enable_codegen()
+7
View File
@@ -0,0 +1,7 @@
from tests.testing_helpers import ComponentManifestOverride
def override_manifest(manifest: ComponentManifestOverride) -> None:
# logger must run its to_code during C++ test builds because it configures
# the logging subsystem used by ESP_LOG* macros throughout component code.
manifest.enable_codegen()
+9
View File
@@ -0,0 +1,9 @@
import esphome.codegen as cg
from tests.testing_helpers import ComponentManifestOverride
def override_manifest(manifest: ComponentManifestOverride) -> None:
async def to_code(config):
cg.add_build_flag("-DUSE_TIME_TIMEZONE")
manifest.to_code = to_code
+148
View File
@@ -1821,3 +1821,151 @@ def test_component_batching_beta_branch_40_per_batch(
all_components.extend(batch_str.split())
assert len(all_components) == 120
assert set(all_components) == set(component_names)
# --- should_run_benchmarks tests ---
def test_should_run_benchmarks_core_change() -> None:
"""Test benchmarks trigger on core C++ file changes."""
with patch.object(
determine_jobs, "changed_files", return_value=["esphome/core/scheduler.cpp"]
):
assert determine_jobs.should_run_benchmarks() is True
def test_should_run_benchmarks_core_header_change() -> None:
"""Test benchmarks trigger on core header changes."""
with patch.object(
determine_jobs, "changed_files", return_value=["esphome/core/helpers.h"]
):
assert determine_jobs.should_run_benchmarks() is True
def test_should_run_benchmarks_benchmark_infra_change() -> None:
"""Test benchmarks trigger on benchmark infrastructure changes."""
for infra_file in [
"script/cpp_benchmark.py",
"script/build_helpers.py",
"script/setup_codspeed_lib.py",
]:
with patch.object(determine_jobs, "changed_files", return_value=[infra_file]):
assert determine_jobs.should_run_benchmarks() is True, (
f"Expected benchmarks to run for {infra_file}"
)
def test_should_run_benchmarks_benchmark_file_change() -> None:
"""Test benchmarks trigger on benchmark file changes."""
with patch.object(
determine_jobs,
"changed_files",
return_value=["tests/benchmarks/components/api/bench_proto_encode.cpp"],
):
assert determine_jobs.should_run_benchmarks() is True
def test_should_run_benchmarks_core_benchmark_file_change() -> None:
"""Test benchmarks trigger on core benchmark file changes."""
with patch.object(
determine_jobs,
"changed_files",
return_value=["tests/benchmarks/core/bench_scheduler.cpp"],
):
assert determine_jobs.should_run_benchmarks() is True
def test_should_run_benchmarks_benchmarked_component_change(tmp_path: Path) -> None:
"""Test benchmarks trigger when a benchmarked component changes."""
# Create a fake benchmarks directory with an 'api' component
benchmarks_dir = tmp_path / "tests" / "benchmarks" / "components" / "api"
benchmarks_dir.mkdir(parents=True)
(benchmarks_dir / "bench_proto_encode.cpp").write_text("// benchmark")
with (
patch.object(
determine_jobs,
"changed_files",
return_value=["esphome/components/api/proto.h"],
),
patch.object(determine_jobs, "root_path", str(tmp_path)),
patch.object(
determine_jobs,
"BENCHMARKS_COMPONENTS_PATH",
"tests/benchmarks/components",
),
):
assert determine_jobs.should_run_benchmarks() is True
def test_should_run_benchmarks_non_benchmarked_component_change(
tmp_path: Path,
) -> None:
"""Test benchmarks do NOT trigger for non-benchmarked component changes."""
# Create a fake benchmarks directory with only 'api'
benchmarks_dir = tmp_path / "tests" / "benchmarks" / "components" / "api"
benchmarks_dir.mkdir(parents=True)
(benchmarks_dir / "bench_proto_encode.cpp").write_text("// benchmark")
with (
patch.object(
determine_jobs,
"changed_files",
return_value=["esphome/components/sensor/__init__.py"],
),
patch.object(determine_jobs, "root_path", str(tmp_path)),
patch.object(
determine_jobs,
"BENCHMARKS_COMPONENTS_PATH",
"tests/benchmarks/components",
),
):
assert determine_jobs.should_run_benchmarks() is False
def test_should_run_benchmarks_no_dependency_expansion(tmp_path: Path) -> None:
"""Test benchmarks do NOT expand to dependent components.
Changing 'sensor' should not trigger 'api' benchmarks even if api
depends on sensor. This is intentional benchmark runs should be
targeted to directly changed components only.
"""
benchmarks_dir = tmp_path / "tests" / "benchmarks" / "components" / "api"
benchmarks_dir.mkdir(parents=True)
(benchmarks_dir / "bench_proto_encode.cpp").write_text("// benchmark")
with (
patch.object(
determine_jobs,
"changed_files",
# sensor is a dependency of api, but benchmarks don't expand
return_value=["esphome/components/sensor/sensor.cpp"],
),
patch.object(determine_jobs, "root_path", str(tmp_path)),
patch.object(
determine_jobs,
"BENCHMARKS_COMPONENTS_PATH",
"tests/benchmarks/components",
),
):
assert determine_jobs.should_run_benchmarks() is False
def test_should_run_benchmarks_unrelated_change() -> None:
"""Test benchmarks do NOT trigger for unrelated changes."""
with patch.object(determine_jobs, "changed_files", return_value=["README.md"]):
assert determine_jobs.should_run_benchmarks() is False
def test_should_run_benchmarks_no_changes() -> None:
"""Test benchmarks do NOT trigger with no changes."""
with patch.object(determine_jobs, "changed_files", return_value=[]):
assert determine_jobs.should_run_benchmarks() is False
def test_should_run_benchmarks_with_branch() -> None:
"""Test should_run_benchmarks passes branch to changed_files."""
with patch.object(determine_jobs, "changed_files") as mock_changed:
mock_changed.return_value = []
determine_jobs.should_run_benchmarks("release")
mock_changed.assert_called_with("release")
-22
View File
@@ -1073,28 +1073,6 @@ def test_get_all_dependencies_platform_component_with_dependencies() -> None:
assert result == {"sensor.bthome", "sensor"}
def test_get_all_dependencies_cpp_testing_flag() -> None:
"""cpp_testing=True propagates to CORE.cpp_testing during resolution."""
from esphome.core import CORE
with (
patch("esphome.loader.get_component") as mock_get_component,
patch("esphome.loader.get_platform"),
):
observed: list[bool] = []
def capturing_get_component(name: str):
observed.append(CORE.cpp_testing)
mock_get_component.side_effect = capturing_get_component
helpers.get_all_dependencies({"some_comp"}, cpp_testing=True)
assert observed and all(observed), (
"CORE.cpp_testing should be True during resolution"
)
def test_get_components_from_integration_fixtures() -> None:
"""Test extraction of components from fixture YAML files."""
yaml_content = {
+260
View File
@@ -0,0 +1,260 @@
"""Unit tests for script/build_helpers.py manifest override and build helpers."""
import os
from pathlib import Path
import sys
import textwrap
from unittest.mock import MagicMock, patch
import pytest
# Add the script directory to Python path so we can import build_helpers
sys.path.insert(
0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", "script"))
)
import build_helpers # noqa: E402
from esphome.loader import ComponentManifest # noqa: E402
from tests.testing_helpers import ComponentManifestOverride # noqa: E402
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _make_component_manifest(*, to_code=None, dependencies=None) -> ComponentManifest:
mod = MagicMock()
mod.to_code = to_code
mod.DEPENDENCIES = dependencies or []
return ComponentManifest(mod)
# ---------------------------------------------------------------------------
# filter_components_with_files
# ---------------------------------------------------------------------------
def test_filter_keeps_components_with_cpp_files(tmp_path: Path) -> None:
comp_dir = tmp_path / "mycomp"
comp_dir.mkdir()
(comp_dir / "mycomp_test.cpp").write_text("")
result = build_helpers.filter_components_with_files(["mycomp"], tmp_path)
assert result == ["mycomp"]
def test_filter_keeps_components_with_h_files(tmp_path: Path) -> None:
comp_dir = tmp_path / "mycomp"
comp_dir.mkdir()
(comp_dir / "helpers.h").write_text("")
result = build_helpers.filter_components_with_files(["mycomp"], tmp_path)
assert result == ["mycomp"]
def test_filter_drops_components_without_test_dir(tmp_path: Path) -> None:
result = build_helpers.filter_components_with_files(["nodir"], tmp_path)
assert result == []
def test_filter_drops_components_with_no_cpp_or_h(tmp_path: Path) -> None:
comp_dir = tmp_path / "mycomp"
comp_dir.mkdir()
(comp_dir / "README.md").write_text("")
result = build_helpers.filter_components_with_files(["mycomp"], tmp_path)
assert result == []
# ---------------------------------------------------------------------------
# get_platform_components
# ---------------------------------------------------------------------------
def test_get_platform_components_discovers_subdirectory(tmp_path: Path) -> None:
(tmp_path / "bthome" / "sensor").mkdir(parents=True)
sensor_mod = MagicMock()
sensor_mod.IS_PLATFORM_COMPONENT = True
with patch(
"build_helpers.get_component", return_value=ComponentManifest(sensor_mod)
):
result = build_helpers.get_platform_components(["bthome"], tmp_path)
assert result == ["sensor.bthome"]
def test_get_platform_components_skips_pycache(tmp_path: Path) -> None:
(tmp_path / "bthome" / "__pycache__").mkdir(parents=True)
result = build_helpers.get_platform_components(["bthome"], tmp_path)
assert result == []
def test_get_platform_components_raises_for_invalid_domain(tmp_path: Path) -> None:
(tmp_path / "bthome" / "notadomain").mkdir(parents=True)
with (
patch("build_helpers.get_component", return_value=None),
pytest.raises(ValueError, match="notadomain"),
):
build_helpers.get_platform_components(["bthome"], tmp_path)
# ---------------------------------------------------------------------------
# load_test_manifest_overrides
# ---------------------------------------------------------------------------
def test_load_suppresses_to_code(tmp_path: Path) -> None:
"""to_code is always set to None before the override is called."""
async def real_to_code(config):
pass
inner = _make_component_manifest(to_code=real_to_code)
with (
patch("build_helpers.get_component", return_value=inner),
patch("build_helpers.set_testing_manifest") as mock_set,
):
build_helpers.load_test_manifest_overrides(["mycomp"], tmp_path)
installed: ComponentManifestOverride = mock_set.call_args[0][1]
assert installed.to_code is None
def test_load_calls_override_fn(tmp_path: Path) -> None:
"""override_manifest() in test_init is called with the ComponentManifestOverride."""
comp_dir = tmp_path / "mycomp"
comp_dir.mkdir()
init_py = comp_dir / "__init__.py"
init_py.write_text(
textwrap.dedent("""\
def override_manifest(manifest):
manifest.dependencies = ["injected"]
""")
)
inner = _make_component_manifest()
override = ComponentManifestOverride(inner)
override.to_code = None
with (
patch("build_helpers.get_component", return_value=inner),
patch("build_helpers.set_testing_manifest") as mock_set,
):
build_helpers.load_test_manifest_overrides(["mycomp"], tmp_path)
installed: ComponentManifestOverride = mock_set.call_args[0][1]
assert installed.dependencies == ["injected"]
def test_load_enable_codegen_in_override(tmp_path: Path) -> None:
"""An override_manifest that calls enable_codegen() restores to_code."""
async def real_to_code(config):
pass
comp_dir = tmp_path / "mycomp"
comp_dir.mkdir()
init_py = comp_dir / "__init__.py"
init_py.write_text(
textwrap.dedent("""\
def override_manifest(manifest):
manifest.enable_codegen()
""")
)
inner = _make_component_manifest(to_code=real_to_code)
with (
patch("build_helpers.get_component", return_value=inner),
patch("build_helpers.set_testing_manifest") as mock_set,
):
build_helpers.load_test_manifest_overrides(["mycomp"], tmp_path)
installed: ComponentManifestOverride = mock_set.call_args[0][1]
assert installed.to_code is real_to_code
def test_load_no_override_file(tmp_path: Path) -> None:
"""No override file: manifest is wrapped and to_code suppressed, nothing else."""
inner = _make_component_manifest()
with (
patch("build_helpers.get_component", return_value=inner),
patch("build_helpers.set_testing_manifest") as mock_set,
):
build_helpers.load_test_manifest_overrides(["mycomp"], tmp_path)
mock_set.assert_called_once()
key, installed = mock_set.call_args[0]
assert key == "mycomp"
assert isinstance(installed, ComponentManifestOverride)
def test_load_skips_already_wrapped(tmp_path: Path) -> None:
"""Components already wrapped as ComponentManifestOverride are not double-wrapped."""
inner = _make_component_manifest()
already_wrapped = ComponentManifestOverride(inner)
with (
patch("build_helpers.get_component", return_value=already_wrapped),
patch("build_helpers.set_testing_manifest") as mock_set,
):
build_helpers.load_test_manifest_overrides(["mycomp"], tmp_path)
mock_set.assert_not_called()
def test_load_skips_platform_component_already_wrapped(tmp_path: Path) -> None:
inner = _make_component_manifest()
already_wrapped = ComponentManifestOverride(inner)
with (
patch("build_helpers.get_platform", return_value=already_wrapped),
patch("build_helpers.set_testing_manifest") as mock_set,
):
build_helpers.load_test_manifest_overrides(["sensor.bthome"], tmp_path)
mock_set.assert_not_called()
def test_load_wraps_top_level_component(tmp_path: Path) -> None:
inner = _make_component_manifest()
with (
patch("build_helpers.get_component", return_value=inner),
patch("build_helpers.set_testing_manifest") as mock_set,
):
build_helpers.load_test_manifest_overrides(["mycomp"], tmp_path)
mock_set.assert_called_once()
key, installed = mock_set.call_args[0]
assert key == "mycomp"
assert isinstance(installed, ComponentManifestOverride)
assert installed.to_code is None
def test_load_wraps_platform_component(tmp_path: Path) -> None:
inner = _make_component_manifest()
with (
patch("build_helpers.get_platform", return_value=inner),
patch("build_helpers.set_testing_manifest") as mock_set,
):
build_helpers.load_test_manifest_overrides(["sensor.bthome"], tmp_path)
mock_set.assert_called_once()
key, installed = mock_set.call_args[0]
assert key == "bthome.sensor"
assert isinstance(installed, ComponentManifestOverride)
assert installed.to_code is None
+63
View File
@@ -0,0 +1,63 @@
from typing import Any
from esphome.loader import ComponentManifest, _replace_component_manifest
class ComponentManifestOverride:
"""Mutable wrapper around ComponentManifest for test-specific attribute overrides.
When ``tests/components/<name>/__init__.py`` defines::
def override_manifest(manifest: ComponentManifestOverride) -> None:
...
the function receives an instance of this class wrapping the real component
manifest. Any attribute assignment stores an override; reads fall back to
the underlying ``ComponentManifest`` when no override has been set.
Example::
def override_manifest(manifest: ComponentManifestOverride) -> None:
async def to_code_testing(config):
pass # lightweight no-op stub for C++ unit tests
manifest.to_code = to_code_testing
manifest.dependencies = manifest.dependencies + ["extra_dep_for_tests"]
"""
def __init__(self, wrapped: "ComponentManifest") -> None:
object.__setattr__(self, "_wrapped", wrapped)
object.__setattr__(self, "_overrides", {})
def __getattr__(self, name: str) -> Any:
overrides: dict[str, Any] = object.__getattribute__(self, "_overrides")
if name in overrides:
return overrides[name]
wrapped: ComponentManifest = object.__getattribute__(self, "_wrapped")
return getattr(wrapped, name)
def __setattr__(self, name: str, value: Any) -> None:
overrides: dict[str, Any] = object.__getattribute__(self, "_overrides")
overrides[name] = value
def enable_codegen(self) -> None:
"""Remove the to_code suppression, re-enabling code generation for this component.
Call this from ``override_manifest`` when the component needs its real (or a
custom stub) ``to_code`` to run during C++ unit test builds.
"""
overrides: dict[str, Any] = object.__getattribute__(self, "_overrides")
overrides.pop("to_code", None)
def restore(self) -> None:
"""Clear all overrides, reverting to the wrapped manifest's values."""
object.__getattribute__(self, "_overrides").clear()
def set_testing_manifest(domain: str, manifest: ComponentManifestOverride) -> None:
"""Install a testing manifest override into the component cache.
Called from the C++ unit test infrastructure when a component's test
directory provides an ``override_manifest`` function.
"""
_replace_component_manifest(domain, manifest)
+98 -1
View File
@@ -2,7 +2,104 @@
from unittest.mock import MagicMock, patch
from esphome.loader import ComponentManifest
from esphome.loader import ComponentManifest, _replace_component_manifest, get_component
from tests.testing_helpers import ComponentManifestOverride
# ---------------------------------------------------------------------------
# ComponentManifestOverride
# ---------------------------------------------------------------------------
def _make_manifest(*, to_code=None, dependencies=None) -> ComponentManifest:
"""Return a ComponentManifest backed by a minimal mock module."""
mod = MagicMock()
mod.to_code = to_code
mod.DEPENDENCIES = dependencies or []
return ComponentManifest(mod)
def test_testing_manifest_delegates_to_wrapped() -> None:
"""Unoverridden attributes fall through to the wrapped manifest."""
inner = _make_manifest(dependencies=["wifi"])
tm = ComponentManifestOverride(inner)
assert tm.dependencies == ["wifi"]
def test_testing_manifest_override_shadows_wrapped() -> None:
"""An assigned attribute shadows the wrapped value."""
inner = _make_manifest(dependencies=["wifi"])
tm = ComponentManifestOverride(inner)
tm.dependencies = ["ble"]
assert tm.dependencies == ["ble"]
# Wrapped value unchanged
assert inner.dependencies == ["wifi"]
def test_testing_manifest_to_code_suppression() -> None:
"""Setting to_code=None suppresses code generation."""
async def real_to_code(config):
pass
inner = _make_manifest(to_code=real_to_code)
tm = ComponentManifestOverride(inner)
tm.to_code = None
assert tm.to_code is None
def test_testing_manifest_enable_codegen_removes_suppression() -> None:
"""enable_codegen() removes the to_code override, restoring the original."""
async def real_to_code(config):
pass
inner = _make_manifest(to_code=real_to_code)
tm = ComponentManifestOverride(inner)
tm.to_code = None
assert tm.to_code is None
tm.enable_codegen()
assert tm.to_code is real_to_code
def test_testing_manifest_enable_codegen_preserves_other_overrides() -> None:
"""enable_codegen() only removes to_code; other overrides survive."""
inner = _make_manifest(dependencies=["wifi"])
tm = ComponentManifestOverride(inner)
tm.to_code = None
tm.dependencies = ["ble"]
tm.enable_codegen()
assert tm.to_code is inner.to_code
assert tm.dependencies == ["ble"]
def test_testing_manifest_restore_clears_all_overrides() -> None:
"""restore() removes every override, reverting all attributes to wrapped values."""
async def real_to_code(config):
pass
inner = _make_manifest(to_code=real_to_code, dependencies=["wifi"])
tm = ComponentManifestOverride(inner)
tm.to_code = None
tm.dependencies = ["ble"]
tm.restore()
assert tm.to_code is real_to_code
assert tm.dependencies == ["wifi"]
def test_replace_component_manifest_installs_override() -> None:
"""_replace_component_manifest replaces the cached manifest for a domain."""
inner = _make_manifest()
override = ComponentManifestOverride(inner)
_replace_component_manifest("_test_dummy_domain", override)
assert get_component("_test_dummy_domain") is override
def test_component_manifest_resources_with_filter_source_files() -> None:
+8
View File
@@ -167,6 +167,13 @@ def mock_run_miniterm() -> Generator[Mock]:
yield mock
@pytest.fixture
def mock_wait_for_serial_port() -> Generator[Mock]:
"""Mock _wait_for_serial_port for testing."""
with patch("esphome.__main__._wait_for_serial_port") as mock:
yield mock
@pytest.fixture
def mock_upload_using_esptool() -> Generator[Mock]:
"""Mock upload_using_esptool for testing."""
@@ -1706,6 +1713,7 @@ def test_show_logs_serial(
mock_get_port_type: Mock,
mock_check_permissions: Mock,
mock_run_miniterm: Mock,
mock_wait_for_serial_port: Mock,
) -> None:
"""Test show_logs with serial port."""
setup_core(config={"logger": {}}, platform=PLATFORM_ESP32)