Merge branch 'dev' of https://github.com/esphome/esphome into copilot/update-cover-component-triggers

This commit is contained in:
clydebarrow
2026-02-04 09:03:50 +11:00
245 changed files with 5632 additions and 2147 deletions
+36
View File
@@ -0,0 +1,36 @@
ch423:
- id: ch423_hub
i2c_id: i2c_bus
binary_sensor:
- platform: gpio
id: ch423_input
name: CH423 Binary Sensor
pin:
ch423: ch423_hub
number: 1
mode: INPUT
inverted: true
- platform: gpio
id: ch423_input_2
name: CH423 Binary Sensor 2
pin:
ch423: ch423_hub
number: 0
mode: INPUT
inverted: false
output:
- platform: gpio
id: ch423_out_11
pin:
ch423: ch423_hub
number: 11
mode: OUTPUT_OPEN_DRAIN
inverted: true
- platform: gpio
id: ch423_out_23
pin:
ch423: ch423_hub
number: 23
mode: OUTPUT_OPEN_DRAIN
inverted: false
@@ -0,0 +1,4 @@
packages:
i2c: !include ../../test_build_components/common/i2c/esp32-idf.yaml
<<: !include common.yaml
@@ -0,0 +1,4 @@
packages:
i2c: !include ../../test_build_components/common/i2c/esp8266-ard.yaml
<<: !include common.yaml
@@ -0,0 +1,4 @@
packages:
i2c: !include ../../test_build_components/common/i2c/rp2040-ard.yaml
<<: !include common.yaml
@@ -0,0 +1,11 @@
dlms_meter:
decryption_key: "36C66639E48A8CA4D6BC8B282A793BBB" # change this to your decryption key!
sensor:
- platform: dlms_meter
reactive_energy_plus:
name: "Reactive energy taken from grid"
reactive_energy_minus:
name: "Reactive energy put into grid"
<<: !include common.yaml
@@ -0,0 +1,17 @@
dlms_meter:
decryption_key: "36C66639E48A8CA4D6BC8B282A793BBB" # change this to your decryption key!
provider: netznoe # (optional) key - only set if using evn
sensor:
- platform: dlms_meter
# EVN
power_factor:
name: "Power Factor"
text_sensor:
- platform: dlms_meter
# EVN
meternumber:
name: "meterNumber"
<<: !include common.yaml
+27
View File
@@ -0,0 +1,27 @@
sensor:
- platform: dlms_meter
voltage_l1:
name: "Voltage L1"
voltage_l2:
name: "Voltage L2"
voltage_l3:
name: "Voltage L3"
current_l1:
name: "Current L1"
current_l2:
name: "Current L2"
current_l3:
name: "Current L3"
active_power_plus:
name: "Active power taken from grid"
active_power_minus:
name: "Active power put into grid"
active_energy_plus:
name: "Active energy taken from grid"
active_energy_minus:
name: "Active energy put into grid"
text_sensor:
- platform: dlms_meter
timestamp:
name: "timestamp"
@@ -0,0 +1,4 @@
packages:
uart_2400: !include ../../test_build_components/common/uart_2400/esp32-ard.yaml
<<: !include common-generic.yaml
@@ -0,0 +1,4 @@
packages:
uart_2400: !include ../../test_build_components/common/uart_2400/esp32-idf.yaml
<<: !include common-netznoe.yaml
@@ -0,0 +1,4 @@
packages:
uart_2400: !include ../../test_build_components/common/uart_2400/esp8266-ard.yaml
<<: !include common-generic.yaml
@@ -8,6 +8,16 @@ esp32:
enable_lwip_bridge_interface: true
disable_libc_locks_in_iram: false # Test explicit opt-out of RAM optimization
use_full_certificate_bundle: false # Test CMN bundle (default)
include_builtin_idf_components:
- freertos # Test escape hatch (freertos is always included anyway)
disable_debug_stubs: true
disable_ocd_aware: true
disable_usb_serial_jtag_secondary: true
disable_dev_null_vfs: true
disable_mbedtls_peer_cert: true
disable_mbedtls_pkcs7: true
disable_regi2c_in_iram: true
disable_fatfs: true
wifi:
ssid: MySSID
@@ -10,6 +10,14 @@ esp32:
ref: 2.7.0
advanced:
enable_idf_experimental_features: yes
disable_debug_stubs: true
disable_ocd_aware: true
disable_usb_serial_jtag_secondary: true
disable_dev_null_vfs: true
disable_mbedtls_peer_cert: true
disable_mbedtls_pkcs7: true
disable_regi2c_in_iram: true
disable_fatfs: true
ota:
platform: esphome
@@ -5,6 +5,14 @@ esp32:
advanced:
execute_from_psram: true
disable_libc_locks_in_iram: true # Test default RAM optimization enabled
disable_debug_stubs: true
disable_ocd_aware: true
disable_usb_serial_jtag_secondary: true
disable_dev_null_vfs: true
disable_mbedtls_peer_cert: true
disable_mbedtls_pkcs7: true
disable_regi2c_in_iram: true
disable_fatfs: true
psram:
mode: octal
@@ -13,3 +13,7 @@ ethernet:
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!"
@@ -13,3 +13,7 @@ ethernet:
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!"
@@ -13,3 +13,7 @@ ethernet:
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!"
@@ -13,3 +13,7 @@ ethernet:
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!"
@@ -13,3 +13,7 @@ ethernet:
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!"
@@ -13,3 +13,7 @@ ethernet:
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!"
@@ -12,3 +12,7 @@ ethernet:
gateway: 192.168.178.1
subnet: 255.255.255.0
domain: .local
on_connect:
- logger.log: "Ethernet connected!"
on_disconnect:
- logger.log: "Ethernet disconnected!"
@@ -13,3 +13,7 @@ ethernet:
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!"
@@ -1,2 +1,6 @@
ethernet:
type: OPENETH
on_connect:
- logger.log: "Ethernet connected!"
on_disconnect:
- logger.log: "Ethernet disconnected!"
@@ -13,3 +13,7 @@ ethernet:
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!"
@@ -13,3 +13,7 @@ ethernet:
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!"
+1
View File
@@ -10,6 +10,7 @@ globals:
type: int
restore_value: true
initial_value: "0"
update_interval: 5s
- id: glob_float
type: float
restore_value: true
@@ -18,14 +18,23 @@ key_collector:
- logger.log:
format: "input progress: '%s', started by '%c'"
args: ['x.c_str()', "(start == 0 ? '~' : start)"]
- logger.log:
format: "second listener - progress: '%s'"
args: ['x.c_str()']
on_result:
- logger.log:
format: "input result: '%s', started by '%c', ended by '%c'"
args: ['x.c_str()', "(start == 0 ? '~' : start)", "(end == 0 ? '~' : end)"]
- logger.log:
format: "second listener - result: '%s'"
args: ['x.c_str()']
on_timeout:
- logger.log:
format: "input timeout: '%s', started by '%c'"
args: ['x.c_str()', "(start == 0 ? '~' : start)"]
- logger.log:
format: "second listener - timeout: '%s'"
args: ['x.c_str()']
enable_on_boot: false
button:
@@ -34,3 +43,8 @@ button:
on_press:
- key_collector.enable:
- key_collector.disable:
text_sensor:
- platform: key_collector
id: collected_keys
source_id: reader
@@ -0,0 +1,10 @@
substitutions:
dc_pin: GPIO15
cs_pin: GPIO5
enable_pin: GPIO4
reset_pin: GPIO16
packages:
spi: !include ../../test_build_components/common/spi/esp8266-ard.yaml
<<: !include common.yaml
+3 -3
View File
@@ -8,11 +8,11 @@ sensor:
pm_10_0:
name: PM 10.0 Concentration
pm_1_0_std:
name: PM 1.0 Standard Atmospher Concentration
name: PM 1.0 Standard Atmospheric Concentration
pm_2_5_std:
name: PM 2.5 Standard Atmospher Concentration
name: PM 2.5 Standard Atmospheric Concentration
pm_10_0_std:
name: PM 10.0 Standard Atmospher Concentration
name: PM 10.0 Standard Atmospheric Concentration
pm_0_3um:
name: Particulate Count >0.3um
pm_0_5um:
@@ -334,6 +334,16 @@ select:
// Migration guide: Store in std::string
std::string stored_option(id(template_select).current_option());
ESP_LOGI("test", "Stored: %s", stored_option.c_str());
- platform: template
id: template_select_with_action
name: "Template select with action"
options:
- option_a
- option_b
set_action:
- logger.log:
format: "Selected: %s"
args: ["x.c_str()"]
lock:
- platform: template
+4
View File
@@ -26,3 +26,7 @@ wifi:
- ssid: MySSID3
password: password3
priority: 0
on_connect:
- logger.log: "WiFi connected!"
on_disconnect:
- logger.log: "WiFi disconnected!"
+8 -2
View File
@@ -6,8 +6,6 @@ binary_sensor:
name: "Garage Door Open 2"
- platform: template
name: "Garage Door Open 3"
- platform: template
name: "Garage Door Open 4"
- platform: template
name: "Garage Door Internal"
internal: True
@@ -44,3 +42,11 @@ switch:
- platform: template
name: "Template Switch"
optimistic: true
number:
- platform: template
name: "Template number"
optimistic: true
min_value: 2
max_value: 100
step: 1
@@ -20,6 +20,9 @@ globals:
- id: retry_counter
type: int
initial_value: '0'
- id: defer_counter
type: int
initial_value: '0'
- id: tests_done
type: bool
initial_value: 'false'
@@ -136,11 +139,49 @@ script:
App.scheduler.cancel_retry(component1, 6002U);
ESP_LOGI("test", "Cancelled numeric retry 6002");
// Test 12: defer with numeric ID (Component method)
class TestDeferComponent : public Component {
public:
void test_defer_methods() {
// Test defer with uint32_t ID - should execute on next loop
this->defer(7001U, []() {
ESP_LOGI("test", "Component numeric defer 7001 fired");
id(defer_counter) += 1;
});
// Test another defer with numeric ID
this->defer(7002U, []() {
ESP_LOGI("test", "Component numeric defer 7002 fired");
id(defer_counter) += 1;
});
}
};
static TestDeferComponent test_defer_component;
test_defer_component.test_defer_methods();
// Test 13: cancel_defer with numeric ID (Component method)
class TestCancelDeferComponent : public Component {
public:
void test_cancel_defer() {
// Set a defer that should be cancelled
this->defer(8001U, []() {
ESP_LOGE("test", "ERROR: Numeric defer 8001 should have been cancelled");
});
// Cancel it immediately
bool cancelled = this->cancel_defer(8001U);
ESP_LOGI("test", "Cancelled numeric defer 8001: %s", cancelled ? "true" : "false");
}
};
static TestCancelDeferComponent test_cancel_defer_component;
test_cancel_defer_component.test_cancel_defer();
- id: report_results
then:
- lambda: |-
ESP_LOGI("test", "Final results - Timeouts: %d, Intervals: %d, Retries: %d",
id(timeout_counter), id(interval_counter), id(retry_counter));
ESP_LOGI("test", "Final results - Timeouts: %d, Intervals: %d, Retries: %d, Defers: %d",
id(timeout_counter), id(interval_counter), id(retry_counter), id(defer_counter));
sensor:
- platform: template
@@ -56,7 +56,21 @@ select:
std::string prefix = x.substr(0, 6);
ESP_LOGI("test", "Substr prefix: %s", prefix.c_str());
# Second select with numeric options to test ADL functions
# Second select with set_action trigger (uses TemplateSelectWithSetAction subclass)
- platform: template
name: "Action Select"
id: action_select
options:
- "Action A"
- "Action B"
set_action:
then:
# Test: set_action trigger receives StringRef
- logger.log:
format: "set_action triggered: %s"
args: ['x.c_str()']
# Third select with numeric options to test ADL functions
- platform: template
name: "Baud Rate"
id: baud_select
@@ -19,6 +19,7 @@ async def test_scheduler_numeric_id_test(
timeout_count = 0
interval_count = 0
retry_count = 0
defer_count = 0
# Events for each test completion
numeric_timeout_1001_fired = asyncio.Event()
@@ -33,6 +34,9 @@ async def test_scheduler_numeric_id_test(
max_id_timeout_fired = asyncio.Event()
numeric_retry_done = asyncio.Event()
numeric_retry_cancelled = asyncio.Event()
numeric_defer_7001_fired = asyncio.Event()
numeric_defer_7002_fired = asyncio.Event()
numeric_defer_cancelled = asyncio.Event()
final_results_logged = asyncio.Event()
# Track interval counts
@@ -40,7 +44,7 @@ async def test_scheduler_numeric_id_test(
numeric_retry_count = 0
def on_log_line(line: str) -> None:
nonlocal timeout_count, interval_count, retry_count
nonlocal timeout_count, interval_count, retry_count, defer_count
nonlocal numeric_interval_count, numeric_retry_count
# Strip ANSI color codes
@@ -105,15 +109,27 @@ async def test_scheduler_numeric_id_test(
elif "Cancelled numeric retry 6002" in clean_line:
numeric_retry_cancelled.set()
# Check for numeric defer tests
elif "Component numeric defer 7001 fired" in clean_line:
numeric_defer_7001_fired.set()
elif "Component numeric defer 7002 fired" in clean_line:
numeric_defer_7002_fired.set()
elif "Cancelled numeric defer 8001: true" in clean_line:
numeric_defer_cancelled.set()
# Check for final results
elif "Final results" in clean_line:
match = re.search(
r"Timeouts: (\d+), Intervals: (\d+), Retries: (\d+)", clean_line
r"Timeouts: (\d+), Intervals: (\d+), Retries: (\d+), Defers: (\d+)",
clean_line,
)
if match:
timeout_count = int(match.group(1))
interval_count = int(match.group(2))
retry_count = int(match.group(3))
defer_count = int(match.group(4))
final_results_logged.set()
async with (
@@ -201,6 +217,23 @@ async def test_scheduler_numeric_id_test(
"Numeric retry 6002 should have been cancelled"
)
# Wait for numeric defer tests
try:
await asyncio.wait_for(numeric_defer_7001_fired.wait(), timeout=0.5)
except TimeoutError:
pytest.fail("Numeric defer 7001 did not fire within 0.5 seconds")
try:
await asyncio.wait_for(numeric_defer_7002_fired.wait(), timeout=0.5)
except TimeoutError:
pytest.fail("Numeric defer 7002 did not fire within 0.5 seconds")
# Verify numeric defer was cancelled
try:
await asyncio.wait_for(numeric_defer_cancelled.wait(), timeout=0.5)
except TimeoutError:
pytest.fail("Numeric defer 8001 cancel confirmation not received")
# Wait for final results
try:
await asyncio.wait_for(final_results_logged.wait(), timeout=3.0)
@@ -215,3 +248,4 @@ async def test_scheduler_numeric_id_test(
assert retry_count >= 2, (
f"Expected at least 2 retry attempts, got {retry_count}"
)
assert defer_count >= 2, f"Expected at least 2 defer fires, got {defer_count}"
@@ -28,6 +28,8 @@ async def test_select_stringref_trigger(
find_substr_future = loop.create_future()
find_char_future = loop.create_future()
substr_future = loop.create_future()
# set_action trigger (TemplateSelectWithSetAction subclass)
set_action_future = loop.create_future()
# ADL functions
stoi_future = loop.create_future()
stol_future = loop.create_future()
@@ -43,6 +45,8 @@ async def test_select_stringref_trigger(
find_substr_pattern = re.compile(r"Found 'Option' in value")
find_char_pattern = re.compile(r"Space at position: 6") # space at index 6
substr_pattern = re.compile(r"Substr prefix: Option")
# set_action trigger pattern (TemplateSelectWithSetAction subclass)
set_action_pattern = re.compile(r"set_action triggered: Action B")
# ADL function patterns (115200 from baud rate select)
stoi_pattern = re.compile(r"stoi result: 115200")
stol_pattern = re.compile(r"stol result: 115200")
@@ -67,6 +71,9 @@ async def test_select_stringref_trigger(
find_char_future.set_result(True)
if not substr_future.done() and substr_pattern.search(line):
substr_future.set_result(True)
# set_action trigger
if not set_action_future.done() and set_action_pattern.search(line):
set_action_future.set_result(True)
# ADL functions
if not stoi_future.done() and stoi_pattern.search(line):
stoi_future.set_result(True)
@@ -89,22 +96,21 @@ async def test_select_stringref_trigger(
# List entities to find our select
entities, _ = await client.list_entities_services()
select_entity = next(
(e for e in entities if hasattr(e, "options") and e.name == "Test Select"),
None,
)
select_entity = next((e for e in entities if e.name == "Test Select"), None)
assert select_entity is not None, "Test Select entity not found"
baud_entity = next(
(e for e in entities if hasattr(e, "options") and e.name == "Baud Rate"),
None,
)
baud_entity = next((e for e in entities if e.name == "Baud Rate"), None)
assert baud_entity is not None, "Baud Rate entity not found"
action_entity = next((e for e in entities if e.name == "Action Select"), None)
assert action_entity is not None, "Action Select entity not found"
# Change select to Option B - this should trigger on_value with StringRef
client.select_command(select_entity.key, "Option B")
# Change baud to 115200 - this tests ADL functions (stoi, stol, stof, stod)
client.select_command(baud_entity.key, "115200")
# Change action select - tests set_action trigger (TemplateSelectWithSetAction)
client.select_command(action_entity.key, "Action B")
# Wait for all log messages confirming StringRef operations work
try:
@@ -118,6 +124,7 @@ async def test_select_stringref_trigger(
find_substr_future,
find_char_future,
substr_future,
set_action_future,
stoi_future,
stol_future,
stof_future,
@@ -135,6 +142,7 @@ async def test_select_stringref_trigger(
"find_substr": find_substr_future.done(),
"find_char": find_char_future.done(),
"substr": substr_future.done(),
"set_action": set_action_future.done(),
"stoi": stoi_future.done(),
"stol": stol_future.done(),
"stof": stof_future.done(),
@@ -0,0 +1,11 @@
# Common UART configuration for ESP32 Arduino tests - 2400 baud
substitutions:
tx_pin: GPIO17
rx_pin: GPIO16
uart:
- id: uart_bus
tx_pin: ${tx_pin}
rx_pin: ${rx_pin}
baud_rate: 2400
@@ -0,0 +1,11 @@
# Common UART configuration for ESP32 IDF tests - 2400 baud
substitutions:
tx_pin: GPIO17
rx_pin: GPIO16
uart:
- id: uart_bus
tx_pin: ${tx_pin}
rx_pin: ${rx_pin}
baud_rate: 2400
@@ -0,0 +1,11 @@
# Common UART configuration for ESP8266 Arduino tests - 2400 baud
substitutions:
tx_pin: GPIO4
rx_pin: GPIO5
uart:
- id: uart_bus
tx_pin: ${tx_pin}
rx_pin: ${rx_pin}
baud_rate: 2400
+58
View File
@@ -0,0 +1,58 @@
"""Tests for ch423 component validation."""
from unittest.mock import patch
from esphome import config, yaml_util
from esphome.core import CORE
def test_ch423_mixed_gpio_modes_fails(tmp_path, capsys):
"""Test that mixing input/output on GPIO pins 0-7 fails validation."""
test_file = tmp_path / "test.yaml"
test_file.write_text("""
esphome:
name: test
esp8266:
board: esp01_1m
i2c:
sda: GPIO4
scl: GPIO5
ch423:
- id: ch423_hub
binary_sensor:
- platform: gpio
name: "CH423 Input 0"
pin:
ch423: ch423_hub
number: 0
mode: input
switch:
- platform: gpio
name: "CH423 Output 1"
pin:
ch423: ch423_hub
number: 1
mode: output
""")
parsed_yaml = yaml_util.load_yaml(test_file)
with (
patch.object(yaml_util, "load_yaml", return_value=parsed_yaml),
patch.object(CORE, "config_path", test_file),
):
result = config.read_config({})
assert result is None, "Expected validation to fail with mixed GPIO modes"
# Check that the error message mentions the GPIO pin restriction
captured = capsys.readouterr()
assert (
"GPIO pins (0-7) must all be configured as input or all as output"
in captured.out
)
+277
View File
@@ -347,3 +347,280 @@ class TestMockObj:
assert isinstance(actual, cg.MockObj)
assert actual.base == "foo.eek"
assert actual.op == "."
class TestStatementFunction:
"""Tests for the statement() function."""
def test_statement__expression_converted_to_statement(self):
"""Test that expressions are converted to ExpressionStatement."""
expr = cg.RawExpression("foo()")
result = cg.statement(expr)
assert isinstance(result, cg.ExpressionStatement)
assert str(result) == "foo();"
def test_statement__statement_unchanged(self):
"""Test that statements are returned unchanged."""
stmt = cg.RawStatement("foo()")
result = cg.statement(stmt)
assert result is stmt
assert str(result) == "foo()"
def test_statement__expression_statement_unchanged(self):
"""Test that ExpressionStatement is returned unchanged."""
stmt = cg.ExpressionStatement(42)
result = cg.statement(stmt)
assert result is stmt
assert str(result) == "42;"
def test_statement__line_comment_unchanged(self):
"""Test that LineComment is returned unchanged."""
stmt = cg.LineComment("This is a comment")
result = cg.statement(stmt)
assert result is stmt
assert str(result) == "// This is a comment"
class TestLiteralFunction:
"""Tests for the literal() function."""
def test_literal__creates_mockobj(self):
"""Test that literal() creates a MockObj."""
result = cg.literal("MY_CONSTANT")
assert isinstance(result, cg.MockObj)
assert result.base == "MY_CONSTANT"
assert result.op == ""
def test_literal__string_representation(self):
"""Test that literal names appear unquoted in generated code."""
result = cg.literal("nullptr")
assert str(result) == "nullptr"
def test_literal__can_be_used_in_expressions(self):
"""Test that literals can be used as part of larger expressions."""
null_lit = cg.literal("nullptr")
expr = cg.CallExpression(cg.RawExpression("my_func"), null_lit)
assert str(expr) == "my_func(nullptr)"
def test_literal__common_cpp_literals(self):
"""Test common C++ literal values."""
test_cases = [
("nullptr", "nullptr"),
("true", "true"),
("false", "false"),
("NULL", "NULL"),
("NAN", "NAN"),
]
for name, expected in test_cases:
result = cg.literal(name)
assert str(result) == expected
class TestLambdaConstructor:
"""Tests for the Lambda class constructor in core/__init__.py."""
def test_lambda__from_string(self):
"""Test Lambda constructor with string argument."""
from esphome.core import Lambda
lambda_obj = Lambda("return x + 1;")
assert lambda_obj.value == "return x + 1;"
assert str(lambda_obj) == "return x + 1;"
def test_lambda__from_expression(self):
"""Test Lambda constructor with Expression argument."""
from esphome.core import Lambda
expr = cg.RawExpression("x + 1")
lambda_obj = Lambda(expr)
# Expression should be converted to statement (with semicolon)
assert lambda_obj.value == "x + 1;"
def test_lambda__from_lambda(self):
"""Test Lambda constructor with another Lambda argument."""
from esphome.core import Lambda
original = Lambda("return x + 1;")
copy = Lambda(original)
assert copy.value == original.value
assert copy.value == "return x + 1;"
def test_lambda__parts_parsing(self):
"""Test that Lambda correctly parses parts with id() references."""
from esphome.core import Lambda
lambda_obj = Lambda("return id(my_sensor).state;")
parts = lambda_obj.parts
# Parts should be split by LAMBDA_PROG regex: text, id, op, text
assert len(parts) == 4
assert parts[0] == "return "
assert parts[1] == "my_sensor"
assert parts[2] == "."
assert parts[3] == "state;"
def test_lambda__requires_ids(self):
"""Test that Lambda correctly extracts required IDs."""
from esphome.core import ID, Lambda
lambda_obj = Lambda("return id(sensor1).state + id(sensor2).value;")
ids = lambda_obj.requires_ids
assert len(ids) == 2
assert all(isinstance(id_obj, ID) for id_obj in ids)
assert ids[0].id == "sensor1"
assert ids[1].id == "sensor2"
def test_lambda__no_ids(self):
"""Test Lambda with no id() references."""
from esphome.core import Lambda
lambda_obj = Lambda("return 42;")
ids = lambda_obj.requires_ids
assert len(ids) == 0
def test_lambda__comment_removal(self):
"""Test that comments are removed when parsing parts."""
from esphome.core import Lambda
lambda_obj = Lambda("return id(sensor).state; // Get sensor state")
parts = lambda_obj.parts
# Comment should be replaced with space, not affect parsing
assert "my_sensor" not in str(parts)
def test_lambda__multiline_string(self):
"""Test Lambda with multiline string."""
from esphome.core import Lambda
code = """if (id(sensor).state > 0) {
return true;
}
return false;"""
lambda_obj = Lambda(code)
assert lambda_obj.value == code
assert "sensor" in [id_obj.id for id_obj in lambda_obj.requires_ids]
@pytest.mark.asyncio
class TestProcessLambda:
"""Tests for the process_lambda() async function."""
async def test_process_lambda__none_value(self):
"""Test that None returns None."""
result = await cg.process_lambda(None, [])
assert result is None
async def test_process_lambda__with_expression(self):
"""Test process_lambda with Expression argument."""
expr = cg.RawExpression("return x + 1")
result = await cg.process_lambda(expr, [(int, "x")])
assert isinstance(result, cg.LambdaExpression)
assert "x + 1" in str(result)
async def test_process_lambda__simple_lambda_no_ids(self):
"""Test process_lambda with simple Lambda without id() references."""
from esphome.core import Lambda
lambda_obj = Lambda("return x + 1;")
result = await cg.process_lambda(lambda_obj, [(int, "x")])
assert isinstance(result, cg.LambdaExpression)
# Should have parameter
lambda_str = str(result)
assert "int32_t x" in lambda_str
assert "return x + 1;" in lambda_str
async def test_process_lambda__with_return_type(self):
"""Test process_lambda with return type specified."""
from esphome.core import Lambda
lambda_obj = Lambda("return x > 0;")
result = await cg.process_lambda(lambda_obj, [(int, "x")], return_type=bool)
assert isinstance(result, cg.LambdaExpression)
lambda_str = str(result)
assert "-> bool" in lambda_str
async def test_process_lambda__with_capture(self):
"""Test process_lambda with capture specified."""
from esphome.core import Lambda
lambda_obj = Lambda("return captured + x;")
result = await cg.process_lambda(lambda_obj, [(int, "x")], capture="captured")
assert isinstance(result, cg.LambdaExpression)
lambda_str = str(result)
assert "[captured]" in lambda_str
async def test_process_lambda__empty_capture(self):
"""Test process_lambda with empty capture (stateless lambda)."""
from esphome.core import Lambda
lambda_obj = Lambda("return x + 1;")
result = await cg.process_lambda(lambda_obj, [(int, "x")], capture="")
assert isinstance(result, cg.LambdaExpression)
lambda_str = str(result)
assert "[]" in lambda_str
async def test_process_lambda__no_parameters(self):
"""Test process_lambda with no parameters."""
from esphome.core import Lambda
lambda_obj = Lambda("return 42;")
result = await cg.process_lambda(lambda_obj, [])
assert isinstance(result, cg.LambdaExpression)
lambda_str = str(result)
# Should have empty parameter list
assert "()" in lambda_str
async def test_process_lambda__multiple_parameters(self):
"""Test process_lambda with multiple parameters."""
from esphome.core import Lambda
lambda_obj = Lambda("return x + y + z;")
result = await cg.process_lambda(
lambda_obj, [(int, "x"), (float, "y"), (bool, "z")]
)
assert isinstance(result, cg.LambdaExpression)
lambda_str = str(result)
assert "int32_t x" in lambda_str
assert "float y" in lambda_str
assert "bool z" in lambda_str
async def test_process_lambda__parameter_validation(self):
"""Test that malformed parameters raise assertion error."""
from esphome.core import Lambda
lambda_obj = Lambda("return x;")
# Test invalid parameter format (not list of tuples)
with pytest.raises(AssertionError):
await cg.process_lambda(lambda_obj, "invalid")
# Test invalid tuple format (not 2-element tuples)
with pytest.raises(AssertionError):
await cg.process_lambda(lambda_obj, [(int, "x", "extra")])
# Test invalid tuple format (single element)
with pytest.raises(AssertionError):
await cg.process_lambda(lambda_obj, [(int,)])
+80 -4
View File
@@ -25,7 +25,6 @@ def default_config() -> dict[str, Any]:
"board": "esp01_1m",
"ssid": "test_ssid",
"psk": "test_psk",
"password": "",
}
@@ -37,7 +36,7 @@ def wizard_answers() -> list[str]:
"nodemcuv2", # board
"SSID", # ssid
"psk", # wifi password
"ota_pass", # ota password
"", # ota password (empty for no password)
]
@@ -105,16 +104,35 @@ def test_config_file_should_include_ota_when_password_set(
default_config: dict[str, Any],
):
"""
The Over-The-Air update should be enabled when a password is set
The Over-The-Air update should be enabled when an OTA password is set
"""
# Given
default_config["password"] = "foo"
default_config["ota_password"] = "foo"
# When
config = wz.wizard_file(**default_config)
# Then
assert "ota:" in config
assert 'password: "foo"' in config
def test_config_file_should_include_api_encryption_key(
default_config: dict[str, Any],
):
"""
The API encryption key should be included when set
"""
# Given
default_config["api_encryption_key"] = "test_encryption_key_base64=="
# When
config = wz.wizard_file(**default_config)
# Then
assert "api:" in config
assert "encryption:" in config
assert 'key: "test_encryption_key_base64=="' in config
def test_wizard_write_sets_platform(
@@ -556,3 +574,61 @@ def test_wizard_write_protects_existing_config(
# Then
assert result is False # Should return False when file exists
assert config_file.read_text() == original_content
def test_wizard_accepts_ota_password(
tmp_path: Path, monkeypatch: MonkeyPatch, wizard_answers: list[str]
):
"""
The wizard should pass ota_password to wizard_write when the user provides one
"""
# Given
wizard_answers[5] = "my_ota_password" # Set OTA password
config_file = tmp_path / "test.yaml"
input_mock = MagicMock(side_effect=wizard_answers)
monkeypatch.setattr("builtins.input", input_mock)
monkeypatch.setattr(wz, "safe_print", lambda t=None, end=None: 0)
monkeypatch.setattr(wz, "sleep", lambda _: 0)
wizard_write_mock = MagicMock(return_value=True)
monkeypatch.setattr(wz, "wizard_write", wizard_write_mock)
# When
retval = wz.wizard(config_file)
# Then
assert retval == 0
call_kwargs = wizard_write_mock.call_args.kwargs
assert "ota_password" in call_kwargs
assert call_kwargs["ota_password"] == "my_ota_password"
def test_wizard_accepts_rpipico_board(tmp_path: Path, monkeypatch: MonkeyPatch):
"""
The wizard should handle rpipico board which doesn't support WiFi.
This tests the branch where api_encryption_key is None.
"""
# Given
wizard_answers_rp2040 = [
"test-node", # Name of the node
"RP2040", # platform
"rpipico", # board (no WiFi support)
]
config_file = tmp_path / "test.yaml"
input_mock = MagicMock(side_effect=wizard_answers_rp2040)
monkeypatch.setattr("builtins.input", input_mock)
monkeypatch.setattr(wz, "safe_print", lambda t=None, end=None: 0)
monkeypatch.setattr(wz, "sleep", lambda _: 0)
wizard_write_mock = MagicMock(return_value=True)
monkeypatch.setattr(wz, "wizard_write", wizard_write_mock)
# When
retval = wz.wizard(config_file)
# Then
assert retval == 0
call_kwargs = wizard_write_mock.call_args.kwargs
# rpipico doesn't support WiFi, so no api_encryption_key or ota_password
assert "api_encryption_key" not in call_kwargs
assert "ota_password" not in call_kwargs