Measure MQTT topics in bytes and mirror the firmware object id

# Conflicts:
#	esphome/components/mqtt/__init__.py

esphome/components/mqtt/__init__.py

@@ -1,3 +1,5 @@
+from collections.abc import Iterator
+
 from esphome import automation
 import esphome.codegen as cg
 from esphome.components import logger, network, socket
@@ -8,6 +10,7 @@ from esphome.components.esp32 import (
     request_tls,
     require_certificate_bundle,
 )
+from esphome.config import iter_ids
 from esphome.config_helpers import (
     filter_source_files_from_defines,
     filter_source_files_from_platform,
@@ -24,6 +27,8 @@ from esphome.const import (
     CONF_CLIENT_ID,
     CONF_COMMAND_RETAIN,
     CONF_COMMAND_TOPIC,
+    CONF_DEVICE_ID,
+    CONF_DEVICES,
     CONF_DISCOVER_IP,
     CONF_DISCOVERY,
     CONF_DISCOVERY_OBJECT_ID_GENERATOR,
@@ -31,11 +36,17 @@ from esphome.const import (
     CONF_DISCOVERY_RETAIN,
     CONF_DISCOVERY_UNIQUE_ID_GENERATOR,
     CONF_ENABLE_ON_BOOT,
+    CONF_ESPHOME,
     CONF_ID,
+    CONF_INTERNAL,
     CONF_KEEPALIVE,
     CONF_LEVEL,
     CONF_LOG_TOPIC,
+    CONF_MODE_COMMAND_TOPIC,
     CONF_MQTT,
+    CONF_MQTT_ID,
+    CONF_NAME,
+    CONF_NAME_ADD_MAC_SUFFIX,
     CONF_ON_CONNECT,
     CONF_ON_DISCONNECT,
     CONF_ON_JSON_MESSAGE,
@@ -44,6 +55,7 @@ from esphome.const import (
     CONF_PAYLOAD,
     CONF_PAYLOAD_AVAILABLE,
     CONF_PAYLOAD_NOT_AVAILABLE,
+    CONF_PLATFORM,
     CONF_PORT,
     CONF_PUBLISH_NAN_AS_NONE,
     CONF_QOS,
@@ -53,6 +65,9 @@ from esphome.const import (
     CONF_SKIP_CERT_CN_CHECK,
     CONF_STATE_TOPIC,
     CONF_SUBSCRIBE_QOS,
+    CONF_TARGET_TEMPERATURE_COMMAND_TOPIC,
+    CONF_TARGET_TEMPERATURE_HIGH_COMMAND_TOPIC,
+    CONF_TARGET_TEMPERATURE_LOW_COMMAND_TOPIC,
     CONF_TOPIC,
     CONF_TOPIC_PREFIX,
     CONF_USE_ABBREVIATIONS,
@@ -67,6 +82,9 @@ from esphome.const import (
     PlatformFramework,
 )
 from esphome.core import CORE, CoroPriority, coroutine_with_priority
+from esphome.cpp_generator import MockObjClass
+import esphome.final_validate as fv
+from esphome.helpers import sanitize, snake_case
 from esphome.types import ConfigType
 
 DEPENDENCIES = ["network"]
@@ -130,28 +148,76 @@ MQTTClientComponent = mqtt_ns.class_("MQTTClientComponent", cg.Component)
 MQTTPublishJsonAction = mqtt_ns.class_("MQTTPublishJsonAction", automation.Action)
 MQTTComponent = mqtt_ns.class_("MQTTComponent", cg.Component)
 
-MQTTAlarmControlPanelComponent = mqtt_ns.class_(
-    "MQTTAlarmControlPanelComponent", MQTTComponent
+# A topic the firmware builds from a suffix, unless the config key replaces the whole topic
+_Topic = tuple[str, str]
+# The firmware subscribes to exactly one of these alternatives, chosen by entity traits
+_Subscription = tuple[tuple[_Topic, ...], ...]
+# The subscriptions each entity MQTT class always makes, keyed by class: component type,
+# then the subscriptions
+_COMMAND_TOPICS: dict[str, tuple[str, tuple[_Subscription, ...]]] = {}
+
+
+def _single_topic(suffix: str, custom_key: str) -> _Subscription:
+    return (((suffix, custom_key),),)
+
+
+_COMMAND = _single_topic("command", CONF_COMMAND_TOPIC)
+
+
+def _entity_class(
+    name: str,
+    component_type: str | None = None,
+    subscriptions: tuple[_Subscription, ...] = (_COMMAND,),
+) -> MockObjClass:
+    """Declare an entity MQTT class; a component_type means it subscribes to command topics."""
+    cls = mqtt_ns.class_(name, MQTTComponent)
+    if component_type is not None:
+        _COMMAND_TOPICS[str(cls)] = (component_type, subscriptions)
+    return cls
+
+
+MQTTAlarmControlPanelComponent = _entity_class(
+    "MQTTAlarmControlPanelComponent", "alarm_control_panel"
 )
-MQTTBinarySensorComponent = mqtt_ns.class_("MQTTBinarySensorComponent", MQTTComponent)
-MQTTClimateComponent = mqtt_ns.class_("MQTTClimateComponent", MQTTComponent)
-MQTTCoverComponent = mqtt_ns.class_("MQTTCoverComponent", MQTTComponent)
-MQTTFanComponent = mqtt_ns.class_("MQTTFanComponent", MQTTComponent)
-MQTTJSONLightComponent = mqtt_ns.class_("MQTTJSONLightComponent", MQTTComponent)
-MQTTSensorComponent = mqtt_ns.class_("MQTTSensorComponent", MQTTComponent)
-MQTTSwitchComponent = mqtt_ns.class_("MQTTSwitchComponent", MQTTComponent)
-MQTTTextSensor = mqtt_ns.class_("MQTTTextSensor", MQTTComponent)
-MQTTNumberComponent = mqtt_ns.class_("MQTTNumberComponent", MQTTComponent)
-MQTTDateComponent = mqtt_ns.class_("MQTTDateComponent", MQTTComponent)
-MQTTTimeComponent = mqtt_ns.class_("MQTTTimeComponent", MQTTComponent)
-MQTTDateTimeComponent = mqtt_ns.class_("MQTTDateTimeComponent", MQTTComponent)
-MQTTTextComponent = mqtt_ns.class_("MQTTTextComponent", MQTTComponent)
-MQTTSelectComponent = mqtt_ns.class_("MQTTSelectComponent", MQTTComponent)
-MQTTButtonComponent = mqtt_ns.class_("MQTTButtonComponent", MQTTComponent)
-MQTTLockComponent = mqtt_ns.class_("MQTTLockComponent", MQTTComponent)
-MQTTEventComponent = mqtt_ns.class_("MQTTEventComponent", MQTTComponent)
-MQTTUpdateComponent = mqtt_ns.class_("MQTTUpdateComponent", MQTTComponent)
-MQTTValveComponent = mqtt_ns.class_("MQTTValveComponent", MQTTComponent)
+MQTTBinarySensorComponent = _entity_class("MQTTBinarySensorComponent")
+MQTTClimateComponent = _entity_class(
+    "MQTTClimateComponent",
+    "climate",
+    (
+        _single_topic("mode/command", CONF_MODE_COMMAND_TOPIC),
+        # One target temperature topic, or the _low and _high pair on two-point devices
+        (
+            (("target_temperature/command", CONF_TARGET_TEMPERATURE_COMMAND_TOPIC),),
+            (
+                (
+                    "target_temperature_low/command",
+                    CONF_TARGET_TEMPERATURE_LOW_COMMAND_TOPIC,
+                ),
+                (
+                    "target_temperature_high/command",
+                    CONF_TARGET_TEMPERATURE_HIGH_COMMAND_TOPIC,
+                ),
+            ),
+        ),
+    ),
+)
+MQTTCoverComponent = _entity_class("MQTTCoverComponent", "cover")
+MQTTFanComponent = _entity_class("MQTTFanComponent", "fan")
+MQTTJSONLightComponent = _entity_class("MQTTJSONLightComponent", "light")
+MQTTSensorComponent = _entity_class("MQTTSensorComponent")
+MQTTSwitchComponent = _entity_class("MQTTSwitchComponent", "switch")
+MQTTTextSensor = _entity_class("MQTTTextSensor")
+MQTTNumberComponent = _entity_class("MQTTNumberComponent", "number")
+MQTTDateComponent = _entity_class("MQTTDateComponent", "date")
+MQTTTimeComponent = _entity_class("MQTTTimeComponent", "time")
+MQTTDateTimeComponent = _entity_class("MQTTDateTimeComponent", "datetime")
+MQTTTextComponent = _entity_class("MQTTTextComponent", "text")
+MQTTSelectComponent = _entity_class("MQTTSelectComponent", "select")
+MQTTButtonComponent = _entity_class("MQTTButtonComponent", "button")
+MQTTLockComponent = _entity_class("MQTTLockComponent", "lock")
+MQTTEventComponent = _entity_class("MQTTEventComponent")
+MQTTUpdateComponent = _entity_class("MQTTUpdateComponent", "update")
+MQTTValveComponent = _entity_class("MQTTValveComponent", "valve")
 
 MQTTDiscoveryUniqueIdGenerator = mqtt_ns.enum("MQTTDiscoveryUniqueIdGenerator")
 MQTT_DISCOVERY_UNIQUE_ID_GENERATOR_OPTIONS = {
@@ -369,6 +435,137 @@ CONFIG_SCHEMA = cv.All(
     network.require_ipv4(CONF_MQTT),
 )
 
+# AsyncMqttClient, the ESP8266 and LibreTiny backend, drops any inbound message whose
+# topic is longer than this many bytes (its default maxTopicLength)
+ASYNC_MQTT_CLIENT_MAX_TOPIC_LEN = 128
+# Keep in sync with OBJECT_ID_MAX_LEN in esphome/core/entity_base.h
+OBJECT_ID_MAX_LEN = 128
+# name_add_mac_suffix appends "-" plus 6 hex digits to the name at runtime
+MAC_SUFFIX = "-xxxxxx"
+
+
+def _entity_object_id(
+    config: ConfigType, mac_suffix: bool, devices: dict[str, str]
+) -> str:
+    """Build the object id the way EntityBase::write_object_id_to() does."""
+    # The name falls back the way EntityBase::configure_entity_() does
+    name = config.get(CONF_NAME) or ""
+    if not name:
+        if (device := config.get(CONF_DEVICE_ID)) is not None:
+            name = devices[device.id]
+        elif mac_suffix:
+            # Only the friendly name gains the MAC suffix, and an empty one stays empty
+            name = f"{CORE.friendly_name}{MAC_SUFFIX}" if CORE.friendly_name else ""
+        else:
+            name = CORE.friendly_name or CORE.name
+    # The firmware sanitizes each UTF-8 byte to one character and caps the result
+    raw = name.encode()[: OBJECT_ID_MAX_LEN - 1]
+    return sanitize(snake_case(raw.decode("ascii", "replace")))
+
+
+def _entity_topic(
+    config: ConfigType, default_topic: str, suffix: str, custom_key: str
+) -> str | None:
+    """Return the topic an entity subscribes to, or None when only known at runtime."""
+    custom = config.get(custom_key)
+    # An empty specialized topic falls back to the default (MQTT_COMPONENT_CUSTOM_TOPIC_)
+    # while an empty command_topic stays empty
+    if custom == "" and custom_key != CONF_COMMAND_TOPIC:
+        custom = None
+    if custom is not None:
+        # A lambda topic is only known at runtime
+        return custom if isinstance(custom, str) else None
+    return f"{default_topic}/{suffix}" if default_topic else None
+
+
+def _entity_subscriptions(
+    config: ConfigType, topic_prefix: str, mac_suffix: bool, devices: dict[str, str]
+) -> Iterator[tuple[tuple[str, ...], ...]]:
+    """Yield the subscriptions an entity's MQTT component always makes, as known topics."""
+    component_type, subscriptions = _COMMAND_TOPICS[str(config[CONF_MQTT_ID].type)]
+    # Mirrors MQTTComponent::compute_is_internal_(): internal components never subscribe
+    if CONF_STATE_TOPIC in config:
+        internal = config[CONF_STATE_TOPIC] == ""
+    elif CONF_COMMAND_TOPIC in config:
+        internal = config[CONF_COMMAND_TOPIC] == ""
+    else:
+        internal = not topic_prefix or config.get(CONF_INTERNAL, False)
+    if internal:
+        return
+    default_topic = ""
+    if topic_prefix:
+        object_id = _entity_object_id(config, mac_suffix, devices)
+        default_topic = f"{topic_prefix}/{component_type}/{object_id}"
+    for subscription in subscriptions:
+        yield tuple(
+            tuple(
+                topic
+                for suffix, custom_key in alternative
+                if (topic := _entity_topic(config, default_topic, suffix, custom_key))
+                is not None
+            )
+            for alternative in subscription
+        )
+
+
+def _subscriptions(config: ConfigType) -> Iterator[tuple[tuple[str, ...], ...]]:
+    """Yield every subscription known at build time, as alternatives of topics."""
+    full_config = fv.full_config.get()
+    esphome_config = full_config[CONF_ESPHOME]
+    mac_suffix = esphome_config.get(CONF_NAME_ADD_MAC_SUFFIX, False)
+    devices = {
+        device[CONF_ID].id: device[CONF_NAME]
+        for device in esphome_config.get(CONF_DEVICES, [])
+    }
+    topic_prefix = config[CONF_TOPIC_PREFIX]
+    if mac_suffix and topic_prefix == CORE.name:
+        topic_prefix += MAC_SUFFIX
+    # Entities can sit at any depth (a platform can declare its own switches), so find them by mqtt_id
+    for id_, path in iter_ids(full_config):
+        if path[-1] != CONF_MQTT_ID or str(id_.type) not in _COMMAND_TOPICS:
+            continue
+        entity = full_config
+        for key in path[:-1]:
+            entity = entity[key]
+        yield from _entity_subscriptions(entity, topic_prefix, mac_suffix, devices)
+    for key in (CONF_ON_MESSAGE, CONF_ON_JSON_MESSAGE):
+        for trigger in config.get(key, []):
+            yield ((trigger[CONF_TOPIC],),)
+    for domain in ("sensor", "text_sensor"):
+        for entity in full_config.get(domain) or []:
+            if entity[CONF_PLATFORM] == "mqtt_subscribe":
+                yield ((entity[CONF_TOPIC],),)
+
+
+def _too_long(topic: str) -> bool:
+    # A wildcard filter matches topics of any length, so only literals are checked
+    if "+" in topic or "#" in topic:
+        return False
+    return len(topic.encode()) > ASYNC_MQTT_CLIENT_MAX_TOPIC_LEN
+
+
+def _validate_topic_lengths(config: ConfigType) -> ConfigType:
+    """Reject topics that the ESP8266 and LibreTiny MQTT client would silently drop."""
+    if not (CORE.is_esp8266 or CORE.is_libretiny):
+        return config
+    for subscription in _subscriptions(config):
+        # The config cannot tell which alternative the firmware picks, so only a
+        # subscription that is too long whichever way is rejected
+        too_long = [
+            next((topic for topic in alternative if _too_long(topic)), None)
+            for alternative in subscription
+        ]
+        if all(too_long):
+            topic = too_long[0]
+            raise cv.Invalid(
+                f"MQTT topic '{topic}' is {len(topic.encode())} bytes; the MQTT client "
+                f"on this platform drops topics longer than {ASYNC_MQTT_CLIENT_MAX_TOPIC_LEN}"
+            )
+    return config
+
+
+FINAL_VALIDATE_SCHEMA = _validate_topic_lengths
+
 
 def exp_mqtt_message(config):
     if config is None:

tests/unit_tests/components/test_mqtt_topic_length.py

@@ -0,0 +1,274 @@
+"""Tests for the mqtt check that subscribed topics fit the AsyncMqttClient limit."""
+
+import pytest
+
+from esphome.components.mqtt import (
+    ASYNC_MQTT_CLIENT_MAX_TOPIC_LEN,
+    OBJECT_ID_MAX_LEN,
+    MQTTClimateComponent,
+    MQTTSwitchComponent,
+    _validate_topic_lengths,
+)
+import esphome.config_validation as cv
+from esphome.const import (
+    CONF_DEVICES,
+    CONF_ESPHOME,
+    CONF_ID,
+    CONF_MQTT_ID,
+    CONF_NAME,
+    CONF_NAME_ADD_MAC_SUFFIX,
+    KEY_CORE,
+    KEY_TARGET_PLATFORM,
+    PLATFORM_BK72XX,
+    PLATFORM_ESP32,
+    PLATFORM_ESP8266,
+)
+from esphome.core import CORE, ID, Lambda
+from esphome.cpp_generator import MockObjClass
+import esphome.final_validate as fv
+from esphome.types import ConfigType
+
+# "dev/switch/<object_id>/command" is 19 bytes around the object id
+FITS = "a" * (ASYNC_MQTT_CLIENT_MAX_TOPIC_LEN - 19)
+TOO_LONG = FITS + "a"
+LONG_LITERAL = "t" * (ASYNC_MQTT_CLIENT_MAX_TOPIC_LEN + 1)
+# Each byte of a two-byte character sanitizes to one underscore on the device
+FITS_MULTIBYTE = "é" * ((ASYNC_MQTT_CLIENT_MAX_TOPIC_LEN - 19) // 2)
+TOO_LONG_MULTIBYTE = FITS_MULTIBYTE + "é"
+LONG_MULTIBYTE_LITERAL = "é" * (ASYNC_MQTT_CLIENT_MAX_TOPIC_LEN // 2 + 1)
+
+
+def _mqtt_id(cls: MockObjClass) -> ID:
+    return ID("mqtt_id", is_declaration=True, type=cls)
+
+
+def _switch(name: str, **extra) -> dict:
+    return {
+        "switch": [
+            {CONF_NAME: name, CONF_MQTT_ID: _mqtt_id(MQTTSwitchComponent), **extra}
+        ]
+    }
+
+
+def _run(
+    monkeypatch: pytest.MonkeyPatch,
+    full_config: dict,
+    mqtt: ConfigType | None = None,
+    platform: str = PLATFORM_ESP8266,
+) -> ConfigType:
+    monkeypatch.setitem(CORE.data, KEY_CORE, {KEY_TARGET_PLATFORM: platform})
+    monkeypatch.setattr(CORE, "name", "dev")
+    monkeypatch.setattr(CORE, "friendly_name", None)
+    config = {"topic_prefix": "dev", **(mqtt or {})}
+    token = fv.full_config.set({CONF_ESPHOME: {}, **full_config})
+    try:
+        return _validate_topic_lengths(config)
+    finally:
+        fv.full_config.reset(token)
+
+
+@pytest.mark.parametrize(
+    ("full_config", "mqtt", "platform"),
+    [
+        pytest.param(_switch(TOO_LONG), None, PLATFORM_ESP8266, id="command_topic"),
+        pytest.param(_switch(TOO_LONG), None, PLATFORM_BK72XX, id="libretiny"),
+        pytest.param(
+            {CONF_ESPHOME: {CONF_NAME_ADD_MAC_SUFFIX: True}, **_switch(FITS)},
+            None,
+            PLATFORM_ESP8266,
+            id="mac_suffix",
+        ),
+        pytest.param(
+            _switch("s", command_topic=LONG_LITERAL),
+            None,
+            PLATFORM_ESP8266,
+            id="custom_command_topic",
+        ),
+        pytest.param(
+            _switch(TOO_LONG_MULTIBYTE), None, PLATFORM_ESP8266, id="multibyte_name"
+        ),
+        pytest.param(
+            _switch("s", command_topic=LONG_MULTIBYTE_LITERAL),
+            None,
+            PLATFORM_ESP8266,
+            id="multibyte_literal",
+        ),
+        pytest.param(
+            {
+                "climate": [
+                    # "dev/climate/<object_id>/mode/command" is 25 characters around it
+                    {CONF_NAME: "a" * 104, CONF_MQTT_ID: _mqtt_id(MQTTClimateComponent)}
+                ]
+            },
+            None,
+            PLATFORM_ESP8266,
+            id="climate_mode_command",
+        ),
+        pytest.param(
+            # An empty specialized topic means the default, unlike an empty command_topic
+            {
+                "climate": [
+                    {
+                        CONF_NAME: "a" * 104,
+                        CONF_MQTT_ID: _mqtt_id(MQTTClimateComponent),
+                        "mode_command_topic": "",
+                        "target_temperature_command_topic": "short/temp",
+                    }
+                ]
+            },
+            None,
+            PLATFORM_ESP8266,
+            id="climate_empty_mode_command_topic",
+        ),
+        pytest.param(
+            {},
+            {"on_message": [{"topic": LONG_LITERAL}]},
+            PLATFORM_ESP8266,
+            id="trigger",
+        ),
+        pytest.param(
+            {"sensor": [{"platform": "mqtt_subscribe", "topic": LONG_LITERAL}]},
+            None,
+            PLATFORM_ESP8266,
+            id="mqtt_subscribe",
+        ),
+        pytest.param(
+            {
+                CONF_ESPHOME: {
+                    CONF_DEVICES: [{CONF_ID: ID("sub"), CONF_NAME: TOO_LONG}]
+                },
+                **_switch("", device_id=ID("sub")),
+            },
+            None,
+            PLATFORM_ESP8266,
+            id="unnamed_on_sub_device",
+        ),
+        pytest.param(
+            _switch("s", command_topic=LONG_LITERAL),
+            {"topic_prefix": ""},
+            PLATFORM_ESP8266,
+            id="custom_topic_without_prefix",
+        ),
+        pytest.param(
+            # A platform that declares its own switches, like sprinkler valves
+            {"sprinkler": [{"valves": [_switch(TOO_LONG)["switch"][0]]}]},
+            None,
+            PLATFORM_ESP8266,
+            id="nested_entity",
+        ),
+        pytest.param(
+            # "dev/climate/<object_id>/target_temperature/command" is 39 around it,
+            # while mode/command still fits
+            {
+                "climate": [
+                    {CONF_NAME: "a" * 90, CONF_MQTT_ID: _mqtt_id(MQTTClimateComponent)}
+                ]
+            },
+            None,
+            PLATFORM_ESP8266,
+            id="climate_target_temperature",
+        ),
+    ],
+)
+def test_rejected(
+    monkeypatch: pytest.MonkeyPatch, full_config: dict, mqtt: dict | None, platform: str
+) -> None:
+    with pytest.raises(cv.Invalid, match="bytes"):
+        _run(monkeypatch, full_config, mqtt, platform)
+
+
+@pytest.mark.parametrize(
+    ("full_config", "mqtt", "platform"),
+    [
+        pytest.param(_switch(FITS), None, PLATFORM_ESP8266, id="at_the_limit"),
+        pytest.param(
+            _switch(FITS_MULTIBYTE), None, PLATFORM_ESP8266, id="multibyte_at_the_limit"
+        ),
+        pytest.param(
+            _switch("s", state_topic="dev/s", command_topic=""),
+            None,
+            PLATFORM_ESP8266,
+            id="empty_command_topic_stays_empty",
+        ),
+        pytest.param(
+            # The firmware subscribes to the short _low and _high overrides on a two-point
+            # device, so the unused default target_temperature topic must not count
+            {
+                "climate": [
+                    {
+                        CONF_NAME: "a" * 90,
+                        CONF_MQTT_ID: _mqtt_id(MQTTClimateComponent),
+                        "target_temperature_low_command_topic": "short/low",
+                        "target_temperature_high_command_topic": "short/high",
+                    }
+                ]
+            },
+            None,
+            PLATFORM_ESP8266,
+            id="climate_two_point_overrides",
+        ),
+        pytest.param(_switch(TOO_LONG), None, PLATFORM_ESP32, id="esp32"),
+        pytest.param(
+            _switch(TOO_LONG, command_topic="short/cmd"),
+            None,
+            PLATFORM_ESP8266,
+            id="short_custom_topic",
+        ),
+        pytest.param(
+            _switch(TOO_LONG, command_topic=Lambda('return "x";')),
+            None,
+            PLATFORM_ESP8266,
+            id="lambda_topic",
+        ),
+        pytest.param(
+            _switch(TOO_LONG, internal=True), None, PLATFORM_ESP8266, id="internal"
+        ),
+        pytest.param(
+            {"switch": [{CONF_NAME: TOO_LONG}]}, None, PLATFORM_ESP8266, id="no_mqtt"
+        ),
+        pytest.param(
+            _switch("s", state_topic="", command_topic=LONG_LITERAL),
+            None,
+            PLATFORM_ESP8266,
+            id="internal_by_empty_state_topic",
+        ),
+        pytest.param(
+            # The MAC suffix goes on the friendly name, never on a sub-device's name
+            {
+                CONF_ESPHOME: {
+                    CONF_NAME_ADD_MAC_SUFFIX: True,
+                    CONF_DEVICES: [{CONF_ID: ID("sub"), CONF_NAME: FITS}],
+                },
+                **_switch("", device_id=ID("sub")),
+            },
+            {"topic_prefix": "abc"},
+            PLATFORM_ESP8266,
+            id="sub_device_has_no_mac_suffix",
+        ),
+        pytest.param(
+            {},
+            {"on_message": [{"topic": LONG_LITERAL + "/#"}]},
+            PLATFORM_ESP8266,
+            id="wildcard",
+        ),
+    ],
+)
+def test_accepted(
+    monkeypatch: pytest.MonkeyPatch, full_config: dict, mqtt: dict | None, platform: str
+) -> None:
+    _run(monkeypatch, full_config, mqtt, platform)
+
+
+def test_message_names_the_topic_and_length(monkeypatch: pytest.MonkeyPatch) -> None:
+    with pytest.raises(
+        cv.Invalid, match=rf"'dev/switch/{TOO_LONG}/command' is 129 bytes"
+    ):
+        _run(monkeypatch, _switch(TOO_LONG))
+
+
+def test_object_id_is_capped_like_the_firmware(monkeypatch: pytest.MonkeyPatch) -> None:
+    capped = "a" * (OBJECT_ID_MAX_LEN - 1)
+    with pytest.raises(
+        cv.Invalid, match=rf"'dev/switch/{capped}/command' is 146 bytes"
+    ):
+        _run(monkeypatch, _switch("a" * 300))