Migrate ZHA when enabling multi-PAN support on HA Yellow (#82213)

* Migrate ZHA when enabling multi-PAN support on HA Yellow

* Refactor BaseZhaFlow.async_step_maybe_confirm_ezsp_restore

* Change data passed to ZHA to initiate migration

* Catch errors during ZHA migration

* Fix ZhaMigrationHelper.async_prepare_yellow_migration return value

* Improve test coverage

* Improve test coverage

* Fix spelling

* Rename some none HA yellow specifics

* Rename again

* Increase number of migration retries + refactor

* Suppress OperationNotAllowed when reloading

* Adjust tests

homeassistant/components/homeassistant_hardware/manifest.json

@@ -3,5 +3,6 @@
   "name": "Home Assistant Hardware",
   "documentation": "https://www.home-assistant.io/integrations/homeassistant_hardware",
   "codeowners": ["@home-assistant/core"],
-  "integration_type": "system"
+  "integration_type": "system",
+  "after_dependencies": ["zha"]
 }

homeassistant/components/homeassistant_hardware/silabs_multiprotocol_addon.py

@@ -17,6 +17,8 @@ from homeassistant.components.hassio import (
     AddonState,
     is_hassio,
 )
+from homeassistant.components.zha import DOMAIN as ZHA_DOMAIN
+from homeassistant.components.zha.radio_manager import ZhaMultiPANMigrationHelper
 from homeassistant.core import HomeAssistant, callback
 from homeassistant.data_entry_flow import (
     AbortFlow,
@@ -77,6 +79,7 @@ class BaseMultiPanFlow(FlowHandler):
         # If we install the add-on we should uninstall it on entry remove.
         self.install_task: asyncio.Task | None = None
         self.start_task: asyncio.Task | None = None
+        self._zha_migration_mgr: ZhaMultiPANMigrationHelper | None = None
 
     @property
     @abstractmethod
@@ -87,6 +90,18 @@ class BaseMultiPanFlow(FlowHandler):
     async def _async_serial_port_settings(self) -> SerialPortSettings:
         """Return the radio serial port settings."""
 
+    @abstractmethod
+    async def _async_zha_physical_discovery(self) -> dict[str, Any]:
+        """Return ZHA discovery data when multiprotocol FW is not used.
+
+        Passed to ZHA do determine if the ZHA config entry is connected to the radio
+        being migrated.
+        """
+
+    @abstractmethod
+    def _zha_name(self) -> str:
+        """Return the ZHA name."""
+
     async def async_step_install_addon(
         self, user_input: dict[str, Any] | None = None
     ) -> FlowResult:
@@ -260,6 +275,31 @@ class OptionsFlowHandler(BaseMultiPanFlow, config_entries.OptionsFlow):
             **dataclasses.asdict(serial_port_settings),
         }
 
+        # Initiate ZHA migration
+        zha_entries = self.hass.config_entries.async_entries(ZHA_DOMAIN)
+
+        if zha_entries:
+            zha_migration_mgr = ZhaMultiPANMigrationHelper(self.hass, zha_entries[0])
+            migration_data = {
+                "new_discovery_info": {
+                    "name": self._zha_name(),
+                    "port": {
+                        "path": get_zigbee_socket(self.hass, addon_info),
+                        "baudrate": 115200,
+                        "flow_control": "hardware",
+                    },
+                    "radio_type": "efr32",
+                },
+                "old_discovery_info": await self._async_zha_physical_discovery(),
+            }
+            _LOGGER.debug("Starting ZHA migration with: %s", migration_data)
+            try:
+                if await zha_migration_mgr.async_initiate_migration(migration_data):
+                    self._zha_migration_mgr = zha_migration_mgr
+            except Exception as err:
+                _LOGGER.exception("Unexpected exception during ZHA migration")
+                raise AbortFlow("zha_migration_failed") from err
+
         if new_addon_config != addon_config:
             # Copy the add-on config to keep the objects separate.
             self.original_addon_config = dict(addon_config)
@@ -277,6 +317,14 @@ class OptionsFlowHandler(BaseMultiPanFlow, config_entries.OptionsFlow):
             self.hass.config_entries.async_reload(self.config_entry.entry_id)
         )
 
