mirror of
https://github.com/esphome/esphome.git
synced 2026-09-14 16:48:40 +00:00
Merge branch 'dev' into alarm-control-panel-trigger-trampoline
This commit is contained in:
@@ -24,6 +24,7 @@ from esphome.const import (
|
||||
CONF_ID,
|
||||
CONF_INITIAL_STATE,
|
||||
CONF_MQTT_ID,
|
||||
CONF_NAME,
|
||||
CONF_ON_STATE,
|
||||
CONF_ON_TURN_OFF,
|
||||
CONF_ON_TURN_ON,
|
||||
@@ -41,6 +42,8 @@ from esphome.const import (
|
||||
from esphome.core import CORE, ID, CoroPriority, HexInt, Lambda, coroutine_with_priority
|
||||
from esphome.core.entity_helpers import entity_duplicate_validator, setup_entity
|
||||
from esphome.cpp_generator import MockObjClass
|
||||
import esphome.final_validate as fv
|
||||
from esphome.types import ConfigType
|
||||
|
||||
from .automation import LIGHT_STATE_SCHEMA
|
||||
from .effects import (
|
||||
@@ -70,9 +73,19 @@ IS_PLATFORM_COMPONENT = True
|
||||
DOMAIN = "light"
|
||||
|
||||
|
||||
@dataclass
|
||||
class EffectRef:
|
||||
"""A pending effect name reference from a light action to validate."""
|
||||
|
||||
light_id: ID
|
||||
effect_name: str
|
||||
component_path: list[str | int] # path_context when the action was validated
|
||||
|
||||
|
||||
@dataclass
|
||||
class LightData:
|
||||
gamma_tables: dict = field(default_factory=dict) # gamma_value -> fwd_arr
|
||||
effect_refs: list[EffectRef] = field(default_factory=list)
|
||||
|
||||
|
||||
def _get_data() -> LightData:
|
||||
@@ -115,6 +128,68 @@ def _get_or_create_gamma_table(gamma_correct):
|
||||
return fwd_arr
|
||||
|
||||
|
||||
def find_effect_index(effects: list, effect_name: str) -> int | None:
|
||||
"""Find the 1-based index of an effect by name (case-insensitive).
|
||||
|
||||
Returns the 1-based index if found, or None if not found.
|
||||
"""
|
||||
effect_name_lower = effect_name.lower()
|
||||
for i, effect_conf in enumerate(effects):
|
||||
key = next(iter(effect_conf))
|
||||
if effect_conf[key][CONF_NAME].lower() == effect_name_lower:
|
||||
return i + 1
|
||||
return None
|
||||
|
||||
|
||||
def available_effects_str(effects: list) -> str:
|
||||
"""Return a comma-separated string of available effect names."""
|
||||
available = [
|
||||
effect_conf[next(iter(effect_conf))][CONF_NAME] for effect_conf in effects
|
||||
]
|
||||
return ", ".join(f"'{name}'" for name in available) if available else "none"
|
||||
|
||||
|
||||
def _final_validate(config: ConfigType) -> ConfigType:
|
||||
"""Validate all recorded effect name references against their target lights.
|
||||
|
||||
This runs once per light platform instance. If no light platform is configured,
|
||||
this never runs — but the ID validator will catch the missing light ID separately.
|
||||
"""
|
||||
data = _get_data()
|
||||
if not data.effect_refs:
|
||||
return config
|
||||
|
||||
# Drain the list so we only validate once even though
|
||||
# FINAL_VALIDATE_SCHEMA runs for each light platform instance.
|
||||
refs = data.effect_refs
|
||||
data.effect_refs = []
|
||||
|
||||
fconf = fv.full_config.get()
|
||||
|
||||
for ref in refs:
|
||||
try:
|
||||
light_path = fconf.get_path_for_id(ref.light_id)[:-1]
|
||||
light_config = fconf.get_config_for_path(light_path)
|
||||
except KeyError:
|
||||
# Light ID not found — ID validation will have already reported this
|
||||
continue
|
||||
|
||||
effects = light_config.get(CONF_EFFECTS, [])
|
||||
|
||||
if find_effect_index(effects, ref.effect_name) is None:
|
||||
raise cv.FinalExternalInvalid(
|
||||
f"Effect '{ref.effect_name}' not found for light "
|
||||
f"'{ref.light_id}'. "
|
||||
f"Available effects: {available_effects_str(effects)}",
|
||||
path=[cv.ROOT_CONFIG_PATH] + ref.component_path,
|
||||
)
|
||||
|
||||
return config
|
||||
|
||||
|
||||
FINAL_VALIDATE_SCHEMA = _final_validate
|
||||
|
||||
|
||||
LightRestoreMode = light_ns.enum("LightRestoreMode")
|
||||
RESTORE_MODES = {
|
||||
"RESTORE_DEFAULT_OFF": LightRestoreMode.LIGHT_RESTORE_DEFAULT_OFF,
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
from esphome import automation
|
||||
import esphome.codegen as cg
|
||||
from esphome.config import path_context
|
||||
import esphome.config_validation as cv
|
||||
from esphome.const import (
|
||||
CONF_BLUE,
|
||||
@@ -17,7 +18,6 @@ from esphome.const import (
|
||||
CONF_LIMIT_MODE,
|
||||
CONF_MAX_BRIGHTNESS,
|
||||
CONF_MIN_BRIGHTNESS,
|
||||
CONF_NAME,
|
||||
CONF_RANGE_FROM,
|
||||
CONF_RANGE_TO,
|
||||
CONF_RED,
|
||||
@@ -26,7 +26,7 @@ from esphome.const import (
|
||||
CONF_WARM_WHITE,
|
||||
CONF_WHITE,
|
||||
)
|
||||
from esphome.core import CORE, Lambda
|
||||
from esphome.core import CORE, EsphomeError, Lambda
|
||||
from esphome.cpp_generator import LambdaExpression
|
||||
from esphome.types import ConfigType
|
||||
|
||||
@@ -98,6 +98,31 @@ LIGHT_CONTROL_ACTION_SCHEMA = LIGHT_STATE_SCHEMA.extend(
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
def _record_effect_ref(config: ConfigType) -> ConfigType:
|
||||
"""Record a static effect name reference for later cross-component validation."""
|
||||
if CONF_EFFECT not in config:
|
||||
return config
|
||||
effect = config[CONF_EFFECT]
|
||||
if isinstance(effect, Lambda):
|
||||
return config # Lambda effects resolved at runtime
|
||||
if effect.lower() == "none":
|
||||
return config # "None" is always valid
|
||||
|
||||
from . import EffectRef, _get_data
|
||||
|
||||
_get_data().effect_refs.append(
|
||||
EffectRef(
|
||||
light_id=config[CONF_ID],
|
||||
effect_name=effect,
|
||||
component_path=path_context.get(),
|
||||
)
|
||||
)
|
||||
return config
|
||||
|
||||
|
||||
LIGHT_CONTROL_ACTION_SCHEMA.add_extra(_record_effect_ref)
|
||||
|
||||
LIGHT_TURN_OFF_ACTION_SCHEMA = automation.maybe_simple_id(
|
||||
{
|
||||
cv.Required(CONF_ID): cv.use_id(LightState),
|
||||
@@ -122,18 +147,24 @@ def _resolve_effect_index(config: ConfigType) -> int:
|
||||
Effect index 0 means "None" (no effect). Effects are 1-indexed matching
|
||||
the C++ convention in LightState.
|
||||
"""
|
||||
from . import available_effects_str, find_effect_index
|
||||
|
||||
original_name = config[CONF_EFFECT]
|
||||
effect_name = original_name.lower()
|
||||
if effect_name == "none":
|
||||
if original_name.lower() == "none":
|
||||
return 0
|
||||
light_id = config[CONF_ID]
|
||||
light_path = CORE.config.get_path_for_id(light_id)[:-1]
|
||||
light_config = CORE.config.get_config_for_path(light_path)
|
||||
for i, effect_conf in enumerate(light_config.get(CONF_EFFECTS, [])):
|
||||
key = next(iter(effect_conf))
|
||||
if effect_conf[key][CONF_NAME].lower() == effect_name:
|
||||
return i + 1
|
||||
raise ValueError(f"Effect '{original_name}' not found in light '{light_id}'")
|
||||
effects = light_config.get(CONF_EFFECTS, [])
|
||||
index = find_effect_index(effects, original_name)
|
||||
if index is not None:
|
||||
return index
|
||||
# Should never reach here — effect names are validated during config
|
||||
# validation in FINAL_VALIDATE_SCHEMA. This is a safety net.
|
||||
raise EsphomeError(
|
||||
f"Effect '{original_name}' not found for light '{light_id}'. "
|
||||
f"Available effects: {available_effects_str(effects)}"
|
||||
)
|
||||
|
||||
|
||||
@automation.register_action(
|
||||
|
||||
@@ -243,6 +243,16 @@ void Logger::dump_config() {
|
||||
#endif
|
||||
#ifdef USE_ZEPHYR
|
||||
dump_crash_();
|
||||
#endif
|
||||
// Warn users that VERBOSE/VERY_VERBOSE logging impacts performance.
|
||||
// Only the compiled log level matters — all log calls up to this level
|
||||
// are in the binary and will be formatted (vsnprintf) and block UART.
|
||||
#if ESPHOME_LOG_LEVEL >= ESPHOME_LOG_LEVEL_VERY_VERBOSE
|
||||
ESP_LOGW(TAG, "VERY_VERBOSE logging is active — significant performance impact, short-term debugging only\n"
|
||||
" May cause connection instability. Set log level to DEBUG or lower for long-term use.");
|
||||
#elif ESPHOME_LOG_LEVEL >= ESPHOME_LOG_LEVEL_VERBOSE
|
||||
ESP_LOGI(TAG, "VERBOSE logging is active — performance impact, short-term debugging only\n"
|
||||
" Set log level to DEBUG or lower for long-term use.");
|
||||
#endif
|
||||
}
|
||||
|
||||
|
||||
@@ -10,6 +10,10 @@ namespace nextion {
|
||||
|
||||
static const char *const TAG = "nextion";
|
||||
|
||||
// Nextion command terminator: three consecutive 0xFF bytes (per Nextion Instruction Set v1.1).
|
||||
static constexpr uint8_t COMMAND_DELIMITER[3] = {0xFF, 0xFF, 0xFF};
|
||||
static constexpr size_t DELIMITER_SIZE = sizeof(COMMAND_DELIMITER);
|
||||
|
||||
void Nextion::setup() {
|
||||
this->is_setup_ = false;
|
||||
this->connection_state_.ignore_is_setup_ = true;
|
||||
@@ -415,7 +419,8 @@ void Nextion::process_nextion_commands_() {
|
||||
#ifdef NEXTION_PROTOCOL_LOG
|
||||
this->print_queue_members_();
|
||||
#endif
|
||||
while ((to_process_length = this->command_data_.find(COMMAND_DELIMITER)) != std::string::npos) {
|
||||
while ((to_process_length = this->command_data_.find(reinterpret_cast<const char *>(COMMAND_DELIMITER), 0,
|
||||
DELIMITER_SIZE)) != std::string::npos) {
|
||||
#ifdef USE_NEXTION_MAX_COMMANDS_PER_LOOP
|
||||
if (++commands_processed > this->max_commands_per_loop_) {
|
||||
ESP_LOGW(TAG, "Command processing limit exceeded");
|
||||
@@ -423,8 +428,8 @@ void Nextion::process_nextion_commands_() {
|
||||
}
|
||||
#endif // USE_NEXTION_MAX_COMMANDS_PER_LOOP
|
||||
ESP_LOGN(TAG, "queue size: %zu", this->nextion_queue_.size());
|
||||
while (to_process_length + COMMAND_DELIMITER.length() < this->command_data_.length() &&
|
||||
static_cast<uint8_t>(this->command_data_[to_process_length + COMMAND_DELIMITER.length()]) == 0xFF) {
|
||||
while (to_process_length + DELIMITER_SIZE < this->command_data_.length() &&
|
||||
static_cast<uint8_t>(this->command_data_[to_process_length + DELIMITER_SIZE]) == 0xFF) {
|
||||
++to_process_length;
|
||||
ESP_LOGN(TAG, "Add 0xFF");
|
||||
}
|
||||
@@ -829,7 +834,7 @@ void Nextion::process_nextion_commands_() {
|
||||
break;
|
||||
}
|
||||
|
||||
this->command_data_.erase(0, to_process_length + COMMAND_DELIMITER.length() + 1);
|
||||
this->command_data_.erase(0, to_process_length + DELIMITER_SIZE + 1);
|
||||
}
|
||||
|
||||
const uint32_t ms = App.get_loop_component_start_time();
|
||||
|
||||
@@ -29,8 +29,6 @@ class NextionComponentBase;
|
||||
|
||||
using nextion_writer_t = display::DisplayWriter<Nextion>;
|
||||
|
||||
static const std::string COMMAND_DELIMITER{static_cast<char>(255), static_cast<char>(255), static_cast<char>(255)};
|
||||
|
||||
#ifdef USE_NEXTION_COMMAND_SPACING
|
||||
class NextionCommandPacer {
|
||||
public:
|
||||
|
||||
@@ -10,24 +10,24 @@ StaticVector<Controller *, CONTROLLER_REGISTRY_MAX> ControllerRegistry::controll
|
||||
|
||||
void ControllerRegistry::register_controller(Controller *controller) { controllers.push_back(controller); }
|
||||
|
||||
void ControllerRegistry::notify(void *obj, DispatchFunc dispatch) {
|
||||
for (auto *controller : controllers) {
|
||||
dispatch(controller, obj);
|
||||
}
|
||||
}
|
||||
|
||||
// Macro for standard registry notification dispatch - calls on_<entity_name>_update()
|
||||
// Each wrapper passes a small trampoline lambda that calls the correct virtual method.
|
||||
// Each notify method directly iterates controllers and calls the virtual method.
|
||||
// This avoids the overhead of a shared noinline dispatch loop with function pointer
|
||||
// indirection. The loop is tiny (~20 bytes per entity type) so the flash cost of
|
||||
// duplicating it is negligible compared to eliminating two levels of indirection
|
||||
// (noinline call + function pointer) from every state publish.
|
||||
// NOLINTBEGIN(bugprone-macro-parentheses)
|
||||
#define CONTROLLER_REGISTRY_NOTIFY(entity_type, entity_name) \
|
||||
void ControllerRegistry::notify_##entity_name##_update(entity_type *obj) { \
|
||||
notify(obj, [](Controller *c, void *o) { c->on_##entity_name##_update(static_cast<entity_type *>(o)); }); \
|
||||
for (auto *controller : controllers) { \
|
||||
controller->on_##entity_name##_update(obj); \
|
||||
} \
|
||||
}
|
||||
|
||||
// Macro for entities where controller method has no "_update" suffix (Event, Update)
|
||||
#define CONTROLLER_REGISTRY_NOTIFY_NO_UPDATE_SUFFIX(entity_type, entity_name) \
|
||||
void ControllerRegistry::notify_##entity_name(entity_type *obj) { \
|
||||
notify(obj, [](Controller *c, void *o) { c->on_##entity_name(static_cast<entity_type *>(o)); }); \
|
||||
for (auto *controller : controllers) { \
|
||||
controller->on_##entity_name(obj); \
|
||||
} \
|
||||
}
|
||||
// NOLINTEND(bugprone-macro-parentheses)
|
||||
|
||||
|
||||
@@ -146,8 +146,8 @@ class UpdateEntity;
|
||||
* entities call ControllerRegistry::notify_*_update() which iterates the small list
|
||||
* of registered controllers (typically 2: API and WebServer).
|
||||
*
|
||||
* Controllers read state directly from entities using existing accessors (obj->state, etc.)
|
||||
* rather than receiving it as callback parameters that were being ignored anyway.
|
||||
* Each notify method directly iterates controllers and calls the virtual method,
|
||||
* avoiding function pointer indirection for minimal dispatch overhead.
|
||||
*
|
||||
* Memory savings: 32 bytes per entity (2 controllers × 16 bytes std::function overhead)
|
||||
* Typical config (25 entities): ~780 bytes saved
|
||||
@@ -247,21 +247,6 @@ class ControllerRegistry {
|
||||
#endif
|
||||
|
||||
protected:
|
||||
/** Type-erased dispatch function pointer.
|
||||
*
|
||||
* Each notify method passes a small trampoline that calls the
|
||||
* correct virtual method on Controller. The shared notify() loop
|
||||
* iterates controllers once, calling the trampoline for each.
|
||||
*/
|
||||
using DispatchFunc = void (*)(Controller *, void *);
|
||||
|
||||
/** Shared dispatch loop - iterates controllers and calls dispatch for each.
|
||||
*
|
||||
* Marked noinline to ensure only one copy of the loop exists in flash,
|
||||
* rather than being duplicated into each notify_*_update wrapper.
|
||||
*/
|
||||
static void __attribute__((noinline)) notify(void *obj, DispatchFunc dispatch);
|
||||
|
||||
static StaticVector<Controller *, CONTROLLER_REGISTRY_MAX> controllers;
|
||||
};
|
||||
|
||||
|
||||
@@ -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,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 == []
|
||||
Reference in New Issue
Block a user