jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-19 03:07:37 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #75758 from home-assistant/rc

Browse Source
This commit is contained in:
Franck Nijhof 2022-07-26 17:22:42 +02:00 committed by GitHub
commit 35f4220f4e
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
19 changed files with 219 additions and 544 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -3,7 +3,7 @@
"name": "Epson",
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/epson",
"requirements": ["epson-projector==0.4.2"],
"requirements": ["epson-projector==0.4.6"],
"codeowners": ["@pszafer"],
"iot_class": "local_polling",
"loggers": ["epson_projector"]

7
homeassistant/components/epson/media_player.py
View File

@ -133,7 +133,10 @@ class EpsonProjectorMediaPlayer(MediaPlayerEntity):
self._source = SOURCE_LIST.get(source, self._source)
volume = await self._projector.get_property(VOLUME)
if volume:
self._volume = volume
try:
self._volume = float(volume)
except ValueError:
self._volume = None
elif power_state == BUSY:
self._state = STATE_ON
else:
@ -176,11 +179,13 @@ class EpsonProjectorMediaPlayer(MediaPlayerEntity):
"""Turn on epson."""
if self._state == STATE_OFF:
await self._projector.send_command(TURN_ON)
self._state = STATE_ON
async def async_turn_off(self):
"""Turn off epson."""
if self._state == STATE_ON:
await self._projector.send_command(TURN_OFF)
self._state = STATE_OFF
@property
def source_list(self):

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

@ -3,7 +3,7 @@
"name": "HVV Departures",
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/hvv_departures",
"requirements": ["pygti==0.9.2"],
"requirements": ["pygti==0.9.3"],
"codeowners": ["@vigonotion"],
"iot_class": "cloud_polling",
"loggers": ["pygti"]

4
homeassistant/components/monoprice/config_flow.py
View File

