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

Fix SimpliSafe to work with new MFA (#38097)

Browse Source 
* Fix SimpliSafe to work with new MFA

* Code review (part 1)

* Input needed from Martin

* Code review

* Code review

* Restore YAML

* Tests

* Code review

* Remove JSON patching in tests

* Add reauth test

* One more reauth test

* Don't abuse the word "conf"

* Update homeassistant/components/simplisafe/config_flow.py

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>

* Test coverage

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>
This commit is contained in:
Aaron Bach 2020-07-23 20:02:29 -06:00 committed by GitHub
parent 2dfd767b8c
commit a5b7a2c228
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
11 changed files with 314 additions and 116 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

77
homeassistant/components/simplisafe/__init__.py
View File

@ -1,6 +1,6 @@
"""Support for SimpliSafe alarm systems."""
import asyncio
import logging
from uuid import UUID
from simplipy import API
from simplipy.errors import InvalidCredentialsError, SimplipyError
@ -55,11 +55,10 @@ from .const import (
DATA_CLIENT,
DEFAULT_SCAN_INTERVAL,
DOMAIN,
LOGGER,
VOLUMES,
)
_LOGGER = logging.getLogger(__name__)
CONF_ACCOUNTS = "accounts"
DATA_LISTENER = "listener"
@ -161,6 +160,13 @@ def _async_save_refresh_token(hass, config_entry, token):
)
async def async_get_client_id(hass):
"""Get a client ID (based on the HASS unique ID) for the SimpliSafe API."""
hass_id = await hass.helpers.instance_id.async_get()
# SimpliSafe requires full, "dashed" versions of UUIDs:
return str(UUID(hass_id))
async def async_register_base_station(hass, system, config_entry_id):
"""Register a new bridge."""
device_registry = await dr.async_get_registry(hass)
@ -220,17 +226,18 @@ async def async_setup_entry(hass, config_entry):
_verify_domain_control = verify_domain_control(hass, DOMAIN)
client_id = await async_get_client_id(hass)
websession = aiohttp_client.async_get_clientsession(hass)
try:
api = await API.login_via_token(
config_entry.data[CONF_TOKEN], session=websession
config_entry.data[CONF_TOKEN], client_id=client_id, session=websession
)
except InvalidCredentialsError:
_LOGGER.error("Invalid credentials provided")
LOGGER.error("Invalid credentials provided")
return False
except SimplipyError as err:
_LOGGER.error("Config entry failed: %s", err)
LOGGER.error("Config entry failed: %s", err)
raise ConfigEntryNotReady
_async_save_refresh_token(hass, config_entry, api.refresh_token)
@ -252,7 +259,7 @@ async def async_setup_entry(hass, config_entry):
"""Decorate."""
system_id = int(call.data[ATTR_SYSTEM_ID])
if system_id not in simplisafe.systems:
_LOGGER.error("Unknown system ID in service call: %s", system_id)
LOGGER.error("Unknown system ID in service call: %s", system_id)
return
await coro(call)
@ -266,7 +273,7 @@ async def async_setup_entry(hass, config_entry):
"""Decorate."""
system = simplisafe.systems[int(call.data[ATTR_SYSTEM_ID])]
if system.version != 3:
_LOGGER.error("Service only available on V3 systems")
LOGGER.error("Service only available on V3 systems")
return
await coro(call)
@ -280,7 +287,7 @@ async def async_setup_entry(hass, config_entry):
try:
await system.clear_notifications()
except SimplipyError as err:
_LOGGER.error("Error during service call: %s", err)
LOGGER.error("Error during service call: %s", err)
return
@verify_system_exists
@ -291,7 +298,7 @@ async def async_setup_entry(hass, config_entry):
try:
await system.remove_pin(call.data[ATTR_PIN_LABEL_OR_VALUE])
except SimplipyError as err:
_LOGGER.error("Error during service call: %s", err)
LOGGER.error("Error during service call: %s", err)
return
@verify_system_exists
@ -302,7 +309,7 @@ async def async_setup_entry(hass, config_entry):
try:
await system.set_pin(call.data[ATTR_PIN_LABEL], call.data[ATTR_PIN_VALUE])
except SimplipyError as err:
_LOGGER.error("Error during service call: %s", err)
LOGGER.error("Error during service call: %s", err)
return
@verify_system_exists
@ -320,7 +327,7 @@ async def async_setup_entry(hass, config_entry):
}
)
except SimplipyError as err:
_LOGGER.error("Error during service call: %s", err)
LOGGER.error("Error during service call: %s", err)
return
for service, method, schema in [
@ -373,16 +380,16 @@ class SimpliSafeWebsocket:
@staticmethod
def _on_connect():
"""Define a handler to fire when the websocket is connected."""
_LOGGER.info("Connected to websocket")
LOGGER.info("Connected to websocket")
@staticmethod
def _on_disconnect():
"""Define a handler to fire when the websocket is disconnected."""
_LOGGER.info("Disconnected from websocket")
LOGGER.info("Disconnected from websocket")
def _on_event(self, event):
"""Define a handler to fire when a new SimpliSafe event arrives."""
_LOGGER.debug("New websocket event: %s", event)
LOGGER.debug("New websocket event: %s", event)
async_dispatcher_send(
self._hass, TOPIC_UPDATE_WEBSOCKET.format(event.system_id), event
)
@ -451,7 +458,7 @@ class SimpliSafe:
if not to_add:
return
_LOGGER.debug("New system notifications: %s", to_add)
LOGGER.debug("New system notifications: %s", to_add)
self._system_notifications[system.system_id].update(to_add)
@ -492,7 +499,7 @@ class SimpliSafe:
system.system_id
] = await system.get_latest_event()
except SimplipyError as err:
_LOGGER.error("Error while fetching initial event: %s", err)
LOGGER.error("Error while fetching initial event: %s", err)
self.initial_event_to_use[system.system_id] = {}
async def refresh(event_time):
@ -512,7 +519,7 @@ class SimpliSafe:
"""Update a system."""
await system.update()
self._async_process_new_notifications(system)
_LOGGER.debug('Updated REST API data for "%s"', system.address)
LOGGER.debug('Updated REST API data for "%s"', system.address)
async_dispatcher_send(
self._hass, TOPIC_UPDATE_REST_API.format(system.system_id)
)
@ -523,27 +530,37 @@ class SimpliSafe:
for result in results:
if isinstance(result, InvalidCredentialsError):
if self._emergency_refresh_token_used:
_LOGGER.error(
"SimpliSafe authentication disconnected. Please restart HASS"
LOGGER.error(
"Token disconnected or invalid. Please re-auth the "
"SimpliSafe integration in HASS"
)
remove_listener = self._hass.data[DOMAIN][DATA_LISTENER].pop(
self._config_entry.entry_id
self._hass.async_create_task(
self._hass.config_entries.flow.async_init(
DOMAIN,
context={"source": "reauth"},
data=self._config_entry.data,
)
)
remove_listener()
return
_LOGGER.warning("SimpliSafe cloud error; trying stored refresh token")
LOGGER.warning("SimpliSafe cloud error; trying stored refresh token")
self._emergency_refresh_token_used = True
return await self._api.refresh_access_token(
self._config_entry.data[CONF_TOKEN]
)
try:
await self._api.refresh_access_token(
self._config_entry.data[CONF_TOKEN]
)
return
except SimplipyError as err:
LOGGER.error("Error while using stored refresh token: %s", err)
return
if isinstance(result, SimplipyError):
_LOGGER.error("SimpliSafe error while updating: %s", result)
LOGGER.error("SimpliSafe error while updating: %s", result)
return
if isinstance(result, SimplipyError):
_LOGGER.error("Unknown error while updating: %s", result)
if isinstance(result, Exception): # pylint: disable=broad-except
LOGGER.error("Unknown error while updating: %s", result)
return
if self._api.refresh_token != self._config_entry.data[CONF_TOKEN]:

12
homeassistant/components/simplisafe/alarm_control_panel.py
View File

@ -1,5 +1,4 @@
"""Support for SimpliSafe alarm control panels."""
import logging
import re
from simplipy.errors import SimplipyError
@ -50,11 +49,10 @@ from .const import (
ATTR_VOICE_PROMPT_VOLUME,
DATA_CLIENT,
DOMAIN,
LOGGER,
VOLUME_STRING_MAP,
)
_LOGGER = logging.getLogger(__name__)
ATTR_BATTERY_BACKUP_POWER_LEVEL = "battery_backup_power_level"
ATTR_GSM_STRENGTH = "gsm_strength"
ATTR_PIN_NAME = "pin_name"
@ -146,7 +144,7 @@ class SimpliSafeAlarm(SimpliSafeEntity, AlarmControlPanelEntity):
return True
if not code or code != self._simplisafe.options[CONF_CODE]:
_LOGGER.warning(
LOGGER.warning(
"Incorrect alarm code entered (target state: %s): %s", state, code
)
return False
@ -161,7 +159,7 @@ class SimpliSafeAlarm(SimpliSafeEntity, AlarmControlPanelEntity):
try:
await self._system.set_off()
except SimplipyError as err:
_LOGGER.error('Error while disarming "%s": %s', self._system.name, err)
LOGGER.error('Error while disarming "%s": %s', self._system.name, err)
return
self._state = STATE_ALARM_DISARMED
@ -174,7 +172,7 @@ class SimpliSafeAlarm(SimpliSafeEntity, AlarmControlPanelEntity):
try:
await self._system.set_home()
except SimplipyError as err:
_LOGGER.error('Error while arming "%s" (home): %s', self._system.name, err)
LOGGER.error('Error while arming "%s" (home): %s', self._system.name, err)
return
self._state = STATE_ALARM_ARMED_HOME
@ -187,7 +185,7 @@ class SimpliSafeAlarm(SimpliSafeEntity, AlarmControlPanelEntity):
try:
await self._system.set_away()
except SimplipyError as err:
_LOGGER.error('Error while arming "%s" (away): %s', self._system.name, err)
LOGGER.error('Error while arming "%s" (away): %s', self._system.name, err)
return
self._state = STATE_ALARM_ARMING

132
homeassistant/components/simplisafe/config_flow.py
View File

@ -1,6 +1,10 @@
"""Config flow to configure the SimpliSafe component."""
from simplipy import API
from simplipy.errors import SimplipyError
from simplipy.errors import (
InvalidCredentialsError,
PendingAuthorizationError,
SimplipyError,
)
import voluptuous as vol
from homeassistant import config_entries
@ -8,7 +12,8 @@ from homeassistant.const import CONF_CODE, CONF_PASSWORD, CONF_TOKEN, CONF_USERN
from homeassistant.core import callback
from homeassistant.helpers import aiohttp_client
from .const import DOMAIN # pylint: disable=unused-import
from . import async_get_client_id
from .const import DOMAIN, LOGGER # pylint: disable=unused-import
class SimpliSafeFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
@ -19,21 +24,18 @@ class SimpliSafeFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
def __init__(self):
"""Initialize the config flow."""
self.data_schema = vol.Schema(
self.full_data_schema = vol.Schema(
{
vol.Required(CONF_USERNAME): str,
vol.Required(CONF_PASSWORD): str,
vol.Optional(CONF_CODE): str,
}
)
self.password_data_schema = vol.Schema({vol.Required(CONF_PASSWORD): str})
async def _show_form(self, errors=None):
"""Show the form to the user."""
return self.async_show_form(
step_id="user",
data_schema=self.data_schema,
errors=errors if errors else {},
)
self._code = None
self._password = None
self._username = None
@staticmethod
@callback
@ -41,34 +43,112 @@ class SimpliSafeFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
"""Define the config flow to handle options."""
return SimpliSafeOptionsFlowHandler(config_entry)
async def _async_get_simplisafe_api(self):
"""Get an authenticated SimpliSafe API client."""
client_id = await async_get_client_id(self.hass)
websession = aiohttp_client.async_get_clientsession(self.hass)
return await API.login_via_credentials(
self._username, self._password, client_id=client_id, session=websession,
)
async def _async_login_during_step(self, *, step_id, form_schema):
"""Attempt to log into the API from within a config flow step."""
errors = {}
try:
simplisafe = await self._async_get_simplisafe_api()
except PendingAuthorizationError:
LOGGER.info("Awaiting confirmation of MFA email click")
return await self.async_step_mfa()
except InvalidCredentialsError:
errors = {"base": "invalid_credentials"}
except SimplipyError as err:
LOGGER.error("Unknown error while logging into SimpliSafe: %s", err)
errors = {"base": "unknown"}
if errors:
return self.async_show_form(
step_id=step_id, data_schema=form_schema, errors=errors,
)
return await self.async_step_finish(
{
CONF_USERNAME: self._username,
CONF_TOKEN: simplisafe.refresh_token,
CONF_CODE: self._code,
}
)
async def async_step_finish(self, user_input=None):
"""Handle finish config entry setup."""
existing_entry = await self.async_set_unique_id(self._username)
if existing_entry:
self.hass.config_entries.async_update_entry(existing_entry, data=user_input)
return self.async_abort(reason="reauth_successful")
return self.async_create_entry(title=self._username, data=user_input)
async def async_step_import(self, import_config):
"""Import a config entry from configuration.yaml."""
return await self.async_step_user(import_config)
async def async_step_mfa(self, user_input=None):
"""Handle multi-factor auth confirmation."""
if user_input is None:
return self.async_show_form(step_id="mfa")
try:
simplisafe = await self._async_get_simplisafe_api()
except PendingAuthorizationError:
LOGGER.error("Still awaiting confirmation of MFA email click")
return self.async_show_form(
step_id="mfa", errors={"base": "still_awaiting_mfa"}
)
return await self.async_step_finish(
{
CONF_USERNAME: self._username,
CONF_TOKEN: simplisafe.refresh_token,
CONF_CODE: self._code,
}
)
async def async_step_reauth(self, config):
"""Handle configuration by re-auth."""
self._code = config.get(CONF_CODE)
self._username = config[CONF_USERNAME]
return await self.async_step_reauth_confirm()
async def async_step_reauth_confirm(self, user_input=None):
"""Handle re-auth completion."""
if not user_input:
return self.async_show_form(
step_id="reauth_confirm", data_schema=self.password_data_schema
)
self._password = user_input[CONF_PASSWORD]
return await self._async_login_during_step(
step_id="reauth_confirm", form_schema=self.password_data_schema
)
async def async_step_user(self, user_input=None):
"""Handle the start of the config flow."""
if not user_input:
return await self._show_form()
return self.async_show_form(
step_id="user", data_schema=self.full_data_schema
)
await self.async_set_unique_id(user_input[CONF_USERNAME])
self._abort_if_unique_id_configured()
websession = aiohttp_client.async_get_clientsession(self.hass)
self._code = user_input.get(CONF_CODE)
self._password = user_input[CONF_PASSWORD]
self._username = user_input[CONF_USERNAME]
try:
simplisafe = await API.login_via_credentials(
user_input[CONF_USERNAME], user_input[CONF_PASSWORD], session=websession
)
except SimplipyError:
return await self._show_form(errors={"base": "invalid_credentials"})
return self.async_create_entry(
title=user_input[CONF_USERNAME],
data={
CONF_USERNAME: user_input[CONF_USERNAME],
CONF_TOKEN: simplisafe.refresh_token,
CONF_CODE: user_input.get(CONF_CODE),
},
return await self._async_login_during_step(
step_id="user", form_schema=self.full_data_schema
)

3
homeassistant/components/simplisafe/const.py
View File

@ -1,8 +1,11 @@
"""Define constants for the SimpliSafe component."""
from datetime import timedelta
import logging
from simplipy.system.v3 import VOLUME_HIGH, VOLUME_LOW, VOLUME_MEDIUM, VOLUME_OFF
LOGGER = logging.getLogger(__package__)
DOMAIN = "simplisafe"
DATA_CLIENT = "client"

10
homeassistant/components/simplisafe/lock.py
View File

@ -1,6 +1,4 @@
"""Support for SimpliSafe locks."""
import logging
from simplipy.errors import SimplipyError
from simplipy.lock import LockStates
from simplipy.websocket import EVENT_LOCK_LOCKED, EVENT_LOCK_UNLOCKED
@ -9,9 +7,7 @@ from homeassistant.components.lock import LockEntity
from homeassistant.core import callback
from . import SimpliSafeEntity
from .const import DATA_CLIENT, DOMAIN
_LOGGER = logging.getLogger(__name__)
from .const import DATA_CLIENT, DOMAIN, LOGGER
ATTR_LOCK_LOW_BATTERY = "lock_low_battery"
ATTR_JAMMED = "jammed"
@ -52,7 +48,7 @@ class SimpliSafeLock(SimpliSafeEntity, LockEntity):
try:
await self._lock.lock()
except SimplipyError as err:
_LOGGER.error('Error while locking "%s": %s', self._lock.name, err)
LOGGER.error('Error while locking "%s": %s', self._lock.name, err)
return
self._is_locked = True
@ -62,7 +58,7 @@ class SimpliSafeLock(SimpliSafeEntity, LockEntity):
try:
await self._lock.unlock()
except SimplipyError as err:
_LOGGER.error('Error while unlocking "%s": %s', self._lock.name, err)
LOGGER.error('Error while unlocking "%s": %s', self._lock.name, err)
return
self._is_locked = False

2
homeassistant/components/simplisafe/manifest.json
View File

@ -3,6 +3,6 @@
"name": "SimpliSafe",
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/simplisafe",
"requirements": ["simplisafe-python==9.2.0"],
"requirements": ["simplisafe-python==9.2.1"],
"codeowners": ["@bachya"]
}

20
homeassistant/components/simplisafe/strings.json
View File

@ -1,6 +1,17 @@
{
"config": {
"step": {
"mfa": {
"title": "SimpliSafe Multi-Factor Authentication",
"description": "Check your email for a link from SimpliSafe. After verifying the link, return here to complete the installation of the integration."
},
"reauth_confirm": {
"title": "Re-link SimpliSafe Account",
"description": "Your access token has expired or been revoked. Enter your password to re-link your account.",
"data": {
"password": "[%key:common::config_flow::data::password%]"
}
},
"user": {
"title": "Fill in your information.",
"data": {
@ -12,10 +23,13 @@
},
"error": {
"identifier_exists": "Account already registered",
"invalid_credentials": "Invalid credentials"
"invalid_credentials": "Invalid credentials",
"still_awaiting_mfa": "Still awaiting MFA email click",
"unknown": "[%key:common::config_flow::error::unknown%]"
},
"abort": {
"already_configured": "This SimpliSafe account is already in use."
"already_configured": "This SimpliSafe account is already in use.",
"reauth_successful": "SimpliSafe successfully reauthenticated."
}
},
"options": {
@ -28,4 +42,4 @@
}
}
}
}
}

22
homeassistant/components/simplisafe/translations/en.json
View File

@ -1,18 +1,32 @@
{
"config": {
"abort": {
"already_configured": "This SimpliSafe account is already in use."
"already_configured": "This SimpliSafe account is already in use.",
"reauth_successful": "SimpliSafe successfully reauthenticated."
},
"error": {
"identifier_exists": "Account already registered",
"invalid_credentials": "Invalid credentials"
"invalid_credentials": "Invalid credentials",
"still_awaiting_mfa": "Still awaiting MFA email click",
"unknown": "[%key:common::config_flow::error::unknown%]"
},
"step": {
"mfa": {
"description": "Check your email for a link from SimpliSafe. After verifying the link, return here to complete the installation of the integration.",
"title": "SimpliSafe Multi-Factor Authentication"
},
"reauth_confirm": {
"data": {
"password": "[%key:common::config_flow::data::password%]"
},
"description": "Your access token has expired or been revoked. Enter your password to re-link your account.",
"title": "Re-link SimpliSafe Account"
},
"user": {
"data": {
"code": "Code (used in Home Assistant UI)",
"password": "Password",
"username": "Email"
"password": "[%key:common::config_flow::data::password%]",
"username": "[%key:common::config_flow::data::email%]"
},
"title": "Fill in your information."
}

2
requirements_all.txt
View File

@ -1951,7 +1951,7 @@ simplehound==0.3
simplepush==1.1.4
# homeassistant.components.simplisafe
simplisafe-python==9.2.0
simplisafe-python==9.2.1
# homeassistant.components.sisyphus
sisyphus-control==2.2.1

2
requirements_test_all.txt
View File

@ -866,7 +866,7 @@ sentry-sdk==0.13.5
simplehound==0.3
# homeassistant.components.simplisafe
simplisafe-python==9.2.0
simplisafe-python==9.2.1
# homeassistant.components.sleepiq
sleepyq==0.7

148
tests/components/simplisafe/test_config_flow.py
View File

@ -1,14 +1,16 @@
"""Define tests for the SimpliSafe config flow."""
import json
from simplipy.errors import SimplipyError
from simplipy.errors import (
InvalidCredentialsError,
PendingAuthorizationError,
SimplipyError,
)
from homeassistant import data_entry_flow
from homeassistant.components.simplisafe import DOMAIN
from homeassistant.config_entries import SOURCE_IMPORT, SOURCE_USER
from homeassistant.const import CONF_CODE, CONF_PASSWORD, CONF_TOKEN, CONF_USERNAME
from tests.async_mock import MagicMock, PropertyMock, mock_open, patch
from tests.async_mock import MagicMock, PropertyMock, patch
from tests.common import MockConfigEntry
@ -21,11 +23,17 @@ def mock_api():
async def test_duplicate_error(hass):
"""Test that errors are shown when duplicates are added."""
conf = {CONF_USERNAME: "user@email.com", CONF_PASSWORD: "password"}
conf = {
CONF_USERNAME: "user@email.com",
CONF_PASSWORD: "password",
CONF_CODE: "1234",
}
MockConfigEntry(domain=DOMAIN, unique_id="user@email.com", data=conf).add_to_hass(
hass
)
MockConfigEntry(
domain=DOMAIN,
unique_id="user@email.com",
data={CONF_USERNAME: "user@email.com", CONF_TOKEN: "12345", CONF_CODE: "1234"},
).add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}, data=conf
@ -39,8 +47,9 @@ async def test_invalid_credentials(hass):
"""Test that invalid credentials throws an error."""
conf = {CONF_USERNAME: "user@email.com", CONF_PASSWORD: "password"}
print("AARON")
with patch(
"simplipy.API.login_via_credentials", side_effect=SimplipyError,
"simplipy.API.login_via_credentials", side_effect=InvalidCredentialsError,
):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}, data=conf
@ -75,15 +84,12 @@ async def test_options_flow(hass):
async def test_show_form(hass):
"""Test that the form is served with no input."""
with patch(
"homeassistant.components.simplisafe.async_setup_entry", return_value=True
):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_IMPORT}
)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_IMPORT}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "user"
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "user"
async def test_step_import(hass):
@ -94,17 +100,9 @@ async def test_step_import(hass):
CONF_CODE: "1234",
}
mop = mock_open(read_data=json.dumps({"refresh_token": "12345"}))
with patch(
"homeassistant.components.simplisafe.async_setup_entry", return_value=True
), patch("simplipy.API.login_via_credentials", return_value=mock_api()), patch(
"homeassistant.util.json.open", mop, create=True
), patch(
"homeassistant.util.json.os.open", return_value=0
), patch(
"homeassistant.util.json.os.replace"
):
), patch("simplipy.API.login_via_credentials", return_value=mock_api()):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}, data=conf
)
@ -118,25 +116,48 @@ async def test_step_import(hass):
}
async def test_step_reauth(hass):
"""Test that the reauth step works."""
MockConfigEntry(
domain=DOMAIN,
unique_id="user@email.com",
data={CONF_USERNAME: "user@email.com", CONF_TOKEN: "12345", CONF_CODE: "1234"},
).add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": "reauth"},
data={CONF_CODE: "1234", CONF_USERNAME: "user@email.com"},
)
assert result["step_id"] == "reauth_confirm"
result = await hass.config_entries.flow.async_configure(result["flow_id"])
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "reauth_confirm"
with patch(
"homeassistant.components.simplisafe.async_setup_entry", return_value=True
), patch("simplipy.API.login_via_credentials", return_value=mock_api()):
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={CONF_PASSWORD: "password"}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "reauth_successful"
assert len(hass.config_entries.async_entries()) == 1
async def test_step_user(hass):
"""Test that the user step works."""
"""Test that the user step works (without MFA)."""
conf = {
CONF_USERNAME: "user@email.com",
CONF_PASSWORD: "password",
CONF_CODE: "1234",
}
mop = mock_open(read_data=json.dumps({"refresh_token": "12345"}))
with patch(
"homeassistant.components.simplisafe.async_setup_entry", return_value=True
), patch("simplipy.API.login_via_credentials", return_value=mock_api()), patch(
"homeassistant.util.json.open", mop, create=True
), patch(
"homeassistant.util.json.os.open", return_value=0
), patch(
"homeassistant.util.json.os.replace"
):
), patch("simplipy.API.login_via_credentials", return_value=mock_api()):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}, data=conf
)
@ -148,3 +169,58 @@ async def test_step_user(hass):
CONF_TOKEN: "12345abc",
CONF_CODE: "1234",
}
async def test_step_user_mfa(hass):
"""Test that the user step works when MFA is in the middle."""
conf = {
CONF_USERNAME: "user@email.com",
CONF_PASSWORD: "password",
CONF_CODE: "1234",
}
with patch(
"simplipy.API.login_via_credentials", side_effect=PendingAuthorizationError
):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}, data=conf
)
assert result["step_id"] == "mfa"
with patch(
"simplipy.API.login_via_credentials", side_effect=PendingAuthorizationError
):
# Simulate the user pressing the MFA submit button without having clicked
# the link in the MFA email:
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={}
)
assert result["step_id"] == "mfa"
with patch(
"homeassistant.components.simplisafe.async_setup_entry", return_value=True
), patch("simplipy.API.login_via_credentials", return_value=mock_api()):
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result["title"] == "user@email.com"
assert result["data"] == {
CONF_USERNAME: "user@email.com",
CONF_TOKEN: "12345abc",
CONF_CODE: "1234",
}
async def test_unknown_error(hass):
"""Test that an unknown error raises the correct error."""
conf = {CONF_USERNAME: "user@email.com", CONF_PASSWORD: "password"}
with patch(
"simplipy.API.login_via_credentials", side_effect=SimplipyError,
):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}, data=conf
)
assert result["errors"] == {"base": "unknown"}