+        # Finish ZHA migration if needed
+        if self._zha_migration_mgr:
+            try:
+                await self._zha_migration_mgr.async_finish_migration()
+            except Exception as err:
+                _LOGGER.exception("Unexpected exception during ZHA migration")
+                raise AbortFlow("zha_migration_failed") from err
+
         return self.async_create_entry(title="", data={})
 
     async def async_step_addon_installed(

homeassistant/components/homeassistant_yellow/__init__.py

@@ -18,6 +18,8 @@ from homeassistant.config_entries import ConfigEntry
 from homeassistant.core import HomeAssistant
 from homeassistant.exceptions import ConfigEntryNotReady
 
+from .const import ZHA_HW_DISCOVERY_DATA
+
 _LOGGER = logging.getLogger(__name__)
 
 
@@ -52,22 +54,22 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
         raise ConfigEntryNotReady
 
     if addon_info.state == AddonState.NOT_INSTALLED:
-        path = "/dev/ttyAMA1"
+        hw_discovery_data = ZHA_HW_DISCOVERY_DATA
     else:
-        path = get_zigbee_socket(hass, addon_info)
+        hw_discovery_data = {
-
+            "name": "Yellow Multi-PAN",
-    await hass.config_entries.flow.async_init(
-        "zha",
-        context={"source": "hardware"},
-        data={
-            "name": "Yellow",
             "port": {
-                "path": path,
+                "path": get_zigbee_socket(hass, addon_info),
                 "baudrate": 115200,
                 "flow_control": "hardware",
             },
             "radio_type": "efr32",
-        },
+        }
+
+    await hass.config_entries.flow.async_init(
+        "zha",
+        context={"source": "hardware"},
+        data=hw_discovery_data,
     )
 
     return True

homeassistant/components/homeassistant_yellow/config_flow.py

@@ -8,7 +8,7 @@ from homeassistant.config_entries import ConfigEntry, ConfigFlow
 from homeassistant.core import callback
 from homeassistant.data_entry_flow import FlowResult
 
-from .const import DOMAIN
+from .const import DOMAIN, ZHA_HW_DISCOVERY_DATA
 
 
 class HomeAssistantYellowConfigFlow(ConfigFlow, domain=DOMAIN):
@@ -44,3 +44,15 @@ class HomeAssistantYellowOptionsFlow(silabs_multiprotocol_addon.OptionsFlowHandl
             baudrate="115200",
             flow_control=True,
         )
+
+    async def _async_zha_physical_discovery(self) -> dict[str, Any]:
+        """Return ZHA discovery data when multiprotocol FW is not used.
+
+        Passed to ZHA do determine if the ZHA config entry is connected to the radio
+        being migrated.
+        """
+        return ZHA_HW_DISCOVERY_DATA
+
+    def _zha_name(self) -> str:
+        """Return the ZHA name."""
+        return "Yellow Multi-PAN"

homeassistant/components/homeassistant_yellow/const.py

@@ -1,3 +1,13 @@
 """Constants for the Home Assistant Yellow integration."""
 
 DOMAIN = "homeassistant_yellow"
+
+ZHA_HW_DISCOVERY_DATA = {
+    "name": "Yellow",
+    "port": {
+        "path": "/dev/ttyAMA1",
+        "baudrate": 115200,
+        "flow_control": "hardware",
+    },
+    "radio_type": "efr32",
+}

homeassistant/components/homeassistant_yellow/strings.json

@@ -3,7 +3,7 @@
     "step": {
       "addon_not_installed": {
         "title": "Enable multiprotocol support on the IEEE 802.15.4 radio",
-        "description": "When multiprotocol support is enabled, the Home Assistant Yellow's IEEE 802.15.4 radio can be used for both Zigbee and Thread (used by Matter) at the same time. Note: This is an experimental feature.",
+        "description": "When multiprotocol support is enabled, the Home Assistant Yellow's IEEE 802.15.4 radio can be used for both Zigbee and Thread (used by Matter) at the same time. If the radio is already used by the ZHA Zigbee integration, ZHA will be reconfigured to use the multiprotocol firmware.\n\nNote: This is an experimental feature.",
         "data": {
           "enable_multi_pan": "Enable multiprotocol support"
         }
@@ -30,7 +30,8 @@
       "addon_install_failed": "Failed to install the Silicon Labs Multiprotocol add-on.",
       "addon_set_config_failed": "Failed to set Silicon Labs Multiprotocol configuration.",
       "addon_start_failed": "Failed to start the Silicon Labs Multiprotocol add-on.",
-      "not_hassio": "The hardware options can only be configured on HassOS installations."
+      "not_hassio": "The hardware options can only be configured on HassOS installations.",
+      "zha_migration_failed": "The ZHA migration did not succeed."
     },
     "progress": {
       "install_addon": "Please wait while the Silicon Labs Multiprotocol add-on installation finishes. This can take several minutes.",

homeassistant/components/homeassistant_yellow/translations/en.json

@@ -5,7 +5,8 @@
             "addon_install_failed": "Failed to install the Silicon Labs Multiprotocol add-on.",
             "addon_set_config_failed": "Failed to set Silicon Labs Multiprotocol configuration.",
             "addon_start_failed": "Failed to start the Silicon Labs Multiprotocol add-on.",
-            "not_hassio": "The hardware options can only be configured on HassOS installations."
+            "not_hassio": "The hardware options can only be configured on HassOS installations.",
+            "zha_migration_failed": "The ZHA migration did not succeed."
         },
         "error": {
             "unknown": "Unexpected error"
@@ -22,7 +23,7 @@
                 "data": {
                     "enable_multi_pan": "Enable multiprotocol support"
                 },
-                "description": "When multiprotocol support is enabled, the Home Assistant Yellow's IEEE 802.15.4 radio can be used for both Zigbee and Thread (used by Matter) at the same time. Note: This is an experimental feature.",
+                "description": "When multiprotocol support is enabled, the Home Assistant Yellow's IEEE 802.15.4 radio can be used for both Zigbee and Thread (used by Matter) at the same time. If the radio is already used by the ZHA Zigbee integration, ZHA will be reconfigured to use the multiprotocol firmware.\n\nNote: This is an experimental feature.",
                 "title": "Enable multiprotocol support on the IEEE 802.15.4 radio"
             },
             "install_addon": {

homeassistant/components/zha/config_flow.py

@@ -2,7 +2,6 @@
 from __future__ import annotations
 
 import collections
-import copy
 import json
 from typing import Any
 
@@ -25,10 +24,9 @@ from .core.const import (
     CONF_FLOWCONTROL,
     CONF_RADIO_TYPE,
     DOMAIN,
-    EZSP_OVERWRITE_EUI64,
     RadioType,
 )
-from .radio_manager import ZhaRadioManager
+from .radio_manager import HARDWARE_DISCOVERY_SCHEMA, ZhaRadioManager
 
 CONF_MANUAL_PATH = "Enter Manually"
 SUPPORTED_PORT_SETTINGS = (
@@ -54,14 +52,6 @@ UPLOADED_BACKUP_FILE = "uploaded_backup_file"
 DEFAULT_ZHA_ZEROCONF_PORT = 6638
 ESPHOME_API_PORT = 6053
 
-HARDWARE_DISCOVERY_SCHEMA = vol.Schema(
-    {
-        vol.Required("name"): str,
-        vol.Required("port"): dict,
-        vol.Required("radio_type"): str,
-    }
-)
-
 
 def _format_backup_choice(
     backup: zigpy.backups.NetworkBackup, *, pan_ids: bool = True
@@ -80,33 +70,6 @@ def _format_backup_choice(
     return f"{dt.as_local(backup.backup_time).strftime('%c')} ({identifier})"
 
 
-def _allow_overwrite_ezsp_ieee(
-    backup: zigpy.backups.NetworkBackup,
-) -> zigpy.backups.NetworkBackup:
-    """Return a new backup with the flag to allow overwriting the EZSP EUI64."""
-    new_stack_specific = copy.deepcopy(backup.network_info.stack_specific)
-    new_stack_specific.setdefault("ezsp", {})[EZSP_OVERWRITE_EUI64] = True
-
-    return backup.replace(
-        network_info=backup.network_info.replace(stack_specific=new_stack_specific)
-    )
-
-
-def _prevent_overwrite_ezsp_ieee(
-    backup: zigpy.backups.NetworkBackup,
-) -> zigpy.backups.NetworkBackup:
-    """Return a new backup without the flag to allow overwriting the EZSP EUI64."""
-    if "ezsp" not in backup.network_info.stack_specific:
-        return backup
-
-    new_stack_specific = copy.deepcopy(backup.network_info.stack_specific)
-    new_stack_specific.setdefault("ezsp", {}).pop(EZSP_OVERWRITE_EUI64, None)
-
-    return backup.replace(
-        network_info=backup.network_info.replace(stack_specific=new_stack_specific)
-    )
-
-
 class BaseZhaFlow(FlowHandler):
     """Mixin for common ZHA flow steps and forms."""
 
@@ -407,46 +370,14 @@ class BaseZhaFlow(FlowHandler):
         self, user_input: dict[str, Any] | None = None
     ) -> FlowResult:
         """Confirm restore for EZSP radios that require permanent IEEE writes."""
-        assert self._radio_mgr.chosen_backup is not None
+        call_step_2 = await self._radio_mgr.async_restore_backup_step_1()
-
+        if not call_step_2:
-        if self._radio_mgr.radio_type != RadioType.ezsp:
-            await self._radio_mgr.restore_backup(self._radio_mgr.chosen_backup)
-            return await self._async_create_radio_entry()
-
-        # We have no way to partially load network settings if no network is formed
-        if self._radio_mgr.current_settings is None:
-            # Since we are going to be restoring the backup anyways, write it to the
-            # radio without overwriting the IEEE but don't take a backup with these
-            # temporary settings
-            temp_backup = _prevent_overwrite_ezsp_ieee(self._radio_mgr.chosen_backup)
-            await self._radio_mgr.restore_backup(temp_backup, create_new=False)
-            await self._radio_mgr.async_load_network_settings()
-
-            assert self._radio_mgr.current_settings is not None
-
-        if (
-            self._radio_mgr.current_settings.node_info.ieee
-            == self._radio_mgr.chosen_backup.node_info.ieee
-            or not self._radio_mgr.current_settings.network_info.metadata["ezsp"][
-                "can_write_custom_eui64"
-            ]
-        ):
-            # No point in prompting the user if the backup doesn't have a new IEEE
-            # address or if there is no way to overwrite the IEEE address a second time
-            await self._radio_mgr.restore_backup(self._radio_mgr.chosen_backup)
-
             return await self._async_create_radio_entry()
 
         if user_input is not None:
-            backup = self._radio_mgr.chosen_backup
+            await self._radio_mgr.async_restore_backup_step_2(
-
+                user_input[OVERWRITE_COORDINATOR_IEEE]
-            if user_input[OVERWRITE_COORDINATOR_IEEE]:
+            )
-                backup = _allow_overwrite_ezsp_ieee(backup)
-
-            # If the user declined to overwrite the IEEE *and* we wrote the backup to
-            # their empty radio above, restoring it again would be redundant.
-            await self._radio_mgr.restore_backup(backup)
-
             return await self._async_create_radio_entry()
 
         return self.async_show_form(

homeassistant/components/zha/radio_manager.py

@@ -3,23 +3,28 @@ from __future__ import annotations
 
 import asyncio
 import contextlib
+import copy
 import logging
 import os
 from typing import Any
 
+import voluptuous as vol
 from zigpy.application import ControllerApplication
 import zigpy.backups
 from zigpy.config import CONF_DEVICE, CONF_DEVICE_PATH
 from zigpy.exceptions import NetworkNotFormed
 
+from homeassistant import config_entries
 from homeassistant.core import HomeAssistant
 
 from .core.const import (
     CONF_DATABASE,
+    CONF_RADIO_TYPE,
     CONF_ZIGPY,
     DATA_ZHA,
     DATA_ZHA_CONFIG,
     DEFAULT_DATABASE_NAME,
+    EZSP_OVERWRITE_EUI64,
     RadioType,
 )
 
@@ -35,9 +40,53 @@ AUTOPROBE_RADIOS = (
 
 CONNECT_DELAY_S = 1.0
 
+MIGRATION_RETRIES = 100
+
+HARDWARE_DISCOVERY_SCHEMA = vol.Schema(
+    {
+        vol.Required("name"): str,
+        vol.Required("port"): dict,
+        vol.Required("radio_type"): str,
+    }
+)
+
+HARDWARE_MIGRATION_SCHEMA = vol.Schema(
+    {
+        vol.Required("new_discovery_info"): HARDWARE_DISCOVERY_SCHEMA,
+        vol.Required("old_discovery_info"): HARDWARE_DISCOVERY_SCHEMA,
+    }
+)
+
 _LOGGER = logging.getLogger(__name__)
 
 
+def _allow_overwrite_ezsp_ieee(
+    backup: zigpy.backups.NetworkBackup,
+) -> zigpy.backups.NetworkBackup:
+    """Return a new backup with the flag to allow overwriting the EZSP EUI64."""
+    new_stack_specific = copy.deepcopy(backup.network_info.stack_specific)
+    new_stack_specific.setdefault("ezsp", {})[EZSP_OVERWRITE_EUI64] = True
+
+    return backup.replace(
+        network_info=backup.network_info.replace(stack_specific=new_stack_specific)
+    )
+
+
+def _prevent_overwrite_ezsp_ieee(
+    backup: zigpy.backups.NetworkBackup,
+) -> zigpy.backups.NetworkBackup:
+    """Return a new backup without the flag to allow overwriting the EZSP EUI64."""
+    if "ezsp" not in backup.network_info.stack_specific:
+        return backup
+
+    new_stack_specific = copy.deepcopy(backup.network_info.stack_specific)
+    new_stack_specific.setdefault("ezsp", {}).pop(EZSP_OVERWRITE_EUI64, None)
+
+    return backup.replace(
+        network_info=backup.network_info.replace(stack_specific=new_stack_specific)
+    )
+
+
 class ZhaRadioManager:
     """Helper class with radio related functionality."""
 
@@ -96,7 +145,8 @@ class ZhaRadioManager:
         async with self._connect_zigpy_app() as app:
             await app.backups.restore_backup(backup, **kwargs)
 
-    def parse_radio_type(self, radio_type: str) -> RadioType:
+    @staticmethod
+    def parse_radio_type(radio_type: str) -> RadioType:
         """Parse a radio type name, accounting for past aliases."""
         if radio_type == "efr32":
             return RadioType.ezsp
@@ -127,8 +177,12 @@ class ZhaRadioManager:
 
         return False
 
-    async def async_load_network_settings(self, create_backup: bool = False) -> None:
+    async def async_load_network_settings(
+        self, *, create_backup: bool = False
+    ) -> zigpy.backups.NetworkBackup | None:
         """Connect to the radio and load its current network settings."""
+        backup = None
+
         async with self._connect_zigpy_app() as app:
             # Check if the stick has any settings and load them
             try:
@@ -142,11 +196,13 @@ class ZhaRadioManager:
                 )
 
                 if create_backup:
-                    await app.backups.create_backup()
+                    backup = await app.backups.create_backup()
 
             # The list of backups will always exist
             self.backups = app.backups.backups.copy()
 
+        return backup
+
     async def async_form_network(self) -> None:
         """Form a brand new network."""
         async with self._connect_zigpy_app() as app:
@@ -156,3 +212,171 @@ class ZhaRadioManager:
         """Reset the current adapter."""
         async with self._connect_zigpy_app() as app:
             await app.reset_network_info()
+
+    async def async_restore_backup_step_1(self) -> bool:
+        """Prepare restoring backup.
+
+        Returns True if async_restore_backup_step_2 should be called.
+        """
+        assert self.chosen_backup is not None
+
+        if self.radio_type != RadioType.ezsp:
+            await self.restore_backup(self.chosen_backup)
+            return False
+
+        # We have no way to partially load network settings if no network is formed
+        if self.current_settings is None:
+            # Since we are going to be restoring the backup anyways, write it to the
+            # radio without overwriting the IEEE but don't take a backup with these
+            # temporary settings
+            temp_backup = _prevent_overwrite_ezsp_ieee(self.chosen_backup)
+            await self.restore_backup(temp_backup, create_new=False)
+            await self.async_load_network_settings()
+
+            assert self.current_settings is not None
+
+        if (
+            self.current_settings.node_info.ieee == self.chosen_backup.node_info.ieee
+            or not self.current_settings.network_info.metadata["ezsp"][
+                "can_write_custom_eui64"
+            ]
+        ):
+            # No point in prompting the user if the backup doesn't have a new IEEE
+            # address or if there is no way to overwrite the IEEE address a second time
+            await self.restore_backup(self.chosen_backup)
+
+            return False
+
+        return True
+
+    async def async_restore_backup_step_2(self, overwrite_ieee: bool) -> None:
+        """Restore backup and optionally overwrite IEEE."""
+        assert self.chosen_backup is not None
+
+        backup = self.chosen_backup
+
+        if overwrite_ieee:
+            backup = _allow_overwrite_ezsp_ieee(backup)
+
+        # If the user declined to overwrite the IEEE *and* we wrote the backup to
+        # their empty radio above, restoring it again would be redundant.
+        await self.restore_backup(backup)
+
+
+class ZhaMultiPANMigrationHelper:
+    """Helper class for automatic migration when upgrading the firmware of a radio.
+
+    This class is currently only intended to be used when changing the firmware on the
+    radio used in the Home Assistant Sky Connect USB stick and the Home Asssistant Yellow
+    from Zigbee only firmware to firmware supporting both Zigbee and Thread.
+    """
+
+    def __init__(
+        self, hass: HomeAssistant, config_entry: config_entries.ConfigEntry
+    ) -> None:
+        """Initialize MigrationHelper instance."""
+        self._config_entry = config_entry
+        self._hass = hass
+        self._radio_mgr = ZhaRadioManager()
+        self._radio_mgr.hass = hass
+
+    async def async_initiate_migration(self, data: dict[str, Any]) -> bool:
+        """Initiate ZHA migration.
+
+        The passed data should contain:
+        - Discovery data identifying the device being firmware updated
+        - Discovery data for connecting to the device after the firmware update is
+          completed.
+
+        Returns True if async_finish_migration should be called after the firmware
+        update is completed.
+        """
+        migration_data = HARDWARE_MIGRATION_SCHEMA(data)
+
+        name = migration_data["new_discovery_info"]["name"]
+        new_radio_type = ZhaRadioManager.parse_radio_type(
+            migration_data["new_discovery_info"]["radio_type"]
+        )
+        old_radio_type = ZhaRadioManager.parse_radio_type(
+            migration_data["old_discovery_info"]["radio_type"]
+        )
+
+        new_device_settings = new_radio_type.controller.SCHEMA_DEVICE(
+            migration_data["new_discovery_info"]["port"]
+        )
+        old_device_settings = old_radio_type.controller.SCHEMA_DEVICE(
+            migration_data["old_discovery_info"]["port"]
+        )
+
+        if (
+            self._config_entry.data[CONF_DEVICE][CONF_DEVICE_PATH]
+            != old_device_settings[CONF_DEVICE_PATH]
+        ):
+            # ZHA is using another radio, do nothing
+            return False
+
+        try:
+            await self._hass.config_entries.async_unload(self._config_entry.entry_id)
+        except config_entries.OperationNotAllowed:
+            # ZHA is not running
+            pass
+
+        # Temporarily connect to the old radio to read its settings
+        old_radio_mgr = ZhaRadioManager()
+        old_radio_mgr.hass = self._hass
+        old_radio_mgr.radio_type = old_radio_type
+        old_radio_mgr.device_path = old_device_settings[CONF_DEVICE_PATH]
+        old_radio_mgr.device_settings = old_device_settings
+        backup = await old_radio_mgr.async_load_network_settings(create_backup=True)
+
+        # Then configure the radio manager for the new radio to use the new settings
+        self._radio_mgr.chosen_backup = backup
+        self._radio_mgr.radio_type = new_radio_type
+        self._radio_mgr.device_path = new_device_settings[CONF_DEVICE_PATH]
+        self._radio_mgr.device_settings = new_device_settings
+        device_settings = self._radio_mgr.device_settings.copy()  # type: ignore[union-attr]
+
+        # Update the config entry settings
+        self._hass.config_entries.async_update_entry(
+            entry=self._config_entry,
+            data={
+                CONF_DEVICE: device_settings,
+                CONF_RADIO_TYPE: self._radio_mgr.radio_type.name,
+            },
+            options=self._config_entry.options,
+            title=name,
+        )
+        return True
+
+    async def async_finish_migration(self) -> None:
+        """Finish ZHA migration.
+
+        Throws an exception if the migration did not succeed.
+        """
+        # Restore the backup, permanently overwriting the device IEEE address
+        for retry in range(MIGRATION_RETRIES):
+            try:
+                if await self._radio_mgr.async_restore_backup_step_1():
+                    await self._radio_mgr.async_restore_backup_step_2(True)
+
+                break
+            except OSError as err:
+                if retry >= MIGRATION_RETRIES - 1:
+                    raise
+
+                _LOGGER.debug(
+                    "Failed to restore backup %r, retrying in %s seconds",
+                    err,
+                    CONNECT_DELAY_S,
+                )
+
+            await asyncio.sleep(CONNECT_DELAY_S)
+
+        _LOGGER.debug("Restored backup after %s retries", retry)
+
+        # Launch ZHA again
+        try:
+            await self._hass.config_entries.async_setup(self._config_entry.entry_id)
+        except config_entries.OperationNotAllowed:
+            # ZHA is not unloaded
+            pass

tests/components/homeassistant_hardware/conftest.py

@@ -1,9 +1,37 @@
 """Test fixtures for the Home Assistant Hardware integration."""
-from unittest.mock import patch
+from collections.abc import Generator
+from typing import Any
+from unittest.mock import MagicMock, patch
 
 import pytest
 
 
+@pytest.fixture(autouse=True)
+def mock_zha_config_flow_setup() -> Generator[None, None, None]:
+    """Mock the radio connection and probing of the ZHA config flow."""
+
+    def mock_probe(config: dict[str, Any]) -> None:
+        # The radio probing will return the correct baudrate
+        return {**config, "baudrate": 115200}
+
+    mock_connect_app = MagicMock()
+    mock_connect_app.__aenter__.return_value.backups.backups = [MagicMock()]
+    mock_connect_app.__aenter__.return_value.backups.create_backup.return_value = (
+        MagicMock()
+    )
+
+    with patch(
+        "bellows.zigbee.application.ControllerApplication.probe", side_effect=mock_probe
+    ), patch(
+        "homeassistant.components.zha.radio_manager.ZhaRadioManager._connect_zigpy_app",
+        return_value=mock_connect_app,
+    ), patch(
+        "homeassistant.components.zha.async_setup_entry",
+        return_value=True,
+    ):
+        yield
+
+
 @pytest.fixture(name="addon_running")
 def mock_addon_running(addon_store_info, addon_info):
     """Mock add-on already running."""

tests/components/homeassistant_hardware/test_silabs_multiprotocol_addon.py

@@ -1,4 +1,4 @@
-"""Test the Home Assistant Yellow config flow."""
+"""Test the Home Assistant Hardware silabs multiprotocol addon manager."""
 from __future__ import annotations
 
 from collections.abc import Generator
@@ -10,6 +10,7 @@ import pytest
 from homeassistant import config_entries
 from homeassistant.components.hassio.handler import HassioAPIError
 from homeassistant.components.homeassistant_hardware import silabs_multiprotocol_addon
+from homeassistant.components.zha.core.const import DOMAIN as ZHA_DOMAIN
 from homeassistant.config_entries import ConfigEntry, ConfigFlow
 from homeassistant.core import HomeAssistant, callback
 from homeassistant.data_entry_flow import FlowResult, FlowResultType
@@ -20,7 +21,7 @@ TEST_DOMAIN = "test"
 
 
 class TestConfigFlow(ConfigFlow, domain=TEST_DOMAIN):
-    """Handle a config flow for Home Assistant Yellow."""
+    """Handle a config flow for the silabs multiprotocol add-on."""
 
     VERSION = 1
 
@@ -37,11 +38,11 @@ class TestConfigFlow(ConfigFlow, domain=TEST_DOMAIN):
         if self._async_current_entries():
             return self.async_abort(reason="single_instance_allowed")
 
-        return self.async_create_entry(title="Home Assistant Yellow", data={})
+        return self.async_create_entry(title="Test HW", data={})
 
 
 class TestOptionsFlow(silabs_multiprotocol_addon.OptionsFlowHandler):
-    """Handle an option flow for Home Assistant Yellow."""
+    """Handle an option flow for the silabs multiprotocol add-on."""
 
     async def _async_serial_port_settings(
         self,
@@ -53,6 +54,26 @@ class TestOptionsFlow(silabs_multiprotocol_addon.OptionsFlowHandler):
             flow_control=True,
         )
 
+    async def _async_zha_physical_discovery(self) -> dict[str, Any]:
+        """Return ZHA discovery data when multiprotocol FW is not used.
+
+        Passed to ZHA do determine if the ZHA config entry is connected to the radio
+        being migrated.
+        """
+        return {
+            "name": "Test",
+            "port": {
+                "path": "/dev/ttyTEST123",
+                "baudrate": 115200,
+                "flow_control": "hardware",
+            },
+            "radio_type": "efr32",
+        }
+
+    def _zha_name(self) -> str:
+        """Return the ZHA name."""
+        return "Test Multi-PAN"
+
 
 @pytest.fixture(autouse=True)
 def config_flow_handler(
@@ -80,7 +101,7 @@ async def test_option_flow_install_multi_pan_addon(
         data={},
         domain=TEST_DOMAIN,
         options={},
-        title="Home Assistant Yellow",
+        title="Test HW",
     )
     config_entry.add_to_hass(hass)
 
@@ -132,6 +153,186 @@ async def test_option_flow_install_multi_pan_addon(
     assert result["type"] == FlowResultType.CREATE_ENTRY
 
 
+async def test_option_flow_install_multi_pan_addon_zha(
+    hass: HomeAssistant,
+    addon_store_info,
+    addon_info,
+    install_addon,
+    set_addon_options,
+    start_addon,
+) -> None:
+    """Test installing the multi pan addon when a zha config entry exists."""
+    mock_integration(hass, MockModule("hassio"))
+
+    # Setup the config entry
+    config_entry = MockConfigEntry(
+        data={},
+        domain=TEST_DOMAIN,
+        options={},
+        title="Test HW",
+    )
+    config_entry.add_to_hass(hass)
+
+    zha_config_entry = MockConfigEntry(
+        data={"device": {"path": "/dev/ttyTEST123"}, "radio_type": "ezsp"},
+        domain=ZHA_DOMAIN,
+        options={},
+        title="Test",
+    )
+    zha_config_entry.add_to_hass(hass)
+
+    with patch(
+        "homeassistant.components.homeassistant_hardware.silabs_multiprotocol_addon.is_hassio",
+        side_effect=Mock(return_value=True),
+    ):
+        result = await hass.config_entries.options.async_init(config_entry.entry_id)
+        assert result["type"] == FlowResultType.FORM
+        assert result["step_id"] == "addon_not_installed"
+
+    result = await hass.config_entries.options.async_configure(
+        result["flow_id"],
+        user_input={
+            "enable_multi_pan": True,
+        },
+    )
+    assert result["type"] == FlowResultType.SHOW_PROGRESS
+    assert result["step_id"] == "install_addon"
+    assert result["progress_action"] == "install_addon"
+
+    result = await hass.config_entries.options.async_configure(result["flow_id"])
+    assert result["type"] == FlowResultType.SHOW_PROGRESS_DONE
+    assert result["step_id"] == "configure_addon"
+    install_addon.assert_called_once_with(hass, "core_silabs_multiprotocol")
+
+    result = await hass.config_entries.options.async_configure(result["flow_id"])
+    assert result["type"] == FlowResultType.SHOW_PROGRESS
+    assert result["step_id"] == "start_addon"
+    set_addon_options.assert_called_once_with(
+        hass,
+        "core_silabs_multiprotocol",
+        {
+            "options": {
+                "autoflash_firmware": True,
+                "device": "/dev/ttyTEST123",
+                "baudrate": "115200",
+                "flow_control": True,
+            }
+        },
+    )
+    # Check the ZHA config entry data is updated
+    assert zha_config_entry.data == {
+        "device": {
+            "path": "socket://core-silabs-multiprotocol:9999",
+            "baudrate": 115200,
+            "flow_control": "hardware",
+        },
+        "radio_type": "ezsp",
+    }
+    assert zha_config_entry.title == "Test Multi-PAN"
+
+    result = await hass.config_entries.options.async_configure(result["flow_id"])
+    assert result["type"] == FlowResultType.SHOW_PROGRESS_DONE
+    assert result["step_id"] == "finish_addon_setup"
+    start_addon.assert_called_once_with(hass, "core_silabs_multiprotocol")
+
+    result = await hass.config_entries.options.async_configure(result["flow_id"])
+    assert result["type"] == FlowResultType.CREATE_ENTRY
+
+
+async def test_option_flow_install_multi_pan_addon_zha_other_radio(
+    hass: HomeAssistant,
+    addon_store_info,
+    addon_info,
+    install_addon,
+    set_addon_options,
+    start_addon,
+) -> None:
+    """Test installing the multi pan addon when a zha config entry exists."""
+    mock_integration(hass, MockModule("hassio"))
+
+    # Setup the config entry
+    config_entry = MockConfigEntry(
+        data={},
+        domain=TEST_DOMAIN,
+        options={},
+        title="Test HW",
+    )
+    config_entry.add_to_hass(hass)
+
+    zha_config_entry = MockConfigEntry(
+        data={
+            "device": {
+                "path": "/dev/other_radio",
+                "baudrate": 115200,
+                "flow_control": "hardware",
+            },
+            "radio_type": "ezsp",
+        },
+        domain=ZHA_DOMAIN,
+        options={},
+        title="Test HW",
+    )
+    zha_config_entry.add_to_hass(hass)
+
+    with patch(
+        "homeassistant.components.homeassistant_hardware.silabs_multiprotocol_addon.is_hassio",
+        side_effect=Mock(return_value=True),
+    ):
+        result = await hass.config_entries.options.async_init(config_entry.entry_id)
+        assert result["type"] == FlowResultType.FORM
+        assert result["step_id"] == "addon_not_installed"
+
+    result = await hass.config_entries.options.async_configure(
+        result["flow_id"],
+        user_input={
+            "enable_multi_pan": True,
+        },
+    )
+    assert result["type"] == FlowResultType.SHOW_PROGRESS
+    assert result["step_id"] == "install_addon"
+    assert result["progress_action"] == "install_addon"
+
+    result = await hass.config_entries.options.async_configure(result["flow_id"])
+    assert result["type"] == FlowResultType.SHOW_PROGRESS_DONE
+    assert result["step_id"] == "configure_addon"
+    install_addon.assert_called_once_with(hass, "core_silabs_multiprotocol")
+
+    addon_info.return_value["hostname"] = "core-silabs-multiprotocol"
+    result = await hass.config_entries.options.async_configure(result["flow_id"])
+    assert result["type"] == FlowResultType.SHOW_PROGRESS
+    assert result["step_id"] == "start_addon"
+    set_addon_options.assert_called_once_with(
+        hass,
+        "core_silabs_multiprotocol",
+        {
+            "options": {
+                "autoflash_firmware": True,
+                "device": "/dev/ttyTEST123",
+                "baudrate": "115200",
+                "flow_control": True,
+            }
+        },
+    )
+
+    result = await hass.config_entries.options.async_configure(result["flow_id"])
+    assert result["type"] == FlowResultType.SHOW_PROGRESS_DONE
+    assert result["step_id"] == "finish_addon_setup"
+    start_addon.assert_called_once_with(hass, "core_silabs_multiprotocol")
+
+    result = await hass.config_entries.options.async_configure(result["flow_id"])
+    assert result["type"] == FlowResultType.CREATE_ENTRY
+
+    # Check the ZHA entry data is not changed
+    assert zha_config_entry.data == {
+        "device": {
+            "path": "/dev/other_radio",
+            "baudrate": 115200,
+            "flow_control": "hardware",
+        },
+        "radio_type": "ezsp",
+    }
+
+
 async def test_option_flow_non_hassio(
     hass: HomeAssistant,
 ) -> None:
@@ -143,7 +344,7 @@ async def test_option_flow_non_hassio(
         data={},
         domain=TEST_DOMAIN,
         options={},
-        title="Home Assistant Yellow",
+        title="Test HW",
     )
     config_entry.add_to_hass(hass)
 
@@ -165,7 +366,7 @@ async def test_option_flow_addon_installed_other_device(
         data={},
         domain=TEST_DOMAIN,
         options={},
-        title="Home Assistant Yellow",
+        title="Test HW",
     )
     config_entry.add_to_hass(hass)
 
@@ -196,7 +397,7 @@ async def test_option_flow_addon_installed_same_device(
         data={},
         domain=TEST_DOMAIN,
         options={},
-        title="Home Assistant Yellow",
+        title="Test HW",
     )
     config_entry.add_to_hass(hass)
 
@@ -225,7 +426,7 @@ async def test_option_flow_do_not_install_multi_pan_addon(
         data={},
         domain=TEST_DOMAIN,
         options={},
-        title="Home Assistant Yellow",
+        title="Test HW",
     )
     config_entry.add_to_hass(hass)
 
@@ -263,7 +464,7 @@ async def test_option_flow_install_multi_pan_addon_install_fails(
         data={},
         domain=TEST_DOMAIN,
         options={},
-        title="Home Assistant Yellow",
+        title="Test HW",
     )
     config_entry.add_to_hass(hass)
 
@@ -312,7 +513,7 @@ async def test_option_flow_install_multi_pan_addon_start_fails(
         data={},
         domain=TEST_DOMAIN,
         options={},
-        title="Home Assistant Yellow",
+        title="Test HW",
     )
     config_entry.add_to_hass(hass)
 
@@ -382,7 +583,7 @@ async def test_option_flow_install_multi_pan_addon_set_options_fails(
         data={},
         domain=TEST_DOMAIN,
         options={},
-        title="Home Assistant Yellow",
+        title="Test HW",
     )
     config_entry.add_to_hass(hass)
 
@@ -428,7 +629,7 @@ async def test_option_flow_addon_info_fails(
         data={},
         domain=TEST_DOMAIN,
         options={},
-        title="Home Assistant Yellow",
+        title="Test HW",
     )
     config_entry.add_to_hass(hass)
 
@@ -439,3 +640,147 @@ async def test_option_flow_addon_info_fails(
         result = await hass.config_entries.options.async_init(config_entry.entry_id)
         assert result["type"] == FlowResultType.ABORT
         assert result["reason"] == "addon_info_failed"
+
+
+@patch(
+    "homeassistant.components.zha.radio_manager.ZhaMultiPANMigrationHelper.async_initiate_migration",
+    side_effect=Exception("Boom!"),
+)
+async def test_option_flow_install_multi_pan_addon_zha_migration_fails_step_1(
+    mock_initiate_migration,
+    hass: HomeAssistant,
+    addon_store_info,
+    addon_info,
+    install_addon,
+    set_addon_options,
+    start_addon,
+) -> None:
+    """Test installing the multi pan addon."""
+    mock_integration(hass, MockModule("hassio"))
+
+    # Setup the config entry
+    config_entry = MockConfigEntry(
+        data={},
+        domain=TEST_DOMAIN,
+        options={},
+        title="Test HW",
+    )
+    config_entry.add_to_hass(hass)
+
+    zha_config_entry = MockConfigEntry(
+        data={"device": {"path": "/dev/ttyTEST123"}, "radio_type": "ezsp"},
+        domain=ZHA_DOMAIN,
+        options={},
+        title="Test",
+    )
+    zha_config_entry.add_to_hass(hass)
+
+    with patch(
+        "homeassistant.components.homeassistant_hardware.silabs_multiprotocol_addon.is_hassio",
+        side_effect=Mock(return_value=True),
+    ):
+        result = await hass.config_entries.options.async_init(config_entry.entry_id)
+        assert result["type"] == FlowResultType.FORM
+        assert result["step_id"] == "addon_not_installed"
+
+    result = await hass.config_entries.options.async_configure(
+        result["flow_id"],
+        user_input={
+            "enable_multi_pan": True,
+        },
+    )
+    assert result["type"] == FlowResultType.SHOW_PROGRESS
+    assert result["step_id"] == "install_addon"
+    assert result["progress_action"] == "install_addon"
+
+    result = await hass.config_entries.options.async_configure(result["flow_id"])
+    assert result["type"] == FlowResultType.SHOW_PROGRESS_DONE
+    assert result["step_id"] == "configure_addon"
+    install_addon.assert_called_once_with(hass, "core_silabs_multiprotocol")
+
+    result = await hass.config_entries.options.async_configure(result["flow_id"])
+    assert result["type"] == FlowResultType.ABORT
+    assert result["reason"] == "zha_migration_failed"
+    set_addon_options.assert_not_called()
+
+
+@patch(
+    "homeassistant.components.zha.radio_manager.ZhaMultiPANMigrationHelper.async_finish_migration",
+    side_effect=Exception("Boom!"),
+)
+async def test_option_flow_install_multi_pan_addon_zha_migration_fails_step_2(
+    mock_finish_migration,
+    hass: HomeAssistant,
+    addon_store_info,
+    addon_info,
+    install_addon,
+    set_addon_options,
+    start_addon,
+) -> None:
+    """Test installing the multi pan addon."""
+    mock_integration(hass, MockModule("hassio"))
+
+    # Setup the config entry
+    config_entry = MockConfigEntry(
+        data={},
+        domain=TEST_DOMAIN,
+        options={},
+        title="Test HW",
+    )
+    config_entry.add_to_hass(hass)
+
+    zha_config_entry = MockConfigEntry(
+        data={"device": {"path": "/dev/ttyTEST123"}, "radio_type": "ezsp"},
+        domain=ZHA_DOMAIN,
+        options={},
+        title="Test",
+    )
+    zha_config_entry.add_to_hass(hass)
+
+    with patch(
+        "homeassistant.components.homeassistant_hardware.silabs_multiprotocol_addon.is_hassio",
+        side_effect=Mock(return_value=True),
+    ):
+        result = await hass.config_entries.options.async_init(config_entry.entry_id)
+        assert result["type"] == FlowResultType.FORM
+        assert result["step_id"] == "addon_not_installed"
+
+    result = await hass.config_entries.options.async_configure(
+        result["flow_id"],
+        user_input={
+            "enable_multi_pan": True,
+        },
+    )
+    assert result["type"] == FlowResultType.SHOW_PROGRESS
+    assert result["step_id"] == "install_addon"
+    assert result["progress_action"] == "install_addon"
+
+    result = await hass.config_entries.options.async_configure(result["flow_id"])
+    assert result["type"] == FlowResultType.SHOW_PROGRESS_DONE
+    assert result["step_id"] == "configure_addon"
+    install_addon.assert_called_once_with(hass, "core_silabs_multiprotocol")
+
+    result = await hass.config_entries.options.async_configure(result["flow_id"])
+    assert result["type"] == FlowResultType.SHOW_PROGRESS
+    assert result["step_id"] == "start_addon"
+    set_addon_options.assert_called_once_with(
+        hass,
+        "core_silabs_multiprotocol",
+        {
+            "options": {
+                "autoflash_firmware": True,
+                "device": "/dev/ttyTEST123",
+                "baudrate": "115200",
+                "flow_control": True,
+            }
+        },
+    )
+
+    result = await hass.config_entries.options.async_configure(result["flow_id"])
+    assert result["type"] == FlowResultType.SHOW_PROGRESS_DONE
+    assert result["step_id"] == "finish_addon_setup"
+    start_addon.assert_called_once_with(hass, "core_silabs_multiprotocol")
+
+    result = await hass.config_entries.options.async_configure(result["flow_id"])
+    assert result["type"] == FlowResultType.ABORT
+    assert result["reason"] == "zha_migration_failed"

tests/components/homeassistant_yellow/conftest.py

@@ -15,7 +15,10 @@ def mock_zha_config_flow_setup() -> Generator[None, None, None]:
         return {**config, "baudrate": 115200}
 
     mock_connect_app = MagicMock()
-    mock_connect_app.__aenter__.return_value.backups.backups = []
+    mock_connect_app.__aenter__.return_value.backups.backups = [MagicMock()]
+    mock_connect_app.__aenter__.return_value.backups.create_backup.return_value = (
+        MagicMock()
+    )
 
     with patch(
         "bellows.zigbee.application.ControllerApplication.probe", side_effect=mock_probe

tests/components/homeassistant_yellow/test_config_flow.py

@@ -2,6 +2,7 @@
 from unittest.mock import Mock, patch
 
 from homeassistant.components.homeassistant_yellow.const import DOMAIN
+from homeassistant.components.zha.core.const import DOMAIN as ZHA_DOMAIN
 from homeassistant.core import HomeAssistant
 from homeassistant.data_entry_flow import FlowResultType
 
@@ -124,3 +125,88 @@ async def test_option_flow_install_multi_pan_addon(
 
     result = await hass.config_entries.options.async_configure(result["flow_id"])
     assert result["type"] == FlowResultType.CREATE_ENTRY
+
+
+async def test_option_flow_install_multi_pan_addon_zha(
+    hass: HomeAssistant,
+    addon_store_info,
+    addon_info,
+    install_addon,
+    set_addon_options,
+    start_addon,
+) -> None:
+    """Test installing the multi pan addon when a zha config entry exists."""
+    mock_integration(hass, MockModule("hassio"))
+
+    # Setup the config entry
+    config_entry = MockConfigEntry(
+        data={},
+        domain=DOMAIN,
+        options={},
+        title="Home Assistant Yellow",
+    )
+    config_entry.add_to_hass(hass)
+
+    zha_config_entry = MockConfigEntry(
+        data={"device": {"path": "/dev/ttyAMA1"}, "radio_type": "ezsp"},
+        domain=ZHA_DOMAIN,
+        options={},
+        title="Yellow",
+    )
+    zha_config_entry.add_to_hass(hass)
+
+    with patch(
+        "homeassistant.components.homeassistant_hardware.silabs_multiprotocol_addon.is_hassio",
+        side_effect=Mock(return_value=True),
+    ):
+        result = await hass.config_entries.options.async_init(config_entry.entry_id)
+        assert result["type"] == FlowResultType.FORM
+        assert result["step_id"] == "addon_not_installed"
+
+    result = await hass.config_entries.options.async_configure(
+        result["flow_id"],
+        user_input={
+            "enable_multi_pan": True,
+        },
+    )
+    assert result["type"] == FlowResultType.SHOW_PROGRESS
+    assert result["step_id"] == "install_addon"
+    assert result["progress_action"] == "install_addon"
+
+    result = await hass.config_entries.options.async_configure(result["flow_id"])
+    assert result["type"] == FlowResultType.SHOW_PROGRESS_DONE
+    assert result["step_id"] == "configure_addon"
+    install_addon.assert_called_once_with(hass, "core_silabs_multiprotocol")
+
+    result = await hass.config_entries.options.async_configure(result["flow_id"])
+    assert result["type"] == FlowResultType.SHOW_PROGRESS
+    assert result["step_id"] == "start_addon"
+    set_addon_options.assert_called_once_with(
+        hass,
+        "core_silabs_multiprotocol",
+        {
+            "options": {
+                "autoflash_firmware": True,
+                "device": "/dev/ttyAMA1",
+                "baudrate": "115200",
+                "flow_control": True,
+            }
+        },
+    )
+    # Check the ZHA config entry data is updated
+    assert zha_config_entry.data == {
+        "device": {
+            "path": "socket://core-silabs-multiprotocol:9999",
+            "baudrate": 115200,
+            "flow_control": "hardware",
+        },
+        "radio_type": "ezsp",
+    }
+
+    result = await hass.config_entries.options.async_configure(result["flow_id"])
+    assert result["type"] == FlowResultType.SHOW_PROGRESS_DONE
+    assert result["step_id"] == "finish_addon_setup"
+    start_addon.assert_called_once_with(hass, "core_silabs_multiprotocol")
+
+    result = await hass.config_entries.options.async_configure(result["flow_id"])
+    assert result["type"] == FlowResultType.CREATE_ENTRY

tests/components/homeassistant_yellow/test_init.py

@@ -149,7 +149,7 @@ async def test_setup_zha_multipan(
         "radio_type": "ezsp",
     }
     assert config_entry.options == {}
-    assert config_entry.title == "Yellow"
+    assert config_entry.title == "Yellow Multi-PAN"
 
 
 async def test_setup_entry_wrong_board(hass: HomeAssistant) -> None:

tests/components/zha/test_config_flow.py

@@ -16,7 +16,7 @@ import zigpy.types
 from homeassistant import config_entries
 from homeassistant.components import ssdp, usb, zeroconf
 from homeassistant.components.ssdp import ATTR_UPNP_MANUFACTURER_URL, ATTR_UPNP_SERIAL
-from homeassistant.components.zha import config_flow
+from homeassistant.components.zha import config_flow, radio_manager
 from homeassistant.components.zha.core.const import (
     CONF_BAUDRATE,
     CONF_FLOWCONTROL,
@@ -1024,7 +1024,7 @@ async def test_hardware_invalid_data(hass, data):
 def test_allow_overwrite_ezsp_ieee():
     """Test modifying the backup to allow bellows to override the IEEE address."""
     backup = zigpy.backups.NetworkBackup()
-    new_backup = config_flow._allow_overwrite_ezsp_ieee(backup)
+    new_backup = radio_manager._allow_overwrite_ezsp_ieee(backup)
 
     assert backup != new_backup
     assert new_backup.network_info.stack_specific["ezsp"][EZSP_OVERWRITE_EUI64] is True
@@ -1034,7 +1034,7 @@ def test_prevent_overwrite_ezsp_ieee():
     """Test modifying the backup to prevent bellows from overriding the IEEE address."""
     backup = zigpy.backups.NetworkBackup()
     backup.network_info.stack_specific["ezsp"] = {EZSP_OVERWRITE_EUI64: True}
-    new_backup = config_flow._prevent_overwrite_ezsp_ieee(backup)
+    new_backup = radio_manager._prevent_overwrite_ezsp_ieee(backup)
 
     assert backup != new_backup
     assert not new_backup.network_info.stack_specific.get("ezsp", {}).get(
@@ -1131,7 +1131,7 @@ def test_parse_uploaded_backup(process_mock):
     assert backup == parsed_backup
 
 
-@patch("homeassistant.components.zha.config_flow._allow_overwrite_ezsp_ieee")
+@patch("homeassistant.components.zha.radio_manager._allow_overwrite_ezsp_ieee")
 async def test_formation_strategy_restore_manual_backup_non_ezsp(
     allow_overwrite_ieee_mock, pick_radio, mock_app, hass
 ):
@@ -1163,7 +1163,7 @@ async def test_formation_strategy_restore_manual_backup_non_ezsp(
     assert result3["data"][CONF_RADIO_TYPE] == "znp"
 
 
-@patch("homeassistant.components.zha.config_flow._allow_overwrite_ezsp_ieee")
+@patch("homeassistant.components.zha.radio_manager._allow_overwrite_ezsp_ieee")
 async def test_formation_strategy_restore_manual_backup_overwrite_ieee_ezsp(
     allow_overwrite_ieee_mock, pick_radio, mock_app, backup, hass
 ):
@@ -1203,7 +1203,7 @@ async def test_formation_strategy_restore_manual_backup_overwrite_ieee_ezsp(
     assert result4["data"][CONF_RADIO_TYPE] == "ezsp"
 
 
-@patch("homeassistant.components.zha.config_flow._allow_overwrite_ezsp_ieee")
+@patch("homeassistant.components.zha.radio_manager._allow_overwrite_ezsp_ieee")
 async def test_formation_strategy_restore_manual_backup_ezsp(
     allow_overwrite_ieee_mock, pick_radio, mock_app, hass
 ):
@@ -1391,7 +1391,7 @@ async def test_formation_strategy_restore_automatic_backup_non_ezsp(
     assert result3["data"][CONF_RADIO_TYPE] == "znp"
 
 
-@patch("homeassistant.components.zha.config_flow._allow_overwrite_ezsp_ieee")
+@patch("homeassistant.components.zha.radio_manager._allow_overwrite_ezsp_ieee")
 async def test_ezsp_restore_without_settings_change_ieee(
     allow_overwrite_ieee_mock, pick_radio, mock_app, backup, hass
 ):

tests/components/zha/test_radio_manager.py

@@ -0,0 +1,311 @@
+"""Tests for ZHA config flow."""
+
+from collections.abc import Generator
+from unittest.mock import AsyncMock, MagicMock, create_autospec, patch
+
+import pytest
+import serial.tools.list_ports
+from zigpy.backups import BackupManager
+import zigpy.config
+from zigpy.config import CONF_DEVICE_PATH
+import zigpy.types
+
+from homeassistant import config_entries
+from homeassistant.components.zha import radio_manager
+from homeassistant.components.zha.core.const import DOMAIN, RadioType
+from homeassistant.core import HomeAssistant
+
+from tests.common import MockConfigEntry
+
+PROBE_FUNCTION_PATH = "zigbee.application.ControllerApplication.probe"
+
+
+@pytest.fixture(autouse=True)
+def disable_platform_only():
+    """Disable platforms to speed up tests."""
+    with patch("homeassistant.components.zha.PLATFORMS", []):
+        yield
+
+
+@pytest.fixture(autouse=True)
+def reduce_reconnect_timeout():
+    """Reduces reconnect timeout to speed up tests."""
+    with patch("homeassistant.components.zha.radio_manager.CONNECT_DELAY_S", 0.0001):
+        yield
+
+
+@pytest.fixture(autouse=True)
+def mock_app():
+    """Mock zigpy app interface."""
+    mock_app = AsyncMock()
+    mock_app.backups = create_autospec(BackupManager, instance=True)
+    mock_app.backups.backups = []
+
+    with patch(
+        "zigpy.application.ControllerApplication.new", AsyncMock(return_value=mock_app)
+    ):
+        yield mock_app
+
+
+@pytest.fixture
+def backup():
+    """Zigpy network backup with non-default settings."""
+    backup = zigpy.backups.NetworkBackup()
+    backup.node_info.ieee = zigpy.types.EUI64.convert("AA:BB:CC:DD:11:22:33:44")
+
+    return backup
+
+
+def mock_detect_radio_type(radio_type=RadioType.ezsp, ret=True):
+    """Mock `detect_radio_type` that just sets the appropriate attributes."""
+
+    async def detect(self):
+        self.radio_type = radio_type
+        self.device_settings = radio_type.controller.SCHEMA_DEVICE(
+            {CONF_DEVICE_PATH: self.device_path}
+        )
+
+        return ret
+
+    return detect
+
+
+def com_port(device="/dev/ttyUSB1234"):
+    """Mock of a serial port."""
+    port = serial.tools.list_ports_common.ListPortInfo("/dev/ttyUSB1234")
+    port.serial_number = "1234"
+    port.manufacturer = "Virtual serial port"
+    port.device = device
+    port.description = "Some serial port"
+
+    return port
+
+
+@pytest.fixture()
+def mock_connect_zigpy_app() -> Generator[None, None, None]:
+    """Mock the radio connection."""
+
+    mock_connect_app = MagicMock()
+    mock_connect_app.__aenter__.return_value.backups.backups = [MagicMock()]
+    mock_connect_app.__aenter__.return_value.backups.create_backup.return_value = (
+        MagicMock()
+    )
+
+    with patch(
+        "homeassistant.components.zha.radio_manager.ZhaRadioManager._connect_zigpy_app",
+        return_value=mock_connect_app,
+    ):
+        yield
+
+
+@patch("homeassistant.components.zha.async_setup_entry", AsyncMock(return_value=True))
+async def test_migrate_matching_port(
+    hass: HomeAssistant,
+    mock_connect_zigpy_app,
+) -> None:
+    """Test automatic migration."""
+    # Setup the config entry
+    config_entry = MockConfigEntry(
+        data={"device": {"path": "/dev/ttyTEST123"}, "radio_type": "ezsp"},
+        domain=DOMAIN,
+        options={},
+        title="Test",
+        version=3,
+    )
+    config_entry.add_to_hass(hass)
+
+    migration_data = {
+        "new_discovery_info": {
+            "name": "Test Updated",
+            "port": {
+                "path": "socket://some/virtual_port",
+                "baudrate": 115200,
+                "flow_control": "hardware",
+            },
+            "radio_type": "efr32",
+        },
+        "old_discovery_info": {
+            "name": "Test",
+            "port": {
+                "path": "/dev/ttyTEST123",
+                "baudrate": 115200,
+                "flow_control": "hardware",
+            },
+            "radio_type": "efr32",
+        },
+    }
+
+    migration_helper = radio_manager.ZhaMultiPANMigrationHelper(hass, config_entry)
+    assert await migration_helper.async_initiate_migration(migration_data)
+
+    # Check the ZHA config entry data is updated
+    assert config_entry.data == {
+        "device": {
+            "path": "socket://some/virtual_port",
+            "baudrate": 115200,
+            "flow_control": "hardware",
+        },
+        "radio_type": "ezsp",
+    }
+    assert config_entry.title == "Test Updated"
+
+    await migration_helper.async_finish_migration()
+
+
+async def test_migrate_matching_port_config_entry_not_loaded(
+    hass: HomeAssistant,
+    mock_connect_zigpy_app,
+) -> None:
+    """Test automatic migration."""
+    # Setup the config entry
+    config_entry = MockConfigEntry(
+        data={"device": {"path": "/dev/ttyTEST123"}, "radio_type": "ezsp"},
+        domain=DOMAIN,
+        options={},
+        title="Test",
+    )
+    config_entry.add_to_hass(hass)
+    config_entry.state = config_entries.ConfigEntryState.SETUP_IN_PROGRESS
+
+    migration_data = {
+        "new_discovery_info": {
+            "name": "Test Updated",
+            "port": {
+                "path": "socket://some/virtual_port",
+                "baudrate": 115200,
+                "flow_control": "hardware",
+            },
+            "radio_type": "efr32",
+        },
+        "old_discovery_info": {
+            "name": "Test",
+            "port": {
+                "path": "/dev/ttyTEST123",
+                "baudrate": 115200,
+                "flow_control": "hardware",
+            },
+            "radio_type": "efr32",
+        },
+    }
+
+    migration_helper = radio_manager.ZhaMultiPANMigrationHelper(hass, config_entry)
+    assert await migration_helper.async_initiate_migration(migration_data)
+
+    # Check the ZHA config entry data is updated
+    assert config_entry.data == {
+        "device": {
+            "path": "socket://some/virtual_port",
+            "baudrate": 115200,
+            "flow_control": "hardware",
+        },
+        "radio_type": "ezsp",
+    }
+    assert config_entry.title == "Test Updated"
+
+    await migration_helper.async_finish_migration()
+
+
+@patch(
+    "homeassistant.components.zha.radio_manager.ZhaRadioManager.async_restore_backup_step_1",
+    side_effect=OSError,
+)
+async def test_migrate_matching_port_retry(
+    mock_restore_backup_step_1,
+    hass: HomeAssistant,
+    mock_connect_zigpy_app,
+) -> None:
+    """Test automatic migration."""
+    # Setup the config entry
+    config_entry = MockConfigEntry(
+        data={"device": {"path": "/dev/ttyTEST123"}, "radio_type": "ezsp"},
+        domain=DOMAIN,
+        options={},
+        title="Test",
+    )
+    config_entry.add_to_hass(hass)
+    config_entry.state = config_entries.ConfigEntryState.SETUP_IN_PROGRESS
+
+    migration_data = {
+        "new_discovery_info": {
+            "name": "Test Updated",
+            "port": {
+                "path": "socket://some/virtual_port",
+                "baudrate": 115200,
+                "flow_control": "hardware",
+            },
+            "radio_type": "efr32",
+        },
+        "old_discovery_info": {
+            "name": "Test",
+            "port": {
+                "path": "/dev/ttyTEST123",
+                "baudrate": 115200,
+                "flow_control": "hardware",
+            },
+            "radio_type": "efr32",
+        },
+    }
+
+    migration_helper = radio_manager.ZhaMultiPANMigrationHelper(hass, config_entry)
+    assert await migration_helper.async_initiate_migration(migration_data)
+
+    # Check the ZHA config entry data is updated
+    assert config_entry.data == {
+        "device": {
+            "path": "socket://some/virtual_port",
+            "baudrate": 115200,
+            "flow_control": "hardware",
+        },
+        "radio_type": "ezsp",
+    }
+    assert config_entry.title == "Test Updated"
+
+    with pytest.raises(OSError):
+        await migration_helper.async_finish_migration()
+    assert mock_restore_backup_step_1.call_count == 100
+
+
+async def test_migrate_non_matching_port(
+    hass: HomeAssistant,
+    mock_connect_zigpy_app,
+) -> None:
+    """Test automatic migration."""
+    # Setup the config entry
+    config_entry = MockConfigEntry(
+        data={"device": {"path": "/dev/ttyTEST123"}, "radio_type": "ezsp"},
+        domain=DOMAIN,
+        options={},
+        title="Test",
+    )
+    config_entry.add_to_hass(hass)
+
+    migration_data = {
+        "new_discovery_info": {
+            "name": "Test Updated",
+            "port": {
+                "path": "socket://some/virtual_port",
+                "baudrate": 115200,
+                "flow_control": "hardware",
+            },
+            "radio_type": "efr32",
+        },
+        "old_discovery_info": {
+            "name": "Test",
+            "port": {
+                "path": "/dev/ttyTEST456",
+                "baudrate": 115200,
+                "flow_control": "hardware",
+            },
+            "radio_type": "efr32",
+        },
+    }
+
+    migration_helper = radio_manager.ZhaMultiPANMigrationHelper(hass, config_entry)
+    assert not await migration_helper.async_initiate_migration(migration_data)
+
+    # Check the ZHA config entry data is not updated
+    assert config_entry.data == {
+        "device": {"path": "/dev/ttyTEST123"},
+        "radio_type": "ezsp",
+    }
+    assert config_entry.title == "Test"