Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Revert SimpliSafe auth flow to the quasi-manual OAuth method from 2021.11.0 (#75641) 

* Revert "Migrate SimpliSafe to new web-based authentication (#57212)"

This reverts commit bf7c99c1f8.

* Tests 100%

* Version bump

* Add manifest version for custom component testing

* Remove manifest version

* Code review

* Fix tests
pull/75758/head
Aaron Bach 2022-07-24 14:09:02 -06:00 committed by Franck Nijhof
parent 0407fc4581
commit 9173aef1ef
No known key found for this signature in database
GPG Key ID: D62583BA8AB11CA3
9 changed files with 192 additions and 521 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -278,10 +278,8 @@ def _async_standardize_config_entry(hass: HomeAssistant, entry: ConfigEntry) ->
"""Bring a config entry up to current standards."""
if CONF_TOKEN not in entry.data:
raise ConfigEntryAuthFailed(
"New SimpliSafe OAuth standard requires re-authentication"
"SimpliSafe OAuth standard requires re-authentication"
)
if CONF_USERNAME not in entry.data:
raise ConfigEntryAuthFailed("Need to re-auth with username/password")
entry_updates = {}
if not entry.unique_id:

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

@ -1,48 +1,52 @@
"""Config flow to configure the SimpliSafe component."""
from __future__ import annotations
import asyncio
from collections.abc import Mapping
from typing import Any
from typing import Any, NamedTuple
import async_timeout
from simplipy import API
from simplipy.api import AuthStates
from simplipy.errors import InvalidCredentialsError, SimplipyError, Verify2FAPending
from simplipy.errors import InvalidCredentialsError, SimplipyError
from simplipy.util.auth import (
get_auth0_code_challenge,
get_auth0_code_verifier,
get_auth_url,
)
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_CODE, CONF_PASSWORD, CONF_TOKEN, CONF_USERNAME
from homeassistant.const import CONF_CODE, CONF_TOKEN, CONF_URL, CONF_USERNAME
from homeassistant.core import callback
from homeassistant.data_entry_flow import FlowResult
from homeassistant.helpers import aiohttp_client, config_validation as cv
from .const import DOMAIN, LOGGER
DEFAULT_EMAIL_2FA_SLEEP = 3
DEFAULT_EMAIL_2FA_TIMEOUT = 600
STEP_REAUTH_SCHEMA = vol.Schema(
{
vol.Required(CONF_PASSWORD): cv.string,
}
)
STEP_SMS_2FA_SCHEMA = vol.Schema(
{
vol.Required(CONF_CODE): cv.string,
}
)
CONF_AUTH_CODE = "auth_code"
STEP_USER_SCHEMA = vol.Schema(
{
vol.Required(CONF_USERNAME): cv.string,
vol.Required(CONF_PASSWORD): cv.string,
vol.Required(CONF_AUTH_CODE): cv.string,
}
)
class SimpliSafeOAuthValues(NamedTuple):
"""Define a named tuple to handle SimpliSafe OAuth strings."""
auth_url: str
code_verifier: str
@callback
def async_get_simplisafe_oauth_values() -> SimpliSafeOAuthValues:
"""Get a SimpliSafe OAuth code verifier and auth URL."""
code_verifier = get_auth0_code_verifier()
code_challenge = get_auth0_code_challenge(code_verifier)
auth_url = get_auth_url(code_challenge)
return SimpliSafeOAuthValues(auth_url, code_verifier)
class SimpliSafeFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle a SimpliSafe config flow."""
@ -50,45 +54,8 @@ class SimpliSafeFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
def __init__(self) -> None:
"""Initialize the config flow."""
self._email_2fa_task: asyncio.Task | None = None
self._password: str | None = None
self._oauth_values: SimpliSafeOAuthValues = async_get_simplisafe_oauth_values()
self._reauth: bool = False
self._simplisafe: API | None = None
self._username: str | None = None
async def _async_authenticate(
self, originating_step_id: str, originating_step_schema: vol.Schema
) -> FlowResult:
"""Attempt to authenticate to the SimpliSafe API."""
assert self._password
assert self._username
errors = {}
session = aiohttp_client.async_get_clientsession(self.hass)
try:
self._simplisafe = await API.async_from_credentials(
self._username, self._password, session=session
)
except InvalidCredentialsError:
errors = {"base": "invalid_auth"}
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=originating_step_id,
data_schema=originating_step_schema,
errors=errors,
description_placeholders={CONF_USERNAME: self._username},
)
assert self._simplisafe
if self._simplisafe.auth_state == AuthStates.PENDING_2FA_SMS:
return await self.async_step_sms_2fa()
return await self.async_step_email_2fa()
@staticmethod
@callback
@ -98,146 +65,66 @@ class SimpliSafeFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
"""Define the config flow to handle options."""
return SimpliSafeOptionsFlowHandler(config_entry)
async def async_step_reauth(self, entry_data: Mapping[str, Any]) -> FlowResult:
async def async_step_reauth(self, config: Mapping[str, Any]) -> FlowResult:
"""Handle configuration by re-auth."""
self._reauth = True
if CONF_USERNAME not in entry_data:
# Old versions of the config flow may not have the username by this point;
# in that case, we reauth them by making them go through the user flow:
return await self.async_step_user()
self._username = entry_data[CONF_USERNAME]
return await self.async_step_reauth_confirm()
async def _async_get_email_2fa(self) -> None:
"""Define a task to wait for email-based 2FA."""
assert self._simplisafe
try:
async with async_timeout.timeout(DEFAULT_EMAIL_2FA_TIMEOUT):
while True:
try:
await self._simplisafe.async_verify_2fa_email()
except Verify2FAPending:
LOGGER.info("Email-based 2FA pending; trying again")
await asyncio.sleep(DEFAULT_EMAIL_2FA_SLEEP)
else:
break
finally:
self.hass.async_create_task(
self.hass.config_entries.flow.async_configure(flow_id=self.flow_id)
)
async def async_step_email_2fa(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle email-based two-factor authentication."""
if not self._email_2fa_task:
self._email_2fa_task = self.hass.async_create_task(
self._async_get_email_2fa()
)
return self.async_show_progress(
step_id="email_2fa", progress_action="email_2fa"
)
try:
await self._email_2fa_task
except asyncio.TimeoutError:
return self.async_show_progress_done(next_step_id="email_2fa_error")
return self.async_show_progress_done(next_step_id="finish")
async def async_step_email_2fa_error(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle an error during email-based two-factor authentication."""
return self.async_abort(reason="email_2fa_timed_out")
async def async_step_finish(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle the final step."""
assert self._simplisafe
assert self._username
data = {
CONF_USERNAME: self._username,
CONF_TOKEN: self._simplisafe.refresh_token,
}
user_id = str(self._simplisafe.user_id)
if self._reauth:
# "Old" config entries utilized the user's email address (username) as the
# unique ID, whereas "new" config entries utilize the SimpliSafe user ID –
# only one can exist at a time, but the presence of either one is a
# candidate for re-auth:
if existing_entries := [
entry
for entry in self.hass.config_entries.async_entries()
if entry.domain == DOMAIN
and entry.unique_id in (self._username, user_id)
]:
existing_entry = existing_entries[0]
self.hass.config_entries.async_update_entry(
existing_entry, unique_id=user_id, title=self._username, data=data
)
self.hass.async_create_task(
self.hass.config_entries.async_reload(existing_entry.entry_id)
)
return self.async_abort(reason="reauth_successful")
await self.async_set_unique_id(user_id)
self._abort_if_unique_id_configured()
return self.async_create_entry(title=self._username, data=data)
async def async_step_reauth_confirm(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle re-auth completion."""
if not user_input:
return self.async_show_form(
step_id="reauth_confirm",
data_schema=STEP_REAUTH_SCHEMA,
description_placeholders={CONF_USERNAME: self._username},
)
self._password = user_input[CONF_PASSWORD]
return await self._async_authenticate("reauth_confirm", STEP_REAUTH_SCHEMA)
async def async_step_sms_2fa(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle SMS-based two-factor authentication."""
if not user_input:
return self.async_show_form(
step_id="sms_2fa",
data_schema=STEP_SMS_2FA_SCHEMA,
)
assert self._simplisafe
try:
await self._simplisafe.async_verify_2fa_sms(user_input[CONF_CODE])
except InvalidCredentialsError:
return self.async_show_form(
step_id="sms_2fa",
data_schema=STEP_SMS_2FA_SCHEMA,
errors={CONF_CODE: "invalid_auth"},
)
return await self.async_step_finish()
return await self.async_step_user()
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle the start of the config flow."""
if user_input is None:
return self.async_show_form(step_id="user", data_schema=STEP_USER_SCHEMA)
return self.async_show_form(
step_id="user",
data_schema=STEP_USER_SCHEMA,
description_placeholders={CONF_URL: self._oauth_values.auth_url},
)
self._username = user_input[CONF_USERNAME]
self._password = user_input[CONF_PASSWORD]
return await self._async_authenticate("user", STEP_USER_SCHEMA)
errors = {}
session = aiohttp_client.async_get_clientsession(self.hass)
try:
simplisafe = await API.async_from_auth(
user_input[CONF_AUTH_CODE],
self._oauth_values.code_verifier,
session=session,
)
except InvalidCredentialsError:
errors = {"base": "invalid_auth"}
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="user",
data_schema=STEP_USER_SCHEMA,
errors=errors,
description_placeholders={CONF_URL: self._oauth_values.auth_url},
)
simplisafe_user_id = str(simplisafe.user_id)
data = {CONF_USERNAME: simplisafe_user_id, CONF_TOKEN: simplisafe.refresh_token}
if self._reauth:
existing_entry = await self.async_set_unique_id(simplisafe_user_id)
if not existing_entry:
# If we don't have an entry that matches this user ID, the user logged
# in with different credentials:
return self.async_abort(reason="wrong_account")
self.hass.config_entries.async_update_entry(
existing_entry, unique_id=simplisafe_user_id, data=data
)
self.hass.async_create_task(
self.hass.config_entries.async_reload(existing_entry.entry_id)
)
return self.async_abort(reason="reauth_successful")
await self.async_set_unique_id(simplisafe_user_id)
self._abort_if_unique_id_configured()
return self.async_create_entry(title=simplisafe_user_id, data=data)
class SimpliSafeOptionsFlowHandler(config_entries.OptionsFlow):

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

@ -3,7 +3,7 @@
"name": "SimpliSafe",
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/simplisafe",
"requirements": ["simplisafe-python==2022.07.0"],
"requirements": ["simplisafe-python==2022.07.1"],
"codeowners": ["@bachya"],
"iot_class": "cloud_polling",
"dhcp": [

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

@ -1,38 +1,22 @@
{
"config": {
"step": {
"reauth_confirm": {
"title": "[%key:common::config_flow::title::reauth%]",
"description": "Please re-enter the password for {username}.",
"data": {
"password": "[%key:common::config_flow::data::password%]"
}
},
"sms_2fa": {
"description": "Input the two-factor authentication code sent to you via SMS.",
"data": {
"code": "Code"
}
},
"user": {
"description": "Input your username and password.",
"description": "SimpliSafe authenticates users via its web app. Due to technical limitations, there is a manual step at the end of this process; please ensure that you read the [documentation](http://home-assistant.io/integrations/simplisafe#getting-an-authorization-code) before starting.\n\nWhen you are ready, click [here]({url}) to open the SimpliSafe web app and input your credentials. When the process is complete, return here and input the authorization code from the SimpliSafe web app URL.",
"data": {
"username": "[%key:common::config_flow::data::username%]",
"password": "[%key:common::config_flow::data::password%]"
"auth_code": "Authorization Code"
}
}
},
"error": {
"identifier_exists": "Account already registered",
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
"unknown": "[%key:common::config_flow::error::unknown%]"
},
"abort": {
"already_configured": "This SimpliSafe account is already in use.",
"email_2fa_timed_out": "Timed out while waiting for email-based two-factor authentication.",
"reauth_successful": "[%key:common::config_flow::abort::reauth_successful%]"
},
"progress": {
"email_2fa": "Check your email for a verification link from Simplisafe."
"reauth_successful": "[%key:common::config_flow::abort::reauth_successful%]",
"wrong_account": "The user credentials provided do not match this SimpliSafe account."
}
},
"options": {

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

@ -2,36 +2,20 @@
"config": {
"abort": {
"already_configured": "This SimpliSafe account is already in use.",
"email_2fa_timed_out": "Timed out while waiting for email-based two-factor authentication.",
"reauth_successful": "Re-authentication was successful"
"reauth_successful": "Re-authentication was successful",
"wrong_account": "The user credentials provided do not match this SimpliSafe account."
},
"error": {
"identifier_exists": "Account already registered",
"invalid_auth": "Invalid authentication",
"unknown": "Unexpected error"
},
"progress": {
"email_2fa": "Check your email for a verification link from Simplisafe."
},
"step": {
"reauth_confirm": {
"data": {
"password": "Password"
},
"description": "Please re-enter the password for {username}.",
"title": "Reauthenticate Integration"
},
"sms_2fa": {
"data": {
"code": "Code"
},
"description": "Input the two-factor authentication code sent to you via SMS."
},
"user": {
"data": {
"password": "Password",
"username": "Username"
"auth_code": "Authorization Code"
},
"description": "Input your username and password."
"description": "SimpliSafe authenticates users via its web app. Due to technical limitations, there is a manual step at the end of this process; please ensure that you read the [documentation](http://home-assistant.io/integrations/simplisafe#getting-an-authorization-code) before starting.\n\nWhen you are ready, click [here]({url}) to open the SimpliSafe web app and input your credentials. When the process is complete, return here and input the authorization code from the SimpliSafe web app URL."
}
}
},

2
requirements_all.txt
View File

@ -2168,7 +2168,7 @@ simplehound==0.3
simplepush==1.1.4
# homeassistant.components.simplisafe
simplisafe-python==2022.07.0
simplisafe-python==2022.07.1
# homeassistant.components.sisyphus
sisyphus-control==3.1.2

2
requirements_test_all.txt
View File

@ -1440,7 +1440,7 @@ simplehound==0.3
simplepush==1.1.4
# homeassistant.components.simplisafe
simplisafe-python==2022.07.0
simplisafe-python==2022.07.1
# homeassistant.components.slack
slackclient==2.5.0

18
tests/components/simplisafe/conftest.py
View File

@ -3,7 +3,6 @@ import json
from unittest.mock import AsyncMock, Mock, patch
import pytest
from simplipy.api import AuthStates
from simplipy.system.v3 import SystemV3
from homeassistant.components.simplisafe.const import DOMAIN
@ -19,20 +18,11 @@ PASSWORD = "password"
SYSTEM_ID = "system_123"
@pytest.fixture(name="api_auth_state")
def api_auth_state_fixture():
"""Define a SimpliSafe API auth state."""
return AuthStates.PENDING_2FA_SMS
@pytest.fixture(name="api")
def api_fixture(api_auth_state, data_subscription, system_v3, websocket):
def api_fixture(data_subscription, system_v3, websocket):
"""Define a simplisafe-python API object."""
return Mock(
async_get_systems=AsyncMock(return_value={SYSTEM_ID: system_v3}),
async_verify_2fa_email=AsyncMock(),
async_verify_2fa_sms=AsyncMock(),
auth_state=api_auth_state,
refresh_token=REFRESH_TOKEN,
subscription_data=data_subscription,
user_id=USER_ID,
@ -104,12 +94,10 @@ def reauth_config_fixture():
async def setup_simplisafe_fixture(hass, api, config):
"""Define a fixture to set up SimpliSafe."""
with patch(
"homeassistant.components.simplisafe.config_flow.DEFAULT_EMAIL_2FA_SLEEP", 0
), patch(
"homeassistant.components.simplisafe.config_flow.API.async_from_credentials",
"homeassistant.components.simplisafe.config_flow.API.async_from_auth",
return_value=api,
), patch(
"homeassistant.components.simplisafe.API.async_from_credentials",
"homeassistant.components.simplisafe.API.async_from_auth",
return_value=api,
), patch(
"homeassistant.components.simplisafe.API.async_from_refresh_token",

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

@ -2,45 +2,53 @@
from unittest.mock import patch
import pytest
from simplipy.api import AuthStates
from simplipy.errors import InvalidCredentialsError, SimplipyError, Verify2FAPending
from simplipy.errors import InvalidCredentialsError, SimplipyError
from homeassistant import data_entry_flow
from homeassistant.components.simplisafe import DOMAIN
from homeassistant.components.simplisafe.config_flow import CONF_AUTH_CODE
from homeassistant.config_entries import SOURCE_REAUTH, SOURCE_USER
from homeassistant.const import CONF_CODE, CONF_TOKEN, CONF_USERNAME
from .common import REFRESH_TOKEN, USER_ID, USERNAME
from tests.common import MockConfigEntry
CONF_USER_ID = "user_id"
async def test_duplicate_error(
hass, config_entry, credentials_config, setup_simplisafe, sms_config
):
async def test_duplicate_error(config_entry, hass, setup_simplisafe):
"""Test that errors are shown when duplicates are added."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["step_id"] == "user"
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
with patch(
"homeassistant.components.simplisafe.async_setup_entry", return_value=True
):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["step_id"] == "user"
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input=credentials_config
)
assert result["step_id"] == "sms_2fa"
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input=sms_config
)
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "already_configured"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={CONF_AUTH_CODE: "code123"}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "already_configured"
async def test_options_flow(hass, config_entry):
async def test_invalid_credentials(hass):
"""Test that invalid credentials show the correct error."""
with patch(
"homeassistant.components.simplisafe.config_flow.API.async_from_auth",
side_effect=InvalidCredentialsError,
):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["step_id"] == "user"
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={CONF_AUTH_CODE: "code123"}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["errors"] == {"base": "invalid_auth"}
async def test_options_flow(config_entry, hass):
"""Test config flow options."""
with patch(
"homeassistant.components.simplisafe.async_setup_entry", return_value=True
@ -53,134 +61,79 @@ async def test_options_flow(hass, config_entry):
result = await hass.config_entries.options.async_configure(
result["flow_id"], user_input={CONF_CODE: "4321"}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert config_entry.options == {CONF_CODE: "4321"}
@pytest.mark.parametrize("unique_id", [USERNAME, USER_ID])
async def test_step_reauth(
hass, config, config_entry, reauth_config, setup_simplisafe, sms_config, unique_id
):
"""Test the re-auth step (testing both username and user ID as unique ID)."""
# Add a second config entry (tied to a random domain, but with the same unique ID
# that could exist in a SimpliSafe entry) to ensure that this reauth process only
# touches the SimpliSafe entry:
entry = MockConfigEntry(domain="random", unique_id=USERNAME, data={"some": "data"})
entry.add_to_hass(hass)
async def test_step_reauth(config_entry, hass, setup_simplisafe):
"""Test the re-auth step."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_REAUTH}, data=config
)
assert result["step_id"] == "reauth_confirm"
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input=reauth_config
)
assert result["step_id"] == "sms_2fa"
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input=sms_config
)
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "reauth_successful"
assert len(hass.config_entries.async_entries()) == 2
# Test that the SimpliSafe config flow is updated:
[config_entry] = hass.config_entries.async_entries(DOMAIN)
assert config_entry.unique_id == USER_ID
assert config_entry.data == config
# Test that the non-SimpliSafe config flow remains the same:
[config_entry] = hass.config_entries.async_entries("random")
assert config_entry == entry
@pytest.mark.parametrize(
"exc,error_string",
[(InvalidCredentialsError, "invalid_auth"), (SimplipyError, "unknown")],
)
async def test_step_reauth_errors(hass, config, error_string, exc, reauth_config):
"""Test that errors during the reauth step are handled."""
with patch(
"homeassistant.components.simplisafe.API.async_from_credentials",
side_effect=exc,
):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_REAUTH}, data=config
)
assert result["step_id"] == "reauth_confirm"
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input=reauth_config
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "reauth_confirm"
assert result["errors"] == {"base": error_string}
@pytest.mark.parametrize(
"config,unique_id",
[
(
{
CONF_TOKEN: REFRESH_TOKEN,
CONF_USER_ID: USER_ID,
},
USERNAME,
),
(
{
CONF_TOKEN: REFRESH_TOKEN,
CONF_USER_ID: USER_ID,
},
USER_ID,
),
],
)
async def test_step_reauth_from_scratch(
hass, config, config_entry, credentials_config, setup_simplisafe, sms_config
):
"""Test the re-auth step when a complete redo is needed."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_REAUTH}, data=config
DOMAIN,
context={"source": SOURCE_REAUTH},
data={CONF_USERNAME: "12345", CONF_TOKEN: "token123"},
)
assert result["step_id"] == "user"
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input=credentials_config
)
assert result["step_id"] == "sms_2fa"
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input=sms_config
)
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "reauth_successful"
with patch(
"homeassistant.components.simplisafe.async_setup_entry", return_value=True
), patch("homeassistant.config_entries.ConfigEntries.async_reload"):
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={CONF_AUTH_CODE: "code123"}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "reauth_successful"
assert len(hass.config_entries.async_entries()) == 1
[config_entry] = hass.config_entries.async_entries(DOMAIN)
assert config_entry.unique_id == USER_ID
assert config_entry.data == {
CONF_TOKEN: REFRESH_TOKEN,
CONF_USERNAME: USERNAME,
}
assert config_entry.data == {CONF_USERNAME: "12345", CONF_TOKEN: "token123"}
@pytest.mark.parametrize(
"exc,error_string",
[(InvalidCredentialsError, "invalid_auth"), (SimplipyError, "unknown")],
)
async def test_step_user_errors(hass, credentials_config, error_string, exc):
"""Test that errors during the user step are handled."""
@pytest.mark.parametrize("unique_id", ["some_other_id"])
async def test_step_reauth_wrong_account(config_entry, hass, setup_simplisafe):
"""Test the re-auth step where the wrong account is used during login."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_REAUTH},
data={CONF_USERNAME: "12345", CONF_TOKEN: "token123"},
)
assert result["step_id"] == "user"
with patch(
"homeassistant.components.simplisafe.API.async_from_credentials",
side_effect=exc,
"homeassistant.components.simplisafe.async_setup_entry", return_value=True
), patch("homeassistant.config_entries.ConfigEntries.async_reload"):
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={CONF_AUTH_CODE: "code123"}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "wrong_account"
async def test_step_user(hass, setup_simplisafe):
"""Test the user step."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["step_id"] == "user"
with patch(
"homeassistant.components.simplisafe.async_setup_entry", return_value=True
), patch("homeassistant.config_entries.ConfigEntries.async_reload"):
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={CONF_AUTH_CODE: "code123"}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert len(hass.config_entries.async_entries()) == 1
[config_entry] = hass.config_entries.async_entries(DOMAIN)
assert config_entry.data == {CONF_USERNAME: "12345", CONF_TOKEN: "token123"}
async def test_unknown_error(hass, setup_simplisafe):
"""Test that an unknown error shows ohe correct error."""
with patch(
"homeassistant.components.simplisafe.config_flow.API.async_from_auth",
side_effect=SimplipyError,
):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
@ -189,130 +142,7 @@ async def test_step_user_errors(hass, credentials_config, error_string, exc):
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input=credentials_config
result["flow_id"], user_input={CONF_AUTH_CODE: "code123"}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "user"
assert result["errors"] == {"base": error_string}
@pytest.mark.parametrize("api_auth_state", [AuthStates.PENDING_2FA_EMAIL])
async def test_step_user_email_2fa(
api, api_auth_state, hass, config, credentials_config, setup_simplisafe
):
"""Test the user step with email-based 2FA."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["step_id"] == "user"
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
# Patch API.async_verify_2fa_email to first return pending, then return all done:
api.async_verify_2fa_email.side_effect = [Verify2FAPending, None]
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input=credentials_config
)
assert result["type"] == data_entry_flow.RESULT_TYPE_SHOW_PROGRESS
result = await hass.config_entries.flow.async_configure(result["flow_id"])
assert result["type"] == data_entry_flow.RESULT_TYPE_SHOW_PROGRESS_DONE
result = await hass.config_entries.flow.async_configure(result["flow_id"])
assert len(hass.config_entries.async_entries()) == 1
[config_entry] = hass.config_entries.async_entries(DOMAIN)
assert config_entry.unique_id == USER_ID
assert config_entry.data == config
@patch("homeassistant.components.simplisafe.config_flow.DEFAULT_EMAIL_2FA_TIMEOUT", 0)
@pytest.mark.parametrize("api_auth_state", [AuthStates.PENDING_2FA_EMAIL])
async def test_step_user_email_2fa_timeout(
api, hass, config, credentials_config, setup_simplisafe
):
"""Test a timeout during the user step with email-based 2FA."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["step_id"] == "user"
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
# Patch API.async_verify_2fa_email to return pending:
api.async_verify_2fa_email.side_effect = Verify2FAPending
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input=credentials_config
)
assert result["type"] == data_entry_flow.RESULT_TYPE_SHOW_PROGRESS
result = await hass.config_entries.flow.async_configure(result["flow_id"])
assert result["type"] == data_entry_flow.RESULT_TYPE_SHOW_PROGRESS_DONE
assert result["step_id"] == "email_2fa_error"
result = await hass.config_entries.flow.async_configure(result["flow_id"])
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "email_2fa_timed_out"
async def test_step_user_sms_2fa(
hass, config, credentials_config, setup_simplisafe, sms_config
):
"""Test the user step with SMS-based 2FA."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["step_id"] == "user"
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input=credentials_config
)
assert result["step_id"] == "sms_2fa"
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input=sms_config
)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert len(hass.config_entries.async_entries()) == 1
[config_entry] = hass.config_entries.async_entries(DOMAIN)
assert config_entry.unique_id == USER_ID
assert config_entry.data == config
@pytest.mark.parametrize(
"exc,error_string", [(InvalidCredentialsError, "invalid_auth")]
)
async def test_step_user_sms_2fa_errors(
api,
hass,
config,
credentials_config,
error_string,
exc,
setup_simplisafe,
sms_config,
):
"""Test that errors during the SMS-based 2FA step are handled."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["step_id"] == "user"
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input=credentials_config
)
assert result["step_id"] == "sms_2fa"
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
# Simulate entering the incorrect SMS code:
api.async_verify_2fa_sms.side_effect = InvalidCredentialsError
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input=sms_config
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["errors"] == {"code": error_string}
assert result["errors"] == {"base": "unknown"}