From 6c55b3148f75b45d5e6ae44b7d1444417ea8cc00 Mon Sep 17 00:00:00 2001 From: clydebarrow <2366188+clydebarrow@users.noreply.github.com> Date: Tue, 20 Oct 2026 08:03:25 +1100 Subject: [PATCH] [tcp] Explain why the rolling code cannot be used on a TCP link Co-Authored-By: Claude Opus 5.5 --- .../components/packet_interface/__init__.py | 40 +++++++++++++------ tests/unit_tests/test_tcp_packet_interface.py | 12 ++++++ 2 files changed, 39 insertions(+), 13 deletions(-) diff --git a/esphome/components/packet_interface/__init__.py b/esphome/components/packet_interface/__init__.py index fc1bb702baa..bff8f690321 100644 --- a/esphome/components/packet_interface/__init__.py +++ b/esphome/components/packet_interface/__init__.py @@ -19,6 +19,7 @@ from .layers import ( CONF_ADDRESSING, CONF_DESTINATION, CONF_PEERS, + CONF_SESSION, STACK_ORDER, PacketInterface, build_stack, @@ -132,6 +133,20 @@ def final_validate_interface(key: str) -> Callable[[ConfigType], ConfigType]: return validator +def _check_provided_session(config: ConfigType) -> ConfigType: + """A link's own session does not send packets through the layers, so none can need them from the peer.""" + for kind in STACK_ORDER: + if (layer := config.get(kind.key)) is not None and kind.get_type( + layer + ).needs_session(layer): + raise cv.Invalid( + f"This '{kind.key}' needs packets from the peer, which this link does not send through its layers, " + "so it cannot be used on this link", + path=[kind.key], + ) + return config + + def packet_interface_schema( class_: MockObjClass, default_max_packet_size: int, @@ -147,19 +162,18 @@ def packet_interface_schema( ) for key, schema in LINK_LAYER_SCHEMAS.items() } - return ( - cv.Schema( - { - cv.GenerateID(): cv.declare_id(class_), - cv.Optional( - CONF_MAX_PACKET_SIZE, default=default_max_packet_size - ): cv.int_range(16, max_packet_size_limit), - **link_layers, - } - ) - .extend(cv.COMPONENT_SCHEMA) - .add_extra(_validate_layers) - ) + schema = cv.Schema( + { + cv.GenerateID(): cv.declare_id(class_), + cv.Optional( + CONF_MAX_PACKET_SIZE, default=default_max_packet_size + ): cv.int_range(16, max_packet_size_limit), + **link_layers, + } + ).extend(cv.COMPONENT_SCHEMA) + if CONF_SESSION in provides: + schema = schema.add_extra(_check_provided_session) + return schema.add_extra(_validate_layers) async def new_packet_interface(config: ConfigType, *args: Any) -> MockObj: diff --git a/tests/unit_tests/test_tcp_packet_interface.py b/tests/unit_tests/test_tcp_packet_interface.py index 77828908972..f1d32a104c7 100644 --- a/tests/unit_tests/test_tcp_packet_interface.py +++ b/tests/unit_tests/test_tcp_packet_interface.py @@ -28,6 +28,18 @@ def test_provided_layers_rejected(setup_core, layer: str) -> None: ) +def test_rolling_code_rejected(setup_core) -> None: + """The rolling code needs the session's keep-alives, which TCP does not send through the layers.""" + with pytest.raises(cv.Invalid, match="cannot be used on this link"): + tcp_packet_interface.CONFIG_SCHEMA( + { + "host": "192.168.1.50", + "port": 6060, + "encryption": {"key": "secret", "rolling_code_enable": True}, + } + ) + + def test_auto_load_binary_sensor_for_connected(setup_core) -> None: """The binary_sensor component is loaded only for the connected sensor.""" assert tcp_packet_interface.AUTO_LOAD() == ["socket", "binary_sensor"]