evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Move remaining of ZHA imports to top level. (#28071)

Browse source 
* Move ZHA import to top level.
* ZHA tests: move imports to top level.
This commit is contained in:
Alexei Chetroi 2019-10-21 19:30:56 -04:00 • committed by GitHub
parent fc8920646b
commit fe7c45b363
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
20 changed files with 261 additions and 272 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
homeassistant/components/zha/api.py
View file

@ -5,6 +5,7 @@ import logging
import voluptuous as vol
from zigpy.types.named import EUI64
import zigpy.zdo.types as zdo_types
from homeassistant.components import websocket_api
from homeassistant.core import callback
@ -514,7 +515,6 @@ async def websocket_unbind_devices(hass, connection, msg):
async def async_binding_operation(zha_gateway, source_ieee, target_ieee, operation):
"""Create or remove a direct zigbee binding between 2 devices."""
from zigpy.zdo import types as zdo_types
source_device = zha_gateway.get_device(source_ieee)
target_device = zha_gateway.get_device(target_ieee)

14
homeassistant/components/zha/core/channels/__init__.py
View file

@ -11,6 +11,8 @@ from functools import wraps
import logging
from random import uniform
import zigpy.exceptions
from homeassistant.core import callback
from homeassistant.helpers.dispatcher import async_dispatcher_send
@ -48,8 +50,6 @@ def decorate_command(channel, command):
@wraps(command)
async def wrapper(*args, **kwds):
from zigpy.exceptions import DeliveryError
try:
result = await command(*args, **kwds)
channel.debug(
@ -61,7 +61,7 @@ def decorate_command(channel, command):
)
return result
except (DeliveryError, Timeout) as ex:
except (zigpy.exceptions.DeliveryError, Timeout) as ex:
channel.debug("command failed: %s exception: %s", command.__name__, str(ex))
return ex
@ -143,12 +143,10 @@ class ZigbeeChannel(LogMixin):
This also swallows DeliveryError exceptions that are thrown when
devices are unreachable.
"""
from zigpy.exceptions import DeliveryError
try:
res = await self.cluster.bind()
self.debug("bound '%s' cluster: %s", self.cluster.ep_attribute, res[0])
except (DeliveryError, Timeout) as ex:
except (zigpy.exceptions.DeliveryError, Timeout) as ex:
self.debug(
"Failed to bind '%s' cluster: %s", self.cluster.ep_attribute, str(ex)
)
@ -167,8 +165,6 @@ class ZigbeeChannel(LogMixin):
This also swallows DeliveryError exceptions that are thrown when
devices are unreachable.
"""
from zigpy.exceptions import DeliveryError
attr_name = self.cluster.attributes.get(attr, [attr])[0]
kwargs = {}
@ -189,7 +185,7 @@ class ZigbeeChannel(LogMixin):
reportable_change,
res,
)
except (DeliveryError, Timeout) as ex:
except (zigpy.exceptions.DeliveryError, Timeout) as ex:
self.debug(
"failed to set reporting for '%s' attr on '%s' cluster: %s",
attr_name,

13
homeassistant/components/zha/core/device.py
View file

@ -10,6 +10,10 @@ from enum import Enum
import logging
import time
import zigpy.exceptions
import zigpy.quirks
from zigpy.profiles import zha, zll
from homeassistant.core import callback
from homeassistant.helpers.dispatcher import (
async_dispatcher_connect,
@ -87,9 +91,7 @@ class ZHADevice(LogMixin):
self._unsub = async_dispatcher_connect(
self.hass, self._available_signal, self.async_initialize
)
from zigpy.quirks import CustomDevice
self.quirk_applied = isinstance(self._zigpy_device, CustomDevice)
self.quirk_applied = isinstance(self._zigpy_device, zigpy.quirks.CustomDevice)
self.quirk_class = "{}.{}".format(
self._zigpy_device.__class__.__module__,
self._zigpy_device.__class__.__name__,
@ -394,7 +396,6 @@ class ZHADevice(LogMixin):
@callback
def async_get_std_clusters(self):
"""Get ZHA and ZLL clusters for this device."""
from zigpy.profiles import zha, zll
return {
ep_id: {
@ -448,8 +449,6 @@ class ZHADevice(LogMixin):
if cluster is None:
return None
from zigpy.exceptions import DeliveryError
try:
response = await cluster.write_attributes(
{attribute: value}, manufacturer=manufacturer
@ -463,7 +462,7 @@ class ZHADevice(LogMixin):
response,
)
return response
except DeliveryError as exc:
except zigpy.exceptions.DeliveryError as exc:
self.debug(
"failed to set attribute: %s %s %s %s %s",
f"{ATTR_VALUE}: {value}",

7
homeassistant/components/zha/core/discovery.py
View file

@ -7,6 +7,9 @@ https://home-assistant.io/integrations/zha/
import logging
import zigpy.profiles
from zigpy.zcl.clusters.general import OnOff, PowerConfiguration
from homeassistant import const as ha_const
from homeassistant.components.binary_sensor import DOMAIN as BINARY_SENSOR
from homeassistant.components.sensor import DOMAIN as SENSOR
@ -52,8 +55,6 @@ def async_process_endpoint(
is_new_join,
):
"""Process an endpoint on a zigpy device."""
import zigpy.profiles
if endpoint_id == 0: # ZDO
_async_create_cluster_channel(
endpoint, zha_device, is_new_join, channel_class=ZDOChannel
@ -179,8 +180,6 @@ def _async_handle_single_cluster_matches(
hass, endpoint, zha_device, profile_clusters, device_key, is_new_join
):
"""Dispatch single cluster matches to HA components."""
from zigpy.zcl.clusters.general import OnOff, PowerConfiguration
cluster_matches = []
cluster_match_results = []
matched_power_configuration = False

6
homeassistant/components/zha/core/gateway.py
View file

@ -108,9 +108,9 @@ class ZHAGateway:
baudrate = self._config.get(CONF_BAUDRATE, DEFAULT_BAUDRATE)
radio_type = self._config_entry.data.get(CONF_RADIO_TYPE)
radio_details = RADIO_TYPES[radio_type][ZHA_GW_RADIO]()
radio = radio_details[ZHA_GW_RADIO]
self.radio_description = RADIO_TYPES[radio_type][ZHA_GW_RADIO_DESCRIPTION]
radio_details = RADIO_TYPES[radio_type]
radio = radio_details[ZHA_GW_RADIO]()
self.radio_description = radio_details[ZHA_GW_RADIO_DESCRIPTION]
await radio.connect(usb_path, baudrate)
if CONF_DATABASE in self._config:

28
homeassistant/components/zha/core/helpers.py
View file

@ -8,7 +8,15 @@ import asyncio
import collections
import logging
from zigpy.types.named import EUI64
import bellows.ezsp
import bellows.zigbee.application
import zigpy.types
import zigpy_deconz.api
import zigpy_deconz.zigbee.application
import zigpy_xbee.api
import zigpy_xbee.zigbee.application
import zigpy_zigate.api
import zigpy_zigate.zigbee.application
from homeassistant.core import callback
@ -51,25 +59,17 @@ async def safe_read(
async def check_zigpy_connection(usb_path, radio_type, database_path):
"""Test zigpy radio connection."""
if radio_type == RadioType.ezsp.name:
import bellows.ezsp
from bellows.zigbee.application import ControllerApplication
radio = bellows.ezsp.EZSP()
ControllerApplication = bellows.zigbee.application.ControllerApplication
elif radio_type == RadioType.xbee.name:
import zigpy_xbee.api
from zigpy_xbee.zigbee.application import ControllerApplication
radio = zigpy_xbee.api.XBee()
ControllerApplication = zigpy_xbee.zigbee.application.ControllerApplication
elif radio_type == RadioType.deconz.name:
import zigpy_deconz.api
from zigpy_deconz.zigbee.application import ControllerApplication
radio = zigpy_deconz.api.Deconz()
ControllerApplication = zigpy_deconz.zigbee.application.ControllerApplication
elif radio_type == RadioType.zigate.name:
import zigpy_zigate.api
from zigpy_zigate.zigbee.application import ControllerApplication
radio = zigpy_zigate.api.ZiGate()
ControllerApplication = zigpy_zigate.zigbee.application.ControllerApplication
try:
await radio.connect(usb_path, DEFAULT_BAUDRATE)
controller = ControllerApplication(radio, database_path)
@ -138,7 +138,7 @@ async def async_get_zha_device(hass, device_id):
registry_device = device_registry.async_get(device_id)
zha_gateway = hass.data[DATA_ZHA][DATA_ZHA_GATEWAY]
ieee_address = list(list(registry_device.identifiers)[0])[1]
ieee = EUI64.convert(ieee_address)
ieee = zigpy.types.EUI64.convert(ieee_address)
return zha_gateway.devices[ieee]

136
homeassistant/components/zha/core/registries.py
View file

@ -6,6 +6,18 @@ https://home-assistant.io/integrations/zha/
"""
import collections
import bellows.ezsp
import bellows.zigbee.application
import zigpy.profiles.zha
import zigpy.profiles.zll
import zigpy.zcl as zcl
import zigpy_deconz.api
import zigpy_deconz.zigbee.application
import zigpy_xbee.api
import zigpy_xbee.zigbee.application
import zigpy_zigate.api
import zigpy_zigate.zigbee.application
from homeassistant.components.binary_sensor import DOMAIN as BINARY_SENSOR
from homeassistant.components.device_tracker import DOMAIN as DEVICE_TRACKER
from homeassistant.components.fan import DOMAIN as FAN
@ -14,6 +26,8 @@ from homeassistant.components.lock import DOMAIN as LOCK
from homeassistant.components.sensor import DOMAIN as SENSOR
from homeassistant.components.switch import DOMAIN as SWITCH
# importing channels updates registries
from . import channels # noqa pylint: disable=wrong-import-position,unused-import
from .const import (
CONTROLLER,
SENSOR_ACCELERATION,
@ -63,9 +77,6 @@ COMPONENT_CLUSTERS = {
ZIGBEE_CHANNEL_REGISTRY = DictRegistry()
# importing channels updates registries
from . import channels # noqa pylint: disable=wrong-import-position,unused-import
def establish_device_mappings():
"""Establish mappings between ZCL objects and HA ZHA objects.
@ -73,56 +84,27 @@ def establish_device_mappings():
These cannot be module level, as importing bellows must be done in a
in a function.
"""
from zigpy import zcl
from zigpy.profiles import zha, zll
def get_ezsp_radio():
import bellows.ezsp
from bellows.zigbee.application import ControllerApplication
return {ZHA_GW_RADIO: bellows.ezsp.EZSP(), CONTROLLER: ControllerApplication}
RADIO_TYPES[RadioType.ezsp.name] = {
ZHA_GW_RADIO: get_ezsp_radio,
ZHA_GW_RADIO: bellows.ezsp.EZSP,
CONTROLLER: bellows.zigbee.application.ControllerApplication,
ZHA_GW_RADIO_DESCRIPTION: "EZSP",
}
def get_deconz_radio():
import zigpy_deconz.api
from zigpy_deconz.zigbee.application import ControllerApplication
return {
ZHA_GW_RADIO: zigpy_deconz.api.Deconz(),
CONTROLLER: ControllerApplication,
}
RADIO_TYPES[RadioType.deconz.name] = {
ZHA_GW_RADIO: get_deconz_radio,
ZHA_GW_RADIO: zigpy_deconz.api.Deconz,
CONTROLLER: zigpy_deconz.zigbee.application.ControllerApplication,
ZHA_GW_RADIO_DESCRIPTION: "Deconz",
}
def get_xbee_radio():
import zigpy_xbee.api
from zigpy_xbee.zigbee.application import ControllerApplication
return {ZHA_GW_RADIO: zigpy_xbee.api.XBee(), CONTROLLER: ControllerApplication}
RADIO_TYPES[RadioType.xbee.name] = {
ZHA_GW_RADIO: get_xbee_radio,
ZHA_GW_RADIO: zigpy_xbee.api.XBee,
CONTROLLER: zigpy_xbee.zigbee.application.ControllerApplication,
ZHA_GW_RADIO_DESCRIPTION: "XBee",
}
def get_zigate_radio():
import zigpy_zigate.api
from zigpy_zigate.zigbee.application import ControllerApplication
return {
ZHA_GW_RADIO: zigpy_zigate.api.ZiGate(),
CONTROLLER: ControllerApplication,
}
RADIO_TYPES[RadioType.zigate.name] = {
ZHA_GW_RADIO: get_zigate_radio,
ZHA_GW_RADIO: zigpy_zigate.api.ZiGate,
CONTROLLER: zigpy_zigate.zigbee.application.ControllerApplication,
ZHA_GW_RADIO_DESCRIPTION: "ZiGate",
}
@ -137,33 +119,33 @@ def establish_device_mappings():
}
)
DEVICE_CLASS[zha.PROFILE_ID].update(
DEVICE_CLASS[zigpy.profiles.zha.PROFILE_ID].update(
{
SMARTTHINGS_ARRIVAL_SENSOR_DEVICE_TYPE: DEVICE_TRACKER,
zha.DeviceType.COLOR_DIMMABLE_LIGHT: LIGHT,
zha.DeviceType.COLOR_TEMPERATURE_LIGHT: LIGHT,
zha.DeviceType.DIMMABLE_BALLAST: LIGHT,
zha.DeviceType.DIMMABLE_LIGHT: LIGHT,
zha.DeviceType.DIMMABLE_PLUG_IN_UNIT: LIGHT,
zha.DeviceType.EXTENDED_COLOR_LIGHT: LIGHT,
zha.DeviceType.LEVEL_CONTROLLABLE_OUTPUT: LIGHT,
zha.DeviceType.ON_OFF_BALLAST: SWITCH,
zha.DeviceType.ON_OFF_LIGHT: LIGHT,
zha.DeviceType.ON_OFF_LIGHT_SWITCH: SWITCH,
zha.DeviceType.ON_OFF_PLUG_IN_UNIT: SWITCH,
zha.DeviceType.SMART_PLUG: SWITCH,
zigpy.profiles.zha.DeviceType.COLOR_DIMMABLE_LIGHT: LIGHT,
zigpy.profiles.zha.DeviceType.COLOR_TEMPERATURE_LIGHT: LIGHT,
zigpy.profiles.zha.DeviceType.DIMMABLE_BALLAST: LIGHT,
zigpy.profiles.zha.DeviceType.DIMMABLE_LIGHT: LIGHT,
zigpy.profiles.zha.DeviceType.DIMMABLE_PLUG_IN_UNIT: LIGHT,
zigpy.profiles.zha.DeviceType.EXTENDED_COLOR_LIGHT: LIGHT,
zigpy.profiles.zha.DeviceType.LEVEL_CONTROLLABLE_OUTPUT: LIGHT,
zigpy.profiles.zha.DeviceType.ON_OFF_BALLAST: SWITCH,
zigpy.profiles.zha.DeviceType.ON_OFF_LIGHT: LIGHT,
zigpy.profiles.zha.DeviceType.ON_OFF_LIGHT_SWITCH: SWITCH,
zigpy.profiles.zha.DeviceType.ON_OFF_PLUG_IN_UNIT: SWITCH,
zigpy.profiles.zha.DeviceType.SMART_PLUG: SWITCH,
}
)
DEVICE_CLASS[zll.PROFILE_ID].update(
DEVICE_CLASS[zigpy.profiles.zll.PROFILE_ID].update(
{
zll.DeviceType.COLOR_LIGHT: LIGHT,
zll.DeviceType.COLOR_TEMPERATURE_LIGHT: LIGHT,
zll.DeviceType.DIMMABLE_LIGHT: LIGHT,
zll.DeviceType.DIMMABLE_PLUGIN_UNIT: LIGHT,
zll.DeviceType.EXTENDED_COLOR_LIGHT: LIGHT,
zll.DeviceType.ON_OFF_LIGHT: LIGHT,
zll.DeviceType.ON_OFF_PLUGIN_UNIT: SWITCH,
zigpy.profiles.zll.DeviceType.COLOR_LIGHT: LIGHT,
zigpy.profiles.zll.DeviceType.COLOR_TEMPERATURE_LIGHT: LIGHT,
zigpy.profiles.zll.DeviceType.DIMMABLE_LIGHT: LIGHT,
zigpy.profiles.zll.DeviceType.DIMMABLE_PLUGIN_UNIT: LIGHT,
zigpy.profiles.zll.DeviceType.EXTENDED_COLOR_LIGHT: LIGHT,
zigpy.profiles.zll.DeviceType.ON_OFF_LIGHT: LIGHT,
zigpy.profiles.zll.DeviceType.ON_OFF_PLUGIN_UNIT: SWITCH,
}
)
@ -207,19 +189,21 @@ def establish_device_mappings():
}
)
zhap = zha.PROFILE_ID
REMOTE_DEVICE_TYPES[zhap].append(zha.DeviceType.COLOR_CONTROLLER)
REMOTE_DEVICE_TYPES[zhap].append(zha.DeviceType.COLOR_DIMMER_SWITCH)
REMOTE_DEVICE_TYPES[zhap].append(zha.DeviceType.COLOR_SCENE_CONTROLLER)
REMOTE_DEVICE_TYPES[zhap].append(zha.DeviceType.DIMMER_SWITCH)
REMOTE_DEVICE_TYPES[zhap].append(zha.DeviceType.NON_COLOR_CONTROLLER)
REMOTE_DEVICE_TYPES[zhap].append(zha.DeviceType.NON_COLOR_SCENE_CONTROLLER)
REMOTE_DEVICE_TYPES[zhap].append(zha.DeviceType.REMOTE_CONTROL)
REMOTE_DEVICE_TYPES[zhap].append(zha.DeviceType.SCENE_SELECTOR)
zha = zigpy.profiles.zha
REMOTE_DEVICE_TYPES[zha.PROFILE_ID].append(zha.DeviceType.COLOR_CONTROLLER)
REMOTE_DEVICE_TYPES[zha.PROFILE_ID].append(zha.DeviceType.COLOR_DIMMER_SWITCH)
REMOTE_DEVICE_TYPES[zha.PROFILE_ID].append(zha.DeviceType.COLOR_SCENE_CONTROLLER)
REMOTE_DEVICE_TYPES[zha.PROFILE_ID].append(zha.DeviceType.DIMMER_SWITCH)
REMOTE_DEVICE_TYPES[zha.PROFILE_ID].append(zha.DeviceType.NON_COLOR_CONTROLLER)
REMOTE_DEVICE_TYPES[zha.PROFILE_ID].append(
zha.DeviceType.NON_COLOR_SCENE_CONTROLLER
)
REMOTE_DEVICE_TYPES[zha.PROFILE_ID].append(zha.DeviceType.REMOTE_CONTROL)
REMOTE_DEVICE_TYPES[zha.PROFILE_ID].append(zha.DeviceType.SCENE_SELECTOR)
zllp = zll.PROFILE_ID
REMOTE_DEVICE_TYPES[zllp].append(zll.DeviceType.COLOR_CONTROLLER)
REMOTE_DEVICE_TYPES[zllp].append(zll.DeviceType.COLOR_SCENE_CONTROLLER)
REMOTE_DEVICE_TYPES[zllp].append(zll.DeviceType.CONTROL_BRIDGE)
REMOTE_DEVICE_TYPES[zllp].append(zll.DeviceType.CONTROLLER)
REMOTE_DEVICE_TYPES[zllp].append(zll.DeviceType.SCENE_CONTROLLER)
zll = zigpy.profiles.zll
REMOTE_DEVICE_TYPES[zll.PROFILE_ID].append(zll.DeviceType.COLOR_CONTROLLER)
REMOTE_DEVICE_TYPES[zll.PROFILE_ID].append(zll.DeviceType.COLOR_SCENE_CONTROLLER)
REMOTE_DEVICE_TYPES[zll.PROFILE_ID].append(zll.DeviceType.CONTROL_BRIDGE)
REMOTE_DEVICE_TYPES[zll.PROFILE_ID].append(zll.DeviceType.CONTROLLER)
REMOTE_DEVICE_TYPES[zll.PROFILE_ID].append(zll.DeviceType.SCENE_CONTROLLER)

41
tests/components/zha/common.py
View file

@ -3,8 +3,12 @@ import time
from unittest.mock import Mock, patch
from asynctest import CoroutineMock
from zigpy.types.named import EUI64
import zigpy.profiles.zha
import zigpy.types
import zigpy.zcl
import zigpy.zcl.clusters.general
import zigpy.zcl.foundation as zcl_f
import zigpy.zdo.types
from homeassistant.components.zha.core.const import (
DATA_ZHA,
@ -22,7 +26,7 @@ class FakeApplication:
def __init__(self):
"""Init fake application."""
self.ieee = EUI64.convert("00:15:8d:00:02:32:4f:32")
self.ieee = zigpy.types.EUI64.convert("00:15:8d:00:02:32:4f:32")
self.nwk = 0x087D
@ -34,8 +38,6 @@ class FakeEndpoint:
def __init__(self, manufacturer, model):
"""Init fake endpoint."""
from zigpy.profiles.zha import PROFILE_ID
self.device = None
self.endpoint_id = 1
self.in_clusters = {}
@ -44,14 +46,12 @@ class FakeEndpoint:
self.status = 1
self.manufacturer = manufacturer
self.model = model
self.profile_id = PROFILE_ID
self.profile_id = zigpy.profiles.zha.PROFILE_ID
self.device_type = None
def add_input_cluster(self, cluster_id):
"""Add an input cluster."""
from zigpy.zcl import Cluster
cluster = Cluster.from_id(self, cluster_id, is_server=True)
cluster = zigpy.zcl.Cluster.from_id(self, cluster_id, is_server=True)
patch_cluster(cluster)
self.in_clusters[cluster_id] = cluster
if hasattr(cluster, "ep_attribute"):
@ -59,9 +59,7 @@ class FakeEndpoint:
def add_output_cluster(self, cluster_id):
"""Add an output cluster."""
from zigpy.zcl import Cluster
cluster = Cluster.from_id(self, cluster_id, is_server=False)
cluster = zigpy.zcl.Cluster.from_id(self, cluster_id, is_server=False)
patch_cluster(cluster)
self.out_clusters[cluster_id] = cluster
@ -83,7 +81,7 @@ class FakeDevice:
def __init__(self, ieee, manufacturer, model):
"""Init fake device."""
self._application = APPLICATION
self.ieee = EUI64.convert(ieee)
self.ieee = zigpy.types.EUI64.convert(ieee)
self.nwk = 0xB79C
self.zdo = Mock()
self.endpoints = {0: self.zdo}
@ -94,9 +92,7 @@ class FakeDevice:
self.initializing = False
self.manufacturer = manufacturer
self.model = model
from zigpy.zdo.types import NodeDescriptor
self.node_desc = NodeDescriptor()
self.node_desc = zigpy.zdo.types.NodeDescriptor()
def make_device(
@ -150,11 +146,9 @@ async def async_init_zigpy_device(
def make_attribute(attrid, value, status=0):
"""Make an attribute."""
from zigpy.zcl.foundation import Attribute, TypeValue
attr = Attribute()
attr = zcl_f.Attribute()
attr.attrid = attrid
attr.value = TypeValue()
attr.value = zcl_f.TypeValue()
attr.value.value = value
return attr
@ -202,21 +196,18 @@ async def async_test_device_join(
simulate pairing a new device to the network so that code pathways that
only trigger during device joins can be tested.
"""
from zigpy.zcl.foundation import Status
from zigpy.zcl.clusters.general import Basic
# create zigpy device mocking out the zigbee network operations
with patch(
"zigpy.zcl.Cluster.configure_reporting",
return_value=mock_coro([Status.SUCCESS, Status.SUCCESS]),
return_value=mock_coro([zcl_f.Status.SUCCESS, zcl_f.Status.SUCCESS]),
):
with patch(
"zigpy.zcl.Cluster.bind",
return_value=mock_coro([Status.SUCCESS, Status.SUCCESS]),
return_value=mock_coro([zcl_f.Status.SUCCESS, zcl_f.Status.SUCCESS]),
):
zigpy_device = await async_init_zigpy_device(
hass,
[cluster_id, Basic.cluster_id],
[cluster_id, zigpy.zcl.clusters.general.Basic.cluster_id],
[],
device_type,
zha_gateway,

9
tests/components/zha/conftest.py
View file

@ -1,13 +1,16 @@
"""Test configuration for the ZHA component."""
from unittest.mock import patch
import pytest
from homeassistant import config_entries
from homeassistant.components.zha.core.const import DOMAIN, DATA_ZHA, COMPONENTS
from homeassistant.helpers.device_registry import async_get_registry as get_dev_reg
from homeassistant.components.zha.core.const import COMPONENTS, DATA_ZHA, DOMAIN
from homeassistant.components.zha.core.gateway import ZHAGateway
from homeassistant.components.zha.core.registries import establish_device_mappings
from .common import async_setup_entry
from homeassistant.components.zha.core.store import async_get_registry
from homeassistant.helpers.device_registry import async_get_registry as get_dev_reg
from .common import async_setup_entry
@pytest.fixture(name="config_entry")

20
tests/components/zha/test_api.py
View file

@ -1,33 +1,39 @@
"""Test ZHA API."""
import pytest
import zigpy.zcl.clusters.general as general
from homeassistant.components.switch import DOMAIN
from homeassistant.components.zha.api import async_load_api, TYPE, ID
from homeassistant.components.websocket_api import const
from homeassistant.components.zha.api import ID, TYPE, async_load_api
from homeassistant.components.zha.core.const import (
ATTR_CLUSTER_ID,
ATTR_CLUSTER_TYPE,
CLUSTER_TYPE_IN,
ATTR_ENDPOINT_ID,
ATTR_IEEE,
ATTR_MANUFACTURER,
ATTR_MODEL,
ATTR_NAME,
ATTR_QUIRK_APPLIED,
ATTR_MANUFACTURER,
ATTR_ENDPOINT_ID,
CLUSTER_TYPE_IN,
)
from homeassistant.components.websocket_api import const
from .common import async_init_zigpy_device
@pytest.fixture
async def zha_client(hass, config_entry, zha_gateway, hass_ws_client):
"""Test zha switch platform."""
from zigpy.zcl.clusters.general import OnOff, Basic
# load the ZHA API
async_load_api(hass)
# create zigpy device
await async_init_zigpy_device(
hass, [OnOff.cluster_id, Basic.cluster_id], [], None, zha_gateway
hass,
[general.OnOff.cluster_id, general.Basic.cluster_id],
[],
None,
zha_gateway,
)
# load up switch domain

22
tests/components/zha/test_binary_sensor.py
View file

@ -1,5 +1,8 @@
"""Test zha binary sensor."""
from zigpy.zcl.foundation import Command
import zigpy.zcl.clusters.general as general
import zigpy.zcl.clusters.measurement as measurement
import zigpy.zcl.clusters.security as security
import zigpy.zcl.foundation as zcl_f
from homeassistant.components.binary_sensor import DOMAIN
from homeassistant.const import STATE_OFF, STATE_ON, STATE_UNAVAILABLE
@ -16,18 +19,19 @@ from .common import (
async def test_binary_sensor(hass, config_entry, zha_gateway):
"""Test zha binary_sensor platform."""
from zigpy.zcl.clusters.security import IasZone
from zigpy.zcl.clusters.measurement import OccupancySensing
from zigpy.zcl.clusters.general import Basic
# create zigpy devices
zigpy_device_zone = await async_init_zigpy_device(
hass, [IasZone.cluster_id, Basic.cluster_id], [], None, zha_gateway
hass,
[security.IasZone.cluster_id, general.Basic.cluster_id],
[],
None,
zha_gateway,
)
zigpy_device_occupancy = await async_init_zigpy_device(
hass,
[OccupancySensing.cluster_id, Basic.cluster_id],
[measurement.OccupancySensing.cluster_id, general.Basic.cluster_id],
[],
None,
zha_gateway,
@ -71,14 +75,16 @@ async def test_binary_sensor(hass, config_entry, zha_gateway):
await async_test_iaszone_on_off(hass, zone_cluster, zone_entity_id)
# test new sensor join
await async_test_device_join(hass, zha_gateway, OccupancySensing.cluster_id, DOMAIN)
await async_test_device_join(
hass, zha_gateway, measurement.OccupancySensing.cluster_id, DOMAIN
)
async def async_test_binary_sensor_on_off(hass, cluster, entity_id):
"""Test getting on and off messages for binary sensors."""
# binary sensor on
attr = make_attribute(0, 1)
hdr = make_zcl_header(Command.Report_Attributes)
hdr = make_zcl_header(zcl_f.Command.Report_Attributes)
cluster.handle_message(hdr, [[attr]])
await hass.async_block_till_done()

2
tests/components/zha/test_config_flow.py
View file

@ -1,7 +1,9 @@
"""Tests for ZHA config flow."""
from asynctest import patch
from homeassistant.components.zha import config_flow
from homeassistant.components.zha.core.const import DOMAIN
from tests.common import MockConfigEntry

26
tests/components/zha/test_device_action.py
View file

@ -2,6 +2,9 @@
from unittest.mock import patch
import pytest
import zigpy.zcl.clusters.general as general
import zigpy.zcl.clusters.security as security
import zigpy.zcl.foundation as zcl_f
import homeassistant.components.automation as automation
from homeassistant.components.device_automation import (
@ -29,13 +32,15 @@ def calls(hass):
async def test_get_actions(hass, config_entry, zha_gateway):
"""Test we get the expected actions from a zha device."""
from zigpy.zcl.clusters.general import Basic
from zigpy.zcl.clusters.security import IasZone, IasWd
# create zigpy device
zigpy_device = await async_init_zigpy_device(
hass,
[Basic.cluster_id, IasZone.cluster_id, IasWd.cluster_id],
[
general.Basic.cluster_id,
security.IasZone.cluster_id,
security.IasWd.cluster_id,
],
[],
None,
zha_gateway,
@ -64,15 +69,15 @@ async def test_get_actions(hass, config_entry, zha_gateway):
async def test_action(hass, config_entry, zha_gateway, calls):
"""Test for executing a zha device action."""
from zigpy.zcl.clusters.general import Basic, OnOff
from zigpy.zcl.clusters.security import IasZone, IasWd
from zigpy.zcl.foundation import Status
# create zigpy device
zigpy_device = await async_init_zigpy_device(
hass,
[Basic.cluster_id, IasZone.cluster_id, IasWd.cluster_id],
[OnOff.cluster_id],
[
general.Basic.cluster_id,
security.IasZone.cluster_id,
security.IasWd.cluster_id,
],
[general.OnOff.cluster_id],
None,
zha_gateway,
)
@ -96,7 +101,8 @@ async def test_action(hass, config_entry, zha_gateway, calls):
await async_enable_traffic(hass, zha_gateway, [zha_device])
with patch(
"zigpy.zcl.Cluster.request", return_value=mock_coro([0x00, Status.SUCCESS])
"zigpy.zcl.Cluster.request",
return_value=mock_coro([0x00, zcl_f.Status.SUCCESS]),
):
assert await async_setup_component(
hass,

27
tests/components/zha/test_device_tracker.py
View file

@ -2,7 +2,8 @@
from datetime import timedelta
import time
from zigpy.zcl.foundation import Command
import zigpy.zcl.clusters.general as general
import zigpy.zcl.foundation as zcl_f
from homeassistant.components.device_tracker import DOMAIN, SOURCE_TYPE_ROUTER
from homeassistant.components.zha.core.registries import (
@ -25,26 +26,18 @@ from tests.common import async_fire_time_changed
async def test_device_tracker(hass, config_entry, zha_gateway):
"""Test zha device tracker platform."""
from zigpy.zcl.clusters.general import (
Basic,
PowerConfiguration,
BinaryInput,
Identify,
Ota,
PollControl,
)
# create zigpy device
zigpy_device = await async_init_zigpy_device(
hass,
[
Basic.cluster_id,
PowerConfiguration.cluster_id,
Identify.cluster_id,
PollControl.cluster_id,
BinaryInput.cluster_id,
general.Basic.cluster_id,
general.PowerConfiguration.cluster_id,
general.Identify.cluster_id,
general.PollControl.cluster_id,
general.BinaryInput.cluster_id,
],
[Identify.cluster_id, Ota.cluster_id],
[general.Identify.cluster_id, general.Ota.cluster_id],
SMARTTHINGS_ARRIVAL_SENSOR_DEVICE_TYPE,
zha_gateway,
)
@ -73,7 +66,7 @@ async def test_device_tracker(hass, config_entry, zha_gateway):
# turn state flip
attr = make_attribute(0x0020, 23)
hdr = make_zcl_header(Command.Report_Attributes)
hdr = make_zcl_header(zcl_f.Command.Report_Attributes)
cluster.handle_message(hdr, [[attr]])
attr = make_attribute(0x0021, 200)
@ -96,7 +89,7 @@ async def test_device_tracker(hass, config_entry, zha_gateway):
await async_test_device_join(
hass,
zha_gateway,
PowerConfiguration.cluster_id,
general.PowerConfiguration.cluster_id,
DOMAIN,
SMARTTHINGS_ARRIVAL_SENSOR_DEVICE_TYPE,
)

18
tests/components/zha/test_device_trigger.py
View file

@ -1,5 +1,6 @@
"""ZHA device automation trigger tests."""
import pytest
import zigpy.zcl.clusters.general as general
import homeassistant.components.automation as automation
from homeassistant.components.switch import DOMAIN
@ -9,7 +10,7 @@ from homeassistant.setup import async_setup_component
from .common import async_enable_traffic, async_init_zigpy_device
from tests.common import async_mock_service, async_get_device_automations
from tests.common import async_get_device_automations, async_mock_service
ON = 1
OFF = 0
@ -43,11 +44,10 @@ def calls(hass):
async def test_triggers(hass, config_entry, zha_gateway):
"""Test zha device triggers."""
from zigpy.zcl.clusters.general import OnOff, Basic
# create zigpy device
zigpy_device = await async_init_zigpy_device(
hass, [Basic.cluster_id], [OnOff.cluster_id], None, zha_gateway
hass, [general.Basic.cluster_id], [general.OnOff.cluster_id], None, zha_gateway
)
zigpy_device.device_automation_triggers = {
@ -112,11 +112,10 @@ async def test_triggers(hass, config_entry, zha_gateway):
async def test_no_triggers(hass, config_entry, zha_gateway):
"""Test zha device with no triggers."""
from zigpy.zcl.clusters.general import OnOff, Basic
# create zigpy device
zigpy_device = await async_init_zigpy_device(
hass, [Basic.cluster_id], [OnOff.cluster_id], None, zha_gateway
hass, [general.Basic.cluster_id], [general.OnOff.cluster_id], None, zha_gateway
)
await hass.config_entries.async_forward_entry_setup(config_entry, DOMAIN)
@ -135,11 +134,10 @@ async def test_no_triggers(hass, config_entry, zha_gateway):
async def test_if_fires_on_event(hass, config_entry, zha_gateway, calls):
"""Test for remote triggers firing."""
from zigpy.zcl.clusters.general import OnOff, Basic
# create zigpy device
zigpy_device = await async_init_zigpy_device(
hass, [Basic.cluster_id], [OnOff.cluster_id], None, zha_gateway
hass, [general.Basic.cluster_id], [general.OnOff.cluster_id], None, zha_gateway
)
zigpy_device.device_automation_triggers = {
@ -197,11 +195,10 @@ async def test_if_fires_on_event(hass, config_entry, zha_gateway, calls):
async def test_exception_no_triggers(hass, config_entry, zha_gateway, calls, caplog):
"""Test for exception on event triggers firing."""
from zigpy.zcl.clusters.general import OnOff, Basic
# create zigpy device
zigpy_device = await async_init_zigpy_device(
hass, [Basic.cluster_id], [OnOff.cluster_id], None, zha_gateway
hass, [general.Basic.cluster_id], [general.OnOff.cluster_id], None, zha_gateway
)
await hass.config_entries.async_forward_entry_setup(config_entry, DOMAIN)
@ -244,11 +241,10 @@ async def test_exception_no_triggers(hass, config_entry, zha_gateway, calls, cap
async def test_exception_bad_trigger(hass, config_entry, zha_gateway, calls, caplog):
"""Test for exception on event triggers firing."""
from zigpy.zcl.clusters.general import OnOff, Basic
# create zigpy device
zigpy_device = await async_init_zigpy_device(
hass, [Basic.cluster_id], [OnOff.cluster_id], None, zha_gateway
hass, [general.Basic.cluster_id], [general.OnOff.cluster_id], None, zha_gateway
)
zigpy_device.device_automation_triggers = {

19
tests/components/zha/test_fan.py
View file

@ -1,7 +1,9 @@
"""Test zha fan."""
from unittest.mock import call, patch
from zigpy.zcl.foundation import Command
import zigpy.zcl.clusters.general as general
import zigpy.zcl.clusters.hvac as hvac
import zigpy.zcl.foundation as zcl_f
from homeassistant.components import fan
from homeassistant.components.fan import ATTR_SPEED, DOMAIN, SERVICE_SET_SPEED
@ -28,13 +30,10 @@ from tests.common import mock_coro
async def test_fan(hass, config_entry, zha_gateway):
"""Test zha fan platform."""
from zigpy.zcl.clusters.hvac import Fan
from zigpy.zcl.clusters.general import Basic
from zigpy.zcl.foundation import Status
# create zigpy device
zigpy_device = await async_init_zigpy_device(
hass, [Fan.cluster_id, Basic.cluster_id], [], None, zha_gateway
hass, [hvac.Fan.cluster_id, general.Basic.cluster_id], [], None, zha_gateway
)
# load up fan domain
@ -56,7 +55,7 @@ async def test_fan(hass, config_entry, zha_gateway):
# turn on at fan
attr = make_attribute(0, 1)
hdr = make_zcl_header(Command.Report_Attributes)
hdr = make_zcl_header(zcl_f.Command.Report_Attributes)
cluster.handle_message(hdr, [[attr]])
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_ON
@ -70,7 +69,7 @@ async def test_fan(hass, config_entry, zha_gateway):
# turn on from HA
with patch(
"zigpy.zcl.Cluster.write_attributes",
return_value=mock_coro([Status.SUCCESS, Status.SUCCESS]),
return_value=mock_coro([zcl_f.Status.SUCCESS, zcl_f.Status.SUCCESS]),
):
# turn on via UI
await async_turn_on(hass, entity_id)
@ -80,7 +79,7 @@ async def test_fan(hass, config_entry, zha_gateway):
# turn off from HA
with patch(
"zigpy.zcl.Cluster.write_attributes",
return_value=mock_coro([Status.SUCCESS, Status.SUCCESS]),
return_value=mock_coro([zcl_f.Status.SUCCESS, zcl_f.Status.SUCCESS]),
):
# turn off via UI
await async_turn_off(hass, entity_id)
@ -90,7 +89,7 @@ async def test_fan(hass, config_entry, zha_gateway):
# change speed from HA
with patch(
"zigpy.zcl.Cluster.write_attributes",
return_value=mock_coro([Status.SUCCESS, Status.SUCCESS]),
return_value=mock_coro([zcl_f.Status.SUCCESS, zcl_f.Status.SUCCESS]),
):
# turn on via UI
await async_set_speed(hass, entity_id, speed=fan.SPEED_HIGH)
@ -98,7 +97,7 @@ async def test_fan(hass, config_entry, zha_gateway):
assert cluster.write_attributes.call_args == call({"fan_mode": 3})
# test adding new fan to the network and HA
await async_test_device_join(hass, zha_gateway, Fan.cluster_id, DOMAIN)
await async_test_device_join(hass, zha_gateway, hvac.Fan.cluster_id, DOMAIN)
async def async_turn_on(hass, entity_id, speed=None):

51
tests/components/zha/test_light.py
View file

@ -2,7 +2,10 @@
import asyncio
from unittest.mock import MagicMock, call, patch, sentinel
from zigpy.zcl.foundation import Command
import zigpy.profiles.zha
import zigpy.types
import zigpy.zcl.clusters.general as general
import zigpy.zcl.foundation as zcl_f
from homeassistant.components.light import DOMAIN
from homeassistant.const import STATE_OFF, STATE_ON, STATE_UNAVAILABLE
@ -24,24 +27,25 @@ OFF = 0
async def test_light(hass, config_entry, zha_gateway, monkeypatch):
"""Test zha light platform."""
from zigpy.zcl.clusters.general import OnOff, LevelControl, Basic
from zigpy.zcl.foundation import Status
from zigpy.profiles.zha import DeviceType
# create zigpy devices
zigpy_device_on_off = await async_init_zigpy_device(
hass,
[OnOff.cluster_id, Basic.cluster_id],
[general.OnOff.cluster_id, general.Basic.cluster_id],
[],
DeviceType.ON_OFF_LIGHT,
zigpy.profiles.zha.DeviceType.ON_OFF_LIGHT,
zha_gateway,
)
zigpy_device_level = await async_init_zigpy_device(
hass,
[OnOff.cluster_id, LevelControl.cluster_id, Basic.cluster_id],
[
general.OnOff.cluster_id,
general.LevelControl.cluster_id,
general.Basic.cluster_id,
],
[],
DeviceType.ON_OFF_LIGHT,
zigpy.profiles.zha.DeviceType.ON_OFF_LIGHT,
zha_gateway,
ieee="00:0d:6f:11:0a:90:69:e7",
manufacturer="FakeLevelManufacturer",
@ -64,12 +68,12 @@ async def test_light(hass, config_entry, zha_gateway, monkeypatch):
level_device_level_cluster = zigpy_device_level.endpoints.get(1).level
on_off_mock = MagicMock(
side_effect=asyncio.coroutine(
MagicMock(return_value=[sentinel.data, Status.SUCCESS])
MagicMock(return_value=[sentinel.data, zcl_f.Status.SUCCESS])
)
)
level_mock = MagicMock(
side_effect=asyncio.coroutine(
MagicMock(return_value=[sentinel.data, Status.SUCCESS])
MagicMock(return_value=[sentinel.data, zcl_f.Status.SUCCESS])
)
)
monkeypatch.setattr(level_device_on_off_cluster, "request", on_off_mock)
@ -118,7 +122,11 @@ async def test_light(hass, config_entry, zha_gateway, monkeypatch):
# test adding a new light to the network and HA
await async_test_device_join(
hass, zha_gateway, OnOff.cluster_id, DOMAIN, device_type=DeviceType.ON_OFF_LIGHT
hass,
zha_gateway,
general.OnOff.cluster_id,
DOMAIN,
device_type=zigpy.profiles.zha.DeviceType.ON_OFF_LIGHT,
)
@ -126,7 +134,7 @@ async def async_test_on_off_from_light(hass, cluster, entity_id):
"""Test on off functionality from the light."""
# turn on at light
attr = make_attribute(0, 1)
hdr = make_zcl_header(Command.Report_Attributes)
hdr = make_zcl_header(zcl_f.Command.Report_Attributes)
cluster.handle_message(hdr, [[attr]])
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_ON
@ -142,7 +150,7 @@ async def async_test_on_from_light(hass, cluster, entity_id):
"""Test on off functionality from the light."""
# turn on at light
attr = make_attribute(0, 1)
hdr = make_zcl_header(Command.Report_Attributes)
hdr = make_zcl_header(zcl_f.Command.Report_Attributes)
cluster.handle_message(hdr, [[attr]])
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_ON
@ -150,10 +158,9 @@ async def async_test_on_from_light(hass, cluster, entity_id):
async def async_test_on_off_from_hass(hass, cluster, entity_id):
"""Test on off functionality from hass."""
from zigpy.zcl.foundation import Status
with patch(
"zigpy.zcl.Cluster.request", return_value=mock_coro([0x00, Status.SUCCESS])
"zigpy.zcl.Cluster.request",
return_value=mock_coro([0x00, zcl_f.Status.SUCCESS]),
):
# turn on via UI
await hass.services.async_call(
@ -169,10 +176,9 @@ async def async_test_on_off_from_hass(hass, cluster, entity_id):
async def async_test_off_from_hass(hass, cluster, entity_id):
"""Test turning off the light from homeassistant."""
from zigpy.zcl.foundation import Status
with patch(
"zigpy.zcl.Cluster.request", return_value=mock_coro([0x01, Status.SUCCESS])
"zigpy.zcl.Cluster.request",
return_value=mock_coro([0x01, zcl_f.Status.SUCCESS]),
):
# turn off via UI
await hass.services.async_call(
@ -188,7 +194,6 @@ async def async_test_level_on_off_from_hass(
hass, on_off_cluster, level_cluster, entity_id
):
"""Test on off functionality from hass."""
from zigpy import types
# turn on via UI
await hass.services.async_call(
@ -213,7 +218,7 @@ async def async_test_level_on_off_from_hass(
assert level_cluster.request.call_args == call(
False,
4,
(types.uint8_t, types.uint16_t),
(zigpy.types.uint8_t, zigpy.types.uint16_t),
254,
100.0,
expect_reply=True,
@ -233,7 +238,7 @@ async def async_test_level_on_off_from_hass(
assert level_cluster.request.call_args == call(
False,
4,
(types.uint8_t, types.uint16_t),
(zigpy.types.uint8_t, zigpy.types.uint16_t),
10,
0,
expect_reply=True,
@ -248,7 +253,7 @@ async def async_test_level_on_off_from_hass(
async def async_test_dimmer_from_light(hass, cluster, entity_id, level, expected_state):
"""Test dimmer functionality from the light."""
attr = make_attribute(0, level)
hdr = make_zcl_header(Command.Report_Attributes)
hdr = make_zcl_header(zcl_f.Command.Report_Attributes)
cluster.handle_message(hdr, [[attr]])
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == expected_state

26
tests/components/zha/test_lock.py
View file

@ -1,7 +1,9 @@
"""Test zha lock."""
from unittest.mock import patch
from zigpy.zcl.foundation import Command
import zigpy.zcl.clusters.closures as closures
import zigpy.zcl.clusters.general as general
import zigpy.zcl.foundation as zcl_f
from homeassistant.components.lock import DOMAIN
from homeassistant.const import STATE_LOCKED, STATE_UNAVAILABLE, STATE_UNLOCKED
@ -22,12 +24,14 @@ UNLOCK_DOOR = 1
async def test_lock(hass, config_entry, zha_gateway):
"""Test zha lock platform."""
from zigpy.zcl.clusters.closures import DoorLock
from zigpy.zcl.clusters.general import Basic
# create zigpy device
zigpy_device = await async_init_zigpy_device(
hass, [DoorLock.cluster_id, Basic.cluster_id], [], None, zha_gateway
hass,
[closures.DoorLock.cluster_id, general.Basic.cluster_id],
[],
None,
zha_gateway,
)
# load up lock domain
@ -49,7 +53,7 @@ async def test_lock(hass, config_entry, zha_gateway):
# set state to locked
attr = make_attribute(0, 1)
hdr = make_zcl_header(Command.Report_Attributes)
hdr = make_zcl_header(zcl_f.Command.Report_Attributes)
cluster.handle_message(hdr, [[attr]])
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_LOCKED
@ -69,9 +73,9 @@ async def test_lock(hass, config_entry, zha_gateway):
async def async_lock(hass, cluster, entity_id):
"""Test lock functionality from hass."""
from zigpy.zcl.foundation import Status
with patch("zigpy.zcl.Cluster.request", return_value=mock_coro([Status.SUCCESS])):
with patch(
"zigpy.zcl.Cluster.request", return_value=mock_coro([zcl_f.Status.SUCCESS])
):
# lock via UI
await hass.services.async_call(
DOMAIN, "lock", {"entity_id": entity_id}, blocking=True
@ -83,9 +87,9 @@ async def async_lock(hass, cluster, entity_id):
async def async_unlock(hass, cluster, entity_id):
"""Test lock functionality from hass."""
from zigpy.zcl.foundation import Status
with patch("zigpy.zcl.Cluster.request", return_value=mock_coro([Status.SUCCESS])):
with patch(
"zigpy.zcl.Cluster.request", return_value=mock_coro([zcl_f.Status.SUCCESS])
):
# lock via UI
await hass.services.async_call(
DOMAIN, "unlock", {"entity_id": entity_id}, blocking=True

45
tests/components/zha/test_sensor.py
View file

@ -1,5 +1,9 @@
"""Test zha sensor."""
from zigpy.zcl.foundation import Command
import zigpy.zcl.clusters.general as general
import zigpy.zcl.clusters.homeautomation as homeautomation
import zigpy.zcl.clusters.measurement as measurement
import zigpy.zcl.clusters.smartenergy as smartenergy
import zigpy.zcl.foundation as zcl_f
from homeassistant.components.sensor import DOMAIN
from homeassistant.const import STATE_UNAVAILABLE, STATE_UNKNOWN
@ -16,23 +20,15 @@ from .common import (
async def test_sensor(hass, config_entry, zha_gateway):
"""Test zha sensor platform."""
from zigpy.zcl.clusters.measurement import (
RelativeHumidity,
TemperatureMeasurement,
PressureMeasurement,
IlluminanceMeasurement,
)
from zigpy.zcl.clusters.smartenergy import Metering
from zigpy.zcl.clusters.homeautomation import ElectricalMeasurement
# list of cluster ids to create devices and sensor entities for
cluster_ids = [
RelativeHumidity.cluster_id,
TemperatureMeasurement.cluster_id,
PressureMeasurement.cluster_id,
IlluminanceMeasurement.cluster_id,
Metering.cluster_id,
ElectricalMeasurement.cluster_id,
measurement.RelativeHumidity.cluster_id,
measurement.TemperatureMeasurement.cluster_id,
measurement.PressureMeasurement.cluster_id,
measurement.IlluminanceMeasurement.cluster_id,
smartenergy.Metering.cluster_id,
homeautomation.ElectricalMeasurement.cluster_id,
]
# devices that were created from cluster_ids list above
@ -63,33 +59,33 @@ async def test_sensor(hass, config_entry, zha_gateway):
assert hass.states.get(entity_id).state == STATE_UNKNOWN
# get the humidity device info and test the associated sensor logic
device_info = zigpy_device_infos[RelativeHumidity.cluster_id]
device_info = zigpy_device_infos[measurement.RelativeHumidity.cluster_id]
await async_test_humidity(hass, device_info)
# get the temperature device info and test the associated sensor logic
device_info = zigpy_device_infos[TemperatureMeasurement.cluster_id]
device_info = zigpy_device_infos[measurement.TemperatureMeasurement.cluster_id]
await async_test_temperature(hass, device_info)
# get the pressure device info and test the associated sensor logic
device_info = zigpy_device_infos[PressureMeasurement.cluster_id]
device_info = zigpy_device_infos[measurement.PressureMeasurement.cluster_id]
await async_test_pressure(hass, device_info)
# get the illuminance device info and test the associated sensor logic
device_info = zigpy_device_infos[IlluminanceMeasurement.cluster_id]
device_info = zigpy_device_infos[measurement.IlluminanceMeasurement.cluster_id]
await async_test_illuminance(hass, device_info)
# get the metering device info and test the associated sensor logic
device_info = zigpy_device_infos[Metering.cluster_id]
device_info = zigpy_device_infos[smartenergy.Metering.cluster_id]
await async_test_metering(hass, device_info)
# get the electrical_measurement device info and test the associated
# sensor logic
device_info = zigpy_device_infos[ElectricalMeasurement.cluster_id]
device_info = zigpy_device_infos[homeautomation.ElectricalMeasurement.cluster_id]
await async_test_electrical_measurement(hass, device_info)
# test joining a new temperature sensor to the network
await async_test_device_join(
hass, zha_gateway, TemperatureMeasurement.cluster_id, DOMAIN
hass, zha_gateway, measurement.TemperatureMeasurement.cluster_id, DOMAIN
)
@ -102,7 +98,6 @@ async def async_build_devices(hass, zha_gateway, config_entry, cluster_ids):
A dict containing relevant device info for testing is returned. It contains
the entity id, zigpy device, and the zigbee cluster for the sensor.
"""
from zigpy.zcl.clusters.general import Basic
device_infos = {}
counter = 0
@ -111,7 +106,7 @@ async def async_build_devices(hass, zha_gateway, config_entry, cluster_ids):
device_infos[cluster_id] = {"zigpy_device": None}
device_infos[cluster_id]["zigpy_device"] = await async_init_zigpy_device(
hass,
[cluster_id, Basic.cluster_id],
[cluster_id, general.Basic.cluster_id],
[],
None,
zha_gateway,
@ -181,7 +176,7 @@ async def send_attribute_report(hass, cluster, attrid, value):
device is paired to the zigbee network.
"""
attr = make_attribute(attrid, value)
hdr = make_zcl_header(Command.Report_Attributes)
hdr = make_zcl_header(zcl_f.Command.Report_Attributes)
cluster.handle_message(hdr, [[attr]])
await hass.async_block_till_done()

21
tests/components/zha/test_switch.py
View file

@ -1,7 +1,8 @@
"""Test zha switch."""
from unittest.mock import call, patch
from zigpy.zcl.foundation import Command
import zigpy.zcl.clusters.general as general
import zigpy.zcl.foundation as zcl_f
from homeassistant.components.switch import DOMAIN
from homeassistant.const import STATE_OFF, STATE_ON, STATE_UNAVAILABLE
@ -23,12 +24,14 @@ OFF = 0
async def test_switch(hass, config_entry, zha_gateway):
"""Test zha switch platform."""
from zigpy.zcl.clusters.general import OnOff, Basic
from zigpy.zcl.foundation import Status
# create zigpy device
zigpy_device = await async_init_zigpy_device(
hass, [OnOff.cluster_id, Basic.cluster_id], [], None, zha_gateway
hass,
[general.OnOff.cluster_id, general.Basic.cluster_id],
[],
None,
zha_gateway,
)
# load up switch domain
@ -50,7 +53,7 @@ async def test_switch(hass, config_entry, zha_gateway):
# turn on at switch
attr = make_attribute(0, 1)
hdr = make_zcl_header(Command.Report_Attributes)
hdr = make_zcl_header(zcl_f.Command.Report_Attributes)
cluster.handle_message(hdr, [[attr]])
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_ON
@ -63,7 +66,8 @@ async def test_switch(hass, config_entry, zha_gateway):
# turn on from HA
with patch(
"zigpy.zcl.Cluster.request", return_value=mock_coro([0x00, Status.SUCCESS])
"zigpy.zcl.Cluster.request",
return_value=mock_coro([0x00, zcl_f.Status.SUCCESS]),
):
# turn on via UI
await hass.services.async_call(
@ -76,7 +80,8 @@ async def test_switch(hass, config_entry, zha_gateway):
# turn off from HA
with patch(
"zigpy.zcl.Cluster.request", return_value=mock_coro([0x01, Status.SUCCESS])
"zigpy.zcl.Cluster.request",
return_value=mock_coro([0x01, zcl_f.Status.SUCCESS]),
):
# turn off via UI
await hass.services.async_call(
@ -88,4 +93,4 @@ async def test_switch(hass, config_entry, zha_gateway):
)
# test joining a new switch to the network and HA
await async_test_device_join(hass, zha_gateway, OnOff.cluster_id, DOMAIN)
await async_test_device_join(hass, zha_gateway, general.OnOff.cluster_id, DOMAIN)