[modbus] Add allow_broadcast_read and expect_broadcast_write_response options (#19304)

tests/component_tests/modbus_controller/test_broadcast_address.py

@@ -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

tests/component_tests/modbus_controller/test_custom_pdu.py

@@ -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,
+        }
+    )