Add initial MQTT subentry support for notify entities (#138461)

* Add initial MQTT subentry support for notify entities

* Fix componts assigment is reset on device config. Translation tweaks

* Rephrase

* Go to summary menu when components are set up already - add test

* Fix suggested device info on config flow

* Invert

* Simplify subentry config flow and omit menu

* Use constants instead of literals

* More constants

* Teak some translations

* Only show save when the the entry is dirty

* Do not trigger an entry reload twice

* Remove encoding, entity_category

* Remove icon from mqtt subentry flow

* Separate entity settings and MQTT specific settings

* Remove object_id and refactor

* Migrate translations

* Make subconfig flow test extensible

* Make sub reconfig flow tests extensible

* Rename entity_platform_config step to mqtt_platform_config

* Make component unique ID independent from the name

* Move code for update of component data to helper

* Follow up on code review

* Skip dirty stuff

* Fix rebase issues #1

* Do not allow reconfig for entity platform/name, default QoS and refactor tests

* Add entity platform and entity name label to basic entity config dialog

* Rename to exclude_from_reconfig and make reconfig option not optional

homeassistant/components/mqtt/__init__.py

@@ -13,7 +13,7 @@ import voluptuous as vol
 from homeassistant import config as conf_util
 from homeassistant.components import websocket_api
 from homeassistant.config_entries import ConfigEntry
-from homeassistant.const import CONF_DISCOVERY, SERVICE_RELOAD
+from homeassistant.const import CONF_DISCOVERY, CONF_PLATFORM, SERVICE_RELOAD
 from homeassistant.core import HomeAssistant, ServiceCall, callback
 from homeassistant.exceptions import (
     ConfigValidationError,
@@ -81,6 +81,7 @@ from .const import (
     ENTRY_OPTION_FIELDS,
     MQTT_CONNECTION_STATE,
     TEMPLATE_ERRORS,
+    Platform,
 )
 from .models import (
     DATA_MQTT,
@@ -293,6 +294,21 @@ async def async_check_config_schema(
                     ) from exc
 
 
+def _platforms_in_use(hass: HomeAssistant, entry: ConfigEntry) -> set[str | Platform]:
+    """Return a set of platforms in use."""
+    domains: set[str | Platform] = {
+        entry.domain
+        for entry in er.async_entries_for_config_entry(
+            er.async_get(hass), entry.entry_id
+        )
+    }
+    # Update with domains from subentries
+    for subentry in entry.subentries.values():
+        components = subentry.data["components"].values()
+        domains.update(component[CONF_PLATFORM] for component in components)
+    return domains
+
+
 async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
     """Set up the actions and websocket API for the MQTT component."""
 
@@ -434,12 +450,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
 
     mqtt_data, conf = await _setup_client()
     platforms_used = platforms_from_config(mqtt_data.config)
-    platforms_used.update(
+    platforms_used.update(_platforms_in_use(hass, entry))
-        entry.domain
-        for entry in er.async_entries_for_config_entry(
-            er.async_get(hass), entry.entry_id
-        )
-    )
     integration = async_get_loaded_integration(hass, DOMAIN)
     # Preload platforms we know we are going to use so
     # discovery can setup each platform synchronously

homeassistant/components/mqtt/config_flow.py

@@ -5,12 +5,15 @@ from __future__ import annotations
 import asyncio
 from collections import OrderedDict
 from collections.abc import Callable, Mapping
+from copy import deepcopy
+from dataclasses import dataclass
 from enum import IntEnum
 import logging
 import queue
 from ssl import PROTOCOL_TLS_CLIENT, SSLContext, SSLError
 from types import MappingProxyType
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, cast
+from uuid import uuid4
 
 from cryptography.hazmat.primitives.serialization import (
     Encoding,
@@ -29,21 +32,32 @@ from homeassistant.config_entries import (
     ConfigEntry,
     ConfigFlow,
     ConfigFlowResult,
+    ConfigSubentryFlow,
     OptionsFlow,
+    SubentryFlowResult,
 )
 from homeassistant.const import (
+    ATTR_CONFIGURATION_URL,
+    ATTR_HW_VERSION,
+    ATTR_MODEL,
+    ATTR_MODEL_ID,
+    ATTR_NAME,
+    ATTR_SW_VERSION,
     CONF_CLIENT_ID,
+    CONF_DEVICE,
     CONF_DISCOVERY,
     CONF_HOST,
+    CONF_NAME,
     CONF_PASSWORD,
     CONF_PAYLOAD,
+    CONF_PLATFORM,
     CONF_PORT,
     CONF_PROTOCOL,
     CONF_USERNAME,
 )
 from homeassistant.core import HomeAssistant, callback
 from homeassistant.data_entry_flow import AbortFlow
-from homeassistant.helpers import config_validation as cv
+from homeassistant.helpers import config_validation as cv, entity_registry as er
 from homeassistant.helpers.hassio import is_hassio
 from homeassistant.helpers.json import json_dumps
 from homeassistant.helpers.selector import (
@@ -54,9 +68,12 @@ from homeassistant.helpers.selector import (
     NumberSelectorConfig,
     NumberSelectorMode,
     SelectOptionDict,
+    Selector,
     SelectSelector,
     SelectSelectorConfig,
     SelectSelectorMode,
+    TemplateSelector,
+    TemplateSelectorConfig,
     TextSelector,
     TextSelectorConfig,
     TextSelectorType,
@@ -76,8 +93,13 @@ from .const import (
     CONF_CERTIFICATE,
     CONF_CLIENT_CERT,
     CONF_CLIENT_KEY,
+    CONF_COMMAND_TEMPLATE,
+    CONF_COMMAND_TOPIC,
     CONF_DISCOVERY_PREFIX,
+    CONF_ENTITY_PICTURE,
     CONF_KEEPALIVE,
+    CONF_QOS,
+    CONF_RETAIN,
     CONF_TLS_INSECURE,
     CONF_TRANSPORT,
     CONF_WILL_MESSAGE,
@@ -99,12 +121,15 @@ from .const import (
     SUPPORTED_PROTOCOLS,
     TRANSPORT_TCP,
     TRANSPORT_WEBSOCKETS,
+    Platform,
 )
+from .models import MqttDeviceData, MqttSubentryData
 from .util import (
     async_create_certificate_temp_files,
     get_file_path,
     valid_birth_will,
     valid_publish_topic,
+    valid_qos_schema,
 )
 
 _LOGGER = logging.getLogger(__name__)
@@ -128,10 +153,10 @@ PORT_SELECTOR = vol.All(
     vol.Coerce(int),
 )
 PASSWORD_SELECTOR = TextSelector(TextSelectorConfig(type=TextSelectorType.PASSWORD))
-QOS_SELECTOR = vol.All(
+QOS_SELECTOR = NumberSelector(
-    NumberSelector(NumberSelectorConfig(mode=NumberSelectorMode.BOX, min=0, max=2)),
+    NumberSelectorConfig(mode=NumberSelectorMode.BOX, min=0, max=2)
-    vol.Coerce(int),
 )
+QOS_DATA_SCHEMA = vol.All(QOS_SELECTOR, valid_qos_schema)
 KEEPALIVE_SELECTOR = vol.All(
     NumberSelector(
         NumberSelectorConfig(
@@ -183,6 +208,65 @@ KEY_UPLOAD_SELECTOR = FileSelector(
     FileSelectorConfig(accept=".pem,.key,.der,.pk8,application/pkcs8")
 )
 
+# Subentry selectors
+SUBENTRY_PLATFORMS = [Platform.NOTIFY]
+SUBENTRY_PLATFORM_SELECTOR = SelectSelector(
+    SelectSelectorConfig(
+        options=[platform.value for platform in SUBENTRY_PLATFORMS],
+        mode=SelectSelectorMode.DROPDOWN,
+        translation_key=CONF_PLATFORM,
+    )
+)
+
+TEMPLATE_SELECTOR = TemplateSelector(TemplateSelectorConfig())
+
+
+@dataclass(frozen=True)
+class PlatformField:
+    """Stores a platform config field schema, required flag and validator."""
+
+    selector: Selector
+    required: bool
+    validator: Callable[..., Any]
+    error: str | None = None
+    default: str | int | vol.Undefined = vol.UNDEFINED
+    exclude_from_reconfig: bool = False
+
+
+COMMON_ENTITY_FIELDS = {
+    CONF_PLATFORM: PlatformField(
+        SUBENTRY_PLATFORM_SELECTOR, True, str, exclude_from_reconfig=True
+    ),
+    CONF_NAME: PlatformField(TEXT_SELECTOR, False, str, exclude_from_reconfig=True),
+    CONF_ENTITY_PICTURE: PlatformField(TEXT_SELECTOR, False, cv.url, "invalid_url"),
+}
+
+COMMON_MQTT_FIELDS = {
+    CONF_QOS: PlatformField(QOS_SELECTOR, False, valid_qos_schema, default=0),
+    CONF_RETAIN: PlatformField(BOOLEAN_SELECTOR, False, bool),
+}
+PLATFORM_MQTT_FIELDS = {
+    Platform.NOTIFY.value: {
+        CONF_COMMAND_TOPIC: PlatformField(
+            TEXT_SELECTOR, True, valid_publish_topic, "invalid_publish_topic"
+        ),
+        CONF_COMMAND_TEMPLATE: PlatformField(
+            TEMPLATE_SELECTOR, False, cv.template, "invalid_template"
+        ),
+    },
+}
+
+MQTT_DEVICE_SCHEMA = vol.Schema(
+    {
+        vol.Required(ATTR_NAME): TEXT_SELECTOR,
+        vol.Optional(ATTR_SW_VERSION): TEXT_SELECTOR,
+        vol.Optional(ATTR_HW_VERSION): TEXT_SELECTOR,
+        vol.Optional(ATTR_MODEL): TEXT_SELECTOR,
+        vol.Optional(ATTR_MODEL_ID): TEXT_SELECTOR,
+        vol.Optional(ATTR_CONFIGURATION_URL): TEXT_SELECTOR,
+    }
+)
+
 REAUTH_SCHEMA = vol.Schema(
     {
         vol.Required(CONF_USERNAME): TEXT_SELECTOR,
@@ -215,6 +299,57 @@ def update_password_from_user_input(
     return substituted_used_data
 
 
+@callback
+def validate_field(
+    field: str,
+    validator: Callable[..., Any],
+    user_input: dict[str, Any] | None,
+    errors: dict[str, str],
+    error: str,
+) -> None:
+    """Validate a single field."""
+    if user_input is None or field not in user_input:
+        return
+    try:
+        validator(user_input[field])
+    except (ValueError, vol.Invalid):
+        errors[field] = error
+
+
+@callback
+def validate_user_input(
+    user_input: dict[str, Any],
+    data_schema_fields: dict[str, PlatformField],
+    errors: dict[str, str],
+) -> None:
+    """Validate user input."""
+    for field, value in user_input.items():
+        validator = data_schema_fields[field].validator
+        try:
+            validator(value)
+        except (ValueError, vol.Invalid):
+            errors[field] = data_schema_fields[field].error or "invalid_input"
+
+
+@callback
+def data_schema_from_fields(
+    data_schema_fields: dict[str, PlatformField],
+    reconfig: bool,
+) -> vol.Schema:
+    """Generate data schema from platform fields."""
+    return vol.Schema(
+        {
+            vol.Required(field_name, default=field_details.default)
+            if field_details.required
+            else vol.Optional(
+                field_name, default=field_details.default
+            ): field_details.selector
+            for field_name, field_details in data_schema_fields.items()
+            if not field_details.exclude_from_reconfig or not reconfig
+        }
+    )
+
+
 class FlowHandler(ConfigFlow, domain=DOMAIN):
     """Handle a config flow."""
 
@@ -230,6 +365,14 @@ class FlowHandler(ConfigFlow, domain=DOMAIN):
         self.install_task: asyncio.Task | None = None
         self.start_task: asyncio.Task | None = None
 
+    @classmethod
+    @callback
+    def async_get_supported_subentry_types(
+        cls, config_entry: ConfigEntry
+    ) -> dict[str, type[ConfigSubentryFlow]]:
+        """Return subentries supported by this handler."""
+        return {CONF_DEVICE: MQTTSubentryFlowHandler}
+
     @staticmethod
     @callback
     def async_get_options_flow(
@@ -685,7 +828,7 @@ class MQTTOptionsFlowHandler(OptionsFlow):
                 "birth_payload", description={"suggested_value": birth[CONF_PAYLOAD]}
             )
         ] = TEXT_SELECTOR
-        fields[vol.Optional("birth_qos", default=birth[ATTR_QOS])] = QOS_SELECTOR
+        fields[vol.Optional("birth_qos", default=birth[ATTR_QOS])] = QOS_DATA_SCHEMA
         fields[vol.Optional("birth_retain", default=birth[ATTR_RETAIN])] = (
             BOOLEAN_SELECTOR
         )
@@ -708,7 +851,7 @@ class MQTTOptionsFlowHandler(OptionsFlow):
                 "will_payload", description={"suggested_value": will[CONF_PAYLOAD]}
             )
         ] = TEXT_SELECTOR
-        fields[vol.Optional("will_qos", default=will[ATTR_QOS])] = QOS_SELECTOR
+        fields[vol.Optional("will_qos", default=will[ATTR_QOS])] = QOS_DATA_SCHEMA
         fields[vol.Optional("will_retain", default=will[ATTR_RETAIN])] = (
             BOOLEAN_SELECTOR
         )
@@ -721,6 +864,288 @@ class MQTTOptionsFlowHandler(OptionsFlow):
         )
 
 
+class MQTTSubentryFlowHandler(ConfigSubentryFlow):
+    """Handle MQTT subentry flow."""
+
+    _subentry_data: MqttSubentryData
+    _component_id: str | None = None
+
+    @callback
+    def update_component_fields(
+        self, data_schema: vol.Schema, user_input: dict[str, Any]
+    ) -> None:
+        """Update the componment fields."""
+        if TYPE_CHECKING:
+            assert self._component_id is not None
+        component_data = self._subentry_data["components"][self._component_id]
+        # Remove the fields from the component data if they are not in the user input
+        for field in [
+            form_field
+            for form_field in data_schema.schema
+            if form_field in component_data and form_field not in user_input
+        ]:
+            component_data.pop(field)
+        component_data.update(user_input)
+
+    async def async_step_user(
+        self, user_input: dict[str, Any] | None = None
+    ) -> SubentryFlowResult:
+        """Add a subentry."""
+        self._subentry_data = MqttSubentryData(device=MqttDeviceData(), components={})
+        return await self.async_step_device()
+
+    async def async_step_reconfigure(
+        self, user_input: dict[str, Any] | None = None
+    ) -> SubentryFlowResult:
+        """Reconfigure a subentry."""
+        reconfigure_subentry = self._get_reconfigure_subentry()
+        self._subentry_data = cast(
+            MqttSubentryData, deepcopy(dict(reconfigure_subentry.data))
+        )
+        return await self.async_step_summary_menu()
+
+    async def async_step_device(
+        self, user_input: dict[str, Any] | None = None
+    ) -> SubentryFlowResult:
+        """Add a new MQTT device."""
+        errors: dict[str, str] = {}
+        validate_field("configuration_url", cv.url, user_input, errors, "invalid_url")
+        if not errors and user_input is not None:
+            self._subentry_data[CONF_DEVICE] = cast(MqttDeviceData, user_input)
+            if self.source == SOURCE_RECONFIGURE:
+                return await self.async_step_summary_menu()
+            return await self.async_step_entity()
+
+        data_schema = self.add_suggested_values_to_schema(
+            MQTT_DEVICE_SCHEMA,
+            self._subentry_data[CONF_DEVICE] if user_input is None else user_input,
+        )
+        return self.async_show_form(
+            step_id=CONF_DEVICE,
+            data_schema=data_schema,
+            errors=errors,
+            last_step=False,
+        )
+
+    async def async_step_entity(
+        self, user_input: dict[str, Any] | None = None
+    ) -> SubentryFlowResult:
+        """Add or edit an mqtt entity."""
+        errors: dict[str, str] = {}
+        data_schema_fields = COMMON_ENTITY_FIELDS
+        entity_name_label: str = ""
+        platform_label: str = ""
+        if reconfig := (self._component_id is not None):
+            name: str | None = self._subentry_data["components"][
+                self._component_id
+            ].get(CONF_NAME)
+            platform_label = f"{self._subentry_data['components'][self._component_id][CONF_PLATFORM]} "
+            entity_name_label = f" ({name})" if name is not None else ""
+        data_schema = data_schema_from_fields(data_schema_fields, reconfig=reconfig)
+        if user_input is not None:
+            validate_user_input(user_input, data_schema_fields, errors)
+            if not errors:
+                if self._component_id is None:
+                    self._component_id = uuid4().hex
+                self._subentry_data["components"].setdefault(self._component_id, {})
+                self.update_component_fields(data_schema, user_input)
+                return await self.async_step_mqtt_platform_config()
+            data_schema = self.add_suggested_values_to_schema(data_schema, user_input)
+        elif self.source == SOURCE_RECONFIGURE and self._component_id is not None:
+            data_schema = self.add_suggested_values_to_schema(
+                data_schema, self._subentry_data["components"][self._component_id]
+            )
+        device_name = self._subentry_data[CONF_DEVICE][CONF_NAME]
+        return self.async_show_form(
+            step_id="entity",
+            data_schema=data_schema,
+            description_placeholders={
+                "mqtt_device": device_name,
+                "entity_name_label": entity_name_label,
+                "platform_label": platform_label,
+            },
+            errors=errors,
+            last_step=False,
+        )
+
+    def _show_update_or_delete_form(self, step_id: str) -> SubentryFlowResult:
+        """Help selecting an entity to update or delete."""
+        device_name = self._subentry_data[CONF_DEVICE][CONF_NAME]
+        entities = [
+            SelectOptionDict(
+                value=key, label=f"{device_name} {component.get(CONF_NAME, '-')}"
+            )
+            for key, component in self._subentry_data["components"].items()
+        ]
+        data_schema = vol.Schema(
+            {
+                vol.Required("component"): SelectSelector(
+                    SelectSelectorConfig(
+                        options=entities,
+                        mode=SelectSelectorMode.LIST,
+                    )
+                )
+            }
+        )
+        return self.async_show_form(
+            step_id=step_id, data_schema=data_schema, last_step=False
+        )
+
+    async def async_step_update_entity(
+        self, user_input: dict[str, Any] | None = None
+    ) -> SubentryFlowResult:
+        """Select the entity to update."""
+        if user_input:
+            self._component_id = user_input["component"]
+            return await self.async_step_entity()
+        if len(self._subentry_data["components"]) == 1:
+            # Return first key
+            self._component_id = next(iter(self._subentry_data["components"]))
+            return await self.async_step_entity()
+        return self._show_update_or_delete_form("update_entity")
+
+    async def async_step_delete_entity(
+        self, user_input: dict[str, Any] | None = None
+    ) -> SubentryFlowResult:
+        """Select the entity to delete."""
+        if user_input:
+            del self._subentry_data["components"][user_input["component"]]
+            return await self.async_step_summary_menu()
+        return self._show_update_or_delete_form("delete_entity")
+
+    async def async_step_mqtt_platform_config(
+        self, user_input: dict[str, Any] | None = None
+    ) -> SubentryFlowResult:
+        """Configure entity platform MQTT details."""
+        errors: dict[str, str] = {}
+        if TYPE_CHECKING:
+            assert self._component_id is not None
+        platform = self._subentry_data["components"][self._component_id][CONF_PLATFORM]
+        data_schema_fields = PLATFORM_MQTT_FIELDS[platform] | COMMON_MQTT_FIELDS
+        data_schema = data_schema_from_fields(
+            data_schema_fields, reconfig=self._component_id is not None
+        )
+        if user_input is not None:
+            # Test entity fields against the validator
+            validate_user_input(user_input, data_schema_fields, errors)
+            if not errors:
+                self.update_component_fields(data_schema, user_input)
+                self._component_id = None
+                if self.source == SOURCE_RECONFIGURE:
+                    return await self.async_step_summary_menu()
+                return self._async_create_subentry()
+
+            data_schema = self.add_suggested_values_to_schema(data_schema, user_input)
+        else:
+            data_schema = self.add_suggested_values_to_schema(
+                data_schema, self._subentry_data["components"][self._component_id]
+            )
+        device_name = self._subentry_data[CONF_DEVICE][CONF_NAME]
+        entity_name: str | None
+        if entity_name := self._subentry_data["components"][self._component_id].get(
+            CONF_NAME
+        ):
+            full_entity_name: str = f"{device_name} {entity_name}"
+        else:
+            full_entity_name = device_name
+        return self.async_show_form(
+            step_id="mqtt_platform_config",
+            data_schema=data_schema,
+            description_placeholders={
+                "mqtt_device": device_name,
+                CONF_PLATFORM: platform,
+                "entity": full_entity_name,
+            },
+            errors=errors,
+            last_step=False,
+        )
+
+    @callback
+    def _async_create_subentry(
+        self, user_input: dict[str, Any] | None = None
+    ) -> SubentryFlowResult:
+        """Create a subentry for a new MQTT device."""
+        device_name = self._subentry_data[CONF_DEVICE][CONF_NAME]
+        component: dict[str, Any] = next(
+            iter(self._subentry_data["components"].values())
+        )
+        platform = component[CONF_PLATFORM]
+        entity_name: str | None
+        if entity_name := component.get(CONF_NAME):
+            full_entity_name: str = f"{device_name} {entity_name}"
+        else:
+            full_entity_name = device_name
+
+        return self.async_create_entry(
+            data=self._subentry_data,
+            title=self._subentry_data[CONF_DEVICE][CONF_NAME],
+            description_placeholders={
+                "entity": full_entity_name,
+                CONF_PLATFORM: platform,
+            },
+        )
+
+    async def async_step_summary_menu(
+        self, user_input: dict[str, Any] | None = None
+    ) -> SubentryFlowResult:
+        """Show summary menu and decide to add more entities or to finish the flow."""
+        self._component_id = None
+        mqtt_device = self._subentry_data[CONF_DEVICE][CONF_NAME]
+        mqtt_items = ", ".join(
+            f"{mqtt_device} {component.get(CONF_NAME, '-')}"
+            for component in self._subentry_data["components"].values()
+        )
+        menu_options = [
+            "entity",
+            "update_entity",
+        ]
+        if len(self._subentry_data["components"]) > 1:
+            menu_options.append("delete_entity")
+        menu_options.append("device")
+        if self._subentry_data != self._get_reconfigure_subentry().data:
+            menu_options.append("save_changes")
+        return self.async_show_menu(
+            step_id="summary_menu",
+            menu_options=menu_options,
+            description_placeholders={
+                "mqtt_device": mqtt_device,
+                "mqtt_items": mqtt_items,
+            },
+        )
+
+    async def async_step_save_changes(
+        self, user_input: dict[str, Any] | None = None
+    ) -> SubentryFlowResult:
+        """Save the changes made to the subentry."""
+        entry = self._get_reconfigure_entry()
+        subentry = self._get_reconfigure_subentry()
+        entity_registry = er.async_get(self.hass)
+
+        # When a component is removed from the MQTT device,
+        # And we save the changes to the subentry,
+        # we need to clean up stale entity registry entries.
+        # The component id is used as a part of the unique id of the entity.
+        for unique_id, platform in [
+            (
+                f"{subentry.subentry_id}_{component_id}",
+                subentry.data["components"][component_id][CONF_PLATFORM],
+            )
+            for component_id in subentry.data["components"]
+            if component_id not in self._subentry_data["components"]
+        ]:
+            if entity_id := entity_registry.async_get_entity_id(
+                platform, DOMAIN, unique_id
+            ):
+                entity_registry.async_remove(entity_id)
+
+        return self.async_update_and_abort(
+            entry,
+            subentry,
+            data=self._subentry_data,
+            title=self._subentry_data[CONF_DEVICE][CONF_NAME],
+        )
+
+
 @callback
 def async_is_pem_data(data: bytes) -> bool:
     """Return True if data is in PEM format."""

homeassistant/components/mqtt/entity.py

@@ -43,7 +43,7 @@ from homeassistant.helpers.dispatcher import (
     async_dispatcher_send,
 )
 from homeassistant.helpers.entity import Entity, async_generate_entity_id
-from homeassistant.helpers.entity_platform import AddEntitiesCallback
+from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
 from homeassistant.helpers.event import (
     async_track_device_registry_updated_event,
     async_track_entity_registry_updated_event,
@@ -111,6 +111,7 @@ from .discovery import (
 from .models import (
     DATA_MQTT,
     MessageCallbackType,
+    MqttSubentryData,
     MqttValueTemplate,
     MqttValueTemplateException,
     PublishPayloadType,
@@ -238,7 +239,7 @@ def async_setup_entity_entry_helper(
     entry: ConfigEntry,
     entity_class: type[MqttEntity] | None,
     domain: str,
-    async_add_entities: AddEntitiesCallback,
+    async_add_entities: AddConfigEntryEntitiesCallback,
     discovery_schema: VolSchemaType,
     platform_schema_modern: VolSchemaType,
     schema_class_mapping: dict[str, type[MqttEntity]] | None = None,
@@ -282,11 +283,10 @@ def async_setup_entity_entry_helper(
 
     @callback
     def _async_setup_entities() -> None:
-        """Set up MQTT items from configuration.yaml."""
+        """Set up MQTT items from subentries and configuration.yaml."""
         nonlocal entity_class
         mqtt_data = hass.data[DATA_MQTT]
-        if not (config_yaml := mqtt_data.config):
+        config_yaml = mqtt_data.config
-            return
         yaml_configs: list[ConfigType] = [
             config
             for config_item in config_yaml
@@ -294,6 +294,41 @@ def async_setup_entity_entry_helper(
             for config in configs
             if config_domain == domain
         ]
+        # process subentry entity setup
+        for config_subentry_id, subentry in entry.subentries.items():
+            subentry_data = cast(MqttSubentryData, subentry.data)
+            subentry_entities: list[Entity] = []
+            device_config = subentry_data["device"].copy()
+            device_config["identifiers"] = config_subentry_id
+            for component_id, component_data in subentry_data["components"].items():
+                if component_data["platform"] != domain:
+                    continue
+                component_config: dict[str, Any] = component_data.copy()
+                component_config[CONF_UNIQUE_ID] = (
+                    f"{config_subentry_id}_{component_id}"
+                )
+                component_config[CONF_DEVICE] = device_config
+                component_config.pop("platform")
+
+                try:
+                    config = platform_schema_modern(component_config)
+                    if schema_class_mapping is not None:
+                        entity_class = schema_class_mapping[config[CONF_SCHEMA]]
+                    if TYPE_CHECKING:
+                        assert entity_class is not None
+                    subentry_entities.append(entity_class(hass, config, entry, None))
+                except vol.Invalid as exc:
+                    _LOGGER.error(
+                        "Schema violation occurred when trying to set up "
+                        "entity from subentry %s %s %s: %s",
+                        config_subentry_id,
+                        subentry.title,
+                        subentry.data,
+                        exc,
+                    )
+
+            async_add_entities(subentry_entities, config_subentry_id=config_subentry_id)
+
         entities: list[Entity] = []
         for yaml_config in yaml_configs:
             try:

homeassistant/components/mqtt/models.py

@@ -420,5 +420,24 @@ class MqttComponentConfig:
     discovery_payload: MQTTDiscoveryPayload
 
 
+class MqttDeviceData(TypedDict, total=False):
+    """Hold the data for an MQTT device."""
+
+    name: str
+    identifiers: str
+    configuration_url: str
+    sw_version: str
+    hw_version: str
+    model: str
+    model_id: str
+
+
+class MqttSubentryData(TypedDict):
+    """Hold the data for a MQTT subentry."""
+
+    device: MqttDeviceData
+    components: dict[str, dict[str, Any]]
+
+
 DATA_MQTT: HassKey[MqttData] = HassKey("mqtt")
 DATA_MQTT_AVAILABLE: HassKey[asyncio.Future[bool]] = HassKey("mqtt_client_available")

homeassistant/components/mqtt/strings.json

@@ -108,6 +108,110 @@
       "invalid_inclusion": "The client certificate and private key must be configured together"
     }
   },
+  "config_subentries": {
+    "device": {
+      "initiate_flow": {
+        "user": "Add MQTT Device",
+        "reconfigure": "Reconfigure MQTT Device"
+      },
+      "entry_type": "MQTT Device",
+      "step": {
+        "device": {
+          "title": "Configure MQTT device details",
+          "description": "Enter the MQTT device details:",
+          "data": {
+            "name": "[%key:common::config_flow::data::name%]",
+            "configuration_url": "Configuration URL",
+            "sw_version": "Software version",
+            "hw_version": "Hardware version",
+            "model": "Model",
+            "model_id": "Model ID"
+          },
+          "data_description": {
+            "name": "The name of the manually added MQTT device.",
+            "configuration_url": "A link to the webpage that can manage the configuration of this device. Can be either a 'http://', 'https://' or an internal 'homeassistant://' URL.",
+            "sw_version": "The software version of the device. E.g. '2025.1.0'.",
+            "hw_version": "The hardware version of the device. E.g. 'v1.0 rev a'.",
+            "model": "E.g. 'Cleanmaster Pro'.",
+            "model_id": "E.g. '123NK2PRO'."
+          }
+        },
+        "summary_menu": {
+          "title": "Reconfigure \"{mqtt_device}\"",
+          "description": "Entities set up:\n{mqtt_items}\n\nDecide what to do next:",
+          "menu_options": {
+            "entity": "Add another entity to \"{mqtt_device}\"",
+            "update_entity": "Update entity properties",
+            "delete_entity": "Delete an entity",
+            "device": "Update device properties",
+            "save_changes": "Save changes"
+          }
+        },
+        "entity": {
+          "title": "Configure MQTT device \"{mqtt_device}\"",
+          "description": "Configure the basic {platform_label}entity settings{entity_name_label}",
+          "data": {
+            "platform": "Type of entity",
+            "name": "Entity name",
+            "entity_picture": "Entity picture"
+          },
+          "data_description": {
+            "platform": "The type of the entity to configure.",
+            "name": "The name of the entity. Leave empty to set it to `None` to [mark it as main feature of the MQTT device](https://www.home-assistant.io/integrations/mqtt/#naming-of-mqtt-entities).",
+            "entity_picture": "An URL to a picture to be assigned."
+          }
+        },
+        "delete_entity": {
+          "title": "Delete entity",
+          "description": "Delete an entity. The entity will be removed from the device. Removing an entity will break any automations or scripts that depend on it.",
+          "data": {
+            "component": "Entity"
+          },
+          "data_description": {
+            "component": "Select the entity you want to delete. Minimal one entity is required."
+          }
+        },
+        "update_entity": {
+          "title": "Select entity",
+          "description": "Select the entity you want to update",
+          "data": {
+            "component": "Entity"
+          },
+          "data_description": {
+            "component": "Select the entity you want to update."
+          }
+        },
+        "mqtt_platform_config": {
+          "title": "Configure MQTT device \"{mqtt_device}\"",
+          "description": "Please configure MQTT specific details for {platform} entity \"{entity}\":",
+          "data": {
+            "command_topic": "Command topic",
+            "command_template": "Command template",
+            "retain": "Retain",
+            "qos": "QoS"
+          },
+          "data_description": {
+            "command_topic": "The publishing topic that will be used to control the {platform} entity.",
+            "command_template": "A [template](https://www.home-assistant.io/docs/configuration/templating/#using-command-templates-with-mqtt) to render the payload to be published at the command topic.",
+            "retain": "Select if values published by the {platform} entity should be retained at the MQTT broker.",
+            "qos": "The QoS value {platform} entity should use."
+          }
+        }
+      },
+      "abort": {
+        "reconfigure_successful": "[%key:common::config_flow::abort::reconfigure_successful%]"
+      },
+      "create_entry": {
+        "default": "MQTT device with {platform} entity \"{entity}\" was set up successfully.\n\nNote that you can reconfigure the MQTT device at any time, e.g. to add more entities."
+      },
+      "error": {
+        "invalid_input": "Invalid value",
+        "invalid_subscribe_topic": "Invalid subscribe topic",
+        "invalid_template": "Invalid template",
+        "invalid_url": "Invalid URL"
+      }
+    }
+  },
   "device_automation": {
     "trigger_type": {
       "button_short_press": "\"{subtype}\" pressed",
@@ -221,6 +325,11 @@
     }
   },
   "selector": {
+    "platform": {
+      "options": {
+        "notify": "Notify"
+      }
+    },
     "set_ca_cert": {
       "options": {
         "off": "[%key:common::state::off%]",

tests/components/mqtt/common.py

@@ -66,6 +66,118 @@ DEFAULT_CONFIG_DEVICE_INFO_MAC = {
     "configuration_url": "http://example.com",
 }
 
+MOCK_SUBENTRY_NOTIFY_COMPONENT1 = {
+    "363a7ecad6be4a19b939a016ea93e994": {
+        "platform": "notify",
+        "name": "Milkman alert",
+        "qos": 0,
+        "command_topic": "test-topic",
+        "command_template": "{{ value_json.value }}",
+        "entity_picture": "https://example.com/363a7ecad6be4a19b939a016ea93e994",
+        "retain": False,
+    },
+}
+MOCK_SUBENTRY_NOTIFY_COMPONENT2 = {
+    "6494827dac294fa0827c54b02459d309": {
+        "platform": "notify",
+        "name": "The second notifier",
+        "qos": 0,
+        "command_topic": "test-topic2",
+        "entity_picture": "https://example.com/6494827dac294fa0827c54b02459d309",
+    },
+}
+MOCK_SUBENTRY_NOTIFY_COMPONENT_NO_NAME = {
+    "5269352dd9534c908d22812ea5d714cd": {
+        "platform": "notify",
+        "qos": 0,
+        "command_topic": "test-topic",
+        "command_template": "{{ value_json.value }}",
+        "entity_picture": "https://example.com/5269352dd9534c908d22812ea5d714cd",
+        "retain": False,
+    },
+}
+
+# Bogus light component just for code coverage
+# Note that light cannot be setup through the UI yet
+# The test is for code coverage
+MOCK_SUBENTRY_LIGHT_COMPONENT = {
+    "8131babc5e8d4f44b82e0761d39091a2": {
+        "platform": "light",
+        "name": "Test light",
+        "qos": 1,
+        "command_topic": "test-topic4",
+        "schema": "basic",
+        "entity_picture": "https://example.com/8131babc5e8d4f44b82e0761d39091a2",
+    },
+}
+MOCK_SUBENTRY_NOTIFY_BAD_SCHEMA = {
+    "b10b531e15244425a74bb0abb1e9d2c6": {
+        "platform": "notify",
+        "name": "Test",
+        "qos": 1,
+        "command_topic": "bad#topic",
+    },
+}
+
+MOCK_NOTIFY_SUBENTRY_DATA_MULTI = {
+    "device": {
+        "name": "Milk notifier",
+        "sw_version": "1.0",
+        "hw_version": "2.1 rev a",
+        "model": "Model XL",
+        "model_id": "mn002",
+        "configuration_url": "https://example.com",
+    },
+    "components": MOCK_SUBENTRY_NOTIFY_COMPONENT1 | MOCK_SUBENTRY_NOTIFY_COMPONENT2,
+}
+
+MOCK_NOTIFY_SUBENTRY_DATA_SINGLE = {
+    "device": {
+        "name": "Milk notifier",
+        "sw_version": "1.0",
+        "hw_version": "2.1 rev a",
+        "model": "Model XL",
+        "model_id": "mn002",
+        "configuration_url": "https://example.com",
+    },
+    "components": MOCK_SUBENTRY_NOTIFY_COMPONENT1,
+}
+MOCK_SUBENTRY_DATA_NOTIFY_NO_NAME = {
+    "device": {
+        "name": "Milk notifier",
+        "sw_version": "1.0",
+        "hw_version": "2.1 rev a",
+        "model": "Model XL",
+        "model_id": "mn002",
+        "configuration_url": "https://example.com",
+    },
+    "components": MOCK_SUBENTRY_NOTIFY_COMPONENT_NO_NAME,
+}
+
+MOCK_SUBENTRY_DATA_BAD_COMPONENT_SCHEMA = {
+    "device": {
+        "name": "Milk notifier",
+        "sw_version": "1.0",
+        "hw_version": "2.1 rev a",
+        "model": "Model XL",
+        "model_id": "mn002",
+        "configuration_url": "https://example.com",
+    },
+    "components": MOCK_SUBENTRY_NOTIFY_BAD_SCHEMA,
+}
+MOCK_SUBENTRY_DATA_SET_MIX = {
+    "device": {
+        "name": "Milk notifier",
+        "sw_version": "1.0",
+        "hw_version": "2.1 rev a",
+        "model": "Model XL",
+        "model_id": "mn002",
+        "configuration_url": "https://example.com",
+    },
+    "components": MOCK_SUBENTRY_NOTIFY_COMPONENT1
+    | MOCK_SUBENTRY_NOTIFY_COMPONENT2
+    | MOCK_SUBENTRY_LIGHT_COMPONENT,
+}
 _SENTINEL = object()
 
 DISCOVERY_COUNT = sum(len(discovery_topic) for discovery_topic in MQTT.values())

tests/components/mqtt/test_config_flow.py

@@ -2,6 +2,7 @@
 
 from collections.abc import Generator, Iterator
 from contextlib import contextmanager
+from copy import deepcopy
 from pathlib import Path
 from ssl import SSLError
 from typing import Any
@@ -17,6 +18,7 @@ from homeassistant import config_entries
 from homeassistant.components import mqtt
 from homeassistant.components.hassio import AddonError
 from homeassistant.components.mqtt.config_flow import PWD_NOT_CHANGED
+from homeassistant.config_entries import ConfigSubentry, ConfigSubentryData
 from homeassistant.const import (
     CONF_CLIENT_ID,
     CONF_PASSWORD,
@@ -26,8 +28,15 @@ from homeassistant.const import (
 )
 from homeassistant.core import HomeAssistant
 from homeassistant.data_entry_flow import FlowResultType
+from homeassistant.helpers import device_registry as dr, entity_registry as er
 from homeassistant.helpers.service_info.hassio import HassioServiceInfo
 
+from .common import (
+    MOCK_NOTIFY_SUBENTRY_DATA_MULTI,
+    MOCK_NOTIFY_SUBENTRY_DATA_SINGLE,
+    MOCK_SUBENTRY_DATA_NOTIFY_NO_NAME,
+)
+
 from tests.common import MockConfigEntry, MockMqttReasonCode
 from tests.typing import MqttMockHAClientGenerator, MqttMockPahoClient
 
@@ -2598,3 +2607,706 @@ async def test_migrate_of_incompatible_config_entry(
         await mqtt_mock_entry()
 
     assert config_entry.state is config_entries.ConfigEntryState.MIGRATION_ERROR
+
+
+@pytest.mark.parametrize(
+    (
+        "config_subentries_data",
+        "mock_entity_user_input",
+        "mock_mqtt_user_input",
+        "mock_failed_mqtt_user_input",
+        "mock_failed_mqtt_user_input_errors",
+        "entity_name",
+    ),
+    [
+        (
+            MOCK_NOTIFY_SUBENTRY_DATA_SINGLE,
+            {"name": "Milkman alert"},
+            {
+                "command_topic": "test-topic",
+                "command_template": "{{ value_json.value }}",
+                "qos": 0,
+                "retain": False,
+            },
+            {"command_topic": "test-topic#invalid"},
+            {"command_topic": "invalid_publish_topic"},
+            "Milk notifier Milkman alert",
+        ),
+        (
+            MOCK_SUBENTRY_DATA_NOTIFY_NO_NAME,
+            {},
+            {
+                "command_topic": "test-topic",
+                "command_template": "{{ value_json.value }}",
+                "qos": 0,
+                "retain": False,
+            },
+            {"command_topic": "test-topic#invalid"},
+            {"command_topic": "invalid_publish_topic"},
+            "Milk notifier",
+        ),
+    ],
+    ids=["notify_with_entity_name", "notify_no_entity_name"],
+)
+async def test_subentry_configflow(
+    hass: HomeAssistant,
+    mqtt_mock_entry: MqttMockHAClientGenerator,
+    config_subentries_data: dict[str, Any],
+    mock_entity_user_input: dict[str, Any],
+    mock_mqtt_user_input: dict[str, Any],
+    mock_failed_mqtt_user_input: dict[str, Any],
+    mock_failed_mqtt_user_input_errors: dict[str, Any],
+    entity_name: str,
+) -> None:
+    """Test the subentry ConfigFlow."""
+    device_name = config_subentries_data["device"]["name"]
+    component = next(iter(config_subentries_data["components"].values()))
+
+    await mqtt_mock_entry()
+    config_entry = hass.config_entries.async_entries(mqtt.DOMAIN)[0]
+
+    result = await hass.config_entries.subentries.async_init(
+        (config_entry.entry_id, "device"),
+        context={"source": config_entries.SOURCE_USER},
+    )
+    assert result["type"] is FlowResultType.FORM
+    assert result["step_id"] == "device"
+
+    # Test the URL validation
+    result = await hass.config_entries.subentries.async_configure(
+        result["flow_id"],
+        user_input={
+            "name": device_name,
+            "configuration_url": "http:/badurl.example.com",
+        },
+    )
+    assert result["type"] is FlowResultType.FORM
+    assert result["step_id"] == "device"
+    assert result["errors"]["configuration_url"] == "invalid_url"
+
+    result = await hass.config_entries.subentries.async_configure(
+        result["flow_id"],
+        user_input={
+            "name": device_name,
+            "sw_version": "1.0",
+            "hw_version": "2.1 rev a",
+            "model": "Model XL",
+            "model_id": "mn002",
+            "configuration_url": "https://example.com",
+        },
+    )
+    assert result["type"] is FlowResultType.FORM
+    assert result["step_id"] == "entity"
+    assert result["errors"] == {}
+
+    # Process entity flow (initial step)
+
+    # Test the entity picture URL validation
+    result = await hass.config_entries.subentries.async_configure(
+        result["flow_id"],
+        user_input={
+            "platform": component["platform"],
+            "entity_picture": "invalid url",
+        }
+        | mock_entity_user_input,
+    )
+    assert result["type"] is FlowResultType.FORM
+    assert result["step_id"] == "entity"
+
+    # Try again with valid data
+    result = await hass.config_entries.subentries.async_configure(
+        result["flow_id"],
+        user_input={
+            "platform": component["platform"],
+            "entity_picture": component["entity_picture"],
+        }
+        | mock_entity_user_input,
+    )
+    assert result["type"] is FlowResultType.FORM
+    assert result["step_id"] == "mqtt_platform_config"
+    assert result["errors"] == {}
+    assert result["description_placeholders"] == {
+        "mqtt_device": "Milk notifier",
+        "platform": "notify",
+        "entity": entity_name,
+    }
+
+    # Process entity platform config flow
+
+    # Test an invalid mqtt user_input case
+    result = await hass.config_entries.subentries.async_configure(
+        result["flow_id"],
+        user_input=mock_failed_mqtt_user_input,
+    )
+    assert result["type"] is FlowResultType.FORM
+    assert result["errors"] == mock_failed_mqtt_user_input_errors
+
+    # Try again with a valid configuration
+    result = await hass.config_entries.subentries.async_configure(
+        result["flow_id"], user_input=mock_mqtt_user_input
+    )
+
+    assert result["type"] is FlowResultType.CREATE_ENTRY
+    assert result["title"] == device_name
+
+    subentry_component = next(
+        iter(next(iter(config_entry.subentries.values())).data["components"].values())
+    )
+    assert subentry_component == next(
+        iter(config_subentries_data["components"].values())
+    )
+
+    await hass.async_block_till_done()
+
+
+@pytest.mark.parametrize(
+    "mqtt_config_subentries_data",
+    [
+        (
+            ConfigSubentryData(
+                data=MOCK_NOTIFY_SUBENTRY_DATA_MULTI,
+                subentry_type="device",
+                title="Mock subentry",
+            ),
+        )
+    ],
+    ids=["notify"],
+)
+async def test_subentry_reconfigure_remove_entity(
+    hass: HomeAssistant,
+    mqtt_mock_entry: MqttMockHAClientGenerator,
+    device_registry: dr.DeviceRegistry,
+    entity_registry: er.EntityRegistry,
+) -> None:
+    """Test the subentry ConfigFlow reconfigure removing an entity."""
+    await mqtt_mock_entry()
+    config_entry: MockConfigEntry = hass.config_entries.async_entries(mqtt.DOMAIN)[0]
+    subentry_id: str
+    subentry: ConfigSubentry
+    subentry_id, subentry = next(iter(config_entry.subentries.items()))
+    result = await config_entry.start_subentry_reconfigure_flow(
+        hass, "device", subentry_id
+    )
+    assert result["type"] is FlowResultType.MENU
+    assert result["step_id"] == "summary_menu"
+
+    # assert we have a device for the subentry
+    device = device_registry.async_get_device(identifiers={(mqtt.DOMAIN, subentry_id)})
+    assert device is not None
+
+    # assert we have an entity for all subentry components
+    components = deepcopy(dict(subentry.data))["components"]
+    assert len(components) == 2
+    object_list = list(components)
+    component_list = list(components.values())
+    entity_name_0 = f"{device.name} {component_list[0]['name']}"
+    entity_name_1 = f"{device.name} {component_list[1]['name']}"
+
+    for key, component in components.items():
+        unique_entity_id = f"{subentry_id}_{key}"
+        entity_id = entity_registry.async_get_entity_id(
+            domain=component["platform"],
+            platform=mqtt.DOMAIN,
+            unique_id=unique_entity_id,
+        )
+        assert entity_id is not None
+        entity_entry = entity_registry.async_get(entity_id)
+        assert entity_entry is not None
+        assert entity_entry.config_subentry_id == subentry_id
+
+    # assert menu options, we have the option to delete one entity
+    # we have no option to save and finish yet
+    assert result["menu_options"] == [
+        "entity",
+        "update_entity",
+        "delete_entity",
+        "device",
+    ]
+
+    # assert we can delete an entity
+    result = await hass.config_entries.subentries.async_configure(
+        result["flow_id"],
+        {"next_step_id": "delete_entity"},
+    )
+    assert result["type"] is FlowResultType.FORM
+    assert result["step_id"] == "delete_entity"
+    assert result["data_schema"].schema["component"].config["options"] == [
+        {"value": object_list[0], "label": entity_name_0},
+        {"value": object_list[1], "label": entity_name_1},
+    ]
+    # remove notify_the_second_notifier
+    result = await hass.config_entries.subentries.async_configure(
+        result["flow_id"],
+        user_input={
+            "component": object_list[1],
+        },
+    )
+
+    # assert menu options, we have only one item left, we cannot delete it
+    assert result["type"] is FlowResultType.MENU
+    assert result["step_id"] == "summary_menu"
+    assert result["menu_options"] == [
+        "entity",
+        "update_entity",
+        "device",
+        "save_changes",
+    ]
+
+    # finish reconfigure flow
+    result = await hass.config_entries.subentries.async_configure(
+        result["flow_id"],
+        {"next_step_id": "save_changes"},
+    )
+    assert result["type"] is FlowResultType.ABORT
+    assert result["reason"] == "reconfigure_successful"
+
+    # check if the second entity was removed from the subentry and entity registry
+    unique_entity_id = f"{subentry_id}_{object_list[1]}"
+    entity_id = entity_registry.async_get_entity_id(
+        domain=components[object_list[1]]["platform"],
+        platform=mqtt.DOMAIN,
+        unique_id=unique_entity_id,
+    )
+    assert entity_id is None
+    new_components = deepcopy(dict(subentry.data))["components"]
+    assert object_list[0] in new_components
+    assert object_list[1] not in new_components
+
+
+@pytest.mark.parametrize(
+    ("mqtt_config_subentries_data", "user_input_mqtt"),
+    [
+        (
+            (
+                ConfigSubentryData(
+                    data=MOCK_NOTIFY_SUBENTRY_DATA_MULTI,
+                    subentry_type="device",
+                    title="Mock subentry",
+                ),
+            ),
+            {"command_topic": "test-topic2-updated"},
+        )
+    ],
+    ids=["notify"],
+)
+async def test_subentry_reconfigure_edit_entity_multi_entitites(
+    hass: HomeAssistant,
+    mqtt_mock_entry: MqttMockHAClientGenerator,
+    device_registry: dr.DeviceRegistry,
+    entity_registry: er.EntityRegistry,
+    user_input_mqtt: dict[str, Any],
+) -> None:
+    """Test the subentry ConfigFlow reconfigure with multi entities."""
+    await mqtt_mock_entry()
+    config_entry: MockConfigEntry = hass.config_entries.async_entries(mqtt.DOMAIN)[0]
+    subentry_id: str
+    subentry: ConfigSubentry
+    subentry_id, subentry = next(iter(config_entry.subentries.items()))
+    result = await config_entry.start_subentry_reconfigure_flow(
+        hass, "device", subentry_id
+    )
+
+    assert result["type"] is FlowResultType.MENU
+    assert result["step_id"] == "summary_menu"
+
+    # assert we have a device for the subentry
+    device = device_registry.async_get_device(identifiers={(mqtt.DOMAIN, subentry_id)})
+    assert device is not None
+
+    # assert we have an entity for all subentry components
+    components = deepcopy(dict(subentry.data))["components"]
+    assert len(components) == 2
+    object_list = list(components)
+    component_list = list(components.values())
+    entity_name_0 = f"{device.name} {component_list[0]['name']}"
+    entity_name_1 = f"{device.name} {component_list[1]['name']}"
+
+    for key in components:
+        unique_entity_id = f"{subentry_id}_{key}"
+        entity_id = entity_registry.async_get_entity_id(
+            domain="notify", platform=mqtt.DOMAIN, unique_id=unique_entity_id
+        )
+        assert entity_id is not None
+        entity_entry = entity_registry.async_get(entity_id)
+        assert entity_entry is not None
+        assert entity_entry.config_subentry_id == subentry_id
+
+    # assert menu options, we have the option to delete one entity
+    # we have no option to save and finish yet
+    assert result["menu_options"] == [
+        "entity",
+        "update_entity",
+        "delete_entity",
+        "device",
+    ]
+
+    # assert we can update an entity
+    result = await hass.config_entries.subentries.async_configure(
+        result["flow_id"],
+        {"next_step_id": "update_entity"},
+    )
+    assert result["type"] is FlowResultType.FORM
+    assert result["step_id"] == "update_entity"
+    assert result["data_schema"].schema["component"].config["options"] == [
+        {"value": object_list[0], "label": entity_name_0},
+        {"value": object_list[1], "label": entity_name_1},
+    ]
+    # select second entity
+    result = await hass.config_entries.subentries.async_configure(
+        result["flow_id"],
+        user_input={
+            "component": object_list[1],
+        },
+    )
+    assert result["type"] is FlowResultType.FORM
+    assert result["step_id"] == "entity"
+
+    # submit the common entity data with changed entity_picture
+    result = await hass.config_entries.subentries.async_configure(
+        result["flow_id"],
+        user_input={
+            "entity_picture": "https://example.com",
+        },
+    )
+    assert result["type"] is FlowResultType.FORM
+    assert result["step_id"] == "mqtt_platform_config"
+
+    # submit the new platform specific entity data
+    result = await hass.config_entries.subentries.async_configure(
+        result["flow_id"],
+        user_input=user_input_mqtt,
+    )
+    assert result["type"] is FlowResultType.MENU
+    assert result["step_id"] == "summary_menu"
+
+    # finish reconfigure flow
+    result = await hass.config_entries.subentries.async_configure(
+        result["flow_id"],
+        {"next_step_id": "save_changes"},
+    )
+    assert result["type"] is FlowResultType.ABORT
+    assert result["reason"] == "reconfigure_successful"
+
+    # Check we still have our components
+    new_components = deepcopy(dict(subentry.data))["components"]
+
+    # Check the second component was updated
+    assert new_components[object_list[0]] == components[object_list[0]]
+    for key, value in user_input_mqtt.items():
+        assert new_components[object_list[1]][key] == value
+
+
+@pytest.mark.parametrize(
+    ("mqtt_config_subentries_data", "user_input_mqtt"),
+    [
+        (
+            (
+                ConfigSubentryData(
+                    data=MOCK_NOTIFY_SUBENTRY_DATA_SINGLE,
+                    subentry_type="device",
+                    title="Mock subentry",
+                ),
+            ),
+            {
+                "command_topic": "test-topic1-updated",
+                "command_template": "{{ value_json.value }}",
+                "retain": True,
+            },
+        )
+    ],
+    ids=["notify"],
+)
+async def test_subentry_reconfigure_edit_entity_single_entity(
+    hass: HomeAssistant,
+    mqtt_mock_entry: MqttMockHAClientGenerator,
+    device_registry: dr.DeviceRegistry,
+    entity_registry: er.EntityRegistry,
+    user_input_mqtt: dict[str, Any],
+) -> None:
+    """Test the subentry ConfigFlow reconfigure with single entity."""
+    await mqtt_mock_entry()
+    config_entry: MockConfigEntry = hass.config_entries.async_entries(mqtt.DOMAIN)[0]
+    subentry_id: str
+    subentry: ConfigSubentry
+    subentry_id, subentry = next(iter(config_entry.subentries.items()))
+    result = await config_entry.start_subentry_reconfigure_flow(
+        hass, "device", subentry_id
+    )
+    assert result["type"] is FlowResultType.MENU
+    assert result["step_id"] == "summary_menu"
+
+    # assert we have a device for the subentry
+    device = device_registry.async_get_device(identifiers={(mqtt.DOMAIN, subentry_id)})
+    assert device is not None
+
+    # assert we have an entity for the subentry component
+    # Check we have "notify_milkman_alert" in our mock data
+    components = deepcopy(dict(subentry.data))["components"]
+    assert len(components) == 1
+
+    component_id, component = next(iter(components.items()))
+
+    unique_entity_id = f"{subentry_id}_{component_id}"
+    entity_id = entity_registry.async_get_entity_id(
+        domain=component["platform"], platform=mqtt.DOMAIN, unique_id=unique_entity_id
+    )
+    assert entity_id is not None
+    entity_entry = entity_registry.async_get(entity_id)
+    assert entity_entry is not None
+    assert entity_entry.config_subentry_id == subentry_id
+
+    # assert menu options, we do not have the option to delete an entity
+    # we have no option to save and finish yet
+    assert result["menu_options"] == [
+        "entity",
+        "update_entity",
+        "device",
+    ]
+
+    # assert we can update the entity, there is no select step
+    result = await hass.config_entries.subentries.async_configure(
+        result["flow_id"],
+        {"next_step_id": "update_entity"},
+    )
+    assert result["type"] is FlowResultType.FORM
+    assert result["step_id"] == "entity"
+
+    # submit the new common entity data, reset entity_picture
+    result = await hass.config_entries.subentries.async_configure(
+        result["flow_id"],
+        user_input={},
+    )
+    assert result["type"] is FlowResultType.FORM
+    assert result["step_id"] == "mqtt_platform_config"
+
+    # submit the new platform specific entity data,
+    result = await hass.config_entries.subentries.async_configure(
+        result["flow_id"],
+        user_input=user_input_mqtt,
+    )
+    assert result["type"] is FlowResultType.MENU
+    assert result["step_id"] == "summary_menu"
+
+    # finish reconfigure flow
+    result = await hass.config_entries.subentries.async_configure(
+        result["flow_id"],
+        {"next_step_id": "save_changes"},
+    )
+    assert result["type"] is FlowResultType.ABORT
+    assert result["reason"] == "reconfigure_successful"
+
+    # Check we still have out components
+    new_components = deepcopy(dict(subentry.data))["components"]
+    assert len(new_components) == 1
+
+    # Check our update was successful
+    assert "entity_picture" not in new_components[component_id]
+
+    # Check the second component was updated
+    for key, value in user_input_mqtt.items():
+        assert new_components[component_id][key] == value
+
+
+@pytest.mark.parametrize(
+    ("mqtt_config_subentries_data", "user_input_entity", "user_input_mqtt"),
+    [
+        (
+            (
+                ConfigSubentryData(
+                    data=MOCK_NOTIFY_SUBENTRY_DATA_SINGLE,
+                    subentry_type="device",
+                    title="Mock subentry",
+                ),
+            ),
+            {
+                "platform": "notify",
+                "name": "The second notifier",
+                "entity_picture": "https://example.com",
+            },
+            {
+                "command_topic": "test-topic2",
+                "qos": 0,
+            },
+        )
+    ],
+    ids=["notify_notify"],
+)
+async def test_subentry_reconfigure_add_entity(
+    hass: HomeAssistant,
+    mqtt_mock_entry: MqttMockHAClientGenerator,
+    device_registry: dr.DeviceRegistry,
+    entity_registry: er.EntityRegistry,
+    user_input_entity: dict[str, Any],
+    user_input_mqtt: dict[str, Any],
+) -> None:
+    """Test the subentry ConfigFlow reconfigure and add an entity."""
+    await mqtt_mock_entry()
+    config_entry: MockConfigEntry = hass.config_entries.async_entries(mqtt.DOMAIN)[0]
+    subentry_id: str
+    subentry: ConfigSubentry
+    subentry_id, subentry = next(iter(config_entry.subentries.items()))
+    result = await config_entry.start_subentry_reconfigure_flow(
+        hass, "device", subentry_id
+    )
+    assert result["type"] is FlowResultType.MENU
+    assert result["step_id"] == "summary_menu"
+
+    # assert we have a device for the subentry
+    device = device_registry.async_get_device(identifiers={(mqtt.DOMAIN, subentry_id)})
+    assert device is not None
+
+    # assert we have an entity for the subentry component
+    components = deepcopy(dict(subentry.data))["components"]
+    assert len(components) == 1
+    component_id_1, component1 = next(iter(components.items()))
+    unique_entity_id = f"{subentry_id}_{component_id_1}"
+    entity_id = entity_registry.async_get_entity_id(
+        domain=component1["platform"], platform=mqtt.DOMAIN, unique_id=unique_entity_id
+    )
+    assert entity_id is not None
+    entity_entry = entity_registry.async_get(entity_id)
+    assert entity_entry is not None
+    assert entity_entry.config_subentry_id == subentry_id
+
+    # assert menu options, we do not have the option to delete an entity
+    # we have no option to save and finish yet
+    assert result["menu_options"] == [
+        "entity",
+        "update_entity",
+        "device",
+    ]
+
+    # assert we can update the entity, there is no select step
+    result = await hass.config_entries.subentries.async_configure(
+        result["flow_id"],
+        {"next_step_id": "entity"},
+    )
+    assert result["type"] is FlowResultType.FORM
+    assert result["step_id"] == "entity"
+
+    # submit the new common entity data
+    result = await hass.config_entries.subentries.async_configure(
+        result["flow_id"],
+        user_input=user_input_entity,
+    )
+    assert result["type"] is FlowResultType.FORM
+    assert result["step_id"] == "mqtt_platform_config"
+
+    # submit the new platform specific entity data
+    result = await hass.config_entries.subentries.async_configure(
+        result["flow_id"],
+        user_input=user_input_mqtt,
+    )
+    assert result["type"] is FlowResultType.MENU
+    assert result["step_id"] == "summary_menu"
+
+    # Finish reconfigure flow
+    result = await hass.config_entries.subentries.async_configure(
+        result["flow_id"],
+        {"next_step_id": "save_changes"},
+    )
+    assert result["type"] is FlowResultType.ABORT
+    assert result["reason"] == "reconfigure_successful"
+
+    # Check we still have out components
+    new_components = deepcopy(dict(subentry.data))["components"]
+    assert len(new_components) == 2
+
+    component_id_2 = next(iter(set(new_components) - {component_id_1}))
+
+    # Check our new entity was added correctly
+    expected_component_config = user_input_entity | user_input_mqtt
+    for key, value in expected_component_config.items():
+        assert new_components[component_id_2][key] == value
+
+
+@pytest.mark.parametrize(
+    "mqtt_config_subentries_data",
+    [
+        (
+            ConfigSubentryData(
+                data=MOCK_NOTIFY_SUBENTRY_DATA_MULTI,
+                subentry_type="device",
+                title="Mock subentry",
+            ),
+        )
+    ],
+)
+async def test_subentry_reconfigure_update_device_properties(
+    hass: HomeAssistant,
+    mqtt_mock_entry: MqttMockHAClientGenerator,
+    device_registry: dr.DeviceRegistry,
+) -> None:
+    """Test the subentry ConfigFlow reconfigure and update device properties."""
+    await mqtt_mock_entry()
+    config_entry: MockConfigEntry = hass.config_entries.async_entries(mqtt.DOMAIN)[0]
+    subentry_id: str
+    subentry: ConfigSubentry
+    subentry_id, subentry = next(iter(config_entry.subentries.items()))
+    result = await config_entry.start_subentry_reconfigure_flow(
+        hass, "device", subentry_id
+    )
+    assert result["type"] is FlowResultType.MENU
+    assert result["step_id"] == "summary_menu"
+
+    # assert we have a device for the subentry
+    device = device_registry.async_get_device(identifiers={(mqtt.DOMAIN, subentry_id)})
+    assert device is not None
+
+    # assert we have an entity for all subentry components
+    components = deepcopy(dict(subentry.data))["components"]
+    assert len(components) == 2
+
+    # Assert initial data
+    device = deepcopy(dict(subentry.data))["device"]
+    assert device["name"] == "Milk notifier"
+    assert device["sw_version"] == "1.0"
+    assert device["hw_version"] == "2.1 rev a"
+    assert device["model"] == "Model XL"
+    assert device["model_id"] == "mn002"
+
+    # assert menu options, we have the option to delete one entity
+    # we have no option to save and finish yet
+    assert result["menu_options"] == [
+        "entity",
+        "update_entity",
+        "delete_entity",
+        "device",
+    ]
+
+    # assert we can update the device properties
+    result = await hass.config_entries.subentries.async_configure(
+        result["flow_id"],
+        {"next_step_id": "device"},
+    )
+    assert result["type"] is FlowResultType.FORM
+    assert result["step_id"] == "device"
+
+    # Update the device details
+    result = await hass.config_entries.subentries.async_configure(
+        result["flow_id"],
+        user_input={
+            "name": "Beer notifier",
+            "sw_version": "1.1",
+            "model": "Beer bottle XL",
+            "model_id": "bn003",
+            "configuration_url": "https://example.com",
+        },
+    )
+    assert result["type"] is FlowResultType.MENU
+    assert result["step_id"] == "summary_menu"
+
+    # finish reconfigure flow
+    result = await hass.config_entries.subentries.async_configure(
+        result["flow_id"],
+        {"next_step_id": "save_changes"},
+    )
+    assert result["type"] is FlowResultType.ABORT
+    assert result["reason"] == "reconfigure_successful"
+
+    # Check our device was updated
+    device = deepcopy(dict(subentry.data))["device"]
+    assert device["name"] == "Beer notifier"
+    assert "hw_version" not in device
+    assert device["model"] == "Beer bottle XL"
+    assert device["model_id"] == "bn003"

tests/components/mqtt/test_mixins.py

@@ -1,18 +1,27 @@
 """The tests for shared code of the MQTT platform."""
 
+from typing import Any
 from unittest.mock import patch
 
 import pytest
 
 from homeassistant.components import mqtt, sensor
 from homeassistant.components.mqtt.sensor import DEFAULT_NAME as DEFAULT_SENSOR_NAME
+from homeassistant.config_entries import ConfigSubentryData
 from homeassistant.const import (
     ATTR_FRIENDLY_NAME,
     EVENT_HOMEASSISTANT_STARTED,
     EVENT_STATE_CHANGED,
 )
 from homeassistant.core import CoreState, HomeAssistant, callback
-from homeassistant.helpers import device_registry as dr, issue_registry as ir
+from homeassistant.helpers import (
+    device_registry as dr,
+    entity_registry as er,
+    issue_registry as ir,
+)
+from homeassistant.util import slugify
+
+from .common import MOCK_SUBENTRY_DATA_BAD_COMPONENT_SCHEMA, MOCK_SUBENTRY_DATA_SET_MIX
 
 from tests.common import MockConfigEntry, async_capture_events, async_fire_mqtt_message
 from tests.typing import MqttMockHAClientGenerator
@@ -453,3 +462,74 @@ async def test_value_template_fails(
         "TypeError: unsupported operand type(s) for *: 'NoneType' and 'int' rendering template"
         in caplog.text
     )
+
+
+@pytest.mark.parametrize(
+    "mqtt_config_subentries_data",
+    [
+        (
+            ConfigSubentryData(
+                data=MOCK_SUBENTRY_DATA_SET_MIX,
+                subentry_type="device",
+                title="Mock subentry",
+            ),
+        )
+    ],
+)
+async def test_loading_subentries(
+    hass: HomeAssistant,
+    mqtt_mock_entry: MqttMockHAClientGenerator,
+    mqtt_config_subentries_data: tuple[dict[str, Any]],
+    device_registry: dr.DeviceRegistry,
+    entity_registry: er.EntityRegistry,
+) -> None:
+    """Test loading subentries."""
+    await mqtt_mock_entry()
+    entry = hass.config_entries.async_entries(mqtt.DOMAIN)[0]
+    subentry_id = next(iter(entry.subentries))
+    # Each subentry has one device
+    device = device_registry.async_get_device({("mqtt", subentry_id)})
+    assert device is not None
+    for object_id, component in mqtt_config_subentries_data[0]["data"][
+        "components"
+    ].items():
+        platform = component["platform"]
+        entity_id = f"{platform}.{slugify(device.name)}_{slugify(component['name'])}"
+        entity_entry_entity_id = entity_registry.async_get_entity_id(
+            platform, mqtt.DOMAIN, f"{subentry_id}_{object_id}"
+        )
+        assert entity_entry_entity_id == entity_id
+        state = hass.states.get(entity_id)
+        assert state is not None
+
+
+@pytest.mark.parametrize(
+    "mqtt_config_subentries_data",
+    [
+        (
+            ConfigSubentryData(
+                data=MOCK_SUBENTRY_DATA_BAD_COMPONENT_SCHEMA,
+                subentry_type="device",
+                title="Mock subentry",
+            ),
+        )
+    ],
+)
+async def test_loading_subentry_with_bad_component_schema(
+    hass: HomeAssistant,
+    mqtt_mock_entry: MqttMockHAClientGenerator,
+    mqtt_config_subentries_data: tuple[dict[str, Any]],
+    device_registry: dr.DeviceRegistry,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """Test loading subentries."""
+    await mqtt_mock_entry()
+    entry = hass.config_entries.async_entries(mqtt.DOMAIN)[0]
+    subentry_id = next(iter(entry.subentries))
+    # Each subentry has one device
+    device = device_registry.async_get_device({("mqtt", subentry_id)})
+    assert device is None
+    assert (
+        "Schema violation occurred when trying to set up entity from subentry"
+        in caplog.text
+    )

tests/conftest.py

@@ -67,7 +67,12 @@ from homeassistant.components.websocket_api.auth import (
 # pylint: disable-next=hass-component-root-import
 from homeassistant.components.websocket_api.http import URL
 from homeassistant.config import YAML_CONFIG_FILE
-from homeassistant.config_entries import ConfigEntries, ConfigEntry, ConfigEntryState
+from homeassistant.config_entries import (
+    ConfigEntries,
+    ConfigEntry,
+    ConfigEntryState,
+    ConfigSubentryData,
+)
 from homeassistant.const import BASE_PLATFORMS, HASSIO_USER_NAME
 from homeassistant.core import (
     Context,
@@ -946,6 +951,12 @@ def mqtt_config_entry_data() -> dict[str, Any] | None:
     return None
 
 
+@pytest.fixture
+def mqtt_config_subentries_data() -> tuple[ConfigSubentryData] | None:
+    """Fixture to allow overriding MQTT subentries data."""
+    return None
+
+
 @pytest.fixture
 def mqtt_config_entry_options() -> dict[str, Any] | None:
     """Fixture to allow overriding MQTT entry options."""
@@ -1032,6 +1043,7 @@ async def mqtt_mock(
     mqtt_client_mock: MqttMockPahoClient,
     mqtt_config_entry_data: dict[str, Any] | None,
     mqtt_config_entry_options: dict[str, Any] | None,
+    mqtt_config_subentries_data: tuple[ConfigSubentryData] | None,
     mqtt_mock_entry: MqttMockHAClientGenerator,
 ) -> AsyncGenerator[MqttMockHAClient]:
     """Fixture to mock MQTT component."""
@@ -1044,6 +1056,7 @@ async def _mqtt_mock_entry(
     mqtt_client_mock: MqttMockPahoClient,
     mqtt_config_entry_data: dict[str, Any] | None,
     mqtt_config_entry_options: dict[str, Any] | None,
+    mqtt_config_subentries_data: tuple[ConfigSubentryData] | None,
 ) -> AsyncGenerator[MqttMockHAClientGenerator]:
     """Fixture to mock a delayed setup of the MQTT config entry."""
     # Local import to avoid processing MQTT modules when running a testcase
@@ -1060,6 +1073,7 @@ async def _mqtt_mock_entry(
     entry = MockConfigEntry(
         data=mqtt_config_entry_data,
         options=mqtt_config_entry_options,
+        subentries_data=mqtt_config_subentries_data,
         domain=mqtt.DOMAIN,
         title="MQTT",
         version=1,
@@ -1174,6 +1188,7 @@ async def mqtt_mock_entry(
     mqtt_client_mock: MqttMockPahoClient,
     mqtt_config_entry_data: dict[str, Any] | None,
     mqtt_config_entry_options: dict[str, Any] | None,
+    mqtt_config_subentries_data: tuple[ConfigSubentryData] | None,
 ) -> AsyncGenerator[MqttMockHAClientGenerator]:
     """Set up an MQTT config entry."""
 
@@ -1190,7 +1205,11 @@ async def mqtt_mock_entry(
         return await mqtt_mock_entry(_async_setup_config_entry)
 
     async with _mqtt_mock_entry(
-        hass, mqtt_client_mock, mqtt_config_entry_data, mqtt_config_entry_options
+        hass,
+        mqtt_client_mock,
+        mqtt_config_entry_data,
+        mqtt_config_entry_options,
+        mqtt_config_subentries_data,
     ) as mqtt_mock_entry:
         yield _setup_mqtt_entry