diff --git a/esphome/components/packet_interface/basic_fragmentation.cpp b/esphome/components/packet_interface/basic_fragmentation.cpp index 2513afbfd74..b2199c9b039 100644 --- a/esphome/components/packet_interface/basic_fragmentation.cpp +++ b/esphome/components/packet_interface/basic_fragmentation.cpp @@ -44,7 +44,8 @@ void BasicFragmentation::setup() { } void BasicFragmentation::dump_config() { - size_t fragments = this->fragments_for_(this->max_packet_size_); + // A link with no room for fragments fails setup, which core reports below + size_t fragments = this->fragment_size_() == 0 ? 0 : this->fragments_for_(this->max_packet_size_); ESP_LOGCONFIG(TAG, "Basic Fragmentation on '%s':\n" " Max Packet Size: %zu\n" diff --git a/esphome/components/packet_interface/layers/basic_addressing.py b/esphome/components/packet_interface/layers/basic_addressing.py index b1003f670f5..76dc4e150d4 100644 --- a/esphome/components/packet_interface/layers/basic_addressing.py +++ b/esphome/components/packet_interface/layers/basic_addressing.py @@ -49,6 +49,22 @@ def _unique_destinations(peers: list[ConfigType]) -> list[ConfigType]: return peers +def _not_own_address(config: ConfigType) -> ConfigType: + """Packets from this device's own address are dropped as echoes, so it cannot be a destination.""" + address = config[CONF_ADDRESS] + if config.get(CONF_DESTINATION) == address: + raise cv.Invalid( + "Must not be this device's own address", path=[CONF_DESTINATION] + ) + for index, peer in enumerate(config.get(CONF_PEERS, [])): + if peer[CONF_DESTINATION] == address: + raise cv.Invalid( + "Must not be this device's own address", + path=[CONF_PEERS, index, CONF_DESTINATION], + ) + return config + + # A function, as each peer holds the other layers, whose schemas exist only once every layer type is registered def _basic_addressing_schema() -> cv.Schema: peer_schema = cv.All( @@ -74,6 +90,7 @@ def _basic_addressing_schema() -> cv.Schema: } ).extend(cv.COMPONENT_SCHEMA), cv.has_at_most_one_key(CONF_DESTINATION, CONF_PEERS), + _not_own_address, ) diff --git a/tests/unit_tests/test_packet_interface.py b/tests/unit_tests/test_packet_interface.py index 6fc1858d58f..895bceab29b 100644 --- a/tests/unit_tests/test_packet_interface.py +++ b/tests/unit_tests/test_packet_interface.py @@ -23,6 +23,8 @@ from esphome.components.packet_interface.layers.hdlc_reliability import ( from esphome.components.uart import packet_interface as uart_packet_interface from esphome.components.udp import packet_interface as udp_packet_interface from esphome.const import CONF_ENCRYPTION +from esphome.core import ID +import esphome.final_validate as fv @pytest.mark.parametrize( @@ -292,3 +294,36 @@ def test_auto_load_follows_layers( def test_auto_load_without_config() -> None: """Tooling that lists dependencies gets everything a layer could load.""" assert packet_interface.AUTO_LOAD() == ["binary_sensor", "xxtea"] + + +@pytest.mark.parametrize( + "addressing", + [ + {"address": 1, "destination": 1}, + {"address": 1, "peers": [_peer("a", 2), _peer("b", 1)]}, + ], +) +def test_addressing_own_address_rejected(setup_core, addressing: dict) -> None: + """A device cannot send to its own address, as its own packets are dropped as echoes.""" + with pytest.raises(cv.Invalid, match="own address"): + uart_packet_interface.CONFIG_SCHEMA({CONF_ADDRESSING: addressing}) + + +def test_application_needs_peer_id_with_several_peers(setup_core) -> None: + """An application must use one peer's id, not the id of a link with several peers.""" + validator = packet_interface.final_validate_interface("packet_interface") + token = fv.full_config.set( + { + "packet_interface": [ + {"id": ID("bus"), CONF_ADDRESSING: {"peers": [_peer("a", 2)]}}, + {"id": ID("single"), CONF_ADDRESSING: {"destination": 2}}, + ] + } + ) + try: + with pytest.raises(cv.Invalid, match="several peers"): + validator({"packet_interface": ID("bus")}) + assert validator({"packet_interface": ID("single")}) + assert validator({"packet_interface": ID("a")}) + finally: + fv.full_config.reset(token)