mirror of https://github.com/home-assistant/core
Add config_flow and stream selection to foscam (#41429)
* Add config_flow and stream selection to foscam * Simplify config_flow steps * Make debug log entry more useful * Deprecate config and platform schemas * Simplify config loading * Add config flow testing * Remove unneeded CONFIG_SCHEMA deprecation * Improve test coverage * Unload service by tracking loaded entries * Address comment Co-authored-by: Paulus Schoutsen <paulus@home-assistant.io>
This commit is contained in:
Sergio Conde Gómez
GitHub
parent
3537a7c3d5
commit
83b210061d
|
|
@ -300,8 +300,8 @@ omit =
|
|||
homeassistant/components/folder_watcher/*
|
||||
homeassistant/components/foobot/sensor.py
|
||||
homeassistant/components/fortios/device_tracker.py
|
||||
homeassistant/components/foscam/__init__.py
|
||||
homeassistant/components/foscam/camera.py
|
||||
homeassistant/components/foscam/const.py
|
||||
homeassistant/components/foursquare/*
|
||||
homeassistant/components/free_mobile/notify.py
|
||||
homeassistant/components/freebox/__init__.py
|
||||
|
|
|
|||
|
|
@ -1 +1,51 @@
|
|||
"""The foscam component."""
|
||||
import asyncio
|
||||
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.core import HomeAssistant
|
||||
|
||||
from .const import DOMAIN, SERVICE_PTZ
|
||||
|
||||
PLATFORMS = ["camera"]
|
||||
|
||||
CONFIG_SCHEMA = vol.Schema({DOMAIN: vol.Schema({})}, extra=vol.ALLOW_EXTRA)
|
||||
|
||||
|
||||
async def async_setup(hass: HomeAssistant, config: dict):
|
||||
"""Set up the foscam component."""
|
||||
hass.data.setdefault(DOMAIN, {})
|
||||
return True
|
||||
|
||||
|
||||
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry):
|
||||
"""Set up foscam from a config entry."""
|
||||
for component in PLATFORMS:
|
||||
hass.async_create_task(
|
||||
hass.config_entries.async_forward_entry_setup(entry, component)
|
||||
)
|
||||
|
||||
hass.data[DOMAIN][entry.unique_id] = entry.data
|
||||
|
||||
return True
|
||||
|
||||
|
||||
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry):
|
||||
"""Unload a config entry."""
|
||||
unload_ok = all(
|
||||
await asyncio.gather(
|
||||
*[
|
||||
hass.config_entries.async_forward_entry_unload(entry, component)
|
||||
for component in PLATFORMS
|
||||
]
|
||||
)
|
||||
)
|
||||
|
||||
if unload_ok:
|
||||
hass.data[DOMAIN].pop(entry.unique_id)
|
||||
|
||||
if not hass.data[DOMAIN]:
|
||||
hass.services.async_remove(domain=DOMAIN, service=SERVICE_PTZ)
|
||||
|
||||
return unload_ok
|
||||
|
|
|
|||
|
|
@ -1,13 +1,14 @@
|
|||
"""This component provides basic support for Foscam IP cameras."""
|
||||
import asyncio
|
||||
import logging
|
||||
|
||||
from libpyfoscam import FoscamCamera
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant.components.camera import PLATFORM_SCHEMA, SUPPORT_STREAM, Camera
|
||||
from homeassistant.config_entries import SOURCE_IMPORT
|
||||
from homeassistant.const import (
|
||||
ATTR_ENTITY_ID,
|
||||
CONF_HOST,
|
||||
CONF_NAME,
|
||||
CONF_PASSWORD,
|
||||
CONF_PORT,
|
||||
|
|
@ -15,21 +16,18 @@ from homeassistant.const import (
|
|||
)
|
||||
from homeassistant.helpers import config_validation as cv, entity_platform
|
||||
|
||||
from .const import DATA as FOSCAM_DATA, ENTITIES as FOSCAM_ENTITIES
|
||||
from .const import CONF_STREAM, DOMAIN, LOGGER, SERVICE_PTZ
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
CONF_IP = "ip"
|
||||
CONF_RTSP_PORT = "rtsp_port"
|
||||
|
||||
DEFAULT_NAME = "Foscam Camera"
|
||||
DEFAULT_PORT = 88
|
||||
|
||||
SERVICE_PTZ = "ptz"
|
||||
ATTR_MOVEMENT = "movement"
|
||||
ATTR_TRAVELTIME = "travel_time"
|
||||
|
||||
DEFAULT_TRAVELTIME = 0.125
|
||||
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
|
||||
{
|
||||
vol.Required("ip"): cv.string,
|
||||
vol.Required(CONF_PASSWORD): cv.string,
|
||||
vol.Required(CONF_USERNAME): cv.string,
|
||||
vol.Optional(CONF_NAME, default="Foscam Camera"): cv.string,
|
||||
vol.Optional(CONF_PORT, default=88): cv.port,
|
||||
vol.Optional("rtsp_port"): cv.port,
|
||||
}
|
||||
)
|
||||
|
||||
DIR_UP = "up"
|
||||
DIR_DOWN = "down"
|
||||
|
|
@ -52,16 +50,11 @@ MOVEMENT_ATTRS = {
|
|||
DIR_BOTTOMRIGHT: "ptz_move_bottom_right",
|
||||
}
|
||||
|
||||
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
|
||||
{
|
||||
vol.Required(CONF_IP): cv.string,
|
||||
vol.Required(CONF_PASSWORD): cv.string,
|
||||
vol.Required(CONF_USERNAME): cv.string,
|
||||
vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
|
||||
vol.Optional(CONF_PORT, default=DEFAULT_PORT): cv.port,
|
||||
vol.Optional(CONF_RTSP_PORT): cv.port,
|
||||
}
|
||||
)
|
||||
DEFAULT_TRAVELTIME = 0.125
|
||||
|
||||
ATTR_MOVEMENT = "movement"
|
||||
ATTR_TRAVELTIME = "travel_time"
|
||||
|
||||
|
||||
SERVICE_PTZ_SCHEMA = vol.Schema(
|
||||
{
|
||||
|
|
@ -85,83 +78,90 @@ SERVICE_PTZ_SCHEMA = vol.Schema(
|
|||
|
||||
async def async_setup_platform(hass, config, async_add_entities, discovery_info=None):
|
||||
"""Set up a Foscam IP Camera."""
|
||||
LOGGER.warning(
|
||||
"Loading foscam via platform config is deprecated, it will be automatically imported. Please remove it afterwards."
|
||||
)
|
||||
|
||||
config_new = {
|
||||
CONF_NAME: config[CONF_NAME],
|
||||
CONF_HOST: config["ip"],
|
||||
CONF_PORT: config[CONF_PORT],
|
||||
CONF_USERNAME: config[CONF_USERNAME],
|
||||
CONF_PASSWORD: config[CONF_PASSWORD],
|
||||
CONF_STREAM: "Main",
|
||||
}
|
||||
|
||||
hass.async_create_task(
|
||||
hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": SOURCE_IMPORT}, data=config_new
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
async def async_setup_entry(hass, config_entry, async_add_entities):
|
||||
"""Add a Foscam IP camera from a config entry."""
|
||||
platform = entity_platform.current_platform.get()
|
||||
assert platform is not None
|
||||
platform.async_register_entity_service(
|
||||
"ptz",
|
||||
{
|
||||
vol.Required(ATTR_MOVEMENT): vol.In(
|
||||
[
|
||||
DIR_UP,
|
||||
DIR_DOWN,
|
||||
DIR_LEFT,
|
||||
DIR_RIGHT,
|
||||
DIR_TOPLEFT,
|
||||
DIR_TOPRIGHT,
|
||||
DIR_BOTTOMLEFT,
|
||||
DIR_BOTTOMRIGHT,
|
||||
]
|
||||
),
|
||||
vol.Optional(ATTR_TRAVELTIME, default=DEFAULT_TRAVELTIME): cv.small_float,
|
||||
},
|
||||
"async_perform_ptz",
|
||||
SERVICE_PTZ, SERVICE_PTZ_SCHEMA, "async_perform_ptz"
|
||||
)
|
||||
|
||||
camera = FoscamCamera(
|
||||
config[CONF_IP],
|
||||
config[CONF_PORT],
|
||||
config[CONF_USERNAME],
|
||||
config[CONF_PASSWORD],
|
||||
config_entry.data[CONF_HOST],
|
||||
config_entry.data[CONF_PORT],
|
||||
config_entry.data[CONF_USERNAME],
|
||||
config_entry.data[CONF_PASSWORD],
|
||||
verbose=False,
|
||||
)
|
||||
|
||||
rtsp_port = config.get(CONF_RTSP_PORT)
|
||||
if not rtsp_port:
|
||||
ret, response = await hass.async_add_executor_job(camera.get_port_info)
|
||||
|
||||
if ret == 0:
|
||||
rtsp_port = response.get("rtspPort") or response.get("mediaPort")
|
||||
|
||||
ret, response = await hass.async_add_executor_job(camera.get_motion_detect_config)
|
||||
|
||||
motion_status = False
|
||||
if ret != 0 and response == 1:
|
||||
motion_status = True
|
||||
|
||||
async_add_entities(
|
||||
[
|
||||
HassFoscamCamera(
|
||||
camera,
|
||||
config[CONF_NAME],
|
||||
config[CONF_USERNAME],
|
||||
config[CONF_PASSWORD],
|
||||
rtsp_port,
|
||||
motion_status,
|
||||
)
|
||||
]
|
||||
)
|
||||
async_add_entities([HassFoscamCamera(camera, config_entry)])
|
||||
|
||||
|
||||
class HassFoscamCamera(Camera):
|
||||
"""An implementation of a Foscam IP camera."""
|
||||
|
||||
def __init__(self, camera, name, username, password, rtsp_port, motion_status):
|
||||
def __init__(self, camera, config_entry):
|
||||
"""Initialize a Foscam camera."""
|
||||
super().__init__()
|
||||
|
||||
self._foscam_session = camera
|
||||
self._name = name
|
||||
self._username = username
|
||||
self._password = password
|
||||
self._rtsp_port = rtsp_port
|
||||
self._motion_status = motion_status
|
||||
self._name = config_entry.title
|
||||
self._username = config_entry.data[CONF_USERNAME]
|
||||
self._password = config_entry.data[CONF_PASSWORD]
|
||||
self._stream = config_entry.data[CONF_STREAM]
|
||||
self._unique_id = config_entry.unique_id
|
||||
self._rtsp_port = None
|
||||
self._motion_status = False
|
||||
|
||||
async def async_added_to_hass(self):
|
||||
"""Handle entity addition to hass."""
|
||||
entities = self.hass.data.setdefault(FOSCAM_DATA, {}).setdefault(
|
||||
FOSCAM_ENTITIES, []
|
||||
# Get motion detection status
|
||||
ret, response = await self.hass.async_add_executor_job(
|
||||
self._foscam_session.get_motion_detect_config
|
||||
)
|
||||
entities.append(self)
|
||||
|
||||
if ret != 0:
|
||||
LOGGER.error(
|
||||
"Error getting motion detection status of %s: %s", self._name, ret
|
||||
)
|
||||
|
||||
else:
|
||||
self._motion_status = response == 1
|
||||
|
||||
# Get RTSP port
|
||||
ret, response = await self.hass.async_add_executor_job(
|
||||
self._foscam_session.get_port_info
|
||||
)
|
||||
|
||||
if ret != 0:
|
||||
LOGGER.error("Error getting RTSP port of %s: %s", self._name, ret)
|
||||
|
||||
else:
|
||||
self._rtsp_port = response.get("rtspPort") or response.get("mediaPort")
|
||||
|
||||
@property
|
||||
def unique_id(self):
|
||||
"""Return the entity unique ID."""
|
||||
return self._unique_id
|
||||
|
||||
def camera_image(self):
|
||||
"""Return a still image response from the camera."""
|
||||
|
|
@ -178,12 +178,14 @@ class HassFoscamCamera(Camera):
|
|||
"""Return supported features."""
|
||||
if self._rtsp_port:
|
||||
return SUPPORT_STREAM
|
||||
return 0
|
||||
|
||||
return None
|
||||
|
||||
async def stream_source(self):
|
||||
"""Return the stream source."""
|
||||
if self._rtsp_port:
|
||||
return f"rtsp://{self._username}:{self._password}@{self._foscam_session.host}:{self._rtsp_port}/videoMain"
|
||||
return f"rtsp://{self._username}:{self._password}@{self._foscam_session.host}:{self._rtsp_port}/video{self._stream}"
|
||||
|
||||
return None
|
||||
|
||||
@property
|
||||
|
|
@ -201,7 +203,10 @@ class HassFoscamCamera(Camera):
|
|||
|
||||
self._motion_status = True
|
||||
except TypeError:
|
||||
_LOGGER.debug("Communication problem")
|
||||
LOGGER.debug(
|
||||
"Failed enabling motion detection on '%s'. Is it supported by the device?",
|
||||
self._name,
|
||||
)
|
||||
|
||||
def disable_motion_detection(self):
|
||||
"""Disable motion detection."""
|
||||
|
|
@ -213,18 +218,21 @@ class HassFoscamCamera(Camera):
|
|||
|
||||
self._motion_status = False
|
||||
except TypeError:
|
||||
_LOGGER.debug("Communication problem")
|
||||
LOGGER.debug(
|
||||
"Failed disabling motion detection on '%s'. Is it supported by the device?",
|
||||
self._name,
|
||||
)
|
||||
|
||||
async def async_perform_ptz(self, movement, travel_time):
|
||||
"""Perform a PTZ action on the camera."""
|
||||
_LOGGER.debug("PTZ action '%s' on %s", movement, self._name)
|
||||
LOGGER.debug("PTZ action '%s' on %s", movement, self._name)
|
||||
|
||||
movement_function = getattr(self._foscam_session, MOVEMENT_ATTRS[movement])
|
||||
|
||||
ret, _ = await self.hass.async_add_executor_job(movement_function)
|
||||
|
||||
if ret != 0:
|
||||
_LOGGER.error("Error moving %s '%s': %s", movement, self._name, ret)
|
||||
LOGGER.error("Error moving %s '%s': %s", movement, self._name, ret)
|
||||
return
|
||||
|
||||
await asyncio.sleep(travel_time)
|
||||
|
|
@ -234,7 +242,7 @@ class HassFoscamCamera(Camera):
|
|||
)
|
||||
|
||||
if ret != 0:
|
||||
_LOGGER.error("Error stopping movement on '%s': %s", self._name, ret)
|
||||
LOGGER.error("Error stopping movement on '%s': %s", self._name, ret)
|
||||
return
|
||||
|
||||
@property
|
||||
|
|
|
|||
|
|
@ -0,0 +1,123 @@
|
|||
"""Config flow for foscam integration."""
|
||||
from libpyfoscam import FoscamCamera
|
||||
from libpyfoscam.foscam import ERROR_FOSCAM_AUTH, ERROR_FOSCAM_UNAVAILABLE
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant import config_entries, exceptions
|
||||
from homeassistant.const import (
|
||||
CONF_HOST,
|
||||
CONF_NAME,
|
||||
CONF_PASSWORD,
|
||||
CONF_PORT,
|
||||
CONF_USERNAME,
|
||||
)
|
||||
from homeassistant.data_entry_flow import AbortFlow
|
||||
|
||||
from .const import CONF_STREAM, LOGGER
|
||||
from .const import DOMAIN # pylint:disable=unused-import
|
||||
|
||||
STREAMS = ["Main", "Sub"]
|
||||
|
||||
DEFAULT_PORT = 88
|
||||
|
||||
|
||||
DATA_SCHEMA = vol.Schema(
|
||||
{
|
||||
vol.Required(CONF_HOST): str,
|
||||
vol.Required(CONF_PORT, default=DEFAULT_PORT): int,
|
||||
vol.Required(CONF_USERNAME): str,
|
||||
vol.Required(CONF_PASSWORD): str,
|
||||
vol.Required(CONF_STREAM, default=STREAMS[0]): vol.In(STREAMS),
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
|
||||
"""Handle a config flow for foscam."""
|
||||
|
||||
VERSION = 1
|
||||
CONNECTION_CLASS = config_entries.CONN_CLASS_LOCAL_POLL
|
||||
|
||||
async def _validate_and_create(self, data):
|
||||
"""Validate the user input allows us to connect.
|
||||
|
||||
Data has the keys from DATA_SCHEMA with values provided by the user.
|
||||
"""
|
||||
camera = FoscamCamera(
|
||||
data[CONF_HOST],
|
||||
data[CONF_PORT],
|
||||
data[CONF_USERNAME],
|
||||
data[CONF_PASSWORD],
|
||||
verbose=False,
|
||||
)
|
||||
|
||||
# Validate data by sending a request to the camera
|
||||
ret, response = await self.hass.async_add_executor_job(camera.get_dev_info)
|
||||
|
||||
if ret == ERROR_FOSCAM_UNAVAILABLE:
|
||||
raise CannotConnect
|
||||
|
||||
if ret == ERROR_FOSCAM_AUTH:
|
||||
raise InvalidAuth
|
||||
|
||||
await self.async_set_unique_id(response["mac"])
|
||||
self._abort_if_unique_id_configured()
|
||||
|
||||
name = data.pop(CONF_NAME, response["devName"])
|
||||
|
||||
return self.async_create_entry(title=name, data=data)
|
||||
|
||||
async def async_step_user(self, user_input=None):
|
||||
"""Handle the initial step."""
|
||||
errors = {}
|
||||
|
||||
if user_input is not None:
|
||||
try:
|
||||
return await self._validate_and_create(user_input)
|
||||
|
||||
except CannotConnect:
|
||||
errors["base"] = "cannot_connect"
|
||||
|
||||
except InvalidAuth:
|
||||
errors["base"] = "invalid_auth"
|
||||
|
||||
except AbortFlow:
|
||||
raise
|
||||
|
||||
except Exception: # pylint: disable=broad-except
|
||||
LOGGER.exception("Unexpected exception")
|
||||
errors["base"] = "unknown"
|
||||
|
||||
return self.async_show_form(
|
||||
step_id="user", data_schema=DATA_SCHEMA, errors=errors
|
||||
)
|
||||
|
||||
async def async_step_import(self, import_config):
|
||||
"""Handle config import from yaml."""
|
||||
try:
|
||||
return await self._validate_and_create(import_config)
|
||||
|
||||
except CannotConnect:
|
||||
LOGGER.error("Error importing foscam platform config: cannot connect.")
|
||||
return self.async_abort(reason="cannot_connect")
|
||||
|
||||
except InvalidAuth:
|
||||
LOGGER.error("Error importing foscam platform config: invalid auth.")
|
||||
return self.async_abort(reason="invalid_auth")
|
||||
|
||||
except AbortFlow:
|
||||
raise
|
||||
|
||||
except Exception: # pylint: disable=broad-except
|
||||
LOGGER.exception(
|
||||
"Error importing foscam platform config: unexpected exception."
|
||||
)
|
||||
return self.async_abort(reason="unknown")
|
||||
|
||||
|
||||
class CannotConnect(exceptions.HomeAssistantError):
|
||||
"""Error to indicate we cannot connect."""
|
||||
|
||||
|
||||
class InvalidAuth(exceptions.HomeAssistantError):
|
||||
"""Error to indicate there is invalid auth."""
|
||||
|
|
@ -1,5 +1,10 @@
|
|||
"""Constants for Foscam component."""
|
||||
import logging
|
||||
|
||||
LOGGER = logging.getLogger(__package__)
|
||||
|
||||
DOMAIN = "foscam"
|
||||
DATA = "foscam"
|
||||
ENTITIES = "entities"
|
||||
|
||||
CONF_STREAM = "stream"
|
||||
|
||||
SERVICE_PTZ = "ptz"
|
||||
|
|
|
|||
|
|
@ -1,6 +1,7 @@
|
|||
{
|
||||
"domain": "foscam",
|
||||
"name": "Foscam",
|
||||
"config_flow": true,
|
||||
"documentation": "https://www.home-assistant.io/integrations/foscam",
|
||||
"requirements": ["libpyfoscam==1.0"],
|
||||
"codeowners": ["@skgsergio"]
|
||||
|
|
|
|||
|
|
@ -0,0 +1,24 @@
|
|||
{
|
||||
"title": "Foscam",
|
||||
"config": {
|
||||
"step": {
|
||||
"user": {
|
||||
"data": {
|
||||
"host": "[%key:common::config_flow::data::host%]",
|
||||
"port": "[%key:common::config_flow::data::port%]",
|
||||
"username": "[%key:common::config_flow::data::username%]",
|
||||
"password": "[%key:common::config_flow::data::password%]",
|
||||
"stream": "Stream"
|
||||
}
|
||||
}
|
||||
},
|
||||
"error": {
|
||||
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
|
||||
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
|
||||
"unknown": "[%key:common::config_flow::error::unknown%]"
|
||||
},
|
||||
"abort": {
|
||||
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]"
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,24 @@
|
|||
{
|
||||
"config": {
|
||||
"abort": {
|
||||
"already_configured": "Device is already configured"
|
||||
},
|
||||
"error": {
|
||||
"cannot_connect": "Failed to connect",
|
||||
"invalid_auth": "Invalid authentication",
|
||||
"unknown": "Unexpected error"
|
||||
},
|
||||
"step": {
|
||||
"user": {
|
||||
"data": {
|
||||
"host": "Host",
|
||||
"password": "Password",
|
||||
"port": "Port",
|
||||
"stream": "Stream",
|
||||
"username": "Username"
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
"title": "Foscam"
|
||||
}
|
||||
|
|
@ -66,6 +66,7 @@ FLOWS = [
|
|||
"flume",
|
||||
"flunearyou",
|
||||
"forked_daapd",
|
||||
"foscam",
|
||||
"freebox",
|
||||
"fritzbox",
|
||||
"garmin_connect",
|
||||
|
|
|
|||
|
|
@ -443,6 +443,9 @@ konnected==1.2.0
|
|||
# homeassistant.components.dyson
|
||||
libpurecool==0.6.4
|
||||
|
||||
# homeassistant.components.foscam
|
||||
libpyfoscam==1.0
|
||||
|
||||
# homeassistant.components.mikrotik
|
||||
librouteros==3.0.0
|
||||
|
||||
|
|
|
|||
|
|
@ -0,0 +1 @@
|
|||
"""Tests for the Foscam integration."""
|
||||
|
|
@ -0,0 +1,358 @@
|
|||
"""Test the Foscam config flow."""
|
||||
from unittest.mock import patch
|
||||
|
||||
from libpyfoscam.foscam import ERROR_FOSCAM_AUTH, ERROR_FOSCAM_UNAVAILABLE
|
||||
|
||||
from homeassistant import config_entries, data_entry_flow, setup
|
||||
from homeassistant.components.foscam import config_flow
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
VALID_CONFIG = {
|
||||
config_flow.CONF_HOST: "10.0.0.2",
|
||||
config_flow.CONF_PORT: 88,
|
||||
config_flow.CONF_USERNAME: "admin",
|
||||
config_flow.CONF_PASSWORD: "1234",
|
||||
config_flow.CONF_STREAM: "Main",
|
||||
}
|
||||
CAMERA_NAME = "Mocked Foscam Camera"
|
||||
CAMERA_MAC = "C0:C1:D0:F4:B4:D4"
|
||||
|
||||
|
||||
def setup_mock_foscam_camera(mock_foscam_camera):
|
||||
"""Mock FoscamCamera simulating behaviour using a base valid config."""
|
||||
|
||||
def configure_mock_on_init(host, port, user, passwd, verbose=False):
|
||||
return_code = 0
|
||||
data = {}
|
||||
|
||||
if (
|
||||
host != VALID_CONFIG[config_flow.CONF_HOST]
|
||||
or port != VALID_CONFIG[config_flow.CONF_PORT]
|
||||
):
|
||||
return_code = ERROR_FOSCAM_UNAVAILABLE
|
||||
|
||||
elif (
|
||||
user != VALID_CONFIG[config_flow.CONF_USERNAME]
|
||||
or passwd != VALID_CONFIG[config_flow.CONF_PASSWORD]
|
||||
):
|
||||
return_code = ERROR_FOSCAM_AUTH
|
||||
|
||||
else:
|
||||
data["devName"] = CAMERA_NAME
|
||||
data["mac"] = CAMERA_MAC
|
||||
|
||||
mock_foscam_camera.get_dev_info.return_value = (return_code, data)
|
||||
|
||||
return mock_foscam_camera
|
||||
|
||||
mock_foscam_camera.side_effect = configure_mock_on_init
|
||||
|
||||
|
||||
async def test_user_valid(hass):
|
||||
"""Test valid config from user input."""
|
||||
await setup.async_setup_component(hass, "persistent_notification", {})
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
config_flow.DOMAIN, context={"source": config_entries.SOURCE_USER}
|
||||
)
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
|
||||
assert result["errors"] == {}
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.foscam.config_flow.FoscamCamera",
|
||||
) as mock_foscam_camera, patch(
|
||||
"homeassistant.components.foscam.async_setup", return_value=True
|
||||
) as mock_setup, patch(
|
||||
"homeassistant.components.foscam.async_setup_entry",
|
||||
return_value=True,
|
||||
) as mock_setup_entry:
|
||||
setup_mock_foscam_camera(mock_foscam_camera)
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
VALID_CONFIG,
|
||||
)
|
||||
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
|
||||
assert result["title"] == CAMERA_NAME
|
||||
assert result["data"] == VALID_CONFIG
|
||||
|
||||
assert len(mock_setup.mock_calls) == 1
|
||||
assert len(mock_setup_entry.mock_calls) == 1
|
||||
|
||||
|
||||
async def test_user_invalid_auth(hass):
|
||||
"""Test we handle invalid auth from user input."""
|
||||
await setup.async_setup_component(hass, "persistent_notification", {})
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
config_flow.DOMAIN, context={"source": config_entries.SOURCE_USER}
|
||||
)
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
|
||||
assert result["errors"] == {}
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.foscam.config_flow.FoscamCamera",
|
||||
) as mock_foscam_camera:
|
||||
setup_mock_foscam_camera(mock_foscam_camera)
|
||||
|
||||
invalid_user = VALID_CONFIG.copy()
|
||||
invalid_user[config_flow.CONF_USERNAME] = "invalid"
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
invalid_user,
|
||||
)
|
||||
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
|
||||
assert result["errors"] == {"base": "invalid_auth"}
|
||||
|
||||
|
||||
async def test_user_cannot_connect(hass):
|
||||
"""Test we handle cannot connect error from user input."""
|
||||
await setup.async_setup_component(hass, "persistent_notification", {})
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
config_flow.DOMAIN, context={"source": config_entries.SOURCE_USER}
|
||||
)
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
|
||||
assert result["errors"] == {}
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.foscam.config_flow.FoscamCamera",
|
||||
) as mock_foscam_camera:
|
||||
setup_mock_foscam_camera(mock_foscam_camera)
|
||||
|
||||
invalid_host = VALID_CONFIG.copy()
|
||||
invalid_host[config_flow.CONF_HOST] = "127.0.0.1"
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
invalid_host,
|
||||
)
|
||||
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
|
||||
assert result["errors"] == {"base": "cannot_connect"}
|
||||
|
||||
|
||||
async def test_user_already_configured(hass):
|
||||
"""Test we handle already configured from user input."""
|
||||
await setup.async_setup_component(hass, "persistent_notification", {})
|
||||
|
||||
entry = MockConfigEntry(
|
||||
domain=config_flow.DOMAIN, data=VALID_CONFIG, unique_id=CAMERA_MAC
|
||||
)
|
||||
entry.add_to_hass(hass)
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
config_flow.DOMAIN, context={"source": config_entries.SOURCE_USER}
|
||||
)
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
|
||||
assert result["errors"] == {}
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.foscam.config_flow.FoscamCamera",
|
||||
) as mock_foscam_camera:
|
||||
setup_mock_foscam_camera(mock_foscam_camera)
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
VALID_CONFIG,
|
||||
)
|
||||
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
|
||||
assert result["reason"] == "already_configured"
|
||||
|
||||
|
||||
async def test_user_unknown_exception(hass):
|
||||
"""Test we handle unknown exceptions from user input."""
|
||||
await setup.async_setup_component(hass, "persistent_notification", {})
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
config_flow.DOMAIN, context={"source": config_entries.SOURCE_USER}
|
||||
)
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
|
||||
assert result["errors"] == {}
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.foscam.config_flow.FoscamCamera",
|
||||
) as mock_foscam_camera:
|
||||
mock_foscam_camera.side_effect = Exception("test")
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
VALID_CONFIG,
|
||||
)
|
||||
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
|
||||
assert result["errors"] == {"base": "unknown"}
|
||||
|
||||
|
||||
async def test_import_user_valid(hass):
|
||||
"""Test valid config from import."""
|
||||
with patch(
|
||||
"homeassistant.components.foscam.config_flow.FoscamCamera",
|
||||
) as mock_foscam_camera, patch(
|
||||
"homeassistant.components.foscam.async_setup", return_value=True
|
||||
) as mock_setup, patch(
|
||||
"homeassistant.components.foscam.async_setup_entry",
|
||||
return_value=True,
|
||||
) as mock_setup_entry:
|
||||
setup_mock_foscam_camera(mock_foscam_camera)
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
config_flow.DOMAIN,
|
||||
context={"source": config_entries.SOURCE_IMPORT},
|
||||
data=VALID_CONFIG,
|
||||
)
|
||||
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
|
||||
assert result["title"] == CAMERA_NAME
|
||||
assert result["data"] == VALID_CONFIG
|
||||
|
||||
assert len(mock_setup.mock_calls) == 1
|
||||
assert len(mock_setup_entry.mock_calls) == 1
|
||||
|
||||
|
||||
async def test_import_user_valid_with_name(hass):
|
||||
"""Test valid config with extra name from import."""
|
||||
with patch(
|
||||
"homeassistant.components.foscam.config_flow.FoscamCamera",
|
||||
) as mock_foscam_camera, patch(
|
||||
"homeassistant.components.foscam.async_setup", return_value=True
|
||||
) as mock_setup, patch(
|
||||
"homeassistant.components.foscam.async_setup_entry",
|
||||
return_value=True,
|
||||
) as mock_setup_entry:
|
||||
setup_mock_foscam_camera(mock_foscam_camera)
|
||||
|
||||
name = CAMERA_NAME + " 1234"
|
||||
with_name = VALID_CONFIG.copy()
|
||||
with_name[config_flow.CONF_NAME] = name
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
config_flow.DOMAIN,
|
||||
context={"source": config_entries.SOURCE_IMPORT},
|
||||
data=with_name,
|
||||
)
|
||||
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
|
||||
assert result["title"] == name
|
||||
assert result["data"] == VALID_CONFIG
|
||||
|
||||
assert len(mock_setup.mock_calls) == 1
|
||||
assert len(mock_setup_entry.mock_calls) == 1
|
||||
|
||||
|
||||
async def test_import_invalid_auth(hass):
|
||||
"""Test we handle invalid auth from import."""
|
||||
entry = MockConfigEntry(
|
||||
domain=config_flow.DOMAIN, data=VALID_CONFIG, unique_id=CAMERA_MAC
|
||||
)
|
||||
entry.add_to_hass(hass)
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.foscam.config_flow.FoscamCamera",
|
||||
) as mock_foscam_camera:
|
||||
setup_mock_foscam_camera(mock_foscam_camera)
|
||||
|
||||
invalid_user = VALID_CONFIG.copy()
|
||||
invalid_user[config_flow.CONF_USERNAME] = "invalid"
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
config_flow.DOMAIN,
|
||||
context={"source": config_entries.SOURCE_IMPORT},
|
||||
data=invalid_user,
|
||||
)
|
||||
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
|
||||
assert result["reason"] == "invalid_auth"
|
||||
|
||||
|
||||
async def test_import_cannot_connect(hass):
|
||||
"""Test we handle invalid auth from import."""
|
||||
entry = MockConfigEntry(
|
||||
domain=config_flow.DOMAIN, data=VALID_CONFIG, unique_id=CAMERA_MAC
|
||||
)
|
||||
entry.add_to_hass(hass)
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.foscam.config_flow.FoscamCamera",
|
||||
) as mock_foscam_camera:
|
||||
setup_mock_foscam_camera(mock_foscam_camera)
|
||||
|
||||
invalid_host = VALID_CONFIG.copy()
|
||||
invalid_host[config_flow.CONF_HOST] = "127.0.0.1"
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
config_flow.DOMAIN,
|
||||
context={"source": config_entries.SOURCE_IMPORT},
|
||||
data=invalid_host,
|
||||
)
|
||||
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
|
||||
assert result["reason"] == "cannot_connect"
|
||||
|
||||
|
||||
async def test_import_already_configured(hass):
|
||||
"""Test we handle already configured from import."""
|
||||
entry = MockConfigEntry(
|
||||
domain=config_flow.DOMAIN, data=VALID_CONFIG, unique_id=CAMERA_MAC
|
||||
)
|
||||
entry.add_to_hass(hass)
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.foscam.config_flow.FoscamCamera",
|
||||
) as mock_foscam_camera:
|
||||
setup_mock_foscam_camera(mock_foscam_camera)
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
config_flow.DOMAIN,
|
||||
context={"source": config_entries.SOURCE_IMPORT},
|
||||
data=VALID_CONFIG,
|
||||
)
|
||||
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
|
||||
assert result["reason"] == "already_configured"
|
||||
|
||||
|
||||
async def test_import_unknown_exception(hass):
|
||||
"""Test we handle unknown exceptions from import."""
|
||||
await setup.async_setup_component(hass, "persistent_notification", {})
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.foscam.config_flow.FoscamCamera",
|
||||
) as mock_foscam_camera:
|
||||
mock_foscam_camera.side_effect = Exception("test")
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
config_flow.DOMAIN,
|
||||
context={"source": config_entries.SOURCE_IMPORT},
|
||||
data=VALID_CONFIG,
|
||||
)
|
||||
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
|
||||
assert result["reason"] == "unknown"
|
||||
Loading…
Reference in New Issue