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

Use auth token in Ezviz (#54663)

Browse Source 
* Initial commit

* Revert "Initial commit"

This reverts commit 452027f1a3c1be186cedd4115cea6928917c9467.

* Change ezviz to token auth

* Bump API version.

* Add fix for token expired. Fix options update and unload.

* Fix tests (PLATFORM to PLATFORM_BY_TYPE)

* Uses and stores token only, added reauth step when token expires.

* Add tests MFA code exceptions.

* Fix tests.

* Remove redundant try/except blocks.

* Rebase fixes.

* Fix errors in reauth config flow

* Implement recommendations

* Fix typing error in config_flow

* Fix tests after rebase, readd camera check on init

* Change to platform setup

* Cleanup init.

* Test for MFA required under user form

* Remove useless if block.

* Fix formating after rebase

* Fix formating.

* No longer stored in the repository

---------

Co-authored-by: Paulus Schoutsen <paulus@home-assistant.io>
This commit is contained in:
RenierM26 2023-03-29 23:43:54 +02:00 committed by GitHub
parent 4c21caa917
commit 93d1961aae
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
9 changed files with 535 additions and 217 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

137
homeassistant/components/ezviz/__init__.py
View File

@ -2,26 +2,26 @@
import logging import logging
from pyezviz.client import EzvizClient from pyezviz.client import EzvizClient
from pyezviz.exceptions import HTTPError, InvalidURL, PyEzvizError from pyezviz.exceptions import (
EzvizAuthTokenExpired,
EzvizAuthVerificationCode,
HTTPError,
InvalidURL,
PyEzvizError,
)
from homeassistant.config_entries import ConfigEntry from homeassistant.config_entries import ConfigEntry
from homeassistant.const import ( from homeassistant.const import CONF_TIMEOUT, CONF_TYPE, CONF_URL, Platform
CONF_PASSWORD,
CONF_TIMEOUT,
CONF_TYPE,
CONF_URL,
CONF_USERNAME,
Platform,
)
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
from .const import ( from .const import (
ATTR_TYPE_CAMERA, ATTR_TYPE_CAMERA,
ATTR_TYPE_CLOUD, ATTR_TYPE_CLOUD,
CONF_FFMPEG_ARGUMENTS, CONF_FFMPEG_ARGUMENTS,
CONF_RFSESSION_ID,
CONF_SESSION_ID,
DATA_COORDINATOR, DATA_COORDINATOR,
DATA_UNDO_UPDATE_LISTENER,
DEFAULT_FFMPEG_ARGUMENTS, DEFAULT_FFMPEG_ARGUMENTS,
DEFAULT_TIMEOUT, DEFAULT_TIMEOUT,
DOMAIN, DOMAIN,
@ -30,17 +30,22 @@ from .coordinator import EzvizDataUpdateCoordinator
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
PLATFORMS = [ PLATFORMS_BY_TYPE: dict[str, list] = {
Platform.BINARY_SENSOR, ATTR_TYPE_CAMERA: [],
Platform.CAMERA, ATTR_TYPE_CLOUD: [
Platform.SENSOR, Platform.BINARY_SENSOR,
Platform.SWITCH, Platform.CAMERA,
] Platform.SENSOR,
Platform.SWITCH,
],
}
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up EZVIZ from a config entry.""" """Set up EZVIZ from a config entry."""
hass.data.setdefault(DOMAIN, {}) hass.data.setdefault(DOMAIN, {})
sensor_type: str = entry.data[CONF_TYPE]
ezviz_client = None
if not entry.options: if not entry.options:
options = { options = {
@ -50,69 +55,71 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
hass.config_entries.async_update_entry(entry, options=options) hass.config_entries.async_update_entry(entry, options=options)
if entry.data.get(CONF_TYPE) == ATTR_TYPE_CAMERA: # Initialize EZVIZ cloud entities
if hass.data.get(DOMAIN): if PLATFORMS_BY_TYPE[sensor_type]:
# Should only execute on addition of new camera entry. # Initiate reauth config flow if account token if not present.
# Fetch Entry id of main account and reload it. if not entry.data.get(CONF_SESSION_ID):
for item in hass.config_entries.async_entries(): raise ConfigEntryAuthFailed
if item.data.get(CONF_TYPE) == ATTR_TYPE_CLOUD:
_LOGGER.info("Reload EZVIZ integration with new camera rtsp entry")
await hass.config_entries.async_reload(item.entry_id)
return True ezviz_client = EzvizClient(
token={
try: CONF_SESSION_ID: entry.data.get(CONF_SESSION_ID),
ezviz_client = await hass.async_add_executor_job( CONF_RFSESSION_ID: entry.data.get(CONF_RFSESSION_ID),
_get_ezviz_client_instance, entry "api_url": entry.data.get(CONF_URL),
},
timeout=entry.options.get(CONF_TIMEOUT, DEFAULT_TIMEOUT),
) )
except (InvalidURL, HTTPError, PyEzvizError) as error:
_LOGGER.error("Unable to connect to EZVIZ service: %s", str(error))
raise ConfigEntryNotReady from error
coordinator = EzvizDataUpdateCoordinator( try:
hass, api=ezviz_client, api_timeout=entry.options[CONF_TIMEOUT] await hass.async_add_executor_job(ezviz_client.login)
except (EzvizAuthTokenExpired, EzvizAuthVerificationCode) as error:
raise ConfigEntryAuthFailed from error
except (InvalidURL, HTTPError, PyEzvizError) as error:
_LOGGER.error("Unable to connect to Ezviz service: %s", str(error))
raise ConfigEntryNotReady from error
coordinator = EzvizDataUpdateCoordinator(
hass, api=ezviz_client, api_timeout=entry.options[CONF_TIMEOUT]
)
await coordinator.async_config_entry_first_refresh()
hass.data[DOMAIN][entry.entry_id] = {DATA_COORDINATOR: coordinator}
entry.async_on_unload(entry.add_update_listener(_async_update_listener))
# Check EZVIZ cloud account entity is present, reload cloud account entities for camera entity change to take effect.
# Cameras are accessed via local RTSP stream with unique credentials per camera.
# Separate camera entities allow for credential changes per camera.
if sensor_type == ATTR_TYPE_CAMERA and hass.data[DOMAIN]:
for item in hass.config_entries.async_entries(domain=DOMAIN):
if item.data.get(CONF_TYPE) == ATTR_TYPE_CLOUD:
_LOGGER.info("Reload Ezviz main account with camera entry")
await hass.config_entries.async_reload(item.entry_id)
return True
await hass.config_entries.async_forward_entry_setups(
entry, PLATFORMS_BY_TYPE[sensor_type]
) )
await coordinator.async_refresh()
if not coordinator.last_update_success:
raise ConfigEntryNotReady
undo_listener = entry.add_update_listener(_async_update_listener)
hass.data[DOMAIN][entry.entry_id] = {
DATA_COORDINATOR: coordinator,
DATA_UNDO_UPDATE_LISTENER: undo_listener,
}
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
return True return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry.""" """Unload a config entry."""
sensor_type = entry.data[CONF_TYPE]
if entry.data.get(CONF_TYPE) == ATTR_TYPE_CAMERA: unload_ok = await hass.config_entries.async_unload_platforms(
return True entry, PLATFORMS_BY_TYPE[sensor_type]
)
unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS) if sensor_type == ATTR_TYPE_CLOUD and unload_ok:
if unload_ok:
hass.data[DOMAIN][entry.entry_id][DATA_UNDO_UPDATE_LISTENER]()
hass.data[DOMAIN].pop(entry.entry_id) hass.data[DOMAIN].pop(entry.entry_id)
return unload_ok return unload_ok
async def _async_update_listener(hass: HomeAssistant, entry: ConfigEntry) -> None: async def _async_update_listener(hass: HomeAssistant, entry: ConfigEntry) -> None:
"""Handle options update.""" """Handle options update."""
await hass.config_entries.async_reload(entry.entry_id) await hass.config_entries.async_reload(entry.entry_id)
def _get_ezviz_client_instance(entry: ConfigEntry) -> EzvizClient:
"""Initialize a new instance of EzvizClientApi."""
ezviz_client = EzvizClient(
entry.data[CONF_USERNAME],
entry.data[CONF_PASSWORD],
entry.data[CONF_URL],
entry.options.get(CONF_TIMEOUT, DEFAULT_TIMEOUT),
)
ezviz_client.login()
return ezviz_client

14
homeassistant/components/ezviz/camera.py
View File

@ -34,7 +34,6 @@ from .const import (
DATA_COORDINATOR, DATA_COORDINATOR,
DEFAULT_CAMERA_USERNAME, DEFAULT_CAMERA_USERNAME,
DEFAULT_FFMPEG_ARGUMENTS, DEFAULT_FFMPEG_ARGUMENTS,
DEFAULT_RTSP_PORT,
DIR_DOWN, DIR_DOWN,
DIR_LEFT, DIR_LEFT,
DIR_RIGHT, DIR_RIGHT,
@ -70,24 +69,17 @@ async def async_setup_entry(
if item.unique_id == camera and item.source != SOURCE_IGNORE if item.unique_id == camera and item.source != SOURCE_IGNORE
] ]
# There seem to be a bug related to localRtspPort in EZVIZ API.
local_rtsp_port = (
value["local_rtsp_port"]
if value["local_rtsp_port"] != 0
else DEFAULT_RTSP_PORT
)
if camera_rtsp_entry: if camera_rtsp_entry:
ffmpeg_arguments = camera_rtsp_entry[0].options[CONF_FFMPEG_ARGUMENTS] ffmpeg_arguments = camera_rtsp_entry[0].options[CONF_FFMPEG_ARGUMENTS]
camera_username = camera_rtsp_entry[0].data[CONF_USERNAME] camera_username = camera_rtsp_entry[0].data[CONF_USERNAME]
camera_password = camera_rtsp_entry[0].data[CONF_PASSWORD] camera_password = camera_rtsp_entry[0].data[CONF_PASSWORD]
camera_rtsp_stream = f"rtsp://{camera_username}:{camera_password}@{value['local_ip']}:{local_rtsp_port}{ffmpeg_arguments}" camera_rtsp_stream = f"rtsp://{camera_username}:{camera_password}@{value['local_ip']}:{value['local_rtsp_port']}{ffmpeg_arguments}"
_LOGGER.debug( _LOGGER.debug(
"Configuring Camera %s with ip: %s rtsp port: %s ffmpeg arguments: %s", "Configuring Camera %s with ip: %s rtsp port: %s ffmpeg arguments: %s",
camera, camera,
value["local_ip"], value["local_ip"],
local_rtsp_port, value["local_rtsp_port"],
ffmpeg_arguments, ffmpeg_arguments,
) )
@ -123,7 +115,7 @@ async def async_setup_entry(
camera_username, camera_username,
camera_password, camera_password,
camera_rtsp_stream, camera_rtsp_stream,
local_rtsp_port, value["local_rtsp_port"],
ffmpeg_arguments, ffmpeg_arguments,
) )
) )

