TelldusLive config flow (#18758)

* update TelldusLive to use config flow

* fixes from Martin

* Update homeassistant/components/tellduslive/config_flow.py

Co-Authored-By: fredrike <fredrik.e@gmail.com>

* revert changes in entry.py

* tox tests

* tox fixes

* woof woof (fix for hound)

* lint ignore

* unload entry

* coverall toxtests

* fix some toxtests
This commit is contained in:
Fredrik Erlandsson 2018-12-10 18:44:45 +01:00 committed by Martin Hjelmare
parent f4f42176bd
commit 92e19f6001
14 changed files with 685 additions and 235 deletions

View File

@ -9,22 +9,35 @@ https://home-assistant.io/components/binary_sensor.tellduslive/
""" """
import logging import logging
from homeassistant.components import tellduslive from homeassistant.components import binary_sensor, tellduslive
from homeassistant.components.binary_sensor import BinarySensorDevice from homeassistant.components.binary_sensor import BinarySensorDevice
from homeassistant.components.tellduslive.entry import TelldusLiveEntity from homeassistant.components.tellduslive.entry import TelldusLiveEntity
from homeassistant.helpers.dispatcher import async_dispatcher_connect
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
def setup_platform(hass, config, add_entities, discovery_info=None): def setup_platform(hass, config, add_entities, discovery_info=None):
"""Set up Tellstick sensors.""" """Old way of setting up TelldusLive.
if discovery_info is None:
return Can only be called when a user accidentally mentions the platform in their
client = hass.data[tellduslive.DOMAIN] config. But even in that case it would have been ignored.
add_entities( """
TelldusLiveSensor(client, binary_sensor) pass
for binary_sensor in discovery_info
)
async def async_setup_entry(hass, config_entry, async_add_entities):
"""Set up tellduslive sensors dynamically."""
async def async_discover_binary_sensor(device_id):
"""Discover and add a discovered sensor."""
client = hass.data[tellduslive.DOMAIN]
async_add_entities([TelldusLiveSensor(client, device_id)])
async_dispatcher_connect(
hass,
tellduslive.TELLDUS_DISCOVERY_NEW.format(binary_sensor.DOMAIN,
tellduslive.DOMAIN),
async_discover_binary_sensor)
class TelldusLiveSensor(TelldusLiveEntity, BinarySensorDevice): class TelldusLiveSensor(TelldusLiveEntity, BinarySensorDevice):

View File

@ -8,20 +8,36 @@ https://home-assistant.io/components/cover.tellduslive/
""" """
import logging import logging
from homeassistant.components import tellduslive from homeassistant.components import cover, tellduslive
from homeassistant.components.cover import CoverDevice from homeassistant.components.cover import CoverDevice
from homeassistant.components.tellduslive.entry import TelldusLiveEntity from homeassistant.components.tellduslive.entry import TelldusLiveEntity
from homeassistant.helpers.dispatcher import async_dispatcher_connect
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
def setup_platform(hass, config, add_entities, discovery_info=None): def setup_platform(hass, config, add_entities, discovery_info=None):
"""Set up the Telldus Live covers.""" """Old way of setting up TelldusLive.
if discovery_info is None:
return
client = hass.data[tellduslive.DOMAIN] Can only be called when a user accidentally mentions the platform in their
add_entities(TelldusLiveCover(client, cover) for cover in discovery_info) config. But even in that case it would have been ignored.
"""
pass
async def async_setup_entry(hass, config_entry, async_add_entities):
"""Set up tellduslive sensors dynamically."""
async def async_discover_cover(device_id):
"""Discover and add a discovered sensor."""
client = hass.data[tellduslive.DOMAIN]
async_add_entities([TelldusLiveCover(client, device_id)])
async_dispatcher_connect(
hass,
tellduslive.TELLDUS_DISCOVERY_NEW.format(cover.DOMAIN,
tellduslive.DOMAIN),
async_discover_cover,
)
class TelldusLiveCover(TelldusLiveEntity, CoverDevice): class TelldusLiveCover(TelldusLiveEntity, CoverDevice):

View File

@ -49,6 +49,7 @@ CONFIG_ENTRY_HANDLERS = {
SERVICE_DECONZ: 'deconz', SERVICE_DECONZ: 'deconz',
'google_cast': 'cast', 'google_cast': 'cast',
SERVICE_HUE: 'hue', SERVICE_HUE: 'hue',
SERVICE_TELLDUSLIVE: 'tellduslive',
SERVICE_IKEA_TRADFRI: 'tradfri', SERVICE_IKEA_TRADFRI: 'tradfri',
'sonos': 'sonos', 'sonos': 'sonos',
} }
@ -62,7 +63,6 @@ SERVICE_HANDLERS = {
SERVICE_APPLE_TV: ('apple_tv', None), SERVICE_APPLE_TV: ('apple_tv', None),
SERVICE_WINK: ('wink', None), SERVICE_WINK: ('wink', None),
SERVICE_XIAOMI_GW: ('xiaomi_aqara', None), SERVICE_XIAOMI_GW: ('xiaomi_aqara', None),
SERVICE_TELLDUSLIVE: ('tellduslive', None),
SERVICE_DAIKIN: ('daikin', None), SERVICE_DAIKIN: ('daikin', None),
SERVICE_SABNZBD: ('sabnzbd', None), SERVICE_SABNZBD: ('sabnzbd', None),
SERVICE_SAMSUNG_PRINTER: ('sensor', 'syncthru'), SERVICE_SAMSUNG_PRINTER: ('sensor', 'syncthru'),

View File

@ -8,20 +8,37 @@ https://home-assistant.io/components/light.tellduslive/
""" """
import logging import logging
from homeassistant.components import tellduslive from homeassistant.components import light, tellduslive
from homeassistant.components.light import ( from homeassistant.components.light import (
ATTR_BRIGHTNESS, SUPPORT_BRIGHTNESS, Light) ATTR_BRIGHTNESS, SUPPORT_BRIGHTNESS, Light)
from homeassistant.components.tellduslive.entry import TelldusLiveEntity from homeassistant.components.tellduslive.entry import TelldusLiveEntity
from homeassistant.helpers.dispatcher import async_dispatcher_connect
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
def setup_platform(hass, config, add_entities, discovery_info=None): def setup_platform(hass, config, add_entities, discovery_info=None):
"""Set up the Tellstick Net lights.""" """Old way of setting up TelldusLive.
if discovery_info is None:
return Can only be called when a user accidentally mentions the platform in their
client = hass.data[tellduslive.DOMAIN] config. But even in that case it would have been ignored.
add_entities(TelldusLiveLight(client, light) for light in discovery_info) """
pass
async def async_setup_entry(hass, config_entry, async_add_entities):
"""Set up tellduslive sensors dynamically."""
async def async_discover_light(device_id):
"""Discover and add a discovered sensor."""
client = hass.data[tellduslive.DOMAIN]
async_add_entities([TelldusLiveLight(client, device_id)])
async_dispatcher_connect(
hass,
tellduslive.TELLDUS_DISCOVERY_NEW.format(light.DOMAIN,
tellduslive.DOMAIN),
async_discover_light,
)
class TelldusLiveLight(TelldusLiveEntity, Light): class TelldusLiveLight(TelldusLiveEntity, Light):

View File

@ -6,11 +6,12 @@ https://home-assistant.io/components/sensor.tellduslive/
""" """
import logging import logging
from homeassistant.components import tellduslive from homeassistant.components import sensor, tellduslive
from homeassistant.components.tellduslive.entry import TelldusLiveEntity from homeassistant.components.tellduslive.entry import TelldusLiveEntity
from homeassistant.const import ( from homeassistant.const import (
DEVICE_CLASS_HUMIDITY, DEVICE_CLASS_ILLUMINANCE, DEVICE_CLASS_TEMPERATURE, DEVICE_CLASS_HUMIDITY, DEVICE_CLASS_ILLUMINANCE, DEVICE_CLASS_TEMPERATURE,
TEMP_CELSIUS) TEMP_CELSIUS)
from homeassistant.helpers.dispatcher import async_dispatcher_connect
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
@ -46,12 +47,25 @@ SENSOR_TYPES = {
def setup_platform(hass, config, add_entities, discovery_info=None): def setup_platform(hass, config, add_entities, discovery_info=None):
"""Set up the Tellstick sensors.""" """Old way of setting up TelldusLive.
if discovery_info is None:
return Can only be called when a user accidentally mentions the platform in their
client = hass.data[tellduslive.DOMAIN] config. But even in that case it would have been ignored.
add_entities( """
TelldusLiveSensor(client, sensor) for sensor in discovery_info) pass
async def async_setup_entry(hass, config_entry, async_add_entities):
"""Set up tellduslive sensors dynamically."""
async def async_discover_sensor(device_id):
"""Discover and add a discovered sensor."""
client = hass.data[tellduslive.DOMAIN]
async_add_entities([TelldusLiveSensor(client, device_id)])
async_dispatcher_connect(
hass,
tellduslive.TELLDUS_DISCOVERY_NEW.format(
sensor.DOMAIN, tellduslive.DOMAIN), async_discover_sensor)
class TelldusLiveSensor(TelldusLiveEntity): class TelldusLiveSensor(TelldusLiveEntity):

View File

@ -9,20 +9,36 @@ https://home-assistant.io/components/switch.tellduslive/
""" """
import logging import logging
from homeassistant.components import tellduslive from homeassistant.components import switch, tellduslive
from homeassistant.components.tellduslive.entry import TelldusLiveEntity from homeassistant.components.tellduslive.entry import TelldusLiveEntity
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity import ToggleEntity from homeassistant.helpers.entity import ToggleEntity
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
def setup_platform(hass, config, add_entities, discovery_info=None): def setup_platform(hass, config, add_entities, discovery_info=None):
"""Set up Tellstick switches.""" """Old way of setting up TelldusLive.
if discovery_info is None:
return Can only be called when a user accidentally mentions the platform in their
client = hass.data[tellduslive.DOMAIN] config. But even in that case it would have been ignored.
add_entities( """
TelldusLiveSwitch(client, switch) for switch in discovery_info) pass
async def async_setup_entry(hass, config_entry, async_add_entities):
"""Set up tellduslive sensors dynamically."""
async def async_discover_switch(device_id):
"""Discover and add a discovered sensor."""
client = hass.data[tellduslive.DOMAIN]
async_add_entities([TelldusLiveSwitch(client, device_id)])
async_dispatcher_connect(
hass,
tellduslive.TELLDUS_DISCOVERY_NEW.format(switch.DOMAIN,
tellduslive.DOMAIN),
async_discover_switch,
)
class TelldusLiveSwitch(TelldusLiveEntity, ToggleEntity): class TelldusLiveSwitch(TelldusLiveEntity, ToggleEntity):

View File

@ -0,0 +1,24 @@
{
"config": {
"title": "Telldus Live",
"step": {
"user": {
"title": "Pick endpoint.",
"description": "",
"data": {
"host": "Host"
}
},
"auth": {
"title": "Authenticate against TelldusLive",
"description": "To link your TelldusLive account:\n 1. Click the link below\n 2. Login to Telldus Live\n 3. Authorize **{app_name}** (click **Yes**).\n 4. Come back here and click **SUBMIT**.\n\n [Link TelldusLive account]({auth_url})"
}
},
"abort": {
"authorize_url_timeout": "Timeout generating authorize url.",
"authorize_url_fail": "Unknown error generating an authorize url.",
"all_configured": "TelldusLive is already configured",
"unknown": "Unknown error occurred"
}
}
}

View File

@ -4,20 +4,22 @@ Support for Telldus Live.
For more details about this component, please refer to the documentation at For more details about this component, please refer to the documentation at
https://home-assistant.io/components/tellduslive/ https://home-assistant.io/components/tellduslive/
""" """
import asyncio
from datetime import timedelta from datetime import timedelta
import logging import logging
import voluptuous as vol import voluptuous as vol
from homeassistant.components.discovery import SERVICE_TELLDUSLIVE from homeassistant import config_entries
from homeassistant.const import CONF_HOST, CONF_TOKEN
from homeassistant.helpers import discovery
import homeassistant.helpers.config_validation as cv import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.dispatcher import dispatcher_send from homeassistant.helpers.dispatcher import async_dispatcher_send
from homeassistant.helpers.event import track_time_interval from homeassistant.helpers.event import async_track_time_interval
from homeassistant.util.json import load_json, save_json
from .const import DOMAIN, SIGNAL_UPDATE_ENTITY from . import config_flow # noqa pylint_disable=unused-import
from .const import (
CONF_HOST, CONF_UPDATE_INTERVAL, DOMAIN, KEY_HOST, KEY_SCAN_INTERVAL,
KEY_SESSION, MIN_UPDATE_INTERVAL, NOT_SO_PRIVATE_KEY, PUBLIC_KEY,
SCAN_INTERVAL, SIGNAL_UPDATE_ENTITY, TELLDUS_DISCOVERY_NEW)
APPLICATION_NAME = 'Home Assistant' APPLICATION_NAME = 'Home Assistant'
@ -25,229 +27,149 @@ REQUIREMENTS = ['tellduslive==0.10.4']
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
TELLLDUS_CONFIG_FILE = 'tellduslive.conf' CONFIG_SCHEMA = vol.Schema(
KEY_CONFIG = 'tellduslive_config' {
DOMAIN:
vol.Schema({
vol.Optional(CONF_HOST, default=DOMAIN):
cv.string,
vol.Optional(CONF_UPDATE_INTERVAL, default=SCAN_INTERVAL):
(vol.All(cv.time_period, vol.Clamp(min=MIN_UPDATE_INTERVAL)))
}),
},
extra=vol.ALLOW_EXTRA,
)
CONF_TOKEN_SECRET = 'token_secret' DATA_CONFIG_ENTRY_LOCK = 'tellduslive_config_entry_lock'
CONF_UPDATE_INTERVAL = 'update_interval' CONFIG_ENTRY_IS_SETUP = 'telldus_config_entry_is_setup'
PUBLIC_KEY = 'THUPUNECH5YEQA3RE6UYUPRUZ2DUGUGA' INTERVAL_TRACKER = '{}_INTERVAL'.format(DOMAIN)
NOT_SO_PRIVATE_KEY = 'PHES7U2RADREWAFEBUSTUBAWRASWUTUS'
MIN_UPDATE_INTERVAL = timedelta(seconds=5)
DEFAULT_UPDATE_INTERVAL = timedelta(minutes=1)
CONFIG_SCHEMA = vol.Schema({
DOMAIN: vol.Schema({
vol.Optional(CONF_HOST): cv.string,
vol.Optional(CONF_UPDATE_INTERVAL, default=DEFAULT_UPDATE_INTERVAL): (
vol.All(cv.time_period, vol.Clamp(min=MIN_UPDATE_INTERVAL)))
}),
}, extra=vol.ALLOW_EXTRA)
CONFIG_INSTRUCTIONS = """
To link your TelldusLive account:
1. Click the link below
2. Login to Telldus Live
3. Authorize {app_name}.
4. Click the Confirm button.
[Link TelldusLive account]({auth_url})
"""
def setup(hass, config, session=None): async def async_setup_entry(hass, entry):
"""Set up the Telldus Live component.""" """Create a tellduslive session."""
from tellduslive import Session, supports_local_api from tellduslive import Session
config_filename = hass.config.path(TELLLDUS_CONFIG_FILE) conf = entry.data[KEY_SESSION]
conf = load_json(config_filename)
def request_configuration(host=None): if KEY_HOST in conf:
"""Request TelldusLive authorization.""" session = Session(**conf)
configurator = hass.components.configurator
hass.data.setdefault(KEY_CONFIG, {})
data_key = host or DOMAIN
# Configuration already in progress
if hass.data[KEY_CONFIG].get(data_key):
return
_LOGGER.info('Configuring TelldusLive %s',
'local client: {}'.format(host) if host else
'cloud service')
session = Session(public_key=PUBLIC_KEY,
private_key=NOT_SO_PRIVATE_KEY,
host=host,
application=APPLICATION_NAME)
auth_url = session.authorize_url
if not auth_url:
_LOGGER.warning('Failed to retrieve authorization URL')
return
_LOGGER.debug('Got authorization URL %s', auth_url)
def configuration_callback(callback_data):
"""Handle the submitted configuration."""
session.authorize()
res = setup(hass, config, session)
if not res:
configurator.notify_errors(
hass.data[KEY_CONFIG].get(data_key),
'Unable to connect.')
return
conf.update(
{host: {CONF_HOST: host,
CONF_TOKEN: session.access_token}} if host else
{DOMAIN: {CONF_TOKEN: session.access_token,
CONF_TOKEN_SECRET: session.access_token_secret}})
save_json(config_filename, conf)
# Close all open configurators: for now, we only support one
# tellstick device, and configuration via either cloud service
# or via local API, not both at the same time
for instance in hass.data[KEY_CONFIG].values():
configurator.request_done(instance)
hass.data[KEY_CONFIG][data_key] = \
configurator.request_config(
'TelldusLive ({})'.format(
'LocalAPI' if host
else 'Cloud service'),
configuration_callback,
description=CONFIG_INSTRUCTIONS.format(
app_name=APPLICATION_NAME,
auth_url=auth_url),
submit_caption='Confirm',
entity_picture='/static/images/logo_tellduslive.png',
)
def tellstick_discovered(service, info):
"""Run when a Tellstick is discovered."""
_LOGGER.info('Discovered tellstick device')
if DOMAIN in hass.data:
_LOGGER.debug('Tellstick already configured')
return
host, device = info[:2]
if not supports_local_api(device):
_LOGGER.debug('Tellstick does not support local API')
# Configure the cloud service
hass.add_job(request_configuration)
return
_LOGGER.debug('Tellstick does support local API')
# Ignore any known devices
if conf and host in conf:
_LOGGER.debug('Discovered already known device: %s', host)
return
# Offer configuration of both live and local API
request_configuration()
request_configuration(host)
discovery.listen(hass, SERVICE_TELLDUSLIVE, tellstick_discovered)
if session:
_LOGGER.debug('Continuing setup configured by configurator')
elif conf and CONF_HOST in next(iter(conf.values())):
# For now, only one local device is supported
_LOGGER.debug('Using Local API pre-configured by configurator')
session = Session(**next(iter(conf.values())))
elif DOMAIN in conf:
_LOGGER.debug('Using TelldusLive cloud service '
'pre-configured by configurator')
session = Session(PUBLIC_KEY, NOT_SO_PRIVATE_KEY,
application=APPLICATION_NAME, **conf[DOMAIN])
elif config.get(DOMAIN):
_LOGGER.info('Found entry in configuration.yaml. '
'Requesting TelldusLive cloud service configuration')
request_configuration()
if CONF_HOST in config.get(DOMAIN, {}):
_LOGGER.info('Found TelldusLive host entry in configuration.yaml. '
'Requesting Telldus Local API configuration')
request_configuration(config.get(DOMAIN).get(CONF_HOST))
return True
else: else:
_LOGGER.info('Tellstick discovered, awaiting discovery callback') session = Session(
return True PUBLIC_KEY,
NOT_SO_PRIVATE_KEY,
application=APPLICATION_NAME,
**conf,
)
if not session.is_authorized: if not session.is_authorized:
_LOGGER.error('Authentication Error') _LOGGER.error('Authentication Error')
return False return False
client = TelldusLiveClient(hass, config, session) hass.data[DATA_CONFIG_ENTRY_LOCK] = asyncio.Lock()
hass.data[CONFIG_ENTRY_IS_SETUP] = set()
client = TelldusLiveClient(hass, entry, session)
hass.data[DOMAIN] = client hass.data[DOMAIN] = client
client.update()
interval = config.get(DOMAIN, {}).get(CONF_UPDATE_INTERVAL, await client.update()
DEFAULT_UPDATE_INTERVAL)
interval = timedelta(seconds=entry.data[KEY_SCAN_INTERVAL])
_LOGGER.debug('Update interval %s', interval) _LOGGER.debug('Update interval %s', interval)
track_time_interval(hass, client.update, interval) hass.data[INTERVAL_TRACKER] = async_track_time_interval(
hass, client.update, interval)
return True return True
async def async_setup(hass, config):
"""Set up the Telldus Live component."""
if DOMAIN not in config:
return True
hass.async_create_task(
hass.config_entries.flow.async_init(
DOMAIN,
context={'source': config_entries.SOURCE_IMPORT},
data={
KEY_HOST: config[DOMAIN].get(CONF_HOST),
KEY_SCAN_INTERVAL: config[DOMAIN].get(CONF_UPDATE_INTERVAL),
}))
return True
async def async_unload_entry(hass, config_entry):
"""Unload a config entry."""
interval_tracker = hass.data.pop(INTERVAL_TRACKER)
interval_tracker()
await asyncio.wait([
hass.config_entries.async_forward_entry_unload(config_entry, component)
for component in hass.data.pop(CONFIG_ENTRY_IS_SETUP)
])
del hass.data[DOMAIN]
del hass.data[DATA_CONFIG_ENTRY_LOCK]
return True
class TelldusLiveClient: class TelldusLiveClient:
"""Get the latest data and update the states.""" """Get the latest data and update the states."""
def __init__(self, hass, config, session): def __init__(self, hass, config_entry, session):
"""Initialize the Tellus data object.""" """Initialize the Tellus data object."""
self._known_devices = set() self._known_devices = set()
self._hass = hass self._hass = hass
self._config = config self._config_entry = config_entry
self._client = session self._client = session
def update(self, *args): @staticmethod
"""Update local list of devices.""" def identify_device(device):
"""Find out what type of HA component to create."""
if device.is_sensor:
return 'sensor'
from tellduslive import (DIM, UP, TURNON)
if device.methods & DIM:
return 'light'
if device.methods & UP:
return 'cover'
if device.methods & TURNON:
return 'switch'
if device.methods == 0:
return 'binary_sensor'
_LOGGER.warning("Unidentified device type (methods: %d)",
device.methods)
return 'switch'
async def _discover(self, device_id):
"""Discover the component."""
device = self._client.device(device_id)
component = self.identify_device(device)
async with self._hass.data[DATA_CONFIG_ENTRY_LOCK]:
if component not in self._hass.data[CONFIG_ENTRY_IS_SETUP]:
await self._hass.config_entries.async_forward_entry_setup(
self._config_entry, component)
self._hass.data[CONFIG_ENTRY_IS_SETUP].add(component)
device_ids = []
if device.is_sensor:
for item in device.items:
device_ids.append((device.device_id, item.name, item.scale))
else:
device_ids.append(device_id)
for _id in device_ids:
async_dispatcher_send(
self._hass, TELLDUS_DISCOVERY_NEW.format(component, DOMAIN),
_id)
async def update(self, *args):
"""Periodically poll the servers for current state."""
_LOGGER.debug('Updating') _LOGGER.debug('Updating')
if not self._client.update(): if not self._client.update():
_LOGGER.warning('Failed request') _LOGGER.warning('Failed request')
def identify_device(device): dev_ids = {dev.device_id for dev in self._client.devices}
"""Find out what type of HA component to create.""" new_devices = dev_ids - self._known_devices
from tellduslive import (DIM, UP, TURNON) await asyncio.gather(*[self._discover(d_id) for d_id in new_devices])
if device.methods & DIM: self._known_devices |= new_devices
return 'light' async_dispatcher_send(self._hass, SIGNAL_UPDATE_ENTITY)
if device.methods & UP:
return 'cover'
if device.methods & TURNON:
return 'switch'
if device.methods == 0:
return 'binary_sensor'
_LOGGER.warning(
"Unidentified device type (methods: %d)", device.methods)
return 'switch'
def discover(device_id, component):
"""Discover the component."""
discovery.load_platform(
self._hass, component, DOMAIN, [device_id], self._config)
for device in self._client.devices:
if device.device_id in self._known_devices:
continue
if device.is_sensor:
for item in device.items:
discover((device.device_id, item.name, item.scale),
'sensor')
else:
discover(device.device_id,
identify_device(device))
self._known_devices.add(device.device_id)
dispatcher_send(self._hass, SIGNAL_UPDATE_ENTITY)
def device(self, device_id): def device(self, device_id):
"""Return device representation.""" """Return device representation."""

View File

@ -0,0 +1,150 @@
"""Config flow for Tellduslive."""
import asyncio
import logging
import os
import async_timeout
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.util.json import load_json
from .const import (
APPLICATION_NAME, CLOUD_NAME, DOMAIN, KEY_HOST, KEY_SCAN_INTERVAL,
KEY_SESSION, NOT_SO_PRIVATE_KEY, PUBLIC_KEY, SCAN_INTERVAL,
TELLDUS_CONFIG_FILE)
KEY_TOKEN = 'token'
KEY_TOKEN_SECRET = 'token_secret'
_LOGGER = logging.getLogger(__name__)
@config_entries.HANDLERS.register('tellduslive')
class FlowHandler(config_entries.ConfigFlow):
"""Handle a config flow."""
VERSION = 1
CONNECTION_CLASS = config_entries.CONN_CLASS_LOCAL_POLL
def __init__(self):
"""Init config flow."""
self._hosts = [CLOUD_NAME]
self._host = None
self._session = None
self._scan_interval = SCAN_INTERVAL
def _get_auth_url(self):
return self._session.authorize_url
async def async_step_user(self, user_input=None):
"""Let user select host or cloud."""
if self.hass.config_entries.async_entries(DOMAIN):
return self.async_abort(reason='already_setup')
if user_input is not None or len(self._hosts) == 1:
if user_input is not None and user_input[KEY_HOST] != CLOUD_NAME:
self._host = user_input[KEY_HOST]
return await self.async_step_auth()
return self.async_show_form(
step_id='user',
data_schema=vol.Schema({
vol.Required(KEY_HOST):
vol.In(list(self._hosts))
}))
async def async_step_auth(self, user_input=None):
"""Handle the submitted configuration."""
if not self._session:
from tellduslive import Session
self._session = Session(
public_key=PUBLIC_KEY,
private_key=NOT_SO_PRIVATE_KEY,
host=self._host,
application=APPLICATION_NAME,
)
if user_input is not None and self._session.authorize():
host = self._host or CLOUD_NAME
if self._host:
session = {
KEY_HOST: host,
KEY_TOKEN: self._session.access_token
}
else:
session = {
KEY_TOKEN: self._session.access_token,
KEY_TOKEN_SECRET: self._session.access_token_secret
}
return self.async_create_entry(
title=host, data={
KEY_HOST: host,
KEY_SCAN_INTERVAL: self._scan_interval.seconds,
KEY_SESSION: session,
})
try:
with async_timeout.timeout(10):
auth_url = await self.hass.async_add_executor_job(
self._get_auth_url)
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')
_LOGGER.debug('Got authorization URL %s', auth_url)
return self.async_show_form(
step_id='auth',
description_placeholders={
'app_name': APPLICATION_NAME,
'auth_url': auth_url,
},
)
async def async_step_discovery(self, user_input):
"""Run when a Tellstick is discovered."""
from tellduslive import supports_local_api
_LOGGER.info('Discovered tellstick device: %s', user_input)
# Ignore any known devices
for entry in self._async_current_entries():
if entry.data[KEY_HOST] == user_input[0]:
return self.async_abort(reason='already_configured')
if not supports_local_api(user_input[1]):
_LOGGER.debug('Tellstick does not support local API')
# Configure the cloud service
return await self.async_step_auth()
self._hosts.append(user_input[0])
return await self.async_step_user()
async def async_step_import(self, user_input):
"""Import a config entry."""
if self.hass.config_entries.async_entries(DOMAIN):
return self.async_abort(reason='already_setup')
self._scan_interval = user_input[KEY_SCAN_INTERVAL]
if user_input[KEY_HOST] != DOMAIN:
self._hosts.append(user_input[KEY_HOST])
if not await self.hass.async_add_executor_job(
os.path.isfile, self.hass.config.path(TELLDUS_CONFIG_FILE)):
return await self.async_step_user()
conf = await self.hass.async_add_executor_job(
load_json, self.hass.config.path(TELLDUS_CONFIG_FILE))
host = next(iter(conf))
if user_input[KEY_HOST] != host:
return await self.async_step_user()
host = CLOUD_NAME if host == 'tellduslive' else host
return self.async_create_entry(
title=host,
data={
KEY_HOST: host,
KEY_SCAN_INTERVAL: self._scan_interval.seconds,
KEY_SESSION: next(iter(conf.values())),
})

View File

@ -1,5 +1,34 @@
"""Consts used by TelldusLive.""" """Consts used by TelldusLive."""
from datetime import timedelta
from homeassistant.const import ( # noqa pylint: disable=unused-import
ATTR_BATTERY_LEVEL, CONF_HOST, CONF_TOKEN, DEVICE_DEFAULT_NAME)
APPLICATION_NAME = 'Home Assistant'
DOMAIN = 'tellduslive' DOMAIN = 'tellduslive'
TELLDUS_CONFIG_FILE = 'tellduslive.conf'
KEY_CONFIG = 'tellduslive_config'
SIGNAL_UPDATE_ENTITY = 'tellduslive_update' SIGNAL_UPDATE_ENTITY = 'tellduslive_update'
KEY_HOST = 'host'
KEY_SESSION = 'session'
KEY_SCAN_INTERVAL = 'scan_interval'
CONF_TOKEN_SECRET = 'token_secret'
CONF_UPDATE_INTERVAL = 'update_interval'
PUBLIC_KEY = 'THUPUNECH5YEQA3RE6UYUPRUZ2DUGUGA'
NOT_SO_PRIVATE_KEY = 'PHES7U2RADREWAFEBUSTUBAWRASWUTUS'
MIN_UPDATE_INTERVAL = timedelta(seconds=5)
SCAN_INTERVAL = timedelta(minutes=1)
ATTR_LAST_UPDATED = 'time_last_updated'
SIGNAL_UPDATE_ENTITY = 'tellduslive_update'
TELLDUS_DISCOVERY_NEW = 'telldus_new_{}_{}'
CLOUD_NAME = 'Cloud API'

