Merge remote-tracking branch 'origin/dev' into web-server-offline-hint

# Conflicts:
#	esphome/components/wifi/wifi_component.cpp
#	esphome/components/wifi/wifi_component.h
This commit is contained in:
J. Nick Koston
2026-09-16 09:31:54 -05:00
228 changed files with 3076 additions and 720 deletions
@@ -0,0 +1,12 @@
---
esphome:
name: test
esp32:
board: esp32dev
wifi:
ssid: test
password: testtest
api:
@@ -0,0 +1,15 @@
---
esphome:
name: test
esp32:
board: esp32dev
wifi:
ssid: test
password: testtest
api:
port: 6054
reboot_timeout: 0s
batch_delay: 0ms
@@ -0,0 +1,15 @@
---
esphome:
name: test
esp32:
board: esp32dev
wifi:
ssid: test
password: testtest
api:
port: 6053
reboot_timeout: 15min
batch_delay: 100ms
@@ -0,0 +1,35 @@
"""Tests that the api component only emits setters for non default values."""
from collections.abc import Callable
from pathlib import Path
import pytest
@pytest.mark.parametrize("config_file", ["bare.yaml", "defaults.yaml"])
def test_default_values_are_not_emitted(
generate_main: Callable[[str | Path], str],
component_config_path: Callable[[str], Path],
config_file: str,
) -> None:
"""Port 6053, a 15 min reboot timeout and 100 ms batch delay are C++ initializers.
Both the schema defaults and the same values written explicitly take the skip path.
"""
main_cpp = generate_main(component_config_path(config_file))
assert "api_apiserver_id->set_port(" not in main_cpp
assert "api_apiserver_id->set_reboot_timeout(" not in main_cpp
assert "api_apiserver_id->set_batch_delay(" not in main_cpp
def test_custom_values_are_emitted(
generate_main: Callable[[str | Path], str],
component_config_path: Callable[[str], Path],
) -> None:
"""Non default values still reach the C++ setters."""
main_cpp = generate_main(component_config_path("custom.yaml"))
assert "api_apiserver_id->set_port(6054);" in main_cpp
assert "api_apiserver_id->set_reboot_timeout(0);" in main_cpp
assert "api_apiserver_id->set_batch_delay(0);" in main_cpp
@@ -0,0 +1,12 @@
esphome:
name: bk-power-save
bk72xx:
board: cb2s
wifi:
ssid: test
password: testtest
power_save_mode: high
bk72xx_ble:
@@ -0,0 +1,20 @@
"""bk72xx_ble keeps WiFi power save off: the Beken SDK's MCU sleep does not
wake up once the station is stopped while the BLE controller runs."""
from collections.abc import Callable
from pathlib import Path
import pytest
def test_power_save_mode_is_not_applied_with_ble(
generate_main: Callable[[str | Path], str],
component_config_path: Callable[[str], Path],
caplog: pytest.LogCaptureFixture,
) -> None:
main_cpp = generate_main(component_config_path("test_power_save.yaml"))
assert "bk72xx_ble::BK72xxBLE" in main_cpp
assert "set_power_save_mode(" not in main_cpp
assert "power_save_mode HIGH is not applied" in caplog.text
assert "issues/18592" in caplog.text
@@ -9,5 +9,5 @@ wifi:
password: password1
# esp32_ble_server is only auto-loaded here, so it has no services of its own.
esp32_improv:
improv_ble:
authorizer: none
@@ -55,7 +55,7 @@ def test_uuid_is_matches_descriptor_short_strings(uuid16) -> None:
@pytest.mark.parametrize(
("config_file", "required"),
[
# Auto-loaded by esp32_improv only: nothing to find until Improv asks for it
# Auto-loaded by improv_ble only: nothing to find until Improv asks for it
("improv_only.yaml", False),
# The configuration defines a service clients are meant to connect to
("own_service.yaml", True),
@@ -0,0 +1,31 @@
esphome:
name: improv-ble-automations
esp32:
variant: esp32
framework:
type: esp-idf
logger:
wifi:
ssid: MySSID
password: password1
binary_sensor:
- platform: gpio
pin: 0
id: io0_button
output:
- platform: gpio
pin: 2
id: built_in_led
improv_ble:
authorizer: io0_button
status_indicator: built_in_led
on_provisioned:
- logger.log: provisioned
on_provisioning:
- logger.log: provisioning
on_start:
- logger.log: start
on_state:
- logger.log: state
on_stop:
- logger.log: stop
@@ -0,0 +1,12 @@
esphome:
name: improv-ble-esp32
esp32:
variant: esp32
framework:
type: esp-idf
logger:
wifi:
ssid: MySSID
password: password1
improv_ble:
authorizer: none
@@ -0,0 +1,10 @@
esphome:
name: improv-ble-esp8266
esp8266:
board: nodemcuv2
logger:
wifi:
ssid: MySSID
password: password1
improv_ble:
authorizer: none
@@ -0,0 +1,12 @@
esphome:
name: improv-ble-legacy-key
esp32:
variant: esp32
framework:
type: esp-idf
logger:
wifi:
ssid: MySSID
password: password1
esp32_improv:
authorizer: none
@@ -0,0 +1,59 @@
"""improv_ble is platform neutral; only its BLE server backends are not.
Covers the platform gate (BLE_SERVER_BACKENDS) and the esp32_improv alias that
keeps pre-rename configurations working.
"""
from collections.abc import Callable
from pathlib import Path
import pytest
from esphome.config import read_config
from esphome.core import CORE
def test_esp32_generates_component(
generate_main: Callable[[str | Path], str],
component_config_path: Callable[[str], Path],
) -> None:
main_cpp = generate_main(component_config_path("esp32.yaml"))
assert "improv_ble::ImprovBLEComponent" in main_cpp
def test_legacy_key_routes_to_improv_ble(
generate_main: Callable[[str | Path], str],
component_config_path: Callable[[str], Path],
caplog: pytest.LogCaptureFixture,
) -> None:
main_cpp = generate_main(component_config_path("legacy_key.yaml"))
assert "improv_ble::ImprovBLEComponent" in main_cpp
assert "'esp32_improv:' top-level key is deprecated" in caplog.text
def test_platform_without_ble_server_rejected(
component_config_path: Callable[[str], Path],
capsys: pytest.CaptureFixture[str],
) -> None:
# AUTO_LOAD finds no backend for esp8266 and pulls in improv_base only, so
# the platform gate in CONFIG_SCHEMA is what has to reject the config.
CORE.config_path = component_config_path("esp8266.yaml")
assert read_config({}) is None
assert "only available on" in capsys.readouterr().out
def test_automations_emit_renamed_triggers(
generate_main: Callable[[str | Path], str],
component_config_path: Callable[[str], Path],
) -> None:
main_cpp = generate_main(component_config_path("automations.yaml"))
for trigger in (
"ImprovBLEProvisionedTrigger",
"ImprovBLEProvisioningTrigger",
"ImprovBLEStartTrigger",
"ImprovBLEStateTrigger",
"ImprovBLEStoppedTrigger",
):
assert f"improv_ble::{trigger}" in main_cpp
assert "set_authorizer" in main_cpp
assert "set_status_indicator" in main_cpp
@@ -0,0 +1,70 @@
"""request_service_enable_disable() only opts in on platforms whose mDNS stack
can add and remove services after setup, and tells the caller so."""
import pytest
from esphome.components import mdns
from esphome.const import CONF_DISABLED, PlatformFramework
from esphome.core import CORE
from tests.component_tests.types import SetCoreConfigCallable
DEFINE = "USE_MDNS_SUPPORTS_ENABLE_DISABLE"
def _defines() -> set[str]:
return {define.name for define in CORE.defines}
def _set_config(
set_core_config: SetCoreConfigCallable,
platform_framework: PlatformFramework,
config: dict,
) -> None:
set_core_config(platform_framework)
CORE.config = config
@pytest.mark.parametrize(
"platform_framework",
[PlatformFramework.ESP32_IDF, PlatformFramework.ESP32_ARDUINO],
)
def test_esp32_adds_define_and_keeps_services_stored(
set_core_config: SetCoreConfigCallable, platform_framework: PlatformFramework
) -> None:
_set_config(set_core_config, platform_framework, {"mdns": {CONF_DISABLED: False}})
assert mdns.request_service_enable_disable() is True
# Disabled services must stay stored so they can be re-registered later.
assert {DEFINE, "USE_MDNS_STORE_SERVICES"} <= _defines()
@pytest.mark.parametrize(
"platform_framework",
[PlatformFramework.ESP8266_ARDUINO, PlatformFramework.RP2_ARDUINO],
)
def test_other_platforms_return_false(
set_core_config: SetCoreConfigCallable, platform_framework: PlatformFramework
) -> None:
_set_config(set_core_config, platform_framework, {"mdns": {CONF_DISABLED: False}})
assert mdns.request_service_enable_disable() is False
assert DEFINE not in _defines()
@pytest.mark.parametrize(
"config",
[
pytest.param({}, id="no_mdns"),
pytest.param({"mdns": {CONF_DISABLED: True}}, id="mdns_disabled"),
pytest.param(
{"mdns": {CONF_DISABLED: False}, "openthread": {}}, id="openthread"
),
],
)
def test_esp32_returns_false_when_services_cannot_be_toggled(
set_core_config: SetCoreConfigCallable, config: dict
) -> None:
_set_config(set_core_config, PlatformFramework.ESP32_IDF, config)
assert mdns.request_service_enable_disable() is False
assert DEFINE not in _defines()
+1 -2
View File
@@ -33,7 +33,6 @@ def test_server_schema_rejects_address_zero() -> None:
def test_client_schema_still_accepts_address_zero() -> None:
# Not rejected for clients today, but not supported either: a client broadcast gets no reply and
# stalls the hub for the full send-wait.
# A client may address 0: writes are broadcast, and reads are allowed with allow_broadcast_read.
schema = modbus.modbus_device_schema(0x01)
assert schema({CONF_MODBUS_ID: "hub", CONF_ADDRESS: 0})[CONF_ADDRESS] == 0
@@ -7,7 +7,7 @@ guard is a safety property: these tests pin it to every handler slot.
import pytest
from esphome import config_validation as cv
from esphome.components import modbus_client
from esphome.components import modbus, modbus_client
from esphome.components.modbus_client import (
CONF_ON_NO_RESPONSE,
CONF_ON_NOT_SENT,
@@ -126,7 +126,7 @@ def test_on_no_response_retry_lambda_accepted() -> None:
def test_continuous_on_write_pdu_rejected() -> None:
"""A literal write-code PDU with continuous: true is rejected at config time (reads only)."""
with pytest.raises(cv.Invalid, match="does not apply to a write PDU"):
with pytest.raises(cv.Invalid, match="does not apply to function code"):
MODBUS_CLIENT_SEND_SCHEMA(
{
CONF_ADDRESS: 0x01,
@@ -185,3 +185,145 @@ def test_multi_conf_no_default_is_set() -> None:
"""
assert modbus_client.MULTI_CONF is True
assert modbus_client.MULTI_CONF_NO_DEFAULT is True
@pytest.mark.parametrize("key", [CONF_CONTINUOUS, modbus.CONF_ALLOW_BROADCAST_READ])
def test_send_rejects_read_option_on_static_write_pdu(key: str) -> None:
# A read option set true on a static write PDU is refused at validation, naming the key.
config = {
CONF_ADDRESS: 1,
CONF_PDU: [0x06, 0x00, 0x10, 0x00, 0x01],
key: True,
}
with pytest.raises(
cv.Invalid, match=f"'{key}: true' does not apply to function code"
):
MODBUS_CLIENT_SEND_SCHEMA(config)
def test_send_accepts_allow_broadcast_read_on_read_pdu() -> None:
# allow_broadcast_read defaults to False and is accepted on a read PDU to address 0.
config = MODBUS_CLIENT_SEND_SCHEMA(
{CONF_ADDRESS: 0, CONF_PDU: [0x03, 0x00, 0x10, 0x00, 0x02]}
)
assert config[modbus.CONF_ALLOW_BROADCAST_READ] is False
config = MODBUS_CLIENT_SEND_SCHEMA(
{
CONF_ADDRESS: 0,
CONF_PDU: [0x03, 0x00, 0x10, 0x00, 0x02],
modbus.CONF_ALLOW_BROADCAST_READ: True,
}
)
assert config[modbus.CONF_ALLOW_BROADCAST_READ] is True
def test_send_rejects_write_option_on_static_read_pdu() -> None:
# The write-side option is refused on a static read PDU, the mirror of the read-option check.
key = modbus.CONF_EXPECT_BROADCAST_WRITE_RESPONSE
with pytest.raises(
cv.Invalid, match=f"'{key}: true' does not apply to function code"
):
MODBUS_CLIENT_SEND_SCHEMA(
{CONF_ADDRESS: 0, CONF_PDU: [0x03, 0x00, 0x10, 0x00, 0x02], key: True}
)
def test_send_accepts_write_option_on_static_write_pdu() -> None:
config = MODBUS_CLIENT_SEND_SCHEMA(
{
CONF_ADDRESS: 0,
CONF_PDU: [0x06, 0x00, 0x10, 0x00, 0x01],
modbus.CONF_EXPECT_BROADCAST_WRITE_RESPONSE: True,
}
)
assert config[modbus.CONF_EXPECT_BROADCAST_WRITE_RESPONSE] is True
def test_write_actions_offer_write_option_only() -> None:
# Every write action takes expect_broadcast_write_response and none of the read options.
from esphome.components.modbus_client import (
_WRITE_MULTIPLE_COILS_SCHEMA,
_WRITE_MULTIPLE_REGISTERS_SCHEMA,
_WRITE_SINGLE_COIL_SCHEMA,
_WRITE_SINGLE_REGISTER_SCHEMA,
CONF_START_ADDRESS,
CONF_VALUE,
CONF_VALUES,
)
write_key = modbus.CONF_EXPECT_BROADCAST_WRITE_RESPONSE
base = {CONF_ADDRESS: 0, CONF_START_ADDRESS: 0x10, write_key: True}
for schema, extra in (
(_WRITE_SINGLE_REGISTER_SCHEMA, {CONF_VALUE: 1}),
(_WRITE_SINGLE_COIL_SCHEMA, {CONF_VALUE: True}),
(_WRITE_MULTIPLE_REGISTERS_SCHEMA, {CONF_VALUES: [1, 2]}),
(_WRITE_MULTIPLE_COILS_SCHEMA, {CONF_VALUES: [True, False]}),
):
config = schema({**base, **extra})
assert config[write_key] is True
assert modbus.CONF_ALLOW_BROADCAST_READ not in config
with pytest.raises(cv.Invalid):
schema({**base, **extra, modbus.CONF_ALLOW_BROADCAST_READ: True})
def test_send_options_follow_the_hub_classification() -> None:
# A vendor code is broadcastable, so it takes the write-side flag and refuses the read-side one;
# 0x17 is a read for broadcast purposes, so the reverse holds.
write_key = modbus.CONF_EXPECT_BROADCAST_WRITE_RESPONSE
read_key = modbus.CONF_ALLOW_BROADCAST_READ
assert MODBUS_CLIENT_SEND_SCHEMA(
{CONF_ADDRESS: 0, CONF_PDU: [0x41, 0x01], write_key: True}
)[write_key]
with pytest.raises(cv.Invalid, match=f"'{read_key}: true' does not apply"):
MODBUS_CLIENT_SEND_SCHEMA(
{CONF_ADDRESS: 0, CONF_PDU: [0x41, 0x01], read_key: True}
)
pdu_0x17 = [0x17, 0x00, 0x10, 0x00, 0x01, 0x00, 0x20, 0x00, 0x01, 0x02, 0x00, 0x01]
assert MODBUS_CLIENT_SEND_SCHEMA(
{CONF_ADDRESS: 0, CONF_PDU: pdu_0x17, read_key: True}
)[read_key]
with pytest.raises(cv.Invalid, match=f"'{write_key}: true' does not apply"):
MODBUS_CLIENT_SEND_SCHEMA(
{CONF_ADDRESS: 0, CONF_PDU: pdu_0x17, write_key: True}
)
def test_read_write_multiple_offers_allow_broadcast_read_only() -> None:
from esphome.components.modbus_client import (
_READ_WRITE_MULTIPLE_REGISTERS_SCHEMA,
CONF_READ_ADDRESS,
CONF_VALUES,
CONF_WRITE_ADDRESS,
)
config = _READ_WRITE_MULTIPLE_REGISTERS_SCHEMA(
{
CONF_ADDRESS: 0,
CONF_READ_ADDRESS: 0x10,
CONF_WRITE_ADDRESS: 0x20,
CONF_VALUES: [1],
modbus.CONF_ALLOW_BROADCAST_READ: True,
}
)
assert config[modbus.CONF_ALLOW_BROADCAST_READ] is True
assert CONF_CONTINUOUS not in config
assert modbus.CONF_EXPECT_BROADCAST_WRITE_RESPONSE not in config
@pytest.mark.parametrize(
"key",
[modbus.CONF_ALLOW_BROADCAST_READ, modbus.CONF_EXPECT_BROADCAST_WRITE_RESPONSE],
)
def test_broadcast_options_rejected_on_literal_unicast_address(key: str) -> None:
# A broadcast-only option on a literal non-zero address would be silently dropped by the hub.
if key == modbus.CONF_ALLOW_BROADCAST_READ:
pdu = [0x03, 0x00, 0x10, 0x00, 0x01]
else:
pdu = [0x06, 0x00, 0x10, 0x00, 0x01]
with pytest.raises(cv.Invalid, match="only applies to the broadcast address"):
MODBUS_CLIENT_SEND_SCHEMA({CONF_ADDRESS: 1, CONF_PDU: pdu, key: True})
# A templated address is not decidable at validation and passes through.
config = MODBUS_CLIENT_SEND_SCHEMA(
{CONF_ADDRESS: Lambda("return 1;"), CONF_PDU: pdu, key: True}
)
assert config[key] is True
@@ -0,0 +1,79 @@
"""A modbus_controller cannot poll the broadcast address (0) unless allow_broadcast_read says the
device answers it."""
import pytest
from esphome import config_validation as cv
from esphome.components import modbus
from esphome.components.modbus_controller import CONFIG_SCHEMA
from esphome.const import CONF_ADDRESS
from esphome.types import ConfigType
def _controller(address: int, **extra: object) -> ConfigType:
return CONFIG_SCHEMA({modbus.CONF_MODBUS_ID: "bus", CONF_ADDRESS: address, **extra})
def test_address_zero_rejected_by_default() -> None:
with pytest.raises(cv.Invalid, match="broadcast address"):
_controller(0)
def test_address_zero_accepted_with_allow_broadcast_read() -> None:
config = _controller(0, **{modbus.CONF_ALLOW_BROADCAST_READ: True})
assert config[CONF_ADDRESS] == 0
assert config[modbus.CONF_ALLOW_BROADCAST_READ] is True
def test_allow_broadcast_read_defaults_false() -> None:
assert _controller(1)[modbus.CONF_ALLOW_BROADCAST_READ] is False
def test_writer_entity_takes_expect_broadcast_write_response() -> None:
# The write-side option lives on the writing platforms, not the controller.
from esphome.components.modbus_controller.const import CONF_MODBUS_CONTROLLER_ID
from esphome.components.modbus_controller.switch import (
CONFIG_SCHEMA as SWITCH_SCHEMA,
)
from esphome.const import CONF_NAME
key = modbus.CONF_EXPECT_BROADCAST_WRITE_RESPONSE
base = {
CONF_MODBUS_CONTROLLER_ID: "ctl",
CONF_NAME: "Switch",
"register_type": "coil",
CONF_ADDRESS: 0x20,
}
assert SWITCH_SCHEMA(base)[key] is False
assert SWITCH_SCHEMA({**base, CONF_NAME: "Switch 2", key: True})[key] is True
with pytest.raises(cv.Invalid):
CONFIG_SCHEMA({modbus.CONF_MODBUS_ID: "bus", CONF_ADDRESS: 1, key: True})
def test_allow_broadcast_read_requires_address_zero() -> None:
# The option only means something at address 0; elsewhere it would be silently inert.
with pytest.raises(cv.Invalid, match="only applies to the broadcast address"):
_controller(5, **{modbus.CONF_ALLOW_BROADCAST_READ: True})
def test_add_command_options_skips_defaults() -> None:
# The setter is only emitted when an option differs from its C++ default.
import esphome.codegen as cg
from esphome.const import CONF_CONTINUOUS
var = cg.MockObj("ctl")
emitted: list = []
original = cg.add
cg.add = emitted.append
try:
modbus.add_command_options(
var, "set_read_options", {CONF_CONTINUOUS: False}, direction="read"
)
assert emitted == []
modbus.add_command_options(
var, "set_read_options", {CONF_CONTINUOUS: True}, direction="read"
)
assert len(emitted) == 1
assert "set_read_options" in str(emitted[0])
finally:
cg.add = original
@@ -9,6 +9,7 @@ test cannot: a write-coded custom_pdu polled continuously is rejected there.
import pytest
from voluptuous import Invalid, MultipleInvalid
from esphome.components import modbus
from esphome.components.modbus_controller import (
ModbusItemBaseSchema,
validate_custom_pdu_item,
@@ -55,14 +56,21 @@ def test_custom_pdu_rejects_non_byte_values() -> None:
ModbusItemBaseSchema({CONF_CUSTOM_PDU: [0x0103, 0x002A]})
def _controller_full_config(*, continuous: bool) -> Config:
def _controller_full_config(
*, continuous: bool, allow_broadcast_read: bool = False
) -> Config:
"""A minimal full-config graph with one modbus_controller declaring id 'ctl', enough for the
final-validate to resolve the controller (and its continuous flag) from an item's
final-validate to resolve the controller (and its option flags) from an item's
modbus_controller_id."""
ctl_id = ID("ctl", is_declaration=True)
config = Config()
config["modbus_controller"] = [
{CONF_ID: ctl_id, CONF_ADDRESS: 1, CONF_CONTINUOUS: continuous}
{
CONF_ID: ctl_id,
CONF_ADDRESS: 0 if allow_broadcast_read else 1,
CONF_CONTINUOUS: continuous,
modbus.CONF_ALLOW_BROADCAST_READ: allow_broadcast_read,
}
]
config.declare_ids.append((ctl_id, ["modbus_controller", 0, CONF_ID]))
return config
@@ -98,3 +106,64 @@ def test_continuous_read_custom_pdu_allowed(reset_full_config) -> None:
CONF_CUSTOM_PDU: [0x03, 0x00, 0x2A, 0x00, 0x01],
}
)
def test_broadcastable_custom_pdu_rejected_under_broadcast_controller(
reset_full_config,
) -> None:
"""A vendor-coded custom_pdu under an allow_broadcast_read controller would be a real broadcast,
never answered, so it is rejected at final validate."""
fv.full_config.set(
_controller_full_config(continuous=False, allow_broadcast_read=True)
)
with pytest.raises(Invalid, match="is a real broadcast at address 0"):
validate_custom_pdu_item(
{
CONF_MODBUS_CONTROLLER_ID: ID("ctl"),
CONF_CUSTOM_PDU: [0x41, 0x00, 0x03],
}
)
def test_read_custom_pdu_allowed_under_broadcast_controller(reset_full_config) -> None:
"""A read-coded custom_pdu (0x03) is answered under allow_broadcast_read, so it is fine."""
fv.full_config.set(
_controller_full_config(continuous=False, allow_broadcast_read=True)
)
validate_custom_pdu_item(
{
CONF_MODBUS_CONTROLLER_ID: ID("ctl"),
CONF_CUSTOM_PDU: [0x03, 0x00, 0x2A, 0x00, 0x01],
}
)
def test_write_option_rejected_under_unicast_controller(reset_full_config) -> None:
"""expect_broadcast_write_response on a writer entity whose controller is not at address 0 is
rejected at final validate, where the controller's address is known."""
from esphome.components.modbus_controller import validate_writer_item
fv.full_config.set(_controller_full_config(continuous=False))
with pytest.raises(
Invalid, match="only applies when the 'ctl' modbus_controller is at address 0"
):
validate_writer_item(
{
CONF_MODBUS_CONTROLLER_ID: ID("ctl"),
modbus.CONF_EXPECT_BROADCAST_WRITE_RESPONSE: True,
}
)
def test_write_option_allowed_under_broadcast_controller(reset_full_config) -> None:
from esphome.components.modbus_controller import validate_writer_item
fv.full_config.set(
_controller_full_config(continuous=False, allow_broadcast_read=True)
)
validate_writer_item(
{
CONF_MODBUS_CONTROLLER_ID: ID("ctl"),
modbus.CONF_EXPECT_BROADCAST_WRITE_RESPONSE: True,
}
)
@@ -0,0 +1,18 @@
---
esphome:
name: test
esp32:
board: esp32dev
logger:
switch:
- platform: template
id: plain_switch
turn_on_action:
- logger.log: "on"
- platform: template
id: enabled_switch
optimistic: true
assumed_state: true
@@ -0,0 +1,17 @@
"""Tests for the template switch codegen."""
from collections.abc import Callable
from pathlib import Path
def test_default_flags_are_not_emitted(
generate_main: Callable[[str | Path], str],
component_config_path: Callable[[str], Path],
) -> None:
"""Only true optimistic and assumed_state are set; false is the C++ initializer."""
main_cpp = generate_main(component_config_path("switch_defaults.yaml"))
assert "plain_switch->set_optimistic(" not in main_cpp
assert "plain_switch->set_assumed_state(" not in main_cpp
assert "enabled_switch->set_optimistic(true);" in main_cpp
assert "enabled_switch->set_assumed_state(true);" in main_cpp
@@ -0,0 +1,42 @@
"""Tests for the uart_mux component's final validation."""
import pytest
from esphome import config_validation as cv
from esphome.const import CONF_ID, PlatformFramework
from esphome.core import ID
from esphome.types import ConfigType
from tests.component_tests.types import SetCoreConfigCallable
CONF_BRIDGE_ID = "bridge_id"
def _set_esp32_s3(set_core_config: SetCoreConfigCallable) -> None:
from esphome.components.esp32 import KEY_VARIANT, VARIANT_ESP32S3
set_core_config(
PlatformFramework.ESP32_IDF, platform_data={KEY_VARIANT: VARIANT_ESP32S3}
)
def _mux_conf(mux_id: str, bridge_id: str) -> ConfigType:
return {CONF_ID: ID(mux_id), CONF_BRIDGE_ID: ID(bridge_id)}
def test_accepts_one_mux_per_bridge(set_core_config: SetCoreConfigCallable) -> None:
_set_esp32_s3(set_core_config)
from esphome.components import uart_mux
uart_mux._final_validate(_mux_conf("mux_0", "bridge_0"))
uart_mux._final_validate(_mux_conf("mux_1", "bridge_1"))
def test_rejects_two_muxes_on_one_bridge(
set_core_config: SetCoreConfigCallable,
) -> None:
_set_esp32_s3(set_core_config)
from esphome.components import uart_mux
uart_mux._final_validate(_mux_conf("mux_0", "bridge_0"))
with pytest.raises(cv.Invalid, match="already routed by another 'uart_mux'"):
uart_mux._final_validate(_mux_conf("mux_1", "bridge_0"))
@@ -0,0 +1,46 @@
"""Tests for wifi.force_power_save_off(), the hook platforms use to keep the
station out of power save."""
from collections.abc import Callable
from pathlib import Path
import pytest
from esphome.components import wifi
from esphome.core import CORE, EsphomeError
def test_reasons_accumulate_without_duplicates() -> None:
"""Every caller's reason is kept once; a repeated reason is not duplicated."""
wifi.force_power_save_off("first")
wifi.force_power_save_off("first")
wifi.force_power_save_off("second")
assert CORE.data[wifi.POWER_SAVE_OFF_REASONS_KEY] == ["first", "second"]
def test_forced_off_skips_the_setter_and_warns(
generate_main: Callable[[str | Path], str],
component_config_path: Callable[[str], Path],
caplog: pytest.LogCaptureFixture,
) -> None:
"""With a reason recorded, power_save_mode is reported and not applied."""
wifi.force_power_save_off("the platform cannot sleep")
main_cpp = generate_main(component_config_path("custom.yaml"))
assert "set_power_save_mode(" not in main_cpp
assert (
"power_save_mode LIGHT is not applied: the platform cannot sleep" in caplog.text
)
def test_call_after_wifi_codegen_raises(
generate_main: Callable[[str | Path], str],
component_config_path: Callable[[str], Path],
) -> None:
"""Once wifi has generated its code the hook cannot take effect any more."""
generate_main(component_config_path("custom.yaml"))
with pytest.raises(EsphomeError, match="before wifi generates its code"):
wifi.force_power_save_off("too late")
@@ -0,0 +1,9 @@
# A wifi power_save_mode other than NONE is forced off with a warning while
# bk72xx_ble is configured (esphome#18592); this config must still validate.
packages:
bk72xx_ble: !include common.yaml
wifi:
ssid: MySSID
password: password1
power_save_mode: high
@@ -52,7 +52,7 @@ TEST(RpcResponseBuilder, GoldenBytes) {
(std::vector<uint8_t>{0x04, 0x03, 0x02, 'a', 'b', 0xCC}));
}
// esp32_improv calls finish() and build_rpc_response() with no checksum flag,
// improv_ble calls finish() and build_rpc_response() with no checksum flag,
// so the two defaults must agree
TEST(RpcResponseBuilder, DefaultChecksumFlagMatches) {
const std::vector<std::string> urls = {"https://example.com"};
@@ -12,7 +12,7 @@ output:
pin: 2
id: built_in_led
esp32_improv:
improv_ble:
authorizer: io0_button
authorized_duration: 1min
status_indicator: built_in_led
@@ -5,6 +5,6 @@ wifi:
logger:
hardware_uart: UART0
# next_url compiles the USE_IMPROV_SERIAL_NEXT_URL branch and add_next_url_
# next_url compiles the USE_IMPROV_NEXT_URL branch and add_next_url_
improv_serial:
next_url: https://example.com/?device_name={{device_name}}&ip_address={{ip_address}}
@@ -792,6 +792,261 @@ TEST(ModbusClientHubBroadcast, RefusesReadBroadcast) {
EXPECT_EQ(device.sent_count_, 0); // never transmitted
}
// allow_broadcast_read lifts the refusal for a device that answers address 0: the read is queued, sent,
// and waits for a reply like a unicast read, so a reply from address 0 completes it with on_response.
TEST(ModbusClientHubBroadcast, AllowBroadcastReadWaitsAndAcceptsReplyFromZero) {
NullUART uart;
NoResponseProbeHub hub;
hub.set_uart_parent(&uart);
hub.setup();
BroadcastProbeDevice device(&hub, BROADCAST_ADDRESS);
const uint8_t read[] = {0x03, 0x00, 0x10, 0x00, 0x02}; // read holding registers 0x0010, count 2
ASSERT_TRUE(device.queue_pdu(read, {.allow_broadcast_read = true}));
EXPECT_TRUE(hub.queued(0).options.allow_broadcast_read);
EXPECT_FALSE(hub.queued(0).fire_and_forget());
hub.send_next_for_test();
EXPECT_EQ(device.sent_count_, 1);
EXPECT_TRUE(hub.waiting()); // not fire-and-forget: the reply is expected
EXPECT_EQ(hub.entries(), 1u);
const uint8_t reply[] = {0x03, 0x04, 0x00, 0x01, 0x00, 0x02};
hub.receive_frame_for_test(BROADCAST_ADDRESS, reply);
EXPECT_EQ(device.response_count_, 1);
EXPECT_EQ(device.last_response_size_, sizeof(reply));
EXPECT_FALSE(hub.waiting());
EXPECT_EQ(hub.entries(), 0u);
}
// The address-0 read waits like a unicast one, so the reply must come from address 0 too: a reply from
// another unit id is an unexpected frame and interrupts the transaction as it would for any address.
TEST(ModbusClientHubBroadcast, AllowBroadcastReadRejectsReplyFromOtherAddress) {
NullUART uart;
NoResponseProbeHub hub;
hub.set_uart_parent(&uart);
hub.setup();
BroadcastProbeDevice device(&hub, BROADCAST_ADDRESS);
const uint8_t read[] = {0x03, 0x00, 0x10, 0x00, 0x02};
ASSERT_TRUE(device.queue_pdu(read, {.allow_broadcast_read = true}));
hub.send_next_for_test();
ASSERT_TRUE(hub.waiting());
const uint8_t reply[] = {0x03, 0x04, 0x00, 0x01, 0x00, 0x02};
hub.receive_frame_for_test(0x07, reply);
EXPECT_EQ(device.response_count_, 0);
EXPECT_EQ(hub.waiting_command().state, FrameState::INTERRUPTED);
}
// An address-scoped clear must not turn a live address-0 entry back into a fire-and-forget broadcast: a
// retry granted after the clear is re-sent with the flag intact, so it still waits and gets its terminal.
TEST(ModbusClientHubBroadcast, AllowBroadcastReadSurvivesClearBeforeRetry) {
NullUART uart;
NoResponseProbeHub hub;
hub.set_uart_parent(&uart);
hub.setup();
RetryingDevice device(&hub, BROADCAST_ADDRESS, true);
const uint8_t read[] = {0x03, 0x00, 0x10, 0x00, 0x02};
ASSERT_TRUE(device.queue_pdu(read, {.allow_broadcast_read = true}));
hub.send_next_for_test();
ASSERT_TRUE(hub.waiting());
hub.clear_tx_queue_for_address(BROADCAST_ADDRESS);
EXPECT_EQ(hub.waiting_command().state, FrameState::WAITING_RETIRED);
EXPECT_TRUE(hub.waiting_command().options.allow_broadcast_read);
hub.timeout_waiting(); // retry granted: the entry is READY again
ASSERT_EQ(hub.queued_frames(), 1u);
EXPECT_FALSE(hub.queued(0).fire_and_forget());
hub.send_next_for_test();
EXPECT_TRUE(hub.waiting()); // the retry still waits for its reply
EXPECT_EQ(hub.entries(), 1u);
}
// The function code check is unchanged by the relaxed address match: a mismatched reply still interrupts.
TEST(ModbusClientHubBroadcast, AllowBroadcastReadStillRejectsWrongFunctionCode) {
NullUART uart;
NoResponseProbeHub hub;
hub.set_uart_parent(&uart);
hub.setup();
BroadcastProbeDevice device(&hub, BROADCAST_ADDRESS);
const uint8_t read[] = {0x03, 0x00, 0x10, 0x00, 0x02};
ASSERT_TRUE(device.queue_pdu(read, {.allow_broadcast_read = true}));
hub.send_next_for_test();
ASSERT_TRUE(hub.waiting());
const uint8_t wrong_reply[] = {0x04, 0x04, 0x00, 0x01, 0x00, 0x02};
hub.receive_frame_for_test(BROADCAST_ADDRESS, wrong_reply); // right address, wrong function code
EXPECT_EQ(device.response_count_, 0);
EXPECT_EQ(hub.waiting_command().state, FrameState::INTERRUPTED);
}
// A silent device leaves the read to the normal send-wait timeout, so on_no_response is delivered.
TEST(ModbusClientHubBroadcast, AllowBroadcastReadTimesOutLikeUnicast) {
NullUART uart;
NoResponseProbeHub hub;
hub.set_uart_parent(&uart);
hub.setup();
BroadcastProbeDevice device(&hub, BROADCAST_ADDRESS);
const uint8_t read[] = {0x03, 0x00, 0x10, 0x00, 0x02};
ASSERT_TRUE(device.queue_pdu(read, {.allow_broadcast_read = true}));
hub.send_next_for_test();
ASSERT_TRUE(hub.waiting());
hub.timeout_waiting();
EXPECT_EQ(device.no_response_count_, 1);
EXPECT_EQ(device.response_count_, 0);
EXPECT_FALSE(hub.waiting());
EXPECT_EQ(hub.entries(), 0u);
}
// allow_broadcast_read is stripped from a broadcastable code (a write or custom code to address 0 is a real broadcast,
// still fire-and-forget) and from a unicast frame (nothing to allow).
TEST(ModbusClientHubBroadcast, AllowBroadcastReadIgnoredForWritesAndUnicast) {
NullUART uart;
NoResponseProbeHub hub;
hub.set_uart_parent(&uart);
hub.setup();
BroadcastProbeDevice broadcast_device(&hub, BROADCAST_ADDRESS);
BroadcastProbeDevice unicast_device(&hub, 0x01);
const uint8_t write[] = {0x06, 0x00, 0x10, 0x00, 0x01};
ASSERT_TRUE(broadcast_device.queue_pdu(write, {.allow_broadcast_read = true}));
EXPECT_FALSE(hub.queued(0).options.allow_broadcast_read);
EXPECT_TRUE(hub.queued(0).fire_and_forget());
hub.send_next_for_test();
EXPECT_EQ(broadcast_device.sent_count_, 1);
EXPECT_FALSE(hub.waiting());
EXPECT_EQ(hub.entries(), 0u);
const uint8_t custom[] = {0x41, 0x01, 0x02};
ASSERT_TRUE(broadcast_device.queue_pdu(custom, {.allow_broadcast_read = true}));
EXPECT_FALSE(hub.queued(0).options.allow_broadcast_read);
EXPECT_TRUE(hub.queued(0).fire_and_forget());
hub.send_next_for_test();
EXPECT_FALSE(hub.waiting());
EXPECT_EQ(hub.entries(), 0u);
const uint8_t read[] = {0x03, 0x00, 0x10, 0x00, 0x02};
ASSERT_TRUE(unicast_device.queue_pdu(read, {.allow_broadcast_read = true}));
EXPECT_FALSE(hub.queued(0).options.allow_broadcast_read);
EXPECT_FALSE(hub.queued(0).fire_and_forget());
}
// expect_broadcast_write_response is the write-side twin: a write to address 0 waits for its reply instead
// of retiring at transmission, and the reply (from address 0) completes it.
TEST(ModbusClientHubBroadcast, ExpectBroadcastWriteResponseWaitsAndAcceptsReply) {
NullUART uart;
NoResponseProbeHub hub;
hub.set_uart_parent(&uart);
hub.setup();
BroadcastProbeDevice device(&hub, BROADCAST_ADDRESS);
const uint8_t write[] = {0x06, 0x00, 0x10, 0x00, 0x01};
ASSERT_TRUE(device.write_single_register(0x0010, 0x0001, {.expect_broadcast_write_response = true}));
EXPECT_TRUE(hub.queued(0).options.expect_broadcast_write_response);
EXPECT_FALSE(hub.queued(0).fire_and_forget());
hub.send_next_for_test();
EXPECT_EQ(device.sent_count_, 1);
EXPECT_TRUE(hub.waiting());
EXPECT_EQ(hub.entries(), 1u);
hub.receive_frame_for_test(BROADCAST_ADDRESS, write); // the echo, as address 0
EXPECT_EQ(device.response_count_, 1);
EXPECT_EQ(device.last_response_size_, sizeof(write));
EXPECT_FALSE(hub.waiting());
EXPECT_EQ(hub.entries(), 0u);
}
// Two requests for the same address-0 write may disagree on expect_broadcast_write_response (a
// broadcastable frame is accepted either way), but a write duplicate is refused at its cap of one in
// flight rather than absorbed, so the queued entry's delivery mode is never changed under it.
TEST(ModbusClientHubBroadcast, ExpectBroadcastWriteResponseDuplicateRefusedNotMerged) {
NullUART uart;
NoResponseProbeHub hub;
hub.set_uart_parent(&uart);
hub.setup();
BroadcastProbeDevice device(&hub, BROADCAST_ADDRESS);
ASSERT_TRUE(device.write_single_register(0x0010, 0x0001)); // fire-and-forget as queued
EXPECT_TRUE(hub.queued(0).fire_and_forget());
EXPECT_FALSE(device.write_single_register(0x0010, 0x0001, {.expect_broadcast_write_response = true}));
EXPECT_EQ(hub.entries(), 1u);
EXPECT_TRUE(hub.queued(0).fire_and_forget()); // the refused request left the entry untouched
hub.send_next_for_test();
EXPECT_FALSE(hub.waiting());
EXPECT_EQ(hub.entries(), 0u);
}
// A custom-code poll at address 0 is a fire-and-forget broadcast that a one-shot duplicate downgrades and
// is absorbed into; if that duplicate wants the reply, the entry waits for it instead of retiring at the
// send, so the absorbed request still gets its terminal callback.
TEST(ModbusClientHubBroadcast, ExpectBroadcastWriteResponseMergesIntoDowngradedPoll) {
NullUART uart;
NoResponseProbeHub hub;
hub.set_uart_parent(&uart);
hub.setup();
BroadcastProbeDevice device(&hub, BROADCAST_ADDRESS);
const uint8_t custom[] = {0x41, 0x01, 0x02};
ASSERT_TRUE(device.queue_pdu(custom, {.continuous = true}));
EXPECT_TRUE(hub.queued(0).fire_and_forget());
ASSERT_TRUE(device.queue_pdu(custom, {.expect_broadcast_write_response = true})); // downgrades, absorbed
EXPECT_EQ(hub.entries(), 1u);
EXPECT_FALSE(hub.queued(0).options.continuous);
EXPECT_FALSE(hub.queued(0).fire_and_forget());
hub.send_next_for_test();
EXPECT_TRUE(hub.waiting());
hub.receive_frame_for_test(BROADCAST_ADDRESS, custom);
EXPECT_EQ(device.response_count_, 1);
}
// A silent device leaves an expected write response to the normal send-wait timeout.
TEST(ModbusClientHubBroadcast, ExpectBroadcastWriteResponseTimesOutLikeUnicast) {
NullUART uart;
NoResponseProbeHub hub;
hub.set_uart_parent(&uart);
hub.setup();
BroadcastProbeDevice device(&hub, BROADCAST_ADDRESS);
ASSERT_TRUE(device.write_single_coil(0x0010, true, {.expect_broadcast_write_response = true}));
hub.send_next_for_test();
ASSERT_TRUE(hub.waiting());
hub.timeout_waiting();
EXPECT_EQ(device.no_response_count_, 1);
EXPECT_EQ(device.response_count_, 0);
EXPECT_FALSE(hub.waiting());
EXPECT_EQ(hub.entries(), 0u);
}
// expect_broadcast_write_response is stripped from a read (allow_broadcast_read is the read-side flag, so
// the broadcast guard still refuses it) and from a unicast frame (nothing to expect).
TEST(ModbusClientHubBroadcast, ExpectBroadcastWriteResponseIgnoredForReadsAndUnicast) {
NullUART uart;
NoResponseProbeHub hub;
hub.set_uart_parent(&uart);
hub.setup();
BroadcastProbeDevice broadcast_device(&hub, BROADCAST_ADDRESS);
BroadcastProbeDevice unicast_device(&hub, 0x01);
const uint8_t read[] = {0x03, 0x00, 0x10, 0x00, 0x02};
EXPECT_FALSE(broadcast_device.queue_pdu(read, {.expect_broadcast_write_response = true}));
EXPECT_EQ(hub.entries(), 0u);
ASSERT_TRUE(unicast_device.write_single_register(0x0010, 0x0001, {.expect_broadcast_write_response = true}));
EXPECT_FALSE(hub.queued(0).options.expect_broadcast_write_response);
EXPECT_FALSE(hub.queued(0).fire_and_forget());
}
// The counterpart to RefusesReadBroadcast: a custom (user-defined) function code carries no reply the
// hub knows how to expect, so a broadcast of one is accepted and completes fire-and-forget like a write.
TEST(ModbusClientHubBroadcast, AcceptsCustomBroadcast) {
+3 -1
View File
@@ -79,7 +79,8 @@ button:
name: "Typed Actions"
on_press:
- modbus_client.write_single_register:
address: 0x01
address: !lambda "return 1;"
expect_broadcast_write_response: true
start_address: 0x0102
value: !lambda "return 42;"
on_response:
@@ -93,6 +94,7 @@ button:
start_address: 0x10
count: 2
continuous: true
allow_broadcast_read: !lambda "return false;"
on_response:
then:
- lambda: 'ESP_LOGI("modbus_client.test", "first=%u n=%u", values[0], (unsigned) values.size());'
@@ -0,0 +1,36 @@
# Config-only: actions that address the broadcast address (0) and wait for a reply, for a device that
# answers it. Never compiled, so the extra action objects do not inflate the memory-impact baseline.
packages:
modbus: !include ../../test_build_components/common/modbus/esp32-idf.yaml
button:
- platform: template
name: Broadcast probe
on_press:
- modbus_client.read_holding_registers:
address: 0
allow_broadcast_read: true
start_address: 0x10
count: 1
on_response:
then:
- lambda: 'ESP_LOGI("modbus_client.test", "broadcast read first=%u", values[0]);'
- modbus_client.write_single_register:
address: 0
expect_broadcast_write_response: true
start_address: 0x0102
value: 42
on_response:
then:
- logger.log: "broadcast write acked"
- modbus_client.read_write_multiple_registers:
address: 0
allow_broadcast_read: true
read_address: 0x10
read_count: 1
write_address: 0x20
values: [1]
- modbus_client.send:
address: 0
expect_broadcast_write_response: true
pdu: [0x41, 0x01]
@@ -6,7 +6,6 @@ modbus_controller:
on_online:
then:
logger.log: "Module Online"
binary_sensor:
- platform: modbus_controller
modbus_controller_id: modbus_controller1
@@ -0,0 +1,29 @@
# Config-only: a controller polling the broadcast address (0), for a device that answers it, with a
# writer entity expecting the reply to its broadcast writes. Never compiled, so the extra entities do
# not inflate the memory-impact baseline.
packages:
modbus: !include ../../test_build_components/common/modbus/esp32-idf.yaml
modbus_controller:
- id: modbus_controller_broadcast
address: 0
allow_broadcast_read: true
modbus_id: modbus_bus
sensor:
- platform: modbus_controller
modbus_controller_id: modbus_controller_broadcast
id: modbus_broadcast_sensor
name: Broadcast Read Sensor
register_type: holding
address: 0x0010
value_type: U_WORD
switch:
- platform: modbus_controller
modbus_controller_id: modbus_controller_broadcast
id: modbus_broadcast_switch
name: Broadcast Write Switch
register_type: coil
address: 0x20
expect_broadcast_write_response: true
@@ -1,6 +1,6 @@
# Exercises the provisioning window: api registers as a provisioning source
# (encryption enabled, no key), the on_timeout automation, and the wifi (AP +
# captive portal) and esp32_improv cross-component guards. improv_serial is
# captive portal) and improv_ble cross-component guards. improv_serial is
# intentionally NOT gated.
provisioning:
timeout: 1min
@@ -26,5 +26,5 @@ binary_sensor:
pin: 0
id: io0_button
esp32_improv:
improv_ble:
authorizer: io0_button
+9
View File
@@ -6,6 +6,15 @@ tinyusb:
usb_product_str: ESPHomeTestProduct
usb_serial_str: ESPHomeTestSerialNumber
usb_vendor_id: 0x2345
on_mount:
- logger.log: USB host mounted
- if:
condition:
tinyusb.is_mounted:
then:
- logger.log: USB host is mounted
on_unmount:
- logger.log: USB host unmounted
# tinyusb requires at least one USB class companion; usb_cdc_acm satisfies that.
usb_cdc_acm:
@@ -1 +1,7 @@
<<: !include common.yaml
packages:
tinyusb: !include common.yaml
# VBUS monitoring is per variant: the OTG hardware watches the pin here, while the
# S31 would need the GPIO ISR path and rejects the key.
tinyusb:
vbus_monitor_pin: 4
@@ -1,4 +1,10 @@
<<: !include common.yaml
packages:
tinyusb: !include common.yaml
# VBUS monitoring is per variant: the OTG hardware watches the pin here, while the
# S31 would need the GPIO ISR path and rejects the key.
tinyusb:
vbus_monitor_pin: 4
# S2 defaults logger to USB_CDC, which conflicts with tinyusb on the shared
# USB OTG peripheral; route the logger to UART0 so the fixture builds.
@@ -1 +1,7 @@
<<: !include common.yaml
packages:
tinyusb: !include common.yaml
# VBUS monitoring is per variant: the OTG hardware watches the pin here, while the
# S31 would need the GPIO ISR path and rejects the key.
tinyusb:
vbus_monitor_pin: 4
+44
View File
@@ -0,0 +1,44 @@
tinyusb:
id: tinyusb_test
on_mount:
- uart_mux.select_bridge: mux_0
on_unmount:
- uart_mux.select_local: mux_0
usb_manufacturer_str: ESPHomeTestManufacturer
usb_product_id: 0x1234
usb_product_str: ESPHomeTestProduct
usb_vendor_id: 0x2345
uart:
- id: uart_0
tx_pin: 14
rx_pin: 13
baud_rate: 115200
usb_cdc_acm:
interfaces:
- id: cdc_acm_1
bridge:
- platform: cdc_acm_uart
id: bridge_0
uart_id: uart_0
usb_cdc_acm_id: cdc_acm_1
uart_mux:
- id: mux_0
bridge_id: bridge_0
initial_route: local
interval:
- interval: 60s
then:
- if:
condition:
uart_mux.is_local: mux_0
then:
- lambda: |-
uint8_t byte;
if (id(mux_0).available() && id(mux_0).read_byte(&byte)) {
id(mux_0).write_byte(byte);
}
@@ -0,0 +1,2 @@
packages:
uart_mux: !include common.yaml
@@ -0,0 +1,7 @@
# ESP32-S2 has no USB_SERIAL_JTAG, so the logger defaults to USB_CDC, which shares
# the USB OTG peripheral with tinyusb. Use a hardware UART for logging instead.
logger:
hardware_uart: UART0
packages:
uart_mux: !include common.yaml
@@ -0,0 +1,2 @@
packages:
uart_mux: !include common.yaml
+1
View File
@@ -6,6 +6,7 @@ usb_uart:
type: cdc_acm
vid: 0x1234
pid: 0x5678
claim_comm_interface: false
channels:
- id: channel_0_1
- id: uart_1
+42 -1
View File
@@ -5,8 +5,13 @@ from __future__ import annotations
from pathlib import Path
from unittest.mock import patch
import pytest
from esphome import yaml_util
from esphome.components.file import image as file_image
from esphome.external_files import RemoteFile
from esphome.const import CONF_PATH
from esphome.core import CORE
from esphome.external_files import RemoteFile, url_cache_key
from esphome.loader import get_component, get_platform
@@ -55,6 +60,42 @@ def test_prefetch_files_yields_remote_refs(setup_core: Path) -> None:
assert files[1].url == "https://example.com/img.png"
def test_validated_file_values_hash_alike_across_data_dirs(
setup_core: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""A CLI and an add-on data dir dump validated image files identically."""
url = "https://example.com/img.png"
(setup_core / "img.png").touch()
dumps: list[str] = []
for data_dir in (
setup_core / ".esphome",
setup_core.parent / f"{setup_core.name}-data",
):
monkeypatch.setenv("ESPHOME_DATA_DIR", str(data_dir))
with patch("esphome.components.file.image.external_files.download_content"):
config = {
"remote": file_image.validate_file_shorthand(url),
"mdi": file_image.validate_file_shorthand("mdi:home"),
"local": file_image.validate_file_shorthand("img.png"),
"local_schema": file_image.LOCAL_SCHEMA({CONF_PATH: "img.png"}),
}
dumps.append(
yaml_util.dump(
config,
sort_keys=True,
relative_to=CORE.config_dir,
data_dir=CORE.data_dir,
)
)
assert dumps[0] == dumps[1]
assert dumps[0].splitlines() == [
"local: img.png",
"local_schema: img.png",
"mdi: .esphome/image/mdi/home.svg",
f"remote: .esphome/image/{url_cache_key(url)}",
]
def test_extractor_matches_validator_path(setup_core: Path) -> None:
"""The path the validator downloads to equals the extractor's path."""
with patch(
+21
View File
@@ -175,6 +175,27 @@ async def test_core_area_recorded_at_config_load(
assert CORE.area == expected_area
@pytest.mark.asyncio
async def test_app_is_default_initialized(
yaml_file: Callable[[str], Path],
) -> None:
"""App is constructed with `new (&App) Application`, no parentheses.
`Application()` would value-initialize and memset the whole object into
storage that is already zero."""
result = load_config_from_fixture(yaml_file, "valid_area_device.yaml", FIXTURES_DIR)
assert result is not None
with patch("esphome.core.config.cg") as mock_cg:
mock_cg.RawStatement.side_effect = lambda *args, **kwargs: MagicMock()
mock_cg.RawExpression.side_effect = lambda *args, **kwargs: MagicMock()
await config.to_code(result[CONF_ESPHOME])
raw_expressions = [c.args[0] for c in mock_cg.RawExpression.call_args_list]
assert "new (&App) Application" in raw_expressions
assert "new (&App) Application()" not in raw_expressions
def test_config_load_without_area_clears_stale_core_area(
yaml_file: Callable[[str], Path],
) -> None:
+122 -3
View File
@@ -1,7 +1,9 @@
"""Tests for esphome.automation module."""
from collections.abc import Generator
from unittest.mock import AsyncMock, call, patch
from collections.abc import Callable, Generator
from functools import partial
from typing import NamedTuple
from unittest.mock import AsyncMock, MagicMock, call, patch
import pytest
@@ -12,9 +14,18 @@ from esphome.automation import (
TriggerOnTrueForwarder,
build_callback_automations,
has_non_synchronous_actions,
register_bare_action,
register_bare_condition,
register_parented_action,
register_parented_condition,
register_simple_action,
register_simple_condition,
)
import esphome.codegen as cg
from esphome.const import CONF_ID
from esphome.core import ID
from esphome.cpp_generator import MockObj, RawExpression
from esphome.util import RegistryEntry
from esphome.util import Registry, RegistryEntry
def _make_registry(non_synchronous_actions: set[str]) -> dict[str, RegistryEntry]:
@@ -475,3 +486,111 @@ async def test_build_callback_automations_defaults(
mock_build_callback.assert_called_once_with(
parent, "add_on_press_callback", [], conf, forwarder=None
)
PARENT_ID = ID("my_component")
PARENT_OBJ = MockObj("parent", "->")
NEW_OBJ = MockObj("var", "->")
ACTION_TYPE = cg.esphome_ns.class_("MyAction")
CONDITION_TYPE = cg.esphome_ns.class_("MyCondition")
TEMPLATE_ARG = cg.TemplateArguments()
class MockCodegen(NamedTuple):
get_variable: AsyncMock
new_pvariable: MagicMock
register_parented: AsyncMock
@pytest.fixture
def mock_cg() -> Generator[MockCodegen]:
"""Patch the codegen calls the shared builders make."""
with (
patch("esphome.codegen.get_variable", new_callable=AsyncMock) as get_variable,
patch("esphome.codegen.new_Pvariable") as new_pvariable,
patch(
"esphome.codegen.register_parented", new_callable=AsyncMock
) as register_parented,
):
get_variable.return_value = PARENT_OBJ
new_pvariable.return_value = NEW_OBJ
yield MockCodegen(get_variable, new_pvariable, register_parented)
@pytest.fixture
def registries() -> Generator[tuple[Registry, Registry]]:
"""Patch both registries so registrations made by a test do not leak."""
actions = Registry()
conditions = Registry()
with (
patch("esphome.automation.ACTION_REGISTRY", actions),
patch("esphome.automation.CONDITION_REGISTRY", conditions),
):
yield actions, conditions
@pytest.mark.asyncio
@pytest.mark.parametrize(
("register", "is_action", "ctor_parent", "parented"),
[
(partial(register_simple_action, synchronous=True), True, True, False),
(partial(register_bare_action, synchronous=True), True, False, False),
(partial(register_parented_action, synchronous=True), True, False, True),
(register_simple_condition, False, True, False),
(register_bare_condition, False, False, False),
(register_parented_condition, False, False, True),
],
ids=[
"simple_action",
"bare_action",
"parented_action",
"simple_condition",
"bare_condition",
"parented_condition",
],
)
async def test_shared_builders(
registries: tuple[Registry, Registry],
mock_cg: MockCodegen,
register: Callable[..., None],
is_action: bool,
ctor_parent: bool,
parented: bool,
) -> None:
"""Each helper constructs the object and wires the parent the way its C++ shape needs."""
actions, conditions = registries
type_id = ACTION_TYPE if is_action else CONDITION_TYPE
register("my.entry", type_id, {})
entry = (actions if is_action else conditions)["my.entry"]
assert entry.type_id is type_id
config = {CONF_ID: PARENT_ID} if ctor_parent or parented else {}
result = await entry.fun(config, ID("obj_1"), TEMPLATE_ARG, [])
assert result is NEW_OBJ
if ctor_parent:
mock_cg.get_variable.assert_awaited_once_with(PARENT_ID)
mock_cg.new_pvariable.assert_called_once_with(
ID("obj_1"), TEMPLATE_ARG, PARENT_OBJ
)
else:
mock_cg.get_variable.assert_not_called()
mock_cg.new_pvariable.assert_called_once_with(ID("obj_1"), TEMPLATE_ARG)
if parented:
mock_cg.register_parented.assert_awaited_once_with(NEW_OBJ, PARENT_ID)
else:
mock_cg.register_parented.assert_not_called()
@pytest.mark.parametrize("synchronous", [True, False])
def test_shared_builders_keep_synchronous_flag(
registries: tuple[Registry, Registry], synchronous: bool
) -> None:
"""The synchronous flag reaches the registry entry unchanged."""
actions, _ = registries
register_simple_action("my.simple", ACTION_TYPE, {}, synchronous=synchronous)
register_bare_action("my.bare", ACTION_TYPE, {}, synchronous=synchronous)
register_parented_action("my.parented", ACTION_TYPE, {}, synchronous=synchronous)
assert actions["my.simple"].synchronous is synchronous
assert actions["my.bare"].synchronous is synchronous
assert actions["my.parented"].synchronous is synchronous