jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-04-24 09:17:53 +00:00
Code Issues Packages Projects Releases Wiki Activity

deCONZ improve light tests (#26697)

Browse Source 
* Improve light tests
* Small improvements on light and group classes
This commit is contained in:
Robert Svensson 2019-09-18 12:47:26 +02:00 committed by GitHub
parent 72baf563fa
commit 8a39924b37
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
6 changed files with 431 additions and 360 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
homeassistant/components/deconz/cover.py
View File

@ -17,7 +17,6 @@ from .gateway import get_gateway_from_config_entry
async def async_setup_platform(hass, config, async_add_entities, discovery_info=None):
"""Old way of setting up deCONZ platforms."""
pass
async def async_setup_entry(hass, config_entry, async_add_entities):

16
homeassistant/components/deconz/light.py
View File

@ -34,7 +34,6 @@ from .gateway import get_gateway_from_config_entry, DeconzEntityHandler
async def async_setup_platform(hass, config, async_add_entities, discovery_info=None):
"""Old way of setting up deCONZ platforms."""
pass
async def async_setup_entry(hass, config_entry, async_add_entities):
@ -194,9 +193,6 @@ class DeconzLight(DeconzDevice, Light):
attributes = {}
attributes["is_deconz_group"] = self._device.type == "LightGroup"
if self._device.type == "LightGroup":
attributes["all_on"] = self._device.all_on
return attributes
@ -207,9 +203,7 @@ class DeconzGroup(DeconzLight):
"""Set up group and create an unique id."""
super().__init__(device, gateway)
self._unique_id = "{}-{}".format(
self.gateway.api.config.bridgeid, self._device.deconz_id
)
self._unique_id = f"{self.gateway.api.config.bridgeid}-{self._device.deconz_id}"
@property
def unique_id(self):
@ -228,3 +222,11 @@ class DeconzGroup(DeconzLight):
"name": self._device.name,
"via_device": (DECONZ_DOMAIN, bridgeid),
}
@property
def device_state_attributes(self):
"""Return the device state attributes."""
attributes = dict(super().device_state_attributes)
attributes["all_on"] = self._device.all_on
return attributes

1
homeassistant/components/deconz/switch.py
View File

@ -10,7 +10,6 @@ from .gateway import get_gateway_from_config_entry
async def async_setup_platform(hass, config, async_add_entities, discovery_info=None):
"""Old way of setting up deCONZ platforms."""
pass
async def async_setup_entry(hass, config_entry, async_add_entities):

186
tests/components/deconz/test_cover.py
View File