@ -3,7 +3,7 @@ from __future__ import annotations
import logging
from pymonoprice import get_async_monoprice
from pymonoprice import get_monoprice
from serial import SerialException
import voluptuous as vol
@ -56,7 +56,7 @@ async def validate_input(hass: core.HomeAssistant, data):
Data has the keys from DATA_SCHEMA with values provided by the user.
"""
try:
await get_async_monoprice(data[CONF_PORT], hass.loop)
await hass.async_add_executor_job(get_monoprice, data[CONF_PORT])
except SerialException as err:
_LOGGER.error("Error connecting to Monoprice controller")
raise CannotConnect from err

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

@ -2,7 +2,7 @@
"domain": "opentherm_gw",
"name": "OpenTherm Gateway",
"documentation": "https://www.home-assistant.io/integrations/opentherm_gw",
"requirements": ["pyotgw==2.0.0"],
"requirements": ["pyotgw==2.0.1"],
"codeowners": ["@mvn23"],
"config_flow": true,
"iot_class": "local_push",

4
homeassistant/components/rainmachine/sensor.py
View File

@ -243,10 +243,8 @@ class TimeRemainingSensor(RainMachineEntity, RestoreSensor):
seconds_remaining = self.calculate_seconds_remaining()
new_timestamp = now + timedelta(seconds=seconds_remaining)
assert isinstance(self._attr_native_value, datetime)
if (
self._attr_native_value
isinstance(self._attr_native_value, datetime)
and new_timestamp - self._attr_native_value
< DEFAULT_ZONE_COMPLETION_TIME_WOBBLE_TOLERANCE
):

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."
}
}
},

7
homeassistant/components/stream/__init__.py
View File

@ -503,15 +503,16 @@ class Stream:
await self.start()
self._logger.debug("Started a stream recording of %s seconds", duration)
# Take advantage of lookback
hls: HlsStreamOutput = cast(HlsStreamOutput, self.outputs().get(HLS_PROVIDER))
if lookback > 0 and hls:
num_segments = min(int(lookback // hls.target_duration), MAX_SEGMENTS)
if hls:
num_segments = min(int(lookback / hls.target_duration) + 1, MAX_SEGMENTS)
# Wait for latest segment, then add the lookback
await hls.recv()
recorder.prepend(list(hls.get_segments())[-num_segments - 1 : -1])
self._logger.debug("Started a stream recording of %s seconds", duration)
await recorder.async_record()
async def async_get_image(

2
homeassistant/const.py
View File

@ -7,7 +7,7 @@ from .backports.enum import StrEnum
MAJOR_VERSION: Final = 2022
MINOR_VERSION: Final = 7
PATCH_VERSION: Final = "6"
PATCH_VERSION: Final = "7"
__short_version__: Final = f"{MAJOR_VERSION}.{MINOR_VERSION}"
__version__: Final = f"{__short_version__}.{PATCH_VERSION}"
REQUIRED_PYTHON_VER: Final[tuple[int, int, int]] = (3, 9, 0)

2
pyproject.toml
View File

@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "homeassistant"
version = "2022.7.6"
version = "2022.7.7"
license = {text = "Apache-2.0"}
description = "Open-source home automation platform running on Python 3."
readme = "README.rst"

8
requirements_all.txt
View File

@ -610,7 +610,7 @@ envoy_reader==0.20.1
ephem==4.1.2
# homeassistant.components.epson
epson-projector==0.4.2
epson-projector==0.4.6
# homeassistant.components.epsonworkforce
epsonprinter==0.0.9
@ -1532,7 +1532,7 @@ pygatt[GATTTOOL]==4.0.5
pygtfs==0.1.6
# homeassistant.components.hvv_departures
pygti==0.9.2
pygti==0.9.3
# homeassistant.components.version
pyhaversion==22.4.1
@ -1715,7 +1715,7 @@ pyopnsense==0.2.0
pyoppleio==1.0.5
# homeassistant.components.opentherm_gw
pyotgw==2.0.0
pyotgw==2.0.1
# homeassistant.auth.mfa_modules.notify
# homeassistant.auth.mfa_modules.totp
@ -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

8
requirements_test_all.txt
View File

@ -447,7 +447,7 @@ envoy_reader==0.20.1
ephem==4.1.2
# homeassistant.components.epson
epson-projector==0.4.2
epson-projector==0.4.6
# homeassistant.components.faa_delays
faadelays==0.0.7
@ -1032,7 +1032,7 @@ pyfttt==0.3
pygatt[GATTTOOL]==4.0.5
# homeassistant.components.hvv_departures
pygti==0.9.2
pygti==0.9.3
# homeassistant.components.version
pyhaversion==22.4.1
@ -1167,7 +1167,7 @@ pyopenuv==2022.04.0
pyopnsense==0.2.0
# homeassistant.components.opentherm_gw
pyotgw==2.0.0
pyotgw==2.0.1
# homeassistant.auth.mfa_modules.notify
# homeassistant.auth.mfa_modules.totp
@ -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

6
tests/components/monoprice/test_config_flow.py
View File

@ -33,7 +33,7 @@ async def test_form(hass):
assert result["errors"] == {}
with patch(
"homeassistant.components.monoprice.config_flow.get_async_monoprice",
"homeassistant.components.monoprice.config_flow.get_monoprice",
return_value=True,
), patch(
"homeassistant.components.monoprice.async_setup_entry",
@ -60,7 +60,7 @@ async def test_form_cannot_connect(hass):
)
with patch(
"homeassistant.components.monoprice.config_flow.get_async_monoprice",
"homeassistant.components.monoprice.config_flow.get_monoprice",
side_effect=SerialException,
):
result2 = await hass.config_entries.flow.async_configure(
@ -78,7 +78,7 @@ async def test_generic_exception(hass):
)
with patch(
"homeassistant.components.monoprice.config_flow.get_async_monoprice",
"homeassistant.components.monoprice.config_flow.get_monoprice",
side_effect=Exception,
):
result2 = await hass.config_entries.flow.async_configure(

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"}