Fix ZHA configuration APIs (#81874)

* Fix ZHA configuration loading and saving issues

* add tests
This commit is contained in:
David F. Mulcahey 2022-11-13 08:30:16 -05:00 committed by GitHub
parent e4ecaa433a
commit ebffe0f33b
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
4 changed files with 239 additions and 3 deletions

View file

@ -1090,11 +1090,17 @@ async def websocket_update_zha_configuration(
): ):
data_to_save[CUSTOM_CONFIGURATION][section].pop(entry) data_to_save[CUSTOM_CONFIGURATION][section].pop(entry)
# remove entire section block if empty # remove entire section block if empty
if not data_to_save[CUSTOM_CONFIGURATION][section]: if (
not data_to_save[CUSTOM_CONFIGURATION].get(section)
and section in data_to_save[CUSTOM_CONFIGURATION]
):
data_to_save[CUSTOM_CONFIGURATION].pop(section) data_to_save[CUSTOM_CONFIGURATION].pop(section)
# remove entire custom_configuration block if empty # remove entire custom_configuration block if empty
if not data_to_save[CUSTOM_CONFIGURATION]: if (
not data_to_save.get(CUSTOM_CONFIGURATION)
and CUSTOM_CONFIGURATION in data_to_save
):
data_to_save.pop(CUSTOM_CONFIGURATION) data_to_save.pop(CUSTOM_CONFIGURATION)
_LOGGER.info( _LOGGER.info(

View file

@ -221,11 +221,13 @@ def async_get_zha_config_value(
) )
def async_cluster_exists(hass, cluster_id): def async_cluster_exists(hass, cluster_id, skip_coordinator=True):
"""Determine if a device containing the specified in cluster is paired.""" """Determine if a device containing the specified in cluster is paired."""
zha_gateway = hass.data[DATA_ZHA][DATA_ZHA_GATEWAY] zha_gateway = hass.data[DATA_ZHA][DATA_ZHA_GATEWAY]
zha_devices = zha_gateway.devices.values() zha_devices = zha_gateway.devices.values()
for zha_device in zha_devices: for zha_device in zha_devices:
if skip_coordinator and zha_device.is_coordinator:
continue
clusters_by_endpoint = zha_device.async_get_clusters() clusters_by_endpoint = zha_device.async_get_clusters()
for clusters in clusters_by_endpoint.values(): for clusters in clusters_by_endpoint.values():
if ( if (

View file

@ -0,0 +1,153 @@
"""Test data for ZHA API tests."""
BASE_CUSTOM_CONFIGURATION = {
"schemas": {
"zha_options": [
{
"type": "integer",
"valueMin": 0,
"name": "default_light_transition",
"optional": True,
"default": 0,
},
{
"type": "boolean",
"name": "enhanced_light_transition",
"required": True,
"default": False,
},
{
"type": "boolean",
"name": "light_transitioning_flag",
"required": True,
"default": True,
},
{
"type": "boolean",
"name": "always_prefer_xy_color_mode",
"required": True,
"default": True,
},
{
"type": "boolean",
"name": "enable_identify_on_join",
"required": True,
"default": True,
},
{
"type": "integer",
"valueMin": 0,
"name": "consider_unavailable_mains",
"optional": True,
"default": 7200,
},
{
"type": "integer",
"valueMin": 0,
"name": "consider_unavailable_battery",
"optional": True,
"default": 21600,
},
]
},
"data": {
"zha_options": {
"enhanced_light_transition": True,
"default_light_transition": 0,
"light_transitioning_flag": True,
"always_prefer_xy_color_mode": True,
"enable_identify_on_join": True,
"consider_unavailable_mains": 7200,
"consider_unavailable_battery": 21600,
}
},
}
CONFIG_WITH_ALARM_OPTIONS = {
"schemas": {
"zha_options": [
{
"type": "integer",
"valueMin": 0,
"name": "default_light_transition",
"optional": True,
"default": 0,
},
{
"type": "boolean",
"name": "enhanced_light_transition",
"required": True,
"default": False,
},
{
"type": "boolean",
"name": "light_transitioning_flag",
"required": True,
"default": True,
},
{
"type": "boolean",
"name": "always_prefer_xy_color_mode",
"required": True,
"default": True,
},
{
"type": "boolean",
"name": "enable_identify_on_join",
"required": True,
"default": True,
},
{
"type": "integer",
"valueMin": 0,
"name": "consider_unavailable_mains",
"optional": True,
"default": 7200,
},
{
"type": "integer",
"valueMin": 0,
"name": "consider_unavailable_battery",
"optional": True,
"default": 21600,
},
],
"zha_alarm_options": [
{
"type": "string",
"name": "alarm_master_code",
"required": True,
"default": "1234",
},
{
"type": "integer",
"valueMin": 0,
"name": "alarm_failed_tries",
"required": True,
"default": 3,
},
{
"type": "boolean",
"name": "alarm_arm_requires_code",
"required": True,
"default": False,
},
],
},
"data": {
"zha_options": {
"enhanced_light_transition": True,
"default_light_transition": 0,
"light_transitioning_flag": True,
"always_prefer_xy_color_mode": True,
"enable_identify_on_join": True,
"consider_unavailable_mains": 7200,
"consider_unavailable_battery": 21600,
},
"zha_alarm_options": {
"alarm_arm_requires_code": False,
"alarm_master_code": "4321",
"alarm_failed_tries": 2,
},
},
}

View file

@ -1,5 +1,6 @@
"""Test ZHA API.""" """Test ZHA API."""
from binascii import unhexlify from binascii import unhexlify
from copy import deepcopy
from unittest.mock import AsyncMock, patch from unittest.mock import AsyncMock, patch
import pytest import pytest
@ -8,6 +9,7 @@ import zigpy.backups
import zigpy.profiles.zha import zigpy.profiles.zha
import zigpy.types import zigpy.types
import zigpy.zcl.clusters.general as general import zigpy.zcl.clusters.general as general
import zigpy.zcl.clusters.security as security
from homeassistant.components.websocket_api import const from homeassistant.components.websocket_api import const
from homeassistant.components.zha import DOMAIN from homeassistant.components.zha import DOMAIN
@ -50,6 +52,7 @@ from .conftest import (
SIG_EP_PROFILE, SIG_EP_PROFILE,
SIG_EP_TYPE, SIG_EP_TYPE,
) )
from .data import BASE_CUSTOM_CONFIGURATION, CONFIG_WITH_ALARM_OPTIONS
IEEE_SWITCH_DEVICE = "01:2d:6f:00:0a:90:69:e7" IEEE_SWITCH_DEVICE = "01:2d:6f:00:0a:90:69:e7"
IEEE_GROUPABLE_DEVICE = "01:2d:6f:00:0a:90:69:e8" IEEE_GROUPABLE_DEVICE = "01:2d:6f:00:0a:90:69:e8"
@ -61,6 +64,7 @@ def required_platform_only():
with patch( with patch(
"homeassistant.components.zha.PLATFORMS", "homeassistant.components.zha.PLATFORMS",
( (
Platform.ALARM_CONTROL_PANEL,
Platform.SELECT, Platform.SELECT,
Platform.SENSOR, Platform.SENSOR,
Platform.SWITCH, Platform.SWITCH,
@ -89,6 +93,25 @@ async def device_switch(hass, zigpy_device_mock, zha_device_joined):
return zha_device 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 @pytest.fixture
async def device_groupable(hass, zigpy_device_mock, zha_device_joined): async def device_groupable(hass, zigpy_device_mock, zha_device_joined):
"""Test zha light platform.""" """Test zha light platform."""
@ -225,6 +248,58 @@ async def test_list_devices(zha_client):
assert device == device2 assert device == device2
async def test_get_zha_config(zha_client):
"""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, zha_client, device_ias_ace):
"""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):
"""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): async def test_device_not_found(zha_client):
"""Test not found response from get device API.""" """Test not found response from get device API."""
await zha_client.send_json( await zha_client.send_json(