@ -1,82 +1,83 @@
"""deCONZ cover platform tests."""
from unittest.mock import Mock, patch
from copy import deepcopy
from asynctest import patch
from homeassistant import config_entries
from homeassistant.components import deconz
from homeassistant.components.deconz.const import COVER_TYPES
from homeassistant.helpers.dispatcher import async_dispatcher_send
from homeassistant.setup import async_setup_component
import homeassistant.components.cover as cover
from tests.common import mock_coro
SUPPORTED_COVERS = {
COVERS = {
"1": {
"id": "Cover 1 id",
"name": "Cover 1 name",
"id": "Level controllable cover id",
"name": "Level controllable cover",
"type": "Level controllable output",
"state": {"bri": 255, "on": False, "reachable": True},
"modelid": "Not zigbee spec",
"uniqueid": "00:00:00:00:00:00:00:00-00",
},
"2": {
"id": "Cover 2 id",
"name": "Cover 2 name",
"id": "Window covering device id",
"name": "Window covering device",
"type": "Window covering device",
"state": {"bri": 255, "on": True, "reachable": True},
"modelid": "lumi.curtain",
"uniqueid": "00:00:00:00:00:00:00:01-00",
},
"3": {
"id": "Unsupported cover id",
"name": "Unsupported cover",
"type": "Not a cover",
"state": {"reachable": True},
"uniqueid": "00:00:00:00:00:00:00:02-00",
},
}
UNSUPPORTED_COVER = {
"1": {
"id": "Cover id",
"name": "Unsupported switch",
"type": "Not a cover",
"state": {},
}
}
BRIDGEID = "0123456789"
ENTRY_CONFIG = {
deconz.const.CONF_ALLOW_CLIP_SENSOR: True,
deconz.const.CONF_ALLOW_DECONZ_GROUPS: True,
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,
}
DECONZ_CONFIG = {
"bridgeid": BRIDGEID,
"mac": "00:11:22:33:44:55",
"name": "deCONZ mock gateway",
"sw_version": "2.05.69",
"websocketport": 1234,
}
async def setup_gateway(hass, data):
"""Load the deCONZ cover platform."""
from pydeconz import DeconzSession
DECONZ_WEB_REQUEST = {"config": DECONZ_CONFIG}
loop = Mock()
session = Mock()
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=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, "cover")
# 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):
@ -92,64 +93,69 @@ async def test_platform_manually_configured(hass):
async def test_no_covers(hass):
"""Test that no cover entities are created."""
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_cover(hass):
"""Test that all supported cover entities are created."""
with patch("pydeconz.DeconzSession.async_put_state", return_value=mock_coro(True)):
gateway = await setup_gateway(hass, {"lights": SUPPORTED_COVERS})
assert "cover.cover_1_name" in gateway.deconz_ids
assert len(SUPPORTED_COVERS) == len(COVER_TYPES)
assert len(hass.states.async_all()) == 3
cover_1 = hass.states.get("cover.cover_1_name")
assert cover_1 is not None
assert cover_1.state == "open"
gateway.api.lights["1"].async_update({})
await hass.services.async_call(
"cover", "open_cover", {"entity_id": "cover.cover_1_name"}, blocking=True
)
await hass.services.async_call(
"cover", "close_cover", {"entity_id": "cover.cover_1_name"}, blocking=True
)
await hass.services.async_call(
"cover", "stop_cover", {"entity_id": "cover.cover_1_name"}, blocking=True
data = deepcopy(DECONZ_WEB_REQUEST)
data["lights"] = deepcopy(COVERS)
gateway = await setup_deconz_integration(
hass, ENTRY_CONFIG, options={}, get_state_response=data
)
assert "cover.level_controllable_cover" in gateway.deconz_ids
assert "cover.window_covering_device" in gateway.deconz_ids
assert "cover.unsupported_cover" not in gateway.deconz_ids
assert len(hass.states.async_all()) == 5
await hass.services.async_call(
"cover", "close_cover", {"entity_id": "cover.cover_2_name"}, blocking=True
)
level_controllable_cover = hass.states.get("cover.level_controllable_cover")
assert level_controllable_cover.state == "open"
level_controllable_cover_device = gateway.api.lights["1"]
async def test_add_new_cover(hass):
"""Test successful creation of cover entity."""
data = {}
gateway = await setup_gateway(hass, data)
cover = Mock()
cover.name = "name"
cover.type = "Level controllable output"
cover.uniqueid = "1"
cover.register_async_callback = Mock()
async_dispatcher_send(hass, gateway.async_signal_new_device("light"), [cover])
level_controllable_cover_device.async_update({"state": {"on": True}})
await hass.async_block_till_done()
assert "cover.name" in gateway.deconz_ids
level_controllable_cover = hass.states.get("cover.level_controllable_cover")
assert level_controllable_cover.state == "closed"
async def test_unsupported_cover(hass):
"""Test that unsupported covers are not created."""
await setup_gateway(hass, {"lights": UNSUPPORTED_COVER})
assert len(hass.states.async_all()) == 0
with patch.object(
level_controllable_cover_device, "_async_set_callback", return_value=True
) as set_callback:
await hass.services.async_call(
cover.DOMAIN,
cover.SERVICE_OPEN_COVER,
{"entity_id": "cover.level_controllable_cover"},
blocking=True,
)
await hass.async_block_till_done()
set_callback.assert_called_with("/lights/1/state", {"on": False})
with patch.object(
level_controllable_cover_device, "_async_set_callback", return_value=True
) as set_callback:
await hass.services.async_call(
cover.DOMAIN,
cover.SERVICE_CLOSE_COVER,
{"entity_id": "cover.level_controllable_cover"},
blocking=True,
)
await hass.async_block_till_done()
set_callback.assert_called_with("/lights/1/state", {"on": True, "bri": 255})
async def test_unload_cover(hass):
"""Test that it works to unload switch entities."""
gateway = await setup_gateway(hass, {"lights": SUPPORTED_COVERS})
await gateway.async_reset()
assert len(hass.states.async_all()) == 1
with patch.object(
level_controllable_cover_device, "_async_set_callback", return_value=True
) as set_callback:
await hass.services.async_call(
cover.DOMAIN,
cover.SERVICE_STOP_COVER,
{"entity_id": "cover.level_controllable_cover"},
blocking=True,
)
await hass.async_block_till_done()
set_callback.assert_called_with("/lights/1/state", {"bri_inc": 0})

369
tests/components/deconz/test_light.py
View File

@ -1,49 +1,28 @@
"""deCONZ light platform tests."""
from unittest.mock import Mock, patch
from copy import deepcopy
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.light as light
from tests.common import mock_coro
LIGHT = {
GROUPS = {
"1": {
"id": "Light 1 id",
"name": "Light 1 name",
"state": {
"on": True,
"bri": 255,
"colormode": "xy",
"xy": (500, 500),
"reachable": True,
},
"uniqueid": "00:00:00:00:00:00:00:00-00",
},
"2": {
"id": "Light 2 id",
"name": "Light 2 name",
"state": {"on": True, "colormode": "ct", "ct": 2500, "reachable": True},
},
}
GROUP = {
"1": {
"id": "Group 1 id",
"name": "Group 1 name",
"id": "Light group id",
"name": "Light group",
"type": "LightGroup",
"state": {},
"state": {"all_on": False, "any_on": True},
"action": {},
"scenes": [],
"lights": ["1", "2"],
},
"2": {
"id": "Group 2 id",
"name": "Group 2 name",
"id": "Empty group id",
"name": "Empty group",
"type": "LightGroup",
"state": {},
"action": {},
"scenes": [],
@ -51,60 +30,80 @@ GROUP = {
},
}
SWITCH = {
LIGHTS = {
"1": {
"id": "Switch 1 id",
"name": "Switch 1 name",
"id": "RGB light id",
"name": "RGB light",
"state": {
"on": True,
"bri": 255,
"colormode": "xy",
"effect": "colorloop",
"xy": (500, 500),
"reachable": True,
},
"type": "Extended color light",
"uniqueid": "00:00:00:00:00:00:00:00-00",
},
"2": {
"id": "Tunable white light id",
"name": "Tunable white light",
"state": {"on": True, "colormode": "ct", "ct": 2500, "reachable": True},
"type": "Tunable white light",
"uniqueid": "00:00:00:00:00:00:00:01-00",
},
"3": {
"id": "On off switch id",
"name": "On off switch",
"type": "On/Off plug-in unit",
"state": {},
}
"state": {"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_deconz_groups=True):
"""Load the deCONZ light platform."""
from pydeconz import DeconzSession
loop = Mock()
session = Mock()
ENTRY_OPTIONS[deconz.const.CONF_ALLOW_DECONZ_GROUPS] = allow_deconz_groups
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, "light")
# 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):
@ -120,119 +119,157 @@ async def test_platform_manually_configured(hass):
async def test_no_lights_or_groups(hass):
"""Test that no lights or groups entities are created."""
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_lights_and_groups(hass):
"""Test that lights or groups entities are created."""
with patch("pydeconz.DeconzSession.async_put_state", return_value=mock_coro(True)):
gateway = await setup_gateway(hass, {"lights": LIGHT, "groups": GROUP})
assert "light.light_1_name" in gateway.deconz_ids
assert "light.light_2_name" in gateway.deconz_ids
assert "light.group_1_name" in gateway.deconz_ids
assert "light.group_2_name" not in gateway.deconz_ids
assert len(hass.states.async_all()) == 4
lamp_1 = hass.states.get("light.light_1_name")
assert lamp_1 is not None
assert lamp_1.state == "on"
assert lamp_1.attributes["brightness"] == 255
assert lamp_1.attributes["hs_color"] == (224.235, 100.0)
light_2 = hass.states.get("light.light_2_name")
assert light_2 is not None
assert light_2.state == "on"
assert light_2.attributes["color_temp"] == 2500
gateway.api.lights["1"].async_update({})
await hass.services.async_call(
"light",
"turn_on",
{
"entity_id": "light.light_1_name",
"color_temp": 2500,
"brightness": 200,
"transition": 5,
"flash": "short",
"effect": "colorloop",
},
blocking=True,
)
await hass.services.async_call(
"light",
"turn_on",
{
"entity_id": "light.light_1_name",
"hs_color": (20, 30),
"flash": "long",
"effect": "None",
},
blocking=True,
)
await hass.services.async_call(
"light",
"turn_off",
{"entity_id": "light.light_1_name", "transition": 5, "flash": "short"},
blocking=True,
)
await hass.services.async_call(
"light",
"turn_off",
{"entity_id": "light.light_1_name", "flash": "long"},
blocking=True,
data = deepcopy(DECONZ_WEB_REQUEST)
data["groups"] = deepcopy(GROUPS)
data["lights"] = deepcopy(LIGHTS)
gateway = await setup_deconz_integration(
hass, ENTRY_CONFIG, options={}, get_state_response=data
)
assert "light.rgb_light" in gateway.deconz_ids
assert "light.tunable_white_light" in gateway.deconz_ids
assert "light.light_group" in gateway.deconz_ids
assert "light.empty_group" not in gateway.deconz_ids
assert "light.on_off_switch" not in gateway.deconz_ids
# 4 entities + 2 groups (one for switches and one for lights)
assert len(hass.states.async_all()) == 6
rgb_light = hass.states.get("light.rgb_light")
assert rgb_light.state == "on"
assert rgb_light.attributes["brightness"] == 255
assert rgb_light.attributes["hs_color"] == (224.235, 100.0)
assert rgb_light.attributes["is_deconz_group"] is False
async def test_add_new_light(hass):
"""Test successful creation of light entities."""
gateway = await setup_gateway(hass, {})
light = Mock()
light.name = "name"
light.uniqueid = "1"
light.register_async_callback = Mock()
async_dispatcher_send(hass, gateway.async_signal_new_device("light"), [light])
tunable_white_light = hass.states.get("light.tunable_white_light")
assert tunable_white_light.state == "on"
assert tunable_white_light.attributes["color_temp"] == 2500
light_group = hass.states.get("light.light_group")
assert light_group.state == "on"
assert light_group.attributes["all_on"] is False
empty_group = hass.states.get("light.empty_group")
assert empty_group is None
rgb_light_device = gateway.api.lights["1"]
rgb_light_device.async_update({"state": {"on": False}})
await hass.async_block_till_done()
assert "light.name" in gateway.deconz_ids
rgb_light = hass.states.get("light.rgb_light")
assert rgb_light.state == "off"
with patch.object(
rgb_light_device, "_async_set_callback", return_value=True
) as set_callback:
await hass.services.async_call(
light.DOMAIN,
light.SERVICE_TURN_ON,
{
"entity_id": "light.rgb_light",
"color_temp": 2500,
"brightness": 200,
"transition": 5,
"flash": "short",
"effect": "colorloop",
},
blocking=True,
)
await hass.async_block_till_done()
set_callback.assert_called_with(
"/lights/1/state",
{
"ct": 2500,
"bri": 200,
"transitiontime": 50,
"alert": "select",
"effect": "colorloop",
},
)
with patch.object(
rgb_light_device, "_async_set_callback", return_value=True
) as set_callback:
await hass.services.async_call(
light.DOMAIN,
light.SERVICE_TURN_ON,
{
"entity_id": "light.rgb_light",
"hs_color": (20, 30),
"flash": "long",
"effect": "None",
},
blocking=True,
)
await hass.async_block_till_done()
set_callback.assert_called_with(
"/lights/1/state",
{"xy": (0.411, 0.351), "alert": "lselect", "effect": "none"},
)
with patch.object(
rgb_light_device, "_async_set_callback", return_value=True
) as set_callback:
await hass.services.async_call(
light.DOMAIN,
light.SERVICE_TURN_OFF,
{"entity_id": "light.rgb_light", "transition": 5, "flash": "short"},
blocking=True,
)
await hass.async_block_till_done()
set_callback.assert_called_with(
"/lights/1/state", {"bri": 0, "transitiontime": 50, "alert": "select"}
)
with patch.object(
rgb_light_device, "_async_set_callback", return_value=True
) as set_callback:
await hass.services.async_call(
light.DOMAIN,
light.SERVICE_TURN_OFF,
{"entity_id": "light.rgb_light", "flash": "long"},
blocking=True,
)
await hass.async_block_till_done()
set_callback.assert_called_with("/lights/1/state", {"alert": "lselect"})
async def test_add_new_group(hass):
"""Test successful creation of group entities."""
gateway = await setup_gateway(hass, {})
group = Mock()
group.name = "name"
group.register_async_callback = Mock()
async_dispatcher_send(hass, gateway.async_signal_new_device("group"), [group])
await hass.async_block_till_done()
assert "light.name" in gateway.deconz_ids
async def test_disable_light_groups(hass):
"""Test successful creation of sensor entities."""
data = deepcopy(DECONZ_WEB_REQUEST)
data["groups"] = deepcopy(GROUPS)
data["lights"] = deepcopy(LIGHTS)
gateway = await setup_deconz_integration(
hass,
ENTRY_CONFIG,
options={deconz.gateway.CONF_ALLOW_DECONZ_GROUPS: False},
get_state_response=data,
)
assert "light.rgb_light" in gateway.deconz_ids
assert "light.tunable_white_light" in gateway.deconz_ids
assert "light.light_group" not in gateway.deconz_ids
assert "light.empty_group" not in gateway.deconz_ids
assert "light.on_off_switch" not in gateway.deconz_ids
# 4 entities + 2 groups (one for switches and one for lights)
assert len(hass.states.async_all()) == 5
rgb_light = hass.states.get("light.rgb_light")
assert rgb_light is not None
async def test_do_not_add_deconz_groups(hass):
"""Test that clip sensors can be ignored."""
gateway = await setup_gateway(hass, {}, allow_deconz_groups=False)
group = Mock()
group.name = "name"
group.type = "LightGroup"
group.register_async_callback = Mock()
async_dispatcher_send(hass, gateway.async_signal_new_device("group"), [group])
await hass.async_block_till_done()
assert len(gateway.deconz_ids) == 0
tunable_white_light = hass.states.get("light.tunable_white_light")
assert tunable_white_light is not None
light_group = hass.states.get("light.light_group")
assert light_group is None
async def test_no_switch(hass):
"""Test that a switch doesn't get created as a light entity."""
gateway = await setup_gateway(hass, {"lights": SWITCH})
assert len(gateway.deconz_ids) == 0
assert len(hass.states.async_all()) == 0
async def test_unload_light(hass):
"""Test that it works to unload switch entities."""
gateway = await setup_gateway(hass, {"lights": LIGHT, "groups": GROUP})
await gateway.async_reset()
# Group.all_lights will not be removed
assert len(hass.states.async_all()) == 1
empty_group = hass.states.get("light.empty_group")
assert empty_group is None

218
tests/components/deconz/test_switch.py
View File

@ -1,86 +1,89 @@
"""deCONZ switch platform tests."""
from unittest.mock import Mock, patch
from copy import deepcopy
from asynctest import patch
from homeassistant import config_entries
from homeassistant.components import deconz
from homeassistant.components.deconz.const import SWITCH_TYPES
from homeassistant.helpers.dispatcher import async_dispatcher_send
from homeassistant.setup import async_setup_component
import homeassistant.components.switch as switch
from tests.common import mock_coro
SUPPORTED_SWITCHES = {
SWITCHES = {
"1": {
"id": "Switch 1 id",
"name": "Switch 1 name",
"id": "On off switch id",
"name": "On off switch",
"type": "On/Off plug-in unit",
"state": {"on": True, "reachable": True},
"uniqueid": "00:00:00:00:00:00:00:00-00",
},
"2": {
"id": "Switch 2 id",
"name": "Switch 2 name",
"id": "Smart plug id",
"name": "Smart plug",
"type": "Smart plug",
"state": {"on": True, "reachable": True},
"state": {"on": False, "reachable": True},
"uniqueid": "00:00:00:00:00:00:00:01-00",
},
"3": {
"id": "Switch 3 id",
"name": "Switch 3 name",
"id": "Warning device id",
"name": "Warning device",
"type": "Warning device",
"state": {"alert": "lselect", "reachable": True},
"uniqueid": "00:00:00:00:00:00:00:02-00",
},
"4": {
"id": "Unsupported switch id",
"name": "Unsupported switch",
"type": "Not a smart plug",
"state": {"reachable": True},
"uniqueid": "00:00:00:00:00:00:00:03-00",
},
}
UNSUPPORTED_SWITCH = {
"1": {
"id": "Switch id",
"name": "Unsupported switch",
"type": "Not a smart plug",
"state": {},
}
}
BRIDGEID = "0123456789"
ENTRY_CONFIG = {
deconz.const.CONF_ALLOW_CLIP_SENSOR: True,
deconz.const.CONF_ALLOW_DECONZ_GROUPS: True,
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,
}
DECONZ_CONFIG = {
"bridgeid": BRIDGEID,
"mac": "00:11:22:33:44:55",
"name": "deCONZ mock gateway",
"sw_version": "2.05.69",
"websocketport": 1234,
}
async def setup_gateway(hass, data):
"""Load the deCONZ switch platform."""
from pydeconz import DeconzSession
DECONZ_WEB_REQUEST = {"config": DECONZ_CONFIG}
loop = Mock()
session = Mock()
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=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, "switch")
# 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):
@ -96,68 +99,93 @@ async def test_platform_manually_configured(hass):
async def test_no_switches(hass):
"""Test that no switch entities are created."""
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_switches(hass):
"""Test that all supported switch entities are created."""
with patch("pydeconz.DeconzSession.async_put_state", return_value=mock_coro(True)):
gateway = await setup_gateway(hass, {"lights": SUPPORTED_SWITCHES})
assert "switch.switch_1_name" in gateway.deconz_ids
assert "switch.switch_2_name" in gateway.deconz_ids
assert "switch.switch_3_name" in gateway.deconz_ids
assert len(SUPPORTED_SWITCHES) == len(SWITCH_TYPES)
assert len(hass.states.async_all()) == 4
switch_1 = hass.states.get("switch.switch_1_name")
assert switch_1 is not None
assert switch_1.state == "on"
switch_3 = hass.states.get("switch.switch_3_name")
assert switch_3 is not None
assert switch_3.state == "on"
gateway.api.lights["1"].async_update({})
await hass.services.async_call(
"switch", "turn_on", {"entity_id": "switch.switch_1_name"}, blocking=True
)
await hass.services.async_call(
"switch", "turn_off", {"entity_id": "switch.switch_1_name"}, blocking=True
data = deepcopy(DECONZ_WEB_REQUEST)
data["lights"] = deepcopy(SWITCHES)
gateway = await setup_deconz_integration(
hass, ENTRY_CONFIG, options={}, get_state_response=data
)
assert "switch.on_off_switch" in gateway.deconz_ids
assert "switch.smart_plug" in gateway.deconz_ids
assert "switch.warning_device" in gateway.deconz_ids
assert "switch.unsupported_switch" not in gateway.deconz_ids
assert len(hass.states.async_all()) == 6
await hass.services.async_call(
"switch", "turn_on", {"entity_id": "switch.switch_3_name"}, blocking=True
)
await hass.services.async_call(
"switch", "turn_off", {"entity_id": "switch.switch_3_name"}, blocking=True
)
on_off_switch = hass.states.get("switch.on_off_switch")
assert on_off_switch.state == "on"
smart_plug = hass.states.get("switch.smart_plug")
assert smart_plug.state == "off"
async def test_add_new_switch(hass):
"""Test successful creation of switch entity."""
gateway = await setup_gateway(hass, {})
switch = Mock()
switch.name = "name"
switch.type = "Smart plug"
switch.uniqueid = "1"
switch.register_async_callback = Mock()
async_dispatcher_send(hass, gateway.async_signal_new_device("light"), [switch])
warning_device = hass.states.get("switch.warning_device")
assert warning_device.state == "on"
on_off_switch_device = gateway.api.lights["1"]
warning_device_device = gateway.api.lights["3"]
on_off_switch_device.async_update({"state": {"on": False}})
warning_device_device.async_update({"state": {"alert": None}})
await hass.async_block_till_done()
assert "switch.name" in gateway.deconz_ids
on_off_switch = hass.states.get("switch.on_off_switch")
assert on_off_switch.state == "off"
async def test_unsupported_switch(hass):
"""Test that unsupported switches are not created."""
await setup_gateway(hass, {"lights": UNSUPPORTED_SWITCH})
assert len(hass.states.async_all()) == 0
warning_device = hass.states.get("switch.warning_device")
assert warning_device.state == "off"
with patch.object(
on_off_switch_device, "_async_set_callback", return_value=True
) as set_callback:
await hass.services.async_call(
switch.DOMAIN,
switch.SERVICE_TURN_ON,
{"entity_id": "switch.on_off_switch"},
blocking=True,
)
await hass.async_block_till_done()
set_callback.assert_called_with("/lights/1/state", {"on": True})
async def test_unload_switch(hass):
"""Test that it works to unload switch entities."""
gateway = await setup_gateway(hass, {"lights": SUPPORTED_SWITCHES})
with patch.object(
on_off_switch_device, "_async_set_callback", return_value=True
) as set_callback:
await hass.services.async_call(
switch.DOMAIN,
switch.SERVICE_TURN_OFF,
{"entity_id": "switch.on_off_switch"},
blocking=True,
)
await hass.async_block_till_done()
set_callback.assert_called_with("/lights/1/state", {"on": False})
await gateway.async_reset()
with patch.object(
warning_device_device, "_async_set_callback", return_value=True
) as set_callback:
await hass.services.async_call(
switch.DOMAIN,
switch.SERVICE_TURN_ON,
{"entity_id": "switch.warning_device"},
blocking=True,
)
await hass.async_block_till_done()
set_callback.assert_called_with("/lights/3/state", {"alert": "lselect"})
assert len(hass.states.async_all()) == 1
with patch.object(
warning_device_device, "_async_set_callback", return_value=True
) as set_callback:
await hass.services.async_call(
switch.DOMAIN,
switch.SERVICE_TURN_OFF,
{"entity_id": "switch.warning_device"},
blocking=True,
)
await hass.async_block_till_done()
set_callback.assert_called_with("/lights/3/state", {"alert": "none"})