mirror of
https://github.com/home-assistant/core.git
synced 2025-06-25 01:21:51 +02:00
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>
This commit is contained in:
@ -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)
|
||||
await zha_gateway.async_initialize()
|
||||
|
||||
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)
|
||||
|
||||
|
@ -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 (
|
||||
self._radio_mgr.radio_type is None
|
||||
and not await self._radio_mgr.detect_radio_type()
|
||||
):
|
||||
if self._radio_mgr.radio_type is None:
|
||||
probe_result = 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
|
||||
|
@ -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": [
|
||||
{
|
||||
|
@ -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
|
||||
|
126
homeassistant/components/zha/repairs.py
Normal file
126
homeassistant/components/zha/repairs.py
Normal file
@ -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)
|
@ -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."
|
||||
}
|
||||
}
|
||||
}
|
||||
|
@ -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
|
||||
|
||||
|
@ -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
|
||||
|
||||
|
@ -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"),
|
||||
|
@ -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(
|
||||
version=2,
|
||||
return MockConfigEntry(
|
||||
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}}
|
||||
)
|
||||
|
@ -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)
|
||||
async def test_discovery_via_usb_no_radio(probe_mock, hass: HomeAssistant) -> None:
|
||||
@patch(
|
||||
"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(
|
||||
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.data = {**config_entry.data, CONF_RADIO_TYPE: old_type}
|
||||
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"
|
||||
|
@ -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))})
|
||||
|
@ -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
|
||||
|
235
tests/components/zha/test_repairs.py
Normal file
235
tests/components/zha/test_repairs.py
Normal file
@ -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
|
Reference in New Issue
Block a user