jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-04-23 08:47:57 +00:00
Code Issues Packages Projects Releases Wiki Activity

Refactor Hue: phue -> aiohue (#13043)

Browse Source 
* phue -> aiohue

* Clean up

* Fix config

* Address comments

* Typo

* Fix rebase error

* Mark light as unavailable when bridge is disconnected

* Tests

* Make Throttle work with double delay and async

* Rework update logic

* Don't resolve host to IP

* Clarify comment

* No longer do unnecessary updates

* Add more doc

* Another comment update

* Wrap up tests

* Lint

* Fix tests

* PyLint does not like mix 'n match async and coroutine

* Lint

* Update aiohue to 1.2

* Lint

* Fix await MagicMock
This commit is contained in:
Paulus Schoutsen 2018-03-16 20:27:05 -07:00 committed by GitHub
parent d78e75db66
commit 5a9013cda5
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
20 changed files with 1289 additions and 1485 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
homeassistant/components/discovery.py
View File

@ -6,7 +6,6 @@ Will emit EVENT_PLATFORM_DISCOVERED whenever a new service has been discovered.
Knows which components handle certain types, will make sure they are
loaded before the EVENT_PLATFORM_DISCOVERED is fired.
"""
import asyncio
import json
from datetime import timedelta
import logging
@ -84,8 +83,7 @@ CONFIG_SCHEMA = vol.Schema({
}, extra=vol.ALLOW_EXTRA)
@asyncio.coroutine
def async_setup(hass, config):
async def async_setup(hass, config):
"""Start a discovery service."""
from netdisco.discovery import NetworkDiscovery
@ -99,8 +97,7 @@ def async_setup(hass, config):
# Platforms ignore by config
ignored_platforms = config[DOMAIN][CONF_IGNORE]
@asyncio.coroutine
def new_service_found(service, info):
async def new_service_found(service, info):
"""Handle a new service if one is found."""
if service in ignored_platforms:
logger.info("Ignoring service: %s %s", service, info)
@ -124,15 +121,14 @@ def async_setup(hass, config):
component, platform = comp_plat
if platform is None:
yield from async_discover(hass, service, info, component, config)
await async_discover(hass, service, info, component, config)
else:
yield from async_load_platform(
await async_load_platform(
hass, component, platform, info, config)
@asyncio.coroutine
def scan_devices(now):
async def scan_devices(now):
"""Scan for devices."""
results = yield from hass.async_add_job(_discover, netdisco)
results = await hass.async_add_job(_discover, netdisco)
for result in results:
hass.async_add_job(new_service_found(*result))

261
homeassistant/components/hue/__init__.py
View File

@ -6,22 +6,22 @@ https://home-assistant.io/components/hue/
"""
import asyncio
import json
from functools import partial
import ipaddress
import logging
import os
import socket
import async_timeout
import requests
import voluptuous as vol
from homeassistant.core import callback
from homeassistant.components.discovery import SERVICE_HUE
from homeassistant.const import CONF_FILENAME, CONF_HOST
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers import discovery, aiohttp_client
from homeassistant import config_entries
from homeassistant.util.json import save_json
REQUIREMENTS = ['phue==1.0', 'aiohue==0.3.0']
REQUIREMENTS = ['aiohue==1.2.0']
_LOGGER = logging.getLogger(__name__)
@ -36,26 +36,23 @@ 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,
BRIDGE_CONFIG_SCHEMA = vol.Schema({
# Validate as IP address and then convert back to a string.
vol.Required(CONF_HOST): vol.All(ipaddress.ip_address, 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): BRIDGE_CONFIG_SCHEMA,
vol.Optional(CONF_BRIDGES):
vol.All(cv.ensure_list, [BRIDGE_CONFIG_SCHEMA]),
}),
}, extra=vol.ALLOW_EXTRA)
@ -73,7 +70,7 @@ Press the button on the bridge to register Philips Hue with Home Assistant.
"""
def setup(hass, config):
async def async_setup(hass, config):
"""Set up the Hue platform."""
conf = config.get(DOMAIN)
if conf is None:
@ -82,135 +79,130 @@ def setup(hass, 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))
async def async_bridge_discovered(service, discovery_info):
"""Dispatcher for Hue discovery events."""
# Ignore emulated hue
if "HASS Bridge" in discovery_info.get('name', ''):
return
await async_setup_bridge(
hass, discovery_info['host'],
'phue-{}.conf'.format(discovery_info['serial']))
discovery.async_listen(hass, SERVICE_HUE, async_bridge_discovered)
# User has configured bridges
if CONF_BRIDGES in conf:
bridges = conf[CONF_BRIDGES]
# Component is part of config but no bridges specified, discover.
elif DOMAIN in config:
# discover from nupnp
hosts = requests.get(API_NUPNP).json()
bridges = [{
websession = aiohttp_client.async_get_clientsession(hass)
async with websession.get(API_NUPNP) as req:
hosts = await req.json()
# Run through config schema to populate defaults
bridges = [BRIDGE_CONFIG_SCHEMA({
CONF_HOST: entry['internalipaddress'],
CONF_FILENAME: '.hue_{}.conf'.format(entry['id']),
} for entry in hosts]
}) for entry in hosts]
else:
# Component not specified in config, we're loaded via discovery
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)
if not bridges:
return True
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)
await asyncio.wait([
async_setup_bridge(
hass, bridge[CONF_HOST], bridge[CONF_FILENAME],
bridge[CONF_ALLOW_UNREACHABLE], bridge[CONF_ALLOW_HUE_GROUPS]
) for bridge in bridges
])
return True
def bridge_discovered(hass, service, discovery_info):
"""Dispatcher for Hue discovery events."""
if "HASS Bridge" in discovery_info.get('name', ''):
return
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,
username=None):
async def async_setup_bridge(
hass, host, filename=None,
allow_unreachable=DEFAULT_ALLOW_UNREACHABLE,
allow_hue_groups=DEFAULT_ALLOW_HUE_GROUPS,
username=None):
"""Set up a given Hue bridge."""
assert filename or username, 'Need to pass at least a username or filename'
# Only register a device once
if socket.gethostbyname(host) in hass.data[DOMAIN]:
if host in hass.data[DOMAIN]:
return
if username is None:
username = await hass.async_add_job(
_find_username_from_config, hass, filename)
bridge = HueBridge(host, hass, filename, username, allow_unreachable,
allow_in_emulated_hue, allow_hue_groups)
bridge.setup()
allow_hue_groups)
await bridge.async_setup()
hass.data[DOMAIN][host] = bridge
def _find_host_from_config(hass, filename=PHUE_CONFIG_FILE):
"""Attempt to detect host based on existing configuration."""
def _find_username_from_config(hass, filename):
"""Load username from config."""
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
with open(path) as inp:
return list(json.load(inp).values())[0]['username']
class HueBridge(object):
"""Manages a single Hue bridge."""
def __init__(self, host, hass, filename, username, allow_unreachable=False,
allow_in_emulated_hue=True, allow_hue_groups=True):
def __init__(self, host, hass, filename, username,
allow_unreachable=False, allow_groups=True):
"""Initialize the system."""
self.host = host
self.bridge_id = socket.gethostbyname(host)
self.hass = hass
self.filename = filename
self.username = username
self.allow_unreachable = allow_unreachable
self.allow_in_emulated_hue = allow_in_emulated_hue
self.allow_hue_groups = allow_hue_groups
self.allow_groups = allow_groups
self.available = True
self.bridge = None
self.lights = {}
self.lightgroups = {}
self.configured = False
self.config_request_id = None
self.api = None
hass.data[DOMAIN][self.bridge_id] = self
def setup(self):
async def async_setup(self):
"""Set up a phue bridge based on host parameter."""
import phue
import aiohue
api = aiohue.Bridge(
self.host,
username=self.username,
websession=aiohttp_client.async_get_clientsession(self.hass)
)
try:
kwargs = {}
if self.username is not None:
kwargs['username'] = self.username
if self.filename is not None:
kwargs['config_file_path'] = \
self.hass.config.path(self.filename)
self.bridge = phue.Bridge(self.host, **kwargs)
except OSError: # Wrong host was given
with async_timeout.timeout(5):
# Initialize bridge and validate our username
if not self.username:
await api.create_user('home-assistant')
await api.initialize()
except (aiohue.LinkButtonNotPressed, aiohue.Unauthorized):
_LOGGER.warning("Connected to Hue at %s but not registered.",
self.host)
self.async_request_configuration()
return
except (asyncio.TimeoutError, aiohue.RequestError):
_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()
except aiohue.AiohueException:
_LOGGER.exception('Unknown Hue linking error occurred')
self.async_request_configuration()
return
except Exception: # pylint: disable=broad-except
_LOGGER.exception("Unknown error connecting with Hue bridge at %s",
@ -221,57 +213,77 @@ class HueBridge(object):
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.hass.components.configurator.async_request_done(request_id)
self.configured = True
self.username = api.username
discovery.load_platform(
# Save config file
await self.hass.async_add_job(
save_json, self.hass.config.path(self.filename),
{self.host: {'username': api.username}})
self.api = api
self.hass.async_add_job(discovery.async_load_platform(
self.hass, 'light', DOMAIN,
{'bridge_id': self.bridge_id})
{'host': 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)
self.hass.services.register(
DOMAIN, SERVICE_HUE_SCENE, hue_activate_scene,
self.hass.services.async_register(
DOMAIN, SERVICE_HUE_SCENE, self.hue_activate_scene,
schema=SCENE_SCHEMA)
def request_configuration(self):
@callback
def async_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(
configurator.async_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(),
async def config_callback(data):
"""Callback for configurator data."""
await self.async_setup()
self.config_request_id = configurator.async_request_config(
"Philips Hue", config_callback,
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()
async def hue_activate_scene(self, call, updated=False):
"""Service to call directly into bridge to set scenes."""
group_name = call.data[ATTR_GROUP_NAME]
scene_name = call.data[ATTR_SCENE_NAME]
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)
group = next(
(group for group in self.api.groups.values()
if group.name == group_name), None)
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)
scene_id = next(
(scene.id for scene in self.api.scenes.values()
if scene.name == scene_name), None)
# If we can't find it, fetch latest info.
if not updated and (group is None or scene_id is None):
await self.api.groups.update()
await self.api.scenes.update()
await self.hue_activate_scene(call, updated=True)
return
if group is None:
_LOGGER.warning('Unable to find group %s', group_name)
return
if scene_id is None:
_LOGGER.warning('Unable to find scene %s', scene_name)
return
await group.set_action(scene=scene_id)
@config_entries.HANDLERS.register(DOMAIN)
@ -374,7 +386,6 @@ class HueFlowHandler(config_entries.ConfigFlowHandler):
async def async_setup_entry(hass, entry):
"""Set up a bridge for a config entry."""
await hass.async_add_job(partial(
setup_bridge, entry.data['host'], hass,
username=entry.data['username']))
await async_setup_bridge(hass, entry.data['host'],
username=entry.data['username'])
return True

367
homeassistant/components/light/hue.py
View File

@ -8,31 +8,23 @@ import asyncio
from datetime import timedelta
import logging
import random
import re
import socket
import voluptuous as vol
import async_timeout
import homeassistant.components.hue as hue
from homeassistant.components.light import (
ATTR_BRIGHTNESS, ATTR_COLOR_TEMP, ATTR_EFFECT, ATTR_FLASH, ATTR_RGB_COLOR,
ATTR_TRANSITION, ATTR_XY_COLOR, EFFECT_COLORLOOP, EFFECT_RANDOM,
FLASH_LONG, FLASH_SHORT, PLATFORM_SCHEMA, SUPPORT_BRIGHTNESS,
FLASH_LONG, FLASH_SHORT, SUPPORT_BRIGHTNESS,
SUPPORT_COLOR_TEMP, SUPPORT_EFFECT, SUPPORT_FLASH, SUPPORT_RGB_COLOR,
SUPPORT_TRANSITION, SUPPORT_XY_COLOR, Light)
from homeassistant.const import CONF_FILENAME, CONF_HOST, DEVICE_DEFAULT_NAME
import homeassistant.helpers.config_validation as cv
import homeassistant.util as util
from homeassistant.util import yaml
import homeassistant.util.color as color_util
DEPENDENCIES = ['hue']
SCAN_INTERVAL = timedelta(seconds=5)
_LOGGER = logging.getLogger(__name__)
MIN_TIME_BETWEEN_SCANS = timedelta(seconds=10)
MIN_TIME_BETWEEN_FORCED_SCANS = timedelta(milliseconds=100)
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)
@ -48,244 +40,232 @@ SUPPORT_HUE = {
'Color temperature light': SUPPORT_HUE_COLOR_TEMP
}
ATTR_EMULATED_HUE_HIDDEN = 'emulated_hue_hidden'
ATTR_IS_HUE_GROUP = 'is_hue_group'
# 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({
vol.Optional(CONF_HOST): cv.string,
vol.Optional(CONF_ALLOW_UNREACHABLE): cv.boolean,
vol.Optional(CONF_FILENAME): cv.string,
vol.Optional(CONF_ALLOW_IN_EMULATED_HUE): cv.boolean,
vol.Optional(CONF_ALLOW_HUE_GROUPS,
default=DEFAULT_ALLOW_HUE_GROUPS): cv.boolean,
})
MIGRATION_ID = 'light_hue_config_migration'
MIGRATION_TITLE = 'Philips Hue Configuration Migration'
MIGRATION_INSTRUCTIONS = """
Configuration for the Philips Hue component has changed; action required.
You have configured at least one bridge:
hue:
{config}
This configuration is deprecated, please check the
[Hue component](https://home-assistant.io/components/hue/) page for more
information.
"""
SIGNAL_CALLBACK = 'hue_light_callback_{}_{}'
# Minimum Hue Bridge API version to support groups
# 1.4.0 introduced extended group info
# 1.12 introduced the state object for groups
# 1.13 introduced "any_on" to group state objects
GROUP_MIN_API_VERSION = (1, 13, 0)
def setup_platform(hass, config, add_devices, discovery_info=None):
async def async_setup_platform(hass, config, async_add_devices,
discovery_info=None):
"""Set up the Hue lights."""
if discovery_info is None or 'bridge_id' not in discovery_info:
if discovery_info is None:
return
if config is not None and config:
# 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 = hass.data[hue.DOMAIN][discovery_info['host']]
cur_lights = {}
cur_groups = {}
bridge_id = discovery_info['bridge_id']
bridge = hass.data[hue.DOMAIN][bridge_id]
unthrottled_update_lights(hass, bridge, add_devices)
api_version = tuple(
int(v) for v in bridge.api.config.apiversion.split('.'))
allow_groups = bridge.allow_groups
if allow_groups and api_version < GROUP_MIN_API_VERSION:
_LOGGER.warning('Please update your Hue bridge to support groups')
allow_groups = False
# Hue updates all lights via a single API call.
#
# If we call a service to update 2 lights, we only want the API to be
# called once.
#
# The throttle decorator will return right away if a call is currently
# in progress. This means that if we are updating 2 lights, the first one
# is in the update method, the second one will skip it and assume the
# update went through and updates it's data, not good!
#
# The current mechanism will make sure that all lights will wait till
# the update call is done before writing their data to the state machine.
#
# An alternative approach would be to disable automatic polling by Home
# Assistant and take control ourselves. This works great for polling as now
# we trigger from 1 time update an update to all entities. However it gets
# tricky from inside async_turn_on and async_turn_off.
#
# If automatic polling is enabled, Home Assistant will call the entity
# update method after it is done calling all the services. This means that
# when we update, we know all commands have been processed. If we trigger
# the update from inside async_turn_on, the update will not capture the
# changes to the second entity until the next polling update because the
# throttle decorator will prevent the call.
progress = None
light_progress = set()
group_progress = set()
async def request_update(is_group, object_id):
"""Request an update.
We will only make 1 request to the server for updating at a time. If a
request is in progress, we will join the request that is in progress.
This approach is possible because should_poll=True. That means that
Home Assistant will ask lights for updates during a polling cycle or
after it has called a service.
We keep track of the lights that are waiting for the request to finish.
When new data comes in, we'll trigger an update for all non-waiting
lights. This covers the case where a service is called to enable 2
lights but in the meanwhile some other light has changed too.
"""
nonlocal progress
progress_set = group_progress if is_group else light_progress
progress_set.add(object_id)
if progress is not None:
return await progress
progress = asyncio.ensure_future(update_bridge())
result = await progress
progress = None
light_progress.clear()
group_progress.clear()
return result
async def update_bridge():
"""Update the values of the bridge.
Will update lights and, if enabled, groups from the bridge.
"""
tasks = []
tasks.append(async_update_items(
hass, bridge, async_add_devices, request_update,
False, cur_lights, light_progress
))
if allow_groups:
tasks.append(async_update_items(
hass, bridge, async_add_devices, request_update,
True, cur_groups, group_progress
))
await asyncio.wait(tasks)
await update_bridge()
@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)
async def async_update_items(hass, bridge, async_add_devices,
request_bridge_update, is_group, current,
progress_waiting):
"""Update either groups or lights from the bridge."""
import aiohue
def unthrottled_update_lights(hass, bridge, add_devices):
"""Update the lights (Internal version of update_lights)."""
import phue
if not bridge.configured:
return
if is_group:
api = bridge.api.groups
else:
api = bridge.api.lights
try:
api = bridge.get_api()
except phue.PhueRequestTimeout:
_LOGGER.warning("Timeout trying to reach the bridge")
bridge.available = False
return
except ConnectionRefusedError:
_LOGGER.error("The bridge refused the connection")
bridge.available = False
return
except socket.error:
# socket.error when we cannot reach Hue
_LOGGER.exception("Cannot reach the bridge")
with async_timeout.timeout(4):
await api.update()
except (asyncio.TimeoutError, aiohue.AiohueException):
if not bridge.available:
return
_LOGGER.error('Unable to reach bridge %s', bridge.host)
bridge.available = False
for light_id, light in current.items():
if light_id not in progress_waiting:
light.async_schedule_update_ha_state()
return
bridge.available = True
if not bridge.available:
_LOGGER.info('Reconnected to bridge %s', bridge.host)
bridge.available = True
new_lights = process_lights(
hass, api, bridge,
lambda **kw: update_lights(hass, bridge, add_devices, **kw))
if bridge.allow_hue_groups:
new_lightgroups = process_groups(
hass, api, bridge,
lambda **kw: update_lights(hass, bridge, add_devices, **kw))
new_lights.extend(new_lightgroups)
new_lights = []
for item_id in api:
if item_id not in current:
current[item_id] = HueLight(
api[item_id], request_bridge_update, bridge, is_group)
new_lights.append(current[item_id])
elif item_id not in progress_waiting:
current[item_id].async_schedule_update_ha_state()
if new_lights:
add_devices(new_lights)
def process_lights(hass, api, bridge, update_lights_cb):
"""Set up HueLight objects for all lights."""
api_lights = api.get('lights')
if not isinstance(api_lights, dict):
_LOGGER.error("Got unexpected result from Hue API")
return []
new_lights = []
for light_id, info in api_lights.items():
if light_id not in bridge.lights:
bridge.lights[light_id] = HueLight(
int(light_id), info, bridge,
update_lights_cb,
bridge.allow_unreachable,
bridge.allow_in_emulated_hue)
new_lights.append(bridge.lights[light_id])
else:
bridge.lights[light_id].info = info
hass.helpers.dispatcher.dispatcher_send(
SIGNAL_CALLBACK.format(
bridge.bridge_id,
bridge.lights[light_id].light_id))
return new_lights
def process_groups(hass, api, bridge, 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 = []
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 bridge.lightgroups:
bridge.lightgroups[lightgroup_id] = HueLight(
int(lightgroup_id), info, bridge,
update_lights_cb,
bridge.allow_unreachable,
bridge.allow_in_emulated_hue, True)
new_lights.append(bridge.lightgroups[lightgroup_id])
else:
bridge.lightgroups[lightgroup_id].info = info
hass.helpers.dispatcher.dispatcher_send(
SIGNAL_CALLBACK.format(
bridge.bridge_id,
bridge.lightgroups[lightgroup_id].light_id))
return new_lights
async_add_devices(new_lights)
class HueLight(Light):
"""Representation of a Hue light."""
def __init__(self, light_id, info, bridge, update_lights_cb,
allow_unreachable, allow_in_emulated_hue, is_group=False):
def __init__(self, light, request_bridge_update, bridge, is_group=False):
"""Initialize the light."""
self.light_id = light_id
self.info = info
self.light = light
self.async_request_bridge_update = request_bridge_update
self.bridge = bridge
self.update_lights = update_lights_cb
self.allow_unreachable = allow_unreachable
self.is_group = is_group
self.allow_in_emulated_hue = allow_in_emulated_hue
if is_group:
self._command_func = self.bridge.set_group
self.is_osram = False
self.is_philips = False
else:
self._command_func = self.bridge.set_light
self.is_osram = light.manufacturername == 'OSRAM'
self.is_philips = light.manufacturername == 'Philips'
@property
def unique_id(self):
"""Return the ID of this Hue light."""
return self.info.get('uniqueid')
return self.light.uniqueid
@property
def name(self):
"""Return the name of the Hue light."""
return self.info.get('name', DEVICE_DEFAULT_NAME)
return self.light.name
@property
def brightness(self):
"""Return the brightness of this light between 0..255."""
if self.is_group:
return self.info['action'].get('bri')
return self.info['state'].get('bri')
return self.light.action.get('bri')
return self.light.state.get('bri')
@property
def xy_color(self):
"""Return the XY color value."""
if self.is_group:
return self.info['action'].get('xy')
return self.info['state'].get('xy')
return self.light.action.get('xy')
return self.light.state.get('xy')
@property
def color_temp(self):
"""Return the CT color value."""
if self.is_group:
return self.info['action'].get('ct')
return self.info['state'].get('ct')
return self.light.action.get('ct')
return self.light.state.get('ct')
@property
def is_on(self):
"""Return true if device is on."""
if self.is_group:
return self.info['state']['any_on']
return self.info['state']['on']
return self.light.state['any_on']
return self.light.state['on']
@property
def available(self):
"""Return if light is available."""
return self.bridge.available and (self.is_group or
self.allow_unreachable or
self.info['state']['reachable'])
self.bridge.allow_unreachable or
self.light.state['reachable'])
@property
def supported_features(self):
"""Flag supported features."""
return SUPPORT_HUE.get(self.info.get('type'), SUPPORT_HUE_EXTENDED)
return SUPPORT_HUE.get(self.light.type, SUPPORT_HUE_EXTENDED)
@property
def effect_list(self):
"""Return the list of supported effects."""
return [EFFECT_COLORLOOP, EFFECT_RANDOM]
def turn_on(self, **kwargs):
async def async_turn_on(self, **kwargs):
"""Turn the specified or all lights on."""
command = {'on': True}
@ -293,7 +273,7 @@ class HueLight(Light):
command['transitiontime'] = int(kwargs[ATTR_TRANSITION] * 10)
if ATTR_XY_COLOR in kwargs:
if self.info.get('manufacturername') == 'OSRAM':
if self.is_osram:
color_hue, sat = color_util.color_xy_to_hs(
*kwargs[ATTR_XY_COLOR])
command['hue'] = color_hue / 360 * 65535
@ -301,7 +281,7 @@ class HueLight(Light):
else:
command['xy'] = kwargs[ATTR_XY_COLOR]
elif ATTR_RGB_COLOR in kwargs:
if self.info.get('manufacturername') == 'OSRAM':
if self.is_osram:
hsv = color_util.color_RGB_to_hsv(
*(int(val) for val in kwargs[ATTR_RGB_COLOR]))
command['hue'] = hsv[0] / 360 * 65535
@ -336,12 +316,15 @@ class HueLight(Light):
elif effect == EFFECT_RANDOM:
command['hue'] = random.randrange(0, 65535)
command['sat'] = random.randrange(150, 254)
elif self.info.get('manufacturername') == 'Philips':
elif self.is_philips:
command['effect'] = 'none'
self._command_func(self.light_id, command)
if self.is_group:
await self.light.set_action(**command)
else:
await self.light.set_state(**command)
def turn_off(self, **kwargs):
async def async_turn_off(self, **kwargs):
"""Turn the specified or all lights off."""
command = {'on': False}
@ -359,27 +342,19 @@ class HueLight(Light):
else:
command['alert'] = 'none'
self._command_func(self.light_id, command)
if self.is_group:
await self.light.set_action(**command)
else:
await self.light.set_state(**command)
def update(self):
async def async_update(self):
"""Synchronize state with bridge."""
self.update_lights(no_throttle=True)
await self.async_request_bridge_update(self.is_group, self.light.id)
@property
def device_state_attributes(self):
"""Return the device state attributes."""
attributes = {}
if not self.allow_in_emulated_hue:
attributes[ATTR_EMULATED_HUE_HIDDEN] = \
not self.allow_in_emulated_hue
if self.is_group:
attributes[ATTR_IS_HUE_GROUP] = self.is_group
return attributes
@asyncio.coroutine
def async_added_to_hass(self):
"""Register update callback."""
dev_id = self.bridge.bridge_id, self.light_id
self.hass.helpers.dispatcher.async_dispatcher_connect(
SIGNAL_CALLBACK.format(*dev_id),
self.async_schedule_update_ha_state)

14
homeassistant/components/mqtt/discovery.py
View File

@ -4,7 +4,6 @@ Support for MQTT discovery.
For more details about this component, please refer to the documentation at
https://home-assistant.io/components/mqtt/#discovery
"""
import asyncio
import json
import logging
import re
@ -35,19 +34,16 @@ ALLOWED_PLATFORMS = {
ALREADY_DISCOVERED = 'mqtt_discovered_components'
@asyncio.coroutine
def async_start(hass, discovery_topic, hass_config):
async def async_start(hass, discovery_topic, hass_config):
"""Initialize of MQTT Discovery."""
# pylint: disable=unused-variable
@asyncio.coroutine
def async_device_message_received(topic, payload, qos):
async def async_device_message_received(topic, payload, qos):
"""Process the received message."""
match = TOPIC_MATCHER.match(topic)
if not match:
return
prefix_topic, component, node_id, object_id = match.groups()
_prefix_topic, component, node_id, object_id = match.groups()
try:
payload = json.loads(payload)
@ -88,10 +84,10 @@ def async_start(hass, discovery_topic, hass_config):
_LOGGER.info("Found new component: %s %s", component, discovery_id)
yield from async_load_platform(
await async_load_platform(
hass, component, platform, payload, hass_config)
yield from mqtt.async_subscribe(
await mqtt.async_subscribe(
hass, discovery_topic + '/#', async_device_message_received, 0)
return True

20
homeassistant/components/zwave/__init__.py
View File

@ -203,8 +203,8 @@ def get_config_value(node, value_index, tries=5):
return None
@asyncio.coroutine
def async_setup_platform(hass, config, async_add_devices, discovery_info=None):
async def async_setup_platform(hass, config, async_add_devices,
discovery_info=None):
"""Set up the Z-Wave platform (generic part)."""
if discovery_info is None or DATA_NETWORK not in hass.data:
return False
@ -504,8 +504,7 @@ def setup(hass, config):
"target node:%s, instance=%s", node_id, group,
target_node_id, instance)
@asyncio.coroutine
def async_refresh_entity(service):
async def async_refresh_entity(service):
"""Refresh values that specific entity depends on."""
entity_id = service.data.get(ATTR_ENTITY_ID)
async_dispatcher_send(
@ -559,8 +558,7 @@ def setup(hass, config):
network.start()
hass.bus.fire(const.EVENT_NETWORK_START)
@asyncio.coroutine
def _check_awaked():
async def _check_awaked():
"""Wait for Z-wave awaked state (or timeout) and finalize start."""
_LOGGER.debug(
"network state: %d %s", network.state,
@ -585,7 +583,7 @@ def setup(hass, config):
network.state_str)
break
else:
yield from asyncio.sleep(1, loop=hass.loop)
await asyncio.sleep(1, loop=hass.loop)
hass.async_add_job(_finalize_start)
@ -798,11 +796,10 @@ class ZWaveDeviceEntityValues():
dict_id = id(self)
@asyncio.coroutine
def discover_device(component, device, dict_id):
async def discover_device(component, device, dict_id):
"""Put device in a dictionary and call discovery on it."""
self._hass.data[DATA_DEVICES][dict_id] = device
yield from discovery.async_load_platform(
await discovery.async_load_platform(
self._hass, component, DOMAIN,
{const.DISCOVERY_DEVICE: dict_id}, self._zwave_config)
self._hass.add_job(discover_device, component, device, dict_id)
@ -844,8 +841,7 @@ class ZWaveDeviceEntity(ZWaveBaseEntity):
self.update_properties()
self.maybe_schedule_update()
@asyncio.coroutine
def async_added_to_hass(self):
async def async_added_to_hass(self):
"""Add device to dict."""
async_dispatcher_connect(
self.hass,

2
homeassistant/core.py
View File

@ -79,7 +79,7 @@ def callback(func: Callable[..., None]) -> Callable[..., None]:
def is_callback(func: Callable[..., Any]) -> bool:
"""Check if function is safe to be called in the event loop."""
return '_hass_callback' in func.__dict__
return '_hass_callback' in getattr(func, '__dict__', {})
@callback

18
homeassistant/helpers/discovery.py
View File

@ -5,8 +5,6 @@ There are two different types of discoveries that can be fired/listened for.
- listen_platform/discover_platform is for platforms. These are used by
components to allow discovery of their platforms.
"""
import asyncio
from homeassistant import setup, core
from homeassistant.loader import bind_hass
from homeassistant.const import (
@ -58,17 +56,16 @@ def discover(hass, service, discovered=None, component=None, hass_config=None):
async_discover(hass, service, discovered, component, hass_config))
@asyncio.coroutine
@bind_hass
def async_discover(hass, service, discovered=None, component=None,
hass_config=None):
async def async_discover(hass, service, discovered=None, component=None,
hass_config=None):
"""Fire discovery event. Can ensure a component is loaded."""
if component in DEPENDENCY_BLACKLIST:
raise HomeAssistantError(
'Cannot discover the {} component.'.format(component))
if component is not None and component not in hass.config.components:
yield from setup.async_setup_component(
await setup.async_setup_component(
hass, component, hass_config)
data = {
@ -134,10 +131,9 @@ def load_platform(hass, component, platform, discovered=None,
hass_config))
@asyncio.coroutine
@bind_hass
def async_load_platform(hass, component, platform, discovered=None,
hass_config=None):
async def async_load_platform(hass, component, platform, discovered=None,
hass_config=None):
"""Load a component and platform dynamically.
Target components will be loaded and an EVENT_PLATFORM_DISCOVERED will be
@ -148,7 +144,7 @@ def async_load_platform(hass, component, platform, discovered=None,
Use `listen_platform` to register a callback for these events.
Warning: Do not yield from this inside a setup method to avoid a dead lock.
Warning: Do not await this inside a setup method to avoid a dead lock.
Use `hass.async_add_job(async_load_platform(..))` instead.
This method is a coroutine.
@ -160,7 +156,7 @@ def async_load_platform(hass, component, platform, discovered=None,
setup_success = True
if component not in hass.config.components:
setup_success = yield from setup.async_setup_component(
setup_success = await setup.async_setup_component(
hass, component, hass_config)
# No need to fire event if we could not setup component

20
homeassistant/util/__init__.py
View File

@ -261,6 +261,16 @@ class Throttle(object):
def __call__(self, method):
"""Caller for the throttle."""
# Make sure we return a coroutine if the method is async.
if asyncio.iscoroutinefunction(method):
async def throttled_value():
"""Stand-in function for when real func is being throttled."""
return None
else:
def throttled_value():
"""Stand-in function for when real func is being throttled."""
return None
if self.limit_no_throttle is not None:
method = Throttle(self.limit_no_throttle)(method)
@ -277,16 +287,6 @@ class Throttle(object):
is_func = (not hasattr(method, '__self__') and
'.' not in method.__qualname__.split('.<locals>.')[-1])
# Make sure we return a coroutine if the method is async.
if asyncio.iscoroutinefunction(method):
async def throttled_value():
"""Stand-in function for when real func is being throttled."""
return None
else:
def throttled_value():
"""Stand-in function for when real func is being throttled."""
return None
@wraps(method)
def wrapper(*args, **kwargs):
"""Wrap that allows wrapped to be called only once per min_time.

5
requirements_all.txt
View File

@ -74,7 +74,7 @@ aiodns==1.1.1
aiohttp_cors==0.6.0
# homeassistant.components.hue
aiohue==0.3.0
aiohue==1.2.0
# homeassistant.components.sensor.imap
aioimaplib==0.7.13
@ -568,9 +568,6 @@ pdunehd==1.3
# homeassistant.components.media_player.pandora
pexpect==4.0.1
# homeassistant.components.hue
phue==1.0
# homeassistant.components.rpi_pfio
pifacecommon==4.1.2

2
requirements_test_all.txt
View File

@ -35,7 +35,7 @@ aioautomatic==0.6.5
aiohttp_cors==0.6.0
# homeassistant.components.hue
aiohue==0.3.0
aiohue==1.2.0
# homeassistant.components.notify.apns
apns2==0.3.0

1
tests/components/hue/__init__.py Normal file
View File

@ -0,0 +1 @@
"""Tests for the Hue component."""

17
tests/components/hue/conftest.py Normal file
View File

@ -0,0 +1,17 @@
"""Fixtures for Hue tests."""
from unittest.mock import patch
import pytest
from tests.common import mock_coro_func
@pytest.fixture
def mock_bridge():
"""Mock the HueBridge from initializing."""
with patch('homeassistant.components.hue._find_username_from_config',
return_value=None), \
patch('homeassistant.components.hue.HueBridge') as mock_bridge:
mock_bridge().async_setup = mock_coro_func()
mock_bridge.reset_mock()
yield mock_bridge

98
tests/components/hue/test_bridge.py Normal file
View File

@ -0,0 +1,98 @@
"""Test Hue bridge."""
import asyncio
from unittest.mock import Mock, patch
import aiohue
import pytest
from homeassistant.components import hue
from tests.common import mock_coro
class MockBridge(hue.HueBridge):
"""Class that sets default for constructor."""
def __init__(self, hass, host='1.2.3.4', filename='mock-bridge.conf',
username=None, **kwargs):
"""Initialize a mock bridge."""
super().__init__(host, hass, filename, username, **kwargs)
@pytest.fixture
def mock_request():
"""Mock configurator.async_request_config."""
with patch('homeassistant.components.configurator.'
'async_request_config') as mock_request:
yield mock_request
async def test_setup_request_config_button_not_pressed(hass, mock_request):
"""Test we request config if link button has not been pressed."""
with patch('aiohue.Bridge.create_user',
side_effect=aiohue.LinkButtonNotPressed):
await MockBridge(hass).async_setup()
assert len(mock_request.mock_calls) == 1
async def test_setup_request_config_invalid_username(hass, mock_request):
"""Test we request config if username is no longer whitelisted."""
with patch('aiohue.Bridge.create_user',
side_effect=aiohue.Unauthorized):
await MockBridge(hass).async_setup()
assert len(mock_request.mock_calls) == 1
async def test_setup_timeout(hass, mock_request):
"""Test we give up when there is a timeout."""
with patch('aiohue.Bridge.create_user',
side_effect=asyncio.TimeoutError):
await MockBridge(hass).async_setup()
assert len(mock_request.mock_calls) == 0
async def test_only_create_no_username(hass):
"""."""
with patch('aiohue.Bridge.create_user') as mock_create, \
patch('aiohue.Bridge.initialize') as mock_init:
await MockBridge(hass, username='bla').async_setup()
assert len(mock_create.mock_calls) == 0
assert len(mock_init.mock_calls) == 1
async def test_configurator_callback(hass, mock_request):
"""."""
with patch('aiohue.Bridge.create_user',
side_effect=aiohue.LinkButtonNotPressed):
await MockBridge(hass).async_setup()
assert len(mock_request.mock_calls) == 1
callback = mock_request.mock_calls[0][1][2]
mock_init = Mock(return_value=mock_coro())
mock_create = Mock(return_value=mock_coro())
with patch('aiohue.Bridge') as mock_bridge, \
patch('homeassistant.helpers.discovery.async_load_platform',
return_value=mock_coro()) as mock_load_platform, \
patch('homeassistant.components.hue.save_json') as mock_save:
inst = mock_bridge()
inst.username = 'mock-user'
inst.create_user = mock_create
inst.initialize = mock_init
await callback(None)
assert len(mock_create.mock_calls) == 1
assert len(mock_init.mock_calls) == 1
assert len(mock_save.mock_calls) == 1
assert mock_save.mock_calls[0][1][1] == {
'1.2.3.4': {
'username': 'mock-user'
}
}
assert len(mock_load_platform.mock_calls) == 1

184
tests/components/hue/test_config_flow.py Normal file
View File

@ -0,0 +1,184 @@
"""Tests for Philips Hue config flow."""
import asyncio
from unittest.mock import patch
import aiohue
import pytest
import voluptuous as vol
from homeassistant.components import hue
from tests.common import MockConfigEntry, mock_coro
async def test_flow_works(hass, aioclient_mock):
"""Test config flow ."""
aioclient_mock.get(hue.API_NUPNP, json=[
{'internalipaddress': '1.2.3.4', 'id': 'bla'}
])
flow = hue.HueFlowHandler()
flow.hass = hass
await flow.async_step_init()
with patch('aiohue.Bridge') as mock_bridge:
def mock_constructor(host, websession):
mock_bridge.host = host
return mock_bridge
mock_bridge.side_effect = mock_constructor
mock_bridge.username = 'username-abc'
mock_bridge.config.name = 'Mock Bridge'
mock_bridge.config.bridgeid = 'bridge-id-1234'
mock_bridge.create_user.return_value = mock_coro()
mock_bridge.initialize.return_value = mock_coro()
result = await flow.async_step_link(user_input={})
assert mock_bridge.host == '1.2.3.4'
assert len(mock_bridge.create_user.mock_calls) == 1
assert len(mock_bridge.initialize.mock_calls) == 1
assert result['type'] == 'create_entry'
assert result['title'] == 'Mock Bridge'
assert result['data'] == {
'host': '1.2.3.4',
'bridge_id': 'bridge-id-1234',
'username': 'username-abc'
}
async def test_flow_no_discovered_bridges(hass, aioclient_mock):
"""Test config flow discovers no bridges."""
aioclient_mock.get(hue.API_NUPNP, json=[])
flow = hue.HueFlowHandler()
flow.hass = hass
result = await flow.async_step_init()
assert result['type'] == 'abort'
async def test_flow_all_discovered_bridges_exist(hass, aioclient_mock):
"""Test config flow discovers only already configured bridges."""
aioclient_mock.get(hue.API_NUPNP, json=[
{'internalipaddress': '1.2.3.4', 'id': 'bla'}
])
MockConfigEntry(domain='hue', data={
'host': '1.2.3.4'
}).add_to_hass(hass)
flow = hue.HueFlowHandler()
flow.hass = hass
result = await flow.async_step_init()
assert result['type'] == 'abort'
async def test_flow_one_bridge_discovered(hass, aioclient_mock):
"""Test config flow discovers one bridge."""
aioclient_mock.get(hue.API_NUPNP, json=[
{'internalipaddress': '1.2.3.4', 'id': 'bla'}
])
flow = hue.HueFlowHandler()
flow.hass = hass
result = await flow.async_step_init()
assert result['type'] == 'form'
assert result['step_id'] == 'link'
async def test_flow_two_bridges_discovered(hass, aioclient_mock):
"""Test config flow discovers two bridges."""
aioclient_mock.get(hue.API_NUPNP, json=[
{'internalipaddress': '1.2.3.4', 'id': 'bla'},
{'internalipaddress': '5.6.7.8', 'id': 'beer'}
])
flow = hue.HueFlowHandler()
flow.hass = hass
result = await flow.async_step_init()
assert result['type'] == 'form'
assert result['step_id'] == 'init'
with pytest.raises(vol.Invalid):
assert result['data_schema']({'host': '0.0.0.0'})
result['data_schema']({'host': '1.2.3.4'})
result['data_schema']({'host': '5.6.7.8'})
async def test_flow_two_bridges_discovered_one_new(hass, aioclient_mock):
"""Test config flow discovers two bridges."""
aioclient_mock.get(hue.API_NUPNP, json=[
{'internalipaddress': '1.2.3.4', 'id': 'bla'},
{'internalipaddress': '5.6.7.8', 'id': 'beer'}
])
MockConfigEntry(domain='hue', data={
'host': '1.2.3.4'
}).add_to_hass(hass)
flow = hue.HueFlowHandler()
flow.hass = hass
result = await flow.async_step_init()
assert result['type'] == 'form'
assert result['step_id'] == 'link'
assert flow.host == '5.6.7.8'
async def test_flow_timeout_discovery(hass):
"""Test config flow ."""
flow = hue.HueFlowHandler()
flow.hass = hass
with patch('aiohue.discovery.discover_nupnp',
side_effect=asyncio.TimeoutError):
result = await flow.async_step_init()
assert result['type'] == 'abort'
async def test_flow_link_timeout(hass):
"""Test config flow ."""
flow = hue.HueFlowHandler()
flow.hass = hass
with patch('aiohue.Bridge.create_user',
side_effect=asyncio.TimeoutError):
result = await flow.async_step_link({})
assert result['type'] == 'form'
assert result['step_id'] == 'link'
assert result['errors'] == {
'base': 'register_failed'
}
async def test_flow_link_button_not_pressed(hass):
"""Test config flow ."""
flow = hue.HueFlowHandler()
flow.hass = hass
with patch('aiohue.Bridge.create_user',
side_effect=aiohue.LinkButtonNotPressed):
result = await flow.async_step_link({})
assert result['type'] == 'form'
assert result['step_id'] == 'link'
assert result['errors'] == {
'base': 'register_failed'
}
async def test_flow_link_unknown_host(hass):
"""Test config flow ."""
flow = hue.HueFlowHandler()
flow.hass = hass
with patch('aiohue.Bridge.create_user',
side_effect=aiohue.RequestError):
result = await flow.async_step_link({})
assert result['type'] == 'form'
assert result['step_id'] == 'link'
assert result['errors'] == {
'base': 'register_failed'
}

74
tests/components/hue/test_setup.py Normal file
View File

@ -0,0 +1,74 @@
"""Test Hue setup process."""
from homeassistant.setup import async_setup_component
from homeassistant.components import hue
from homeassistant.components.discovery import SERVICE_HUE
async def test_setup_with_multiple_hosts(hass, mock_bridge):
"""Multiple hosts specified in the config file."""
assert await async_setup_component(hass, hue.DOMAIN, {
hue.DOMAIN: {
hue.CONF_BRIDGES: [
{hue.CONF_HOST: '127.0.0.1'},
{hue.CONF_HOST: '192.168.1.10'},
]
}
})
assert len(mock_bridge.mock_calls) == 2
hosts = sorted(mock_call[1][0] for mock_call in mock_bridge.mock_calls)
assert hosts == ['127.0.0.1', '192.168.1.10']
assert len(hass.data[hue.DOMAIN]) == 2
async def test_bridge_discovered(hass, mock_bridge):
"""Bridge discovery."""
assert await async_setup_component(hass, hue.DOMAIN, {})
await hass.helpers.discovery.async_discover(SERVICE_HUE, {
'host': '192.168.1.10',
'serial': '1234567',
})
await hass.async_block_till_done()
assert len(mock_bridge.mock_calls) == 1
assert mock_bridge.mock_calls[0][1][0] == '192.168.1.10'
assert len(hass.data[hue.DOMAIN]) == 1
async def test_bridge_configure_and_discovered(hass, mock_bridge):
"""Bridge is in the config file, then we discover it."""
assert await async_setup_component(hass, hue.DOMAIN, {
hue.DOMAIN: {
hue.CONF_BRIDGES: {
hue.CONF_HOST: '192.168.1.10'
}
}
})
assert len(mock_bridge.mock_calls) == 1
assert mock_bridge.mock_calls[0][1][0] == '192.168.1.10'
assert len(hass.data[hue.DOMAIN]) == 1
mock_bridge.reset_mock()
await hass.helpers.discovery.async_discover(SERVICE_HUE, {
'host': '192.168.1.10',
'serial': '1234567',
})
await hass.async_block_till_done()
assert len(mock_bridge.mock_calls) == 0
assert len(hass.data[hue.DOMAIN]) == 1
async def test_setup_no_host(hass, aioclient_mock):
"""Check we call discovery if domain specified but no bridges."""
aioclient_mock.get(hue.API_NUPNP, json=[])
result = await async_setup_component(
hass, hue.DOMAIN, {hue.DOMAIN: {}})
assert result
assert len(aioclient_mock.mock_calls) == 1
assert len(hass.data[hue.DOMAIN]) == 0

1051
tests/components/light/test_hue.py
View File

File diff suppressed because it is too large Load Diff

588
tests/components/test_hue.py
View File

@ -1,588 +0,0 @@
"""Generic Philips Hue component tests."""
import asyncio
import logging
import unittest
from unittest.mock import call, MagicMock, patch
import aiohue
import pytest
import voluptuous as vol
from homeassistant.components import configurator, hue
from homeassistant.const import CONF_FILENAME, CONF_HOST
from homeassistant.setup import setup_component, async_setup_component
from tests.common import (
assert_setup_component, get_test_home_assistant, get_test_config_dir,
MockDependency, MockConfigEntry, mock_coro
)
_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.assertEqual({}, 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.assertEqual(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.assertEqual(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.assertEqual(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.assertEqual(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.assertEqual(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.assertEqual(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, None)
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, None)
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, None)
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, None)
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, None)
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, None)
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')
async def test_setup_no_host(hass, requests_mock):
"""No host specified in any way."""
requests_mock.get(hue.API_NUPNP, json=[])
with MockDependency('phue') as mock_phue:
result = await async_setup_component(
hass, hue.DOMAIN, {hue.DOMAIN: {}})
assert result
mock_phue.Bridge.assert_not_called()
assert hass.data[hue.DOMAIN] == {}
async def test_flow_works(hass, aioclient_mock):
"""Test config flow ."""
aioclient_mock.get(hue.API_NUPNP, json=[
{'internalipaddress': '1.2.3.4', 'id': 'bla'}
])
flow = hue.HueFlowHandler()
flow.hass = hass
await flow.async_step_init()
with patch('aiohue.Bridge') as mock_bridge:
def mock_constructor(host, websession):
mock_bridge.host = host
return mock_bridge
mock_bridge.side_effect = mock_constructor
mock_bridge.username = 'username-abc'
mock_bridge.config.name = 'Mock Bridge'
mock_bridge.config.bridgeid = 'bridge-id-1234'
mock_bridge.create_user.return_value = mock_coro()
mock_bridge.initialize.return_value = mock_coro()
result = await flow.async_step_link(user_input={})
assert mock_bridge.host == '1.2.3.4'
assert len(mock_bridge.create_user.mock_calls) == 1
assert len(mock_bridge.initialize.mock_calls) == 1
assert result['type'] == 'create_entry'
assert result['title'] == 'Mock Bridge'
assert result['data'] == {
'host': '1.2.3.4',
'bridge_id': 'bridge-id-1234',
'username': 'username-abc'
}
async def test_flow_no_discovered_bridges(hass, aioclient_mock):
"""Test config flow discovers no bridges."""
aioclient_mock.get(hue.API_NUPNP, json=[])
flow = hue.HueFlowHandler()
flow.hass = hass
result = await flow.async_step_init()
assert result['type'] == 'abort'
async def test_flow_all_discovered_bridges_exist(hass, aioclient_mock):
"""Test config flow discovers only already configured bridges."""
aioclient_mock.get(hue.API_NUPNP, json=[
{'internalipaddress': '1.2.3.4', 'id': 'bla'}
])
MockConfigEntry(domain='hue', data={
'host': '1.2.3.4'
}).add_to_hass(hass)
flow = hue.HueFlowHandler()
flow.hass = hass
result = await flow.async_step_init()
assert result['type'] == 'abort'
async def test_flow_one_bridge_discovered(hass, aioclient_mock):
"""Test config flow discovers one bridge."""
aioclient_mock.get(hue.API_NUPNP, json=[
{'internalipaddress': '1.2.3.4', 'id': 'bla'}
])
flow = hue.HueFlowHandler()
flow.hass = hass
result = await flow.async_step_init()
assert result['type'] == 'form'
assert result['step_id'] == 'link'
async def test_flow_two_bridges_discovered(hass, aioclient_mock):
"""Test config flow discovers two bridges."""
aioclient_mock.get(hue.API_NUPNP, json=[
{'internalipaddress': '1.2.3.4', 'id': 'bla'},
{'internalipaddress': '5.6.7.8', 'id': 'beer'}
])
flow = hue.HueFlowHandler()
flow.hass = hass
result = await flow.async_step_init()
assert result['type'] == 'form'
assert result['step_id'] == 'init'
with pytest.raises(vol.Invalid):
assert result['data_schema']({'host': '0.0.0.0'})
result['data_schema']({'host': '1.2.3.4'})
result['data_schema']({'host': '5.6.7.8'})
async def test_flow_two_bridges_discovered_one_new(hass, aioclient_mock):
"""Test config flow discovers two bridges."""
aioclient_mock.get(hue.API_NUPNP, json=[
{'internalipaddress': '1.2.3.4', 'id': 'bla'},
{'internalipaddress': '5.6.7.8', 'id': 'beer'}
])
MockConfigEntry(domain='hue', data={
'host': '1.2.3.4'
}).add_to_hass(hass)
flow = hue.HueFlowHandler()
flow.hass = hass
result = await flow.async_step_init()
assert result['type'] == 'form'
assert result['step_id'] == 'link'
assert flow.host == '5.6.7.8'
async def test_flow_timeout_discovery(hass):
"""Test config flow ."""
flow = hue.HueFlowHandler()
flow.hass = hass
with patch('aiohue.discovery.discover_nupnp',
side_effect=asyncio.TimeoutError):
result = await flow.async_step_init()
assert result['type'] == 'abort'
async def test_flow_link_timeout(hass):
"""Test config flow ."""
flow = hue.HueFlowHandler()
flow.hass = hass
with patch('aiohue.Bridge.create_user',
side_effect=asyncio.TimeoutError):
result = await flow.async_step_link({})
assert result['type'] == 'form'
assert result['step_id'] == 'link'
assert result['errors'] == {
'base': 'register_failed'
}
async def test_flow_link_button_not_pressed(hass):
"""Test config flow ."""
flow = hue.HueFlowHandler()
flow.hass = hass
with patch('aiohue.Bridge.create_user',
side_effect=aiohue.LinkButtonNotPressed):
result = await flow.async_step_link({})
assert result['type'] == 'form'
assert result['step_id'] == 'link'
assert result['errors'] == {
'base': 'register_failed'
}
async def test_flow_link_unknown_host(hass):
"""Test config flow ."""
flow = hue.HueFlowHandler()
flow.hass = hass
with patch('aiohue.Bridge.create_user',
side_effect=aiohue.RequestError):
result = await flow.async_step_link({})
assert result['type'] == 'form'
assert result['step_id'] == 'link'
assert result['errors'] == {
'base': 'register_failed'
}

14
tests/components/zwave/test_init.py
View File

@ -17,7 +17,7 @@ from homeassistant.setup import setup_component
import pytest
from tests.common import (
get_test_home_assistant, async_fire_time_changed)
get_test_home_assistant, async_fire_time_changed, mock_coro)
from tests.mock.zwave import MockNetwork, MockNode, MockValue, MockEntityValues
@ -468,6 +468,7 @@ class TestZWaveDeviceEntityValues(unittest.TestCase):
@patch.object(zwave, 'discovery')
def test_entity_discovery(self, discovery, get_platform):
"""Test the creation of a new entity."""
discovery.async_load_platform.return_value = mock_coro()
mock_platform = MagicMock()
get_platform.return_value = mock_platform
mock_device = MagicMock()
@ -500,8 +501,7 @@ class TestZWaveDeviceEntityValues(unittest.TestCase):
key=lambda a: id(a)))
assert discovery.async_load_platform.called
# Second call is to async yield from
assert len(discovery.async_load_platform.mock_calls) == 2
assert len(discovery.async_load_platform.mock_calls) == 1
args = discovery.async_load_platform.mock_calls[0][1]
assert args[0] == self.hass
assert args[1] == 'mock_component'
@ -532,6 +532,7 @@ class TestZWaveDeviceEntityValues(unittest.TestCase):
@patch.object(zwave, 'discovery')
def test_entity_existing_values(self, discovery, get_platform):
"""Test the loading of already discovered values."""
discovery.async_load_platform.return_value = mock_coro()
mock_platform = MagicMock()
get_platform.return_value = mock_platform
mock_device = MagicMock()
@ -563,8 +564,7 @@ class TestZWaveDeviceEntityValues(unittest.TestCase):
key=lambda a: id(a)))
assert discovery.async_load_platform.called
# Second call is to async yield from
assert len(discovery.async_load_platform.mock_calls) == 2
assert len(discovery.async_load_platform.mock_calls) == 1
args = discovery.async_load_platform.mock_calls[0][1]
assert args[0] == self.hass
assert args[1] == 'mock_component'
@ -599,6 +599,7 @@ class TestZWaveDeviceEntityValues(unittest.TestCase):
@patch.object(zwave, 'discovery')
def test_entity_workaround_component(self, discovery, get_platform):
"""Test ignore workaround."""
discovery.async_load_platform.return_value = mock_coro()
mock_platform = MagicMock()
get_platform.return_value = mock_platform
mock_device = MagicMock()
@ -629,8 +630,7 @@ class TestZWaveDeviceEntityValues(unittest.TestCase):
self.hass.block_till_done()
assert discovery.async_load_platform.called
# Second call is to async yield from
assert len(discovery.async_load_platform.mock_calls) == 2
assert len(discovery.async_load_platform.mock_calls) == 1
args = discovery.async_load_platform.mock_calls[0][1]
assert args[1] == 'binary_sensor'

14
tests/helpers/test_discovery.py
View File

@ -1,5 +1,4 @@
"""Test discovery helpers."""
import asyncio
from unittest.mock import patch
import pytest
@ -24,7 +23,8 @@ class TestHelpersDiscovery:
"""Stop everything that was started."""
self.hass.stop()
@patch('homeassistant.setup.async_setup_component')
@patch('homeassistant.setup.async_setup_component',
return_value=mock_coro())
def test_listen(self, mock_setup_component):
"""Test discovery listen/discover combo."""
helpers = self.hass.helpers
@ -199,15 +199,13 @@ class TestHelpersDiscovery:
assert len(component_calls) == 1
@asyncio.coroutine
def test_load_platform_forbids_config():
async def test_load_platform_forbids_config():
"""Test you cannot setup config component with load_platform."""
with pytest.raises(HomeAssistantError):
yield from discovery.async_load_platform(None, 'config', 'zwave')
await discovery.async_load_platform(None, 'config', 'zwave')
@asyncio.coroutine
def test_discover_forbids_config():
async def test_discover_forbids_config():
"""Test you cannot setup config component with load_platform."""
with pytest.raises(HomeAssistantError):
yield from discovery.async_discover(None, None, None, 'config')
await discovery.async_discover(None, None, None, 'config')

8
tests/util/test_init.py
View File

@ -291,3 +291,11 @@ async def test_throttle_async():
assert (await test_method()) is True
assert (await test_method()) is None
@util.Throttle(timedelta(seconds=2), timedelta(seconds=0.1))
async def test_method2():
"""Only first call should return a value."""
return True
assert (await test_method2()) is True
assert (await test_method2()) is None