Use websocket fixture in deCONZ event related tests (#47823)

Localize test data
Improve asserts
This commit is contained in:
Robert Svensson 2021-03-18 10:44:31 +01:00 committed by GitHub
parent a57d340037
commit 7ff9610e67
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 236 additions and 166 deletions

View File

@ -1,125 +1,164 @@
"""Test deCONZ remote events."""
from copy import deepcopy
from unittest.mock import patch
from homeassistant.components.deconz.const import DOMAIN as DECONZ_DOMAIN
from homeassistant.components.deconz.deconz_event import CONF_DECONZ_EVENT
from homeassistant.components.deconz.gateway import get_gateway_from_config_entry
from homeassistant.const import STATE_UNAVAILABLE
from homeassistant.helpers.device_registry import async_entries_for_config_entry
from .test_gateway import DECONZ_WEB_REQUEST, setup_deconz_integration
from tests.common import async_capture_events
SENSORS = {
"1": {
"id": "Switch 1 id",
"name": "Switch 1",
"type": "ZHASwitch",
"state": {"buttonevent": 1000},
"config": {},
"uniqueid": "00:00:00:00:00:00:00:01-00",
},
"2": {
"id": "Switch 2 id",
"name": "Switch 2",
"type": "ZHASwitch",
"state": {"buttonevent": 1000},
"config": {"battery": 100},
"uniqueid": "00:00:00:00:00:00:00:02-00",
},
"3": {
"id": "Switch 3 id",
"name": "Switch 3",
"type": "ZHASwitch",
"state": {"buttonevent": 1000, "gesture": 1},
"config": {"battery": 100},
"uniqueid": "00:00:00:00:00:00:00:03-00",
},
"4": {
"id": "Switch 4 id",
"name": "Switch 4",
"type": "ZHASwitch",
"state": {"buttonevent": 1000, "gesture": 1},
"config": {"battery": 100},
"uniqueid": "00:00:00:00:00:00:00:04-00",
},
"5": {
"id": "ZHA remote 1 id",
"name": "ZHA remote 1",
"type": "ZHASwitch",
"state": {"angle": 0, "buttonevent": 1000, "xy": [0.0, 0.0]},
"config": {"group": "4,5,6", "reachable": True, "on": True},
"uniqueid": "00:00:00:00:00:00:00:05-00",
},
}
async def test_deconz_events(hass, aioclient_mock):
async def test_deconz_events(hass, aioclient_mock, mock_deconz_websocket):
"""Test successful creation of deconz events."""
data = deepcopy(DECONZ_WEB_REQUEST)
data["sensors"] = deepcopy(SENSORS)
config_entry = await setup_deconz_integration(
hass, aioclient_mock, get_state_response=data
)
gateway = get_gateway_from_config_entry(hass, config_entry)
data = {
"sensors": {
"1": {
"name": "Switch 1",
"type": "ZHASwitch",
"state": {"buttonevent": 1000},
"config": {},
"uniqueid": "00:00:00:00:00:00:00:01-00",
},
"2": {
"name": "Switch 2",
"type": "ZHASwitch",
"state": {"buttonevent": 1000},
"config": {"battery": 100},
"uniqueid": "00:00:00:00:00:00:00:02-00",
},
"3": {
"name": "Switch 3",
"type": "ZHASwitch",
"state": {"buttonevent": 1000, "gesture": 1},
"config": {"battery": 100},
"uniqueid": "00:00:00:00:00:00:00:03-00",
},
"4": {
"name": "Switch 4",
"type": "ZHASwitch",
"state": {"buttonevent": 1000, "gesture": 1},
"config": {"battery": 100},
"uniqueid": "00:00:00:00:00:00:00:04-00",
},
"5": {
"name": "ZHA remote 1",
"type": "ZHASwitch",
"state": {"angle": 0, "buttonevent": 1000, "xy": [0.0, 0.0]},
"config": {"group": "4,5,6", "reachable": True, "on": True},
"uniqueid": "00:00:00:00:00:00:00:05-00",
},
}
}
with patch.dict(DECONZ_WEB_REQUEST, data):
config_entry = await setup_deconz_integration(hass, aioclient_mock)
device_registry = await hass.helpers.device_registry.async_get_registry()
assert len(hass.states.async_all()) == 3
assert len(gateway.events) == 5
assert hass.states.get("sensor.switch_1") is None
assert hass.states.get("sensor.switch_1_battery_level") is None
assert hass.states.get("sensor.switch_2") is None
# 5 switches + 2 additional devices for deconz service and host
assert (
len(async_entries_for_config_entry(device_registry, config_entry.entry_id)) == 7
)
assert hass.states.get("sensor.switch_2_battery_level").state == "100"
assert hass.states.get("sensor.switch_3_battery_level").state == "100"
assert hass.states.get("sensor.switch_4_battery_level").state == "100"
events = async_capture_events(hass, CONF_DECONZ_EVENT)
captured_events = async_capture_events(hass, CONF_DECONZ_EVENT)
gateway.api.sensors["1"].update({"state": {"buttonevent": 2000}})
event_changed_sensor = {
"t": "event",
"e": "changed",
"r": "sensors",
"id": "1",
"state": {"buttonevent": 2000},
}
await mock_deconz_websocket(data=event_changed_sensor)
await hass.async_block_till_done()
assert len(events) == 1
assert events[0].data == {
device = device_registry.async_get_device(
identifiers={(DECONZ_DOMAIN, "00:00:00:00:00:00:00:01")}
)
assert len(captured_events) == 1
assert captured_events[0].data == {
"id": "switch_1",
"unique_id": "00:00:00:00:00:00:00:01",
"event": 2000,
"device_id": gateway.events[0].device_id,
"device_id": device.id,
}
gateway.api.sensors["3"].update({"state": {"buttonevent": 2000}})
event_changed_sensor = {
"t": "event",
"e": "changed",
"r": "sensors",
"id": "3",
"state": {"buttonevent": 2000},
}
await mock_deconz_websocket(data=event_changed_sensor)
await hass.async_block_till_done()
assert len(events) == 2
assert events[1].data == {
device = device_registry.async_get_device(
identifiers={(DECONZ_DOMAIN, "00:00:00:00:00:00:00:03")}
)
assert len(captured_events) == 2
assert captured_events[1].data == {
"id": "switch_3",
"unique_id": "00:00:00:00:00:00:00:03",
"event": 2000,
"gesture": 1,
"device_id": gateway.events[2].device_id,
"device_id": device.id,
}
gateway.api.sensors["4"].update({"state": {"gesture": 0}})
event_changed_sensor = {
"t": "event",
"e": "changed",
"r": "sensors",
"id": "4",
"state": {"gesture": 0},
}
await mock_deconz_websocket(data=event_changed_sensor)
await hass.async_block_till_done()
assert len(events) == 3
assert events[2].data == {
device = device_registry.async_get_device(
identifiers={(DECONZ_DOMAIN, "00:00:00:00:00:00:00:04")}
)
assert len(captured_events) == 3
assert captured_events[2].data == {
"id": "switch_4",
"unique_id": "00:00:00:00:00:00:00:04",
"event": 1000,
"gesture": 0,
"device_id": gateway.events[3].device_id,
"device_id": device.id,
}
gateway.api.sensors["5"].update(
{"state": {"buttonevent": 6002, "angle": 110, "xy": [0.5982, 0.3897]}}
)
event_changed_sensor = {
"t": "event",
"e": "changed",
"r": "sensors",
"id": "5",
"state": {"buttonevent": 6002, "angle": 110, "xy": [0.5982, 0.3897]},
}
await mock_deconz_websocket(data=event_changed_sensor)
await hass.async_block_till_done()
assert len(events) == 4
assert events[3].data == {
device = device_registry.async_get_device(
identifiers={(DECONZ_DOMAIN, "00:00:00:00:00:00:00:05")}
)
assert len(captured_events) == 4
assert captured_events[3].data == {
"id": "zha_remote_1",
"unique_id": "00:00:00:00:00:00:00:05",
"event": 6002,
"angle": 110,
"xy": [0.5982, 0.3897],
"device_id": gateway.events[4].device_id,
"device_id": device.id,
}
await hass.config_entries.async_unload(config_entry.entry_id)
@ -128,9 +167,7 @@ async def test_deconz_events(hass, aioclient_mock):
assert len(hass.states.async_all()) == 3
for state in states:
assert state.state == STATE_UNAVAILABLE
assert len(gateway.events) == 0
await hass.config_entries.async_remove(config_entry.entry_id)
await hass.async_block_till_done()
assert len(hass.states.async_all()) == 0
assert len(gateway.events) == 0

View File

@ -1,11 +1,10 @@
"""deCONZ device automation tests."""
from copy import deepcopy
from unittest.mock import patch
from homeassistant.components.deconz import device_trigger
from homeassistant.components.deconz.const import DOMAIN as DECONZ_DOMAIN
from homeassistant.components.deconz.device_trigger import CONF_SUBTYPE
from homeassistant.components.deconz.gateway import get_gateway_from_config_entry
from homeassistant.components.sensor import DOMAIN as SENSOR_DOMAIN
from homeassistant.const import (
ATTR_BATTERY_LEVEL,
@ -38,7 +37,7 @@ SENSORS = {
"name": "TRÅDFRI on/off switch ",
"state": {"buttonevent": 2002, "lastupdated": "2019-09-07T07:39:39"},
"swversion": "1.4.018",
CONF_TYPE: "ZHASwitch",
"type": "ZHASwitch",
"uniqueid": "d0:cf:5e:ff:fe:71:a4:3a-01-1000",
}
}
@ -46,60 +45,86 @@ SENSORS = {
async def test_get_triggers(hass, aioclient_mock):
"""Test triggers work."""
data = deepcopy(DECONZ_WEB_REQUEST)
data["sensors"] = deepcopy(SENSORS)
config_entry = await setup_deconz_integration(
hass, aioclient_mock, get_state_response=data
data = {
"sensors": {
"1": {
"config": {
"alert": "none",
"battery": 60,
"group": "10",
"on": True,
"reachable": True,
},
"ep": 1,
"etag": "1b355c0b6d2af28febd7ca9165881952",
"manufacturername": "IKEA of Sweden",
"mode": 1,
"modelid": "TRADFRI on/off switch",
"name": "TRÅDFRI on/off switch ",
"state": {"buttonevent": 2002, "lastupdated": "2019-09-07T07:39:39"},
"swversion": "1.4.018",
"type": "ZHASwitch",
"uniqueid": "d0:cf:5e:ff:fe:71:a4:3a-01-1000",
}
}
}
with patch.dict(DECONZ_WEB_REQUEST, data):
await setup_deconz_integration(hass, aioclient_mock)
device_registry = await hass.helpers.device_registry.async_get_registry()
device = device_registry.async_get_device(
identifiers={(DECONZ_DOMAIN, "d0:cf:5e:ff:fe:71:a4:3a")}
)
gateway = get_gateway_from_config_entry(hass, config_entry)
device_id = gateway.events[0].device_id
triggers = await async_get_device_automations(hass, "trigger", device_id)
assert device_trigger._get_deconz_event_from_device_id(hass, device.id)
triggers = await async_get_device_automations(hass, "trigger", device.id)
expected_triggers = [
{
CONF_DEVICE_ID: device_id,
CONF_DEVICE_ID: device.id,
CONF_DOMAIN: DECONZ_DOMAIN,
CONF_PLATFORM: "device",
CONF_TYPE: device_trigger.CONF_SHORT_PRESS,
CONF_SUBTYPE: device_trigger.CONF_TURN_ON,
},
{
CONF_DEVICE_ID: device_id,
CONF_DEVICE_ID: device.id,
CONF_DOMAIN: DECONZ_DOMAIN,
CONF_PLATFORM: "device",
CONF_TYPE: device_trigger.CONF_LONG_PRESS,
CONF_SUBTYPE: device_trigger.CONF_TURN_ON,
},
{
CONF_DEVICE_ID: device_id,
CONF_DEVICE_ID: device.id,
CONF_DOMAIN: DECONZ_DOMAIN,
CONF_PLATFORM: "device",
CONF_TYPE: device_trigger.CONF_LONG_RELEASE,
CONF_SUBTYPE: device_trigger.CONF_TURN_ON,
},
{
CONF_DEVICE_ID: device_id,
CONF_DEVICE_ID: device.id,
CONF_DOMAIN: DECONZ_DOMAIN,
CONF_PLATFORM: "device",
CONF_TYPE: device_trigger.CONF_SHORT_PRESS,
CONF_SUBTYPE: device_trigger.CONF_TURN_OFF,
},
{
CONF_DEVICE_ID: device_id,
CONF_DEVICE_ID: device.id,
CONF_DOMAIN: DECONZ_DOMAIN,
CONF_PLATFORM: "device",
CONF_TYPE: device_trigger.CONF_LONG_PRESS,
CONF_SUBTYPE: device_trigger.CONF_TURN_OFF,
},
{
CONF_DEVICE_ID: device_id,
CONF_DEVICE_ID: device.id,
CONF_DOMAIN: DECONZ_DOMAIN,
CONF_PLATFORM: "device",
CONF_TYPE: device_trigger.CONF_LONG_RELEASE,
CONF_SUBTYPE: device_trigger.CONF_TURN_OFF,
},
{
CONF_DEVICE_ID: device_id,
CONF_DEVICE_ID: device.id,
CONF_DOMAIN: SENSOR_DOMAIN,
ATTR_ENTITY_ID: "sensor.tradfri_on_off_switch_battery_level",
CONF_PLATFORM: "device",
@ -110,27 +135,14 @@ async def test_get_triggers(hass, aioclient_mock):
assert_lists_same(triggers, expected_triggers)
async def test_helper_successful(hass, aioclient_mock):
"""Verify trigger helper."""
data = deepcopy(DECONZ_WEB_REQUEST)
data["sensors"] = deepcopy(SENSORS)
config_entry = await setup_deconz_integration(
hass, aioclient_mock, get_state_response=data
)
gateway = get_gateway_from_config_entry(hass, config_entry)
device_id = gateway.events[0].device_id
deconz_event = device_trigger._get_deconz_event_from_device_id(hass, device_id)
assert deconz_event == gateway.events[0]
async def test_helper_no_match(hass, aioclient_mock):
"""Verify trigger helper returns None when no event could be matched."""
await setup_deconz_integration(hass, aioclient_mock)
deconz_event = device_trigger._get_deconz_event_from_device_id(hass, "mock-id")
assert deconz_event is None
assert not deconz_event
async def test_helper_no_gateway_exist(hass):
"""Verify trigger helper returns None when no gateway exist."""
deconz_event = device_trigger._get_deconz_event_from_device_id(hass, "mock-id")
assert deconz_event is None
assert not deconz_event

View File

@ -1,13 +1,13 @@
"""The tests for deCONZ logbook."""
from copy import deepcopy
from unittest.mock import patch
from homeassistant.components import logbook
from homeassistant.components.deconz.const import CONF_GESTURE
from homeassistant.components.deconz.const import CONF_GESTURE, DOMAIN as DECONZ_DOMAIN
from homeassistant.components.deconz.deconz_event import CONF_DECONZ_EVENT
from homeassistant.components.deconz.gateway import get_gateway_from_config_entry
from homeassistant.const import CONF_DEVICE_ID, CONF_EVENT, CONF_ID, CONF_UNIQUE_ID
from homeassistant.setup import async_setup_component
from homeassistant.util import slugify
from .test_gateway import DECONZ_WEB_REQUEST, setup_deconz_integration
@ -16,47 +16,68 @@ from tests.components.logbook.test_init import MockLazyEventPartialState
async def test_humanifying_deconz_event(hass, aioclient_mock):
"""Test humanifying deCONZ event."""
data = deepcopy(DECONZ_WEB_REQUEST)
data["sensors"] = {
"0": {
"id": "Switch 1 id",
"name": "Switch 1",
"type": "ZHASwitch",
"state": {"buttonevent": 1000},
"config": {},
"uniqueid": "00:00:00:00:00:00:00:01-00",
},
"1": {
"id": "Hue remote id",
"name": "Hue remote",
"type": "ZHASwitch",
"modelid": "RWL021",
"state": {"buttonevent": 1000},
"config": {},
"uniqueid": "00:00:00:00:00:00:00:02-00",
},
"2": {
"id": "Xiaomi cube id",
"name": "Xiaomi cube",
"type": "ZHASwitch",
"modelid": "lumi.sensor_cube",
"state": {"buttonevent": 1000, "gesture": 1},
"config": {},
"uniqueid": "00:00:00:00:00:00:00:03-00",
},
"3": {
"id": "faulty",
"name": "Faulty event",
"type": "ZHASwitch",
"state": {},
"config": {},
"uniqueid": "00:00:00:00:00:00:00:04-00",
},
data = {
"sensors": {
"1": {
"name": "Switch 1",
"type": "ZHASwitch",
"state": {"buttonevent": 1000},
"config": {},
"uniqueid": "00:00:00:00:00:00:00:01-00",
},
"2": {
"name": "Hue remote",
"type": "ZHASwitch",
"modelid": "RWL021",
"state": {"buttonevent": 1000},
"config": {},
"uniqueid": "00:00:00:00:00:00:00:02-00",
},
"3": {
"name": "Xiaomi cube",
"type": "ZHASwitch",
"modelid": "lumi.sensor_cube",
"state": {"buttonevent": 1000, "gesture": 1},
"config": {},
"uniqueid": "00:00:00:00:00:00:00:03-00",
},
"4": {
"name": "Faulty event",
"type": "ZHASwitch",
"state": {},
"config": {},
"uniqueid": "00:00:00:00:00:00:00:04-00",
},
}
}
config_entry = await setup_deconz_integration(
hass, aioclient_mock, get_state_response=data
with patch.dict(DECONZ_WEB_REQUEST, data):
await setup_deconz_integration(hass, aioclient_mock)
device_registry = await hass.helpers.device_registry.async_get_registry()
switch_event_id = slugify(data["sensors"]["1"]["name"])
switch_serial = data["sensors"]["1"]["uniqueid"].split("-", 1)[0]
switch_entry = device_registry.async_get_device(
identifiers={(DECONZ_DOMAIN, switch_serial)}
)
hue_remote_event_id = slugify(data["sensors"]["2"]["name"])
hue_remote_serial = data["sensors"]["2"]["uniqueid"].split("-", 1)[0]
hue_remote_entry = device_registry.async_get_device(
identifiers={(DECONZ_DOMAIN, hue_remote_serial)}
)
xiaomi_cube_event_id = slugify(data["sensors"]["3"]["name"])
xiaomi_cube_serial = data["sensors"]["3"]["uniqueid"].split("-", 1)[0]
xiaomi_cube_entry = device_registry.async_get_device(
identifiers={(DECONZ_DOMAIN, xiaomi_cube_serial)}
)
faulty_event_id = slugify(data["sensors"]["4"]["name"])
faulty_serial = data["sensors"]["4"]["uniqueid"].split("-", 1)[0]
faulty_entry = device_registry.async_get_device(
identifiers={(DECONZ_DOMAIN, faulty_serial)}
)
gateway = get_gateway_from_config_entry(hass, config_entry)
hass.config.components.add("recorder")
assert await async_setup_component(hass, "logbook", {})
@ -70,50 +91,50 @@ async def test_humanifying_deconz_event(hass, aioclient_mock):
MockLazyEventPartialState(
CONF_DECONZ_EVENT,
{
CONF_DEVICE_ID: gateway.events[0].device_id,
CONF_DEVICE_ID: switch_entry.id,
CONF_EVENT: 2000,
CONF_ID: gateway.events[0].event_id,
CONF_UNIQUE_ID: gateway.events[0].serial,
CONF_ID: switch_event_id,
CONF_UNIQUE_ID: switch_serial,
},
),
# Event with matching device trigger
MockLazyEventPartialState(
CONF_DECONZ_EVENT,
{
CONF_DEVICE_ID: gateway.events[1].device_id,
CONF_DEVICE_ID: hue_remote_entry.id,
CONF_EVENT: 2001,
CONF_ID: gateway.events[1].event_id,
CONF_UNIQUE_ID: gateway.events[1].serial,
CONF_ID: hue_remote_event_id,
CONF_UNIQUE_ID: hue_remote_serial,
},
),
# Gesture with matching device trigger
MockLazyEventPartialState(
CONF_DECONZ_EVENT,
{
CONF_DEVICE_ID: gateway.events[2].device_id,
CONF_DEVICE_ID: xiaomi_cube_entry.id,
CONF_GESTURE: 1,
CONF_ID: gateway.events[2].event_id,
CONF_UNIQUE_ID: gateway.events[2].serial,
CONF_ID: xiaomi_cube_event_id,
CONF_UNIQUE_ID: xiaomi_cube_serial,
},
),
# Unsupported device trigger
MockLazyEventPartialState(
CONF_DECONZ_EVENT,
{
CONF_DEVICE_ID: gateway.events[2].device_id,
CONF_DEVICE_ID: xiaomi_cube_entry.id,
CONF_GESTURE: "unsupported_gesture",
CONF_ID: gateway.events[2].event_id,
CONF_UNIQUE_ID: gateway.events[2].serial,
CONF_ID: xiaomi_cube_event_id,
CONF_UNIQUE_ID: xiaomi_cube_serial,
},
),
# Unknown event
MockLazyEventPartialState(
CONF_DECONZ_EVENT,
{
CONF_DEVICE_ID: gateway.events[3].device_id,
CONF_DEVICE_ID: faulty_entry.id,
"unknown_event": None,
CONF_ID: gateway.events[3].event_id,
CONF_UNIQUE_ID: gateway.events[3].serial,
CONF_ID: faulty_event_id,
CONF_UNIQUE_ID: faulty_serial,
},
),
],