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

Cache device trigger info during ZHA startup (#99764)

Browse Source 
* Do not connect to the radio hardware within `_connect_zigpy_app`

* Make `connect_zigpy_app` public

* Create radio manager instances from config entries

* Cache device triggers on startup

* reorg zha init

* don't reuse gateway

* don't nuke yaml configuration

* review comments

* Fix existing unit tests

* Ensure `app.shutdown` is called, not just `app.disconnect`

* Revert creating group entities and device registry entries early

* Add unit tests

---------

Co-authored-by: David F. Mulcahey <david.mulcahey@icloud.com>
This commit is contained in:
puddly 2023-09-11 14:36:01 +02:00 committed by GitHub
parent 42046a3ce2
commit a6f325d05a
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
15 changed files with 299 additions and 134 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -1,5 +1,6 @@
"""Support for Zigbee Home Automation devices."""
import asyncio
import contextlib
import copy
import logging
import os
@ -33,13 +34,16 @@ from .core.const import (
CONF_ZIGPY,
DATA_ZHA,
DATA_ZHA_CONFIG,
DATA_ZHA_DEVICE_TRIGGER_CACHE,
DATA_ZHA_GATEWAY,
DOMAIN,
PLATFORMS,
SIGNAL_ADD_ENTITIES,
RadioType,
)
from .core.device import get_device_automation_triggers
from .core.discovery import GROUP_PROBE
from .radio_manager import ZhaRadioManager
DEVICE_CONFIG_SCHEMA_ENTRY = vol.Schema({vol.Optional(CONF_TYPE): cv.string})
ZHA_CONFIG_SCHEMA = {
@ -134,9 +138,43 @@ async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> b
else:
_LOGGER.debug("ZHA storage file does not exist or was already removed")
# Re-use the gateway object between ZHA reloads
if (zha_gateway := zha_data.get(DATA_ZHA_GATEWAY)) is None:
zha_gateway = ZHAGateway(hass, config, config_entry)
# Load and cache device trigger information early
zha_data.setdefault(DATA_ZHA_DEVICE_TRIGGER_CACHE, {})
device_registry = dr.async_get(hass)
radio_mgr = ZhaRadioManager.from_config_entry(hass, config_entry)
async with radio_mgr.connect_zigpy_app() as app:
for dev in app.devices.values():
dev_entry = device_registry.async_get_device(
identifiers={(DOMAIN, str(dev.ieee))},
connections={(dr.CONNECTION_ZIGBEE, str(dev.ieee))},
)
if dev_entry is None:
continue
zha_data[DATA_ZHA_DEVICE_TRIGGER_CACHE][dev_entry.id] = (
str(dev.ieee),
get_device_automation_triggers(dev),
)
_LOGGER.debug("Trigger cache: %s", zha_data[DATA_ZHA_DEVICE_TRIGGER_CACHE])
zha_gateway = ZHAGateway(hass, config, config_entry)
async def async_zha_shutdown():
"""Handle shutdown tasks."""
await zha_gateway.shutdown()
# clean up any remaining entity metadata
# (entities that have been discovered but not yet added to HA)
# suppress KeyError because we don't know what state we may
# be in when we get here in failure cases
with contextlib.suppress(KeyError):
for platform in PLATFORMS:
del hass.data[DATA_ZHA][platform]
config_entry.async_on_unload(async_zha_shutdown)
try:
await zha_gateway.async_initialize()
@ -155,9 +193,6 @@ async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> b
repairs.async_delete_blocking_issues(hass)
config_entry.async_on_unload(zha_gateway.shutdown)
device_registry = dr.async_get(hass)
device_registry.async_get_or_create(
config_entry_id=config_entry.entry_id,
connections={(dr.CONNECTION_ZIGBEE, str(zha_gateway.coordinator_ieee))},

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

@ -186,6 +186,7 @@ DATA_ZHA = "zha"
DATA_ZHA_CONFIG = "config"
DATA_ZHA_BRIDGE_ID = "zha_bridge_id"
DATA_ZHA_CORE_EVENTS = "zha_core_events"
DATA_ZHA_DEVICE_TRIGGER_CACHE = "zha_device_trigger_cache"
DATA_ZHA_GATEWAY = "zha_gateway"
DEBUG_COMP_BELLOWS = "bellows"

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

@ -93,6 +93,16 @@ _UPDATE_ALIVE_INTERVAL = (60, 90)
_CHECKIN_GRACE_PERIODS = 2
def get_device_automation_triggers(
device: zigpy.device.Device,
) -> dict[tuple[str, str], dict[str, str]]:
"""Get the supported device automation triggers for a zigpy device."""
return {
("device_offline", "device_offline"): {"device_event_type": "device_offline"},
**getattr(device, "device_automation_triggers", {}),
}
class DeviceStatus(Enum):
"""Status of a device."""
@ -311,16 +321,7 @@ class ZHADevice(LogMixin):
@cached_property
def device_automation_triggers(self) -> dict[tuple[str, str], dict[str, str]]:
"""Return the device automation triggers for this device."""
triggers = {
("device_offline", "device_offline"): {
"device_event_type": "device_offline"
}
}
if hasattr(self._zigpy_device, "device_automation_triggers"):
triggers.update(self._zigpy_device.device_automation_triggers)
return triggers
return get_device_automation_triggers(self._zigpy_device)
@property
def available_signal(self) -> str:

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

@ -149,12 +149,6 @@ class ZHAGateway:
self.config_entry = config_entry
self._unsubs: list[Callable[[], None]] = []
discovery.PROBE.initialize(self._hass)
discovery.GROUP_PROBE.initialize(self._hass)
self.ha_device_registry = dr.async_get(self._hass)
self.ha_entity_registry = er.async_get(self._hass)
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]
@ -197,6 +191,12 @@ class ZHAGateway:
async def async_initialize(self) -> None:
"""Initialize controller and connect radio."""
discovery.PROBE.initialize(self._hass)
discovery.GROUP_PROBE.initialize(self._hass)
self.ha_device_registry = dr.async_get(self._hass)
self.ha_entity_registry = er.async_get(self._hass)
app_controller_cls, app_config = self.get_application_controller_data()
self.application_controller = await app_controller_cls.new(
config=app_config,
@ -204,23 +204,6 @@ class ZHAGateway:
start_radio=False,
)
self._hass.data[DATA_ZHA][DATA_ZHA_GATEWAY] = self
self.async_load_devices()
# Groups are attached to the coordinator device so we need to load it early
coordinator = self._find_coordinator_device()
loaded_groups = False
# We can only load groups early if the coordinator's model info has been stored
# in the zigpy database
if coordinator.model is not None:
self.coordinator_zha_device = self._async_get_or_create_device(
coordinator, restored=True
)
self.async_load_groups()
loaded_groups = True
for attempt in range(STARTUP_RETRIES):
try:
await self.application_controller.startup(auto_form=True)
@ -242,14 +225,15 @@ class ZHAGateway:
else:
break
self._hass.data[DATA_ZHA][DATA_ZHA_GATEWAY] = self
self._hass.data[DATA_ZHA][DATA_ZHA_BRIDGE_ID] = str(self.coordinator_ieee)
self.coordinator_zha_device = self._async_get_or_create_device(
self._find_coordinator_device(), restored=True
)
self._hass.data[DATA_ZHA][DATA_ZHA_BRIDGE_ID] = str(self.coordinator_ieee)
# If ZHA groups could not load early, we can safely load them now
if not loaded_groups:
self.async_load_groups()
self.async_load_devices()
self.async_load_groups()
self.application_controller.add_listener(self)
self.application_controller.groups.add_listener(self)
@ -766,7 +750,15 @@ class ZHAGateway:
unsubscribe()
for device in self.devices.values():
device.async_cleanup_handles()
await self.application_controller.shutdown()
# 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()
def handle_message(
self,

91
homeassistant/components/zha/device_trigger.py
View File

@ -9,12 +9,12 @@ from homeassistant.components.device_automation.exceptions import (
from homeassistant.components.homeassistant.triggers import event as event_trigger
from homeassistant.const import CONF_DEVICE_ID, CONF_DOMAIN, CONF_PLATFORM, CONF_TYPE
from homeassistant.core import CALLBACK_TYPE, HomeAssistant
from homeassistant.exceptions import HomeAssistantError, IntegrationError
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.trigger import TriggerActionType, TriggerInfo
from homeassistant.helpers.typing import ConfigType
from . import DOMAIN as ZHA_DOMAIN
from .core.const import ZHA_EVENT
from .core.const import DATA_ZHA, DATA_ZHA_DEVICE_TRIGGER_CACHE, ZHA_EVENT
from .core.helpers import async_get_zha_device
CONF_SUBTYPE = "subtype"
@ -26,21 +26,32 @@ TRIGGER_SCHEMA = DEVICE_TRIGGER_BASE_SCHEMA.extend(
)
def _get_device_trigger_data(hass: HomeAssistant, device_id: str) -> tuple[str, dict]:
"""Get device trigger data for a device, falling back to the cache if possible."""
# First, try checking to see if the device itself is accessible
try:
zha_device = async_get_zha_device(hass, device_id)
except KeyError:
pass
else:
return str(zha_device.ieee), zha_device.device_automation_triggers
# If not, check the trigger cache but allow any `KeyError`s to propagate
return hass.data[DATA_ZHA][DATA_ZHA_DEVICE_TRIGGER_CACHE][device_id]
async def async_validate_trigger_config(
hass: HomeAssistant, config: ConfigType
) -> ConfigType:
"""Validate config."""
config = TRIGGER_SCHEMA(config)
# Trigger validation will not occur if the config entry is not loaded
_, triggers = _get_device_trigger_data(hass, config[CONF_DEVICE_ID])
trigger = (config[CONF_TYPE], config[CONF_SUBTYPE])
try:
zha_device = async_get_zha_device(hass, config[CONF_DEVICE_ID])
except (KeyError, AttributeError, IntegrationError) as err:
raise InvalidDeviceAutomationConfig from err
if (
zha_device.device_automation_triggers is None
or trigger not in zha_device.device_automation_triggers
):
if trigger not in triggers:
raise InvalidDeviceAutomationConfig(f"device does not have trigger {trigger}")
return config
@ -53,26 +64,26 @@ async def async_attach_trigger(
trigger_info: TriggerInfo,
) -> CALLBACK_TYPE:
"""Listen for state changes based on configuration."""
trigger_key: tuple[str, str] = (config[CONF_TYPE], config[CONF_SUBTYPE])
try:
zha_device = async_get_zha_device(hass, config[CONF_DEVICE_ID])
except (KeyError, AttributeError) as err:
ieee, triggers = _get_device_trigger_data(hass, config[CONF_DEVICE_ID])
except KeyError as err:
raise HomeAssistantError(
f"Unable to get zha device {config[CONF_DEVICE_ID]}"
) from err
if trigger_key not in zha_device.device_automation_triggers:
trigger_key: tuple[str, str] = (config[CONF_TYPE], config[CONF_SUBTYPE])
if trigger_key not in triggers:
raise HomeAssistantError(f"Unable to find trigger {trigger_key}")
trigger = zha_device.device_automation_triggers[trigger_key]
event_config = {
event_trigger.CONF_PLATFORM: "event",
event_trigger.CONF_EVENT_TYPE: ZHA_EVENT,
event_trigger.CONF_EVENT_DATA: {DEVICE_IEEE: str(zha_device.ieee), **trigger},
}
event_config = event_trigger.TRIGGER_SCHEMA(event_config)
event_config = event_trigger.TRIGGER_SCHEMA(
{
event_trigger.CONF_PLATFORM: "event",
event_trigger.CONF_EVENT_TYPE: ZHA_EVENT,
event_trigger.CONF_EVENT_DATA: {DEVICE_IEEE: ieee, **triggers[trigger_key]},
}
)
return await event_trigger.async_attach_trigger(
hass, event_config, action, trigger_info, platform_type="device"
)
@ -83,24 +94,20 @@ async def async_get_triggers(
) -> list[dict[str, str]]:
"""List device triggers.
Make sure the device supports device automations and
if it does return the trigger list.
Make sure the device supports device automations and return the trigger list.
"""
zha_device = async_get_zha_device(hass, device_id)
try:
_, triggers = _get_device_trigger_data(hass, device_id)
except KeyError as err:
raise InvalidDeviceAutomationConfig from err
if not zha_device.device_automation_triggers:
return []
triggers = []
for trigger, subtype in zha_device.device_automation_triggers:
triggers.append(
{
CONF_DEVICE_ID: device_id,
CONF_DOMAIN: ZHA_DOMAIN,
CONF_PLATFORM: DEVICE,
CONF_TYPE: trigger,
CONF_SUBTYPE: subtype,
}
)
return triggers
return [
{
CONF_DEVICE_ID: device_id,
CONF_DOMAIN: ZHA_DOMAIN,
CONF_PLATFORM: DEVICE,
CONF_TYPE: trigger,
CONF_SUBTYPE: subtype,
}
for trigger, subtype in triggers
]

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

@ -8,7 +8,7 @@ import copy
import enum
import logging
import os
from typing import Any
from typing import Any, Self
from bellows.config import CONF_USE_THREAD
import voluptuous as vol
@ -127,8 +127,21 @@ class ZhaRadioManager:
self.backups: list[zigpy.backups.NetworkBackup] = []
self.chosen_backup: zigpy.backups.NetworkBackup | None = None
@classmethod
def from_config_entry(
cls, hass: HomeAssistant, config_entry: config_entries.ConfigEntry
) -> Self:
"""Create an instance from a config entry."""
mgr = cls()
mgr.hass = hass
mgr.device_path = config_entry.data[CONF_DEVICE][CONF_DEVICE_PATH]
mgr.device_settings = config_entry.data[CONF_DEVICE]
mgr.radio_type = RadioType[config_entry.data[CONF_RADIO_TYPE]]
return mgr
@contextlib.asynccontextmanager
async def _connect_zigpy_app(self) -> ControllerApplication:
async def connect_zigpy_app(self) -> ControllerApplication:
"""Connect to the radio with the current config and then clean up."""
assert self.radio_type is not None
@ -155,10 +168,9 @@ class ZhaRadioManager:
)
try:
await app.connect()
yield app
finally:
await app.disconnect()
await app.shutdown()
await asyncio.sleep(CONNECT_DELAY_S)
async def restore_backup(
@ -170,7 +182,8 @@ class ZhaRadioManager:
):
return
async with self._connect_zigpy_app() as app:
async with self.connect_zigpy_app() as app:
await app.connect()
await app.backups.restore_backup(backup, **kwargs)
@staticmethod
@ -218,7 +231,9 @@ class ZhaRadioManager:
"""Connect to the radio and load its current network settings."""
backup = None
async with self._connect_zigpy_app() as app:
async with self.connect_zigpy_app() as app:
await app.connect()
# Check if the stick has any settings and load them
try:
await app.load_network_info()
@ -241,12 +256,14 @@ class ZhaRadioManager:
async def async_form_network(self) -> None:
"""Form a brand-new network."""
async with self._connect_zigpy_app() as app:
async with self.connect_zigpy_app() as app:
await app.connect()
await app.form_network()
async def async_reset_adapter(self) -> None:
"""Reset the current adapter."""
async with self._connect_zigpy_app() as app:
async with self.connect_zigpy_app() as app:
await app.connect()
await app.reset_network_info()
async def async_restore_backup_step_1(self) -> bool:

2
tests/components/homeassistant_hardware/conftest.py
View File

@ -23,7 +23,7 @@ def mock_zha_config_flow_setup() -> Generator[None, None, None]:
with patch(
"bellows.zigbee.application.ControllerApplication.probe", side_effect=mock_probe
), patch(
"homeassistant.components.zha.radio_manager.ZhaRadioManager._connect_zigpy_app",
"homeassistant.components.zha.radio_manager.ZhaRadioManager.connect_zigpy_app",
return_value=mock_connect_app,
), patch(
"homeassistant.components.zha.async_setup_entry",

2
tests/components/homeassistant_sky_connect/conftest.py
View File

@ -25,7 +25,7 @@ def mock_zha():
)
with patch(
"homeassistant.components.zha.radio_manager.ZhaRadioManager._connect_zigpy_app",
"homeassistant.components.zha.radio_manager.ZhaRadioManager.connect_zigpy_app",
return_value=mock_connect_app,
), patch(
"homeassistant.components.zha.async_setup_entry",

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

@ -45,7 +45,7 @@ def mock_zha_config_flow_setup() -> Generator[None, None, None]:
with patch(
"bellows.zigbee.application.ControllerApplication.probe", side_effect=mock_probe
), patch(
"homeassistant.components.zha.radio_manager.ZhaRadioManager._connect_zigpy_app",
"homeassistant.components.zha.radio_manager.ZhaRadioManager.connect_zigpy_app",
return_value=mock_connect_app,
):
yield

2
tests/components/homeassistant_yellow/conftest.py
View File

@ -23,7 +23,7 @@ def mock_zha_config_flow_setup() -> Generator[None, None, None]:
with patch(
"bellows.zigbee.application.ControllerApplication.probe", side_effect=mock_probe
), patch(
"homeassistant.components.zha.radio_manager.ZhaRadioManager._connect_zigpy_app",
"homeassistant.components.zha.radio_manager.ZhaRadioManager.connect_zigpy_app",
return_value=mock_connect_app,
), patch(
"homeassistant.components.zha.async_setup_entry",

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

@ -293,14 +293,20 @@ def zigpy_device_mock(zigpy_app_controller):
return _mock_dev
@patch("homeassistant.components.zha.setup_quirks", MagicMock(return_value=True))
@pytest.fixture
def zha_device_joined(hass, setup_zha):
"""Return a newly joined ZHA device."""
setup_zha_fixture = setup_zha
async def _zha_device(zigpy_dev):
async def _zha_device(zigpy_dev, *, setup_zha: bool = True):
zigpy_dev.last_seen = time.time()
await setup_zha()
if setup_zha:
await setup_zha_fixture()
zha_gateway = common.get_zha_gateway(hass)
zha_gateway.application_controller.devices[zigpy_dev.ieee] = zigpy_dev
await zha_gateway.async_device_initialized(zigpy_dev)
await hass.async_block_till_done()
return zha_gateway.get_device(zigpy_dev.ieee)
@ -308,17 +314,21 @@ def zha_device_joined(hass, setup_zha):
return _zha_device
@patch("homeassistant.components.zha.setup_quirks", MagicMock(return_value=True))
@pytest.fixture
def zha_device_restored(hass, zigpy_app_controller, setup_zha):
"""Return a restored ZHA device."""
setup_zha_fixture = setup_zha
async def _zha_device(zigpy_dev, last_seen=None):
async def _zha_device(zigpy_dev, *, last_seen=None, setup_zha: bool = True):
zigpy_app_controller.devices[zigpy_dev.ieee] = zigpy_dev
if last_seen is not None:
zigpy_dev.last_seen = last_seen
await setup_zha()
if setup_zha:
await setup_zha_fixture()
zha_gateway = hass.data[zha_const.DATA_ZHA][zha_const.DATA_ZHA_GATEWAY]
return zha_gateway.get_device(zigpy_dev.ieee)
@ -376,3 +386,10 @@ def hass_disable_services(hass):
hass, "services", MagicMock(has_service=MagicMock(return_value=True))
):
yield hass
@pytest.fixture(autouse=True)
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

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

@ -63,13 +63,6 @@ def mock_multipan_platform():
yield
@pytest.fixture(autouse=True)
def reduce_reconnect_timeout():
"""Reduces reconnect timeout to speed up tests."""
with patch("homeassistant.components.zha.radio_manager.CONNECT_DELAY_S", 0.01):
yield
@pytest.fixture(autouse=True)
def mock_app():
"""Mock zigpy app interface."""

130
tests/components/zha/test_device_trigger.py
View File

@ -9,6 +9,9 @@ import zigpy.zcl.clusters.general as general
import homeassistant.components.automation as automation
from homeassistant.components.device_automation import DeviceAutomationType
from homeassistant.components.device_automation.exceptions import (
InvalidDeviceAutomationConfig,
)
from homeassistant.components.zha.core.const import ATTR_ENDPOINT_ID
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
@ -20,6 +23,7 @@ from .common import async_enable_traffic
from .conftest import SIG_EP_INPUT, SIG_EP_OUTPUT, SIG_EP_PROFILE, SIG_EP_TYPE
from tests.common import (
MockConfigEntry,
async_fire_time_changed,
async_get_device_automations,
async_mock_service,
@ -45,6 +49,16 @@ LONG_PRESS = "remote_button_long_press"
LONG_RELEASE = "remote_button_long_release"
SWITCH_SIGNATURE = {
1: {
SIG_EP_INPUT: [general.Basic.cluster_id],
SIG_EP_OUTPUT: [general.OnOff.cluster_id],
SIG_EP_TYPE: zigpy.profiles.zha.DeviceType.ON_OFF_SWITCH,
SIG_EP_PROFILE: zigpy.profiles.zha.PROFILE_ID,
}
}
@pytest.fixture(autouse=True)
def sensor_platforms_only():
"""Only set up the sensor platform and required base platforms to speed up tests."""
@ -72,16 +86,7 @@ def calls(hass):
async def mock_devices(hass, zigpy_device_mock, zha_device_joined_restored):
"""IAS device fixture."""
zigpy_device = zigpy_device_mock(
{
1: {
SIG_EP_INPUT: [general.Basic.cluster_id],
SIG_EP_OUTPUT: [general.OnOff.cluster_id],
SIG_EP_TYPE: zigpy.profiles.zha.DeviceType.ON_OFF_SWITCH,
SIG_EP_PROFILE: zigpy.profiles.zha.PROFILE_ID,
}
}
)
zigpy_device = zigpy_device_mock(SWITCH_SIGNATURE)
zha_device = await zha_device_joined_restored(zigpy_device)
zha_device.update_available(True)
@ -397,3 +402,108 @@ async def test_exception_bad_trigger(
"Unnamed automation failed to setup triggers and has been disabled: "
"device does not have trigger ('junk', 'junk')" in caplog.text
)
async def test_validate_trigger_config_missing_info(
hass: HomeAssistant,
config_entry: MockConfigEntry,
zigpy_device_mock,
mock_zigpy_connect,
zha_device_joined,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test device triggers referring to a missing device."""
# Join a device
switch = zigpy_device_mock(SWITCH_SIGNATURE)
await zha_device_joined(switch)
# After we unload the config entry, trigger info was not cached on startup, nor can
# it be pulled from the current device, making it impossible to validate triggers
await hass.config_entries.async_unload(config_entry.entry_id)
ha_device_registry = dr.async_get(hass)
reg_device = ha_device_registry.async_get_device(
identifiers={("zha", str(switch.ieee))}
)
assert await async_setup_component(
hass,
automation.DOMAIN,
{
automation.DOMAIN: [
{
"trigger": {
"device_id": reg_device.id,
"domain": "zha",
"platform": "device",
"type": "junk",
"subtype": "junk",
},
"action": {
"service": "test.automation",
"data": {"message": "service called"},
},
}
]
},
)
assert "Unable to get zha device" in caplog.text
with pytest.raises(InvalidDeviceAutomationConfig):
await async_get_device_automations(
hass, DeviceAutomationType.TRIGGER, reg_device.id
)
async def test_validate_trigger_config_unloaded_bad_info(
hass: HomeAssistant,
config_entry: MockConfigEntry,
zigpy_device_mock,
mock_zigpy_connect,
zha_device_joined,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test device triggers referring to a missing device."""
# Join a device
switch = zigpy_device_mock(SWITCH_SIGNATURE)
await zha_device_joined(switch)
# After we unload the config entry, trigger info was not cached on startup, nor can
# it be pulled from the current device, making it impossible to validate triggers
await hass.config_entries.async_unload(config_entry.entry_id)
# Reload ZHA to persist the device info in the cache
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.config_entries.async_unload(config_entry.entry_id)
ha_device_registry = dr.async_get(hass)
reg_device = ha_device_registry.async_get_device(
identifiers={("zha", str(switch.ieee))}
)
assert await async_setup_component(
hass,
automation.DOMAIN,
{
automation.DOMAIN: [
{
"trigger": {
"device_id": reg_device.id,
"domain": "zha",
"platform": "device",
"type": "junk",
"subtype": "junk",
},
"action": {
"service": "test.automation",
"data": {"message": "service called"},
},
}
]
},
)
assert "Unable to find trigger" in caplog.text

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

@ -6,7 +6,6 @@ import pytest
from zigpy.config import CONF_DEVICE, CONF_DEVICE_PATH
from zigpy.exceptions import TransientConnectionError
from homeassistant.components.zha import async_setup_entry
from homeassistant.components.zha.core.const import (
CONF_BAUDRATE,
CONF_RADIO_TYPE,
@ -22,7 +21,7 @@ from .test_light import LIGHT_ON_OFF
from tests.common import MockConfigEntry
DATA_RADIO_TYPE = "deconz"
DATA_RADIO_TYPE = "ezsp"
DATA_PORT_PATH = "/dev/serial/by-id/FTDI_USB__-__Serial_Cable_12345678-if00-port0"
@ -137,7 +136,7 @@ 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
hass: HomeAssistant, path: str, cleaned_path: str, mock_zigpy_connect
) -> None:
"""Test migration of config entry from v3, applying corrections to the port path."""
config_entry_v3 = MockConfigEntry(
@ -150,14 +149,9 @@ async def test_setup_with_v3_cleaning_uri(
)
config_entry_v3.add_to_hass(hass)
with patch(
"homeassistant.components.zha.ZHAGateway", return_value=AsyncMock()
) as mock_gateway:
mock_gateway.return_value.coordinator_ieee = "mock_ieee"
mock_gateway.return_value.radio_description = "mock_radio"
assert await async_setup_entry(hass, config_entry_v3)
hass.data[DOMAIN]["zha_gateway"] = mock_gateway.return_value
await hass.config_entries.async_setup(config_entry_v3.entry_id)
await hass.async_block_till_done()
await hass.config_entries.async_unload(config_entry_v3.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

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

@ -32,9 +32,7 @@ def disable_platform_only():
@pytest.fixture(autouse=True)
def reduce_reconnect_timeout():
"""Reduces reconnect timeout to speed up tests."""
with patch(
"homeassistant.components.zha.radio_manager.CONNECT_DELAY_S", 0.0001
), patch("homeassistant.components.zha.radio_manager.RETRY_DELAY_S", 0.0001):
with patch("homeassistant.components.zha.radio_manager.RETRY_DELAY_S", 0.0001):
yield
@ -99,7 +97,7 @@ def mock_connect_zigpy_app() -> Generator[MagicMock, None, None]:
)
with patch(
"homeassistant.components.zha.radio_manager.ZhaRadioManager._connect_zigpy_app",
"homeassistant.components.zha.radio_manager.ZhaRadioManager.connect_zigpy_app",
return_value=mock_connect_app,
):
yield mock_connect_app