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

Update ZHA API to be device oriented (#20990)

Browse Source 
* update cluster API

* swap to device focused API

* update test
This commit is contained in:
David F. Mulcahey 2019-02-12 15:05:02 -05:00 committed by GitHub
parent 6b46ed850b
commit 80442e655d
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 101 additions and 161 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -9,12 +9,11 @@ import logging
import voluptuous as vol
from homeassistant.components import websocket_api
from homeassistant.const import ATTR_ENTITY_ID
import homeassistant.helpers.config_validation as cv
from .core.const import (
DOMAIN, ATTR_CLUSTER_ID, ATTR_CLUSTER_TYPE, ATTR_ATTRIBUTE, ATTR_VALUE,
ATTR_MANUFACTURER, ATTR_COMMAND, ATTR_COMMAND_TYPE, ATTR_ARGS, IN, OUT,
CLIENT_COMMANDS, SERVER_COMMANDS, SERVER, NAME)
CLIENT_COMMANDS, SERVER_COMMANDS, SERVER, NAME, ATTR_ENDPOINT_ID)
_LOGGER = logging.getLogger(__name__)
@ -32,7 +31,6 @@ SERVICE_PERMIT = 'permit'
SERVICE_REMOVE = 'remove'
SERVICE_SET_ZIGBEE_CLUSTER_ATTRIBUTE = 'set_zigbee_cluster_attribute'
SERVICE_ISSUE_ZIGBEE_CLUSTER_COMMAND = 'issue_zigbee_cluster_command'
ZIGBEE_CLUSTER_SERVICE = 'zigbee_cluster_service'
IEEE_SERVICE = 'ieee_based_service'
SERVICE_SCHEMAS = {
@ -43,13 +41,9 @@ SERVICE_SCHEMAS = {
IEEE_SERVICE: vol.Schema({
vol.Required(ATTR_IEEE_ADDRESS): cv.string,
}),
ZIGBEE_CLUSTER_SERVICE: vol.Schema({
vol.Required(ATTR_ENTITY_ID): cv.entity_id,
vol.Required(ATTR_CLUSTER_ID): cv.positive_int,
vol.Optional(ATTR_CLUSTER_TYPE, default=IN): cv.string
}),
SERVICE_SET_ZIGBEE_CLUSTER_ATTRIBUTE: vol.Schema({
vol.Required(ATTR_ENTITY_ID): cv.entity_id,
vol.Required(ATTR_IEEE): cv.string,
vol.Required(ATTR_ENDPOINT_ID): cv.positive_int,
vol.Required(ATTR_CLUSTER_ID): cv.positive_int,
vol.Optional(ATTR_CLUSTER_TYPE, default=IN): cv.string,
vol.Required(ATTR_ATTRIBUTE): cv.positive_int,
@ -57,7 +51,8 @@ SERVICE_SCHEMAS = {
vol.Optional(ATTR_MANUFACTURER): cv.positive_int,
}),
SERVICE_ISSUE_ZIGBEE_CLUSTER_COMMAND: vol.Schema({
vol.Required(ATTR_ENTITY_ID): cv.entity_id,
vol.Required(ATTR_IEEE): cv.string,
vol.Required(ATTR_ENDPOINT_ID): cv.positive_int,
vol.Required(ATTR_CLUSTER_ID): cv.positive_int,
vol.Optional(ATTR_CLUSTER_TYPE, default=IN): cv.string,
vol.Required(ATTR_COMMAND): cv.positive_int,
@ -67,7 +62,7 @@ SERVICE_SCHEMAS = {
}),
}
WS_RECONFIGURE_NODE = 'zha/nodes/reconfigure'
WS_RECONFIGURE_NODE = 'zha/devices/reconfigure'
SCHEMA_WS_RECONFIGURE_NODE = websocket_api.BASE_COMMAND_MESSAGE_SCHEMA.extend({
vol.Required(TYPE): WS_RECONFIGURE_NODE,
vol.Required(ATTR_IEEE): str
@ -78,44 +73,39 @@ SCHEMA_WS_LIST_DEVICES = websocket_api.BASE_COMMAND_MESSAGE_SCHEMA.extend({
vol.Required(TYPE): WS_DEVICES,
})
WS_ENTITIES_BY_IEEE = 'zha/entities'
SCHEMA_WS_LIST = websocket_api.BASE_COMMAND_MESSAGE_SCHEMA.extend({
vol.Required(TYPE): WS_ENTITIES_BY_IEEE,
})
WS_ENTITY_CLUSTERS = 'zha/entities/clusters'
WS_DEVICE_CLUSTERS = 'zha/devices/clusters'
SCHEMA_WS_CLUSTERS = websocket_api.BASE_COMMAND_MESSAGE_SCHEMA.extend({
vol.Required(TYPE): WS_ENTITY_CLUSTERS,
vol.Required(ATTR_ENTITY_ID): cv.entity_id,
vol.Required(TYPE): WS_DEVICE_CLUSTERS,
vol.Required(ATTR_IEEE): str
})
WS_ENTITY_CLUSTER_ATTRIBUTES = 'zha/entities/clusters/attributes'
WS_DEVICE_CLUSTER_ATTRIBUTES = 'zha/devices/clusters/attributes'
SCHEMA_WS_CLUSTER_ATTRIBUTES = \
websocket_api.BASE_COMMAND_MESSAGE_SCHEMA.extend({
vol.Required(TYPE): WS_ENTITY_CLUSTER_ATTRIBUTES,
vol.Required(ATTR_ENTITY_ID): cv.entity_id,
vol.Required(TYPE): WS_DEVICE_CLUSTER_ATTRIBUTES,
vol.Required(ATTR_IEEE): str,
vol.Required(ATTR_ENDPOINT_ID): int,
vol.Required(ATTR_CLUSTER_ID): int,
vol.Required(ATTR_CLUSTER_TYPE): str
})
WS_READ_CLUSTER_ATTRIBUTE = 'zha/entities/clusters/attributes/value'
WS_READ_CLUSTER_ATTRIBUTE = 'zha/devices/clusters/attributes/value'
SCHEMA_WS_READ_CLUSTER_ATTRIBUTE = \
websocket_api.BASE_COMMAND_MESSAGE_SCHEMA.extend({
vol.Required(TYPE): WS_READ_CLUSTER_ATTRIBUTE,
vol.Required(ATTR_ENTITY_ID): cv.entity_id,
vol.Required(ATTR_IEEE): str,
vol.Required(ATTR_ENDPOINT_ID): int,
vol.Required(ATTR_CLUSTER_ID): int,
vol.Required(ATTR_CLUSTER_TYPE): str,
vol.Required(ATTR_ATTRIBUTE): int,
vol.Optional(ATTR_MANUFACTURER): object,
})
WS_ENTITY_CLUSTER_COMMANDS = 'zha/entities/clusters/commands'
WS_DEVICE_CLUSTER_COMMANDS = 'zha/devices/clusters/commands'
SCHEMA_WS_CLUSTER_COMMANDS = websocket_api.BASE_COMMAND_MESSAGE_SCHEMA.extend({
vol.Required(TYPE): WS_ENTITY_CLUSTER_COMMANDS,
vol.Required(ATTR_ENTITY_ID): cv.entity_id,
vol.Required(TYPE): WS_DEVICE_CLUSTER_COMMANDS,
vol.Required(ATTR_IEEE): str,
vol.Required(ATTR_ENDPOINT_ID): int,
vol.Required(ATTR_CLUSTER_ID): int,
vol.Required(ATTR_CLUSTER_TYPE): str
})
@ -145,18 +135,18 @@ def async_load_api(hass, application_controller, zha_gateway):
async def set_zigbee_cluster_attributes(service):
"""Set zigbee attribute for cluster on zha entity."""
entity_id = service.data.get(ATTR_ENTITY_ID)
ieee = service.data.get(ATTR_IEEE)
endpoint_id = service.data.get(ATTR_ENDPOINT_ID)
cluster_id = service.data.get(ATTR_CLUSTER_ID)
cluster_type = service.data.get(ATTR_CLUSTER_TYPE)
attribute = service.data.get(ATTR_ATTRIBUTE)
value = service.data.get(ATTR_VALUE)
manufacturer = service.data.get(ATTR_MANUFACTURER) or None
entity_ref = zha_gateway.get_entity_reference(entity_id)
zha_device = zha_gateway.get_device(ieee)
response = None
if entity_ref is not None:
response = await entity_ref.zha_device.write_zigbee_attribute(
list(entity_ref.cluster_listeners.values())[
0].cluster.endpoint.endpoint_id,
if zha_device is not None:
response = await zha_device.write_zigbee_attribute(
endpoint_id,
cluster_id,
attribute,
value,
@ -166,7 +156,7 @@ def async_load_api(hass, application_controller, zha_gateway):
_LOGGER.debug("Set attribute for: %s %s %s %s %s %s %s",
"{}: [{}]".format(ATTR_CLUSTER_ID, cluster_id),
"{}: [{}]".format(ATTR_CLUSTER_TYPE, cluster_type),
"{}: [{}]".format(ATTR_ENTITY_ID, entity_id),
"{}: [{}]".format(ATTR_ENDPOINT_ID, endpoint_id),
"{}: [{}]".format(ATTR_ATTRIBUTE, attribute),
"{}: [{}]".format(ATTR_VALUE, value),
"{}: [{}]".format(ATTR_MANUFACTURER, manufacturer),
@ -181,20 +171,19 @@ def async_load_api(hass, application_controller, zha_gateway):
async def issue_zigbee_cluster_command(service):
"""Issue command on zigbee cluster on zha entity."""
entity_id = service.data.get(ATTR_ENTITY_ID)
ieee = service.data.get(ATTR_IEEE)
endpoint_id = service.data.get(ATTR_ENDPOINT_ID)
cluster_id = service.data.get(ATTR_CLUSTER_ID)
cluster_type = service.data.get(ATTR_CLUSTER_TYPE)
command = service.data.get(ATTR_COMMAND)
command_type = service.data.get(ATTR_COMMAND_TYPE)
args = service.data.get(ATTR_ARGS)
manufacturer = service.data.get(ATTR_MANUFACTURER) or None
entity_ref = zha_gateway.get_entity_reference(entity_id)
zha_device = entity_ref.zha_device
zha_device = zha_gateway.get_device(ieee)
response = None
if entity_ref is not None:
if zha_device is not None:
response = await zha_device.issue_cluster_command(
list(entity_ref.cluster_listeners.values())[
0].cluster.endpoint.endpoint_id,
endpoint_id,
cluster_id,
command,
command_type,
@ -205,7 +194,7 @@ def async_load_api(hass, application_controller, zha_gateway):
_LOGGER.debug("Issue command for: %s %s %s %s %s %s %s %s",
"{}: [{}]".format(ATTR_CLUSTER_ID, cluster_id),
"{}: [{}]".format(ATTR_CLUSTER_TYPE, cluster_type),
"{}: [{}]".format(ATTR_ENTITY_ID, entity_id),
"{}: [{}]".format(ATTR_ENDPOINT_ID, endpoint_id),
"{}: [{}]".format(ATTR_COMMAND, command),
"{}: [{}]".format(ATTR_COMMAND_TYPE, command_type),
"{}: [{}]".format(ATTR_ARGS, args),
@ -256,77 +245,52 @@ def async_load_api(hass, application_controller, zha_gateway):
)
@websocket_api.async_response
async def websocket_entities_by_ieee(hass, connection, msg):
"""Return a dict of all zha entities grouped by ieee."""
entities_by_ieee = {}
for ieee, entities in zha_gateway.device_registry.items():
ieee_string = str(ieee)
entities_by_ieee[ieee_string] = []
for entity in entities:
entities_by_ieee[ieee_string].append({
ATTR_ENTITY_ID: entity.reference_id,
DEVICE_INFO: entity.device_info
})
connection.send_message(websocket_api.result_message(
msg[ID],
entities_by_ieee
))
hass.components.websocket_api.async_register_command(
WS_ENTITIES_BY_IEEE, websocket_entities_by_ieee,
SCHEMA_WS_LIST
)
@websocket_api.async_response
async def websocket_entity_clusters(hass, connection, msg):
"""Return a list of entity clusters."""
entity_id = msg[ATTR_ENTITY_ID]
entity_ref = zha_gateway.get_entity_reference(entity_id)
clusters = []
if entity_ref is not None:
for listener in entity_ref.cluster_listeners.values():
cluster = listener.cluster
in_clusters = cluster.endpoint.in_clusters.values()
out_clusters = cluster.endpoint.out_clusters.values()
if cluster in in_clusters:
clusters.append({
async def websocket_device_clusters(hass, connection, msg):
"""Return a list of device clusters."""
ieee = msg[ATTR_IEEE]
zha_device = zha_gateway.get_device(ieee)
response_clusters = []
if zha_device is not None:
clusters_by_endpoint = await zha_device.get_clusters()
for ep_id, clusters in clusters_by_endpoint.items():
for c_id, cluster in clusters[IN].items():
response_clusters.append({
TYPE: IN,
ID: cluster.cluster_id,
NAME: cluster.__class__.__name__
ID: c_id,
NAME: cluster.__class__.__name__,
'endpoint_id': ep_id
})
elif cluster in out_clusters:
clusters.append({
for c_id, cluster in clusters[OUT].items():
response_clusters.append({
TYPE: OUT,
ID: cluster.cluster_id,
NAME: cluster.__class__.__name__
ID: c_id,
NAME: cluster.__class__.__name__,
'endpoint_id': ep_id
})
connection.send_message(websocket_api.result_message(
msg[ID],
clusters
response_clusters
))
hass.components.websocket_api.async_register_command(
WS_ENTITY_CLUSTERS, websocket_entity_clusters,
WS_DEVICE_CLUSTERS, websocket_device_clusters,
SCHEMA_WS_CLUSTERS
)
@websocket_api.async_response
async def websocket_entity_cluster_attributes(hass, connection, msg):
async def websocket_device_cluster_attributes(hass, connection, msg):
"""Return a list of cluster attributes."""
entity_id = msg[ATTR_ENTITY_ID]
ieee = msg[ATTR_IEEE]
endpoint_id = msg[ATTR_ENDPOINT_ID]
cluster_id = msg[ATTR_CLUSTER_ID]
cluster_type = msg[ATTR_CLUSTER_TYPE]
ieee = msg[ATTR_IEEE]
cluster_attributes = []
entity_ref = zha_gateway.get_entity_reference(entity_id)
device = zha_gateway.get_device(ieee)
zha_device = zha_gateway.get_device(ieee)
attributes = None
if entity_ref is not None:
attributes = await device.get_cluster_attributes(
list(entity_ref.cluster_listeners.values())[
0].cluster.endpoint.endpoint_id,
if zha_device is not None:
attributes = await zha_device.get_cluster_attributes(
endpoint_id,
cluster_id,
cluster_type)
if attributes is not None:
@ -340,7 +304,7 @@ def async_load_api(hass, application_controller, zha_gateway):
_LOGGER.debug("Requested attributes for: %s %s %s %s",
"{}: [{}]".format(ATTR_CLUSTER_ID, cluster_id),
"{}: [{}]".format(ATTR_CLUSTER_TYPE, cluster_type),
"{}: [{}]".format(ATTR_ENTITY_ID, entity_id),
"{}: [{}]".format(ATTR_ENDPOINT_ID, endpoint_id),
"{}: [{}]".format(RESPONSE, cluster_attributes)
)
@ -350,25 +314,23 @@ def async_load_api(hass, application_controller, zha_gateway):
))
hass.components.websocket_api.async_register_command(
WS_ENTITY_CLUSTER_ATTRIBUTES, websocket_entity_cluster_attributes,
WS_DEVICE_CLUSTER_ATTRIBUTES, websocket_device_cluster_attributes,
SCHEMA_WS_CLUSTER_ATTRIBUTES
)
@websocket_api.async_response
async def websocket_entity_cluster_commands(hass, connection, msg):
async def websocket_device_cluster_commands(hass, connection, msg):
"""Return a list of cluster commands."""
entity_id = msg[ATTR_ENTITY_ID]
cluster_id = msg[ATTR_CLUSTER_ID]
cluster_type = msg[ATTR_CLUSTER_TYPE]
ieee = msg[ATTR_IEEE]
entity_ref = zha_gateway.get_entity_reference(entity_id)
device = zha_gateway.get_device(ieee)
endpoint_id = msg[ATTR_ENDPOINT_ID]
zha_device = zha_gateway.get_device(ieee)
cluster_commands = []
commands = None
if entity_ref is not None:
commands = await device.get_cluster_commands(
list(entity_ref.cluster_listeners.values())[
0].cluster.endpoint.endpoint_id,
if zha_device is not None:
commands = await zha_device.get_cluster_commands(
endpoint_id,
cluster_id,
cluster_type)
@ -392,7 +354,7 @@ def async_load_api(hass, application_controller, zha_gateway):
_LOGGER.debug("Requested commands for: %s %s %s %s",
"{}: [{}]".format(ATTR_CLUSTER_ID, cluster_id),
"{}: [{}]".format(ATTR_CLUSTER_TYPE, cluster_type),
"{}: [{}]".format(ATTR_ENTITY_ID, entity_id),
"{}: [{}]".format(ATTR_ENDPOINT_ID, endpoint_id),
"{}: [{}]".format(RESPONSE, cluster_commands)
)
@ -402,31 +364,24 @@ def async_load_api(hass, application_controller, zha_gateway):
))
hass.components.websocket_api.async_register_command(
WS_ENTITY_CLUSTER_COMMANDS, websocket_entity_cluster_commands,
WS_DEVICE_CLUSTER_COMMANDS, websocket_device_cluster_commands,
SCHEMA_WS_CLUSTER_COMMANDS
)
@websocket_api.async_response
async def websocket_read_zigbee_cluster_attributes(hass, connection, msg):
"""Read zigbee attribute for cluster on zha entity."""
entity_id = msg[ATTR_ENTITY_ID]
ieee = msg[ATTR_IEEE]
endpoint_id = msg[ATTR_ENDPOINT_ID]
cluster_id = msg[ATTR_CLUSTER_ID]
cluster_type = msg[ATTR_CLUSTER_TYPE]
attribute = msg[ATTR_ATTRIBUTE]
entity_ref = zha_gateway.get_entity_reference(entity_id)
manufacturer = msg.get(ATTR_MANUFACTURER) or None
zha_device = zha_gateway.get_device(ieee)
success = failure = None
clusters = []
if cluster_type == IN:
clusters = \
list(entity_ref.cluster_listeners.values())[
0].cluster.endpoint.in_clusters
else:
clusters = \
list(entity_ref.cluster_listeners.values())[
0].cluster.endpoint.out_clusters
cluster = clusters[cluster_id]
if entity_ref is not None:
if zha_device is not None:
cluster = await zha_device.get_cluster(
endpoint_id, cluster_id, cluster_type=cluster_type)
success, failure = await cluster.read_attributes(
[attribute],
allow_cache=False,
@ -436,7 +391,7 @@ def async_load_api(hass, application_controller, zha_gateway):
_LOGGER.debug("Read attribute for: %s %s %s %s %s %s %s",
"{}: [{}]".format(ATTR_CLUSTER_ID, cluster_id),
"{}: [{}]".format(ATTR_CLUSTER_TYPE, cluster_type),
"{}: [{}]".format(ATTR_ENTITY_ID, entity_id),
"{}: [{}]".format(ATTR_ENDPOINT_ID, endpoint_id),
"{}: [{}]".format(ATTR_ATTRIBUTE, attribute),
"{}: [{}]".format(ATTR_MANUFACTURER, manufacturer),
"{}: [{}]".format(RESPONSE, str(success.get(attribute))),

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

@ -220,24 +220,24 @@ class ZHADevice:
if ep_id != 0
}
async def get_cluster(self, endpooint_id, cluster_id, cluster_type=IN):
async def get_cluster(self, endpoint_id, cluster_id, cluster_type=IN):
"""Get zigbee cluster from this entity."""
clusters = await self.get_clusters()
return clusters[endpooint_id][cluster_type][cluster_id]
return clusters[endpoint_id][cluster_type][cluster_id]
async def get_cluster_attributes(self, endpooint_id, cluster_id,
async def get_cluster_attributes(self, endpoint_id, cluster_id,
cluster_type=IN):
"""Get zigbee attributes for specified cluster."""
cluster = await self.get_cluster(endpooint_id, cluster_id,
cluster = await self.get_cluster(endpoint_id, cluster_id,
cluster_type)
if cluster is None:
return None
return cluster.attributes
async def get_cluster_commands(self, endpooint_id, cluster_id,
async def get_cluster_commands(self, endpoint_id, cluster_id,
cluster_type=IN):
"""Get zigbee commands for specified cluster."""
cluster = await self.get_cluster(endpooint_id, cluster_id,
cluster = await self.get_cluster(endpoint_id, cluster_id,
cluster_type)
if cluster is None:
return None
@ -246,12 +246,12 @@ class ZHADevice:
SERVER_COMMANDS: cluster.server_commands,
}
async def write_zigbee_attribute(self, endpooint_id, cluster_id,
async def write_zigbee_attribute(self, endpoint_id, cluster_id,
attribute, value, cluster_type=IN,
manufacturer=None):
"""Write a value to a zigbee attribute for a cluster in this entity."""
cluster = await self.get_cluster(
endpooint_id, cluster_id, cluster_type)
endpoint_id, cluster_id, cluster_type)
if cluster is None:
return None
@ -266,7 +266,7 @@ class ZHADevice:
value,
attribute,
cluster_id,
endpooint_id,
endpoint_id,
response
)
return response
@ -276,17 +276,17 @@ class ZHADevice:
'{}: {}'.format(ATTR_VALUE, value),
'{}: {}'.format(ATTR_ATTRIBUTE, attribute),
'{}: {}'.format(ATTR_CLUSTER_ID, cluster_id),
'{}: {}'.format(ATTR_ENDPOINT_ID, endpooint_id),
'{}: {}'.format(ATTR_ENDPOINT_ID, endpoint_id),
exc
)
return None
async def issue_cluster_command(self, endpooint_id, cluster_id, command,
async def issue_cluster_command(self, endpoint_id, cluster_id, command,
command_type, args, cluster_type=IN,
manufacturer=None):
"""Issue a command against specified zigbee cluster on this entity."""
cluster = await self.get_cluster(
endpooint_id, cluster_id, cluster_type)
endpoint_id, cluster_id, cluster_type)
if cluster is None:
return None
response = None
@ -305,6 +305,6 @@ class ZHADevice:
'{}: {}'.format(ATTR_ARGS, args),
'{}: {}'.format(ATTR_CLUSTER_ID, cluster_type),
'{}: {}'.format(ATTR_MANUFACTURER, manufacturer),
'{}: {}'.format(ATTR_ENDPOINT_ID, endpooint_id)
'{}: {}'.format(ATTR_ENDPOINT_ID, endpoint_id)
)
return response

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

@ -1,16 +1,15 @@
"""Test ZHA API."""
from unittest.mock import Mock
import pytest
from homeassistant.const import ATTR_ENTITY_ID
from homeassistant.components.switch import DOMAIN
from homeassistant.components.zha.api import (
async_load_api, WS_ENTITIES_BY_IEEE, WS_ENTITY_CLUSTERS, ATTR_IEEE, TYPE,
ID, WS_ENTITY_CLUSTER_ATTRIBUTES, WS_ENTITY_CLUSTER_COMMANDS,
async_load_api, WS_DEVICE_CLUSTERS, ATTR_IEEE, TYPE,
ID, WS_DEVICE_CLUSTER_ATTRIBUTES, WS_DEVICE_CLUSTER_COMMANDS,
WS_DEVICES
)
from homeassistant.components.zha.core.const import (
ATTR_CLUSTER_ID, ATTR_CLUSTER_TYPE, IN, IEEE, MODEL, NAME, QUIRK_APPLIED,
ATTR_MANUFACTURER
ATTR_MANUFACTURER, ATTR_ENDPOINT_ID
)
from .common import async_init_zigpy_device
@ -35,25 +34,11 @@ async def zha_client(hass, config_entry, zha_gateway, hass_ws_client):
return await hass_ws_client(hass)
async def test_entities_by_ieee(hass, config_entry, zha_gateway, zha_client):
"""Test getting entity refs by ieee address."""
async def test_device_clusters(hass, config_entry, zha_gateway, zha_client):
"""Test getting device cluster info."""
await zha_client.send_json({
ID: 5,
TYPE: WS_ENTITIES_BY_IEEE,
})
msg = await zha_client.receive_json()
assert '00:0d:6f:00:0a:90:69:e7' in msg['result']
assert len(msg['result']['00:0d:6f:00:0a:90:69:e7']) == 2
async def test_entity_clusters(hass, config_entry, zha_gateway, zha_client):
"""Test getting entity cluster info."""
await zha_client.send_json({
ID: 5,
TYPE: WS_ENTITY_CLUSTERS,
ATTR_ENTITY_ID: 'switch.fakemanufacturer_fakemodel_0a9069e7_1_6',
TYPE: WS_DEVICE_CLUSTERS,
ATTR_IEEE: '00:0d:6f:00:0a:90:69:e7'
})
@ -68,13 +53,13 @@ async def test_entity_clusters(hass, config_entry, zha_gateway, zha_client):
assert cluster_info[NAME] == 'OnOff'
async def test_entity_cluster_attributes(
async def test_device_cluster_attributes(
hass, config_entry, zha_gateway, zha_client):
"""Test getting entity cluster attributes."""
"""Test getting device cluster attributes."""
await zha_client.send_json({
ID: 5,
TYPE: WS_ENTITY_CLUSTER_ATTRIBUTES,
ATTR_ENTITY_ID: 'switch.fakemanufacturer_fakemodel_0a9069e7_1_6',
TYPE: WS_DEVICE_CLUSTER_ATTRIBUTES,
ATTR_ENDPOINT_ID: 1,
ATTR_IEEE: '00:0d:6f:00:0a:90:69:e7',
ATTR_CLUSTER_ID: 6,
ATTR_CLUSTER_TYPE: IN
@ -90,13 +75,13 @@ async def test_entity_cluster_attributes(
assert attribute[NAME] is not None
async def test_entity_cluster_commands(
async def test_device_cluster_commands(
hass, config_entry, zha_gateway, zha_client):
"""Test getting entity cluster commands."""
"""Test getting device cluster commands."""
await zha_client.send_json({
ID: 5,
TYPE: WS_ENTITY_CLUSTER_COMMANDS,
ATTR_ENTITY_ID: 'switch.fakemanufacturer_fakemodel_0a9069e7_1_6',
TYPE: WS_DEVICE_CLUSTER_COMMANDS,
ATTR_ENDPOINT_ID: 1,
ATTR_IEEE: '00:0d:6f:00:0a:90:69:e7',
ATTR_CLUSTER_ID: 6,
ATTR_CLUSTER_TYPE: IN