Mqtt support config_entry unload (#70149)

* squashed commits for rebase

* Flake

* Fix reloading issue manual legacy items

* Improve ACS sync for unsubscribe at disconnect

* Processed review comments

* Update homeassistant/components/mqtt/client.py

Co-authored-by: Erik Montnemery <erik@montnemery.com>

* No need to await entry setup

* Remove complication is_connected

* Update homeassistant/components/mqtt/__init__.py

Co-authored-by: Erik Montnemery <erik@montnemery.com>

homeassistant/components/mqtt/__init__.py

@@ -11,7 +11,7 @@ from typing import Any, cast
 import jinja2
 import voluptuous as vol
 
-from homeassistant import config_entries
+from homeassistant import config as conf_util, config_entries
 from homeassistant.components import websocket_api
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.const import (
@@ -20,10 +20,9 @@ from homeassistant.const import (
     CONF_PAYLOAD,
     CONF_PORT,
     CONF_USERNAME,
-    EVENT_HOMEASSISTANT_STOP,
     SERVICE_RELOAD,
 )
-from homeassistant.core import Event, HassJob, HomeAssistant, ServiceCall, callback
+from homeassistant.core import HassJob, HomeAssistant, ServiceCall, callback
 from homeassistant.data_entry_flow import BaseServiceInfo
 from homeassistant.exceptions import TemplateError, Unauthorized
 from homeassistant.helpers import config_validation as cv, event, template
@@ -65,9 +64,10 @@ from .const import (  # noqa: F401
     CONF_TOPIC,
     CONF_WILL_MESSAGE,
     CONFIG_ENTRY_IS_SETUP,
-    DATA_CONFIG_ENTRY_LOCK,
     DATA_MQTT,
     DATA_MQTT_CONFIG,
+    DATA_MQTT_RELOAD_DISPATCHERS,
+    DATA_MQTT_RELOAD_ENTRY,
     DATA_MQTT_RELOAD_NEEDED,
     DATA_MQTT_UPDATED_CONFIG,
     DEFAULT_ENCODING,
@@ -87,7 +87,12 @@ from .models import (  # noqa: F401
     ReceiveMessage,
     ReceivePayloadType,
 )
-from .util import _VALID_QOS_SCHEMA, valid_publish_topic, valid_subscribe_topic
+from .util import (
+    _VALID_QOS_SCHEMA,
+    mqtt_config_entry_enabled,
+    valid_publish_topic,
+    valid_subscribe_topic,
+)
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -174,7 +179,7 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
         conf = dict(conf)
         hass.data[DATA_MQTT_CONFIG] = conf
 
-    if not bool(hass.config_entries.async_entries(DOMAIN)):
+    if (mqtt_entry_status := mqtt_config_entry_enabled(hass)) is None:
         # Create an import flow if the user has yaml configured entities etc.
         # but no broker configuration. Note: The intention is not for this to
         # import broker configuration from YAML because that has been deprecated.
@@ -185,6 +190,13 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
                 data={},
             )
         )
+        hass.data[DATA_MQTT_RELOAD_NEEDED] = True
+    elif mqtt_entry_status is False:
+        _LOGGER.info(
+            "MQTT will be not available until the config entry is enabled",
+        )
+        hass.data[DATA_MQTT_RELOAD_NEEDED] = True
+
     return True
 
 
@@ -239,17 +251,18 @@ async def _async_config_entry_updated(hass: HomeAssistant, entry: ConfigEntry) -
         await _async_setup_discovery(hass, mqtt_client.conf, entry)
 
 
-async def async_setup_entry(  # noqa: C901
+async def async_fetch_config(hass: HomeAssistant, entry: ConfigEntry) -> dict | None:
-    hass: HomeAssistant, entry: ConfigEntry
+    """Fetch fresh MQTT yaml config from the hass config when (re)loading the entry."""
-) -> bool:
+    if DATA_MQTT_RELOAD_ENTRY in hass.data:
-    """Load a config entry."""
+        hass_config = await conf_util.async_hass_config_yaml(hass)
-    # Merge basic configuration, and add missing defaults for basic options
+        mqtt_config = CONFIG_SCHEMA_BASE(hass_config.get(DOMAIN, {}))
-    _merge_basic_config(hass, entry, hass.data.get(DATA_MQTT_CONFIG, {}))
+        hass.data[DATA_MQTT_CONFIG] = mqtt_config
 
+    _merge_basic_config(hass, entry, hass.data.get(DATA_MQTT_CONFIG, {}))
     # Bail out if broker setting is missing
     if CONF_BROKER not in entry.data:
         _LOGGER.error("MQTT broker is not configured, please configure it")
-        return False
+        return None
 
     # If user doesn't have configuration.yaml config, generate default values
     # for options not in config entry data
@@ -271,22 +284,21 @@ async def async_setup_entry(  # noqa: C901
 
     # Merge advanced configuration values from configuration.yaml
     conf = _merge_extended_config(entry, conf)
+    return conf
 
-    hass.data[DATA_MQTT] = MQTT(
+
-        hass,
+async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
-        entry,
+    """Load a config entry."""
-        conf,
+    # Merge basic configuration, and add missing defaults for basic options
-    )
+    if (conf := await async_fetch_config(hass, entry)) is None:
+        # Bail out
+        return False
+
+    hass.data[DATA_MQTT] = MQTT(hass, entry, conf)
     entry.add_update_listener(_async_config_entry_updated)
 
     await hass.data[DATA_MQTT].async_connect()
 
-    async def async_stop_mqtt(_event: Event):
-        """Stop MQTT component."""
-        await hass.data[DATA_MQTT].async_disconnect()
-
-    hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, async_stop_mqtt)
-
     async def async_publish_service(call: ServiceCall) -> None:
         """Handle MQTT publish service calls."""
         msg_topic = call.data.get(ATTR_TOPIC)
@@ -375,7 +387,6 @@ async def async_setup_entry(  # noqa: C901
     )
 
     # setup platforms and discovery
-    hass.data[DATA_CONFIG_ENTRY_LOCK] = asyncio.Lock()
     hass.data[CONFIG_ENTRY_IS_SETUP] = set()
 
     async def async_setup_reload_service() -> None:
@@ -411,6 +422,7 @@ async def async_setup_entry(  # noqa: C901
         # pylint: disable-next=import-outside-toplevel
         from . import device_automation, tag
 
+        # Forward the entry setup to the MQTT platforms
         await asyncio.gather(
             *(
                 [
@@ -428,21 +440,25 @@ async def async_setup_entry(  # noqa: C901
             await _async_setup_discovery(hass, conf, entry)
         # Setup reload service after all platforms have loaded
         await async_setup_reload_service()
-
         if DATA_MQTT_RELOAD_NEEDED in hass.data:
             hass.data.pop(DATA_MQTT_RELOAD_NEEDED)
-            await hass.services.async_call(
+            await async_reload_manual_mqtt_items(hass)
-                DOMAIN,
-                SERVICE_RELOAD,
-                {},
-                blocking=False,
-            )
 
     await async_forward_entry_setup_and_setup_discovery(entry)
 
     return True
 
 
+async def async_reload_manual_mqtt_items(hass: HomeAssistant) -> None:
+    """Reload manual configured MQTT items."""
+    await hass.services.async_call(
+        DOMAIN,
+        SERVICE_RELOAD,
+        {},
+        blocking=True,
+    )
+
+
 @websocket_api.websocket_command(
     {vol.Required("type"): "mqtt/device/debug_info", vol.Required("device_id"): str}
 )
@@ -544,3 +560,49 @@ async def async_remove_config_entry_device(
 
     await device_automation.async_removed_from_device(hass, device_entry.id)
     return True
+
+
+async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Unload MQTT dump and publish service when the config entry is unloaded."""
+    # Unload publish and dump services.
+    hass.services.async_remove(
+        DOMAIN,
+        SERVICE_PUBLISH,
+    )
+    hass.services.async_remove(
+        DOMAIN,
+        SERVICE_DUMP,
+    )
+
+    # Stop the discovery
+    await discovery.async_stop(hass)
+    mqtt_client: MQTT = hass.data[DATA_MQTT]
+    # Unload the platforms
+    await asyncio.gather(
+        *(
+            hass.config_entries.async_forward_entry_unload(entry, component)
+            for component in PLATFORMS
+        )
+    )
+    await hass.async_block_till_done()
+    # Unsubscribe reload dispatchers
+    while reload_dispatchers := hass.data.setdefault(DATA_MQTT_RELOAD_DISPATCHERS, []):
+        reload_dispatchers.pop()()
+    hass.data[CONFIG_ENTRY_IS_SETUP] = set()
+    # Cleanup listeners
+    mqtt_client.cleanup()
+
+    # Trigger reload manual MQTT items at entry setup
+    # Reload the legacy yaml platform
+    await async_reload_integration_platforms(hass, DOMAIN, RELOADABLE_PLATFORMS)
+    if (mqtt_entry_status := mqtt_config_entry_enabled(hass)) is False:
+        # The entry is disabled reload legacy manual items when the entry is enabled again
+        hass.data[DATA_MQTT_RELOAD_NEEDED] = True
+    elif mqtt_entry_status is True:
+        # The entry is reloaded:
+        # Trigger re-fetching the yaml config at entry setup
+        hass.data[DATA_MQTT_RELOAD_ENTRY] = True
+    # Stop the loop
+    await mqtt_client.async_disconnect()
+
+    return True

homeassistant/components/mqtt/alarm_control_panel.py

@@ -156,8 +156,12 @@ async def async_setup_entry(
 
 
 async def _async_setup_entity(
-    hass, async_add_entities, config, config_entry=None, discovery_data=None
+    hass: HomeAssistant,
-):
+    async_add_entities: AddEntitiesCallback,
+    config: ConfigType,
+    config_entry: ConfigEntry | None = None,
+    discovery_data: dict | None = None,
+) -> None:
     """Set up the MQTT Alarm Control Panel platform."""
     async_add_entities([MqttAlarm(hass, config, config_entry, discovery_data)])
 

homeassistant/components/mqtt/binary_sensor.py

@@ -112,8 +112,12 @@ async def async_setup_entry(
 
 
 async def _async_setup_entity(
-    hass, async_add_entities, config, config_entry=None, discovery_data=None
+    hass: HomeAssistant,
-):
+    async_add_entities: AddEntitiesCallback,
+    config: ConfigType,
+    config_entry: ConfigEntry | None = None,
+    discovery_data: dict | None = None,
+) -> None:
     """Set up the MQTT binary sensor."""
     async_add_entities([MqttBinarySensor(hass, config, config_entry, discovery_data)])
 

homeassistant/components/mqtt/button.py

@@ -91,8 +91,12 @@ async def async_setup_entry(
 
 
 async def _async_setup_entity(
-    hass, async_add_entities, config, config_entry=None, discovery_data=None
+    hass: HomeAssistant,
-):
+    async_add_entities: AddEntitiesCallback,
+    config: ConfigType,
+    config_entry: ConfigEntry | None = None,
+    discovery_data: dict | None = None,
+) -> None:
     """Set up the MQTT button."""
     async_add_entities([MqttButton(hass, config, config_entry, discovery_data)])
 

homeassistant/components/mqtt/camera.py

@@ -89,8 +89,12 @@ async def async_setup_entry(
 
 
 async def _async_setup_entity(
-    hass, async_add_entities, config, config_entry=None, discovery_data=None
+    hass: HomeAssistant,
-):
+    async_add_entities: AddEntitiesCallback,
+    config: ConfigType,
+    config_entry: ConfigEntry | None = None,
+    discovery_data: dict | None = None,
+) -> None:
     """Set up the MQTT Camera."""
     async_add_entities([MqttCamera(hass, config, config_entry, discovery_data)])
 

homeassistant/components/mqtt/client.py

@@ -23,8 +23,9 @@ from homeassistant.const import (
     CONF_PROTOCOL,
     CONF_USERNAME,
     EVENT_HOMEASSISTANT_STARTED,
+    EVENT_HOMEASSISTANT_STOP,
 )
-from homeassistant.core import CoreState, HassJob, HomeAssistant, callback
+from homeassistant.core import CoreState, Event, HassJob, HomeAssistant, callback
 from homeassistant.exceptions import HomeAssistantError
 from homeassistant.helpers.dispatcher import dispatcher_send
 from homeassistant.helpers.typing import ConfigType
@@ -59,6 +60,7 @@ from .models import (
     ReceiveMessage,
     ReceivePayloadType,
 )
+from .util import mqtt_config_entry_enabled
 
 if TYPE_CHECKING:
     # Only import for paho-mqtt type checking here, imports are done locally
@@ -95,6 +97,10 @@ async def async_publish(
 ) -> None:
     """Publish message to a MQTT topic."""
 
+    if DATA_MQTT not in hass.data or not mqtt_config_entry_enabled(hass):
+        raise HomeAssistantError(
+            f"Cannot publish to topic '{topic}', MQTT is not enabled"
+        )
     outgoing_payload = payload
     if not isinstance(payload, bytes):
         if not encoding:
@@ -174,6 +180,10 @@ async def async_subscribe(
 
     Call the return value to unsubscribe.
     """
+    if DATA_MQTT not in hass.data or not mqtt_config_entry_enabled(hass):
+        raise HomeAssistantError(
+            f"Cannot subscribe to topic '{topic}', MQTT is not enabled"
+        )
     # Count callback parameters which don't have a default value
     non_default = 0
     if msg_callback:
@@ -316,6 +326,8 @@ class MQTT:
         self._last_subscribe = time.time()
         self._mqttc: mqtt.Client = None
         self._paho_lock = asyncio.Lock()
+        self._pending_acks: set[int] = set()
+        self._cleanup_on_unload: list[Callable] = []
 
         self._pending_operations: dict[str, asyncio.Event] = {}
 
@@ -331,6 +343,20 @@ class MQTT:
 
         self.init_client()
 
+        @callback
+        async def async_stop_mqtt(_event: Event):
+            """Stop MQTT component."""
+            await self.async_disconnect()
+
+        self._cleanup_on_unload.append(
+            hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, async_stop_mqtt)
+        )
+
+    def cleanup(self):
+        """Clean up listeners."""
+        while self._cleanup_on_unload:
+            self._cleanup_on_unload.pop()()
+
     def init_client(self):
         """Initialize paho client."""
         self._mqttc = MqttClientSetup(self.conf).client
@@ -405,6 +431,15 @@ class MQTT:
             # Do not disconnect, we want the broker to always publish will
             self._mqttc.loop_stop()
 
+        # wait for ACK-s to be processes (unsubscribe only)
+        async with self._paho_lock:
+            tasks = [
+                self.hass.async_create_task(self._wait_for_mid(mid))
+                for mid in self._pending_acks
+            ]
+        await asyncio.gather(*tasks)
+
+        # stop the MQTT loop
         await self.hass.async_add_executor_job(stop)
 
     async def async_subscribe(
@@ -440,7 +475,7 @@ class MQTT:
             self.subscriptions.remove(subscription)
             self._matching_subscriptions.cache_clear()
 
-            # Only unsubscribe if currently connected.
+            # Only unsubscribe if currently connected
             if self.connected:
                 self.hass.async_create_task(self._async_unsubscribe(topic))
 
@@ -451,18 +486,20 @@ class MQTT:
 
         This method is a coroutine.
         """
+
+        def _client_unsubscribe(topic: str) -> None:
+            result: int | None = None
+            result, mid = self._mqttc.unsubscribe(topic)
+            _LOGGER.debug("Unsubscribing from %s, mid: %s", topic, mid)
+            _raise_on_error(result)
+            self._pending_acks.add(mid)
+
         if any(other.topic == topic for other in self.subscriptions):
             # Other subscriptions on topic remaining - don't unsubscribe.
             return
 
         async with self._paho_lock:
-            result: int | None = None
+            await self.hass.async_add_executor_job(_client_unsubscribe, topic)
-            result, mid = await self.hass.async_add_executor_job(
-                self._mqttc.unsubscribe, topic
-            )
-            _LOGGER.debug("Unsubscribing from %s, mid: %s", topic, mid)
-            _raise_on_error(result)
-        await self._wait_for_mid(mid)
 
     async def _async_perform_subscriptions(
         self, subscriptions: Iterable[tuple[str, int]]
@@ -643,6 +680,10 @@ class MQTT:
             )
         finally:
             del self._pending_operations[mid]
+            # Cleanup ACK sync buffer
+            async with self._paho_lock:
+                if mid in self._pending_acks:
+                    self._pending_acks.remove(mid)
 
     async def _discovery_cooldown(self):
         now = time.time()

homeassistant/components/mqtt/climate.py

@@ -401,8 +401,12 @@ async def async_setup_entry(
 
 
 async def _async_setup_entity(
-    hass, async_add_entities, config, config_entry=None, discovery_data=None
+    hass: HomeAssistant,
-):
+    async_add_entities: AddEntitiesCallback,
+    config: ConfigType,
+    config_entry: ConfigEntry | None = None,
+    discovery_data: dict | None = None,
+) -> None:
     """Set up the MQTT climate devices."""
     async_add_entities([MqttClimate(hass, config, config_entry, discovery_data)])
 

homeassistant/components/mqtt/const.py

@@ -31,9 +31,11 @@ CONF_TLS_INSECURE = "tls_insecure"
 CONF_TLS_VERSION = "tls_version"
 
 CONFIG_ENTRY_IS_SETUP = "mqtt_config_entry_is_setup"
-DATA_CONFIG_ENTRY_LOCK = "mqtt_config_entry_lock"
 DATA_MQTT = "mqtt"
 DATA_MQTT_CONFIG = "mqtt_config"
+MQTT_DATA_DEVICE_TRACKER_LEGACY = "mqtt_device_tracker_legacy"
+DATA_MQTT_RELOAD_DISPATCHERS = "mqtt_reload_dispatchers"
+DATA_MQTT_RELOAD_ENTRY = "mqtt_reload_entry"
 DATA_MQTT_RELOAD_NEEDED = "mqtt_reload_needed"
 DATA_MQTT_UPDATED_CONFIG = "mqtt_updated_config"
 

homeassistant/components/mqtt/cover.py

@@ -251,8 +251,12 @@ async def async_setup_entry(
 
 
 async def _async_setup_entity(
-    hass, async_add_entities, config, config_entry=None, discovery_data=None
+    hass: HomeAssistant,
-):
+    async_add_entities: AddEntitiesCallback,
+    config: ConfigType,
+    config_entry: ConfigEntry | None = None,
+    discovery_data: dict | None = None,
+) -> None:
     """Set up the MQTT Cover."""
     async_add_entities([MqttCover(hass, config, config_entry, discovery_data)])
 

homeassistant/components/mqtt/device_tracker/__init__.py

@@ -2,7 +2,11 @@
 import voluptuous as vol
 
 from homeassistant.components import device_tracker
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
 
+from ..const import MQTT_DATA_DEVICE_TRACKER_LEGACY
 from ..mixins import warn_for_legacy_schema
 from .schema_discovery import PLATFORM_SCHEMA_MODERN  # noqa: F401
 from .schema_discovery import async_setup_entry_from_discovery
@@ -12,5 +16,20 @@ from .schema_yaml import PLATFORM_SCHEMA_YAML, async_setup_scanner_from_yaml
 PLATFORM_SCHEMA = vol.All(
     PLATFORM_SCHEMA_YAML, warn_for_legacy_schema(device_tracker.DOMAIN)
 )
+
+# Legacy setup
 async_setup_scanner = async_setup_scanner_from_yaml
-async_setup_entry = async_setup_entry_from_discovery
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    config_entry: ConfigEntry,
+    async_add_entities: AddEntitiesCallback,
+) -> None:
+    """Set up MQTT device_tracker through configuration.yaml and dynamically through MQTT discovery."""
+    await async_setup_entry_from_discovery(hass, config_entry, async_add_entities)
+    # (re)load legacy service
+    if MQTT_DATA_DEVICE_TRACKER_LEGACY in hass.data:
+        await async_setup_scanner_from_yaml(
+            hass, **hass.data[MQTT_DATA_DEVICE_TRACKER_LEGACY]
+        )

homeassistant/components/mqtt/device_tracker/schema_discovery.py

@@ -1,4 +1,6 @@
-"""Support for tracking MQTT enabled devices."""
+"""Support for tracking MQTT enabled devices identified through discovery."""
+from __future__ import annotations
+
 import asyncio
 import functools
 
@@ -7,6 +9,7 @@ import voluptuous as vol
 from homeassistant.components import device_tracker
 from homeassistant.components.device_tracker import SOURCE_TYPES
 from homeassistant.components.device_tracker.config_entry import TrackerEntity
+from homeassistant.config_entries import ConfigEntry
 from homeassistant.const import (
     ATTR_GPS_ACCURACY,
     ATTR_LATITUDE,
@@ -16,8 +19,10 @@ from homeassistant.const import (
     STATE_HOME,
     STATE_NOT_HOME,
 )
-from homeassistant.core import callback
+from homeassistant.core import HomeAssistant, callback
 import homeassistant.helpers.config_validation as cv
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+from homeassistant.helpers.typing import ConfigType
 
 from .. import subscription
 from ..config import MQTT_RO_SCHEMA
@@ -47,7 +52,11 @@ PLATFORM_SCHEMA_MODERN = MQTT_RO_SCHEMA.extend(
 DISCOVERY_SCHEMA = PLATFORM_SCHEMA_MODERN.extend({}, extra=vol.REMOVE_EXTRA)
 
 
-async def async_setup_entry_from_discovery(hass, config_entry, async_add_entities):
+async def async_setup_entry_from_discovery(
+    hass: HomeAssistant,
+    config_entry: ConfigEntry,
+    async_add_entities: AddEntitiesCallback,
+) -> None:
     """Set up MQTT device tracker configuration.yaml and dynamically through MQTT discovery."""
     # load and initialize platform config from configuration.yaml
     await asyncio.gather(
@@ -66,8 +75,12 @@ async def async_setup_entry_from_discovery(hass, config_entry, async_add_entitie
 
 
 async def _async_setup_entity(
-    hass, async_add_entities, config, config_entry=None, discovery_data=None
+    hass: HomeAssistant,
-):
+    async_add_entities: AddEntitiesCallback,
+    config: ConfigType,
+    config_entry: ConfigEntry | None = None,
+    discovery_data: dict | None = None,
+) -> None:
     """Set up the MQTT Device Tracker entity."""
     async_add_entities([MqttDeviceTracker(hass, config, config_entry, discovery_data)])
 

homeassistant/components/mqtt/device_tracker/schema_yaml.py

@@ -1,16 +1,23 @@
 """Support for tracking MQTT enabled devices defined in YAML."""
 
+from collections.abc import Callable
+import logging
+from typing import Any
+
 import voluptuous as vol
 
 from homeassistant.components.device_tracker import PLATFORM_SCHEMA, SOURCE_TYPES
 from homeassistant.const import CONF_DEVICES, STATE_HOME, STATE_NOT_HOME
-from homeassistant.core import callback
+from homeassistant.core import HomeAssistant, callback
-import homeassistant.helpers.config_validation as cv
+from homeassistant.helpers import config_validation as cv
 
+from ... import mqtt
 from ..client import async_subscribe
 from ..config import SCHEMA_BASE
-from ..const import CONF_QOS
+from ..const import CONF_QOS, MQTT_DATA_DEVICE_TRACKER_LEGACY
-from ..util import valid_subscribe_topic
+from ..util import mqtt_config_entry_enabled, valid_subscribe_topic
+
+_LOGGER = logging.getLogger(__name__)
 
 CONF_PAYLOAD_HOME = "payload_home"
 CONF_PAYLOAD_NOT_HOME = "payload_not_home"
@@ -26,13 +33,34 @@ PLATFORM_SCHEMA_YAML = PLATFORM_SCHEMA.extend(SCHEMA_BASE).extend(
 )
 
 
-async def async_setup_scanner_from_yaml(hass, config, async_see, discovery_info=None):
+async def async_setup_scanner_from_yaml(
+    hass: HomeAssistant, config, async_see, discovery_info=None
+):
     """Set up the MQTT tracker."""
     devices = config[CONF_DEVICES]
     qos = config[CONF_QOS]
     payload_home = config[CONF_PAYLOAD_HOME]
     payload_not_home = config[CONF_PAYLOAD_NOT_HOME]
     source_type = config.get(CONF_SOURCE_TYPE)
+    config_entry = hass.config_entries.async_entries(mqtt.DOMAIN)[0]
+    subscriptions: list[Callable] = []
+
+    hass.data[MQTT_DATA_DEVICE_TRACKER_LEGACY] = {
+        "async_see": async_see,
+        "config": config,
+    }
+    if not mqtt_config_entry_enabled(hass):
+        _LOGGER.info(
+            "MQTT device trackers will be not available until the config entry is enabled",
+        )
+        return
+
+    @callback
+    def _entry_unload(*_: Any) -> None:
+        """Handle the unload of the config entry."""
+        # Unsubscribe from mqtt
+        for unsubscribe in subscriptions:
+            unsubscribe()
 
     for dev_id, topic in devices.items():
 
@@ -52,6 +80,10 @@ async def async_setup_scanner_from_yaml(hass, config, async_see, discovery_info=
 
             hass.async_create_task(async_see(**see_args))
 
-        await async_subscribe(hass, topic, async_message_received, qos)
+        subscriptions.append(
+            await async_subscribe(hass, topic, async_message_received, qos)
+        )
+
+    config_entry.async_on_unload(_entry_unload)
 
     return True

homeassistant/components/mqtt/discovery.py

@@ -234,8 +234,7 @@ async def async_start(  # noqa: C901
                 hass, MQTT_DISCOVERY_DONE.format(discovery_hash), None
             )
 
-    hass.data[DATA_CONFIG_FLOW_LOCK] = asyncio.Lock()
+    hass.data.setdefault(DATA_CONFIG_FLOW_LOCK, asyncio.Lock())
-
     hass.data[ALREADY_DISCOVERED] = {}
     hass.data[PENDING_DISCOVERED] = {}
 

homeassistant/components/mqtt/fan.py

@@ -241,8 +241,12 @@ async def async_setup_entry(
 
 
 async def _async_setup_entity(
-    hass, async_add_entities, config, config_entry=None, discovery_data=None
+    hass: HomeAssistant,
-):
+    async_add_entities: AddEntitiesCallback,
+    config: ConfigType,
+    config_entry: ConfigEntry | None = None,
+    discovery_data: dict | None = None,
+) -> None:
     """Set up the MQTT fan."""
     async_add_entities([MqttFan(hass, config, config_entry, discovery_data)])
 

homeassistant/components/mqtt/humidifier.py

@@ -197,8 +197,12 @@ async def async_setup_entry(
 
 
 async def _async_setup_entity(
-    hass, async_add_entities, config, config_entry=None, discovery_data=None
+    hass: HomeAssistant,
-):
+    async_add_entities: AddEntitiesCallback,
+    config: ConfigType,
+    config_entry: ConfigEntry | None = None,
+    discovery_data: dict | None = None,
+) -> None:
     """Set up the MQTT humidifier."""
     async_add_entities([MqttHumidifier(hass, config, config_entry, discovery_data)])
 

homeassistant/components/mqtt/light/__init__.py

@@ -1,6 +1,7 @@
 """Support for MQTT lights."""
 from __future__ import annotations
 
+from collections.abc import Callable
 import functools
 
 import voluptuous as vol
@@ -120,10 +121,14 @@ async def async_setup_entry(
 
 
 async def _async_setup_entity(
-    hass, async_add_entities, config, config_entry=None, discovery_data=None
+    hass: HomeAssistant,
-):
+    async_add_entities: AddEntitiesCallback,
+    config: ConfigType,
+    config_entry: ConfigEntry | None = None,
+    discovery_data: dict | None = None,
+) -> None:
     """Set up a MQTT Light."""
-    setup_entity = {
+    setup_entity: dict[str, Callable] = {
         "basic": async_setup_entity_basic,
         "json": async_setup_entity_json,
         "template": async_setup_entity_template,

homeassistant/components/mqtt/lock.py

@@ -112,8 +112,12 @@ async def async_setup_entry(
 
 
 async def _async_setup_entity(
-    hass, async_add_entities, config, config_entry=None, discovery_data=None
+    hass: HomeAssistant,
-):
+    async_add_entities: AddEntitiesCallback,
+    config: ConfigType,
+    config_entry: ConfigEntry | None = None,
+    discovery_data: dict | None = None,
+) -> None:
     """Set up the MQTT Lock platform."""
     async_add_entities([MqttLock(hass, config, config_entry, discovery_data)])
 

homeassistant/components/mqtt/mixins.py

@@ -3,7 +3,8 @@ from __future__ import annotations
 
 from abc import abstractmethod
 import asyncio
-from collections.abc import Callable
+from collections.abc import Callable, Coroutine
+from functools import partial
 import logging
 from typing import Any, Protocol, cast, final
 
@@ -61,7 +62,7 @@ from .const import (
     CONF_TOPIC,
     DATA_MQTT,
     DATA_MQTT_CONFIG,
-    DATA_MQTT_RELOAD_NEEDED,
+    DATA_MQTT_RELOAD_DISPATCHERS,
     DATA_MQTT_UPDATED_CONFIG,
     DEFAULT_ENCODING,
     DEFAULT_PAYLOAD_AVAILABLE,
@@ -84,7 +85,7 @@ from .subscription import (
     async_subscribe_topics,
     async_unsubscribe_topics,
 )
-from .util import valid_subscribe_topic
+from .util import mqtt_config_entry_enabled, valid_subscribe_topic
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -299,11 +300,24 @@ async def async_get_platform_config_from_yaml(
     return platform_configs
 
 
-async def async_setup_entry_helper(hass, domain, async_setup, schema):
+async def async_setup_entry_helper(
+    hass: HomeAssistant,
+    domain: str,
+    async_setup: partial[Coroutine[HomeAssistant, str, None]],
+    schema: vol.Schema,
+) -> None:
     """Set up entity, automation or tag creation dynamically through MQTT discovery."""
 
     async def async_discover(discovery_payload):
         """Discover and add an MQTT entity, automation or tag."""
+        if not mqtt_config_entry_enabled(hass):
+            _LOGGER.warning(
+                "MQTT integration is disabled, skipping setup of discovered item "
+                "MQTT %s, payload %s",
+                domain,
+                discovery_payload,
+            )
+            return
         discovery_data = discovery_payload.discovery_data
         try:
             config = schema(discovery_payload)
@@ -316,8 +330,10 @@ async def async_setup_entry_helper(hass, domain, async_setup, schema):
             )
             raise
 
-    async_dispatcher_connect(
+    hass.data.setdefault(DATA_MQTT_RELOAD_DISPATCHERS, []).append(
-        hass, MQTT_DISCOVERY_NEW.format(domain, "mqtt"), async_discover
+        async_dispatcher_connect(
+            hass, MQTT_DISCOVERY_NEW.format(domain, "mqtt"), async_discover
+        )
     )
 
 
@@ -328,16 +344,17 @@ async def async_setup_platform_helper(
     async_add_entities: AddEntitiesCallback,
     async_setup_entities: SetupEntity,
 ) -> None:
-    """Return true if platform setup should be aborted."""
+    """Help to set up the platform for manual configured MQTT entities."""
-    if not bool(hass.config_entries.async_entries(DOMAIN)):
+    if not (entry_status := mqtt_config_entry_enabled(hass)):
-        hass.data[DATA_MQTT_RELOAD_NEEDED] = None
         _LOGGER.warning(
-            "MQTT integration is not setup, skipping setup of manually configured "
+            "MQTT integration is %s, skipping setup of manually configured MQTT %s",
-            "MQTT %s",
+            "not setup" if entry_status is None else "disabled",
             platform_domain,
         )
         return
-    await async_setup_entities(hass, async_add_entities, config)
+    # Ensure we set config_entry when entries are set up to enable clean up
+    config_entry: ConfigEntry = hass.config_entries.async_entries(DOMAIN)[0]
+    await async_setup_entities(hass, async_add_entities, config, config_entry)
 
 
 def init_entity_id_from_config(hass, entity, config, entity_id_format):
@@ -640,6 +657,7 @@ class MqttDiscoveryDeviceUpdate:
             MQTT_DISCOVERY_UPDATED.format(discovery_hash),
             self.async_discovery_update,
         )
+        config_entry.async_on_unload(self._entry_unload)
         if device_id is not None:
             self._remove_device_updated = hass.bus.async_listen(
                 EVENT_DEVICE_REGISTRY_UPDATED, self._async_device_removed
@@ -650,6 +668,14 @@ class MqttDiscoveryDeviceUpdate:
             discovery_hash,
         )
 
+    @callback
+    def _entry_unload(self, *_: Any) -> None:
+        """Handle cleanup when the config entry is unloaded."""
+        stop_discovery_updates(
+            self.hass, self._discovery_data, self._remove_discovery_updated
+        )
+        self.hass.async_add_job(self.async_tear_down())
+
     async def async_discovery_update(
         self,
         discovery_payload: DiscoveryInfoType | None,
@@ -734,7 +760,11 @@ class MqttDiscoveryDeviceUpdate:
 class MqttDiscoveryUpdate(Entity):
     """Mixin used to handle updated discovery message for entity based platforms."""
 
-    def __init__(self, discovery_data, discovery_update=None) -> None:
+    def __init__(
+        self,
+        discovery_data: dict,
+        discovery_update: Callable | None = None,
+    ) -> None:
         """Initialize the discovery update mixin."""
         self._discovery_data = discovery_data
         self._discovery_update = discovery_update

homeassistant/components/mqtt/number.py

@@ -145,8 +145,12 @@ async def async_setup_entry(
 
 
 async def _async_setup_entity(
-    hass, async_add_entities, config, config_entry=None, discovery_data=None
+    hass: HomeAssistant,
-):
+    async_add_entities: AddEntitiesCallback,
+    config: ConfigType,
+    config_entry: ConfigEntry | None = None,
+    discovery_data: dict | None = None,
+) -> None:
     """Set up the MQTT number."""
     async_add_entities([MqttNumber(hass, config, config_entry, discovery_data)])
 

homeassistant/components/mqtt/scene.py

@@ -88,8 +88,12 @@ async def async_setup_entry(
 
 
 async def _async_setup_entity(
-    hass, async_add_entities, config, config_entry=None, discovery_data=None
+    hass: HomeAssistant,
-):
+    async_add_entities: AddEntitiesCallback,
+    config: ConfigType,
+    config_entry: ConfigEntry | None = None,
+    discovery_data: dict | None = None,
+) -> None:
     """Set up the MQTT scene."""
     async_add_entities([MqttScene(hass, config, config_entry, discovery_data)])
 

homeassistant/components/mqtt/select.py

@@ -103,8 +103,12 @@ async def async_setup_entry(
 
 
 async def _async_setup_entity(
-    hass, async_add_entities, config, config_entry=None, discovery_data=None
+    hass: HomeAssistant,
-):
+    async_add_entities: AddEntitiesCallback,
+    config: ConfigType,
+    config_entry: ConfigEntry | None = None,
+    discovery_data: dict | None = None,
+) -> None:
     """Set up the MQTT select."""
     async_add_entities([MqttSelect(hass, config, config_entry, discovery_data)])
 

homeassistant/components/mqtt/sensor.py

@@ -156,8 +156,12 @@ async def async_setup_entry(
 
 
 async def _async_setup_entity(
-    hass, async_add_entities, config: ConfigType, config_entry=None, discovery_data=None
+    hass: HomeAssistant,
-):
+    async_add_entities: AddEntitiesCallback,
+    config: ConfigType,
+    config_entry: ConfigEntry | None = None,
+    discovery_data: dict | None = None,
+) -> None:
     """Set up MQTT sensor."""
     async_add_entities([MqttSensor(hass, config, config_entry, discovery_data)])
 

homeassistant/components/mqtt/siren.py

@@ -152,8 +152,12 @@ async def async_setup_entry(
 
 
 async def _async_setup_entity(
-    hass, async_add_entities, config, config_entry=None, discovery_data=None
+    hass: HomeAssistant,
-):
+    async_add_entities: AddEntitiesCallback,
+    config: ConfigType,
+    config_entry: ConfigEntry | None = None,
+    discovery_data: dict | None = None,
+) -> None:
     """Set up the MQTT siren."""
     async_add_entities([MqttSiren(hass, config, config_entry, discovery_data)])
 

homeassistant/components/mqtt/switch.py

@@ -111,8 +111,12 @@ async def async_setup_entry(
 
 
 async def _async_setup_entity(
-    hass, async_add_entities, config, config_entry=None, discovery_data=None
+    hass: HomeAssistant,
-):
+    async_add_entities: AddEntitiesCallback,
+    config: ConfigType,
+    config_entry: ConfigEntry | None = None,
+    discovery_data: dict | None = None,
+) -> None:
     """Set up the MQTT switch."""
     async_add_entities([MqttSwitch(hass, config, config_entry, discovery_data)])
 

homeassistant/components/mqtt/util.py

@@ -1,9 +1,13 @@
 """Utility functions for the MQTT integration."""
+
+from __future__ import annotations
+
 from typing import Any
 
 import voluptuous as vol
 
 from homeassistant.const import CONF_PAYLOAD
+from homeassistant.core import HomeAssistant
 from homeassistant.helpers import config_validation as cv, template
 
 from .const import (
@@ -13,9 +17,17 @@ from .const import (
     ATTR_TOPIC,
     DEFAULT_QOS,
     DEFAULT_RETAIN,
+    DOMAIN,
 )
 
 
+def mqtt_config_entry_enabled(hass: HomeAssistant) -> bool | None:
+    """Return true when the MQTT config entry is enabled."""
+    if not bool(hass.config_entries.async_entries(DOMAIN)):
+        return None
+    return not bool(hass.config_entries.async_entries(DOMAIN)[0].disabled_by)
+
+
 def valid_topic(value: Any) -> str:
     """Validate that this is a valid topic name/filter."""
     value = cv.string(value)

homeassistant/components/mqtt/vacuum/__init__.py

@@ -100,10 +100,17 @@ async def async_setup_entry(
 
 
 async def _async_setup_entity(
-    hass, async_add_entities, config, config_entry=None, discovery_data=None
+    hass: HomeAssistant,
-):
+    async_add_entities: AddEntitiesCallback,
+    config: ConfigType,
+    config_entry: ConfigEntry | None = None,
+    discovery_data: dict | None = None,
+) -> None:
     """Set up the MQTT vacuum."""
-    setup_entity = {LEGACY: async_setup_entity_legacy, STATE: async_setup_entity_state}
+    setup_entity = {
+        LEGACY: async_setup_entity_legacy,
+        STATE: async_setup_entity_state,
+    }
     await setup_entity[config[CONF_SCHEMA]](
         hass, config, async_add_entities, config_entry, discovery_data
     )

tests/components/mqtt/test_alarm_control_panel.py

@@ -60,6 +60,7 @@ from .test_common import (
     help_test_setting_blocked_attribute_via_mqtt_json_message,
     help_test_setup_manual_entity_from_yaml,
     help_test_unique_id,
+    help_test_unload_config_entry_with_platform,
     help_test_update_with_json_attrs_bad_JSON,
     help_test_update_with_json_attrs_not_dict,
 )
@@ -971,3 +972,12 @@ async def test_setup_manual_entity_from_yaml(hass):
     del config["platform"]
     await help_test_setup_manual_entity_from_yaml(hass, platform, config)
     assert hass.states.get(f"{platform}.test") is not None
+
+
+async def test_unload_entry(hass, mqtt_mock_entry_with_yaml_config, tmp_path):
+    """Test unloading the config entry."""
+    domain = alarm_control_panel.DOMAIN
+    config = DEFAULT_CONFIG[domain]
+    await help_test_unload_config_entry_with_platform(
+        hass, mqtt_mock_entry_with_yaml_config, tmp_path, domain, config
+    )

tests/components/mqtt/test_binary_sensor.py

@@ -44,6 +44,7 @@ from .test_common import (
     help_test_setting_attribute_with_template,
     help_test_setup_manual_entity_from_yaml,
     help_test_unique_id,
+    help_test_unload_config_entry_with_platform,
     help_test_update_with_json_attrs_bad_JSON,
     help_test_update_with_json_attrs_not_dict,
 )
@@ -1079,3 +1080,12 @@ async def test_setup_manual_entity_from_yaml(hass):
     del config["platform"]
     await help_test_setup_manual_entity_from_yaml(hass, platform, config)
     assert hass.states.get(f"{platform}.test") is not None
+
+
+async def test_unload_entry(hass, mqtt_mock_entry_with_yaml_config, tmp_path):
+    """Test unloading the config entry."""
+    domain = binary_sensor.DOMAIN
+    config = DEFAULT_CONFIG[domain]
+    await help_test_unload_config_entry_with_platform(
+        hass, mqtt_mock_entry_with_yaml_config, tmp_path, domain, config
+    )

tests/components/mqtt/test_button.py

@@ -37,6 +37,7 @@ from .test_common import (
     help_test_setting_blocked_attribute_via_mqtt_json_message,
     help_test_setup_manual_entity_from_yaml,
     help_test_unique_id,
+    help_test_unload_config_entry_with_platform,
     help_test_update_with_json_attrs_bad_JSON,
     help_test_update_with_json_attrs_not_dict,
 )
@@ -482,3 +483,12 @@ async def test_setup_manual_entity_from_yaml(hass):
     del config["platform"]
     await help_test_setup_manual_entity_from_yaml(hass, platform, config)
     assert hass.states.get(f"{platform}.test") is not None
+
+
+async def test_unload_entry(hass, mqtt_mock_entry_with_yaml_config, tmp_path):
+    """Test unloading the config entry."""
+    domain = button.DOMAIN
+    config = DEFAULT_CONFIG[domain]
+    await help_test_unload_config_entry_with_platform(
+        hass, mqtt_mock_entry_with_yaml_config, tmp_path, domain, config
+    )

tests/components/mqtt/test_camera.py

@@ -36,6 +36,7 @@ from .test_common import (
     help_test_setting_blocked_attribute_via_mqtt_json_message,
     help_test_setup_manual_entity_from_yaml,
     help_test_unique_id,
+    help_test_unload_config_entry_with_platform,
     help_test_update_with_json_attrs_bad_JSON,
     help_test_update_with_json_attrs_not_dict,
 )
@@ -346,3 +347,12 @@ async def test_setup_manual_entity_from_yaml(hass):
     del config["platform"]
     await help_test_setup_manual_entity_from_yaml(hass, platform, config)
     assert hass.states.get(f"{platform}.test") is not None
+
+
+async def test_unload_entry(hass, mqtt_mock_entry_with_yaml_config, tmp_path):
+    """Test unloading the config entry."""
+    domain = camera.DOMAIN
+    config = DEFAULT_CONFIG[domain]
+    await help_test_unload_config_entry_with_platform(
+        hass, mqtt_mock_entry_with_yaml_config, tmp_path, domain, config
+    )

tests/components/mqtt/test_climate.py

@@ -55,6 +55,7 @@ from .test_common import (
     help_test_setting_blocked_attribute_via_mqtt_json_message,
     help_test_setup_manual_entity_from_yaml,
     help_test_unique_id,
+    help_test_unload_config_entry_with_platform,
     help_test_update_with_json_attrs_bad_JSON,
     help_test_update_with_json_attrs_not_dict,
 )
@@ -1881,3 +1882,12 @@ async def test_setup_manual_entity_from_yaml(hass):
     del config["platform"]
     await help_test_setup_manual_entity_from_yaml(hass, platform, config)
     assert hass.states.get(f"{platform}.test") is not None
+
+
+async def test_unload_entry(hass, mqtt_mock_entry_with_yaml_config, tmp_path):
+    """Test unloading the config entry."""
+    domain = climate.DOMAIN
+    config = DEFAULT_CONFIG[domain]
+    await help_test_unload_config_entry_with_platform(
+        hass, mqtt_mock_entry_with_yaml_config, tmp_path, domain, config
+    )

tests/components/mqtt/test_common.py

@@ -2,7 +2,7 @@
 import copy
 from datetime import datetime
 import json
-from unittest.mock import ANY, patch
+from unittest.mock import ANY, MagicMock, patch
 
 import yaml
 
@@ -11,6 +11,7 @@ from homeassistant.components import mqtt
 from homeassistant.components.mqtt import debug_info
 from homeassistant.components.mqtt.const import MQTT_DISCONNECTED
 from homeassistant.components.mqtt.mixins import MQTT_ATTRIBUTES_BLOCKED
+from homeassistant.config_entries import ConfigEntryState
 from homeassistant.const import (
     ATTR_ASSUMED_STATE,
     ATTR_ENTITY_ID,
@@ -1670,6 +1671,25 @@ async def help_test_reload_with_config(hass, caplog, tmp_path, config):
     assert "<Event event_mqtt_reloaded[L]>" in caplog.text
 
 
+async def help_test_entry_reload_with_new_config(hass, tmp_path, new_config):
+    """Test reloading with supplied config."""
+    mqtt_config_entry = hass.config_entries.async_entries(mqtt.DOMAIN)[0]
+    assert mqtt_config_entry.state is ConfigEntryState.LOADED
+    new_yaml_config_file = tmp_path / "configuration.yaml"
+    new_yaml_config = yaml.dump(new_config)
+    new_yaml_config_file.write_text(new_yaml_config)
+    assert new_yaml_config_file.read_text() == new_yaml_config
+
+    with patch.object(hass_config, "YAML_CONFIG_FILE", new_yaml_config_file), patch(
+        "paho.mqtt.client.Client"
+    ) as mock_client:
+        mock_client().connect = lambda *args: 0
+        # reload the config entry
+        assert await hass.config_entries.async_reload(mqtt_config_entry.entry_id)
+        assert mqtt_config_entry.state is ConfigEntryState.LOADED
+        await hass.async_block_till_done()
+
+
 async def help_test_reloadable(
     hass, mqtt_mock_entry_with_yaml_config, caplog, tmp_path, domain, config
 ):
@@ -1782,6 +1802,7 @@ async def help_test_reloadable_late(hass, caplog, tmp_path, domain, config):
         domain: [new_config_1, new_config_2, new_config_3],
     }
     await help_test_reload_with_config(hass, caplog, tmp_path, new_config)
+    await hass.async_block_till_done()
 
     assert len(hass.states.async_all(domain)) == 3
 
@@ -1792,6 +1813,12 @@ async def help_test_reloadable_late(hass, caplog, tmp_path, domain, config):
 
 async def help_test_setup_manual_entity_from_yaml(hass, platform, config):
     """Help to test setup from yaml through configuration entry."""
+    calls = MagicMock()
+
+    async def mock_reload(hass):
+        """Mock reload."""
+        calls()
+
     config_structure = {mqtt.DOMAIN: {platform: config}}
 
     await async_setup_component(hass, mqtt.DOMAIN, config_structure)
@@ -1799,7 +1826,71 @@ async def help_test_setup_manual_entity_from_yaml(hass, platform, config):
     entry = MockConfigEntry(domain=mqtt.DOMAIN, data={mqtt.CONF_BROKER: "test-broker"})
     entry.add_to_hass(hass)
 
-    with patch("paho.mqtt.client.Client") as mock_client:
+    with patch(
+        "homeassistant.components.mqtt.async_reload_manual_mqtt_items",
+        side_effect=mock_reload,
+    ), patch("paho.mqtt.client.Client") as mock_client:
         mock_client().connect = lambda *args: 0
         assert await hass.config_entries.async_setup(entry.entry_id)
         await hass.async_block_till_done()
+        calls.assert_called_once()
+
+
+async def help_test_unload_config_entry(hass, tmp_path, newconfig):
+    """Test unloading the MQTT config entry."""
+    mqtt_config_entry = hass.config_entries.async_entries(mqtt.DOMAIN)[0]
+    assert mqtt_config_entry.state is ConfigEntryState.LOADED
+
+    new_yaml_config_file = tmp_path / "configuration.yaml"
+    new_yaml_config = yaml.dump(newconfig)
+    new_yaml_config_file.write_text(new_yaml_config)
+    with patch.object(hass_config, "YAML_CONFIG_FILE", new_yaml_config_file):
+        assert await hass.config_entries.async_unload(mqtt_config_entry.entry_id)
+        assert mqtt_config_entry.state is ConfigEntryState.NOT_LOADED
+        await hass.async_block_till_done()
+
+
+async def help_test_unload_config_entry_with_platform(
+    hass,
+    mqtt_mock_entry_with_yaml_config,
+    tmp_path,
+    domain,
+    config,
+):
+    """Test unloading the MQTT config entry with a specific platform domain."""
+    # prepare setup through configuration.yaml
+    config_setup = copy.deepcopy(config)
+    config_setup["name"] = "config_setup"
+    config_name = config_setup
+    assert await async_setup_component(hass, domain, {domain: [config_setup]})
+    await hass.async_block_till_done()
+    await mqtt_mock_entry_with_yaml_config()
+
+    # prepare setup through discovery
+    discovery_setup = copy.deepcopy(config)
+    discovery_setup["name"] = "discovery_setup"
+    async_fire_mqtt_message(
+        hass, f"homeassistant/{domain}/bla/config", json.dumps(discovery_setup)
+    )
+    await hass.async_block_till_done()
+
+    # check if both entities were setup correctly
+    config_setup_entity = hass.states.get(f"{domain}.config_setup")
+    assert config_setup_entity
+
+    discovery_setup_entity = hass.states.get(f"{domain}.discovery_setup")
+    assert discovery_setup_entity
+
+    await help_test_unload_config_entry(hass, tmp_path, config_setup)
+
+    async_fire_mqtt_message(
+        hass, f"homeassistant/{domain}/bla/config", json.dumps(discovery_setup)
+    )
+    await hass.async_block_till_done()
+
+    # check if both entities were unloaded correctly
+    config_setup_entity = hass.states.get(f"{domain}.{config_name}")
+    assert config_setup_entity is None
+
+    discovery_setup_entity = hass.states.get(f"{domain}.discovery_setup")
+    assert discovery_setup_entity is None

tests/components/mqtt/test_config_flow.py

@@ -145,12 +145,17 @@ async def test_manual_config_starts_discovery_flow(
 
 
 async def test_manual_config_set(
-    hass, mock_try_connection, mock_finish_setup, mqtt_client_mock
+    hass,
+    mock_try_connection,
+    mock_finish_setup,
+    mqtt_client_mock,
 ):
     """Test manual config does not create an entry, and entry can be setup late."""
     # MQTT config present in yaml config
     assert await async_setup_component(hass, "mqtt", {"mqtt": {"broker": "bla"}})
     await hass.async_block_till_done()
+    # do not try to reload
+    del hass.data["mqtt_reload_needed"]
     assert len(mock_finish_setup.mock_calls) == 0
 
     mock_try_connection.return_value = True

tests/components/mqtt/test_cover.py

@@ -73,6 +73,7 @@ from .test_common import (
     help_test_setting_blocked_attribute_via_mqtt_json_message,
     help_test_setup_manual_entity_from_yaml,
     help_test_unique_id,
+    help_test_unload_config_entry_with_platform,
     help_test_update_with_json_attrs_bad_JSON,
     help_test_update_with_json_attrs_not_dict,
 )
@@ -3364,3 +3365,12 @@ async def test_setup_manual_entity_from_yaml(hass):
     del config["platform"]
     await help_test_setup_manual_entity_from_yaml(hass, platform, config)
     assert hass.states.get(f"{platform}.test") is not None
+
+
+async def test_unload_entry(hass, mqtt_mock_entry_with_yaml_config, tmp_path):
+    """Test unloading the config entry."""
+    domain = cover.DOMAIN
+    config = DEFAULT_CONFIG[domain]
+    await help_test_unload_config_entry_with_platform(
+        hass, mqtt_mock_entry_with_yaml_config, tmp_path, domain, config
+    )

tests/components/mqtt/test_device_tracker.py

@@ -1,13 +1,20 @@
 """The tests for the MQTT device tracker platform using configuration.yaml."""
+import json
 from unittest.mock import patch
 
 import pytest
 
 from homeassistant.components.device_tracker.const import DOMAIN, SOURCE_TYPE_BLUETOOTH
+from homeassistant.config_entries import ConfigEntryDisabler
 from homeassistant.const import CONF_PLATFORM, STATE_HOME, STATE_NOT_HOME, Platform
 from homeassistant.setup import async_setup_component
 
-from .test_common import help_test_setup_manual_entity_from_yaml
+from .test_common import (
+    MockConfigEntry,
+    help_test_entry_reload_with_new_config,
+    help_test_setup_manual_entity_from_yaml,
+    help_test_unload_config_entry,
+)
 
 from tests.common import async_fire_mqtt_message
 
@@ -265,3 +272,114 @@ async def test_setup_with_modern_schema(hass, mock_device_tracker_conf):
     await help_test_setup_manual_entity_from_yaml(hass, DOMAIN, config)
 
     assert hass.states.get(entity_id) is not None
+
+
+async def test_unload_entry(
+    hass, mock_device_tracker_conf, mqtt_mock_entry_no_yaml_config, tmp_path
+):
+    """Test unloading the config entry."""
+    # setup through configuration.yaml
+    await mqtt_mock_entry_no_yaml_config()
+    dev_id = "jan"
+    entity_id = f"{DOMAIN}.{dev_id}"
+    topic = "/location/jan"
+    location = "home"
+
+    hass.config.components = {"mqtt", "zone"}
+    assert await async_setup_component(
+        hass, DOMAIN, {DOMAIN: {CONF_PLATFORM: "mqtt", "devices": {dev_id: topic}}}
+    )
+    async_fire_mqtt_message(hass, topic, location)
+    await hass.async_block_till_done()
+    assert hass.states.get(entity_id).state == location
+
+    # setup through discovery
+    dev_id = "piet"
+    subscription = "/location/#"
+    domain = DOMAIN
+    discovery_config = {
+        "devices": {dev_id: subscription},
+        "state_topic": "some-state",
+        "name": "piet",
+    }
+    async_fire_mqtt_message(
+        hass, f"homeassistant/{domain}/bla/config", json.dumps(discovery_config)
+    )
+    await hass.async_block_till_done()
+
+    # check that both entities were created
+    config_setup_entity = hass.states.get(f"{domain}.jan")
+    assert config_setup_entity
+
+    discovery_setup_entity = hass.states.get(f"{domain}.piet")
+    assert discovery_setup_entity
+
+    await help_test_unload_config_entry(hass, tmp_path, {})
+    await hass.async_block_till_done()
+
+    # check that both entities were unsubscribed and that the location was not processed
+    async_fire_mqtt_message(hass, "some-state", "not_home")
+    async_fire_mqtt_message(hass, "location/jan", "not_home")
+    await hass.async_block_till_done()
+
+    config_setup_entity = hass.states.get(f"{domain}.jan")
+    assert config_setup_entity.state == location
+
+    # the discovered tracker is an entity which state is removed at unload
+    discovery_setup_entity = hass.states.get(f"{domain}.piet")
+    assert discovery_setup_entity is None
+
+
+async def test_reload_entry_legacy(
+    hass, mock_device_tracker_conf, mqtt_mock_entry_no_yaml_config, tmp_path
+):
+    """Test reloading the config entry with manual MQTT items."""
+    # setup through configuration.yaml
+    await mqtt_mock_entry_no_yaml_config()
+    entity_id = f"{DOMAIN}.jan"
+    topic = "location/jan"
+    location = "home"
+
+    config = {
+        DOMAIN: {CONF_PLATFORM: "mqtt", "devices": {"jan": topic}},
+    }
+    hass.config.components = {"mqtt", "zone"}
+    assert await async_setup_component(hass, DOMAIN, config)
+    await hass.async_block_till_done()
+
+    async_fire_mqtt_message(hass, topic, location)
+    await hass.async_block_till_done()
+    assert hass.states.get(entity_id).state == location
+
+    await help_test_entry_reload_with_new_config(hass, tmp_path, config)
+    await hass.async_block_till_done()
+
+    location = "not_home"
+    async_fire_mqtt_message(hass, topic, location)
+    await hass.async_block_till_done()
+    assert hass.states.get(entity_id).state == location
+
+
+async def test_setup_with_disabled_entry(
+    hass, mock_device_tracker_conf, caplog
+) -> None:
+    """Test setting up the platform with a disabled config entry."""
+    # Try to setup the platform with a disabled config entry
+    config_entry = MockConfigEntry(
+        domain="mqtt", data={}, disabled_by=ConfigEntryDisabler.USER
+    )
+    config_entry.add_to_hass(hass)
+    topic = "location/jan"
+
+    config = {
+        DOMAIN: {CONF_PLATFORM: "mqtt", "devices": {"jan": topic}},
+    }
+    hass.config.components = {"mqtt", "zone"}
+
+    await async_setup_component(hass, DOMAIN, config)
+    await hass.async_block_till_done()
+
+    assert (
+        "MQTT device trackers will be not available until the config entry is enabled"
+        in caplog.text
+    )

tests/components/mqtt/test_device_trigger.py

@@ -12,6 +12,8 @@ from homeassistant.helpers import device_registry as dr
 from homeassistant.helpers.trigger import async_initialize_triggers
 from homeassistant.setup import async_setup_component
 
+from .test_common import help_test_unload_config_entry
+
 from tests.common import (
     assert_lists_same,
     async_fire_mqtt_message,
@@ -1372,3 +1374,53 @@ async def test_trigger_debug_info(hass, mqtt_mock_entry_no_yaml_config):
         == "homeassistant/device_automation/bla2/config"
     )
     assert debug_info_data["triggers"][0]["discovery_data"]["payload"] == config2
+
+
+async def test_unload_entry(hass, calls, device_reg, mqtt_mock, tmp_path) -> None:
+    """Test unloading the MQTT entry."""
+
+    data1 = (
+        '{ "automation_type":"trigger",'
+        '  "device":{"identifiers":["0AFFD2"]},'
+        '  "topic": "foobar/triggers/button1",'
+        '  "type": "button_short_press",'
+        '  "subtype": "button_1" }'
+    )
+    async_fire_mqtt_message(hass, "homeassistant/device_automation/bla1/config", data1)
+    await hass.async_block_till_done()
+    device_entry = device_reg.async_get_device({("mqtt", "0AFFD2")})
+
+    assert await async_setup_component(
+        hass,
+        automation.DOMAIN,
+        {
+            automation.DOMAIN: [
+                {
+                    "trigger": {
+                        "platform": "device",
+                        "domain": DOMAIN,
+                        "device_id": device_entry.id,
+                        "discovery_id": "bla1",
+                        "type": "button_short_press",
+                        "subtype": "button_1",
+                    },
+                    "action": {
+                        "service": "test.automation",
+                        "data_template": {"some": ("short_press")},
+                    },
+                },
+            ]
+        },
+    )
+
+    # Fake short press 1
+    async_fire_mqtt_message(hass, "foobar/triggers/button1", "short_press")
+    await hass.async_block_till_done()
+    assert len(calls) == 1
+
+    await help_test_unload_config_entry(hass, tmp_path, {})
+
+    # Fake short press 2
+    async_fire_mqtt_message(hass, "foobar/triggers/button1", "short_press")
+    await hass.async_block_till_done()
+    assert len(calls) == 1

tests/components/mqtt/test_fan.py

@@ -58,6 +58,7 @@ from .test_common import (
     help_test_setting_blocked_attribute_via_mqtt_json_message,
     help_test_setup_manual_entity_from_yaml,
     help_test_unique_id,
+    help_test_unload_config_entry_with_platform,
     help_test_update_with_json_attrs_bad_JSON,
     help_test_update_with_json_attrs_not_dict,
 )
@@ -1910,3 +1911,12 @@ async def test_setup_manual_entity_from_yaml(hass):
     del config["platform"]
     await help_test_setup_manual_entity_from_yaml(hass, platform, config)
     assert hass.states.get(f"{platform}.test") is not None
+
+
+async def test_unload_entry(hass, mqtt_mock_entry_with_yaml_config, tmp_path):
+    """Test unloading the config entry."""
+    domain = fan.DOMAIN
+    config = DEFAULT_CONFIG[domain]
+    await help_test_unload_config_entry_with_platform(
+        hass, mqtt_mock_entry_with_yaml_config, tmp_path, domain, config
+    )

tests/components/mqtt/test_humidifier.py

@@ -60,6 +60,7 @@ from .test_common import (
     help_test_setting_blocked_attribute_via_mqtt_json_message,
     help_test_setup_manual_entity_from_yaml,
     help_test_unique_id,
+    help_test_unload_config_entry_with_platform,
     help_test_update_with_json_attrs_bad_JSON,
     help_test_update_with_json_attrs_not_dict,
 )
@@ -1296,3 +1297,12 @@ async def test_config_schema_validation(hass):
     CONFIG_SCHEMA({DOMAIN: {platform: [config]}})
     with pytest.raises(MultipleInvalid):
         CONFIG_SCHEMA({"mqtt": {"humidifier": [{"bla": "bla"}]}})
+
+
+async def test_unload_config_entry(hass, mqtt_mock_entry_with_yaml_config, tmp_path):
+    """Test unloading the config entry."""
+    domain = humidifier.DOMAIN
+    config = DEFAULT_CONFIG[domain]
+    await help_test_unload_config_entry_with_platform(
+        hass, mqtt_mock_entry_with_yaml_config, tmp_path, domain, config
+    )

tests/components/mqtt/test_init.py

@@ -17,6 +17,7 @@ from homeassistant.components import mqtt
 from homeassistant.components.mqtt import CONFIG_SCHEMA, debug_info
 from homeassistant.components.mqtt.mixins import MQTT_ENTITY_DEVICE_INFO_SCHEMA
 from homeassistant.components.mqtt.models import ReceiveMessage
+from homeassistant.config_entries import ConfigEntryDisabler, ConfigEntryState
 from homeassistant.const import (
     ATTR_ASSUMED_STATE,
     EVENT_HOMEASSISTANT_STARTED,
@@ -32,7 +33,10 @@ from homeassistant.helpers.entity import Entity
 from homeassistant.setup import async_setup_component
 from homeassistant.util.dt import utcnow
 
-from .test_common import help_test_setup_manual_entity_from_yaml
+from .test_common import (
+    help_test_entry_reload_with_new_config,
+    help_test_setup_manual_entity_from_yaml,
+)
 
 from tests.common import (
     MockConfigEntry,
@@ -106,6 +110,18 @@ def record_calls(calls):
     return record_calls
 
 
+@pytest.fixture
+def empty_mqtt_config(hass, tmp_path):
+    """Fixture to provide an empty config from yaml."""
+    new_yaml_config_file = tmp_path / "configuration.yaml"
+    new_yaml_config_file.write_text("")
+
+    with patch.object(
+        hass_config, "YAML_CONFIG_FILE", new_yaml_config_file
+    ) as empty_config:
+        yield empty_config
+
+
 async def test_mqtt_connects_on_home_assistant_mqtt_setup(
     hass, mqtt_client_mock, mqtt_mock_entry_no_yaml_config
 ):
@@ -115,14 +131,14 @@ async def test_mqtt_connects_on_home_assistant_mqtt_setup(
 
 
 async def test_mqtt_disconnects_on_home_assistant_stop(
-    hass, mqtt_mock_entry_no_yaml_config
+    hass, mqtt_mock_entry_no_yaml_config, mqtt_client_mock
 ):
     """Test if client stops on HA stop."""
-    mqtt_mock = await mqtt_mock_entry_no_yaml_config()
+    await mqtt_mock_entry_no_yaml_config()
     hass.bus.fire(EVENT_HOMEASSISTANT_STOP)
     await hass.async_block_till_done()
     await hass.async_block_till_done()
-    assert mqtt_mock.async_disconnect.called
+    assert mqtt_client_mock.loop_stop.call_count == 1
 
 
 async def test_publish(hass, mqtt_mock_entry_no_yaml_config):
@@ -521,8 +537,11 @@ async def test_service_call_with_ascii_qos_retain_flags(
     assert not mqtt_mock.async_publish.call_args[0][3]
 
 
-async def test_publish_function_with_bad_encoding_conditions(hass, caplog):
+async def test_publish_function_with_bad_encoding_conditions(
-    """Test internal publish function with bas use cases."""
+    hass, caplog, mqtt_mock_entry_no_yaml_config
+):
+    """Test internal publish function with basic use cases."""
+    await mqtt_mock_entry_no_yaml_config()
     await mqtt.async_publish(
         hass, "some-topic", "test-payload", qos=0, retain=False, encoding=None
     )
@@ -1249,13 +1268,18 @@ async def test_restore_all_active_subscriptions_on_reconnect(
     assert mqtt_client_mock.subscribe.mock_calls == expected
 
 
-async def test_initial_setup_logs_error(hass, caplog, mqtt_client_mock):
+async def test_initial_setup_logs_error(
+    hass, caplog, mqtt_client_mock, empty_mqtt_config
+):
     """Test for setup failure if initial client connection fails."""
     entry = MockConfigEntry(domain=mqtt.DOMAIN, data={mqtt.CONF_BROKER: "test-broker"})
-
+    entry.add_to_hass(hass)
     mqtt_client_mock.connect.return_value = 1
-    assert await mqtt.async_setup_entry(hass, entry)
+    try:
-    await hass.async_block_till_done()
+        assert await mqtt.async_setup_entry(hass, entry)
+        await hass.async_block_till_done()
+    except HomeAssistantError:
+        assert True
     assert "Failed to connect to MQTT server:" in caplog.text
 
 
@@ -1298,6 +1322,7 @@ async def test_handle_mqtt_on_callback(
 async def test_publish_error(hass, caplog):
     """Test publish error."""
     entry = MockConfigEntry(domain=mqtt.DOMAIN, data={mqtt.CONF_BROKER: "test-broker"})
+    entry.add_to_hass(hass)
 
     # simulate an Out of memory error
     with patch("paho.mqtt.client.Client") as mock_client:
@@ -1365,6 +1390,7 @@ async def test_setup_override_configuration(hass, caplog, tmp_path):
             domain=mqtt.DOMAIN,
             data={mqtt.CONF_BROKER: "test-broker", "password": "somepassword"},
         )
+        entry.add_to_hass(hass)
 
         with patch("paho.mqtt.client.Client") as mock_client:
             mock_client().username_pw_set = mock_usename_password_set
@@ -1429,9 +1455,11 @@ async def test_setup_mqtt_client_protocol(hass):
             mqtt.config_integration.CONF_PROTOCOL: "3.1",
         },
     )
+    entry.add_to_hass(hass)
     with patch("paho.mqtt.client.Client") as mock_client:
         mock_client.on_connect(return_value=0)
         assert await mqtt.async_setup_entry(hass, entry)
+        await hass.async_block_till_done()
 
         # check if protocol setup was correctly
         assert mock_client.call_args[1]["protocol"] == 3
@@ -1467,15 +1495,18 @@ async def test_handle_mqtt_timeout_on_callback(hass, caplog):
         entry = MockConfigEntry(
             domain=mqtt.DOMAIN, data={mqtt.CONF_BROKER: "test-broker"}
         )
-        # Set up the integration
+        entry.add_to_hass(hass)
-        assert await mqtt.async_setup_entry(hass, entry)
+
         # Make sure we are connected correctly
         mock_client.on_connect(mock_client, None, None, 0)
+        # Set up the integration
+        assert await mqtt.async_setup_entry(hass, entry)
+        await hass.async_block_till_done()
 
         # Now call we publish without simulating and ACK callback
         await mqtt.async_publish(hass, "no_callback/test-topic", "test-payload")
         await hass.async_block_till_done()
-        # The is no ACK so we should see a timeout in the log after publishing
+        # There is no ACK so we should see a timeout in the log after publishing
         assert len(mock_client.publish.mock_calls) == 1
         assert "No ACK from MQTT server" in caplog.text
 
@@ -1483,10 +1514,12 @@ async def test_handle_mqtt_timeout_on_callback(hass, caplog):
 async def test_setup_raises_ConfigEntryNotReady_if_no_connect_broker(hass, caplog):
     """Test for setup failure if connection to broker is missing."""
     entry = MockConfigEntry(domain=mqtt.DOMAIN, data={mqtt.CONF_BROKER: "test-broker"})
+    entry.add_to_hass(hass)
 
     with patch("paho.mqtt.client.Client") as mock_client:
         mock_client().connect = MagicMock(side_effect=OSError("Connection error"))
         assert await mqtt.async_setup_entry(hass, entry)
+        await hass.async_block_till_done()
         assert "Failed to connect to MQTT server due to exception:" in caplog.text
 
 
@@ -1514,8 +1547,9 @@ async def test_setup_uses_certificate_on_certificate_set_to_auto_and_insecure(
             domain=mqtt.DOMAIN,
             data=config_item_data,
         )
-
+        entry.add_to_hass(hass)
         assert await mqtt.async_setup_entry(hass, entry)
+        await hass.async_block_till_done()
 
         assert calls
 
@@ -1546,8 +1580,9 @@ async def test_setup_without_tls_config_uses_tlsv1_under_python36(hass):
             domain=mqtt.DOMAIN,
             data={"certificate": "auto", mqtt.CONF_BROKER: "test-broker"},
         )
-
+        entry.add_to_hass(hass)
         assert await mqtt.async_setup_entry(hass, entry)
+        await hass.async_block_till_done()
 
         assert calls
 
@@ -2644,3 +2679,206 @@ async def test_config_schema_validation(hass):
     config = {"mqtt": {"sensor": [{"some_illegal_topic": "mystate/topic/path"}]}}
     with pytest.raises(vol.MultipleInvalid):
         CONFIG_SCHEMA(config)
+
+
+@patch("homeassistant.components.mqtt.PLATFORMS", [Platform.LIGHT])
+async def test_unload_config_entry(
+    hass, mqtt_mock, mqtt_client_mock, tmp_path, caplog
+) -> None:
+    """Test unloading the MQTT entry."""
+    assert hass.services.has_service(mqtt.DOMAIN, "dump")
+    assert hass.services.has_service(mqtt.DOMAIN, "publish")
+
+    mqtt_config_entry = hass.config_entries.async_entries(mqtt.DOMAIN)[0]
+    assert mqtt_config_entry.state is ConfigEntryState.LOADED
+
+    # Publish just before unloading to test await cleanup
+    mqtt_client_mock.reset_mock()
+    mqtt.publish(hass, "just_in_time", "published", qos=0, retain=False)
+
+    new_yaml_config_file = tmp_path / "configuration.yaml"
+    new_yaml_config = yaml.dump({})
+    new_yaml_config_file.write_text(new_yaml_config)
+    with patch.object(hass_config, "YAML_CONFIG_FILE", new_yaml_config_file):
+        assert await hass.config_entries.async_unload(mqtt_config_entry.entry_id)
+        mqtt_client_mock.publish.assert_any_call("just_in_time", "published", 0, False)
+        assert mqtt_config_entry.state is ConfigEntryState.NOT_LOADED
+        await hass.async_block_till_done()
+    assert not hass.services.has_service(mqtt.DOMAIN, "dump")
+    assert not hass.services.has_service(mqtt.DOMAIN, "publish")
+    assert "No ACK from MQTT server" not in caplog.text
+
+
+@patch("homeassistant.components.mqtt.PLATFORMS", [])
+async def test_setup_with_disabled_entry(hass, caplog) -> None:
+    """Test setting up the platform with a disabled config entry."""
+    # Try to setup the platform with a disabled config entry
+    config_entry = MockConfigEntry(
+        domain=mqtt.DOMAIN, data={}, disabled_by=ConfigEntryDisabler.USER
+    )
+    config_entry.add_to_hass(hass)
+
+    config = {mqtt.DOMAIN: {}}
+    await async_setup_component(hass, mqtt.DOMAIN, config)
+    await hass.async_block_till_done()
+
+    assert "MQTT will be not available until the config entry is enabled" in caplog.text
+
+
+@patch("homeassistant.components.mqtt.PLATFORMS", [])
+async def test_publish_or_subscribe_without_valid_config_entry(hass, caplog):
+    """Test internal publish function with bas use cases."""
+    with pytest.raises(HomeAssistantError):
+        await mqtt.async_publish(
+            hass, "some-topic", "test-payload", qos=0, retain=False, encoding=None
+        )
+    with pytest.raises(HomeAssistantError):
+        await mqtt.async_subscribe(hass, "some-topic", lambda: None, qos=0)
+
+
+@patch("homeassistant.components.mqtt.PLATFORMS", [Platform.LIGHT])
+async def test_reload_entry_with_new_config(hass, tmp_path):
+    """Test reloading the config entry with a new yaml config."""
+    config_old = [{"name": "test_old1", "command_topic": "test-topic_old"}]
+    config_yaml_new = {
+        "mqtt": {
+            "light": [{"name": "test_new_modern", "command_topic": "test-topic_new"}]
+        },
+        "light": [
+            {
+                "platform": "mqtt",
+                "name": "test_new_legacy",
+                "command_topic": "test-topic_new",
+            }
+        ],
+    }
+    await help_test_setup_manual_entity_from_yaml(hass, "light", config_old)
+    assert hass.states.get("light.test_old1") is not None
+
+    await help_test_entry_reload_with_new_config(hass, tmp_path, config_yaml_new)
+    assert hass.states.get("light.test_old1") is None
+    assert hass.states.get("light.test_new_modern") is not None
+    assert hass.states.get("light.test_new_legacy") is not None
+
+
+@patch("homeassistant.components.mqtt.PLATFORMS", [Platform.LIGHT])
+async def test_disabling_and_enabling_entry(hass, tmp_path, caplog):
+    """Test disabling and enabling the config entry."""
+    config_old = [{"name": "test_old1", "command_topic": "test-topic_old"}]
+    config_yaml_new = {
+        "mqtt": {
+            "light": [{"name": "test_new_modern", "command_topic": "test-topic_new"}]
+        },
+        "light": [
+            {
+                "platform": "mqtt",
+                "name": "test_new_legacy",
+                "command_topic": "test-topic_new",
+            }
+        ],
+    }
+    await help_test_setup_manual_entity_from_yaml(hass, "light", config_old)
+    assert hass.states.get("light.test_old1") is not None
+
+    mqtt_config_entry = hass.config_entries.async_entries(mqtt.DOMAIN)[0]
+
+    assert mqtt_config_entry.state is ConfigEntryState.LOADED
+    new_yaml_config_file = tmp_path / "configuration.yaml"
+    new_yaml_config = yaml.dump(config_yaml_new)
+    new_yaml_config_file.write_text(new_yaml_config)
+    assert new_yaml_config_file.read_text() == new_yaml_config
+
+    with patch.object(hass_config, "YAML_CONFIG_FILE", new_yaml_config_file), patch(
+        "paho.mqtt.client.Client"
+    ) as mock_client:
+        mock_client().connect = lambda *args: 0
+
+        # Late discovery of a light
+        config = '{"name": "abc", "command_topic": "test-topic"}'
+        async_fire_mqtt_message(hass, "homeassistant/light/abc/config", config)
+
+        # Disable MQTT config entry
+        await hass.config_entries.async_set_disabled_by(
+            mqtt_config_entry.entry_id, ConfigEntryDisabler.USER
+        )
+
+        await hass.async_block_till_done()
+        await hass.async_block_till_done()
+        # Assert that the discovery was still received
+        # but kipped the setup
+        assert (
+            "MQTT integration is disabled, skipping setup of manually configured MQTT light"
+            in caplog.text
+        )
+
+        assert mqtt_config_entry.state is ConfigEntryState.NOT_LOADED
+        assert hass.states.get("light.test_old1") is None
+
+        # Enable the entry again
+        await hass.config_entries.async_set_disabled_by(
+            mqtt_config_entry.entry_id, None
+        )
+        await hass.async_block_till_done()
+        await hass.async_block_till_done()
+        assert mqtt_config_entry.state is ConfigEntryState.LOADED
+
+        assert hass.states.get("light.test_old1") is None
+        assert hass.states.get("light.test_new_modern") is not None
+        assert hass.states.get("light.test_new_legacy") is not None
+
+
+@patch("homeassistant.components.mqtt.PLATFORMS", [Platform.LIGHT])
+@pytest.mark.parametrize(
+    "config, unique",
+    [
+        (
+            [
+                {
+                    "name": "test1",
+                    "unique_id": "very_not_unique_deadbeef",
+                    "command_topic": "test-topic_unique",
+                },
+                {
+                    "name": "test2",
+                    "unique_id": "very_not_unique_deadbeef",
+                    "command_topic": "test-topic_unique",
+                },
+            ],
+            False,
+        ),
+        (
+            [
+                {
+                    "name": "test1",
+                    "unique_id": "very_unique_deadbeef1",
+                    "command_topic": "test-topic_unique",
+                },
+                {
+                    "name": "test2",
+                    "unique_id": "very_unique_deadbeef2",
+                    "command_topic": "test-topic_unique",
+                },
+            ],
+            True,
+        ),
+    ],
+)
+async def test_setup_manual_items_with_unique_ids(
+    hass, tmp_path, caplog, config, unique
+):
+    """Test setup manual items is generating unique id's."""
+    await help_test_setup_manual_entity_from_yaml(hass, "light", config)
+
+    assert hass.states.get("light.test1") is not None
+    assert (hass.states.get("light.test2") is not None) == unique
+    assert bool("Platform mqtt does not generate unique IDs." in caplog.text) != unique
+
+    # reload and assert again
+    caplog.clear()
+    await help_test_entry_reload_with_new_config(
+        hass, tmp_path, {"mqtt": {"light": config}}
+    )
+
+    assert hass.states.get("light.test1") is not None
+    assert (hass.states.get("light.test2") is not None) == unique
+    assert bool("Platform mqtt does not generate unique IDs." in caplog.text) != unique

tests/components/mqtt/test_light.py

@@ -240,6 +240,7 @@ from .test_common import (
     help_test_setting_blocked_attribute_via_mqtt_json_message,
     help_test_setup_manual_entity_from_yaml,
     help_test_unique_id,
+    help_test_unload_config_entry_with_platform,
     help_test_update_with_json_attrs_bad_JSON,
     help_test_update_with_json_attrs_not_dict,
 )
@@ -3803,3 +3804,12 @@ async def test_setup_manual_entity_from_yaml(hass):
     del config["platform"]
     await help_test_setup_manual_entity_from_yaml(hass, platform, config)
     assert hass.states.get(f"{platform}.test") is not None
+
+
+async def test_unload_entry(hass, mqtt_mock_entry_with_yaml_config, tmp_path):
+    """Test unloading the config entry."""
+    domain = light.DOMAIN
+    config = DEFAULT_CONFIG[domain]
+    await help_test_unload_config_entry_with_platform(
+        hass, mqtt_mock_entry_with_yaml_config, tmp_path, domain, config
+    )

tests/components/mqtt/test_light_template.py

@@ -72,6 +72,7 @@ from .test_common import (
     help_test_setting_blocked_attribute_via_mqtt_json_message,
     help_test_setup_manual_entity_from_yaml,
     help_test_unique_id,
+    help_test_unload_config_entry_with_platform,
     help_test_update_with_json_attrs_bad_JSON,
     help_test_update_with_json_attrs_not_dict,
 )
@@ -1266,3 +1267,12 @@ async def test_setup_manual_entity_from_yaml(hass):
     del config["platform"]
     await help_test_setup_manual_entity_from_yaml(hass, platform, config)
     assert hass.states.get(f"{platform}.test") is not None
+
+
+async def test_unload_entry(hass, mqtt_mock_entry_with_yaml_config, tmp_path):
+    """Test unloading the config entry."""
+    domain = light.DOMAIN
+    config = DEFAULT_CONFIG[domain]
+    await help_test_unload_config_entry_with_platform(
+        hass, mqtt_mock_entry_with_yaml_config, tmp_path, domain, config
+    )

tests/components/mqtt/test_lock.py

@@ -48,6 +48,7 @@ from .test_common import (
     help_test_setting_blocked_attribute_via_mqtt_json_message,
     help_test_setup_manual_entity_from_yaml,
     help_test_unique_id,
+    help_test_unload_config_entry_with_platform,
     help_test_update_with_json_attrs_bad_JSON,
     help_test_update_with_json_attrs_not_dict,
 )
@@ -748,3 +749,12 @@ async def test_setup_manual_entity_from_yaml(hass, caplog, tmp_path):
     del config["platform"]
     await help_test_setup_manual_entity_from_yaml(hass, platform, config)
     assert hass.states.get(f"{platform}.test") is not None
+
+
+async def test_unload_entry(hass, mqtt_mock_entry_with_yaml_config, tmp_path):
+    """Test unloading the config entry."""
+    domain = LOCK_DOMAIN
+    config = DEFAULT_CONFIG[domain]
+    await help_test_unload_config_entry_with_platform(
+        hass, mqtt_mock_entry_with_yaml_config, tmp_path, domain, config
+    )

tests/components/mqtt/test_number.py

@@ -57,6 +57,7 @@ from .test_common import (
     help_test_setting_blocked_attribute_via_mqtt_json_message,
     help_test_setup_manual_entity_from_yaml,
     help_test_unique_id,
+    help_test_unload_config_entry_with_platform,
     help_test_update_with_json_attrs_bad_JSON,
     help_test_update_with_json_attrs_not_dict,
 )
@@ -853,3 +854,12 @@ async def test_setup_manual_entity_from_yaml(hass):
     del config["platform"]
     await help_test_setup_manual_entity_from_yaml(hass, platform, config)
     assert hass.states.get(f"{platform}.test") is not None
+
+
+async def test_unload_entry(hass, mqtt_mock_entry_with_yaml_config, tmp_path):
+    """Test unloading the config entry."""
+    domain = number.DOMAIN
+    config = DEFAULT_CONFIG[domain]
+    await help_test_unload_config_entry_with_platform(
+        hass, mqtt_mock_entry_with_yaml_config, tmp_path, domain, config
+    )

tests/components/mqtt/test_scene.py

@@ -22,6 +22,7 @@ from .test_common import (
     help_test_reloadable_late,
     help_test_setup_manual_entity_from_yaml,
     help_test_unique_id,
+    help_test_unload_config_entry_with_platform,
 )
 
 DEFAULT_CONFIG = {
@@ -237,3 +238,12 @@ async def test_setup_manual_entity_from_yaml(hass):
     del config["platform"]
     await help_test_setup_manual_entity_from_yaml(hass, platform, config)
     assert hass.states.get(f"{platform}.test") is not None
+
+
+async def test_unload_entry(hass, mqtt_mock_entry_with_yaml_config, tmp_path):
+    """Test unloading the config entry."""
+    domain = scene.DOMAIN
+    config = DEFAULT_CONFIG[domain]
+    await help_test_unload_config_entry_with_platform(
+        hass, mqtt_mock_entry_with_yaml_config, tmp_path, domain, config
+    )

tests/components/mqtt/test_select.py

@@ -48,6 +48,7 @@ from .test_common import (
     help_test_setting_blocked_attribute_via_mqtt_json_message,
     help_test_setup_manual_entity_from_yaml,
     help_test_unique_id,
+    help_test_unload_config_entry_with_platform,
     help_test_update_with_json_attrs_bad_JSON,
     help_test_update_with_json_attrs_not_dict,
 )
@@ -687,3 +688,12 @@ async def test_setup_manual_entity_from_yaml(hass):
     del config["platform"]
     await help_test_setup_manual_entity_from_yaml(hass, platform, config)
     assert hass.states.get(f"{platform}.test") is not None
+
+
+async def test_unload_entry(hass, mqtt_mock_entry_with_yaml_config, tmp_path):
+    """Test unloading the config entry."""
+    domain = select.DOMAIN
+    config = DEFAULT_CONFIG[domain]
+    await help_test_unload_config_entry_with_platform(
+        hass, mqtt_mock_entry_with_yaml_config, tmp_path, domain, config
+    )

tests/components/mqtt/test_sensor.py

@@ -58,6 +58,7 @@ from .test_common import (
     help_test_setting_blocked_attribute_via_mqtt_json_message,
     help_test_setup_manual_entity_from_yaml,
     help_test_unique_id,
+    help_test_unload_config_entry_with_platform,
     help_test_update_with_json_attrs_bad_JSON,
     help_test_update_with_json_attrs_not_dict,
 )
@@ -1213,3 +1214,12 @@ async def test_setup_manual_entity_from_yaml(hass):
     del config["platform"]
     await help_test_setup_manual_entity_from_yaml(hass, platform, config)
     assert hass.states.get(f"{platform}.test") is not None
+
+
+async def test_unload_entry(hass, mqtt_mock_entry_with_yaml_config, tmp_path):
+    """Test unloading the config entry."""
+    domain = sensor.DOMAIN
+    config = DEFAULT_CONFIG[domain]
+    await help_test_unload_config_entry_with_platform(
+        hass, mqtt_mock_entry_with_yaml_config, tmp_path, domain, config
+    )

tests/components/mqtt/test_siren.py

@@ -45,6 +45,7 @@ from .test_common import (
     help_test_setting_blocked_attribute_via_mqtt_json_message,
     help_test_setup_manual_entity_from_yaml,
     help_test_unique_id,
+    help_test_unload_config_entry_with_platform,
     help_test_update_with_json_attrs_bad_JSON,
     help_test_update_with_json_attrs_not_dict,
 )
@@ -975,3 +976,12 @@ async def test_setup_manual_entity_from_yaml(hass):
     del config["platform"]
     await help_test_setup_manual_entity_from_yaml(hass, platform, config)
     assert hass.states.get(f"{platform}.test") is not None
+
+
+async def test_unload_entry(hass, mqtt_mock_entry_with_yaml_config, tmp_path):
+    """Test unloading the config entry."""
+    domain = siren.DOMAIN
+    config = DEFAULT_CONFIG[domain]
+    await help_test_unload_config_entry_with_platform(
+        hass, mqtt_mock_entry_with_yaml_config, tmp_path, domain, config
+    )

tests/components/mqtt/test_switch.py

@@ -42,6 +42,7 @@ from .test_common import (
     help_test_setting_blocked_attribute_via_mqtt_json_message,
     help_test_setup_manual_entity_from_yaml,
     help_test_unique_id,
+    help_test_unload_config_entry_with_platform,
     help_test_update_with_json_attrs_bad_JSON,
     help_test_update_with_json_attrs_not_dict,
 )
@@ -664,3 +665,12 @@ async def test_setup_manual_entity_from_yaml(hass):
     del config["platform"]
     await help_test_setup_manual_entity_from_yaml(hass, platform, config)
     assert hass.states.get(f"{platform}.test") is not None
+
+
+async def test_unload_entry(hass, mqtt_mock_entry_with_yaml_config, tmp_path):
+    """Test unloading the config entry."""
+    domain = switch.DOMAIN
+    config = DEFAULT_CONFIG[domain]
+    await help_test_unload_config_entry_with_platform(
+        hass, mqtt_mock_entry_with_yaml_config, tmp_path, domain, config
+    )

tests/components/mqtt/test_tag.py

@@ -11,6 +11,8 @@ from homeassistant.const import Platform
 from homeassistant.helpers import device_registry as dr
 from homeassistant.setup import async_setup_component
 
+from .test_common import help_test_unload_config_entry
+
 from tests.common import (
     MockConfigEntry,
     async_fire_mqtt_message,
@@ -797,3 +799,28 @@ async def test_cleanup_device_with_entity2(
     # Verify device registry entry is cleared
     device_entry = device_reg.async_get_device({("mqtt", "helloworld")})
     assert device_entry is None
+
+
+async def test_unload_entry(hass, device_reg, mqtt_mock, tag_mock, tmp_path) -> None:
+    """Test unloading the MQTT entry."""
+
+    config = copy.deepcopy(DEFAULT_CONFIG_DEVICE)
+
+    async_fire_mqtt_message(hass, "homeassistant/tag/bla1/config", json.dumps(config))
+    await hass.async_block_till_done()
+    device_entry = device_reg.async_get_device({("mqtt", "0AFFD2")})
+
+    # Fake tag scan, should be processed
+    async_fire_mqtt_message(hass, "foobar/tag_scanned", DEFAULT_TAG_SCAN)
+    await hass.async_block_till_done()
+    tag_mock.assert_called_once_with(ANY, DEFAULT_TAG_ID, device_entry.id)
+
+    tag_mock.reset_mock()
+
+    await help_test_unload_config_entry(hass, tmp_path, {})
+    await hass.async_block_till_done()
+
+    # Fake tag scan, should not be processed
+    async_fire_mqtt_message(hass, "foobar/tag_scanned", DEFAULT_TAG_SCAN)
+    await hass.async_block_till_done()
+    tag_mock.assert_not_called()