Improve error reporting in onvif in config flow (#91876)

This commit is contained in:
J. Nick Koston 2023-04-24 08:20:37 -05:00 committed by GitHub
parent 1c3e1d2e13
commit e25885b943
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
9 changed files with 376 additions and 60 deletions

View file

@ -830,6 +830,7 @@ omit =
homeassistant/components/onvif/event.py homeassistant/components/onvif/event.py
homeassistant/components/onvif/parsers.py homeassistant/components/onvif/parsers.py
homeassistant/components/onvif/sensor.py homeassistant/components/onvif/sensor.py
homeassistant/components/onvif/util.py
homeassistant/components/open_meteo/weather.py homeassistant/components/open_meteo/weather.py
homeassistant/components/opencv/* homeassistant/components/opencv/*
homeassistant/components/openevse/sensor.py homeassistant/components/openevse/sensor.py

View file

@ -5,7 +5,6 @@ from pprint import pformat
from typing import Any from typing import Any
from urllib.parse import urlparse from urllib.parse import urlparse
from onvif.exceptions import ONVIFError
import voluptuous as vol import voluptuous as vol
from wsdiscovery.discovery import ThreadedWSDiscovery as WSDiscovery from wsdiscovery.discovery import ThreadedWSDiscovery as WSDiscovery
from wsdiscovery.scope import Scope from wsdiscovery.scope import Scope
@ -28,11 +27,19 @@ from homeassistant.const import (
CONF_USERNAME, CONF_USERNAME,
) )
from homeassistant.core import HomeAssistant, callback from homeassistant.core import HomeAssistant, callback
from homeassistant.data_entry_flow import FlowResult from homeassistant.data_entry_flow import AbortFlow, FlowResult
from homeassistant.helpers import device_registry as dr from homeassistant.helpers import device_registry as dr
from .const import CONF_DEVICE_ID, DEFAULT_ARGUMENTS, DEFAULT_PORT, DOMAIN, LOGGER from .const import (
CONF_DEVICE_ID,
DEFAULT_ARGUMENTS,
DEFAULT_PORT,
DOMAIN,
GET_CAPABILITIES_EXCEPTIONS,
LOGGER,
)
from .device import get_device from .device import get_device
from .util import stringify_onvif_error
CONF_MANUAL_INPUT = "Manually configure ONVIF device" CONF_MANUAL_INPUT = "Manually configure ONVIF device"
@ -175,15 +182,18 @@ class OnvifFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
return await self.async_step_configure() return await self.async_step_configure()
async def async_step_configure(self, user_input=None): async def async_step_configure(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Device configuration.""" """Device configuration."""
errors = {} errors: dict[str, str] = {}
description_placeholders: dict[str, str] = {}
if user_input: if user_input:
self.onvif_config = user_input self.onvif_config = user_input
try: errors, description_placeholders = await self.async_setup_profiles()
return await self.async_setup_profiles() if not errors:
except Fault: title = f"{self.onvif_config[CONF_NAME]} - {self.device_id}"
errors["base"] = "cannot_connect" return self.async_create_entry(title=title, data=self.onvif_config)
def conf(name, default=None): def conf(name, default=None):
return self.onvif_config.get(name, default) return self.onvif_config.get(name, default)
@ -204,9 +214,10 @@ class OnvifFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
} }
), ),
errors=errors, errors=errors,
description_placeholders=description_placeholders,
) )
async def async_setup_profiles(self): async def async_setup_profiles(self) -> tuple[dict[str, str], dict[str, str]]:
"""Fetch ONVIF device profiles.""" """Fetch ONVIF device profiles."""
LOGGER.debug( LOGGER.debug(
"Fetching profiles from ONVIF device %s", pformat(self.onvif_config) "Fetching profiles from ONVIF device %s", pformat(self.onvif_config)
@ -223,7 +234,6 @@ class OnvifFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
try: try:
await device.update_xaddrs() await device.update_xaddrs()
device_mgmt = device.create_devicemgmt_service() device_mgmt = device.create_devicemgmt_service()
# Get the MAC address to use as the unique ID for the config flow # Get the MAC address to use as the unique ID for the config flow
if not self.device_id: if not self.device_id:
try: try:
@ -237,23 +247,18 @@ class OnvifFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
except Fault as fault: except Fault as fault:
if "not implemented" not in fault.message: if "not implemented" not in fault.message:
raise fault raise fault
LOGGER.debug( LOGGER.debug(
( "%s: Could not get network interfaces: %s",
"Couldn't get network interfaces from ONVIF deivice '%s'."
" Error: %s"
),
self.onvif_config[CONF_NAME], self.onvif_config[CONF_NAME],
fault, stringify_onvif_error(fault),
) )
# If no network interfaces are exposed, fallback to serial number # If no network interfaces are exposed, fallback to serial number
if not self.device_id: if not self.device_id:
device_info = await device_mgmt.GetDeviceInformation() device_info = await device_mgmt.GetDeviceInformation()
self.device_id = device_info.SerialNumber self.device_id = device_info.SerialNumber
if not self.device_id: if not self.device_id:
return self.async_abort(reason="no_mac") raise AbortFlow(reason="no_mac")
await self.async_set_unique_id(self.device_id, raise_on_progress=False) await self.async_set_unique_id(self.device_id, raise_on_progress=False)
self._abort_if_unique_id_configured( self._abort_if_unique_id_configured(
@ -263,30 +268,42 @@ class OnvifFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
CONF_NAME: self.onvif_config[CONF_NAME], CONF_NAME: self.onvif_config[CONF_NAME],
} }
) )
# Verify there is an H264 profile # Verify there is an H264 profile
media_service = device.create_media_service() media_service = device.create_media_service()
profiles = await media_service.GetProfiles() profiles = await media_service.GetProfiles()
h264 = any( except Fault as err:
stringified_error = stringify_onvif_error(err)
description_placeholders = {"error": stringified_error}
if "auth" in stringified_error.lower():
LOGGER.debug(
"%s: Could not authenticate with camera: %s",
self.onvif_config[CONF_NAME],
stringified_error,
)
return {CONF_PASSWORD: "auth_failed"}, description_placeholders
LOGGER.debug(
"%s: Could not determine camera capabilities: %s",
self.onvif_config[CONF_NAME],
stringified_error,
exc_info=True,
)
return {"base": "onvif_error"}, description_placeholders
except GET_CAPABILITIES_EXCEPTIONS as err:
LOGGER.debug(
"%s: Could not determine camera capabilities: %s",
self.onvif_config[CONF_NAME],
stringify_onvif_error(err),
exc_info=True,
)
return {"base": "onvif_error"}, {"error": stringify_onvif_error(err)}
else:
if not any(
profile.VideoEncoderConfiguration profile.VideoEncoderConfiguration
and profile.VideoEncoderConfiguration.Encoding == "H264" and profile.VideoEncoderConfiguration.Encoding == "H264"
for profile in profiles for profile in profiles
) ):
raise AbortFlow(reason="no_h264")
if not h264: return {}, {}
return self.async_abort(reason="no_h264")
title = f"{self.onvif_config[CONF_NAME]} - {self.device_id}"
return self.async_create_entry(title=title, data=self.onvif_config)
except ONVIFError as err:
LOGGER.error(
"Couldn't setup ONVIF device '%s'. Error: %s",
self.onvif_config[CONF_NAME],
err,
)
return self.async_abort(reason="onvif_error")
finally: finally:
await device.close() await device.close()

View file

@ -1,6 +1,10 @@
"""Constants for the onvif component.""" """Constants for the onvif component."""
import logging import logging
from httpx import RequestError
from onvif.exceptions import ONVIFError
from zeep.exceptions import Fault, TransportError
LOGGER = logging.getLogger(__package__) LOGGER = logging.getLogger(__package__)
DOMAIN = "onvif" DOMAIN = "onvif"
@ -36,3 +40,8 @@ GOTOPRESET_MOVE = "GotoPreset"
STOP_MOVE = "Stop" STOP_MOVE = "Stop"
SERVICE_PTZ = "ptz" SERVICE_PTZ = "ptz"
# Some cameras don't support the GetServiceCapabilities call
# and will return a 404 error which is caught by TransportError
GET_CAPABILITIES_EXCEPTIONS = (ONVIFError, Fault, RequestError, TransportError)

View file

@ -11,7 +11,7 @@ from httpx import RequestError
import onvif import onvif
from onvif import ONVIFCamera from onvif import ONVIFCamera
from onvif.exceptions import ONVIFError from onvif.exceptions import ONVIFError
from zeep.exceptions import Fault, TransportError, XMLParseError from zeep.exceptions import Fault, XMLParseError
from homeassistant.config_entries import ConfigEntry from homeassistant.config_entries import ConfigEntry
from homeassistant.const import ( from homeassistant.const import (
@ -28,6 +28,7 @@ import homeassistant.util.dt as dt_util
from .const import ( from .const import (
ABSOLUTE_MOVE, ABSOLUTE_MOVE,
CONTINUOUS_MOVE, CONTINUOUS_MOVE,
GET_CAPABILITIES_EXCEPTIONS,
GOTOPRESET_MOVE, GOTOPRESET_MOVE,
LOGGER, LOGGER,
PAN_FACTOR, PAN_FACTOR,
@ -39,10 +40,6 @@ from .const import (
from .event import EventManager from .event import EventManager
from .models import PTZ, Capabilities, DeviceInfo, Profile, Resolution, Video from .models import PTZ, Capabilities, DeviceInfo, Profile, Resolution, Video
# Some cameras don't support the GetServiceCapabilities call
# and will return a 404 error which is caught by TransportError
GET_CAPABILITIES_EXCEPTIONS = (ONVIFError, Fault, RequestError, TransportError)
class ONVIFDevice: class ONVIFDevice:
"""Manages an ONVIF device.""" """Manages an ONVIF device."""

View file

@ -28,6 +28,7 @@ from homeassistant.helpers.network import NoURLAvailableError, get_url
from .const import DOMAIN, LOGGER from .const import DOMAIN, LOGGER
from .models import Event, PullPointManagerState, WebHookManagerState from .models import Event, PullPointManagerState, WebHookManagerState
from .parsers import PARSERS from .parsers import PARSERS
from .util import stringify_onvif_error
# Topics in this list are ignored because we do not want to create # Topics in this list are ignored because we do not want to create
# entities for them. # entities for them.
@ -60,13 +61,6 @@ PULLPOINT_MESSAGE_LIMIT = 100
PULLPOINT_COOLDOWN_TIME = 0.75 PULLPOINT_COOLDOWN_TIME = 0.75
def _stringify_onvif_error(error: Exception) -> str:
"""Stringify ONVIF error."""
if isinstance(error, Fault):
return error.message or str(error) or "Device sent empty error"
return str(error)
class EventManager: class EventManager:
"""ONVIF Event Manager.""" """ONVIF Event Manager."""
@ -349,7 +343,7 @@ class PullPointManager:
LOGGER.debug( LOGGER.debug(
"%s: Device does not support PullPoint service or has too many subscriptions: %s", "%s: Device does not support PullPoint service or has too many subscriptions: %s",
self._name, self._name,
_stringify_onvif_error(err), stringify_onvif_error(err),
) )
return False return False
@ -447,7 +441,7 @@ class PullPointManager:
" This is normal if the device restarted: %s" " This is normal if the device restarted: %s"
), ),
self._name, self._name,
_stringify_onvif_error(err), stringify_onvif_error(err),
) )
self._pullpoint_subscription = None self._pullpoint_subscription = None
@ -483,7 +477,7 @@ class PullPointManager:
LOGGER.debug( LOGGER.debug(
"%s: Failed to renew PullPoint subscription; %s", "%s: Failed to renew PullPoint subscription; %s",
self._name, self._name,
_stringify_onvif_error(err), stringify_onvif_error(err),
) )
return False return False
@ -522,7 +516,7 @@ class PullPointManager:
"%s: PullPoint subscription encountered a remote protocol error " "%s: PullPoint subscription encountered a remote protocol error "
"(this is normal for some cameras): %s", "(this is normal for some cameras): %s",
self._name, self._name,
_stringify_onvif_error(err), stringify_onvif_error(err),
) )
return True return True
except (XMLParseError, *SUBSCRIPTION_ERRORS) as err: except (XMLParseError, *SUBSCRIPTION_ERRORS) as err:
@ -531,7 +525,7 @@ class PullPointManager:
LOGGER.debug( LOGGER.debug(
"%s: Failed to fetch PullPoint subscription messages: %s", "%s: Failed to fetch PullPoint subscription messages: %s",
self._name, self._name,
_stringify_onvif_error(err), stringify_onvif_error(err),
) )
# Treat errors as if the camera restarted. Assume that the pullpoint # Treat errors as if the camera restarted. Assume that the pullpoint
# subscription is no longer valid. # subscription is no longer valid.
@ -694,7 +688,7 @@ class WebHookManager:
LOGGER.debug( LOGGER.debug(
"%s: Device does not support notification service or too many subscriptions: %s", "%s: Device does not support notification service or too many subscriptions: %s",
self._name, self._name,
_stringify_onvif_error(err), stringify_onvif_error(err),
) )
return False return False
@ -735,7 +729,7 @@ class WebHookManager:
LOGGER.debug( LOGGER.debug(
"%s: Failed to renew webhook subscription %s", "%s: Failed to renew webhook subscription %s",
self._name, self._name,
_stringify_onvif_error(err), stringify_onvif_error(err),
) )
return False return False
@ -863,6 +857,6 @@ class WebHookManager:
" This is normal if the device restarted: %s" " This is normal if the device restarted: %s"
), ),
self._name, self._name,
_stringify_onvif_error(err), stringify_onvif_error(err),
) )
self._webhook_subscription = None self._webhook_subscription = None

View file

@ -4,11 +4,12 @@
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]", "already_configured": "[%key:common::config_flow::abort::already_configured_device%]",
"no_devices_found": "[%key:common::config_flow::abort::no_devices_found%]", "no_devices_found": "[%key:common::config_flow::abort::no_devices_found%]",
"already_in_progress": "[%key:common::config_flow::abort::already_in_progress%]", "already_in_progress": "[%key:common::config_flow::abort::already_in_progress%]",
"onvif_error": "Error setting up ONVIF device. Check logs for more information.",
"no_h264": "There were no H264 streams available. Check the profile configuration on your device.", "no_h264": "There were no H264 streams available. Check the profile configuration on your device.",
"no_mac": "Could not configure unique ID for ONVIF device." "no_mac": "Could not configure unique ID for ONVIF device."
}, },
"error": { "error": {
"onvif_error": "Error setting up ONVIF device: {error}. Check logs for more information.",
"auth_failed": "Could not authenticate: {error}",
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]" "cannot_connect": "[%key:common::config_flow::error::cannot_connect%]"
}, },
"step": { "step": {

View file

@ -0,0 +1,11 @@
"""ONVIF util."""
from __future__ import annotations
from zeep.exceptions import Fault
def stringify_onvif_error(error: Exception) -> str:
"""Stringify ONVIF error."""
if isinstance(error, Fault):
return error.message or str(error) or "Device sent empty error"
return str(error)

View file

@ -1,6 +1,7 @@
"""Tests for the ONVIF integration.""" """Tests for the ONVIF integration."""
from unittest.mock import AsyncMock, MagicMock, patch from unittest.mock import AsyncMock, MagicMock, patch
from onvif.exceptions import ONVIFError
from zeep.exceptions import Fault from zeep.exceptions import Fault
from homeassistant import config_entries from homeassistant import config_entries
@ -39,6 +40,10 @@ def setup_mock_onvif_camera(
with_interfaces=True, with_interfaces=True,
with_interfaces_not_implemented=False, with_interfaces_not_implemented=False,
with_serial=True, with_serial=True,
profiles_transient_failure=False,
auth_fail=False,
update_xaddrs_fail=False,
no_profiles=False,
): ):
"""Prepare mock onvif.ONVIFCamera.""" """Prepare mock onvif.ONVIFCamera."""
devicemgmt = MagicMock() devicemgmt = MagicMock()
@ -67,9 +72,21 @@ def setup_mock_onvif_camera(
profile2 = MagicMock() profile2 = MagicMock()
profile2.VideoEncoderConfiguration.Encoding = "H264" if two_profiles else "MJPEG" profile2.VideoEncoderConfiguration.Encoding = "H264" if two_profiles else "MJPEG"
media_service.GetProfiles = AsyncMock(return_value=[profile1, profile2]) if auth_fail:
media_service.GetProfiles = AsyncMock(side_effect=Fault("Authority failure"))
elif profiles_transient_failure:
media_service.GetProfiles = AsyncMock(side_effect=Fault("camera not ready"))
elif no_profiles:
media_service.GetProfiles = AsyncMock(return_value=[])
else:
media_service.GetProfiles = AsyncMock(return_value=[profile1, profile2])
mock_onvif_camera.update_xaddrs = AsyncMock(return_value=True) if update_xaddrs_fail:
mock_onvif_camera.update_xaddrs = AsyncMock(
side_effect=ONVIFError("camera not ready")
)
else:
mock_onvif_camera.update_xaddrs = AsyncMock(return_value=True)
mock_onvif_camera.create_devicemgmt_service = MagicMock(return_value=devicemgmt) mock_onvif_camera.create_devicemgmt_service = MagicMock(return_value=devicemgmt)
mock_onvif_camera.create_media_service = MagicMock(return_value=media_service) mock_onvif_camera.create_media_service = MagicMock(return_value=media_service)
mock_onvif_camera.close = AsyncMock(return_value=None) mock_onvif_camera.close = AsyncMock(return_value=None)

View file

@ -328,6 +328,275 @@ async def test_flow_manual_entry(hass: HomeAssistant) -> None:
} }
async def test_flow_manual_entry_no_profiles(hass: HomeAssistant) -> None:
"""Test that config flow when no profiles are returned."""
result = await hass.config_entries.flow.async_init(
config_flow.DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["step_id"] == "user"
with patch(
"homeassistant.components.onvif.config_flow.get_device"
) as mock_onvif_camera, patch(
"homeassistant.components.onvif.config_flow.wsdiscovery"
) as mock_discovery, patch(
"homeassistant.components.onvif.ONVIFDevice"
) as mock_device:
setup_mock_onvif_camera(mock_onvif_camera, no_profiles=True)
# no discovery
mock_discovery.return_value = []
setup_mock_device(mock_device)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={"auto": False},
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
config_flow.CONF_NAME: NAME,
config_flow.CONF_HOST: HOST,
config_flow.CONF_PORT: PORT,
config_flow.CONF_USERNAME: USERNAME,
config_flow.CONF_PASSWORD: PASSWORD,
},
)
assert result["type"] == data_entry_flow.FlowResultType.ABORT
assert result["reason"] == "no_h264"
async def test_flow_manual_entry_no_mac(hass: HomeAssistant) -> None:
"""Test that config flow when no mac address is returned."""
result = await hass.config_entries.flow.async_init(
config_flow.DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["step_id"] == "user"
with patch(
"homeassistant.components.onvif.config_flow.get_device"
) as mock_onvif_camera, patch(
"homeassistant.components.onvif.config_flow.wsdiscovery"
) as mock_discovery, patch(
"homeassistant.components.onvif.ONVIFDevice"
) as mock_device:
setup_mock_onvif_camera(
mock_onvif_camera, with_serial=False, with_interfaces=False
)
# no discovery
mock_discovery.return_value = []
setup_mock_device(mock_device)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={"auto": False},
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
config_flow.CONF_NAME: NAME,
config_flow.CONF_HOST: HOST,
config_flow.CONF_PORT: PORT,
config_flow.CONF_USERNAME: USERNAME,
config_flow.CONF_PASSWORD: PASSWORD,
},
)
assert result["type"] == data_entry_flow.FlowResultType.ABORT
assert result["reason"] == "no_mac"
async def test_flow_manual_entry_fails(hass: HomeAssistant) -> None:
"""Test that we get a good error when manual entry fails."""
result = await hass.config_entries.flow.async_init(
config_flow.DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["step_id"] == "user"
with patch(
"homeassistant.components.onvif.config_flow.get_device"
) as mock_onvif_camera, patch(
"homeassistant.components.onvif.config_flow.wsdiscovery"
) as mock_discovery, patch(
"homeassistant.components.onvif.ONVIFDevice"
) as mock_device:
setup_mock_onvif_camera(
mock_onvif_camera, two_profiles=True, profiles_transient_failure=True
)
# no discovery
mock_discovery.return_value = []
setup_mock_device(mock_device)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={"auto": False},
)
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["step_id"] == "configure"
with patch(
"homeassistant.components.onvif.async_setup_entry", return_value=True
) as mock_setup_entry:
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
config_flow.CONF_NAME: NAME,
config_flow.CONF_HOST: HOST,
config_flow.CONF_PORT: PORT,
config_flow.CONF_USERNAME: USERNAME,
config_flow.CONF_PASSWORD: PASSWORD,
},
)
await hass.async_block_till_done()
assert len(mock_setup_entry.mock_calls) == 0
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["step_id"] == "configure"
assert result["errors"] == {"base": "onvif_error"}
assert result["description_placeholders"] == {"error": "camera not ready"}
setup_mock_onvif_camera(
mock_onvif_camera, two_profiles=True, update_xaddrs_fail=True
)
with patch(
"homeassistant.components.onvif.async_setup_entry", return_value=True
) as mock_setup_entry:
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
config_flow.CONF_NAME: NAME,
config_flow.CONF_HOST: HOST,
config_flow.CONF_PORT: PORT,
config_flow.CONF_USERNAME: USERNAME,
config_flow.CONF_PASSWORD: PASSWORD,
},
)
await hass.async_block_till_done()
assert len(mock_setup_entry.mock_calls) == 0
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["step_id"] == "configure"
assert result["errors"] == {"base": "onvif_error"}
assert result["description_placeholders"] == {
"error": "Unknown error: camera not ready"
}
setup_mock_onvif_camera(mock_onvif_camera, two_profiles=True)
with patch(
"homeassistant.components.onvif.async_setup_entry", return_value=True
) as mock_setup_entry:
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
config_flow.CONF_NAME: NAME,
config_flow.CONF_HOST: HOST,
config_flow.CONF_PORT: PORT,
config_flow.CONF_USERNAME: USERNAME,
config_flow.CONF_PASSWORD: PASSWORD,
},
)
await hass.async_block_till_done()
assert len(mock_setup_entry.mock_calls) == 1
assert result["title"] == f"{NAME} - {MAC}"
assert result["data"] == {
config_flow.CONF_NAME: NAME,
config_flow.CONF_HOST: HOST,
config_flow.CONF_PORT: PORT,
config_flow.CONF_USERNAME: USERNAME,
config_flow.CONF_PASSWORD: PASSWORD,
}
async def test_flow_manual_entry_wrong_password(hass: HomeAssistant) -> None:
"""Test that we get a an auth error with the wrong password."""
result = await hass.config_entries.flow.async_init(
config_flow.DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["step_id"] == "user"
with patch(
"homeassistant.components.onvif.config_flow.get_device"
) as mock_onvif_camera, patch(
"homeassistant.components.onvif.config_flow.wsdiscovery"
) as mock_discovery, patch(
"homeassistant.components.onvif.ONVIFDevice"
) as mock_device:
setup_mock_onvif_camera(mock_onvif_camera, two_profiles=True, auth_fail=True)
# no discovery
mock_discovery.return_value = []
setup_mock_device(mock_device)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={"auto": False},
)
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["step_id"] == "configure"
with patch(
"homeassistant.components.onvif.async_setup_entry", return_value=True
) as mock_setup_entry:
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
config_flow.CONF_NAME: NAME,
config_flow.CONF_HOST: HOST,
config_flow.CONF_PORT: PORT,
config_flow.CONF_USERNAME: USERNAME,
config_flow.CONF_PASSWORD: PASSWORD,
},
)
await hass.async_block_till_done()
assert len(mock_setup_entry.mock_calls) == 0
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["step_id"] == "configure"
assert result["errors"] == {"password": "auth_failed"}
assert result["description_placeholders"] == {"error": "Authority failure"}
setup_mock_onvif_camera(mock_onvif_camera, two_profiles=True)
with patch(
"homeassistant.components.onvif.async_setup_entry", return_value=True
) as mock_setup_entry:
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
config_flow.CONF_NAME: NAME,
config_flow.CONF_HOST: HOST,
config_flow.CONF_PORT: PORT,
config_flow.CONF_USERNAME: USERNAME,
config_flow.CONF_PASSWORD: PASSWORD,
},
)
await hass.async_block_till_done()
assert len(mock_setup_entry.mock_calls) == 1
assert result["title"] == f"{NAME} - {MAC}"
assert result["data"] == {
config_flow.CONF_NAME: NAME,
config_flow.CONF_HOST: HOST,
config_flow.CONF_PORT: PORT,
config_flow.CONF_USERNAME: USERNAME,
config_flow.CONF_PASSWORD: PASSWORD,
}
async def test_option_flow(hass: HomeAssistant) -> None: async def test_option_flow(hass: HomeAssistant) -> None:
"""Test config flow options.""" """Test config flow options."""
entry, _, _ = await setup_onvif_integration(hass) entry, _, _ = await setup_onvif_integration(hass)