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

Nest config flow (#14921)

Browse source 
* Move nest to dir based component

* Add config flow for Nest

* Load Nest platforms via config entry

* Add tests for Nest config flow

* Import existing access tokens as config entries

* Lint

* Update coverage

* Update translation

* Fix tests

* Address strings

* Use python-nest token resolution

* Lint

* Do not do I/O inside constructor

* Lint

* Update test requirements
This commit is contained in:
Paulus Schoutsen 2018-06-13 11:14:52 -04:00 • committed by GitHub
parent d549e26a9b
commit e014a84215
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
21 changed files with 666 additions and 153 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
.coveragerc
View file

@ -195,7 +195,7 @@ omit =
homeassistant/components/neato.py
homeassistant/components/*/neato.py
homeassistant/components/nest.py
homeassistant/components/nest/__init__.py
homeassistant/components/*/nest.py
homeassistant/components/netatmo.py

69
homeassistant/components/binary_sensor/nest.py
View file

@ -8,7 +8,8 @@ from itertools import chain
import logging
from homeassistant.components.binary_sensor import BinarySensorDevice
from homeassistant.components.nest import DATA_NEST, NestSensorDevice
from homeassistant.components.nest import (
DATA_NEST, DATA_NEST_CONFIG, CONF_BINARY_SENSORS, NestSensorDevice)
from homeassistant.const import CONF_MONITORED_CONDITIONS
DEPENDENCIES = ['nest']
@ -56,12 +57,19 @@ _LOGGER = logging.getLogger(__name__)
def setup_platform(hass, config, add_devices, discovery_info=None):
"""Set up the Nest binary sensors."""
if discovery_info is None:
return
"""Set up the Nest binary sensors.
No longer used.
"""
async def async_setup_entry(hass, entry, async_add_devices):
"""Set up a Nest binary sensor based on a config entry."""
nest = hass.data[DATA_NEST]
discovery_info = \
hass.data.get(DATA_NEST_CONFIG, {}).get(CONF_BINARY_SENSORS, {})
# Add all available binary sensors if no Nest binary sensor config is set
if discovery_info == {}:
conditions = _VALID_BINARY_SENSOR_TYPES
@ -76,32 +84,37 @@ def setup_platform(hass, config, add_devices, discovery_info=None):
"for valid options.")
_LOGGER.error(wstr)
sensors = []
for structure in nest.structures():
sensors += [NestBinarySensor(structure, None, variable)
for variable in conditions
if variable in STRUCTURE_BINARY_TYPES]
device_chain = chain(nest.thermostats(),
nest.smoke_co_alarms(),
nest.cameras())
for structure, device in device_chain:
sensors += [NestBinarySensor(structure, device, variable)
for variable in conditions
if variable in BINARY_TYPES]
sensors += [NestBinarySensor(structure, device, variable)
for variable in conditions
if variable in CLIMATE_BINARY_TYPES
and device.is_thermostat]
if device.is_camera:
def get_binary_sensors():
"""Get the Nest binary sensors."""
sensors = []
for structure in nest.structures():
sensors += [NestBinarySensor(structure, None, variable)
for variable in conditions
if variable in STRUCTURE_BINARY_TYPES]
device_chain = chain(nest.thermostats(),
nest.smoke_co_alarms(),
nest.cameras())
for structure, device in device_chain:
sensors += [NestBinarySensor(structure, device, variable)
for variable in conditions
if variable in CAMERA_BINARY_TYPES]
for activity_zone in device.activity_zones:
sensors += [NestActivityZoneSensor(structure,
device,
activity_zone)]
add_devices(sensors, True)
if variable in BINARY_TYPES]
sensors += [NestBinarySensor(structure, device, variable)
for variable in conditions
if variable in CLIMATE_BINARY_TYPES
and device.is_thermostat]
if device.is_camera:
sensors += [NestBinarySensor(structure, device, variable)
for variable in conditions
if variable in CAMERA_BINARY_TYPES]
for activity_zone in device.activity_zones:
sensors += [NestActivityZoneSensor(structure,
device,
activity_zone)]
return sensors
async_add_devices(await hass.async_add_job(get_binary_sensors), True)
class NestBinarySensor(NestSensorDevice, BinarySensorDevice):

10
homeassistant/components/camera/__init__.py
View file

@ -216,6 +216,16 @@ def async_setup(hass, config):
return True
async def async_setup_entry(hass, entry):
"""Setup a config entry."""
return await hass.data[DOMAIN].async_setup_entry(entry)
async def async_unload_entry(hass, entry):
"""Unload a config entry."""
return await hass.data[DOMAIN].async_unload_entry(entry)
class Camera(Entity):
"""The base class for camera entities."""

15
homeassistant/components/camera/nest.py
View file

@ -23,14 +23,19 @@ PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({})
def setup_platform(hass, config, add_devices, discovery_info=None):
"""Set up a Nest Cam."""
if discovery_info is None:
return
"""Set up a Nest Cam.
camera_devices = hass.data[nest.DATA_NEST].cameras()
No longer in use.
"""
async def async_setup_entry(hass, entry, async_add_devices):
"""Set up a Nest sensor based on a config entry."""
camera_devices = \
await hass.async_add_job(hass.data[nest.DATA_NEST].cameras)
cameras = [NestCamera(structure, device)
for structure, device in camera_devices]
add_devices(cameras, True)
async_add_devices(cameras, True)
class NestCamera(Camera):

13
homeassistant/components/climate/__init__.py
View file

@ -246,7 +246,8 @@ def set_swing_mode(hass, swing_mode, entity_id=None):
async def async_setup(hass, config):
"""Set up climate devices."""
component = EntityComponent(_LOGGER, DOMAIN, hass, SCAN_INTERVAL)
component = hass.data[DOMAIN] = \
EntityComponent(_LOGGER, DOMAIN, hass, SCAN_INTERVAL)
await component.async_setup(config)
async def async_away_mode_set_service(service):
@ -456,6 +457,16 @@ async def async_setup(hass, config):
return True
async def async_setup_entry(hass, entry):
"""Setup a config entry."""
return await hass.data[DOMAIN].async_setup_entry(entry)
async def async_unload_entry(hass, entry):
"""Unload a config entry."""
return await hass.data[DOMAIN].async_unload_entry(entry)
class ClimateDevice(Entity):
"""Representation of a climate device."""

18
homeassistant/components/climate/nest.py
View file

@ -32,16 +32,22 @@ NEST_MODE_HEAT_COOL = 'heat-cool'
def setup_platform(hass, config, add_devices, discovery_info=None):
"""Set up the Nest thermostat."""
if discovery_info is None:
return
"""Set up the Nest thermostat.
No longer in use.
"""
async def async_setup_entry(hass, entry, async_add_devices):
"""Set up the Nest climate device based on a config entry."""
temp_unit = hass.config.units.temperature_unit
all_devices = [NestThermostat(structure, device, temp_unit)
for structure, device in hass.data[DATA_NEST].thermostats()]
thermostats = await hass.async_add_job(hass.data[DATA_NEST].thermostats)
add_devices(all_devices, True)
all_devices = [NestThermostat(structure, device, temp_unit)
for structure, device in thermostats]
async_add_devices(all_devices, True)
class NestThermostat(ClimateDevice):

33
homeassistant/components/nest/.translations/en.json Normal file
View file

@ -0,0 +1,33 @@
{
"config": {
"abort": {
"already_setup": "You can only configure a single Nest account.",
"authorize_url_fail": "Unknown error generating an authorize url.",
"authorize_url_timeout": "Timeout generating authorize url.",
"no_flows": "You need to configure Nest before being able to authenticate with it. [Please read the instructions](https://www.home-assistant.io/components/nest/)."
},
"error": {
"internal_error": "Internal error validating code",
"invalid_code": "Invalid code",
"timeout": "Timeout validating code",
"unknown": "Unknown error validating code"
},
"step": {
"init": {
"data": {
"flow_impl": "Provider"
},
"description": "Pick via which authentication provider you want to authenticate with Nest.",
"title": "Authentication Provider"
},
"link": {
"data": {
"code": "Pin code"
},
"description": "To link your Nest account, [authorize your account]({url}).\n\nAfter authorization, copy-paste the provided pin code below.",
"title": "Link Nest Account"
}
},
"title": "Nest"
}
}

130
homeassistant/components/nest.py → homeassistant/components/nest/__init__.py
View file

@ -6,6 +6,7 @@ https://home-assistant.io/components/nest/
"""
from concurrent.futures import ThreadPoolExecutor
import logging
import os.path
import socket
from datetime import datetime, timedelta
@ -15,19 +16,22 @@ from homeassistant.const import (
CONF_STRUCTURE, CONF_FILENAME, CONF_BINARY_SENSORS, CONF_SENSORS,
CONF_MONITORED_CONDITIONS,
EVENT_HOMEASSISTANT_START, EVENT_HOMEASSISTANT_STOP)
from homeassistant.helpers import discovery, config_validation as cv
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.dispatcher import async_dispatcher_send, \
async_dispatcher_connect
from homeassistant.helpers.entity import Entity
from .const import DOMAIN
from . import local_auth
REQUIREMENTS = ['python-nest==4.0.2']
_CONFIGURING = {}
_LOGGER = logging.getLogger(__name__)
DOMAIN = 'nest'
DATA_NEST = 'nest'
DATA_NEST_CONFIG = 'nest_config'
SIGNAL_NEST_UPDATE = 'nest_update'
@ -86,76 +90,45 @@ async def async_nest_update_event_broker(hass, nest):
return
async def async_request_configuration(nest, hass, config):
"""Request configuration steps from the user."""
configurator = hass.components.configurator
if 'nest' in _CONFIGURING:
_LOGGER.debug("configurator failed")
configurator.async_notify_errors(
_CONFIGURING['nest'], "Failed to configure, please try again.")
async def async_setup(hass, config):
"""Set up Nest components."""
if DOMAIN not in config:
return
async def async_nest_config_callback(data):
"""Run when the configuration callback is called."""
_LOGGER.debug("configurator callback")
pin = data.get('pin')
if await async_setup_nest(hass, nest, config, pin=pin):
# start nest update event listener as we missed startup hook
hass.async_add_job(async_nest_update_event_broker, hass, nest)
conf = config[DOMAIN]
_CONFIGURING['nest'] = configurator.async_request_config(
"Nest", async_nest_config_callback,
description=('To configure Nest, click Request Authorization below, '
'log into your Nest account, '
'and then enter the resulting PIN'),
link_name='Request Authorization',
link_url=nest.authorize_url,
submit_caption="Confirm",
fields=[{'id': 'pin', 'name': 'Enter the PIN', 'type': ''}]
)
local_auth.initialize(hass, conf[CONF_CLIENT_ID], conf[CONF_CLIENT_SECRET])
filename = config.get(CONF_FILENAME, NEST_CONFIG_FILE)
access_token_cache_file = hass.config.path(filename)
if await hass.async_add_job(os.path.isfile, access_token_cache_file):
hass.async_add_job(hass.config_entries.flow.async_init(
DOMAIN, source='import', data={
'nest_conf_path': access_token_cache_file,
}
))
# Store config to be used during entry setup
hass.data[DATA_NEST_CONFIG] = conf
return True
async def async_setup_nest(hass, nest, config, pin=None):
"""Set up the Nest devices."""
from nest.nest import AuthorizationError, APIError
if pin is not None:
_LOGGER.debug("pin acquired, requesting access token")
error_message = None
try:
nest.request_token(pin)
except AuthorizationError as auth_error:
error_message = "Nest authorization failed: {}".format(auth_error)
except APIError as api_error:
error_message = "Failed to call Nest API: {}".format(api_error)
async def async_setup_entry(hass, entry):
"""Setup Nest from a config entry."""
from nest import Nest
if error_message is not None:
_LOGGER.warning(error_message)
hass.components.configurator.async_notify_errors(
_CONFIGURING['nest'], error_message)
return False
if nest.access_token is None:
_LOGGER.debug("no access_token, requesting configuration")
await async_request_configuration(nest, hass, config)
return False
if 'nest' in _CONFIGURING:
_LOGGER.debug("configuration done")
configurator = hass.components.configurator
configurator.async_request_done(_CONFIGURING.pop('nest'))
nest = Nest(access_token=entry.data['tokens']['access_token'])
_LOGGER.debug("proceeding with setup")
conf = config[DOMAIN]
conf = hass.data.get(DATA_NEST_CONFIG, {})
hass.data[DATA_NEST] = NestDevice(hass, conf, nest)
await hass.async_add_job(hass.data[DATA_NEST].initialize)
for component, discovered in [
('climate', {}),
('camera', {}),
('sensor', conf.get(CONF_SENSORS, {})),
('binary_sensor', conf.get(CONF_BINARY_SENSORS, {}))]:
_LOGGER.debug("proceeding with discovery -- %s", component)
hass.async_add_job(discovery.async_load_platform,
hass, component, DOMAIN, discovered, config)
for component in 'climate', 'camera', 'sensor', 'binary_sensor':
hass.async_add_job(hass.config_entries.async_forward_entry_setup(
entry, component))
def set_mode(service):
"""
@ -210,29 +183,6 @@ async def async_setup_nest(hass, nest, config, pin=None):
return True
async def async_setup(hass, config):
"""Set up Nest components."""
from nest import Nest
if 'nest' in _CONFIGURING:
return
conf = config[DOMAIN]
client_id = conf[CONF_CLIENT_ID]
client_secret = conf[CONF_CLIENT_SECRET]
filename = config.get(CONF_FILENAME, NEST_CONFIG_FILE)
access_token_cache_file = hass.config.path(filename)
nest = Nest(
access_token_cache_file=access_token_cache_file,
client_id=client_id, client_secret=client_secret)
await async_setup_nest(hass, nest, config)
return True
class NestDevice(object):
"""Structure Nest functions for hass."""
@ -240,12 +190,12 @@ class NestDevice(object):
"""Init Nest Devices."""
self.hass = hass
self.nest = nest
self.local_structure = conf.get(CONF_STRUCTURE)
if CONF_STRUCTURE not in conf:
self.local_structure = [s.name for s in nest.structures]
else:
self.local_structure = conf[CONF_STRUCTURE]
_LOGGER.debug("Structures to include: %s", self.local_structure)
def initialize(self):
"""Initialize Nest."""
if self.local_structure is None:
self.local_structure = [s.name for s in self.nest.structures]
def structures(self):
"""Generate a list of structures."""

154
homeassistant/components/nest/config_flow.py Normal file
View file

@ -0,0 +1,154 @@
"""Config flow to configure Nest."""
import asyncio
from collections import OrderedDict
import logging
import async_timeout
import voluptuous as vol
from homeassistant import config_entries, data_entry_flow
from homeassistant.core import callback
from homeassistant.exceptions import HomeAssistantError
from homeassistant.util.json import load_json
from .const import DOMAIN
DATA_FLOW_IMPL = 'nest_flow_implementation'
_LOGGER = logging.getLogger(__name__)
@callback
def register_flow_implementation(hass, domain, name, gen_authorize_url,
convert_code):
"""Register a flow implementation.
domain: Domain of the component responsible for the implementation.
name: Name of the component.
gen_authorize_url: Coroutine function to generate the authorize url.
convert_code: Coroutine function to convert a code to an access token.
"""
if DATA_FLOW_IMPL not in hass.data:
hass.data[DATA_FLOW_IMPL] = OrderedDict()
hass.data[DATA_FLOW_IMPL][domain] = {
'domain': domain,
'name': name,
'gen_authorize_url': gen_authorize_url,
'convert_code': convert_code,
}
class NestAuthError(HomeAssistantError):
"""Base class for Nest auth errors."""
class CodeInvalid(NestAuthError):
"""Raised when invalid authorization code."""
@config_entries.HANDLERS.register(DOMAIN)
class NestFlowHandler(data_entry_flow.FlowHandler):
"""Handle a Nest config flow."""
VERSION = 1
def __init__(self):
"""Initialize the Nest config flow."""
self.flow_impl = None
async def async_step_init(self, user_input=None):
"""Handle a flow start."""
flows = self.hass.data.get(DATA_FLOW_IMPL, {})
if self.hass.config_entries.async_entries(DOMAIN):
return self.async_abort(reason='already_setup')
elif not flows:
return self.async_abort(reason='no_flows')
elif len(flows) == 1:
self.flow_impl = list(flows)[0]
return await self.async_step_link()
elif user_input is not None:
self.flow_impl = user_input['flow_impl']
return await self.async_step_link()
return self.async_show_form(
step_id='init',
data_schema=vol.Schema({
vol.Required('flow_impl'): vol.In(list(flows))
})
)
async def async_step_link(self, user_input=None):
"""Attempt to link with the Nest account.
Route the user to a website to authenticate with Nest. Depending on
implementation type we expect a pin or an external component to
deliver the authentication code.
"""
flow = self.hass.data[DATA_FLOW_IMPL][self.flow_impl]
errors = {}
if user_input is not None:
try:
with async_timeout.timeout(10):
tokens = await flow['convert_code'](user_input['code'])
return self._entry_from_tokens(
'Nest (via {})'.format(flow['name']), flow, tokens)
except asyncio.TimeoutError:
errors['code'] = 'timeout'
except CodeInvalid:
errors['code'] = 'invalid_code'
except NestAuthError:
errors['code'] = 'unknown'
except Exception: # pylint: disable=broad-except
errors['code'] = 'internal_error'
_LOGGER.exception("Unexpected error resolving code")
try:
with async_timeout.timeout(10):
url = await flow['gen_authorize_url'](self.flow_id)
except asyncio.TimeoutError:
return self.async_abort(reason='authorize_url_timeout')
except Exception: # pylint: disable=broad-except
_LOGGER.exception("Unexpected error generating auth url")
return self.async_abort(reason='authorize_url_fail')
return self.async_show_form(
step_id='link',
description_placeholders={
'url': url
},
data_schema=vol.Schema({
vol.Required('code'): str,
}),
errors=errors,
)
async def async_step_import(self, info):
"""Import existing auth from Nest."""
if self.hass.config_entries.async_entries(DOMAIN):
return self.async_abort(reason='already_setup')
flow = self.hass.data[DATA_FLOW_IMPL][DOMAIN]
tokens = await self.hass.async_add_job(
load_json, info['nest_conf_path'])
return self._entry_from_tokens(
'Nest (import from configuration.yaml)', flow, tokens)
@callback
def _entry_from_tokens(self, title, flow, tokens):
"""Create an entry from tokens."""
return self.async_create_entry(
title=title,
data={
'tokens': tokens,
'impl_domain': flow['domain'],
},
)

2
homeassistant/components/nest/const.py Normal file
View file

@ -0,0 +1,2 @@
"""Constants used by the Nest component."""
DOMAIN = 'nest'

45
homeassistant/components/nest/local_auth.py Normal file
View file

@ -0,0 +1,45 @@
"""Local Nest authentication."""
import asyncio
from functools import partial
from homeassistant.core import callback
from . import config_flow
from .const import DOMAIN
@callback
def initialize(hass, client_id, client_secret):
"""Initialize a local auth provider."""
config_flow.register_flow_implementation(
hass, DOMAIN, 'local', partial(generate_auth_url, client_id),
partial(resolve_auth_code, hass, client_id, client_secret)
)
async def generate_auth_url(client_id, flow_id):
"""Generate an authorize url."""
from nest.nest import AUTHORIZE_URL
return AUTHORIZE_URL.format(client_id, flow_id)
async def resolve_auth_code(hass, client_id, client_secret, code):
"""Resolve an authorization code."""
from nest.nest import NestAuth, AuthorizationError
result = asyncio.Future()
auth = NestAuth(
client_id=client_id,
client_secret=client_secret,
auth_callback=result.set_result,
)
auth.pin = code
try:
await hass.async_add_job(auth.login)
return await result
except AuthorizationError as err:
if err.response.status_code == 401:
raise config_flow.CodeInvalid()
else:
raise config_flow.NestAuthError('Unknown error: {} ({})'.format(
err, err.response.status_code))

33
homeassistant/components/nest/strings.json Normal file
View file

@ -0,0 +1,33 @@
{
"config": {
"title": "Nest",
"step": {
"init": {
"title": "Authentication Provider",
"description": "Pick via which authentication provider you want to authenticate with Nest.",
"data": {
"flow_impl": "Provider"
}
},
"link": {
"title": "Link Nest Account",
"description": "To link your Nest account, [authorize your account]({url}).\n\nAfter authorization, copy-paste the provided pin code below.",
"data": {
"code": "Pin code"
}
}
},
"error": {
"timeout": "Timeout validating code",
"invalid_code": "Invalid code",
"unknown": "Unknown error validating code",
"internal_error": "Internal error validating code"
},
"abort": {
"already_setup": "You can only configure a single Nest account.",
"no_flows": "You need to configure Nest before being able to authenticate with it. [Please read the instructions](https://www.home-assistant.io/components/nest/).",
"authorize_url_timeout": "Timeout generating authorize url.",
"authorize_url_fail": "Unknown error generating an authorize url."
}
}
}

53
homeassistant/components/sensor/nest.py
View file

@ -6,7 +6,8 @@ https://home-assistant.io/components/sensor.nest/
"""
import logging
from homeassistant.components.nest import DATA_NEST, NestSensorDevice
from homeassistant.components.nest import (
DATA_NEST, DATA_NEST_CONFIG, CONF_SENSORS, NestSensorDevice)
from homeassistant.const import (
TEMP_CELSIUS, TEMP_FAHRENHEIT, CONF_MONITORED_CONDITIONS,
DEVICE_CLASS_TEMPERATURE, DEVICE_CLASS_HUMIDITY)
@ -51,12 +52,18 @@ _LOGGER = logging.getLogger(__name__)
def setup_platform(hass, config, add_devices, discovery_info=None):
"""Set up the Nest Sensor."""
if discovery_info is None:
return
"""Set up the Nest Sensor.
No longer used.
"""
async def async_setup_entry(hass, entry, async_add_devices):
"""Set up a Nest sensor based on a config entry."""
nest = hass.data[DATA_NEST]
discovery_info = hass.data.get(DATA_NEST_CONFIG, {}).get(CONF_SENSORS, {})
# Add all available sensors if no Nest sensor config is set
if discovery_info == {}:
conditions = _VALID_SENSOR_TYPES
@ -77,26 +84,30 @@ def setup_platform(hass, config, add_devices, discovery_info=None):
"binary_sensor.nest/ for valid options.")
_LOGGER.error(wstr)
all_sensors = []
for structure in nest.structures():
all_sensors += [NestBasicSensor(structure, None, variable)
for variable in conditions
if variable in STRUCTURE_SENSOR_TYPES]
def get_sensors():
"""Get the Nest sensors."""
all_sensors = []
for structure in nest.structures():
all_sensors += [NestBasicSensor(structure, None, variable)
for variable in conditions
if variable in STRUCTURE_SENSOR_TYPES]
for structure, device in nest.thermostats():
all_sensors += [NestBasicSensor(structure, device, variable)
for variable in conditions
if variable in SENSOR_TYPES]
all_sensors += [NestTempSensor(structure, device, variable)
for variable in conditions
if variable in TEMP_SENSOR_TYPES]
for structure, device in nest.thermostats():
all_sensors += [NestBasicSensor(structure, device, variable)
for variable in conditions
if variable in SENSOR_TYPES]
all_sensors += [NestTempSensor(structure, device, variable)
for variable in conditions
if variable in TEMP_SENSOR_TYPES]
for structure, device in nest.smoke_co_alarms():
all_sensors += [NestBasicSensor(structure, device, variable)
for variable in conditions
if variable in PROTECT_SENSOR_TYPES]
for structure, device in nest.smoke_co_alarms():
all_sensors += [NestBasicSensor(structure, device, variable)
for variable in conditions
if variable in PROTECT_SENSOR_TYPES]
add_devices(all_sensors, True)
return all_sensors
async_add_devices(await hass.async_add_job(get_sensors), True)
class NestBasicSensor(NestSensorDevice):

1
homeassistant/config_entries.py
View file

@ -129,6 +129,7 @@ HANDLERS = Registry()
FLOWS = [
'deconz',
'hue',
'nest',
'zone',
]

4
homeassistant/data_entry_flow.py
View file

@ -132,7 +132,8 @@ class FlowHandler:
VERSION = 1
@callback
def async_show_form(self, *, step_id, data_schema=None, errors=None):
def async_show_form(self, *, step_id, data_schema=None, errors=None,
description_placeholders=None):
"""Return the definition of a form to gather user input."""
return {
'type': RESULT_TYPE_FORM,
@ -141,6 +142,7 @@ class FlowHandler:
'step_id': step_id,
'data_schema': data_schema,
'errors': errors,
'description_placeholders': description_placeholders,
}
@callback

3
requirements_test_all.txt
View file

@ -155,6 +155,9 @@ pyqwikswitch==0.8
# homeassistant.components.weather.darksky
python-forecastio==1.4.0
# homeassistant.components.nest
python-nest==4.0.2
# homeassistant.components.sensor.whois
pythonwhois==2.4.3

1
script/gen_requirements_all.py
View file

@ -77,6 +77,7 @@ TEST_REQUIREMENTS = (
'pynx584',
'pyqwikswitch',
'python-forecastio',
'python-nest',
'pytradfri\[async\]',
'pyunifi',
'pyupnp-async',

7
tests/components/config/test_config_entries.py
View file

@ -110,6 +110,9 @@ def test_initialize_flow(hass, client):
return self.async_show_form(
step_id='init',
data_schema=schema,
description_placeholders={
'url': 'https://example.com',
},
errors={
'username': 'Should be unique.'
}
@ -140,6 +143,9 @@ def test_initialize_flow(hass, client):
'type': 'string'
}
],
'description_placeholders': {
'url': 'https://example.com',
},
'errors': {
'username': 'Should be unique.'
}
@ -242,6 +248,7 @@ def test_two_step_flow(hass, client):
'type': 'string'
}
],
'description_placeholders': None,
'errors': None
}

1
tests/components/nest/__init__.py Normal file
View file

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

174
tests/components/nest/test_config_flow.py Normal file
View file

@ -0,0 +1,174 @@
"""Tests for the Nest config flow."""
import asyncio
from unittest.mock import Mock, patch
from homeassistant import data_entry_flow
from homeassistant.components.nest import config_flow
from tests.common import mock_coro
async def test_abort_if_no_implementation_registered(hass):
"""Test we abort if no implementation is registered."""
flow = config_flow.NestFlowHandler()
flow.hass = hass
result = await flow.async_step_init()
assert result['type'] == data_entry_flow.RESULT_TYPE_ABORT
assert result['reason'] == 'no_flows'
async def test_abort_if_already_setup(hass):
"""Test we abort if Nest is already setup."""
flow = config_flow.NestFlowHandler()
flow.hass = hass
with patch.object(hass.config_entries, 'async_entries', return_value=[{}]):
result = await flow.async_step_init()
assert result['type'] == data_entry_flow.RESULT_TYPE_ABORT
assert result['reason'] == 'already_setup'
async def test_full_flow_implementation(hass):
"""Test registering an implementation and finishing flow works."""
gen_authorize_url = Mock(return_value=mock_coro('https://example.com'))
convert_code = Mock(return_value=mock_coro({'access_token': 'yoo'}))
config_flow.register_flow_implementation(
hass, 'test', 'Test', gen_authorize_url, convert_code)
config_flow.register_flow_implementation(
hass, 'test-other', 'Test Other', None, None)
flow = config_flow.NestFlowHandler()
flow.hass = hass
result = await flow.async_step_init()
assert result['type'] == data_entry_flow.RESULT_TYPE_FORM
assert result['step_id'] == 'init'
result = await flow.async_step_init({'flow_impl': 'test'})
assert result['type'] == data_entry_flow.RESULT_TYPE_FORM
assert result['step_id'] == 'link'
assert result['description_placeholders'] == {
'url': 'https://example.com',
}
result = await flow.async_step_link({'code': '123ABC'})
assert result['type'] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result['data']['tokens'] == {'access_token': 'yoo'}
assert result['data']['impl_domain'] == 'test'
assert result['title'] == 'Nest (via Test)'
async def test_not_pick_implementation_if_only_one(hass):
"""Test we allow picking implementation if we have two."""
gen_authorize_url = Mock(return_value=mock_coro('https://example.com'))
config_flow.register_flow_implementation(
hass, 'test', 'Test', gen_authorize_url, None)
flow = config_flow.NestFlowHandler()
flow.hass = hass
result = await flow.async_step_init()
assert result['type'] == data_entry_flow.RESULT_TYPE_FORM
assert result['step_id'] == 'link'
async def test_abort_if_timeout_generating_auth_url(hass):
"""Test we abort if generating authorize url fails."""
gen_authorize_url = Mock(side_effect=asyncio.TimeoutError)
config_flow.register_flow_implementation(
hass, 'test', 'Test', gen_authorize_url, None)
flow = config_flow.NestFlowHandler()
flow.hass = hass
result = await flow.async_step_init()
assert result['type'] == data_entry_flow.RESULT_TYPE_ABORT
assert result['reason'] == 'authorize_url_timeout'
async def test_abort_if_exception_generating_auth_url(hass):
"""Test we abort if generating authorize url blows up."""
gen_authorize_url = Mock(side_effect=ValueError)
config_flow.register_flow_implementation(
hass, 'test', 'Test', gen_authorize_url, None)
flow = config_flow.NestFlowHandler()
flow.hass = hass
result = await flow.async_step_init()
assert result['type'] == data_entry_flow.RESULT_TYPE_ABORT
assert result['reason'] == 'authorize_url_fail'
async def test_verify_code_timeout(hass):
"""Test verify code timing out."""
gen_authorize_url = Mock(return_value=mock_coro('https://example.com'))
convert_code = Mock(side_effect=asyncio.TimeoutError)
config_flow.register_flow_implementation(
hass, 'test', 'Test', gen_authorize_url, convert_code)
flow = config_flow.NestFlowHandler()
flow.hass = hass
result = await flow.async_step_init()
assert result['type'] == data_entry_flow.RESULT_TYPE_FORM
assert result['step_id'] == 'link'
result = await flow.async_step_link({'code': '123ABC'})
assert result['type'] == data_entry_flow.RESULT_TYPE_FORM
assert result['step_id'] == 'link'
assert result['errors'] == {'code': 'timeout'}
async def test_verify_code_invalid(hass):
"""Test verify code invalid."""
gen_authorize_url = Mock(return_value=mock_coro('https://example.com'))
convert_code = Mock(side_effect=config_flow.CodeInvalid)
config_flow.register_flow_implementation(
hass, 'test', 'Test', gen_authorize_url, convert_code)
flow = config_flow.NestFlowHandler()
flow.hass = hass
result = await flow.async_step_init()
assert result['type'] == data_entry_flow.RESULT_TYPE_FORM
assert result['step_id'] == 'link'
result = await flow.async_step_link({'code': '123ABC'})
assert result['type'] == data_entry_flow.RESULT_TYPE_FORM
assert result['step_id'] == 'link'
assert result['errors'] == {'code': 'invalid_code'}
async def test_verify_code_unknown_error(hass):
"""Test verify code unknown error."""
gen_authorize_url = Mock(return_value=mock_coro('https://example.com'))
convert_code = Mock(side_effect=config_flow.NestAuthError)
config_flow.register_flow_implementation(
hass, 'test', 'Test', gen_authorize_url, convert_code)
flow = config_flow.NestFlowHandler()
flow.hass = hass
result = await flow.async_step_init()
assert result['type'] == data_entry_flow.RESULT_TYPE_FORM
assert result['step_id'] == 'link'
result = await flow.async_step_link({'code': '123ABC'})
assert result['type'] == data_entry_flow.RESULT_TYPE_FORM
assert result['step_id'] == 'link'
assert result['errors'] == {'code': 'unknown'}
async def test_verify_code_exception(hass):
"""Test verify code blows up."""
gen_authorize_url = Mock(return_value=mock_coro('https://example.com'))
convert_code = Mock(side_effect=ValueError)
config_flow.register_flow_implementation(
hass, 'test', 'Test', gen_authorize_url, convert_code)
flow = config_flow.NestFlowHandler()
flow.hass = hass
result = await flow.async_step_init()
assert result['type'] == data_entry_flow.RESULT_TYPE_FORM
assert result['step_id'] == 'link'
result = await flow.async_step_link({'code': '123ABC'})
assert result['type'] == data_entry_flow.RESULT_TYPE_FORM
assert result['step_id'] == 'link'
assert result['errors'] == {'code': 'internal_error'}

51
tests/components/nest/test_local_auth.py Normal file
View file

@ -0,0 +1,51 @@
"""Test Nest local auth."""
from homeassistant.components.nest import const, config_flow, local_auth
from urllib.parse import parse_qsl
import pytest
import requests_mock as rmock
@pytest.fixture
def registered_flow(hass):
"""Mock a registered flow."""
local_auth.initialize(hass, 'TEST-CLIENT-ID', 'TEST-CLIENT-SECRET')
return hass.data[config_flow.DATA_FLOW_IMPL][const.DOMAIN]
async def test_generate_auth_url(registered_flow):
"""Test generating an auth url.
Mainly testing that it doesn't blow up.
"""
url = await registered_flow['gen_authorize_url']('TEST-FLOW-ID')
assert url is not None
async def test_convert_code(requests_mock, registered_flow):
"""Test converting a code."""
from nest.nest import ACCESS_TOKEN_URL
def token_matcher(request):
"""Match a fetch token request."""
if request.url != ACCESS_TOKEN_URL:
return None
assert dict(parse_qsl(request.text)) == {
'client_id': 'TEST-CLIENT-ID',
'client_secret': 'TEST-CLIENT-SECRET',
'code': 'TEST-CODE',
'grant_type': 'authorization_code'
}
return rmock.create_response(request, json={
'access_token': 'TEST-ACCESS-TOKEN'
})
requests_mock.add_matcher(token_matcher)
tokens = await registered_flow['convert_code']('TEST-CODE')
assert tokens == {
'access_token': 'TEST-ACCESS-TOKEN'
}