Merge branch 'dev' into component-8byte-optimization

This commit is contained in:
J. Nick Koston
2026-03-29 08:25:57 -10:00
committed by GitHub
337 changed files with 7663 additions and 4481 deletions
@@ -172,6 +172,135 @@ BENCHMARK(NoiseDecrypt_MediumMessage);
static void NoiseDecrypt_LargeMessage(benchmark::State &state) { noise_decrypt_bench(state, 1024); }
BENCHMARK(NoiseDecrypt_LargeMessage);
// --- Full Noise_NNpsk0 handshake benchmark ---
// Measures the complete handshake between initiator and responder:
// - Create handshake states for both sides
// - Set PSK and prologue
// - Exchange messages (initiator write -> responder read -> responder write -> initiator read)
// - Split to get cipher states
// This is dominated by Curve25519 DH operations (expensive on ESP8266).
// No inner iterations — each handshake is already expensive enough.
static void NoiseHandshake_Full(benchmark::State &state) {
// Matching ESPHome's protocol: Noise_NNpsk0_25519_ChaChaPoly_SHA256
NoiseProtocolId nid;
memset(&nid, 0, sizeof(nid));
nid.pattern_id = NOISE_PATTERN_NN;
nid.cipher_id = NOISE_CIPHER_CHACHAPOLY;
nid.dh_id = NOISE_DH_CURVE25519;
nid.prefix_id = NOISE_PREFIX_STANDARD;
nid.hybrid_id = NOISE_DH_NONE;
nid.hash_id = NOISE_HASH_SHA256;
nid.modifier_ids[0] = NOISE_MODIFIER_PSK0;
// Dummy PSK (32 bytes) and prologue matching production setup
static constexpr uint8_t PSK[32] = {0xAB, 0xAB, 0xAB, 0xAB, 0xAB, 0xAB, 0xAB, 0xAB, 0xAB, 0xAB, 0xAB,
0xAB, 0xAB, 0xAB, 0xAB, 0xAB, 0xAB, 0xAB, 0xAB, 0xAB, 0xAB, 0xAB,
0xAB, 0xAB, 0xAB, 0xAB, 0xAB, 0xAB, 0xAB, 0xAB, 0xAB, 0xAB};
static constexpr uint8_t PROLOGUE[] = "NoESPHome";
// Message buffer for handshake exchange (max handshake message ~96 bytes)
uint8_t msg_buf[128];
for (auto _ : state) {
NoiseHandshakeState *initiator = nullptr;
NoiseHandshakeState *responder = nullptr;
NoiseCipherState *init_send = nullptr, *init_recv = nullptr;
NoiseCipherState *resp_send = nullptr, *resp_recv = nullptr;
int err;
// Create both handshake states
err = noise_handshakestate_new_by_id(&initiator, &nid, NOISE_ROLE_INITIATOR);
if (err != NOISE_ERROR_NONE) {
state.SkipWithError("Failed to create initiator");
return;
}
err = noise_handshakestate_new_by_id(&responder, &nid, NOISE_ROLE_RESPONDER);
if (err != NOISE_ERROR_NONE) {
state.SkipWithError("Failed to create responder");
noise_handshakestate_free(initiator);
return;
}
// Set PSK and prologue on both sides
noise_handshakestate_set_pre_shared_key(initiator, PSK, sizeof(PSK));
noise_handshakestate_set_pre_shared_key(responder, PSK, sizeof(PSK));
noise_handshakestate_set_prologue(initiator, PROLOGUE, sizeof(PROLOGUE) - 1);
noise_handshakestate_set_prologue(responder, PROLOGUE, sizeof(PROLOGUE) - 1);
noise_handshakestate_start(initiator);
noise_handshakestate_start(responder);
// Message 1: Initiator -> Responder
NoiseBuffer write_buf, read_buf;
noise_buffer_set_output(write_buf, msg_buf, sizeof(msg_buf));
err = noise_handshakestate_write_message(initiator, &write_buf, nullptr);
if (err != NOISE_ERROR_NONE) {
state.SkipWithError("Initiator write_message failed");
noise_handshakestate_free(initiator);
noise_handshakestate_free(responder);
return;
}
noise_buffer_set_input(read_buf, msg_buf, write_buf.size);
err = noise_handshakestate_read_message(responder, &read_buf, nullptr);
if (err != NOISE_ERROR_NONE) {
state.SkipWithError("Responder read_message failed");
noise_handshakestate_free(initiator);
noise_handshakestate_free(responder);
return;
}
// Message 2: Responder -> Initiator
noise_buffer_set_output(write_buf, msg_buf, sizeof(msg_buf));
err = noise_handshakestate_write_message(responder, &write_buf, nullptr);
if (err != NOISE_ERROR_NONE) {
state.SkipWithError("Responder write_message failed");
noise_handshakestate_free(initiator);
noise_handshakestate_free(responder);
return;
}
noise_buffer_set_input(read_buf, msg_buf, write_buf.size);
err = noise_handshakestate_read_message(initiator, &read_buf, nullptr);
if (err != NOISE_ERROR_NONE) {
state.SkipWithError("Initiator read_message failed");
noise_handshakestate_free(initiator);
noise_handshakestate_free(responder);
return;
}
// Split to get cipher states
err = noise_handshakestate_split(initiator, &init_send, &init_recv);
if (err != NOISE_ERROR_NONE) {
state.SkipWithError("Initiator split failed");
noise_handshakestate_free(initiator);
noise_handshakestate_free(responder);
return;
}
err = noise_handshakestate_split(responder, &resp_send, &resp_recv);
if (err != NOISE_ERROR_NONE) {
state.SkipWithError("Responder split failed");
noise_handshakestate_free(initiator);
noise_handshakestate_free(responder);
noise_cipherstate_free(init_send);
noise_cipherstate_free(init_recv);
return;
}
benchmark::DoNotOptimize(init_send);
// Cleanup
noise_handshakestate_free(initiator);
noise_handshakestate_free(responder);
noise_cipherstate_free(init_send);
noise_cipherstate_free(init_recv);
noise_cipherstate_free(resp_send);
noise_cipherstate_free(resp_recv);
}
}
BENCHMARK(NoiseHandshake_Full);
} // namespace esphome::api::benchmarks
#endif // USE_API_NOISE
@@ -0,0 +1,5 @@
from tests.testing_helpers import ComponentManifestOverride
def override_manifest(manifest: ComponentManifestOverride) -> None:
manifest.enable_codegen()
@@ -0,0 +1,142 @@
#include <benchmark/benchmark.h>
#include "esphome/components/climate/climate.h"
namespace esphome::benchmarks {
// Inner iteration count to amortize CodSpeed instrumentation overhead.
static constexpr int kInnerIterations = 2000;
// Minimal Climate for benchmarking — control() is a no-op.
class BenchClimate : public climate::Climate {
public:
void configure(const char *name) { this->configure_entity_(name, 0x12345678, 0); }
climate::ClimateTraits traits() override { return this->traits_; }
climate::ClimateTraits traits_;
protected:
void control(const climate::ClimateCall & /*call*/) override {}
};
// Helper to create a typical HVAC climate device for benchmarks.
// Note: setup() is not called (no preferences backend), so save_state_()
// is effectively a no-op. This benchmarks the call/validation path, not persistence.
static void setup_hvac_climate(BenchClimate &climate) {
climate.configure("test_climate");
climate.traits_.set_supported_modes({
climate::CLIMATE_MODE_OFF,
climate::CLIMATE_MODE_HEAT_COOL,
climate::CLIMATE_MODE_COOL,
climate::CLIMATE_MODE_HEAT,
climate::CLIMATE_MODE_FAN_ONLY,
});
climate.traits_.set_supported_fan_modes({
climate::CLIMATE_FAN_AUTO,
climate::CLIMATE_FAN_LOW,
climate::CLIMATE_FAN_MEDIUM,
climate::CLIMATE_FAN_HIGH,
});
climate.traits_.set_supported_swing_modes({
climate::CLIMATE_SWING_OFF,
climate::CLIMATE_SWING_BOTH,
climate::CLIMATE_SWING_VERTICAL,
climate::CLIMATE_SWING_HORIZONTAL,
});
climate.traits_.set_supported_presets({
climate::CLIMATE_PRESET_NONE,
climate::CLIMATE_PRESET_HOME,
climate::CLIMATE_PRESET_AWAY,
});
climate.traits_.set_visual_min_temperature(16.0f);
climate.traits_.set_visual_max_temperature(30.0f);
climate.traits_.set_visual_target_temperature_step(0.5f);
climate.traits_.set_visual_current_temperature_step(0.1f);
climate.traits_.add_feature_flags(climate::CLIMATE_SUPPORTS_CURRENT_TEMPERATURE | climate::CLIMATE_SUPPORTS_ACTION);
}
// --- Climate::publish_state() with temperature update ---
// Measures the publish path for a thermostat reporting state —
// the hot path during HVAC operation.
static void ClimatePublish_State(benchmark::State &state) {
BenchClimate climate;
setup_hvac_climate(climate);
climate.mode = climate::CLIMATE_MODE_HEAT;
climate.action = climate::CLIMATE_ACTION_HEATING;
climate.target_temperature = 22.0f;
for (auto _ : state) {
for (int i = 0; i < kInnerIterations; i++) {
climate.current_temperature = 20.0f + static_cast<float>(i % 100) / 10.0f;
climate.publish_state();
}
benchmark::DoNotOptimize(climate.current_temperature);
}
state.SetItemsProcessed(state.iterations() * kInnerIterations);
}
BENCHMARK(ClimatePublish_State);
// --- Climate::publish_state() with callback ---
// Measures callback dispatch overhead.
static void ClimatePublish_WithCallback(benchmark::State &state) {
BenchClimate climate;
setup_hvac_climate(climate);
climate.mode = climate::CLIMATE_MODE_HEAT;
climate.target_temperature = 22.0f;
uint64_t callback_count = 0;
climate.add_on_state_callback([&callback_count](climate::Climate & /*c*/) { callback_count++; });
for (auto _ : state) {
for (int i = 0; i < kInnerIterations; i++) {
climate.current_temperature = 20.0f + static_cast<float>(i % 100) / 10.0f;
climate.publish_state();
}
benchmark::DoNotOptimize(callback_count);
}
state.SetItemsProcessed(state.iterations() * kInnerIterations);
}
BENCHMARK(ClimatePublish_WithCallback);
// --- ClimateCall::perform() set target temperature ---
// The most common climate call — adjusting the thermostat setpoint.
static void ClimateCall_SetTemperature(benchmark::State &state) {
BenchClimate climate;
setup_hvac_climate(climate);
climate.mode = climate::CLIMATE_MODE_HEAT;
for (auto _ : state) {
for (int i = 0; i < kInnerIterations; i++) {
float temp = 18.0f + static_cast<float>(i % 25) * 0.5f;
climate.make_call().set_target_temperature(temp).perform();
}
benchmark::DoNotOptimize(climate.target_temperature);
}
state.SetItemsProcessed(state.iterations() * kInnerIterations);
}
BENCHMARK(ClimateCall_SetTemperature);
// --- ClimateCall::perform() mode change with fan ---
// Exercises the validation path with multiple fields set.
static void ClimateCall_ModeChange(benchmark::State &state) {
BenchClimate climate;
setup_hvac_climate(climate);
for (auto _ : state) {
for (int i = 0; i < kInnerIterations; i++) {
auto mode = (i % 2 == 0) ? climate::CLIMATE_MODE_HEAT : climate::CLIMATE_MODE_COOL;
auto fan = (i % 2 == 0) ? climate::CLIMATE_FAN_HIGH : climate::CLIMATE_FAN_LOW;
climate.make_call().set_mode(mode).set_fan_mode(fan).set_target_temperature(22.0f).perform();
}
benchmark::DoNotOptimize(climate.mode);
}
state.SetItemsProcessed(state.iterations() * kInnerIterations);
}
BENCHMARK(ClimateCall_ModeChange);
} // namespace esphome::benchmarks
@@ -0,0 +1 @@
climate:
@@ -0,0 +1,5 @@
from tests.testing_helpers import ComponentManifestOverride
def override_manifest(manifest: ComponentManifestOverride) -> None:
manifest.enable_codegen()
@@ -0,0 +1,107 @@
#include <benchmark/benchmark.h>
#include "esphome/components/cover/cover.h"
namespace esphome::benchmarks {
// Inner iteration count to amortize CodSpeed instrumentation overhead.
static constexpr int kInnerIterations = 2000;
// Minimal Cover for benchmarking — control() is a no-op.
class BenchCover : public cover::Cover {
public:
cover::CoverTraits get_traits() override { return this->traits_; }
void configure(const char *name) { this->configure_entity_(name, 0x12345678, 0); }
cover::CoverTraits traits_;
protected:
void control(const cover::CoverCall & /*call*/) override {}
};
// --- Cover::publish_state() with position updates ---
// Measures the publish path for a garage door reporting position
// during open/close — the hot path during movement.
static void CoverPublish_Position(benchmark::State &state) {
BenchCover cover;
cover.configure("test_cover");
cover.traits_.set_supports_position(true);
cover.traits_.set_supports_tilt(false);
for (auto _ : state) {
for (int i = 0; i < kInnerIterations; i++) {
cover.position = static_cast<float>(i % 101) / 100.0f;
cover.current_operation = (i % 2 == 0) ? cover::COVER_OPERATION_OPENING : cover::COVER_OPERATION_CLOSING;
cover.publish_state(false);
}
benchmark::DoNotOptimize(cover.position);
}
state.SetItemsProcessed(state.iterations() * kInnerIterations);
}
BENCHMARK(CoverPublish_Position);
// --- Cover::publish_state() with callback ---
// Measures callback dispatch overhead.
static void CoverPublish_WithCallback(benchmark::State &state) {
BenchCover cover;
cover.configure("test_cover");
cover.traits_.set_supports_position(true);
uint64_t callback_count = 0;
cover.add_on_state_callback([&callback_count]() { callback_count++; });
for (auto _ : state) {
for (int i = 0; i < kInnerIterations; i++) {
cover.position = static_cast<float>(i % 101) / 100.0f;
cover.publish_state(false);
}
benchmark::DoNotOptimize(callback_count);
}
state.SetItemsProcessed(state.iterations() * kInnerIterations);
}
BENCHMARK(CoverPublish_WithCallback);
// --- CoverCall::perform() open/close cycle ---
// Measures the full call path: validation + control delegation.
static void CoverCall_OpenClose(benchmark::State &state) {
BenchCover cover;
cover.configure("test_cover");
cover.traits_.set_supports_position(true);
for (auto _ : state) {
for (int i = 0; i < kInnerIterations; i++) {
if (i % 2 == 0) {
cover.make_call().set_command_open().perform();
} else {
cover.make_call().set_command_close().perform();
}
}
benchmark::DoNotOptimize(cover.position);
}
state.SetItemsProcessed(state.iterations() * kInnerIterations);
}
BENCHMARK(CoverCall_OpenClose);
// --- CoverCall::perform() set position ---
// Measures the position-setting call path.
static void CoverCall_SetPosition(benchmark::State &state) {
BenchCover cover;
cover.configure("test_cover");
cover.traits_.set_supports_position(true);
for (auto _ : state) {
for (int i = 0; i < kInnerIterations; i++) {
float pos = static_cast<float>(i % 101) / 100.0f;
cover.make_call().set_position(pos).perform();
}
benchmark::DoNotOptimize(cover.position);
}
state.SetItemsProcessed(state.iterations() * kInnerIterations);
}
BENCHMARK(CoverCall_SetPosition);
} // namespace esphome::benchmarks
@@ -0,0 +1 @@
cover:
@@ -0,0 +1,5 @@
from tests.testing_helpers import ComponentManifestOverride
def override_manifest(manifest: ComponentManifestOverride) -> None:
manifest.enable_codegen()
@@ -0,0 +1,122 @@
#include <benchmark/benchmark.h>
#include "esphome/components/fan/fan.h"
namespace esphome::benchmarks {
// Inner iteration count to amortize CodSpeed instrumentation overhead.
static constexpr int kInnerIterations = 2000;
// Minimal Fan for benchmarking — control() is a no-op.
class BenchFan : public fan::Fan {
public:
void configure(const char *name) { this->configure_entity_(name, 0x12345678, 0); }
fan::FanTraits get_traits() override { return this->traits_; }
fan::FanTraits traits_;
protected:
void control(const fan::FanCall & /*call*/) override {}
};
// Helper to create a typical fan device for benchmarks.
// Note: setup() is not called (no preferences backend), so save_state_()
// is effectively a no-op. This benchmarks the call/validation path, not persistence.
static void setup_fan(BenchFan &fan) {
fan.configure("test_fan");
fan.traits_.set_oscillation(true);
fan.traits_.set_speed(true);
fan.traits_.set_supported_speed_count(6);
fan.traits_.set_direction(true);
fan.set_restore_mode(fan::FanRestoreMode::NO_RESTORE);
fan.traits_.set_supported_preset_modes({
"auto",
"sleep",
"nature",
"turbo",
});
}
// --- Fan::publish_state() with speed update ---
// Measures the publish path for a fan reporting state —
// the hot path during fan operation.
static void FanPublish_State(benchmark::State &state) {
BenchFan fan;
setup_fan(fan);
fan.state = true;
fan.direction = fan::FanDirection::FORWARD;
for (auto _ : state) {
for (int i = 0; i < kInnerIterations; i++) {
fan.speed = (i % 6) + 1;
fan.publish_state();
}
benchmark::DoNotOptimize(fan.speed);
}
state.SetItemsProcessed(state.iterations() * kInnerIterations);
}
BENCHMARK(FanPublish_State);
// --- Fan::publish_state() with callback ---
// Measures callback dispatch overhead.
static void FanPublish_WithCallback(benchmark::State &state) {
BenchFan fan;
setup_fan(fan);
fan.state = true;
uint64_t callback_count = 0;
fan.add_on_state_callback([&callback_count]() { callback_count++; });
for (auto _ : state) {
for (int i = 0; i < kInnerIterations; i++) {
fan.speed = (i % 6) + 1;
fan.publish_state();
}
benchmark::DoNotOptimize(callback_count);
}
state.SetItemsProcessed(state.iterations() * kInnerIterations);
}
BENCHMARK(FanPublish_WithCallback);
// --- FanCall::perform() set speed ---
// The most common fan call — adjusting the speed level.
static void FanCall_SetSpeed(benchmark::State &state) {
BenchFan fan;
setup_fan(fan);
fan.state = true;
for (auto _ : state) {
for (int i = 0; i < kInnerIterations; i++) {
int speed = (i % 6) + 1;
fan.make_call().set_speed(speed).perform();
}
benchmark::DoNotOptimize(fan.speed);
}
state.SetItemsProcessed(state.iterations() * kInnerIterations);
}
BENCHMARK(FanCall_SetSpeed);
// --- FanCall::perform() with multiple fields ---
// Exercises the validation path with state, speed, oscillation, and direction.
static void FanCall_MultiField(benchmark::State &state) {
BenchFan fan;
setup_fan(fan);
for (auto _ : state) {
for (int i = 0; i < kInnerIterations; i++) {
auto dir = (i % 2 == 0) ? fan::FanDirection::FORWARD : fan::FanDirection::REVERSE;
int speed = (i % 6) + 1;
fan.make_call().set_state(true).set_speed(speed).set_oscillating(i % 2 == 0).set_direction(dir).perform();
}
benchmark::DoNotOptimize(fan.state);
}
state.SetItemsProcessed(state.iterations() * kInnerIterations);
}
BENCHMARK(FanCall_MultiField);
} // namespace esphome::benchmarks
@@ -0,0 +1 @@
fan:
@@ -0,0 +1,28 @@
import esphome.codegen as cg
from esphome.components.light import generate_gamma_table
from tests.testing_helpers import ComponentManifestOverride
def override_manifest(manifest: ComponentManifestOverride) -> None:
# Light benchmarks need USE_LIGHT_GAMMA_LUT defined and a gamma table
# with external linkage that the benchmark .cpp can reference.
manifest.enable_codegen()
original_to_code = manifest.to_code
async def to_code(config):
await original_to_code(config)
cg.add_define("USE_LIGHT_GAMMA_LUT")
# Use the light component's own generate_gamma_table() so the
# benchmark stays in sync with any formula changes.
forward = generate_gamma_table(2.8)
values = ", ".join(f"0x{int(v):04X}" for v in forward)
# Use extern-visible (non-static) array so the benchmark .cpp
# can reference it via extern declaration.
cg.add_global(
cg.RawStatement(
f"extern const uint16_t bench_gamma_2_8_fwd[256] PROGMEM = {{{values}}};"
)
)
to_code.priority = original_to_code.priority
manifest.to_code = to_code
@@ -0,0 +1,253 @@
#include <benchmark/benchmark.h>
#include "esphome/components/light/light_output.h"
#include "esphome/components/light/light_state.h"
// Gamma 2.8 forward LUT generated by the light component's Python codegen
// (see tests/benchmarks/components/light/__init__.py which calls generate_gamma_table())
extern const uint16_t bench_gamma_2_8_fwd[256];
namespace esphome::benchmarks {
// Inner iteration count to amortize CodSpeed instrumentation overhead.
static constexpr int kInnerIterations = 2000;
// Minimal LightOutput for benchmarking — no real hardware interaction.
class BenchLightOutput : public light::LightOutput {
public:
light::LightTraits get_traits() override { return this->traits_; }
void write_state(light::LightState * /*state*/) override {}
light::LightTraits traits_;
};
// Test subclass to access protected configure_entity_() for benchmark setup.
class TestLightState : public light::LightState {
public:
using LightState::LightState;
void configure(const char *name) { this->configure_entity_(name, 0x12345678, 0); }
};
// Helper to create a configured RGBWW light state for benchmarks.
// Note: setup() is not called (no preferences backend), so save_remote_values_()
// is effectively a no-op. This benchmarks the call/validation path, not persistence.
static void setup_rgbww_light(BenchLightOutput &output, TestLightState &light) {
output.traits_.set_supported_color_modes({light::ColorMode::RGB_COLD_WARM_WHITE});
output.traits_.set_min_mireds(153.0f);
output.traits_.set_max_mireds(500.0f);
light.configure("test_light");
light.set_default_transition_length(0);
light.set_gamma_correct(2.8f);
light.set_gamma_table(bench_gamma_2_8_fwd);
light.set_restore_mode(light::LIGHT_ALWAYS_OFF);
}
// --- LightCall::perform() with instant RGB color change (Home Assistant API path) ---
// Measures the full call path: validation, set_immediately_, publish, and save.
// HA sends color_mode explicitly since API 1.6.
static void LightCall_RGBInstant(benchmark::State &state) {
BenchLightOutput output;
TestLightState light(&output);
setup_rgbww_light(output, light);
// Turn on first so subsequent calls are color changes
light.make_call().set_state(true).set_brightness(1.0f).set_color_brightness(1.0f).set_transition_length(0).perform();
for (auto _ : state) {
for (int i = 0; i < kInnerIterations; i++) {
float v = static_cast<float>(i % 256) / 255.0f;
light.make_call()
.set_color_mode(light::ColorMode::RGB_COLD_WARM_WHITE)
.set_red(v)
.set_green(1.0f - v)
.set_blue(v * 0.5f)
.set_transition_length(0)
.perform();
}
benchmark::DoNotOptimize(light.remote_values);
}
state.SetItemsProcessed(state.iterations() * kInnerIterations);
}
BENCHMARK(LightCall_RGBInstant);
// --- LightCall::perform() turn on/off cycle (Home Assistant API path) ---
// HA sends color_mode explicitly since API 1.6, skipping compute_color_mode_().
static void LightCall_ToggleOnOff(benchmark::State &state) {
BenchLightOutput output;
TestLightState light(&output);
setup_rgbww_light(output, light);
for (auto _ : state) {
for (int i = 0; i < kInnerIterations; i++) {
light.make_call()
.set_state(i % 2 == 0)
.set_color_mode(light::ColorMode::RGB_COLD_WARM_WHITE)
.set_transition_length(0)
.perform();
}
benchmark::DoNotOptimize(light.remote_values);
}
state.SetItemsProcessed(state.iterations() * kInnerIterations);
}
BENCHMARK(LightCall_ToggleOnOff);
// --- LightCall::perform() turn on/off via MQTT ---
// MQTT never sends color_mode, so compute_color_mode_() runs every call.
static void LightCall_ToggleOnOff_MQTT(benchmark::State &state) {
BenchLightOutput output;
TestLightState light(&output);
setup_rgbww_light(output, light);
for (auto _ : state) {
for (int i = 0; i < kInnerIterations; i++) {
light.make_call().set_state(i % 2 == 0).set_transition_length(0).perform();
}
benchmark::DoNotOptimize(light.remote_values);
}
state.SetItemsProcessed(state.iterations() * kInnerIterations);
}
BENCHMARK(LightCall_ToggleOnOff_MQTT);
// --- LightCall::perform() with color temperature via MQTT ---
// Exercises the transform_parameters_() path that converts color_temperature
// to cold/warm white fractions. MQTT never sends color_mode, so this also
// hits compute_color_mode_() every call. Modern HA avoids this path entirely
// by converting color temp to CW/WW client-side.
static void LightCall_ColorTemperature_MQTT(benchmark::State &state) {
BenchLightOutput output;
TestLightState light(&output);
setup_rgbww_light(output, light);
light.make_call().set_state(true).set_brightness(1.0f).set_transition_length(0).perform();
for (auto _ : state) {
for (int i = 0; i < kInnerIterations; i++) {
// Sweep through color temperature range
float ct = 153.0f + static_cast<float>(i % 348);
light.make_call().set_color_temperature(ct).set_transition_length(0).perform();
}
benchmark::DoNotOptimize(light.remote_values);
}
state.SetItemsProcessed(state.iterations() * kInnerIterations);
}
BENCHMARK(LightCall_ColorTemperature_MQTT);
// --- LightCall::perform() with 1s transition (Home Assistant API path) ---
// Exercises start_transition_() which allocates a LightTransformer.
// This is the default HA path when transition_length > 0.
static void LightCall_Transition(benchmark::State &state) {
BenchLightOutput output;
TestLightState light(&output);
setup_rgbww_light(output, light);
light.make_call().set_state(true).set_brightness(1.0f).set_transition_length(0).perform();
for (auto _ : state) {
for (int i = 0; i < kInnerIterations; i++) {
float v = static_cast<float>(i % 256) / 255.0f;
light.make_call()
.set_color_mode(light::ColorMode::RGB_COLD_WARM_WHITE)
.set_red(v)
.set_green(1.0f - v)
.set_blue(v * 0.5f)
.set_transition_length(1000)
.perform();
}
benchmark::DoNotOptimize(light.remote_values);
}
state.SetItemsProcessed(state.iterations() * kInnerIterations);
}
BENCHMARK(LightCall_Transition);
// --- LightCall::perform() with cold/warm white (Home Assistant API path) ---
// Mirrors what modern HA sends: explicit color_mode with direct cold_white
// and warm_white values. HA converts color temp to CW/WW client-side for
// CWWW lights (API >= 1.6), so this is the primary HA path.
static void LightCall_ColdWarmWhite(benchmark::State &state) {
BenchLightOutput output;
TestLightState light(&output);
setup_rgbww_light(output, light);
light.make_call().set_state(true).set_brightness(1.0f).set_transition_length(0).perform();
for (auto _ : state) {
for (int i = 0; i < kInnerIterations; i++) {
float frac = static_cast<float>(i % 256) / 255.0f;
light.make_call()
.set_color_mode(light::ColorMode::RGB_COLD_WARM_WHITE)
.set_cold_white(1.0f - frac)
.set_warm_white(frac)
.set_transition_length(0)
.perform();
}
benchmark::DoNotOptimize(light.remote_values);
}
state.SetItemsProcessed(state.iterations() * kInnerIterations);
}
BENCHMARK(LightCall_ColdWarmWhite);
// --- LightState::publish_state() with a remote values listener ---
// Measures listener notification overhead.
static void LightPublish_WithListener(benchmark::State &state) {
BenchLightOutput output;
TestLightState light(&output);
setup_rgbww_light(output, light);
struct TestListener : public light::LightRemoteValuesListener {
void on_light_remote_values_update() override { count_++; }
uint64_t count_{0};
} listener;
light.add_remote_values_listener(&listener);
for (auto _ : state) {
for (int i = 0; i < kInnerIterations; i++) {
light.publish_state();
}
benchmark::DoNotOptimize(listener.count_);
}
state.SetItemsProcessed(state.iterations() * kInnerIterations);
}
BENCHMARK(LightPublish_WithListener);
// --- current_values_as_rgbww output conversion with gamma LUT ---
// Measures the output conversion path that real light drivers call
// from write_state() to get hardware PWM values, including gamma
// table lookups via the LUT generated by Python codegen.
static void LightOutput_RGBWW(benchmark::State &state) {
BenchLightOutput output;
TestLightState light(&output);
setup_rgbww_light(output, light);
light.make_call()
.set_state(true)
.set_brightness(0.8f)
.set_color_brightness(0.6f)
.set_red(1.0f)
.set_green(0.5f)
.set_blue(0.2f)
.set_cold_white(0.7f)
.set_warm_white(0.3f)
.set_transition_length(0)
.perform();
float r, g, b, cw, ww;
for (auto _ : state) {
for (int i = 0; i < kInnerIterations; i++) {
light.current_values_as_rgbww(&r, &g, &b, &cw, &ww);
}
benchmark::DoNotOptimize(r);
benchmark::DoNotOptimize(cw);
}
state.SetItemsProcessed(state.iterations() * kInnerIterations);
}
BENCHMARK(LightOutput_RGBWW);
} // namespace esphome::benchmarks
@@ -0,0 +1 @@
light:
+1 -1
View File
@@ -26,7 +26,7 @@ void setup() {
// Log functions call global_logger->log_vprintf_() without a null check,
// so we must set up a Logger before any test that triggers logging.
static esphome::logger::Logger test_logger(0, 64);
static esphome::logger::Logger test_logger(0);
test_logger.set_log_level(ESPHOME_LOG_LEVEL);
test_logger.pre_setup();
@@ -0,0 +1,79 @@
#include <benchmark/benchmark.h>
#include "esphome/components/sensor/sensor.h"
namespace esphome::benchmarks {
// Inner iteration count to amortize CodSpeed instrumentation overhead.
// Without this, the ~60ns per-iteration valgrind start/stop cost dominates
// sub-microsecond benchmarks.
static constexpr int kInnerIterations = 2000;
// Test subclass to access protected configure_entity_() for benchmark setup.
class TestSensor : public sensor::Sensor {
public:
void configure(const char *name) { this->configure_entity_(name, 0x12345678, 0); }
};
// --- Sensor::publish_state() with no callbacks registered ---
// Measures baseline publish overhead: state assignment, logging,
// internal_send_state_to_frontend, ControllerRegistry notification.
static void SensorPublish_NoCallbacks(benchmark::State &state) {
TestSensor sensor;
sensor.configure("test_sensor");
for (auto _ : state) {
for (int i = 0; i < kInnerIterations; i++) {
sensor.publish_state(static_cast<float>(i));
}
benchmark::DoNotOptimize(sensor.state);
}
state.SetItemsProcessed(state.iterations() * kInnerIterations);
}
BENCHMARK(SensorPublish_NoCallbacks);
// --- Sensor::publish_state() with one state callback ---
// Measures callback dispatch overhead through LazyCallbackManager.
static void SensorPublish_WithCallback(benchmark::State &state) {
TestSensor sensor;
sensor.configure("test_sensor");
float callback_value = 0.0f;
sensor.add_on_state_callback([&callback_value](float value) { callback_value = value; });
for (auto _ : state) {
for (int i = 0; i < kInnerIterations; i++) {
sensor.publish_state(static_cast<float>(i));
}
benchmark::DoNotOptimize(callback_value);
}
state.SetItemsProcessed(state.iterations() * kInnerIterations);
}
BENCHMARK(SensorPublish_WithCallback);
// --- Sensor::publish_state() with the same value every time ---
// Steady-state pattern: sensor reports an unchanged reading.
// Sensor doesn't dedup today, so this exercises the same code path
// as changing values, but tracks the common real-world pattern
// separately for regression detection.
static void SensorPublish_SameValue(benchmark::State &state) {
TestSensor sensor;
sensor.configure("test_sensor");
// Warm up so has_state is already set
sensor.publish_state(23.5f);
for (auto _ : state) {
for (int i = 0; i < kInnerIterations; i++) {
sensor.publish_state(23.5f);
}
benchmark::DoNotOptimize(sensor.state);
}
state.SetItemsProcessed(state.iterations() * kInnerIterations);
}
BENCHMARK(SensorPublish_SameValue);
} // namespace esphome::benchmarks
@@ -0,0 +1,2 @@
*.pcf -text
*.ttf -text
Binary file not shown.
+337
View File
@@ -0,0 +1,337 @@
"""Tests for the font component.
Focuses on verifying that long multi-byte (Chinese/CJK) glyph strings
are correctly processed through the font configuration pipeline.
"""
import functools
from pathlib import Path
from unittest.mock import MagicMock, patch
import pytest
from esphome.components.font import (
CONF_BPP,
CONF_EXTRAS,
CONF_GLYPHSETS,
CONF_IGNORE_MISSING_GLYPHS,
CONF_RAW_GLYPH_ID,
FONT_CACHE,
flatten,
glyph_comparator,
to_code,
validate_font_config,
)
import esphome.config_validation as cv
from esphome.const import (
CONF_FILE,
CONF_GLYPHS,
CONF_ID,
CONF_PATH,
CONF_RAW_DATA_ID,
CONF_SIZE,
CONF_TYPE,
)
FONT_DIR = Path(__file__).parent
FONT_PATH = FONT_DIR / "NotoSans-Regular.ttf"
# 200 unique CJK Unified Ideograph characters (U+4E00..U+4EC7)
CHINESE_200 = "".join(chr(cp) for cp in range(0x4E00, 0x4EC8))
def _file_conf() -> dict:
return {CONF_PATH: str(FONT_PATH), CONF_TYPE: "local"}
def _make_config(
glyphs: list[str],
*,
ignore_missing: bool = False,
size: int = 20,
bpp: int = 1,
extras: list | None = None,
glyphsets: list | None = None,
) -> dict:
"""Build a config dict matching what FONT_SCHEMA produces."""
return {
CONF_FILE: _file_conf(),
CONF_GLYPHS: glyphs,
CONF_GLYPHSETS: glyphsets or [],
CONF_IGNORE_MISSING_GLYPHS: ignore_missing,
CONF_SIZE: size,
CONF_BPP: bpp,
CONF_EXTRAS: extras or [],
}
@pytest.fixture(autouse=True)
def _load_font():
"""Load the test font into FONT_CACHE and clean up afterwards."""
fc = _file_conf()
FONT_CACHE[fc] = FONT_PATH
yield
FONT_CACHE.store.clear()
# ---------- flatten / glyph_comparator helpers ----------
def test_flatten_splits_chinese_string_into_chars():
"""A single string of 200 Chinese characters must become 200 individual chars."""
result = flatten([CHINESE_200])
assert len(result) == 200
assert all(len(c) == 1 for c in result)
assert result[0] == "\u4e00"
assert result[-1] == "\u4ec7"
def test_flatten_multiple_chinese_strings():
"""Multiple glyph strings are concatenated then split correctly."""
s1 = CHINESE_200[:100]
s2 = CHINESE_200[100:]
result = flatten([list(s1), list(s2)])
assert len(result) == 200
def test_glyph_comparator_orders_chinese_by_utf8():
"""glyph_comparator must order CJK characters by their UTF-8 byte sequence."""
chars = list(CHINESE_200[:10])
sorted_chars = sorted(chars, key=functools.cmp_to_key(glyph_comparator))
# CJK block is contiguous and UTF-8 order matches codepoint order here
assert sorted_chars == chars
def test_glyph_comparator_mixed_ascii_and_chinese():
"""ASCII characters sort before CJK characters (lower UTF-8 bytes)."""
assert glyph_comparator("A", "\u4e00") == -1
assert glyph_comparator("\u4e00", "A") == 1
assert glyph_comparator("\u4e00", "\u4e00") == 0
# ---------- validate_font_config ----------
def test_long_chinese_glyphs_raises_missing_error():
"""200 Chinese chars not present in NotoSans must raise Invalid with the correct count."""
config = _make_config([CHINESE_200])
with pytest.raises(cv.Invalid, match=r"missing 200 glyphs"):
validate_font_config(config)
def test_long_chinese_glyphs_error_mentions_overflow():
"""When more than 10 glyphs are missing the error should mention the remainder."""
config = _make_config([CHINESE_200])
with pytest.raises(cv.Invalid, match=r"and 190 more"):
validate_font_config(config)
def test_duplicate_chinese_glyphs_detected():
"""Duplicate CJK characters within a single glyph string must be caught."""
duped = "\u4e00\u4e01\u4e00" # first char repeated
config = _make_config([duped])
with pytest.raises(cv.Invalid, match="duplicate"):
validate_font_config(config)
def test_duplicate_chinese_across_strings():
"""Duplicates across separate glyph strings are also caught."""
config = _make_config(["\u4e00\u4e01", "\u4e01\u4e02"])
with pytest.raises(cv.Invalid, match="duplicate"):
validate_font_config(config)
def test_no_false_duplicates_in_200_unique_chinese():
"""200 unique CJK characters must not trigger the duplicate check."""
config = _make_config([CHINESE_200])
# Should not raise duplicate error — it should reach the missing-glyph check instead
with pytest.raises(cv.Invalid, match="missing"):
validate_font_config(config)
def test_valid_latin_glyphs_pass_validation():
"""Latin characters present in NotoSans-Regular pass validation without error."""
config = _make_config(["ABCabc123"])
result = validate_font_config(config)
assert result is not None
assert result[CONF_SIZE] == 20
def test_long_latin_glyphs_pass_validation():
"""A long string of supported Latin glyphs passes validation."""
# 95 printable ASCII characters that NotoSans supports
latin = "".join(chr(cp) for cp in range(0x21, 0x7F))
config = _make_config([latin])
result = validate_font_config(config)
assert result is not None
def test_mixed_latin_and_chinese_glyphs_error():
"""Mixing valid Latin and invalid Chinese chars reports missing Chinese glyphs."""
chinese_10 = CHINESE_200[:10]
config = _make_config(["ABC", chinese_10])
with pytest.raises(cv.Invalid, match=r"missing 10 glyphs"):
validate_font_config(config)
def test_single_chinese_char_glyph():
"""A single Chinese character is correctly handled as one glyph."""
config = _make_config(["\u4e00"])
with pytest.raises(cv.Invalid, match=r"missing 1 glyph[^s]"):
validate_font_config(config)
def test_chinese_glyphs_as_individual_list_items():
"""Chinese chars provided as separate list items are handled the same as a single string."""
chars_as_list = list(CHINESE_200[:50])
config = _make_config(chars_as_list)
with pytest.raises(cv.Invalid, match=r"missing 50 glyphs"):
validate_font_config(config)
# ---------- YAML parsing ----------
def test_yaml_long_latin_glyphs_parsed_and_validated(tmp_path):
"""200 Latin Extended chars on a single YAML line are parsed intact and pass validation."""
from esphome.yaml_util import load_yaml
latin_long = "".join(chr(cp) for cp in range(0x100, 0x1C8))
yaml_file = tmp_path / "font_test.yaml"
yaml_file.write_text(
f'font:\n - file: "NotoSans-Regular.ttf"\n glyphs: "{latin_long}"\n',
encoding="utf-8",
)
parsed = load_yaml(yaml_file)
raw_glyphs = parsed["font"][0]["glyphs"]
# YAML must preserve every Unicode character on the single line
assert raw_glyphs == latin_long
assert len(raw_glyphs) == 200
# Feed through validate_font_config to confirm all glyphs are accepted
config = _make_config([raw_glyphs])
result = validate_font_config(config)
assert result is not None
@pytest.mark.parametrize(
"glyphs_str",
[
" ABC", # space at start
"AB CD", # space in middle
"ABC ", # space at end
],
ids=["start", "middle", "end"],
)
def test_yaml_space_in_glyphs_preserved(tmp_path, glyphs_str):
"""A space character in a glyphs string must survive YAML round-trip and validation."""
from esphome.yaml_util import load_yaml
yaml_file = tmp_path / "font_test.yaml"
yaml_file.write_text(
f'font:\n - file: "NotoSans-Regular.ttf"\n glyphs: "{glyphs_str}"\n',
encoding="utf-8",
)
parsed = load_yaml(yaml_file)
raw_glyphs = parsed["font"][0]["glyphs"]
assert raw_glyphs == glyphs_str
assert " " in raw_glyphs
# Space and ASCII letters are all in NotoSans — validation must pass
config = _make_config([raw_glyphs])
result = validate_font_config(config)
assert result is not None
# ---------- to_code generation ----------
# 200 unique Latin Extended characters (U+0100..U+01C7), all present in NotoSans
LATIN_LONG = "".join(chr(cp) for cp in range(0x100, 0x1C8))
@pytest.fixture
def mock_cg():
"""Mock all cg codegen functions used by to_code."""
with (
patch("esphome.components.font.cg.add_define") as mock_define,
patch("esphome.components.font.cg.progmem_array") as mock_progmem,
patch("esphome.components.font.cg.static_const_array") as mock_static,
patch("esphome.components.font.cg.new_Pvariable") as mock_new_pvar,
):
mock_progmem.return_value = MagicMock()
mock_static.return_value = MagicMock()
yield {
"add_define": mock_define,
"progmem_array": mock_progmem,
"static_const_array": mock_static,
"new_Pvariable": mock_new_pvar,
}
@pytest.mark.asyncio
async def test_to_code_long_latin_generates_all_glyphs(mock_cg):
"""to_code must generate glyph data for every character in a long Latin string."""
glyph_count = len(LATIN_LONG) # 200
config = _make_config([LATIN_LONG])
config[CONF_ID] = MagicMock()
config[CONF_RAW_DATA_ID] = MagicMock()
config[CONF_RAW_GLYPH_ID] = MagicMock()
await to_code(config)
# USE_FONT define must be emitted
mock_cg["add_define"].assert_any_call("USE_FONT")
# progmem_array receives the combined bitmap data (non-empty)
mock_cg["progmem_array"].assert_called_once()
bitmap_data = mock_cg["progmem_array"].call_args.args[1]
assert len(bitmap_data) > 0
# static_const_array receives one entry per unique glyph
mock_cg["static_const_array"].assert_called_once()
glyph_initializer = mock_cg["static_const_array"].call_args.args[1]
assert len(glyph_initializer) == glyph_count
# new_Pvariable is called with the correct glyph count
mock_cg["new_Pvariable"].assert_called_once()
pvar_args = mock_cg["new_Pvariable"].call_args.args
assert pvar_args[2] == glyph_count # len(glyph_initializer)
assert pvar_args[8] == 1 # bpp
@pytest.mark.asyncio
async def test_to_code_glyph_entries_contain_expected_fields(mock_cg):
"""Each glyph initializer entry must have 7 fields: codepoint, data ptr, advance, offset_x, offset_y, w, h."""
config = _make_config([LATIN_LONG])
config[CONF_ID] = MagicMock()
config[CONF_RAW_DATA_ID] = MagicMock()
config[CONF_RAW_GLYPH_ID] = MagicMock()
await to_code(config)
glyph_initializer = mock_cg["static_const_array"].call_args.args[1]
for entry in glyph_initializer:
assert len(entry) == 7, f"Glyph entry should have 7 fields, got {len(entry)}"
codepoint = entry[0]
assert isinstance(codepoint, int)
assert 0x100 <= codepoint <= 0x1C7
@pytest.mark.asyncio
async def test_to_code_glyphs_sorted_by_utf8(mock_cg):
"""Glyphs in the initializer must be sorted by UTF-8 byte order."""
config = _make_config([LATIN_LONG])
config[CONF_ID] = MagicMock()
config[CONF_RAW_DATA_ID] = MagicMock()
config[CONF_RAW_GLYPH_ID] = MagicMock()
await to_code(config)
glyph_initializer = mock_cg["static_const_array"].call_args.args[1]
codepoints = [entry[0] for entry in glyph_initializer]
assert codepoints == sorted(codepoints)
@@ -0,0 +1,280 @@
"""Tests for light effect name validation."""
from __future__ import annotations
from collections.abc import Generator
from contextvars import Token
import pytest
from esphome import config_validation as cv
from esphome.components.light import (
EffectRef,
_final_validate,
_get_data,
available_effects_str,
find_effect_index,
)
from esphome.components.light.automation import _record_effect_ref
from esphome.config import Config, path_context
from esphome.const import CONF_EFFECT, CONF_EFFECTS, CONF_ID, CONF_NAME
from esphome.core import ID, Lambda
import esphome.final_validate as fv
from esphome.types import ConfigType
def _make_effects(*names: str) -> list[dict[str, dict[str, str]]]:
"""Create a list of effect config dicts from names."""
return [{f"effect_{i}": {CONF_NAME: name}} for i, name in enumerate(names)]
# --- find_effect_index ---
def test_find_effect_index_found() -> None:
effects = _make_effects("Fast Pulse", "Slow Pulse")
assert find_effect_index(effects, "Fast Pulse") == 1
assert find_effect_index(effects, "Slow Pulse") == 2
def test_find_effect_index_case_insensitive() -> None:
effects = _make_effects("Fast Pulse")
assert find_effect_index(effects, "fast pulse") == 1
assert find_effect_index(effects, "FAST PULSE") == 1
def test_find_effect_index_not_found() -> None:
effects = _make_effects("Fast Pulse", "Slow Pulse")
assert find_effect_index(effects, "Missing") is None
def test_find_effect_index_empty() -> None:
assert find_effect_index([], "anything") is None
# --- available_effects_str ---
def test_available_effects_str_multiple() -> None:
effects = _make_effects("Fast Pulse", "Slow Pulse")
assert available_effects_str(effects) == "'Fast Pulse', 'Slow Pulse'"
def test_available_effects_str_single() -> None:
effects = _make_effects("Fast Pulse")
assert available_effects_str(effects) == "'Fast Pulse'"
def test_available_effects_str_empty() -> None:
assert available_effects_str([]) == "none"
# --- _final_validate ---
def _setup_final_validate(
effect_refs: list[EffectRef],
light_configs: list[ConfigType],
declare_ids: list[tuple[ID, list[str | int]]],
) -> Token:
"""Set up CORE.data and fv.full_config for _final_validate tests."""
data = _get_data()
data.effect_refs = effect_refs
full_conf = Config()
full_conf["light"] = light_configs
for id_, path in declare_ids:
full_conf.declare_ids.append((id_, path))
return fv.full_config.set(full_conf)
def test_final_validate_valid_effect() -> None:
"""Valid effect name should not raise."""
light_id = ID("led1", is_declaration=True)
token = _setup_final_validate(
effect_refs=[
EffectRef(
light_id=light_id, effect_name="Fast Pulse", component_path=["esphome"]
),
],
light_configs=[
{CONF_ID: light_id, CONF_EFFECTS: _make_effects("Fast Pulse", "Slow Pulse")}
],
declare_ids=[(light_id, ["light", 0, CONF_ID])],
)
try:
_final_validate({})
finally:
fv.full_config.reset(token)
def test_final_validate_invalid_effect_raises() -> None:
"""Invalid effect name should raise FinalExternalInvalid."""
light_id = ID("led1", is_declaration=True)
token = _setup_final_validate(
effect_refs=[
EffectRef(
light_id=light_id, effect_name="Nonexistent", component_path=["esphome"]
),
],
light_configs=[
{CONF_ID: light_id, CONF_EFFECTS: _make_effects("Fast Pulse", "Slow Pulse")}
],
declare_ids=[(light_id, ["light", 0, CONF_ID])],
)
try:
with pytest.raises(cv.FinalExternalInvalid, match="Nonexistent"):
_final_validate({})
finally:
fv.full_config.reset(token)
def test_final_validate_lists_available_effects() -> None:
"""Error message should list available effects."""
light_id = ID("led1", is_declaration=True)
token = _setup_final_validate(
effect_refs=[
EffectRef(
light_id=light_id, effect_name="Missing", component_path=["esphome"]
),
],
light_configs=[
{CONF_ID: light_id, CONF_EFFECTS: _make_effects("Fast Pulse", "Slow Pulse")}
],
declare_ids=[(light_id, ["light", 0, CONF_ID])],
)
try:
with pytest.raises(cv.FinalExternalInvalid, match="'Fast Pulse', 'Slow Pulse'"):
_final_validate({})
finally:
fv.full_config.reset(token)
def test_final_validate_no_effects_on_light() -> None:
"""Light with no effects should report 'none' as available."""
light_id = ID("led1", is_declaration=True)
token = _setup_final_validate(
effect_refs=[
EffectRef(
light_id=light_id, effect_name="Missing", component_path=["esphome"]
),
],
light_configs=[{CONF_ID: light_id}],
declare_ids=[(light_id, ["light", 0, CONF_ID])],
)
try:
with pytest.raises(cv.FinalExternalInvalid, match="Available effects: none"):
_final_validate({})
finally:
fv.full_config.reset(token)
def test_final_validate_no_refs_is_noop() -> None:
"""No stored refs should pass without error."""
data = _get_data()
data.effect_refs = []
_final_validate({})
def test_final_validate_unknown_light_id_skipped() -> None:
"""Refs to unknown light IDs should be silently skipped."""
data = _get_data()
data.effect_refs = [
EffectRef(
light_id=ID("nonexistent", is_declaration=True),
effect_name="Missing",
component_path=["esphome"],
)
]
full_conf = Config()
token = fv.full_config.set(full_conf)
try:
_final_validate({})
finally:
fv.full_config.reset(token)
def test_final_validate_drains_refs() -> None:
"""Refs should be drained after validation to avoid redundant runs."""
light_id = ID("led1", is_declaration=True)
token = _setup_final_validate(
effect_refs=[
EffectRef(
light_id=light_id, effect_name="Fast Pulse", component_path=["esphome"]
),
],
light_configs=[{CONF_ID: light_id, CONF_EFFECTS: _make_effects("Fast Pulse")}],
declare_ids=[(light_id, ["light", 0, CONF_ID])],
)
try:
_final_validate({})
assert _get_data().effect_refs == []
finally:
fv.full_config.reset(token)
# --- _record_effect_ref ---
@pytest.fixture
def _path_ctx() -> Generator[None]:
"""Set path_context for _record_effect_ref tests."""
token = path_context.set(["esphome"])
yield
path_context.reset(token)
@pytest.mark.usefixtures("_path_ctx")
def test_record_effect_ref_static() -> None:
"""Static effect name should be recorded."""
light_id = ID("led1", is_declaration=True)
config: ConfigType = {CONF_ID: light_id, CONF_EFFECT: "Fast Pulse"}
result = _record_effect_ref(config)
assert result is config
data = _get_data()
assert len(data.effect_refs) == 1
assert data.effect_refs[0].effect_name == "Fast Pulse"
assert data.effect_refs[0].light_id is light_id
assert data.effect_refs[0].component_path == ["esphome"]
@pytest.mark.usefixtures("_path_ctx")
def test_record_effect_ref_skips_lambda() -> None:
"""Lambda effect should not be recorded."""
config: ConfigType = {
CONF_ID: ID("led1", is_declaration=True),
CONF_EFFECT: Lambda("return effect;"),
}
_record_effect_ref(config)
assert _get_data().effect_refs == []
@pytest.mark.usefixtures("_path_ctx")
def test_record_effect_ref_skips_none() -> None:
"""Effect 'None' should not be recorded."""
config: ConfigType = {
CONF_ID: ID("led1", is_declaration=True),
CONF_EFFECT: "None",
}
_record_effect_ref(config)
assert _get_data().effect_refs == []
@pytest.mark.usefixtures("_path_ctx")
def test_record_effect_ref_skips_none_case_insensitive() -> None:
"""Effect 'none' (lowercase) should not be recorded."""
config: ConfigType = {
CONF_ID: ID("led1", is_declaration=True),
CONF_EFFECT: "none",
}
_record_effect_ref(config)
assert _get_data().effect_refs == []
def test_record_effect_ref_skips_no_effect_key() -> None:
"""Config without effect key should be a no-op."""
config: ConfigType = {CONF_ID: ID("led1", is_declaration=True)}
_record_effect_ref(config)
assert _get_data().effect_refs == []
+2 -2
View File
@@ -69,7 +69,7 @@ def test_packages_skip_update_false(
}
# Call with skip_update=False (default)
do_packages_pass(config, skip_update=False)
do_packages_pass(config, command_line_substitutions={}, skip_update=False)
# Verify clone_or_update was called with actual refresh value
mock_clone_or_update.assert_called_once()
@@ -104,7 +104,7 @@ def test_packages_default_no_skip(
}
# Call without skip_update parameter
do_packages_pass(config)
do_packages_pass(config, command_line_substitutions={})
# Verify clone_or_update was called with actual refresh value
mock_clone_or_update.assert_called_once()
+168 -12
View File
@@ -6,6 +6,7 @@ from unittest.mock import MagicMock, patch
import pytest
from esphome.components.packages import CONFIG_SCHEMA, do_packages_pass, merge_packages
from esphome.components.substitutions import do_substitution_pass
import esphome.config as config_module
from esphome.config import resolve_extend_remove
from esphome.config_helpers import Extend, Remove
@@ -36,6 +37,7 @@ from esphome.const import (
)
from esphome.core import CORE
from esphome.util import OrderedDict
from esphome.yaml_util import add_context
# Test strings
TEST_DEVICE_NAME = "test_device_name"
@@ -69,8 +71,9 @@ def fixture_basic_esphome():
def packages_pass(config):
"""Wrapper around packages_pass that also resolves Extend and Remove."""
"""Passes the config through the packages processing steps."""
config = do_packages_pass(config)
config = do_substitution_pass(config)
config = merge_packages(config)
resolve_extend_remove(config)
return config
@@ -703,6 +706,85 @@ def test_remote_packages_with_files_list(
assert actual == expected
@patch("esphome.yaml_util.load_yaml")
@patch("pathlib.Path.is_file")
@patch("esphome.git.clone_or_update")
def test_remote_packages_with_files_list_and_substitutions(
mock_clone_or_update, mock_is_file, mock_load_yaml
) -> None:
"""
Ensures that packages are loaded as mixed list of dictionary and strings
"""
# Mock the response from git.clone_or_update
mock_revert = MagicMock()
mock_clone_or_update.return_value = (Path("/tmp/noexists"), mock_revert)
# Mock the response from pathlib.Path.is_file
mock_is_file.return_value = True
# Mock the response from esphome.yaml_util.load_yaml
mock_load_yaml.side_effect = [
OrderedDict(
{
CONF_SENSOR: [
{
CONF_PLATFORM: TEST_SENSOR_PLATFORM_1,
CONF_NAME: TEST_SENSOR_NAME_1,
}
]
}
),
OrderedDict(
{
CONF_SENSOR: [
{
CONF_PLATFORM: TEST_SENSOR_PLATFORM_1,
CONF_NAME: TEST_SENSOR_NAME_2,
}
]
}
),
]
# Define the input config
config = {
CONF_PACKAGES: {
"package1": add_context(
{
CONF_URL: r"${url}",
CONF_REF: r"${branch}",
CONF_FILES: [
{CONF_PATH: r"$file"},
"sensor2.yaml",
],
CONF_REFRESH: "1d",
},
{
"branch": "main",
"file": TEST_YAML_FILENAME,
"url": "https://github.com/esphome/non-existant-repo",
},
)
}
}
expected = {
CONF_SENSOR: [
{
CONF_PLATFORM: TEST_SENSOR_PLATFORM_1,
CONF_NAME: TEST_SENSOR_NAME_1,
},
{
CONF_PLATFORM: TEST_SENSOR_PLATFORM_1,
CONF_NAME: TEST_SENSOR_NAME_2,
},
]
}
actual = packages_pass(config)
assert actual == expected
@patch("esphome.yaml_util.load_yaml")
@patch("pathlib.Path.is_file")
@patch("esphome.git.clone_or_update")
@@ -904,7 +986,7 @@ def test_packages_merge_substitutions() -> None:
},
}
actual = do_packages_pass(config)
actual = do_packages_pass(config, command_line_substitutions={})
assert actual == expected
@@ -968,33 +1050,107 @@ def test_package_merge() -> None:
assert actual == expected
def test_packages_invalid_type_raises() -> None:
"""Packages that are not a dict or list raise cv.Invalid."""
config = {
CONF_PACKAGES: "not_a_dict_or_list",
}
with pytest.raises(
cv.Invalid, match="Packages must be a key to value mapping or list"
):
do_packages_pass(config)
@pytest.mark.parametrize(
"invalid_package",
[
6,
"some string",
["some string"],
None,
True,
{"some_component": 8},
{3: 2},
{"some_component": r"${unevaluated expression}"},
],
)
def test_package_merge_invalid(invalid_package) -> None:
"""
Tests that trying to merge an invalid package raises an error.
"""
def test_invalid_package_contents_rejected(invalid_package: object) -> None:
"""Invalid package contents are rejected by PACKAGE_SCHEMA during do_packages_pass."""
config = {
CONF_PACKAGES: {
"some_package": invalid_package,
},
}
with pytest.raises(cv.Invalid):
do_packages_pass(config)
@pytest.mark.xfail(
reason="Deprecated single-package fallback swallows these errors. "
"Remove xfail when single-package deprecation is removed (2026.7.0).",
strict=True,
)
@pytest.mark.parametrize(
"invalid_package",
[
None,
["some string"],
{"some_component": 8},
{3: 2},
],
)
def test_invalid_package_contents_masked_by_deprecation(
invalid_package: object,
) -> None:
"""These invalid packages are swallowed by the deprecated single-package fallback."""
config = {
CONF_PACKAGES: {
"some_package": invalid_package,
},
}
with pytest.raises(cv.Invalid):
do_packages_pass(config)
def test_merge_packages_invalid_nested_type_raises() -> None:
"""Invalid nested packages type during merge raises cv.Invalid."""
config = {
CONF_PACKAGES: {
"pkg": {
CONF_PACKAGES: "invalid",
},
},
}
with pytest.raises(
cv.Invalid, match="Packages must be a key to value mapping or list"
):
merge_packages(config)
@patch("esphome.yaml_util.load_yaml")
@patch("pathlib.Path.is_file")
@patch("esphome.git.clone_or_update")
def test_remote_packages_no_revert(
mock_clone_or_update, mock_is_file, mock_load_yaml
) -> None:
"""Remote packages with revert=None load without retry logic."""
mock_clone_or_update.return_value = (Path("/tmp/noexists"), None)
mock_is_file.return_value = True
mock_load_yaml.return_value = OrderedDict(
{CONF_SENSOR: [{CONF_PLATFORM: TEST_SENSOR_PLATFORM_1, CONF_NAME: "test"}]}
)
config = {
CONF_PACKAGES: {
"pkg": {
CONF_URL: "https://github.com/esphome/repo",
CONF_REF: "main",
CONF_FILES: [{CONF_PATH: "file.yaml"}],
CONF_REFRESH: "1d",
}
}
}
actual = packages_pass(config)
assert actual[CONF_SENSOR] == [
{CONF_PLATFORM: TEST_SENSOR_PLATFORM_1, CONF_NAME: "test"}
]
def test_raw_config_contains_merged_esphome_from_package(tmp_path) -> None:
"""Test that CORE.raw_config contains esphome section from merged package.
+9
View File
@@ -0,0 +1,9 @@
sensor:
- platform: bmp581_spi
cs_pin: ${cs_pin}
temperature:
name: BMP581 Temperature
iir_filter: 2x
pressure:
name: BMP581 Pressure
oversampling: 128x
@@ -0,0 +1,7 @@
substitutions:
cs_pin: GPIO5
packages:
spi: !include ../../test_build_components/common/spi/esp32-idf.yaml
<<: !include common.yaml
@@ -0,0 +1,7 @@
substitutions:
cs_pin: GPIO15
packages:
spi: !include ../../test_build_components/common/spi/esp8266-ard.yaml
<<: !include common.yaml
@@ -0,0 +1,7 @@
substitutions:
cs_pin: GPIO5
packages:
spi: !include ../../test_build_components/common/spi/rp2040-ard.yaml
<<: !include common.yaml
@@ -19,6 +19,7 @@ esp32:
disable_mbedtls_pkcs7: true
disable_regi2c_in_iram: true
disable_fatfs: true
sram1_as_iram: true
wifi:
ssid: MySSID
@@ -69,3 +69,11 @@ esp32_ble_server:
- ble_server.descriptor.set_value:
id: test_change_descriptor
value: !lambda return bytebuffer::ByteBuffer::wrap({0x03, 0x04, 0x05}).get_data();
- ble_server.characteristic.set_value:
id: test_change_characteristic
value:
data: [0xfc, 0xef, 0xfe, 0x86]
- ble_server.descriptor.set_value:
id: test_change_descriptor
value:
data: [0x01, 0x02, 0x03]
@@ -0,0 +1,18 @@
ethernet:
type: ENC28J60
clk_pin: 18
mosi_pin: 19
miso_pin: 16
cs_pin: 17
interrupt_pin: 21
reset_pin: 20
manual_ip:
static_ip: 192.168.178.56
gateway: 192.168.178.1
subnet: 255.255.255.0
domain: .local
mac_address: "02:AA:BB:CC:DD:01"
on_connect:
- logger.log: "Ethernet connected!"
on_disconnect:
- logger.log: "Ethernet disconnected!"
@@ -0,0 +1,19 @@
ethernet:
type: ENC28J60
clk_pin: 19
mosi_pin: 21
miso_pin: 23
cs_pin: 18
interrupt_pin: 36
reset_pin: 22
clock_speed: 10Mhz
manual_ip:
static_ip: 192.168.178.56
gateway: 192.168.178.1
subnet: 255.255.255.0
domain: .local
mac_address: "02:AA:BB:CC:DD:01"
on_connect:
- logger.log: "Ethernet connected!"
on_disconnect:
- logger.log: "Ethernet disconnected!"
@@ -0,0 +1,18 @@
ethernet:
type: W5100
clk_pin: 18
mosi_pin: 19
miso_pin: 16
cs_pin: 17
interrupt_pin: 21
reset_pin: 20
manual_ip:
static_ip: 192.168.178.56
gateway: 192.168.178.1
subnet: 255.255.255.0
domain: .local
mac_address: "02:AA:BB:CC:DD:01"
on_connect:
- logger.log: "Ethernet connected!"
on_disconnect:
- logger.log: "Ethernet disconnected!"
@@ -0,0 +1 @@
<<: !include common-enc28j60.yaml
@@ -0,0 +1 @@
<<: !include common-enc28j60-rp2040.yaml
@@ -0,0 +1 @@
<<: !include common-w5100-rp2040.yaml
+2 -1
View File
@@ -1 +1,2 @@
*.pcf -text
*.pcf -text
*.ttf -text
@@ -8,6 +8,7 @@ infrared:
- platform: ir_rf_proxy
id: ir_rx
name: "IR Receiver"
receiver_frequency: 38kHz
remote_receiver_id: ir_receiver
# RF 900MHz receiver
+11 -20
View File
@@ -43,9 +43,6 @@ lvgl:
start_value: 0
end_value: 180
bg_color: light_blue
disp_bg_color: color_id
disp_bg_image: cat_image
disp_bg_opa: cover
bottom_layer:
widgets:
- obj:
@@ -58,7 +55,6 @@ lvgl:
gradients:
- id: color_bar
direction: hor
# dither: err_diff
stops:
- color: 0xFF0000
position: 0
@@ -143,12 +139,11 @@ lvgl:
body:
text: This is a sample messagebox
bg_color: 0x808080
button_style:
bg_color: 0xff00
border_width: 4
buttons:
- id: msgbox_button
text: Button
bg_color: 0x00ff00
border_width: 4
- id: msgbox_apply
text: "Close"
on_click:
@@ -160,8 +155,8 @@ lvgl:
bg_opa: !lambda return 0.5;
- lvgl.image.update:
id: lv_image
zoom: !lambda return 512;
angle: !lambda return 100;
scale: !lambda return 512;
rotation: !lambda return 100;
pivot_x: !lambda return 20;
pivot_y: !lambda return 20;
offset_x: !lambda return 20;
@@ -287,8 +282,8 @@ lvgl:
then:
- lvgl.animimg.stop: anim_img
- lvgl.update:
disp_bg_color: 0xffff00
disp_bg_image: none
bottom_layer:
bg_color: 0xffff00
- lvgl.widget.show: message_box
- label:
text: "Hello shiny day"
@@ -361,8 +356,6 @@ lvgl:
pad_right: 10px
pad_top: 10px
shadow_color: light_blue
shadow_ofs_x: 5
shadow_ofs_y: 5
shadow_opa: cover
shadow_spread: 5
shadow_width: 10
@@ -373,12 +366,10 @@ lvgl:
text_letter_space: 4
text_line_space: 4
text_opa: cover
transform_angle: 180
transform_rotation: 90
transform_height: 100
transform_pivot_x: 50%
transform_pivot_y: 50%
transform_zoom: 0.5
transform_scale: 2.0
transform_scale_x: 1.5
transform_scale_y: 0.8
@@ -470,11 +461,11 @@ lvgl:
id: button_button
width: 20%
height: 10%
transform_angle: !lambda return(180*100);
transform_rotation: !lambda return(180*100);
arc_width: !lambda return 4;
border_width: !lambda return 6;
shadow_ofs_x: !lambda return 6;
shadow_ofs_y: !lambda return 6;
shadow_offset_x: !lambda return 6;
shadow_offset_y: !lambda return 6;
shadow_spread: !lambda return 6;
shadow_width: !lambda return 6;
pressed:
@@ -646,8 +637,8 @@ lvgl:
border_opa: 80%
shadow_color: black
shadow_width: 10
shadow_ofs_x: 5
shadow_ofs_y: 5
shadow_offset_x: 5
shadow_offset_y: 5
shadow_spread: 4
shadow_opa: cover
outline_color: red
+1 -1
View File
@@ -22,7 +22,7 @@ void original_setup() {
void setup() {
// Log functions call global_logger->log_vprintf_() without a null check,
// so we must set up a Logger before any test that triggers logging.
static esphome::logger::Logger test_logger(0, 64);
static esphome::logger::Logger test_logger(0);
test_logger.set_log_level(ESPHOME_LOG_LEVEL);
test_logger.pre_setup();
+30 -17
View File
@@ -273,26 +273,39 @@ text_sensor:
display:
- platform: nextion
id: main_lcd
auto_wake_on_touch: true
brightness: 80%
command_spacing: 5ms
dump_device_info: true
exit_reparse_on_start: true
lambda: |-
ESP_LOGD("display","Display is being tested!");
max_commands_per_loop: 20
max_queue_age: 5000ms # Remove queue items after 5s
max_queue_size: 50
update_interval: 5s
on_sleep:
then:
lambda: 'ESP_LOGD("display","Display went to sleep");'
on_wake:
then:
lambda: 'ESP_LOGD("display","Display woke up");'
on_setup:
then:
lambda: 'ESP_LOGD("display","Display setup completed");'
on_page:
then:
lambda: 'ESP_LOGD("display","Display shows new page %u", x);'
on_buffer_overflow:
then:
logger.log: "Nextion reported a buffer overflow!"
command_spacing: 5ms
dump_device_info: true
max_queue_age: 5000ms # Remove queue items after 5s
on_page:
then:
lambda: 'ESP_LOGD("display","Display shows new page %u", x);'
on_setup:
then:
lambda: 'ESP_LOGD("display","Display setup completed");'
on_sleep:
then:
lambda: 'ESP_LOGD("display","Display went to sleep");'
on_touch:
then:
lambda: |-
ESP_LOGD("display",
"Display was touched at page %u, component %u, touch event: %s",
page_id, component_id, touch_event ? "press" : "release");
on_wake:
then:
lambda: 'ESP_LOGD("display","Display woke up");'
update_interval: 5s
start_up_page: 1
startup_override_ms: 10000ms # Wait 10s for display ready
touch_sleep_timeout: 3
wake_up_page: 2
+13
View File
@@ -0,0 +1,13 @@
number:
- platform: template
name: "Test Number"
id: test_number
optimistic: true
min_value: 0
max_value: 100
step: 1
sensor:
- platform: number
name: "Test Number Value"
source_id: test_number
@@ -0,0 +1,2 @@
packages:
common: !include common.yaml
@@ -0,0 +1,2 @@
packages:
common: !include common.yaml
+15
View File
@@ -0,0 +1,15 @@
sensor:
- platform: spa06_spi
spi_id: spi_bus
cs_pin: ${cs_pin}
temperature:
id: spa06_spi_temperature
name: Outside Temperature
sample_rate: 1
oversampling: NONE
pressure:
name: Outside Pressure
id: spa06_spi_pressure
sample_rate: 25p4
oversampling: 16X
update_interval: 15s
@@ -0,0 +1,7 @@
substitutions:
cs_pin: GPIO5
packages:
spi: !include ../../test_build_components/common/spi/esp32-idf.yaml
<<: !include common.yaml
@@ -0,0 +1,7 @@
substitutions:
cs_pin: GPIO15
packages:
spi: !include ../../test_build_components/common/spi/esp8266-ard.yaml
<<: !include common.yaml
@@ -0,0 +1,7 @@
substitutions:
cs_pin: GPIO17
packages:
spi: !include ../../test_build_components/common/spi/rp2040-ard.yaml
<<: !include common.yaml
+5
View File
@@ -23,3 +23,8 @@ text:
min_length: 8
max_length: 32
mode: password
text_sensor:
- platform: text
name: "Test Text State"
source_id: test_text
+54 -2
View File
@@ -1036,8 +1036,6 @@ static time_t esptime_recalc_local(int year, int month, int day, int hour, int m
t.hour = hour;
t.minute = min;
t.second = sec;
t.day_of_week = 1; // Placeholder for fields_in_range()
t.day_of_year = 1;
t.recalc_timestamp_local();
return t.timestamp;
}
@@ -1187,6 +1185,60 @@ TEST(RecalcTimestampLocal, NonDefaultTransitionTime) {
EXPECT_EQ(esp_result, libc_result);
}
TEST(RecalcTimestampLocal, MinimalFieldsWithoutDayOfWeekOrYear) {
// Regression test for issue #15115: DateTimeEntity::state_as_esptime() constructs
// an ESPTime with only year/month/day/hour/minute/second set (no day_of_week or
// day_of_year). recalc_timestamp_local() must work without those fields.
const char *tz_str = "CET-1CEST,M3.5.0,M10.5.0";
setenv("TZ", tz_str, 1);
tzset();
time::ParsedTimezone tz{};
ASSERT_TRUE(parse_posix_tz(tz_str, tz));
set_global_tz(tz);
// Construct ESPTime with only date/time fields (like state_as_esptime does)
ESPTime t{};
t.year = 2026;
t.month = 3;
t.day_of_month = 20;
t.hour = 23;
t.minute = 14;
t.second = 55;
// day_of_week and day_of_year are deliberately left as 0
t.recalc_timestamp_local();
// Must NOT return -1 (the bug: fields_in_range() rejected valid times)
EXPECT_NE(t.timestamp, -1);
// Verify against libc
time_t libc_result = libc_mktime(2026, 3, 20, 23, 14, 55);
EXPECT_EQ(t.timestamp, libc_result);
}
TEST(RecalcTimestampLocal, MinimalFieldsNoDST) {
// Same test but with a timezone that has no DST
const char *tz_str = "IST-5:30";
setenv("TZ", tz_str, 1);
tzset();
time::ParsedTimezone tz{};
ASSERT_TRUE(parse_posix_tz(tz_str, tz));
set_global_tz(tz);
ESPTime t{};
t.year = 2026;
t.month = 3;
t.day_of_month = 23;
t.hour = 10;
t.minute = 0;
t.second = 0;
t.recalc_timestamp_local();
EXPECT_NE(t.timestamp, -1);
time_t libc_result = libc_mktime(2026, 3, 23, 10, 0, 0);
EXPECT_EQ(t.timestamp, libc_result);
}
TEST(RecalcTimestampLocal, YearBoundaryDST) {
// Test southern hemisphere DST across year boundary
// Australia/Sydney: DST active from October to April (spans Jan 1)
+2
View File
@@ -5,3 +5,5 @@ display:
intensity: 3
lambda: |-
it.print("1234");
static const uint8_t buf[] = {0x3f, 0x06, 0x5b, 0x4f | 0x80};
it.set_buffer(buf, sizeof(buf));
+1 -1
View File
@@ -15,7 +15,7 @@ void setup() {
static char name[] = "livingroom";
static char friendly_name[] = "LivingRoom";
App.pre_setup(name, sizeof(name) - 1, friendly_name, sizeof(friendly_name) - 1);
auto *log = new logger::Logger(115200, 512); // NOLINT
auto *log = new logger::Logger(115200); // NOLINT
log->pre_setup();
log->set_uart_selection(logger::UART_SELECTION_UART0);
App.register_component_(log);
@@ -0,0 +1,39 @@
esphome:
name: light-initial-state-test
host:
api: # Port will be automatically injected
logger:
level: DEBUG
output:
- platform: template
id: test_red
type: float
write_action:
- lambda: ""
- platform: template
id: test_green
type: float
write_action:
- lambda: ""
- platform: template
id: test_blue
type: float
write_action:
- lambda: ""
light:
- platform: rgb
name: "Test Light"
id: test_light
red: test_red
green: test_green
blue: test_blue
restore_mode: ALWAYS_OFF
initial_state:
color_mode: RGB
state: true
brightness: 0.75
red: 1.0
green: 0.5
blue: 0.0
@@ -0,0 +1,38 @@
"""Integration test for light initial_state configuration.
Tests that the initial_state values are correctly applied at boot when
no saved preferences exist. The initial_state callback populates defaults
that the restore logic uses as a fallback.
"""
import pytest
from .state_utils import InitialStateHelper, require_entity
from .types import APIClientConnectedFactory, RunCompiledFunction
@pytest.mark.asyncio
async def test_light_initial_state(
yaml_config: str,
run_compiled: RunCompiledFunction,
api_client_connected: APIClientConnectedFactory,
) -> None:
"""Test that initial_state values are applied at boot."""
async with run_compiled(yaml_config), api_client_connected() as client:
entities, _ = await client.list_entities_services()
light = require_entity(entities, "test_light")
helper = InitialStateHelper(entities)
client.subscribe_states(helper.on_state_wrapper(lambda s: None))
await helper.wait_for_initial_states()
state = helper.initial_states[light.key]
# restore_mode: ALWAYS_OFF overrides state to false
assert state.state is False
# But the color values from initial_state should be applied
assert state.brightness == pytest.approx(0.75, abs=0.05)
assert state.red == pytest.approx(1.0, abs=0.01)
assert state.green == pytest.approx(0.5, abs=0.01)
assert state.blue == pytest.approx(0.0, abs=0.01)
+8 -1
View File
@@ -83,11 +83,18 @@ async def test_uart_mock_ld2450(
],
)
# Signal when we see recovery frame values (target 1 distance ≈ 500mm)
# Signal when we see all recovery frame values
# Must wait for ALL values to avoid race where some arrive after the waiter fires
recovery_received = collector.add_waiter(
lambda: (
pytest.approx(500.0, abs=1.0)
in collector.sensor_states["target_1_distance"]
and pytest.approx(300.0) in collector.sensor_states["target_1_x"]
and pytest.approx(400.0) in collector.sensor_states["target_1_y"]
and pytest.approx(30.0) in collector.sensor_states["target_1_speed"]
and pytest.approx(1.0) in collector.sensor_states["target_count"]
and pytest.approx(1.0) in collector.sensor_states["moving_target_count"]
and pytest.approx(0.0) in collector.sensor_states["still_target_count"]
)
)
+18
View File
@@ -248,6 +248,24 @@ def test_area_id_hash_collision(
)
def test_area_singular_hash_collision(
yaml_file: Callable[[str], str], capsys: pytest.CaptureFixture[str]
) -> None:
"""Test that area hash collisions between singular area: and areas: list are detected."""
result = load_config_from_fixture(
yaml_file, "area_singular_hash_collision.yaml", FIXTURES_DIR
)
assert result is None
captured = capsys.readouterr()
assert (
"Area ID 'd6ka' with hash 3082558663 collides with existing area ID 'test_2258'"
in captured.out
)
# Error path should point to 'areas' (where the colliding entry is), not 'area'
assert "areas" in captured.out
def test_device_duplicate_id(
yaml_file: Callable[[str], str], capsys: pytest.CaptureFixture[str]
) -> None:
@@ -0,0 +1,10 @@
esphome:
name: test
area:
id: test_2258
name: "Area 1"
areas:
- id: d6ka
name: "Area 2"
host:
@@ -38,3 +38,20 @@ test_list:
- '{ 79, 82 }'
- a: 15 should be 15, overridden from command line
b: 20 should stay as 20, not overridden
- aa:
- 1
- 2
- 3
- 4
- 5
- 6
bb:
- 7
- 8
- 9
- aa:
x: 1
y: 3
z: 4
bb:
w: 5
@@ -44,3 +44,13 @@ test_list:
- '{ ${position.x}, ${position.y} }'
- a: ${a} should be 15, overridden from command line
b: ${b} should stay as 20, not overridden
# Test merging lists when substituted keys resolve to an existing key
- ${ "aa" }: [1, 2, 3]
${ "a" + "a" }: [4, 5, 6]
${ "bb" }: [7, 8, 9]
# Test merging dicts when substituted keys resolve to an existing key
- ${ "aa" }: {"x": 1, "y": 2}
${ "a" + "a" }: {"y": 3, "z": 4}
${ "bb" }: {"w": 5}
@@ -9,6 +9,11 @@ substitutions:
numberOne: 1
var1: 79
double_width: 14
double_height: 16
y: ${x}
x: ${y}
b: 79
c: 80
test_list:
- The area is 56
- 56
@@ -27,3 +32,4 @@ test_list:
- chr(97) = a
- len([1,2,3]) = 3
- width = 7, double_width = 14
- a = ${a}
@@ -1,4 +1,7 @@
substitutions:
y: ${x} # Circular reference, expect to pass unresolved.
x: ${y} # Circular reference, expect to pass unresolved.
double_height: ${height * 2}
width: 7
height: 8
enabled: true
@@ -9,6 +12,8 @@ substitutions:
numberOne: 1
var1: 79
double_width: ${width * 2}
c: ${b+1}
b: ${undefined_variable | default(79) }
test_list:
- "The area is ${width * height}"
@@ -25,3 +30,4 @@ test_list:
- chr(97) = ${ chr(97) }
- len([1,2,3]) = ${ len([1,2,3]) }
- width = ${width}, double_width = ${double_width}
- a = ${a}
@@ -1,7 +1,3 @@
substitutions:
x: 10
y: 20
z: 30
values_from_repo1_main:
- package_name: package1
x: 3
@@ -28,3 +24,20 @@ values_from_repo1_main:
y: 20
z: 5
volume: 1000
- package_name: package6
x: 12
y: 13
z: 5
volume: 780
- package_name: default
x: 10
y: 20
z: 5
volume: 1000
substitutions:
x: 10
y: 20
z: 30
my_repo: repo1
my_file: file1
my_ref: main
@@ -2,16 +2,26 @@ substitutions:
x: 10
y: 20
z: 30
my_repo: default_repo
my_file: default_file
my_ref: main
# The following key is only used by the test framework
# to simulate command line substitutions
command_line_substitutions:
my_repo: repo1
my_file: file1
packages:
package1:
url: https://github.com/esphome/repo1
ref: main
files:
- path: file1.yaml
vars:
package_name: package1
x: 3
y: 4
ref: main
package2: !include # a package that just includes the given remote package
file: remote_package_proxy.yaml
vars:
@@ -41,3 +51,13 @@ packages:
repo: repo1
file: file1.yaml
ref: main
package6:
url: https://github.com/esphome/${my_repo}
ref: ${my_ref}
files:
- path: ${my_file + ".yaml"}
vars:
package_name: package6
x: 12
y: 13
package7: github://esphome/${my_repo}/${my_file + ".yaml"}@${my_ref}
@@ -0,0 +1,44 @@
fancy_component: &id001
- id: component9
value: 9
some_component:
- id: component1
value: 1
- id: component2
value: 2
- id: component3
value: 3
- id: component4
value: 4
- id: component5
value: 79
power: 200
- id: component6
value: 6
- id: component7
value: 7
switch: &id002
- platform: gpio
id: switch1
pin: 12
- platform: gpio
id: switch2
pin: 13
display:
- platform: ili9xxx
dimensions:
width: 100
height: 480
substitutions:
extended_component: component5
package_options:
alternative_package:
alternative_component:
- id: component8
value: 8
fancy_package:
fancy_component: *id001
pin: 12
some_switches: *id002
package_selection: fancy_package
fancy_subst: 42
@@ -0,0 +1,63 @@
substitutions:
package_options:
alternative_package:
alternative_component:
- id: component8
value: 8
fancy_package:
substitutions:
fancy_subst: 42
fancy_component:
- id: component9
value: 9
pin: 12
some_switches:
- platform: gpio
id: switch1
pin: ${pin}
- platform: gpio
id: switch2
pin: ${pin+1}
package_selection: fancy_package
packages:
- ${ package_options[package_selection] }
- some_component:
- id: component1
value: 1
- some_component:
- id: component2
value: 2
- switch: ${ some_switches }
- packages:
package_with_defaults: !include
file: display.yaml
vars:
native_width: 100
high_dpi: false
my_package:
packages:
- packages:
special_package:
substitutions:
extended_component: component5
some_component:
- id: component3
value: 3
some_component:
- id: component4
value: 4
- id: !extend ${ extended_component }
power: 200
value: 79
some_component:
- id: component5
value: 5
some_component:
- id: component6
value: 6
- id: component7
value: 7
@@ -0,0 +1,49 @@
substitutions:
a: 10
b: 20
x: 79
test_list:
- level1:
a: 10
b: 20
c: 10
d: 20
e: ${e}
f: ${f}
g: ${g}
h: ${h}
i: ${i}
j: ${j}
x: 80
y: 40
level2:
- level2:
a: 10
b: 20
c: 10
d: 20
e: 20
f: 40
g: ${g}
h: ${h}
i: ${i}
j: ${j}
x: 81
y: 40
level3:
- level3:
a: 10
b: 20
c: 10
d: 20
e: 20
f: 40
g: 100
h: 200
i: 30
j: ${undefined_variable}
x: 82
y: 40
- a: 10
b: 20
x: 79
@@ -0,0 +1,16 @@
substitutions:
a: 10
b: 20
x: 79
test_list:
- !include
file: level1_package.yaml
vars:
x: ${x+1}
y: ${d*2}
c: ${a}
d: ${b}
- a: ${a}
b: ${b}
x: ${x}
@@ -0,0 +1,5 @@
values:
- var1: $var1
- a: 10
- b: B-default
- c: The value of C is 79
@@ -0,0 +1,7 @@
# Test that include_vars with vars works even when there are no substitutions key defined.
packages:
- !include
file: inc1.yaml
vars:
a: 10
c: 79
@@ -0,0 +1,69 @@
substitutions:
a: from base config
b: from package3
c: from nested package4
nested_package:
nested_package_test_list:
- a: from base config
- b: from package3
- c: from nested package4
package1:
package1_test_list:
- a: from base config
- b: from package3
- c: from nested package4
package2:
package2_test_list:
- a: from package2 vars
- b: from package3
- c: from nested package4
package3:
package3_test_list:
- a: from base config
- b: from package3
- c: from nested package4
package4:
packages:
- nested_package_test_list:
- a: from base config
- b: from package3
- c: from nested package4
package_map:
package1:
package1_test_list:
- a: from base config
- b: from package3
- c: from nested package4
package2:
package2_test_list:
- a: from package2 vars
- b: from package3
- c: from nested package4
package3: &id001
package3_test_list:
- a: from base config
- b: from package3
- c: from nested package4
selected_package_number: 3
selected_package_name: package3
selected_package: *id001
base_test_list:
- a: from base config
- b: from package3
- c: from nested package4
package1_test_list:
- a: from base config
- b: from package3
- c: from nested package4
package2_test_list:
- a: from package2 vars
- b: from package3
- c: from nested package4
package3_test_list:
- a: from base config
- b: from package3
- c: from nested package4
nested_package_test_list:
- a: from base config
- b: from package3
- c: from nested package4
@@ -0,0 +1,62 @@
command_line_substitutions:
selected_package_number: 3
substitutions:
a: from base config
package1: &p1
substitutions:
a: from package1
b: from package1
c: from package1
package1_test_list:
- a: ${ a }
- b: ${ b }
- c: ${ c }
package2: &p2 !include
file: package2.yaml
vars:
a: from package2 vars
package3: &p3
substitutions:
a: from package3
b: from package3
c: from package3
package3_test_list:
- a: ${ a }
- b: ${ b }
- c: ${ c }
package4:
substitutions:
nested_package:
substitutions:
c: from nested package4
nested_package_test_list:
- a: ${ a }
- b: ${ b }
- c: ${ c }
packages:
- ${ nested_package }
package_map:
package1: *p1
package2: *p2
package3: *p3
selected_package_number: 2 # will be overridden by command line substitutions
selected_package_name: package${ selected_package_number }
selected_package: ${ package_map[selected_package_name] }
packages:
- ${ package1 }
- ${ package2 }
- ${ selected_package }
- ${ package4 }
base_test_list:
- a: ${ a }
- b: ${ b }
- c: ${ c }
@@ -0,0 +1,21 @@
# this file is included by 07-include_hierarchy.input.yaml
level1:
a: ${a} # top-level substitution
b: ${b} # top-level substitution
c: ${c} # from vars when including
d: ${d} # from vars when including
e: ${e} # undefined at this level
f: ${f} # undefined at this level
g: ${g} # undefined at this level
h: ${h} # undefined at this level
i: ${i} # undefined at this level
j: ${j} # undefined at this level
x: ${x} # from vars when including, calculated
y: ${y} # from vars when including, calculated
level2:
- !include
file: level2_package.yaml
vars:
e: ${c*2}
f: ${d*2}
x: ${x+1}
@@ -0,0 +1,21 @@
# this file is included by level1_package.yaml
level2:
a: ${a} # top-level substitution
b: ${b} # top-level substitution
c: ${c} # visible from level1 vars
d: ${d} # visible from level1 vars
e: ${e} # from vars when including
f: ${f} # from vars when including
g: ${g} # undefined at this level
h: ${h} # undefined at this level
i: ${i} # undefined at this level
j: ${j} # undefined at this level
x: ${x} # from vars when including, calculated
y: ${y} # from vars when including, calculated
level3:
- !include
file: level3_package.yaml
vars:
g: ${e*5}
h: ${f*5}
x: ${x+1}
@@ -0,0 +1,16 @@
# this file is included by level2_package.yaml
defaults:
i: 30
level3:
a: ${a} # top-level substitution
b: ${b} # top-level substitution
c: ${c} # visible from level1 vars
d: ${d} # visible from level1 vars
e: ${e} # visible from level2 vars
f: ${f} # visible from level2 vars
g: ${g} # from vars when including
h: ${h} # from vars when including
i: ${i} # Should take the default value of 30
j: ${undefined_variable} # Does not exist, should be output as-is
x: ${x} # from vars when including, calculated
y: ${y} # from vars when including, calculated
@@ -0,0 +1,10 @@
# included from 10-dynamic_packages.input.yaml
substitutions:
a: from package2 # must not override base config's a
# b not defined here, won't override package1's b
c: from package2 # will override package1's c
package2_test_list:
- a: ${ a }
- b: ${ b }
- c: ${ c }
+80 -1
View File
@@ -5,7 +5,13 @@ from unittest.mock import patch
import pytest
from esphome.automation import has_non_synchronous_actions
from esphome.automation import (
TriggerForwarder,
TriggerOnFalseForwarder,
TriggerOnTrueForwarder,
has_non_synchronous_actions,
)
from esphome.cpp_generator import MockObj, RawExpression
from esphome.util import RegistryEntry
@@ -175,3 +181,76 @@ def test_has_non_synchronous_actions_dict_input(
"""Direct dict input (single action)."""
assert has_non_synchronous_actions({"delay": "1s"}) is True
assert has_non_synchronous_actions({"logger.log": "hello"}) is False
def _build_forwarder(
automation_name: str,
args: list[tuple[str, str]],
forwarder: MockObj | None = None,
) -> str:
"""Build a trigger forwarder expression the same way build_callback_automation does.
Mirrors the forwarder selection logic in automation.build_callback_automation.
"""
import esphome.codegen as cg
obj = MockObj(automation_name, "->")
if forwarder is None:
arg_types = [RawExpression(t) for t, _ in args]
templ = (
cg.TemplateArguments(*arg_types) if arg_types else cg.TemplateArguments()
)
forwarder = TriggerForwarder.template(templ)
return f"{forwarder}{{{obj}}}"
def test_trigger_forwarder_no_args() -> None:
"""Button on_press: TriggerForwarder<> with no args."""
result = _build_forwarder("auto_1", [])
assert result == "TriggerForwarder<>{auto_1}"
def test_trigger_forwarder_single_float_arg() -> None:
"""Sensor on_value: TriggerForwarder<float>."""
result = _build_forwarder("auto_1", [("float", "x")])
assert result == "TriggerForwarder<float>{auto_1}"
def test_trigger_forwarder_single_bool_arg() -> None:
"""Switch on_state: TriggerForwarder<bool>."""
result = _build_forwarder("auto_1", [("bool", "x")])
assert result == "TriggerForwarder<bool>{auto_1}"
def test_trigger_forwarder_on_true() -> None:
"""Binary_sensor on_press / switch on_turn_on: TriggerOnTrueForwarder."""
result = _build_forwarder("auto_1", [], forwarder=TriggerOnTrueForwarder)
assert result == "TriggerOnTrueForwarder{auto_1}"
def test_trigger_forwarder_on_false() -> None:
"""Binary_sensor on_release / switch on_turn_off: TriggerOnFalseForwarder."""
result = _build_forwarder("auto_1", [], forwarder=TriggerOnFalseForwarder)
assert result == "TriggerOnFalseForwarder{auto_1}"
def test_trigger_forwarder_multiple_args() -> None:
"""Binary_sensor on_state_change: TriggerForwarder with two args."""
result = _build_forwarder(
"auto_1",
[("optional<bool>", "x_previous"), ("optional<bool>", "x")],
)
assert result == "TriggerForwarder<optional<bool>, optional<bool>>{auto_1}"
def test_trigger_forwarder_string_arg() -> None:
"""Text_sensor on_value: TriggerForwarder<std::string>."""
result = _build_forwarder("auto_1", [("std::string", "x")])
assert result == "TriggerForwarder<std::string>{auto_1}"
def test_trigger_forwarder_custom_type() -> None:
"""Custom forwarder type passed directly."""
custom = MockObj("MyForwarder", "")
result = _build_forwarder("auto_1", [], forwarder=custom)
assert result == "MyForwarder{auto_1}"
+41 -6
View File
@@ -1762,7 +1762,34 @@ def test_show_logs_api(
assert result == 0
mock_run_logs.assert_called_once_with(
CORE.config, ["192.168.1.100", "192.168.1.101"]
CORE.config, ["192.168.1.100", "192.168.1.101"], subscribe_states=True
)
@patch("esphome.components.api.client.run_logs")
def test_show_logs_api_no_states(
mock_run_logs: Mock,
) -> None:
"""Test show_logs with --no-states flag."""
setup_core(
config={
"logger": {},
CONF_API: {},
CONF_MDNS: {CONF_DISABLED: False},
},
platform=PLATFORM_ESP32,
)
mock_run_logs.return_value = 0
args = MockArgs()
args.no_states = True
devices = ["192.168.1.100"]
result = show_logs(CORE.config, args, devices)
assert result == 0
mock_run_logs.assert_called_once_with(
CORE.config, ["192.168.1.100"], subscribe_states=False
)
@@ -1788,7 +1815,9 @@ def test_show_logs_api_with_fqdn_mdns_disabled(
assert result == 0
# Should use the FQDN directly, not try MQTT lookup
mock_run_logs.assert_called_once_with(CORE.config, ["device.example.com"])
mock_run_logs.assert_called_once_with(
CORE.config, ["device.example.com"], subscribe_states=True
)
@patch("esphome.components.api.client.run_logs")
@@ -1816,7 +1845,9 @@ def test_show_logs_api_with_mqtt_fallback(
assert result == 0
mock_mqtt_get_ip.assert_called_once_with(CORE.config, "user", "pass", "client")
mock_run_logs.assert_called_once_with(CORE.config, ["192.168.1.200"])
mock_run_logs.assert_called_once_with(
CORE.config, ["192.168.1.200"], subscribe_states=True
)
@patch("esphome.mqtt.show_logs")
@@ -2746,7 +2777,7 @@ def test_show_logs_api_static_ip_with_mqttip(
# Verify run_logs was called with both IPs
mock_run_logs.assert_called_once_with(
CORE.config, ["192.168.1.100", "192.168.2.50"]
CORE.config, ["192.168.1.100", "192.168.2.50"], subscribe_states=True
)
@@ -2782,7 +2813,9 @@ def test_show_logs_api_multiple_mqttip_resolves_once(
# Note: "MQTT" is a different magic string from "MQTTIP", but both trigger MQTT resolution
# The _resolve_network_devices helper filters out both after first resolution
mock_run_logs.assert_called_once_with(
CORE.config, ["192.168.2.50", "192.168.2.51", "192.168.1.100"]
CORE.config,
["192.168.2.50", "192.168.2.51", "192.168.1.100"],
subscribe_states=True,
)
@@ -2862,7 +2895,9 @@ def test_show_logs_api_mqtt_timeout_fallback(
mock_mqtt_get_ip.assert_called_once_with(CORE.config, "user", "pass", "client")
# Verify run_logs was called with only the static IP (MQTT failed)
mock_run_logs.assert_called_once_with(CORE.config, ["192.168.1.100"])
mock_run_logs.assert_called_once_with(
CORE.config, ["192.168.1.100"], subscribe_states=True
)
def test_detect_external_components_no_external(
+264 -18
View File
@@ -10,9 +10,10 @@ from esphome import config as config_module, yaml_util
from esphome.components import substitutions
from esphome.components.packages import do_packages_pass, merge_packages
from esphome.config import resolve_extend_remove
from esphome.config_helpers import merge_config
from esphome.config_helpers import Extend, merge_config
import esphome.config_validation as cv
from esphome.const import CONF_SUBSTITUTIONS
from esphome.core import CORE
from esphome.core import CORE, Lambda
from esphome.util import OrderedDict
_LOGGER = logging.getLogger(__name__)
@@ -142,9 +143,11 @@ def test_substitutions_fixtures(
command_line_substitutions = config.pop("command_line_substitutions", None)
config = do_packages_pass(config)
config = do_packages_pass(
config, command_line_substitutions=command_line_substitutions
)
substitutions.do_substitution_pass(config, command_line_substitutions)
config = substitutions.do_substitution_pass(config, command_line_substitutions)
config = merge_packages(config)
@@ -206,7 +209,7 @@ def test_substitutions_with_command_line_maintains_ordered_dict() -> None:
command_line_subs = {"var2": "override", "var3": "new_value"}
# Call do_substitution_pass with command line substitutions
substitutions.do_substitution_pass(config, command_line_subs)
config = substitutions.do_substitution_pass(config, command_line_subs)
# Verify that config is still an OrderedDict
assert isinstance(config, OrderedDict), "Config should remain an OrderedDict"
@@ -234,7 +237,7 @@ def test_substitutions_without_command_line_maintains_ordered_dict() -> None:
config["other_key"] = "other_value"
# Call without command line substitutions
substitutions.do_substitution_pass(config, None)
config = substitutions.do_substitution_pass(config, None)
# Verify that config is still an OrderedDict
assert isinstance(config, OrderedDict), "Config should remain an OrderedDict"
@@ -268,7 +271,7 @@ def test_substitutions_after_merge_config_maintains_ordered_dict() -> None:
)
# Now try to run substitution pass on the merged config
substitutions.do_substitution_pass(merged_config, None)
merged_config = substitutions.do_substitution_pass(merged_config, None)
# Should not raise AttributeError
assert isinstance(merged_config, OrderedDict), (
@@ -279,7 +282,7 @@ def test_substitutions_after_merge_config_maintains_ordered_dict() -> None:
def test_validate_config_with_command_line_substitutions_maintains_ordered_dict(
tmp_path,
tmp_path: Path,
) -> None:
"""Test that validate_config preserves OrderedDict when merging command-line substitutions.
@@ -288,7 +291,7 @@ def test_validate_config_with_command_line_substitutions_maintains_ordered_dict(
"""
# Create a minimal valid config
test_config = OrderedDict()
test_config["esphome"] = {"name": "test_device", "platform": "ESP32"}
test_config["esphome"] = {"name": "test_device"}
test_config[CONF_SUBSTITUTIONS] = OrderedDict({"var1": "value1", "var2": "value2"})
test_config["esp32"] = {"board": "esp32dev"}
@@ -314,17 +317,11 @@ def test_validate_config_with_command_line_substitutions_maintains_ordered_dict(
assert result[CONF_SUBSTITUTIONS]["var3"] == "new_value"
def test_validate_config_without_command_line_substitutions_maintains_ordered_dict(
tmp_path,
) -> None:
"""Test that validate_config preserves OrderedDict without command-line substitutions.
This tests the code path in config.py where result[CONF_SUBSTITUTIONS] is set
using merge_dicts_ordered() when command_line_substitutions is None.
"""
def _get_test_minimal_valid_config(tmp_path: Path) -> OrderedDict:
"""Helper to create a minimal valid config for testing."""
# Create a minimal valid config
test_config = OrderedDict()
test_config["esphome"] = {"name": "test_device", "platform": "ESP32"}
test_config["esphome"] = {"name": "test_device"}
test_config[CONF_SUBSTITUTIONS] = OrderedDict({"var1": "value1", "var2": "value2"})
test_config["esp32"] = {"board": "esp32dev"}
@@ -332,6 +329,19 @@ def test_validate_config_without_command_line_substitutions_maintains_ordered_di
test_yaml = tmp_path / "test.yaml"
test_yaml.write_text("# test config")
CORE.config_path = test_yaml
return test_config
def test_validate_config_without_command_line_substitutions_maintains_ordered_dict(
tmp_path: Path,
) -> None:
"""Test that validate_config preserves OrderedDict without command-line substitutions.
This tests the code path in config.py where result[CONF_SUBSTITUTIONS] is set
using merge_dicts_ordered() when command_line_substitutions is None.
"""
test_config = _get_test_minimal_valid_config(tmp_path)
# Call validate_config without command line substitutions
result = config_module.validate_config(test_config, None)
@@ -384,3 +394,239 @@ def test_merge_config_preserves_ordered_dict() -> None:
assert not isinstance(result, OrderedDict), (
"dict + dict should not return OrderedDict"
)
def test_substitution_pass_error_gets_captured(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
"""vol.Invalid from do_substitution_pass is captured by validate_config."""
# Patch the target: in config_module.do_substitution_pass (NOT where it's defined)
def fake_do_substitution_pass(*args, **kwargs):
raise cv.Invalid("Error in do_substitutions_pass!!")
monkeypatch.setattr(
config_module, "do_substitution_pass", fake_do_substitution_pass
)
# Prepare minimal config + no CLI substitutions
config = _get_test_minimal_valid_config(tmp_path)
# Call the function under test
result = config_module.validate_config(config, None)
# Now assert that add_error was called with the vol.Invalid
assert "Error in do_substitutions_pass!!" in str(result.get_error_for_path([]))
@pytest.mark.parametrize(
"value", ["", " ", "1foo", "9VAR", "0abc", "$1foo", "$9VAR", "$0abc"]
)
def test_validate_substitution_key_empty_raises(value: str) -> None:
"""Empty (or all-whitespace) substitution keys are rejected."""
with pytest.raises(cv.Invalid):
substitutions.validate_substitution_key(value)
@pytest.mark.parametrize(
"input_value, expected_output",
[
("$FOO_bar9", "FOO_bar9"), # Valid key with leading '$'
("Foo_bar9", "Foo_bar9"), # Normal valid key
],
)
def test_validate_substitution_key_valid(
input_value: str, expected_output: str
) -> None:
"""Valid substitution keys are accepted with optional leading '$'."""
result = substitutions.validate_substitution_key(input_value)
assert result == expected_output
def test_circular_dependency_warnings(
caplog: pytest.LogCaptureFixture,
) -> None:
"""Circular substitution references produce warnings naming the cause."""
config = OrderedDict(
{
CONF_SUBSTITUTIONS: OrderedDict({"x": "${y}", "y": "${x}"}),
"key": "value",
}
)
with caplog.at_level(logging.WARNING):
substitutions.do_substitution_pass(config)
assert "Could not resolve substitution variable 'x'" in caplog.text
assert "'y' is undefined" in caplog.text
assert "Could not resolve substitution variable 'y'" in caplog.text
assert "'x' is undefined" in caplog.text
# Verify path includes location
assert "substitutions->x" in caplog.text
assert "substitutions->y" in caplog.text
def test_missing_dependency_warning(
caplog: pytest.LogCaptureFixture,
) -> None:
"""A substitution referencing an undefined variable warns with the cause."""
config = OrderedDict(
{
CONF_SUBSTITUTIONS: OrderedDict({"a": "${missing}"}),
"key": "value",
}
)
with caplog.at_level(logging.WARNING):
substitutions.do_substitution_pass(config)
assert "Could not resolve substitution variable 'a'" in caplog.text
assert "'missing' is undefined" in caplog.text
assert "substitutions->a" in caplog.text
def test_undefined_variable_warning(
caplog: pytest.LogCaptureFixture,
) -> None:
"""A reference to an undefined variable in config values produces a warning."""
config = OrderedDict(
{
"key": "${undefined_var}",
}
)
with caplog.at_level(logging.WARNING):
substitutions.do_substitution_pass(config)
assert "'undefined_var' is undefined" in caplog.text
def test_password_field_warnings_suppressed(
caplog: pytest.LogCaptureFixture,
) -> None:
"""Undefined variables in password fields should not produce warnings."""
config = OrderedDict(
{
"password": "${undefined_var}",
}
)
with caplog.at_level(logging.WARNING):
substitutions.do_substitution_pass(config)
assert caplog.text == ""
def test_config_context_unresolvable_warns(
caplog: pytest.LogCaptureFixture,
) -> None:
"""Unresolvable vars in a ConfigContext produce warnings via push_context."""
inner = OrderedDict({"key": "${a}"})
yaml_util.add_context(inner, {"a": "${undefined}"})
config = OrderedDict({"items": [inner]})
with caplog.at_level(logging.WARNING):
substitutions.do_substitution_pass(config)
assert "Could not resolve substitution variable 'a'" in caplog.text
assert "'undefined' is undefined" in caplog.text
def test_non_string_substitution_value_warning(
caplog: pytest.LogCaptureFixture,
) -> None:
"""Undefined vars in non-string contexts (e.g. dict keys) produce warnings."""
config = OrderedDict(
{
"items": {"${undefined_key}": "value"},
}
)
with caplog.at_level(logging.WARNING):
substitutions.do_substitution_pass(config)
assert "'undefined_key' is undefined" in caplog.text
def test_lambda_substitution() -> None:
"""Substitution inside a Lambda value should be expanded."""
lam = Lambda("return ${var};")
config = OrderedDict(
{
CONF_SUBSTITUTIONS: OrderedDict({"var": "42"}),
"lambda": lam,
}
)
config = substitutions.do_substitution_pass(config)
assert config["lambda"].value == "return 42;"
def test_lambda_no_substitution_unchanged() -> None:
"""A Lambda with no variable references should not be mutated."""
lam = Lambda("return 1;")
original_value = lam.value
config = OrderedDict(
{
CONF_SUBSTITUTIONS: OrderedDict({"var": "42"}),
"lambda": lam,
}
)
config = substitutions.do_substitution_pass(config)
assert config["lambda"].value is original_value
def test_extend_substitution() -> None:
"""Substitution inside an Extend value should be expanded."""
ext = Extend("${component_id}")
config = OrderedDict(
{
CONF_SUBSTITUTIONS: OrderedDict({"component_id": "my_sensor"}),
"sensor": ext,
}
)
config = substitutions.do_substitution_pass(config)
assert config["sensor"].value == "my_sensor"
def test_substitute_does_not_mutate_input() -> None:
"""substitute() must return a new tree without modifying the original."""
inner_list = ["${var}", "static"]
inner_dict = OrderedDict({"key": "${var}"})
lam = Lambda("return ${var};")
config = OrderedDict(
{
"a_list": inner_list,
"a_dict": inner_dict,
"a_lambda": lam,
"plain": "${var}",
}
)
context = substitutions.ContextVars({"var": "replaced"})
result = substitutions.substitute(config, [], context, strict_undefined=True)
# Result has substitutions applied
assert result["plain"] == "replaced"
assert result["a_list"] == ["replaced", "static"]
assert result["a_dict"]["key"] == "replaced"
assert result["a_lambda"].value == "return replaced;"
# Original input is untouched
assert config["plain"] == "${var}"
assert inner_list == ["${var}", "static"]
assert inner_dict["key"] == "${var}"
assert lam.value == "return ${var};"
# Containers are new objects, not the originals
assert result["a_list"] is not inner_list
assert result["a_dict"] is not inner_dict
assert result["a_lambda"] is not lam
def test_do_substitution_pass_substitutions_must_be_mapping_from_config() -> None:
"""Non-mapping substitutions raises cv.Invalid."""
config = OrderedDict(
{
CONF_SUBSTITUTIONS: ["not", "a", "mapping"],
"other": "value",
}
)
with pytest.raises(
cv.Invalid, match="Substitutions must be a key to value mapping"
):
substitutions.do_substitution_pass(config)
+165
View File
@@ -866,6 +866,171 @@ def test_clean_all_with_yaml_file(
assert str(build_dir) in caplog.text
@patch("esphome.writer.CORE")
def test_clean_all_with_yaml_build_path(
mock_core: MagicMock,
tmp_path: Path,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test clean_all cleans absolute build_path specified in YAML config."""
config_dir = tmp_path / "config"
config_dir.mkdir()
# Create an absolute custom build path directory with contents
custom_build = tmp_path / "custom_build"
custom_build.mkdir()
(custom_build / "firmware.bin").write_text("x")
sub = custom_build / "subdir"
sub.mkdir()
(sub / "file.txt").write_text("x")
yaml_file = config_dir / "test.yaml"
# Absolute build_path: data_dir / absolute = absolute (Python Path behavior)
yaml_file.write_text(f"esphome:\n name: test\n build_path: {custom_build}\n")
# Also create the normal .esphome dir
build_dir = config_dir / ".esphome"
build_dir.mkdir()
(build_dir / "dummy.txt").write_text("x")
from esphome.writer import clean_all
with caplog.at_level("INFO"):
clean_all([str(yaml_file)])
# Both .esphome and custom build_path should be cleaned
assert build_dir.exists()
assert not (build_dir / "dummy.txt").exists()
assert custom_build.exists()
assert not (custom_build / "firmware.bin").exists()
assert not sub.exists()
@patch("esphome.writer.CORE")
def test_clean_all_with_yaml_parse_error(
mock_core: MagicMock,
tmp_path: Path,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test clean_all still cleans .esphome when YAML parse fails."""
config_dir = tmp_path / "config"
config_dir.mkdir()
yaml_file = config_dir / "test.yaml"
yaml_file.write_text("invalid: yaml: content: [")
build_dir = config_dir / ".esphome"
build_dir.mkdir()
(build_dir / "dummy.txt").write_text("x")
from esphome.writer import clean_all
with caplog.at_level("INFO"):
clean_all([str(yaml_file)])
# .esphome should still be cleaned despite YAML parse failure
assert build_dir.exists()
assert not (build_dir / "dummy.txt").exists()
@patch("esphome.writer.CORE")
def test_clean_all_with_env_build_path(
mock_core: MagicMock,
tmp_path: Path,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test clean_all cleans ESPHOME_BUILD_PATH directory."""
config_dir = tmp_path / "config"
config_dir.mkdir()
build_dir = config_dir / ".esphome"
build_dir.mkdir()
(build_dir / "dummy.txt").write_text("x")
# Create env build path directory
env_build = tmp_path / "env_build"
env_build.mkdir()
(env_build / "firmware.bin").write_text("x")
from esphome.writer import clean_all
with (
caplog.at_level("INFO"),
patch.dict(os.environ, {"ESPHOME_BUILD_PATH": str(env_build)}),
):
clean_all([str(config_dir)])
# Both should be cleaned
assert not (build_dir / "dummy.txt").exists()
assert env_build.exists()
assert not (env_build / "firmware.bin").exists()
@patch("esphome.writer.CORE")
def test_clean_all_ignores_empty_env_vars(
mock_core: MagicMock,
tmp_path: Path,
) -> None:
"""Test clean_all ignores empty ESPHOME_BUILD_PATH/ESPHOME_DATA_DIR."""
config_dir = tmp_path / "config"
config_dir.mkdir()
# Create a file in cwd that must NOT be cleaned
marker = tmp_path / "important.txt"
marker.write_text("do not delete")
from esphome.writer import clean_all
with patch.dict(
os.environ,
{"ESPHOME_BUILD_PATH": "", "ESPHOME_DATA_DIR": ""},
):
clean_all([str(config_dir)])
# Empty env vars must not cause cwd to be cleaned
assert marker.exists()
@patch("esphome.writer.CORE")
def test_clean_all_no_args_with_esphome_dir(
mock_core: MagicMock,
tmp_path: Path,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test clean_all with no args cleans .esphome in cwd."""
esphome_dir = tmp_path / ".esphome"
esphome_dir.mkdir()
(esphome_dir / "dummy.txt").write_text("x")
from esphome.writer import clean_all
with (
caplog.at_level("INFO"),
patch("esphome.writer.Path.cwd", return_value=tmp_path),
):
clean_all([])
assert esphome_dir.exists()
assert not (esphome_dir / "dummy.txt").exists()
@patch("esphome.writer.CORE")
def test_clean_all_no_args_no_esphome_dir(
mock_core: MagicMock,
tmp_path: Path,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test clean_all with no args and no .esphome dir warns."""
from esphome.writer import clean_all
with (
caplog.at_level("WARNING"),
patch("esphome.writer.Path.cwd", return_value=tmp_path),
):
clean_all([])
assert "No configuration files specified" in caplog.text
@patch("esphome.writer.CORE")
def test_clean_all(
mock_core: MagicMock,
+29 -15
View File
@@ -25,7 +25,7 @@ def test_include_with_vars(fixture_path: Path) -> None:
yaml_file = fixture_path / "yaml_util" / "includetest.yaml"
actual = yaml_util.load_yaml(yaml_file)
substitutions.do_substitution_pass(actual, None)
actual = substitutions.do_substitution_pass(actual, None)
assert actual["esphome"]["name"] == "original"
assert actual["esphome"]["libraries"][0] == "Wire"
assert actual["esp8266"]["board"] == "nodemcu"
@@ -98,13 +98,15 @@ def test_construct_secret_missing(fixture_path: Path, tmp_path: Path) -> None:
"""Test that missing secrets raise proper errors."""
# Create a YAML file with a secret that doesn't exist
test_yaml = tmp_path / "test.yaml"
test_yaml.write_text("""
test_yaml.write_text(
"""
esphome:
name: test
wifi:
password: !secret nonexistent_secret
""")
"""
)
# Create an empty secrets file
secrets_yaml = tmp_path / "secrets.yaml"
@@ -118,10 +120,12 @@ def test_construct_secret_no_secrets_file(tmp_path: Path) -> None:
"""Test that missing secrets.yaml file raises proper error."""
# Create a YAML file with a secret but no secrets.yaml
test_yaml = tmp_path / "test.yaml"
test_yaml.write_text("""
test_yaml.write_text(
"""
wifi:
password: !secret some_secret
""")
"""
)
# Mock CORE.config_path to avoid NoneType error
with (
@@ -140,10 +144,12 @@ def test_construct_secret_fallback_to_main_config_dir(
subdir.mkdir()
test_yaml = subdir / "test.yaml"
test_yaml.write_text("""
test_yaml.write_text(
"""
wifi:
password: !secret test_secret
""")
"""
)
# Create secrets.yaml in the main directory
main_secrets = tmp_path / "secrets.yaml"
@@ -164,9 +170,11 @@ def test_construct_include_dir_named(fixture_path: Path, tmp_path: Path) -> None
# Create test YAML that uses include_dir_named
test_yaml = dst_dir / "test_include_named.yaml"
test_yaml.write_text("""
test_yaml.write_text(
"""
sensor: !include_dir_named named_dir
""")
"""
)
actual = yaml_util.load_yaml(test_yaml)
actual_sensor = actual["sensor"]
@@ -199,9 +207,11 @@ def test_construct_include_dir_named_empty_dir(tmp_path: Path) -> None:
empty_dir.mkdir()
test_yaml = tmp_path / "test.yaml"
test_yaml.write_text("""
test_yaml.write_text(
"""
sensor: !include_dir_named empty_dir
""")
"""
)
actual = yaml_util.load_yaml(test_yaml)
@@ -231,9 +241,11 @@ def test_construct_include_dir_named_with_dots(tmp_path: Path) -> None:
hidden_subfile.write_text("key: hidden_subfile_value")
test_yaml = tmp_path / "test.yaml"
test_yaml.write_text("""
test_yaml.write_text(
"""
test: !include_dir_named test_dir
""")
"""
)
actual = yaml_util.load_yaml(test_yaml)
@@ -255,9 +267,11 @@ def test_find_files_recursive(fixture_path: Path, tmp_path: Path) -> None:
# This indirectly tests _find_files by using include_dir_named
test_yaml = dst_dir / "test_include_recursive.yaml"
test_yaml.write_text("""
test_yaml.write_text(
"""
all_sensors: !include_dir_named named_dir
""")
"""
)
actual = yaml_util.load_yaml(test_yaml)