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

This commit is contained in:
Aaron Bach 2021-10-19 14:09:48 -06:00 committed by GitHub
parent 8e0cb5fcec
commit bf7c99c1f8
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
13 changed files with 366 additions and 388 deletions

View file

@ -3,25 +3,30 @@ from __future__ import annotations
import asyncio import asyncio
from collections.abc import Awaitable, Callable from collections.abc import Awaitable, Callable
from typing import cast from datetime import timedelta
from uuid import UUID from typing import TYPE_CHECKING, cast
from simplipy import get_api from simplipy import API
from simplipy.api import API from simplipy.device.sensor.v2 import SensorV2
from simplipy.device.sensor.v3 import SensorV3
from simplipy.errors import ( from simplipy.errors import (
EndpointUnavailableError, EndpointUnavailableError,
InvalidCredentialsError, InvalidCredentialsError,
SimplipyError, SimplipyError,
) )
from simplipy.sensor.v2 import SensorV2
from simplipy.sensor.v3 import SensorV3
from simplipy.system import SystemNotification from simplipy.system import SystemNotification
from simplipy.system.v2 import SystemV2 from simplipy.system.v2 import SystemV2
from simplipy.system.v3 import SystemV3 from simplipy.system.v3 import (
VOLUME_HIGH,
VOLUME_LOW,
VOLUME_MEDIUM,
VOLUME_OFF,
SystemV3,
)
import voluptuous as vol import voluptuous as vol
from homeassistant.config_entries import ConfigEntry from homeassistant.config_entries import ConfigEntry
from homeassistant.const import ATTR_CODE, CONF_CODE, CONF_PASSWORD, CONF_USERNAME from homeassistant.const import ATTR_CODE, CONF_CODE, CONF_TOKEN
from homeassistant.core import CoreState, HomeAssistant, ServiceCall, callback from homeassistant.core import CoreState, HomeAssistant, ServiceCall, callback
from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
from homeassistant.helpers import ( from homeassistant.helpers import (
@ -49,15 +54,15 @@ from .const import (
ATTR_EXIT_DELAY_HOME, ATTR_EXIT_DELAY_HOME,
ATTR_LIGHT, ATTR_LIGHT,
ATTR_VOICE_PROMPT_VOLUME, ATTR_VOICE_PROMPT_VOLUME,
CONF_USER_ID,
DATA_CLIENT, DATA_CLIENT,
DEFAULT_SCAN_INTERVAL,
DOMAIN, DOMAIN,
LOGGER, LOGGER,
VOLUMES,
) )
EVENT_SIMPLISAFE_NOTIFICATION = "SIMPLISAFE_NOTIFICATION" EVENT_SIMPLISAFE_NOTIFICATION = "SIMPLISAFE_NOTIFICATION"
DEFAULT_SCAN_INTERVAL = timedelta(seconds=30)
DEFAULT_SOCKET_MIN_RETRY = 15 DEFAULT_SOCKET_MIN_RETRY = 15
PLATFORMS = ( PLATFORMS = (
@ -75,6 +80,8 @@ ATTR_PIN_VALUE = "pin"
ATTR_SYSTEM_ID = "system_id" ATTR_SYSTEM_ID = "system_id"
ATTR_TIMESTAMP = "timestamp" ATTR_TIMESTAMP = "timestamp"
VOLUMES = [VOLUME_OFF, VOLUME_LOW, VOLUME_MEDIUM, VOLUME_HIGH]
SERVICE_BASE_SCHEMA = vol.Schema({vol.Required(ATTR_SYSTEM_ID): cv.positive_int}) SERVICE_BASE_SCHEMA = vol.Schema({vol.Required(ATTR_SYSTEM_ID): cv.positive_int})
SERVICE_REMOVE_PIN_SCHEMA = SERVICE_BASE_SCHEMA.extend( SERVICE_REMOVE_PIN_SCHEMA = SERVICE_BASE_SCHEMA.extend(
@ -120,13 +127,29 @@ SERVICE_SET_SYSTEM_PROPERTIES_SCHEMA = SERVICE_BASE_SCHEMA.extend(
CONFIG_SCHEMA = cv.deprecated(DOMAIN) CONFIG_SCHEMA = cv.deprecated(DOMAIN)
async def async_get_client_id(hass: HomeAssistant) -> str: @callback
"""Get a client ID (based on the HASS unique ID) for the SimpliSafe API. def _async_standardize_config_entry(hass: HomeAssistant, entry: ConfigEntry) -> None:
"""Bring a config entry up to current standards."""
if CONF_TOKEN not in entry.data:
raise ConfigEntryAuthFailed(
"New SimpliSafe OAuth standard requires re-authentication"
)
Note that SimpliSafe requires full, "dashed" versions of UUIDs. entry_updates = {}
""" if not entry.unique_id:
hass_id = await hass.helpers.instance_id.async_get() # If the config entry doesn't already have a unique ID, set one:
return str(UUID(hass_id)) entry_updates["unique_id"] = entry.data[CONF_USER_ID]
if CONF_CODE in entry.data:
# If an alarm code was provided as part of configuration.yaml, pop it out of
# the config entry's data and move it to options:
data = {**entry.data}
entry_updates["data"] = data
entry_updates["options"] = {
**entry.options,
CONF_CODE: data.pop(CONF_CODE),
}
if entry_updates:
hass.config_entries.async_update_entry(entry, **entry_updates)
async def async_register_base_station( async def async_register_base_station(
@ -143,47 +166,19 @@ async def async_register_base_station(
) )
@callback
def _async_standardize_config_entry(hass: HomeAssistant, entry: ConfigEntry) -> None:
"""Bring a config entry up to current standards."""
if CONF_PASSWORD not in entry.data:
raise ConfigEntryAuthFailed("Config schema change requires re-authentication")
entry_updates = {}
if not entry.unique_id:
# If the config entry doesn't already have a unique ID, set one:
entry_updates["unique_id"] = entry.data[CONF_USERNAME]
if CONF_CODE in entry.data:
# If an alarm code was provided as part of configuration.yaml, pop it out of
# the config entry's data and move it to options:
data = {**entry.data}
entry_updates["data"] = data
entry_updates["options"] = {
**entry.options,
CONF_CODE: data.pop(CONF_CODE),
}
if entry_updates:
hass.config_entries.async_update_entry(entry, **entry_updates)
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up SimpliSafe as config entry.""" """Set up SimpliSafe as config entry."""
hass.data.setdefault(DOMAIN, {DATA_CLIENT: {}}) hass.data.setdefault(DOMAIN, {})
hass.data[DOMAIN][DATA_CLIENT][entry.entry_id] = [] hass.data[DOMAIN][entry.entry_id] = {}
_async_standardize_config_entry(hass, entry) _async_standardize_config_entry(hass, entry)
_verify_domain_control = verify_domain_control(hass, DOMAIN) _verify_domain_control = verify_domain_control(hass, DOMAIN)
client_id = await async_get_client_id(hass)
websession = aiohttp_client.async_get_clientsession(hass) websession = aiohttp_client.async_get_clientsession(hass)
try: try:
api = await get_api( api = await API.async_from_refresh_token(
entry.data[CONF_USERNAME], entry.data[CONF_TOKEN], session=websession
entry.data[CONF_PASSWORD],
client_id=client_id,
session=websession,
) )
except InvalidCredentialsError as err: except InvalidCredentialsError as err:
raise ConfigEntryAuthFailed from err raise ConfigEntryAuthFailed from err
@ -198,7 +193,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
except SimplipyError as err: except SimplipyError as err:
raise ConfigEntryNotReady from err raise ConfigEntryNotReady from err
hass.data[DOMAIN][DATA_CLIENT][entry.entry_id] = simplisafe hass.data[DOMAIN][entry.entry_id][DATA_CLIENT] = simplisafe
hass.config_entries.async_setup_platforms(entry, PLATFORMS) hass.config_entries.async_setup_platforms(entry, PLATFORMS)
@callback @callback
@ -237,7 +232,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Clear all active notifications.""" """Clear all active notifications."""
system = simplisafe.systems[call.data[ATTR_SYSTEM_ID]] system = simplisafe.systems[call.data[ATTR_SYSTEM_ID]]
try: try:
await system.clear_notifications() await system.async_clear_notifications()
except SimplipyError as err: except SimplipyError as err:
LOGGER.error("Error during service call: %s", err) LOGGER.error("Error during service call: %s", err)
@ -247,7 +242,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Remove a PIN.""" """Remove a PIN."""
system = simplisafe.systems[call.data[ATTR_SYSTEM_ID]] system = simplisafe.systems[call.data[ATTR_SYSTEM_ID]]
try: try:
await system.remove_pin(call.data[ATTR_PIN_LABEL_OR_VALUE]) await system.async_remove_pin(call.data[ATTR_PIN_LABEL_OR_VALUE])
except SimplipyError as err: except SimplipyError as err:
LOGGER.error("Error during service call: %s", err) LOGGER.error("Error during service call: %s", err)
@ -257,7 +252,9 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set a PIN.""" """Set a PIN."""
system = simplisafe.systems[call.data[ATTR_SYSTEM_ID]] system = simplisafe.systems[call.data[ATTR_SYSTEM_ID]]
try: try:
await system.set_pin(call.data[ATTR_PIN_LABEL], call.data[ATTR_PIN_VALUE]) await system.async_set_pin(
call.data[ATTR_PIN_LABEL], call.data[ATTR_PIN_VALUE]
)
except SimplipyError as err: except SimplipyError as err:
LOGGER.error("Error during service call: %s", err) LOGGER.error("Error during service call: %s", err)
@ -268,7 +265,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set one or more system parameters.""" """Set one or more system parameters."""
system = cast(SystemV3, simplisafe.systems[call.data[ATTR_SYSTEM_ID]]) system = cast(SystemV3, simplisafe.systems[call.data[ATTR_SYSTEM_ID]])
try: try:
await system.set_properties( await system.async_set_properties(
{ {
prop: value prop: value
for prop, value in call.data.items() for prop, value in call.data.items()
@ -299,7 +296,7 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a SimpliSafe config entry.""" """Unload a SimpliSafe config entry."""
unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS) unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
if unload_ok: if unload_ok:
hass.data[DOMAIN][DATA_CLIENT].pop(entry.entry_id) hass.data[DOMAIN].pop(entry.entry_id)
return unload_ok return unload_ok
@ -362,7 +359,7 @@ class SimpliSafe:
async def async_init(self) -> None: async def async_init(self) -> None:
"""Initialize the data class.""" """Initialize the data class."""
self.systems = await self._api.get_systems() self.systems = await self._api.async_get_systems()
for system in self.systems.values(): for system in self.systems.values():
self._system_notifications[system.system_id] = set() self._system_notifications[system.system_id] = set()
@ -373,17 +370,34 @@ class SimpliSafe:
self.coordinator = DataUpdateCoordinator( self.coordinator = DataUpdateCoordinator(
self._hass, self._hass,
LOGGER, LOGGER,
name=self.entry.data[CONF_USERNAME], name=self.entry.data[CONF_USER_ID],
update_interval=DEFAULT_SCAN_INTERVAL, update_interval=DEFAULT_SCAN_INTERVAL,
update_method=self.async_update, update_method=self.async_update,
) )
@callback
def async_save_refresh_token(token: str) -> None:
"""Save a refresh token to the config entry."""
LOGGER.info("Saving new refresh token to HASS storage")
self._hass.config_entries.async_update_entry(
self.entry,
data={**self.entry.data, CONF_TOKEN: token},
)
self.entry.async_on_unload(
self._api.add_refresh_token_listener(async_save_refresh_token)
)
if TYPE_CHECKING:
assert self._api.refresh_token
async_save_refresh_token(self._api.refresh_token)
async def async_update(self) -> None: async def async_update(self) -> None:
"""Get updated data from SimpliSafe.""" """Get updated data from SimpliSafe."""
async def async_update_system(system: SystemV2 | SystemV3) -> None: async def async_update_system(system: SystemV2 | SystemV3) -> None:
"""Update a system.""" """Update a system."""
await system.update(cached=system.version != 3) await system.async_update(cached=system.version != 3)
self._async_process_new_notifications(system) self._async_process_new_notifications(system)
tasks = [async_update_system(system) for system in self.systems.values()] tasks = [async_update_system(system) for system in self.systems.values()]

View file

@ -4,7 +4,13 @@ from __future__ import annotations
from simplipy.errors import SimplipyError from simplipy.errors import SimplipyError
from simplipy.system import SystemStates from simplipy.system import SystemStates
from simplipy.system.v2 import SystemV2 from simplipy.system.v2 import SystemV2
from simplipy.system.v3 import SystemV3 from simplipy.system.v3 import (
VOLUME_HIGH,
VOLUME_LOW,
VOLUME_MEDIUM,
VOLUME_OFF,
SystemV3,
)
from homeassistant.components.alarm_control_panel import ( from homeassistant.components.alarm_control_panel import (
FORMAT_NUMBER, FORMAT_NUMBER,
@ -41,7 +47,6 @@ from .const import (
DATA_CLIENT, DATA_CLIENT,
DOMAIN, DOMAIN,
LOGGER, LOGGER,
VOLUME_STRING_MAP,
) )
ATTR_BATTERY_BACKUP_POWER_LEVEL = "battery_backup_power_level" ATTR_BATTERY_BACKUP_POWER_LEVEL = "battery_backup_power_level"
@ -51,12 +56,19 @@ ATTR_RF_JAMMING = "rf_jamming"
ATTR_WALL_POWER_LEVEL = "wall_power_level" ATTR_WALL_POWER_LEVEL = "wall_power_level"
ATTR_WIFI_STRENGTH = "wifi_strength" ATTR_WIFI_STRENGTH = "wifi_strength"
VOLUME_STRING_MAP = {
VOLUME_HIGH: "high",
VOLUME_LOW: "low",
VOLUME_MEDIUM: "medium",
VOLUME_OFF: "off",
}
async def async_setup_entry( async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None: ) -> None:
"""Set up a SimpliSafe alarm control panel based on a config entry.""" """Set up a SimpliSafe alarm control panel based on a config entry."""
simplisafe = hass.data[DOMAIN][DATA_CLIENT][entry.entry_id] simplisafe = hass.data[DOMAIN][entry.entry_id][DATA_CLIENT]
async_add_entities( async_add_entities(
[SimpliSafeAlarm(simplisafe, system) for system in simplisafe.systems.values()], [SimpliSafeAlarm(simplisafe, system) for system in simplisafe.systems.values()],
True, True,
@ -115,7 +127,7 @@ class SimpliSafeAlarm(SimpliSafeEntity, AlarmControlPanelEntity):
return return
try: try:
await self._system.set_off() await self._system.async_set_off()
except SimplipyError as err: except SimplipyError as err:
LOGGER.error('Error while disarming "%s": %s', self._system.system_id, err) LOGGER.error('Error while disarming "%s": %s', self._system.system_id, err)
return return
@ -129,7 +141,7 @@ class SimpliSafeAlarm(SimpliSafeEntity, AlarmControlPanelEntity):
return return
try: try:
await self._system.set_home() await self._system.async_set_home()
except SimplipyError as err: except SimplipyError as err:
LOGGER.error( LOGGER.error(
'Error while arming "%s" (home): %s', self._system.system_id, err 'Error while arming "%s" (home): %s', self._system.system_id, err
@ -145,7 +157,7 @@ class SimpliSafeAlarm(SimpliSafeEntity, AlarmControlPanelEntity):
return return
try: try:
await self._system.set_away() await self._system.async_set_away()
except SimplipyError as err: except SimplipyError as err:
LOGGER.error( LOGGER.error(
'Error while arming "%s" (away): %s', self._system.system_id, err 'Error while arming "%s" (away): %s', self._system.system_id, err

View file

@ -1,7 +1,9 @@
"""Support for SimpliSafe binary sensors.""" """Support for SimpliSafe binary sensors."""
from __future__ import annotations from __future__ import annotations
from simplipy.entity import Entity as SimplipyEntity, EntityTypes from simplipy.device import DeviceTypes
from simplipy.device.sensor.v2 import SensorV2
from simplipy.device.sensor.v3 import SensorV3
from simplipy.system.v2 import SystemV2 from simplipy.system.v2 import SystemV2
from simplipy.system.v3 import SystemV3 from simplipy.system.v3 import SystemV3
@ -23,25 +25,25 @@ from . import SimpliSafe, SimpliSafeBaseSensor
from .const import DATA_CLIENT, DOMAIN, LOGGER from .const import DATA_CLIENT, DOMAIN, LOGGER
SUPPORTED_BATTERY_SENSOR_TYPES = [ SUPPORTED_BATTERY_SENSOR_TYPES = [
EntityTypes.carbon_monoxide, DeviceTypes.carbon_monoxide,
EntityTypes.entry, DeviceTypes.entry,
EntityTypes.glass_break, DeviceTypes.glass_break,
EntityTypes.leak, DeviceTypes.leak,
EntityTypes.lock_keypad, DeviceTypes.lock_keypad,
EntityTypes.motion, DeviceTypes.motion,
EntityTypes.siren, DeviceTypes.siren,
EntityTypes.smoke, DeviceTypes.smoke,
EntityTypes.temperature, DeviceTypes.temperature,
] ]
TRIGGERED_SENSOR_TYPES = { TRIGGERED_SENSOR_TYPES = {
EntityTypes.carbon_monoxide: DEVICE_CLASS_GAS, DeviceTypes.carbon_monoxide: DEVICE_CLASS_GAS,
EntityTypes.entry: DEVICE_CLASS_DOOR, DeviceTypes.entry: DEVICE_CLASS_DOOR,
EntityTypes.glass_break: DEVICE_CLASS_SAFETY, DeviceTypes.glass_break: DEVICE_CLASS_SAFETY,
EntityTypes.leak: DEVICE_CLASS_MOISTURE, DeviceTypes.leak: DEVICE_CLASS_MOISTURE,
EntityTypes.motion: DEVICE_CLASS_MOTION, DeviceTypes.motion: DEVICE_CLASS_MOTION,
EntityTypes.siren: DEVICE_CLASS_SAFETY, DeviceTypes.siren: DEVICE_CLASS_SAFETY,
EntityTypes.smoke: DEVICE_CLASS_SMOKE, DeviceTypes.smoke: DEVICE_CLASS_SMOKE,
} }
@ -49,7 +51,7 @@ async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None: ) -> None:
"""Set up SimpliSafe binary sensors based on a config entry.""" """Set up SimpliSafe binary sensors based on a config entry."""
simplisafe = hass.data[DOMAIN][DATA_CLIENT][entry.entry_id] simplisafe = hass.data[DOMAIN][entry.entry_id][DATA_CLIENT]
sensors: list[BatteryBinarySensor | TriggeredBinarySensor] = [] sensors: list[BatteryBinarySensor | TriggeredBinarySensor] = []
@ -81,7 +83,7 @@ class TriggeredBinarySensor(SimpliSafeBaseSensor, BinarySensorEntity):
self, self,
simplisafe: SimpliSafe, simplisafe: SimpliSafe,
system: SystemV2 | SystemV3, system: SystemV2 | SystemV3,
sensor: SimplipyEntity, sensor: SensorV2 | SensorV3,
device_class: str, device_class: str,
) -> None: ) -> None:
"""Initialize.""" """Initialize."""
@ -104,7 +106,7 @@ class BatteryBinarySensor(SimpliSafeBaseSensor, BinarySensorEntity):
self, self,
simplisafe: SimpliSafe, simplisafe: SimpliSafe,
system: SystemV2 | SystemV3, system: SystemV2 | SystemV3,
sensor: SimplipyEntity, sensor: SensorV2 | SensorV3,
) -> None: ) -> None:
"""Initialize.""" """Initialize."""
super().__init__(simplisafe, system, sensor) super().__init__(simplisafe, system, sensor)

View file

@ -1,36 +1,50 @@
"""Config flow to configure the SimpliSafe component.""" """Config flow to configure the SimpliSafe component."""
from __future__ import annotations from __future__ import annotations
from typing import Any from typing import TYPE_CHECKING, Any, NamedTuple
from simplipy import get_api from simplipy import API
from simplipy.api import API from simplipy.errors import InvalidCredentialsError, SimplipyError
from simplipy.errors import ( from simplipy.util.auth import (
InvalidCredentialsError, get_auth0_code_challenge,
PendingAuthorizationError, get_auth0_code_verifier,
SimplipyError, get_auth_url,
) )
import voluptuous as vol import voluptuous as vol
from homeassistant import config_entries from homeassistant import config_entries
from homeassistant.config_entries import ConfigEntry from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_CODE, CONF_PASSWORD, CONF_USERNAME from homeassistant.const import CONF_CODE, CONF_TOKEN, CONF_URL, CONF_USERNAME
from homeassistant.core import callback from homeassistant.core import callback
from homeassistant.data_entry_flow import FlowResult from homeassistant.data_entry_flow import FlowResult
from homeassistant.helpers import aiohttp_client from homeassistant.helpers import aiohttp_client, config_validation as cv
from homeassistant.helpers.typing import ConfigType from homeassistant.helpers.typing import ConfigType
from . import async_get_client_id from .const import CONF_USER_ID, DOMAIN, LOGGER
from .const import DOMAIN, LOGGER
FULL_DATA_SCHEMA = vol.Schema( CONF_AUTH_CODE = "auth_code"
STEP_INPUT_AUTH_CODE_SCHEMA = vol.Schema(
{ {
vol.Required(CONF_USERNAME): str, vol.Required(CONF_AUTH_CODE): cv.string,
vol.Required(CONF_PASSWORD): str,
vol.Optional(CONF_CODE): str,
} }
) )
PASSWORD_DATA_SCHEMA = vol.Schema({vol.Required(CONF_PASSWORD): str})
class SimpliSafeOAuthValues(NamedTuple):
"""Define a named tuple to handle SimpliSafe OAuth strings."""
code_verifier: str
auth_url: 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(code_verifier, auth_url)
class SimpliSafeFlowHandler(config_entries.ConfigFlow, domain=DOMAIN): class SimpliSafeFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
@ -40,8 +54,9 @@ class SimpliSafeFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
def __init__(self) -> None: def __init__(self) -> None:
"""Initialize the config flow.""" """Initialize the config flow."""
self._code: str | None = None self._errors: dict[str, Any] = {}
self._password: str | None = None self._oauth_values: SimpliSafeOAuthValues = async_get_simplisafe_oauth_values()
self._reauth: bool = False
self._username: str | None = None self._username: str | None = None
@staticmethod @staticmethod
@ -52,128 +67,79 @@ class SimpliSafeFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
"""Define the config flow to handle options.""" """Define the config flow to handle options."""
return SimpliSafeOptionsFlowHandler(config_entry) return SimpliSafeOptionsFlowHandler(config_entry)
async def _async_get_simplisafe_api(self) -> API: async def async_step_input_auth_code(
"""Get an authenticated SimpliSafe API client.""" self, user_input: dict[str, Any] | None = None
assert self._username
assert self._password
client_id = await async_get_client_id(self.hass)
websession = aiohttp_client.async_get_clientsession(self.hass)
return await get_api(
self._username,
self._password,
client_id=client_id,
session=websession,
)
async def _async_login_during_step(
self, *, step_id: str, form_schema: vol.Schema
) -> FlowResult: ) -> FlowResult:
"""Attempt to log into the API from within a config flow step.""" """Handle the input of a SimpliSafe OAuth authorization code."""
errors = {} if user_input is None:
try:
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_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( return self.async_show_form(
step_id=step_id, step_id="input_auth_code", data_schema=STEP_INPUT_AUTH_CODE_SCHEMA
data_schema=form_schema,
errors=errors,
) )
return await self.async_step_finish( if TYPE_CHECKING:
{ assert self._oauth_values
CONF_USERNAME: self._username,
CONF_PASSWORD: self._password,
CONF_CODE: self._code,
}
)
async def async_step_finish(self, user_input: dict[str, Any]) -> FlowResult: self._errors = {}
"""Handle finish config entry setup.""" session = aiohttp_client.async_get_clientsession(self.hass)
assert self._username
existing_entry = await self.async_set_unique_id(self._username) try:
if existing_entry: simplisafe = await API.async_from_auth(
self.hass.config_entries.async_update_entry(existing_entry, data=user_input) user_input[CONF_AUTH_CODE],
self._oauth_values.code_verifier,
session=session,
)
except InvalidCredentialsError:
self._errors = {"base": "invalid_auth"}
except SimplipyError as err:
LOGGER.error("Unknown error while logging into SimpliSafe: %s", err)
self._errors = {"base": "unknown"}
if self._errors:
return await self.async_step_user()
data = {CONF_USER_ID: simplisafe.user_id, CONF_TOKEN: simplisafe.refresh_token}
unique_id = str(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
# either one is a candidate for re-auth:
existing_entry = await self.async_set_unique_id(self._username or unique_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=unique_id, data=data
)
self.hass.async_create_task( self.hass.async_create_task(
self.hass.config_entries.async_reload(existing_entry.entry_id) self.hass.config_entries.async_reload(existing_entry.entry_id)
) )
return self.async_abort(reason="reauth_successful") return self.async_abort(reason="reauth_successful")
return self.async_create_entry(title=self._username, data=user_input)
async def async_step_mfa( await self.async_set_unique_id(unique_id)
self, user_input: dict[str, Any] | None = None self._abort_if_unique_id_configured()
) -> FlowResult: return self.async_create_entry(title=unique_id, data=data)
"""Handle multi-factor auth confirmation."""
if user_input is None:
return self.async_show_form(step_id="mfa")
try:
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_PASSWORD: self._password,
CONF_CODE: self._code,
}
)
async def async_step_reauth(self, config: ConfigType) -> FlowResult: async def async_step_reauth(self, config: ConfigType) -> FlowResult:
"""Handle configuration by re-auth.""" """Handle configuration by re-auth."""
self._code = config.get(CONF_CODE) self._username = config.get(CONF_USERNAME)
self._username = config[CONF_USERNAME] self._reauth = True
return await self.async_step_user()
return await self.async_step_reauth_confirm()
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=PASSWORD_DATA_SCHEMA
)
self._password = user_input[CONF_PASSWORD]
return await self._async_login_during_step(
step_id="reauth_confirm", form_schema=PASSWORD_DATA_SCHEMA
)
async def async_step_user( async def async_step_user(
self, user_input: dict[str, Any] | None = None self, user_input: dict[str, Any] | None = None
) -> FlowResult: ) -> FlowResult:
"""Handle the start of the config flow.""" """Handle the start of the config flow."""
if not user_input: if user_input is None:
return self.async_show_form(step_id="user", data_schema=FULL_DATA_SCHEMA) return self.async_show_form(
step_id="user",
errors=self._errors,
description_placeholders={CONF_URL: self._oauth_values.auth_url},
)
await self.async_set_unique_id(user_input[CONF_USERNAME]) return await self.async_step_input_auth_code()
self._abort_if_unique_id_configured()
self._code = user_input.get(CONF_CODE)
self._password = user_input[CONF_PASSWORD]
self._username = user_input[CONF_USERNAME]
return await self._async_login_during_step(
step_id="user", form_schema=FULL_DATA_SCHEMA
)
class SimpliSafeOptionsFlowHandler(config_entries.OptionsFlow): class SimpliSafeOptionsFlowHandler(config_entries.OptionsFlow):

View file

@ -1,17 +1,10 @@
"""Define constants for the SimpliSafe component.""" """Define constants for the SimpliSafe component."""
from datetime import timedelta
import logging import logging
from simplipy.system.v3 import VOLUME_HIGH, VOLUME_LOW, VOLUME_MEDIUM, VOLUME_OFF
LOGGER = logging.getLogger(__package__) LOGGER = logging.getLogger(__package__)
DOMAIN = "simplisafe" DOMAIN = "simplisafe"
DATA_CLIENT = "client"
DEFAULT_SCAN_INTERVAL = timedelta(seconds=30)
ATTR_ALARM_DURATION = "alarm_duration" ATTR_ALARM_DURATION = "alarm_duration"
ATTR_ALARM_VOLUME = "alarm_volume" ATTR_ALARM_VOLUME = "alarm_volume"
ATTR_CHIME_VOLUME = "chime_volume" ATTR_CHIME_VOLUME = "chime_volume"
@ -22,10 +15,6 @@ ATTR_EXIT_DELAY_HOME = "exit_delay_home"
ATTR_LIGHT = "light" ATTR_LIGHT = "light"
ATTR_VOICE_PROMPT_VOLUME = "voice_prompt_volume" ATTR_VOICE_PROMPT_VOLUME = "voice_prompt_volume"
VOLUMES = [VOLUME_OFF, VOLUME_LOW, VOLUME_MEDIUM, VOLUME_HIGH] CONF_USER_ID = "user_id"
VOLUME_STRING_MAP = {
VOLUME_HIGH: "high", DATA_CLIENT = "client"
VOLUME_LOW: "low",
VOLUME_MEDIUM: "medium",
VOLUME_OFF: "off",
}

View file

@ -3,8 +3,8 @@ from __future__ import annotations
from typing import Any from typing import Any
from simplipy.device.lock import Lock, LockStates
from simplipy.errors import SimplipyError from simplipy.errors import SimplipyError
from simplipy.lock import Lock, LockStates
from simplipy.system.v3 import SystemV3 from simplipy.system.v3 import SystemV3
from homeassistant.components.lock import LockEntity from homeassistant.components.lock import LockEntity
@ -23,7 +23,7 @@ async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None: ) -> None:
"""Set up SimpliSafe locks based on a config entry.""" """Set up SimpliSafe locks based on a config entry."""
simplisafe = hass.data[DOMAIN][DATA_CLIENT][entry.entry_id] simplisafe = hass.data[DOMAIN][entry.entry_id][DATA_CLIENT]
locks = [] locks = []
for system in simplisafe.systems.values(): for system in simplisafe.systems.values():
@ -49,7 +49,7 @@ class SimpliSafeLock(SimpliSafeEntity, LockEntity):
async def async_lock(self, **kwargs: Any) -> None: async def async_lock(self, **kwargs: Any) -> None:
"""Lock the lock.""" """Lock the lock."""
try: try:
await self._lock.lock() await self._lock.async_lock()
except SimplipyError as err: 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 return
@ -60,7 +60,7 @@ class SimpliSafeLock(SimpliSafeEntity, LockEntity):
async def async_unlock(self, **kwargs: Any) -> None: async def async_unlock(self, **kwargs: Any) -> None:
"""Unlock the lock.""" """Unlock the lock."""
try: try:
await self._lock.unlock() await self._lock.async_unlock()
except SimplipyError as err: 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 return

View file

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

View file

@ -1,5 +1,8 @@
"""Support for SimpliSafe freeze sensor.""" """Support for SimpliSafe freeze sensor."""
from simplipy.entity import EntityTypes from typing import TYPE_CHECKING
from simplipy.device import DeviceTypes
from simplipy.device.sensor.v3 import SensorV3
from homeassistant.components.sensor import STATE_CLASS_MEASUREMENT, SensorEntity from homeassistant.components.sensor import STATE_CLASS_MEASUREMENT, SensorEntity
from homeassistant.config_entries import ConfigEntry from homeassistant.config_entries import ConfigEntry
@ -15,7 +18,7 @@ async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None: ) -> None:
"""Set up SimpliSafe freeze sensors based on a config entry.""" """Set up SimpliSafe freeze sensors based on a config entry."""
simplisafe = hass.data[DOMAIN][DATA_CLIENT][entry.entry_id] simplisafe = hass.data[DOMAIN][entry.entry_id][DATA_CLIENT]
sensors = [] sensors = []
for system in simplisafe.systems.values(): for system in simplisafe.systems.values():
@ -24,7 +27,7 @@ async def async_setup_entry(
continue continue
for sensor in system.sensors.values(): for sensor in system.sensors.values():
if sensor.type == EntityTypes.temperature: if sensor.type == DeviceTypes.temperature:
sensors.append(SimplisafeFreezeSensor(simplisafe, system, sensor)) sensors.append(SimplisafeFreezeSensor(simplisafe, system, sensor))
async_add_entities(sensors) async_add_entities(sensors)
@ -40,4 +43,6 @@ class SimplisafeFreezeSensor(SimpliSafeBaseSensor, SensorEntity):
@callback @callback
def async_update_from_rest_api(self) -> None: def async_update_from_rest_api(self) -> None:
"""Update the entity with the provided REST API data.""" """Update the entity with the provided REST API data."""
if TYPE_CHECKING:
assert isinstance(self._sensor, SensorV3)
self._attr_native_value = self._sensor.temperature self._attr_native_value = self._sensor.temperature

View file

@ -1,24 +1,15 @@
{ {
"config": { "config": {
"step": { "step": {
"mfa": { "input_auth_code": {
"title": "SimpliSafe Multi-Factor Authentication", "title": "Finish Authorization",
"description": "Check your email for a link from SimpliSafe. After verifying the link, return here to complete the installation of the integration." "description": "Input the authorization code from the SimpliSafe web app URL:",
},
"reauth_confirm": {
"title": "[%key:common::config_flow::title::reauth%]",
"description": "Your access has expired or been revoked. Enter your password to re-link your account.",
"data": { "data": {
"password": "[%key:common::config_flow::data::password%]" "auth_code": "Authorization Code"
} }
}, },
"user": { "user": {
"title": "Fill in your information.", "description": "Starting in 2021, SimpliSafe has moved to a new authentication mechanism 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 click Submit."
"data": {
"username": "[%key:common::config_flow::data::email%]",
"password": "[%key:common::config_flow::data::password%]",
"code": "Code (used in Home Assistant UI)"
}
} }
}, },
"error": { "error": {
@ -29,7 +20,8 @@
}, },
"abort": { "abort": {
"already_configured": "This SimpliSafe account is already in use.", "already_configured": "This SimpliSafe account is already in use.",
"reauth_successful": "[%key:common::config_flow::abort::reauth_successful%]" "reauth_successful": "[%key:common::config_flow::abort::reauth_successful%]",
"wrong_account": "The user credentials provided do not match this SimpliSafe account."
} }
}, },
"options": { "options": {

View file

@ -2,7 +2,8 @@
"config": { "config": {
"abort": { "abort": {
"already_configured": "This SimpliSafe account is already in use.", "already_configured": "This SimpliSafe account is already in use.",
"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": { "error": {
"identifier_exists": "Account already registered", "identifier_exists": "Account already registered",
@ -11,24 +12,15 @@
"unknown": "Unexpected error" "unknown": "Unexpected error"
}, },
"step": { "step": {
"mfa": { "input_auth_code": {
"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": { "data": {
"password": "Password" "auth_code": "Authorization Code"
}, },
"description": "Your access has expired or been revoked. Enter your password to re-link your account.", "description": "Input the authorization code from the SimpliSafe web app URL:",
"title": "Reauthenticate Integration" "title": "Finish Authorization"
}, },
"user": { "user": {
"data": { "description": "Starting in 2021, SimpliSafe has moved to a new authentication mechanism 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 click Submit."
"code": "Code (used in Home Assistant UI)",
"password": "Password",
"username": "Email"
},
"title": "Fill in your information."
} }
} }
}, },

View file

@ -2137,7 +2137,7 @@ simplehound==0.3
simplepush==1.1.4 simplepush==1.1.4
# homeassistant.components.simplisafe # homeassistant.components.simplisafe
simplisafe-python==11.0.7 simplisafe-python==12.0.0
# homeassistant.components.sisyphus # homeassistant.components.sisyphus
sisyphus-control==3.0 sisyphus-control==3.0

View file

@ -1233,7 +1233,7 @@ sharkiqpy==0.1.8
simplehound==0.3 simplehound==0.3
# homeassistant.components.simplisafe # homeassistant.components.simplisafe
simplisafe-python==11.0.7 simplisafe-python==12.0.0
# homeassistant.components.slack # homeassistant.components.slack
slackclient==2.5.0 slackclient==2.5.0

View file

@ -1,68 +1,94 @@
"""Define tests for the SimpliSafe config flow.""" """Define tests for the SimpliSafe config flow."""
from unittest.mock import AsyncMock, patch from unittest.mock import AsyncMock, Mock, patch
from simplipy.errors import ( import pytest
InvalidCredentialsError, from simplipy.errors import InvalidCredentialsError, SimplipyError
PendingAuthorizationError,
SimplipyError,
)
from homeassistant import data_entry_flow from homeassistant import data_entry_flow
from homeassistant.components.simplisafe import DOMAIN from homeassistant.components.simplisafe import DOMAIN
from homeassistant.components.simplisafe.config_flow import CONF_AUTH_CODE
from homeassistant.components.simplisafe.const import CONF_USER_ID
from homeassistant.config_entries import SOURCE_REAUTH, SOURCE_USER from homeassistant.config_entries import SOURCE_REAUTH, SOURCE_USER
from homeassistant.const import CONF_CODE, CONF_PASSWORD, CONF_USERNAME from homeassistant.const import CONF_CODE, CONF_PASSWORD, CONF_TOKEN, CONF_USERNAME
from tests.common import MockConfigEntry from tests.common import MockConfigEntry
async def test_duplicate_error(hass): @pytest.fixture(name="api")
"""Test that errors are shown when duplicates are added.""" def api_fixture():
conf = { """Define a fixture for simplisafe-python API object."""
CONF_USERNAME: "user@email.com", api = Mock()
CONF_PASSWORD: "password", api.refresh_token = "token123"
CONF_CODE: "1234", api.user_id = "12345"
} return api
@pytest.fixture(name="mock_async_from_auth")
def mock_async_from_auth_fixture(api):
"""Define a fixture for simplipy.API.async_from_auth."""
with patch(
"homeassistant.components.simplisafe.config_flow.API.async_from_auth",
) as mock_async_from_auth:
mock_async_from_auth.side_effect = AsyncMock(return_value=api)
yield mock_async_from_auth
async def test_duplicate_error(hass, mock_async_from_auth):
"""Test that errors are shown when duplicates are added."""
MockConfigEntry( MockConfigEntry(
domain=DOMAIN, domain=DOMAIN,
unique_id="user@email.com", unique_id="12345",
data={ data={
CONF_USERNAME: "user@email.com", CONF_USER_ID: "12345",
CONF_PASSWORD: "password", CONF_TOKEN: "token123",
CONF_CODE: "1234",
}, },
).add_to_hass(hass) ).add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}, data=conf
)
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "already_configured"
async def test_invalid_credentials(hass):
"""Test that invalid credentials throws an error."""
conf = {CONF_USERNAME: "user@email.com", CONF_PASSWORD: "password"}
with patch( with patch(
"homeassistant.components.simplisafe.config_flow.get_api", "homeassistant.components.simplisafe.async_setup_entry", return_value=True
new=AsyncMock(side_effect=InvalidCredentialsError),
): ):
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}, data=conf DOMAIN, context={"source": SOURCE_USER}
) )
assert result["errors"] == {"base": "invalid_auth"} 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={}
)
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_invalid_credentials(hass, mock_async_from_auth):
"""Test that invalid credentials show the correct error."""
mock_async_from_auth.side_effect = AsyncMock(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={}
)
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(hass): async def test_options_flow(hass):
"""Test config flow options.""" """Test config flow options."""
conf = {CONF_USERNAME: "user@email.com", CONF_PASSWORD: "password"}
entry = MockConfigEntry( entry = MockConfigEntry(
domain=DOMAIN, domain=DOMAIN,
unique_id="abcde12345", unique_id="abcde12345",
data=conf, data={CONF_USER_ID: "12345", CONF_TOKEN: "token456"},
options={CONF_CODE: "1234"}, options={CONF_CODE: "1234"},
) )
entry.add_to_hass(hass) entry.add_to_hass(hass)
@ -84,134 +110,114 @@ async def test_options_flow(hass):
assert entry.options == {CONF_CODE: "4321"} assert entry.options == {CONF_CODE: "4321"}
async def test_show_form(hass): async def test_step_reauth_old_format(hass, mock_async_from_auth):
"""Test that the form is served with no input.""" """Test the re-auth step with "old" config entries (those with user IDs)."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "user"
async def test_step_reauth(hass):
"""Test that the reauth step works."""
MockConfigEntry( MockConfigEntry(
domain=DOMAIN, domain=DOMAIN,
unique_id="user@email.com", unique_id="user@email.com",
data={ data={
CONF_USERNAME: "user@email.com", CONF_USERNAME: "user@email.com",
CONF_PASSWORD: "password", CONF_PASSWORD: "password",
CONF_CODE: "1234",
}, },
).add_to_hass(hass) ).add_to_hass(hass)
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
DOMAIN, DOMAIN,
context={"source": SOURCE_REAUTH}, context={"source": SOURCE_REAUTH},
data={CONF_CODE: "1234", CONF_USERNAME: "user@email.com"}, data={CONF_USERNAME: "user@email.com", CONF_PASSWORD: "password"},
) )
assert result["step_id"] == "reauth_confirm" assert result["step_id"] == "user"
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( with patch(
"homeassistant.components.simplisafe.async_setup_entry", return_value=True "homeassistant.components.simplisafe.async_setup_entry", return_value=True
), patch("homeassistant.components.simplisafe.config_flow.get_api"), patch( ), patch("homeassistant.config_entries.ConfigEntries.async_reload"):
"homeassistant.config_entries.ConfigEntries.async_reload"
):
result = await hass.config_entries.flow.async_configure( result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={CONF_PASSWORD: "password"} result["flow_id"], user_input={}
)
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["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "reauth_successful" assert result["reason"] == "reauth_successful"
assert len(hass.config_entries.async_entries()) == 1 assert len(hass.config_entries.async_entries()) == 1
[config_entry] = hass.config_entries.async_entries(DOMAIN)
assert config_entry.data == {CONF_USER_ID: "12345", CONF_TOKEN: "token123"}
async def test_step_user(hass): async def test_step_reauth_new_format(hass, mock_async_from_auth):
"""Test that the user step works (without MFA).""" """Test the re-auth step with "new" config entries (those with user IDs)."""
conf = { MockConfigEntry(
CONF_USERNAME: "user@email.com", domain=DOMAIN,
CONF_PASSWORD: "password", unique_id="12345",
CONF_CODE: "1234", data={
} CONF_USER_ID: "12345",
CONF_TOKEN: "token123",
},
).add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_REAUTH},
data={CONF_USER_ID: "12345", CONF_TOKEN: "token123"},
)
assert result["step_id"] == "user"
with patch( with patch(
"homeassistant.components.simplisafe.async_setup_entry", return_value=True "homeassistant.components.simplisafe.async_setup_entry", return_value=True
), patch( ), patch("homeassistant.config_entries.ConfigEntries.async_reload"):
"homeassistant.components.simplisafe.config_flow.get_api", new=AsyncMock()
):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}, data=conf
)
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_PASSWORD: "password",
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(
"homeassistant.components.simplisafe.config_flow.get_api",
new=AsyncMock(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(
"homeassistant.components.simplisafe.config_flow.get_api",
new=AsyncMock(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 = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={} result["flow_id"], user_input={}
) )
assert result["step_id"] == "mfa" 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.data == {CONF_USER_ID: "12345", CONF_TOKEN: "token123"}
async def test_step_user(hass, mock_async_from_auth):
"""Test the user step."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["step_id"] == "user"
with patch( with patch(
"homeassistant.components.simplisafe.async_setup_entry", return_value=True "homeassistant.components.simplisafe.async_setup_entry", return_value=True
), patch( ), patch("homeassistant.config_entries.ConfigEntries.async_reload"):
"homeassistant.components.simplisafe.config_flow.get_api", new=AsyncMock()
):
result = await hass.config_entries.flow.async_configure( result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={} result["flow_id"], user_input={}
) )
result = await hass.config_entries.flow.async_configure(
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY result["flow_id"], user_input={CONF_AUTH_CODE: "code123"}
assert result["title"] == "user@email.com"
assert result["data"] == {
CONF_USERNAME: "user@email.com",
CONF_PASSWORD: "password",
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(
"homeassistant.components.simplisafe.config_flow.get_api",
new=AsyncMock(side_effect=SimplipyError),
):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}, data=conf
) )
assert result["errors"] == {"base": "unknown"} 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_USER_ID: "12345", CONF_TOKEN: "token123"}
async def test_unknown_error(hass, mock_async_from_auth):
"""Test that an unknown error shows ohe correct error."""
mock_async_from_auth.side_effect = AsyncMock(side_effect=SimplipyError)
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={}
)
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": "unknown"}