146 lines
5 KiB
Python
146 lines
5 KiB
Python
"""Test zha device switch."""
|
|
from datetime import timedelta
|
|
import time
|
|
from unittest import mock
|
|
|
|
import asynctest
|
|
import pytest
|
|
import zigpy.zcl.clusters.general as general
|
|
|
|
import homeassistant.components.zha.core.device as zha_core_device
|
|
import homeassistant.core as ha
|
|
import homeassistant.util.dt as dt_util
|
|
|
|
from .common import async_enable_traffic
|
|
|
|
|
|
@pytest.fixture
|
|
def zigpy_device(zigpy_device_mock):
|
|
"""Device tracker zigpy device."""
|
|
|
|
def _dev(with_basic_channel: bool = True):
|
|
in_clusters = [general.OnOff.cluster_id]
|
|
if with_basic_channel:
|
|
in_clusters.append(general.Basic.cluster_id)
|
|
|
|
endpoints = {
|
|
3: {"in_clusters": in_clusters, "out_clusters": [], "device_type": 0}
|
|
}
|
|
return zigpy_device_mock(endpoints)
|
|
|
|
return _dev
|
|
|
|
|
|
@pytest.fixture
|
|
def device_with_basic_channel(zigpy_device):
|
|
"""Return a zha device with a basic channel present."""
|
|
return zigpy_device(with_basic_channel=True)
|
|
|
|
|
|
@pytest.fixture
|
|
def device_without_basic_channel(zigpy_device):
|
|
"""Return a zha device with a basic channel present."""
|
|
return zigpy_device(with_basic_channel=False)
|
|
|
|
|
|
def _send_time_changed(hass, seconds):
|
|
"""Send a time changed event."""
|
|
now = dt_util.utcnow() + timedelta(seconds)
|
|
hass.bus.async_fire(ha.EVENT_TIME_CHANGED, {ha.ATTR_NOW: now})
|
|
|
|
|
|
@asynctest.patch(
|
|
"homeassistant.components.zha.core.channels.general.BasicChannel.async_initialize",
|
|
new=mock.MagicMock(),
|
|
)
|
|
async def test_check_available_success(
|
|
hass, device_with_basic_channel, zha_device_restored
|
|
):
|
|
"""Check device availability success on 1st try."""
|
|
|
|
# pylint: disable=protected-access
|
|
zha_device = await zha_device_restored(device_with_basic_channel)
|
|
await async_enable_traffic(hass, [zha_device])
|
|
basic_ch = device_with_basic_channel.endpoints[3].basic
|
|
|
|
basic_ch.read_attributes.reset_mock()
|
|
device_with_basic_channel.last_seen = None
|
|
assert zha_device.available is True
|
|
_send_time_changed(hass, 61)
|
|
await hass.async_block_till_done()
|
|
assert zha_device.available is False
|
|
assert basic_ch.read_attributes.await_count == 0
|
|
|
|
device_with_basic_channel.last_seen = (
|
|
time.time() - zha_core_device._KEEP_ALIVE_INTERVAL - 2
|
|
)
|
|
_seens = [time.time(), device_with_basic_channel.last_seen]
|
|
|
|
def _update_last_seen(*args, **kwargs):
|
|
device_with_basic_channel.last_seen = _seens.pop()
|
|
|
|
basic_ch.read_attributes.side_effect = _update_last_seen
|
|
|
|
# successfully ping zigpy device, but zha_device is not yet available
|
|
_send_time_changed(hass, 61)
|
|
await hass.async_block_till_done()
|
|
assert basic_ch.read_attributes.await_count == 1
|
|
assert basic_ch.read_attributes.await_args[0][0] == ["manufacturer"]
|
|
assert zha_device.available is False
|
|
|
|
# There was traffic from the device: pings, but not yet available
|
|
_send_time_changed(hass, 61)
|
|
await hass.async_block_till_done()
|
|
assert basic_ch.read_attributes.await_count == 2
|
|
assert basic_ch.read_attributes.await_args[0][0] == ["manufacturer"]
|
|
assert zha_device.available is False
|
|
|
|
# There was traffic from the device: don't try to ping, marked as available
|
|
_send_time_changed(hass, 61)
|
|
await hass.async_block_till_done()
|
|
assert basic_ch.read_attributes.await_count == 2
|
|
assert basic_ch.read_attributes.await_args[0][0] == ["manufacturer"]
|
|
assert zha_device.available is True
|
|
|
|
|
|
@asynctest.patch(
|
|
"homeassistant.components.zha.core.channels.general.BasicChannel.async_initialize",
|
|
new=mock.MagicMock(),
|
|
)
|
|
async def test_check_available_unsuccessful(
|
|
hass, device_with_basic_channel, zha_device_restored
|
|
):
|
|
"""Check device availability all tries fail."""
|
|
|
|
# pylint: disable=protected-access
|
|
zha_device = await zha_device_restored(device_with_basic_channel)
|
|
await async_enable_traffic(hass, [zha_device])
|
|
basic_ch = device_with_basic_channel.endpoints[3].basic
|
|
|
|
assert zha_device.available is True
|
|
assert basic_ch.read_attributes.await_count == 0
|
|
|
|
device_with_basic_channel.last_seen = (
|
|
time.time() - zha_core_device._KEEP_ALIVE_INTERVAL - 2
|
|
)
|
|
|
|
# unsuccessfuly ping zigpy device, but zha_device is still available
|
|
_send_time_changed(hass, 61)
|
|
await hass.async_block_till_done()
|
|
assert basic_ch.read_attributes.await_count == 1
|
|
assert basic_ch.read_attributes.await_args[0][0] == ["manufacturer"]
|
|
assert zha_device.available is True
|
|
|
|
# still no traffic, but zha_device is still available
|
|
_send_time_changed(hass, 61)
|
|
await hass.async_block_till_done()
|
|
assert basic_ch.read_attributes.await_count == 2
|
|
assert basic_ch.read_attributes.await_args[0][0] == ["manufacturer"]
|
|
assert zha_device.available is True
|
|
|
|
# not even trying to update, device is unavailble
|
|
_send_time_changed(hass, 61)
|
|
await hass.async_block_till_done()
|
|
assert basic_ch.read_attributes.await_count == 2
|
|
assert basic_ch.read_attributes.await_args[0][0] == ["manufacturer"]
|
|
assert zha_device.available is False
|