Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

deCONZ rewrite sensor tests () 

* Improve binary sensor tests

* Fix sensor tests

* Improve readability of binary sensor

* Fix climate tests
Fix sensor platform not loading climate devices as sensors

* Add test to verify adding new sensor after start up
pull/26736/head
Robert Svensson 2019-09-18 19:07:32 +02:00 committed by Paulus Schoutsen
parent ce42b46ccd
commit 886d8bd6e2
4 changed files with 527 additions and 310 deletions
homeassistant/components/deconz

4
homeassistant/components/deconz/sensor.py
View File

@ -1,5 +1,5 @@
"""Support for deCONZ sensors."""
from pydeconz.sensor import Consumption, Daylight, LightLevel, Power, Switch
from pydeconz.sensor import Consumption, Daylight, LightLevel, Power, Switch, Thermostat
from homeassistant.const import ATTR_TEMPERATURE, ATTR_VOLTAGE, DEVICE_CLASS_BATTERY
from homeassistant.core import callback
@ -48,7 +48,7 @@ async def async_setup_entry(hass, config_entry, async_add_entities):
hass.async_create_task(new_event.async_update_device_registry())
gateway.events.append(new_event)
elif not sensor.BINARY:
elif not sensor.BINARY and sensor.type not in Thermostat.ZHATYPE:
new_sensor = DeconzSensor(sensor, gateway)
entity_handler.add_entity(new_sensor)

213
tests/components/deconz/test_binary_sensor.py
View File

