Somfy open api (#19548)
* CREATE Somfy component * CREATE cover Somfy platform * USE somfy id as unique id * UPDATE all the devices in one call to limit the number of call * FIX Don't load devices if not yet configured * IMP Replace configurator by a simple notification * ADD log in case state does not match * IMP wording * REMOVE debug stuf * ADD support for tilt position * UPDATE requirements * FIX Use code instead of authorization response - Will allow to setup Somfy without https * HANDLE stateless devices (Somfy RTS) * FIX import locally 3rd party library * UPDATE pymfy to 0.4.3 * ADD missing docstring * FIX For Somfy 100 means closed and 0 opened * FIX position can be None * ENHANCE error management when error 500 occurs at setup * FIX indent * ROLLBACK tilt modification - See https://community.home-assistant.io/t/somfy-tahoma-official-api/61448/90?u=tetienne * FIX Look for capability instead of state * DON'T use exception to test if a feature is available * UPDATE dependency * ADD device_info property * AVOID object creation in each method * REMOVE unused constants * ADD missing doc * IMP Only make one call to add_entities * USE dict[key] instead of get method * IMP Don't pass hass object to the entities * FIX Don't end logging messages with period * USE config entries instead of a cache file * IMPLEMENT async_unload_entry * CONSOLIDATE package - see home-assistant/architecture#124 * UPDATE to pymfy 0.5.1 * SIMPLIFY config flow * ADD French translation * FIX 80 vs 79 max length * ABORT flow asap * FIX A tupple was returned * MIGRATE to manifest.json * ADD a placeholder async_setup_platform coroutine - It's currently required and expected by the platform helper. * FIX codeowner * ADD missing translations file * USE new external step * UPGRADE pymfy version * Close Somfy tab automatically * ADD manufacturer - Somfy only for the moment. * HANDLE missing code or state in Somfy request * REMOVE unused strings * DECLARE somfy component to use config_flow * APPLY static check remarks * FIX async method cannot be called from sync context * FIX only unload what has been loaded during entry setup * DON't catch them all * DON'T log full stacktrace * ABORT conflig flow if configuration missing * OMIT Somfy files for coverage * ADD tests about Somfy config flow * ADD pymfy to the test dependenciespull/24485/head
tetienne
Paulus Schoutsen
parent
046a4fc401
commit
0a7919a279
|
|
@ -561,6 +561,7 @@ omit =
|
|||
homeassistant/components/solaredge/sensor.py
|
||||
homeassistant/components/solaredge_local/sensor.py
|
||||
homeassistant/components/solax/sensor.py
|
||||
homeassistant/components/somfy/*
|
||||
homeassistant/components/somfy_mylink/*
|
||||
homeassistant/components/sonarr/sensor.py
|
||||
homeassistant/components/songpal/media_player.py
|
||||
|
|
|
|||
|
|
@ -223,6 +223,7 @@ homeassistant/components/smarty/* @z0mbieprocess
|
|||
homeassistant/components/smtp/* @fabaff
|
||||
homeassistant/components/solaredge_local/* @drobtravels
|
||||
homeassistant/components/solax/* @squishykid
|
||||
homeassistant/components/somfy/* @tetienne
|
||||
homeassistant/components/sonos/* @amelchio
|
||||
homeassistant/components/spaceapi/* @fabaff
|
||||
homeassistant/components/spider/* @peternijssen
|
||||
|
|
|
|||
|
|
@ -0,0 +1,13 @@
|
|||
{
|
||||
"config": {
|
||||
"abort": {
|
||||
"already_setup": "You can only configure one Somfy account.",
|
||||
"authorize_url_timeout": "Timeout generating authorize url.",
|
||||
"missing_configuration": "The Somfy component is not configured. Please follow the documentation."
|
||||
},
|
||||
"create_entry": {
|
||||
"default": "Successfully authenticated with Somfy."
|
||||
},
|
||||
"title": "Somfy"
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,13 @@
|
|||
{
|
||||
"config": {
|
||||
"abort": {
|
||||
"already_setup": "Vous ne pouvez configurer qu'un seul compte Somfy.",
|
||||
"authorize_url_timeout": "Durée expirée pour la génération de l'url d'autorisation.",
|
||||
"missing_configuration": "Le composant Somfy n'est pas configuré. Merci de suivre la documentation."
|
||||
},
|
||||
"create_entry": {
|
||||
"default": "Authentification réussie avec Somfy."
|
||||
},
|
||||
"title": "Somfy"
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,160 @@
|
|||
"""
|
||||
Support for Somfy hubs.
|
||||
|
||||
For more details about this component, please refer to the documentation at
|
||||
https://home-assistant.io/components/somfy/
|
||||
"""
|
||||
import logging
|
||||
from datetime import timedelta
|
||||
from functools import partial
|
||||
|
||||
import voluptuous as vol
|
||||
|
||||
import homeassistant.helpers.config_validation as cv
|
||||
from homeassistant import config_entries
|
||||
from homeassistant.components.somfy import config_flow
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.const import CONF_TOKEN
|
||||
from homeassistant.helpers.entity import Entity
|
||||
from homeassistant.helpers.typing import HomeAssistantType
|
||||
from homeassistant.util import Throttle
|
||||
|
||||
API = 'api'
|
||||
|
||||
DEVICES = 'devices'
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
MIN_TIME_BETWEEN_UPDATES = timedelta(seconds=10)
|
||||
|
||||
DOMAIN = 'somfy'
|
||||
|
||||
CONF_CLIENT_ID = 'client_id'
|
||||
CONF_CLIENT_SECRET = 'client_secret'
|
||||
|
||||
SOMFY_AUTH_CALLBACK_PATH = '/auth/somfy/callback'
|
||||
SOMFY_AUTH_START = '/auth/somfy'
|
||||
|
||||
CONFIG_SCHEMA = vol.Schema({
|
||||
DOMAIN: vol.Schema({
|
||||
vol.Required(CONF_CLIENT_ID): cv.string,
|
||||
vol.Required(CONF_CLIENT_SECRET): cv.string
|
||||
})
|
||||
}, extra=vol.ALLOW_EXTRA)
|
||||
|
||||
SOMFY_COMPONENTS = ['cover']
|
||||
|
||||
|
||||
async def async_setup(hass, config):
|
||||
"""Set up the Somfy component."""
|
||||
if DOMAIN not in config:
|
||||
return True
|
||||
|
||||
hass.data[DOMAIN] = {}
|
||||
|
||||
config_flow.register_flow_implementation(
|
||||
hass, config[DOMAIN][CONF_CLIENT_ID],
|
||||
config[DOMAIN][CONF_CLIENT_SECRET])
|
||||
|
||||
hass.async_create_task(
|
||||
hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={'source': config_entries.SOURCE_IMPORT},
|
||||
))
|
||||
|
||||
return True
|
||||
|
||||
|
||||
async def async_setup_entry(hass: HomeAssistantType, entry: ConfigEntry):
|
||||
"""Set up Somfy from a config entry."""
|
||||
def token_saver(token):
|
||||
_LOGGER.debug('Saving updated token')
|
||||
entry.data[CONF_TOKEN] = token
|
||||
update_entry = partial(
|
||||
hass.config_entries.async_update_entry,
|
||||
data={**entry.data}
|
||||
)
|
||||
hass.add_job(update_entry, entry)
|
||||
|
||||
# Force token update.
|
||||
from pymfy.api.somfy_api import SomfyApi
|
||||
hass.data[DOMAIN][API] = SomfyApi(
|
||||
entry.data['refresh_args']['client_id'],
|
||||
entry.data['refresh_args']['client_secret'],
|
||||
token=entry.data[CONF_TOKEN],
|
||||
token_updater=token_saver
|
||||
)
|
||||
|
||||
await update_all_devices(hass)
|
||||
|
||||
for component in SOMFY_COMPONENTS:
|
||||
hass.async_create_task(
|
||||
hass.config_entries.async_forward_entry_setup(entry, component))
|
||||
|
||||
return True
|
||||
|
||||
|
||||
async def async_unload_entry(hass: HomeAssistantType, entry: ConfigEntry):
|
||||
"""Unload a config entry."""
|
||||
hass.data[DOMAIN].pop(API, None)
|
||||
return True
|
||||
|
||||
|
||||
class SomfyEntity(Entity):
|
||||
"""Representation of a generic Somfy device."""
|
||||
|
||||
def __init__(self, device, api):
|
||||
"""Initialize the Somfy device."""
|
||||
self.device = device
|
||||
self.api = api
|
||||
|
||||
@property
|
||||
def unique_id(self):
|
||||
"""Return the unique id base on the id returned by Somfy."""
|
||||
return self.device.id
|
||||
|
||||
@property
|
||||
def name(self):
|
||||
"""Return the name of the device."""
|
||||
return self.device.name
|
||||
|
||||
@property
|
||||
def device_info(self):
|
||||
"""Return device specific attributes.
|
||||
|
||||
Implemented by platform classes.
|
||||
"""
|
||||
return {
|
||||
'identifiers': {(DOMAIN, self.unique_id)},
|
||||
'name': self.name,
|
||||
'model': self.device.type,
|
||||
'via_hub': (DOMAIN, self.device.site_id),
|
||||
# For the moment, Somfy only returns their own device.
|
||||
'manufacturer': 'Somfy'
|
||||
}
|
||||
|
||||
async def async_update(self):
|
||||
"""Update the device with the latest data."""
|
||||
await update_all_devices(self.hass)
|
||||
devices = self.hass.data[DOMAIN][DEVICES]
|
||||
self.device = next((d for d in devices if d.id == self.device.id),
|
||||
self.device)
|
||||
|
||||
def has_capability(self, capability):
|
||||
"""Test if device has a capability."""
|
||||
capabilities = self.device.capabilities
|
||||
return bool([c for c in capabilities if c.name == capability])
|
||||
|
||||
|
||||
@Throttle(MIN_TIME_BETWEEN_UPDATES)
|
||||
async def update_all_devices(hass):
|
||||
"""Update all the devices."""
|
||||
from requests import HTTPError
|
||||
try:
|
||||
data = hass.data[DOMAIN]
|
||||
data[DEVICES] = await hass.async_add_executor_job(
|
||||
data[API].get_devices)
|
||||
except HTTPError:
|
||||
_LOGGER.warning("Cannot update devices")
|
||||
return False
|
||||
return True
|
||||
|
|
@ -0,0 +1,146 @@
|
|||
"""Config flow for Somfy."""
|
||||
import asyncio
|
||||
import logging
|
||||
|
||||
import async_timeout
|
||||
|
||||
from homeassistant import config_entries
|
||||
from homeassistant.components.http import HomeAssistantView
|
||||
from homeassistant.core import callback
|
||||
from .const import CLIENT_ID, CLIENT_SECRET, DOMAIN
|
||||
|
||||
AUTH_CALLBACK_PATH = '/auth/somfy/callback'
|
||||
AUTH_CALLBACK_NAME = 'auth:somfy:callback'
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@callback
|
||||
def register_flow_implementation(hass, client_id, client_secret):
|
||||
"""Register a flow implementation.
|
||||
|
||||
client_id: Client id.
|
||||
client_secret: Client secret.
|
||||
"""
|
||||
hass.data[DOMAIN][CLIENT_ID] = client_id
|
||||
hass.data[DOMAIN][CLIENT_SECRET] = client_secret
|
||||
|
||||
|
||||
@config_entries.HANDLERS.register('somfy')
|
||||
class SomfyFlowHandler(config_entries.ConfigFlow):
|
||||
"""Handle a config flow."""
|
||||
|
||||
VERSION = 1
|
||||
CONNECTION_CLASS = config_entries.CONN_CLASS_CLOUD_POLL
|
||||
|
||||
def __init__(self):
|
||||
"""Instantiate config flow."""
|
||||
self.code = None
|
||||
|
||||
async def async_step_import(self, user_input=None):
|
||||
"""Handle external yaml configuration."""
|
||||
if self.hass.config_entries.async_entries(DOMAIN):
|
||||
return self.async_abort(reason='already_setup')
|
||||
return await self.async_step_auth()
|
||||
|
||||
async def async_step_user(self, user_input=None):
|
||||
"""Handle a flow start."""
|
||||
if self.hass.config_entries.async_entries(DOMAIN):
|
||||
return self.async_abort(reason='already_setup')
|
||||
|
||||
if DOMAIN not in self.hass.data:
|
||||
return self.async_abort(reason='missing_configuration')
|
||||
|
||||
return await self.async_step_auth()
|
||||
|
||||
async def async_step_auth(self, user_input=None):
|
||||
"""Create an entry for auth."""
|
||||
# Flow has been triggered from Somfy website
|
||||
if user_input:
|
||||
return await self.async_step_code(user_input)
|
||||
|
||||
try:
|
||||
with async_timeout.timeout(10):
|
||||
url, _ = await self._get_authorization_url()
|
||||
except asyncio.TimeoutError:
|
||||
return self.async_abort(reason='authorize_url_timeout')
|
||||
|
||||
return self.async_external_step(
|
||||
step_id='auth',
|
||||
url=url
|
||||
)
|
||||
|
||||
async def _get_authorization_url(self):
|
||||
"""Get Somfy authorization url."""
|
||||
from pymfy.api.somfy_api import SomfyApi
|
||||
client_id = self.hass.data[DOMAIN][CLIENT_ID]
|
||||
client_secret = self.hass.data[DOMAIN][CLIENT_SECRET]
|
||||
redirect_uri = '{}{}'.format(
|
||||
self.hass.config.api.base_url, AUTH_CALLBACK_PATH)
|
||||
api = SomfyApi(client_id, client_secret, redirect_uri)
|
||||
|
||||
self.hass.http.register_view(SomfyAuthCallbackView())
|
||||
# Thanks to the state, we can forward the flow id to Somfy that will
|
||||
# add it in the callback.
|
||||
return await self.hass.async_add_executor_job(
|
||||
api.get_authorization_url, self.flow_id)
|
||||
|
||||
async def async_step_code(self, code):
|
||||
"""Received code for authentication."""
|
||||
self.code = code
|
||||
return self.async_external_step_done(next_step_id="creation")
|
||||
|
||||
async def async_step_creation(self, user_input=None):
|
||||
"""Create Somfy api and entries."""
|
||||
client_id = self.hass.data[DOMAIN][CLIENT_ID]
|
||||
client_secret = self.hass.data[DOMAIN][CLIENT_SECRET]
|
||||
code = self.code
|
||||
from pymfy.api.somfy_api import SomfyApi
|
||||
redirect_uri = '{}{}'.format(
|
||||
self.hass.config.api.base_url, AUTH_CALLBACK_PATH)
|
||||
api = SomfyApi(client_id, client_secret, redirect_uri)
|
||||
token = await self.hass.async_add_executor_job(api.request_token, None,
|
||||
code)
|
||||
_LOGGER.info('Successfully authenticated Somfy')
|
||||
return self.async_create_entry(
|
||||
title='Somfy',
|
||||
data={
|
||||
'token': token,
|
||||
'refresh_args': {
|
||||
'client_id': client_id,
|
||||
'client_secret': client_secret
|
||||
}
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
class SomfyAuthCallbackView(HomeAssistantView):
|
||||
"""Somfy Authorization Callback View."""
|
||||
|
||||
requires_auth = False
|
||||
url = AUTH_CALLBACK_PATH
|
||||
name = AUTH_CALLBACK_NAME
|
||||
|
||||
@staticmethod
|
||||
async def get(request):
|
||||
"""Receive authorization code."""
|
||||
from aiohttp import web_response
|
||||
|
||||
if 'code' not in request.query or 'state' not in request.query:
|
||||
return web_response.Response(
|
||||
text="Missing code or state parameter in " + request.url
|
||||
)
|
||||
|
||||
hass = request.app['hass']
|
||||
hass.async_create_task(
|
||||
hass.config_entries.flow.async_configure(
|
||||
flow_id=request.query['state'],
|
||||
user_input=request.query['code'],
|
||||
))
|
||||
|
||||
return web_response.Response(
|
||||
headers={
|
||||
'content-type': 'text/html'
|
||||
},
|
||||
text="<script>window.close()</script>"
|
||||
)
|
||||
|
|
@ -0,0 +1,5 @@
|
|||
"""Define constants for the Somfy component."""
|
||||
|
||||
DOMAIN = 'somfy'
|
||||
CLIENT_ID = 'client_id'
|
||||
CLIENT_SECRET = 'client_secret'
|
||||
|
|
@ -0,0 +1,114 @@
|
|||
"""
|
||||
Support for Somfy Covers.
|
||||
|
||||
For more details about this platform, please refer to the documentation at
|
||||
https://home-assistant.io/components/cover.somfy/
|
||||
"""
|
||||
|
||||
from homeassistant.components.cover import CoverDevice, ATTR_POSITION, \
|
||||
ATTR_TILT_POSITION
|
||||
from homeassistant.components.somfy import DOMAIN, SomfyEntity, DEVICES, API
|
||||
|
||||
|
||||
async def async_setup_entry(hass, config_entry, async_add_entities):
|
||||
"""Set up the Somfy cover platform."""
|
||||
def get_covers():
|
||||
"""Retrieve covers."""
|
||||
from pymfy.api.devices.category import Category
|
||||
|
||||
categories = {Category.ROLLER_SHUTTER.value,
|
||||
Category.INTERIOR_BLIND.value,
|
||||
Category.EXTERIOR_BLIND.value}
|
||||
|
||||
devices = hass.data[DOMAIN][DEVICES]
|
||||
|
||||
return [SomfyCover(cover, hass.data[DOMAIN][API]) for cover in
|
||||
devices if
|
||||
categories & set(cover.categories)]
|
||||
|
||||
async_add_entities(await hass.async_add_executor_job(get_covers), True)
|
||||
|
||||
|
||||
async def async_setup_platform(hass, config, async_add_entities,
|
||||
discovery_info=None):
|
||||
"""Old way of setting up platform.
|
||||
|
||||
Can only be called when a user accidentally mentions the platform in their
|
||||
config. But even in that case it would have been ignored.
|
||||
"""
|
||||
pass
|
||||
|
||||
|
||||
class SomfyCover(SomfyEntity, CoverDevice):
|
||||
"""Representation of a Somfy cover device."""
|
||||
|
||||
def __init__(self, device, api):
|
||||
"""Initialize the Somfy device."""
|
||||
from pymfy.api.devices.blind import Blind
|
||||
super().__init__(device, api)
|
||||
self.cover = Blind(self.device, self.api)
|
||||
|
||||
async def async_update(self):
|
||||
"""Update the device with the latest data."""
|
||||
from pymfy.api.devices.blind import Blind
|
||||
await super().async_update()
|
||||
self.cover = Blind(self.device, self.api)
|
||||
|
||||
def close_cover(self, **kwargs):
|
||||
"""Close the cover."""
|
||||
self.cover.close()
|
||||
|
||||
def open_cover(self, **kwargs):
|
||||
"""Open the cover."""
|
||||
self.cover.open()
|
||||
|
||||
def stop_cover(self, **kwargs):
|
||||
"""Stop the cover."""
|
||||
self.cover.stop()
|
||||
|
||||
def set_cover_position(self, **kwargs):
|
||||
"""Move the cover shutter to a specific position."""
|
||||
self.cover.set_position(100 - kwargs[ATTR_POSITION])
|
||||
|
||||
@property
|
||||
def current_cover_position(self):
|
||||
"""Return the current position of cover shutter."""
|
||||
position = None
|
||||
if self.has_capability('position'):
|
||||
position = 100 - self.cover.get_position()
|
||||
return position
|
||||
|
||||
@property
|
||||
def is_closed(self):
|
||||
"""Return if the cover is closed."""
|
||||
is_closed = None
|
||||
if self.has_capability('position'):
|
||||
is_closed = self.cover.is_closed()
|
||||
return is_closed
|
||||
|
||||
@property
|
||||
def current_cover_tilt_position(self):
|
||||
"""Return current position of cover tilt.
|
||||
|
||||
None is unknown, 0 is closed, 100 is fully open.
|
||||
"""
|
||||
orientation = None
|
||||
if self.has_capability('rotation'):
|
||||
orientation = 100 - self.cover.orientation
|
||||
return orientation
|
||||
|
||||
def set_cover_tilt_position(self, **kwargs):
|
||||
"""Move the cover tilt to a specific position."""
|
||||
self.cover.orientation = kwargs[ATTR_TILT_POSITION]
|
||||
|
||||
def open_cover_tilt(self, **kwargs):
|
||||
"""Open the cover tilt."""
|
||||
self.cover.orientation = 100
|
||||
|
||||
def close_cover_tilt(self, **kwargs):
|
||||
"""Close the cover tilt."""
|
||||
self.cover.orientation = 0
|
||||
|
||||
def stop_cover_tilt(self, **kwargs):
|
||||
"""Stop the cover."""
|
||||
self.cover.stop()
|
||||
|
|
@ -0,0 +1,13 @@
|
|||
{
|
||||
"domain": "somfy",
|
||||
"name": "Somfy Open API",
|
||||
"config_flow": true,
|
||||
"documentation": "https://www.home-assistant.io/components/somfy",
|
||||
"dependencies": [],
|
||||
"codeowners": [
|
||||
"@tetienne"
|
||||
],
|
||||
"requirements": [
|
||||
"pymfy==0.5.2"
|
||||
]
|
||||
}
|
||||
|
|
@ -0,0 +1,13 @@
|
|||
{
|
||||
"config": {
|
||||
"abort": {
|
||||
"already_setup": "You can only configure one Somfy account.",
|
||||
"authorize_url_timeout": "Timeout generating authorize url.",
|
||||
"missing_configuration": "The Somfy component is not configured. Please follow the documentation."
|
||||
},
|
||||
"create_entry": {
|
||||
"default": "Successfully authenticated with Somfy."
|
||||
},
|
||||
"title": "Somfy"
|
||||
}
|
||||
}
|
||||
|
|
@ -43,6 +43,7 @@ FLOWS = [
|
|||
"simplisafe",
|
||||
"smartthings",
|
||||
"smhi",
|
||||
"somfy",
|
||||
"sonos",
|
||||
"tellduslive",
|
||||
"toon",
|
||||
|
|
|
|||
|
|
@ -1210,6 +1210,9 @@ pymailgunner==1.4
|
|||
# homeassistant.components.mediaroom
|
||||
pymediaroom==0.6.4
|
||||
|
||||
# homeassistant.components.somfy
|
||||
pymfy==0.5.2
|
||||
|
||||
# homeassistant.components.xiaomi_tv
|
||||
pymitv==1.4.3
|
||||
|
||||
|
|
|
|||
|
|
@ -250,6 +250,9 @@ pyiqvia==0.2.1
|
|||
# homeassistant.components.litejet
|
||||
pylitejet==0.1
|
||||
|
||||
# homeassistant.components.somfy
|
||||
pymfy==0.5.2
|
||||
|
||||
# homeassistant.components.monoprice
|
||||
pymonoprice==0.3
|
||||
|
||||
|
|
|
|||
|
|
@ -107,6 +107,7 @@ TEST_REQUIREMENTS = (
|
|||
'pyhomematic',
|
||||
'pyiqvia',
|
||||
'pylitejet',
|
||||
'pymfy',
|
||||
'pymonoprice',
|
||||
'pynx584',
|
||||
'pyopenuv',
|
||||
|
|
|
|||
|
|
@ -0,0 +1 @@
|
|||
"""Tests for the Somfy component."""
|
||||
|
|
@ -0,0 +1,77 @@
|
|||
"""Tests for the Somfy config flow."""
|
||||
import asyncio
|
||||
from unittest.mock import Mock, patch
|
||||
|
||||
from pymfy.api.somfy_api import SomfyApi
|
||||
|
||||
from homeassistant import data_entry_flow
|
||||
from homeassistant.components.somfy import config_flow, DOMAIN
|
||||
from homeassistant.components.somfy.config_flow import \
|
||||
register_flow_implementation
|
||||
from tests.common import MockConfigEntry, mock_coro
|
||||
|
||||
CLIENT_SECRET_VALUE = "5678"
|
||||
|
||||
CLIENT_ID_VALUE = "1234"
|
||||
|
||||
AUTH_URL = 'http://somfy.com'
|
||||
|
||||
|
||||
async def test_abort_if_no_configuration(hass):
|
||||
"""Check flow abort when no configuration."""
|
||||
flow = config_flow.SomfyFlowHandler()
|
||||
flow.hass = hass
|
||||
result = await flow.async_step_user()
|
||||
assert result['type'] == data_entry_flow.RESULT_TYPE_ABORT
|
||||
assert result['reason'] == 'missing_configuration'
|
||||
|
||||
|
||||
async def test_abort_if_existing_entry(hass):
|
||||
"""Check flow abort when an entry already exist."""
|
||||
flow = config_flow.SomfyFlowHandler()
|
||||
flow.hass = hass
|
||||
MockConfigEntry(domain=DOMAIN).add_to_hass(hass)
|
||||
result = await flow.async_step_import()
|
||||
assert result['type'] == data_entry_flow.RESULT_TYPE_ABORT
|
||||
assert result['reason'] == 'already_setup'
|
||||
result = await flow.async_step_user()
|
||||
assert result['type'] == data_entry_flow.RESULT_TYPE_ABORT
|
||||
assert result['reason'] == 'already_setup'
|
||||
|
||||
|
||||
async def test_full_flow(hass):
|
||||
"""Check classic use case."""
|
||||
hass.data[DOMAIN] = {}
|
||||
register_flow_implementation(hass, CLIENT_ID_VALUE, CLIENT_SECRET_VALUE)
|
||||
flow = config_flow.SomfyFlowHandler()
|
||||
flow.hass = hass
|
||||
hass.config.api = Mock(base_url='https://example.com')
|
||||
flow._get_authorization_url = Mock(
|
||||
return_value=mock_coro((AUTH_URL, 'state')))
|
||||
result = await flow.async_step_import()
|
||||
assert result['type'] == data_entry_flow.RESULT_TYPE_EXTERNAL_STEP
|
||||
assert result['url'] == AUTH_URL
|
||||
result = await flow.async_step_auth("my_super_code")
|
||||
assert result['type'] == data_entry_flow.RESULT_TYPE_EXTERNAL_STEP_DONE
|
||||
assert result['step_id'] == 'creation'
|
||||
assert flow.code == 'my_super_code'
|
||||
with patch.object(SomfyApi, 'request_token',
|
||||
return_value={"access_token": "super_token"}):
|
||||
result = await flow.async_step_creation()
|
||||
assert result['type'] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
|
||||
assert result['data']['refresh_args'] == {
|
||||
'client_id': CLIENT_ID_VALUE,
|
||||
'client_secret': CLIENT_SECRET_VALUE
|
||||
}
|
||||
assert result['title'] == 'Somfy'
|
||||
assert result['data']['token'] == {"access_token": "super_token"}
|
||||
|
||||
|
||||
async def test_abort_if_authorization_timeout(hass):
|
||||
"""Check Somfy authorization timeout."""
|
||||
flow = config_flow.SomfyFlowHandler()
|
||||
flow.hass = hass
|
||||
flow._get_authorization_url = Mock(side_effect=asyncio.TimeoutError)
|
||||
result = await flow.async_step_auth()
|
||||
assert result['type'] == data_entry_flow.RESULT_TYPE_ABORT
|
||||
assert result['reason'] == 'authorize_url_timeout'
|
||||
Loading…
Reference in New Issue