jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-23 05:07:41 +00:00
Code Issues Packages Projects Releases Wiki Activity

Bump ZHA dependencies (#104335)

Browse Source
This commit is contained in:
puddly 2023-11-29 05:30:15 -05:00 committed by GitHub
parent 999875d0e4
commit bd8f01bd35
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
21 changed files with 349 additions and 322 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

106
homeassistant/components/zha/__init__.py
View File

@ -9,12 +9,12 @@ import re
import voluptuous as vol
from zhaquirks import setup as setup_quirks
from zigpy.config import CONF_DATABASE, CONF_DEVICE, CONF_DEVICE_PATH
from zigpy.exceptions import NetworkSettingsInconsistent
from zigpy.exceptions import NetworkSettingsInconsistent, TransientConnectionError
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_TYPE, EVENT_HOMEASSISTANT_STOP
from homeassistant.core import Event, HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady, HomeAssistantError
from homeassistant.exceptions import ConfigEntryError, ConfigEntryNotReady
from homeassistant.helpers import device_registry as dr
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.dispatcher import async_dispatcher_send
@ -29,6 +29,7 @@ from .core.const import (
CONF_CUSTOM_QUIRKS_PATH,
CONF_DEVICE_CONFIG,
CONF_ENABLE_QUIRKS,
CONF_FLOW_CONTROL,
CONF_RADIO_TYPE,
CONF_USB_PATH,
CONF_ZIGPY,
@ -36,6 +37,8 @@ from .core.const import (
DOMAIN,
PLATFORMS,
SIGNAL_ADD_ENTITIES,
STARTUP_FAILURE_DELAY_S,
STARTUP_RETRIES,
RadioType,
)
from .core.device import get_device_automation_triggers
@ -158,42 +161,67 @@ async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> b
_LOGGER.debug("Trigger cache: %s", zha_data.device_trigger_cache)
zha_gateway = ZHAGateway(hass, zha_data.yaml_config, config_entry)
# Retry setup a few times before giving up to deal with missing serial ports in VMs
for attempt in range(STARTUP_RETRIES):
try:
zha_gateway = await ZHAGateway.async_from_config(
hass=hass,
config=zha_data.yaml_config,
config_entry=config_entry,
)
break
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 ConfigEntryError(
"Network settings do not match most recent backup"
) from exc
except TransientConnectionError as exc:
raise ConfigEntryNotReady from exc
except Exception as exc: # pylint: disable=broad-except
_LOGGER.debug(
"Couldn't start coordinator (attempt %s of %s)",
attempt + 1,
STARTUP_RETRIES,
exc_info=exc,
)
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 warn_on_wrong_silabs_firmware(
hass, config_entry.data[CONF_DEVICE][CONF_DEVICE_PATH]
)
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
if attempt < STARTUP_RETRIES - 1:
await asyncio.sleep(STARTUP_FAILURE_DELAY_S)
continue
raise
if RadioType[config_entry.data[CONF_RADIO_TYPE]] == RadioType.ezsp:
try:
# Ignore all exceptions during probing, they shouldn't halt setup
await warn_on_wrong_silabs_firmware(
hass, config_entry.data[CONF_DEVICE][CONF_DEVICE_PATH]
)
except AlreadyRunningEZSP as ezsp_exc:
raise ConfigEntryNotReady from ezsp_exc
raise
repairs.async_delete_blocking_issues(hass)
manufacturer = zha_gateway.state.node_info.manufacturer
model = zha_gateway.state.node_info.model
if manufacturer is None and model is None:
manufacturer = "Unknown"
model = "Unknown"
device_registry.async_get_or_create(
config_entry_id=config_entry.entry_id,
connections={(dr.CONNECTION_ZIGBEE, str(zha_gateway.coordinator_ieee))},
identifiers={(DOMAIN, str(zha_gateway.coordinator_ieee))},
connections={(dr.CONNECTION_ZIGBEE, str(zha_gateway.state.node_info.ieee))},
identifiers={(DOMAIN, str(zha_gateway.state.node_info.ieee))},
name="Zigbee Coordinator",
manufacturer="ZHA",
model=zha_gateway.radio_description,
manufacturer=manufacturer,
model=model,
sw_version=zha_gateway.state.node_info.version,
)
websocket_api.async_load_api(hass)
@ -267,5 +295,23 @@ async def async_migrate_entry(hass: HomeAssistant, config_entry: ConfigEntry) ->
config_entry.version = 3
hass.config_entries.async_update_entry(config_entry, data=data)
if config_entry.version == 3:
data = {**config_entry.data}
if not data[CONF_DEVICE].get(CONF_BAUDRATE):
data[CONF_DEVICE][CONF_BAUDRATE] = {
"deconz": 38400,
"xbee": 57600,
"ezsp": 57600,
"znp": 115200,
"zigate": 115200,
}[data[CONF_RADIO_TYPE]]
if not data[CONF_DEVICE].get(CONF_FLOW_CONTROL):
data[CONF_DEVICE][CONF_FLOW_CONTROL] = None
config_entry.version = 4
hass.config_entries.async_update_entry(config_entry, data=data)
_LOGGER.info("Migration to version %s successful", config_entry.version)
return True

26
homeassistant/components/zha/config_flow.py
View File

@ -27,12 +27,13 @@ from homeassistant.util import dt as dt_util
from .core.const import (
CONF_BAUDRATE,
CONF_FLOWCONTROL,
CONF_FLOW_CONTROL,
CONF_RADIO_TYPE,
DOMAIN,
RadioType,
)
from .radio_manager import (
DEVICE_SCHEMA,
HARDWARE_DISCOVERY_SCHEMA,
RECOMMENDED_RADIOS,
ProbeResult,
@ -42,7 +43,7 @@ from .radio_manager import (
CONF_MANUAL_PATH = "Enter Manually"
SUPPORTED_PORT_SETTINGS = (
CONF_BAUDRATE,
CONF_FLOWCONTROL,
CONF_FLOW_CONTROL,
)
DECONZ_DOMAIN = "deconz"
@ -160,7 +161,7 @@ class BaseZhaFlow(FlowHandler):
return self.async_create_entry(
title=self._title,
data={
CONF_DEVICE: device_settings,
CONF_DEVICE: DEVICE_SCHEMA(device_settings),
CONF_RADIO_TYPE: self._radio_mgr.radio_type.name,
},
)
@ -281,7 +282,7 @@ class BaseZhaFlow(FlowHandler):
for (
param,
value,
) in self._radio_mgr.radio_type.controller.SCHEMA_DEVICE.schema.items():
) in DEVICE_SCHEMA.schema.items():
if param not in SUPPORTED_PORT_SETTINGS:
continue
@ -488,7 +489,7 @@ class BaseZhaFlow(FlowHandler):
class ZhaConfigFlowHandler(BaseZhaFlow, config_entries.ConfigFlow, domain=DOMAIN):
"""Handle a config flow."""
VERSION = 3
VERSION = 4
async def _set_unique_id_or_update_path(
self, unique_id: str, device_path: str
@ -646,22 +647,17 @@ class ZhaConfigFlowHandler(BaseZhaFlow, config_entries.ConfigFlow, domain=DOMAIN
name = discovery_data["name"]
radio_type = self._radio_mgr.parse_radio_type(discovery_data["radio_type"])
try:
device_settings = radio_type.controller.SCHEMA_DEVICE(
discovery_data["port"]
)
except vol.Invalid:
return self.async_abort(reason="invalid_hardware_data")
device_settings = discovery_data["port"]
device_path = device_settings[CONF_DEVICE_PATH]
await self._set_unique_id_or_update_path(
unique_id=f"{name}_{radio_type.name}_{device_settings[CONF_DEVICE_PATH]}",
device_path=device_settings[CONF_DEVICE_PATH],
unique_id=f"{name}_{radio_type.name}_{device_path}",
device_path=device_path,
)
self._title = name
self._radio_mgr.radio_type = radio_type
self._radio_mgr.device_path = device_settings[CONF_DEVICE_PATH]
self._radio_mgr.device_path = device_path
self._radio_mgr.device_settings = device_settings
self.context["title_placeholders"] = {CONF_NAME: name}

2
homeassistant/components/zha/core/const.py
View File

@ -127,6 +127,7 @@ CONF_ALARM_FAILED_TRIES = "alarm_failed_tries"
CONF_ALARM_ARM_REQUIRES_CODE = "alarm_arm_requires_code"
CONF_BAUDRATE = "baudrate"
CONF_FLOW_CONTROL = "flow_control"
CONF_CUSTOM_QUIRKS_PATH = "custom_quirks_path"
CONF_DEFAULT_LIGHT_TRANSITION = "default_light_transition"
CONF_DEVICE_CONFIG = "device_config"
@ -136,7 +137,6 @@ CONF_ALWAYS_PREFER_XY_COLOR_MODE = "always_prefer_xy_color_mode"
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_RADIO_TYPE = "radio_type"
CONF_USB_PATH = "usb_path"
CONF_USE_THREAD = "use_thread"

2
homeassistant/components/zha/core/device.py
View File

@ -285,7 +285,7 @@ class ZHADevice(LogMixin):
if not self.is_coordinator:
return False
return self.ieee == self.gateway.coordinator_ieee
return self.ieee == self.gateway.state.node_info.ieee
@property
def is_end_device(self) -> bool | None:

96
homeassistant/components/zha/core/gateway.py
View File

@ -11,7 +11,7 @@ import itertools
import logging
import re
import time
from typing import TYPE_CHECKING, Any, NamedTuple
from typing import TYPE_CHECKING, Any, NamedTuple, Self
from zigpy.application import ControllerApplication
from zigpy.config import (
@ -24,15 +24,14 @@ from zigpy.config import (
)
import zigpy.device
import zigpy.endpoint
from zigpy.exceptions import NetworkSettingsInconsistent, TransientConnectionError
import zigpy.group
from zigpy.state import State
from zigpy.types.named import EUI64
from homeassistant import __path__ as HOMEASSISTANT_PATH
from homeassistant.components.system_log import LogEntry, _figure_out_source
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers import device_registry as dr, entity_registry as er
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.dispatcher import async_dispatcher_send
@ -66,8 +65,6 @@ from .const import (
SIGNAL_ADD_ENTITIES,
SIGNAL_GROUP_MEMBERSHIP_CHANGE,
SIGNAL_REMOVE,
STARTUP_FAILURE_DELAY_S,
STARTUP_RETRIES,
UNKNOWN_MANUFACTURER,
UNKNOWN_MODEL,
ZHA_GW_MSG,
@ -123,10 +120,6 @@ class DevicePairingStatus(Enum):
class ZHAGateway:
"""Gateway that handles events that happen on the ZHA Zigbee network."""
# -- Set in async_initialize --
application_controller: ControllerApplication
radio_description: str
def __init__(
self, hass: HomeAssistant, config: ConfigType, config_entry: ConfigEntry
) -> None:
@ -135,7 +128,8 @@ class ZHAGateway:
self._config = config
self._devices: dict[EUI64, ZHADevice] = {}
self._groups: dict[int, ZHAGroup] = {}
self.coordinator_zha_device: ZHADevice | None = None
self.application_controller: ControllerApplication = None
self.coordinator_zha_device: ZHADevice = None # type: ignore[assignment]
self._device_registry: collections.defaultdict[
EUI64, list[EntityReference]
] = collections.defaultdict(list)
@ -147,13 +141,11 @@ class ZHAGateway:
self._log_relay_handler = LogRelayHandler(hass, self)
self.config_entry = config_entry
self._unsubs: list[Callable[[], None]] = []
self.shutting_down = False
def get_application_controller_data(self) -> tuple[ControllerApplication, dict]:
"""Get an uninitialized instance of a zigpy `ControllerApplication`."""
radio_type = self.config_entry.data[CONF_RADIO_TYPE]
app_controller_cls = RadioType[radio_type].controller
self.radio_description = RadioType[radio_type].description
radio_type = RadioType[self.config_entry.data[CONF_RADIO_TYPE]]
app_config = self._config.get(CONF_ZIGPY, {})
database = self._config.get(
@ -170,7 +162,7 @@ class ZHAGateway:
# event loop, when a connection to a TCP coordinator fails in a specific way
if (
CONF_USE_THREAD not in app_config
and RadioType[radio_type] is RadioType.ezsp
and radio_type is RadioType.ezsp
and app_config[CONF_DEVICE][CONF_DEVICE_PATH].startswith("socket://")
):
app_config[CONF_USE_THREAD] = False
@ -189,48 +181,40 @@ class ZHAGateway:
):
app_config.setdefault(CONF_NWK, {})[CONF_NWK_CHANNEL] = 15
return app_controller_cls, app_controller_cls.SCHEMA(app_config)
return radio_type.controller, radio_type.controller.SCHEMA(app_config)
@classmethod
async def async_from_config(
cls, hass: HomeAssistant, config: ConfigType, config_entry: ConfigEntry
) -> Self:
"""Create an instance of a gateway from config objects."""
instance = cls(hass, config, config_entry)
await instance.async_initialize()
return instance
async def async_initialize(self) -> None:
"""Initialize controller and connect radio."""
discovery.PROBE.initialize(self.hass)
discovery.GROUP_PROBE.initialize(self.hass)
self.shutting_down = False
app_controller_cls, app_config = self.get_application_controller_data()
self.application_controller = await app_controller_cls.new(
app = await app_controller_cls.new(
config=app_config,
auto_form=False,
start_radio=False,
)
try:
for attempt in range(STARTUP_RETRIES):
try:
await self.application_controller.startup(auto_form=True)
except TransientConnectionError as exc:
raise ConfigEntryNotReady from exc
except NetworkSettingsInconsistent:
raise
except Exception as exc: # pylint: disable=broad-except
_LOGGER.debug(
"Couldn't start %s coordinator (attempt %s of %s)",
self.radio_description,
attempt + 1,
STARTUP_RETRIES,
exc_info=exc,
)
if attempt == STARTUP_RETRIES - 1:
raise exc
await asyncio.sleep(STARTUP_FAILURE_DELAY_S)
else:
break
await app.startup(auto_form=True)
except Exception:
# Explicitly shut down the controller application on failure
await self.application_controller.shutdown()
await app.shutdown()
raise
self.application_controller = app
zha_data = get_zha_data(self.hass)
zha_data.gateway = self
@ -244,6 +228,17 @@ class ZHAGateway:
self.application_controller.add_listener(self)
self.application_controller.groups.add_listener(self)
def connection_lost(self, exc: Exception) -> None:
"""Handle connection lost event."""
if self.shutting_down:
return
_LOGGER.debug("Connection to the radio was lost: %r", exc)
self.hass.async_create_task(
self.hass.config_entries.async_reload(self.config_entry.entry_id)
)
def _find_coordinator_device(self) -> zigpy.device.Device:
zigpy_coordinator = self.application_controller.get_device(nwk=0x0000)
@ -258,6 +253,7 @@ class ZHAGateway:
@callback
def async_load_devices(self) -> None:
"""Restore ZHA devices from zigpy application state."""
for zigpy_device in self.application_controller.devices.values():
zha_device = self._async_get_or_create_device(zigpy_device, restored=True)
delta_msg = "not known"
@ -280,6 +276,7 @@ class ZHAGateway:
@callback
def async_load_groups(self) -> None:
"""Initialize ZHA groups."""
for group_id in self.application_controller.groups:
group = self.application_controller.groups[group_id]
zha_group = self._async_get_or_create_group(group)
@ -521,9 +518,9 @@ class ZHAGateway:
entity_registry.async_remove(entry.entity_id)
@property
def coordinator_ieee(self) -> EUI64:
"""Return the active coordinator's IEEE address."""
return self.application_controller.state.node_info.ieee
def state(self) -> State:
"""Return the active coordinator's network state."""
return self.application_controller.state
@property
def devices(self) -> dict[EUI64, ZHADevice]:
@ -711,6 +708,7 @@ class ZHAGateway:
group_id: int | None = None,
) -> ZHAGroup | None:
"""Create a new Zigpy Zigbee group."""
# we start with two to fill any gaps from a user removing existing groups
if group_id is None:
@ -758,19 +756,13 @@ class ZHAGateway:
async def shutdown(self) -> None:
"""Stop ZHA Controller Application."""
_LOGGER.debug("Shutting down ZHA ControllerApplication")
self.shutting_down = True
for unsubscribe in self._unsubs:
unsubscribe()
for device in self.devices.values():
device.async_cleanup_handles()
# shutdown is called when the config entry unloads are processed
# there are cases where unloads are processed because of a failure of
# some sort and the application controller may not have been
# created yet
if (
hasattr(self, "application_controller")
and self.application_controller is not None
):
await self.application_controller.shutdown()
await self.application_controller.shutdown()
def handle_message(
self,

2
homeassistant/components/zha/entity.py
View File

@ -92,7 +92,7 @@ class BaseZhaEntity(LogMixin, entity.Entity):
manufacturer=zha_device_info[ATTR_MANUFACTURER],
model=zha_device_info[ATTR_MODEL],
name=zha_device_info[ATTR_NAME],
via_device=(DOMAIN, zha_gateway.coordinator_ieee),
via_device=(DOMAIN, zha_gateway.state.node_info.ieee),
)
@callback

14
homeassistant/components/zha/manifest.json
View File

@ -21,16 +21,16 @@
"universal_silabs_flasher"
],
"requirements": [
"bellows==0.36.8",
"bellows==0.37.1",
"pyserial==3.5",
"pyserial-asyncio==0.6",
"zha-quirks==0.0.107",
"zigpy-deconz==0.21.1",
"zigpy==0.59.0",
"zigpy-xbee==0.19.0",
"zigpy-zigate==0.11.0",
"zigpy-znp==0.11.6",
"universal-silabs-flasher==0.0.14",
"zigpy-deconz==0.22.0",
"zigpy==0.60.0",
"zigpy-xbee==0.20.0",
"zigpy-zigate==0.12.0",
"zigpy-znp==0.12.0",
"universal-silabs-flasher==0.0.15",
"pyserial-asyncio-fast==0.11"
],
"usb": [

20
homeassistant/components/zha/radio_manager.py
View File

@ -19,6 +19,7 @@ from zigpy.config import (
CONF_DEVICE,
CONF_DEVICE_PATH,
CONF_NWK_BACKUP_ENABLED,
SCHEMA_DEVICE,
)
from zigpy.exceptions import NetworkNotFormed
@ -58,10 +59,21 @@ RETRY_DELAY_S = 1.0
BACKUP_RETRIES = 5
MIGRATION_RETRIES = 100
DEVICE_SCHEMA = vol.Schema(
{
vol.Required("path"): str,
vol.Optional("baudrate", default=115200): int,
vol.Optional("flow_control", default=None): vol.In(
["hardware", "software", None]
),
}
)
HARDWARE_DISCOVERY_SCHEMA = vol.Schema(
{
vol.Required("name"): str,
vol.Required("port"): dict,
vol.Required("port"): DEVICE_SCHEMA,
vol.Required("radio_type"): str,
}
)
@ -204,9 +216,7 @@ class ZhaRadioManager:
for radio in AUTOPROBE_RADIOS:
_LOGGER.debug("Attempting to probe radio type %s", radio)
dev_config = radio.controller.SCHEMA_DEVICE(
{CONF_DEVICE_PATH: self.device_path}
)
dev_config = SCHEMA_DEVICE({CONF_DEVICE_PATH: self.device_path})
probe_result = await radio.controller.probe(dev_config)
if not probe_result:
@ -357,7 +367,7 @@ class ZhaMultiPANMigrationHelper:
migration_data["new_discovery_info"]["radio_type"]
)
new_device_settings = new_radio_type.controller.SCHEMA_DEVICE(
new_device_settings = SCHEMA_DEVICE(
migration_data["new_discovery_info"]["port"]
)

14
requirements_all.txt
View File

@ -523,7 +523,7 @@ beautifulsoup4==4.12.2
# beewi-smartclim==0.0.10
# homeassistant.components.zha
bellows==0.36.8
bellows==0.37.1
# homeassistant.components.bmw_connected_drive
bimmer-connected==0.14.3
@ -2660,7 +2660,7 @@ unifi-discovery==1.1.7
unifiled==0.11
# homeassistant.components.zha
universal-silabs-flasher==0.0.14
universal-silabs-flasher==0.0.15
# homeassistant.components.upb
upb-lib==0.5.4
@ -2828,19 +2828,19 @@ zhong-hong-hvac==1.0.9
ziggo-mediabox-xl==1.1.0
# homeassistant.components.zha
zigpy-deconz==0.21.1
zigpy-deconz==0.22.0
# homeassistant.components.zha
zigpy-xbee==0.19.0
zigpy-xbee==0.20.0
# homeassistant.components.zha
zigpy-zigate==0.11.0
zigpy-zigate==0.12.0
# homeassistant.components.zha
zigpy-znp==0.11.6
zigpy-znp==0.12.0
# homeassistant.components.zha
zigpy==0.59.0
zigpy==0.60.0
# homeassistant.components.zoneminder
zm-py==0.5.2

14
requirements_test_all.txt
View File

@ -445,7 +445,7 @@ base36==0.1.1
beautifulsoup4==4.12.2
# homeassistant.components.zha
bellows==0.36.8
bellows==0.37.1
# homeassistant.components.bmw_connected_drive
bimmer-connected==0.14.3
@ -1979,7 +1979,7 @@ ultraheat-api==0.5.7
unifi-discovery==1.1.7
# homeassistant.components.zha
universal-silabs-flasher==0.0.14
universal-silabs-flasher==0.0.15
# homeassistant.components.upb
upb-lib==0.5.4
@ -2117,19 +2117,19 @@ zeversolar==0.3.1
zha-quirks==0.0.107
# homeassistant.components.zha
zigpy-deconz==0.21.1
zigpy-deconz==0.22.0
# homeassistant.components.zha
zigpy-xbee==0.19.0
zigpy-xbee==0.20.0
# homeassistant.components.zha
zigpy-zigate==0.11.0
zigpy-zigate==0.12.0
# homeassistant.components.zha
zigpy-znp==0.11.6
zigpy-znp==0.12.0
# homeassistant.components.zha
zigpy==0.59.0
zigpy==0.60.0
# homeassistant.components.zwave_js
zwave-js-server-python==0.54.0

13
tests/components/homeassistant_hardware/test_silabs_multiprotocol_addon.py
View File

@ -293,7 +293,14 @@ async def test_option_flow_install_multi_pan_addon_zha(
config_entry.add_to_hass(hass)
zha_config_entry = MockConfigEntry(
data={"device": {"path": "/dev/ttyTEST123"}, "radio_type": "ezsp"},
data={
"device": {
"path": "/dev/ttyTEST123",
"baudrate": 115200,
"flow_control": None,
},
"radio_type": "ezsp",
},
domain=ZHA_DOMAIN,
options={},
title="Test",
@ -348,8 +355,8 @@ async def test_option_flow_install_multi_pan_addon_zha(
assert zha_config_entry.data == {
"device": {
"path": "socket://core-silabs-multiprotocol:9999",
"baudrate": 57600, # ZHA default
"flow_control": "software", # ZHA default
"baudrate": 115200,
"flow_control": None,
},
"radio_type": "ezsp",
}

4
tests/components/homeassistant_sky_connect/test_config_flow.py
View File

@ -337,8 +337,8 @@ async def test_option_flow_install_multi_pan_addon_zha(
assert zha_config_entry.data == {
"device": {
"path": "socket://core-silabs-multiprotocol:9999",
"baudrate": 57600, # ZHA default
"flow_control": "software", # ZHA default
"baudrate": 115200,
"flow_control": None,
},
"radio_type": "ezsp",
}

8
tests/components/homeassistant_sky_connect/test_init.py
View File

@ -147,7 +147,7 @@ async def test_setup_zha(
assert config_entry.data == {
"device": {
"baudrate": 115200,
"flow_control": "software",
"flow_control": None,
"path": CONFIG_ENTRY_DATA["device"],
},
"radio_type": "ezsp",
@ -200,8 +200,8 @@ async def test_setup_zha_multipan(
config_entry = hass.config_entries.async_entries("zha")[0]
assert config_entry.data == {
"device": {
"baudrate": 57600, # ZHA default
"flow_control": "software", # ZHA default
"baudrate": 115200,
"flow_control": None,
"path": "socket://core-silabs-multiprotocol:9999",
},
"radio_type": "ezsp",
@ -255,7 +255,7 @@ async def test_setup_zha_multipan_other_device(
assert config_entry.data == {
"device": {
"baudrate": 115200,
"flow_control": "software",
"flow_control": None,
"path": CONFIG_ENTRY_DATA["device"],
},
"radio_type": "ezsp",

4
tests/components/homeassistant_yellow/test_config_flow.py
View File

@ -249,8 +249,8 @@ async def test_option_flow_install_multi_pan_addon_zha(
assert zha_config_entry.data == {
"device": {
"path": "socket://core-silabs-multiprotocol:9999",
"baudrate": 57600, # ZHA default
"flow_control": "software", # ZHA default
"baudrate": 115200,
"flow_control": None,
},
"radio_type": "ezsp",
}

4
tests/components/homeassistant_yellow/test_init.py
View File

@ -145,8 +145,8 @@ async def test_setup_zha_multipan(
config_entry = hass.config_entries.async_entries("zha")[0]
assert config_entry.data == {
"device": {
"baudrate": 57600, # ZHA default
"flow_control": "software", # ZHA default
"baudrate": 115200,
"flow_control": None,
"path": "socket://core-silabs-multiprotocol:9999",
},
"radio_type": "ezsp",

6
tests/components/zha/conftest.py
View File

@ -46,7 +46,7 @@ def disable_request_retry_delay():
with patch(
"homeassistant.components.zha.core.cluster_handlers.RETRYABLE_REQUEST_DECORATOR",
zigpy.util.retryable_request(tries=3, delay=0),
):
), patch("homeassistant.components.zha.STARTUP_FAILURE_DELAY_S", 0.01):
yield
@ -83,8 +83,8 @@ class _FakeApp(ControllerApplication):
async def permit_ncp(self, time_s: int = 60):
pass
async def permit_with_key(
self, node: zigpy.types.EUI64, code: bytes, time_s: int = 60
async def permit_with_link_key(
self, node: zigpy.types.EUI64, link_key: zigpy.types.KeyData, time_s: int = 60
):
pass

97
tests/components/zha/test_config_flow.py
View File

@ -10,7 +10,7 @@ import pytest
import serial.tools.list_ports
from zigpy.backups import BackupManager
import zigpy.config
from zigpy.config import CONF_DEVICE, CONF_DEVICE_PATH
from zigpy.config import CONF_DEVICE, CONF_DEVICE_PATH, SCHEMA_DEVICE
import zigpy.device
from zigpy.exceptions import NetworkNotFormed
import zigpy.types
@ -22,7 +22,7 @@ from homeassistant.components.ssdp import ATTR_UPNP_MANUFACTURER_URL, ATTR_UPNP_
from homeassistant.components.zha import config_flow, radio_manager
from homeassistant.components.zha.core.const import (
CONF_BAUDRATE,
CONF_FLOWCONTROL,
CONF_FLOW_CONTROL,
CONF_RADIO_TYPE,
DOMAIN,
EZSP_OVERWRITE_EUI64,
@ -118,9 +118,7 @@ def mock_detect_radio_type(
async def detect(self):
self.radio_type = radio_type
self.device_settings = radio_type.controller.SCHEMA_DEVICE(
{CONF_DEVICE_PATH: self.device_path}
)
self.device_settings = SCHEMA_DEVICE({CONF_DEVICE_PATH: self.device_path})
return ret
@ -181,7 +179,7 @@ async def test_zeroconf_discovery_znp(hass: HomeAssistant) -> None:
assert result3["data"] == {
CONF_DEVICE: {
CONF_BAUDRATE: 115200,
CONF_FLOWCONTROL: None,
CONF_FLOW_CONTROL: None,
CONF_DEVICE_PATH: "socket://192.168.1.200:6638",
},
CONF_RADIO_TYPE: "znp",
@ -238,6 +236,8 @@ async def test_zigate_via_zeroconf(setup_entry_mock, hass: HomeAssistant) -> Non
assert result4["data"] == {
CONF_DEVICE: {
CONF_DEVICE_PATH: "socket://192.168.1.200:1234",
CONF_BAUDRATE: 115200,
CONF_FLOW_CONTROL: None,
},
CONF_RADIO_TYPE: "zigate",
}
@ -287,7 +287,7 @@ async def test_efr32_via_zeroconf(hass: HomeAssistant) -> None:
CONF_DEVICE: {
CONF_DEVICE_PATH: "socket://192.168.1.200:1234",
CONF_BAUDRATE: 115200,
CONF_FLOWCONTROL: "software",
CONF_FLOW_CONTROL: None,
},
CONF_RADIO_TYPE: "ezsp",
}
@ -304,7 +304,7 @@ async def test_discovery_via_zeroconf_ip_change(hass: HomeAssistant) -> None:
CONF_DEVICE: {
CONF_DEVICE_PATH: "socket://192.168.1.5:6638",
CONF_BAUDRATE: 115200,
CONF_FLOWCONTROL: None,
CONF_FLOW_CONTROL: None,
}
},
)
@ -328,7 +328,7 @@ async def test_discovery_via_zeroconf_ip_change(hass: HomeAssistant) -> None:
assert entry.data[CONF_DEVICE] == {
CONF_DEVICE_PATH: "socket://192.168.1.22:6638",
CONF_BAUDRATE: 115200,
CONF_FLOWCONTROL: None,
CONF_FLOW_CONTROL: None,
}
@ -483,6 +483,8 @@ async def test_zigate_discovery_via_usb(probe_mock, hass: HomeAssistant) -> None
assert result4["data"] == {
"device": {
"path": "/dev/ttyZIGBEE",
"baudrate": 115200,
"flow_control": None,
},
CONF_RADIO_TYPE: "zigate",
}
@ -555,7 +557,7 @@ async def test_discovery_via_usb_path_changes(hass: HomeAssistant) -> None:
CONF_DEVICE: {
CONF_DEVICE_PATH: "/dev/ttyUSB1",
CONF_BAUDRATE: 115200,
CONF_FLOWCONTROL: None,
CONF_FLOW_CONTROL: None,
}
},
)
@ -579,7 +581,7 @@ async def test_discovery_via_usb_path_changes(hass: HomeAssistant) -> None:
assert entry.data[CONF_DEVICE] == {
CONF_DEVICE_PATH: "/dev/ttyZIGBEE",
CONF_BAUDRATE: 115200,
CONF_FLOWCONTROL: None,
CONF_FLOW_CONTROL: None,
}
@ -754,6 +756,8 @@ async def test_user_flow(hass: HomeAssistant) -> None:
assert result2["data"] == {
"device": {
"path": port.device,
CONF_BAUDRATE: 115200,
CONF_FLOW_CONTROL: None,
},
CONF_RADIO_TYPE: "deconz",
}
@ -773,7 +777,11 @@ async def test_user_flow_not_detected(hass: HomeAssistant) -> None:
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={CONF_SOURCE: SOURCE_USER},
data={zigpy.config.CONF_DEVICE_PATH: port_select},
data={
zigpy.config.CONF_DEVICE_PATH: port_select,
CONF_BAUDRATE: 115200,
CONF_FLOW_CONTROL: None,
},
)
assert result["type"] == FlowResultType.FORM
@ -951,31 +959,6 @@ async def test_user_port_config(probe_mock, hass: HomeAssistant) -> None:
assert probe_mock.await_count == 1
@pytest.mark.parametrize(
("old_type", "new_type"),
[
("ezsp", "ezsp"),
("ti_cc", "znp"), # only one that should change
("znp", "znp"),
("deconz", "deconz"),
],
)
async def test_migration_ti_cc_to_znp(
old_type, new_type, hass: HomeAssistant, config_entry: MockConfigEntry
) -> None:
"""Test zigpy-cc to zigpy-znp config migration."""
config_entry.data = {**config_entry.data, CONF_RADIO_TYPE: old_type}
config_entry.version = 2
config_entry.add_to_hass(hass)
with patch("homeassistant.components.zha.async_setup_entry", return_value=True):
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert config_entry.version > 2
assert config_entry.data[CONF_RADIO_TYPE] == new_type
@pytest.mark.parametrize("onboarded", [True, False])
@patch("homeassistant.components.zha.async_setup_entry", AsyncMock(return_value=True))
async def test_hardware(onboarded, hass: HomeAssistant) -> None:
@ -1022,7 +1005,7 @@ async def test_hardware(onboarded, hass: HomeAssistant) -> None:
assert result3["data"] == {
CONF_DEVICE: {
CONF_BAUDRATE: 115200,
CONF_FLOWCONTROL: "hardware",
CONF_FLOW_CONTROL: "hardware",
CONF_DEVICE_PATH: "/dev/ttyAMA1",
},
CONF_RADIO_TYPE: "ezsp",
@ -1171,6 +1154,7 @@ async def test_formation_strategy_form_initial_network(
@patch(f"zigpy_znp.{PROBE_FUNCTION_PATH}", AsyncMock(return_value=True))
@patch("homeassistant.components.zha.async_setup_entry", AsyncMock(return_value=True))
async def test_onboarding_auto_formation_new_hardware(
mock_app, hass: HomeAssistant
) -> None:
@ -1577,7 +1561,7 @@ async def test_options_flow_defaults(
CONF_DEVICE: {
CONF_DEVICE_PATH: "/dev/ttyUSB0",
CONF_BAUDRATE: 12345,
CONF_FLOWCONTROL: None,
CONF_FLOW_CONTROL: None,
},
CONF_RADIO_TYPE: "znp",
},
@ -1645,7 +1629,7 @@ async def test_options_flow_defaults(
# Change everything
CONF_DEVICE_PATH: "/dev/new_serial_port",
CONF_BAUDRATE: 54321,
CONF_FLOWCONTROL: "software",
CONF_FLOW_CONTROL: "software",
},
)
@ -1668,7 +1652,7 @@ async def test_options_flow_defaults(
CONF_DEVICE: {
CONF_DEVICE_PATH: "/dev/new_serial_port",
CONF_BAUDRATE: 54321,
CONF_FLOWCONTROL: "software",
CONF_FLOW_CONTROL: "software",
},
CONF_RADIO_TYPE: "znp",
}
@ -1697,7 +1681,7 @@ async def test_options_flow_defaults_socket(hass: HomeAssistant) -> None:
CONF_DEVICE: {
CONF_DEVICE_PATH: "socket://localhost:5678",
CONF_BAUDRATE: 12345,
CONF_FLOWCONTROL: None,
CONF_FLOW_CONTROL: None,
},
CONF_RADIO_TYPE: "znp",
},
@ -1766,7 +1750,7 @@ async def test_options_flow_restarts_running_zha_if_cancelled(
CONF_DEVICE: {
CONF_DEVICE_PATH: "socket://localhost:5678",
CONF_BAUDRATE: 12345,
CONF_FLOWCONTROL: None,
CONF_FLOW_CONTROL: None,
},
CONF_RADIO_TYPE: "znp",
},
@ -1821,7 +1805,7 @@ async def test_options_flow_migration_reset_old_adapter(
CONF_DEVICE: {
CONF_DEVICE_PATH: "/dev/serial/by-id/old_radio",
CONF_BAUDRATE: 12345,
CONF_FLOWCONTROL: None,
CONF_FLOW_CONTROL: None,
},
CONF_RADIO_TYPE: "znp",
},
@ -1954,3 +1938,28 @@ async def test_discovery_wrong_firmware_installed(hass: HomeAssistant) -> None:
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "wrong_firmware_installed"
@pytest.mark.parametrize(
("old_type", "new_type"),
[
("ezsp", "ezsp"),
("ti_cc", "znp"), # only one that should change
("znp", "znp"),
("deconz", "deconz"),
],
)
async def test_migration_ti_cc_to_znp(
old_type: str, new_type: str, hass: HomeAssistant, config_entry: MockConfigEntry
) -> None:
"""Test zigpy-cc to zigpy-znp config migration."""
config_entry.data = {**config_entry.data, CONF_RADIO_TYPE: old_type}
config_entry.version = 2
config_entry.add_to_hass(hass)
with patch("homeassistant.components.zha.async_setup_entry", return_value=True):
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert config_entry.version > 2
assert config_entry.data[CONF_RADIO_TYPE] == new_type

136
tests/components/zha/test_gateway.py
View File

@ -4,22 +4,21 @@ from unittest.mock import MagicMock, patch
import pytest
from zigpy.application import ControllerApplication
import zigpy.exceptions
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.gateway import ZHAGateway
from homeassistant.components.zha.core.group import GroupMember
from homeassistant.components.zha.core.helpers import get_zha_gateway
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
from .common import async_find_group_entity_id
from .conftest import SIG_EP_INPUT, SIG_EP_OUTPUT, SIG_EP_PROFILE, SIG_EP_TYPE
from tests.common import MockConfigEntry
IEEE_GROUPABLE_DEVICE = "01:2d:6f:00:0a:90:69:e8"
IEEE_GROUPABLE_DEVICE2 = "02:2d:6f:00:0a:90:69:e8"
@ -224,101 +223,6 @@ async def test_gateway_create_group_with_id(
assert zha_group.group_id == 0x1234
@patch(
"homeassistant.components.zha.core.gateway.ZHAGateway.async_load_devices",
MagicMock(),
)
@patch(
"homeassistant.components.zha.core.gateway.ZHAGateway.async_load_groups",
MagicMock(),
)
@patch("homeassistant.components.zha.core.gateway.STARTUP_FAILURE_DELAY_S", 0.01)
@pytest.mark.parametrize(
"startup_effect",
[
[asyncio.TimeoutError(), FileNotFoundError(), None],
[asyncio.TimeoutError(), None],
[None],
],
)
async def test_gateway_initialize_success(
startup_effect: list[Exception | None],
hass: HomeAssistant,
device_light_1: ZHADevice,
coordinator: ZHADevice,
zigpy_app_controller: ControllerApplication,
) -> None:
"""Test ZHA initializing the gateway successfully."""
zha_gateway = get_zha_gateway(hass)
assert zha_gateway is not None
zigpy_app_controller.startup.side_effect = startup_effect
zigpy_app_controller.startup.reset_mock()
with patch(
"bellows.zigbee.application.ControllerApplication.new",
return_value=zigpy_app_controller,
):
await zha_gateway.async_initialize()
assert zigpy_app_controller.startup.call_count == len(startup_effect)
device_light_1.async_cleanup_handles()
@patch("homeassistant.components.zha.core.gateway.STARTUP_FAILURE_DELAY_S", 0.01)
async def test_gateway_initialize_failure(
hass: HomeAssistant,
device_light_1: ZHADevice,
coordinator: ZHADevice,
zigpy_app_controller: ControllerApplication,
) -> None:
"""Test ZHA failing to initialize the gateway."""
zha_gateway = get_zha_gateway(hass)
assert zha_gateway is not None
zigpy_app_controller.startup.side_effect = [
asyncio.TimeoutError(),
RuntimeError(),
FileNotFoundError(),
]
zigpy_app_controller.startup.reset_mock()
with patch(
"bellows.zigbee.application.ControllerApplication.new",
return_value=zigpy_app_controller,
), pytest.raises(FileNotFoundError):
await zha_gateway.async_initialize()
assert zigpy_app_controller.startup.call_count == 3
@patch("homeassistant.components.zha.core.gateway.STARTUP_FAILURE_DELAY_S", 0.01)
async def test_gateway_initialize_failure_transient(
hass: HomeAssistant,
device_light_1: ZHADevice,
coordinator: ZHADevice,
zigpy_app_controller: ControllerApplication,
) -> None:
"""Test ZHA failing to initialize the gateway but with a transient error."""
zha_gateway = get_zha_gateway(hass)
assert zha_gateway is not None
zigpy_app_controller.startup.side_effect = [
RuntimeError(),
zigpy.exceptions.TransientConnectionError(),
]
zigpy_app_controller.startup.reset_mock()
with patch(
"bellows.zigbee.application.ControllerApplication.new",
return_value=zigpy_app_controller,
), pytest.raises(ConfigEntryNotReady):
await zha_gateway.async_initialize()
# Initialization immediately stops and is retried after TransientConnectionError
assert zigpy_app_controller.startup.call_count == 2
@patch(
"homeassistant.components.zha.core.gateway.ZHAGateway.async_load_devices",
MagicMock(),
@ -340,22 +244,25 @@ async def test_gateway_initialize_bellows_thread(
thread_state: bool,
config_override: dict,
hass: HomeAssistant,
coordinator: ZHADevice,
zigpy_app_controller: ControllerApplication,
config_entry: MockConfigEntry,
) -> None:
"""Test ZHA disabling the UART thread when connecting to a TCP coordinator."""
zha_gateway = get_zha_gateway(hass)
assert zha_gateway is not None
config_entry.data = dict(config_entry.data)
config_entry.data["device"]["path"] = device_path
config_entry.add_to_hass(hass)
zha_gateway.config_entry.data = dict(zha_gateway.config_entry.data)
zha_gateway.config_entry.data["device"]["path"] = device_path
zha_gateway._config.setdefault("zigpy_config", {}).update(config_override)
zha_gateway = ZHAGateway(hass, {"zigpy_config": config_override}, config_entry)
await zha_gateway.async_initialize()
with patch(
"bellows.zigbee.application.ControllerApplication.new",
return_value=zigpy_app_controller,
) as mock_new:
await zha_gateway.async_initialize()
RadioType.ezsp.controller.new.mock_calls[-1].kwargs["config"][
"use_thread"
] is thread_state
mock_new.mock_calls[-1].kwargs["config"]["use_thread"] is thread_state
await zha_gateway.shutdown()
@pytest.mark.parametrize(
@ -373,15 +280,14 @@ async def test_gateway_force_multi_pan_channel(
config_override: dict,
expected_channel: int | None,
hass: HomeAssistant,
coordinator,
config_entry: MockConfigEntry,
) -> None:
"""Test ZHA disabling the UART thread when connecting to a TCP coordinator."""
zha_gateway = get_zha_gateway(hass)
assert zha_gateway is not None
config_entry.data = dict(config_entry.data)
config_entry.data["device"]["path"] = device_path
config_entry.add_to_hass(hass)
zha_gateway.config_entry.data = dict(zha_gateway.config_entry.data)
zha_gateway.config_entry.data["device"]["path"] = device_path
zha_gateway._config.setdefault("zigpy_config", {}).update(config_override)
zha_gateway = ZHAGateway(hass, {"zigpy_config": config_override}, config_entry)
_, config = zha_gateway.get_application_controller_data()
assert config["network"]["channel"] == expected_channel

79
tests/components/zha/test_init.py
View File

@ -1,5 +1,6 @@
"""Tests for ZHA integration init."""
import asyncio
import typing
from unittest.mock import AsyncMock, Mock, patch
import pytest
@ -9,6 +10,7 @@ from zigpy.exceptions import TransientConnectionError
from homeassistant.components.zha.core.const import (
CONF_BAUDRATE,
CONF_FLOW_CONTROL,
CONF_RADIO_TYPE,
CONF_USB_PATH,
DOMAIN,
@ -61,9 +63,8 @@ async def test_migration_from_v1_no_baudrate(
assert config_entry_v1.data[CONF_RADIO_TYPE] == DATA_RADIO_TYPE
assert CONF_DEVICE in config_entry_v1.data
assert config_entry_v1.data[CONF_DEVICE][CONF_DEVICE_PATH] == DATA_PORT_PATH
assert CONF_BAUDRATE not in config_entry_v1.data[CONF_DEVICE]
assert CONF_USB_PATH not in config_entry_v1.data
assert config_entry_v1.version == 3
assert config_entry_v1.version == 4
@patch("homeassistant.components.zha.async_setup_entry", AsyncMock(return_value=True))
@ -80,7 +81,7 @@ async def test_migration_from_v1_with_baudrate(
assert CONF_USB_PATH not in config_entry_v1.data
assert CONF_BAUDRATE in config_entry_v1.data[CONF_DEVICE]
assert config_entry_v1.data[CONF_DEVICE][CONF_BAUDRATE] == 115200
assert config_entry_v1.version == 3
assert config_entry_v1.version == 4
@patch("homeassistant.components.zha.async_setup_entry", AsyncMock(return_value=True))
@ -95,8 +96,7 @@ async def test_migration_from_v1_wrong_baudrate(
assert CONF_DEVICE in config_entry_v1.data
assert config_entry_v1.data[CONF_DEVICE][CONF_DEVICE_PATH] == DATA_PORT_PATH
assert CONF_USB_PATH not in config_entry_v1.data
assert CONF_BAUDRATE not in config_entry_v1.data[CONF_DEVICE]
assert config_entry_v1.version == 3
assert config_entry_v1.version == 4
@pytest.mark.skipif(
@ -149,23 +149,74 @@ async def test_setup_with_v3_cleaning_uri(
mock_zigpy_connect: ControllerApplication,
) -> None:
"""Test migration of config entry from v3, applying corrections to the port path."""
config_entry_v3 = MockConfigEntry(
config_entry_v4 = MockConfigEntry(
domain=DOMAIN,
data={
CONF_RADIO_TYPE: DATA_RADIO_TYPE,
CONF_DEVICE: {CONF_DEVICE_PATH: path, CONF_BAUDRATE: 115200},
CONF_DEVICE: {
CONF_DEVICE_PATH: path,
CONF_BAUDRATE: 115200,
CONF_FLOW_CONTROL: None,
},
},
version=3,
version=4,
)
config_entry_v3.add_to_hass(hass)
config_entry_v4.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry_v3.entry_id)
await hass.config_entries.async_setup(config_entry_v4.entry_id)
await hass.async_block_till_done()
await hass.config_entries.async_unload(config_entry_v3.entry_id)
await hass.config_entries.async_unload(config_entry_v4.entry_id)
assert config_entry_v3.data[CONF_RADIO_TYPE] == DATA_RADIO_TYPE
assert config_entry_v3.data[CONF_DEVICE][CONF_DEVICE_PATH] == cleaned_path
assert config_entry_v3.version == 3
assert config_entry_v4.data[CONF_RADIO_TYPE] == DATA_RADIO_TYPE
assert config_entry_v4.data[CONF_DEVICE][CONF_DEVICE_PATH] == cleaned_path
assert config_entry_v4.version == 4
@pytest.mark.parametrize(
(
"radio_type",
"old_baudrate",
"old_flow_control",
"new_baudrate",
"new_flow_control",
),
[
("znp", None, None, 115200, None),
("znp", None, "software", 115200, "software"),
("znp", 57600, "software", 57600, "software"),
("deconz", None, None, 38400, None),
("deconz", 115200, None, 115200, None),
],
)
@patch("homeassistant.components.zha.async_setup_entry", AsyncMock(return_value=True))
async def test_migration_baudrate_and_flow_control(
radio_type: str,
old_baudrate: int,
old_flow_control: typing.Literal["hardware", "software", None],
new_baudrate: int,
new_flow_control: typing.Literal["hardware", "software", None],
hass: HomeAssistant,
config_entry: MockConfigEntry,
) -> None:
"""Test baudrate and flow control migration."""
config_entry.data = {
**config_entry.data,
CONF_RADIO_TYPE: radio_type,
CONF_DEVICE: {
CONF_BAUDRATE: old_baudrate,
CONF_FLOW_CONTROL: old_flow_control,
CONF_DEVICE_PATH: "/dev/null",
},
}
config_entry.version = 3
config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert config_entry.version > 3
assert config_entry.data[CONF_DEVICE][CONF_BAUDRATE] == new_baudrate
assert config_entry.data[CONF_DEVICE][CONF_FLOW_CONTROL] == new_flow_control
@patch(

6
tests/components/zha/test_repairs.py
View File

@ -95,6 +95,7 @@ def test_detect_radio_hardware_failure(hass: HomeAssistant) -> None:
assert _detect_radio_hardware(hass, SKYCONNECT_DEVICE) == HardwareType.OTHER
@patch("homeassistant.components.zha.STARTUP_RETRIES", new=1)
@pytest.mark.parametrize(
("detected_hardware", "expected_learn_more_url"),
[
@ -188,6 +189,7 @@ async def test_multipan_firmware_no_repair_on_probe_failure(
assert issue is None
@patch("homeassistant.components.zha.STARTUP_RETRIES", new=1)
async def test_multipan_firmware_retry_on_probe_ezsp(
hass: HomeAssistant,
config_entry: MockConfigEntry,
@ -312,6 +314,8 @@ async def test_inconsistent_settings_keep_new(
data = await resp.json()
assert data["type"] == "create_entry"
await hass.config_entries.async_unload(config_entry.entry_id)
assert (
issue_registry.async_get_issue(
domain=DOMAIN,
@ -388,6 +392,8 @@ async def test_inconsistent_settings_restore_old(
data = await resp.json()
assert data["type"] == "create_entry"
await hass.config_entries.async_unload(config_entry.entry_id)
assert (
issue_registry.async_get_issue(
domain=DOMAIN,

18
tests/components/zha/test_websocket_api.py
View File

@ -62,7 +62,7 @@ from .conftest import (
)
from .data import BASE_CUSTOM_CONFIGURATION, CONFIG_WITH_ALARM_OPTIONS
from tests.common import MockUser
from tests.common import MockConfigEntry, MockUser
IEEE_SWITCH_DEVICE = "01:2d:6f:00:0a:90:69:e7"
IEEE_GROUPABLE_DEVICE = "01:2d:6f:00:0a:90:69:e8"
@ -295,10 +295,12 @@ async def test_get_zha_config_with_alarm(
async def test_update_zha_config(
zha_client, app_controller: ControllerApplication
hass: HomeAssistant,
config_entry: MockConfigEntry,
zha_client,
app_controller: ControllerApplication,
) -> None:
"""Test updating ZHA custom configuration."""
configuration: dict = deepcopy(CONFIG_WITH_ALARM_OPTIONS)
configuration["data"]["zha_options"]["default_light_transition"] = 10
@ -312,10 +314,12 @@ async def test_update_zha_config(
msg = await zha_client.receive_json()
assert msg["success"]
await zha_client.send_json({ID: 6, TYPE: "zha/configuration"})
msg = await zha_client.receive_json()
configuration = msg["result"]
assert configuration == configuration
await zha_client.send_json({ID: 6, TYPE: "zha/configuration"})
msg = await zha_client.receive_json()
configuration = msg["result"]
assert configuration == configuration
await hass.config_entries.async_unload(config_entry.entry_id)
async def test_device_not_found(zha_client) -> None: