226 lines
7.0 KiB
Python
226 lines
7.0 KiB
Python
"""Common test objects."""
|
|
import time
|
|
from unittest.mock import Mock, patch
|
|
|
|
from asynctest import CoroutineMock
|
|
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,
|
|
DATA_ZHA_BRIDGE_ID,
|
|
DATA_ZHA_CONFIG,
|
|
DATA_ZHA_DISPATCHERS,
|
|
)
|
|
from homeassistant.util import slugify
|
|
|
|
from tests.common import mock_coro
|
|
|
|
|
|
class FakeApplication:
|
|
"""Fake application for mocking zigpy."""
|
|
|
|
def __init__(self):
|
|
"""Init fake application."""
|
|
self.ieee = zigpy.types.EUI64.convert("00:15:8d:00:02:32:4f:32")
|
|
self.nwk = 0x087D
|
|
|
|
|
|
APPLICATION = FakeApplication()
|
|
|
|
|
|
class FakeEndpoint:
|
|
"""Fake endpoint for moking zigpy."""
|
|
|
|
def __init__(self, manufacturer, model):
|
|
"""Init fake endpoint."""
|
|
self.device = None
|
|
self.endpoint_id = 1
|
|
self.in_clusters = {}
|
|
self.out_clusters = {}
|
|
self._cluster_attr = {}
|
|
self.status = 1
|
|
self.manufacturer = manufacturer
|
|
self.model = model
|
|
self.profile_id = zigpy.profiles.zha.PROFILE_ID
|
|
self.device_type = None
|
|
|
|
def add_input_cluster(self, cluster_id):
|
|
"""Add an input cluster."""
|
|
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"):
|
|
setattr(self, cluster.ep_attribute, cluster)
|
|
|
|
def add_output_cluster(self, cluster_id):
|
|
"""Add an output cluster."""
|
|
cluster = zigpy.zcl.Cluster.from_id(self, cluster_id, is_server=False)
|
|
patch_cluster(cluster)
|
|
self.out_clusters[cluster_id] = cluster
|
|
|
|
|
|
def patch_cluster(cluster):
|
|
"""Patch a cluster for testing."""
|
|
cluster.bind = CoroutineMock(return_value=[0])
|
|
cluster.configure_reporting = CoroutineMock(return_value=[0])
|
|
cluster.deserialize = Mock()
|
|
cluster.handle_cluster_request = Mock()
|
|
cluster.read_attributes = CoroutineMock()
|
|
cluster.read_attributes_raw = Mock()
|
|
cluster.unbind = CoroutineMock(return_value=[0])
|
|
|
|
|
|
class FakeDevice:
|
|
"""Fake device for mocking zigpy."""
|
|
|
|
def __init__(self, ieee, manufacturer, model):
|
|
"""Init fake device."""
|
|
self._application = APPLICATION
|
|
self.ieee = zigpy.types.EUI64.convert(ieee)
|
|
self.nwk = 0xB79C
|
|
self.zdo = Mock()
|
|
self.endpoints = {0: self.zdo}
|
|
self.lqi = 255
|
|
self.rssi = 8
|
|
self.last_seen = time.time()
|
|
self.status = 2
|
|
self.initializing = False
|
|
self.manufacturer = manufacturer
|
|
self.model = model
|
|
self.node_desc = zigpy.zdo.types.NodeDescriptor()
|
|
|
|
|
|
def make_device(
|
|
in_cluster_ids, out_cluster_ids, device_type, ieee, manufacturer, model
|
|
):
|
|
"""Make a fake device using the specified cluster classes."""
|
|
device = FakeDevice(ieee, manufacturer, model)
|
|
endpoint = FakeEndpoint(manufacturer, model)
|
|
endpoint.device = device
|
|
device.endpoints[endpoint.endpoint_id] = endpoint
|
|
endpoint.device_type = device_type
|
|
|
|
for cluster_id in in_cluster_ids:
|
|
endpoint.add_input_cluster(cluster_id)
|
|
|
|
for cluster_id in out_cluster_ids:
|
|
endpoint.add_output_cluster(cluster_id)
|
|
|
|
return device
|
|
|
|
|
|
async def async_init_zigpy_device(
|
|
hass,
|
|
in_cluster_ids,
|
|
out_cluster_ids,
|
|
device_type,
|
|
gateway,
|
|
ieee="00:0d:6f:00:0a:90:69:e7",
|
|
manufacturer="FakeManufacturer",
|
|
model="FakeModel",
|
|
is_new_join=False,
|
|
):
|
|
"""Create and initialize a device.
|
|
|
|
This creates a fake device and adds it to the "network". It can be used to
|
|
test existing device functionality and new device pairing functionality.
|
|
The is_new_join parameter influences whether or not the device will go
|
|
through cluster binding and zigbee cluster configure reporting. That only
|
|
happens when the device is paired to the network for the first time.
|
|
"""
|
|
device = make_device(
|
|
in_cluster_ids, out_cluster_ids, device_type, ieee, manufacturer, model
|
|
)
|
|
if is_new_join:
|
|
await gateway.async_device_initialized(device)
|
|
else:
|
|
await gateway.async_device_restored(device)
|
|
await hass.async_block_till_done()
|
|
return device
|
|
|
|
|
|
def make_attribute(attrid, value, status=0):
|
|
"""Make an attribute."""
|
|
attr = zcl_f.Attribute()
|
|
attr.attrid = attrid
|
|
attr.value = zcl_f.TypeValue()
|
|
attr.value.value = value
|
|
return attr
|
|
|
|
|
|
async def async_setup_entry(hass, config_entry):
|
|
"""Mock setup entry for zha."""
|
|
hass.data[DATA_ZHA][DATA_ZHA_CONFIG] = {}
|
|
hass.data[DATA_ZHA][DATA_ZHA_DISPATCHERS] = []
|
|
hass.data[DATA_ZHA][DATA_ZHA_BRIDGE_ID] = APPLICATION.ieee
|
|
return True
|
|
|
|
|
|
async def find_entity_id(domain, zha_device, hass):
|
|
"""Find the entity id under the testing.
|
|
|
|
This is used to get the entity id in order to get the state from the state
|
|
machine so that we can test state changes.
|
|
"""
|
|
ieeetail = "".join([f"{o:02x}" for o in zha_device.ieee[:4]])
|
|
head = f"{domain}." + slugify(f"{zha_device.name} {ieeetail}")
|
|
|
|
enitiy_ids = hass.states.async_entity_ids(domain)
|
|
await hass.async_block_till_done()
|
|
|
|
for entity_id in enitiy_ids:
|
|
if entity_id.startswith(head):
|
|
return entity_id
|
|
return None
|
|
|
|
|
|
async def async_enable_traffic(hass, zha_gateway, zha_devices):
|
|
"""Allow traffic to flow through the gateway and the zha device."""
|
|
for zha_device in zha_devices:
|
|
zha_device.update_available(True)
|
|
await hass.async_block_till_done()
|
|
|
|
|
|
async def async_test_device_join(
|
|
hass, zha_gateway, cluster_id, entity_id, device_type=None
|
|
):
|
|
"""Test a newly joining device.
|
|
|
|
This creates a new fake device and adds it to the network. It is meant to
|
|
simulate pairing a new device to the network so that code pathways that
|
|
only trigger during device joins can be tested.
|
|
"""
|
|
# create zigpy device mocking out the zigbee network operations
|
|
with patch(
|
|
"zigpy.zcl.Cluster.configure_reporting",
|
|
return_value=mock_coro([zcl_f.Status.SUCCESS, zcl_f.Status.SUCCESS]),
|
|
):
|
|
with patch(
|
|
"zigpy.zcl.Cluster.bind",
|
|
return_value=mock_coro([zcl_f.Status.SUCCESS, zcl_f.Status.SUCCESS]),
|
|
):
|
|
await async_init_zigpy_device(
|
|
hass,
|
|
[cluster_id, zigpy.zcl.clusters.general.Basic.cluster_id],
|
|
[],
|
|
device_type,
|
|
zha_gateway,
|
|
ieee="00:0d:6f:00:0a:90:69:f7",
|
|
is_new_join=True,
|
|
)
|
|
assert hass.states.get(entity_id) is not None
|
|
|
|
|
|
def make_zcl_header(command_id: int, global_command: bool = True) -> zcl_f.ZCLHeader:
|
|
"""Cluster.handle_message() ZCL Header helper."""
|
|
if global_command:
|
|
frc = zcl_f.FrameControl(zcl_f.FrameType.GLOBAL_COMMAND)
|
|
else:
|
|
frc = zcl_f.FrameControl(zcl_f.FrameType.CLUSTER_COMMAND)
|
|
return zcl_f.ZCLHeader(frc, tsn=1, command_id=command_id)
|