@ -1,79 +1,98 @@
"""deCONZ binary sensor platform tests."""
from unittest.mock import Mock, patch
from copy import deepcopy
from tests.common import mock_coro
from asynctest import patch
from homeassistant import config_entries
from homeassistant.components import deconz
from homeassistant.helpers.dispatcher import async_dispatcher_send
from homeassistant.setup import async_setup_component
import homeassistant.components.binary_sensor as binary_sensor
SENSOR = {
SENSORS = {
"1": {
"id": "Sensor 1 id",
"name": "Sensor 1 name",
"id": "Presence sensor id",
"name": "Presence sensor",
"type": "ZHAPresence",
"state": {"presence": False},
"config": {},
"state": {"dark": False, "presence": False},
"config": {"on": True, "reachable": True, "temperature": 10},
"uniqueid": "00:00:00:00:00:00:00:00-00",
},
"2": {
"id": "Sensor 2 id",
"name": "Sensor 2 name",
"id": "Temperature sensor id",
"name": "Temperature sensor",
"type": "ZHATemperature",
"state": {"temperature": False},
"config": {},
"uniqueid": "00:00:00:00:00:00:00:01-00",
},
"3": {
"id": "CLIP presence sensor id",
"name": "CLIP presence sensor",
"type": "CLIPPresence",
"state": {},
"config": {},
"uniqueid": "00:00:00:00:00:00:00:02-00",
},
"4": {
"id": "Vibration sensor id",
"name": "Vibration sensor",
"type": "ZHAVibration",
"state": {
"orientation": [1, 2, 3],
"tiltangle": 36,
"vibration": True,
"vibrationstrength": 10,
},
"config": {"on": True, "reachable": True, "temperature": 10},
"uniqueid": "00:00:00:00:00:00:00:03-00",
},
}
BRIDGEID = "0123456789"
ENTRY_CONFIG = {
deconz.config_flow.CONF_API_KEY: "ABCDEF",
deconz.config_flow.CONF_BRIDGEID: "0123456789",
deconz.config_flow.CONF_BRIDGEID: BRIDGEID,
deconz.config_flow.CONF_HOST: "1.2.3.4",
deconz.config_flow.CONF_PORT: 80,
}
ENTRY_OPTIONS = {
deconz.const.CONF_ALLOW_CLIP_SENSOR: True,
deconz.const.CONF_ALLOW_DECONZ_GROUPS: True,
DECONZ_CONFIG = {
"bridgeid": BRIDGEID,
"mac": "00:11:22:33:44:55",
"name": "deCONZ mock gateway",
"sw_version": "2.05.69",
"websocketport": 1234,
}
DECONZ_WEB_REQUEST = {"config": DECONZ_CONFIG}
async def setup_gateway(hass, data, allow_clip_sensor=True):
"""Load the deCONZ binary sensor platform."""
from pydeconz import DeconzSession
loop = Mock()
session = Mock()
ENTRY_OPTIONS[deconz.const.CONF_ALLOW_CLIP_SENSOR] = allow_clip_sensor
async def setup_deconz_integration(hass, config, options, get_state_response):
"""Create the deCONZ gateway."""
config_entry = config_entries.ConfigEntry(
1,
deconz.DOMAIN,
"Mock Title",
ENTRY_CONFIG,
"test",
config_entries.CONN_CLASS_LOCAL_PUSH,
version=1,
domain=deconz.DOMAIN,
title="Mock Title",
data=config,
source="test",
connection_class=config_entries.CONN_CLASS_LOCAL_PUSH,
system_options={},
options=ENTRY_OPTIONS,
options=options,
entry_id="1",
)
gateway = deconz.DeconzGateway(hass, config_entry)
gateway.api = DeconzSession(loop, session, **config_entry.data)
gateway.api.config = Mock()
hass.data[deconz.DOMAIN] = {gateway.bridgeid: gateway}
with patch("pydeconz.DeconzSession.async_get_state", return_value=mock_coro(data)):
await gateway.api.async_load_parameters()
await hass.config_entries.async_forward_entry_setup(config_entry, "binary_sensor")
# To flush out the service call to update the group
with patch(
"pydeconz.DeconzSession.async_get_state", return_value=get_state_response
), patch("pydeconz.DeconzSession.start", return_value=True):
await deconz.async_setup_entry(hass, config_entry)
await hass.async_block_till_done()
return gateway
hass.config_entries._entries.append(config_entry)
return hass.data[deconz.DOMAIN][config[deconz.CONF_BRIDGEID]]
async def test_platform_manually_configured(hass):
@ -89,58 +108,94 @@ async def test_platform_manually_configured(hass):
async def test_no_binary_sensors(hass):
"""Test that no sensors in deconz results in no sensor entities."""
data = {}
gateway = await setup_gateway(hass, data)
assert len(hass.data[deconz.DOMAIN][gateway.bridgeid].deconz_ids) == 0
data = deepcopy(DECONZ_WEB_REQUEST)
gateway = await setup_deconz_integration(
hass, ENTRY_CONFIG, options={}, get_state_response=data
)
assert len(gateway.deconz_ids) == 0
assert len(hass.states.async_all()) == 0
async def test_binary_sensors(hass):
"""Test successful creation of binary sensor entities."""
data = {"sensors": SENSOR}
gateway = await setup_gateway(hass, data)
assert "binary_sensor.sensor_1_name" in gateway.deconz_ids
assert "binary_sensor.sensor_2_name" not in gateway.deconz_ids
assert len(hass.states.async_all()) == 1
hass.data[deconz.DOMAIN][gateway.bridgeid].api.sensors["1"].async_update(
{"state": {"on": False}}
data = deepcopy(DECONZ_WEB_REQUEST)
data["sensors"] = deepcopy(SENSORS)
gateway = await setup_deconz_integration(
hass, ENTRY_CONFIG, options={}, get_state_response=data
)
assert "binary_sensor.presence_sensor" in gateway.deconz_ids
assert "binary_sensor.temperature_sensor" not in gateway.deconz_ids
assert "binary_sensor.clip_presence_sensor" not in gateway.deconz_ids
assert "binary_sensor.vibration_sensor" in gateway.deconz_ids
assert len(hass.states.async_all()) == 3
presence_sensor = hass.states.get("binary_sensor.presence_sensor")
assert presence_sensor.state == "off"
async def test_add_new_sensor(hass):
"""Test successful creation of sensor entities."""
data = {}
gateway = await setup_gateway(hass, data)
sensor = Mock()
sensor.name = "name"
sensor.type = "ZHAPresence"
sensor.BINARY = True
sensor.uniqueid = "1"
sensor.register_async_callback = Mock()
async_dispatcher_send(hass, gateway.async_signal_new_device("sensor"), [sensor])
temperature_sensor = hass.states.get("binary_sensor.temperature_sensor")
assert temperature_sensor is None
clip_presence_sensor = hass.states.get("binary_sensor.clip_presence_sensor")
assert clip_presence_sensor is None
vibration_sensor = hass.states.get("binary_sensor.vibration_sensor")
assert vibration_sensor.state == "on"
gateway.api.sensors["1"].async_update({"state": {"presence": True}})
await hass.async_block_till_done()
assert "binary_sensor.name" in gateway.deconz_ids
presence_sensor = hass.states.get("binary_sensor.presence_sensor")
assert presence_sensor.state == "on"
async def test_do_not_allow_clip_sensor(hass):
"""Test that clip sensors can be ignored."""
data = {}
gateway = await setup_gateway(hass, data, allow_clip_sensor=False)
sensor = Mock()
sensor.name = "name"
sensor.type = "CLIPPresence"
sensor.register_async_callback = Mock()
async_dispatcher_send(hass, gateway.async_signal_new_device("sensor"), [sensor])
await hass.async_block_till_done()
async def test_allow_clip_sensor(hass):
"""Test that CLIP sensors can be allowed."""
data = deepcopy(DECONZ_WEB_REQUEST)
data["sensors"] = deepcopy(SENSORS)
gateway = await setup_deconz_integration(
hass,
ENTRY_CONFIG,
options={deconz.gateway.CONF_ALLOW_CLIP_SENSOR: True},
get_state_response=data,
)
assert "binary_sensor.presence_sensor" in gateway.deconz_ids
assert "binary_sensor.temperature_sensor" not in gateway.deconz_ids
assert "binary_sensor.clip_presence_sensor" in gateway.deconz_ids
assert "binary_sensor.vibration_sensor" in gateway.deconz_ids
assert len(hass.states.async_all()) == 4
presence_sensor = hass.states.get("binary_sensor.presence_sensor")
assert presence_sensor.state == "off"
temperature_sensor = hass.states.get("binary_sensor.temperature_sensor")
assert temperature_sensor is None
clip_presence_sensor = hass.states.get("binary_sensor.clip_presence_sensor")
assert clip_presence_sensor.state == "off"
vibration_sensor = hass.states.get("binary_sensor.vibration_sensor")
assert vibration_sensor.state == "on"
async def test_add_new_binary_sensor(hass):
"""Test that adding a new binary sensor works."""
data = deepcopy(DECONZ_WEB_REQUEST)
gateway = await setup_deconz_integration(
hass, ENTRY_CONFIG, options={}, get_state_response=data
)
assert len(gateway.deconz_ids) == 0
state_added = {
"t": "event",
"e": "added",
"r": "sensors",
"id": "1",
"sensor": deepcopy(SENSORS["1"]),
}
gateway.api.async_event_handler(state_added)
await hass.async_block_till_done()
async def test_unload_switch(hass):
"""Test that it works to unload switch entities."""
data = {"sensors": SENSOR}
gateway = await setup_gateway(hass, data)
assert "binary_sensor.presence_sensor" in gateway.deconz_ids
await gateway.async_reset()
assert len(hass.states.async_all()) == 0
presence_sensor = hass.states.get("binary_sensor.presence_sensor")
assert presence_sensor.state == "off"

319
tests/components/deconz/test_climate.py
View File

@ -1,23 +1,18 @@
"""deCONZ climate platform tests."""
from copy import deepcopy
from unittest.mock import Mock, patch
import asynctest
from asynctest import patch
from homeassistant import config_entries
from homeassistant.components import deconz
from homeassistant.helpers.dispatcher import async_dispatcher_send
from homeassistant.setup import async_setup_component
import homeassistant.components.climate as climate
from tests.common import mock_coro
SENSOR = {
SENSORS = {
"1": {
"id": "Climate 1 id",
"name": "Climate 1 name",
"id": "Thermostat id",
"name": "Thermostat",
"type": "ZHAThermostat",
"state": {"on": True, "temperature": 2260, "valve": 30},
"config": {
@ -30,62 +25,66 @@ SENSOR = {
"uniqueid": "00:00:00:00:00:00:00:00-00",
},
"2": {
"id": "Sensor 2 id",
"name": "Sensor 2 name",
"id": "Presence sensor id",
"name": "Presence sensor",
"type": "ZHAPresence",
"state": {"presence": False},
"config": {},
"config": {"reachable": True},
"uniqueid": "00:00:00:00:00:00:00:01-00",
},
"3": {
"id": "CLIP thermostat id",
"name": "CLIP thermostat",
"type": "CLIPThermostat",
"state": {"on": True, "temperature": 2260, "valve": 30},
"config": {"reachable": True},
"uniqueid": "00:00:00:00:00:00:00:02-00",
},
}
BRIDGEID = "0123456789"
ENTRY_CONFIG = {
deconz.config_flow.CONF_API_KEY: "ABCDEF",
deconz.config_flow.CONF_BRIDGEID: "0123456789",
deconz.config_flow.CONF_BRIDGEID: BRIDGEID,
deconz.config_flow.CONF_HOST: "1.2.3.4",
deconz.config_flow.CONF_PORT: 80,
}
ENTRY_OPTIONS = {
deconz.const.CONF_ALLOW_CLIP_SENSOR: True,
deconz.const.CONF_ALLOW_DECONZ_GROUPS: True,
DECONZ_CONFIG = {
"bridgeid": BRIDGEID,
"mac": "00:11:22:33:44:55",
"name": "deCONZ mock gateway",
"sw_version": "2.05.69",
"websocketport": 1234,
}
DECONZ_WEB_REQUEST = {"config": DECONZ_CONFIG}
async def setup_gateway(hass, data, allow_clip_sensor=True):
"""Load the deCONZ sensor platform."""
from pydeconz import DeconzSession
response = Mock(
status=200, json=asynctest.CoroutineMock(), text=asynctest.CoroutineMock()
)
response.content_type = "application/json"
session = Mock(put=asynctest.CoroutineMock(return_value=response))
ENTRY_OPTIONS[deconz.const.CONF_ALLOW_CLIP_SENSOR] = allow_clip_sensor
async def setup_deconz_integration(hass, config, options, get_state_response):
"""Create the deCONZ gateway."""
config_entry = config_entries.ConfigEntry(
1,
deconz.DOMAIN,
"Mock Title",
ENTRY_CONFIG,
"test",
config_entries.CONN_CLASS_LOCAL_PUSH,
version=1,
domain=deconz.DOMAIN,
title="Mock Title",
data=config,
source="test",
connection_class=config_entries.CONN_CLASS_LOCAL_PUSH,
system_options={},
options=ENTRY_OPTIONS,
options=options,
entry_id="1",
)
gateway = deconz.DeconzGateway(hass, config_entry)
gateway.api = DeconzSession(hass.loop, session, **config_entry.data)
gateway.api.config = Mock()
hass.data[deconz.DOMAIN] = {gateway.bridgeid: gateway}
with patch("pydeconz.DeconzSession.async_get_state", return_value=mock_coro(data)):
await gateway.api.async_load_parameters()
await hass.config_entries.async_forward_entry_setup(config_entry, "climate")
# To flush out the service call to update the group
with patch(
"pydeconz.DeconzSession.async_get_state", return_value=get_state_response
), patch("pydeconz.DeconzSession.start", return_value=True):
await deconz.async_setup_entry(hass, config_entry)
await hass.async_block_till_done()
return gateway
hass.config_entries._entries.append(config_entry)
return hass.data[deconz.DOMAIN][config[deconz.CONF_BRIDGEID]]
async def test_platform_manually_configured(hass):
@ -101,69 +100,155 @@ async def test_platform_manually_configured(hass):
async def test_no_sensors(hass):
"""Test that no sensors in deconz results in no climate entities."""
gateway = await setup_gateway(hass, {})
assert not hass.data[deconz.DOMAIN][gateway.bridgeid].deconz_ids
assert not hass.states.async_all()
data = deepcopy(DECONZ_WEB_REQUEST)
gateway = await setup_deconz_integration(
hass, ENTRY_CONFIG, options={}, get_state_response=data
)
assert len(gateway.deconz_ids) == 0
assert len(hass.states.async_all()) == 0
async def test_climate_devices(hass):
"""Test successful creation of sensor entities."""
gateway = await setup_gateway(hass, {"sensors": deepcopy(SENSOR)})
assert "climate.climate_1_name" in gateway.deconz_ids
assert "sensor.sensor_2_name" not in gateway.deconz_ids
assert len(hass.states.async_all()) == 1
data = deepcopy(DECONZ_WEB_REQUEST)
data["sensors"] = deepcopy(SENSORS)
gateway = await setup_deconz_integration(
hass, ENTRY_CONFIG, options={}, get_state_response=data
)
assert "climate.thermostat" in gateway.deconz_ids
assert "sensor.thermostat" not in gateway.deconz_ids
assert "sensor.thermostat_battery_level" in gateway.deconz_ids
assert "climate.presence_sensor" not in gateway.deconz_ids
assert "climate.clip_thermostat" not in gateway.deconz_ids
assert len(hass.states.async_all()) == 3
gateway.api.sensors["1"].async_update({"state": {"on": False}})
thermostat = hass.states.get("climate.thermostat")
assert thermostat.state == "auto"
await hass.services.async_call(
"climate",
"set_hvac_mode",
{"entity_id": "climate.climate_1_name", "hvac_mode": "auto"},
blocking=True,
)
gateway.api.session.put.assert_called_with(
"http://1.2.3.4:80/api/ABCDEF/sensors/1/config", data='{"mode": "auto"}'
)
thermostat = hass.states.get("sensor.thermostat")
assert thermostat is None
await hass.services.async_call(
"climate",
"set_hvac_mode",
{"entity_id": "climate.climate_1_name", "hvac_mode": "heat"},
blocking=True,
)
gateway.api.session.put.assert_called_with(
"http://1.2.3.4:80/api/ABCDEF/sensors/1/config", data='{"mode": "heat"}'
)
thermostat_battery_level = hass.states.get("sensor.thermostat_battery_level")
assert thermostat_battery_level.state == "100"
await hass.services.async_call(
"climate",
"set_hvac_mode",
{"entity_id": "climate.climate_1_name", "hvac_mode": "off"},
blocking=True,
)
gateway.api.session.put.assert_called_with(
"http://1.2.3.4:80/api/ABCDEF/sensors/1/config", data='{"mode": "off"}'
)
presence_sensor = hass.states.get("climate.presence_sensor")
assert presence_sensor is None
await hass.services.async_call(
"climate",
"set_temperature",
{"entity_id": "climate.climate_1_name", "temperature": 20},
blocking=True,
)
gateway.api.session.put.assert_called_with(
"http://1.2.3.4:80/api/ABCDEF/sensors/1/config", data='{"heatsetpoint": 2000.0}'
)
clip_thermostat = hass.states.get("climate.clip_thermostat")
assert clip_thermostat is None
assert len(gateway.api.session.put.mock_calls) == 4
thermostat_device = gateway.api.sensors["1"]
thermostat_device.async_update({"config": {"mode": "off"}})
await hass.async_block_till_done()
thermostat = hass.states.get("climate.thermostat")
assert thermostat.state == "off"
thermostat_device.async_update({"config": {"mode": "other"}, "state": {"on": True}})
await hass.async_block_till_done()
thermostat = hass.states.get("climate.thermostat")
assert thermostat.state == "heat"
thermostat_device.async_update({"state": {"on": False}})
await hass.async_block_till_done()
thermostat = hass.states.get("climate.thermostat")
assert thermostat.state == "off"
# Verify service calls
with patch.object(
thermostat_device, "_async_set_callback", return_value=True
) as set_callback:
await hass.services.async_call(
climate.DOMAIN,
climate.SERVICE_SET_HVAC_MODE,
{"entity_id": "climate.thermostat", "hvac_mode": "auto"},
blocking=True,
)
await hass.async_block_till_done()
set_callback.assert_called_with("/sensors/1/config", {"mode": "auto"})
with patch.object(
thermostat_device, "_async_set_callback", return_value=True
) as set_callback:
await hass.services.async_call(
climate.DOMAIN,
climate.SERVICE_SET_HVAC_MODE,
{"entity_id": "climate.thermostat", "hvac_mode": "heat"},
blocking=True,
)
await hass.async_block_till_done()
set_callback.assert_called_with("/sensors/1/config", {"mode": "heat"})
with patch.object(
thermostat_device, "_async_set_callback", return_value=True
) as set_callback:
await hass.services.async_call(
climate.DOMAIN,
climate.SERVICE_SET_HVAC_MODE,
{"entity_id": "climate.thermostat", "hvac_mode": "off"},
blocking=True,
)
set_callback.assert_called_with("/sensors/1/config", {"mode": "off"})
with patch.object(
thermostat_device, "_async_set_callback", return_value=True
) as set_callback:
await hass.services.async_call(
climate.DOMAIN,
climate.SERVICE_SET_TEMPERATURE,
{"entity_id": "climate.thermostat", "temperature": 20},
blocking=True,
)
set_callback.assert_called_with("/sensors/1/config", {"heatsetpoint": 2000.0})
async def test_clip_climate_device(hass):
"""Test successful creation of sensor entities."""
data = deepcopy(DECONZ_WEB_REQUEST)
data["sensors"] = deepcopy(SENSORS)
gateway = await setup_deconz_integration(
hass,
ENTRY_CONFIG,
options={deconz.gateway.CONF_ALLOW_CLIP_SENSOR: True},
get_state_response=data,
)
assert "climate.thermostat" in gateway.deconz_ids
assert "sensor.thermostat" not in gateway.deconz_ids
assert "sensor.thermostat_battery_level" in gateway.deconz_ids
assert "climate.presence_sensor" not in gateway.deconz_ids
assert "climate.clip_thermostat" in gateway.deconz_ids
assert len(hass.states.async_all()) == 4
thermostat = hass.states.get("climate.thermostat")
assert thermostat.state == "auto"
thermostat = hass.states.get("sensor.thermostat")
assert thermostat is None
thermostat_battery_level = hass.states.get("sensor.thermostat_battery_level")
assert thermostat_battery_level.state == "100"
presence_sensor = hass.states.get("climate.presence_sensor")
assert presence_sensor is None
clip_thermostat = hass.states.get("climate.clip_thermostat")
assert clip_thermostat.state == "heat"
async def test_verify_state_update(hass):
"""Test that state update properly."""
gateway = await setup_gateway(hass, {"sensors": deepcopy(SENSOR)})
assert "climate.climate_1_name" in gateway.deconz_ids
data = deepcopy(DECONZ_WEB_REQUEST)
data["sensors"] = deepcopy(SENSORS)
gateway = await setup_deconz_integration(
hass, ENTRY_CONFIG, options={}, get_state_response=data
)
assert "climate.thermostat" in gateway.deconz_ids
thermostat = hass.states.get("climate.climate_1_name")
thermostat = hass.states.get("climate.thermostat")
assert thermostat.state == "auto"
state_update = {
@ -174,44 +259,32 @@ async def test_verify_state_update(hass):
"state": {"on": False},
}
gateway.api.async_event_handler(state_update)
await hass.async_block_till_done()
assert len(hass.states.async_all()) == 1
thermostat = hass.states.get("climate.climate_1_name")
thermostat = hass.states.get("climate.thermostat")
assert thermostat.state == "auto"
assert gateway.api.sensors["1"].changed_keys == {"state", "r", "t", "on", "e", "id"}
async def test_add_new_climate_device(hass):
"""Test successful creation of climate entities."""
gateway = await setup_gateway(hass, {})
sensor = Mock()
sensor.name = "name"
sensor.type = "ZHAThermostat"
sensor.uniqueid = "1"
sensor.register_async_callback = Mock()
async_dispatcher_send(hass, gateway.async_signal_new_device("sensor"), [sensor])
await hass.async_block_till_done()
assert "climate.name" in gateway.deconz_ids
async def test_do_not_allow_clipsensor(hass):
"""Test that clip sensors can be ignored."""
gateway = await setup_gateway(hass, {}, allow_clip_sensor=False)
sensor = Mock()
sensor.name = "name"
sensor.type = "CLIPThermostat"
sensor.register_async_callback = Mock()
async_dispatcher_send(hass, gateway.async_signal_new_device("sensor"), [sensor])
await hass.async_block_till_done()
"""Test that adding a new climate device works."""
data = deepcopy(DECONZ_WEB_REQUEST)
gateway = await setup_deconz_integration(
hass, ENTRY_CONFIG, options={}, get_state_response=data
)
assert len(gateway.deconz_ids) == 0
state_added = {
"t": "event",
"e": "added",
"r": "sensors",
"id": "1",
"sensor": deepcopy(SENSORS["1"]),
}
gateway.api.async_event_handler(state_added)
await hass.async_block_till_done()
async def test_unload_sensor(hass):
"""Test that it works to unload sensor entities."""
gateway = await setup_gateway(hass, {"sensors": SENSOR})
assert "climate.thermostat" in gateway.deconz_ids
await gateway.async_reset()
assert len(hass.states.async_all()) == 0
thermostat = hass.states.get("climate.thermostat")
assert thermostat.state == "auto"

301
tests/components/deconz/test_sensor.py
View File

@ -1,123 +1,125 @@
"""deCONZ sensor platform tests."""
from unittest.mock import Mock, patch
from copy import deepcopy
from tests.common import mock_coro
from asynctest import patch
from homeassistant import config_entries
from homeassistant.components import deconz
from homeassistant.helpers.dispatcher import async_dispatcher_send
from homeassistant.setup import async_setup_component
import homeassistant.components.sensor as sensor
SENSOR = {
SENSORS = {
"1": {
"id": "Sensor 1 id",
"name": "Sensor 1 name",
"id": "Light sensor id",
"name": "Light level sensor",
"type": "ZHALightLevel",
"state": {"lightlevel": 30000, "dark": False},
"config": {"reachable": True},
"config": {"on": True, "reachable": True, "temperature": 10},
"uniqueid": "00:00:00:00:00:00:00:00-00",
},
"2": {
"id": "Sensor 2 id",
"name": "Sensor 2 name",
"id": "Presence sensor id",
"name": "Presence sensor",
"type": "ZHAPresence",
"state": {"presence": False},
"config": {},
},
"3": {
"id": "Sensor 3 id",
"name": "Sensor 3 name",
"type": "ZHASwitch",
"state": {"buttonevent": 1000},
"config": {},
},
"4": {
"id": "Sensor 4 id",
"name": "Sensor 4 name",
"type": "ZHASwitch",
"state": {"buttonevent": 1000},
"config": {"battery": 100},
"uniqueid": "00:00:00:00:00:00:00:01-00",
},
"5": {
"id": "Sensor 5 id",
"name": "Sensor 5 name",
"3": {
"id": "Switch 1 id",
"name": "Switch 1",
"type": "ZHASwitch",
"state": {"buttonevent": 1000},
"config": {},
"uniqueid": "00:00:00:00:00:00:00:02-00",
},
"4": {
"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-00",
"uniqueid": "00:00:00:00:00:00:00:03-00",
},
"5": {
"id": "Daylight sensor id",
"name": "Daylight sensor",
"type": "Daylight",
"state": {"daylight": True, "status": 130},
"config": {},
"uniqueid": "00:00:00:00:00:00:00:04-00",
},
"6": {
"id": "Sensor 6 id",
"name": "Sensor 6 name",
"type": "Daylight",
"state": {"daylight": True},
"config": {},
},
"7": {
"id": "Sensor 7 id",
"name": "Sensor 7 name",
"id": "Power sensor id",
"name": "Power sensor",
"type": "ZHAPower",
"state": {"current": 2, "power": 6, "voltage": 3},
"config": {"reachable": True},
"uniqueid": "00:00:00:00:00:00:00:05-00",
},
"8": {
"id": "Sensor 8 id",
"name": "Sensor 8 name",
"7": {
"id": "Consumption id",
"name": "Consumption sensor",
"type": "ZHAConsumption",
"state": {"consumption": 2, "power": 6},
"config": {"reachable": True},
"uniqueid": "00:00:00:00:00:00:00:06-00",
},
"8": {
"id": "CLIP light sensor id",
"name": "CLIP light level sensor",
"type": "CLIPLightLevel",
"state": {"lightlevel": 30000},
"config": {"reachable": True},
"uniqueid": "00:00:00:00:00:00:00:07-00",
},
}
BRIDGEID = "0123456789"
ENTRY_CONFIG = {
deconz.config_flow.CONF_API_KEY: "ABCDEF",
deconz.config_flow.CONF_BRIDGEID: "0123456789",
deconz.config_flow.CONF_BRIDGEID: BRIDGEID,
deconz.config_flow.CONF_HOST: "1.2.3.4",
deconz.config_flow.CONF_PORT: 80,
}
ENTRY_OPTIONS = {
deconz.const.CONF_ALLOW_CLIP_SENSOR: True,
deconz.const.CONF_ALLOW_DECONZ_GROUPS: True,
DECONZ_CONFIG = {
"bridgeid": BRIDGEID,
"mac": "00:11:22:33:44:55",
"name": "deCONZ mock gateway",
"sw_version": "2.05.69",
"websocketport": 1234,
}
DECONZ_WEB_REQUEST = {"config": DECONZ_CONFIG}
async def setup_gateway(hass, data, allow_clip_sensor=True):
"""Load the deCONZ sensor platform."""
from pydeconz import DeconzSession
loop = Mock()
session = Mock()
ENTRY_OPTIONS[deconz.const.CONF_ALLOW_CLIP_SENSOR] = allow_clip_sensor
async def setup_deconz_integration(hass, config, options, get_state_response):
"""Create the deCONZ gateway."""
config_entry = config_entries.ConfigEntry(
1,
deconz.DOMAIN,
"Mock Title",
ENTRY_CONFIG,
"test",
config_entries.CONN_CLASS_LOCAL_PUSH,
version=1,
domain=deconz.DOMAIN,
title="Mock Title",
data=config,
source="test",
connection_class=config_entries.CONN_CLASS_LOCAL_PUSH,
system_options={},
options=ENTRY_OPTIONS,
options=options,
entry_id="1",
)
gateway = deconz.DeconzGateway(hass, config_entry)
gateway.api = DeconzSession(loop, session, **config_entry.data)
gateway.api.config = Mock()
hass.data[deconz.DOMAIN] = {gateway.bridgeid: gateway}
with patch("pydeconz.DeconzSession.async_get_state", return_value=mock_coro(data)):
await gateway.api.async_load_parameters()
await hass.config_entries.async_forward_entry_setup(config_entry, "sensor")
# To flush out the service call to update the group
with patch(
"pydeconz.DeconzSession.async_get_state", return_value=get_state_response
), patch("pydeconz.DeconzSession.start", return_value=True):
await deconz.async_setup_entry(hass, config_entry)
await hass.async_block_till_done()
return gateway
hass.config_entries._entries.append(config_entry)
return hass.data[deconz.DOMAIN][config[deconz.CONF_BRIDGEID]]
async def test_platform_manually_configured(hass):
@ -133,56 +135,143 @@ async def test_platform_manually_configured(hass):
async def test_no_sensors(hass):
"""Test that no sensors in deconz results in no sensor entities."""
gateway = await setup_gateway(hass, {})
assert not hass.data[deconz.DOMAIN][gateway.bridgeid].deconz_ids
data = deepcopy(DECONZ_WEB_REQUEST)
gateway = await setup_deconz_integration(
hass, ENTRY_CONFIG, options={}, get_state_response=data
)
assert len(gateway.deconz_ids) == 0
assert len(hass.states.async_all()) == 0
async def test_sensors(hass):
"""Test successful creation of sensor entities."""
gateway = await setup_gateway(hass, {"sensors": SENSOR})
assert "sensor.sensor_1_name" in gateway.deconz_ids
assert "sensor.sensor_2_name" not in gateway.deconz_ids
assert "sensor.sensor_3_name" not in gateway.deconz_ids
assert "sensor.sensor_3_name_battery_level" not in gateway.deconz_ids
assert "sensor.sensor_4_name" not in gateway.deconz_ids
assert "sensor.sensor_4_name_battery_level" in gateway.deconz_ids
data = deepcopy(DECONZ_WEB_REQUEST)
data["sensors"] = deepcopy(SENSORS)
gateway = await setup_deconz_integration(
hass, ENTRY_CONFIG, options={}, get_state_response=data
)
assert "sensor.light_level_sensor" in gateway.deconz_ids
assert "sensor.presence_sensor" not in gateway.deconz_ids
assert "sensor.switch_1" not in gateway.deconz_ids
assert "sensor.switch_1_battery_level" not in gateway.deconz_ids
assert "sensor.switch_2" not in gateway.deconz_ids
assert "sensor.switch_2_battery_level" in gateway.deconz_ids
assert "sensor.daylight_sensor" in gateway.deconz_ids
assert "sensor.power_sensor" in gateway.deconz_ids
assert "sensor.consumption_sensor" in gateway.deconz_ids
assert "sensor.clip_light_level_sensor" not in gateway.deconz_ids
assert len(hass.states.async_all()) == 6
gateway.api.sensors["1"].async_update({"state": {"on": False}})
light_level_sensor = hass.states.get("sensor.light_level_sensor")
assert light_level_sensor.state == "999.8"
presence_sensor = hass.states.get("sensor.presence_sensor")
assert presence_sensor is None
switch_1 = hass.states.get("sensor.switch_1")
assert switch_1 is None
switch_1_battery_level = hass.states.get("sensor.switch_1_battery_level")
assert switch_1_battery_level is None
switch_2 = hass.states.get("sensor.switch_2")
assert switch_2 is None
switch_2_battery_level = hass.states.get("sensor.switch_2_battery_level")
assert switch_2_battery_level.state == "100"
daylight_sensor = hass.states.get("sensor.daylight_sensor")
assert daylight_sensor.state == "dawn"
power_sensor = hass.states.get("sensor.power_sensor")
assert power_sensor.state == "6"
consumption_sensor = hass.states.get("sensor.consumption_sensor")
assert consumption_sensor.state == "0.002"
gateway.api.sensors["1"].async_update({"state": {"lightlevel": 2000}})
gateway.api.sensors["4"].async_update({"config": {"battery": 75}})
await hass.async_block_till_done()
light_level_sensor = hass.states.get("sensor.light_level_sensor")
assert light_level_sensor.state == "1.6"
switch_2_battery_level = hass.states.get("sensor.switch_2_battery_level")
assert switch_2_battery_level.state == "75"
async def test_allow_clip_sensors(hass):
"""Test that CLIP sensors can be allowed."""
data = deepcopy(DECONZ_WEB_REQUEST)
data["sensors"] = deepcopy(SENSORS)
gateway = await setup_deconz_integration(
hass,
ENTRY_CONFIG,
options={deconz.gateway.CONF_ALLOW_CLIP_SENSOR: True},
get_state_response=data,
)
assert "sensor.light_level_sensor" in gateway.deconz_ids
assert "sensor.presence_sensor" not in gateway.deconz_ids
assert "sensor.switch_1" not in gateway.deconz_ids
assert "sensor.switch_1_battery_level" not in gateway.deconz_ids
assert "sensor.switch_2" not in gateway.deconz_ids
assert "sensor.switch_2_battery_level" in gateway.deconz_ids
assert "sensor.daylight_sensor" in gateway.deconz_ids
assert "sensor.power_sensor" in gateway.deconz_ids
assert "sensor.consumption_sensor" in gateway.deconz_ids
assert "sensor.clip_light_level_sensor" in gateway.deconz_ids
assert len(hass.states.async_all()) == 7
light_level_sensor = hass.states.get("sensor.light_level_sensor")
assert light_level_sensor.state == "999.8"
presence_sensor = hass.states.get("sensor.presence_sensor")
assert presence_sensor is None
switch_1 = hass.states.get("sensor.switch_1")
assert switch_1 is None
switch_1_battery_level = hass.states.get("sensor.switch_1_battery_level")
assert switch_1_battery_level is None
switch_2 = hass.states.get("sensor.switch_2")
assert switch_2 is None
switch_2_battery_level = hass.states.get("sensor.switch_2_battery_level")
assert switch_2_battery_level.state == "100"
daylight_sensor = hass.states.get("sensor.daylight_sensor")
assert daylight_sensor.state == "dawn"
power_sensor = hass.states.get("sensor.power_sensor")
assert power_sensor.state == "6"
consumption_sensor = hass.states.get("sensor.consumption_sensor")
assert consumption_sensor.state == "0.002"
clip_light_level_sensor = hass.states.get("sensor.clip_light_level_sensor")
assert clip_light_level_sensor.state == "999.8"
async def test_add_new_sensor(hass):
"""Test successful creation of sensor entities."""
gateway = await setup_gateway(hass, {})
sensor = Mock()
sensor.name = "name"
sensor.type = "ZHATemperature"
sensor.uniqueid = "1"
sensor.BINARY = False
sensor.register_async_callback = Mock()
async_dispatcher_send(hass, gateway.async_signal_new_device("sensor"), [sensor])
await hass.async_block_till_done()
assert "sensor.name" in gateway.deconz_ids
async def test_do_not_allow_clipsensor(hass):
"""Test that clip sensors can be ignored."""
gateway = await setup_gateway(hass, {}, allow_clip_sensor=False)
sensor = Mock()
sensor.name = "name"
sensor.type = "CLIPTemperature"
sensor.register_async_callback = Mock()
async_dispatcher_send(hass, gateway.async_signal_new_device("sensor"), [sensor])
await hass.async_block_till_done()
"""Test that adding a new sensor works."""
data = deepcopy(DECONZ_WEB_REQUEST)
gateway = await setup_deconz_integration(
hass, ENTRY_CONFIG, options={}, get_state_response=data
)
assert len(gateway.deconz_ids) == 0
state_added = {
"t": "event",
"e": "added",
"r": "sensors",
"id": "1",
"sensor": deepcopy(SENSORS["1"]),
}
gateway.api.async_event_handler(state_added)
await hass.async_block_till_done()
async def test_unload_sensor(hass):
"""Test that it works to unload sensor entities."""
gateway = await setup_gateway(hass, {"sensors": SENSOR})
assert "sensor.light_level_sensor" in gateway.deconz_ids
await gateway.async_reset()
assert len(hass.states.async_all()) == 0
light_level_sensor = hass.states.get("sensor.light_level_sensor")
assert light_level_sensor.state == "999.8"