Create a ZHA repair when directly accessing a radio with multi-PAN firmware (#98275)

* Add the SiLabs flasher as a dependency

* Create a repair if the wrong firmware is detected on an EZSP device

* Update homeassistant/components/zha/strings.json

Co-authored-by: c0ffeeca7 <38767475+c0ffeeca7@users.noreply.github.com>

* Provide the ZHA config entry as a reusable fixture

* Create a separate repair when using non-Nabu Casa hardware

* Add unit tests

* Drop extraneous `config_entry.add_to_hass` added in 021def44

* Fully unit test all edge cases

* Move `socket://`-ignoring logic into repair function

* Open a repair from ZHA flows when the wrong firmware is running

* Fix existing unit tests

* Link to the flashing section in the documentation

* Reduce repair severity to `ERROR`

* Make issue persistent

* Add unit tests for new radio probing states

* Add unit tests for new config flow steps

* Handle probing failure raising an exception

* Implement review suggestions

* Address review comments

---------

Co-authored-by: c0ffeeca7 <38767475+c0ffeeca7@users.noreply.github.com>

homeassistant/components/zha/__init__.py

@@ -12,13 +12,14 @@ from zigpy.config import CONF_DEVICE, CONF_DEVICE_PATH
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.const import CONF_TYPE
 from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import ConfigEntryNotReady
 from homeassistant.helpers import device_registry as dr
 import homeassistant.helpers.config_validation as cv
 from homeassistant.helpers.dispatcher import async_dispatcher_send
 from homeassistant.helpers.storage import STORAGE_DIR
 from homeassistant.helpers.typing import ConfigType
 
-from . import websocket_api
+from . import repairs, websocket_api
 from .core import ZHAGateway
 from .core.const import (
     BAUD_RATES,
@@ -134,7 +135,23 @@ async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> b
         _LOGGER.debug("ZHA storage file does not exist or was already removed")
 
     zha_gateway = ZHAGateway(hass, config, config_entry)
+
+    try:
         await zha_gateway.async_initialize()
+    except Exception:  # pylint: disable=broad-except
+        if RadioType[config_entry.data[CONF_RADIO_TYPE]] == RadioType.ezsp:
+            try:
+                await repairs.warn_on_wrong_silabs_firmware(
+                    hass, config_entry.data[CONF_DEVICE][CONF_DEVICE_PATH]
+                )
+            except repairs.AlreadyRunningEZSP as exc:
+                # If connecting fails but we somehow probe EZSP (e.g. stuck in the
+                # bootloader), reconnect, it should work
+                raise ConfigEntryNotReady from exc
+
+        raise
+
+    repairs.async_delete_blocking_issues(hass)
 
     config_entry.async_on_unload(zha_gateway.shutdown)
 

homeassistant/components/zha/config_flow.py

@@ -35,6 +35,7 @@ from .core.const import (
 from .radio_manager import (
     HARDWARE_DISCOVERY_SCHEMA,
     RECOMMENDED_RADIOS,
+    ProbeResult,
     ZhaRadioManager,
 )
 
@@ -60,6 +61,8 @@ OPTIONS_INTENT_RECONFIGURE = "intent_reconfigure"
 
 UPLOADED_BACKUP_FILE = "uploaded_backup_file"
 
+REPAIR_MY_URL = "https://my.home-assistant.io/redirect/repairs/"
+
 DEFAULT_ZHA_ZEROCONF_PORT = 6638
 ESPHOME_API_PORT = 6053
 
@@ -187,7 +190,13 @@ class BaseZhaFlow(FlowHandler):
             port = ports[list_of_ports.index(user_selection)]
             self._radio_mgr.device_path = port.device
 
-            if not await self._radio_mgr.detect_radio_type():
+            probe_result = await self._radio_mgr.detect_radio_type()
+            if probe_result == ProbeResult.WRONG_FIRMWARE_INSTALLED:
+                return self.async_abort(
+                    reason="wrong_firmware_installed",
+                    description_placeholders={"repair_url": REPAIR_MY_URL},
+                )
+            if probe_result == ProbeResult.PROBING_FAILED:
                 # Did not autodetect anything, proceed to manual selection
                 return await self.async_step_manual_pick_radio_type()
 
@@ -530,10 +539,17 @@ class ZhaConfigFlowHandler(BaseZhaFlow, config_entries.ConfigFlow, domain=DOMAIN
         # config flow logic that interacts with hardware.
         if user_input is not None or not onboarding.async_is_onboarded(self.hass):
             # Probe the radio type if we don't have one yet
-            if (
+            if self._radio_mgr.radio_type is None:
-                self._radio_mgr.radio_type is None
+                probe_result = await self._radio_mgr.detect_radio_type()
-                and not await self._radio_mgr.detect_radio_type()
+            else:
-            ):
+                probe_result = ProbeResult.RADIO_TYPE_DETECTED
+
+            if probe_result == ProbeResult.WRONG_FIRMWARE_INSTALLED:
+                return self.async_abort(
+                    reason="wrong_firmware_installed",
+                    description_placeholders={"repair_url": REPAIR_MY_URL},
+                )
+            if probe_result == ProbeResult.PROBING_FAILED:
                 # This path probably will not happen now that we have
                 # more precise USB matching unless there is a problem
                 # with the device

homeassistant/components/zha/manifest.json

@@ -17,7 +17,8 @@
     "zigpy_deconz",
     "zigpy_xbee",
     "zigpy_zigate",
-    "zigpy_znp"
+    "zigpy_znp",
+    "universal_silabs_flasher"
   ],
   "requirements": [
     "bellows==0.36.1",
@@ -28,7 +29,8 @@
     "zigpy==0.57.0",
     "zigpy-xbee==0.18.1",
     "zigpy-zigate==0.11.0",
-    "zigpy-znp==0.11.4"
+    "zigpy-znp==0.11.4",
+    "universal-silabs-flasher==0.0.13"
   ],
   "usb": [
     {

homeassistant/components/zha/radio_manager.py

@@ -5,6 +5,7 @@ import asyncio
 import contextlib
 from contextlib import suppress
 import copy
+import enum
 import logging
 import os
 from typing import Any
@@ -20,6 +21,7 @@ from homeassistant import config_entries
 from homeassistant.components import usb
 from homeassistant.core import HomeAssistant
 
+from . import repairs
 from .core.const import (
     CONF_DATABASE,
     CONF_RADIO_TYPE,
@@ -76,6 +78,14 @@ HARDWARE_MIGRATION_SCHEMA = vol.Schema(
 _LOGGER = logging.getLogger(__name__)
 
 
+class ProbeResult(enum.StrEnum):
+    """Radio firmware probing result."""
+
+    RADIO_TYPE_DETECTED = "radio_type_detected"
+    WRONG_FIRMWARE_INSTALLED = "wrong_firmware_installed"
+    PROBING_FAILED = "probing_failed"
+
+
 def _allow_overwrite_ezsp_ieee(
     backup: zigpy.backups.NetworkBackup,
 ) -> zigpy.backups.NetworkBackup:
@@ -171,8 +181,10 @@ class ZhaRadioManager:
 
         return RadioType[radio_type]
 
-    async def detect_radio_type(self) -> bool:
+    async def detect_radio_type(self) -> ProbeResult:
         """Probe all radio types on the current port."""
+        assert self.device_path is not None
+
         for radio in AUTOPROBE_RADIOS:
             _LOGGER.debug("Attempting to probe radio type %s", radio)
 
@@ -191,9 +203,14 @@ class ZhaRadioManager:
             self.radio_type = radio
             self.device_settings = dev_config
 
-            return True
+            repairs.async_delete_blocking_issues(self.hass)
+            return ProbeResult.RADIO_TYPE_DETECTED
 
-        return False
+        with suppress(repairs.AlreadyRunningEZSP):
+            if await repairs.warn_on_wrong_silabs_firmware(self.hass, self.device_path):
+                return ProbeResult.WRONG_FIRMWARE_INSTALLED
+
+        return ProbeResult.PROBING_FAILED
 
     async def async_load_network_settings(
         self, *, create_backup: bool = False

homeassistant/components/zha/repairs.py

@@ -0,0 +1,126 @@
+"""ZHA repairs for common environmental and device problems."""
+from __future__ import annotations
+
+import enum
+import logging
+
+from universal_silabs_flasher.const import ApplicationType
+from universal_silabs_flasher.flasher import Flasher
+
+from homeassistant.components.homeassistant_sky_connect import (
+    hardware as skyconnect_hardware,
+)
+from homeassistant.components.homeassistant_yellow import (
+    RADIO_DEVICE as YELLOW_RADIO_DEVICE,
+    hardware as yellow_hardware,
+)
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import HomeAssistantError
+from homeassistant.helpers import issue_registry as ir
+
+from .core.const import DOMAIN
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class AlreadyRunningEZSP(Exception):
+    """The device is already running EZSP firmware."""
+
+
+class HardwareType(enum.StrEnum):
+    """Detected Zigbee hardware type."""
+
+    SKYCONNECT = "skyconnect"
+    YELLOW = "yellow"
+    OTHER = "other"
+
+
+DISABLE_MULTIPAN_URL = {
+    HardwareType.YELLOW: (
+        "https://yellow.home-assistant.io/guides/disable-multiprotocol/#flash-the-silicon-labs-radio-firmware"
+    ),
+    HardwareType.SKYCONNECT: (
+        "https://skyconnect.home-assistant.io/procedures/disable-multiprotocol/#step-flash-the-silicon-labs-radio-firmware"
+    ),
+    HardwareType.OTHER: None,
+}
+
+ISSUE_WRONG_SILABS_FIRMWARE_INSTALLED = "wrong_silabs_firmware_installed"
+
+
+def _detect_radio_hardware(hass: HomeAssistant, device: str) -> HardwareType:
+    """Identify the radio hardware with the given serial port."""
+    try:
+        yellow_hardware.async_info(hass)
+    except HomeAssistantError:
+        pass
+    else:
+        if device == YELLOW_RADIO_DEVICE:
+            return HardwareType.YELLOW
+
+    try:
+        info = skyconnect_hardware.async_info(hass)
+    except HomeAssistantError:
+        pass
+    else:
+        for hardware_info in info:
+            for entry_id in hardware_info.config_entries or []:
+                entry = hass.config_entries.async_get_entry(entry_id)
+
+                if entry is not None and entry.data["device"] == device:
+                    return HardwareType.SKYCONNECT
+
+    return HardwareType.OTHER
+
+
+async def probe_silabs_firmware_type(device: str) -> ApplicationType | None:
+    """Probe the running firmware on a Silabs device."""
+    flasher = Flasher(device=device)
+
+    try:
+        await flasher.probe_app_type()
+    except Exception:  # pylint: disable=broad-except
+        _LOGGER.debug("Failed to probe application type", exc_info=True)
+
+    return flasher.app_type
+
+
+async def warn_on_wrong_silabs_firmware(hass: HomeAssistant, device: str) -> bool:
+    """Create a repair issue if the wrong type of SiLabs firmware is detected."""
+    # Only consider actual serial ports
+    if device.startswith("socket://"):
+        return False
+
+    app_type = await probe_silabs_firmware_type(device)
+
+    if app_type is None:
+        # Failed to probe, we can't tell if the wrong firmware is installed
+        return False
+
+    if app_type == ApplicationType.EZSP:
+        # If connecting fails but we somehow probe EZSP (e.g. stuck in bootloader),
+        # reconnect, it should work
+        raise AlreadyRunningEZSP()
+
+    hardware_type = _detect_radio_hardware(hass, device)
+    ir.async_create_issue(
+        hass,
+        domain=DOMAIN,
+        issue_id=ISSUE_WRONG_SILABS_FIRMWARE_INSTALLED,
+        is_fixable=False,
+        is_persistent=True,
+        learn_more_url=DISABLE_MULTIPAN_URL[hardware_type],
+        severity=ir.IssueSeverity.ERROR,
+        translation_key=(
+            ISSUE_WRONG_SILABS_FIRMWARE_INSTALLED
+            + ("_nabucasa" if hardware_type != HardwareType.OTHER else "_other")
+        ),
+        translation_placeholders={"firmware_type": app_type.name},
+    )
+
+    return True
+
+
+def async_delete_blocking_issues(hass: HomeAssistant) -> None:
+    """Delete repair issues that should disappear on a successful startup."""
+    ir.async_delete_issue(hass, DOMAIN, ISSUE_WRONG_SILABS_FIRMWARE_INSTALLED)

homeassistant/components/zha/strings.json

@@ -75,7 +75,8 @@
     "abort": {
       "single_instance_allowed": "[%key:common::config_flow::abort::single_instance_allowed%]",
       "not_zha_device": "This device is not a zha device",
-      "usb_probe_failed": "Failed to probe the usb device"
+      "usb_probe_failed": "Failed to probe the usb device",
+      "wrong_firmware_installed": "Your device is running the wrong firmware and cannot be used with ZHA until the correct firmware is installed. [A repair has been created]({repair_url}) with more information and instructions for how to fix this."
     }
   },
   "options": {
@@ -168,7 +169,8 @@
     "abort": {
       "single_instance_allowed": "[%key:component::zha::config::abort::single_instance_allowed%]",
       "not_zha_device": "[%key:component::zha::config::abort::not_zha_device%]",
-      "usb_probe_failed": "[%key:component::zha::config::abort::usb_probe_failed%]"
+      "usb_probe_failed": "[%key:component::zha::config::abort::usb_probe_failed%]",
+      "wrong_firmware_installed": "[%key:component::zha::config::abort::wrong_firmware_installed%]"
     }
   },
   "config_panel": {
@@ -502,5 +504,15 @@
         }
       }
     }
+  },
+  "issues": {
+    "wrong_silabs_firmware_installed_nabucasa": {
+      "title": "Zigbee radio with multiprotocol firmware detected",
+      "description": "Your Zigbee radio was previously used with multiprotocol (Zigbee and Thread) and still has multiprotocol firmware installed: ({firmware_type}). \n Option 1: To run your radio exclusively with ZHA, you need to install the Zigbee firmware:\n - Open the documentation by selecting the link under \"Learn More\".\n -. Follow the instructions described in the step to flash the Zigbee firmware.\n Option 2: To run your radio with multiprotocol, follow these steps: \n - Go to Settings > System > Hardware, select the device and select Configure. \n - Select the Configure IEEE 802.15.4 radio multiprotocol support option. \n - Select the checkbox and select Submit. \n - Once installed, configure the newly discovered ZHA integration."
+    },
+    "wrong_silabs_firmware_installed_other": {
+      "title": "[%key:component::zha::issues::wrong_silabs_firmware_installed_nabucasa::title%]",
+      "description": "Your Zigbee radio was previously used with multiprotocol (Zigbee and Thread) and still has multiprotocol firmware installed: ({firmware_type}). To run your radio exclusively with ZHA, you need to install Zigbee firmware. Follow your Zigbee radio manufacturer's instructions for how to do this."
+    }
   }
 }

requirements_all.txt

@@ -2611,6 +2611,9 @@ unifi-discovery==1.1.7
 # homeassistant.components.unifiled
 unifiled==0.11
 
+# homeassistant.components.zha
+universal-silabs-flasher==0.0.13
+
 # homeassistant.components.upb
 upb-lib==0.5.4
 

requirements_test_all.txt

@@ -1908,6 +1908,9 @@ ultraheat-api==0.5.1
 # homeassistant.components.unifiprotect
 unifi-discovery==1.1.7
 
+# homeassistant.components.zha
+universal-silabs-flasher==0.0.13
+
 # homeassistant.components.upb
 upb-lib==0.5.4
 

script/hassfest/dependencies.py

@@ -149,6 +149,7 @@ IGNORE_VIOLATIONS = {
     ("http", "network"),
     # This would be a circular dep
     ("zha", "homeassistant_hardware"),
+    ("zha", "homeassistant_sky_connect"),
     ("zha", "homeassistant_yellow"),
     # This should become a helper method that integrations can submit data to
     ("websocket_api", "lovelace"),

tests/components/zha/conftest.py

@@ -1,5 +1,5 @@
 """Test configuration for the ZHA component."""
-from collections.abc import Callable
+from collections.abc import Callable, Generator
 import itertools
 import time
 from unittest.mock import AsyncMock, MagicMock, patch
@@ -155,10 +155,10 @@ def zigpy_app_controller():
 
 
 @pytest.fixture(name="config_entry")
-async def config_entry_fixture(hass):
+async def config_entry_fixture(hass) -> MockConfigEntry:
     """Fixture representing a config entry."""
-    entry = MockConfigEntry(
+    return MockConfigEntry(
-        version=2,
+        version=3,
         domain=zha_const.DOMAIN,
         data={
             zigpy.config.CONF_DEVICE: {zigpy.config.CONF_DEVICE_PATH: "/dev/ttyUSB0"},
@@ -178,23 +178,30 @@ async def config_entry_fixture(hass):
             }
         },
     )
-    entry.add_to_hass(hass)
-    return entry
 
 
 @pytest.fixture
-def setup_zha(hass, config_entry, zigpy_app_controller):
+def mock_zigpy_connect(
+    zigpy_app_controller: ControllerApplication,
+) -> Generator[ControllerApplication, None, None]:
+    """Patch the zigpy radio connection with our mock application."""
+    with patch(
+        "bellows.zigbee.application.ControllerApplication.new",
+        return_value=zigpy_app_controller,
+    ) as mock_app:
+        yield mock_app
+
+
+@pytest.fixture
+def setup_zha(hass, config_entry: MockConfigEntry, mock_zigpy_connect):
     """Set up ZHA component."""
     zha_config = {zha_const.CONF_ENABLE_QUIRKS: False}
 
-    p1 = patch(
-        "bellows.zigbee.application.ControllerApplication.new",
-        return_value=zigpy_app_controller,
-    )
-
     async def _setup(config=None):
+        config_entry.add_to_hass(hass)
         config = config or {}
-        with p1:
+
+        with mock_zigpy_connect:
             status = await async_setup_component(
                 hass, zha_const.DOMAIN, {zha_const.DOMAIN: {**zha_config, **config}}
             )

tests/components/zha/test_config_flow.py

@@ -26,6 +26,7 @@ from homeassistant.components.zha.core.const import (
     EZSP_OVERWRITE_EUI64,
     RadioType,
 )
+from homeassistant.components.zha.radio_manager import ProbeResult
 from homeassistant.config_entries import (
     SOURCE_SSDP,
     SOURCE_USB,
@@ -114,7 +115,10 @@ def backup(make_backup):
     return make_backup()
 
 
-def mock_detect_radio_type(radio_type=RadioType.ezsp, ret=True):
+def mock_detect_radio_type(
+    radio_type: RadioType = RadioType.ezsp,
+    ret: ProbeResult = ProbeResult.RADIO_TYPE_DETECTED,
+):
     """Mock `detect_radio_type` that just sets the appropriate attributes."""
 
     async def detect(self):
@@ -489,8 +493,11 @@ async def test_zigate_discovery_via_usb(probe_mock, hass: HomeAssistant) -> None
     }
 
 
-@patch(f"bellows.{PROBE_FUNCTION_PATH}", return_value=False)
+@patch(
-async def test_discovery_via_usb_no_radio(probe_mock, hass: HomeAssistant) -> None:
+    "homeassistant.components.zha.radio_manager.ZhaRadioManager.detect_radio_type",
+    AsyncMock(return_value=ProbeResult.PROBING_FAILED),
+)
+async def test_discovery_via_usb_no_radio(hass: HomeAssistant) -> None:
     """Test usb flow -- no radio detected."""
     discovery_info = usb.UsbServiceInfo(
         device="/dev/null",
@@ -759,7 +766,7 @@ async def test_user_flow(hass: HomeAssistant) -> None:
 
 @patch(
     "homeassistant.components.zha.radio_manager.ZhaRadioManager.detect_radio_type",
-    mock_detect_radio_type(ret=False),
+    AsyncMock(return_value=ProbeResult.PROBING_FAILED),
 )
 @patch("serial.tools.list_ports.comports", MagicMock(return_value=[com_port()]))
 async def test_user_flow_not_detected(hass: HomeAssistant) -> None:
@@ -851,6 +858,7 @@ async def test_detect_radio_type_success(
 
     handler = config_flow.ZhaConfigFlowHandler()
     handler._radio_mgr.device_path = "/dev/null"
+    handler.hass = hass
 
     await handler._radio_mgr.detect_radio_type()
 
@@ -879,6 +887,8 @@ async def test_detect_radio_type_success_with_settings(
 
     handler = config_flow.ZhaConfigFlowHandler()
     handler._radio_mgr.device_path = "/dev/null"
+    handler.hass = hass
+
     await handler._radio_mgr.detect_radio_type()
 
     assert handler._radio_mgr.radio_type == RadioType.ezsp
@@ -956,22 +966,10 @@ async def test_user_port_config(probe_mock, hass: HomeAssistant) -> None:
     ],
 )
 async def test_migration_ti_cc_to_znp(
-    old_type, new_type, hass: HomeAssistant, config_entry
+    old_type, new_type, hass: HomeAssistant, config_entry: MockConfigEntry
 ) -> None:
     """Test zigpy-cc to zigpy-znp config migration."""
-    config_entry = MockConfigEntry(
+    config_entry.data = {**config_entry.data, CONF_RADIO_TYPE: old_type}
-        domain=DOMAIN,
-        unique_id=old_type + new_type,
-        data={
-            CONF_RADIO_TYPE: old_type,
-            CONF_DEVICE: {
-                CONF_DEVICE_PATH: "/dev/ttyUSB1",
-                CONF_BAUDRATE: 115200,
-                CONF_FLOWCONTROL: None,
-            },
-        },
-    )
-
     config_entry.version = 2
     config_entry.add_to_hass(hass)
 
@@ -1919,3 +1917,44 @@ async def test_config_flow_port_multiprotocol_port_name(hass: HomeAssistant) ->
         result["data_schema"].schema["path"].container[0]
         == "socket://core-silabs-multiprotocol:9999 - Multiprotocol add-on - Nabu Casa"
     )
+
+
+@patch("serial.tools.list_ports.comports", MagicMock(return_value=[com_port()]))
+async def test_probe_wrong_firmware_installed(hass: HomeAssistant) -> None:
+    """Test auto-probing failing because the wrong firmware is installed."""
+
+    with patch(
+        "homeassistant.components.zha.radio_manager.ZhaRadioManager.detect_radio_type",
+        return_value=ProbeResult.WRONG_FIRMWARE_INSTALLED,
+    ):
+        result = await hass.config_entries.flow.async_init(
+            DOMAIN,
+            context={CONF_SOURCE: "choose_serial_port"},
+            data={
+                CONF_DEVICE_PATH: (
+                    "/dev/ttyUSB1234 - Some serial port, s/n: 1234 - Virtual serial port"
+                )
+            },
+        )
+
+    assert result["type"] == FlowResultType.ABORT
+    assert result["reason"] == "wrong_firmware_installed"
+
+
+async def test_discovery_wrong_firmware_installed(hass: HomeAssistant) -> None:
+    """Test auto-probing failing because the wrong firmware is installed."""
+
+    with patch(
+        "homeassistant.components.zha.radio_manager.ZhaRadioManager.detect_radio_type",
+        return_value=ProbeResult.WRONG_FIRMWARE_INSTALLED,
+    ), patch(
+        "homeassistant.components.onboarding.async_is_onboarded", return_value=False
+    ):
+        result = await hass.config_entries.flow.async_init(
+            DOMAIN,
+            context={CONF_SOURCE: "confirm"},
+            data={},
+        )
+
+    assert result["type"] == FlowResultType.ABORT
+    assert result["reason"] == "wrong_firmware_installed"

tests/components/zha/test_diagnostics.py

@@ -15,6 +15,7 @@ from homeassistant.helpers.device_registry import async_get
 
 from .conftest import SIG_EP_INPUT, SIG_EP_OUTPUT, SIG_EP_PROFILE, SIG_EP_TYPE
 
+from tests.common import MockConfigEntry
 from tests.components.diagnostics import (
     get_diagnostics_for_config_entry,
     get_diagnostics_for_device,
@@ -57,7 +58,7 @@ def zigpy_device(zigpy_device_mock):
 async def test_diagnostics_for_config_entry(
     hass: HomeAssistant,
     hass_client: ClientSessionGenerator,
-    config_entry,
+    config_entry: MockConfigEntry,
     zha_device_joined,
     zigpy_device,
 ) -> None:
@@ -86,12 +87,11 @@ async def test_diagnostics_for_config_entry(
 async def test_diagnostics_for_device(
     hass: HomeAssistant,
     hass_client: ClientSessionGenerator,
-    config_entry,
+    config_entry: MockConfigEntry,
     zha_device_joined,
     zigpy_device,
 ) -> None:
     """Test diagnostics for device."""
-
     zha_device: ZHADevice = await zha_device_joined(zigpy_device)
     dev_reg = async_get(hass)
     device = dev_reg.async_get_device(identifiers={("zha", str(zha_device.ieee))})

tests/components/zha/test_radio_manager.py

@@ -14,6 +14,7 @@ from homeassistant import config_entries
 from homeassistant.components.usb import UsbServiceInfo
 from homeassistant.components.zha import radio_manager
 from homeassistant.components.zha.core.const import DOMAIN, RadioType
+from homeassistant.components.zha.radio_manager import ProbeResult, ZhaRadioManager
 from homeassistant.core import HomeAssistant
 
 from tests.common import MockConfigEntry
@@ -59,10 +60,13 @@ def backup():
     return backup
 
 
-def mock_detect_radio_type(radio_type=RadioType.ezsp, ret=True):
+def mock_detect_radio_type(
+    radio_type: RadioType = RadioType.ezsp,
+    ret: ProbeResult = ProbeResult.RADIO_TYPE_DETECTED,
+):
     """Mock `detect_radio_type` that just sets the appropriate attributes."""
 
-    async def detect(self):
+    async def detect(self) -> ProbeResult:
         self.radio_type = radio_type
         self.device_settings = radio_type.controller.SCHEMA_DEVICE(
             {CONF_DEVICE_PATH: self.device_path}
@@ -421,3 +425,58 @@ async def test_migrate_initiate_failure(
         await migration_helper.async_initiate_migration(migration_data)
 
     assert len(mock_load_info.mock_calls) == radio_manager.BACKUP_RETRIES
+
+
+@pytest.fixture(name="radio_manager")
+def zha_radio_manager(hass: HomeAssistant) -> ZhaRadioManager:
+    """Fixture for an instance of `ZhaRadioManager`."""
+    radio_manager = ZhaRadioManager()
+    radio_manager.hass = hass
+    radio_manager.device_path = "/dev/ttyZigbee"
+    return radio_manager
+
+
+async def test_detect_radio_type_success(radio_manager: ZhaRadioManager) -> None:
+    """Test radio type detection, success."""
+    with patch(
+        "bellows.zigbee.application.ControllerApplication.probe", return_value=False
+    ), patch(
+        # Intentionally probe only the second radio type
+        "zigpy_znp.zigbee.application.ControllerApplication.probe",
+        return_value=True,
+    ):
+        assert (
+            await radio_manager.detect_radio_type() == ProbeResult.RADIO_TYPE_DETECTED
+        )
+        assert radio_manager.radio_type == RadioType.znp
+
+
+async def test_detect_radio_type_failure_wrong_firmware(
+    radio_manager: ZhaRadioManager,
+) -> None:
+    """Test radio type detection, wrong firmware."""
+    with patch(
+        "homeassistant.components.zha.radio_manager.AUTOPROBE_RADIOS", ()
+    ), patch(
+        "homeassistant.components.zha.radio_manager.repairs.warn_on_wrong_silabs_firmware",
+        return_value=True,
+    ):
+        assert (
+            await radio_manager.detect_radio_type()
+            == ProbeResult.WRONG_FIRMWARE_INSTALLED
+        )
+        assert radio_manager.radio_type is None
+
+
+async def test_detect_radio_type_failure_no_detect(
+    radio_manager: ZhaRadioManager,
+) -> None:
+    """Test radio type detection, no firmware detected."""
+    with patch(
+        "homeassistant.components.zha.radio_manager.AUTOPROBE_RADIOS", ()
+    ), patch(
+        "homeassistant.components.zha.radio_manager.repairs.warn_on_wrong_silabs_firmware",
+        return_value=False,
+    ):
+        assert await radio_manager.detect_radio_type() == ProbeResult.PROBING_FAILED
+        assert radio_manager.radio_type is None

tests/components/zha/test_repairs.py

@@ -0,0 +1,235 @@
+"""Test ZHA repairs."""
+from collections.abc import Callable
+import logging
+from unittest.mock import patch
+
+import pytest
+from universal_silabs_flasher.const import ApplicationType
+from universal_silabs_flasher.flasher import Flasher
+
+from homeassistant.components.homeassistant_sky_connect import (
+    DOMAIN as SKYCONNECT_DOMAIN,
+)
+from homeassistant.components.zha.core.const import DOMAIN
+from homeassistant.components.zha.repairs import (
+    DISABLE_MULTIPAN_URL,
+    ISSUE_WRONG_SILABS_FIRMWARE_INSTALLED,
+    HardwareType,
+    _detect_radio_hardware,
+    probe_silabs_firmware_type,
+    warn_on_wrong_silabs_firmware,
+)
+from homeassistant.config_entries import ConfigEntryState
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import HomeAssistantError
+from homeassistant.helpers import issue_registry as ir
+
+from tests.common import MockConfigEntry
+
+SKYCONNECT_DEVICE = "/dev/serial/by-id/usb-Nabu_Casa_SkyConnect_v1.0_9e2adbd75b8beb119fe564a0f320645d-if00-port0"
+
+
+def set_flasher_app_type(app_type: ApplicationType) -> Callable[[Flasher], None]:
+    """Set the app type on the flasher."""
+
+    def replacement(self: Flasher) -> None:
+        self.app_type = app_type
+
+    return replacement
+
+
+def test_detect_radio_hardware(hass: HomeAssistant) -> None:
+    """Test logic to detect radio hardware."""
+    skyconnect_config_entry = MockConfigEntry(
+        data={
+            "device": SKYCONNECT_DEVICE,
+            "vid": "10C4",
+            "pid": "EA60",
+            "serial_number": "3c0ed67c628beb11b1cd64a0f320645d",
+            "manufacturer": "Nabu Casa",
+            "description": "SkyConnect v1.0",
+        },
+        domain=SKYCONNECT_DOMAIN,
+        options={},
+        title="Home Assistant SkyConnect",
+    )
+    skyconnect_config_entry.add_to_hass(hass)
+
+    assert _detect_radio_hardware(hass, SKYCONNECT_DEVICE) == HardwareType.SKYCONNECT
+    assert (
+        _detect_radio_hardware(hass, SKYCONNECT_DEVICE + "_foo") == HardwareType.OTHER
+    )
+    assert _detect_radio_hardware(hass, "/dev/ttyAMA1") == HardwareType.OTHER
+
+    with patch(
+        "homeassistant.components.homeassistant_yellow.hardware.get_os_info",
+        return_value={"board": "yellow"},
+    ):
+        assert _detect_radio_hardware(hass, "/dev/ttyAMA1") == HardwareType.YELLOW
+        assert _detect_radio_hardware(hass, "/dev/ttyAMA2") == HardwareType.OTHER
+        assert (
+            _detect_radio_hardware(hass, SKYCONNECT_DEVICE) == HardwareType.SKYCONNECT
+        )
+
+
+def test_detect_radio_hardware_failure(hass: HomeAssistant) -> None:
+    """Test radio hardware detection failure."""
+
+    with patch(
+        "homeassistant.components.homeassistant_yellow.hardware.async_info",
+        side_effect=HomeAssistantError(),
+    ), patch(
+        "homeassistant.components.homeassistant_sky_connect.hardware.async_info",
+        side_effect=HomeAssistantError(),
+    ):
+        assert _detect_radio_hardware(hass, SKYCONNECT_DEVICE) == HardwareType.OTHER
+
+
+@pytest.mark.parametrize(
+    ("detected_hardware", "expected_learn_more_url"),
+    [
+        (HardwareType.SKYCONNECT, DISABLE_MULTIPAN_URL[HardwareType.SKYCONNECT]),
+        (HardwareType.YELLOW, DISABLE_MULTIPAN_URL[HardwareType.YELLOW]),
+        (HardwareType.OTHER, None),
+    ],
+)
+async def test_multipan_firmware_repair(
+    hass: HomeAssistant,
+    detected_hardware: HardwareType,
+    expected_learn_more_url: str,
+    config_entry: MockConfigEntry,
+    mock_zigpy_connect,
+) -> None:
+    """Test creating a repair when multi-PAN firmware is installed and probed."""
+
+    config_entry.add_to_hass(hass)
+
+    # ZHA fails to set up
+    with patch(
+        "homeassistant.components.zha.repairs.Flasher.probe_app_type",
+        side_effect=set_flasher_app_type(ApplicationType.CPC),
+        autospec=True,
+    ), patch(
+        "homeassistant.components.zha.core.gateway.ZHAGateway.async_initialize",
+        side_effect=RuntimeError(),
+    ), patch(
+        "homeassistant.components.zha.repairs._detect_radio_hardware",
+        return_value=detected_hardware,
+    ):
+        await hass.config_entries.async_setup(config_entry.entry_id)
+        await hass.async_block_till_done()
+
+        assert config_entry.state == ConfigEntryState.SETUP_ERROR
+
+    await hass.config_entries.async_unload(config_entry.entry_id)
+
+    issue_registry = ir.async_get(hass)
+
+    issue = issue_registry.async_get_issue(
+        domain=DOMAIN,
+        issue_id=ISSUE_WRONG_SILABS_FIRMWARE_INSTALLED,
+    )
+
+    # The issue is created when we fail to probe
+    assert issue is not None
+    assert issue.translation_placeholders["firmware_type"] == "CPC"
+    assert issue.learn_more_url == expected_learn_more_url
+
+    # If ZHA manages to start up normally after this, the issue will be deleted
+    with mock_zigpy_connect:
+        await hass.config_entries.async_setup(config_entry.entry_id)
+        await hass.async_block_till_done()
+
+    issue = issue_registry.async_get_issue(
+        domain=DOMAIN,
+        issue_id=ISSUE_WRONG_SILABS_FIRMWARE_INSTALLED,
+    )
+    assert issue is None
+
+
+async def test_multipan_firmware_no_repair_on_probe_failure(
+    hass: HomeAssistant, config_entry: MockConfigEntry
+) -> None:
+    """Test that a repair is not created when multi-PAN firmware cannot be probed."""
+
+    config_entry.add_to_hass(hass)
+
+    # ZHA fails to set up
+    with patch(
+        "homeassistant.components.zha.repairs.Flasher.probe_app_type",
+        side_effect=set_flasher_app_type(None),
+        autospec=True,
+    ), patch(
+        "homeassistant.components.zha.core.gateway.ZHAGateway.async_initialize",
+        side_effect=RuntimeError(),
+    ):
+        await hass.config_entries.async_setup(config_entry.entry_id)
+        await hass.async_block_till_done()
+
+        assert config_entry.state == ConfigEntryState.SETUP_ERROR
+
+    await hass.config_entries.async_unload(config_entry.entry_id)
+
+    # No repair is created
+    issue_registry = ir.async_get(hass)
+    issue = issue_registry.async_get_issue(
+        domain=DOMAIN,
+        issue_id=ISSUE_WRONG_SILABS_FIRMWARE_INSTALLED,
+    )
+    assert issue is None
+
+
+async def test_multipan_firmware_retry_on_probe_ezsp(
+    hass: HomeAssistant,
+    config_entry: MockConfigEntry,
+    mock_zigpy_connect,
+) -> None:
+    """Test that ZHA is reloaded when EZSP firmware is probed."""
+
+    config_entry.add_to_hass(hass)
+
+    # ZHA fails to set up
+    with patch(
+        "homeassistant.components.zha.repairs.Flasher.probe_app_type",
+        side_effect=set_flasher_app_type(ApplicationType.EZSP),
+        autospec=True,
+    ), patch(
+        "homeassistant.components.zha.core.gateway.ZHAGateway.async_initialize",
+        side_effect=RuntimeError(),
+    ):
+        await hass.config_entries.async_setup(config_entry.entry_id)
+        await hass.async_block_till_done()
+
+        # The config entry state is `SETUP_RETRY`, not `SETUP_ERROR`!
+        assert config_entry.state == ConfigEntryState.SETUP_RETRY
+
+    await hass.config_entries.async_unload(config_entry.entry_id)
+
+    # No repair is created
+    issue_registry = ir.async_get(hass)
+    issue = issue_registry.async_get_issue(
+        domain=DOMAIN,
+        issue_id=ISSUE_WRONG_SILABS_FIRMWARE_INSTALLED,
+    )
+    assert issue is None
+
+
+async def test_no_warn_on_socket(hass: HomeAssistant) -> None:
+    """Test that no warning is issued when the device is a socket."""
+    with patch(
+        "homeassistant.components.zha.repairs.probe_silabs_firmware_type", autospec=True
+    ) as mock_probe:
+        await warn_on_wrong_silabs_firmware(hass, device="socket://1.2.3.4:5678")
+
+    mock_probe.assert_not_called()
+
+
+async def test_probe_failure_exception_handling(caplog) -> None:
+    """Test that probe failures are handled gracefully."""
+    with patch(
+        "homeassistant.components.zha.repairs.Flasher.probe_app_type",
+        side_effect=RuntimeError(),
+    ), caplog.at_level(logging.DEBUG):
+        await probe_silabs_firmware_type("/dev/ttyZigbee")
+
+    assert "Failed to probe application type" in caplog.text