evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Open a ZHA repair when network settings change (#99482)

Browse source
This commit is contained in:
puddly 2023-10-09 09:01:05 -04:00 • committed by GitHub
parent b9090452de
commit 2e4df6d2f2
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
18 changed files with 617 additions and 152 deletions
Download patch file Download diff file Expand all files Collapse all files

25
homeassistant/components/zha/__init__.py
View file

@ -8,12 +8,13 @@ import re
import voluptuous as vol
from zhaquirks import setup as setup_quirks
from zigpy.config import CONF_DEVICE, CONF_DEVICE_PATH
from zigpy.config import CONF_DATABASE, CONF_DEVICE, CONF_DEVICE_PATH
from zigpy.exceptions import NetworkSettingsInconsistent
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_TYPE
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.exceptions import ConfigEntryNotReady, HomeAssistantError
from homeassistant.helpers import device_registry as dr
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.dispatcher import async_dispatcher_send
@ -26,7 +27,6 @@ from .core.const import (
BAUD_RATES,
CONF_BAUDRATE,
CONF_CUSTOM_QUIRKS_PATH,
CONF_DATABASE,
CONF_DEVICE_CONFIG,
CONF_ENABLE_QUIRKS,
CONF_RADIO_TYPE,
@ -42,6 +42,11 @@ from .core.device import get_device_automation_triggers
from .core.discovery import GROUP_PROBE
from .core.helpers import ZHAData, get_zha_data
from .radio_manager import ZhaRadioManager
from .repairs.network_settings_inconsistent import warn_on_inconsistent_network_settings
from .repairs.wrong_silabs_firmware import (
AlreadyRunningEZSP,
warn_on_wrong_silabs_firmware,
)
DEVICE_CONFIG_SCHEMA_ENTRY = vol.Schema({vol.Optional(CONF_TYPE): cv.string})
ZHA_CONFIG_SCHEMA = {
@ -170,13 +175,23 @@ async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> b
try:
await zha_gateway.async_initialize()
except NetworkSettingsInconsistent as exc:
await warn_on_inconsistent_network_settings(
hass,
config_entry=config_entry,
old_state=exc.old_state,
new_state=exc.new_state,
)
raise HomeAssistantError(
"Network settings do not match most recent backup"
) from exc
except Exception:
if RadioType[config_entry.data[CONF_RADIO_TYPE]] == RadioType.ezsp:
try:
await repairs.warn_on_wrong_silabs_firmware(
await warn_on_wrong_silabs_firmware(
hass, config_entry.data[CONF_DEVICE][CONF_DEVICE_PATH]
)
except repairs.AlreadyRunningEZSP as exc:
except 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

22
homeassistant/components/zha/api.py
View file

@ -10,8 +10,8 @@ from zigpy.types import Channels
from zigpy.util import pick_optimal_channel
from .core.const import CONF_RADIO_TYPE, DOMAIN, RadioType
from .core.gateway import ZHAGateway
from .core.helpers import get_zha_data, get_zha_gateway
from .core.helpers import get_zha_gateway
from .radio_manager import ZhaRadioManager
if TYPE_CHECKING:
from homeassistant.config_entries import ConfigEntry
@ -55,19 +55,13 @@ async def async_get_last_network_settings(
if config_entry is None:
config_entry = _get_config_entry(hass)
config = get_zha_data(hass).yaml_config
zha_gateway = ZHAGateway(hass, config, config_entry)
radio_mgr = ZhaRadioManager.from_config_entry(hass, config_entry)
app_controller_cls, app_config = zha_gateway.get_application_controller_data()
app = app_controller_cls(app_config)
try:
await app._load_db() # pylint: disable=protected-access
settings = max(app.backups, key=lambda b: b.backup_time)
except ValueError:
settings = None
finally:
await app.shutdown()
async with radio_mgr.connect_zigpy_app() as app:
try:
settings = max(app.backups, key=lambda b: b.backup_time)
except ValueError:
settings = None
return settings

4
homeassistant/components/zha/core/const.py
View file

@ -7,7 +7,6 @@ import logging
import bellows.zigbee.application
import voluptuous as vol
import zigpy.application
from zigpy.config import CONF_DEVICE_PATH # noqa: F401
import zigpy.types as t
import zigpy_deconz.zigbee.application
import zigpy_xbee.zigbee.application
@ -128,7 +127,6 @@ CONF_ALARM_ARM_REQUIRES_CODE = "alarm_arm_requires_code"
CONF_BAUDRATE = "baudrate"
CONF_CUSTOM_QUIRKS_PATH = "custom_quirks_path"
CONF_DATABASE = "database_path"
CONF_DEFAULT_LIGHT_TRANSITION = "default_light_transition"
CONF_DEVICE_CONFIG = "device_config"
CONF_ENABLE_ENHANCED_LIGHT_TRANSITION = "enhanced_light_transition"
@ -138,8 +136,6 @@ CONF_GROUP_MEMBERS_ASSUME_STATE = "group_members_assume_state"
CONF_ENABLE_IDENTIFY_ON_JOIN = "enable_identify_on_join"
CONF_ENABLE_QUIRKS = "enable_quirks"
CONF_FLOWCONTROL = "flow_control"
CONF_NWK = "network"
CONF_NWK_CHANNEL = "channel"
CONF_RADIO_TYPE = "radio_type"
CONF_USB_PATH = "usb_path"
CONF_USE_THREAD = "use_thread"

34
homeassistant/components/zha/core/gateway.py
View file

@ -4,6 +4,7 @@ from __future__ import annotations
import asyncio
import collections
from collections.abc import Callable
from contextlib import suppress
from datetime import timedelta
from enum import Enum
import itertools
@ -13,10 +14,17 @@ import time
from typing import TYPE_CHECKING, Any, NamedTuple
from zigpy.application import ControllerApplication
from zigpy.config import CONF_DEVICE
from zigpy.config import (
CONF_DATABASE,
CONF_DEVICE,
CONF_DEVICE_PATH,
CONF_NWK,
CONF_NWK_CHANNEL,
CONF_NWK_VALIDATE_SETTINGS,
)
import zigpy.device
import zigpy.endpoint
import zigpy.exceptions
from zigpy.exceptions import NetworkSettingsInconsistent, TransientConnectionError
import zigpy.group
from zigpy.types.named import EUI64
@ -38,10 +46,6 @@ from .const import (
ATTR_NWK,
ATTR_SIGNATURE,
ATTR_TYPE,
CONF_DATABASE,
CONF_DEVICE_PATH,
CONF_NWK,
CONF_NWK_CHANNEL,
CONF_RADIO_TYPE,
CONF_USE_THREAD,
CONF_ZIGPY,
@ -159,6 +163,9 @@ class ZHAGateway:
app_config[CONF_DATABASE] = database
app_config[CONF_DEVICE] = self.config_entry.data[CONF_DEVICE]
if CONF_NWK_VALIDATE_SETTINGS not in app_config:
app_config[CONF_NWK_VALIDATE_SETTINGS] = True
# The bellows UART thread sometimes propagates a cancellation into the main Core
# event loop, when a connection to a TCP coordinator fails in a specific way
if (
@ -199,7 +206,9 @@ class ZHAGateway:
for attempt in range(STARTUP_RETRIES):
try:
await self.application_controller.startup(auto_form=True)
except zigpy.exceptions.TransientConnectionError as exc:
except NetworkSettingsInconsistent:
raise
except TransientConnectionError as exc:
raise ConfigEntryNotReady from exc
except Exception as exc: # pylint: disable=broad-except
_LOGGER.warning(
@ -231,12 +240,13 @@ class ZHAGateway:
self.application_controller.groups.add_listener(self)
def _find_coordinator_device(self) -> zigpy.device.Device:
zigpy_coordinator = self.application_controller.get_device(nwk=0x0000)
if last_backup := self.application_controller.backups.most_recent_backup():
zigpy_coordinator = self.application_controller.get_device(
ieee=last_backup.node_info.ieee
)
else:
zigpy_coordinator = self.application_controller.get_device(nwk=0x0000)
with suppress(KeyError):
zigpy_coordinator = self.application_controller.get_device(
ieee=last_backup.node_info.ieee
)
return zigpy_coordinator

14
homeassistant/components/zha/radio_manager.py
View file

@ -14,7 +14,12 @@ from bellows.config import CONF_USE_THREAD
import voluptuous as vol
from zigpy.application import ControllerApplication
import zigpy.backups
from zigpy.config import CONF_DEVICE, CONF_DEVICE_PATH, CONF_NWK_BACKUP_ENABLED
from zigpy.config import (
CONF_DATABASE,
CONF_DEVICE,
CONF_DEVICE_PATH,
CONF_NWK_BACKUP_ENABLED,
)
from zigpy.exceptions import NetworkNotFormed
from homeassistant import config_entries
@ -23,7 +28,6 @@ from homeassistant.core import HomeAssistant
from . import repairs
from .core.const import (
CONF_DATABASE,
CONF_RADIO_TYPE,
CONF_ZIGPY,
DEFAULT_DATABASE_NAME,
@ -218,8 +222,10 @@ class ZhaRadioManager:
repairs.async_delete_blocking_issues(self.hass)
return ProbeResult.RADIO_TYPE_DETECTED
with suppress(repairs.AlreadyRunningEZSP):
if await repairs.warn_on_wrong_silabs_firmware(self.hass, self.device_path):
with suppress(repairs.wrong_silabs_firmware.AlreadyRunningEZSP):
if await repairs.wrong_silabs_firmware.warn_on_wrong_silabs_firmware(
self.hass, self.device_path
):
return ProbeResult.WRONG_FIRMWARE_INSTALLED
return ProbeResult.PROBING_FAILED

33
homeassistant/components/zha/repairs/__init__.py Normal file
View file

@ -0,0 +1,33 @@
"""ZHA repairs for common environmental and device problems."""
from __future__ import annotations
from typing import Any, cast
from homeassistant.components.repairs import ConfirmRepairFlow, RepairsFlow
from homeassistant.core import HomeAssistant
from homeassistant.helpers import issue_registry as ir
from ..core.const import DOMAIN
from .network_settings_inconsistent import (
ISSUE_INCONSISTENT_NETWORK_SETTINGS,
NetworkSettingsInconsistentFlow,
)
from .wrong_silabs_firmware import ISSUE_WRONG_SILABS_FIRMWARE_INSTALLED
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)
ir.async_delete_issue(hass, DOMAIN, ISSUE_INCONSISTENT_NETWORK_SETTINGS)
async def async_create_fix_flow(
hass: HomeAssistant,
issue_id: str,
data: dict[str, str | int | float | None] | None,
) -> RepairsFlow:
"""Create flow."""
if issue_id == ISSUE_INCONSISTENT_NETWORK_SETTINGS:
return NetworkSettingsInconsistentFlow(hass, cast(dict[str, Any], data))
return ConfirmRepairFlow()

151
homeassistant/components/zha/repairs/network_settings_inconsistent.py Normal file
View file

@ -0,0 +1,151 @@
"""ZHA repair for inconsistent network settings."""
from __future__ import annotations
import logging
from typing import Any
from zigpy.backups import NetworkBackup
from homeassistant.components.repairs import RepairsFlow
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResult
from homeassistant.helpers import issue_registry as ir
from ..core.const import DOMAIN
from ..radio_manager import ZhaRadioManager
_LOGGER = logging.getLogger(__name__)
ISSUE_INCONSISTENT_NETWORK_SETTINGS = "inconsistent_network_settings"
def _format_settings_diff(old_state: NetworkBackup, new_state: NetworkBackup) -> str:
"""Format the difference between two network backups."""
lines: list[str] = []
def _add_difference(
lines: list[str], text: str, old: Any, new: Any, pre: bool = True
) -> None:
"""Add a line to the list if the values are different."""
wrap = "`" if pre else ""
if old != new:
lines.append(f"{text}: {wrap}{old}{wrap} \u2192 {wrap}{new}{wrap}")
_add_difference(
lines,
"Channel",
old=old_state.network_info.channel,
new=new_state.network_info.channel,
pre=False,
)
_add_difference(
lines,
"Node IEEE",
old=old_state.node_info.ieee,
new=new_state.node_info.ieee,
)
_add_difference(
lines,
"PAN ID",
old=old_state.network_info.pan_id,
new=new_state.network_info.pan_id,
)
_add_difference(
lines,
"Extended PAN ID",
old=old_state.network_info.extended_pan_id,
new=new_state.network_info.extended_pan_id,
)
_add_difference(
lines,
"NWK update ID",
old=old_state.network_info.nwk_update_id,
new=new_state.network_info.nwk_update_id,
pre=False,
)
_add_difference(
lines,
"TC Link Key",
old=old_state.network_info.tc_link_key.key,
new=new_state.network_info.tc_link_key.key,
)
_add_difference(
lines,
"Network Key",
old=old_state.network_info.network_key.key,
new=new_state.network_info.network_key.key,
)
return "\n".join([f"- {line}" for line in lines])
async def warn_on_inconsistent_network_settings(
hass: HomeAssistant,
config_entry: ConfigEntry,
old_state: NetworkBackup,
new_state: NetworkBackup,
) -> None:
"""Create a repair if the network settings are inconsistent with the last backup."""
ir.async_create_issue(
hass,
domain=DOMAIN,
issue_id=ISSUE_INCONSISTENT_NETWORK_SETTINGS,
is_fixable=True,
severity=ir.IssueSeverity.ERROR,
translation_key=ISSUE_INCONSISTENT_NETWORK_SETTINGS,
data={
"config_entry_id": config_entry.entry_id,
"old_state": old_state.as_dict(),
"new_state": new_state.as_dict(),
},
)
class NetworkSettingsInconsistentFlow(RepairsFlow):
"""Handler for an issue fixing flow."""
def __init__(self, hass: HomeAssistant, data: dict[str, Any]) -> None:
"""Initialize the flow."""
self.hass = hass
self._old_state = NetworkBackup.from_dict(data["old_state"])
self._new_state = NetworkBackup.from_dict(data["new_state"])
self._entry_id: str = data["config_entry_id"]
config_entry = self.hass.config_entries.async_get_entry(self._entry_id)
assert config_entry is not None
self._radio_mgr = ZhaRadioManager.from_config_entry(self.hass, config_entry)
async def async_step_init(
self, user_input: dict[str, str] | None = None
) -> FlowResult:
"""Handle the first step of a fix flow."""
return self.async_show_menu(
step_id="init",
menu_options=["restore_old_settings", "use_new_settings"],
description_placeholders={
"diff": _format_settings_diff(self._old_state, self._new_state)
},
)
async def async_step_use_new_settings(
self, user_input: dict[str, str] | None = None
) -> FlowResult:
"""Step to use the new settings found on the radio."""
async with self._radio_mgr.connect_zigpy_app() as app:
app.backups.add_backup(self._new_state)
await self.hass.config_entries.async_reload(self._entry_id)
return self.async_create_entry(title="", data={})
async def async_step_restore_old_settings(
self, user_input: dict[str, str] | None = None
) -> FlowResult:
"""Step to restore the most recent backup."""
await self._radio_mgr.restore_backup(self._old_state)
await self.hass.config_entries.async_reload(self._entry_id)
return self.async_create_entry(title="", data={})

7
homeassistant/components/zha/repairs.py → homeassistant/components/zha/repairs/wrong_silabs_firmware.py
View file

@ -18,7 +18,7 @@ from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import issue_registry as ir
from .core.const import DOMAIN
from ..core.const import DOMAIN
_LOGGER = logging.getLogger(__name__)
@ -119,8 +119,3 @@ async def warn_on_wrong_silabs_firmware(hass: HomeAssistant, device: str) -> boo
)
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)

15
homeassistant/components/zha/strings.json
View file

@ -513,6 +513,21 @@
"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."
},
"inconsistent_network_settings": {
"title": "Zigbee network settings have changed",
"fix_flow": {
"step": {
"init": {
"title": "[%key:component::zha::issues::inconsistent_network_settings::title%]",
"description": "Your Zigbee radio's network settings are inconsistent with the most recent network backup. This usually happens if another Zigbee integration (e.g. Zigbee2MQTT or deCONZ) has overwritten them.\n\n{diff}\n\nIf you did not intentionally change your network settings, restore from the most recent backup: your devices will not work otherwise.",
"menu_options": {
"use_new_settings": "Keep the new settings",
"restore_old_settings": "Restore backup (recommended)"
}
}
}
}
}
}
}

144
tests/components/zha/conftest.py
View file

@ -2,7 +2,9 @@
from collections.abc import Callable, Generator
import itertools
import time
from unittest.mock import AsyncMock, MagicMock, patch
from typing import Any
from unittest.mock import AsyncMock, MagicMock, create_autospec, patch
import warnings
import pytest
import zigpy
@ -14,6 +16,7 @@ import zigpy.device
import zigpy.group
import zigpy.profiles
import zigpy.quirks
import zigpy.state
import zigpy.types
import zigpy.util
from zigpy.zcl.clusters.general import Basic, Groups
@ -92,7 +95,9 @@ class _FakeApp(ControllerApplication):
async def start_network(self):
pass
async def write_network_info(self):
async def write_network_info(
self, *, network_info: zigpy.state.NetworkInfo, node_info: zigpy.state.NodeInfo
) -> None:
pass
async def request(
@ -111,9 +116,33 @@ class _FakeApp(ControllerApplication):
):
pass
async def move_network_to_channel(
self, new_channel: int, *, num_broadcasts: int = 5
) -> None:
pass
def _wrap_mock_instance(obj: Any) -> MagicMock:
"""Auto-mock every attribute and method in an object."""
mock = create_autospec(obj, spec_set=True, instance=True)
for attr_name in dir(obj):
if attr_name.startswith("__") and attr_name not in {"__getitem__"}:
continue
real_attr = getattr(obj, attr_name)
mock_attr = getattr(mock, attr_name)
if callable(real_attr):
mock_attr.side_effect = real_attr
else:
setattr(mock, attr_name, real_attr)
return mock
@pytest.fixture
def zigpy_app_controller():
async def zigpy_app_controller():
"""Zigpy ApplicationController fixture."""
app = _FakeApp(
{
@ -145,14 +174,14 @@ def zigpy_app_controller():
ep.add_input_cluster(Basic.cluster_id)
ep.add_input_cluster(Groups.cluster_id)
with patch(
"zigpy.device.Device.request", return_value=[Status.SUCCESS]
), patch.object(app, "permit", autospec=True), patch.object(
app, "startup", wraps=app.startup
), patch.object(
app, "permit_with_key", autospec=True
):
yield app
with patch("zigpy.device.Device.request", return_value=[Status.SUCCESS]):
# The mock wrapping accesses deprecated attributes, so we suppress the warnings
with warnings.catch_warnings():
warnings.simplefilter("ignore", DeprecationWarning)
mock_app = _wrap_mock_instance(app)
mock_app.backups = _wrap_mock_instance(app.backups)
yield mock_app
@pytest.fixture(name="config_entry")
@ -189,12 +218,17 @@ def mock_zigpy_connect(
with patch(
"bellows.zigbee.application.ControllerApplication.new",
return_value=zigpy_app_controller,
) as mock_app:
yield mock_app
), patch(
"bellows.zigbee.application.ControllerApplication",
return_value=zigpy_app_controller,
):
yield zigpy_app_controller
@pytest.fixture
def setup_zha(hass, config_entry: MockConfigEntry, mock_zigpy_connect):
def setup_zha(
hass, config_entry: MockConfigEntry, mock_zigpy_connect: ControllerApplication
):
"""Set up ZHA component."""
zha_config = {zha_const.CONF_ENABLE_QUIRKS: False}
@ -202,12 +236,11 @@ def setup_zha(hass, config_entry: MockConfigEntry, mock_zigpy_connect):
config_entry.add_to_hass(hass)
config = config or {}
with mock_zigpy_connect:
status = await async_setup_component(
hass, zha_const.DOMAIN, {zha_const.DOMAIN: {**zha_config, **config}}
)
assert status is True
await hass.async_block_till_done()
status = await async_setup_component(
hass, zha_const.DOMAIN, {zha_const.DOMAIN: {**zha_config, **config}}
)
assert status is True
await hass.async_block_till_done()
return _setup
@ -394,3 +427,74 @@ def speed_up_radio_mgr():
"""Speed up the radio manager connection time by removing delays."""
with patch("homeassistant.components.zha.radio_manager.CONNECT_DELAY_S", 0.00001):
yield
@pytest.fixture
def network_backup() -> zigpy.backups.NetworkBackup:
"""Real ZHA network backup taken from an active instance."""
return zigpy.backups.NetworkBackup.from_dict(
{
"backup_time": "2022-11-16T03:16:49.427675+00:00",
"network_info": {
"extended_pan_id": "2f:73:58:bd:fe:78:91:11",
"pan_id": "2DB4",
"nwk_update_id": 0,
"nwk_manager_id": "0000",
"channel": 15,
"channel_mask": [
11,
12,
13,
14,
15,
16,
17,
18,
19,
20,
21,
22,
23,
24,
25,
26,
],
"security_level": 5,
"network_key": {
"key": "4a:c7:9d:50:51:09:16:37:2e:34:66:c6:ed:9b:23:85",
"tx_counter": 14131,
"rx_counter": 0,
"seq": 0,
"partner_ieee": "ff:ff:ff:ff:ff:ff:ff:ff",
},
"tc_link_key": {
"key": "5a:69:67:42:65:65:41:6c:6c:69:61:6e:63:65:30:39",
"tx_counter": 0,
"rx_counter": 0,
"seq": 0,
"partner_ieee": "84:ba:20:ff:fe:59:f5:ff",
},
"key_table": [],
"children": [],
"nwk_addresses": {"cc:cc:cc:ff:fe:e6:8e:ca": "1431"},
"stack_specific": {
"ezsp": {"hashed_tclk": "e9bd3ac165233d95923613c608beb147"}
},
"metadata": {
"ezsp": {
"manufacturer": "",
"board": "",
"version": "7.1.3.0 build 0",
"stack_version": 9,
"can_write_custom_eui64": False,
}
},
"source": "bellows@0.34.2",
},
"node_info": {
"nwk": "0000",
"ieee": "84:ba:20:ff:fe:59:f5:ff",
"logical_type": "coordinator",
},
}
)

53
tests/components/zha/test_api.py
View file

@ -2,7 +2,7 @@
from __future__ import annotations
from typing import TYPE_CHECKING
from unittest.mock import call, patch
from unittest.mock import AsyncMock, MagicMock, call, patch
import pytest
import zigpy.backups
@ -48,21 +48,18 @@ async def test_async_get_network_settings_inactive(
backup.network_info.channel = 20
zigpy_app_controller.backups.backups.append(backup)
with patch(
"bellows.zigbee.application.ControllerApplication.__new__",
return_value=zigpy_app_controller,
), patch.object(
zigpy_app_controller, "_load_db", wraps=zigpy_app_controller._load_db
) as mock_load_db, patch.object(
zigpy_app_controller,
"start_network",
wraps=zigpy_app_controller.start_network,
) as mock_start_network:
controller = AsyncMock()
controller.SCHEMA = zigpy_app_controller.SCHEMA
controller.new = AsyncMock(return_value=zigpy_app_controller)
with patch.dict(
"homeassistant.components.zha.core.const.RadioType._member_map_",
ezsp=MagicMock(controller=controller, description="EZSP"),
):
settings = await api.async_get_network_settings(hass)
assert len(mock_load_db.mock_calls) == 1
assert len(mock_start_network.mock_calls) == 0
assert settings.network_info.channel == 20
assert len(zigpy_app_controller.start_network.mock_calls) == 0
async def test_async_get_network_settings_missing(
@ -78,11 +75,7 @@ async def test_async_get_network_settings_missing(
zigpy_app_controller.state.network_info = zigpy.state.NetworkInfo()
zigpy_app_controller.state.node_info = zigpy.state.NodeInfo()
with patch(
"bellows.zigbee.application.ControllerApplication.__new__",
return_value=zigpy_app_controller,
):
settings = await api.async_get_network_settings(hass)
settings = await api.async_get_network_settings(hass)
assert settings is None
@ -115,12 +108,8 @@ async def test_change_channel(
"""Test changing the channel."""
await setup_zha()
with patch.object(
zigpy_app_controller, "move_network_to_channel", autospec=True
) as mock_move_network_to_channel:
await api.async_change_channel(hass, 20)
assert mock_move_network_to_channel.mock_calls == [call(20)]
await api.async_change_channel(hass, 20)
assert zigpy_app_controller.move_network_to_channel.mock_calls == [call(20)]
async def test_change_channel_auto(
@ -129,16 +118,10 @@ async def test_change_channel_auto(
"""Test changing the channel automatically using an energy scan."""
await setup_zha()
with patch.object(
zigpy_app_controller, "move_network_to_channel", autospec=True
) as mock_move_network_to_channel, patch.object(
zigpy_app_controller,
"energy_scan",
autospec=True,
return_value={c: c for c in range(11, 26 + 1)},
), patch.object(
api, "pick_optimal_channel", autospec=True, return_value=25
):
zigpy_app_controller.energy_scan.side_effect = None
zigpy_app_controller.energy_scan.return_value = {c: c for c in range(11, 26 + 1)}
with patch.object(api, "pick_optimal_channel", autospec=True, return_value=25):
await api.async_change_channel(hass, "auto")
assert mock_move_network_to_channel.mock_calls == [call(25)]
assert zigpy_app_controller.move_network_to_channel.mock_calls == [call(25)]

19
tests/components/zha/test_backup.py
View file

@ -1,17 +1,24 @@
"""Unit tests for ZHA backup platform."""
from unittest.mock import AsyncMock, patch
from unittest.mock import AsyncMock
from zigpy.application import ControllerApplication
from homeassistant.components.zha.backup import async_post_backup, async_pre_backup
from homeassistant.core import HomeAssistant
async def test_pre_backup(hass: HomeAssistant, setup_zha) -> None:
async def test_pre_backup(
hass: HomeAssistant, zigpy_app_controller: ControllerApplication, setup_zha
) -> None:
"""Test backup creation when `async_pre_backup` is called."""
with patch("zigpy.backups.BackupManager.create_backup", AsyncMock()) as backup_mock:
await setup_zha()
await async_pre_backup(hass)
await setup_zha()
backup_mock.assert_called_once_with(load_devices=True)
zigpy_app_controller.backups.create_backup = AsyncMock()
await async_pre_backup(hass)
zigpy_app_controller.backups.create_backup.assert_called_once_with(
load_devices=True
)
async def test_post_backup(hass: HomeAssistant, setup_zha) -> None:

5
tests/components/zha/test_device_trigger.py
View file

@ -4,6 +4,7 @@ import time
from unittest.mock import patch
import pytest
from zigpy.application import ControllerApplication
import zigpy.profiles.zha
import zigpy.zcl.clusters.general as general
@ -408,7 +409,7 @@ async def test_validate_trigger_config_missing_info(
hass: HomeAssistant,
config_entry: MockConfigEntry,
zigpy_device_mock,
mock_zigpy_connect,
mock_zigpy_connect: ControllerApplication,
zha_device_joined,
caplog: pytest.LogCaptureFixture,
) -> None:
@ -461,7 +462,7 @@ async def test_validate_trigger_config_unloaded_bad_info(
hass: HomeAssistant,
config_entry: MockConfigEntry,
zigpy_device_mock,
mock_zigpy_connect,
mock_zigpy_connect: ControllerApplication,
zha_device_joined,
caplog: pytest.LogCaptureFixture,
) -> None:

11
tests/components/zha/test_gateway.py
View file

@ -9,6 +9,7 @@ import zigpy.profiles.zha as zha
import zigpy.zcl.clusters.general as general
import zigpy.zcl.clusters.lighting as lighting
from homeassistant.components.zha.core.const import RadioType
from homeassistant.components.zha.core.device import ZHADevice
from homeassistant.components.zha.core.group import GroupMember
from homeassistant.components.zha.core.helpers import get_zha_gateway
@ -350,13 +351,11 @@ async def test_gateway_initialize_bellows_thread(
zha_gateway.config_entry.data["device"]["path"] = device_path
zha_gateway._config.setdefault("zigpy_config", {}).update(config_override)
with patch(
"bellows.zigbee.application.ControllerApplication.new",
return_value=zigpy_app_controller,
) as mock_new:
await zha_gateway.async_initialize()
await zha_gateway.async_initialize()
assert mock_new.mock_calls[0].kwargs["config"]["use_thread"] is thread_state
RadioType.ezsp.controller.new.mock_calls[-1].kwargs["config"][
"use_thread"
] is thread_state
@pytest.mark.parametrize(

10
tests/components/zha/test_init.py
View file

@ -3,6 +3,7 @@ import asyncio
from unittest.mock import AsyncMock, Mock, patch
import pytest
from zigpy.application import ControllerApplication
from zigpy.config import CONF_DEVICE, CONF_DEVICE_PATH
from zigpy.exceptions import TransientConnectionError
@ -136,7 +137,10 @@ async def test_config_depreciation(hass: HomeAssistant, zha_config) -> None:
"homeassistant.components.zha.websocket_api.async_load_api", Mock(return_value=True)
)
async def test_setup_with_v3_cleaning_uri(
hass: HomeAssistant, path: str, cleaned_path: str, mock_zigpy_connect
hass: HomeAssistant,
path: str,
cleaned_path: str,
mock_zigpy_connect: ControllerApplication,
) -> None:
"""Test migration of config entry from v3, applying corrections to the port path."""
config_entry_v3 = MockConfigEntry(
@ -166,7 +170,7 @@ async def test_zha_retry_unique_ids(
hass: HomeAssistant,
config_entry: MockConfigEntry,
zigpy_device_mock,
mock_zigpy_connect,
mock_zigpy_connect: ControllerApplication,
caplog,
) -> None:
"""Test that ZHA retrying creates unique entity IDs."""
@ -174,7 +178,7 @@ async def test_zha_retry_unique_ids(
config_entry.add_to_hass(hass)
# Ensure we have some device to try to load
app = mock_zigpy_connect.return_value
app = mock_zigpy_connect
light = zigpy_device_mock(LIGHT_ON_OFF)
app.devices[light.ieee] = light

4
tests/components/zha/test_radio_manager.py
View file

@ -456,7 +456,7 @@ async def test_detect_radio_type_failure_wrong_firmware(
with patch(
"homeassistant.components.zha.radio_manager.AUTOPROBE_RADIOS", ()
), patch(
"homeassistant.components.zha.radio_manager.repairs.warn_on_wrong_silabs_firmware",
"homeassistant.components.zha.radio_manager.repairs.wrong_silabs_firmware.warn_on_wrong_silabs_firmware",
return_value=True,
):
assert (
@ -473,7 +473,7 @@ async def test_detect_radio_type_failure_no_detect(
with patch(
"homeassistant.components.zha.radio_manager.AUTOPROBE_RADIOS", ()
), patch(
"homeassistant.components.zha.radio_manager.repairs.warn_on_wrong_silabs_firmware",
"homeassistant.components.zha.radio_manager.repairs.wrong_silabs_firmware.warn_on_wrong_silabs_firmware",
return_value=False,
):
assert await radio_manager.detect_radio_type() == ProbeResult.PROBING_FAILED

190
tests/components/zha/test_repairs.py
View file

@ -1,17 +1,25 @@
"""Test ZHA repairs."""
from collections.abc import Callable
from http import HTTPStatus
import logging
from unittest.mock import patch
from unittest.mock import Mock, call, patch
import pytest
from universal_silabs_flasher.const import ApplicationType
from universal_silabs_flasher.flasher import Flasher
from zigpy.application import ControllerApplication
import zigpy.backups
from zigpy.exceptions import NetworkSettingsInconsistent
from homeassistant.components.homeassistant_sky_connect import (
DOMAIN as SKYCONNECT_DOMAIN,
)
from homeassistant.components.repairs import DOMAIN as REPAIRS_DOMAIN
from homeassistant.components.zha.core.const import DOMAIN
from homeassistant.components.zha.repairs import (
from homeassistant.components.zha.repairs.network_settings_inconsistent import (
ISSUE_INCONSISTENT_NETWORK_SETTINGS,
)
from homeassistant.components.zha.repairs.wrong_silabs_firmware import (
DISABLE_MULTIPAN_URL,
ISSUE_WRONG_SILABS_FIRMWARE_INSTALLED,
HardwareType,
@ -23,8 +31,10 @@ 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 homeassistant.setup import async_setup_component
from tests.common import MockConfigEntry
from tests.typing import ClientSessionGenerator
SKYCONNECT_DEVICE = "/dev/serial/by-id/usb-Nabu_Casa_SkyConnect_v1.0_9e2adbd75b8beb119fe564a0f320645d-if00-port0"
@ -98,7 +108,7 @@ async def test_multipan_firmware_repair(
detected_hardware: HardwareType,
expected_learn_more_url: str,
config_entry: MockConfigEntry,
mock_zigpy_connect,
mock_zigpy_connect: ControllerApplication,
) -> None:
"""Test creating a repair when multi-PAN firmware is installed and probed."""
@ -106,14 +116,14 @@ async def test_multipan_firmware_repair(
# ZHA fails to set up
with patch(
"homeassistant.components.zha.repairs.Flasher.probe_app_type",
"homeassistant.components.zha.repairs.wrong_silabs_firmware.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",
"homeassistant.components.zha.repairs.wrong_silabs_firmware._detect_radio_hardware",
return_value=detected_hardware,
):
await hass.config_entries.async_setup(config_entry.entry_id)
@ -136,9 +146,8 @@ async def test_multipan_firmware_repair(
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()
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
issue = issue_registry.async_get_issue(
domain=DOMAIN,
@ -156,7 +165,7 @@ async def test_multipan_firmware_no_repair_on_probe_failure(
# ZHA fails to set up
with patch(
"homeassistant.components.zha.repairs.Flasher.probe_app_type",
"homeassistant.components.zha.repairs.wrong_silabs_firmware.Flasher.probe_app_type",
side_effect=set_flasher_app_type(None),
autospec=True,
), patch(
@ -182,7 +191,7 @@ async def test_multipan_firmware_no_repair_on_probe_failure(
async def test_multipan_firmware_retry_on_probe_ezsp(
hass: HomeAssistant,
config_entry: MockConfigEntry,
mock_zigpy_connect,
mock_zigpy_connect: ControllerApplication,
) -> None:
"""Test that ZHA is reloaded when EZSP firmware is probed."""
@ -190,7 +199,7 @@ async def test_multipan_firmware_retry_on_probe_ezsp(
# ZHA fails to set up
with patch(
"homeassistant.components.zha.repairs.Flasher.probe_app_type",
"homeassistant.components.zha.repairs.wrong_silabs_firmware.Flasher.probe_app_type",
side_effect=set_flasher_app_type(ApplicationType.EZSP),
autospec=True,
), patch(
@ -217,7 +226,8 @@ async def test_multipan_firmware_retry_on_probe_ezsp(
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
"homeassistant.components.zha.repairs.wrong_silabs_firmware.probe_silabs_firmware_type",
autospec=True,
) as mock_probe:
await warn_on_wrong_silabs_firmware(hass, device="socket://1.2.3.4:5678")
@ -227,9 +237,163 @@ async def test_no_warn_on_socket(hass: HomeAssistant) -> None:
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",
"homeassistant.components.zha.repairs.wrong_silabs_firmware.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
async def test_inconsistent_settings_keep_new(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
config_entry: MockConfigEntry,
mock_zigpy_connect: ControllerApplication,
network_backup: zigpy.backups.NetworkBackup,
) -> None:
"""Test inconsistent ZHA network settings: keep new settings."""
assert await async_setup_component(hass, REPAIRS_DOMAIN, {REPAIRS_DOMAIN: {}})
config_entry.add_to_hass(hass)
new_state = network_backup.replace(
network_info=network_backup.network_info.replace(pan_id=0xBBBB)
)
old_state = network_backup
with patch(
"homeassistant.components.zha.core.gateway.ZHAGateway.async_initialize",
side_effect=NetworkSettingsInconsistent(
message="Network settings are inconsistent",
new_state=new_state,
old_state=old_state,
),
):
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_INCONSISTENT_NETWORK_SETTINGS,
)
# The issue is created
assert issue is not None
client = await hass_client()
resp = await client.post(
"/api/repairs/issues/fix",
json={"handler": DOMAIN, "issue_id": issue.issue_id},
)
assert resp.status == HTTPStatus.OK
data = await resp.json()
flow_id = data["flow_id"]
assert data["description_placeholders"]["diff"] == "- PAN ID: `0x2DB4` → `0xBBBB`"
mock_zigpy_connect.backups.add_backup = Mock()
resp = await client.post(
f"/api/repairs/issues/fix/{flow_id}",
json={"next_step_id": "use_new_settings"},
)
await hass.async_block_till_done()
assert resp.status == HTTPStatus.OK
data = await resp.json()
assert data["type"] == "create_entry"
assert (
issue_registry.async_get_issue(
domain=DOMAIN,
issue_id=ISSUE_INCONSISTENT_NETWORK_SETTINGS,
)
is None
)
assert mock_zigpy_connect.backups.add_backup.mock_calls == [call(new_state)]
async def test_inconsistent_settings_restore_old(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
config_entry: MockConfigEntry,
mock_zigpy_connect: ControllerApplication,
network_backup: zigpy.backups.NetworkBackup,
) -> None:
"""Test inconsistent ZHA network settings: restore last backup."""
assert await async_setup_component(hass, REPAIRS_DOMAIN, {REPAIRS_DOMAIN: {}})
config_entry.add_to_hass(hass)
new_state = network_backup.replace(
network_info=network_backup.network_info.replace(pan_id=0xBBBB)
)
old_state = network_backup
with patch(
"homeassistant.components.zha.core.gateway.ZHAGateway.async_initialize",
side_effect=NetworkSettingsInconsistent(
message="Network settings are inconsistent",
new_state=new_state,
old_state=old_state,
),
):
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_INCONSISTENT_NETWORK_SETTINGS,
)
# The issue is created
assert issue is not None
client = await hass_client()
resp = await client.post(
"/api/repairs/issues/fix",
json={"handler": DOMAIN, "issue_id": issue.issue_id},
)
assert resp.status == HTTPStatus.OK
data = await resp.json()
flow_id = data["flow_id"]
assert data["description_placeholders"]["diff"] == "- PAN ID: `0x2DB4` → `0xBBBB`"
resp = await client.post(
f"/api/repairs/issues/fix/{flow_id}",
json={"next_step_id": "restore_old_settings"},
)
await hass.async_block_till_done()
assert resp.status == HTTPStatus.OK
data = await resp.json()
assert data["type"] == "create_entry"
assert (
issue_registry.async_get_issue(
domain=DOMAIN,
issue_id=ISSUE_INCONSISTENT_NETWORK_SETTINGS,
)
is None
)
assert mock_zigpy_connect.backups.restore_backup.mock_calls == [call(old_state)]

28
tests/components/zha/test_silabs_multiprotocol.py
View file

@ -44,11 +44,7 @@ async def test_async_get_channel_missing(
zigpy_app_controller.state.network_info = zigpy.state.NetworkInfo()
zigpy_app_controller.state.node_info = zigpy.state.NodeInfo()
with patch(
"bellows.zigbee.application.ControllerApplication.__new__",
return_value=zigpy_app_controller,
):
assert await silabs_multiprotocol.async_get_channel(hass) is None
assert await silabs_multiprotocol.async_get_channel(hass) is None
async def test_async_get_channel_no_zha(hass: HomeAssistant) -> None:
@ -74,26 +70,20 @@ async def test_change_channel(
"""Test changing the channel."""
await setup_zha()
with patch.object(
zigpy_app_controller, "move_network_to_channel", autospec=True
) as mock_move_network_to_channel:
task = await silabs_multiprotocol.async_change_channel(hass, 20)
await task
task = await silabs_multiprotocol.async_change_channel(hass, 20)
await task
assert mock_move_network_to_channel.mock_calls == [call(20)]
assert zigpy_app_controller.move_network_to_channel.mock_calls == [call(20)]
async def test_change_channel_no_zha(
hass: HomeAssistant, zigpy_app_controller: ControllerApplication
) -> None:
"""Test changing the channel with no ZHA config entries and no database."""
with patch.object(
zigpy_app_controller, "move_network_to_channel", autospec=True
) as mock_move_network_to_channel:
task = await silabs_multiprotocol.async_change_channel(hass, 20)
task = await silabs_multiprotocol.async_change_channel(hass, 20)
assert task is None
assert mock_move_network_to_channel.mock_calls == []
assert zigpy_app_controller.mock_calls == []
@pytest.mark.parametrize(("delay", "sleep"), [(0, 0), (5, 0), (15, 15 - 10.27)])
@ -107,13 +97,11 @@ async def test_change_channel_delay(
"""Test changing the channel with a delay."""
await setup_zha()
with patch.object(
zigpy_app_controller, "move_network_to_channel", autospec=True
) as mock_move_network_to_channel, patch(
with patch(
"homeassistant.components.zha.silabs_multiprotocol.asyncio.sleep", autospec=True
) as mock_sleep:
task = await silabs_multiprotocol.async_change_channel(hass, 20, delay=delay)
await task
assert mock_move_network_to_channel.mock_calls == [call(20)]
assert zigpy_app_controller.move_network_to_channel.mock_calls == [call(20)]
assert mock_sleep.mock_calls == [call(sleep)]