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

# Conflicts:
#	esphome/components/web_server/__init__.py
This commit is contained in:
J. Nick Koston
2026-09-14 10:47:19 -05:00
286 changed files with 6919 additions and 1224 deletions
@@ -40,6 +40,9 @@ class SerialProxy {
return SerialProxyResult::SERIAL_PROXY_RESULT_OK;
}
void write_from_client(api::APIConnection *api_connection, const uint8_t *data, size_t len) {}
SerialProxyResult set_mode_from_client(api::APIConnection *api_connection, api::enums::SerialProxyMode mode) {
return SerialProxyResult::SERIAL_PROXY_RESULT_OK;
}
SerialProxyResult set_modem_pins(api::APIConnection *api_connection, uint32_t line_states) {
return SerialProxyResult::SERIAL_PROXY_RESULT_OK;
}
@@ -0,0 +1,13 @@
"""The codegen hook external filtering components use to turn on the filter slot."""
from esphome.components import bluetooth_proxy
from esphome.core import CORE
def test_enable_advertisement_filter_emits_define() -> None:
"""External components call this rather than emitting the define."""
bluetooth_proxy.enable_advertisement_filter()
assert "USE_BLUETOOTH_PROXY_ADVERTISEMENT_FILTER" in {
define.name for define in CORE.defines
}
@@ -0,0 +1,154 @@
"""Tests for the bridge cdc_acm_uart platform's final validation."""
import pytest
from esphome import config_validation as cv
from esphome.components.cdc_acm_uart import bridge
from esphome.components.cdc_acm_uart.bridge import CONF_USB_CDC_ACM_ID
from esphome.config import Config
from esphome.const import CONF_DEBUG, CONF_ID, CONF_UART_ID, PlatformFramework
from esphome.core import ID
from esphome.types import ConfigType
from tests.component_tests.types import SetCoreConfigCallable
_final_validate = bridge._final_validate
def _set_esp32_s3(set_core_config: SetCoreConfigCallable, **kwargs) -> None:
from esphome.components.esp32 import KEY_VARIANT, VARIANT_ESP32S3
set_core_config(
PlatformFramework.ESP32_IDF,
platform_data={KEY_VARIANT: VARIANT_ESP32S3},
**kwargs,
)
def _full_config(uarts: list[ConfigType] | None = None, **domains) -> Config:
"""A full config declaring uart_0 and uart_1 (plus any extra entries), as the ID
pass leaves it, so the debug check can resolve a uart_id to its declaration."""
uarts = uarts or [{CONF_ID: ID("uart_0")}, {CONF_ID: ID("uart_1")}]
full = Config()
full["uart"] = uarts
for index, uart_conf in enumerate(uarts):
full.declare_ids.append((uart_conf[CONF_ID], ["uart", index, CONF_ID]))
full.update(domains)
return full
def _bridge_config(uart_id: str, cdc_id: str) -> dict:
return {CONF_UART_ID: ID(uart_id), CONF_USB_CDC_ACM_ID: ID(cdc_id)}
def test_accepts_distinct_uart_and_cdc_interfaces(
set_core_config: SetCoreConfigCallable,
) -> None:
_set_esp32_s3(set_core_config, full_config=_full_config())
_final_validate(_bridge_config("uart_0", "cdc_acm_1"))
_final_validate(_bridge_config("uart_1", "cdc_acm_2"))
def test_rejects_two_bridges_sharing_a_uart(
set_core_config: SetCoreConfigCallable,
) -> None:
_set_esp32_s3(set_core_config, full_config=_full_config())
_final_validate(_bridge_config("uart_0", "cdc_acm_1"))
with pytest.raises(cv.Invalid, match="already bridged"):
_final_validate(_bridge_config("uart_0", "cdc_acm_2"))
def test_rejects_two_bridges_sharing_a_cdc_interface(
set_core_config: SetCoreConfigCallable,
) -> None:
_set_esp32_s3(set_core_config, full_config=_full_config())
_final_validate(_bridge_config("uart_0", "cdc_acm_1"))
with pytest.raises(cv.Invalid, match="already bridged"):
_final_validate(_bridge_config("uart_1", "cdc_acm_1"))
def test_rejects_uart_shared_with_another_component(
set_core_config: SetCoreConfigCallable,
) -> None:
_set_esp32_s3(
set_core_config,
full_config=_full_config(
sensor=[{"platform": "pzemac", CONF_UART_ID: ID("uart_0")}],
),
)
with pytest.raises(cv.Invalid, match="exclusive"):
_final_validate(_bridge_config("uart_0", "cdc_acm_1"))
def test_rejects_cdc_interface_shared_with_another_component(
set_core_config: SetCoreConfigCallable,
) -> None:
# The CDC instance is itself a uart::UARTComponent, so other components can bind
# it as a plain UART via uart_id -- that must be rejected just like UART sharing.
_set_esp32_s3(
set_core_config,
full_config=_full_config(
sensor=[{"platform": "pzemac", CONF_UART_ID: ID("cdc_acm_1")}],
),
)
with pytest.raises(cv.Invalid, match="exclusive"):
_final_validate(_bridge_config("uart_0", "cdc_acm_1"))
def test_rejects_uart_referenced_from_nested_config(
set_core_config: SetCoreConfigCallable,
) -> None:
# References can sit arbitrarily deep, e.g. inside an automation's action list.
_set_esp32_s3(
set_core_config,
full_config=_full_config(
binary_sensor=[
{
"platform": "gpio",
"on_press": [{"then": [{CONF_UART_ID: ID("uart_0")}]}],
}
],
),
)
with pytest.raises(cv.Invalid, match="exclusive"):
_final_validate(_bridge_config("uart_0", "cdc_acm_1"))
def test_ignores_other_components_on_other_uarts(
set_core_config: SetCoreConfigCallable,
) -> None:
_set_esp32_s3(
set_core_config,
full_config=_full_config(
sensor=[{"platform": "pzemac", CONF_UART_ID: ID("uart_1")}],
# The bridge domain itself is skipped: this bridge's own entry (and any
# bridge-vs-bridge sharing, which the seen-set already rejects) must not
# trip the exclusivity scan.
bridge=[_bridge_config("uart_0", "cdc_acm_1")],
),
)
_final_validate(_bridge_config("uart_0", "cdc_acm_1"))
def test_rejects_debug_on_bridged_uart(
set_core_config: SetCoreConfigCallable,
) -> None:
# The bridge talks to the IDF driver directly, so the uart debugger would see
# nothing and its dummy_receiver would steal RX bytes.
_set_esp32_s3(
set_core_config,
full_config=_full_config(uarts=[{CONF_ID: ID("uart_0"), CONF_DEBUG: {}}]),
)
with pytest.raises(cv.Invalid, match="debug"):
_final_validate(_bridge_config("uart_0", "cdc_acm_1"))
def test_allows_debug_on_other_uart(
set_core_config: SetCoreConfigCallable,
) -> None:
_set_esp32_s3(
set_core_config,
full_config=_full_config(
uarts=[{CONF_ID: ID("uart_0")}, {CONF_ID: ID("uart_1"), CONF_DEBUG: {}}]
),
)
_final_validate(_bridge_config("uart_0", "cdc_acm_1"))
@@ -0,0 +1,53 @@
"""Power-user fields are marked as advanced on the shared schemas.
``filters``, ``manual_ip`` and the GPIO switch interlock options are knobs
whose defaults suit nearly every user, so a schema-aware editor should keep
them behind its "advanced settings" disclosure rather than on the main form.
"""
from __future__ import annotations
import importlib
import pytest
from esphome.components import binary_sensor, ethernet, sensor, text_sensor, wifi
import esphome.config_validation as cv
def _markers(schema: cv.Schema) -> dict[str, object]:
s = schema
if hasattr(s, "validators"):
# cv.All -> the schema is the first validator.
s = s.validators[0]
return {str(k): k for k in s.schema}
def _gpio_switch_schema() -> cv.Schema:
return importlib.import_module("esphome.components.gpio.switch").CONFIG_SCHEMA
@pytest.mark.parametrize(
("label", "schema_factory", "fields"),
[
("sensor", sensor.sensor_schema, ["filters"]),
("binary_sensor", binary_sensor.binary_sensor_schema, ["filters"]),
("text_sensor", text_sensor.text_sensor_schema, ["filters"]),
("wifi_network", lambda: wifi.WIFI_NETWORK_BASE, ["manual_ip"]),
("wifi", lambda: wifi.CONFIG_SCHEMA, ["manual_ip"]),
("ethernet", lambda: ethernet.BASE_SCHEMA, ["manual_ip"]),
("gpio_switch", _gpio_switch_schema, ["interlock", "interlock_wait_time"]),
],
)
def test_power_user_fields_are_advanced(
label: str, schema_factory, fields: list[str]
) -> None:
markers = _markers(schema_factory())
for field in fields:
assert markers[field].visibility is cv.Visibility.ADVANCED, f"{label}.{field}"
def test_interlock_wait_time_keeps_its_default() -> None:
"""Marking the field advanced must not drop its default."""
markers = _markers(_gpio_switch_schema())
assert markers["interlock_wait_time"].default() == "0ms"
+8 -3
View File
@@ -60,7 +60,7 @@ def reset_core() -> Generator[None]:
@pytest.fixture(autouse=True)
def reset_full_config() -> Generator[None]:
"""Give each test a clean final-validate config and restore it after."""
token = final_validate.full_config.set({})
token = final_validate.full_config.set(Config())
yield
final_validate.full_config.reset(token)
@@ -75,7 +75,7 @@ def set_core_config() -> Generator[SetCoreConfigCallable]:
*,
core_data: ConfigType | None = None,
platform_data: ConfigType | None = None,
full_config: dict[str, ConfigType] | None = None,
full_config: dict[str, ConfigType] | Config | None = None,
) -> None:
platform, framework = platform_framework.value
@@ -94,7 +94,12 @@ def set_core_config() -> Generator[SetCoreConfigCallable]:
CORE.data[platform.value] = platform_data
config.path_context.set([])
final_validate.full_config.set(full_config or Config())
# Production always installs a Config (a FinalValidateConfig), never a plain dict.
if not isinstance(full_config, Config):
full = Config()
full.update(full_config or {})
full_config = full
final_validate.full_config.set(full_config)
yield setter
@@ -0,0 +1,14 @@
esphome:
name: test
esp32:
board: esp32dev
framework:
type: esp-idf
wifi:
ssid: "test_ssid"
password: "test_password"
http_request:
verify_ssl: true
@@ -0,0 +1,19 @@
esphome:
name: test
esp32:
variant: esp32c6
framework:
type: esp-idf
network:
enable_ipv6: true
openthread:
channel: 13
network_name: OpenThread-8f28
network_key: 0xdfd34f0f05cad978ec4e32b0413038ff
pan_id: 0x8f28
ext_pan_id: 0xd63e8e3e495ebbc3
pskc: 0xc23a76e98f1a6483639b1ac1271e2e27
mesh_local_prefix: fd53:145f:ed22:ad81::/64
@@ -0,0 +1,17 @@
esphome:
name: test
esp32:
board: esp32dev
framework:
type: esp-idf
advanced:
disable_mbedtls_tls_server: false
disable_mbedtls_tls_extras: false
wifi:
ssid: "test_ssid"
password: "test_password"
http_request:
verify_ssl: true
@@ -0,0 +1,17 @@
esphome:
name: test
esp32:
board: esp32dev
framework:
type: esp-idf
sdkconfig_options:
CONFIG_MBEDTLS_TLS_SERVER_AND_CLIENT: y
CONFIG_MBEDTLS_CCM_C: y
wifi:
ssid: "test_ssid"
password: "test_password"
http_request:
verify_ssl: true
@@ -0,0 +1,17 @@
esphome:
name: test
esp32:
board: esp32dev
framework:
type: esp-idf
wifi:
ssid: "test_ssid"
eap:
identity: "user@example.org"
username: "user"
password: "secret"
http_request:
verify_ssl: true
@@ -0,0 +1,7 @@
esphome:
name: test
esp32:
variant: esp32c6
framework:
type: esp-idf
@@ -0,0 +1,9 @@
esphome:
name: test
esp32:
variant: esp32c6
framework:
type: esp-idf
advanced:
enable_full_printf: true
+99
View File
@@ -11,9 +11,12 @@ import pytest
from esphome.components.esp32 import (
KEY_FATFS_REQUIRED,
KEY_MBEDTLS_TLS_EXTRAS_REQUIRED,
KEY_MBEDTLS_TLS_SERVER_REQUIRED,
KEY_VFS_DIR_REQUIRED,
KEY_VFS_SELECT_REQUIRED,
KEY_VFS_TERMIOS_REQUIRED,
MBEDTLS_TLS_EXTRA_OPTIONS,
VARIANT_ESP32,
VARIANTS,
NetworkSdkconfigData,
@@ -1339,3 +1342,99 @@ def test_esp32_s31_gpio_validation(
with caplog.at_level("WARNING"):
validate_supports(pin)
assert "GPIO36 is a strapping PIN" in caplog.text
_TLS_SERVER_OPTIONS = (
"CONFIG_MBEDTLS_TLS_CLIENT_ONLY",
"CONFIG_MBEDTLS_TLS_SERVER_AND_CLIENT",
)
@pytest.mark.parametrize(
("config_file", "server", "extras"),
[
pytest.param("mbedtls_tls_default.yaml", (True, False), False, id="default"),
pytest.param("mbedtls_tls_opt_out.yaml", (None, None), None, id="opt_out"),
pytest.param("mbedtls_tls_wifi_eap.yaml", (True, False), None, id="wifi_eap"),
],
)
def test_mbedtls_tls_trim_sdkconfig(
generate_main: Callable[[str | Path], str],
component_config_path: Callable[[str], Path],
config_file: str,
server: tuple[bool | None, bool | None],
extras: bool | None,
) -> None:
"""Client-only TLS and the unused-feature trims apply unless opted out or required."""
generate_main(component_config_path(config_file))
sdkconfig = CORE.data[KEY_ESP32][KEY_SDKCONFIG_OPTIONS]
assert tuple(sdkconfig.get(name) for name in _TLS_SERVER_OPTIONS) == server
assert {sdkconfig.get(name) for name in MBEDTLS_TLS_EXTRA_OPTIONS} == {extras}
_OPENTHREAD_EXTRAS = {"CONFIG_MBEDTLS_CCM_C", "CONFIG_MBEDTLS_ECDSA_DETERMINISTIC"}
def test_mbedtls_tls_openthread_keeps_only_what_it_uses(
generate_main: Callable[[str | Path], str],
component_config_path: Callable[[str], Path],
) -> None:
"""The OpenThread config keeps the DTLS server, CCM and deterministic ECDSA; the rest is trimmed."""
generate_main(component_config_path("mbedtls_tls_openthread.yaml"))
sdkconfig = CORE.data[KEY_ESP32][KEY_SDKCONFIG_OPTIONS]
assert tuple(sdkconfig.get(name) for name in _TLS_SERVER_OPTIONS) == (None, None)
for name in MBEDTLS_TLS_EXTRA_OPTIONS:
assert sdkconfig.get(name) is (None if name in _OPENTHREAD_EXTRAS else False)
def test_mbedtls_tls_user_sdkconfig_wins(
generate_main: Callable[[str | Path], str],
component_config_path: Callable[[str], Path],
) -> None:
"""A user-set TLS role member leaves the whole choice alone; other user values are kept."""
generate_main(component_config_path("mbedtls_tls_user_sdkconfig.yaml"))
sdkconfig = CORE.data[KEY_ESP32][KEY_SDKCONFIG_OPTIONS]
assert sdkconfig.get("CONFIG_MBEDTLS_TLS_CLIENT_ONLY") is None
role = sdkconfig["CONFIG_MBEDTLS_TLS_SERVER_AND_CLIENT"]
assert isinstance(role, RawSdkconfigValue) and role.value == "y"
ccm = sdkconfig["CONFIG_MBEDTLS_CCM_C"]
assert isinstance(ccm, RawSdkconfigValue) and ccm.value == "y"
assert {
sdkconfig.get(name)
for name in MBEDTLS_TLS_EXTRA_OPTIONS
if name != "CONFIG_MBEDTLS_CCM_C"
} == {False}
def test_mbedtls_tls_openthread_requires_server_and_extras(
generate_main: Callable[[str | Path], str],
component_config_path: Callable[[str], Path],
) -> None:
"""The OpenThread hooks mark the DTLS server and CCM/deterministic ECDSA as required."""
generate_main(component_config_path("mbedtls_tls_openthread.yaml"))
assert CORE.data[KEY_ESP32][KEY_MBEDTLS_TLS_SERVER_REQUIRED] is True
assert CORE.data[KEY_ESP32][KEY_MBEDTLS_TLS_EXTRAS_REQUIRED] == _OPENTHREAD_EXTRAS
_VASPRINTF_STUB_FLAGS = {"-Wl,--wrap=vasprintf", "-Wl,--undefined=__wrap_vasprintf"}
@pytest.mark.parametrize(
("config_file", "expected"),
[
pytest.param("vasprintf_stub_c6.yaml", True, id="c6"),
pytest.param("vasprintf_stub_c6_full_printf.yaml", False, id="c6_full_printf"),
pytest.param("exclusion_reincludes.yaml", False, id="esp32"),
],
)
def test_vasprintf_stub_only_on_rom_vsnprintf_variants(
generate_main: Callable[[str | Path], str],
component_config_path: Callable[[str], Path],
config_file: str,
expected: bool,
) -> None:
"""The vasprintf wrap is emitted only where the ROM lacks vasprintf but has vsnprintf."""
generate_main(component_config_path(config_file))
assert (CORE.build_flags >= _VASPRINTF_STUB_FLAGS) is expected
defines = {define.name for define in CORE.defines}
assert ("USE_ESP32_VASPRINTF_STUB" in defines) is expected
@@ -0,0 +1,19 @@
---
esphome:
name: test
esp8266:
board: d1_mini
output:
- platform: esp8266_pwm
id: default_frequency
pin: GPIO4
frequency: 1kHz
- platform: esp8266_pwm
id: custom_frequency
pin: GPIO5
frequency: 2kHz
- platform: esp8266_pwm
id: schema_default_frequency
pin: GPIO12
@@ -0,0 +1,16 @@
"""Tests for the esp8266_pwm output codegen."""
from collections.abc import Callable
from pathlib import Path
def test_default_frequency_is_not_emitted(
generate_main: Callable[[str | Path], str],
component_config_path: Callable[[str], Path],
) -> None:
"""The 1 kHz default already lives in the C++ initializer."""
main_cpp = generate_main(component_config_path("frequency.yaml"))
assert "default_frequency->set_frequency(" not in main_cpp
assert "schema_default_frequency->set_frequency(" not in main_cpp
assert "custom_frequency->set_frequency(2000.0f);" in main_cpp
@@ -0,0 +1,29 @@
---
esphome:
name: test
esp32:
board: esp32dev
output:
- platform: ledc
id: out_a
pin: GPIO4
- platform: ledc
id: out_b
pin: GPIO5
light:
- platform: monochromatic
id: plain_light
output: out_a
flash_transition_length: 0s
- platform: monochromatic
id: fancy_light
output: out_b
flash_transition_length: 500ms
effects:
- pulse:
- platform: monochromatic
id: bare_light
output: out_a
@@ -0,0 +1,19 @@
"""Tests that light codegen skips setters for default values."""
from collections.abc import Callable
from pathlib import Path
def test_default_flash_length_and_empty_effects_are_not_emitted(
generate_main: Callable[[str | Path], str],
component_config_path: Callable[[str], Path],
) -> None:
"""A 0 ms flash transition and an empty effect list match the C++ defaults."""
main_cpp = generate_main(component_config_path("transitions.yaml"))
assert "plain_light->set_flash_transition_length(" not in main_cpp
assert "plain_light->add_effects(" not in main_cpp
assert "bare_light->set_flash_transition_length(" not in main_cpp
assert "bare_light->add_effects(" not in main_cpp
assert "fancy_light->set_flash_transition_length(500);" in main_cpp
assert "fancy_light->add_effects({" in main_cpp
@@ -52,3 +52,35 @@ def test_logger_pre_setup_before_other_components(generate_main):
f"Component allocation '{alloc.group()}' at position {alloc.start()} "
f"appears before logger pre_setup() at position {logger_pre_setup.start()}"
)
def test_default_uart_selection_is_not_emitted(generate_main):
"""UART0 is the C++ initializer on ESP8266, so the setter is skipped."""
main_cpp = generate_main("tests/component_tests/logger/test_logger.yaml")
assert "set_uart_selection(" not in main_cpp
def test_custom_uart_selection_is_emitted(generate_main):
"""A non default UART still reaches the setter before pre_setup()."""
main_cpp = generate_main("tests/component_tests/logger/test_logger_uart1.yaml")
assert "set_uart_selection(logger::UART_SELECTION_UART1);" in main_cpp
def test_libretiny_default_uart_selection_is_not_emitted(generate_main):
"""DEFAULT is the C++ initializer on LibreTiny, so the setter is skipped."""
main_cpp = generate_main(
"tests/component_tests/logger/test_logger_libretiny_default.yaml"
)
assert "set_uart_selection(" not in main_cpp
def test_libretiny_uart0_is_emitted(generate_main):
"""UART0 is not the LibreTiny initializer, so it must still be set."""
main_cpp = generate_main(
"tests/component_tests/logger/test_logger_libretiny_uart0.yaml"
)
assert "set_uart_selection(logger::UART_SELECTION_UART0);" in main_cpp
@@ -0,0 +1,8 @@
---
esphome:
name: test
rtl87xx:
board: generic-rtl8710bn-2mb-788k
logger:
@@ -0,0 +1,9 @@
---
esphome:
name: test
rtl87xx:
board: generic-rtl8710bn-2mb-788k
logger:
hardware_uart: UART0
@@ -0,0 +1,9 @@
---
esphome:
name: test
esp8266:
board: d1_mini_lite
logger:
hardware_uart: UART1
@@ -0,0 +1,28 @@
---
esphome:
name: test
esp32:
board: esp32dev
number:
- platform: template
id: auto_number
min_value: 0
max_value: 10
step: 1
optimistic: true
- platform: template
id: box_number
min_value: 0
max_value: 10
step: 1
mode: box
optimistic: true
- platform: template
id: explicit_auto_number
min_value: 0
max_value: 10
step: 1
mode: auto
optimistic: true
@@ -0,0 +1,16 @@
"""Tests for the number component codegen."""
from collections.abc import Callable
from pathlib import Path
def test_default_mode_is_not_emitted(
generate_main: Callable[[str | Path], str],
component_config_path: Callable[[str], Path],
) -> None:
"""Mode auto is the C++ initializer, so only a non default mode is set."""
main_cpp = generate_main(component_config_path("mode.yaml"))
assert "auto_number->traits.set_mode(" not in main_cpp
assert "explicit_auto_number->traits.set_mode(" not in main_cpp
assert "box_number->traits.set_mode(number::NUMBER_MODE_BOX);" in main_cpp
@@ -0,0 +1,13 @@
---
esphome:
name: test
esp32:
board: esp32dev
output:
- platform: ac_dimmer
id: dimmer
gate_pin: GPIO4
zero_cross_pin: GPIO5
min_power: 0%
@@ -0,0 +1,18 @@
---
esphome:
name: test
esp32:
board: esp32dev
output:
- platform: ledc
id: default_power
pin: GPIO4
max_power: 100%
min_power: 0%
- platform: ledc
id: custom_power
pin: GPIO5
max_power: 90%
min_power: 1%
@@ -0,0 +1,31 @@
"""Tests for the output platform codegen."""
from collections.abc import Callable
from pathlib import Path
from esphome.core import CORE
def test_default_power_limits_are_not_emitted(
generate_main: Callable[[str | Path], str],
component_config_path: Callable[[str], Path],
) -> None:
"""max_power 100% and min_power 0% already live in the C++ initializers."""
main_cpp = generate_main(component_config_path("power_limits.yaml"))
assert "default_power->set_max_power(" not in main_cpp
assert "default_power->set_min_power(" not in main_cpp
assert "custom_power->set_max_power(0.9f);" in main_cpp
assert "custom_power->set_min_power(0.01f);" in main_cpp
assert "USE_OUTPUT_FLOAT_POWER_SCALING" in {d.name for d in CORE.defines}
def test_default_min_power_keeps_scaling_fields(
generate_main: Callable[[str | Path], str],
component_config_path: Callable[[str], Path],
) -> None:
"""ac_dimmer reads min_power_ directly, so the define must stay on for min_power 0%."""
main_cpp = generate_main(component_config_path("ac_dimmer_min_power_zero.yaml"))
assert "dimmer->set_min_power(" not in main_cpp
assert "USE_OUTPUT_FLOAT_POWER_SCALING" in {d.name for d in CORE.defines}
@@ -0,0 +1,9 @@
esphome:
name: test
esp32:
board: esp32dev
remote_receiver:
- id: rcvr
pin: GPIO4
@@ -0,0 +1,10 @@
esphome:
name: test
esp32:
board: esp32dev
remote_receiver:
- id: rcvr
pin: GPIO4
buffer_size: 2kb
@@ -0,0 +1,12 @@
esphome:
name: test
esp32:
board: esp32-c2-devkitm-1
variant: esp32c2
framework:
type: esp-idf
remote_receiver:
- id: rcvr
pin: GPIO4
@@ -0,0 +1,9 @@
esphome:
name: test
esp8266:
board: d1_mini
remote_receiver:
- id: rcvr
pin: GPIO4
@@ -0,0 +1,24 @@
esphome:
name: test
esp32:
board: esp32dev
logger:
remote_receiver:
- id: rcvr
pin: GPIO4
dump:
- nec
- rc_switch
on_nec:
then:
- logger.log: nec
binary_sensor:
- platform: remote_receiver
name: Remote Input
nec:
address: 0x1234
command: 0x5678
@@ -0,0 +1,22 @@
esphome:
name: test
esp32:
board: esp32dev
remote_receiver:
- id: rcvr_ir
pin: GPIO4
- id: rcvr_rf
pin: GPIO5
infrared:
- platform: ir_rf_proxy
name: IR Receiver
remote_receiver_id: rcvr_ir
radio_frequency:
- platform: ir_rf_proxy
name: RF Receiver
frequency: 433.92MHz
remote_receiver_id: rcvr_rf
@@ -0,0 +1,28 @@
"""buffer_size reaches the receiver when set, and always on the pulse ring targets."""
from collections.abc import Callable
from pathlib import Path
def test_explicit_buffer_size_is_passed_through(
generate_main: Callable[[str | Path], str],
component_config_path: Callable[[str], Path],
) -> None:
main_cpp = generate_main(component_config_path("receiver_buffer_size.yaml"))
assert "rcvr->set_buffer_size(2000);" in main_cpp
def test_pulse_ring_target_keeps_a_default(
generate_main: Callable[[str | Path], str],
component_config_path: Callable[[str], Path],
) -> None:
main_cpp = generate_main(component_config_path("receiver_esp8266.yaml"))
assert "rcvr->set_buffer_size(1000);" in main_cpp
def test_esp32_variant_without_rmt_keeps_a_default(
generate_main: Callable[[str | Path], str],
component_config_path: Callable[[str], Path],
) -> None:
main_cpp = generate_main(component_config_path("receiver_esp32_c2.yaml"))
assert "rcvr->set_buffer_size(1000);" in main_cpp
@@ -0,0 +1,93 @@
"""Listener and dumper StaticVector sizes come from codegen slot counts."""
from collections.abc import Callable
from pathlib import Path
import pytest
from esphome.automation import ACTION_REGISTRY
from esphome.components import remote_base
import esphome.config_validation as cv
from ..helpers import get_define_value
def test_dumper_and_listener_counts(
generate_main: Callable[[str | Path], str],
component_config_path: Callable[[str], Path],
) -> None:
generate_main(component_config_path("receiver_with_dumpers.yaml"))
# nec and rc_switch dumpers
assert get_define_value("REMOTE_BASE_DUMPER_COUNT") == "2"
# on_nec trigger plus the remote_receiver binary sensor
assert get_define_value("REMOTE_BASE_LISTENER_COUNT") == "2"
def test_bare_receiver_emits_no_counts(
generate_main: Callable[[str | Path], str],
component_config_path: Callable[[str], Path],
) -> None:
main_cpp = generate_main(component_config_path("receiver_bare.yaml"))
# the RMT ring is sized in setup() unless buffer_size is set
assert "set_buffer_size" not in main_cpp
assert get_define_value("REMOTE_BASE_DUMPER_COUNT") is None
assert get_define_value("REMOTE_BASE_LISTENER_COUNT") is None
def test_proxy_receivers_count_as_listeners(
generate_main: Callable[[str | Path], str],
component_config_path: Callable[[str], Path],
) -> None:
generate_main(component_config_path("receiver_with_proxies.yaml"))
# one proxy entity listens on each of the two receivers; every receiver's list gets the
# capacity of the busiest one, so this is the largest per receiver count, not the sum
assert get_define_value("REMOTE_BASE_LISTENER_COUNT") == "1"
assert get_define_value("REMOTE_BASE_DUMPER_COUNT") is None
def test_only_used_protocol_sources_are_compiled(
generate_main: Callable[[str | Path], str],
component_config_path: Callable[[str], Path],
) -> None:
generate_main(component_config_path("receiver_with_dumpers.yaml"))
excluded = set(remote_base.FILTER_SOURCE_FILES())
assert "nec_protocol.cpp" not in excluded
assert "rc_switch_protocol.cpp" not in excluded
assert "sony_protocol.cpp" in excluded
assert "remote_base.cpp" not in excluded
def test_every_registry_name_maps_to_a_protocol_source() -> None:
"""A registry name must resolve to a source file or request_protocol rejects it."""
names = (
set(remote_base.BINARY_SENSOR_REGISTRY)
| set(remote_base.DUMPER_REGISTRY)
| {key.removeprefix("on_") for key in remote_base.TRIGGER_REGISTRY}
| {
key.removeprefix("remote_transmitter.transmit_")
for key in ACTION_REGISTRY
if key.startswith("remote_transmitter.transmit_")
}
)
assert len(names) > 40
for name in names:
assert remote_base._protocol_stem(name) in remote_base._PROTOCOL_STEMS, name
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'"):
remote_base.request_protocol("toshiba")
def test_dump_list_is_deduplicated_across_forms() -> None:
dumpers = remote_base.validate_dumpers(["raw", {"raw": None}, "nec", "nec"])
assert [
next(k for k in entry if k in remote_base.DUMPER_REGISTRY) for entry in dumpers
] == ["raw", "nec"]
@pytest.mark.parametrize("bad", [["nec", None], [5]])
def test_dump_list_rejects_invalid_entries_with_a_validation_error(bad: list) -> None:
with pytest.raises(cv.Invalid):
remote_base.validate_dumpers(bad)
@@ -0,0 +1,12 @@
esphome:
name: test
esp32:
board: esp32dev
framework:
type: esp-idf
wifi:
ap:
sendspin:
@@ -0,0 +1,18 @@
esphome:
name: test
project:
name: project_manufacturer.project_model
version: 9.9.9
esp32:
board: esp32dev
framework:
type: esp-idf
wifi:
ap:
sendspin:
manufacturer: Explicit Manufacturer
model: Explicit Model
firmware_version: 1.2.3
@@ -0,0 +1,15 @@
esphome:
name: test
project:
name: project_manufacturer.project_model
version: 9.9.9
esp32:
board: esp32dev
framework:
type: esp-idf
wifi:
ap:
sendspin:
@@ -0,0 +1,80 @@
"""Tests for the device information the sendspin hub reports to the server."""
from __future__ import annotations
from collections.abc import Callable
from pathlib import Path
import pytest
from esphome import config_validation as cv
from esphome.components.const import CONF_MANUFACTURER
from esphome.components.sendspin import CONF_FIRMWARE_VERSION, CONFIG_SCHEMA
from esphome.const import CONF_MODEL, PlatformFramework
from tests.component_tests.types import SetCoreConfigCallable
def test_explicit_device_info_wins_over_project(
generate_main: Callable[[str | Path], str],
component_config_path: Callable[[str], Path],
) -> None:
"""Configured values take precedence over the project information."""
main_cpp = generate_main(component_config_path("device_info_explicit.yaml"))
assert 'set_manufacturer("Explicit Manufacturer")' in main_cpp
assert 'set_model("Explicit Model")' in main_cpp
assert 'set_firmware_version("1.2.3")' in main_cpp
def test_project_supplies_device_info(
generate_main: Callable[[str | Path], str],
component_config_path: Callable[[str], Path],
) -> None:
"""Without configured values, the project name splits into manufacturer and model."""
main_cpp = generate_main(component_config_path("device_info_project.yaml"))
assert 'set_manufacturer("project_manufacturer")' in main_cpp
assert 'set_model("project_model")' in main_cpp
assert 'set_firmware_version("9.9.9")' in main_cpp
def test_no_device_info_leaves_hub_defaults(
generate_main: Callable[[str | Path], str],
component_config_path: Callable[[str], Path],
) -> None:
"""With neither source, nothing is emitted and the hub keeps its own defaults."""
main_cpp = generate_main(component_config_path("device_info_default.yaml"))
assert "set_manufacturer(" not in main_cpp
assert "set_model(" not in main_cpp
assert "set_firmware_version(" not in main_cpp
@pytest.mark.parametrize(
"conf_key", [CONF_MANUFACTURER, CONF_MODEL, CONF_FIRMWARE_VERSION]
)
def test_empty_device_info_rejected(
set_core_config: SetCoreConfigCallable, conf_key: str
) -> None:
"""An empty string would be sent to the server as an empty value, so it is not accepted."""
set_core_config(PlatformFramework.ESP32_IDF)
with pytest.raises(cv.Invalid):
CONFIG_SCHEMA({conf_key: ""})
@pytest.mark.parametrize(
"conf_key", [CONF_MANUFACTURER, CONF_MODEL, CONF_FIRMWARE_VERSION]
)
def test_device_info_capped_at_127_bytes(
set_core_config: SetCoreConfigCallable, conf_key: str
) -> None:
"""The cap is in bytes so the protobuf length prefix stays a single byte."""
set_core_config(PlatformFramework.ESP32_IDF)
CONFIG_SCHEMA({conf_key: "a" * 127})
with pytest.raises(cv.Invalid):
CONFIG_SCHEMA({conf_key: "a" * 128})
# 64 two-byte characters is 128 bytes.
with pytest.raises(cv.Invalid):
CONFIG_SCHEMA({conf_key: "é" * 64})
@@ -0,0 +1,90 @@
"""Validation tests for the sendspin media_source platform.
These cover the codec preference list, whose rejection branches a compile test
cannot reach: a `test*.yaml` can only assert that a configuration is accepted.
"""
from typing import Any
import pytest
from esphome import config_validation as cv
from esphome.components.sendspin import CONF_CODECS, _get_data
from esphome.components.sendspin.media_source import CONFIG_SCHEMA
from esphome.const import PlatformFramework
from esphome.types import ConfigType
from tests.component_tests.types import SetCoreConfigCallable
def _media_source_config(**overrides: Any) -> ConfigType:
"""Build a minimal valid media source config, allowing field overrides."""
config: ConfigType = {
"id": "sendspin_media_source",
"sendspin_id": "sendspin_hub",
}
config.update(overrides)
return config
def test_default_codecs_at_48_khz(set_core_config: SetCoreConfigCallable) -> None:
"""Every codec is advertised when the sample rate suits all of them."""
set_core_config(PlatformFramework.ESP32_IDF)
config = CONFIG_SCHEMA(_media_source_config())
assert config[CONF_CODECS] == ["flac", "opus", "pcm"]
def test_default_codecs_drop_opus_at_other_rates(
set_core_config: SetCoreConfigCallable,
) -> None:
"""Opus only supports 48 kHz, so it leaves the default list at other rates."""
set_core_config(PlatformFramework.ESP32_IDF)
config = CONFIG_SCHEMA(_media_source_config(sample_rate=44100))
assert config[CONF_CODECS] == ["flac", "pcm"]
def test_configured_order_is_preserved(set_core_config: SetCoreConfigCallable) -> None:
"""The list is a preference order, so it reaches the player role as written."""
set_core_config(PlatformFramework.ESP32_IDF)
CONFIG_SCHEMA(_media_source_config(codecs=["pcm", "flac"]))
assert _get_data().player_config[CONF_CODECS] == ["pcm", "flac"]
def test_empty_codec_list_rejected(set_core_config: SetCoreConfigCallable) -> None:
"""A player with no codecs at all could never be given a stream."""
set_core_config(PlatformFramework.ESP32_IDF)
with pytest.raises(cv.Invalid, match="length of value must be at least 1"):
CONFIG_SCHEMA(_media_source_config(codecs=[]))
def test_duplicate_codec_rejected(set_core_config: SetCoreConfigCallable) -> None:
"""A repeated codec has no meaning in a preference order."""
set_core_config(PlatformFramework.ESP32_IDF)
with pytest.raises(cv.Invalid, match="may only be listed once"):
CONFIG_SCHEMA(_media_source_config(codecs=["flac", "flac"]))
def test_unknown_codec_rejected(set_core_config: SetCoreConfigCallable) -> None:
"""Only codecs the player role can decode are accepted."""
set_core_config(PlatformFramework.ESP32_IDF)
with pytest.raises(cv.Invalid, match="Unknown value"):
CONFIG_SCHEMA(_media_source_config(codecs=["mp3"]))
def test_opus_at_wrong_sample_rate_rejected(
set_core_config: SetCoreConfigCallable,
) -> None:
"""Asking for Opus at a rate it cannot handle fails rather than silently
dropping the stated preference."""
set_core_config(PlatformFramework.ESP32_IDF)
with pytest.raises(cv.Invalid, match="requires a sample_rate of 48000"):
CONFIG_SCHEMA(_media_source_config(codecs=["opus"], sample_rate=44100))
@@ -0,0 +1,76 @@
"""The template platforms surface value-describing metadata on the main form.
Hardware platforms get sensible defaults for unit/device_class/etc., so those
fields fall through to the editor's advanced disclosure. A ``template`` entity
has no such defaults -- the user is expected to define them -- so the template
platforms pass ``visibility=cv.Visibility.UI`` to promote them onto the form.
"""
from __future__ import annotations
import importlib
import pytest
import esphome.config_validation as cv
def _markers(schema: cv.Schema) -> dict[str, object]:
s = schema
if hasattr(s, "validators"):
# cv.All -> the schema is the first validator.
s = s.validators[0]
return {str(k): k for k in s.schema}
@pytest.mark.parametrize(
("platform", "fields"),
[
(
"sensor",
[
"unit_of_measurement",
"accuracy_decimals",
"device_class",
"state_class",
"force_update",
],
),
("binary_sensor", ["device_class"]),
("switch", ["device_class"]),
("cover", ["device_class"]),
("button", ["device_class"]),
("valve", ["device_class"]),
("event", ["device_class"]),
("text_sensor", ["device_class"]),
("number", ["device_class", "unit_of_measurement"]),
],
)
def test_template_metadata_is_ui(platform: str, fields: list[str]) -> None:
mod = importlib.import_module(f"esphome.components.template.{platform}")
markers = _markers(mod.CONFIG_SCHEMA)
for field in fields:
assert markers[field].visibility is cv.Visibility.UI, f"{platform}.{field}"
def test_template_sensor_promotion_preserves_defaults() -> None:
"""Promoting to UI must not drop the fields' defaults."""
from esphome.components.template.sensor import CONFIG_SCHEMA
markers = _markers(CONFIG_SCHEMA)
assert markers["accuracy_decimals"].default() == 1
assert markers["force_update"].default() is False
def test_hardware_platform_metadata_not_promoted() -> None:
"""Without ``visibility=`` the builders leave metadata unset.
Unset markers fall through to the consumer's ``Optional`` default of
advanced, so hardware platforms are unaffected by the template promotion.
"""
from esphome.components import binary_sensor, sensor
hw_sensor = _markers(sensor.sensor_schema(device_class="temperature"))
assert hw_sensor["device_class"].visibility is None
hw_bs = _markers(binary_sensor.binary_sensor_schema(device_class="motion"))
assert hw_bs["device_class"].visibility is None
+2 -1
View File
@@ -4,6 +4,7 @@ from __future__ import annotations
from typing import Protocol
from esphome.config import Config
from esphome.const import PlatformFramework
from esphome.types import ConfigType
@@ -18,5 +19,5 @@ class SetCoreConfigCallable(Protocol):
*,
core_data: ConfigType | None = None,
platform_data: ConfigType | None = None,
full_config: dict[str, ConfigType] | None = None,
full_config: dict[str, ConfigType] | Config | None = None,
) -> None: ...
@@ -0,0 +1,12 @@
---
esphome:
name: test
esp32:
board: esp32dev
wifi:
ssid: test
password: testtest
web_server:
@@ -0,0 +1,15 @@
---
esphome:
name: test
esp32:
board: esp32dev
wifi:
ssid: test
password: testtest
web_server:
port: 8080
log: false
include_internal: true
@@ -0,0 +1,15 @@
---
esphome:
name: test
esp32:
board: esp32dev
wifi:
ssid: test
password: testtest
web_server:
port: 80
log: true
include_internal: false
@@ -0,0 +1,35 @@
"""Tests that web_server 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 80, log on and include_internal off already live in the 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 "set_port(" not in main_cpp
assert "set_expose_log(" not in main_cpp
assert "set_include_internal(" 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 "set_port(8080);" in main_cpp
assert "set_expose_log(false);" in main_cpp
assert "set_include_internal(true);" in main_cpp
@@ -0,0 +1,12 @@
---
esphome:
name: test
esp8266:
board: d1_mini
wifi:
ssid: test
password: testtest
ap:
ssid: fallback
@@ -0,0 +1,18 @@
---
esphome:
name: test
esp8266:
board: d1_mini
wifi:
networks:
- ssid: test
password: testtest
priority: 5
ap:
ssid: fallback
ap_timeout: 2min
reboot_timeout: 0s
power_save_mode: light
min_auth_mode: wpa
@@ -0,0 +1,18 @@
---
esphome:
name: test
esp8266:
board: d1_mini
wifi:
networks:
- ssid: test
password: testtest
priority: 0
ap:
ssid: fallback
ap_timeout: 90s
reboot_timeout: 15min
power_save_mode: none
min_auth_mode: wpa2
@@ -0,0 +1,39 @@
"""Tests that wifi codegen skips setters for 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:
"""Priority 0, 90 s AP timeout, 15 min reboot, power save none, WPA2 are C++ defaults.
Both the schema defaults and the same values written explicitly take the skip path.
"""
main_cpp = generate_main(component_config_path(config_file))
assert "set_priority(" not in main_cpp
assert "set_ap_timeout(" not in main_cpp
assert "set_reboot_timeout(" not in main_cpp
assert "set_power_save_mode(" not in main_cpp
assert "set_min_auth_mode(" 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 "set_priority(5);" in main_cpp
assert "set_ap_timeout(120000);" in main_cpp
assert "set_reboot_timeout(0);" in main_cpp
assert "set_power_save_mode(wifi::WIFI_POWER_SAVE_LIGHT);" in main_cpp
assert "set_min_auth_mode(wifi::WIFI_MIN_AUTH_MODE_WPA);" in main_cpp
+1
View File
@@ -4,3 +4,4 @@ output:
gate_pin: ${gate_pin}
zero_cross_pin: ${zero_cross_pin}
zero_cross_interrupt_type: ANY
min_power: 0%
+17
View File
@@ -0,0 +1,17 @@
import esphome.codegen as cg
from esphome.core import CORE
from tests.testing_helpers import ComponentManifestOverride
def override_manifest(manifest: ComponentManifestOverride) -> None:
# USE_API compiles every api source, so emit what they need. No socket
# override: an __init__.py there makes pytest import its conftest as socket.conftest.
async def to_code_testing(config):
cg.add_define("USE_API")
cg.add_define("USE_API_PLAINTEXT")
cg.add_define("API_MAX_SEND_QUEUE", 8)
cg.add_define("MAX_API_CONNECTIONS", 1)
cg.add_define("USE_SOCKET_IMPL_BSD_SOCKETS")
CORE.register_controller() # api_server registers with the controller registry
manifest.to_code = to_code_testing
+65
View File
@@ -0,0 +1,65 @@
#include <gtest/gtest.h>
#include <cstdint>
#include <cstring>
#include "esphome/components/api/api_buffer.h"
namespace esphome::api::testing {
// Pointer plus two 16 bit sizes
static_assert(sizeof(APIBuffer) <= 2 * sizeof(void *));
TEST(APIBuffer, RefusesSizesAbove16Bits) {
APIBuffer buf;
ASSERT_TRUE(buf.resize(16));
EXPECT_FALSE(buf.reserve(UINT16_MAX + 1));
EXPECT_EQ(buf.size(), 16u);
EXPECT_EQ(buf.capacity(), 16u);
EXPECT_TRUE(buf.reserve(UINT16_MAX));
EXPECT_EQ(buf.capacity(), UINT16_MAX);
}
static const uint8_t BYTES[] = {1, 2, 3, 4, 5, 6};
TEST(APIBuffer, AppendReturnsTheNewBytes) {
APIBuffer buf;
ASSERT_TRUE(buf.reserve(8));
uint8_t *first = buf.append(3);
ASSERT_NE(first, nullptr);
std::memcpy(first, BYTES, 3);
EXPECT_EQ(buf.size(), 3u);
EXPECT_EQ(buf.capacity(), 8u);
// Grows through realloc and keeps what was there
uint8_t *second = buf.append(6);
ASSERT_EQ(second, buf.data() + 3);
std::memcpy(second, BYTES + 3, 3);
EXPECT_EQ(buf.size(), 9u);
EXPECT_EQ(buf.capacity(), 9u);
EXPECT_EQ(std::memcmp(buf.data(), BYTES, 6), 0);
}
TEST(APIBuffer, DropFrontSlidesTheRestDown) {
APIBuffer buf;
uint8_t *bytes = buf.append(6);
ASSERT_NE(bytes, nullptr);
std::memcpy(bytes, BYTES, 6);
buf.drop_front(2);
EXPECT_EQ(buf.size(), 4u);
EXPECT_EQ(buf.capacity(), 6u);
EXPECT_EQ(std::memcmp(buf.data(), BYTES + 2, 4), 0);
// Growing afterwards keeps the slid bytes
ASSERT_TRUE(buf.reserve(64));
EXPECT_EQ(buf.size(), 4u);
EXPECT_EQ(std::memcmp(buf.data(), BYTES + 2, 4), 0);
// Dropping everything leaves an empty buffer with its capacity
buf.drop_front(4);
EXPECT_EQ(buf.size(), 0u);
EXPECT_EQ(buf.capacity(), 64u);
}
} // namespace esphome::api::testing
@@ -0,0 +1,510 @@
#include <gtest/gtest.h>
#include <fcntl.h>
#include <sys/socket.h>
#include <unistd.h>
#include <algorithm>
#include <cerrno>
#include <cstdint>
#include <cstring>
#include <initializer_list>
#include <memory>
#include <vector>
#include "esphome/components/api/api_overflow_buffer.h"
#ifdef USE_HOST
namespace esphome::api::testing {
// Idle cost is the buffer plus one word of bookkeeping
static_assert(sizeof(APIOverflowBuffer) <= sizeof(APIBuffer) + sizeof(void *));
// Exposes storage so tests can check it is reused, not reallocated
class TestOverflowBuffer : public APIOverflowBuffer {
public:
using APIOverflowBuffer::LEN_PREFIX;
using APIOverflowBuffer::MAX_BYTES;
using APIOverflowBuffer::MAX_LONE_BYTES;
struct Storage {
size_t capacity;
const uint8_t *data;
bool operator==(const Storage &) const = default;
};
size_t capacity() const { return this->buf_.capacity(); }
Storage storage() const { return {this->buf_.capacity(), this->buf_.data()}; }
uint8_t count() const { return this->count_; }
size_t live() const { return this->buf_.size() - this->head_; }
/// Simulates a socket write inside try_drain() re-entering the send path
void set_draining(bool draining) { this->draining_ = draining; }
};
static std::vector<uint8_t> make_message(size_t len, uint8_t seed) {
std::vector<uint8_t> msg(len);
for (size_t i = 0; i < len; i++)
msg[i] = static_cast<uint8_t>(seed + i);
return msg;
}
static bool enqueue(TestOverflowBuffer &buf, const std::vector<uint8_t> &msg, uint16_t skip = 0) {
struct iovec iov = {const_cast<uint8_t *>(msg.data()), msg.size()};
return buf.enqueue_iov(&iov, 1, static_cast<uint16_t>(msg.size()), skip);
}
static void append(std::vector<uint8_t> &dst, const std::vector<uint8_t> &src, size_t skip = 0) {
dst.insert(dst.end(), src.begin() + skip, src.end());
}
static std::vector<uint8_t> concat(std::initializer_list<std::vector<uint8_t>> parts) {
std::vector<uint8_t> out;
for (const auto &part : parts)
append(out, part);
return out;
}
/// The pipe delivers the filler first, then the drained messages.
static void expect_after_filler(const std::vector<uint8_t> &received, size_t filler,
const std::vector<uint8_t> &expected) {
ASSERT_EQ(received.size(), filler + expected.size());
EXPECT_TRUE(std::equal(expected.begin(), expected.end(), received.begin() + filler));
}
// Non-blocking socket pair with small buffers, so the writer fills like a stalled TCP connection
class OverflowBufferTest : public ::testing::Test {
protected:
void SetUp() override {
int fds[2];
ASSERT_EQ(::socketpair(AF_UNIX, SOCK_STREAM, 0, fds), 0);
int size = 4096;
ASSERT_EQ(::setsockopt(fds[0], SOL_SOCKET, SO_SNDBUF, &size, sizeof(size)), 0);
ASSERT_EQ(::setsockopt(fds[1], SOL_SOCKET, SO_RCVBUF, &size, sizeof(size)), 0);
ASSERT_EQ(::fcntl(fds[1], F_SETFL, O_NONBLOCK), 0);
this->reader_ = fds[1];
this->sock_ = std::make_unique<socket::Socket>(fds[0]);
ASSERT_EQ(this->sock_->setblocking(false), 0);
}
void TearDown() override { ::close(this->reader_); }
/// Write filler until the socket refuses; returns the bytes accepted
size_t fill_pipe_() {
uint8_t junk[512];
std::memset(junk, 0xEE, sizeof(junk));
size_t total = 0;
for (;;) {
ssize_t written = this->sock_->write(junk, sizeof(junk));
if (written <= 0)
break;
total += static_cast<size_t>(written);
}
return total;
}
/// Append whatever the pipe currently holds.
void read_into_(std::vector<uint8_t> &out) {
uint8_t tmp[1024];
for (;;) {
ssize_t n = ::read(this->reader_, tmp, sizeof(tmp));
if (n <= 0)
break;
out.insert(out.end(), tmp, tmp + n);
}
}
/// Drain once; a refusal must be a would-block, never a hard error.
ssize_t drain_(TestOverflowBuffer &buf) {
ssize_t sent = buf.try_drain(this->sock_.get());
if (sent == -1) {
EXPECT_TRUE(errno == EWOULDBLOCK || errno == EAGAIN);
}
return sent;
}
/// Read and drain until the backlog is empty; returns all bytes received
std::vector<uint8_t> drain_all_(TestOverflowBuffer &buf) {
std::vector<uint8_t> received;
for (int i = 0; i < 10000 && !buf.empty(); i++) {
this->read_into_(received);
// A hard socket error would never clear the backlog; stop instead of spinning
if (this->drain_(buf) == -1 && errno != EWOULDBLOCK && errno != EAGAIN)
break;
}
EXPECT_TRUE(buf.empty());
this->read_into_(received);
return received;
}
struct Stall {
size_t filler;
std::vector<uint8_t> first, second, received;
TestOverflowBuffer::Storage before;
};
/// Park two messages, then drain the first fully and the second part way
void stall_mid_message_(TestOverflowBuffer &buf, Stall &s) {
s.filler = this->fill_pipe_();
s.first = make_message(1500, 20);
ASSERT_GT(s.filler, s.first.size()); // the first message must drain in one go
// Larger than the whole pipe, so a drain always stops inside it
s.second = make_message(std::max<size_t>(s.filler + 1, std::min<size_t>(s.filler * 3, 12000)), 60);
ASSERT_GT(s.second.size(), s.filler);
ASSERT_TRUE(enqueue(buf, s.first));
ASSERT_TRUE(enqueue(buf, s.second));
s.before = buf.storage();
this->read_into_(s.received);
ASSERT_GT(this->drain_(buf), 0);
ASSERT_EQ(buf.count(), 1);
}
int reader_{-1};
std::unique_ptr<socket::Socket> sock_;
};
TEST_F(OverflowBufferTest, IdleBufferOwnsNoStorage) {
TestOverflowBuffer buf;
EXPECT_TRUE(buf.empty());
EXPECT_EQ(buf.capacity(), 0u);
EXPECT_EQ(buf.storage().data, nullptr);
}
TEST_F(OverflowBufferTest, StorageIsReusedAcrossStalls) {
TestOverflowBuffer buf;
auto msg = make_message(1000, 1);
size_t filler = this->fill_pipe_();
ASSERT_TRUE(enqueue(buf, msg));
const auto storage = buf.storage();
EXPECT_GE(storage.capacity, msg.size() + TestOverflowBuffer::LEN_PREFIX);
for (int stall = 0; stall < 5; stall++) {
expect_after_filler(this->drain_all_(buf), filler, msg);
EXPECT_TRUE(buf.empty());
// Same allocation every time: no free, no new allocation
EXPECT_EQ(buf.storage(), storage);
filler = this->fill_pipe_();
ASSERT_TRUE(enqueue(buf, msg));
EXPECT_EQ(buf.storage(), storage);
}
}
TEST_F(OverflowBufferTest, ReleaseWhileQueuedFreesOnceDrained) {
TestOverflowBuffer buf;
auto msg = make_message(1000, 7);
size_t filler = this->fill_pipe_();
ASSERT_TRUE(enqueue(buf, msg));
const size_t capacity = buf.capacity();
// Requested while the backlog still holds data: storage must stay until sent
buf.release();
EXPECT_FALSE(buf.empty());
EXPECT_EQ(buf.capacity(), capacity);
expect_after_filler(this->drain_all_(buf), filler, msg);
EXPECT_TRUE(buf.empty());
EXPECT_EQ(buf.capacity(), 0u);
EXPECT_EQ(buf.storage().data, nullptr);
// A later stall allocates again and keeps it, since nobody asked for a release
filler = this->fill_pipe_();
ASSERT_TRUE(enqueue(buf, msg));
EXPECT_GT(buf.capacity(), 0u);
this->drain_all_(buf);
EXPECT_GT(buf.capacity(), 0u);
}
TEST_F(OverflowBufferTest, ReleaseWhenEmptyFreesImmediately) {
TestOverflowBuffer buf;
auto msg = make_message(100, 3);
this->fill_pipe_();
ASSERT_TRUE(enqueue(buf, msg));
this->drain_all_(buf);
EXPECT_GT(buf.capacity(), 0u);
buf.release();
EXPECT_EQ(buf.capacity(), 0u);
EXPECT_EQ(buf.storage().data, nullptr);
}
TEST_F(OverflowBufferTest, PreservesOrderAndSkipsSentPrefix) {
TestOverflowBuffer buf;
auto first = make_message(700, 10);
auto second_a = make_message(300, 50);
auto second_b = make_message(400, 90);
auto third = make_message(200, 130);
size_t filler = this->fill_pipe_();
// 100 bytes of the first message were already accepted by the socket
ASSERT_TRUE(enqueue(buf, first, 100));
// Two iovecs with the skip covering all of the first one plus part of the second
struct iovec iov[2] = {{second_a.data(), second_a.size()}, {second_b.data(), second_b.size()}};
const uint16_t second_skip = static_cast<uint16_t>(second_a.size() + 5);
ASSERT_TRUE(buf.enqueue_iov(iov, 2, static_cast<uint16_t>(second_a.size() + second_b.size()), second_skip));
ASSERT_TRUE(enqueue(buf, third));
EXPECT_EQ(buf.count(), 3);
// Nothing can go out while the pipe is full
EXPECT_EQ(this->drain_(buf), -1);
EXPECT_EQ(buf.count(), 3);
std::vector<uint8_t> expected;
append(expected, first, 100);
append(expected, second_b, 5);
append(expected, third);
expect_after_filler(this->drain_all_(buf), filler, expected);
}
TEST_F(OverflowBufferTest, RefusesWhenQueueIsFull) {
TestOverflowBuffer buf;
auto msg = make_message(16, 1);
size_t filler = this->fill_pipe_();
for (int i = 0; i < API_MAX_SEND_QUEUE; i++) {
ASSERT_TRUE(enqueue(buf, msg)) << "message " << i;
}
EXPECT_FALSE(enqueue(buf, msg));
EXPECT_EQ(buf.count(), API_MAX_SEND_QUEUE);
// Draining frees the slots again
std::vector<uint8_t> expected;
for (int i = 0; i < API_MAX_SEND_QUEUE; i++)
append(expected, msg);
expect_after_filler(this->drain_all_(buf), filler, expected);
this->fill_pipe_();
EXPECT_TRUE(enqueue(buf, msg));
EXPECT_EQ(buf.count(), 1);
}
TEST_F(OverflowBufferTest, SkipAtIovecBoundary) {
TestOverflowBuffer buf;
auto sent = make_message(300, 50);
auto unsent = make_message(400, 90);
size_t filler = this->fill_pipe_();
// The skip covers the first iovec exactly, so only the second is copied
struct iovec iov[2] = {{sent.data(), sent.size()}, {unsent.data(), unsent.size()}};
ASSERT_TRUE(
buf.enqueue_iov(iov, 2, static_cast<uint16_t>(sent.size() + unsent.size()), static_cast<uint16_t>(sent.size())));
EXPECT_EQ(buf.live(), unsent.size() + TestOverflowBuffer::LEN_PREFIX);
expect_after_filler(this->drain_all_(buf), filler, unsent);
}
TEST_F(OverflowBufferTest, AppendsBehindSentPrefixWhenItFits) {
TestOverflowBuffer buf;
size_t filler = this->fill_pipe_();
auto first = make_message(200, 20);
// Size the second message so the two land half way into a 256 byte step,
// leaving exactly 128 bytes of slack whatever the pipe accepted
const size_t base = std::max<size_t>(filler + 1, std::min<size_t>(filler * 3, 12000));
const size_t second_len = (base / 256 + 1) * 256 + 128 - first.size() - 2 * TestOverflowBuffer::LEN_PREFIX;
auto second = make_message(second_len, 60);
ASSERT_GT(second.size(), filler);
ASSERT_TRUE(enqueue(buf, first));
ASSERT_TRUE(enqueue(buf, second));
const auto storage = buf.storage();
const size_t slack = storage.capacity - first.size() - second.size() - 2 * TestOverflowBuffer::LEN_PREFIX;
ASSERT_EQ(slack, 128u);
auto third = make_message(slack - TestOverflowBuffer::LEN_PREFIX, 200);
std::vector<uint8_t> received;
this->read_into_(received);
ASSERT_GT(this->drain_(buf), 0);
ASSERT_EQ(buf.count(), 1);
const size_t live = buf.live();
// Fits in the tail, so the sent prefix is left alone
ASSERT_TRUE(enqueue(buf, third));
EXPECT_EQ(buf.storage(), storage);
EXPECT_EQ(buf.live(), live + third.size() + TestOverflowBuffer::LEN_PREFIX);
append(received, this->drain_all_(buf));
expect_after_filler(received, filler, concat({first, second, third}));
}
TEST_F(OverflowBufferTest, ReleaseSurvivesFurtherEnqueues) {
TestOverflowBuffer buf;
auto first = make_message(300, 7);
auto second = make_message(300, 70);
size_t filler = this->fill_pipe_();
ASSERT_TRUE(enqueue(buf, first));
buf.release();
ASSERT_TRUE(enqueue(buf, second));
EXPECT_GT(buf.capacity(), 0u);
expect_after_filler(this->drain_all_(buf), filler, concat({first, second}));
EXPECT_EQ(buf.capacity(), 0u);
}
TEST_F(OverflowBufferTest, RefusesWhenByteLimitIsExceeded) {
TestOverflowBuffer buf;
// Two of these fill the byte budget exactly, well before the slot count is reached
static_assert(API_MAX_SEND_QUEUE >= 3);
auto msg = make_message(TestOverflowBuffer::MAX_BYTES / 2 - TestOverflowBuffer::LEN_PREFIX, 1);
this->fill_pipe_();
ASSERT_TRUE(enqueue(buf, msg));
ASSERT_TRUE(enqueue(buf, msg));
EXPECT_FALSE(enqueue(buf, msg));
EXPECT_EQ(buf.count(), 2);
}
TEST_F(OverflowBufferTest, LoneMessageMayExceedByteLimit) {
TestOverflowBuffer buf;
// The oversized message must still fit under the lone message ceiling
static_assert(TestOverflowBuffer::MAX_BYTES + 100 + TestOverflowBuffer::LEN_PREFIX <=
TestOverflowBuffer::MAX_LONE_BYTES);
auto big = make_message(TestOverflowBuffer::MAX_BYTES + 100, 5);
auto small = make_message(16, 9);
// Refusing the only message would drop the connection for nothing
size_t filler = this->fill_pipe_();
ASSERT_TRUE(enqueue(buf, big));
EXPECT_EQ(buf.count(), 1);
// With a backlog present the byte limit applies again
EXPECT_FALSE(enqueue(buf, small));
EXPECT_EQ(buf.count(), 1);
expect_after_filler(this->drain_all_(buf), filler, big);
}
TEST_F(OverflowBufferTest, LoneMessageAboveOffsetLimitIsRefused) {
TestOverflowBuffer buf;
// Payload plus prefix is past the lone message ceiling
auto msg = make_message(TestOverflowBuffer::MAX_LONE_BYTES, 3);
this->fill_pipe_();
EXPECT_FALSE(enqueue(buf, msg));
EXPECT_TRUE(buf.empty());
EXPECT_EQ(buf.capacity(), 0u);
}
TEST_F(OverflowBufferTest, HardSocketErrorLeavesBacklogIntact) {
TestOverflowBuffer buf;
auto msg = make_message(300, 40);
this->fill_pipe_();
ASSERT_TRUE(enqueue(buf, msg));
// A closed socket fails every write outright, unlike a full one
ASSERT_EQ(this->sock_->close(), 0);
errno = 0;
EXPECT_EQ(buf.try_drain(this->sock_.get()), -1);
EXPECT_NE(errno, EWOULDBLOCK);
EXPECT_NE(errno, EAGAIN);
EXPECT_EQ(buf.count(), 1);
EXPECT_EQ(buf.live(), msg.size() + TestOverflowBuffer::LEN_PREFIX);
}
TEST_F(OverflowBufferTest, GrowsWhileReclaimingSentPrefix) {
TestOverflowBuffer buf;
Stall s;
ASSERT_NO_FATAL_FAILURE(this->stall_mid_message_(buf, s));
// One byte too many to fit even after the sent prefix is reclaimed: grows in one copy
auto third = make_message(s.before.capacity - buf.live() + 1, 200);
ASSERT_TRUE(enqueue(buf, third));
EXPECT_GT(buf.capacity(), s.before.capacity);
EXPECT_EQ(buf.count(), 2);
append(s.received, this->drain_all_(buf));
expect_after_filler(s.received, s.filler, concat({s.first, s.second, third}));
}
TEST_F(OverflowBufferTest, NestedDrainMakesNoProgress) {
TestOverflowBuffer buf;
auto msg = make_message(300, 40);
size_t filler = this->fill_pipe_();
ASSERT_TRUE(enqueue(buf, msg));
std::vector<uint8_t> received;
this->read_into_(received);
// Room is available, but a nested drain must leave the outer one's message alone
buf.set_draining(true);
EXPECT_EQ(this->drain_(buf), 0);
EXPECT_EQ(buf.count(), 1);
std::vector<uint8_t> nothing;
this->read_into_(nothing);
EXPECT_TRUE(nothing.empty());
buf.set_draining(false);
append(received, this->drain_all_(buf));
expect_after_filler(received, filler, msg);
}
TEST_F(OverflowBufferTest, NestedEnqueueAppendsWithinCapacity) {
TestOverflowBuffer buf;
auto first = make_message(500, 10);
auto second = make_message(4, 90);
size_t filler = this->fill_pipe_();
ASSERT_TRUE(enqueue(buf, first));
const auto storage = buf.storage();
ASSERT_GE(storage.capacity, first.size() + second.size() + 2 * TestOverflowBuffer::LEN_PREFIX);
buf.set_draining(true);
EXPECT_TRUE(enqueue(buf, second));
EXPECT_EQ(buf.count(), 2);
EXPECT_EQ(buf.storage(), storage);
buf.set_draining(false);
expect_after_filler(this->drain_all_(buf), filler, concat({first, second}));
}
TEST_F(OverflowBufferTest, NestedEnqueueRefusesToGrow) {
TestOverflowBuffer buf;
auto first = make_message(500, 10);
auto second = make_message(100, 90);
size_t filler = this->fill_pipe_();
ASSERT_TRUE(enqueue(buf, first));
const auto storage = buf.storage();
ASSERT_LT(storage.capacity, first.size() + second.size() + 2 * TestOverflowBuffer::LEN_PREFIX);
// Growing would free the bytes the outer write() is sending from
buf.set_draining(true);
EXPECT_FALSE(enqueue(buf, second));
EXPECT_EQ(buf.count(), 1);
EXPECT_EQ(buf.storage(), storage);
buf.set_draining(false);
expect_after_filler(this->drain_all_(buf), filler, first);
}
TEST_F(OverflowBufferTest, NestedEnqueueRefusesToCompact) {
TestOverflowBuffer buf;
Stall s;
ASSERT_NO_FATAL_FAILURE(this->stall_mid_message_(buf, s));
auto third = make_message(1000, 200);
// Sliding the remainder down would move the bytes the outer write() points at
buf.set_draining(true);
EXPECT_FALSE(enqueue(buf, third));
EXPECT_EQ(buf.count(), 1);
EXPECT_EQ(buf.storage(), s.before);
buf.set_draining(false);
// Once the drain is over the same enqueue compacts and succeeds
ASSERT_TRUE(enqueue(buf, third));
EXPECT_EQ(buf.storage(), s.before);
append(s.received, this->drain_all_(buf));
expect_after_filler(s.received, s.filler, concat({s.first, s.second, third}));
}
TEST_F(OverflowBufferTest, CompactsInsteadOfGrowingAfterPartialDrain) {
TestOverflowBuffer buf;
Stall s;
ASSERT_NO_FATAL_FAILURE(this->stall_mid_message_(buf, s));
auto third = make_message(1000, 200);
// The sent first message is reclaimed by sliding the remainder down, not by reallocating
ASSERT_TRUE(enqueue(buf, third));
EXPECT_EQ(buf.storage(), s.before);
append(s.received, this->drain_all_(buf));
expect_after_filler(s.received, s.filler, concat({s.first, s.second, third}));
}
} // namespace esphome::api::testing
#endif // USE_HOST
@@ -6,15 +6,14 @@ def override_manifest(manifest: ComponentManifestOverride) -> None:
# close_service_batch compiles only under USE_BLUETOOTH_PROXY_CONNECTIONS;
# emit the backend define so the host build exercises it.
async def to_code_testing(config):
# These defines are global to the merged host test binary; safe
# because no co-compiled test observes them.
# These defines are global to the merged host test binary. The api sources are
# compiled in it too (the api tests define USE_API), and USE_BLUETOOTH_PROXY would make
# them include and call bluetooth_proxy, which has no host build without a BLE hub.
cg.add_define("USE_BLE_GATT_CLIENT")
cg.add_define("USE_BLE_GATT_CLIENT_STUB_BACKEND")
cg.add_define("USE_BLUETOOTH_PROXY")
# Gates the connection half of the API surface, which is what
# close_service_batch and the GATT response types live behind.
cg.add_define("USE_BLUETOOTH_PROXY_CONNECTIONS")
cg.add_define("BLUETOOTH_PROXY_ADVERTISEMENT_BATCH_SIZE", 16)
cg.add_define("BLUETOOTH_PROXY_MAX_CONNECTIONS", 1)
manifest.to_code = to_code_testing
@@ -0,0 +1,12 @@
# Compile the gated filter path; no external component is in-tree to call
# enable_advertisement_filter(), so the define is forced here.
<<: !include common.yaml
esphome:
build_flags:
- "-DUSE_BLUETOOTH_PROXY_ADVERTISEMENT_FILTER"
esp32_ble_tracker:
bluetooth_proxy:
active: true
+18
View File
@@ -0,0 +1,18 @@
tinyusb:
id: tinyusb_test
usb_lang_id: 0x0123
usb_manufacturer_str: ESPHomeTestManufacturer
usb_product_id: 0x1234
usb_product_str: ESPHomeTestProduct
usb_serial_str: ESPHomeTestSerialNumber
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
@@ -0,0 +1,12 @@
# Second UART/CDC pair for a two-bridge setup. Kept out of common.yaml because the
# ESP32-S2 has only two UART controllers and the logger occupies one, so a second
# uart there would fail at runtime.
uart:
- id: uart_1
tx_pin: 15
rx_pin: 16
baud_rate: 115200
usb_cdc_acm:
interfaces:
- id: cdc_acm_2
@@ -0,0 +1,15 @@
packages:
cdc_acm_uart: !include common.yaml
cdc_acm_uart_dual: !include common_dual.yaml
bridge:
- platform: cdc_acm_uart
uart_id: uart_0
usb_cdc_acm_id: cdc_acm_1
dtr_pin: 40
rts_pin: 41
- platform: cdc_acm_uart
uart_id: uart_1
usb_cdc_acm_id: cdc_acm_2
dtr_pin: 20
rts_pin: 21
@@ -0,0 +1,14 @@
# 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:
cdc_acm_uart: !include common.yaml
bridge:
- platform: cdc_acm_uart
uart_id: uart_0
usb_cdc_acm_id: cdc_acm_1
dtr_pin: 40
rts_pin: 41
@@ -0,0 +1,17 @@
packages:
cdc_acm_uart: !include common.yaml
cdc_acm_uart_dual: !include common_dual.yaml
bridge:
- platform: cdc_acm_uart
uart_id: uart_0
usb_cdc_acm_id: cdc_acm_1
dtr_pin: 40
rts_pin: 41
- platform: cdc_acm_uart
uart_id: uart_1
usb_cdc_acm_id: cdc_acm_2
# GPIO19/20 are USB D-/D+ on the S3 (which the CDC side itself uses); use
# unrelated free pins here.
dtr_pin: 17
rts_pin: 18
+71
View File
@@ -348,4 +348,75 @@ TEST(StepToAccuracyDecimals, NonFiniteAndZero) {
EXPECT_EQ(step_to_accuracy_decimals(-INFINITY), 0);
}
// --- FixedVector::try_init() ---
// Keeps the block observable, else the compiler may drop the malloc and free pair and fold the check
static void escape(const void *p) { asm volatile("" : : "g"(p) : "memory"); }
TEST(FixedVectorTryInit, ReportsExhaustionAndStaysEmpty) {
FixedVector<uint32_t> v;
const bool ok = v.try_init(SIZE_MAX / sizeof(uint32_t));
escape(&v);
EXPECT_FALSE(ok);
EXPECT_EQ(v.capacity(), 0u);
EXPECT_FALSE(v.try_init(SIZE_MAX / sizeof(uint32_t) + 1)); // byte count would wrap
EXPECT_EQ(v.capacity(), 0u);
EXPECT_TRUE(v.try_init(0));
EXPECT_TRUE(v.try_init(4));
v.push_back(7);
EXPECT_EQ(v.size(), 1u);
}
// --- RAMAllocator::make_unique() ---
namespace {
struct Probe {
static inline int live = 0;
int a;
int b;
Probe(int a, int b) : a(a), b(b) { live++; }
~Probe() { live--; }
};
} // namespace
static_assert(sizeof(RAMUniquePtr<Probe>) == sizeof(Probe *), "the deleter must not add storage");
TEST(RAMAllocatorMakeUnique, ForwardsArgsAndDestroysOnce) {
auto p = RAMAllocator<Probe>().make_unique(3, 4);
ASSERT_NE(p, nullptr);
EXPECT_EQ(p->a, 3);
EXPECT_EQ(p->b, 4);
EXPECT_EQ(Probe::live, 1);
p.reset();
EXPECT_EQ(Probe::live, 0);
}
TEST(RAMAllocatorMakeUnique, ValueInitializesLikeMakeUnique) {
struct Plain {
uint32_t words[8];
};
// Dirty a block of the same size first so a recycled allocation is not zero by chance
auto dirty = RAMAllocator<uint8_t>().make_unique_array_for_overwrite(sizeof(Plain));
std::memset(dirty.get(), 0xFF, sizeof(Plain));
dirty.reset();
auto p = RAMAllocator<Plain>().make_unique();
ASSERT_NE(p, nullptr);
// Under ASan fresh blocks are filled with 0xbe, so this holds even when the dirtied block is not reused
EXPECT_TRUE(std::all_of(std::begin(p->words), std::end(p->words), [](uint32_t w) { return w == 0; }));
}
TEST(RAMAllocatorMakeUnique, ArrayFormRejectsOverflowAndZero) {
EXPECT_EQ(RAMAllocator<uint32_t>().make_unique_array_for_overwrite(SIZE_MAX / sizeof(uint32_t) + 1), nullptr);
EXPECT_EQ(RAMAllocator<uint32_t>().make_unique_array_for_overwrite(0), nullptr);
EXPECT_NE(RAMAllocator<uint32_t>().make_unique_array_for_overwrite(1), nullptr);
}
TEST(RAMAllocatorMakeUnique, ArrayFormAllocatesElements) {
RAMUniquePtr<uint8_t[]> buf = RAMAllocator<uint8_t>().make_unique_array_for_overwrite(256);
ASSERT_NE(buf, nullptr);
std::memset(buf.get(), 0xA5, 256);
EXPECT_EQ(buf[0], 0xA5);
EXPECT_EQ(buf[255], 0xA5);
}
} // namespace esphome::core::testing
@@ -17,6 +17,8 @@ esp32:
disable_dev_null_vfs: true
disable_mbedtls_peer_cert: true
disable_mbedtls_pkcs7: true
disable_mbedtls_tls_server: true
disable_mbedtls_tls_extras: true
disable_regi2c_in_iram: true
disable_fatfs: true
sram1_as_iram: true
@@ -0,0 +1,4 @@
substitutions:
verify_ssl: "true"
<<: !include common.yaml
+15 -1
View File
@@ -30,6 +30,18 @@ binary_sensor:
widget: button_button
state: pressed
globals:
- id: counter
type: int
script:
- id: add_row
then:
- lvgl.list.add:
id: test_list_id
label:
text: row
lvgl:
id: lvgl_id
rotation: 90
@@ -1291,7 +1303,7 @@ lvgl:
then:
- logger.log:
format: "table selected row %u col %u"
args: [row, column]
args: [(unsigned)row, (unsigned)column]
on_click:
then:
- lvgl.table.cell.update:
@@ -1347,10 +1359,12 @@ lvgl:
- logger.log:
format: "list entry added at %d"
args: [list_index]
- lambda: "id(counter)++;"
on_remove:
- logger.log:
format: "list entry removed at %d"
args: [list_index]
- lambda: "id(counter)--;"
on_click:
- lvgl.list.add_text:
id: test_list_id
@@ -0,0 +1,6 @@
# A receiver with no dumpers and no listeners compiles both lists out.
# Only built while remote_receiver is tested in isolation: the counts are global defines,
# so this variant cannot be merged with configs that register any.
remote_receiver:
- id: rcvr_bare
pin: ${pin}
@@ -0,0 +1,5 @@
substitutions:
pin: GPIO2
packages:
bare: !include bare-common.yaml
@@ -4,3 +4,6 @@ psram:
sendspin:
id: sendspin_hub_id
task_stack_in_psram: true
manufacturer: Test Manufacturer
model: Test Model
firmware_version: 1.2.3
@@ -0,0 +1,14 @@
substitutions:
tx_pin: GPIO4
rx_pin: GPIO5
# Compile the tap code paths; no tap is attached, so this exercises the
# null-tap branches that a normal build never defines.
esphome:
platformio_options:
build_flags:
- "-DUSE_SERIAL_PROXY_TAP"
packages:
uart: !include ../../test_build_components/common/uart/esp32-idf.yaml
serial_proxy: !include common.yaml
@@ -0,0 +1,34 @@
esphome:
name: set-internal-at-boot
on_boot:
then:
- lambda: |-
id(hidden_at_boot).set_internal(true);
id(shown_at_boot).set_internal(false);
host:
api:
actions:
- action: set_internal_late
then:
- lambda: id(untouched).set_internal(true);
logger:
sensor:
- platform: template
name: "Hidden At Boot"
id: hidden_at_boot
lambda: return 1.0;
- platform: template
name: "Shown At Boot"
id: shown_at_boot
internal: true
lambda: return 2.0;
- platform: template
name: "Untouched"
id: untouched
lambda: return 3.0;
@@ -0,0 +1,16 @@
esphome:
name: wh-template-unknown-test
host:
api:
logger:
water_heater:
- platform: template
id: unknown_boiler
name: Unknown Boiler
# Both temperatures stay unknown, as they do before an upstream component reports a value.
current_temperature: !lambda "return NAN;"
target_temperature: !lambda "return NAN;"
supported_modes:
- "off"
- eco
+153 -141
View File
@@ -1,143 +1,155 @@
{
"tests/integration/test_action_concurrent_reentry.py": 57.91,
"tests/integration/test_addressable_light_transition.py": 21.25,
"tests/integration/test_alarm_control_panel_state_transitions.py": 70.71,
"tests/integration/test_api_action_metadata.py": 66.6,
"tests/integration/test_api_action_responses.py": 36.1,
"tests/integration/test_api_action_timeout.py": 68.86,
"tests/integration/test_api_conditional_memory.py": 15.48,
"tests/integration/test_api_custom_services.py": 18.77,
"tests/integration/test_api_get_time_response_timezone.py": 21.08,
"tests/integration/test_api_homeassistant.py": 65.59,
"tests/integration/test_api_homeassistant_action_no_subscriber.py": 18.44,
"tests/integration/test_api_homeassistant_binary_sensor_initial_state.py": 15.05,
"tests/integration/test_api_list_entities_backpressure.py": 13.88,
"tests/integration/test_api_message_size_batching.py": 29.98,
"tests/integration/test_api_reboot_timeout.py": 16.05,
"tests/integration/test_api_string_lambda.py": 15.31,
"tests/integration/test_api_vv_logging.py": 19.28,
"tests/integration/test_api_zero_psk_provisioning.py": 31.5,
"tests/integration/test_areas_and_devices.py": 24.95,
"tests/integration/test_automation_wait_actions.py": 20.92,
"tests/integration/test_automations.py": 35.19,
"tests/integration/test_batch_delay_zero_rapid_transitions.py": 17.99,
"tests/integration/test_binary_sensor_autorepeat_filter.py": 20.39,
"tests/integration/test_binary_sensor_invalidate_state.py": 18.41,
"tests/integration/test_blocking_warning_log_time_not_charged_to_next_operation.py": 24.69,
"tests/integration/test_build_info.py": 18.7,
"tests/integration/test_camera_mock.py": 16.23,
"tests/integration/test_climate_control_action.py": 21.14,
"tests/integration/test_climate_custom_modes.py": 20.74,
"tests/integration/test_continuation_actions.py": 16.81,
"tests/integration/test_cover_control_action.py": 20.34,
"tests/integration/test_crc8_helper.py": 9.36,
"tests/integration/test_device_id_in_state.py": 44.67,
"tests/integration/test_duplicate_entities.py": 23.58,
"tests/integration/test_entity_icon.py": 34.35,
"tests/integration/test_fan_turn_on_action.py": 24.23,
"tests/integration/test_fnv1_hash_object_id.py": 16.21,
"tests/integration/test_fnv1a_hash.py": 13.38,
"tests/integration/test_gpio_expander_cache.py": 13.06,
"tests/integration/test_host_logger_thread_safety.py": 23.66,
"tests/integration/test_host_mode_basic.py": 8.01,
"tests/integration/test_host_mode_batch_delay.py": 21.0,
"tests/integration/test_host_mode_climate_basic_state.py": 22.14,
"tests/integration/test_host_mode_climate_control.py": 19.39,
"tests/integration/test_host_mode_empty_string_options.py": 21.76,
"tests/integration/test_host_mode_entity_fields.py": 29.61,
"tests/integration/test_host_mode_fan_preset.py": 20.01,
"tests/integration/test_host_mode_many_entities.py": 39.08,
"tests/integration/test_host_mode_many_entities_multiple_connections.py": 23.92,
"tests/integration/test_host_mode_noise_encryption.py": 42.42,
"tests/integration/test_host_mode_reconnect.py": 3.41,
"tests/integration/test_host_mode_sensor.py": 22.96,
"tests/integration/test_host_ota.py": 29.5,
"tests/integration/test_host_preferences.py": 16.06,
"tests/integration/test_host_preferences_suspend_resume.py": 18.71,
"tests/integration/test_improv_serial_uart.py": 20.22,
"tests/integration/test_large_message_batching.py": 26.56,
"tests/integration/test_legacy_area.py": 22.72,
"tests/integration/test_legacy_climate_compat.py": 14.13,
"tests/integration/test_legacy_fan_compat.py": 14.33,
"tests/integration/test_light_automations.py": 18.81,
"tests/integration/test_light_binary_effect_off_phase.py": 8.38,
"tests/integration/test_light_calls.py": 21.88,
"tests/integration/test_light_constant_brightness.py": 59.45,
"tests/integration/test_light_control_action.py": 31.91,
"tests/integration/test_light_dim_relative_action.py": 14.43,
"tests/integration/test_light_effect_zero_brightness.py": 25.05,
"tests/integration/test_light_initial_state.py": 18.97,
"tests/integration/test_light_toggle_action.py": 17.44,
"tests/integration/test_lock_automations.py": 18.9,
"tests/integration/test_logger_buffered_recursion_guard.py": 18.2,
"tests/integration/test_loop_disable_enable.py": 63.35,
"tests/integration/test_loop_interval_decoupling.py": 17.7,
"tests/integration/test_loop_interval_default_not_pulled_forward.py": 21.56,
"tests/integration/test_micros_to_millis.py": 15.89,
"tests/integration/test_multi_click_trigger.py": 17.23,
"tests/integration/test_multi_device_preferences.py": 19.4,
"tests/integration/test_noise_encryption_key_protection.py": 72.59,
"tests/integration/test_object_id_api_verification.py": 19.22,
"tests/integration/test_object_id_friendly_name_no_mac_suffix.py": 16.77,
"tests/integration/test_object_id_no_friendly_name.py": 45.8,
"tests/integration/test_online_image_auto_detects_image_bmp_mime.py": 86.73,
"tests/integration/test_online_image_auto_detects_redirected_image_bmp_mime.py": 40.4,
"tests/integration/test_online_image_bmp.py": 37.24,
"tests/integration/test_oversized_payloads.py": 55.75,
"tests/integration/test_preference_key_stability.py": 25.49,
"tests/integration/test_runtime_stats.py": 29.81,
"tests/integration/test_safe_mode_loop_runs.py": 6.26,
"tests/integration/test_scheduler_blocking_warning.py": 37.98,
"tests/integration/test_scheduler_bulk_cleanup.py": 18.67,
"tests/integration/test_scheduler_defer_cancel.py": 18.46,
"tests/integration/test_scheduler_defer_cancel_regular.py": 16.34,
"tests/integration/test_scheduler_defer_fifo_simple.py": 18.26,
"tests/integration/test_scheduler_defer_stress.py": 17.74,
"tests/integration/test_scheduler_heap_stress.py": 3.89,
"tests/integration/test_scheduler_internal_id_no_collision.py": 20.01,
"tests/integration/test_scheduler_interval_reschedule.py": 16.29,
"tests/integration/test_scheduler_interval_zero_coerced.py": 16.09,
"tests/integration/test_scheduler_null_name.py": 14.69,
"tests/integration/test_scheduler_numeric_id_test.py": 17.08,
"tests/integration/test_scheduler_pool.py": 19.88,
"tests/integration/test_scheduler_rapid_cancellation.py": 4.42,
"tests/integration/test_scheduler_recursive_timeout.py": 4.3,
"tests/integration/test_scheduler_removed_item_race.py": 15.49,
"tests/integration/test_scheduler_self_keyed.py": 25.77,
"tests/integration/test_scheduler_simultaneous_callbacks.py": 14.84,
"tests/integration/test_scheduler_string_test.py": 15.42,
"tests/integration/test_script_array_params.py": 12.73,
"tests/integration/test_script_delay_params.py": 12.69,
"tests/integration/test_script_queued.py": 20.38,
"tests/integration/test_script_queued_idle_loop.py": 25.06,
"tests/integration/test_script_wait_on_boot.py": 15.67,
"tests/integration/test_select_stringref_trigger.py": 19.48,
"tests/integration/test_sensor_filters_delta.py": 27.62,
"tests/integration/test_sensor_filters_ring_buffer.py": 20.27,
"tests/integration/test_sensor_filters_sliding_window.py": 56.28,
"tests/integration/test_sensor_filters_value_list.py": 20.6,
"tests/integration/test_sensor_timeout_filter.py": 22.21,
"tests/integration/test_socket_wake_gate_tcp.py": 16.37,
"tests/integration/test_status_flags.py": 29.68,
"tests/integration/test_strftime_to.py": 17.42,
"tests/integration/test_syslog.py": 18.39,
"tests/integration/test_template_alarm_control_panel_many_sensors.py": 25.61,
"tests/integration/test_template_text_save.py": 19.16,
"tests/integration/test_text_command.py": 16.43,
"tests/integration/test_text_sensor_raw_state.py": 17.19,
"tests/integration/test_uart_mock_ld2410.py": 37.0,
"tests/integration/test_uart_mock_ld2412.py": 40.82,
"tests/integration/test_uart_mock_ld2420.py": 32.7,
"tests/integration/test_uart_mock_ld2450.py": 32.84,
"tests/integration/test_uart_mock_modbus.py": 548.87,
"tests/integration/test_udp.py": 16.67,
"tests/integration/test_use_address_runtime.py": 27.26,
"tests/integration/test_valve_control_action.py": 24.58,
"tests/integration/test_varint_five_byte_device_id.py": 22.5,
"tests/integration/test_wait_until_mid_loop_timing.py": 22.05,
"tests/integration/test_wait_until_on_boot.py": 10.37,
"tests/integration/test_wait_until_ordering.py": 18.23,
"tests/integration/test_wait_until_reentrant_restart.py": 19.35,
"tests/integration/test_wake_loop_forces_phase_b.py": 17.83,
"tests/integration/test_water_heater_template.py": 25.7
"tests/integration/test_action_concurrent_reentry.py": 34.72,
"tests/integration/test_addressable_light_transition.py": 33.71,
"tests/integration/test_alarm_control_panel_state_transitions.py": 38.96,
"tests/integration/test_api_action_metadata.py": 33.36,
"tests/integration/test_api_action_responses.py": 26.22,
"tests/integration/test_api_action_timeout.py": 25.41,
"tests/integration/test_api_conditional_memory.py": 20.74,
"tests/integration/test_api_custom_services.py": 23.21,
"tests/integration/test_api_get_time_response_timezone.py": 25.04,
"tests/integration/test_api_homeassistant.py": 24.18,
"tests/integration/test_api_homeassistant_action_no_subscriber.py": 23.47,
"tests/integration/test_api_homeassistant_binary_sensor_initial_state.py": 22.86,
"tests/integration/test_api_list_entities_backpressure.py": 18.8,
"tests/integration/test_api_message_size_batching.py": 28.8,
"tests/integration/test_api_reboot_timeout.py": 9.47,
"tests/integration/test_api_string_lambda.py": 16.88,
"tests/integration/test_api_vv_logging.py": 17.99,
"tests/integration/test_api_zero_psk_provisioning.py": 47.07,
"tests/integration/test_areas_and_devices.py": 20.77,
"tests/integration/test_automation_wait_actions.py": 20.07,
"tests/integration/test_automations.py": 27.41,
"tests/integration/test_batch_delay_zero_rapid_transitions.py": 20.5,
"tests/integration/test_binary_sensor_autorepeat_filter.py": 26.54,
"tests/integration/test_binary_sensor_invalidate_state.py": 16.09,
"tests/integration/test_blocking_warning_log_time_not_charged_to_next_operation.py": 14.7,
"tests/integration/test_build_info.py": 18.07,
"tests/integration/test_camera_mock.py": 20.44,
"tests/integration/test_climate_control_action.py": 27.97,
"tests/integration/test_climate_custom_modes.py": 26.77,
"tests/integration/test_continuation_actions.py": 12.09,
"tests/integration/test_cover_control_action.py": 19.77,
"tests/integration/test_crc8_helper.py": 12.64,
"tests/integration/test_device_id_in_state.py": 63.19,
"tests/integration/test_duplicate_entities.py": 29.26,
"tests/integration/test_entity_icon.py": 34.95,
"tests/integration/test_fan_turn_on_action.py": 25.98,
"tests/integration/test_fnv1_hash_object_id.py": 4.85,
"tests/integration/test_fnv1a_hash.py": 5.14,
"tests/integration/test_gpio_expander_cache.py": 21.0,
"tests/integration/test_host_logger_thread_safety.py": 17.51,
"tests/integration/test_host_mode_basic.py": 21.2,
"tests/integration/test_host_mode_batch_delay.py": 26.68,
"tests/integration/test_host_mode_climate_basic_state.py": 18.04,
"tests/integration/test_host_mode_climate_control.py": 29.64,
"tests/integration/test_host_mode_empty_string_options.py": 28.8,
"tests/integration/test_host_mode_entity_fields.py": 28.68,
"tests/integration/test_host_mode_fan_preset.py": 16.98,
"tests/integration/test_host_mode_many_entities.py": 39.8,
"tests/integration/test_host_mode_many_entities_multiple_connections.py": 33.12,
"tests/integration/test_host_mode_noise_encryption.py": 52.53,
"tests/integration/test_host_mode_reconnect.py": 14.48,
"tests/integration/test_host_mode_sensor.py": 17.38,
"tests/integration/test_host_ota.py": 94.96,
"tests/integration/test_host_preferences.py": 27.11,
"tests/integration/test_host_preferences_suspend_resume.py": 21.48,
"tests/integration/test_improv_serial_uart.py": 19.43,
"tests/integration/test_large_message_batching.py": 25.67,
"tests/integration/test_legacy_area.py": 15.59,
"tests/integration/test_legacy_climate_compat.py": 18.59,
"tests/integration/test_legacy_fan_compat.py": 18.34,
"tests/integration/test_light_automations.py": 16.74,
"tests/integration/test_light_binary_effect_off_phase.py": 57.7,
"tests/integration/test_light_calls.py": 25.88,
"tests/integration/test_light_constant_brightness.py": 22.32,
"tests/integration/test_light_control_action.py": 18.15,
"tests/integration/test_light_dim_relative_action.py": 30.35,
"tests/integration/test_light_effect_zero_brightness.py": 18.38,
"tests/integration/test_light_initial_state.py": 23.93,
"tests/integration/test_light_toggle_action.py": 26.16,
"tests/integration/test_lock_automations.py": 34.13,
"tests/integration/test_logger_buffered_recursion_guard.py": 25.77,
"tests/integration/test_loop_disable_enable.py": 14.71,
"tests/integration/test_loop_interval_decoupling.py": 26.16,
"tests/integration/test_loop_interval_default_not_pulled_forward.py": 28.47,
"tests/integration/test_lvgl_headless_render.py": 96.36,
"tests/integration/test_micros_to_millis.py": 28.76,
"tests/integration/test_multi_click_trigger.py": 19.8,
"tests/integration/test_multi_device_preferences.py": 38.85,
"tests/integration/test_noise_encryption_key_protection.py": 25.81,
"tests/integration/test_object_id_api_verification.py": 28.46,
"tests/integration/test_object_id_friendly_name_no_mac_suffix.py": 16.14,
"tests/integration/test_object_id_no_friendly_name.py": 18.82,
"tests/integration/test_online_image_auto_detects_image_bmp_mime.py": 30.16,
"tests/integration/test_online_image_auto_detects_redirected_image_bmp_mime.py": 40.67,
"tests/integration/test_online_image_bmp.py": 7.41,
"tests/integration/test_oversized_payloads.py": 59.52,
"tests/integration/test_preference_key_stability.py": 27.24,
"tests/integration/test_runtime_stats.py": 20.53,
"tests/integration/test_safe_mode_loop_runs.py": 10.17,
"tests/integration/test_scheduler_blocking_warning.py": 51.45,
"tests/integration/test_scheduler_bulk_cleanup.py": 22.59,
"tests/integration/test_scheduler_defer_cancel.py": 17.64,
"tests/integration/test_scheduler_defer_cancel_regular.py": 21.61,
"tests/integration/test_scheduler_defer_fifo_simple.py": 24.73,
"tests/integration/test_scheduler_defer_stress.py": 23.91,
"tests/integration/test_scheduler_heap_stress.py": 25.77,
"tests/integration/test_scheduler_internal_id_no_collision.py": 19.83,
"tests/integration/test_scheduler_interval_reschedule.py": 23.4,
"tests/integration/test_scheduler_interval_zero_coerced.py": 5.11,
"tests/integration/test_scheduler_null_name.py": 17.36,
"tests/integration/test_scheduler_numeric_id_test.py": 20.81,
"tests/integration/test_scheduler_pool.py": 17.42,
"tests/integration/test_scheduler_rapid_cancellation.py": 25.28,
"tests/integration/test_scheduler_recursive_timeout.py": 16.48,
"tests/integration/test_scheduler_removed_item_race.py": 16.14,
"tests/integration/test_scheduler_self_keyed.py": 26.19,
"tests/integration/test_scheduler_simultaneous_callbacks.py": 23.26,
"tests/integration/test_scheduler_string_test.py": 27.09,
"tests/integration/test_script_array_params.py": 3.6,
"tests/integration/test_script_delay_params.py": 24.74,
"tests/integration/test_script_queued.py": 17.21,
"tests/integration/test_script_queued_idle_loop.py": 3.4,
"tests/integration/test_script_wait_on_boot.py": 23.7,
"tests/integration/test_sdl_headless_screenshot.py": 19.53,
"tests/integration/test_select_stringref_trigger.py": 18.93,
"tests/integration/test_sensor_filters_delta.py": 20.85,
"tests/integration/test_sensor_filters_ring_buffer.py": 16.82,
"tests/integration/test_sensor_filters_sliding_window.py": 54.78,
"tests/integration/test_sensor_filters_value_list.py": 19.46,
"tests/integration/test_sensor_timeout_filter.py": 18.39,
"tests/integration/test_set_internal_at_boot.py": 21.69,
"tests/integration/test_snapshot_display.py": 12.64,
"tests/integration/test_socket_wake_gate_tcp.py": 13.08,
"tests/integration/test_status_flags.py": 29.54,
"tests/integration/test_strftime_to.py": 25.62,
"tests/integration/test_syslog.py": 16.25,
"tests/integration/test_template_alarm_control_panel_many_sensors.py": 27.18,
"tests/integration/test_template_climate_basic.py": 20.51,
"tests/integration/test_template_climate_custom_modes.py": 27.73,
"tests/integration/test_template_climate_nonoptimistic.py": 26.35,
"tests/integration/test_template_climate_on_control_ordering.py": 26.55,
"tests/integration/test_template_climate_publish_all_fields.py": 17.59,
"tests/integration/test_template_climate_sensor_push.py": 22.04,
"tests/integration/test_template_climate_set_actions.py": 16.82,
"tests/integration/test_template_climate_two_point_temperature.py": 25.13,
"tests/integration/test_template_text_save.py": 25.36,
"tests/integration/test_text_command.py": 18.79,
"tests/integration/test_text_sensor_raw_state.py": 17.07,
"tests/integration/test_uart_mock_ld2410.py": 59.58,
"tests/integration/test_uart_mock_ld2412.py": 59.4,
"tests/integration/test_uart_mock_ld2420.py": 45.27,
"tests/integration/test_uart_mock_ld2450.py": 27.96,
"tests/integration/test_uart_mock_modbus.py": 562.45,
"tests/integration/test_udp.py": 7.48,
"tests/integration/test_use_address_runtime.py": 17.27,
"tests/integration/test_valve_control_action.py": 18.33,
"tests/integration/test_varint_five_byte_device_id.py": 17.59,
"tests/integration/test_wait_until_mid_loop_timing.py": 16.93,
"tests/integration/test_wait_until_on_boot.py": 19.96,
"tests/integration/test_wait_until_ordering.py": 16.19,
"tests/integration/test_wait_until_reentrant_restart.py": 23.67,
"tests/integration/test_wake_loop_forces_phase_b.py": 17.58,
"tests/integration/test_water_heater_template.py": 21.96
}
@@ -0,0 +1,41 @@
"""Integration test for set_internal() called during and after setup."""
from __future__ import annotations
import pytest
from .log_utils import LineWaiter
from .types import APIClientConnectedFactory, RunCompiledFunction
@pytest.mark.asyncio
async def test_set_internal_at_boot(
yaml_config: str,
run_compiled: RunCompiledFunction,
api_client_connected: APIClientConnectedFactory,
) -> None:
"""set_internal() in on_boot changes API exposure, later calls log an error."""
waiter = LineWaiter()
async with (
run_compiled(yaml_config, line_callback=waiter.callback),
api_client_connected() as client,
):
entities, services = await client.list_entities_services()
names = {entity.name for entity in entities}
assert "Hidden At Boot" not in names
assert "Shown At Boot" in names
assert "Untouched" in names
late = next(s for s in services if s.name == "set_internal_late")
await client.execute_service(late, {})
await waiter.wait_for(
"'Untouched'",
"set_internal() after setup is undefined behavior",
timeout=5.0,
)
# Still written during the deprecation window, ignored from 2027.3.0
entities, _ = await client.list_entities_services()
assert "Untouched" not in {entity.name for entity in entities}
@@ -155,3 +155,36 @@ async def test_water_heater_template(
client.water_heater_command(test_water_heater.key, mode=WaterHeaterMode.ECO)
eco_state = await wait_for_state()
assert eco_state.mode == WaterHeaterMode.ECO
@pytest.mark.asyncio
async def test_water_heater_template_unknown_temperature(
yaml_config: str,
run_compiled: RunCompiledFunction,
api_client_connected: APIClientConnectedFactory,
) -> None:
"""Test a template water heater whose temperature lambdas stay unknown.
NAN never compares equal to itself, so a lambda that keeps returning NAN must not be
mistaken for a changed value and republish the state on every loop iteration.
"""
async with run_compiled(yaml_config), api_client_connected() as client:
state_count = 0
def on_state(state: aioesphomeapi.EntityState) -> None:
nonlocal state_count
if isinstance(state, WaterHeaterState):
state_count += 1
entities, _ = await client.list_entities_services()
water_heater_infos = [e for e in entities if isinstance(e, WaterHeaterInfo)]
assert len(water_heater_infos) == 1
client.subscribe_states(on_state)
# Let the device run for a while; only the single initial state may arrive.
await asyncio.sleep(1.0)
assert state_count <= 1, (
f"Expected at most 1 state publish, got {state_count} - "
"an unknown (NAN) temperature is republishing every loop"
)
+17 -17
View File
@@ -35,8 +35,8 @@ def _load_script():
def test_spec_key_collapses_destinations() -> None:
"""Two specs delivering one package share a directory and one key."""
mod = _load_script()
assert mod.spec_key("esphome/noise-c @ 0.1.24") == "noise-c"
assert mod.spec_key("esphome/noise-c@0.1.24") == "noise-c"
assert mod.spec_key("esphome/noise-c @ 1.0") == "noise-c"
assert mod.spec_key("esphome/noise-c@1.0") == "noise-c"
assert mod.spec_key("ESP32Async/AsyncTCP @ ^3.4.10") == mod.spec_key(
"esp32async/asynctcp @ 3.5.0"
)
@@ -54,23 +54,23 @@ def test_parse_specs_and_cli_args(tmp_path: Path) -> None:
"[env:a]\n"
"platform = fake/platform@1\n"
"lib_deps =\n"
" esphome/noise-c @ 0.1.24\n"
" esphome/noise-c @ 1.0\n"
" ${common.lib_deps}\n"
" internal_lib\n"
"[env:b]\n"
"lib_deps =\n"
" esphome/noise-c @ 0.1.24\n"
" esphome/noise-c @ 1.0\n"
)
mod = _load_script()
args = Namespace(libraries=True, platforms=True, tools=False)
libs, platforms, tools = mod.parse_specs(str(ini), args)
# exact-string duplicates collapse; distinct version pins survive
assert libs == ["esphome/noise-c @ 0.1.24"]
assert libs == ["esphome/noise-c @ 1.0"]
assert platforms == ["fake/platform@1"]
assert tools == []
assert mod.build_cli_args(libs, platforms, tools) == [
"-l",
"esphome/noise-c @ 0.1.24",
"esphome/noise-c @ 1.0",
"-p",
"fake/platform@1",
]
@@ -162,13 +162,13 @@ def test_parallel_install_behavior(tmp_path: Path) -> None:
mod.parallel_install(
cls,
[
"esphome/noise-c @ 0.1.24",
"esphome/noise-c @ 0.1.24",
"esphome/noise-c @ 1.0",
"esphome/noise-c @ 1.0",
"esphome/already @ 1.0",
"https://x/framework.tar.xz",
],
)
assert cls.calls == ["esphome/noise-c @ 0.1.24"]
assert cls.calls == ["esphome/noise-c @ 1.0"]
assert cls.lock_events == ["lock", "unlock"]
@@ -205,7 +205,7 @@ def test_parallel_install_runs_dependency_waves(tmp_path: Path) -> None:
mod = _load_script()
cls = _reset_fake(str(tmp_path))
cls.deps = {
"esphome/noise-c @ 0.1.24": [
"esphome/noise-c @ 1.0": [
{"owner": "esphome", "name": "libsodium", "version": "^1.0"},
{"name": "SPI"},
],
@@ -213,12 +213,12 @@ def test_parallel_install_runs_dependency_waves(tmp_path: Path) -> None:
{"owner": "esphome", "name": "libsodium", "version": "^1.0"},
],
}
mod.parallel_install(cls, ["esphome/noise-c @ 0.1.24", "esphome/wg @ 1.0"])
mod.parallel_install(cls, ["esphome/noise-c @ 1.0", "esphome/wg @ 1.0"])
assert len(cls.calls) == 3 # the shared dep installs exactly once
assert {mod.spec_key(c) for c in cls.calls} == {"noise-c", "wg", "libsodium"}
# Wave-1 strings carry no compatibility; the dependency wave does
compats = dict(cls.compat_calls)
assert compats["esphome/noise-c @ 0.1.24"] is None
assert compats["esphome/noise-c @ 1.0"] is None
dep_compat = next(v for k, v in cls.compat_calls if "libsodium" in k)
assert dep_compat is not None # mirrors pio's install_dependency
@@ -229,11 +229,11 @@ def test_dependency_wave_excludes_url_specs(tmp_path: Path) -> None:
mod = _load_script()
cls = _reset_fake(str(tmp_path))
cls.deps = {
"esphome/noise-c @ 0.1.24": [
"esphome/noise-c @ 1.0": [
{"name": "vendored", "version": "https://github.com/x/y.git"},
],
}
mod.parallel_install(cls, ["esphome/noise-c @ 0.1.24"])
mod.parallel_install(cls, ["esphome/noise-c @ 1.0"])
assert {mod.spec_key(c) for c in cls.calls} == {"noise-c"}
@@ -348,13 +348,13 @@ def test_warm_store_still_walks_dependencies(tmp_path: Path) -> None:
"""Already-installed top-level packages still feed the dependency
wave; a warm store can be missing a transitive dep."""
mod = _load_script()
cls = _reset_fake(str(tmp_path), installed={"esphome/noise-c @ 0.1.24"})
cls = _reset_fake(str(tmp_path), installed={"esphome/noise-c @ 1.0"})
cls.deps = {
"esphome/noise-c @ 0.1.24": [
"esphome/noise-c @ 1.0": [
{"owner": "esphome", "name": "libsodium", "version": "^1.0"},
],
}
mod.parallel_install(cls, ["esphome/noise-c @ 0.1.24"])
mod.parallel_install(cls, ["esphome/noise-c @ 1.0"])
assert [mod.spec_key(c) for c in cls.calls] == ["libsodium"]
+562
View File
@@ -0,0 +1,562 @@
"""Tests for script/setup.py."""
import importlib.util
import os
from pathlib import Path, PurePosixPath, PureWindowsPath
import runpy
import sys
from types import ModuleType
from unittest.mock import Mock, call, patch
import pytest
_SCRIPT = Path(__file__).parents[2] / "script" / "setup.py"
def _load_module() -> ModuleType:
spec = importlib.util.spec_from_file_location("script_setup", _SCRIPT)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
@pytest.fixture
def script_setup() -> ModuleType:
"""Fresh import of script/setup.py, isolated from other tests."""
return _load_module()
# --- bin_dir / venv_python / activate_hint -----------------------------------
def test_bin_dir_matches_host_layout(script_setup: ModuleType, tmp_path: Path) -> None:
"""The venv scheme resolves to Scripts on Windows and bin everywhere else."""
expected = "Scripts" if os.name == "nt" else "bin"
assert script_setup.bin_dir(tmp_path) == tmp_path / expected
# Both flavours are exercised on every host. Pure paths are used because a real
# Path refuses to change flavour: PosixPath cannot be built on Windows, and
# WindowsPath cannot be built on Unix.
def test_venv_python_posix(script_setup: ModuleType, tmp_path: Path) -> None:
with (
patch.object(
script_setup, "bin_dir", return_value=PurePosixPath("/x/venv/bin")
),
patch.object(script_setup.os, "name", "posix"),
):
result = script_setup.venv_python(tmp_path)
assert result == PurePosixPath("/x/venv/bin/python")
def test_venv_python_nt(script_setup: ModuleType, tmp_path: Path) -> None:
with (
patch.object(
script_setup, "bin_dir", return_value=PureWindowsPath(r"C:\x\venv\Scripts")
),
patch.object(script_setup.os, "name", "nt"),
):
result = script_setup.venv_python(tmp_path)
assert result == PureWindowsPath(r"C:\x\venv\Scripts\python.exe")
def test_activate_hint_posix(script_setup: ModuleType) -> None:
with (
patch.object(script_setup, "ROOT", PurePosixPath("/x")),
patch.object(
script_setup, "bin_dir", return_value=PurePosixPath("/x/venv/bin")
),
patch.object(script_setup.os, "name", "posix"),
):
hint = script_setup.activate_hint()
assert hint == "source venv/bin/activate"
def test_activate_hint_nt(script_setup: ModuleType) -> None:
with (
patch.object(script_setup, "ROOT", PureWindowsPath(r"C:\x")),
patch.object(
script_setup, "bin_dir", return_value=PureWindowsPath(r"C:\x\venv\Scripts")
),
patch.object(script_setup.os, "name", "nt"),
):
hint = script_setup.activate_hint()
# The nt branch returns str(activate) as-is, skipping the "source " prefix.
assert hint == r"venv\Scripts\activate"
# --- run -----------------------------------------------------------------
def test_run_success(script_setup: ModuleType) -> None:
with patch.object(
script_setup.subprocess, "run", return_value=Mock(returncode=0)
) as mock_run:
script_setup.run(["echo", "hi"])
mock_run.assert_called_once_with(
["echo", "hi"], cwd=script_setup.ROOT, env=None, check=False
)
def test_run_failure_raises_system_exit_with_code(
script_setup: ModuleType, capsys: pytest.CaptureFixture[str]
) -> None:
with (
patch.object(script_setup.subprocess, "run", return_value=Mock(returncode=7)),
pytest.raises(SystemExit) as excinfo,
):
script_setup.run(["false"])
assert excinfo.value.code == 7
assert "Failed with exit code 7: false" in capsys.readouterr().err
# --- git_output ------------------------------------------------------------
def test_git_output_success_strips_stdout(script_setup: ModuleType) -> None:
with patch.object(
script_setup.subprocess,
"run",
return_value=Mock(returncode=0, stdout=" /repo/.git \n"),
) as mock_run:
result = script_setup.git_output("rev-parse", "--absolute-git-dir")
assert result == "/repo/.git"
mock_run.assert_called_once_with(
["git", "rev-parse", "--absolute-git-dir"],
cwd=script_setup.ROOT,
capture_output=True,
text=True,
check=False,
)
def test_git_output_nonzero_returncode_is_empty(script_setup: ModuleType) -> None:
with patch.object(
script_setup.subprocess,
"run",
return_value=Mock(returncode=1, stdout="whatever"),
):
assert script_setup.git_output("status") == ""
def test_git_output_oserror_is_empty(script_setup: ModuleType) -> None:
with patch.object(script_setup.subprocess, "run", side_effect=OSError("no git")):
assert script_setup.git_output("status") == ""
# --- create_venv -----------------------------------------------------------
def test_create_venv_uses_uv_when_present(
script_setup: ModuleType, tmp_path: Path
) -> None:
venv = tmp_path / "venv"
with (
patch.object(script_setup.shutil, "which", return_value="/usr/bin/uv"),
patch.object(
script_setup.subprocess, "run", return_value=Mock(returncode=0)
) as mock_run,
):
script_setup.create_venv(venv)
mock_run.assert_called_once_with(
["/usr/bin/uv", "venv", "--clear", "--seed", str(venv)],
cwd=script_setup.ROOT,
env=None,
check=False,
)
def test_create_venv_falls_back_to_venv_module(
script_setup: ModuleType, tmp_path: Path
) -> None:
venv = tmp_path / "venv"
with (
patch.object(script_setup.shutil, "which", return_value=None),
patch.object(
script_setup.subprocess, "run", return_value=Mock(returncode=0)
) as mock_run,
):
script_setup.create_venv(venv)
mock_run.assert_called_once_with(
[sys.executable, "-m", "venv", "--clear", str(venv)],
cwd=script_setup.ROOT,
env=None,
check=False,
)
# --- venv_environment --------------------------------------------------------
def test_venv_environment_sets_virtual_env_and_prepends_path(
script_setup: ModuleType, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
venv = tmp_path / "venv"
monkeypatch.setenv("PYTHONHOME", "/somewhere")
monkeypatch.setenv("PATH", "/usr/bin:/bin")
env = script_setup.venv_environment(venv)
assert env["VIRTUAL_ENV"] == str(venv)
assert "PYTHONHOME" not in env
expected_prefix = str(script_setup.bin_dir(venv)) + os.pathsep
assert env["PATH"] == expected_prefix + "/usr/bin:/bin"
def test_venv_environment_path_fallback_when_unset(
script_setup: ModuleType, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
venv = tmp_path / "venv"
monkeypatch.delenv("PATH", raising=False)
env = script_setup.venv_environment(venv)
# No trailing separator: an empty PATH entry means "search the cwd".
assert env["PATH"] == str(script_setup.bin_dir(venv))
# --- find_uv -----------------------------------------------------------------
def test_find_uv_found_immediately(script_setup: ModuleType, tmp_path: Path) -> None:
venv = tmp_path / "venv"
env = {"PATH": "/usr/bin"}
with (
patch.object(script_setup.shutil, "which", return_value="/usr/bin/uv"),
patch.object(script_setup.subprocess, "run") as mock_run,
):
result = script_setup.find_uv(venv, env)
assert result == "/usr/bin/uv"
mock_run.assert_not_called()
def test_find_uv_installed_then_found(script_setup: ModuleType, tmp_path: Path) -> None:
venv = tmp_path / "venv"
env = {"PATH": "/usr/bin"}
with (
patch.object(script_setup.shutil, "which", side_effect=[None, "/usr/bin/uv"]),
patch.object(
script_setup.subprocess, "run", return_value=Mock(returncode=0)
) as mock_run,
):
result = script_setup.find_uv(venv, env)
assert result == "/usr/bin/uv"
mock_run.assert_called_once_with(
[str(script_setup.venv_python(venv)), "-m", "pip", "install", "uv"],
cwd=script_setup.ROOT,
env=env,
check=False,
)
def test_find_uv_still_missing_raises_system_exit(
script_setup: ModuleType, tmp_path: Path
) -> None:
venv = tmp_path / "venv"
env = {"PATH": "/usr/bin"}
with (
patch.object(script_setup.shutil, "which", side_effect=[None, None]),
patch.object(script_setup.subprocess, "run", return_value=Mock(returncode=0)),
pytest.raises(SystemExit, match="uv could not be installed"),
):
script_setup.find_uv(venv, env)
# --- install_dependencies -----------------------------------------------------
def test_install_dependencies_installs_setuptools_then_project(
script_setup: ModuleType, tmp_path: Path
) -> None:
venv = tmp_path / "venv"
env = {"PATH": "/usr/bin"}
with (
patch.object(script_setup.shutil, "which", return_value="/usr/bin/uv"),
patch.object(
script_setup.subprocess, "run", return_value=Mock(returncode=0)
) as mock_run,
):
script_setup.install_dependencies(venv, env)
assert mock_run.call_args_list == [
call(
["/usr/bin/uv", "pip", "install", "setuptools", "wheel"],
cwd=script_setup.ROOT,
env=env,
check=False,
),
call(
[
"/usr/bin/uv",
"pip",
"install",
"-e",
".[dev,test]",
"--config-settings",
"editable_mode=compat",
],
cwd=script_setup.ROOT,
env=env,
check=False,
),
]
# --- install_git_hooks ---------------------------------------------------------
def _fake_git_output(git_dir: str, common_dir: str):
def _run(*args: str) -> str:
if "--absolute-git-dir" in args:
return git_dir
return common_dir
return _run
def test_install_git_hooks_returns_early_when_git_dir_empty(
script_setup: ModuleType,
) -> None:
env = {"PATH": "/usr/bin"}
with (
patch.object(
script_setup, "git_output", side_effect=_fake_git_output("", "/repo/.git")
),
patch.object(script_setup.subprocess, "run") as mock_run,
):
script_setup.install_git_hooks(env)
mock_run.assert_not_called()
def test_install_git_hooks_returns_early_when_common_dir_empty(
script_setup: ModuleType,
) -> None:
env = {"PATH": "/usr/bin"}
with (
patch.object(
script_setup, "git_output", side_effect=_fake_git_output("/repo/.git", "")
),
patch.object(script_setup.subprocess, "run") as mock_run,
):
script_setup.install_git_hooks(env)
mock_run.assert_not_called()
def test_install_git_hooks_returns_early_for_worktree(
script_setup: ModuleType,
) -> None:
"""A worktree's git-dir differs from the shared common-dir."""
env = {"PATH": "/usr/bin"}
with (
patch.object(
script_setup,
"git_output",
side_effect=_fake_git_output("/repo/.git/worktrees/wt", "/repo/.git"),
),
patch.object(script_setup.subprocess, "run") as mock_run,
):
script_setup.install_git_hooks(env)
mock_run.assert_not_called()
def test_install_git_hooks_missing_prek_raises_system_exit(
script_setup: ModuleType,
) -> None:
env = {"PATH": "/usr/bin"}
with (
patch.object(
script_setup,
"git_output",
side_effect=_fake_git_output("/repo/.git", "/repo/.git"),
),
patch.object(script_setup.shutil, "which", return_value=None),
patch.object(script_setup.subprocess, "run") as mock_run,
pytest.raises(SystemExit, match="prek was not installed"),
):
script_setup.install_git_hooks(env)
mock_run.assert_not_called()
def test_install_git_hooks_happy_path_installs_hook(
script_setup: ModuleType, tmp_path: Path
) -> None:
env = {"PATH": "/usr/bin"}
common_dir = tmp_path / "repo" / ".git"
hooks_dir = common_dir / "hooks"
hooks_dir.mkdir(parents=True)
source_hook = tmp_path / "post-checkout"
source_hook.write_text("#!/bin/sh\necho post-checkout\n")
with (
patch.object(script_setup, "POST_CHECKOUT_HOOK", source_hook),
patch.object(
script_setup,
"git_output",
side_effect=_fake_git_output(str(common_dir), str(common_dir)),
),
patch.object(script_setup.shutil, "which", return_value="/usr/bin/prek"),
patch.object(
script_setup.subprocess, "run", return_value=Mock(returncode=0)
) as mock_run,
):
script_setup.install_git_hooks(env)
mock_run.assert_called_once_with(
["/usr/bin/prek", "install", "--overwrite"],
cwd=script_setup.ROOT,
env=env,
check=False,
)
installed = hooks_dir / "post-checkout"
assert installed.read_text() == source_hook.read_text()
if os.name != "nt":
# Windows has no POSIX permission bits for chmod to set.
assert (installed.stat().st_mode & 0o777) == 0o755
def test_install_git_hooks_skips_copy_when_hooks_dir_missing(
script_setup: ModuleType, tmp_path: Path
) -> None:
"""The prek install still runs when the hooks directory does not exist."""
env = {"PATH": "/usr/bin"}
common_dir = tmp_path / "repo" / ".git"
common_dir.mkdir(parents=True) # no "hooks" subdirectory created
with (
patch.object(
script_setup,
"git_output",
side_effect=_fake_git_output(str(common_dir), str(common_dir)),
),
patch.object(script_setup.shutil, "which", return_value="/usr/bin/prek"),
patch.object(
script_setup.subprocess, "run", return_value=Mock(returncode=0)
) as mock_run,
):
script_setup.install_git_hooks(env)
mock_run.assert_called_once()
assert not (common_dir / "hooks").exists()
# --- report ------------------------------------------------------------------
def test_report_active_state(
script_setup: ModuleType, capsys: pytest.CaptureFixture
) -> None:
venv = Path("/opt/esphome-venv")
script_setup.report(script_setup.VENV_ACTIVE, venv)
out = capsys.readouterr().out
assert "Dependencies installed into the active virtual environment:" in out
assert str(venv) in out
assert "is already active in this shell" in out
def test_report_reused_state(
script_setup: ModuleType, capsys: pytest.CaptureFixture
) -> None:
script_setup.report(script_setup.VENV_REUSED, script_setup.DEFAULT_VENV)
out = capsys.readouterr().out
assert "Dependencies updated in the existing ./venv" in out
def test_report_created_state(
script_setup: ModuleType, capsys: pytest.CaptureFixture
) -> None:
script_setup.report(script_setup.VENV_CREATED, script_setup.DEFAULT_VENV)
out = capsys.readouterr().out
assert "Virtual environment created at ./venv" in out
# --- main --------------------------------------------------------------------
def test_main_raises_system_exit_when_python_too_old(
script_setup: ModuleType,
) -> None:
with (
patch.object(script_setup.sys, "version_info", (3, 11, 5)),
pytest.raises(SystemExit, match="ESPHome needs Python 3.12"),
):
script_setup.main()
def test_main_uses_active_virtual_env(
script_setup: ModuleType, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
active_venv = tmp_path / "active-venv"
monkeypatch.setenv("VIRTUAL_ENV", str(active_venv))
with (
patch.object(script_setup, "ROOT", tmp_path),
patch.object(script_setup, "create_venv") as mock_create_venv,
patch.object(script_setup, "install_dependencies") as mock_install_deps,
patch.object(script_setup, "install_git_hooks") as mock_install_hooks,
patch.object(script_setup, "report") as mock_report,
):
script_setup.main()
mock_create_venv.assert_not_called()
mock_install_deps.assert_called_once()
mock_install_hooks.assert_called_once()
mock_report.assert_called_once_with(script_setup.VENV_ACTIVE, active_venv)
assert (tmp_path / ".temp").is_dir()
def test_main_reuses_existing_venv(
script_setup: ModuleType, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.delenv("VIRTUAL_ENV", raising=False)
default_venv = tmp_path / "venv"
python_path = script_setup.venv_python(default_venv)
python_path.parent.mkdir(parents=True)
python_path.touch()
with (
patch.object(script_setup, "ROOT", tmp_path),
patch.object(script_setup, "DEFAULT_VENV", default_venv),
patch.object(script_setup, "create_venv") as mock_create_venv,
patch.object(script_setup, "install_dependencies") as mock_install_deps,
patch.object(script_setup, "install_git_hooks") as mock_install_hooks,
patch.object(script_setup, "report") as mock_report,
):
script_setup.main()
mock_create_venv.assert_not_called()
mock_install_deps.assert_called_once()
mock_install_hooks.assert_called_once()
mock_report.assert_called_once_with(script_setup.VENV_REUSED, default_venv)
assert (tmp_path / ".temp").is_dir()
def test_main_creates_new_venv(
script_setup: ModuleType, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.delenv("VIRTUAL_ENV", raising=False)
default_venv = tmp_path / "venv" # does not exist yet
with (
patch.object(script_setup, "ROOT", tmp_path),
patch.object(script_setup, "DEFAULT_VENV", default_venv),
patch.object(script_setup, "create_venv") as mock_create_venv,
patch.object(script_setup, "install_dependencies") as mock_install_deps,
patch.object(script_setup, "install_git_hooks") as mock_install_hooks,
patch.object(script_setup, "report") as mock_report,
):
script_setup.main()
mock_create_venv.assert_called_once_with(default_venv)
mock_install_deps.assert_called_once()
mock_install_hooks.assert_called_once()
mock_report.assert_called_once_with(script_setup.VENV_CREATED, default_venv)
assert (tmp_path / ".temp").is_dir()
def test_run_as_script_calls_main(tmp_path: Path) -> None:
"""The __main__ guard runs the whole flow, with every side effect stubbed."""
completed = Mock(returncode=0, stdout="")
with (
patch("subprocess.run", return_value=completed) as mock_run,
patch("shutil.which", return_value="/usr/bin/uv"),
patch("pathlib.Path.mkdir") as mock_mkdir,
patch.dict(os.environ, {"VIRTUAL_ENV": str(tmp_path / "env")}),
):
runpy.run_path(str(_SCRIPT), run_name="__main__")
# The dependency install ran, and git reported no hooks directory to touch.
assert mock_run.called
mock_mkdir.assert_called_once_with(exist_ok=True)
+148 -1
View File
@@ -1,4 +1,5 @@
import importlib
import io
import json
import logging
from pathlib import Path
@@ -20,6 +21,7 @@ from esphome.components.esp32 import (
VARIANT_ESP32S2,
VARIANT_ESP32S3,
)
from esphome.components.substitutions import do_substitution_pass
from esphome.config_validation import Invalid
from esphome.const import (
CONF_DAY,
@@ -65,7 +67,13 @@ from esphome.core import (
)
from esphome.schema_extractors import SCHEMA_EXTRACT
from esphome.util import Registry
from esphome.yaml_util import ESPHomeDataBase, SensitiveStr, make_data_base
from esphome.yaml_util import (
ESPHomeDataBase,
SensitiveStr,
load_yaml,
make_data_base,
parse_yaml,
)
def test_check_not_templatable__invalid():
@@ -1394,6 +1402,35 @@ def test_entity_metadata_visibility_hints() -> None:
assert web["web_server"].visibility is advanced
def test_with_visibility_remarks_keys() -> None:
"""``with_visibility`` re-marks the named keys, preserving each field's
default and validator, without touching the other keys or the input schema.
"""
base = cv.Schema(
{
cv.Optional("a", default=7): cv.int_,
cv.Optional("b", visibility=cv.Visibility.ADVANCED): cv.string,
}
)
promoted = cv.with_visibility(base, cv.Visibility.UI, "a")
pm = {str(k): k for k in promoted.schema}
assert pm["a"].visibility is cv.Visibility.UI # re-marked
assert pm["a"].default() == 7 # default preserved
assert pm["b"].visibility is cv.Visibility.ADVANCED # sibling untouched
assert promoted({}) == {"a": 7} # validator/default still applied
# The input schema is left untouched (no shared-marker mutation).
assert {str(k): k for k in base.schema}["a"].visibility is None
def test_with_visibility_unknown_key_raises() -> None:
"""A key not present in the schema is a typo — fail at build time."""
base = cv.Schema({cv.Optional("a"): cv.int_})
with pytest.raises(ValueError, match="not in schema"):
cv.with_visibility(base, cv.Visibility.UI, "nope")
def _wrap_str(value: str) -> ESPHomeDataBase:
"""Wrap a raw string as an ESPHomeDataBase, mimicking a YAML-loaded value."""
return make_data_base(value)
@@ -3145,6 +3182,116 @@ def test_file__existing_relative_path(setup_core: Path) -> None:
assert cv.file_("partitions.csv") == setup_core / "partitions.csv"
def _package_value(setup_core: Path, path: str = "assets/ui.js") -> tuple[Path, str]:
"""Write a package file next to an ``assets/`` dir; return the dir and its loaded *path* value."""
package_dir = setup_core / ".esphome" / "packages" / "abc123" / "vendor"
(package_dir / "assets").mkdir(parents=True)
(package_dir / "assets" / "ui.js").write_text("js\n")
(package_dir / "device.yaml").write_text(f"path: {path}\n")
return package_dir, load_yaml(package_dir / "device.yaml")["path"]
def test_file__resolves_relative_to_the_declaring_document(setup_core: Path) -> None:
"""A package's own asset path resolves against the package file when the config dir lacks it."""
package_dir, value = _package_value(setup_core)
assert cv.file_(value) == package_dir / "assets" / "ui.js"
def test_file__resolves_a_substituted_path_against_the_use_site(
setup_core: Path,
) -> None:
package_dir, _ = _package_value(setup_core)
(package_dir / "device.yaml").write_text(
"substitutions:\n ui: assets/ui.js\npath: ${ui}\n"
)
config = do_substitution_pass(load_yaml(package_dir / "device.yaml"))
assert cv.file_(config["path"]) == package_dir / "assets" / "ui.js"
def test_file__result_is_absolute_for_a_relative_document(
setup_core: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""A document loaded by a cwd-relative path still yields an absolute result."""
package_dir, _ = _package_value(setup_core)
monkeypatch.chdir(setup_core)
value = load_yaml(Path(".esphome/packages/abc123/vendor/device.yaml"))["path"]
result = cv.file_(value)
assert result.is_absolute()
assert result == package_dir / "assets" / "ui.js"
def test_file__config_dir_entry_of_the_wrong_kind_does_not_shadow_the_package(
setup_core: Path,
) -> None:
package_dir, value = _package_value(setup_core)
(setup_core / "assets" / "ui.js").mkdir(parents=True)
assert cv.file_(value) == package_dir / "assets" / "ui.js"
def test_file__miss_names_the_declaring_document(setup_core: Path) -> None:
package_dir, value = _package_value(setup_core, "assets/other.js")
with pytest.raises(Invalid, match="Could not find file") as excinfo:
cv.file_(value)
assert f"Also looked next to {package_dir / 'device.yaml'}" in str(excinfo.value)
def test_file__document_spelled_through_dotdot_in_the_config_dir_adds_no_hint(
setup_core: Path,
) -> None:
(setup_core / "sub").mkdir()
(setup_core / "device.yaml").write_text("path: assets/other.js\n")
value = load_yaml(setup_core / "sub" / ".." / "device.yaml")["path"]
with pytest.raises(Invalid) as excinfo:
cv.file_(value)
assert "Also looked" not in str(excinfo.value)
def test_file__wrong_kind_beside_the_document_is_reported(setup_core: Path) -> None:
package_dir, value = _package_value(setup_core, "assets")
with pytest.raises(Invalid, match="is not a file") as excinfo:
cv.file_(value)
assert str(package_dir / "assets") in str(excinfo.value)
def test_file__config_dir_wins_over_the_declaring_document(setup_core: Path) -> None:
_, value = _package_value(setup_core)
(setup_core / "assets").mkdir()
(setup_core / "assets" / "ui.js").write_text("local\n")
assert cv.file_(value) == setup_core / "assets" / "ui.js"
def test_file__declared_in_an_in_memory_document_is_not_resolved(
setup_core: Path,
) -> None:
"""A value whose source document isn't on disk falls through to the config-dir error."""
value = parse_yaml(Path("<unicode string>"), io.StringIO("path: assets/ui.js\n"))[
"path"
]
with pytest.raises(Invalid, match="Could not find file"):
cv.file_(value)
def test_directory_resolves_relative_to_the_declaring_document(
setup_core: Path,
) -> None:
package_dir, value = _package_value(setup_core, "assets")
assert cv.directory(value) == package_dir / "assets"
def test_file__missing_raises(setup_core: Path) -> None:
with pytest.raises(Invalid, match="Could not find file"):
cv.file_("partitions.csv")
+45
View File
@@ -1,5 +1,7 @@
"""Tests for the coroutine module."""
import contextvars
import pytest
from esphome.coroutine import CoroPriority, FakeEventLoop, coroutine_with_priority
@@ -217,3 +219,46 @@ def test_custom_priority_between_enum_values() -> None:
# Check execution order
assert execution_order == ["core", "custom", "diagnostics"]
def test_context_isolated_between_interleaved_tasks() -> None:
"""Test that a contextvar set in one task does not leak into another task that the scheduler interleaves with it."""
my_var: contextvars.ContextVar[str] = contextvars.ContextVar("my_var")
seen: dict[str, str] = {}
def task_a():
my_var.set("a")
yield # suspend so task_b can run before task_a resumes
seen["a"] = my_var.get()
def task_b():
my_var.set("b")
yield
seen["b"] = my_var.get()
loop = FakeEventLoop()
loop.add_job(task_a)
loop.add_job(task_b)
loop.flush_tasks()
assert seen == {"a": "a", "b": "b"}
def test_context_inherits_ambient_value_at_schedule_time() -> None:
"""Test that a job sees whatever contextvar value was set before it was scheduled."""
my_var: contextvars.ContextVar[str] = contextvars.ContextVar("my_var")
token = my_var.set("ambient")
seen: dict[str, str] = {}
def task():
seen["value"] = my_var.get()
yield
try:
loop = FakeEventLoop()
loop.add_job(task)
loop.flush_tasks()
finally:
my_var.reset(token)
assert seen == {"value": "ambient"}
+79
View File
@@ -85,6 +85,15 @@ class TestCallExpression:
assert actual == 'my_function<int32_t, float>(1, "2", false)'
class TestStaticCastExpression:
def test_str(self):
target = cg.StaticCastExpression(ct.bool_, 42)
actual = str(target)
assert actual == "static_cast<bool>(42)"
class TestStructInitializer:
def test_str(self):
target = cg.StructInitializer(
@@ -229,6 +238,76 @@ class TestLambdaExpression:
)
class TestCallLambda:
"""Tests for the call_lambda() function."""
def test_call_lambda__return_expression_casts_to_return_type(self):
"""A lambda body that is just a return statement reduces to the
expression, cast to the lambda's return type."""
lamb = cg.LambdaExpression(("return foo + 1;",), (), "", ct.bool_)
result = cg.call_lambda(lamb)
assert isinstance(result, cg.StaticCastExpression)
assert str(result) == "static_cast<bool>(foo + 1)"
def test_call_lambda__return_expression_with_class_return_type_no_cast(self):
"""A class return type is not cast, since static_cast doesn't apply
to arbitrary class types."""
mock_class = cg.MockObjClass("foo::Bar", parents=())
lamb = cg.LambdaExpression(("return get_bar();",), (), "", mock_class)
result = cg.call_lambda(lamb)
assert isinstance(result, cg.RawExpression)
assert str(result) == "get_bar()"
def test_call_lambda__no_return_with_parameters_calls_with_names(self):
"""A multi-statement lambda with parameters is called with the
parameter names as arguments."""
lamb = cg.LambdaExpression(
("do_something(x, y);",), ((int, "x"), (float, "y")), "=", ct.bool_
)
result = cg.call_lambda(lamb)
assert isinstance(result, cg.CallExpression)
assert str(result) == (
"[=](int32_t x, float y) -> bool {\n do_something(x, y);\n}(x, y)"
)
def test_call_lambda__no_return_type_raises(self):
"""Calling a lambda with no declared return type is a developer
error: call_lambda is only for value-returning lambdas."""
lamb = cg.LambdaExpression(("do_something();",), (), "=")
with pytest.raises(AssertionError):
cg.call_lambda(lamb)
def test_call_lambda__identifier_starting_with_return_is_not_a_return_statement(
self,
):
"""A body that merely starts with the substring "return" (e.g. a call
to a function named returnValue()) must not be mistaken for a return
statement -- the match requires a word boundary after "return"."""
lamb = cg.LambdaExpression(("returnValue();",), (), "=", ct.bool_)
result = cg.call_lambda(lamb)
assert isinstance(result, cg.CallExpression)
assert str(result) == "[=]() -> bool {\n returnValue();\n}()"
def test_call_lambda__no_return_no_parameters_calls_with_no_args(self):
"""A multi-statement lambda without parameters is called with no
arguments."""
lamb = cg.LambdaExpression(("do_something();",), (), "", ct.bool_)
result = cg.call_lambda(lamb)
assert isinstance(result, cg.CallExpression)
assert str(result) == "[]() -> bool {\n do_something();\n}()"
class TestLiterals:
@pytest.mark.parametrize(
"target, expected",
+25
View File
@@ -187,6 +187,31 @@ def test_slot_counter_emits_requested_count() -> None:
assert _define_value("TEST_SLOT_COUNT") == "2"
def test_slot_counter_keyed_emits_largest_count() -> None:
"""Keyed requests size storage every key declares at the same capacity:
the define is the busiest key's count, not the total over all keys."""
request = ch.slot_counter("TEST_SLOT_COUNT_KEYED")
request("rx_a")
request("rx_a")
request("rx_a")
request("rx_b")
assert ch.get_slot_count("TEST_SLOT_COUNT_KEYED") == 3
ch.CORE.flush_tasks()
assert _define_value("TEST_SLOT_COUNT_KEYED") == "3"
def test_slot_counter_rejects_mixed_keyed_and_unkeyed_requests() -> None:
"""A keyed and an unkeyed request for one define cannot be sized together."""
request = ch.slot_counter("TEST_SLOT_COUNT_MIXED")
request("rx_a")
with pytest.raises(ValueError, match="TEST_SLOT_COUNT_MIXED"):
request()
unkeyed = ch.slot_counter("TEST_SLOT_COUNT_MIXED_2")
unkeyed()
with pytest.raises(ValueError, match="TEST_SLOT_COUNT_MIXED_2"):
unkeyed("rx_a")
def test_slot_counter_without_requests_emits_nothing() -> None:
"""No requests, no job, no define — the guarded storage compiles out."""
ch.slot_counter("TEST_SLOT_COUNT_UNUSED")
+3
View File
@@ -416,6 +416,9 @@ def test_perform_ota_no_auth(
"Update took 14.00 seconds (prepare 2.00, upload 5.00, commit 7.00)"
in caplog.text
)
# The data phase timeout must outlast the device's 105 s data timeout
mock_socket.settimeout.assert_any_call(espota2.DATA_PHASE_TIMEOUT)
assert espota2.DATA_PHASE_TIMEOUT > 105.0
@pytest.mark.usefixtures("mock_time")
+2 -2
View File
@@ -1663,7 +1663,7 @@ def test_preinstall_runs_dependency_waves(tmp_path: Path) -> None:
{"name": "SPI"},
]
m.dependency_to_spec.side_effect = lambda dep: _FakeSpec(name=dep["name"])
pf._preinstall(m, [("noise-c@0.1.24", _FakeSpec(name="noise-c"))])
pf._preinstall(m, [("noise-c@1.0", _FakeSpec(name="noise-c"))])
assert installed == ["noise-c", "libsodium"] # dep deduped, SPI left out
# The dep wave carries its compatibility so _install searches qualified
dep_call = m._install.call_args_list[-1]
@@ -1683,7 +1683,7 @@ def test_preinstall_dependency_wave_skips_seen_names(tmp_path: Path) -> None:
m._install.side_effect = lambda spec, skip_dependencies, compatibility=None: (
installed.append(getattr(spec, "name", str(spec)))
)
pf._preinstall(m, [("noise-c@0.1.24", _FakeSpec(name="noise-c"))])
pf._preinstall(m, [("noise-c@1.0", _FakeSpec(name="noise-c"))])
assert installed == ["noise-c"]