View File

@ -0,0 +1,24 @@
{
"config": {
"title": "Telldus Live",
"step": {
"user": {
"title": "Pick endpoint.",
"description": "",
"data": {
"host": "Host"
}
},
"auth": {
"title": "Authenticate against TelldusLive",
"description": "To link your TelldusLive account:\n 1. Click the link below\n 2. Login to Telldus Live\n 3. Authorize **{app_name}** (click **Yes**).\n 4. Come back here and click **SUBMIT**.\n\n [Link TelldusLive account]({auth_url})"
}
},
"abort": {
"authorize_url_timeout": "Timeout generating authorize url.",
"authorize_url_fail": "Unknown error generating an authorize url.",
"all_configured": "TelldusLive is already configured",
"unknown": "Unknown error occurred"
}
}
}

View File

@ -155,6 +155,7 @@ FLOWS = [
'simplisafe', 'simplisafe',
'smhi', 'smhi',
'sonos', 'sonos',
'tellduslive',
'tradfri', 'tradfri',
'twilio', 'twilio',
'unifi', 'unifi',

View File

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

View File

@ -0,0 +1,223 @@
# flake8: noqa pylint: skip-file
"""Tests for the TelldusLive config flow."""
import asyncio
from unittest.mock import Mock, patch
import pytest
from homeassistant import data_entry_flow
from homeassistant.components.tellduslive import (
APPLICATION_NAME, DOMAIN, KEY_HOST, KEY_SCAN_INTERVAL, SCAN_INTERVAL,
config_flow)
from tests.common import MockConfigEntry, MockDependency, mock_coro
def init_config_flow(hass, side_effect=None):
"""Init a configuration flow."""
flow = config_flow.FlowHandler()
flow.hass = hass
if side_effect:
flow._get_auth_url = Mock(side_effect=side_effect)
return flow
@pytest.fixture
def supports_local_api():
"""Set TelldusLive supports_local_api."""
return True
@pytest.fixture
def authorize():
"""Set TelldusLive authorize."""
return True
@pytest.fixture
def mock_tellduslive(supports_local_api, authorize):
"""Mock tellduslive."""
with MockDependency('tellduslive') as mock_tellduslive_:
mock_tellduslive_.supports_local_api.return_value = supports_local_api
mock_tellduslive_.Session().authorize.return_value = authorize
mock_tellduslive_.Session().access_token = 'token'
mock_tellduslive_.Session().access_token_secret = 'token_secret'
mock_tellduslive_.Session().authorize_url = 'https://example.com'
yield mock_tellduslive_
async def test_abort_if_already_setup(hass):
"""Test we abort if TelldusLive is already setup."""
flow = init_config_flow(hass)
with patch.object(hass.config_entries, 'async_entries', return_value=[{}]):
result = await flow.async_step_user()
assert result['type'] == data_entry_flow.RESULT_TYPE_ABORT
assert result['reason'] == 'already_setup'
with patch.object(hass.config_entries, 'async_entries', return_value=[{}]):
result = await flow.async_step_import(None)
assert result['type'] == data_entry_flow.RESULT_TYPE_ABORT
assert result['reason'] == 'already_setup'
async def test_full_flow_implementation(hass, mock_tellduslive):
"""Test registering an implementation and finishing flow works."""
flow = init_config_flow(hass)
result = await flow.async_step_discovery(['localhost', 'tellstick'])
assert result['type'] == data_entry_flow.RESULT_TYPE_FORM
assert result['step_id'] == 'user'
result = await flow.async_step_user()
assert result['type'] == data_entry_flow.RESULT_TYPE_FORM
assert result['step_id'] == 'user'
result = await flow.async_step_user({'host': 'localhost'})
assert result['type'] == data_entry_flow.RESULT_TYPE_FORM
assert result['step_id'] == 'auth'
assert result['description_placeholders'] == {
'auth_url': 'https://example.com',
'app_name': APPLICATION_NAME,
}
result = await flow.async_step_auth('')
assert result['type'] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result['title'] == 'localhost'
assert result['data']['host'] == 'localhost'
assert result['data']['scan_interval'] == 60
assert result['data']['session'] == {'token': 'token', 'host': 'localhost'}
async def test_step_import(hass, mock_tellduslive):
"""Test that we trigger auth when configuring from import."""
flow = init_config_flow(hass)
result = await flow.async_step_import({
KEY_HOST: DOMAIN,
KEY_SCAN_INTERVAL: 0,
})
assert result['type'] == data_entry_flow.RESULT_TYPE_FORM
assert result['step_id'] == 'auth'
async def test_step_import_add_host(hass, mock_tellduslive):
"""Test that we add host and trigger user when configuring from import."""
flow = init_config_flow(hass)
result = await flow.async_step_import({
KEY_HOST: 'localhost',
KEY_SCAN_INTERVAL: 0,
})
assert result['type'] == data_entry_flow.RESULT_TYPE_FORM
assert result['step_id'] == 'user'
async def test_step_import_no_config_file(hass, mock_tellduslive):
"""Test that we trigger user with no config_file configuring from import."""
flow = init_config_flow(hass)
result = await flow.async_step_import({ KEY_HOST: 'localhost', KEY_SCAN_INTERVAL: 0, })
assert result['type'] == data_entry_flow.RESULT_TYPE_FORM
assert result['step_id'] == 'user'
async def test_step_import_load_json_matching_host(hass, mock_tellduslive):
"""Test that we add host and trigger user when configuring from import."""
flow = init_config_flow(hass)
with patch('homeassistant.components.tellduslive.config_flow.load_json',
return_value={'tellduslive': {}}), \
patch('os.path.isfile'):
result = await flow.async_step_import({ KEY_HOST: 'Cloud API', KEY_SCAN_INTERVAL: 0, })
assert result['type'] == data_entry_flow.RESULT_TYPE_FORM
assert result['step_id'] == 'user'
async def test_step_import_load_json(hass, mock_tellduslive):
"""Test that we create entry when configuring from import."""
flow = init_config_flow(hass)
with patch('homeassistant.components.tellduslive.config_flow.load_json',
return_value={'localhost': {}}), \
patch('os.path.isfile'):
result = await flow.async_step_import({ KEY_HOST: 'localhost', KEY_SCAN_INTERVAL: SCAN_INTERVAL, })
assert result['type'] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result['title'] == 'localhost'
assert result['data']['host'] == 'localhost'
assert result['data']['scan_interval'] == 60
assert result['data']['session'] == {}
@pytest.mark.parametrize('supports_local_api', [False])
async def test_step_disco_no_local_api(hass, mock_tellduslive):
"""Test that we trigger when configuring from discovery, not supporting local api."""
flow = init_config_flow(hass)
result = await flow.async_step_discovery(['localhost', 'tellstick'])
assert result['type'] == data_entry_flow.RESULT_TYPE_FORM
assert result['step_id'] == 'auth'
async def test_step_auth(hass, mock_tellduslive):
"""Test that create cloud entity from auth."""
flow = init_config_flow(hass)
result = await flow.async_step_auth(['localhost', 'tellstick'])
assert result['type'] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result['title'] == 'Cloud API'
assert result['data']['host'] == 'Cloud API'
assert result['data']['scan_interval'] == 60
assert result['data']['session'] == {
'token': 'token',
'token_secret': 'token_secret',
}
@pytest.mark.parametrize('authorize', [False])
async def test_wrong_auth_flow_implementation(hass, mock_tellduslive):
"""Test wrong auth."""
flow = init_config_flow(hass)
await flow.async_step_user()
result = await flow.async_step_auth('')
assert result['type'] == data_entry_flow.RESULT_TYPE_FORM
assert result['step_id'] == 'auth'
async def test_not_pick_host_if_only_one(hass, mock_tellduslive):
"""Test not picking host if we have just one."""
flow = init_config_flow(hass)
result = await flow.async_step_user()
assert result['type'] == data_entry_flow.RESULT_TYPE_FORM
assert result['step_id'] == 'auth'
async def test_abort_if_timeout_generating_auth_url(hass, mock_tellduslive):
"""Test abort if generating authorize url timeout."""
flow = init_config_flow(hass, side_effect=asyncio.TimeoutError)
result = await flow.async_step_user()
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, mock_tellduslive):
"""Test we abort if generating authorize url blows up."""
flow = init_config_flow(hass, side_effect=ValueError)
result = await flow.async_step_user()
assert result['type'] == data_entry_flow.RESULT_TYPE_ABORT
assert result['reason'] == 'authorize_url_fail'
async def test_discovery_already_configured(hass, mock_tellduslive):
"""Test abort if alredy configured fires from discovery."""
MockConfigEntry(
domain='tellduslive',
data={'host': 'some-host'}
).add_to_hass(hass)
flow = init_config_flow(hass)
result = await flow.async_step_discovery(['some-host', ''])
assert result['type'] == data_entry_flow.RESULT_TYPE_ABORT
assert result['reason'] == 'already_configured'