mirror of
https://github.com/esphome/esphome.git
synced 2026-10-07 03:16:37 +00:00
Merge remote-tracking branch 'upstream/dev' into neutral-ble-client
This commit is contained in:
@@ -0,0 +1,12 @@
|
||||
---
|
||||
esphome:
|
||||
name: test
|
||||
|
||||
esp32:
|
||||
board: esp32dev
|
||||
|
||||
wifi:
|
||||
ssid: test
|
||||
password: testtest
|
||||
|
||||
api:
|
||||
@@ -0,0 +1,15 @@
|
||||
---
|
||||
esphome:
|
||||
name: test
|
||||
|
||||
esp32:
|
||||
board: esp32dev
|
||||
|
||||
wifi:
|
||||
ssid: test
|
||||
password: testtest
|
||||
|
||||
api:
|
||||
port: 6054
|
||||
reboot_timeout: 0s
|
||||
batch_delay: 0ms
|
||||
@@ -0,0 +1,15 @@
|
||||
---
|
||||
esphome:
|
||||
name: test
|
||||
|
||||
esp32:
|
||||
board: esp32dev
|
||||
|
||||
wifi:
|
||||
ssid: test
|
||||
password: testtest
|
||||
|
||||
api:
|
||||
port: 6053
|
||||
reboot_timeout: 15min
|
||||
batch_delay: 100ms
|
||||
@@ -0,0 +1,35 @@
|
||||
"""Tests that the api component only emits setters for non default values."""
|
||||
|
||||
from collections.abc import Callable
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
@pytest.mark.parametrize("config_file", ["bare.yaml", "defaults.yaml"])
|
||||
def test_default_values_are_not_emitted(
|
||||
generate_main: Callable[[str | Path], str],
|
||||
component_config_path: Callable[[str], Path],
|
||||
config_file: str,
|
||||
) -> None:
|
||||
"""Port 6053, a 15 min reboot timeout and 100 ms batch delay are C++ initializers.
|
||||
|
||||
Both the schema defaults and the same values written explicitly take the skip path.
|
||||
"""
|
||||
main_cpp = generate_main(component_config_path(config_file))
|
||||
|
||||
assert "api_apiserver_id->set_port(" not in main_cpp
|
||||
assert "api_apiserver_id->set_reboot_timeout(" not in main_cpp
|
||||
assert "api_apiserver_id->set_batch_delay(" not in main_cpp
|
||||
|
||||
|
||||
def test_custom_values_are_emitted(
|
||||
generate_main: Callable[[str | Path], str],
|
||||
component_config_path: Callable[[str], Path],
|
||||
) -> None:
|
||||
"""Non default values still reach the C++ setters."""
|
||||
main_cpp = generate_main(component_config_path("custom.yaml"))
|
||||
|
||||
assert "api_apiserver_id->set_port(6054);" in main_cpp
|
||||
assert "api_apiserver_id->set_reboot_timeout(0);" in main_cpp
|
||||
assert "api_apiserver_id->set_batch_delay(0);" in main_cpp
|
||||
@@ -0,0 +1,12 @@
|
||||
esphome:
|
||||
name: bk-power-save
|
||||
|
||||
bk72xx:
|
||||
board: cb2s
|
||||
|
||||
wifi:
|
||||
ssid: test
|
||||
password: testtest
|
||||
power_save_mode: high
|
||||
|
||||
bk72xx_ble:
|
||||
@@ -0,0 +1,20 @@
|
||||
"""bk72xx_ble keeps WiFi power save off: the Beken SDK's MCU sleep does not
|
||||
wake up once the station is stopped while the BLE controller runs."""
|
||||
|
||||
from collections.abc import Callable
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
def test_power_save_mode_is_not_applied_with_ble(
|
||||
generate_main: Callable[[str | Path], str],
|
||||
component_config_path: Callable[[str], Path],
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
) -> None:
|
||||
main_cpp = generate_main(component_config_path("test_power_save.yaml"))
|
||||
|
||||
assert "bk72xx_ble::BK72xxBLE" in main_cpp
|
||||
assert "set_power_save_mode(" not in main_cpp
|
||||
assert "power_save_mode HIGH is not applied" in caplog.text
|
||||
assert "issues/18592" in caplog.text
|
||||
@@ -9,5 +9,5 @@ wifi:
|
||||
password: password1
|
||||
|
||||
# esp32_ble_server is only auto-loaded here, so it has no services of its own.
|
||||
esp32_improv:
|
||||
improv_ble:
|
||||
authorizer: none
|
||||
|
||||
@@ -55,7 +55,7 @@ def test_uuid_is_matches_descriptor_short_strings(uuid16) -> None:
|
||||
@pytest.mark.parametrize(
|
||||
("config_file", "required"),
|
||||
[
|
||||
# Auto-loaded by esp32_improv only: nothing to find until Improv asks for it
|
||||
# Auto-loaded by improv_ble only: nothing to find until Improv asks for it
|
||||
("improv_only.yaml", False),
|
||||
# The configuration defines a service clients are meant to connect to
|
||||
("own_service.yaml", True),
|
||||
|
||||
@@ -0,0 +1,31 @@
|
||||
esphome:
|
||||
name: improv-ble-automations
|
||||
esp32:
|
||||
variant: esp32
|
||||
framework:
|
||||
type: esp-idf
|
||||
logger:
|
||||
wifi:
|
||||
ssid: MySSID
|
||||
password: password1
|
||||
binary_sensor:
|
||||
- platform: gpio
|
||||
pin: 0
|
||||
id: io0_button
|
||||
output:
|
||||
- platform: gpio
|
||||
pin: 2
|
||||
id: built_in_led
|
||||
improv_ble:
|
||||
authorizer: io0_button
|
||||
status_indicator: built_in_led
|
||||
on_provisioned:
|
||||
- logger.log: provisioned
|
||||
on_provisioning:
|
||||
- logger.log: provisioning
|
||||
on_start:
|
||||
- logger.log: start
|
||||
on_state:
|
||||
- logger.log: state
|
||||
on_stop:
|
||||
- logger.log: stop
|
||||
@@ -0,0 +1,12 @@
|
||||
esphome:
|
||||
name: improv-ble-esp32
|
||||
esp32:
|
||||
variant: esp32
|
||||
framework:
|
||||
type: esp-idf
|
||||
logger:
|
||||
wifi:
|
||||
ssid: MySSID
|
||||
password: password1
|
||||
improv_ble:
|
||||
authorizer: none
|
||||
@@ -0,0 +1,10 @@
|
||||
esphome:
|
||||
name: improv-ble-esp8266
|
||||
esp8266:
|
||||
board: nodemcuv2
|
||||
logger:
|
||||
wifi:
|
||||
ssid: MySSID
|
||||
password: password1
|
||||
improv_ble:
|
||||
authorizer: none
|
||||
@@ -0,0 +1,12 @@
|
||||
esphome:
|
||||
name: improv-ble-legacy-key
|
||||
esp32:
|
||||
variant: esp32
|
||||
framework:
|
||||
type: esp-idf
|
||||
logger:
|
||||
wifi:
|
||||
ssid: MySSID
|
||||
password: password1
|
||||
esp32_improv:
|
||||
authorizer: none
|
||||
@@ -0,0 +1,59 @@
|
||||
"""improv_ble is platform neutral; only its BLE server backends are not.
|
||||
|
||||
Covers the platform gate (BLE_SERVER_BACKENDS) and the esp32_improv alias that
|
||||
keeps pre-rename configurations working.
|
||||
"""
|
||||
|
||||
from collections.abc import Callable
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from esphome.config import read_config
|
||||
from esphome.core import CORE
|
||||
|
||||
|
||||
def test_esp32_generates_component(
|
||||
generate_main: Callable[[str | Path], str],
|
||||
component_config_path: Callable[[str], Path],
|
||||
) -> None:
|
||||
main_cpp = generate_main(component_config_path("esp32.yaml"))
|
||||
assert "improv_ble::ImprovBLEComponent" in main_cpp
|
||||
|
||||
|
||||
def test_legacy_key_routes_to_improv_ble(
|
||||
generate_main: Callable[[str | Path], str],
|
||||
component_config_path: Callable[[str], Path],
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
) -> None:
|
||||
main_cpp = generate_main(component_config_path("legacy_key.yaml"))
|
||||
assert "improv_ble::ImprovBLEComponent" in main_cpp
|
||||
assert "'esp32_improv:' top-level key is deprecated" in caplog.text
|
||||
|
||||
|
||||
def test_platform_without_ble_server_rejected(
|
||||
component_config_path: Callable[[str], Path],
|
||||
capsys: pytest.CaptureFixture[str],
|
||||
) -> None:
|
||||
# AUTO_LOAD finds no backend for esp8266 and pulls in improv_base only, so
|
||||
# the platform gate in CONFIG_SCHEMA is what has to reject the config.
|
||||
CORE.config_path = component_config_path("esp8266.yaml")
|
||||
assert read_config({}) is None
|
||||
assert "only available on" in capsys.readouterr().out
|
||||
|
||||
|
||||
def test_automations_emit_renamed_triggers(
|
||||
generate_main: Callable[[str | Path], str],
|
||||
component_config_path: Callable[[str], Path],
|
||||
) -> None:
|
||||
main_cpp = generate_main(component_config_path("automations.yaml"))
|
||||
for trigger in (
|
||||
"ImprovBLEProvisionedTrigger",
|
||||
"ImprovBLEProvisioningTrigger",
|
||||
"ImprovBLEStartTrigger",
|
||||
"ImprovBLEStateTrigger",
|
||||
"ImprovBLEStoppedTrigger",
|
||||
):
|
||||
assert f"improv_ble::{trigger}" in main_cpp
|
||||
assert "set_authorizer" in main_cpp
|
||||
assert "set_status_indicator" in main_cpp
|
||||
@@ -0,0 +1,70 @@
|
||||
"""request_service_enable_disable() only opts in on platforms whose mDNS stack
|
||||
can add and remove services after setup, and tells the caller so."""
|
||||
|
||||
import pytest
|
||||
|
||||
from esphome.components import mdns
|
||||
from esphome.const import CONF_DISABLED, PlatformFramework
|
||||
from esphome.core import CORE
|
||||
from tests.component_tests.types import SetCoreConfigCallable
|
||||
|
||||
DEFINE = "USE_MDNS_SUPPORTS_ENABLE_DISABLE"
|
||||
|
||||
|
||||
def _defines() -> set[str]:
|
||||
return {define.name for define in CORE.defines}
|
||||
|
||||
|
||||
def _set_config(
|
||||
set_core_config: SetCoreConfigCallable,
|
||||
platform_framework: PlatformFramework,
|
||||
config: dict,
|
||||
) -> None:
|
||||
set_core_config(platform_framework)
|
||||
CORE.config = config
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"platform_framework",
|
||||
[PlatformFramework.ESP32_IDF, PlatformFramework.ESP32_ARDUINO],
|
||||
)
|
||||
def test_esp32_adds_define_and_keeps_services_stored(
|
||||
set_core_config: SetCoreConfigCallable, platform_framework: PlatformFramework
|
||||
) -> None:
|
||||
_set_config(set_core_config, platform_framework, {"mdns": {CONF_DISABLED: False}})
|
||||
|
||||
assert mdns.request_service_enable_disable() is True
|
||||
# Disabled services must stay stored so they can be re-registered later.
|
||||
assert {DEFINE, "USE_MDNS_STORE_SERVICES"} <= _defines()
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"platform_framework",
|
||||
[PlatformFramework.ESP8266_ARDUINO, PlatformFramework.RP2_ARDUINO],
|
||||
)
|
||||
def test_other_platforms_return_false(
|
||||
set_core_config: SetCoreConfigCallable, platform_framework: PlatformFramework
|
||||
) -> None:
|
||||
_set_config(set_core_config, platform_framework, {"mdns": {CONF_DISABLED: False}})
|
||||
|
||||
assert mdns.request_service_enable_disable() is False
|
||||
assert DEFINE not in _defines()
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"config",
|
||||
[
|
||||
pytest.param({}, id="no_mdns"),
|
||||
pytest.param({"mdns": {CONF_DISABLED: True}}, id="mdns_disabled"),
|
||||
pytest.param(
|
||||
{"mdns": {CONF_DISABLED: False}, "openthread": {}}, id="openthread"
|
||||
),
|
||||
],
|
||||
)
|
||||
def test_esp32_returns_false_when_services_cannot_be_toggled(
|
||||
set_core_config: SetCoreConfigCallable, config: dict
|
||||
) -> None:
|
||||
_set_config(set_core_config, PlatformFramework.ESP32_IDF, config)
|
||||
|
||||
assert mdns.request_service_enable_disable() is False
|
||||
assert DEFINE not in _defines()
|
||||
@@ -33,7 +33,6 @@ def test_server_schema_rejects_address_zero() -> None:
|
||||
|
||||
|
||||
def test_client_schema_still_accepts_address_zero() -> None:
|
||||
# Not rejected for clients today, but not supported either: a client broadcast gets no reply and
|
||||
# stalls the hub for the full send-wait.
|
||||
# A client may address 0: writes are broadcast, and reads are allowed with allow_broadcast_read.
|
||||
schema = modbus.modbus_device_schema(0x01)
|
||||
assert schema({CONF_MODBUS_ID: "hub", CONF_ADDRESS: 0})[CONF_ADDRESS] == 0
|
||||
|
||||
@@ -7,7 +7,7 @@ guard is a safety property: these tests pin it to every handler slot.
|
||||
import pytest
|
||||
|
||||
from esphome import config_validation as cv
|
||||
from esphome.components import modbus_client
|
||||
from esphome.components import modbus, modbus_client
|
||||
from esphome.components.modbus_client import (
|
||||
CONF_ON_NO_RESPONSE,
|
||||
CONF_ON_NOT_SENT,
|
||||
@@ -126,7 +126,7 @@ def test_on_no_response_retry_lambda_accepted() -> None:
|
||||
|
||||
def test_continuous_on_write_pdu_rejected() -> None:
|
||||
"""A literal write-code PDU with continuous: true is rejected at config time (reads only)."""
|
||||
with pytest.raises(cv.Invalid, match="does not apply to a write PDU"):
|
||||
with pytest.raises(cv.Invalid, match="does not apply to function code"):
|
||||
MODBUS_CLIENT_SEND_SCHEMA(
|
||||
{
|
||||
CONF_ADDRESS: 0x01,
|
||||
@@ -185,3 +185,145 @@ def test_multi_conf_no_default_is_set() -> None:
|
||||
"""
|
||||
assert modbus_client.MULTI_CONF is True
|
||||
assert modbus_client.MULTI_CONF_NO_DEFAULT is True
|
||||
|
||||
|
||||
@pytest.mark.parametrize("key", [CONF_CONTINUOUS, modbus.CONF_ALLOW_BROADCAST_READ])
|
||||
def test_send_rejects_read_option_on_static_write_pdu(key: str) -> None:
|
||||
# A read option set true on a static write PDU is refused at validation, naming the key.
|
||||
config = {
|
||||
CONF_ADDRESS: 1,
|
||||
CONF_PDU: [0x06, 0x00, 0x10, 0x00, 0x01],
|
||||
key: True,
|
||||
}
|
||||
with pytest.raises(
|
||||
cv.Invalid, match=f"'{key}: true' does not apply to function code"
|
||||
):
|
||||
MODBUS_CLIENT_SEND_SCHEMA(config)
|
||||
|
||||
|
||||
def test_send_accepts_allow_broadcast_read_on_read_pdu() -> None:
|
||||
# allow_broadcast_read defaults to False and is accepted on a read PDU to address 0.
|
||||
config = MODBUS_CLIENT_SEND_SCHEMA(
|
||||
{CONF_ADDRESS: 0, CONF_PDU: [0x03, 0x00, 0x10, 0x00, 0x02]}
|
||||
)
|
||||
assert config[modbus.CONF_ALLOW_BROADCAST_READ] is False
|
||||
config = MODBUS_CLIENT_SEND_SCHEMA(
|
||||
{
|
||||
CONF_ADDRESS: 0,
|
||||
CONF_PDU: [0x03, 0x00, 0x10, 0x00, 0x02],
|
||||
modbus.CONF_ALLOW_BROADCAST_READ: True,
|
||||
}
|
||||
)
|
||||
assert config[modbus.CONF_ALLOW_BROADCAST_READ] is True
|
||||
|
||||
|
||||
def test_send_rejects_write_option_on_static_read_pdu() -> None:
|
||||
# The write-side option is refused on a static read PDU, the mirror of the read-option check.
|
||||
key = modbus.CONF_EXPECT_BROADCAST_WRITE_RESPONSE
|
||||
with pytest.raises(
|
||||
cv.Invalid, match=f"'{key}: true' does not apply to function code"
|
||||
):
|
||||
MODBUS_CLIENT_SEND_SCHEMA(
|
||||
{CONF_ADDRESS: 0, CONF_PDU: [0x03, 0x00, 0x10, 0x00, 0x02], key: True}
|
||||
)
|
||||
|
||||
|
||||
def test_send_accepts_write_option_on_static_write_pdu() -> None:
|
||||
config = MODBUS_CLIENT_SEND_SCHEMA(
|
||||
{
|
||||
CONF_ADDRESS: 0,
|
||||
CONF_PDU: [0x06, 0x00, 0x10, 0x00, 0x01],
|
||||
modbus.CONF_EXPECT_BROADCAST_WRITE_RESPONSE: True,
|
||||
}
|
||||
)
|
||||
assert config[modbus.CONF_EXPECT_BROADCAST_WRITE_RESPONSE] is True
|
||||
|
||||
|
||||
def test_write_actions_offer_write_option_only() -> None:
|
||||
# Every write action takes expect_broadcast_write_response and none of the read options.
|
||||
from esphome.components.modbus_client import (
|
||||
_WRITE_MULTIPLE_COILS_SCHEMA,
|
||||
_WRITE_MULTIPLE_REGISTERS_SCHEMA,
|
||||
_WRITE_SINGLE_COIL_SCHEMA,
|
||||
_WRITE_SINGLE_REGISTER_SCHEMA,
|
||||
CONF_START_ADDRESS,
|
||||
CONF_VALUE,
|
||||
CONF_VALUES,
|
||||
)
|
||||
|
||||
write_key = modbus.CONF_EXPECT_BROADCAST_WRITE_RESPONSE
|
||||
base = {CONF_ADDRESS: 0, CONF_START_ADDRESS: 0x10, write_key: True}
|
||||
for schema, extra in (
|
||||
(_WRITE_SINGLE_REGISTER_SCHEMA, {CONF_VALUE: 1}),
|
||||
(_WRITE_SINGLE_COIL_SCHEMA, {CONF_VALUE: True}),
|
||||
(_WRITE_MULTIPLE_REGISTERS_SCHEMA, {CONF_VALUES: [1, 2]}),
|
||||
(_WRITE_MULTIPLE_COILS_SCHEMA, {CONF_VALUES: [True, False]}),
|
||||
):
|
||||
config = schema({**base, **extra})
|
||||
assert config[write_key] is True
|
||||
assert modbus.CONF_ALLOW_BROADCAST_READ not in config
|
||||
with pytest.raises(cv.Invalid):
|
||||
schema({**base, **extra, modbus.CONF_ALLOW_BROADCAST_READ: True})
|
||||
|
||||
|
||||
def test_send_options_follow_the_hub_classification() -> None:
|
||||
# A vendor code is broadcastable, so it takes the write-side flag and refuses the read-side one;
|
||||
# 0x17 is a read for broadcast purposes, so the reverse holds.
|
||||
write_key = modbus.CONF_EXPECT_BROADCAST_WRITE_RESPONSE
|
||||
read_key = modbus.CONF_ALLOW_BROADCAST_READ
|
||||
assert MODBUS_CLIENT_SEND_SCHEMA(
|
||||
{CONF_ADDRESS: 0, CONF_PDU: [0x41, 0x01], write_key: True}
|
||||
)[write_key]
|
||||
with pytest.raises(cv.Invalid, match=f"'{read_key}: true' does not apply"):
|
||||
MODBUS_CLIENT_SEND_SCHEMA(
|
||||
{CONF_ADDRESS: 0, CONF_PDU: [0x41, 0x01], read_key: True}
|
||||
)
|
||||
pdu_0x17 = [0x17, 0x00, 0x10, 0x00, 0x01, 0x00, 0x20, 0x00, 0x01, 0x02, 0x00, 0x01]
|
||||
assert MODBUS_CLIENT_SEND_SCHEMA(
|
||||
{CONF_ADDRESS: 0, CONF_PDU: pdu_0x17, read_key: True}
|
||||
)[read_key]
|
||||
with pytest.raises(cv.Invalid, match=f"'{write_key}: true' does not apply"):
|
||||
MODBUS_CLIENT_SEND_SCHEMA(
|
||||
{CONF_ADDRESS: 0, CONF_PDU: pdu_0x17, write_key: True}
|
||||
)
|
||||
|
||||
|
||||
def test_read_write_multiple_offers_allow_broadcast_read_only() -> None:
|
||||
from esphome.components.modbus_client import (
|
||||
_READ_WRITE_MULTIPLE_REGISTERS_SCHEMA,
|
||||
CONF_READ_ADDRESS,
|
||||
CONF_VALUES,
|
||||
CONF_WRITE_ADDRESS,
|
||||
)
|
||||
|
||||
config = _READ_WRITE_MULTIPLE_REGISTERS_SCHEMA(
|
||||
{
|
||||
CONF_ADDRESS: 0,
|
||||
CONF_READ_ADDRESS: 0x10,
|
||||
CONF_WRITE_ADDRESS: 0x20,
|
||||
CONF_VALUES: [1],
|
||||
modbus.CONF_ALLOW_BROADCAST_READ: True,
|
||||
}
|
||||
)
|
||||
assert config[modbus.CONF_ALLOW_BROADCAST_READ] is True
|
||||
assert CONF_CONTINUOUS not in config
|
||||
assert modbus.CONF_EXPECT_BROADCAST_WRITE_RESPONSE not in config
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"key",
|
||||
[modbus.CONF_ALLOW_BROADCAST_READ, modbus.CONF_EXPECT_BROADCAST_WRITE_RESPONSE],
|
||||
)
|
||||
def test_broadcast_options_rejected_on_literal_unicast_address(key: str) -> None:
|
||||
# A broadcast-only option on a literal non-zero address would be silently dropped by the hub.
|
||||
if key == modbus.CONF_ALLOW_BROADCAST_READ:
|
||||
pdu = [0x03, 0x00, 0x10, 0x00, 0x01]
|
||||
else:
|
||||
pdu = [0x06, 0x00, 0x10, 0x00, 0x01]
|
||||
with pytest.raises(cv.Invalid, match="only applies to the broadcast address"):
|
||||
MODBUS_CLIENT_SEND_SCHEMA({CONF_ADDRESS: 1, CONF_PDU: pdu, key: True})
|
||||
# A templated address is not decidable at validation and passes through.
|
||||
config = MODBUS_CLIENT_SEND_SCHEMA(
|
||||
{CONF_ADDRESS: Lambda("return 1;"), CONF_PDU: pdu, key: True}
|
||||
)
|
||||
assert config[key] is True
|
||||
|
||||
@@ -0,0 +1,79 @@
|
||||
"""A modbus_controller cannot poll the broadcast address (0) unless allow_broadcast_read says the
|
||||
device answers it."""
|
||||
|
||||
import pytest
|
||||
|
||||
from esphome import config_validation as cv
|
||||
from esphome.components import modbus
|
||||
from esphome.components.modbus_controller import CONFIG_SCHEMA
|
||||
from esphome.const import CONF_ADDRESS
|
||||
from esphome.types import ConfigType
|
||||
|
||||
|
||||
def _controller(address: int, **extra: object) -> ConfigType:
|
||||
return CONFIG_SCHEMA({modbus.CONF_MODBUS_ID: "bus", CONF_ADDRESS: address, **extra})
|
||||
|
||||
|
||||
def test_address_zero_rejected_by_default() -> None:
|
||||
with pytest.raises(cv.Invalid, match="broadcast address"):
|
||||
_controller(0)
|
||||
|
||||
|
||||
def test_address_zero_accepted_with_allow_broadcast_read() -> None:
|
||||
config = _controller(0, **{modbus.CONF_ALLOW_BROADCAST_READ: True})
|
||||
assert config[CONF_ADDRESS] == 0
|
||||
assert config[modbus.CONF_ALLOW_BROADCAST_READ] is True
|
||||
|
||||
|
||||
def test_allow_broadcast_read_defaults_false() -> None:
|
||||
assert _controller(1)[modbus.CONF_ALLOW_BROADCAST_READ] is False
|
||||
|
||||
|
||||
def test_writer_entity_takes_expect_broadcast_write_response() -> None:
|
||||
# The write-side option lives on the writing platforms, not the controller.
|
||||
from esphome.components.modbus_controller.const import CONF_MODBUS_CONTROLLER_ID
|
||||
from esphome.components.modbus_controller.switch import (
|
||||
CONFIG_SCHEMA as SWITCH_SCHEMA,
|
||||
)
|
||||
from esphome.const import CONF_NAME
|
||||
|
||||
key = modbus.CONF_EXPECT_BROADCAST_WRITE_RESPONSE
|
||||
base = {
|
||||
CONF_MODBUS_CONTROLLER_ID: "ctl",
|
||||
CONF_NAME: "Switch",
|
||||
"register_type": "coil",
|
||||
CONF_ADDRESS: 0x20,
|
||||
}
|
||||
assert SWITCH_SCHEMA(base)[key] is False
|
||||
assert SWITCH_SCHEMA({**base, CONF_NAME: "Switch 2", key: True})[key] is True
|
||||
with pytest.raises(cv.Invalid):
|
||||
CONFIG_SCHEMA({modbus.CONF_MODBUS_ID: "bus", CONF_ADDRESS: 1, key: True})
|
||||
|
||||
|
||||
def test_allow_broadcast_read_requires_address_zero() -> None:
|
||||
# The option only means something at address 0; elsewhere it would be silently inert.
|
||||
with pytest.raises(cv.Invalid, match="only applies to the broadcast address"):
|
||||
_controller(5, **{modbus.CONF_ALLOW_BROADCAST_READ: True})
|
||||
|
||||
|
||||
def test_add_command_options_skips_defaults() -> None:
|
||||
# The setter is only emitted when an option differs from its C++ default.
|
||||
import esphome.codegen as cg
|
||||
from esphome.const import CONF_CONTINUOUS
|
||||
|
||||
var = cg.MockObj("ctl")
|
||||
emitted: list = []
|
||||
original = cg.add
|
||||
cg.add = emitted.append
|
||||
try:
|
||||
modbus.add_command_options(
|
||||
var, "set_read_options", {CONF_CONTINUOUS: False}, direction="read"
|
||||
)
|
||||
assert emitted == []
|
||||
modbus.add_command_options(
|
||||
var, "set_read_options", {CONF_CONTINUOUS: True}, direction="read"
|
||||
)
|
||||
assert len(emitted) == 1
|
||||
assert "set_read_options" in str(emitted[0])
|
||||
finally:
|
||||
cg.add = original
|
||||
@@ -9,6 +9,7 @@ test cannot: a write-coded custom_pdu polled continuously is rejected there.
|
||||
import pytest
|
||||
from voluptuous import Invalid, MultipleInvalid
|
||||
|
||||
from esphome.components import modbus
|
||||
from esphome.components.modbus_controller import (
|
||||
ModbusItemBaseSchema,
|
||||
validate_custom_pdu_item,
|
||||
@@ -55,14 +56,21 @@ def test_custom_pdu_rejects_non_byte_values() -> None:
|
||||
ModbusItemBaseSchema({CONF_CUSTOM_PDU: [0x0103, 0x002A]})
|
||||
|
||||
|
||||
def _controller_full_config(*, continuous: bool) -> Config:
|
||||
def _controller_full_config(
|
||||
*, continuous: bool, allow_broadcast_read: bool = False
|
||||
) -> Config:
|
||||
"""A minimal full-config graph with one modbus_controller declaring id 'ctl', enough for the
|
||||
final-validate to resolve the controller (and its continuous flag) from an item's
|
||||
final-validate to resolve the controller (and its option flags) from an item's
|
||||
modbus_controller_id."""
|
||||
ctl_id = ID("ctl", is_declaration=True)
|
||||
config = Config()
|
||||
config["modbus_controller"] = [
|
||||
{CONF_ID: ctl_id, CONF_ADDRESS: 1, CONF_CONTINUOUS: continuous}
|
||||
{
|
||||
CONF_ID: ctl_id,
|
||||
CONF_ADDRESS: 0 if allow_broadcast_read else 1,
|
||||
CONF_CONTINUOUS: continuous,
|
||||
modbus.CONF_ALLOW_BROADCAST_READ: allow_broadcast_read,
|
||||
}
|
||||
]
|
||||
config.declare_ids.append((ctl_id, ["modbus_controller", 0, CONF_ID]))
|
||||
return config
|
||||
@@ -98,3 +106,64 @@ def test_continuous_read_custom_pdu_allowed(reset_full_config) -> None:
|
||||
CONF_CUSTOM_PDU: [0x03, 0x00, 0x2A, 0x00, 0x01],
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
def test_broadcastable_custom_pdu_rejected_under_broadcast_controller(
|
||||
reset_full_config,
|
||||
) -> None:
|
||||
"""A vendor-coded custom_pdu under an allow_broadcast_read controller would be a real broadcast,
|
||||
never answered, so it is rejected at final validate."""
|
||||
fv.full_config.set(
|
||||
_controller_full_config(continuous=False, allow_broadcast_read=True)
|
||||
)
|
||||
with pytest.raises(Invalid, match="is a real broadcast at address 0"):
|
||||
validate_custom_pdu_item(
|
||||
{
|
||||
CONF_MODBUS_CONTROLLER_ID: ID("ctl"),
|
||||
CONF_CUSTOM_PDU: [0x41, 0x00, 0x03],
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
def test_read_custom_pdu_allowed_under_broadcast_controller(reset_full_config) -> None:
|
||||
"""A read-coded custom_pdu (0x03) is answered under allow_broadcast_read, so it is fine."""
|
||||
fv.full_config.set(
|
||||
_controller_full_config(continuous=False, allow_broadcast_read=True)
|
||||
)
|
||||
validate_custom_pdu_item(
|
||||
{
|
||||
CONF_MODBUS_CONTROLLER_ID: ID("ctl"),
|
||||
CONF_CUSTOM_PDU: [0x03, 0x00, 0x2A, 0x00, 0x01],
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
def test_write_option_rejected_under_unicast_controller(reset_full_config) -> None:
|
||||
"""expect_broadcast_write_response on a writer entity whose controller is not at address 0 is
|
||||
rejected at final validate, where the controller's address is known."""
|
||||
from esphome.components.modbus_controller import validate_writer_item
|
||||
|
||||
fv.full_config.set(_controller_full_config(continuous=False))
|
||||
with pytest.raises(
|
||||
Invalid, match="only applies when the 'ctl' modbus_controller is at address 0"
|
||||
):
|
||||
validate_writer_item(
|
||||
{
|
||||
CONF_MODBUS_CONTROLLER_ID: ID("ctl"),
|
||||
modbus.CONF_EXPECT_BROADCAST_WRITE_RESPONSE: True,
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
def test_write_option_allowed_under_broadcast_controller(reset_full_config) -> None:
|
||||
from esphome.components.modbus_controller import validate_writer_item
|
||||
|
||||
fv.full_config.set(
|
||||
_controller_full_config(continuous=False, allow_broadcast_read=True)
|
||||
)
|
||||
validate_writer_item(
|
||||
{
|
||||
CONF_MODBUS_CONTROLLER_ID: ID("ctl"),
|
||||
modbus.CONF_EXPECT_BROADCAST_WRITE_RESPONSE: True,
|
||||
}
|
||||
)
|
||||
|
||||
@@ -0,0 +1,18 @@
|
||||
---
|
||||
esphome:
|
||||
name: test
|
||||
|
||||
esp32:
|
||||
board: esp32dev
|
||||
|
||||
logger:
|
||||
|
||||
switch:
|
||||
- platform: template
|
||||
id: plain_switch
|
||||
turn_on_action:
|
||||
- logger.log: "on"
|
||||
- platform: template
|
||||
id: enabled_switch
|
||||
optimistic: true
|
||||
assumed_state: true
|
||||
@@ -0,0 +1,17 @@
|
||||
"""Tests for the template switch codegen."""
|
||||
|
||||
from collections.abc import Callable
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
def test_default_flags_are_not_emitted(
|
||||
generate_main: Callable[[str | Path], str],
|
||||
component_config_path: Callable[[str], Path],
|
||||
) -> None:
|
||||
"""Only true optimistic and assumed_state are set; false is the C++ initializer."""
|
||||
main_cpp = generate_main(component_config_path("switch_defaults.yaml"))
|
||||
|
||||
assert "plain_switch->set_optimistic(" not in main_cpp
|
||||
assert "plain_switch->set_assumed_state(" not in main_cpp
|
||||
assert "enabled_switch->set_optimistic(true);" in main_cpp
|
||||
assert "enabled_switch->set_assumed_state(true);" in main_cpp
|
||||
@@ -0,0 +1,42 @@
|
||||
"""Tests for the uart_mux component's final validation."""
|
||||
|
||||
import pytest
|
||||
|
||||
from esphome import config_validation as cv
|
||||
from esphome.const import CONF_ID, PlatformFramework
|
||||
from esphome.core import ID
|
||||
from esphome.types import ConfigType
|
||||
from tests.component_tests.types import SetCoreConfigCallable
|
||||
|
||||
CONF_BRIDGE_ID = "bridge_id"
|
||||
|
||||
|
||||
def _set_esp32_s3(set_core_config: SetCoreConfigCallable) -> None:
|
||||
from esphome.components.esp32 import KEY_VARIANT, VARIANT_ESP32S3
|
||||
|
||||
set_core_config(
|
||||
PlatformFramework.ESP32_IDF, platform_data={KEY_VARIANT: VARIANT_ESP32S3}
|
||||
)
|
||||
|
||||
|
||||
def _mux_conf(mux_id: str, bridge_id: str) -> ConfigType:
|
||||
return {CONF_ID: ID(mux_id), CONF_BRIDGE_ID: ID(bridge_id)}
|
||||
|
||||
|
||||
def test_accepts_one_mux_per_bridge(set_core_config: SetCoreConfigCallable) -> None:
|
||||
_set_esp32_s3(set_core_config)
|
||||
from esphome.components import uart_mux
|
||||
|
||||
uart_mux._final_validate(_mux_conf("mux_0", "bridge_0"))
|
||||
uart_mux._final_validate(_mux_conf("mux_1", "bridge_1"))
|
||||
|
||||
|
||||
def test_rejects_two_muxes_on_one_bridge(
|
||||
set_core_config: SetCoreConfigCallable,
|
||||
) -> None:
|
||||
_set_esp32_s3(set_core_config)
|
||||
from esphome.components import uart_mux
|
||||
|
||||
uart_mux._final_validate(_mux_conf("mux_0", "bridge_0"))
|
||||
with pytest.raises(cv.Invalid, match="already routed by another 'uart_mux'"):
|
||||
uart_mux._final_validate(_mux_conf("mux_1", "bridge_0"))
|
||||
@@ -0,0 +1,46 @@
|
||||
"""Tests for wifi.force_power_save_off(), the hook platforms use to keep the
|
||||
station out of power save."""
|
||||
|
||||
from collections.abc import Callable
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from esphome.components import wifi
|
||||
from esphome.core import CORE, EsphomeError
|
||||
|
||||
|
||||
def test_reasons_accumulate_without_duplicates() -> None:
|
||||
"""Every caller's reason is kept once; a repeated reason is not duplicated."""
|
||||
wifi.force_power_save_off("first")
|
||||
wifi.force_power_save_off("first")
|
||||
wifi.force_power_save_off("second")
|
||||
|
||||
assert CORE.data[wifi.POWER_SAVE_OFF_REASONS_KEY] == ["first", "second"]
|
||||
|
||||
|
||||
def test_forced_off_skips_the_setter_and_warns(
|
||||
generate_main: Callable[[str | Path], str],
|
||||
component_config_path: Callable[[str], Path],
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
) -> None:
|
||||
"""With a reason recorded, power_save_mode is reported and not applied."""
|
||||
wifi.force_power_save_off("the platform cannot sleep")
|
||||
|
||||
main_cpp = generate_main(component_config_path("custom.yaml"))
|
||||
|
||||
assert "set_power_save_mode(" not in main_cpp
|
||||
assert (
|
||||
"power_save_mode LIGHT is not applied: the platform cannot sleep" in caplog.text
|
||||
)
|
||||
|
||||
|
||||
def test_call_after_wifi_codegen_raises(
|
||||
generate_main: Callable[[str | Path], str],
|
||||
component_config_path: Callable[[str], Path],
|
||||
) -> None:
|
||||
"""Once wifi has generated its code the hook cannot take effect any more."""
|
||||
generate_main(component_config_path("custom.yaml"))
|
||||
|
||||
with pytest.raises(EsphomeError, match="before wifi generates its code"):
|
||||
wifi.force_power_save_off("too late")
|
||||
Reference in New Issue
Block a user