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

Add OAuth to Neato (#44031)

Browse Source 
Co-authored-by: Martin Hjelmare <marhje52@gmail.com>
This commit is contained in:
Santobert 2020-12-16 23:39:41 +01:00 committed by GitHub
parent fd24baa1f6
commit d0ebc00684
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
17 changed files with 363 additions and 504 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.coveragerc
View File

@ -567,6 +567,8 @@ omit =
homeassistant/components/n26/*
homeassistant/components/nad/media_player.py
homeassistant/components/nanoleaf/light.py
homeassistant/components/neato/__init__.py
homeassistant/components/neato/api.py
homeassistant/components/neato/camera.py
homeassistant/components/neato/sensor.py
homeassistant/components/neato/switch.py

162
homeassistant/components/neato/__init__.py
View File

@ -3,26 +3,30 @@ import asyncio
from datetime import timedelta
import logging
from pybotvac import Account, Neato, Vorwerk
from pybotvac.exceptions import NeatoException, NeatoLoginException, NeatoRobotException
from pybotvac import Account, Neato
from pybotvac.exceptions import NeatoException
import voluptuous as vol
from homeassistant.config_entries import SOURCE_IMPORT
from homeassistant.const import CONF_PASSWORD, CONF_USERNAME
from homeassistant.config_entries import SOURCE_REAUTH, ConfigEntry
from homeassistant.const import (
CONF_CLIENT_ID,
CONF_CLIENT_SECRET,
CONF_SOURCE,
CONF_TOKEN,
)
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers import config_entry_oauth2_flow, config_validation as cv
from homeassistant.helpers.typing import ConfigType, HomeAssistantType
from homeassistant.util import Throttle
from .config_flow import NeatoConfigFlow
from . import api, config_flow
from .const import (
CONF_VENDOR,
NEATO_CONFIG,
NEATO_DOMAIN,
NEATO_LOGIN,
NEATO_MAP_DATA,
NEATO_PERSISTENT_MAPS,
NEATO_ROBOTS,
VALID_VENDORS,
)
_LOGGER = logging.getLogger(__name__)
@ -32,82 +36,74 @@ CONFIG_SCHEMA = vol.Schema(
{
NEATO_DOMAIN: vol.Schema(
{
vol.Required(CONF_USERNAME): cv.string,
vol.Required(CONF_PASSWORD): cv.string,
vol.Optional(CONF_VENDOR, default="neato"): vol.In(VALID_VENDORS),
vol.Required(CONF_CLIENT_ID): cv.string,
vol.Required(CONF_CLIENT_SECRET): cv.string,
}
)
},
extra=vol.ALLOW_EXTRA,
)
PLATFORMS = ["camera", "vacuum", "switch", "sensor"]
async def async_setup(hass, config):
async def async_setup(hass: HomeAssistantType, config: ConfigType) -> bool:
"""Set up the Neato component."""
hass.data[NEATO_DOMAIN] = {}
if NEATO_DOMAIN not in config:
# There is an entry and nothing in configuration.yaml
return True
entries = hass.config_entries.async_entries(NEATO_DOMAIN)
hass.data[NEATO_CONFIG] = config[NEATO_DOMAIN]
if entries:
# There is an entry and something in the configuration.yaml
entry = entries[0]
conf = config[NEATO_DOMAIN]
if (
entry.data[CONF_USERNAME] == conf[CONF_USERNAME]
and entry.data[CONF_PASSWORD] == conf[CONF_PASSWORD]
and entry.data[CONF_VENDOR] == conf[CONF_VENDOR]
):
# The entry is not outdated
return True
# The entry is outdated
error = await hass.async_add_executor_job(
NeatoConfigFlow.try_login,
conf[CONF_USERNAME],
conf[CONF_PASSWORD],
conf[CONF_VENDOR],
)
if error is not None:
_LOGGER.error(error)
return False
# Update the entry
hass.config_entries.async_update_entry(entry, data=config[NEATO_DOMAIN])
else:
# Create the new entry
hass.async_create_task(
hass.config_entries.flow.async_init(
NEATO_DOMAIN,
context={"source": SOURCE_IMPORT},
data=config[NEATO_DOMAIN],
)
)
vendor = Neato()
config_flow.OAuth2FlowHandler.async_register_implementation(
hass,
api.NeatoImplementation(
hass,
NEATO_DOMAIN,
config[NEATO_DOMAIN][CONF_CLIENT_ID],
config[NEATO_DOMAIN][CONF_CLIENT_SECRET],
vendor.auth_endpoint,
vendor.token_endpoint,
),
)
return True
async def async_setup_entry(hass, entry):
async def async_setup_entry(hass: HomeAssistantType, entry: ConfigEntry) -> bool:
"""Set up config entry."""
hub = NeatoHub(hass, entry.data, Account)
await hass.async_add_executor_job(hub.login)
if not hub.logged_in:
_LOGGER.debug("Failed to login to Neato API")
if CONF_TOKEN not in entry.data:
# Init reauth flow
hass.async_create_task(
hass.config_entries.flow.async_init(
NEATO_DOMAIN,
context={CONF_SOURCE: SOURCE_REAUTH},
)
)
return False
implementation = (
await config_entry_oauth2_flow.async_get_config_entry_implementation(
hass, entry
)
)
session = config_entry_oauth2_flow.OAuth2Session(hass, entry, implementation)
neato_session = api.ConfigEntryAuth(hass, entry, session)
hass.data[NEATO_DOMAIN][entry.entry_id] = neato_session
hub = NeatoHub(hass, Account(neato_session))
try:
await hass.async_add_executor_job(hub.update_robots)
except NeatoRobotException as ex:
except NeatoException as ex:
_LOGGER.debug("Failed to connect to Neato API")
raise ConfigEntryNotReady from ex
hass.data[NEATO_LOGIN] = hub
for component in ("camera", "vacuum", "switch", "sensor"):
for component in PLATFORMS:
hass.async_create_task(
hass.config_entries.async_forward_entry_setup(entry, component)
)
@ -115,53 +111,27 @@ async def async_setup_entry(hass, entry):
return True
async def async_unload_entry(hass, entry):
async def async_unload_entry(hass: HomeAssistantType, entry: ConfigType) -> bool:
"""Unload config entry."""
hass.data.pop(NEATO_LOGIN)
await asyncio.gather(
hass.config_entries.async_forward_entry_unload(entry, "camera"),
hass.config_entries.async_forward_entry_unload(entry, "vacuum"),
hass.config_entries.async_forward_entry_unload(entry, "switch"),
hass.config_entries.async_forward_entry_unload(entry, "sensor"),
unload_functions = (
hass.config_entries.async_forward_entry_unload(entry, platform)
for platform in PLATFORMS
)
return True
unload_ok = all(await asyncio.gather(*unload_functions))
if unload_ok:
hass.data[NEATO_DOMAIN].pop(entry.entry_id)
return unload_ok
class NeatoHub:
"""A My Neato hub wrapper class."""
def __init__(self, hass, domain_config, neato):
def __init__(self, hass: HomeAssistantType, neato: Account):
"""Initialize the Neato hub."""
self.config = domain_config
self._neato = neato
self._hass = hass
if self.config[CONF_VENDOR] == "vorwerk":
self._vendor = Vorwerk()
else: # Neato
self._vendor = Neato()
self.my_neato = None
self.logged_in = False
def login(self):
"""Login to My Neato."""
_LOGGER.debug("Trying to connect to Neato API")
try:
self.my_neato = self._neato(
self.config[CONF_USERNAME], self.config[CONF_PASSWORD], self._vendor
)
except NeatoException as ex:
if isinstance(ex, NeatoLoginException):
_LOGGER.error("Invalid credentials")
else:
_LOGGER.error("Unable to connect to Neato API")
raise ConfigEntryNotReady from ex
self.logged_in = False
return
self.logged_in = True
_LOGGER.debug("Successfully connected to Neato API")
self._hass: HomeAssistantType = hass
self.my_neato: Account = neato
@Throttle(timedelta(minutes=1))
def update_robots(self):

55
homeassistant/components/neato/api.py Normal file
View File

@ -0,0 +1,55 @@
"""API for Neato Botvac bound to Home Assistant OAuth."""
from asyncio import run_coroutine_threadsafe
import logging
import pybotvac
from homeassistant import config_entries, core
from homeassistant.helpers import config_entry_oauth2_flow
_LOGGER = logging.getLogger(__name__)
class ConfigEntryAuth(pybotvac.OAuthSession):
"""Provide Neato Botvac authentication tied to an OAuth2 based config entry."""
def __init__(
self,
hass: core.HomeAssistant,
config_entry: config_entries.ConfigEntry,
implementation: config_entry_oauth2_flow.AbstractOAuth2Implementation,
):
"""Initialize Neato Botvac Auth."""
self.hass = hass
self.session = config_entry_oauth2_flow.OAuth2Session(
hass, config_entry, implementation
)
super().__init__(self.session.token, vendor=pybotvac.Neato())
def refresh_tokens(self) -> str:
"""Refresh and return new Neato Botvac tokens using Home Assistant OAuth2 session."""
run_coroutine_threadsafe(
self.session.async_ensure_token_valid(), self.hass.loop
).result()
return self.session.token["access_token"]
class NeatoImplementation(config_entry_oauth2_flow.LocalOAuth2Implementation):
"""Neato implementation of LocalOAuth2Implementation.
We need this class because we have to add client_secret and scope to the authorization request.
"""
@property
def extra_authorize_data(self) -> dict:
"""Extra data that needs to be appended to the authorize url."""
return {"client_secret": self.client_secret}
async def async_generate_authorize_url(self, flow_id: str) -> str:
"""Generate a url for the user to authorize.
We must make sure that the plus signs are not encoded.
"""
url = await super().async_generate_authorize_url(flow_id)
return f"{url}&scope=public_profile+control_robots+maps"

2
homeassistant/components/neato/camera.py
View File

@ -45,7 +45,7 @@ class NeatoCleaningMap(Camera):
self.robot = robot
self.neato = neato
self._mapdata = mapdata
self._available = self.neato.logged_in if self.neato is not None else False
self._available = neato is not None
self._robot_name = f"{self.robot.name} Cleaning Map"
self._robot_serial = self.robot.serial
self._generated_at = None

137
homeassistant/components/neato/config_flow.py
View File

@ -1,112 +1,65 @@
"""Config flow to configure Neato integration."""
"""Config flow for Neato Botvac."""
import logging
from typing import Optional
from pybotvac import Account, Neato, Vorwerk
from pybotvac.exceptions import NeatoLoginException, NeatoRobotException
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.const import CONF_PASSWORD, CONF_USERNAME
from homeassistant.const import CONF_TOKEN
from homeassistant.helpers import config_entry_oauth2_flow
# pylint: disable=unused-import
from .const import CONF_VENDOR, NEATO_DOMAIN, VALID_VENDORS
DOCS_URL = "https://www.home-assistant.io/integrations/neato"
DEFAULT_VENDOR = "neato"
from .const import NEATO_DOMAIN
_LOGGER = logging.getLogger(__name__)
class NeatoConfigFlow(config_entries.ConfigFlow, domain=NEATO_DOMAIN):
"""Neato integration config flow."""
class OAuth2FlowHandler(
config_entry_oauth2_flow.AbstractOAuth2FlowHandler, domain=NEATO_DOMAIN
):
"""Config flow to handle Neato Botvac OAuth2 authentication."""
VERSION = 1
DOMAIN = NEATO_DOMAIN
CONNECTION_CLASS = config_entries.CONN_CLASS_CLOUD_POLL
def __init__(self):
"""Initialize flow."""
self._username = vol.UNDEFINED
self._password = vol.UNDEFINED
self._vendor = vol.UNDEFINED
@property
def logger(self) -> logging.Logger:
"""Return logger."""
return logging.getLogger(__name__)
async def async_step_user(self, user_input=None):
"""Handle a flow initialized by the user."""
errors = {}
if self._async_current_entries():
async def async_step_user(self, user_input: Optional[dict] = None) -> dict:
"""Create an entry for the flow."""
current_entries = self._async_current_entries()
if current_entries and CONF_TOKEN in current_entries[0].data:
# Already configured
return self.async_abort(reason="already_configured")
if user_input is not None:
self._username = user_input["username"]
self._password = user_input["password"]
self._vendor = user_input["vendor"]
return await super().async_step_user(user_input=user_input)
error = await self.hass.async_add_executor_job(
self.try_login, self._username, self._password, self._vendor
async def async_step_reauth(self, data) -> dict:
"""Perform reauth upon migration of old entries."""
return await self.async_step_reauth_confirm()
async def async_step_reauth_confirm(
self, user_input: Optional[dict] = None
) -> dict:
"""Confirm reauth upon migration of old entries."""
if user_input is None:
return self.async_show_form(
step_id="reauth_confirm", data_schema=vol.Schema({})
)
if error:
errors["base"] = error
else:
return self.async_create_entry(
title=user_input[CONF_USERNAME],
data=user_input,
description_placeholders={"docs_url": DOCS_URL},
)
return await self.async_step_user()
return self.async_show_form(
step_id="user",
data_schema=vol.Schema(
{
vol.Required(CONF_USERNAME): str,
vol.Required(CONF_PASSWORD): str,
vol.Optional(CONF_VENDOR, default="neato"): vol.In(VALID_VENDORS),
}
),
description_placeholders={"docs_url": DOCS_URL},
errors=errors,
)
async def async_step_import(self, user_input):
"""Import a config flow from configuration."""
if self._async_current_entries():
return self.async_abort(reason="already_configured")
username = user_input[CONF_USERNAME]
password = user_input[CONF_PASSWORD]
vendor = user_input[CONF_VENDOR]
error = await self.hass.async_add_executor_job(
self.try_login, username, password, vendor
)
if error is not None:
_LOGGER.error(error)
return self.async_abort(reason=error)
return self.async_create_entry(
title=f"{username} (from configuration)",
data={
CONF_USERNAME: username,
CONF_PASSWORD: password,
CONF_VENDOR: vendor,
},
)
@staticmethod
def try_login(username, password, vendor):
"""Try logging in to device and return any errors."""
this_vendor = None
if vendor == "vorwerk":
this_vendor = Vorwerk()
else: # Neato
this_vendor = Neato()
try:
Account(username, password, this_vendor)
except NeatoLoginException:
return "invalid_auth"
except NeatoRobotException:
return "unknown"
return None
async def async_oauth_create_entry(self, data: dict) -> dict:
"""Create an entry for the flow. Update an entry if one already exist."""
current_entries = self._async_current_entries()
if current_entries and CONF_TOKEN not in current_entries[0].data:
# Update entry
self.hass.config_entries.async_update_entry(
current_entries[0], title=self.flow_impl.name, data=data
)
self.hass.async_create_task(
self.hass.config_entries.async_reload(current_entries[0].entry_id)
)
return self.async_abort(reason="reauth_successful")
return self.async_create_entry(title=self.flow_impl.name, data=data)

2
homeassistant/components/neato/const.py
View File

@ -11,8 +11,6 @@ NEATO_ROBOTS = "neato_robots"
SCAN_INTERVAL_MINUTES = 1
VALID_VENDORS = ["neato", "vorwerk"]
MODE = {1: "Eco", 2: "Turbo"}
ACTION = {

14
homeassistant/components/neato/manifest.json
View File

@ -3,6 +3,14 @@
"name": "Neato Botvac",
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/neato",
"requirements": ["pybotvac==0.0.17"],
"codeowners": ["@dshokouhi", "@Santobert"]
}
"requirements": [
"pybotvac==0.0.19"
],
"codeowners": [
"@dshokouhi",
"@Santobert"
],
"dependencies": [
"http"
]
}

2
homeassistant/components/neato/sensor.py
View File

@ -37,7 +37,7 @@ class NeatoSensor(Entity):
def __init__(self, neato, robot):
"""Initialize Neato sensor."""
self.robot = robot
self._available = neato.logged_in if neato is not None else False
self._available = neato is not None
self._robot_name = f"{self.robot.name} {BATTERY}"
self._robot_serial = self.robot.serial
self._state = None

31
homeassistant/components/neato/strings.json
View File

@ -1,26 +1,23 @@
{
"config": {
"step": {
"user": {
"title": "Neato Account Info",
"data": {
"username": "[%key:common::config_flow::data::username%]",
"password": "[%key:common::config_flow::data::password%]",
"vendor": "Vendor"
},
"description": "See [Neato documentation]({docs_url})."
"pick_implementation": {
"title": "[%key:common::config_flow::title::oauth2_pick_implementation%]"
},
"reauth_confirm": {
"title": "[%key:common::config_flow::description::confirm_setup%]"
}
},
"error": {
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
"unknown": "[%key:common::config_flow::error::unknown%]"
"abort": {
"missing_configuration": "[%key:common::config_flow::abort::oauth2_missing_configuration%]",
"authorize_url_timeout": "[%key:common::config_flow::abort::oauth2_authorize_url_timeout%]",
"no_url_available": "[%key:common::config_flow::abort::oauth2_no_url_available%]",
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]",
"reauth_successful": "[%key:common::config_flow::abort::reauth_successful%]"
},
"create_entry": {
"default": "See [Neato documentation]({docs_url})."
},
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]",
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]"
"default": "[%key:common::config_flow::create_entry::authenticated%]"
}
}
},
"title": "Neato Botvac"
}

2
homeassistant/components/neato/switch.py
View File

@ -40,7 +40,7 @@ class NeatoConnectedSwitch(ToggleEntity):
"""Initialize the Neato Connected switches."""
self.type = switch_type
self.robot = robot
self._available = neato.logged_in if neato is not None else False
self._available = neato is not None
self._robot_name = f"{self.robot.name} {SWITCH_TYPES[self.type][0]}"
self._state = None
self._schedule_state = None

24
homeassistant/components/neato/translations/de.json
View File

@ -1,21 +1,23 @@
{
"config": {
"abort": {
"already_configured": "Bereits konfiguriert"
"already_configured": "Konto ist bereits konfiguriert.",
"authorize_url_timeout": "Timeout beim Erzeugen der Autorisierungs-URL.",
"missing_configuration": "Die Komponente ist nicht konfiguriert. Bitte beachten Sie die Dokumentation.",
"no_url_available": "Keine URL verfügbar. Informationen zu diesem Fehler finden Sie [im Hilfebereich]({docs_url})",
"reauth_successful": "Re-Authentifizierung war erfolgreich"
},
"create_entry": {
"default": "Siehe [Neato-Dokumentation]({docs_url})."
"default": "Erfolgreich authentifiziert"
},
"step": {
"user": {
"data": {
"password": "Passwort",
"username": "Benutzername",
"vendor": "Hersteller"
},
"description": "Siehe [Neato-Dokumentation]({docs_url}).",
"title": "Neato-Kontoinformationen"
"pick_implementation": {
"title": "Authentifizierungsmethode auswählen"
},
"reauth_confirm": {
"title": "Einrichtung bestätigen?"
}
}
}
},
"title": "Neato Botvac"
}

29
homeassistant/components/neato/translations/en.json
View File

@ -1,26 +1,23 @@
{
"config": {
"abort": {
"already_configured": "Device is already configured",
"invalid_auth": "Invalid authentication"
"already_configured": "Account is already configured.",
"authorize_url_timeout": "Timeout generating authorize URL.",
"missing_configuration": "The component is not configured. Please follow the documentation.",
"no_url_available": "No URL available. For information about this error, [check the help section]({docs_url})",
"reauth_successful": "Re-authentication was successful"
},
"create_entry": {
"default": "See [Neato documentation]({docs_url})."
},
"error": {
"invalid_auth": "Invalid authentication",
"unknown": "Unexpected error"
"default": "Successfully authenticated"
},
"step": {
"user": {
"data": {
"password": "Password",
"username": "Username",
"vendor": "Vendor"
},
"description": "See [Neato documentation]({docs_url}).",
"title": "Neato Account Info"
"pick_implementation": {
"title": "Pick Authentication Method"
},
"reauth_confirm": {
"title": "Confirm setup?"
}
}
}
},
"title": "Neato Botvac"
}

5
homeassistant/components/neato/vacuum.py
View File

@ -24,7 +24,7 @@ from homeassistant.components.vacuum import (
SUPPORT_STOP,
StateVacuumEntity,
)
from homeassistant.const import ATTR_ENTITY_ID, ATTR_MODE
from homeassistant.const import ATTR_MODE
from homeassistant.helpers import config_validation as cv, entity_platform
from .const import (
@ -93,7 +93,6 @@ async def async_setup_entry(hass, entry, async_add_entities):
platform.async_register_entity_service(
"custom_cleaning",
{
vol.Required(ATTR_ENTITY_ID): cv.entity_ids,
vol.Optional(ATTR_MODE, default=2): cv.positive_int,
vol.Optional(ATTR_NAVIGATION, default=1): cv.positive_int,
vol.Optional(ATTR_CATEGORY, default=4): cv.positive_int,
@ -109,7 +108,7 @@ class NeatoConnectedVacuum(StateVacuumEntity):
def __init__(self, neato, robot, mapdata, persistent_maps):
"""Initialize the Neato Connected Vacuum."""
self.robot = robot
self._available = neato.logged_in if neato is not None else False
self._available = neato is not None
self._mapdata = mapdata
self._name = f"{self.robot.name}"
self._robot_has_map = self.robot.has_persistent_maps

2
requirements_all.txt
View File

@ -1292,7 +1292,7 @@ pyblackbird==0.5
# pybluez==0.22
# homeassistant.components.neato
pybotvac==0.0.17
pybotvac==0.0.19
# homeassistant.components.nissan_leaf
pycarwings2==2.9

2
requirements_test_all.txt
View File

@ -655,7 +655,7 @@ pyatv==0.7.5
pyblackbird==0.5
# homeassistant.components.neato
pybotvac==0.0.17
pybotvac==0.0.19
# homeassistant.components.cloudflare
pycfdns==1.2.1

278
tests/components/neato/test_config_flow.py
View File

@ -1,160 +1,156 @@
"""Tests for the Neato config flow."""
from pybotvac.exceptions import NeatoLoginException, NeatoRobotException
import pytest
"""Test the Neato Botvac config flow."""
from pybotvac.neato import Neato
from homeassistant import data_entry_flow
from homeassistant.components.neato import config_flow
from homeassistant.components.neato.const import CONF_VENDOR, NEATO_DOMAIN
from homeassistant.const import CONF_PASSWORD, CONF_USERNAME
from homeassistant import config_entries, data_entry_flow, setup
from homeassistant.components.neato.const import NEATO_DOMAIN
from homeassistant.helpers import config_entry_oauth2_flow
from homeassistant.helpers.typing import HomeAssistantType
from tests.async_mock import patch
from tests.common import MockConfigEntry
USERNAME = "myUsername"
PASSWORD = "myPassword"
VENDOR_NEATO = "neato"
VENDOR_VORWERK = "vorwerk"
VENDOR_INVALID = "invalid"
CLIENT_ID = "1234"
CLIENT_SECRET = "5678"
VENDOR = Neato()
OAUTH2_AUTHORIZE = VENDOR.auth_endpoint
OAUTH2_TOKEN = VENDOR.token_endpoint
@pytest.fixture(name="account")
def mock_controller_login():
"""Mock a successful login."""
with patch("homeassistant.components.neato.config_flow.Account", return_value=True):
yield
def init_config_flow(hass):
"""Init a configuration flow."""
flow = config_flow.NeatoConfigFlow()
flow.hass = hass
return flow
async def test_user(hass, account):
"""Test user config."""
flow = init_config_flow(hass)
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(
{CONF_USERNAME: USERNAME, CONF_PASSWORD: PASSWORD, CONF_VENDOR: VENDOR_NEATO}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result["title"] == USERNAME
assert result["data"][CONF_USERNAME] == USERNAME
assert result["data"][CONF_PASSWORD] == PASSWORD
assert result["data"][CONF_VENDOR] == VENDOR_NEATO
result = await flow.async_step_user(
{CONF_USERNAME: USERNAME, CONF_PASSWORD: PASSWORD, CONF_VENDOR: VENDOR_VORWERK}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result["title"] == USERNAME
assert result["data"][CONF_USERNAME] == USERNAME
assert result["data"][CONF_PASSWORD] == PASSWORD
assert result["data"][CONF_VENDOR] == VENDOR_VORWERK
async def test_import(hass, account):
"""Test import step."""
flow = init_config_flow(hass)
result = await flow.async_step_import(
{CONF_USERNAME: USERNAME, CONF_PASSWORD: PASSWORD, CONF_VENDOR: VENDOR_NEATO}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result["title"] == f"{USERNAME} (from configuration)"
assert result["data"][CONF_USERNAME] == USERNAME
assert result["data"][CONF_PASSWORD] == PASSWORD
assert result["data"][CONF_VENDOR] == VENDOR_NEATO
async def test_abort_if_already_setup(hass, account):
"""Test we abort if Neato is already setup."""
flow = init_config_flow(hass)
MockConfigEntry(
domain=NEATO_DOMAIN,
data={
CONF_USERNAME: USERNAME,
CONF_PASSWORD: PASSWORD,
CONF_VENDOR: VENDOR_NEATO,
async def test_full_flow(
hass, aiohttp_client, aioclient_mock, current_request_with_host
):
"""Check full flow."""
assert await setup.async_setup_component(
hass,
"neato",
{
"neato": {"client_id": CLIENT_ID, "client_secret": CLIENT_SECRET},
"http": {"base_url": "https://example.com"},
},
)
result = await hass.config_entries.flow.async_init(
"neato", context={"source": config_entries.SOURCE_USER}
)
state = config_entry_oauth2_flow._encode_jwt(
hass,
{
"flow_id": result["flow_id"],
"redirect_uri": "https://example.com/auth/external/callback",
},
)
assert result["url"] == (
f"{OAUTH2_AUTHORIZE}?response_type=code&client_id={CLIENT_ID}"
"&redirect_uri=https://example.com/auth/external/callback"
f"&state={state}"
f"&client_secret={CLIENT_SECRET}"
"&scope=public_profile+control_robots+maps"
)
client = await aiohttp_client(hass.http.app)
resp = await client.get(f"/auth/external/callback?code=abcd&state={state}")
assert resp.status == 200
assert resp.headers["content-type"] == "text/html; charset=utf-8"
aioclient_mock.post(
OAUTH2_TOKEN,
json={
"refresh_token": "mock-refresh-token",
"access_token": "mock-access-token",
"type": "Bearer",
"expires_in": 60,
},
)
with patch(
"homeassistant.components.neato.async_setup_entry", return_value=True
) as mock_setup:
await hass.config_entries.flow.async_configure(result["flow_id"])
assert len(hass.config_entries.async_entries(NEATO_DOMAIN)) == 1
assert len(mock_setup.mock_calls) == 1
async def test_abort_if_already_setup(hass: HomeAssistantType):
"""Test we abort if Neato is already setup."""
entry = MockConfigEntry(
domain=NEATO_DOMAIN,
data={"auth_implementation": "neato", "token": {"some": "data"}},
)
entry.add_to_hass(hass)
# Should fail
result = await hass.config_entries.flow.async_init(
"neato", context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "already_configured"
async def test_reauth(
hass: HomeAssistantType, aiohttp_client, aioclient_mock, current_request_with_host
):
"""Test initialization of the reauth flow."""
assert await setup.async_setup_component(
hass,
"neato",
{
"neato": {"client_id": CLIENT_ID, "client_secret": CLIENT_SECRET},
"http": {"base_url": "https://example.com"},
},
)
MockConfigEntry(
entry_id="my_entry",
domain=NEATO_DOMAIN,
data={"username": "abcdef", "password": "123456", "vendor": "neato"},
).add_to_hass(hass)
# Should fail, same USERNAME (import)
result = await flow.async_step_import(
{CONF_USERNAME: USERNAME, CONF_PASSWORD: PASSWORD, CONF_VENDOR: VENDOR_NEATO}
# Should show form
result = await hass.config_entries.flow.async_init(
"neato", context={"source": config_entries.SOURCE_REAUTH}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "already_configured"
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "reauth_confirm"
# Should fail, same USERNAME (flow)
result = await flow.async_step_user(
{CONF_USERNAME: USERNAME, CONF_PASSWORD: PASSWORD, CONF_VENDOR: VENDOR_NEATO}
# Confirm reauth flow
result2 = await hass.config_entries.flow.async_configure(result["flow_id"], {})
state = config_entry_oauth2_flow._encode_jwt(
hass,
{
"flow_id": result["flow_id"],
"redirect_uri": "https://example.com/auth/external/callback",
},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "already_configured"
client = await aiohttp_client(hass.http.app)
resp = await client.get(f"/auth/external/callback?code=abcd&state={state}")
assert resp.status == 200
async def test_abort_on_invalid_credentials(hass):
"""Test when we have invalid credentials."""
flow = init_config_flow(hass)
aioclient_mock.post(
OAUTH2_TOKEN,
json={
"refresh_token": "mock-refresh-token",
"access_token": "mock-access-token",
"type": "Bearer",
"expires_in": 60,
},
)
# Update entry
with patch(
"homeassistant.components.neato.config_flow.Account",
side_effect=NeatoLoginException(),
):
result = await flow.async_step_user(
{
CONF_USERNAME: USERNAME,
CONF_PASSWORD: PASSWORD,
CONF_VENDOR: VENDOR_NEATO,
}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["errors"] == {"base": "invalid_auth"}
"homeassistant.components.neato.async_setup_entry", return_value=True
) as mock_setup:
result3 = await hass.config_entries.flow.async_configure(result2["flow_id"])
await hass.async_block_till_done()
result = await flow.async_step_import(
{
CONF_USERNAME: USERNAME,
CONF_PASSWORD: PASSWORD,
CONF_VENDOR: VENDOR_NEATO,
}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "invalid_auth"
new_entry = hass.config_entries.async_get_entry("my_entry")
async def test_abort_on_unexpected_error(hass):
"""Test when we have an unexpected error."""
flow = init_config_flow(hass)
with patch(
"homeassistant.components.neato.config_flow.Account",
side_effect=NeatoRobotException(),
):
result = await flow.async_step_user(
{
CONF_USERNAME: USERNAME,
CONF_PASSWORD: PASSWORD,
CONF_VENDOR: VENDOR_NEATO,
}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["errors"] == {"base": "unknown"}
result = await flow.async_step_import(
{
CONF_USERNAME: USERNAME,
CONF_PASSWORD: PASSWORD,
CONF_VENDOR: VENDOR_NEATO,
}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "unknown"
assert result3["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result3["reason"] == "reauth_successful"
assert new_entry.state == "loaded"
assert len(hass.config_entries.async_entries(NEATO_DOMAIN)) == 1
assert len(mock_setup.mock_calls) == 1

118
tests/components/neato/test_init.py
View File

@ -1,118 +0,0 @@
"""Tests for the Neato init file."""
from pybotvac.exceptions import NeatoLoginException
import pytest
from homeassistant.components.neato.const import CONF_VENDOR, NEATO_DOMAIN
from homeassistant.const import CONF_PASSWORD, CONF_USERNAME
from homeassistant.setup import async_setup_component
from tests.async_mock import patch
from tests.common import MockConfigEntry
USERNAME = "myUsername"
PASSWORD = "myPassword"
VENDOR_NEATO = "neato"
VENDOR_VORWERK = "vorwerk"
VENDOR_INVALID = "invalid"
VALID_CONFIG = {
CONF_USERNAME: USERNAME,
CONF_PASSWORD: PASSWORD,
CONF_VENDOR: VENDOR_NEATO,
}
DIFFERENT_CONFIG = {
CONF_USERNAME: "anotherUsername",
CONF_PASSWORD: "anotherPassword",
CONF_VENDOR: VENDOR_VORWERK,
}
INVALID_CONFIG = {
CONF_USERNAME: USERNAME,
CONF_PASSWORD: PASSWORD,
CONF_VENDOR: VENDOR_INVALID,
}
@pytest.fixture(name="config_flow")
def mock_config_flow_login():
"""Mock a successful login."""
with patch("homeassistant.components.neato.config_flow.Account", return_value=True):
yield
@pytest.fixture(name="hub")
def mock_controller_login():
"""Mock a successful login."""
with patch("homeassistant.components.neato.Account", return_value=True):
yield
async def test_no_config_entry(hass):
"""There is nothing in configuration.yaml."""
res = await async_setup_component(hass, NEATO_DOMAIN, {})
assert res is True
async def test_create_valid_config_entry(hass, config_flow, hub):
"""There is something in configuration.yaml."""
assert hass.config_entries.async_entries(NEATO_DOMAIN) == []
assert await async_setup_component(hass, NEATO_DOMAIN, {NEATO_DOMAIN: VALID_CONFIG})
await hass.async_block_till_done()
entries = hass.config_entries.async_entries(NEATO_DOMAIN)
assert entries
assert entries[0].data[CONF_USERNAME] == USERNAME
assert entries[0].data[CONF_PASSWORD] == PASSWORD
assert entries[0].data[CONF_VENDOR] == VENDOR_NEATO
async def test_config_entries_in_sync(hass, hub):
"""The config entry and configuration.yaml are in sync."""
MockConfigEntry(domain=NEATO_DOMAIN, data=VALID_CONFIG).add_to_hass(hass)
assert hass.config_entries.async_entries(NEATO_DOMAIN)
assert await async_setup_component(hass, NEATO_DOMAIN, {NEATO_DOMAIN: VALID_CONFIG})
await hass.async_block_till_done()
entries = hass.config_entries.async_entries(NEATO_DOMAIN)
assert entries
assert entries[0].data[CONF_USERNAME] == USERNAME
assert entries[0].data[CONF_PASSWORD] == PASSWORD
assert entries[0].data[CONF_VENDOR] == VENDOR_NEATO
async def test_config_entries_not_in_sync(hass, config_flow, hub):
"""The config entry and configuration.yaml are not in sync."""
MockConfigEntry(domain=NEATO_DOMAIN, data=DIFFERENT_CONFIG).add_to_hass(hass)
assert hass.config_entries.async_entries(NEATO_DOMAIN)
assert await async_setup_component(hass, NEATO_DOMAIN, {NEATO_DOMAIN: VALID_CONFIG})
await hass.async_block_till_done()
entries = hass.config_entries.async_entries(NEATO_DOMAIN)
assert entries
assert entries[0].data[CONF_USERNAME] == USERNAME
assert entries[0].data[CONF_PASSWORD] == PASSWORD
assert entries[0].data[CONF_VENDOR] == VENDOR_NEATO
async def test_config_entries_not_in_sync_error(hass):
"""The config entry and configuration.yaml are not in sync, the new configuration is wrong."""
MockConfigEntry(domain=NEATO_DOMAIN, data=VALID_CONFIG).add_to_hass(hass)
assert hass.config_entries.async_entries(NEATO_DOMAIN)
with patch(
"homeassistant.components.neato.config_flow.Account",
side_effect=NeatoLoginException(),
):
assert not await async_setup_component(
hass, NEATO_DOMAIN, {NEATO_DOMAIN: DIFFERENT_CONFIG}
)
await hass.async_block_till_done()
entries = hass.config_entries.async_entries(NEATO_DOMAIN)
assert entries
assert entries[0].data[CONF_USERNAME] == USERNAME
assert entries[0].data[CONF_PASSWORD] == PASSWORD
assert entries[0].data[CONF_VENDOR] == VENDOR_NEATO