[entity_transport] Reject a timeout on a packet interface with a session layer

With a session and a reliability layer, the update interval defaults to
never, and unchanged states are not sent again, so a receiver's timeout
cleared them on a working link. The session layer already clears the
states when the other device cannot be reached.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
clydebarrow
2026-10-20 08:41:48 +11:00
co-authored by Claude Opus 5.5
parent b17cb0c48c
commit 75c91fb6e5
4 changed files with 57 additions and 3 deletions
@@ -7,7 +7,8 @@ import esphome.codegen as cg
from esphome.components import packet_interface
from esphome.components.binary_sensor import BinarySensor
from esphome.components.const import CONF_PROVIDER, CONF_REMOTE_ID
from esphome.components.packet_interface import CONF_PACKET_INTERFACE
from esphome.components.packet_interface import CONF_PACKET_INTERFACE, interface_layers
from esphome.components.packet_interface.layers import CONF_SESSION
from esphome.components.sensor import Sensor
import esphome.config_validation as cv
from esphome.const import (
@@ -95,6 +96,21 @@ CONFIG_SCHEMA = cv.Schema(
).extend(cv.COMPONENT_SCHEMA)
def _final_validate(config: ConfigType) -> ConfigType:
"""With a session layer, which clears the states when the peer is lost, nothing would refresh an unchanged state."""
layers = interface_layers(config[CONF_PACKET_INTERFACE])
if CONF_TIMEOUT in config and layers is not None and CONF_SESSION in layers:
raise cv.Invalid(
f"Not allowed with a '{CONF_SESSION}' layer, which already clears the states when the other device "
"cannot be reached; unchanged states may never be sent again, so they would time out on a working link",
path=[CONF_TIMEOUT],
)
return config
FINAL_VALIDATE_SCHEMA = _final_validate
def _validate_remote_id(config: ConfigType) -> ConfigType:
if CONF_REMOTE_ID not in config and not config[CONF_ID].is_manual:
raise cv.Invalid(f"'{CONF_REMOTE_ID}' is required unless '{CONF_ID}' is given")
@@ -6,7 +6,9 @@ import esphome.codegen as cg
from esphome.components.const import CONF_MAX_PACKET_SIZE
import esphome.config_validation as cv
from esphome.const import CONF_ID
from esphome.core import ID
from esphome.cpp_generator import MockObj, MockObjClass
import esphome.final_validate as fv
from esphome.types import ConfigType
from . import layers
@@ -34,6 +36,7 @@ LINK_LAYER_SCHEMAS = stack_schema()
__all__ = [
"CONF_MAX_PACKET_SIZE",
"CONF_PACKET_INTERFACE",
"interface_layers",
"PacketInterface",
"new_packet_interface",
"packet_interface_ns",
@@ -59,6 +62,14 @@ def _validate_layers(config: ConfigType) -> ConfigType:
return config
def interface_layers(interface_id: ID) -> ConfigType | None:
"""During final validation, the configuration holding the layers of the packet interface with this id."""
for interface in fv.full_config.get().get(DOMAIN, []):
if interface[CONF_ID].id == interface_id.id:
return interface
return None
def packet_interface_schema(
class_: MockObjClass,
default_max_packet_size: int,
@@ -2,7 +2,6 @@ packet_interface:
- platform: uart
id: entity_transport_link
reliability:
session:
sensor:
- platform: template
+29 -1
View File
@@ -4,7 +4,11 @@ import pytest
from esphome import config_validation as cv
from esphome.components.const import CONF_PROVIDER, CONF_REMOTE_ID
from esphome.components.entity_transport import CONFIG_SCHEMA, remote_entity_schema
from esphome.components.entity_transport import (
CONFIG_SCHEMA,
FINAL_VALIDATE_SCHEMA,
remote_entity_schema,
)
from esphome.components.sensor import sensor_schema
from esphome.const import (
CONF_ID,
@@ -14,6 +18,7 @@ from esphome.const import (
SCHEDULER_DONT_RUN,
)
from esphome.core import ID
import esphome.final_validate as fv
def test_remote_id_required_without_manual_id(setup_core) -> None:
@@ -51,6 +56,29 @@ def test_timeout(setup_core) -> None:
CONFIG_SCHEMA({**base, CONF_TIMEOUT: "500ms"})
@pytest.mark.parametrize(
("layers", "allowed"),
[({}, True), ({"reliability": None}, True), ({"session": {}}, False)],
)
def test_timeout_not_with_session(setup_core, layers: dict, allowed: bool) -> None:
"""With a session layer, unchanged states are never sent again, so a timeout would clear them on a working link."""
config = CONFIG_SCHEMA({"packet_interface": "link", CONF_TIMEOUT: "1min"})
token = fv.full_config.set(
{"packet_interface": [{CONF_ID: ID("link", is_manual=True), **layers}]}
)
try:
if allowed:
FINAL_VALIDATE_SCHEMA(config)
else:
with pytest.raises(cv.Invalid, match="session") as error:
FINAL_VALIDATE_SCHEMA(config)
assert error.value.path == [CONF_TIMEOUT]
# Without a timeout, any layers are fine
FINAL_VALIDATE_SCHEMA(CONFIG_SCHEMA({"packet_interface": "link"}))
finally:
fv.full_config.reset(token)
def test_advertised_ids_unique(setup_core) -> None:
"""Two entities of one type cannot be sent with the same id."""
base = {"packet_interface": "link"}