Merge remote-tracking branch 'upstream/dev' into HEAD

This commit is contained in:
J. Nick Koston
2026-09-17 08:48:55 -05:00
349 changed files with 5339 additions and 2003 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
@@ -0,0 +1,8 @@
esphome:
name: test
esp32:
board: esp32dev
framework:
type: esp-idf
version: 5.0.6
@@ -0,0 +1,9 @@
esphome:
name: test
esp32:
board: esp32dev
framework:
type: esp-idf
advanced:
flash_chip: gd
@@ -0,0 +1,9 @@
esphome:
name: test
esp32:
board: esp32dev
framework:
type: esp-idf
advanced:
flash_chip: generic
@@ -0,0 +1,10 @@
esphome:
name: test
esp32:
variant: esp32s3
flash_mode: opi
framework:
type: esp-idf
advanced:
flash_chip: mxic_opi
@@ -0,0 +1,8 @@
esphome:
name: test
esp32:
variant: esp32s3
flash_mode: opi
framework:
type: esp-idf
+116
View File
@@ -10,6 +10,7 @@ from typing import Any
import pytest
from esphome.components.esp32 import (
ESP32_FLASH_CHIPS,
KEY_FATFS_REQUIRED,
KEY_MBEDTLS_TLS_EXTRAS_REQUIRED,
KEY_MBEDTLS_TLS_SERVER_REQUIRED,
@@ -252,6 +253,51 @@ def test_esp32_rejects_unsupported_cli_toolchain(
r"value must be at most 5 .* @ data\['framework'\]\['advanced'\]\['nvs_encryption'\]\['key_id'\]",
id="nvs_encryption_key_id_out_of_range",
),
pytest.param(
{
"variant": "esp32",
"board": "esp32dev",
"framework": {
"type": "esp-idf",
"advanced": {"flash_chip": "mxic_opi"},
},
},
r"'flash_chip: mxic_opi' is only supported on ESP32S3 @ data\['framework'\]\['advanced'\]\['flash_chip'\]",
id="flash_chip_mxic_opi_only_on_s3",
),
pytest.param(
{
"variant": "esp32s3",
"flash_mode": "opi",
"framework": {
"type": "esp-idf",
"advanced": {"flash_chip": "gd"},
},
},
r"'flash_chip: gd' does not match 'flash_mode: opi'; octal flash uses mxic_opi @ data\['framework'\]\['advanced'\]\['flash_chip'\]",
id="flash_chip_must_match_opi_mode",
),
pytest.param(
{
"variant": "esp32s3",
"framework": {
"type": "esp-idf",
"advanced": {"flash_chip": "mxic_opi"},
},
},
r"'flash_chip: mxic_opi' requires 'flash_mode: opi' @ data\['framework'\]\['advanced'\]\['flash_chip'\]",
id="flash_chip_mxic_opi_requires_opi_mode",
),
pytest.param(
{
"variant": "esp32",
"board": "esp32dev",
"flash_mode": "opi",
"framework": {"type": "esp-idf"},
},
r"'flash_mode: opi' is only supported on ESP32S3 @ data\['flash_mode'\]",
id="flash_mode_opi_only_on_s3",
),
],
)
def test_esp32_configuration_errors(
@@ -658,6 +704,27 @@ def test_platformio_arduino_enables_reproducible_build(
assert sdkconfig.get("CONFIG_APP_REPRODUCIBLE_BUILD") is True
@pytest.mark.parametrize(
("config_file", "expected"),
[
("reproducible_build.yaml", True),
("reproducible_build_arduino.yaml", True),
("file_macro_idf_5_0.yaml", False),
],
)
def test_file_macro_is_basename_only(
generate_main: Callable[[str | Path], str],
component_config_path: Callable[[str], Path],
config_file: str,
expected: bool,
) -> None:
"""__FILE__ becomes the basename on GCC 12 toolchains; IDF 5.0 (GCC 11) is skipped."""
generate_main(component_config_path(config_file))
assert ("-D__FILE__=__FILE_NAME__" in CORE.build_flags) is expected
assert ("-Wno-builtin-macro-redefined" in CORE.build_flags) is expected
def test_native_idf_enables_reproducible_build(
component_config_path: Callable[[str], Path],
) -> None:
@@ -683,10 +750,59 @@ def test_flash_mode_sets_sdkconfig_and_pio_option(
sdkconfig = CORE.data[KEY_ESP32][KEY_SDKCONFIG_OPTIONS]
assert sdkconfig.get("CONFIG_ESPTOOLPY_FLASHMODE_QIO") is True
assert sdkconfig.get("CONFIG_ESPTOOLPY_FLASHFREQ_80M") is True
assert sdkconfig.get("CONFIG_ESPTOOLPY_OCT_FLASH") is False
assert CORE.platformio_options.get("board_build.flash_mode") == "qio"
assert CORE.platformio_options.get("board_build.f_flash") == "80000000L"
@pytest.mark.parametrize(
("config_file", "enabled"),
[
pytest.param("flash_chip_gd.yaml", "CONFIG_SPI_FLASH_SUPPORT_GD_CHIP", id="gd"),
pytest.param("flash_chip_generic.yaml", None, id="generic"),
pytest.param(
"flash_chip_mxic_opi_s3.yaml",
"CONFIG_SPI_FLASH_SUPPORT_MXIC_OPI_CHIP",
id="mxic_opi_s3",
),
],
)
def test_flash_chip_keeps_one_vendor_driver(
generate_main: Callable[[str | Path], str],
component_config_path: Callable[[str], Path],
config_file: str,
enabled: str | None,
) -> None:
"""flash_chip enables only the chosen vendor driver."""
generate_main(component_config_path(config_file))
sdkconfig = CORE.data[KEY_ESP32][KEY_SDKCONFIG_OPTIONS]
vendors = {
k: v for k, v in sdkconfig.items() if k.startswith("CONFIG_SPI_FLASH_SUPPORT_")
}
assert vendors == {flag: flag == enabled for flag in ESP32_FLASH_CHIPS.values()}
def test_flash_chip_unset_keeps_idf_defaults(
generate_main: Callable[[str | Path], str],
component_config_path: Callable[[str], Path],
) -> None:
"""Without flash_chip every vendor driver stays at its ESP-IDF default."""
generate_main(component_config_path("flash_mode_default.yaml"))
sdkconfig = CORE.data[KEY_ESP32][KEY_SDKCONFIG_OPTIONS]
assert not any(key.startswith("CONFIG_SPI_FLASH_SUPPORT_") for key in sdkconfig)
def test_flash_mode_opi_enables_octal_flash(
generate_main: Callable[[str | Path], str],
component_config_path: Callable[[str], Path],
) -> None:
"""flash_mode: opi needs the octal flash switch or ESP-IDF ignores the mode."""
generate_main(component_config_path("flash_mode_opi_s3.yaml"))
sdkconfig = CORE.data[KEY_ESP32][KEY_SDKCONFIG_OPTIONS]
assert sdkconfig.get("CONFIG_ESPTOOLPY_FLASHMODE_OPI") is True
assert sdkconfig.get("CONFIG_ESPTOOLPY_OCT_FLASH") is True
def test_flash_mode_unset_leaves_defaults(
generate_main: Callable[[str | Path], str],
component_config_path: Callable[[str], Path],
@@ -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,
}
)
@@ -319,6 +319,32 @@ def test_encryption_with_captive_portal_does_not_warn(
fv.full_config.reset(token)
@pytest.mark.parametrize("extra", [{}, {"prometheus": {}}])
def test_encryption_with_web_server_ota_disabled_does_not_warn(
caplog: pytest.LogCaptureFixture, extra: dict[str, Any]
) -> None:
"""web_server `ota: false` only serves /update while the captive portal is
active, on every listener, so there is no plaintext endpoint to warn about."""
full_conf = {
"web_server": {CONF_OTA: False},
**extra,
CONF_OTA: [
_make_ota_config(port=3232, **{CONF_ENCRYPTION: {CONF_KEY: OTHER_KEY}}),
{CONF_PLATFORM: "web_server", CONF_ID: ID("ota_ws", is_manual=False)},
],
}
token = fv.full_config.set(full_conf)
try:
with caplog.at_level(logging.WARNING):
ota_esphome_final_validate({})
assert not any(
"OTA encryption does not cover" in record.message
for record in caplog.records
)
finally:
fv.full_config.reset(token)
def test_password_with_api_key_warns(caplog: pytest.LogCaptureFixture) -> None:
"""A static api key makes the device offer encryption and the CLI take
it, so the password is dead weight; the config validates with a warning."""
@@ -0,0 +1,9 @@
esphome:
name: test
bk72xx:
board: generic-bk7252
remote_receiver:
- id: rcvr
pin: P6
@@ -0,0 +1,12 @@
esphome:
name: test
esp32:
board: esp32-c61-devkitc1
variant: esp32c61
framework:
type: esp-idf
remote_receiver:
- id: rcvr
pin: GPIO4
@@ -0,0 +1,9 @@
esphome:
name: test
ln882x:
board: generic-ln882h
remote_receiver:
- id: rcvr
pin: PA4
@@ -0,0 +1,9 @@
esphome:
name: test
rp2:
board: rpipicow
remote_receiver:
- id: rcvr
pin: GPIO4
@@ -0,0 +1,9 @@
esphome:
name: test
rtl87xx:
board: generic-rtl8710bn-2mb-788k
remote_receiver:
- id: rcvr
pin: PA12
@@ -0,0 +1,36 @@
esphome:
name: test
esp32:
board: esp32dev
logger:
external_components:
- source:
type: local
path: ../external_components
fake_protocol:
remote_receiver:
- id: rcvr
pin: GPIO4
dump:
- fake
- nec
on_fake:
then:
- remote_transmitter.transmit_fake:
on_nec:
then:
- logger.log: nec
remote_transmitter:
pin: GPIO5
carrier_duty_percent: 50%
binary_sensor:
- platform: remote_receiver
name: Fake Input
fake:
@@ -0,0 +1,39 @@
"""External component registering a protocol that has no source file in remote_base."""
import esphome.codegen as cg
from esphome.components import remote_base
import esphome.config_validation as cv
from esphome.types import ConfigType
DEPENDENCIES = ["remote_base"]
ns = cg.esphome_ns.namespace("fake_protocol")
FakeData = ns.struct("FakeData")
FakeBinarySensor = ns.class_(
"FakeBinarySensor", remote_base.RemoteReceiverBinarySensorBase
)
FakeTrigger = ns.class_("FakeTrigger", remote_base.RemoteReceiverTrigger)
FakeAction = ns.class_("FakeAction", remote_base.RemoteTransmitterActionBase)
FakeDumper = ns.class_("FakeDumper", remote_base.RemoteReceiverDumperBase)
CONFIG_SCHEMA = cv.Schema({})
@remote_base.register_binary_sensor("fake", FakeBinarySensor, {})
def fake_binary_sensor(var: cg.MockObj, config: ConfigType) -> None:
pass
@remote_base.register_trigger("fake", FakeTrigger, FakeData)
def fake_trigger(var: cg.MockObj, config: ConfigType) -> None:
pass
@remote_base.register_dumper("fake", FakeDumper)
def fake_dumper(var: cg.MockObj, config: ConfigType) -> None:
pass
@remote_base.register_action("fake", FakeAction, {})
async def fake_action(var: cg.MockObj, config: ConfigType, args: list) -> None:
pass
@@ -1,8 +1,16 @@
"""buffer_size reaches the receiver when set, and always on the pulse ring targets."""
"""buffer_size is bytes on the pulse ring targets and only reaches RMT targets when set."""
from collections.abc import Callable
from pathlib import Path
import pytest
from esphome.components import remote_receiver
from esphome.components.esp8266 import gpio as esp8266_gpio # noqa: F401 registers the pin schema
from esphome.config_validation import Invalid
from esphome.const import PlatformFramework
from tests.component_tests.types import SetCoreConfigCallable
def test_explicit_buffer_size_is_passed_through(
generate_main: Callable[[str | Path], str],
@@ -12,17 +20,29 @@ def test_explicit_buffer_size_is_passed_through(
assert "rcvr->set_buffer_size(2000);" in main_cpp
def test_pulse_ring_target_keeps_a_default(
@pytest.mark.parametrize(
"target", ["esp8266", "rp2", "bk72xx", "rtl87xx", "ln882x", "esp32_c2", "esp32_c61"]
)
def test_pulse_ring_default_holds_1000_pulses(
generate_main: Callable[[str | Path], str],
component_config_path: Callable[[str], Path],
target: str,
) -> None:
main_cpp = generate_main(component_config_path("receiver_esp8266.yaml"))
assert "rcvr->set_buffer_size(1000);" in main_cpp
main_cpp = generate_main(component_config_path(f"receiver_{target}.yaml"))
assert "rcvr->set_buffer_size(4000);" in main_cpp
def test_esp32_variant_without_rmt_keeps_a_default(
generate_main: Callable[[str | Path], str],
component_config_path: Callable[[str], Path],
@pytest.mark.parametrize(
("value", "expected"),
[("32b", None), ("64b", 64), ("65b", 65), ("65535b", 65535), ("65536b", None)],
)
def test_buffer_size_range(
set_core_config: SetCoreConfigCallable, value: str, expected: int | None
) -> None:
main_cpp = generate_main(component_config_path("receiver_esp32_c2.yaml"))
assert "rcvr->set_buffer_size(1000);" in main_cpp
set_core_config(PlatformFramework.ESP8266_ARDUINO)
config = {"pin": "GPIO4", "buffer_size": value}
if expected is None:
with pytest.raises(Invalid):
remote_receiver.CONFIG_SCHEMA(config)
else:
assert remote_receiver.CONFIG_SCHEMA(config)["buffer_size"] == expected
@@ -1,13 +1,16 @@
"""Listener and dumper StaticVector sizes come from codegen slot counts."""
from collections.abc import Callable
from collections.abc import Callable, Generator
from pathlib import Path
import sys
import pytest
from esphome import loader
from esphome.automation import ACTION_REGISTRY
from esphome.components import remote_base
import esphome.config_validation as cv
from esphome.core import CORE
from ..helpers import get_define_value
@@ -74,6 +77,47 @@ def test_every_registry_name_maps_to_a_protocol_source() -> None:
assert remote_base._protocol_stem(name) in remote_base._PROTOCOL_STEMS, name
@pytest.fixture
def restore_protocol_registries() -> Generator[None]:
"""Loading an external protocol component adds to module-level registries; undo that.
The loader caches the component too, so drop it or a second load would skip the
decorators and leave the restored registries without the external names.
"""
registries = (
remote_base.BINARY_SENSOR_REGISTRY,
remote_base.TRIGGER_REGISTRY,
remote_base.DUMPER_REGISTRY,
ACTION_REGISTRY,
)
saved = [dict(registry) for registry in registries]
yield
for registry, entries in zip(registries, saved, strict=True):
registry.clear()
registry.update(entries)
loader._COMPONENT_CACHE.pop("fake_protocol", None)
sys.modules.pop("esphome.components.fake_protocol", None)
@pytest.mark.usefixtures("restore_protocol_registries")
def test_external_protocols_register_without_a_remote_base_source(
generate_main: Callable[[str | Path], str],
component_config_path: Callable[[str], Path],
) -> None:
"""An external protocol goes through all four decorators without a source file here, so no define is emitted."""
main_cpp = generate_main(
component_config_path("receiver_with_external_protocol.yaml")
)
defines = {define.name for define in CORE.defines}
assert "USE_REMOTE_PROTOCOL_NEC" in defines
assert "USE_REMOTE_PROTOCOL_FAKE" not in defines
for cls in ("FakeBinarySensor", "FakeTrigger", "FakeDumper", "FakeAction"):
assert f"fake_protocol::{cls}" in main_cpp, cls
# fake and nec dumpers; on_fake and on_nec triggers plus the fake binary sensor
assert get_define_value("REMOTE_BASE_DUMPER_COUNT") == "2"
assert get_define_value("REMOTE_BASE_LISTENER_COUNT") == "3"
def test_request_protocol_rejects_unknown_names() -> None:
"""A misspelled protocol would otherwise surface only as a link error."""
with pytest.raises(ValueError, match="Unknown remote protocol 'toshiba'"):
@@ -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
@@ -22,6 +22,7 @@ esp32:
disable_regi2c_in_iram: true
disable_fatfs: true
sram1_as_iram: true
flash_chip: gd
watchdog_timeout: 7s
wifi:
@@ -9,6 +9,7 @@ esp32:
type: esp-idf
advanced:
execute_from_psram: true
flash_chip: gd
disable_libc_locks_in_iram: true # Test default RAM optimization enabled
disable_debug_stubs: true
disable_ocd_aware: true
+24
View File
@@ -0,0 +1,24 @@
touchscreen:
- platform: icnt86
i2c_id: i2c_bus
interrupt_pin: ${interrupt_pin_touch}
reset_pin: ${reset_pin_touch}
display: epaper
on_touch:
- logger.log:
format: Touch at (%d, %d)
args: [touch.x, touch.y]
display:
- platform: waveshare_epaper
id: epaper
rotation: 90
cs_pin: ${cs_pin_display}
dc_pin: ${dc_pin_display}
busy_pin: ${busy_pin_display}
reset_pin: ${reset_pin_display}
model: 2.90inv2-r2
pages:
- id: icnt86_page
lambda: |-
it.rectangle(0, 0, it.get_width(), it.get_height());
@@ -0,0 +1,14 @@
substitutions:
interrupt_pin_touch: GPIO4
reset_pin_touch: GPIO32
cs_pin_display: GPIO33
dc_pin_display: GPIO21
busy_pin_display: GPIO27
reset_pin_display: GPIO14
clk_pin: GPIO25
mosi_pin: GPIO26
packages:
i2c: !include ../../test_build_components/common/i2c/esp32-idf.yaml
spi: !include ../../test_build_components/common/spi/esp32-idf.yaml
icnt86: !include common.yaml
@@ -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}}
+1 -1
View File
@@ -3,7 +3,7 @@ esphome:
then:
- mixer_speaker.apply_ducking:
id: source_speaker_1_id
decibel_reduction: 10
decibel_reduction: 255
duration: 1s
speaker:
@@ -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
@@ -0,0 +1,6 @@
packages:
sendspin: !include common.yaml
switch:
- platform: sendspin
name: "Sendspin Enabled"
@@ -0,0 +1,2 @@
packages:
sendspin: !include common-switch.yaml
+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
@@ -0,0 +1,53 @@
esphome:
name: test-sensor-raw-state
host:
api:
batch_delay: 0ms # Disable batching to receive all state updates
logger:
level: DEBUG
# Filters are compiled in for this config (USE_SENSOR_FILTER), so raw storage exists
sensor:
# No filters on this sensor: get_raw_state() must equal state
- platform: template
name: "No Filter Sensor"
id: no_filter_sensor
accuracy_decimals: 1
# Filtered sensor: get_raw_state() must be the pre-filter value
- platform: template
name: "With Filter Sensor"
id: with_filter_sensor
accuracy_decimals: 1
filters:
- multiply: 2.0
button:
- platform: template
name: "Test No Filter Button"
id: test_no_filter_button
on_press:
- sensor.template.publish:
id: no_filter_sensor
state: 21.5
- delay: 50ms
- logger.log:
format: "NO_FILTER: state=%.1f raw_state=%.1f"
args:
- id(no_filter_sensor).state
- id(no_filter_sensor).get_raw_state()
- platform: template
name: "Test With Filter Button"
id: test_with_filter_button
on_press:
- sensor.template.publish:
id: with_filter_sensor
state: 21.5
- delay: 50ms
- logger.log:
format: "WITH_FILTER: state=%.1f raw_state=%.1f"
args:
- id(with_filter_sensor).state
- id(with_filter_sensor).get_raw_state()
@@ -0,0 +1,31 @@
esphome:
name: test-sensor-raw-state-no-filter
host:
api:
batch_delay: 0ms # Disable batching to receive all state updates
logger:
level: DEBUG
# No sensor in this config has filters, so USE_SENSOR_FILTER is not defined and
# get_raw_state() falls back to state
sensor:
- platform: template
name: "No Filter Sensor"
id: no_filter_sensor
accuracy_decimals: 1
button:
- platform: template
name: "Test No Filter Button"
id: test_no_filter_button
on_press:
- sensor.template.publish:
id: no_filter_sensor
state: 21.5
- delay: 50ms
- logger.log:
format: "NO_FILTER: state=%.1f raw_state=%.1f"
args:
- id(no_filter_sensor).state
- id(no_filter_sensor).get_raw_state()
@@ -17,10 +17,10 @@ uart:
baud_rate: 115200
port: /dev/null
# Shared 3-bus mesh (see the shared_yaml markers): addr 1 = typed read-only
# registers, addr 5 = the read/write 0x17 target, addr 2/3 on the second
# server hub. auto_start everywhere: the controller polls at boot, so the
# forwarding must already be live or early requests generate warnings.
# Shared 3-bus mesh (see the shared_yaml markers): addr 1 = typed registers
# backed by writable globals, addr 5 = the read/write 0x17 target, addr 2/3/6
# on the second server hub. auto_start everywhere: the controller polls at
# boot, so the forwarding must already be live or early requests generate warnings.
# Every test presses Start Scenario, so all merged actions fire in every test.
uart_mock:
- id: virtual_uart_server
@@ -64,6 +64,54 @@ globals:
- id: stored_1
type: uint16_t
initial_value: "0"
- id: stored_u_word
type: uint16_t
initial_value: "99"
- id: stored_u_word_s
type: uint16_t
initial_value: "4660"
- id: stored_s_word
type: int16_t
initial_value: "-99"
- id: stored_s_word_s
type: int16_t
initial_value: "-2"
- id: stored_u_dword
type: uint32_t
initial_value: "16909060"
- id: stored_s_dword
type: int32_t
initial_value: "-16909060"
- id: stored_u_dword_r
type: uint32_t
initial_value: "67305985"
- id: stored_s_dword_r
type: int32_t
initial_value: "-67305985"
- id: stored_u_qword
type: uint64_t
initial_value: "72623859790382856"
- id: stored_s_qword
type: int64_t
initial_value: "-72623859790382856"
- id: stored_u_qword_r
type: uint64_t
initial_value: "578437695752307201"
- id: stored_s_qword_r
type: int64_t
initial_value: "-578437695752307201"
- id: stored_fp32
type: float
initial_value: "3.14"
- id: stored_fp32_r
type: float
initial_value: "2.5"
- id: stored_bit_2
type: bool
initial_value: "false"
- id: stored_bit_3
type: bool
initial_value: "true"
modbus:
- uart_id: virtual_uart_server
@@ -90,6 +138,10 @@ modbus_controller:
modbus_id: virtual_modbus_client
id: modbus_controller_3
update_interval: 1s
- address: 6
modbus_id: virtual_modbus_client
id: modbus_controller_6
update_interval: 1s
modbus_server:
- address: 1
@@ -97,46 +149,60 @@ modbus_server:
registers:
- address: 0x01
value_type: U_WORD
read_lambda: return 99;
read_lambda: return id(stored_u_word);
write_lambda: id(stored_u_word) = x; return true;
- address: 0x02
value_type: U_WORD_S
read_lambda: return 4660;
read_lambda: return id(stored_u_word_s);
write_lambda: id(stored_u_word_s) = x; return true;
- address: 0x03
value_type: S_WORD
read_lambda: return -99;
read_lambda: return id(stored_s_word);
write_lambda: id(stored_s_word) = x; return true;
- address: 0x04
value_type: S_WORD_S
read_lambda: return -2;
read_lambda: return id(stored_s_word_s);
write_lambda: id(stored_s_word_s) = x; return true;
- address: 0x05
value_type: U_DWORD
read_lambda: return 16909060;
read_lambda: return id(stored_u_dword);
write_lambda: id(stored_u_dword) = x; return true;
- address: 0x08
value_type: S_DWORD
read_lambda: return -16909060;
read_lambda: return id(stored_s_dword);
write_lambda: id(stored_s_dword) = x; return true;
- address: 0x0B
value_type: U_DWORD_R
read_lambda: return 67305985;
read_lambda: return id(stored_u_dword_r);
write_lambda: id(stored_u_dword_r) = x; return true;
- address: 0x0E
value_type: S_DWORD_R
read_lambda: return -67305985;
read_lambda: return id(stored_s_dword_r);
write_lambda: id(stored_s_dword_r) = x; return true;
- address: 0x11
value_type: U_QWORD
read_lambda: return 72623859790382856;
read_lambda: return id(stored_u_qword);
write_lambda: id(stored_u_qword) = x; return true;
- address: 0x16
value_type: S_QWORD
read_lambda: return -72623859790382856;
read_lambda: return id(stored_s_qword);
write_lambda: id(stored_s_qword) = x; return true;
- address: 0x1B
value_type: U_QWORD_R
read_lambda: return 578437695752307201;
read_lambda: return id(stored_u_qword_r);
write_lambda: id(stored_u_qword_r) = x; return true;
- address: 0x20
value_type: S_QWORD_R
read_lambda: return -578437695752307201;
read_lambda: return id(stored_s_qword_r);
write_lambda: id(stored_s_qword_r) = x; return true;
- address: 0x25
value_type: FP32
read_lambda: return 3.14;
read_lambda: return id(stored_fp32);
write_lambda: id(stored_fp32) = x; return true;
- address: 0x28
value_type: FP32_R
read_lambda: return 3.14;
read_lambda: return id(stored_fp32_r);
write_lambda: id(stored_fp32_r) = x; return true;
- address: 5
modbus_id: virtual_modbus_server
registers:
@@ -165,6 +231,19 @@ modbus_server:
- address: 0x01
value_type: U_WORD
read_lambda: return 929;
- address: 6
modbus_id: virtual_modbus_server_2
bits:
- address: 0x00
read_lambda: return true;
- address: 0x01
read_lambda: return false;
- address: 0x02
read_lambda: return id(stored_bit_2);
write_lambda: id(stored_bit_2) = x; return true;
- address: 0x03
read_lambda: return id(stored_bit_3);
write_lambda: id(stored_bit_3) = x; return true;
sensor:
- platform: modbus_controller
@@ -280,6 +359,183 @@ sensor:
name: "client_read_1"
id: client_read_1
# The number schema caps min/max at 16777215 (float32 integer precision), so
# the large dword/qword baselines cannot be written back through these numbers.
number:
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "write_u_word"
address: 0x01
register_type: holding
value_type: U_WORD
min_value: 0
max_value: 65535
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "write_u_word_s"
address: 0x02
register_type: holding
value_type: U_WORD_S
min_value: 0
max_value: 65535
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "write_s_word"
address: 0x03
register_type: holding
value_type: S_WORD
min_value: -16777215
max_value: 16777215
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "write_s_word_s"
address: 0x04
register_type: holding
value_type: S_WORD_S
min_value: -16777215
max_value: 16777215
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "write_u_dword"
address: 0x05
register_type: holding
value_type: U_DWORD
min_value: 0
max_value: 16777215
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "write_s_dword"
address: 0x08
register_type: holding
value_type: S_DWORD
min_value: -16777215
max_value: 16777215
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "write_u_dword_r"
address: 0x0B
register_type: holding
value_type: U_DWORD_R
min_value: 0
max_value: 16777215
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "write_s_dword_r"
address: 0x0E
register_type: holding
value_type: S_DWORD_R
min_value: -16777215
max_value: 16777215
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "write_u_qword"
address: 0x11
register_type: holding
value_type: U_QWORD
min_value: 0
max_value: 16777215
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "write_s_qword"
address: 0x16
register_type: holding
value_type: S_QWORD
min_value: -16777215
max_value: 16777215
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "write_u_qword_r"
address: 0x1B
register_type: holding
value_type: U_QWORD_R
min_value: 0
max_value: 16777215
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "write_s_qword_r"
address: 0x20
register_type: holding
value_type: S_QWORD_R
min_value: -16777215
max_value: 16777215
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "write_fp32"
address: 0x25
register_type: holding
value_type: FP32
min_value: -16777215
max_value: 16777215
step: 0.01
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "write_fp32_r"
address: 0x28
register_type: holding
value_type: FP32_R
min_value: -16777215
max_value: 16777215
step: 0.01
# The four bits are read both as coils (FC 0x01) and discrete inputs (FC 0x02);
# the server serves both from one shared table, so the two views must agree.
binary_sensor:
- platform: modbus_controller
modbus_controller_id: modbus_controller_6
name: "bit_coil_0"
address: 0x00
register_type: coil
- platform: modbus_controller
modbus_controller_id: modbus_controller_6
name: "bit_coil_1"
address: 0x01
register_type: coil
- platform: modbus_controller
modbus_controller_id: modbus_controller_6
name: "bit_coil_2"
address: 0x02
register_type: coil
- platform: modbus_controller
modbus_controller_id: modbus_controller_6
name: "bit_coil_3"
address: 0x03
register_type: coil
- platform: modbus_controller
modbus_controller_id: modbus_controller_6
name: "bit_di_0"
address: 0x00
register_type: discrete_input
- platform: modbus_controller
modbus_controller_id: modbus_controller_6
name: "bit_di_1"
address: 0x01
register_type: discrete_input
- platform: modbus_controller
modbus_controller_id: modbus_controller_6
name: "bit_di_2"
address: 0x02
register_type: discrete_input
- platform: modbus_controller
modbus_controller_id: modbus_controller_6
name: "bit_di_3"
address: 0x03
register_type: discrete_input
# write_bit_2 uses the single-coil write (FC 0x05); write_bit_3 opts into the
# multiple-coils write (FC 0x0F) so both server write paths are exercised.
switch:
- platform: modbus_controller
modbus_controller_id: modbus_controller_6
name: "write_bit_2"
address: 0x02
register_type: coil
- platform: modbus_controller
modbus_controller_id: modbus_controller_6
name: "write_bit_3"
address: 0x03
register_type: coil
use_write_multiple: true
button:
- platform: template
name: "Start Scenario"
@@ -1,147 +0,0 @@
esphome:
name: uart-mock-modbus-srv-bits
host:
api:
logger:
level: VERBOSE
external_components:
- source:
type: local
path: EXTERNAL_COMPONENT_PATH
# Dummy uart entry to satisfy modbus's DEPENDENCIES = ["uart"]
# The actual UART bus used is the uart_mock component below
uart:
baud_rate: 115200
port: /dev/null
uart_mock:
- id: virtual_uart_server
baud_rate: 9600
# auto_start must be true for loopback fixtures: the modbus controller
# polls on its update_interval immediately at boot, so the uart_mock
# forwarding must already be active or early requests are lost and
# generate modbus warnings.
auto_start: true
debug:
on_tx:
- then:
- uart_mock.inject_rx:
id: virtual_uart_controller
data: !lambda return data;
- id: virtual_uart_controller
baud_rate: 9600
auto_start: true # See comment on virtual_uart_server above
debug:
on_tx:
- then:
- uart_mock.inject_rx:
id: virtual_uart_server
data: !lambda return data;
globals:
- id: stored_bit_2
type: bool
initial_value: "false"
- id: stored_bit_3
type: bool
initial_value: "true"
modbus:
- uart_id: virtual_uart_server
id: virtual_modbus_server
role: server
- uart_id: virtual_uart_controller
id: virtual_modbus_controller
role: client
turnaround_time: 10ms
modbus_controller:
- address: 1
modbus_id: virtual_modbus_controller
update_interval: 1s
id: modbus_controller_1
modbus_server:
- address: 1
modbus_id: virtual_modbus_server
id: modbus_server_1
bits:
- address: 0x00
read_lambda: return true;
- address: 0x01
read_lambda: return false;
- address: 0x02
read_lambda: return id(stored_bit_2);
write_lambda: id(stored_bit_2) = x; return true;
- address: 0x03
read_lambda: return id(stored_bit_3);
write_lambda: id(stored_bit_3) = x; return true;
# The same four bits are read both as coils (FC 0x01) and as discrete inputs
# (FC 0x02): the server serves both from one shared bit table, so the two
# views must always agree.
binary_sensor:
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "bit_coil_0"
address: 0x00
register_type: coil
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "bit_coil_1"
address: 0x01
register_type: coil
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "bit_coil_2"
address: 0x02
register_type: coil
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "bit_coil_3"
address: 0x03
register_type: coil
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "bit_di_0"
address: 0x00
register_type: discrete_input
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "bit_di_1"
address: 0x01
register_type: discrete_input
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "bit_di_2"
address: 0x02
register_type: discrete_input
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "bit_di_3"
address: 0x03
register_type: discrete_input
# write_bit_2 uses the single-coil write (FC 0x05); write_bit_3 opts into the
# multiple-coils write (FC 0x0F) so both server write paths are exercised.
switch:
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "write_bit_2"
address: 0x02
register_type: coil
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "write_bit_3"
address: 0x03
register_type: coil
use_write_multiple: true
button:
- platform: template
name: "Start Scenario"
id: start_scenario_btn
# This test does not have anything to start (mock is autostart)
@@ -1,371 +0,0 @@
esphome:
name: uart-mock-modbus-srv-write
host:
api:
logger:
level: VERBOSE
external_components:
- source:
type: local
path: EXTERNAL_COMPONENT_PATH
# Dummy uart entry to satisfy modbus's DEPENDENCIES = ["uart"]
# The actual UART bus used is the uart_mock component below
uart:
baud_rate: 115200
port: /dev/null
uart_mock:
- id: virtual_uart_server
baud_rate: 9600
# auto_start must be true for loopback fixtures: the modbus controller
# polls on its update_interval immediately at boot, so the uart_mock
# forwarding must already be active or early requests are lost and
# generate modbus warnings.
auto_start: true
debug:
on_tx:
- then:
- uart_mock.inject_rx:
id: virtual_uart_controller
data: !lambda return data;
- id: virtual_uart_controller
baud_rate: 9600
auto_start: true # See comment on virtual_uart_server above
debug:
on_tx:
- then:
- uart_mock.inject_rx:
id: virtual_uart_server
data: !lambda return data;
globals:
- id: stored_u_word
type: uint16_t
initial_value: "11"
- id: stored_u_word_s
type: uint16_t
initial_value: "4660"
- id: stored_s_word
type: int16_t
initial_value: "-11"
- id: stored_s_word_s
type: int16_t
initial_value: "-2"
- id: stored_u_dword
type: uint32_t
initial_value: "1001"
- id: stored_s_dword
type: int32_t
initial_value: "-1001"
- id: stored_u_dword_r
type: uint32_t
initial_value: "3003"
- id: stored_s_dword_r
type: int32_t
initial_value: "-3003"
- id: stored_u_qword
type: uint64_t
initial_value: "5005"
- id: stored_s_qword
type: int64_t
initial_value: "-5005"
- id: stored_u_qword_r
type: uint64_t
initial_value: "7007"
- id: stored_s_qword_r
type: int64_t
initial_value: "-7007"
- id: stored_fp32
type: float
initial_value: "1.5"
- id: stored_fp32_r
type: float
initial_value: "2.5"
modbus:
- uart_id: virtual_uart_server
id: virtual_modbus_server
role: server
- uart_id: virtual_uart_controller
id: virtual_modbus_controller
role: client
turnaround_time: 10ms
modbus_controller:
- address: 1
modbus_id: virtual_modbus_controller
update_interval: 2s
id: modbus_controller_1
modbus_server:
- address: 1
modbus_id: virtual_modbus_server
id: modbus_server_1
registers:
- address: 0x01
value_type: U_WORD
read_lambda: return id(stored_u_word);
write_lambda: id(stored_u_word) = x; return true;
- address: 0x02
value_type: U_WORD_S
read_lambda: return id(stored_u_word_s);
write_lambda: id(stored_u_word_s) = x; return true;
- address: 0x03
value_type: S_WORD
read_lambda: return id(stored_s_word);
write_lambda: id(stored_s_word) = x; return true;
- address: 0x04
value_type: S_WORD_S
read_lambda: return id(stored_s_word_s);
write_lambda: id(stored_s_word_s) = x; return true;
- address: 0x05
value_type: U_DWORD
read_lambda: return id(stored_u_dword);
write_lambda: id(stored_u_dword) = x; return true;
- address: 0x08
value_type: S_DWORD
read_lambda: return id(stored_s_dword);
write_lambda: id(stored_s_dword) = x; return true;
- address: 0x0B
value_type: U_DWORD_R
read_lambda: return id(stored_u_dword_r);
write_lambda: id(stored_u_dword_r) = x; return true;
- address: 0x0E
value_type: S_DWORD_R
read_lambda: return id(stored_s_dword_r);
write_lambda: id(stored_s_dword_r) = x; return true;
- address: 0x11
value_type: U_QWORD
read_lambda: return id(stored_u_qword);
write_lambda: id(stored_u_qword) = x; return true;
- address: 0x16
value_type: S_QWORD
read_lambda: return id(stored_s_qword);
write_lambda: id(stored_s_qword) = x; return true;
- address: 0x1B
value_type: U_QWORD_R
read_lambda: return id(stored_u_qword_r);
write_lambda: id(stored_u_qword_r) = x; return true;
- address: 0x20
value_type: S_QWORD_R
read_lambda: return id(stored_s_qword_r);
write_lambda: id(stored_s_qword_r) = x; return true;
- address: 0x25
value_type: FP32
read_lambda: return id(stored_fp32);
write_lambda: id(stored_fp32) = x; return true;
- address: 0x28
value_type: FP32_R
read_lambda: return id(stored_fp32_r);
write_lambda: id(stored_fp32_r) = x; return true;
sensor:
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "reg_u_word"
address: 0x01
register_type: holding
value_type: U_WORD
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "reg_u_word_s"
address: 0x02
register_type: holding
value_type: U_WORD_S
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "reg_s_word"
address: 0x03
register_type: holding
value_type: S_WORD
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "reg_s_word_s"
address: 0x04
register_type: holding
value_type: S_WORD_S
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "reg_u_dword"
address: 0x05
register_type: holding
value_type: U_DWORD
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "reg_s_dword"
address: 0x08
register_type: holding
value_type: S_DWORD
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "reg_u_dword_r"
address: 0x0B
register_type: holding
value_type: U_DWORD_R
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "reg_s_dword_r"
address: 0x0E
register_type: holding
value_type: S_DWORD_R
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "reg_u_qword"
address: 0x11
register_type: holding
value_type: U_QWORD
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "reg_s_qword"
address: 0x16
register_type: holding
value_type: S_QWORD
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "reg_u_qword_r"
address: 0x1B
register_type: holding
value_type: U_QWORD_R
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "reg_s_qword_r"
address: 0x20
register_type: holding
value_type: S_QWORD_R
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "reg_fp32"
address: 0x25
register_type: holding
value_type: FP32
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "reg_fp32_r"
address: 0x28
register_type: holding
value_type: FP32_R
number:
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "write_u_word"
address: 0x01
register_type: holding
value_type: U_WORD
min_value: 0
max_value: 65535
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "write_u_word_s"
address: 0x02
register_type: holding
value_type: U_WORD_S
min_value: 0
max_value: 65535
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "write_s_word"
address: 0x03
register_type: holding
value_type: S_WORD
min_value: -16777215
max_value: 16777215
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "write_s_word_s"
address: 0x04
register_type: holding
value_type: S_WORD_S
min_value: -16777215
max_value: 16777215
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "write_u_dword"
address: 0x05
register_type: holding
value_type: U_DWORD
min_value: 0
max_value: 16777215
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "write_s_dword"
address: 0x08
register_type: holding
value_type: S_DWORD
min_value: -16777215
max_value: 16777215
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "write_u_dword_r"
address: 0x0B
register_type: holding
value_type: U_DWORD_R
min_value: 0
max_value: 16777215
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "write_s_dword_r"
address: 0x0E
register_type: holding
value_type: S_DWORD_R
min_value: -16777215
max_value: 16777215
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "write_u_qword"
address: 0x11
register_type: holding
value_type: U_QWORD
min_value: 0
max_value: 16777215
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "write_s_qword"
address: 0x16
register_type: holding
value_type: S_QWORD
min_value: -16777215
max_value: 16777215
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "write_u_qword_r"
address: 0x1B
register_type: holding
value_type: U_QWORD_R
min_value: 0
max_value: 16777215
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "write_s_qword_r"
address: 0x20
register_type: holding
value_type: S_QWORD_R
min_value: -16777215
max_value: 16777215
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "write_fp32"
address: 0x25
register_type: holding
value_type: FP32
min_value: -16777215
max_value: 16777215
step: 0.01
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "write_fp32_r"
address: 0x28
register_type: holding
value_type: FP32_R
min_value: -16777215
max_value: 16777215
step: 0.01
button:
- platform: template
name: "Start Scenario"
id: start_scenario_btn
# This test does not have anything to start (mock is autostart)
+108
View File
@@ -0,0 +1,108 @@
"""Integration tests for Sensor::get_raw_state().
Raw state storage only exists when filters are compiled in (USE_SENSOR_FILTER).
Without it, get_raw_state() returns state, so both build configurations are covered:
one fixture with a filtered sensor and one with no filters at all.
"""
from __future__ import annotations
import asyncio
import re
from aioesphomeapi import APIClient, EntityInfo
import pytest
from .types import APIClientConnectedFactory, RunCompiledFunction
NO_FILTER_PATTERN = re.compile(r"NO_FILTER: state=([\d.]+) raw_state=([\d.]+)")
WITH_FILTER_PATTERN = re.compile(r"WITH_FILTER: state=([\d.]+) raw_state=([\d.]+)")
async def _press_and_read(
client: APIClient,
entities: list[EntityInfo],
button_object_id: str,
future: asyncio.Future[tuple[float, float]],
label: str,
) -> tuple[float, float]:
button = next(
(e for e in entities if button_object_id in e.object_id.lower()), None
)
assert button is not None, f"{button_object_id} not found"
client.button_command(button.key)
try:
return await asyncio.wait_for(future, timeout=5.0)
except TimeoutError:
pytest.fail(f"Timeout waiting for {label} log message")
@pytest.mark.asyncio
async def test_sensor_raw_state(
yaml_config: str,
run_compiled: RunCompiledFunction,
api_client_connected: APIClientConnectedFactory,
) -> None:
"""With filters compiled in, raw state is stored separately from state."""
loop = asyncio.get_running_loop()
no_filter_future: asyncio.Future[tuple[float, float]] = loop.create_future()
with_filter_future: asyncio.Future[tuple[float, float]] = loop.create_future()
def check_output(line: str) -> None:
if not no_filter_future.done() and (match := NO_FILTER_PATTERN.search(line)):
no_filter_future.set_result((float(match.group(1)), float(match.group(2))))
if not with_filter_future.done() and (
match := WITH_FILTER_PATTERN.search(line)
):
with_filter_future.set_result(
(float(match.group(1)), float(match.group(2)))
)
async with (
run_compiled(yaml_config, line_callback=check_output),
api_client_connected() as client,
):
entities, _ = await client.list_entities_services()
state, raw_state = await _press_and_read(
client, entities, "test_no_filter_button", no_filter_future, "NO_FILTER"
)
assert state == 21.5
assert raw_state == 21.5
state, raw_state = await _press_and_read(
client,
entities,
"test_with_filter_button",
with_filter_future,
"WITH_FILTER",
)
assert state == 43.0
assert raw_state == 21.5
@pytest.mark.asyncio
async def test_sensor_raw_state_no_filter(
yaml_config: str,
run_compiled: RunCompiledFunction,
api_client_connected: APIClientConnectedFactory,
) -> None:
"""Without filters compiled in, get_raw_state() returns state."""
loop = asyncio.get_running_loop()
no_filter_future: asyncio.Future[tuple[float, float]] = loop.create_future()
def check_output(line: str) -> None:
if not no_filter_future.done() and (match := NO_FILTER_PATTERN.search(line)):
no_filter_future.set_result((float(match.group(1)), float(match.group(2))))
async with (
run_compiled(yaml_config, line_callback=check_output),
api_client_connected() as client,
):
entities, _ = await client.list_entities_services()
state, raw_state = await _press_and_read(
client, entities, "test_no_filter_button", no_filter_future, "NO_FILTER"
)
assert state == 21.5
assert raw_state == 21.5
+66 -74
View File
@@ -19,23 +19,40 @@ from __future__ import annotations
import asyncio
from collections.abc import Callable
from dataclasses import dataclass
from aioesphomeapi import ButtonInfo, NumberInfo, SwitchInfo, TextSensorState
import pytest
from .state_utils import SensorTracker, find_entity, wait_for_state
from .state_utils import SensorTracker, find_entity, require_entity, wait_for_state
from .types import APIClientConnectedFactory, RunCompiledFunction
@dataclass
class RegisterTestCase:
"""Test parameters for a single modbus register write/read round-trip."""
def _swap16(value: int) -> int:
"""Byte-swapped view of a 16-bit register as the raw U_WORD wire value."""
return ((value & 0xFF) << 8) | (value >> 8)
initial_value: object
write_number_name: str
write_value: float
post_write_value: object
# Raw U_WORD view of reg_u_word_s's initial 0x1234
MESH_RAW_U_WORD_S = _swap16(4660)
# Initial values of the mesh fixture's address 1 registers; the
# server_controller test reads them and the write test uses them as baseline.
MESH_INITIAL_VALUES: dict[str, object] = {
"reg_u_word": 99,
"reg_u_word_s": 4660,
"reg_s_word": -99,
"reg_s_word_s": -2,
"reg_u_dword": 16909060,
"reg_s_dword": -16909060,
"reg_u_dword_r": pytest.approx(67305985),
"reg_s_dword_r": pytest.approx(-67305985),
"reg_u_qword": pytest.approx(72623859790382856),
"reg_s_qword": pytest.approx(-72623859790382856),
"reg_u_qword_r": pytest.approx(578437695752307201),
"reg_s_qword_r": pytest.approx(-578437695752307201),
"reg_fp32": pytest.approx(3.14),
"reg_fp32_r": pytest.approx(2.5),
}
# ---------------------------------------------------------------------------
@@ -310,23 +327,7 @@ async def test_uart_mock_modbus_server_controller(
line_callback, error_log_lines, warning_log_lines = _make_modbus_line_callback()
expected_values = {
"reg_u_word": 99,
"reg_u_word_s": 4660,
"reg_u_word_s_raw": 13330,
"reg_s_word": -99,
"reg_s_word_s": -2,
"reg_u_dword": 16909060,
"reg_s_dword": -16909060,
"reg_u_dword_r": pytest.approx(67305985),
"reg_s_dword_r": pytest.approx(-67305985),
"reg_u_qword": pytest.approx(72623859790382856),
"reg_s_qword": pytest.approx(-72623859790382856),
"reg_u_qword_r": pytest.approx(578437695752307201),
"reg_s_qword_r": pytest.approx(-578437695752307201),
"reg_fp32": pytest.approx(3.14),
"reg_fp32_r": pytest.approx(3.14),
}
expected_values = MESH_INITIAL_VALUES | {"reg_u_word_s_raw": MESH_RAW_U_WORD_S}
tracker = SensorTracker(list(expected_values.keys()))
futures = tracker.expect_all(expected_values)
@@ -334,14 +335,12 @@ async def test_uart_mock_modbus_server_controller(
run_compiled(yaml_config, line_callback=line_callback),
api_client_connected() as client,
):
# The controller polls from boot, so the first values can already be in
# the states the device sends on connect; matching them there saves
# waiting for the next poll
await tracker.setup_and_start_scenario(client, match_initial_states=True)
await tracker.await_all(futures)
_assert_no_modbus_errors(error_log_lines, warning_log_lines)
@pytest.mark.shared_yaml("uart_mock_modbus_mesh")
@pytest.mark.asyncio
async def test_uart_mock_modbus_server_controller_write(
yaml_config: str,
@@ -357,51 +356,47 @@ async def test_uart_mock_modbus_server_controller_write(
line_callback, error_log_lines, warning_log_lines = _make_modbus_line_callback()
register_test_cases: dict[str, RegisterTestCase] = {
"reg_u_word": RegisterTestCase(11, "write_u_word", 42, 42),
"reg_u_word_s": RegisterTestCase(4660, "write_u_word_s", 17185, 17185),
"reg_s_word": RegisterTestCase(-11, "write_s_word", -42, -42),
"reg_s_word_s": RegisterTestCase(-2, "write_s_word_s", -257, -257),
"reg_u_dword": RegisterTestCase(1001, "write_u_dword", 2002, 2002),
"reg_s_dword": RegisterTestCase(-1001, "write_s_dword", -2002, -2002),
"reg_u_dword_r": RegisterTestCase(3003, "write_u_dword_r", 4004, 4004),
"reg_s_dword_r": RegisterTestCase(-3003, "write_s_dword_r", -4004, -4004),
"reg_u_qword": RegisterTestCase(5005, "write_u_qword", 6006, 6006),
"reg_s_qword": RegisterTestCase(-5005, "write_s_qword", -6006, -6006),
"reg_u_qword_r": RegisterTestCase(7007, "write_u_qword_r", 8008, 8008),
"reg_s_qword_r": RegisterTestCase(-7007, "write_s_qword_r", -8008, -8008),
"reg_fp32": RegisterTestCase(
pytest.approx(1.5, abs=0.01),
"write_fp32",
3.14,
pytest.approx(3.14, abs=0.01),
),
"reg_fp32_r": RegisterTestCase(
pytest.approx(2.5, abs=0.01),
"write_fp32_r",
6.28,
pytest.approx(6.28, abs=0.01),
),
# Per read-back sensor: the number entity to write through and the value;
# floats read back within tolerance, everything else exactly
register_writes: dict[str, tuple[str, int | float]] = {
"reg_u_word": ("write_u_word", 42),
"reg_u_word_s": ("write_u_word_s", 17185),
"reg_s_word": ("write_s_word", -42),
"reg_s_word_s": ("write_s_word_s", -257),
"reg_u_dword": ("write_u_dword", 2002),
"reg_s_dword": ("write_s_dword", -2002),
"reg_u_dword_r": ("write_u_dword_r", 4004),
"reg_s_dword_r": ("write_s_dword_r", -4004),
"reg_u_qword": ("write_u_qword", 6006),
"reg_s_qword": ("write_s_qword", -6006),
"reg_u_qword_r": ("write_u_qword_r", 8008),
"reg_s_qword_r": ("write_s_qword_r", -8008),
"reg_fp32": ("write_fp32", 6.28),
"reg_fp32_r": ("write_fp32_r", 9.42),
}
tracker = SensorTracker(list(register_test_cases.keys()))
tracker = SensorTracker([*register_writes, "reg_u_word_s_raw"])
# The raw U_WORD view of 0x02 pins the byte swap on the write path: the
# round trip through write_u_word_s applies the swap an even number of
# times, so only the raw sensor can catch a symmetrically dropped swap.
# Phase 1: expect initial baseline values
initial_futures = tracker.expect_all(
{name: case.initial_value for name, case in register_test_cases.items()}
MESH_INITIAL_VALUES | {"reg_u_word_s_raw": MESH_RAW_U_WORD_S}
)
# Phase 2: expect post-write values (registered now so on_state can match them)
written_futures = tracker.expect_all(
{name: case.post_write_value for name, case in register_test_cases.items()}
{
name: pytest.approx(value, abs=0.01) if isinstance(value, float) else value
for name, (_, value) in register_writes.items()
}
| {"reg_u_word_s_raw": _swap16(register_writes["reg_u_word_s"][1])}
)
async with (
run_compiled(yaml_config, line_callback=line_callback),
api_client_connected() as client,
):
# The controller polls from boot, so the baseline can already be in the
# states the device sends on connect; matching it there saves waiting for
# the next poll
entities = await tracker.setup_and_start_scenario(
client, match_initial_states=True
)
@@ -410,19 +405,22 @@ async def test_uart_mock_modbus_server_controller_write(
# connection is working before issuing writes
await tracker.await_all(initial_futures, timeout=4.0)
# Issue write commands for all register types
for case in register_test_cases.values():
entity = find_entity(entities, case.write_number_name, NumberInfo)
assert entity is not None, (
f"{case.write_number_name} number entity not found"
)
client.number_command(entity.key, case.write_value)
# Issue write commands for all register types; exact object_id match,
# since several write_* names are prefixes of a sibling
numbers = {
e.object_id.lower(): e for e in entities if isinstance(e, NumberInfo)
}
for number_name, value in register_writes.values():
entity = numbers.get(number_name)
assert entity is not None, f"{number_name} number entity not found"
client.number_command(entity.key, value)
# Wait for sensors to reflect the written values (round-trip write+read)
await tracker.await_all(written_futures, timeout=4.0)
_assert_no_modbus_errors(error_log_lines, warning_log_lines)
@pytest.mark.shared_yaml("uart_mock_modbus_mesh")
@pytest.mark.asyncio
async def test_uart_mock_modbus_server_controller_bits(
yaml_config: str,
@@ -468,8 +466,6 @@ async def test_uart_mock_modbus_server_controller_bits(
run_compiled(yaml_config, line_callback=line_callback),
api_client_connected() as client,
):
# The controller polls from boot and binary sensors drop repeats, so the
# baseline can arrive only in the states the device sends on connect
entities = await tracker.setup_and_start_scenario(
client, match_initial_states=True
)
@@ -480,8 +476,7 @@ async def test_uart_mock_modbus_server_controller_bits(
# Flip both writable bits: 0x02 false -> true, 0x03 true -> false
for switch_name, value in (("write_bit_2", True), ("write_bit_3", False)):
entity = find_entity(entities, switch_name, SwitchInfo)
assert entity is not None, f"{switch_name} switch entity not found"
entity = require_entity(entities, switch_name, SwitchInfo)
client.switch_command(entity.key, value)
# Wait for both read views to reflect the written values
@@ -508,9 +503,6 @@ async def test_uart_mock_modbus_server_controller_multiple(
run_compiled(yaml_config, line_callback=line_callback),
api_client_connected() as client,
):
# The controller polls from boot, so the first values can already be in
# the states the device sends on connect; matching them there saves
# waiting for the next poll
await tracker.setup_and_start_scenario(client, match_initial_states=True)
await tracker.await_all(futures)
_assert_no_modbus_errors(error_log_lines, warning_log_lines)
+12
View File
@@ -163,6 +163,18 @@ def test_has_discovered_components_after_configure(tmp_path: Path) -> None:
assert has_discovered_components()
def test_get_project_cmakelists_size_command_uses_json2() -> None:
"""The POST_BUILD size command uses the cheap json2 format, with --ng
only on the 1.x tool bundled with IDF < 6."""
content = _render()
assert "-m esp_idf_size --ng --format=json2" in content
CORE.data[KEY_ESP32][KEY_IDF_VERSION] = cv.Version(6, 0, 0)
content = _render()
assert "--ng" not in content
assert "--format=json2" in content
def test_get_project_cmakelists_uses_supplied_builtin_components() -> None:
"""A cached list replaces project_description.json and is still filtered
by EXCLUDE_COMPONENTS."""
+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
+37
View File
@@ -638,6 +638,43 @@ def test_run_compile_passes_compile_process_limit(setup_core: Path) -> None:
mock_run.assert_called_once_with("build", "size", jobs=1)
def test_run_compile_passes_size_summary_paths(setup_core: Path) -> None:
"""print_summary receives the size json, partitions.csv, and the built
ELF from get_built_elf_path, which must stay in lockstep with the
project() name in the generated CMakeLists."""
_setup_build(setup_core)
config = {CONF_ESPHOME: {}}
with (
patch.object(toolchain, "need_reconfigure", return_value=False),
patch.object(toolchain, "run_idf_py", return_value=0),
patch.object(toolchain, "print_summary") as mock_summary,
):
assert toolchain.run_compile(config, verbose=False) == 0
mock_summary.assert_called_once_with(
CORE.relative_build_path("build", "esp_idf_size.json"),
CORE.relative_build_path("partitions.csv"),
CORE.relative_build_path("build", f"{CORE.name}.elf"),
)
def test_create_elf_copy(setup_core: Path) -> None:
"""The built <name>.elf is copied to the firmware.elf dashboard name."""
_setup_build(setup_core)
src = toolchain.get_built_elf_path()
src.parent.mkdir(parents=True, exist_ok=True)
src.write_bytes(b"elf")
assert toolchain.create_elf_copy() is True
assert toolchain.get_elf_path().read_bytes() == b"elf"
def test_create_elf_copy_missing_source(setup_core: Path) -> None:
"""A missing built ELF is a warning and False, not a crash."""
_setup_build(setup_core)
assert toolchain.create_elf_copy() is False
def test_run_compile_without_compile_process_limit(setup_core: Path) -> None:
"""When no compile_process_limit is set, no job limit is passed to idf.py."""
_setup_build(setup_core)
+236 -62
View File
@@ -4,6 +4,8 @@ from __future__ import annotations
import json
from pathlib import Path
import struct
from unittest.mock import patch
import pytest
@@ -17,64 +19,106 @@ def _write_size_json(tmp_path: Path, data: dict) -> Path:
return out
def _write_partitions(tmp_path: Path) -> Path:
"""Drop a partitions.csv with a 0x1C0000 (1835008 byte) app slot."""
out = tmp_path / "partitions.csv"
out.write_text(
"# name, type, subtype, offset, size, flags\n"
"app0, app, ota_0, 0x10000, 0x1C0000,\n"
)
return out
def _elf_bytes(sections: list[tuple[int, int, int]], shentsize: int = 40) -> bytes:
"""Build a minimal ELF32 LE whose section headers carry the given
(sh_type, sh_flags, sh_size) triples."""
out = bytearray(52)
out[0:4] = b"\x7fELF"
out[4] = out[5] = 1 # 32-bit, little-endian
struct.pack_into("<I", out, 0x20, 52) # e_shoff
struct.pack_into("<HH", out, 0x2E, shentsize, len(sections))
for sh_type, sh_flags, sh_size in sections:
shdr = bytearray(40)
struct.pack_into("<II", shdr, 4, sh_type, sh_flags)
struct.pack_into("<I", shdr, 20, sh_size)
out += shdr
return bytes(out)
def _esp32_size_data() -> dict:
"""Synthetic esp_idf_size.json for the original ESP32 (split IRAM/DRAM)."""
"""Synthetic json2 for the original ESP32 (split IRAM/DRAM), in the
esp-idf-size >= 2.1 shape that carries ``total_size``."""
return {
"image_size": 827455,
"memory_types": {
"DRAM": {
"size": 180736,
"version": "1.1",
"total_size": 827455,
"layout": [
{
"name": "DRAM",
"total": 180736,
"used": 47332,
"sections": {
".dram0.bss": {"abbrev_name": ".bss", "size": 30616},
".dram0.data": {"abbrev_name": ".data", "size": 16716},
"free": 133404,
"parts": {
".bss": {"size": 30616},
".data": {"size": 16716},
},
},
"IRAM": {
"size": 131072,
{
"name": "IRAM",
"total": 131072,
"used": 80351,
"sections": {
".iram0.text": {"abbrev_name": ".text", "size": 79323},
".iram0.vectors": {"abbrev_name": ".vectors", "size": 1028},
"free": 50721,
"parts": {
".text": {"size": 79323},
".vectors": {"size": 1028},
},
},
},
],
}
def _s3_size_data() -> dict:
"""Synthetic esp_idf_size.json for ESP32-S3 (unified DIRAM)."""
"""Synthetic json2 for ESP32-S3 (unified DIRAM), in the esp-idf-size 1.x
shape without ``total_size``."""
return {
"image_size": 724215,
"memory_types": {
"DIRAM": {
"size": 341760,
"version": "1.1",
"layout": [
{
"name": "DIRAM",
"total": 341760,
"used": 104999,
"sections": {
".iram0.text": {"abbrev_name": ".text", "size": 58051},
".dram0.bss": {"abbrev_name": ".bss", "size": 27088},
".dram0.data": {"abbrev_name": ".data", "size": 19708},
".noinit": {"abbrev_name": ".noinit", "size": 152},
"free": 236761,
"parts": {
".text": {"size": 58051},
".bss": {"size": 27088},
".data": {"size": 19708},
".noinit": {"size": 152},
},
},
"IRAM": {
"size": 16384,
{
"name": "IRAM",
"total": 16384,
"used": 16384,
"sections": {
".iram0.text": {"abbrev_name": ".text", "size": 15356},
".iram0.vectors": {"abbrev_name": ".vectors", "size": 1028},
"free": 0,
"parts": {
".text": {"size": 15356},
".vectors": {"size": 1028},
},
},
},
],
}
def _print_summary_ram_only(tmp_path: Path, size_json: Path) -> None:
"""Call print_summary with no partitions.csv or ELF on disk."""
print_summary(size_json, tmp_path / "partitions.csv", tmp_path / "firmware.elf")
def test_print_summary_esp32_uses_dram(
tmp_path: Path, capsys: pytest.CaptureFixture[str]
) -> None:
"""Original ESP32: DRAM has no ``.text``, so RAM = DRAM.used / DRAM.size unchanged."""
"""Original ESP32: RAM = DRAM.used / DRAM.total."""
size_json = _write_size_json(tmp_path, _esp32_size_data())
print_summary(size_json, partitions_csv=None)
_print_summary_ram_only(tmp_path, size_json)
out = capsys.readouterr().out
assert "RAM:" in out
assert "used 47332 bytes from 180736 bytes" in out
@@ -83,63 +127,193 @@ def test_print_summary_esp32_uses_dram(
def test_print_summary_s3_falls_back_to_diram(
tmp_path: Path, capsys: pytest.CaptureFixture[str]
) -> None:
"""ESP32-S3 with no DRAM key falls back to DIRAM and reports raw region usage."""
"""ESP32-S3 with no DRAM entry falls back to DIRAM and reports raw region usage."""
size_json = _write_size_json(tmp_path, _s3_size_data())
print_summary(size_json, partitions_csv=None)
_print_summary_ram_only(tmp_path, size_json)
out = capsys.readouterr().out
assert "used 104999 bytes from 341760 bytes" in out
def test_print_summary_skips_when_diram_total_collapses(
tmp_path: Path, capsys: pytest.CaptureFixture[str]
tmp_path: Path,
capsys: pytest.CaptureFixture[str],
caplog: pytest.LogCaptureFixture,
) -> None:
"""A zero-size region drops the RAM line rather than divide by zero."""
size_json = _write_size_json(
tmp_path,
{
"memory_types": {
"DIRAM": {
"size": 0,
"used": 0,
"sections": {},
},
},
"version": "1.1",
"layout": [{"name": "DIRAM", "total": 0, "used": 0}],
},
)
print_summary(size_json, partitions_csv=None)
_print_summary_ram_only(tmp_path, size_json)
out = capsys.readouterr().out
assert "RAM:" not in out
assert "unusable region" in caplog.text
def test_print_summary_handles_missing_json(
tmp_path: Path, capsys: pytest.CaptureFixture[str]
) -> None:
"""Missing size json is non-fatal and prints nothing."""
print_summary(tmp_path / "does_not_exist.json", partitions_csv=None)
_print_summary_ram_only(tmp_path, tmp_path / "does_not_exist.json")
assert capsys.readouterr().out == ""
def test_print_summary_handles_no_memory_types(
tmp_path: Path, capsys: pytest.CaptureFixture[str]
def test_print_summary_handles_no_layout(
tmp_path: Path,
capsys: pytest.CaptureFixture[str],
caplog: pytest.LogCaptureFixture,
) -> None:
"""A size json without ``memory_types`` still doesn't crash."""
size_json = _write_size_json(tmp_path, {"image_size": 0})
print_summary(size_json, partitions_csv=None)
"""A size json without ``layout`` warns so schema drift is visible."""
size_json = _write_size_json(tmp_path, {"version": "1.1"})
_print_summary_ram_only(tmp_path, size_json)
assert capsys.readouterr().out == ""
def test_print_summary_flash_line(
tmp_path: Path, capsys: pytest.CaptureFixture[str]
) -> None:
"""A partition table with an app row yields the Flash line in the exact
padded shape script/ci_memory_impact_extract.py greps."""
size_json = _write_size_json(tmp_path, _esp32_size_data())
partitions = tmp_path / "partitions.csv"
partitions.write_text(
"# name, type, subtype, offset, size, flags\n"
"app0, app, ota_0, 0x10000, 0x1C0000,\n"
assert any(
r.levelname == "WARNING" and "no DRAM/DIRAM region" in r.message
for r in caplog.records
)
print_summary(size_json, partitions)
def test_print_summary_flash_line_prefers_total_size(
tmp_path: Path, capsys: pytest.CaptureFixture[str]
) -> None:
"""With ``total_size`` in the json, that figure wins without reading the
ELF, in the exact shape script/ci_memory_impact_extract.py greps."""
size_json = _write_size_json(tmp_path, _esp32_size_data())
partitions = _write_partitions(tmp_path)
print_summary(size_json, partitions, tmp_path / "firmware.elf")
out = capsys.readouterr().out
assert "Flash: " in out
assert "(used 827455 bytes from 1835008 bytes)" in out
def test_print_summary_flash_line_derives_from_elf(
tmp_path: Path, capsys: pytest.CaptureFixture[str]
) -> None:
"""A 1.x json without ``total_size`` sums the ELF's loadable PROGBITS
sections; NOBITS and non-alloc sections are excluded."""
size_json = _write_size_json(tmp_path, _s3_size_data())
partitions = _write_partitions(tmp_path)
firmware_elf = tmp_path / "firmware.elf"
firmware_elf.write_bytes(
_elf_bytes(
[
(1, 0x6, 700000), # PROGBITS, alloc+exec: counted
(1, 0x2, 24215), # PROGBITS, alloc: counted
(8, 0x2, 50000), # NOBITS (.bss): excluded
(1, 0x0, 12345), # PROGBITS, no alloc (.debug_*): excluded
]
)
)
print_summary(size_json, partitions, firmware_elf)
out = capsys.readouterr().out
assert "(used 724215 bytes from 1835008 bytes)" in out
@pytest.mark.parametrize(
"data",
[
pytest.param([1, 2], id="top_level_list"),
pytest.param({"version": "1.1", "layout": None}, id="layout_null"),
pytest.param({"version": "1.1", "layout": 7}, id="layout_scalar"),
],
)
def test_print_summary_handles_unexpected_shapes(
data: object, tmp_path: Path, capsys: pytest.CaptureFixture[str]
) -> None:
"""A foreign-schema size json degrades to a warning, never a traceback."""
size_json = _write_size_json(tmp_path, data)
_print_summary_ram_only(tmp_path, size_json)
assert capsys.readouterr().out == ""
def test_print_summary_skips_flash_on_zero_app_partition(
tmp_path: Path, capsys: pytest.CaptureFixture[str]
) -> None:
"""A zero-size app partition skips the Flash line rather than printing
a from-0-bytes figure CI would record."""
size_json = _write_size_json(tmp_path, _esp32_size_data())
partitions = tmp_path / "partitions.csv"
partitions.write_text(
"# name, type, subtype, offset, size, flags\napp0, app, ota_0, 0x10000, 0x0,\n"
)
print_summary(size_json, partitions, tmp_path / "firmware.elf")
out = capsys.readouterr().out
assert "Flash:" not in out
def test_print_summary_skips_flash_on_unreadable_partitions(
tmp_path: Path, capsys: pytest.CaptureFixture[str]
) -> None:
"""An unreadable partitions.csv is non-fatal (chmod tricks don't work
for root in CI containers, so simulate the OSError instead)."""
size_json = _write_size_json(tmp_path, _esp32_size_data())
partitions = _write_partitions(tmp_path)
with patch(
"esphome.espidf.size_summary._find_app_partition_size",
side_effect=PermissionError("denied"),
):
print_summary(size_json, partitions, tmp_path / "firmware.elf")
assert "Flash:" not in capsys.readouterr().out
def test_print_summary_flash_falls_back_on_bad_total_size(
tmp_path: Path, capsys: pytest.CaptureFixture[str]
) -> None:
"""A zero or non-int total_size falls back to the ELF instead of
printing a used-0-bytes line CI would read as a real measurement."""
data = _s3_size_data()
data["total_size"] = 0
size_json = _write_size_json(tmp_path, data)
partitions = _write_partitions(tmp_path)
firmware_elf = tmp_path / "firmware.elf"
firmware_elf.write_bytes(_elf_bytes([(1, 0x2, 4096)]))
print_summary(size_json, partitions, firmware_elf)
out = capsys.readouterr().out
assert "(used 4096 bytes from 1835008 bytes)" in out
_GOOD_ELF = _elf_bytes([(1, 0x2, 1024)])
@pytest.mark.parametrize(
("elf_bytes", "with_partitions"),
[
pytest.param(None, True, id="missing_elf"),
pytest.param(b"junk", True, id="not_an_elf"),
pytest.param(
_elf_bytes([(1, 0x2, 1024)], shentsize=0), True, id="bad_shentsize"
),
pytest.param(_GOOD_ELF[:60], True, id="truncated_table"),
pytest.param(_elf_bytes([]), True, id="no_sections"),
pytest.param(_elf_bytes([(8, 0x2, 50000)]), True, id="no_progbits"),
pytest.param(_GOOD_ELF, False, id="missing_partitions"),
],
)
def test_print_summary_skips_flash_on_bad_input(
elf_bytes: bytes | None,
with_partitions: bool,
tmp_path: Path,
capsys: pytest.CaptureFixture[str],
caplog: pytest.LogCaptureFixture,
) -> None:
"""An unusable ELF or missing partitions.csv skips the Flash line, not the RAM line."""
size_json = _write_size_json(tmp_path, _s3_size_data())
firmware_elf = tmp_path / "firmware.elf"
if elf_bytes is not None:
firmware_elf.write_bytes(elf_bytes)
if with_partitions:
_write_partitions(tmp_path)
print_summary(size_json, tmp_path / "partitions.csv", firmware_elf)
out = capsys.readouterr().out
assert "RAM:" in out
assert "Flash:" not in out
# ELF problems warn (anomaly after a successful build); a missing
# partitions.csv stays at debug
warned = any(
r.levelname == "WARNING" and "Skipping Flash summary" in r.message
for r in caplog.records
)
assert warned == with_partitions