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

ZHA network settings API (#88564)

Browse Source 
* Rename `zha.api` to `zha.websocket_api`

* Implement a ZHA network settings API

* Use the enum name as the radio type

* Don't filter out ignored config entries

* [WIP] Start unit tests

* Add unit tests

* Rename ZHA websocket API module in `.coveragerc`

* Rename `api` to `websocket_api`

* Increase test coverage to 100%
This commit is contained in:
puddly 2023-03-22 11:15:46 -04:00 committed by GitHub
parent 130c8ea5f5
commit c581116c82
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
10 changed files with 2558 additions and 2349 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.coveragerc
View File

@ -1508,7 +1508,7 @@ omit =
homeassistant/components/zeversolar/coordinator.py homeassistant/components/zeversolar/coordinator.py
homeassistant/components/zeversolar/entity.py homeassistant/components/zeversolar/entity.py
homeassistant/components/zeversolar/sensor.py homeassistant/components/zeversolar/sensor.py
homeassistant/components/zha/api.py homeassistant/components/zha/websocket_api.py
homeassistant/components/zha/core/channels/* homeassistant/components/zha/core/channels/*
homeassistant/components/zha/core/device.py homeassistant/components/zha/core/device.py
homeassistant/components/zha/core/gateway.py homeassistant/components/zha/core/gateway.py

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

@ -17,7 +17,7 @@ from homeassistant.helpers.dispatcher import async_dispatcher_send
from homeassistant.helpers.storage import STORAGE_DIR from homeassistant.helpers.storage import STORAGE_DIR
from homeassistant.helpers.typing import ConfigType from homeassistant.helpers.typing import ConfigType
from . import api from . import websocket_api
from .core import ZHAGateway from .core import ZHAGateway
from .core.const import ( from .core.const import (
BAUD_RATES, BAUD_RATES,
@ -131,7 +131,7 @@ async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> b
model=zha_gateway.radio_description, model=zha_gateway.radio_description,
) )
api.async_load_api(hass) websocket_api.async_load_api(hass)
async def async_zha_shutdown(event): async def async_zha_shutdown(event):
"""Handle shutdown tasks.""" """Handle shutdown tasks."""
@ -150,11 +150,11 @@ async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> b
async def async_unload_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> bool: async def async_unload_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> bool:
"""Unload ZHA config entry.""" """Unload ZHA config entry."""
zha_gateway: ZHAGateway = hass.data[DATA_ZHA][DATA_ZHA_GATEWAY] zha_gateway: ZHAGateway = hass.data[DATA_ZHA].pop(DATA_ZHA_GATEWAY)
await zha_gateway.shutdown() await zha_gateway.shutdown()
GROUP_PROBE.cleanup() GROUP_PROBE.cleanup()
api.async_unload_api(hass) websocket_api.async_unload_api(hass)
# our components don't have unload methods so no need to look at return values # our components don't have unload methods so no need to look at return values
await asyncio.gather( await asyncio.gather(

1615
homeassistant/components/zha/api.py
View File

File diff suppressed because it is too large Load Diff

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

@ -84,7 +84,7 @@ from .const import (
from .helpers import LogMixin, async_get_zha_config_value, convert_to_zcl_values from .helpers import LogMixin, async_get_zha_config_value, convert_to_zcl_values
if TYPE_CHECKING: if TYPE_CHECKING:
from ..api import ClusterBinding from ..websocket_api import ClusterBinding
from .gateway import ZHAGateway from .gateway import ZHAGateway
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)

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

@ -148,14 +148,8 @@ class ZHAGateway:
self._unsubs: list[Callable[[], None]] = [] self._unsubs: list[Callable[[], None]] = []
self.initialized: bool = False self.initialized: bool = False
async def async_initialize(self) -> None: def get_application_controller_data(self) -> tuple[ControllerApplication, dict]:
"""Initialize controller and connect radio.""" """Get an uninitialized instance of a zigpy `ControllerApplication`."""
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)
radio_type = self.config_entry.data[CONF_RADIO_TYPE] radio_type = self.config_entry.data[CONF_RADIO_TYPE]
app_controller_cls = RadioType[radio_type].controller app_controller_cls = RadioType[radio_type].controller
@ -178,7 +172,17 @@ class ZHAGateway:
): ):
app_config[CONF_USE_THREAD] = False app_config[CONF_USE_THREAD] = False
app_config = app_controller_cls.SCHEMA(app_config) return app_controller_cls, app_controller_cls.SCHEMA(app_config)
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()
for attempt in range(STARTUP_RETRIES): for attempt in range(STARTUP_RETRIES):
try: try:

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

@ -12,10 +12,10 @@ from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.typing import ConfigType, TemplateVarsType from homeassistant.helpers.typing import ConfigType, TemplateVarsType
from . import DOMAIN from . import DOMAIN
from .api import SERVICE_WARNING_DEVICE_SQUAWK, SERVICE_WARNING_DEVICE_WARN
from .core.channels.manufacturerspecific import AllLEDEffectType, SingleLEDEffectType from .core.channels.manufacturerspecific import AllLEDEffectType, SingleLEDEffectType
from .core.const import CHANNEL_IAS_WD, CHANNEL_INOVELLI from .core.const import CHANNEL_IAS_WD, CHANNEL_INOVELLI
from .core.helpers import async_get_zha_device from .core.helpers import async_get_zha_device
from .websocket_api import SERVICE_WARNING_DEVICE_SQUAWK, SERVICE_WARNING_DEVICE_WARN
# mypy: disallow-any-generics # mypy: disallow-any-generics

1541
homeassistant/components/zha/websocket_api.py Normal file
View File

File diff suppressed because it is too large Load Diff

869
tests/components/zha/test_api.py
View File

@ -1,842 +1,91 @@
"""Test ZHA API.""" """Test ZHA API."""
from binascii import unhexlify from unittest.mock import patch
from copy import deepcopy
from unittest.mock import AsyncMock, patch
import pytest import pytest
import voluptuous as vol import zigpy.state
import zigpy.backups
import zigpy.profiles.zha
import zigpy.types
import zigpy.zcl.clusters.general as general
import zigpy.zcl.clusters.security as security
from homeassistant.components.websocket_api import const from homeassistant.components import zha
from homeassistant.components.zha import DOMAIN from homeassistant.components.zha import api
from homeassistant.components.zha.api import ( from homeassistant.components.zha.core.const import RadioType
ATTR_DURATION,
ATTR_INSTALL_CODE,
ATTR_QR_CODE,
ATTR_SOURCE_IEEE,
ID,
SERVICE_PERMIT,
TYPE,
async_load_api,
)
from homeassistant.components.zha.core.const import (
ATTR_CLUSTER_ID,
ATTR_CLUSTER_TYPE,
ATTR_ENDPOINT_ID,
ATTR_ENDPOINT_NAMES,
ATTR_IEEE,
ATTR_MANUFACTURER,
ATTR_MODEL,
ATTR_NEIGHBORS,
ATTR_QUIRK_APPLIED,
CLUSTER_TYPE_IN,
DATA_ZHA,
DATA_ZHA_GATEWAY,
EZSP_OVERWRITE_EUI64,
GROUP_ID,
GROUP_IDS,
GROUP_NAME,
)
from homeassistant.const import ATTR_NAME, Platform
from homeassistant.core import Context, HomeAssistant
from .conftest import (
FIXTURE_GRP_ID,
FIXTURE_GRP_NAME,
SIG_EP_INPUT,
SIG_EP_OUTPUT,
SIG_EP_PROFILE,
SIG_EP_TYPE,
)
from .data import BASE_CUSTOM_CONFIGURATION, CONFIG_WITH_ALARM_OPTIONS
from tests.common import MockUser
IEEE_SWITCH_DEVICE = "01:2d:6f:00:0a:90:69:e7"
IEEE_GROUPABLE_DEVICE = "01:2d:6f:00:0a:90:69:e8"
@pytest.fixture(autouse=True) @pytest.fixture(autouse=True)
def required_platform_only(): def required_platform_only():
"""Only set up the required and required base platforms to speed up tests.""" """Only set up the required and required base platforms to speed up tests."""
with patch( with patch("homeassistant.components.zha.PLATFORMS", ()):
"homeassistant.components.zha.PLATFORMS",
(
Platform.ALARM_CONTROL_PANEL,
Platform.SELECT,
Platform.SENSOR,
Platform.SWITCH,
),
):
yield yield
@pytest.fixture async def test_async_get_network_settings_active(hass, setup_zha):
async def device_switch(hass, zigpy_device_mock, zha_device_joined): """Test reading settings with an active ZHA installation."""
"""Test ZHA switch platform.""" await setup_zha()
zigpy_device = zigpy_device_mock( settings = await api.async_get_network_settings(hass)
{ assert settings.network_info.channel == 15
1: {
SIG_EP_INPUT: [general.OnOff.cluster_id, general.Basic.cluster_id],
SIG_EP_OUTPUT: [],
SIG_EP_TYPE: zigpy.profiles.zha.DeviceType.ON_OFF_SWITCH,
SIG_EP_PROFILE: zigpy.profiles.zha.PROFILE_ID,
}
},
ieee=IEEE_SWITCH_DEVICE,
)
zha_device = await zha_device_joined(zigpy_device)
zha_device.available = True
return zha_device
@pytest.fixture async def test_async_get_network_settings_inactive(
async def device_ias_ace(hass, zigpy_device_mock, zha_device_joined): hass, setup_zha, zigpy_app_controller
"""Test alarm control panel device.""" ):
"""Test reading settings with an inactive ZHA installation."""
await setup_zha()
zigpy_device = zigpy_device_mock( gateway = api._get_gateway(hass)
{ await zha.async_unload_entry(hass, gateway.config_entry)
1: {
SIG_EP_INPUT: [security.IasAce.cluster_id],
SIG_EP_OUTPUT: [],
SIG_EP_TYPE: zigpy.profiles.zha.DeviceType.IAS_ANCILLARY_CONTROL,
SIG_EP_PROFILE: zigpy.profiles.zha.PROFILE_ID,
}
},
)
zha_device = await zha_device_joined(zigpy_device)
zha_device.available = True
return zha_device
zigpy_app_controller.state.network_info.channel = 20
@pytest.fixture
async def device_groupable(hass, zigpy_device_mock, zha_device_joined):
"""Test ZHA light platform."""
zigpy_device = zigpy_device_mock(
{
1: {
SIG_EP_INPUT: [
general.OnOff.cluster_id,
general.Basic.cluster_id,
general.Groups.cluster_id,
],
SIG_EP_OUTPUT: [],
SIG_EP_TYPE: zigpy.profiles.zha.DeviceType.ON_OFF_SWITCH,
SIG_EP_PROFILE: zigpy.profiles.zha.PROFILE_ID,
}
},
ieee=IEEE_GROUPABLE_DEVICE,
)
zha_device = await zha_device_joined(zigpy_device)
zha_device.available = True
return zha_device
@pytest.fixture
async def zha_client(hass, hass_ws_client, device_switch, device_groupable):
"""Get ZHA WebSocket client."""
# load the ZHA API
async_load_api(hass)
return await hass_ws_client(hass)
async def test_device_clusters(hass: HomeAssistant, zha_client) -> None:
"""Test getting device cluster info."""
await zha_client.send_json(
{ID: 5, TYPE: "zha/devices/clusters", ATTR_IEEE: IEEE_SWITCH_DEVICE}
)
msg = await zha_client.receive_json()
assert len(msg["result"]) == 2
cluster_infos = sorted(msg["result"], key=lambda k: k[ID])
cluster_info = cluster_infos[0]
assert cluster_info[TYPE] == CLUSTER_TYPE_IN
assert cluster_info[ID] == 0
assert cluster_info[ATTR_NAME] == "Basic"
cluster_info = cluster_infos[1]
assert cluster_info[TYPE] == CLUSTER_TYPE_IN
assert cluster_info[ID] == 6
assert cluster_info[ATTR_NAME] == "OnOff"
async def test_device_cluster_attributes(zha_client) -> None:
"""Test getting device cluster attributes."""
await zha_client.send_json(
{
ID: 5,
TYPE: "zha/devices/clusters/attributes",
ATTR_ENDPOINT_ID: 1,
ATTR_IEEE: IEEE_SWITCH_DEVICE,
ATTR_CLUSTER_ID: 6,
ATTR_CLUSTER_TYPE: CLUSTER_TYPE_IN,
}
)
msg = await zha_client.receive_json()
attributes = msg["result"]
assert len(attributes) == 7
for attribute in attributes:
assert attribute[ID] is not None
assert attribute[ATTR_NAME] is not None
async def test_device_cluster_commands(zha_client) -> None:
"""Test getting device cluster commands."""
await zha_client.send_json(
{
ID: 5,
TYPE: "zha/devices/clusters/commands",
ATTR_ENDPOINT_ID: 1,
ATTR_IEEE: IEEE_SWITCH_DEVICE,
ATTR_CLUSTER_ID: 6,
ATTR_CLUSTER_TYPE: CLUSTER_TYPE_IN,
}
)
msg = await zha_client.receive_json()
commands = msg["result"]
assert len(commands) == 6
for command in commands:
assert command[ID] is not None
assert command[ATTR_NAME] is not None
assert command[TYPE] is not None
async def test_list_devices(zha_client) -> None:
"""Test getting ZHA devices."""
await zha_client.send_json({ID: 5, TYPE: "zha/devices"})
msg = await zha_client.receive_json()
devices = msg["result"]
assert len(devices) == 2
msg_id = 100
for device in devices:
msg_id += 1
assert device[ATTR_IEEE] is not None
assert device[ATTR_MANUFACTURER] is not None
assert device[ATTR_MODEL] is not None
assert device[ATTR_NAME] is not None
assert device[ATTR_QUIRK_APPLIED] is not None
assert device["entities"] is not None
assert device[ATTR_NEIGHBORS] is not None
assert device[ATTR_ENDPOINT_NAMES] is not None
for entity_reference in device["entities"]:
assert entity_reference[ATTR_NAME] is not None
assert entity_reference["entity_id"] is not None
await zha_client.send_json(
{ID: msg_id, TYPE: "zha/device", ATTR_IEEE: device[ATTR_IEEE]}
)
msg = await zha_client.receive_json()
device2 = msg["result"]
assert device == device2
async def test_get_zha_config(zha_client) -> None:
"""Test getting ZHA custom configuration."""
await zha_client.send_json({ID: 5, TYPE: "zha/configuration"})
msg = await zha_client.receive_json()
configuration = msg["result"]
assert configuration == BASE_CUSTOM_CONFIGURATION
async def test_get_zha_config_with_alarm(
hass: HomeAssistant, zha_client, device_ias_ace
) -> None:
"""Test getting ZHA custom configuration."""
await zha_client.send_json({ID: 5, TYPE: "zha/configuration"})
msg = await zha_client.receive_json()
configuration = msg["result"]
assert configuration == CONFIG_WITH_ALARM_OPTIONS
# test that the alarm options are not in the config when we remove the device
device_ias_ace.gateway.device_removed(device_ias_ace.device)
await hass.async_block_till_done()
await zha_client.send_json({ID: 6, TYPE: "zha/configuration"})
msg = await zha_client.receive_json()
configuration = msg["result"]
assert configuration == BASE_CUSTOM_CONFIGURATION
async def test_update_zha_config(zha_client, zigpy_app_controller) -> None:
"""Test updating ZHA custom configuration."""
configuration = deepcopy(CONFIG_WITH_ALARM_OPTIONS)
configuration["data"]["zha_options"]["default_light_transition"] = 10
with patch( with patch(
"bellows.zigbee.application.ControllerApplication.new", "bellows.zigbee.application.ControllerApplication.__new__",
return_value=zigpy_app_controller, return_value=zigpy_app_controller,
): ):
await zha_client.send_json( settings = await api.async_get_network_settings(hass)
{ID: 5, TYPE: "zha/configuration/update", "data": configuration["data"]}
)
msg = await zha_client.receive_json()
assert msg["success"]
await zha_client.send_json({ID: 6, TYPE: "zha/configuration"}) assert len(zigpy_app_controller._load_db.mock_calls) == 1
msg = await zha_client.receive_json() assert len(zigpy_app_controller.start_network.mock_calls) == 0
configuration = msg["result"]
assert configuration == configuration assert settings.network_info.channel == 20
async def test_device_not_found(zha_client) -> None: async def test_async_get_network_settings_missing(
"""Test not found response from get device API.""" hass, setup_zha, zigpy_app_controller
await zha_client.send_json( ):
{ID: 6, TYPE: "zha/device", ATTR_IEEE: "28:6d:97:00:01:04:11:8c"} """Test reading settings with an inactive ZHA installation, no valid channel."""
)
msg = await zha_client.receive_json()
assert msg["id"] == 6
assert msg["type"] == const.TYPE_RESULT
assert not msg["success"]
assert msg["error"]["code"] == const.ERR_NOT_FOUND
async def test_list_groups(zha_client) -> None:
"""Test getting ZHA zigbee groups."""
await zha_client.send_json({ID: 7, TYPE: "zha/groups"})
msg = await zha_client.receive_json()
assert msg["id"] == 7
assert msg["type"] == const.TYPE_RESULT
groups = msg["result"]
assert len(groups) == 1
for group in groups:
assert group["group_id"] == FIXTURE_GRP_ID
assert group["name"] == FIXTURE_GRP_NAME
assert group["members"] == []
async def test_get_group(zha_client) -> None:
"""Test getting a specific ZHA zigbee group."""
await zha_client.send_json({ID: 8, TYPE: "zha/group", GROUP_ID: FIXTURE_GRP_ID})
msg = await zha_client.receive_json()
assert msg["id"] == 8
assert msg["type"] == const.TYPE_RESULT
group = msg["result"]
assert group is not None
assert group["group_id"] == FIXTURE_GRP_ID
assert group["name"] == FIXTURE_GRP_NAME
assert group["members"] == []
async def test_get_group_not_found(zha_client) -> None:
"""Test not found response from get group API."""
await zha_client.send_json({ID: 9, TYPE: "zha/group", GROUP_ID: 1_234_567})
msg = await zha_client.receive_json()
assert msg["id"] == 9
assert msg["type"] == const.TYPE_RESULT
assert not msg["success"]
assert msg["error"]["code"] == const.ERR_NOT_FOUND
async def test_list_groupable_devices(zha_client, device_groupable) -> None:
"""Test getting ZHA devices that have a group cluster."""
await zha_client.send_json({ID: 10, TYPE: "zha/devices/groupable"})
msg = await zha_client.receive_json()
assert msg["id"] == 10
assert msg["type"] == const.TYPE_RESULT
device_endpoints = msg["result"]
assert len(device_endpoints) == 1
for endpoint in device_endpoints:
assert endpoint["device"][ATTR_IEEE] == "01:2d:6f:00:0a:90:69:e8"
assert endpoint["device"][ATTR_MANUFACTURER] is not None
assert endpoint["device"][ATTR_MODEL] is not None
assert endpoint["device"][ATTR_NAME] is not None
assert endpoint["device"][ATTR_QUIRK_APPLIED] is not None
assert endpoint["device"]["entities"] is not None
assert endpoint["endpoint_id"] is not None
assert endpoint["entities"] is not None
for entity_reference in endpoint["device"]["entities"]:
assert entity_reference[ATTR_NAME] is not None
assert entity_reference["entity_id"] is not None
for entity_reference in endpoint["entities"]:
assert entity_reference["original_name"] is not None
# Make sure there are no groupable devices when the device is unavailable
# Make device unavailable
device_groupable.available = False
await zha_client.send_json({ID: 11, TYPE: "zha/devices/groupable"})
msg = await zha_client.receive_json()
assert msg["id"] == 11
assert msg["type"] == const.TYPE_RESULT
device_endpoints = msg["result"]
assert len(device_endpoints) == 0
async def test_add_group(zha_client) -> None:
"""Test adding and getting a new ZHA zigbee group."""
await zha_client.send_json({ID: 12, TYPE: "zha/group/add", GROUP_NAME: "new_group"})
msg = await zha_client.receive_json()
assert msg["id"] == 12
assert msg["type"] == const.TYPE_RESULT
added_group = msg["result"]
assert added_group["name"] == "new_group"
assert added_group["members"] == []
await zha_client.send_json({ID: 13, TYPE: "zha/groups"})
msg = await zha_client.receive_json()
assert msg["id"] == 13
assert msg["type"] == const.TYPE_RESULT
groups = msg["result"]
assert len(groups) == 2
for group in groups:
assert group["name"] == FIXTURE_GRP_NAME or group["name"] == "new_group"
async def test_remove_group(zha_client) -> None:
"""Test removing a new ZHA zigbee group."""
await zha_client.send_json({ID: 14, TYPE: "zha/groups"})
msg = await zha_client.receive_json()
assert msg["id"] == 14
assert msg["type"] == const.TYPE_RESULT
groups = msg["result"]
assert len(groups) == 1
await zha_client.send_json(
{ID: 15, TYPE: "zha/group/remove", GROUP_IDS: [FIXTURE_GRP_ID]}
)
msg = await zha_client.receive_json()
assert msg["id"] == 15
assert msg["type"] == const.TYPE_RESULT
groups_remaining = msg["result"]
assert len(groups_remaining) == 0
await zha_client.send_json({ID: 16, TYPE: "zha/groups"})
msg = await zha_client.receive_json()
assert msg["id"] == 16
assert msg["type"] == const.TYPE_RESULT
groups = msg["result"]
assert len(groups) == 0
@pytest.fixture
async def app_controller(hass, setup_zha):
"""Fixture for zigpy Application Controller."""
await setup_zha() await setup_zha()
controller = hass.data[DATA_ZHA][DATA_ZHA_GATEWAY].application_controller
p1 = patch.object(controller, "permit") gateway = api._get_gateway(hass)
p2 = patch.object(controller, "permit_with_key", new=AsyncMock()) await zha.async_unload_entry(hass, gateway.config_entry)
with p1, p2:
yield controller # Network settings were never loaded for whatever reason
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)
assert settings is None
@pytest.mark.parametrize( async def test_async_get_network_settings_failure(hass):
("params", "duration", "node"), """Test reading settings with no ZHA config entries and no database."""
( with pytest.raises(ValueError):
({}, 60, None), await api.async_get_network_settings(hass)
({ATTR_DURATION: 30}, 30, None),
(
{ATTR_DURATION: 33, ATTR_IEEE: "aa:bb:cc:dd:aa:bb:cc:dd"},
33,
zigpy.types.EUI64.convert("aa:bb:cc:dd:aa:bb:cc:dd"),
),
(
{ATTR_IEEE: "aa:bb:cc:dd:aa:bb:cc:d1"},
60,
zigpy.types.EUI64.convert("aa:bb:cc:dd:aa:bb:cc:d1"),
),
),
)
async def test_permit_ha12(
hass: HomeAssistant,
app_controller,
hass_admin_user: MockUser,
params,
duration,
node,
) -> None:
"""Test permit service."""
await hass.services.async_call(
DOMAIN, SERVICE_PERMIT, params, True, Context(user_id=hass_admin_user.id)
)
assert app_controller.permit.await_count == 1
assert app_controller.permit.await_args[1]["time_s"] == duration
assert app_controller.permit.await_args[1]["node"] == node
assert app_controller.permit_with_key.call_count == 0
IC_TEST_PARAMS = ( async def test_async_get_radio_type_active(hass, setup_zha):
( """Test reading the radio type with an active ZHA installation."""
{ await setup_zha()
ATTR_SOURCE_IEEE: IEEE_SWITCH_DEVICE,
ATTR_INSTALL_CODE: "5279-7BF4-A508-4DAA-8E17-12B6-1741-CA02-4051", radio_type = api.async_get_radio_type(hass)
}, assert radio_type == RadioType.ezsp
zigpy.types.EUI64.convert(IEEE_SWITCH_DEVICE),
unhexlify("52797BF4A5084DAA8E1712B61741CA024051"),
),
(
{
ATTR_SOURCE_IEEE: IEEE_SWITCH_DEVICE,
ATTR_INSTALL_CODE: "52797BF4A5084DAA8E1712B61741CA024051",
},
zigpy.types.EUI64.convert(IEEE_SWITCH_DEVICE),
unhexlify("52797BF4A5084DAA8E1712B61741CA024051"),
),
)
@pytest.mark.parametrize(("params", "src_ieee", "code"), IC_TEST_PARAMS) async def test_async_get_radio_path_active(hass, setup_zha):
async def test_permit_with_install_code( """Test reading the radio path with an active ZHA installation."""
hass: HomeAssistant, await setup_zha()
app_controller,
hass_admin_user: MockUser,
params,
src_ieee,
code,
) -> None:
"""Test permit service with install code."""
await hass.services.async_call( radio_path = api.async_get_radio_path(hass)
DOMAIN, SERVICE_PERMIT, params, True, Context(user_id=hass_admin_user.id) assert radio_path == "/dev/ttyUSB0"
)
assert app_controller.permit.await_count == 0
assert app_controller.permit_with_key.call_count == 1
assert app_controller.permit_with_key.await_args[1]["time_s"] == 60
assert app_controller.permit_with_key.await_args[1]["node"] == src_ieee
assert app_controller.permit_with_key.await_args[1]["code"] == code
IC_FAIL_PARAMS = (
{
# wrong install code
ATTR_SOURCE_IEEE: IEEE_SWITCH_DEVICE,
ATTR_INSTALL_CODE: "5279-7BF4-A508-4DAA-8E17-12B6-1741-CA02-4052",
},
# incorrect service params
{ATTR_INSTALL_CODE: "5279-7BF4-A508-4DAA-8E17-12B6-1741-CA02-4051"},
{ATTR_SOURCE_IEEE: IEEE_SWITCH_DEVICE},
{
# incorrect service params
ATTR_INSTALL_CODE: "5279-7BF4-A508-4DAA-8E17-12B6-1741-CA02-4051",
ATTR_QR_CODE: "Z:000D6FFFFED4163B$I:52797BF4A5084DAA8E1712B61741CA024051",
},
{
# incorrect service params
ATTR_SOURCE_IEEE: IEEE_SWITCH_DEVICE,
ATTR_QR_CODE: "Z:000D6FFFFED4163B$I:52797BF4A5084DAA8E1712B61741CA024051",
},
{
# good regex match, but bad code
ATTR_QR_CODE: "Z:000D6FFFFED4163B$I:52797BF4A5084DAA8E1712B61741CA024052"
},
{
# good aqara regex match, but bad code
ATTR_QR_CODE: (
"G$M:751$S:357S00001579$D:000000000F350FFD%Z$A:04CF8CDF"
"3C3C3C3C$I:52797BF4A5084DAA8E1712B61741CA024052"
)
},
# good consciot regex match, but bad code
{ATTR_QR_CODE: "000D6FFFFED4163B|52797BF4A5084DAA8E1712B61741CA024052"},
)
@pytest.mark.parametrize("params", IC_FAIL_PARAMS)
async def test_permit_with_install_code_fail(
hass: HomeAssistant, app_controller, hass_admin_user: MockUser, params
) -> None:
"""Test permit service with install code."""
with pytest.raises(vol.Invalid):
await hass.services.async_call(
DOMAIN, SERVICE_PERMIT, params, True, Context(user_id=hass_admin_user.id)
)
assert app_controller.permit.await_count == 0
assert app_controller.permit_with_key.call_count == 0
IC_QR_CODE_TEST_PARAMS = (
(
{ATTR_QR_CODE: "000D6FFFFED4163B|52797BF4A5084DAA8E1712B61741CA024051"},
zigpy.types.EUI64.convert("00:0D:6F:FF:FE:D4:16:3B"),
unhexlify("52797BF4A5084DAA8E1712B61741CA024051"),
),
(
{ATTR_QR_CODE: "Z:000D6FFFFED4163B$I:52797BF4A5084DAA8E1712B61741CA024051"},
zigpy.types.EUI64.convert("00:0D:6F:FF:FE:D4:16:3B"),
unhexlify("52797BF4A5084DAA8E1712B61741CA024051"),
),
(
{
ATTR_QR_CODE: (
"G$M:751$S:357S00001579$D:000000000F350FFD%Z$A:04CF8CDF"
"3C3C3C3C$I:52797BF4A5084DAA8E1712B61741CA024051"
)
},
zigpy.types.EUI64.convert("04:CF:8C:DF:3C:3C:3C:3C"),
unhexlify("52797BF4A5084DAA8E1712B61741CA024051"),
),
)
@pytest.mark.parametrize(("params", "src_ieee", "code"), IC_QR_CODE_TEST_PARAMS)
async def test_permit_with_qr_code(
hass: HomeAssistant,
app_controller,
hass_admin_user: MockUser,
params,
src_ieee,
code,
) -> None:
"""Test permit service with install code from qr code."""
await hass.services.async_call(
DOMAIN, SERVICE_PERMIT, params, True, Context(user_id=hass_admin_user.id)
)
assert app_controller.permit.await_count == 0
assert app_controller.permit_with_key.call_count == 1
assert app_controller.permit_with_key.await_args[1]["time_s"] == 60
assert app_controller.permit_with_key.await_args[1]["node"] == src_ieee
assert app_controller.permit_with_key.await_args[1]["code"] == code
@pytest.mark.parametrize(("params", "src_ieee", "code"), IC_QR_CODE_TEST_PARAMS)
async def test_ws_permit_with_qr_code(
app_controller, zha_client, params, src_ieee, code
) -> None:
"""Test permit service with install code from qr code."""
await zha_client.send_json(
{ID: 14, TYPE: f"{DOMAIN}/devices/{SERVICE_PERMIT}", **params}
)
msg = await zha_client.receive_json()
assert msg["id"] == 14
assert msg["type"] == const.TYPE_RESULT
assert msg["success"]
assert app_controller.permit.await_count == 0
assert app_controller.permit_with_key.call_count == 1
assert app_controller.permit_with_key.await_args[1]["time_s"] == 60
assert app_controller.permit_with_key.await_args[1]["node"] == src_ieee
assert app_controller.permit_with_key.await_args[1]["code"] == code
@pytest.mark.parametrize("params", IC_FAIL_PARAMS)
async def test_ws_permit_with_install_code_fail(
app_controller, zha_client, params
) -> None:
"""Test permit ws service with install code."""
await zha_client.send_json(
{ID: 14, TYPE: f"{DOMAIN}/devices/{SERVICE_PERMIT}", **params}
)
msg = await zha_client.receive_json()
assert msg["id"] == 14
assert msg["type"] == const.TYPE_RESULT
assert msg["success"] is False
assert app_controller.permit.await_count == 0
assert app_controller.permit_with_key.call_count == 0
@pytest.mark.parametrize(
("params", "duration", "node"),
(
({}, 60, None),
({ATTR_DURATION: 30}, 30, None),
(
{ATTR_DURATION: 33, ATTR_IEEE: "aa:bb:cc:dd:aa:bb:cc:dd"},
33,
zigpy.types.EUI64.convert("aa:bb:cc:dd:aa:bb:cc:dd"),
),
(
{ATTR_IEEE: "aa:bb:cc:dd:aa:bb:cc:d1"},
60,
zigpy.types.EUI64.convert("aa:bb:cc:dd:aa:bb:cc:d1"),
),
),
)
async def test_ws_permit_ha12(
app_controller, zha_client, params, duration, node
) -> None:
"""Test permit ws service."""
await zha_client.send_json(
{ID: 14, TYPE: f"{DOMAIN}/devices/{SERVICE_PERMIT}", **params}
)
msg = await zha_client.receive_json()
assert msg["id"] == 14
assert msg["type"] == const.TYPE_RESULT
assert msg["success"]
assert app_controller.permit.await_count == 1
assert app_controller.permit.await_args[1]["time_s"] == duration
assert app_controller.permit.await_args[1]["node"] == node
assert app_controller.permit_with_key.call_count == 0
async def test_get_network_settings(app_controller, zha_client) -> None:
"""Test current network settings are returned."""
await app_controller.backups.create_backup()
await zha_client.send_json({ID: 6, TYPE: f"{DOMAIN}/network/settings"})
msg = await zha_client.receive_json()
assert msg["id"] == 6
assert msg["type"] == const.TYPE_RESULT
assert msg["success"]
assert "radio_type" in msg["result"]
assert "network_info" in msg["result"]["settings"]
async def test_list_network_backups(app_controller, zha_client) -> None:
"""Test backups are serialized."""
await app_controller.backups.create_backup()
await zha_client.send_json({ID: 6, TYPE: f"{DOMAIN}/network/backups/list"})
msg = await zha_client.receive_json()
assert msg["id"] == 6
assert msg["type"] == const.TYPE_RESULT
assert msg["success"]
assert "network_info" in msg["result"][0]
async def test_create_network_backup(app_controller, zha_client) -> None:
"""Test creating backup."""
assert not app_controller.backups.backups
await zha_client.send_json({ID: 6, TYPE: f"{DOMAIN}/network/backups/create"})
msg = await zha_client.receive_json()
assert len(app_controller.backups.backups) == 1
assert msg["id"] == 6
assert msg["type"] == const.TYPE_RESULT
assert msg["success"]
assert "backup" in msg["result"] and "is_complete" in msg["result"]
async def test_restore_network_backup_success(app_controller, zha_client) -> None:
"""Test successfully restoring a backup."""
backup = zigpy.backups.NetworkBackup()
with patch.object(app_controller.backups, "restore_backup", new=AsyncMock()) as p:
await zha_client.send_json(
{
ID: 6,
TYPE: f"{DOMAIN}/network/backups/restore",
"backup": backup.as_dict(),
}
)
msg = await zha_client.receive_json()
p.assert_called_once_with(backup)
assert "ezsp" not in backup.network_info.stack_specific
assert msg["id"] == 6
assert msg["type"] == const.TYPE_RESULT
assert msg["success"]
async def test_restore_network_backup_force_write_eui64(
app_controller, zha_client
) -> None:
"""Test successfully restoring a backup."""
backup = zigpy.backups.NetworkBackup()
with patch.object(app_controller.backups, "restore_backup", new=AsyncMock()) as p:
await zha_client.send_json(
{
ID: 6,
TYPE: f"{DOMAIN}/network/backups/restore",
"backup": backup.as_dict(),
"ezsp_force_write_eui64": True,
}
)
msg = await zha_client.receive_json()
# EUI64 will be overwritten
p.assert_called_once_with(
backup.replace(
network_info=backup.network_info.replace(
stack_specific={"ezsp": {EZSP_OVERWRITE_EUI64: True}}
)
)
)
assert msg["id"] == 6
assert msg["type"] == const.TYPE_RESULT
assert msg["success"]
@patch("zigpy.backups.NetworkBackup.from_dict", new=lambda v: v)
async def test_restore_network_backup_failure(app_controller, zha_client) -> None:
"""Test successfully restoring a backup."""
with patch.object(
app_controller.backups,
"restore_backup",
new=AsyncMock(side_effect=ValueError("Restore failed")),
) as p:
await zha_client.send_json(
{ID: 6, TYPE: f"{DOMAIN}/network/backups/restore", "backup": "a backup"}
)
msg = await zha_client.receive_json()
p.assert_called_once_with("a backup")
assert msg["id"] == 6
assert msg["type"] == const.TYPE_RESULT
assert not msg["success"]
assert msg["error"]["code"] == const.ERR_INVALID_FORMAT

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

@ -120,7 +120,9 @@ async def test_config_depreciation(hass: HomeAssistant, zha_config) -> None:
], ],
) )
@patch("homeassistant.components.zha.setup_quirks", Mock(return_value=True)) @patch("homeassistant.components.zha.setup_quirks", Mock(return_value=True))
@patch("homeassistant.components.zha.api.async_load_api", Mock(return_value=True)) @patch(
"homeassistant.components.zha.websocket_api.async_load_api", Mock(return_value=True)
)
async def test_setup_with_v3_spaces_in_uri( async def test_setup_with_v3_spaces_in_uri(
hass: HomeAssistant, path: str, cleaned_path: str hass: HomeAssistant, path: str, cleaned_path: str
) -> None: ) -> None:

842
tests/components/zha/test_websocket_api.py Normal file
View File

@ -0,0 +1,842 @@
"""Test ZHA WebSocket API."""
from binascii import unhexlify
from copy import deepcopy
from unittest.mock import AsyncMock, patch
import pytest
import voluptuous as vol
import zigpy.backups
import zigpy.profiles.zha
import zigpy.types
import zigpy.zcl.clusters.general as general
import zigpy.zcl.clusters.security as security
from homeassistant.components.websocket_api import const
from homeassistant.components.zha import DOMAIN
from homeassistant.components.zha.core.const import (
ATTR_CLUSTER_ID,
ATTR_CLUSTER_TYPE,
ATTR_ENDPOINT_ID,
ATTR_ENDPOINT_NAMES,
ATTR_IEEE,
ATTR_MANUFACTURER,
ATTR_MODEL,
ATTR_NEIGHBORS,
ATTR_QUIRK_APPLIED,
CLUSTER_TYPE_IN,
DATA_ZHA,
DATA_ZHA_GATEWAY,
EZSP_OVERWRITE_EUI64,
GROUP_ID,
GROUP_IDS,
GROUP_NAME,
)
from homeassistant.components.zha.websocket_api import (
ATTR_DURATION,
ATTR_INSTALL_CODE,
ATTR_QR_CODE,
ATTR_SOURCE_IEEE,
ID,
SERVICE_PERMIT,
TYPE,
async_load_api,
)
from homeassistant.const import ATTR_NAME, Platform
from homeassistant.core import Context, HomeAssistant
from .conftest import (
FIXTURE_GRP_ID,
FIXTURE_GRP_NAME,
SIG_EP_INPUT,
SIG_EP_OUTPUT,
SIG_EP_PROFILE,
SIG_EP_TYPE,
)
from .data import BASE_CUSTOM_CONFIGURATION, CONFIG_WITH_ALARM_OPTIONS
from tests.common import MockUser
IEEE_SWITCH_DEVICE = "01:2d:6f:00:0a:90:69:e7"
IEEE_GROUPABLE_DEVICE = "01:2d:6f:00:0a:90:69:e8"
@pytest.fixture(autouse=True)
def required_platform_only():
"""Only set up the required and required base platforms to speed up tests."""
with patch(
"homeassistant.components.zha.PLATFORMS",
(
Platform.ALARM_CONTROL_PANEL,
Platform.SELECT,
Platform.SENSOR,
Platform.SWITCH,
),
):
yield
@pytest.fixture
async def device_switch(hass, zigpy_device_mock, zha_device_joined):
"""Test ZHA switch platform."""
zigpy_device = zigpy_device_mock(
{
1: {
SIG_EP_INPUT: [general.OnOff.cluster_id, general.Basic.cluster_id],
SIG_EP_OUTPUT: [],
SIG_EP_TYPE: zigpy.profiles.zha.DeviceType.ON_OFF_SWITCH,
SIG_EP_PROFILE: zigpy.profiles.zha.PROFILE_ID,
}
},
ieee=IEEE_SWITCH_DEVICE,
)
zha_device = await zha_device_joined(zigpy_device)
zha_device.available = True
return zha_device
@pytest.fixture
async def device_ias_ace(hass, zigpy_device_mock, zha_device_joined):
"""Test alarm control panel device."""
zigpy_device = zigpy_device_mock(
{
1: {
SIG_EP_INPUT: [security.IasAce.cluster_id],
SIG_EP_OUTPUT: [],
SIG_EP_TYPE: zigpy.profiles.zha.DeviceType.IAS_ANCILLARY_CONTROL,
SIG_EP_PROFILE: zigpy.profiles.zha.PROFILE_ID,
}
},
)
zha_device = await zha_device_joined(zigpy_device)
zha_device.available = True
return zha_device
@pytest.fixture
async def device_groupable(hass, zigpy_device_mock, zha_device_joined):
"""Test ZHA light platform."""
zigpy_device = zigpy_device_mock(
{
1: {
SIG_EP_INPUT: [
general.OnOff.cluster_id,
general.Basic.cluster_id,
general.Groups.cluster_id,
],
SIG_EP_OUTPUT: [],
SIG_EP_TYPE: zigpy.profiles.zha.DeviceType.ON_OFF_SWITCH,
SIG_EP_PROFILE: zigpy.profiles.zha.PROFILE_ID,
}
},
ieee=IEEE_GROUPABLE_DEVICE,
)
zha_device = await zha_device_joined(zigpy_device)
zha_device.available = True
return zha_device
@pytest.fixture
async def zha_client(hass, hass_ws_client, device_switch, device_groupable):
"""Get ZHA WebSocket client."""
# load the ZHA API
async_load_api(hass)
return await hass_ws_client(hass)
async def test_device_clusters(hass: HomeAssistant, zha_client) -> None:
"""Test getting device cluster info."""
await zha_client.send_json(
{ID: 5, TYPE: "zha/devices/clusters", ATTR_IEEE: IEEE_SWITCH_DEVICE}
)
msg = await zha_client.receive_json()
assert len(msg["result"]) == 2
cluster_infos = sorted(msg["result"], key=lambda k: k[ID])
cluster_info = cluster_infos[0]
assert cluster_info[TYPE] == CLUSTER_TYPE_IN
assert cluster_info[ID] == 0
assert cluster_info[ATTR_NAME] == "Basic"
cluster_info = cluster_infos[1]
assert cluster_info[TYPE] == CLUSTER_TYPE_IN
assert cluster_info[ID] == 6
assert cluster_info[ATTR_NAME] == "OnOff"
async def test_device_cluster_attributes(zha_client) -> None:
"""Test getting device cluster attributes."""
await zha_client.send_json(
{
ID: 5,
TYPE: "zha/devices/clusters/attributes",
ATTR_ENDPOINT_ID: 1,
ATTR_IEEE: IEEE_SWITCH_DEVICE,
ATTR_CLUSTER_ID: 6,
ATTR_CLUSTER_TYPE: CLUSTER_TYPE_IN,
}
)
msg = await zha_client.receive_json()
attributes = msg["result"]
assert len(attributes) == 7
for attribute in attributes:
assert attribute[ID] is not None
assert attribute[ATTR_NAME] is not None
async def test_device_cluster_commands(zha_client) -> None:
"""Test getting device cluster commands."""
await zha_client.send_json(
{
ID: 5,
TYPE: "zha/devices/clusters/commands",
ATTR_ENDPOINT_ID: 1,
ATTR_IEEE: IEEE_SWITCH_DEVICE,
ATTR_CLUSTER_ID: 6,
ATTR_CLUSTER_TYPE: CLUSTER_TYPE_IN,
}
)
msg = await zha_client.receive_json()
commands = msg["result"]
assert len(commands) == 6
for command in commands:
assert command[ID] is not None
assert command[ATTR_NAME] is not None
assert command[TYPE] is not None
async def test_list_devices(zha_client) -> None:
"""Test getting ZHA devices."""
await zha_client.send_json({ID: 5, TYPE: "zha/devices"})
msg = await zha_client.receive_json()
devices = msg["result"]
assert len(devices) == 2
msg_id = 100
for device in devices:
msg_id += 1
assert device[ATTR_IEEE] is not None
assert device[ATTR_MANUFACTURER] is not None
assert device[ATTR_MODEL] is not None
assert device[ATTR_NAME] is not None
assert device[ATTR_QUIRK_APPLIED] is not None
assert device["entities"] is not None
assert device[ATTR_NEIGHBORS] is not None
assert device[ATTR_ENDPOINT_NAMES] is not None
for entity_reference in device["entities"]:
assert entity_reference[ATTR_NAME] is not None
assert entity_reference["entity_id"] is not None
await zha_client.send_json(
{ID: msg_id, TYPE: "zha/device", ATTR_IEEE: device[ATTR_IEEE]}
)
msg = await zha_client.receive_json()
device2 = msg["result"]
assert device == device2
async def test_get_zha_config(zha_client) -> None:
"""Test getting ZHA custom configuration."""
await zha_client.send_json({ID: 5, TYPE: "zha/configuration"})
msg = await zha_client.receive_json()
configuration = msg["result"]
assert configuration == BASE_CUSTOM_CONFIGURATION
async def test_get_zha_config_with_alarm(
hass: HomeAssistant, zha_client, device_ias_ace
) -> None:
"""Test getting ZHA custom configuration."""
await zha_client.send_json({ID: 5, TYPE: "zha/configuration"})
msg = await zha_client.receive_json()
configuration = msg["result"]
assert configuration == CONFIG_WITH_ALARM_OPTIONS
# test that the alarm options are not in the config when we remove the device
device_ias_ace.gateway.device_removed(device_ias_ace.device)
await hass.async_block_till_done()
await zha_client.send_json({ID: 6, TYPE: "zha/configuration"})
msg = await zha_client.receive_json()
configuration = msg["result"]
assert configuration == BASE_CUSTOM_CONFIGURATION
async def test_update_zha_config(zha_client, zigpy_app_controller) -> None:
"""Test updating ZHA custom configuration."""
configuration = deepcopy(CONFIG_WITH_ALARM_OPTIONS)
configuration["data"]["zha_options"]["default_light_transition"] = 10
with patch(
"bellows.zigbee.application.ControllerApplication.new",
return_value=zigpy_app_controller,
):
await zha_client.send_json(
{ID: 5, TYPE: "zha/configuration/update", "data": configuration["data"]}
)
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
async def test_device_not_found(zha_client) -> None:
"""Test not found response from get device API."""
await zha_client.send_json(
{ID: 6, TYPE: "zha/device", ATTR_IEEE: "28:6d:97:00:01:04:11:8c"}
)
msg = await zha_client.receive_json()
assert msg["id"] == 6
assert msg["type"] == const.TYPE_RESULT
assert not msg["success"]
assert msg["error"]["code"] == const.ERR_NOT_FOUND
async def test_list_groups(zha_client) -> None:
"""Test getting ZHA zigbee groups."""
await zha_client.send_json({ID: 7, TYPE: "zha/groups"})
msg = await zha_client.receive_json()
assert msg["id"] == 7
assert msg["type"] == const.TYPE_RESULT
groups = msg["result"]
assert len(groups) == 1
for group in groups:
assert group["group_id"] == FIXTURE_GRP_ID
assert group["name"] == FIXTURE_GRP_NAME
assert group["members"] == []
async def test_get_group(zha_client) -> None:
"""Test getting a specific ZHA zigbee group."""
await zha_client.send_json({ID: 8, TYPE: "zha/group", GROUP_ID: FIXTURE_GRP_ID})
msg = await zha_client.receive_json()
assert msg["id"] == 8
assert msg["type"] == const.TYPE_RESULT
group = msg["result"]
assert group is not None
assert group["group_id"] == FIXTURE_GRP_ID
assert group["name"] == FIXTURE_GRP_NAME
assert group["members"] == []
async def test_get_group_not_found(zha_client) -> None:
"""Test not found response from get group API."""
await zha_client.send_json({ID: 9, TYPE: "zha/group", GROUP_ID: 1_234_567})
msg = await zha_client.receive_json()
assert msg["id"] == 9
assert msg["type"] == const.TYPE_RESULT
assert not msg["success"]
assert msg["error"]["code"] == const.ERR_NOT_FOUND
async def test_list_groupable_devices(zha_client, device_groupable) -> None:
"""Test getting ZHA devices that have a group cluster."""
await zha_client.send_json({ID: 10, TYPE: "zha/devices/groupable"})
msg = await zha_client.receive_json()
assert msg["id"] == 10
assert msg["type"] == const.TYPE_RESULT
device_endpoints = msg["result"]
assert len(device_endpoints) == 1
for endpoint in device_endpoints:
assert endpoint["device"][ATTR_IEEE] == "01:2d:6f:00:0a:90:69:e8"
assert endpoint["device"][ATTR_MANUFACTURER] is not None
assert endpoint["device"][ATTR_MODEL] is not None
assert endpoint["device"][ATTR_NAME] is not None
assert endpoint["device"][ATTR_QUIRK_APPLIED] is not None
assert endpoint["device"]["entities"] is not None
assert endpoint["endpoint_id"] is not None
assert endpoint["entities"] is not None
for entity_reference in endpoint["device"]["entities"]:
assert entity_reference[ATTR_NAME] is not None
assert entity_reference["entity_id"] is not None
for entity_reference in endpoint["entities"]:
assert entity_reference["original_name"] is not None
# Make sure there are no groupable devices when the device is unavailable
# Make device unavailable
device_groupable.available = False
await zha_client.send_json({ID: 11, TYPE: "zha/devices/groupable"})
msg = await zha_client.receive_json()
assert msg["id"] == 11
assert msg["type"] == const.TYPE_RESULT
device_endpoints = msg["result"]
assert len(device_endpoints) == 0
async def test_add_group(zha_client) -> None:
"""Test adding and getting a new ZHA zigbee group."""
await zha_client.send_json({ID: 12, TYPE: "zha/group/add", GROUP_NAME: "new_group"})
msg = await zha_client.receive_json()
assert msg["id"] == 12
assert msg["type"] == const.TYPE_RESULT
added_group = msg["result"]
assert added_group["name"] == "new_group"
assert added_group["members"] == []
await zha_client.send_json({ID: 13, TYPE: "zha/groups"})
msg = await zha_client.receive_json()
assert msg["id"] == 13
assert msg["type"] == const.TYPE_RESULT
groups = msg["result"]
assert len(groups) == 2
for group in groups:
assert group["name"] == FIXTURE_GRP_NAME or group["name"] == "new_group"
async def test_remove_group(zha_client) -> None:
"""Test removing a new ZHA zigbee group."""
await zha_client.send_json({ID: 14, TYPE: "zha/groups"})
msg = await zha_client.receive_json()
assert msg["id"] == 14
assert msg["type"] == const.TYPE_RESULT
groups = msg["result"]
assert len(groups) == 1
await zha_client.send_json(
{ID: 15, TYPE: "zha/group/remove", GROUP_IDS: [FIXTURE_GRP_ID]}
)
msg = await zha_client.receive_json()
assert msg["id"] == 15
assert msg["type"] == const.TYPE_RESULT
groups_remaining = msg["result"]
assert len(groups_remaining) == 0
await zha_client.send_json({ID: 16, TYPE: "zha/groups"})
msg = await zha_client.receive_json()
assert msg["id"] == 16
assert msg["type"] == const.TYPE_RESULT
groups = msg["result"]
assert len(groups) == 0
@pytest.fixture
async def app_controller(hass, setup_zha):
"""Fixture for zigpy Application Controller."""
await setup_zha()
controller = hass.data[DATA_ZHA][DATA_ZHA_GATEWAY].application_controller
p1 = patch.object(controller, "permit")
p2 = patch.object(controller, "permit_with_key", new=AsyncMock())
with p1, p2:
yield controller
@pytest.mark.parametrize(
("params", "duration", "node"),
(
({}, 60, None),
({ATTR_DURATION: 30}, 30, None),
(
{ATTR_DURATION: 33, ATTR_IEEE: "aa:bb:cc:dd:aa:bb:cc:dd"},
33,
zigpy.types.EUI64.convert("aa:bb:cc:dd:aa:bb:cc:dd"),
),
(
{ATTR_IEEE: "aa:bb:cc:dd:aa:bb:cc:d1"},
60,
zigpy.types.EUI64.convert("aa:bb:cc:dd:aa:bb:cc:d1"),
),
),
)
async def test_permit_ha12(
hass: HomeAssistant,
app_controller,
hass_admin_user: MockUser,
params,
duration,
node,
) -> None:
"""Test permit service."""
await hass.services.async_call(
DOMAIN, SERVICE_PERMIT, params, True, Context(user_id=hass_admin_user.id)
)
assert app_controller.permit.await_count == 1
assert app_controller.permit.await_args[1]["time_s"] == duration
assert app_controller.permit.await_args[1]["node"] == node
assert app_controller.permit_with_key.call_count == 0
IC_TEST_PARAMS = (
(
{
ATTR_SOURCE_IEEE: IEEE_SWITCH_DEVICE,
ATTR_INSTALL_CODE: "5279-7BF4-A508-4DAA-8E17-12B6-1741-CA02-4051",
},
zigpy.types.EUI64.convert(IEEE_SWITCH_DEVICE),
unhexlify("52797BF4A5084DAA8E1712B61741CA024051"),
),
(
{
ATTR_SOURCE_IEEE: IEEE_SWITCH_DEVICE,
ATTR_INSTALL_CODE: "52797BF4A5084DAA8E1712B61741CA024051",
},
zigpy.types.EUI64.convert(IEEE_SWITCH_DEVICE),
unhexlify("52797BF4A5084DAA8E1712B61741CA024051"),
),
)
@pytest.mark.parametrize(("params", "src_ieee", "code"), IC_TEST_PARAMS)
async def test_permit_with_install_code(
hass: HomeAssistant,
app_controller,
hass_admin_user: MockUser,
params,
src_ieee,
code,
) -> None:
"""Test permit service with install code."""
await hass.services.async_call(
DOMAIN, SERVICE_PERMIT, params, True, Context(user_id=hass_admin_user.id)
)
assert app_controller.permit.await_count == 0
assert app_controller.permit_with_key.call_count == 1
assert app_controller.permit_with_key.await_args[1]["time_s"] == 60
assert app_controller.permit_with_key.await_args[1]["node"] == src_ieee
assert app_controller.permit_with_key.await_args[1]["code"] == code
IC_FAIL_PARAMS = (
{
# wrong install code
ATTR_SOURCE_IEEE: IEEE_SWITCH_DEVICE,
ATTR_INSTALL_CODE: "5279-7BF4-A508-4DAA-8E17-12B6-1741-CA02-4052",
},
# incorrect service params
{ATTR_INSTALL_CODE: "5279-7BF4-A508-4DAA-8E17-12B6-1741-CA02-4051"},
{ATTR_SOURCE_IEEE: IEEE_SWITCH_DEVICE},
{
# incorrect service params
ATTR_INSTALL_CODE: "5279-7BF4-A508-4DAA-8E17-12B6-1741-CA02-4051",
ATTR_QR_CODE: "Z:000D6FFFFED4163B$I:52797BF4A5084DAA8E1712B61741CA024051",
},
{
# incorrect service params
ATTR_SOURCE_IEEE: IEEE_SWITCH_DEVICE,
ATTR_QR_CODE: "Z:000D6FFFFED4163B$I:52797BF4A5084DAA8E1712B61741CA024051",
},
{
# good regex match, but bad code
ATTR_QR_CODE: "Z:000D6FFFFED4163B$I:52797BF4A5084DAA8E1712B61741CA024052"
},
{
# good aqara regex match, but bad code
ATTR_QR_CODE: (
"G$M:751$S:357S00001579$D:000000000F350FFD%Z$A:04CF8CDF"
"3C3C3C3C$I:52797BF4A5084DAA8E1712B61741CA024052"
)
},
# good consciot regex match, but bad code
{ATTR_QR_CODE: "000D6FFFFED4163B|52797BF4A5084DAA8E1712B61741CA024052"},
)
@pytest.mark.parametrize("params", IC_FAIL_PARAMS)
async def test_permit_with_install_code_fail(
hass: HomeAssistant, app_controller, hass_admin_user: MockUser, params
) -> None:
"""Test permit service with install code."""
with pytest.raises(vol.Invalid):
await hass.services.async_call(
DOMAIN, SERVICE_PERMIT, params, True, Context(user_id=hass_admin_user.id)
)
assert app_controller.permit.await_count == 0
assert app_controller.permit_with_key.call_count == 0
IC_QR_CODE_TEST_PARAMS = (
(
{ATTR_QR_CODE: "000D6FFFFED4163B|52797BF4A5084DAA8E1712B61741CA024051"},
zigpy.types.EUI64.convert("00:0D:6F:FF:FE:D4:16:3B"),
unhexlify("52797BF4A5084DAA8E1712B61741CA024051"),
),
(
{ATTR_QR_CODE: "Z:000D6FFFFED4163B$I:52797BF4A5084DAA8E1712B61741CA024051"},
zigpy.types.EUI64.convert("00:0D:6F:FF:FE:D4:16:3B"),
unhexlify("52797BF4A5084DAA8E1712B61741CA024051"),
),
(
{
ATTR_QR_CODE: (
"G$M:751$S:357S00001579$D:000000000F350FFD%Z$A:04CF8CDF"
"3C3C3C3C$I:52797BF4A5084DAA8E1712B61741CA024051"
)
},
zigpy.types.EUI64.convert("04:CF:8C:DF:3C:3C:3C:3C"),
unhexlify("52797BF4A5084DAA8E1712B61741CA024051"),
),
)
@pytest.mark.parametrize(("params", "src_ieee", "code"), IC_QR_CODE_TEST_PARAMS)
async def test_permit_with_qr_code(
hass: HomeAssistant,
app_controller,
hass_admin_user: MockUser,
params,
src_ieee,
code,
) -> None:
"""Test permit service with install code from qr code."""
await hass.services.async_call(
DOMAIN, SERVICE_PERMIT, params, True, Context(user_id=hass_admin_user.id)
)
assert app_controller.permit.await_count == 0
assert app_controller.permit_with_key.call_count == 1
assert app_controller.permit_with_key.await_args[1]["time_s"] == 60
assert app_controller.permit_with_key.await_args[1]["node"] == src_ieee
assert app_controller.permit_with_key.await_args[1]["code"] == code
@pytest.mark.parametrize(("params", "src_ieee", "code"), IC_QR_CODE_TEST_PARAMS)
async def test_ws_permit_with_qr_code(
app_controller, zha_client, params, src_ieee, code
) -> None:
"""Test permit service with install code from qr code."""
await zha_client.send_json(
{ID: 14, TYPE: f"{DOMAIN}/devices/{SERVICE_PERMIT}", **params}
)
msg = await zha_client.receive_json()
assert msg["id"] == 14
assert msg["type"] == const.TYPE_RESULT
assert msg["success"]
assert app_controller.permit.await_count == 0
assert app_controller.permit_with_key.call_count == 1
assert app_controller.permit_with_key.await_args[1]["time_s"] == 60
assert app_controller.permit_with_key.await_args[1]["node"] == src_ieee
assert app_controller.permit_with_key.await_args[1]["code"] == code
@pytest.mark.parametrize("params", IC_FAIL_PARAMS)
async def test_ws_permit_with_install_code_fail(
app_controller, zha_client, params
) -> None:
"""Test permit ws service with install code."""
await zha_client.send_json(
{ID: 14, TYPE: f"{DOMAIN}/devices/{SERVICE_PERMIT}", **params}
)
msg = await zha_client.receive_json()
assert msg["id"] == 14
assert msg["type"] == const.TYPE_RESULT
assert msg["success"] is False
assert app_controller.permit.await_count == 0
assert app_controller.permit_with_key.call_count == 0
@pytest.mark.parametrize(
("params", "duration", "node"),
(
({}, 60, None),
({ATTR_DURATION: 30}, 30, None),
(
{ATTR_DURATION: 33, ATTR_IEEE: "aa:bb:cc:dd:aa:bb:cc:dd"},
33,
zigpy.types.EUI64.convert("aa:bb:cc:dd:aa:bb:cc:dd"),
),
(
{ATTR_IEEE: "aa:bb:cc:dd:aa:bb:cc:d1"},
60,
zigpy.types.EUI64.convert("aa:bb:cc:dd:aa:bb:cc:d1"),
),
),
)
async def test_ws_permit_ha12(
app_controller, zha_client, params, duration, node
) -> None:
"""Test permit ws service."""
await zha_client.send_json(
{ID: 14, TYPE: f"{DOMAIN}/devices/{SERVICE_PERMIT}", **params}
)
msg = await zha_client.receive_json()
assert msg["id"] == 14
assert msg["type"] == const.TYPE_RESULT
assert msg["success"]
assert app_controller.permit.await_count == 1
assert app_controller.permit.await_args[1]["time_s"] == duration
assert app_controller.permit.await_args[1]["node"] == node
assert app_controller.permit_with_key.call_count == 0
async def test_get_network_settings(app_controller, zha_client) -> None:
"""Test current network settings are returned."""
await app_controller.backups.create_backup()
await zha_client.send_json({ID: 6, TYPE: f"{DOMAIN}/network/settings"})
msg = await zha_client.receive_json()
assert msg["id"] == 6
assert msg["type"] == const.TYPE_RESULT
assert msg["success"]
assert "radio_type" in msg["result"]
assert "network_info" in msg["result"]["settings"]
async def test_list_network_backups(app_controller, zha_client) -> None:
"""Test backups are serialized."""
await app_controller.backups.create_backup()
await zha_client.send_json({ID: 6, TYPE: f"{DOMAIN}/network/backups/list"})
msg = await zha_client.receive_json()
assert msg["id"] == 6
assert msg["type"] == const.TYPE_RESULT
assert msg["success"]
assert "network_info" in msg["result"][0]
async def test_create_network_backup(app_controller, zha_client) -> None:
"""Test creating backup."""
assert not app_controller.backups.backups
await zha_client.send_json({ID: 6, TYPE: f"{DOMAIN}/network/backups/create"})
msg = await zha_client.receive_json()
assert len(app_controller.backups.backups) == 1
assert msg["id"] == 6
assert msg["type"] == const.TYPE_RESULT
assert msg["success"]
assert "backup" in msg["result"] and "is_complete" in msg["result"]
async def test_restore_network_backup_success(app_controller, zha_client) -> None:
"""Test successfully restoring a backup."""
backup = zigpy.backups.NetworkBackup()
with patch.object(app_controller.backups, "restore_backup", new=AsyncMock()) as p:
await zha_client.send_json(
{
ID: 6,
TYPE: f"{DOMAIN}/network/backups/restore",
"backup": backup.as_dict(),
}
)
msg = await zha_client.receive_json()
p.assert_called_once_with(backup)
assert "ezsp" not in backup.network_info.stack_specific
assert msg["id"] == 6
assert msg["type"] == const.TYPE_RESULT
assert msg["success"]
async def test_restore_network_backup_force_write_eui64(
app_controller, zha_client
) -> None:
"""Test successfully restoring a backup."""
backup = zigpy.backups.NetworkBackup()
with patch.object(app_controller.backups, "restore_backup", new=AsyncMock()) as p:
await zha_client.send_json(
{
ID: 6,
TYPE: f"{DOMAIN}/network/backups/restore",
"backup": backup.as_dict(),
"ezsp_force_write_eui64": True,
}
)
msg = await zha_client.receive_json()
# EUI64 will be overwritten
p.assert_called_once_with(
backup.replace(
network_info=backup.network_info.replace(
stack_specific={"ezsp": {EZSP_OVERWRITE_EUI64: True}}
)
)
)
assert msg["id"] == 6
assert msg["type"] == const.TYPE_RESULT
assert msg["success"]
@patch("zigpy.backups.NetworkBackup.from_dict", new=lambda v: v)
async def test_restore_network_backup_failure(app_controller, zha_client) -> None:
"""Test successfully restoring a backup."""
with patch.object(
app_controller.backups,
"restore_backup",
new=AsyncMock(side_effect=ValueError("Restore failed")),
) as p:
await zha_client.send_json(
{ID: 6, TYPE: f"{DOMAIN}/network/backups/restore", "backup": "a backup"}
)
msg = await zha_client.receive_json()
p.assert_called_once_with("a backup")
assert msg["id"] == 6
assert msg["type"] == const.TYPE_RESULT
assert not msg["success"]
assert msg["error"]["code"] == const.ERR_INVALID_FORMAT