evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Refactor hue to split bridge support from light platform (#10691)

Browse source 
* Introduce a new Hue component that knows how to talk to a Hue bridge, but doesn't actually set up lights.

* Refactor the hue lights platform to use the HueBridge class from the hue component.

* Reimplement support for multiple bridges

* Auto discover bridges.

* Provide some migration support by showing a persistent notification.

* Address most feedback from code review.

* Call load_platform from inside HueBridge.setup passing the bridge id.

Not only this looks nicer, but it also nicely solves additional bridges being added after initial setup (e.g. pairing a second bridge should work now, I believe it required a restart before).

* Add a unit test for hue_activate_scene

* Address feedback from code review.

* After feedback from @andrey-git I was able to find a way to not import phue in tests, yay!

* Inject a mock phue in a couple of places
This commit is contained in:
Andrea Campi 2017-12-10 18:15:01 +00:00 • committed by Paulus Schoutsen
parent b2c5a9f5fe
commit 81974885ee
6 changed files with 1269 additions and 216 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
homeassistant/components/discovery.py
View file

@ -36,6 +36,7 @@ SERVICE_APPLE_TV = 'apple_tv'
SERVICE_WINK = 'wink'
SERVICE_XIAOMI_GW = 'xiaomi_gw'
SERVICE_TELLDUSLIVE = 'tellstick'
SERVICE_HUE = 'philips_hue'
SERVICE_HANDLERS = {
SERVICE_HASS_IOS_APP: ('ios', None),
@ -48,7 +49,7 @@ SERVICE_HANDLERS = {
SERVICE_WINK: ('wink', None),
SERVICE_XIAOMI_GW: ('xiaomi_aqara', None),
SERVICE_TELLDUSLIVE: ('tellduslive', None),
'philips_hue': ('light', 'hue'),
SERVICE_HUE: ('hue', None),
'google_cast': ('media_player', 'cast'),
'panasonic_viera': ('media_player', 'panasonic_viera'),
'plex_mediaserver': ('media_player', 'plex'),

241
homeassistant/components/hue.py Normal file
View file

@ -0,0 +1,241 @@
"""
This component provides basic support for the Philips Hue system.
For more details about this component, please refer to the documentation at
https://home-assistant.io/components/hue/
"""
import json
import logging
import os
import socket
import voluptuous as vol
from homeassistant.components.discovery import SERVICE_HUE
from homeassistant.config import load_yaml_config_file
from homeassistant.const import CONF_FILENAME, CONF_HOST
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers import discovery
REQUIREMENTS = ['phue==1.0']
_LOGGER = logging.getLogger(__name__)
DOMAIN = "hue"
SERVICE_HUE_SCENE = "hue_activate_scene"
CONF_BRIDGES = "bridges"
CONF_ALLOW_UNREACHABLE = 'allow_unreachable'
DEFAULT_ALLOW_UNREACHABLE = False
PHUE_CONFIG_FILE = 'phue.conf'
CONF_ALLOW_IN_EMULATED_HUE = "allow_in_emulated_hue"
DEFAULT_ALLOW_IN_EMULATED_HUE = True
CONF_ALLOW_HUE_GROUPS = "allow_hue_groups"
DEFAULT_ALLOW_HUE_GROUPS = True
BRIDGE_CONFIG_SCHEMA = vol.Schema([{
vol.Optional(CONF_HOST): cv.string,
vol.Optional(CONF_FILENAME, default=PHUE_CONFIG_FILE): cv.string,
vol.Optional(CONF_ALLOW_UNREACHABLE,
default=DEFAULT_ALLOW_UNREACHABLE): cv.boolean,
vol.Optional(CONF_ALLOW_IN_EMULATED_HUE,
default=DEFAULT_ALLOW_IN_EMULATED_HUE): cv.boolean,
vol.Optional(CONF_ALLOW_HUE_GROUPS,
default=DEFAULT_ALLOW_HUE_GROUPS): cv.boolean,
}])
CONFIG_SCHEMA = vol.Schema({
DOMAIN: vol.Schema({
vol.Optional(CONF_BRIDGES, default=[]): BRIDGE_CONFIG_SCHEMA,
}),
}, extra=vol.ALLOW_EXTRA)
ATTR_GROUP_NAME = "group_name"
ATTR_SCENE_NAME = "scene_name"
SCENE_SCHEMA = vol.Schema({
vol.Required(ATTR_GROUP_NAME): cv.string,
vol.Required(ATTR_SCENE_NAME): cv.string,
})
CONFIG_INSTRUCTIONS = """
Press the button on the bridge to register Philips Hue with Home Assistant.
![Location of button on bridge](/static/images/config_philips_hue.jpg)
"""
def setup(hass, config):
"""Set up the Hue platform."""
config = config.get(DOMAIN)
if config is None:
config = {}
if DOMAIN not in hass.data:
hass.data[DOMAIN] = {}
discovery.listen(
hass,
SERVICE_HUE,
lambda service, discovery_info:
bridge_discovered(hass, service, discovery_info))
bridges = config.get(CONF_BRIDGES, [])
for bridge in bridges:
filename = bridge.get(CONF_FILENAME)
allow_unreachable = bridge.get(CONF_ALLOW_UNREACHABLE)
allow_in_emulated_hue = bridge.get(CONF_ALLOW_IN_EMULATED_HUE)
allow_hue_groups = bridge.get(CONF_ALLOW_HUE_GROUPS)
host = bridge.get(CONF_HOST)
if host is None:
host = _find_host_from_config(hass, filename)
if host is None:
_LOGGER.error("No host found in configuration")
return False
setup_bridge(host, hass, filename, allow_unreachable,
allow_in_emulated_hue, allow_hue_groups)
return True
def bridge_discovered(hass, service, discovery_info):
"""Dispatcher for Hue discovery events."""
host = discovery_info.get('host')
serial = discovery_info.get('serial')
filename = 'phue-{}.conf'.format(serial)
setup_bridge(host, hass, filename)
def setup_bridge(host, hass, filename=None, allow_unreachable=False,
allow_in_emulated_hue=True, allow_hue_groups=True):
"""Set up a given Hue bridge."""
# Only register a device once
if socket.gethostbyname(host) in hass.data[DOMAIN]:
return
bridge = HueBridge(host, hass, filename, allow_unreachable,
allow_in_emulated_hue, allow_hue_groups)
bridge.setup()
def _find_host_from_config(hass, filename=PHUE_CONFIG_FILE):
"""Attempt to detect host based on existing configuration."""
path = hass.config.path(filename)
if not os.path.isfile(path):
return None
try:
with open(path) as inp:
return next(iter(json.load(inp).keys()))
except (ValueError, AttributeError, StopIteration):
# ValueError if can't parse as JSON
# AttributeError if JSON value is not a dict
# StopIteration if no keys
return None
class HueBridge(object):
"""Manages a single Hue bridge."""
def __init__(self, host, hass, filename, allow_unreachable=False,
allow_in_emulated_hue=True, allow_hue_groups=True):
"""Initialize the system."""
self.host = host
self.hass = hass
self.filename = filename
self.allow_unreachable = allow_unreachable
self.allow_in_emulated_hue = allow_in_emulated_hue
self.allow_hue_groups = allow_hue_groups
self.bridge = None
self.configured = False
self.config_request_id = None
hass.data[DOMAIN][socket.gethostbyname(host)] = self
def setup(self):
"""Set up a phue bridge based on host parameter."""
import phue
try:
self.bridge = phue.Bridge(
self.host,
config_file_path=self.hass.config.path(self.filename))
except ConnectionRefusedError: # Wrong host was given
_LOGGER.error("Error connecting to the Hue bridge at %s",
self.host)
return
except phue.PhueRegistrationException:
_LOGGER.warning("Connected to Hue at %s but not registered.",
self.host)
self.request_configuration()
return
# If we came here and configuring this host, mark as done
if self.config_request_id:
request_id = self.config_request_id
self.config_request_id = None
configurator = self.hass.components.configurator
configurator.request_done(request_id)
self.configured = True
discovery.load_platform(
self.hass, 'light', DOMAIN,
{'bridge_id': socket.gethostbyname(self.host)})
# create a service for calling run_scene directly on the bridge,
# used to simplify automation rules.
def hue_activate_scene(call):
"""Service to call directly into bridge to set scenes."""
group_name = call.data[ATTR_GROUP_NAME]
scene_name = call.data[ATTR_SCENE_NAME]
self.bridge.run_scene(group_name, scene_name)
descriptions = load_yaml_config_file(
os.path.join(os.path.dirname(__file__), 'services.yaml'))
self.hass.services.register(
DOMAIN, SERVICE_HUE_SCENE, hue_activate_scene,
descriptions.get(SERVICE_HUE_SCENE),
schema=SCENE_SCHEMA)
def request_configuration(self):
"""Request configuration steps from the user."""
configurator = self.hass.components.configurator
# We got an error if this method is called while we are configuring
if self.config_request_id:
configurator.notify_errors(
self.config_request_id,
"Failed to register, please try again.")
return
self.config_request_id = configurator.request_config(
"Philips Hue",
lambda data: self.setup(),
description=CONFIG_INSTRUCTIONS,
entity_picture="/static/images/logo_philips_hue.png",
submit_caption="I have pressed the button"
)
def get_api(self):
"""Return the full api dictionary from phue."""
return self.bridge.get_api()
def set_light(self, light_id, command):
"""Adjust properties of one or more lights. See phue for details."""
return self.bridge.set_light(light_id, command)
def set_group(self, light_id, command):
"""Change light settings for a group. See phue for detail."""
return self.bridge.set_group(light_id, command)

358
homeassistant/components/light/hue.py
View file

@ -1,19 +1,21 @@
"""
Support for Hue lights.
This component provides light support for the Philips Hue system.
For more details about this platform, please refer to the documentation at
https://home-assistant.io/components/light.hue/
"""
import json
import logging
import os
import random
import socket
from datetime import timedelta
import logging
import random
import re
import socket
import voluptuous as vol
import homeassistant.components.hue as hue
import homeassistant.util as util
from homeassistant.util import yaml
import homeassistant.util.color as color_util
from homeassistant.components.light import (
ATTR_BRIGHTNESS, ATTR_COLOR_TEMP, ATTR_EFFECT, ATTR_FLASH, ATTR_RGB_COLOR,
@ -21,30 +23,21 @@ from homeassistant.components.light import (
FLASH_LONG, FLASH_SHORT, SUPPORT_BRIGHTNESS, SUPPORT_COLOR_TEMP,
SUPPORT_EFFECT, SUPPORT_FLASH, SUPPORT_RGB_COLOR, SUPPORT_TRANSITION,
SUPPORT_XY_COLOR, Light, PLATFORM_SCHEMA)
from homeassistant.config import load_yaml_config_file
from homeassistant.const import (CONF_FILENAME, CONF_HOST, DEVICE_DEFAULT_NAME)
from homeassistant.const import CONF_FILENAME, CONF_HOST, DEVICE_DEFAULT_NAME
from homeassistant.components.emulated_hue import ATTR_EMULATED_HUE_HIDDEN
import homeassistant.helpers.config_validation as cv
REQUIREMENTS = ['phue==1.0']
DEPENDENCIES = ['hue']
# Track previously setup bridges
_CONFIGURED_BRIDGES = {}
# Map ip to request id for configuring
_CONFIGURING = {}
_LOGGER = logging.getLogger(__name__)
CONF_ALLOW_UNREACHABLE = 'allow_unreachable'
DEFAULT_ALLOW_UNREACHABLE = False
DOMAIN = "light"
SERVICE_HUE_SCENE = "hue_activate_scene"
DATA_KEY = 'hue_lights'
DATA_LIGHTS = 'lights'
DATA_LIGHTGROUPS = 'lightgroups'
MIN_TIME_BETWEEN_SCANS = timedelta(seconds=10)
MIN_TIME_BETWEEN_FORCED_SCANS = timedelta(milliseconds=100)
PHUE_CONFIG_FILE = 'phue.conf'
SUPPORT_HUE_ON_OFF = (SUPPORT_FLASH | SUPPORT_TRANSITION)
SUPPORT_HUE_DIMMABLE = (SUPPORT_HUE_ON_OFF | SUPPORT_BRIGHTNESS)
SUPPORT_HUE_COLOR_TEMP = (SUPPORT_HUE_DIMMABLE | SUPPORT_COLOR_TEMP)
@ -60,10 +53,14 @@ SUPPORT_HUE = {
'Color temperature light': SUPPORT_HUE_COLOR_TEMP
}
CONF_ALLOW_IN_EMULATED_HUE = "allow_in_emulated_hue"
DEFAULT_ALLOW_IN_EMULATED_HUE = True
ATTR_IS_HUE_GROUP = 'is_hue_group'
CONF_ALLOW_HUE_GROUPS = "allow_hue_groups"
# Legacy configuration, will be removed in 0.60
CONF_ALLOW_UNREACHABLE = 'allow_unreachable'
DEFAULT_ALLOW_UNREACHABLE = False
CONF_ALLOW_IN_EMULATED_HUE = 'allow_in_emulated_hue'
DEFAULT_ALLOW_IN_EMULATED_HUE = True
CONF_ALLOW_HUE_GROUPS = 'allow_hue_groups'
DEFAULT_ALLOW_HUE_GROUPS = True
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
@ -75,236 +72,168 @@ PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
default=DEFAULT_ALLOW_HUE_GROUPS): cv.boolean,
})
ATTR_GROUP_NAME = "group_name"
ATTR_SCENE_NAME = "scene_name"
SCENE_SCHEMA = vol.Schema({
vol.Required(ATTR_GROUP_NAME): cv.string,
vol.Required(ATTR_SCENE_NAME): cv.string,
})
MIGRATION_ID = 'light_hue_config_migration'
MIGRATION_TITLE = 'Philips Hue Configuration Migration'
MIGRATION_INSTRUCTIONS = """
Configuration for the Philips Hue component has changed; action required.
ATTR_IS_HUE_GROUP = "is_hue_group"
You have configured at least one bridge:
CONFIG_INSTRUCTIONS = """
Press the button on the bridge to register Philips Hue with Home Assistant.
hue:
{config}
![Location of button on bridge](/static/images/config_philips_hue.jpg)
This configuration is deprecated, please check the
[Hue component](https://home-assistant.io/components/hue/) page for more
information.
"""
def _find_host_from_config(hass, filename=PHUE_CONFIG_FILE):
"""Attempt to detect host based on existing configuration."""
path = hass.config.path(filename)
if not os.path.isfile(path):
return None
try:
with open(path) as inp:
return next(json.loads(''.join(inp)).keys().__iter__())
except (ValueError, AttributeError, StopIteration):
# ValueError if can't parse as JSON
# AttributeError if JSON value is not a dict
# StopIteration if no keys
return None
def setup_platform(hass, config, add_devices, discovery_info=None):
"""Set up the Hue lights."""
# Default needed in case of discovery
filename = config.get(CONF_FILENAME, PHUE_CONFIG_FILE)
allow_unreachable = config.get(CONF_ALLOW_UNREACHABLE,
DEFAULT_ALLOW_UNREACHABLE)
allow_in_emulated_hue = config.get(CONF_ALLOW_IN_EMULATED_HUE,
DEFAULT_ALLOW_IN_EMULATED_HUE)
allow_hue_groups = config.get(CONF_ALLOW_HUE_GROUPS)
if discovery_info is not None:
if "HASS Bridge" in discovery_info.get('name', ''):
_LOGGER.info("Emulated hue found, will not add")
return False
host = discovery_info.get('host')
else:
host = config.get(CONF_HOST, None)
if host is None:
host = _find_host_from_config(hass, filename)
if host is None:
_LOGGER.error("No host found in configuration")
return False
# Only act if we are not already configuring this host
if host in _CONFIGURING or \
socket.gethostbyname(host) in _CONFIGURED_BRIDGES:
if discovery_info is None or 'bridge_id' not in discovery_info:
return
setup_bridge(host, hass, add_devices, filename, allow_unreachable,
allow_in_emulated_hue, allow_hue_groups)
setup_data(hass)
if config is not None and len(config) > 0:
# Legacy configuration, will be removed in 0.60
config_str = yaml.dump([config])
# Indent so it renders in a fixed-width font
config_str = re.sub('(?m)^', ' ', config_str)
hass.components.persistent_notification.async_create(
MIGRATION_INSTRUCTIONS.format(config=config_str),
title=MIGRATION_TITLE,
notification_id=MIGRATION_ID)
bridge_id = discovery_info['bridge_id']
bridge = hass.data[hue.DOMAIN][bridge_id]
unthrottled_update_lights(hass, bridge, add_devices)
def setup_bridge(host, hass, add_devices, filename, allow_unreachable,
allow_in_emulated_hue, allow_hue_groups):
"""Set up a phue bridge based on host parameter."""
def setup_data(hass):
"""Initialize internal data. Useful from tests."""
if DATA_KEY not in hass.data:
hass.data[DATA_KEY] = {DATA_LIGHTS: {}, DATA_LIGHTGROUPS: {}}
@util.Throttle(MIN_TIME_BETWEEN_SCANS, MIN_TIME_BETWEEN_FORCED_SCANS)
def update_lights(hass, bridge, add_devices):
"""Update the Hue light objects with latest info from the bridge."""
return unthrottled_update_lights(hass, bridge, add_devices)
def unthrottled_update_lights(hass, bridge, add_devices):
"""Internal version of update_lights."""
import phue
if not bridge.configured:
return
try:
bridge = phue.Bridge(
host,
config_file_path=hass.config.path(filename))
except ConnectionRefusedError: # Wrong host was given
_LOGGER.error("Error connecting to the Hue bridge at %s", host)
api = bridge.get_api()
except phue.PhueRequestTimeout:
_LOGGER.warning('Timeout trying to reach the bridge')
return
except ConnectionRefusedError:
_LOGGER.error('The bridge refused the connection')
return
except socket.error:
# socket.error when we cannot reach Hue
_LOGGER.exception('Cannot reach the bridge')
return
except phue.PhueRegistrationException:
_LOGGER.warning("Connected to Hue at %s but not registered.", host)
bridge_type = get_bridge_type(api)
request_configuration(host, hass, add_devices, filename,
allow_unreachable, allow_in_emulated_hue,
allow_hue_groups)
new_lights = process_lights(
hass, api, bridge, bridge_type,
lambda **kw: update_lights(hass, bridge, add_devices, **kw))
if bridge.allow_hue_groups:
new_lightgroups = process_groups(
hass, api, bridge, bridge_type,
lambda **kw: update_lights(hass, bridge, add_devices, **kw))
new_lights.extend(new_lightgroups)
return
if new_lights:
add_devices(new_lights)
# If we came here and configuring this host, mark as done
if host in _CONFIGURING:
request_id = _CONFIGURING.pop(host)
configurator = hass.components.configurator
configurator.request_done(request_id)
lights = {}
lightgroups = {}
skip_groups = not allow_hue_groups
def get_bridge_type(api):
"""Return the bridge type."""
api_name = api.get('config').get('name')
if api_name in ('RaspBee-GW', 'deCONZ-GW'):
return 'deconz'
else:
return 'hue'
@util.Throttle(MIN_TIME_BETWEEN_SCANS, MIN_TIME_BETWEEN_FORCED_SCANS)
def update_lights():
"""Update the Hue light objects with latest info from the bridge."""
nonlocal skip_groups
try:
api = bridge.get_api()
except phue.PhueRequestTimeout:
_LOGGER.warning("Timeout trying to reach the bridge")
return
except ConnectionRefusedError:
_LOGGER.error("The bridge refused the connection")
return
except socket.error:
# socket.error when we cannot reach Hue
_LOGGER.exception("Cannot reach the bridge")
return
def process_lights(hass, api, bridge, bridge_type, update_lights_cb):
"""Set up HueLight objects for all lights."""
api_lights = api.get('lights')
api_lights = api.get('lights')
if not isinstance(api_lights, dict):
_LOGGER.error('Got unexpected result from Hue API')
return []
if not isinstance(api_lights, dict):
_LOGGER.error("Got unexpected result from Hue API")
return
new_lights = []
if skip_groups:
api_groups = {}
lights = hass.data[DATA_KEY][DATA_LIGHTS]
for light_id, info in api_lights.items():
if light_id not in lights:
lights[light_id] = HueLight(
int(light_id), info, bridge,
update_lights_cb,
bridge_type, bridge.allow_unreachable,
bridge.allow_in_emulated_hue)
new_lights.append(lights[light_id])
else:
api_groups = api.get('groups')
lights[light_id].info = info
lights[light_id].schedule_update_ha_state()
if not isinstance(api_groups, dict):
_LOGGER.error("Got unexpected result from Hue API")
return
return new_lights
new_lights = []
api_name = api.get('config').get('name')
if api_name in ('RaspBee-GW', 'deCONZ-GW'):
bridge_type = 'deconz'
def process_groups(hass, api, bridge, bridge_type, update_lights_cb):
"""Set up HueLight objects for all groups."""
api_groups = api.get('groups')
if not isinstance(api_groups, dict):
_LOGGER.error('Got unexpected result from Hue API')
return []
new_lights = []
groups = hass.data[DATA_KEY][DATA_LIGHTGROUPS]
for lightgroup_id, info in api_groups.items():
if 'state' not in info:
_LOGGER.warning('Group info does not contain state. '
'Please update your hub.')
return []
if lightgroup_id not in groups:
groups[lightgroup_id] = HueLight(
int(lightgroup_id), info, bridge,
update_lights_cb,
bridge_type, bridge.allow_unreachable,
bridge.allow_in_emulated_hue, True)
new_lights.append(groups[lightgroup_id])
else:
bridge_type = 'hue'
groups[lightgroup_id].info = info
groups[lightgroup_id].schedule_update_ha_state()
for light_id, info in api_lights.items():
if light_id not in lights:
lights[light_id] = HueLight(int(light_id), info,
bridge, update_lights,
bridge_type, allow_unreachable,
allow_in_emulated_hue)
new_lights.append(lights[light_id])
else:
lights[light_id].info = info
lights[light_id].schedule_update_ha_state()
for lightgroup_id, info in api_groups.items():
if 'state' not in info:
_LOGGER.warning("Group info does not contain state. "
"Please update your hub.")
skip_groups = True
break
if lightgroup_id not in lightgroups:
lightgroups[lightgroup_id] = HueLight(
int(lightgroup_id), info, bridge, update_lights,
bridge_type, allow_unreachable, allow_in_emulated_hue,
True)
new_lights.append(lightgroups[lightgroup_id])
else:
lightgroups[lightgroup_id].info = info
lightgroups[lightgroup_id].schedule_update_ha_state()
if new_lights:
add_devices(new_lights)
_CONFIGURED_BRIDGES[socket.gethostbyname(host)] = True
# create a service for calling run_scene directly on the bridge,
# used to simplify automation rules.
def hue_activate_scene(call):
"""Service to call directly into bridge to set scenes."""
group_name = call.data[ATTR_GROUP_NAME]
scene_name = call.data[ATTR_SCENE_NAME]
bridge.run_scene(group_name, scene_name)
descriptions = load_yaml_config_file(
os.path.join(os.path.dirname(__file__), 'services.yaml'))
hass.services.register(DOMAIN, SERVICE_HUE_SCENE, hue_activate_scene,
descriptions.get(SERVICE_HUE_SCENE),
schema=SCENE_SCHEMA)
update_lights()
def request_configuration(host, hass, add_devices, filename,
allow_unreachable, allow_in_emulated_hue,
allow_hue_groups):
"""Request configuration steps from the user."""
configurator = hass.components.configurator
# We got an error if this method is called while we are configuring
if host in _CONFIGURING:
configurator.notify_errors(
_CONFIGURING[host], "Failed to register, please try again.")
return
# pylint: disable=unused-argument
def hue_configuration_callback(data):
"""Set up actions to do when our configuration callback is called."""
setup_bridge(host, hass, add_devices, filename, allow_unreachable,
allow_in_emulated_hue, allow_hue_groups)
_CONFIGURING[host] = configurator.request_config(
"Philips Hue", hue_configuration_callback,
description=CONFIG_INSTRUCTIONS,
entity_picture="/static/images/logo_philips_hue.png",
submit_caption="I have pressed the button"
)
return new_lights
class HueLight(Light):
"""Representation of a Hue light."""
def __init__(self, light_id, info, bridge, update_lights,
def __init__(self, light_id, info, bridge, update_lights_cb,
bridge_type, allow_unreachable, allow_in_emulated_hue,
is_group=False):
"""Initialize the light."""
self.light_id = light_id
self.info = info
self.bridge = bridge
self.update_lights = update_lights
self.update_lights = update_lights_cb
self.bridge_type = bridge_type
self.allow_unreachable = allow_unreachable
self.is_group = is_group
@ -381,14 +310,15 @@ class HueLight(Light):
command['transitiontime'] = int(kwargs[ATTR_TRANSITION] * 10)
if ATTR_XY_COLOR in kwargs:
if self.info.get('manufacturername') == "OSRAM":
hue, sat = color_util.color_xy_to_hs(*kwargs[ATTR_XY_COLOR])
command['hue'] = hue
if self.info.get('manufacturername') == 'OSRAM':
color_hue, sat = color_util.color_xy_to_hs(
*kwargs[ATTR_XY_COLOR])
command['hue'] = color_hue
command['sat'] = sat
else:
command['xy'] = kwargs[ATTR_XY_COLOR]
elif ATTR_RGB_COLOR in kwargs:
if self.info.get('manufacturername') == "OSRAM":
if self.info.get('manufacturername') == 'OSRAM':
hsv = color_util.color_RGB_to_hsv(
*(int(val) for val in kwargs[ATTR_RGB_COLOR]))
command['hue'] = hsv[0]

2
requirements_all.txt
View file

@ -533,7 +533,7 @@ pdunehd==1.3
# homeassistant.components.media_player.pandora
pexpect==4.0.1
# homeassistant.components.light.hue
# homeassistant.components.hue
phue==1.0
# homeassistant.components.rpi_pfio

479
tests/components/light/test_hue.py Normal file
View file

@ -0,0 +1,479 @@
"""Philips Hue lights platform tests."""
import logging
import unittest
import unittest.mock as mock
from unittest.mock import call, MagicMock, patch
from homeassistant.components import hue
import homeassistant.components.light.hue as hue_light
from tests.common import get_test_home_assistant, MockDependency
_LOGGER = logging.getLogger(__name__)
class TestSetup(unittest.TestCase):
"""Test the Hue light platform."""
def setUp(self): # pylint: disable=invalid-name
"""Setup things to be run when tests are started."""
self.hass = get_test_home_assistant()
self.skip_teardown_stop = False
def tearDown(self):
"""Stop everything that was started."""
if not self.skip_teardown_stop:
self.hass.stop()
def setup_mocks_for_update_lights(self):
"""Set up all mocks for update_lights tests."""
self.mock_bridge = MagicMock()
self.mock_bridge.allow_hue_groups = False
self.mock_api = MagicMock()
self.mock_bridge.get_api.return_value = self.mock_api
self.mock_bridge_type = MagicMock()
self.mock_lights = []
self.mock_groups = []
self.mock_add_devices = MagicMock()
hue_light.setup_data(self.hass)
def setup_mocks_for_process_lights(self):
"""Set up all mocks for process_lights tests."""
self.mock_bridge = MagicMock()
self.mock_api = MagicMock()
self.mock_api.get.return_value = {}
self.mock_bridge.get_api.return_value = self.mock_api
self.mock_bridge_type = MagicMock()
hue_light.setup_data(self.hass)
def setup_mocks_for_process_groups(self):
"""Set up all mocks for process_groups tests."""
self.mock_bridge = MagicMock()
self.mock_bridge.get_group.return_value = {
'name': 'Group 0', 'state': {'any_on': True}}
self.mock_api = MagicMock()
self.mock_api.get.return_value = {}
self.mock_bridge.get_api.return_value = self.mock_api
self.mock_bridge_type = MagicMock()
hue_light.setup_data(self.hass)
def test_setup_platform_no_discovery_info(self):
"""Test setup_platform without discovery info."""
self.hass.data[hue.DOMAIN] = {}
mock_add_devices = MagicMock()
hue_light.setup_platform(self.hass, {}, mock_add_devices)
mock_add_devices.assert_not_called()
def test_setup_platform_no_bridge_id(self):
"""Test setup_platform without a bridge."""
self.hass.data[hue.DOMAIN] = {}
mock_add_devices = MagicMock()
hue_light.setup_platform(self.hass, {}, mock_add_devices, {})
mock_add_devices.assert_not_called()
def test_setup_platform_one_bridge(self):
"""Test setup_platform with one bridge."""
mock_bridge = MagicMock()
self.hass.data[hue.DOMAIN] = {'10.0.0.1': mock_bridge}
mock_add_devices = MagicMock()
with patch('homeassistant.components.light.hue.' +
'unthrottled_update_lights') as mock_update_lights:
hue_light.setup_platform(
self.hass, {}, mock_add_devices,
{'bridge_id': '10.0.0.1'})
mock_update_lights.assert_called_once_with(
self.hass, mock_bridge, mock_add_devices)
def test_setup_platform_multiple_bridges(self):
"""Test setup_platform wuth multiple bridges."""
mock_bridge = MagicMock()
mock_bridge2 = MagicMock()
self.hass.data[hue.DOMAIN] = {
'10.0.0.1': mock_bridge,
'192.168.0.10': mock_bridge2,
}
mock_add_devices = MagicMock()
with patch('homeassistant.components.light.hue.' +
'unthrottled_update_lights') as mock_update_lights:
hue_light.setup_platform(
self.hass, {}, mock_add_devices,
{'bridge_id': '10.0.0.1'})
hue_light.setup_platform(
self.hass, {}, mock_add_devices,
{'bridge_id': '192.168.0.10'})
mock_update_lights.assert_has_calls([
call(self.hass, mock_bridge, mock_add_devices),
call(self.hass, mock_bridge2, mock_add_devices),
])
@MockDependency('phue')
def test_update_lights_with_no_lights(self, mock_phue):
"""Test the update_lights function when no lights are found."""
self.setup_mocks_for_update_lights()
with patch('homeassistant.components.light.hue.get_bridge_type',
return_value=self.mock_bridge_type):
with patch('homeassistant.components.light.hue.process_lights',
return_value=[]) as mock_process_lights:
with patch('homeassistant.components.light.hue.process_groups',
return_value=self.mock_groups) \
as mock_process_groups:
hue_light.unthrottled_update_lights(
self.hass, self.mock_bridge, self.mock_add_devices)
mock_process_lights.assert_called_once_with(
self.hass, self.mock_api, self.mock_bridge,
self.mock_bridge_type, mock.ANY)
mock_process_groups.assert_not_called()
self.mock_add_devices.assert_not_called()
@MockDependency('phue')
def test_update_lights_with_some_lights(self, mock_phue):
"""Test the update_lights function with some lights."""
self.setup_mocks_for_update_lights()
self.mock_lights = ['some', 'light']
with patch('homeassistant.components.light.hue.get_bridge_type',
return_value=self.mock_bridge_type):
with patch('homeassistant.components.light.hue.process_lights',
return_value=self.mock_lights) as mock_process_lights:
with patch('homeassistant.components.light.hue.process_groups',
return_value=self.mock_groups) \
as mock_process_groups:
hue_light.unthrottled_update_lights(
self.hass, self.mock_bridge, self.mock_add_devices)
mock_process_lights.assert_called_once_with(
self.hass, self.mock_api, self.mock_bridge,
self.mock_bridge_type, mock.ANY)
mock_process_groups.assert_not_called()
self.mock_add_devices.assert_called_once_with(
self.mock_lights)
@MockDependency('phue')
def test_update_lights_no_groups(self, mock_phue):
"""Test the update_lights function when no groups are found."""
self.setup_mocks_for_update_lights()
self.mock_bridge.allow_hue_groups = True
self.mock_lights = ['some', 'light']
with patch('homeassistant.components.light.hue.get_bridge_type',
return_value=self.mock_bridge_type):
with patch('homeassistant.components.light.hue.process_lights',
return_value=self.mock_lights) as mock_process_lights:
with patch('homeassistant.components.light.hue.process_groups',
return_value=self.mock_groups) \
as mock_process_groups:
hue_light.unthrottled_update_lights(
self.hass, self.mock_bridge, self.mock_add_devices)
mock_process_lights.assert_called_once_with(
self.hass, self.mock_api, self.mock_bridge,
self.mock_bridge_type, mock.ANY)
mock_process_groups.assert_called_once_with(
self.hass, self.mock_api, self.mock_bridge,
self.mock_bridge_type, mock.ANY)
self.mock_add_devices.assert_called_once_with(
self.mock_lights)
@MockDependency('phue')
def test_update_lights_with_lights_and_groups(self, mock_phue):
"""Test the update_lights function with both lights and groups."""
self.setup_mocks_for_update_lights()
self.mock_bridge.allow_hue_groups = True
self.mock_lights = ['some', 'light']
self.mock_groups = ['and', 'groups']
with patch('homeassistant.components.light.hue.get_bridge_type',
return_value=self.mock_bridge_type):
with patch('homeassistant.components.light.hue.process_lights',
return_value=self.mock_lights) as mock_process_lights:
with patch('homeassistant.components.light.hue.process_groups',
return_value=self.mock_groups) \
as mock_process_groups:
hue_light.unthrottled_update_lights(
self.hass, self.mock_bridge, self.mock_add_devices)
mock_process_lights.assert_called_once_with(
self.hass, self.mock_api, self.mock_bridge,
self.mock_bridge_type, mock.ANY)
mock_process_groups.assert_called_once_with(
self.hass, self.mock_api, self.mock_bridge,
self.mock_bridge_type, mock.ANY)
self.mock_add_devices.assert_called_once_with(
self.mock_lights)
def test_process_lights_api_error(self):
"""Test the process_lights function when the bridge errors out."""
self.setup_mocks_for_process_lights()
self.mock_api.get.return_value = None
ret = hue_light.process_lights(
self.hass, self.mock_api, self.mock_bridge, self.mock_bridge_type,
None)
self.assertEquals([], ret)
self.assertEquals(
{},
self.hass.data[hue_light.DATA_KEY][hue_light.DATA_LIGHTS])
def test_process_lights_no_lights(self):
"""Test the process_lights function when bridge returns no lights."""
self.setup_mocks_for_process_lights()
ret = hue_light.process_lights(
self.hass, self.mock_api, self.mock_bridge, self.mock_bridge_type,
None)
self.assertEquals([], ret)
self.assertEquals(
{},
self.hass.data[hue_light.DATA_KEY][hue_light.DATA_LIGHTS])
@patch('homeassistant.components.light.hue.HueLight')
def test_process_lights_some_lights(self, mock_hue_light):
"""Test the process_lights function with multiple groups."""
self.setup_mocks_for_process_lights()
self.mock_api.get.return_value = {
1: {'state': 'on'}, 2: {'state': 'off'}}
ret = hue_light.process_lights(
self.hass, self.mock_api, self.mock_bridge, self.mock_bridge_type,
None)
self.assertEquals(len(ret), 2)
mock_hue_light.assert_has_calls([
call(
1, {'state': 'on'}, self.mock_bridge, mock.ANY,
self.mock_bridge_type, self.mock_bridge.allow_unreachable,
self.mock_bridge.allow_in_emulated_hue),
call(
2, {'state': 'off'}, self.mock_bridge, mock.ANY,
self.mock_bridge_type, self.mock_bridge.allow_unreachable,
self.mock_bridge.allow_in_emulated_hue),
])
self.assertEquals(
len(self.hass.data[hue_light.DATA_KEY][hue_light.DATA_LIGHTS]),
2)
@patch('homeassistant.components.light.hue.HueLight')
def test_process_lights_new_light(self, mock_hue_light):
"""
Test the process_lights function with new groups.
Test what happens when we already have a light and a new one shows up.
"""
self.setup_mocks_for_process_lights()
self.mock_api.get.return_value = {
1: {'state': 'on'}, 2: {'state': 'off'}}
self.hass.data[
hue_light.DATA_KEY][hue_light.DATA_LIGHTS][1] = MagicMock()
ret = hue_light.process_lights(
self.hass, self.mock_api, self.mock_bridge, self.mock_bridge_type,
None)
self.assertEquals(len(ret), 1)
mock_hue_light.assert_has_calls([
call(
2, {'state': 'off'}, self.mock_bridge, mock.ANY,
self.mock_bridge_type, self.mock_bridge.allow_unreachable,
self.mock_bridge.allow_in_emulated_hue),
])
self.hass.data[hue_light.DATA_KEY][hue_light.DATA_LIGHTS][
1].schedule_update_ha_state.assert_called_once_with()
self.assertEquals(
len(self.hass.data[hue_light.DATA_KEY][hue_light.DATA_LIGHTS]),
2)
def test_process_groups_api_error(self):
"""Test the process_groups function when the bridge errors out."""
self.setup_mocks_for_process_groups()
self.mock_api.get.return_value = None
ret = hue_light.process_groups(
self.hass, self.mock_api, self.mock_bridge, self.mock_bridge_type,
None)
self.assertEquals([], ret)
self.assertEquals(
{},
self.hass.data[hue_light.DATA_KEY][hue_light.DATA_LIGHTGROUPS])
def test_process_groups_no_state(self):
"""Test the process_groups function when bridge returns no status."""
self.setup_mocks_for_process_groups()
self.mock_bridge.get_group.return_value = {'name': 'Group 0'}
ret = hue_light.process_groups(
self.hass, self.mock_api, self.mock_bridge, self.mock_bridge_type,
None)
self.assertEquals([], ret)
self.assertEquals(
{},
self.hass.data[hue_light.DATA_KEY][hue_light.DATA_LIGHTGROUPS])
@patch('homeassistant.components.light.hue.HueLight')
def test_process_groups_some_groups(self, mock_hue_light):
"""Test the process_groups function with multiple groups."""
self.setup_mocks_for_process_groups()
self.mock_api.get.return_value = {
1: {'state': 'on'}, 2: {'state': 'off'}}
ret = hue_light.process_groups(
self.hass, self.mock_api, self.mock_bridge, self.mock_bridge_type,
None)
self.assertEquals(len(ret), 2)
mock_hue_light.assert_has_calls([
call(
1, {'state': 'on'}, self.mock_bridge, mock.ANY,
self.mock_bridge_type, self.mock_bridge.allow_unreachable,
self.mock_bridge.allow_in_emulated_hue, True),
call(
2, {'state': 'off'}, self.mock_bridge, mock.ANY,
self.mock_bridge_type, self.mock_bridge.allow_unreachable,
self.mock_bridge.allow_in_emulated_hue, True),
])
self.assertEquals(
len(self.hass.data[
hue_light.DATA_KEY][hue_light.DATA_LIGHTGROUPS]),
2)
@patch('homeassistant.components.light.hue.HueLight')
def test_process_groups_new_group(self, mock_hue_light):
"""
Test the process_groups function with new groups.
Test what happens when we already have a light and a new one shows up.
"""
self.setup_mocks_for_process_groups()
self.mock_api.get.return_value = {
1: {'state': 'on'}, 2: {'state': 'off'}}
self.hass.data[
hue_light.DATA_KEY][hue_light.DATA_LIGHTGROUPS][1] = MagicMock()
ret = hue_light.process_groups(
self.hass, self.mock_api, self.mock_bridge, self.mock_bridge_type,
None)
self.assertEquals(len(ret), 1)
mock_hue_light.assert_has_calls([
call(
2, {'state': 'off'}, self.mock_bridge, mock.ANY,
self.mock_bridge_type, self.mock_bridge.allow_unreachable,
self.mock_bridge.allow_in_emulated_hue, True),
])
self.hass.data[hue_light.DATA_KEY][hue_light.DATA_LIGHTGROUPS][
1].schedule_update_ha_state.assert_called_once_with()
self.assertEquals(
len(self.hass.data[
hue_light.DATA_KEY][hue_light.DATA_LIGHTGROUPS]),
2)
class TestHueLight(unittest.TestCase):
"""Test the HueLight class."""
def setUp(self): # pylint: disable=invalid-name
"""Setup things to be run when tests are started."""
self.hass = get_test_home_assistant()
self.skip_teardown_stop = False
self.light_id = 42
self.mock_info = MagicMock()
self.mock_bridge = MagicMock()
self.mock_update_lights = MagicMock()
self.mock_bridge_type = MagicMock()
self.mock_allow_unreachable = MagicMock()
self.mock_is_group = MagicMock()
self.mock_allow_in_emulated_hue = MagicMock()
self.mock_is_group = False
def tearDown(self):
"""Stop everything that was started."""
if not self.skip_teardown_stop:
self.hass.stop()
def buildLight(
self, light_id=None, info=None, update_lights=None, is_group=None):
"""Helper to build a HueLight object with minimal fuss."""
return hue_light.HueLight(
light_id if light_id is not None else self.light_id,
info if info is not None else self.mock_info,
self.mock_bridge,
(update_lights
if update_lights is not None
else self.mock_update_lights),
self.mock_bridge_type,
self.mock_allow_unreachable, self.mock_allow_in_emulated_hue,
is_group if is_group is not None else self.mock_is_group)
def test_unique_id_for_light(self):
"""Test the unique_id method with lights."""
class_name = "<class 'homeassistant.components.light.hue.HueLight'>"
light = self.buildLight(info={'uniqueid': 'foobar'})
self.assertEquals(
class_name+'.foobar',
light.unique_id)
light = self.buildLight(info={})
self.assertEquals(
class_name+'.Unnamed Device.Light.42',
light.unique_id)
light = self.buildLight(info={'name': 'my-name'})
self.assertEquals(
class_name+'.my-name.Light.42',
light.unique_id)
light = self.buildLight(info={'type': 'my-type'})
self.assertEquals(
class_name+'.Unnamed Device.my-type.42',
light.unique_id)
light = self.buildLight(info={'name': 'a name', 'type': 'my-type'})
self.assertEquals(
class_name+'.a name.my-type.42',
light.unique_id)
def test_unique_id_for_group(self):
"""Test the unique_id method with groups."""
class_name = "<class 'homeassistant.components.light.hue.HueLight'>"
light = self.buildLight(info={'uniqueid': 'foobar'}, is_group=True)
self.assertEquals(
class_name+'.foobar',
light.unique_id)
light = self.buildLight(info={}, is_group=True)
self.assertEquals(
class_name+'.Unnamed Device.Group.42',
light.unique_id)
light = self.buildLight(info={'name': 'my-name'}, is_group=True)
self.assertEquals(
class_name+'.my-name.Group.42',
light.unique_id)
light = self.buildLight(info={'type': 'my-type'}, is_group=True)
self.assertEquals(
class_name+'.Unnamed Device.my-type.42',
light.unique_id)
light = self.buildLight(
info={'name': 'a name', 'type': 'my-type'},
is_group=True)
self.assertEquals(
class_name+'.a name.my-type.42',
light.unique_id)

402
tests/components/test_hue.py Normal file
View file

@ -0,0 +1,402 @@
"""Generic Philips Hue component tests."""
import logging
import unittest
from unittest.mock import call, MagicMock, patch
from homeassistant.components import configurator, hue
from homeassistant.const import CONF_FILENAME, CONF_HOST
from homeassistant.setup import setup_component
from tests.common import (
assert_setup_component, get_test_home_assistant, get_test_config_dir,
MockDependency
)
_LOGGER = logging.getLogger(__name__)
class TestSetup(unittest.TestCase):
"""Test the Hue component."""
def setUp(self): # pylint: disable=invalid-name
"""Setup things to be run when tests are started."""
self.hass = get_test_home_assistant()
self.skip_teardown_stop = False
def tearDown(self):
"""Stop everything that was started."""
if not self.skip_teardown_stop:
self.hass.stop()
@MockDependency('phue')
def test_setup_no_domain(self, mock_phue):
"""If it's not in the config we won't even try."""
with assert_setup_component(0):
self.assertTrue(setup_component(
self.hass, hue.DOMAIN, {}))
mock_phue.Bridge.assert_not_called()
self.assertEquals({}, self.hass.data[hue.DOMAIN])
@MockDependency('phue')
def test_setup_no_host(self, mock_phue):
"""No host specified in any way."""
with assert_setup_component(1):
self.assertTrue(setup_component(
self.hass, hue.DOMAIN, {hue.DOMAIN: {}}))
mock_phue.Bridge.assert_not_called()
self.assertEquals({}, self.hass.data[hue.DOMAIN])
@MockDependency('phue')
def test_setup_with_host(self, mock_phue):
"""Host specified in the config file."""
mock_bridge = mock_phue.Bridge
with assert_setup_component(1):
with patch('homeassistant.helpers.discovery.load_platform') \
as mock_load:
self.assertTrue(setup_component(
self.hass, hue.DOMAIN,
{hue.DOMAIN: {hue.CONF_BRIDGES: [
{CONF_HOST: 'localhost'}]}}))
mock_bridge.assert_called_once_with(
'localhost',
config_file_path=get_test_config_dir(hue.PHUE_CONFIG_FILE))
mock_load.assert_called_once_with(
self.hass, 'light', hue.DOMAIN,
{'bridge_id': '127.0.0.1'})
self.assertTrue(hue.DOMAIN in self.hass.data)
self.assertEquals(1, len(self.hass.data[hue.DOMAIN]))
@MockDependency('phue')
def test_setup_with_phue_conf(self, mock_phue):
"""No host in the config file, but one is cached in phue.conf."""
mock_bridge = mock_phue.Bridge
with assert_setup_component(1):
with patch(
'homeassistant.components.hue._find_host_from_config',
return_value='localhost'):
with patch('homeassistant.helpers.discovery.load_platform') \
as mock_load:
self.assertTrue(setup_component(
self.hass, hue.DOMAIN,
{hue.DOMAIN: {hue.CONF_BRIDGES: [
{CONF_FILENAME: 'phue.conf'}]}}))
mock_bridge.assert_called_once_with(
'localhost',
config_file_path=get_test_config_dir(
hue.PHUE_CONFIG_FILE))
mock_load.assert_called_once_with(
self.hass, 'light', hue.DOMAIN,
{'bridge_id': '127.0.0.1'})
self.assertTrue(hue.DOMAIN in self.hass.data)
self.assertEquals(1, len(self.hass.data[hue.DOMAIN]))
@MockDependency('phue')
def test_setup_with_multiple_hosts(self, mock_phue):
"""Multiple hosts specified in the config file."""
mock_bridge = mock_phue.Bridge
with assert_setup_component(1):
with patch('homeassistant.helpers.discovery.load_platform') \
as mock_load:
self.assertTrue(setup_component(
self.hass, hue.DOMAIN,
{hue.DOMAIN: {hue.CONF_BRIDGES: [
{CONF_HOST: 'localhost'},
{CONF_HOST: '192.168.0.1'}]}}))
mock_bridge.assert_has_calls([
call(
'localhost',
config_file_path=get_test_config_dir(
hue.PHUE_CONFIG_FILE)),
call(
'192.168.0.1',
config_file_path=get_test_config_dir(
hue.PHUE_CONFIG_FILE))])
mock_load.mock_bridge.assert_not_called()
mock_load.assert_has_calls([
call(
self.hass, 'light', hue.DOMAIN,
{'bridge_id': '127.0.0.1'}),
call(
self.hass, 'light', hue.DOMAIN,
{'bridge_id': '192.168.0.1'}),
], any_order=True)
self.assertTrue(hue.DOMAIN in self.hass.data)
self.assertEquals(2, len(self.hass.data[hue.DOMAIN]))
@MockDependency('phue')
def test_bridge_discovered(self, mock_phue):
"""Bridge discovery."""
mock_bridge = mock_phue.Bridge
mock_service = MagicMock()
discovery_info = {'host': '192.168.0.10', 'serial': 'foobar'}
with patch('homeassistant.helpers.discovery.load_platform') \
as mock_load:
self.assertTrue(setup_component(
self.hass, hue.DOMAIN, {}))
hue.bridge_discovered(self.hass, mock_service, discovery_info)
mock_bridge.assert_called_once_with(
'192.168.0.10',
config_file_path=get_test_config_dir('phue-foobar.conf'))
mock_load.assert_called_once_with(
self.hass, 'light', hue.DOMAIN,
{'bridge_id': '192.168.0.10'})
self.assertTrue(hue.DOMAIN in self.hass.data)
self.assertEquals(1, len(self.hass.data[hue.DOMAIN]))
@MockDependency('phue')
def test_bridge_configure_and_discovered(self, mock_phue):
"""Bridge is in the config file, then we discover it."""
mock_bridge = mock_phue.Bridge
mock_service = MagicMock()
discovery_info = {'host': '192.168.1.10', 'serial': 'foobar'}
with assert_setup_component(1):
with patch('homeassistant.helpers.discovery.load_platform') \
as mock_load:
# First we set up the component from config
self.assertTrue(setup_component(
self.hass, hue.DOMAIN,
{hue.DOMAIN: {hue.CONF_BRIDGES: [
{CONF_HOST: '192.168.1.10'}]}}))
mock_bridge.assert_called_once_with(
'192.168.1.10',
config_file_path=get_test_config_dir(
hue.PHUE_CONFIG_FILE))
calls_to_mock_load = [
call(
self.hass, 'light', hue.DOMAIN,
{'bridge_id': '192.168.1.10'}),
]
mock_load.assert_has_calls(calls_to_mock_load)
self.assertTrue(hue.DOMAIN in self.hass.data)
self.assertEquals(1, len(self.hass.data[hue.DOMAIN]))
# Then we discover the same bridge
hue.bridge_discovered(self.hass, mock_service, discovery_info)
# No additional calls
mock_bridge.assert_called_once_with(
'192.168.1.10',
config_file_path=get_test_config_dir(
hue.PHUE_CONFIG_FILE))
mock_load.assert_has_calls(calls_to_mock_load)
# Still only one
self.assertTrue(hue.DOMAIN in self.hass.data)
self.assertEquals(1, len(self.hass.data[hue.DOMAIN]))
class TestHueBridge(unittest.TestCase):
"""Test the HueBridge class."""
def setUp(self): # pylint: disable=invalid-name
"""Setup things to be run when tests are started."""
self.hass = get_test_home_assistant()
self.hass.data[hue.DOMAIN] = {}
self.skip_teardown_stop = False
def tearDown(self):
"""Stop everything that was started."""
if not self.skip_teardown_stop:
self.hass.stop()
@MockDependency('phue')
def test_setup_bridge_connection_refused(self, mock_phue):
"""Test a registration failed with a connection refused exception."""
mock_bridge = mock_phue.Bridge
mock_bridge.side_effect = ConnectionRefusedError()
bridge = hue.HueBridge('localhost', self.hass, hue.PHUE_CONFIG_FILE)
bridge.setup()
self.assertFalse(bridge.configured)
self.assertTrue(bridge.config_request_id is None)
mock_bridge.assert_called_once_with(
'localhost',
config_file_path=get_test_config_dir(hue.PHUE_CONFIG_FILE))
@MockDependency('phue')
def test_setup_bridge_registration_exception(self, mock_phue):
"""Test a registration failed with an exception."""
mock_bridge = mock_phue.Bridge
mock_phue.PhueRegistrationException = Exception
mock_bridge.side_effect = mock_phue.PhueRegistrationException(1, 2)
bridge = hue.HueBridge('localhost', self.hass, hue.PHUE_CONFIG_FILE)
bridge.setup()
self.assertFalse(bridge.configured)
self.assertFalse(bridge.config_request_id is None)
self.assertTrue(isinstance(bridge.config_request_id, str))
mock_bridge.assert_called_once_with(
'localhost',
config_file_path=get_test_config_dir(hue.PHUE_CONFIG_FILE))
@MockDependency('phue')
def test_setup_bridge_registration_succeeds(self, mock_phue):
"""Test a registration success sequence."""
mock_bridge = mock_phue.Bridge
mock_phue.PhueRegistrationException = Exception
mock_bridge.side_effect = [
# First call, raise because not registered
mock_phue.PhueRegistrationException(1, 2),
# Second call, registration is done
None,
]
bridge = hue.HueBridge('localhost', self.hass, hue.PHUE_CONFIG_FILE)
bridge.setup()
self.assertFalse(bridge.configured)
self.assertFalse(bridge.config_request_id is None)
# Simulate the user confirming the registration
self.hass.services.call(
configurator.DOMAIN, configurator.SERVICE_CONFIGURE,
{configurator.ATTR_CONFIGURE_ID: bridge.config_request_id})
self.hass.block_till_done()
self.assertTrue(bridge.configured)
self.assertTrue(bridge.config_request_id is None)
# We should see a total of two identical calls
args = call(
'localhost',
config_file_path=get_test_config_dir(hue.PHUE_CONFIG_FILE))
mock_bridge.assert_has_calls([args, args])
# Make sure the request is done
self.assertEqual(1, len(self.hass.states.all()))
self.assertEqual('configured', self.hass.states.all()[0].state)
@MockDependency('phue')
def test_setup_bridge_registration_fails(self, mock_phue):
"""
Test a registration failure sequence.
This may happen when we start the registration process, the user
responds to the request but the bridge has become unreachable.
"""
mock_bridge = mock_phue.Bridge
mock_phue.PhueRegistrationException = Exception
mock_bridge.side_effect = [
# First call, raise because not registered
mock_phue.PhueRegistrationException(1, 2),
# Second call, the bridge has gone away
ConnectionRefusedError(),
]
bridge = hue.HueBridge('localhost', self.hass, hue.PHUE_CONFIG_FILE)
bridge.setup()
self.assertFalse(bridge.configured)
self.assertFalse(bridge.config_request_id is None)
# Simulate the user confirming the registration
self.hass.services.call(
configurator.DOMAIN, configurator.SERVICE_CONFIGURE,
{configurator.ATTR_CONFIGURE_ID: bridge.config_request_id})
self.hass.block_till_done()
self.assertFalse(bridge.configured)
self.assertFalse(bridge.config_request_id is None)
# We should see a total of two identical calls
args = call(
'localhost',
config_file_path=get_test_config_dir(hue.PHUE_CONFIG_FILE))
mock_bridge.assert_has_calls([args, args])
# The request should still be pending
self.assertEqual(1, len(self.hass.states.all()))
self.assertEqual('configure', self.hass.states.all()[0].state)
@MockDependency('phue')
def test_setup_bridge_registration_retry(self, mock_phue):
"""
Test a registration retry sequence.
This may happen when we start the registration process, the user
responds to the request but we fail to confirm it with the bridge.
"""
mock_bridge = mock_phue.Bridge
mock_phue.PhueRegistrationException = Exception
mock_bridge.side_effect = [
# First call, raise because not registered
mock_phue.PhueRegistrationException(1, 2),
# Second call, for whatever reason authentication fails
mock_phue.PhueRegistrationException(1, 2),
]
bridge = hue.HueBridge('localhost', self.hass, hue.PHUE_CONFIG_FILE)
bridge.setup()
self.assertFalse(bridge.configured)
self.assertFalse(bridge.config_request_id is None)
# Simulate the user confirming the registration
self.hass.services.call(
configurator.DOMAIN, configurator.SERVICE_CONFIGURE,
{configurator.ATTR_CONFIGURE_ID: bridge.config_request_id})
self.hass.block_till_done()
self.assertFalse(bridge.configured)
self.assertFalse(bridge.config_request_id is None)
# We should see a total of two identical calls
args = call(
'localhost',
config_file_path=get_test_config_dir(hue.PHUE_CONFIG_FILE))
mock_bridge.assert_has_calls([args, args])
# Make sure the request is done
self.assertEqual(1, len(self.hass.states.all()))
self.assertEqual('configure', self.hass.states.all()[0].state)
self.assertEqual(
'Failed to register, please try again.',
self.hass.states.all()[0].attributes.get(configurator.ATTR_ERRORS))
@MockDependency('phue')
def test_hue_activate_scene(self, mock_phue):
"""Test the hue_activate_scene service."""
with patch('homeassistant.helpers.discovery.load_platform'):
bridge = hue.HueBridge('localhost', self.hass,
hue.PHUE_CONFIG_FILE)
bridge.setup()
# No args
self.hass.services.call(hue.DOMAIN, hue.SERVICE_HUE_SCENE,
blocking=True)
bridge.bridge.run_scene.assert_not_called()
# Only one arg
self.hass.services.call(
hue.DOMAIN, hue.SERVICE_HUE_SCENE,
{hue.ATTR_GROUP_NAME: 'group'},
blocking=True)
bridge.bridge.run_scene.assert_not_called()
self.hass.services.call(
hue.DOMAIN, hue.SERVICE_HUE_SCENE,
{hue.ATTR_SCENE_NAME: 'scene'},
blocking=True)
bridge.bridge.run_scene.assert_not_called()
# Both required args
self.hass.services.call(
hue.DOMAIN, hue.SERVICE_HUE_SCENE,
{hue.ATTR_GROUP_NAME: 'group', hue.ATTR_SCENE_NAME: 'scene'},
blocking=True)
bridge.bridge.run_scene.assert_called_once_with('group', 'scene')