286
homeassistant/components/ezviz/config_flow.py
View File

@ -1,12 +1,14 @@
"""Config flow for ezviz.""" """Config flow for EZVIZ."""
from __future__ import annotations from __future__ import annotations
from collections.abc import Mapping
import logging import logging
from typing import Any
from pyezviz.client import EzvizClient from pyezviz.client import EzvizClient
from pyezviz.exceptions import ( from pyezviz.exceptions import (
AuthTestResultFailed, AuthTestResultFailed,
HTTPError, EzvizAuthVerificationCode,
InvalidHost, InvalidHost,
InvalidURL, InvalidURL,
PyEzvizError, PyEzvizError,
@ -25,12 +27,15 @@ from homeassistant.const import (
CONF_USERNAME, CONF_USERNAME,
) )
from homeassistant.core import callback from homeassistant.core import callback
from homeassistant.data_entry_flow import FlowResult
from .const import ( from .const import (
ATTR_SERIAL, ATTR_SERIAL,
ATTR_TYPE_CAMERA, ATTR_TYPE_CAMERA,
ATTR_TYPE_CLOUD, ATTR_TYPE_CLOUD,
CONF_FFMPEG_ARGUMENTS, CONF_FFMPEG_ARGUMENTS,
CONF_RFSESSION_ID,
CONF_SESSION_ID,
DEFAULT_CAMERA_USERNAME, DEFAULT_CAMERA_USERNAME,
DEFAULT_FFMPEG_ARGUMENTS, DEFAULT_FFMPEG_ARGUMENTS,
DEFAULT_TIMEOUT, DEFAULT_TIMEOUT,
@ -40,23 +45,37 @@ from .const import (
) )
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
DEFAULT_OPTIONS = {
CONF_FFMPEG_ARGUMENTS: DEFAULT_FFMPEG_ARGUMENTS,
CONF_TIMEOUT: DEFAULT_TIMEOUT,
}
def _get_ezviz_client_instance(data): def _validate_and_create_auth(data: dict) -> dict[str, Any]:
"""Initialize a new instance of EzvizClientApi.""" """Try to login to EZVIZ cloud account and return token."""
# Verify cloud credentials by attempting a login request with username and password.
# Return login token.
ezviz_client = EzvizClient( ezviz_client = EzvizClient(
data[CONF_USERNAME], data[CONF_USERNAME],
data[CONF_PASSWORD], data[CONF_PASSWORD],
data.get(CONF_URL, EU_URL), data[CONF_URL],
data.get(CONF_TIMEOUT, DEFAULT_TIMEOUT), data.get(CONF_TIMEOUT, DEFAULT_TIMEOUT),
) )
ezviz_client.login() ezviz_token = ezviz_client.login()
return ezviz_client
auth_data = {
CONF_SESSION_ID: ezviz_token[CONF_SESSION_ID],
CONF_RFSESSION_ID: ezviz_token[CONF_RFSESSION_ID],
CONF_URL: ezviz_token["api_url"],
CONF_TYPE: ATTR_TYPE_CLOUD,
}
return auth_data
def _test_camera_rtsp_creds(data): def _test_camera_rtsp_creds(data: dict) -> None:
"""Try DESCRIBE on RTSP camera with credentials.""" """Try DESCRIBE on RTSP camera with credentials."""
test_rtsp = TestRTSPAuth( test_rtsp = TestRTSPAuth(
@ -71,89 +90,43 @@ class EzvizConfigFlow(ConfigFlow, domain=DOMAIN):
VERSION = 1 VERSION = 1
async def _validate_and_create_auth(self, data): async def _validate_and_create_camera_rtsp(self, data: dict) -> FlowResult:
"""Try to login to ezviz cloud account and create entry if successful."""
await self.async_set_unique_id(data[CONF_USERNAME])
self._abort_if_unique_id_configured()
# Verify cloud credentials by attempting a login request.
try:
await self.hass.async_add_executor_job(_get_ezviz_client_instance, data)
except InvalidURL as err:
raise InvalidURL from err
except HTTPError as err:
raise InvalidHost from err
except PyEzvizError as err:
raise PyEzvizError from err
auth_data = {
CONF_USERNAME: data[CONF_USERNAME],
CONF_PASSWORD: data[CONF_PASSWORD],
CONF_URL: data.get(CONF_URL, EU_URL),
CONF_TYPE: ATTR_TYPE_CLOUD,
}
return self.async_create_entry(title=data[CONF_USERNAME], data=auth_data)
async def _validate_and_create_camera_rtsp(self, data):
"""Try DESCRIBE on RTSP camera with credentials.""" """Try DESCRIBE on RTSP camera with credentials."""
# Get EZVIZ cloud credentials from config entry # Get EZVIZ cloud credentials from config entry
ezviz_client_creds = { ezviz_token = {
CONF_USERNAME: None, CONF_SESSION_ID: None,
CONF_PASSWORD: None, CONF_RFSESSION_ID: None,
CONF_URL: None, "api_url": None,
} }
ezviz_timeout = DEFAULT_TIMEOUT
for item in self._async_current_entries(): for item in self._async_current_entries():
if item.data.get(CONF_TYPE) == ATTR_TYPE_CLOUD: if item.data.get(CONF_TYPE) == ATTR_TYPE_CLOUD:
ezviz_client_creds = { ezviz_token = {
CONF_USERNAME: item.data.get(CONF_USERNAME), CONF_SESSION_ID: item.data.get(CONF_SESSION_ID),
CONF_PASSWORD: item.data.get(CONF_PASSWORD), CONF_RFSESSION_ID: item.data.get(CONF_RFSESSION_ID),
CONF_URL: item.data.get(CONF_URL), "api_url": item.data.get(CONF_URL),
} }
ezviz_timeout = item.data.get(CONF_TIMEOUT, DEFAULT_TIMEOUT)
# Abort flow if user removed cloud account before adding camera. # Abort flow if user removed cloud account before adding camera.
if ezviz_client_creds[CONF_USERNAME] is None: if ezviz_token.get(CONF_SESSION_ID) is None:
return self.async_abort(reason="ezviz_cloud_account_missing") return self.async_abort(reason="ezviz_cloud_account_missing")
ezviz_client = EzvizClient(token=ezviz_token, timeout=ezviz_timeout)
# We need to wake hibernating cameras. # We need to wake hibernating cameras.
# First create EZVIZ API instance. # First create EZVIZ API instance.
try: await self.hass.async_add_executor_job(ezviz_client.login)
ezviz_client = await self.hass.async_add_executor_job(
_get_ezviz_client_instance, ezviz_client_creds
)
except InvalidURL as err: # Secondly try to wake hybernating camera.
raise InvalidURL from err await self.hass.async_add_executor_job(
ezviz_client.get_detection_sensibility, data[ATTR_SERIAL]
except HTTPError as err: )
raise InvalidHost from err
except PyEzvizError as err:
raise PyEzvizError from err
# Secondly try to wake hibernating camera.
try:
await self.hass.async_add_executor_job(
ezviz_client.get_detection_sensibility, data[ATTR_SERIAL]
)
except HTTPError as err:
raise InvalidHost from err
# Thirdly attempts an authenticated RTSP DESCRIBE request. # Thirdly attempts an authenticated RTSP DESCRIBE request.
try: await self.hass.async_add_executor_job(_test_camera_rtsp_creds, data)
await self.hass.async_add_executor_job(_test_camera_rtsp_creds, data)
except InvalidHost as err:
raise InvalidHost from err
except AuthTestResultFailed as err:
raise AuthTestResultFailed from err
return self.async_create_entry( return self.async_create_entry(
title=data[ATTR_SERIAL], title=data[ATTR_SERIAL],
@ -162,6 +135,7 @@ class EzvizConfigFlow(ConfigFlow, domain=DOMAIN):
CONF_PASSWORD: data[CONF_PASSWORD], CONF_PASSWORD: data[CONF_PASSWORD],
CONF_TYPE: ATTR_TYPE_CAMERA, CONF_TYPE: ATTR_TYPE_CAMERA,
}, },
options=DEFAULT_OPTIONS,
) )
@staticmethod @staticmethod
@ -170,18 +144,24 @@ class EzvizConfigFlow(ConfigFlow, domain=DOMAIN):
"""Get the options flow for this handler.""" """Get the options flow for this handler."""
return EzvizOptionsFlowHandler(config_entry) return EzvizOptionsFlowHandler(config_entry)
async def async_step_user(self, user_input=None): async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle a flow initiated by the user.""" """Handle a flow initiated by the user."""
# Check if ezviz cloud account is present in entry config, # Check if EZVIZ cloud account is present in entry config,
# abort if already configured. # abort if already configured.
for item in self._async_current_entries(): for item in self._async_current_entries():
if item.data.get(CONF_TYPE) == ATTR_TYPE_CLOUD: if item.data.get(CONF_TYPE) == ATTR_TYPE_CLOUD:
return self.async_abort(reason="already_configured_account") return self.async_abort(reason="already_configured_account")
errors = {} errors = {}
auth_data = {}
if user_input is not None: if user_input is not None:
await self.async_set_unique_id(user_input[CONF_USERNAME])
self._abort_if_unique_id_configured()
if user_input[CONF_URL] == CONF_CUSTOMIZE: if user_input[CONF_URL] == CONF_CUSTOMIZE:
self.context["data"] = { self.context["data"] = {
CONF_USERNAME: user_input[CONF_USERNAME], CONF_USERNAME: user_input[CONF_USERNAME],
@ -189,11 +169,10 @@ class EzvizConfigFlow(ConfigFlow, domain=DOMAIN):
} }
return await self.async_step_user_custom_url() return await self.async_step_user_custom_url()
if CONF_TIMEOUT not in user_input:
user_input[CONF_TIMEOUT] = DEFAULT_TIMEOUT
try: try:
return await self._validate_and_create_auth(user_input) auth_data = await self.hass.async_add_executor_job(
_validate_and_create_auth, user_input
)
except InvalidURL: except InvalidURL:
errors["base"] = "invalid_host" errors["base"] = "invalid_host"
@ -201,6 +180,9 @@ class EzvizConfigFlow(ConfigFlow, domain=DOMAIN):
except InvalidHost: except InvalidHost:
errors["base"] = "cannot_connect" errors["base"] = "cannot_connect"
except EzvizAuthVerificationCode:
errors["base"] = "mfa_required"
except PyEzvizError: except PyEzvizError:
errors["base"] = "invalid_auth" errors["base"] = "invalid_auth"
@ -208,6 +190,13 @@ class EzvizConfigFlow(ConfigFlow, domain=DOMAIN):
_LOGGER.exception("Unexpected exception") _LOGGER.exception("Unexpected exception")
return self.async_abort(reason="unknown") return self.async_abort(reason="unknown")
else:
return self.async_create_entry(
title=user_input[CONF_USERNAME],
data=auth_data,
options=DEFAULT_OPTIONS,
)
data_schema = vol.Schema( data_schema = vol.Schema(
{ {
vol.Required(CONF_USERNAME): str, vol.Required(CONF_USERNAME): str,
@ -222,20 +211,21 @@ class EzvizConfigFlow(ConfigFlow, domain=DOMAIN):
step_id="user", data_schema=data_schema, errors=errors step_id="user", data_schema=data_schema, errors=errors
) )
async def async_step_user_custom_url(self, user_input=None): async def async_step_user_custom_url(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle a flow initiated by the user for custom region url.""" """Handle a flow initiated by the user for custom region url."""
errors = {} errors = {}
auth_data = {}
if user_input is not None: if user_input is not None:
user_input[CONF_USERNAME] = self.context["data"][CONF_USERNAME] user_input[CONF_USERNAME] = self.context["data"][CONF_USERNAME]
user_input[CONF_PASSWORD] = self.context["data"][CONF_PASSWORD] user_input[CONF_PASSWORD] = self.context["data"][CONF_PASSWORD]
if CONF_TIMEOUT not in user_input:
user_input[CONF_TIMEOUT] = DEFAULT_TIMEOUT
try: try:
return await self._validate_and_create_auth(user_input) auth_data = await self.hass.async_add_executor_job(
_validate_and_create_auth, user_input
)
except InvalidURL: except InvalidURL:
errors["base"] = "invalid_host" errors["base"] = "invalid_host"
@ -243,6 +233,9 @@ class EzvizConfigFlow(ConfigFlow, domain=DOMAIN):
except InvalidHost: except InvalidHost:
errors["base"] = "cannot_connect" errors["base"] = "cannot_connect"
except EzvizAuthVerificationCode:
errors["base"] = "mfa_required"
except PyEzvizError: except PyEzvizError:
errors["base"] = "invalid_auth" errors["base"] = "invalid_auth"
@ -250,6 +243,13 @@ class EzvizConfigFlow(ConfigFlow, domain=DOMAIN):
_LOGGER.exception("Unexpected exception") _LOGGER.exception("Unexpected exception")
return self.async_abort(reason="unknown") return self.async_abort(reason="unknown")
else:
return self.async_create_entry(
title=user_input[CONF_USERNAME],
data=auth_data,
options=DEFAULT_OPTIONS,
)
data_schema_custom_url = vol.Schema( data_schema_custom_url = vol.Schema(
{ {
vol.Required(CONF_URL, default=EU_URL): str, vol.Required(CONF_URL, default=EU_URL): str,
@ -260,18 +260,22 @@ class EzvizConfigFlow(ConfigFlow, domain=DOMAIN):
step_id="user_custom_url", data_schema=data_schema_custom_url, errors=errors step_id="user_custom_url", data_schema=data_schema_custom_url, errors=errors
) )
async def async_step_integration_discovery(self, discovery_info): async def async_step_integration_discovery(
self, discovery_info: dict[str, Any]
) -> FlowResult:
"""Handle a flow for discovered camera without rtsp config entry.""" """Handle a flow for discovered camera without rtsp config entry."""
await self.async_set_unique_id(discovery_info[ATTR_SERIAL]) await self.async_set_unique_id(discovery_info[ATTR_SERIAL])
self._abort_if_unique_id_configured() self._abort_if_unique_id_configured()
self.context["title_placeholders"] = {"serial": self.unique_id} self.context["title_placeholders"] = {ATTR_SERIAL: self.unique_id}
self.context["data"] = {CONF_IP_ADDRESS: discovery_info[CONF_IP_ADDRESS]} self.context["data"] = {CONF_IP_ADDRESS: discovery_info[CONF_IP_ADDRESS]}
return await self.async_step_confirm() return await self.async_step_confirm()
async def async_step_confirm(self, user_input=None): async def async_step_confirm(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Confirm and create entry from discovery step.""" """Confirm and create entry from discovery step."""
errors = {} errors = {}
@ -284,6 +288,9 @@ class EzvizConfigFlow(ConfigFlow, domain=DOMAIN):
except (InvalidHost, InvalidURL): except (InvalidHost, InvalidURL):
errors["base"] = "invalid_host" errors["base"] = "invalid_host"
except EzvizAuthVerificationCode:
errors["base"] = "mfa_required"
except (PyEzvizError, AuthTestResultFailed): except (PyEzvizError, AuthTestResultFailed):
errors["base"] = "invalid_auth" errors["base"] = "invalid_auth"
@ -303,11 +310,76 @@ class EzvizConfigFlow(ConfigFlow, domain=DOMAIN):
data_schema=discovered_camera_schema, data_schema=discovered_camera_schema,
errors=errors, errors=errors,
description_placeholders={ description_placeholders={
"serial": self.unique_id, ATTR_SERIAL: self.unique_id,
CONF_IP_ADDRESS: self.context["data"][CONF_IP_ADDRESS], CONF_IP_ADDRESS: self.context["data"][CONF_IP_ADDRESS],
}, },
) )
async def async_step_reauth(self, user_input: Mapping[str, Any]) -> FlowResult:
"""Handle a flow for reauthentication with password."""
return await self.async_step_reauth_confirm()
async def async_step_reauth_confirm(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle a Confirm flow for reauthentication with password."""
auth_data = {}
errors = {}
entry = None
for item in self._async_current_entries():
if item.data.get(CONF_TYPE) == ATTR_TYPE_CLOUD:
self.context["title_placeholders"] = {ATTR_SERIAL: item.title}
entry = await self.async_set_unique_id(item.title)
if not entry:
return self.async_abort(reason="ezviz_cloud_account_missing")
if user_input is not None:
user_input[CONF_URL] = entry.data[CONF_URL]
try:
auth_data = await self.hass.async_add_executor_job(
_validate_and_create_auth, user_input
)
except (InvalidHost, InvalidURL):
errors["base"] = "invalid_host"
except EzvizAuthVerificationCode:
errors["base"] = "mfa_required"
except (PyEzvizError, AuthTestResultFailed):
errors["base"] = "invalid_auth"
except Exception: # pylint: disable=broad-except
_LOGGER.exception("Unexpected exception")
return self.async_abort(reason="unknown")
else:
self.hass.config_entries.async_update_entry(
entry,
data=auth_data,
)
await self.hass.config_entries.async_reload(entry.entry_id)
return self.async_abort(reason="reauth_successful")
data_schema = vol.Schema(
{
vol.Required(CONF_USERNAME, default=entry.title): vol.In([entry.title]),
vol.Required(CONF_PASSWORD): str,
}
)
return self.async_show_form(
step_id="reauth_confirm",
data_schema=data_schema,
errors=errors,
)
class EzvizOptionsFlowHandler(OptionsFlow): class EzvizOptionsFlowHandler(OptionsFlow):
"""Handle EZVIZ client options.""" """Handle EZVIZ client options."""
@ -316,22 +388,28 @@ class EzvizOptionsFlowHandler(OptionsFlow):
"""Initialize options flow.""" """Initialize options flow."""
self.config_entry = config_entry self.config_entry = config_entry
async def async_step_init(self, user_input=None): async def async_step_init(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Manage EZVIZ options.""" """Manage EZVIZ options."""
if user_input is not None: if user_input is not None:
return self.async_create_entry(title="", data=user_input) return self.async_create_entry(title="", data=user_input)
options = { options = vol.Schema(
vol.Optional( {
CONF_TIMEOUT, vol.Optional(
default=self.config_entry.options.get(CONF_TIMEOUT, DEFAULT_TIMEOUT), CONF_TIMEOUT,
): int, default=self.config_entry.options.get(
vol.Optional( CONF_TIMEOUT, DEFAULT_TIMEOUT
CONF_FFMPEG_ARGUMENTS, ),
default=self.config_entry.options.get( ): int,
CONF_FFMPEG_ARGUMENTS, DEFAULT_FFMPEG_ARGUMENTS vol.Optional(
), CONF_FFMPEG_ARGUMENTS,
): str, default=self.config_entry.options.get(
} CONF_FFMPEG_ARGUMENTS, DEFAULT_FFMPEG_ARGUMENTS
),
): str,
}
)
return self.async_show_form(step_id="init", data_schema=vol.Schema(options)) return self.async_show_form(step_id="init", data_schema=options)

5
homeassistant/components/ezviz/const.py
View File

@ -10,6 +10,9 @@ ATTR_HOME = "HOME_MODE"
ATTR_AWAY = "AWAY_MODE" ATTR_AWAY = "AWAY_MODE"
ATTR_TYPE_CLOUD = "EZVIZ_CLOUD_ACCOUNT" ATTR_TYPE_CLOUD = "EZVIZ_CLOUD_ACCOUNT"
ATTR_TYPE_CAMERA = "CAMERA_ACCOUNT" ATTR_TYPE_CAMERA = "CAMERA_ACCOUNT"
CONF_SESSION_ID = "session_id"
CONF_RFSESSION_ID = "rf_session_id"
CONF_EZVIZ_ACCOUNT = "ezviz_account"
# Services data # Services data
DIR_UP = "up" DIR_UP = "up"
@ -33,10 +36,8 @@ SERVICE_DETECTION_SENSITIVITY = "set_alarm_detection_sensibility"
EU_URL = "apiieu.ezvizlife.com" EU_URL = "apiieu.ezvizlife.com"
RUSSIA_URL = "apirus.ezvizru.com" RUSSIA_URL = "apirus.ezvizru.com"
DEFAULT_CAMERA_USERNAME = "admin" DEFAULT_CAMERA_USERNAME = "admin"
DEFAULT_RTSP_PORT = 554
DEFAULT_TIMEOUT = 25 DEFAULT_TIMEOUT = 25
DEFAULT_FFMPEG_ARGUMENTS = "" DEFAULT_FFMPEG_ARGUMENTS = ""
# Data # Data
DATA_COORDINATOR = "coordinator" DATA_COORDINATOR = "coordinator"
DATA_UNDO_UPDATE_LISTENER = "undo_update_listener"

20
homeassistant/components/ezviz/coordinator.py
View File

@ -4,9 +4,16 @@ import logging
from async_timeout import timeout from async_timeout import timeout
from pyezviz.client import EzvizClient from pyezviz.client import EzvizClient
from pyezviz.exceptions import HTTPError, InvalidURL, PyEzvizError from pyezviz.exceptions import (
EzvizAuthTokenExpired,
EzvizAuthVerificationCode,
HTTPError,
InvalidURL,
PyEzvizError,
)
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import DOMAIN from .const import DOMAIN
@ -27,15 +34,16 @@ class EzvizDataUpdateCoordinator(DataUpdateCoordinator):
super().__init__(hass, _LOGGER, name=DOMAIN, update_interval=update_interval) super().__init__(hass, _LOGGER, name=DOMAIN, update_interval=update_interval)
def _update_data(self) -> dict:
"""Fetch data from EZVIZ via camera load function."""
return self.ezviz_client.load_cameras()
async def _async_update_data(self) -> dict: async def _async_update_data(self) -> dict:
"""Fetch data from EZVIZ.""" """Fetch data from EZVIZ."""
try: try:
async with timeout(self._api_timeout): async with timeout(self._api_timeout):
return await self.hass.async_add_executor_job(self._update_data) return await self.hass.async_add_executor_job(
self.ezviz_client.load_cameras
)
except (EzvizAuthTokenExpired, EzvizAuthVerificationCode) as error:
raise ConfigEntryAuthFailed from error
except (InvalidURL, HTTPError, PyEzvizError) as error: except (InvalidURL, HTTPError, PyEzvizError) as error:
raise UpdateFailed(f"Invalid response from API: {error}") from error raise UpdateFailed(f"Invalid response from API: {error}") from error

14
homeassistant/components/ezviz/strings.json
View File

@ -26,17 +26,27 @@
"username": "[%key:common::config_flow::data::username%]", "username": "[%key:common::config_flow::data::username%]",
"password": "[%key:common::config_flow::data::password%]" "password": "[%key:common::config_flow::data::password%]"
} }
},
"reauth_confirm": {
"title": "[%key:common::config_flow::title::reauth%]",
"description": "Enter credentials to reauthenticate to ezviz cloud account",
"data": {
"username": "[%key:common::config_flow::data::username%]",
"password": "[%key:common::config_flow::data::password%]"
}
} }
}, },
"error": { "error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]", "cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]", "invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
"invalid_host": "[%key:common::config_flow::error::invalid_host%]" "invalid_host": "[%key:common::config_flow::error::invalid_host%]",
"mfa_required": "2FA enabled on account, please disable and retry"
}, },
"abort": { "abort": {
"already_configured_account": "[%key:common::config_flow::abort::already_configured_account%]", "already_configured_account": "[%key:common::config_flow::abort::already_configured_account%]",
"unknown": "[%key:common::config_flow::error::unknown%]", "unknown": "[%key:common::config_flow::error::unknown%]",
"ezviz_cloud_account_missing": "EZVIZ cloud account missing. Please reconfigure EZVIZ cloud account" "ezviz_cloud_account_missing": "Ezviz cloud account missing. Please reconfigure Ezviz cloud account",
"reauth_successful": "[%key:common::config_flow::abort::reauth_successful%]"
} }
}, },
"options": { "options": {

26
tests/components/ezviz/__init__.py
View File

@ -3,8 +3,11 @@ from unittest.mock import patch
from homeassistant.components.ezviz.const import ( from homeassistant.components.ezviz.const import (
ATTR_SERIAL, ATTR_SERIAL,
ATTR_TYPE_CAMERA,
ATTR_TYPE_CLOUD, ATTR_TYPE_CLOUD,
CONF_FFMPEG_ARGUMENTS, CONF_FFMPEG_ARGUMENTS,
CONF_RFSESSION_ID,
CONF_SESSION_ID,
DEFAULT_FFMPEG_ARGUMENTS, DEFAULT_FFMPEG_ARGUMENTS,
DEFAULT_TIMEOUT, DEFAULT_TIMEOUT,
DOMAIN, DOMAIN,
@ -22,8 +25,8 @@ from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry from tests.common import MockConfigEntry
ENTRY_CONFIG = { ENTRY_CONFIG = {
CONF_USERNAME: "test-username", CONF_SESSION_ID: "test-username",
CONF_PASSWORD: "test-password", CONF_RFSESSION_ID: "test-password",
CONF_URL: "apiieu.ezvizlife.com", CONF_URL: "apiieu.ezvizlife.com",
CONF_TYPE: ATTR_TYPE_CLOUD, CONF_TYPE: ATTR_TYPE_CLOUD,
} }
@ -46,6 +49,18 @@ USER_INPUT = {
CONF_TYPE: ATTR_TYPE_CLOUD, CONF_TYPE: ATTR_TYPE_CLOUD,
} }
USER_INPUT_CAMERA_VALIDATE = {
ATTR_SERIAL: "C666666",
CONF_PASSWORD: "test-password",
CONF_USERNAME: "test-username",
}
USER_INPUT_CAMERA = {
CONF_PASSWORD: "test-password",
CONF_USERNAME: "test-username",
CONF_TYPE: ATTR_TYPE_CAMERA,
}
DISCOVERY_INFO = { DISCOVERY_INFO = {
ATTR_SERIAL: "C666666", ATTR_SERIAL: "C666666",
CONF_USERNAME: None, CONF_USERNAME: None,
@ -59,6 +74,13 @@ TEST = {
CONF_IP_ADDRESS: "127.0.0.1", CONF_IP_ADDRESS: "127.0.0.1",
} }
API_LOGIN_RETURN_VALIDATE = {
CONF_SESSION_ID: "fake_token",
CONF_RFSESSION_ID: "fake_rf_token",
CONF_URL: "apiieu.ezvizlife.com",
CONF_TYPE: ATTR_TYPE_CLOUD,
}
def _patch_async_setup_entry(return_value=True): def _patch_async_setup_entry(return_value=True):
return patch( return patch(

8
tests/components/ezviz/conftest.py
View File

@ -5,6 +5,12 @@ from pyezviz import EzvizClient
from pyezviz.test_cam_rtsp import TestRTSPAuth from pyezviz.test_cam_rtsp import TestRTSPAuth
import pytest import pytest
ezviz_login_token_return = {
"session_id": "fake_token",
"rf_session_id": "fake_rf_token",
"api_url": "apiieu.ezvizlife.com",
}
@pytest.fixture(autouse=True) @pytest.fixture(autouse=True)
def mock_ffmpeg(hass): def mock_ffmpeg(hass):
@ -42,7 +48,7 @@ def ezviz_config_flow(hass):
"1", "1",
) )
instance.login = MagicMock(return_value=True) instance.login = MagicMock(return_value=ezviz_login_token_return)
instance.get_detection_sensibility = MagicMock(return_value=True) instance.get_detection_sensibility = MagicMock(return_value=True)
yield mock_ezviz yield mock_ezviz

242
tests/components/ezviz/test_config_flow.py
View File

@ -3,6 +3,7 @@ from unittest.mock import patch
from pyezviz.exceptions import ( from pyezviz.exceptions import (
AuthTestResultFailed, AuthTestResultFailed,
EzvizAuthVerificationCode,
HTTPError, HTTPError,
InvalidHost, InvalidHost,
InvalidURL, InvalidURL,
@ -12,13 +13,16 @@ from pyezviz.exceptions import (
from homeassistant.components.ezviz.const import ( from homeassistant.components.ezviz.const import (
ATTR_SERIAL, ATTR_SERIAL,
ATTR_TYPE_CAMERA, ATTR_TYPE_CAMERA,
ATTR_TYPE_CLOUD,
CONF_FFMPEG_ARGUMENTS, CONF_FFMPEG_ARGUMENTS,
DEFAULT_FFMPEG_ARGUMENTS, DEFAULT_FFMPEG_ARGUMENTS,
DEFAULT_TIMEOUT, DEFAULT_TIMEOUT,
DOMAIN, DOMAIN,
) )
from homeassistant.config_entries import SOURCE_INTEGRATION_DISCOVERY, SOURCE_USER from homeassistant.config_entries import (
SOURCE_INTEGRATION_DISCOVERY,
SOURCE_REAUTH,
SOURCE_USER,
)
from homeassistant.const import ( from homeassistant.const import (
CONF_CUSTOMIZE, CONF_CUSTOMIZE,
CONF_IP_ADDRESS, CONF_IP_ADDRESS,
@ -32,8 +36,8 @@ from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType from homeassistant.data_entry_flow import FlowResultType
from . import ( from . import (
API_LOGIN_RETURN_VALIDATE,
DISCOVERY_INFO, DISCOVERY_INFO,
USER_INPUT,
USER_INPUT_VALIDATE, USER_INPUT_VALIDATE,
_patch_async_setup_entry, _patch_async_setup_entry,
init_integration, init_integration,
@ -59,7 +63,7 @@ async def test_user_form(hass: HomeAssistant, ezviz_config_flow) -> None:
assert result["type"] == FlowResultType.CREATE_ENTRY assert result["type"] == FlowResultType.CREATE_ENTRY
assert result["title"] == "test-username" assert result["title"] == "test-username"
assert result["data"] == {**USER_INPUT} assert result["data"] == {**API_LOGIN_RETURN_VALIDATE}
assert len(mock_setup_entry.mock_calls) == 1 assert len(mock_setup_entry.mock_calls) == 1
@ -78,7 +82,11 @@ async def test_user_custom_url(hass: HomeAssistant, ezviz_config_flow) -> None:
result = await hass.config_entries.flow.async_configure( result = await hass.config_entries.flow.async_configure(
result["flow_id"], result["flow_id"],
{CONF_USERNAME: "test-user", CONF_PASSWORD: "test-pass", CONF_URL: "customize"}, {
CONF_USERNAME: "test-username",
CONF_PASSWORD: "test-password",
CONF_URL: CONF_CUSTOMIZE,
},
) )
assert result["type"] == FlowResultType.FORM assert result["type"] == FlowResultType.FORM
@ -90,21 +98,58 @@ async def test_user_custom_url(hass: HomeAssistant, ezviz_config_flow) -> None:
result["flow_id"], result["flow_id"],
{CONF_URL: "test-user"}, {CONF_URL: "test-user"},
) )
await hass.async_block_till_done()
assert result["type"] == FlowResultType.CREATE_ENTRY assert result["type"] == FlowResultType.CREATE_ENTRY
assert result["data"] == { assert result["data"] == API_LOGIN_RETURN_VALIDATE
CONF_PASSWORD: "test-pass",
CONF_TYPE: ATTR_TYPE_CLOUD,
CONF_URL: "test-user",
CONF_USERNAME: "test-user",
}
assert len(mock_setup_entry.mock_calls) == 1 assert len(mock_setup_entry.mock_calls) == 1
async def test_step_discovery_abort_if_cloud_account_missing( async def test_async_step_reauth(hass, ezviz_config_flow):
hass: HomeAssistant, """Test the reauth step."""
) -> None:
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "user"
assert result["errors"] == {}
with _patch_async_setup_entry() as mock_setup_entry:
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
USER_INPUT_VALIDATE,
)
await hass.async_block_till_done()
assert result["type"] == FlowResultType.CREATE_ENTRY
assert result["title"] == "test-username"
assert result["data"] == {**API_LOGIN_RETURN_VALIDATE}
assert len(mock_setup_entry.mock_calls) == 1
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_REAUTH}, data=USER_INPUT_VALIDATE
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "reauth_confirm"
assert result["errors"] == {}
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_USERNAME: "test-username",
CONF_PASSWORD: "test-password",
},
)
await hass.async_block_till_done()
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "reauth_successful"
async def test_step_discovery_abort_if_cloud_account_missing(hass):
"""Test discovery and confirm step, abort if cloud account was removed.""" """Test discovery and confirm step, abort if cloud account was removed."""
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
@ -127,11 +172,21 @@ async def test_step_discovery_abort_if_cloud_account_missing(
assert result["reason"] == "ezviz_cloud_account_missing" assert result["reason"] == "ezviz_cloud_account_missing"
async def test_step_reauth_abort_if_cloud_account_missing(hass):
"""Test reauth and confirm step, abort if cloud account was removed."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_REAUTH}, data=USER_INPUT_VALIDATE
)
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "ezviz_cloud_account_missing"
async def test_async_step_integration_discovery( async def test_async_step_integration_discovery(
hass: HomeAssistant, ezviz_config_flow, ezviz_test_rtsp_config_flow hass, ezviz_config_flow, ezviz_test_rtsp_config_flow
) -> None: ):
"""Test discovery and confirm step.""" """Test discovery and confirm step."""
with patch("homeassistant.components.ezviz.PLATFORMS", []): with patch("homeassistant.components.ezviz.PLATFORMS_BY_TYPE", []):
await init_integration(hass) await init_integration(hass)
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
@ -189,11 +244,14 @@ async def test_options_flow(hass: HomeAssistant) -> None:
async def test_user_form_exception(hass: HomeAssistant, ezviz_config_flow) -> None: async def test_user_form_exception(hass: HomeAssistant, ezviz_config_flow) -> None:
"""Test we handle exception on user form.""" """Test we handle exception on user form."""
ezviz_config_flow.side_effect = PyEzvizError
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER} DOMAIN, context={"source": SOURCE_USER}
) )
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "user"
assert result["errors"] == {}
ezviz_config_flow.side_effect = PyEzvizError
result = await hass.config_entries.flow.async_configure( result = await hass.config_entries.flow.async_configure(
result["flow_id"], result["flow_id"],
@ -215,6 +273,17 @@ async def test_user_form_exception(hass: HomeAssistant, ezviz_config_flow) -> No
assert result["step_id"] == "user" assert result["step_id"] == "user"
assert result["errors"] == {"base": "invalid_host"} assert result["errors"] == {"base": "invalid_host"}
ezviz_config_flow.side_effect = EzvizAuthVerificationCode
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
USER_INPUT_VALIDATE,
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "user"
assert result["errors"] == {"base": "mfa_required"}
ezviz_config_flow.side_effect = HTTPError ezviz_config_flow.side_effect = HTTPError
result = await hass.config_entries.flow.async_configure( result = await hass.config_entries.flow.async_configure(
@ -224,7 +293,7 @@ async def test_user_form_exception(hass: HomeAssistant, ezviz_config_flow) -> No
assert result["type"] == FlowResultType.FORM assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "user" assert result["step_id"] == "user"
assert result["errors"] == {"base": "cannot_connect"} assert result["errors"] == {"base": "invalid_auth"}
ezviz_config_flow.side_effect = Exception ezviz_config_flow.side_effect = Exception
@ -242,7 +311,7 @@ async def test_discover_exception_step1(
ezviz_config_flow, ezviz_config_flow,
) -> None: ) -> None:
"""Test we handle unexpected exception on discovery.""" """Test we handle unexpected exception on discovery."""
with patch("homeassistant.components.ezviz.PLATFORMS", []): with patch("homeassistant.components.ezviz.PLATFORMS_BY_TYPE", []):
await init_integration(hass) await init_integration(hass)
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
@ -295,7 +364,21 @@ async def test_discover_exception_step1(
assert result["type"] == FlowResultType.FORM assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "confirm" assert result["step_id"] == "confirm"
assert result["errors"] == {"base": "invalid_host"} assert result["errors"] == {"base": "invalid_auth"}
ezviz_config_flow.side_effect = EzvizAuthVerificationCode
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_USERNAME: "test-user",
CONF_PASSWORD: "test-pass",
},
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "confirm"
assert result["errors"] == {"base": "mfa_required"}
ezviz_config_flow.side_effect = Exception ezviz_config_flow.side_effect = Exception
@ -317,7 +400,7 @@ async def test_discover_exception_step3(
ezviz_test_rtsp_config_flow, ezviz_test_rtsp_config_flow,
) -> None: ) -> None:
"""Test we handle unexpected exception on discovery.""" """Test we handle unexpected exception on discovery."""
with patch("homeassistant.components.ezviz.PLATFORMS", []): with patch("homeassistant.components.ezviz.PLATFORMS_BY_TYPE", []):
await init_integration(hass) await init_integration(hass)
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
@ -423,7 +506,18 @@ async def test_user_custom_url_exception(
assert result["type"] == FlowResultType.FORM assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "user_custom_url" assert result["step_id"] == "user_custom_url"
assert result["errors"] == {"base": "cannot_connect"} assert result["errors"] == {"base": "invalid_auth"}
ezviz_config_flow.side_effect = EzvizAuthVerificationCode
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_URL: "test-user"},
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "user_custom_url"
assert result["errors"] == {"base": "mfa_required"}
ezviz_config_flow.side_effect = Exception ezviz_config_flow.side_effect = Exception
@ -434,3 +528,103 @@ async def test_user_custom_url_exception(
assert result["type"] == FlowResultType.ABORT assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "unknown" assert result["reason"] == "unknown"
async def test_async_step_reauth_exception(hass, ezviz_config_flow):
"""Test the reauth step exceptions."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "user"
assert result["errors"] == {}
with _patch_async_setup_entry() as mock_setup_entry:
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
USER_INPUT_VALIDATE,
)
await hass.async_block_till_done()
assert result["type"] == FlowResultType.CREATE_ENTRY
assert result["title"] == "test-username"
assert result["data"] == {**API_LOGIN_RETURN_VALIDATE}
assert len(mock_setup_entry.mock_calls) == 1
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_REAUTH}, data=USER_INPUT_VALIDATE
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "reauth_confirm"
assert result["errors"] == {}
ezviz_config_flow.side_effect = InvalidURL()
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_USERNAME: "test-username",
CONF_PASSWORD: "test-password",
},
)
await hass.async_block_till_done()
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "reauth_confirm"
assert result["errors"] == {"base": "invalid_host"}
ezviz_config_flow.side_effect = InvalidHost()
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_USERNAME: "test-username",
CONF_PASSWORD: "test-password",
},
)
await hass.async_block_till_done()
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "reauth_confirm"
assert result["errors"] == {"base": "invalid_host"}
ezviz_config_flow.side_effect = EzvizAuthVerificationCode()
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_USERNAME: "test-username",
CONF_PASSWORD: "test-password",
},
)
await hass.async_block_till_done()
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "reauth_confirm"
assert result["errors"] == {"base": "mfa_required"}
ezviz_config_flow.side_effect = PyEzvizError()
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_USERNAME: "test-username",
CONF_PASSWORD: "test-password",
},
)
await hass.async_block_till_done()
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "reauth_confirm"
assert result["errors"] == {"base": "invalid_auth"}
ezviz_config_flow.side_effect = Exception()
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_USERNAME: "test-username",
CONF_PASSWORD: "test-password",
},
)
await hass.async_block_till_done()
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "unknown"