Add config flow to MJPEG IP Camera (#66607)

This commit is contained in:
Franck Nijhof 2022-02-18 09:03:41 +01:00 committed by GitHub
parent 82ebb7047f
commit e26488b1ca
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
19 changed files with 1076 additions and 40 deletions

View File

@ -722,6 +722,7 @@ omit =
homeassistant/components/minio/*
homeassistant/components/mitemp_bt/sensor.py
homeassistant/components/mjpeg/camera.py
homeassistant/components/mjpeg/util.py
homeassistant/components/mochad/*
homeassistant/components/modbus/climate.py
homeassistant/components/modem_callerid/sensor.py

View File

@ -5,7 +5,7 @@ import logging
from agent import AgentError
from homeassistant.components.camera import SUPPORT_ON_OFF
from homeassistant.components.mjpeg.camera import MjpegCamera, filter_urllib3_logging
from homeassistant.components.mjpeg import MjpegCamera, filter_urllib3_logging
from homeassistant.const import ATTR_ATTRIBUTION
from homeassistant.helpers import entity_platform
from homeassistant.helpers.entity import DeviceInfo

View File

@ -5,7 +5,7 @@ from datetime import timedelta
from pydroid_ipcam import PyDroidIPCam
import voluptuous as vol
from homeassistant.components.mjpeg.camera import CONF_MJPEG_URL, CONF_STILL_IMAGE_URL
from homeassistant.components.mjpeg import CONF_MJPEG_URL, CONF_STILL_IMAGE_URL
from homeassistant.const import (
CONF_HOST,
CONF_NAME,

View File

@ -2,7 +2,7 @@
from urllib.parse import urlencode
from homeassistant.components.camera import SUPPORT_STREAM
from homeassistant.components.mjpeg.camera import MjpegCamera, filter_urllib3_logging
from homeassistant.components.mjpeg import MjpegCamera, filter_urllib3_logging
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import HTTP_DIGEST_AUTHENTICATION
from homeassistant.core import HomeAssistant

View File

@ -1 +1,42 @@
"""The mjpeg component."""
"""The MJPEG IP Camera integration."""
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.typing import ConfigType
from .camera import MjpegCamera
from .const import CONF_MJPEG_URL, CONF_STILL_IMAGE_URL, PLATFORMS
from .util import filter_urllib3_logging
__all__ = [
"CONF_MJPEG_URL",
"CONF_STILL_IMAGE_URL",
"MjpegCamera",
"filter_urllib3_logging",
]
def setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up the MJPEG IP Camera integration."""
filter_urllib3_logging()
return True
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up from a config entry."""
hass.config_entries.async_setup_platforms(entry, PLATFORMS)
# Reload entry when its updated.
entry.async_on_unload(entry.add_update_listener(async_reload_entry))
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry."""
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
async def async_reload_entry(hass: HomeAssistant, entry: ConfigEntry) -> None:
"""Reload the config entry when it changed."""
await hass.config_entries.async_reload(entry.entry_id)

View File

@ -4,7 +4,6 @@ from __future__ import annotations
import asyncio
from collections.abc import Iterable
from contextlib import closing
import logging
import aiohttp
from aiohttp import web
@ -14,6 +13,7 @@ from requests.auth import HTTPBasicAuth, HTTPDigestAuth
import voluptuous as vol
from homeassistant.components.camera import PLATFORM_SCHEMA, Camera
from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
from homeassistant.const import (
CONF_AUTHENTICATION,
CONF_NAME,
@ -29,13 +29,12 @@ from homeassistant.helpers.aiohttp_client import (
async_aiohttp_proxy_web,
async_get_clientsession,
)
from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
_LOGGER = logging.getLogger(__name__)
from .const import CONF_MJPEG_URL, CONF_STILL_IMAGE_URL, DOMAIN, LOGGER
CONF_MJPEG_URL = "mjpeg_url"
CONF_STILL_IMAGE_URL = "still_image_url"
CONTENT_TYPE_HEADER = "Content-Type"
DEFAULT_NAME = "Mjpeg Camera"
@ -62,32 +61,50 @@ async def async_setup_platform(
async_add_entities: AddEntitiesCallback,
discovery_info: DiscoveryInfoType | None = None,
) -> None:
"""Set up a MJPEG IP Camera."""
filter_urllib3_logging()
"""Set up the MJPEG IP camera from platform."""
LOGGER.warning(
"Configuration of the MJPEG IP Camera platform in YAML is deprecated "
"and will be removed in Home Assistant 2022.5; Your existing "
"configuration has been imported into the UI automatically and can be "
"safely removed from your configuration.yaml file"
)
if discovery_info:
config = PLATFORM_SCHEMA(discovery_info)
async_add_entities(
[
MjpegCamera(
name=config[CONF_NAME],
authentication=config[CONF_AUTHENTICATION],
username=config.get(CONF_USERNAME),
password=config[CONF_PASSWORD],
mjpeg_url=config[CONF_MJPEG_URL],
still_image_url=config.get(CONF_STILL_IMAGE_URL),
verify_ssl=config[CONF_VERIFY_SSL],
)
]
hass.async_create_task(
hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_IMPORT},
data=config,
)
)
def filter_urllib3_logging() -> None:
"""Filter header errors from urllib3 due to a urllib3 bug."""
urllib3_logger = logging.getLogger("urllib3.connectionpool")
if not any(isinstance(x, NoHeaderErrorFilter) for x in urllib3_logger.filters):
urllib3_logger.addFilter(NoHeaderErrorFilter())
async def async_setup_entry(
hass: HomeAssistant,
entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up a MJPEG IP Camera based on a config entry."""
async_add_entities(
[
MjpegCamera(
name=entry.title,
authentication=entry.options[CONF_AUTHENTICATION],
username=entry.options.get(CONF_USERNAME),
password=entry.options[CONF_PASSWORD],
mjpeg_url=entry.options[CONF_MJPEG_URL],
still_image_url=entry.options.get(CONF_STILL_IMAGE_URL),
verify_ssl=entry.options[CONF_VERIFY_SSL],
unique_id=entry.entry_id,
device_info=DeviceInfo(
name=entry.title,
identifiers={(DOMAIN, entry.entry_id)},
),
)
]
)
def extract_image_from_mjpeg(stream: Iterable[bytes]) -> bytes | None:
@ -124,6 +141,8 @@ class MjpegCamera(Camera):
username: str | None = None,
password: str = "",
verify_ssl: bool = True,
unique_id: str | None = None,
device_info: DeviceInfo | None = None,
) -> None:
"""Initialize a MJPEG camera."""
super().__init__()
@ -143,6 +162,11 @@ class MjpegCamera(Camera):
self._auth = aiohttp.BasicAuth(self._username, password=self._password)
self._verify_ssl = verify_ssl
if unique_id is not None:
self._attr_unique_id = unique_id
if device_info is not None:
self._attr_device_info = device_info
async def async_camera_image(
self, width: int | None = None, height: int | None = None
) -> bytes | None:
@ -164,10 +188,10 @@ class MjpegCamera(Camera):
return image
except asyncio.TimeoutError:
_LOGGER.error("Timeout getting camera image from %s", self.name)
LOGGER.error("Timeout getting camera image from %s", self.name)
except aiohttp.ClientError as err:
_LOGGER.error("Error getting new camera image from %s: %s", self.name, err)
LOGGER.error("Error getting new camera image from %s: %s", self.name, err)
return None
@ -208,11 +232,3 @@ class MjpegCamera(Camera):
stream_coro = websession.get(self._mjpeg_url, auth=self._auth)
return await async_aiohttp_proxy_web(self.hass, request, stream_coro)
class NoHeaderErrorFilter(logging.Filter):
"""Filter out urllib3 Header Parsing Errors due to a urllib3 bug."""
def filter(self, record: logging.LogRecord) -> bool:
"""Filter out Header Parsing Errors."""
return "Failed to parse headers" not in record.getMessage()

View File

@ -0,0 +1,240 @@
"""Config flow to configure the MJPEG IP Camera integration."""
from __future__ import annotations
from http import HTTPStatus
from types import MappingProxyType
from typing import Any
import requests
from requests.auth import HTTPBasicAuth, HTTPDigestAuth
from requests.exceptions import HTTPError, Timeout
import voluptuous as vol
from homeassistant.config_entries import ConfigEntry, ConfigFlow, OptionsFlow
from homeassistant.const import (
CONF_AUTHENTICATION,
CONF_NAME,
CONF_PASSWORD,
CONF_USERNAME,
CONF_VERIFY_SSL,
HTTP_BASIC_AUTHENTICATION,
HTTP_DIGEST_AUTHENTICATION,
)
from homeassistant.core import HomeAssistant, callback
from homeassistant.data_entry_flow import FlowResult
from homeassistant.exceptions import HomeAssistantError
from .const import CONF_MJPEG_URL, CONF_STILL_IMAGE_URL, DOMAIN, LOGGER
@callback
def async_get_schema(
defaults: dict[str, Any] | MappingProxyType[str, Any], show_name: bool = False
) -> vol.Schema:
"""Return MJPEG IP Camera schema."""
schema = {
vol.Required(CONF_MJPEG_URL, default=defaults.get(CONF_MJPEG_URL)): str,
vol.Optional(
CONF_STILL_IMAGE_URL,
description={"suggested_value": defaults.get(CONF_STILL_IMAGE_URL)},
): str,
vol.Optional(
CONF_USERNAME,
description={"suggested_value": defaults.get(CONF_USERNAME)},
): str,
vol.Optional(
CONF_PASSWORD,
default=defaults.get(CONF_PASSWORD, ""),
): str,
vol.Optional(
CONF_VERIFY_SSL,
default=defaults.get(CONF_VERIFY_SSL, True),
): bool,
}
if show_name:
schema = {
vol.Optional(CONF_NAME, default=defaults.get(CONF_NAME)): str,
**schema,
}
return vol.Schema(schema)
def validate_url(
url: str,
username: str | None,
password: str,
verify_ssl: bool,
authentication: str = HTTP_BASIC_AUTHENTICATION,
) -> str:
"""Test if the given setting works as expected."""
auth: HTTPDigestAuth | HTTPBasicAuth | None = None
if username and password:
if authentication == HTTP_DIGEST_AUTHENTICATION:
auth = HTTPDigestAuth(username, password)
else:
auth = HTTPBasicAuth(username, password)
response = requests.get(
url,
auth=auth,
stream=True,
timeout=10,
verify=verify_ssl,
)
if response.status_code == HTTPStatus.UNAUTHORIZED:
# If unauthorized, try again using digest auth
if authentication == HTTP_BASIC_AUTHENTICATION:
return validate_url(
url, username, password, verify_ssl, HTTP_DIGEST_AUTHENTICATION
)
raise InvalidAuth
response.raise_for_status()
response.close()
return authentication
async def async_validate_input(
hass: HomeAssistant, user_input: dict[str, Any]
) -> tuple[dict[str, str], str]:
"""Manage MJPEG IP Camera options."""
errors = {}
field = "base"
authentication = HTTP_BASIC_AUTHENTICATION
try:
for field in (CONF_MJPEG_URL, CONF_STILL_IMAGE_URL):
if not (url := user_input.get(field)):
continue
authentication = await hass.async_add_executor_job(
validate_url,
url,
user_input.get(CONF_USERNAME),
user_input[CONF_PASSWORD],
user_input[CONF_VERIFY_SSL],
)
except InvalidAuth:
errors["username"] = "invalid_auth"
except (OSError, HTTPError, Timeout):
LOGGER.exception("Cannot connect to %s", user_input[CONF_MJPEG_URL])
errors[field] = "cannot_connect"
return (errors, authentication)
class MJPEGFlowHandler(ConfigFlow, domain=DOMAIN):
"""Config flow for MJPEG IP Camera."""
VERSION = 1
@staticmethod
@callback
def async_get_options_flow(
config_entry: ConfigEntry,
) -> MJPEGOptionsFlowHandler:
"""Get the options flow for this handler."""
return MJPEGOptionsFlowHandler(config_entry)
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle a flow initialized by the user."""
errors: dict[str, str] = {}
if user_input is not None:
errors, authentication = await async_validate_input(self.hass, user_input)
if not errors:
self._async_abort_entries_match(
{CONF_MJPEG_URL: user_input[CONF_MJPEG_URL]}
)
# Storing data in option, to allow for changing them later
# using an options flow.
return self.async_create_entry(
title=user_input.get(CONF_NAME, user_input[CONF_MJPEG_URL]),
data={},
options={
CONF_AUTHENTICATION: authentication,
CONF_MJPEG_URL: user_input[CONF_MJPEG_URL],
CONF_PASSWORD: user_input[CONF_PASSWORD],
CONF_STILL_IMAGE_URL: user_input.get(CONF_STILL_IMAGE_URL),
CONF_USERNAME: user_input.get(CONF_USERNAME),
CONF_VERIFY_SSL: user_input[CONF_VERIFY_SSL],
},
)
else:
user_input = {}
return self.async_show_form(
step_id="user",
data_schema=async_get_schema(user_input, show_name=True),
errors=errors,
)
async def async_step_import(self, config: dict[str, Any]) -> FlowResult:
"""Handle a flow initialized by importing a config."""
self._async_abort_entries_match({CONF_MJPEG_URL: config[CONF_MJPEG_URL]})
return self.async_create_entry(
title=config[CONF_NAME],
data={},
options={
CONF_AUTHENTICATION: config[CONF_AUTHENTICATION],
CONF_MJPEG_URL: config[CONF_MJPEG_URL],
CONF_PASSWORD: config[CONF_PASSWORD],
CONF_STILL_IMAGE_URL: config.get(CONF_STILL_IMAGE_URL),
CONF_USERNAME: config.get(CONF_USERNAME),
CONF_VERIFY_SSL: config[CONF_VERIFY_SSL],
},
)
class MJPEGOptionsFlowHandler(OptionsFlow):
"""Handle MJPEG IP Camera options."""
def __init__(self, config_entry: ConfigEntry) -> None:
"""Initialize MJPEG IP Camera options flow."""
self.config_entry = config_entry
async def async_step_init(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Manage MJPEG IP Camera options."""
errors: dict[str, str] = {}
if user_input is not None:
errors, authentication = await async_validate_input(self.hass, user_input)
if not errors:
for entry in self.hass.config_entries.async_entries(DOMAIN):
if (
entry.entry_id != self.config_entry.entry_id
and entry.options[CONF_MJPEG_URL] == user_input[CONF_MJPEG_URL]
):
errors = {CONF_MJPEG_URL: "already_configured"}
if not errors:
return self.async_create_entry(
title=user_input.get(CONF_NAME, user_input[CONF_MJPEG_URL]),
data={
CONF_AUTHENTICATION: authentication,
CONF_MJPEG_URL: user_input[CONF_MJPEG_URL],
CONF_PASSWORD: user_input[CONF_PASSWORD],
CONF_STILL_IMAGE_URL: user_input.get(CONF_STILL_IMAGE_URL),
CONF_USERNAME: user_input.get(CONF_USERNAME),
CONF_VERIFY_SSL: user_input[CONF_VERIFY_SSL],
},
)
else:
user_input = {}
return self.async_show_form(
step_id="init",
data_schema=async_get_schema(user_input or self.config_entry.options),
errors=errors,
)
class InvalidAuth(HomeAssistantError):
"""Error to indicate there is invalid auth."""

View File

@ -0,0 +1,14 @@
"""Constants for the MJPEG integration."""
import logging
from typing import Final
from homeassistant.const import Platform
DOMAIN: Final = "mjpeg"
PLATFORMS: Final = [Platform.CAMERA]
LOGGER = logging.getLogger(__package__)
CONF_MJPEG_URL: Final = "mjpeg_url"
CONF_STILL_IMAGE_URL: Final = "still_image_url"

View File

@ -3,5 +3,6 @@
"name": "MJPEG IP Camera",
"documentation": "https://www.home-assistant.io/integrations/mjpeg",
"codeowners": [],
"iot_class": "local_push"
"iot_class": "local_push",
"config_flow": true
}

View File

@ -0,0 +1,42 @@
{
"config": {
"step": {
"user": {
"data": {
"mjpeg_url": "MJPEG URL",
"name": "[%key:common::config_flow::data::name%]",
"password": "[%key:common::config_flow::data::password%]",
"still_image_url": "Still Image URL",
"username": "[%key:common::config_flow::data::username%]",
"verify_ssl": "[%key:common::config_flow::data::verify_ssl%]"
}
}
},
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]"
},
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]"
}
},
"options": {
"step": {
"init": {
"data": {
"mjpeg_url": "MJPEG URL",
"name": "[%key:common::config_flow::data::name%]",
"password": "[%key:common::config_flow::data::password%]",
"still_image_url": "Still Image URL",
"username": "[%key:common::config_flow::data::username%]",
"verify_ssl": "[%key:common::config_flow::data::verify_ssl%]"
}
}
},
"error": {
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]",
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]"
}
}
}

View File

@ -0,0 +1,42 @@
{
"config": {
"abort": {
"already_configured": "Device is already configured"
},
"error": {
"cannot_connect": "Failed to connect",
"invalid_auth": "Invalid authentication"
},
"step": {
"user": {
"data": {
"mjpeg_url": "MJPEG URL",
"name": "Name",
"password": "Password",
"still_image_url": "Still Image URL",
"username": "Username",
"verify_ssl": "Verify SSL certificate"
}
}
}
},
"options": {
"error": {
"already_configured": "Device is already configured",
"cannot_connect": "Failed to connect",
"invalid_auth": "Invalid authentication"
},
"step": {
"init": {
"data": {
"mjpeg_url": "MJPEG URL",
"name": "Name",
"password": "Password",
"still_image_url": "Still Image URL",
"username": "Username",
"verify_ssl": "Verify SSL certificate"
}
}
}
}
}

View File

@ -0,0 +1,18 @@
"""Utilities for MJPEG IP Camera."""
import logging
class NoHeaderErrorFilter(logging.Filter):
"""Filter out urllib3 Header Parsing Errors due to a urllib3 bug."""
def filter(self, record: logging.LogRecord) -> bool:
"""Filter out Header Parsing Errors."""
return "Failed to parse headers" not in record.getMessage()
def filter_urllib3_logging() -> None:
"""Filter header errors from urllib3 due to a urllib3 bug."""
urllib3_logger = logging.getLogger("urllib3.connectionpool")
if not any(isinstance(x, NoHeaderErrorFilter) for x in urllib3_logger.filters):
urllib3_logger.addFilter(NoHeaderErrorFilter())

View File

@ -24,7 +24,7 @@ from motioneye_client.const import (
)
import voluptuous as vol
from homeassistant.components.mjpeg.camera import (
from homeassistant.components.mjpeg import (
CONF_MJPEG_URL,
CONF_STILL_IMAGE_URL,
MjpegCamera,

View File

@ -3,7 +3,7 @@ from __future__ import annotations
import logging
from homeassistant.components.mjpeg.camera import MjpegCamera, filter_urllib3_logging
from homeassistant.components.mjpeg import MjpegCamera, filter_urllib3_logging
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType

View File

@ -196,6 +196,7 @@ FLOWS = [
"mikrotik",
"mill",
"minecraft_server",
"mjpeg",
"mobile_app",
"modem_callerid",
"modern_forms",

View File

@ -0,0 +1 @@
"""Tests for the MJPEG IP Camera integration."""

View File

@ -0,0 +1,79 @@
"""Fixtures for MJPEG IP Camera integration tests."""
from __future__ import annotations
from collections.abc import Generator
from unittest.mock import AsyncMock, patch
import pytest
from requests_mock import Mocker
from homeassistant.components.mjpeg.const import (
CONF_MJPEG_URL,
CONF_STILL_IMAGE_URL,
DOMAIN,
)
from homeassistant.const import (
CONF_AUTHENTICATION,
CONF_PASSWORD,
CONF_USERNAME,
CONF_VERIFY_SSL,
HTTP_BASIC_AUTHENTICATION,
)
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry
@pytest.fixture
def mock_config_entry() -> MockConfigEntry:
"""Return the default mocked config entry."""
return MockConfigEntry(
title="My MJPEG Camera",
domain=DOMAIN,
data={},
options={
CONF_AUTHENTICATION: HTTP_BASIC_AUTHENTICATION,
CONF_MJPEG_URL: "https://example.com/mjpeg",
CONF_PASSWORD: "supersecret",
CONF_STILL_IMAGE_URL: "http://example.com/still",
CONF_USERNAME: "frenck",
CONF_VERIFY_SSL: True,
},
)
@pytest.fixture
def mock_setup_entry() -> Generator[AsyncMock, None, None]:
"""Mock setting up a config entry."""
with patch(
"homeassistant.components.mjpeg.async_setup_entry", return_value=True
) as mock_setup:
yield mock_setup
@pytest.fixture
def mock_reload_entry() -> Generator[AsyncMock, None, None]:
"""Mock setting up a config entry."""
with patch("homeassistant.components.mjpeg.async_reload_entry") as mock_reload:
yield mock_reload
@pytest.fixture
def mock_mjpeg_requests(requests_mock: Mocker) -> Generator[Mocker, None, None]:
"""Fixture to provide a requests mocker."""
requests_mock.get("https://example.com/mjpeg", text="resp")
requests_mock.get("https://example.com/still", text="resp")
yield requests_mock
@pytest.fixture
async def init_integration(
hass: HomeAssistant, mock_config_entry: MockConfigEntry, mock_mjpeg_requests: Mocker
) -> MockConfigEntry:
"""Set up the MJPEG IP Camera integration for testing."""
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
return mock_config_entry

View File

@ -0,0 +1,441 @@
"""Tests for the MJPEG IP Camera config flow."""
from unittest.mock import AsyncMock
import requests
from requests_mock import Mocker
from homeassistant.components.mjpeg.const import (
CONF_MJPEG_URL,
CONF_STILL_IMAGE_URL,
DOMAIN,
)
from homeassistant.config_entries import SOURCE_IMPORT, SOURCE_USER
from homeassistant.const import (
CONF_AUTHENTICATION,
CONF_NAME,
CONF_PASSWORD,
CONF_USERNAME,
CONF_VERIFY_SSL,
HTTP_BASIC_AUTHENTICATION,
HTTP_DIGEST_AUTHENTICATION,
)
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import (
RESULT_TYPE_ABORT,
RESULT_TYPE_CREATE_ENTRY,
RESULT_TYPE_FORM,
)
from tests.common import MockConfigEntry
async def test_full_user_flow(
hass: HomeAssistant,
mock_mjpeg_requests: Mocker,
mock_setup_entry: AsyncMock,
) -> None:
"""Test the full user configuration flow."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result.get("type") == RESULT_TYPE_FORM
assert result.get("step_id") == SOURCE_USER
assert "flow_id" in result
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
CONF_NAME: "Spy cam",
CONF_MJPEG_URL: "https://example.com/mjpeg",
CONF_STILL_IMAGE_URL: "https://example.com/still",
CONF_USERNAME: "frenck",
CONF_PASSWORD: "omgpuppies",
CONF_VERIFY_SSL: False,
},
)
assert result2.get("type") == RESULT_TYPE_CREATE_ENTRY
assert result2.get("title") == "Spy cam"
assert result2.get("data") == {}
assert result2.get("options") == {
CONF_AUTHENTICATION: HTTP_BASIC_AUTHENTICATION,
CONF_MJPEG_URL: "https://example.com/mjpeg",
CONF_PASSWORD: "omgpuppies",
CONF_STILL_IMAGE_URL: "https://example.com/still",
CONF_USERNAME: "frenck",
CONF_VERIFY_SSL: False,
}
assert len(mock_setup_entry.mock_calls) == 1
assert mock_mjpeg_requests.call_count == 2
async def test_full_flow_with_authentication_error(
hass: HomeAssistant,
mock_mjpeg_requests: Mocker,
mock_setup_entry: AsyncMock,
) -> None:
"""Test the full user configuration flow with invalid credentials.
This tests tests a full config flow, with a case the user enters an invalid
credentials, but recovers by entering the correct ones.
"""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result.get("type") == RESULT_TYPE_FORM
assert result.get("step_id") == SOURCE_USER
assert "flow_id" in result
mock_mjpeg_requests.get(
"https://example.com/mjpeg", text="Access Denied!", status_code=401
)
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
CONF_NAME: "Sky cam",
CONF_MJPEG_URL: "https://example.com/mjpeg",
CONF_PASSWORD: "omgpuppies",
CONF_USERNAME: "frenck",
},
)
assert result2.get("type") == RESULT_TYPE_FORM
assert result2.get("step_id") == SOURCE_USER
assert result2.get("errors") == {"username": "invalid_auth"}
assert "flow_id" in result2
assert len(mock_setup_entry.mock_calls) == 0
assert mock_mjpeg_requests.call_count == 2
mock_mjpeg_requests.get("https://example.com/mjpeg", text="resp")
result3 = await hass.config_entries.flow.async_configure(
result2["flow_id"],
user_input={
CONF_NAME: "Sky cam",
CONF_MJPEG_URL: "https://example.com/mjpeg",
CONF_PASSWORD: "supersecret",
CONF_USERNAME: "frenck",
},
)
assert result3.get("type") == RESULT_TYPE_CREATE_ENTRY
assert result3.get("title") == "Sky cam"
assert result3.get("data") == {}
assert result3.get("options") == {
CONF_AUTHENTICATION: HTTP_BASIC_AUTHENTICATION,
CONF_MJPEG_URL: "https://example.com/mjpeg",
CONF_PASSWORD: "supersecret",
CONF_STILL_IMAGE_URL: None,
CONF_USERNAME: "frenck",
CONF_VERIFY_SSL: True,
}
assert len(mock_setup_entry.mock_calls) == 1
assert mock_mjpeg_requests.call_count == 3
async def test_connection_error(
hass: HomeAssistant,
mock_mjpeg_requests: Mocker,
mock_setup_entry: AsyncMock,
) -> None:
"""Test connection error."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result.get("type") == RESULT_TYPE_FORM
assert result.get("step_id") == SOURCE_USER
assert "flow_id" in result
# Test connectione error on MJPEG url
mock_mjpeg_requests.get(
"https://example.com/mjpeg", exc=requests.exceptions.ConnectionError
)
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
CONF_NAME: "My cam",
CONF_MJPEG_URL: "https://example.com/mjpeg",
CONF_STILL_IMAGE_URL: "https://example.com/still",
},
)
assert result2.get("type") == RESULT_TYPE_FORM
assert result2.get("step_id") == SOURCE_USER
assert result2.get("errors") == {"mjpeg_url": "cannot_connect"}
assert "flow_id" in result2
assert len(mock_setup_entry.mock_calls) == 0
assert mock_mjpeg_requests.call_count == 1
# Reset
mock_mjpeg_requests.get("https://example.com/mjpeg", text="resp")
# Test connectione error on still url
mock_mjpeg_requests.get(
"https://example.com/still", exc=requests.exceptions.ConnectionError
)
result3 = await hass.config_entries.flow.async_configure(
result2["flow_id"],
user_input={
CONF_NAME: "My cam",
CONF_MJPEG_URL: "https://example.com/mjpeg",
CONF_STILL_IMAGE_URL: "https://example.com/still",
},
)
assert result3.get("type") == RESULT_TYPE_FORM
assert result3.get("step_id") == SOURCE_USER
assert result3.get("errors") == {"still_image_url": "cannot_connect"}
assert "flow_id" in result3
assert len(mock_setup_entry.mock_calls) == 0
assert mock_mjpeg_requests.call_count == 3
# Reset
mock_mjpeg_requests.get("https://example.com/still", text="resp")
# Finish
result4 = await hass.config_entries.flow.async_configure(
result3["flow_id"],
user_input={
CONF_NAME: "My cam",
CONF_MJPEG_URL: "https://example.com/mjpeg",
CONF_STILL_IMAGE_URL: "https://example.com/still",
},
)
assert result4.get("type") == RESULT_TYPE_CREATE_ENTRY
assert result4.get("title") == "My cam"
assert result4.get("data") == {}
assert result4.get("options") == {
CONF_AUTHENTICATION: HTTP_BASIC_AUTHENTICATION,
CONF_MJPEG_URL: "https://example.com/mjpeg",
CONF_PASSWORD: "",
CONF_STILL_IMAGE_URL: "https://example.com/still",
CONF_USERNAME: None,
CONF_VERIFY_SSL: True,
}
assert len(mock_setup_entry.mock_calls) == 1
assert mock_mjpeg_requests.call_count == 5
async def test_already_configured(
hass: HomeAssistant,
mock_mjpeg_requests: Mocker,
mock_config_entry: MockConfigEntry,
mock_setup_entry: AsyncMock,
) -> None:
"""Test we abort if the MJPEG IP Camera is already configured."""
mock_config_entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert "flow_id" in result
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
CONF_NAME: "My cam",
CONF_MJPEG_URL: "https://example.com/mjpeg",
},
)
assert result2.get("type") == RESULT_TYPE_ABORT
assert result2.get("reason") == "already_configured"
async def test_import_flow(
hass: HomeAssistant,
mock_mjpeg_requests: Mocker,
mock_setup_entry: AsyncMock,
) -> None:
"""Test the import configuration flow."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_IMPORT},
data={
CONF_AUTHENTICATION: HTTP_DIGEST_AUTHENTICATION,
CONF_MJPEG_URL: "http://example.com/mjpeg",
CONF_NAME: "Imported Camera",
CONF_PASSWORD: "omgpuppies",
CONF_STILL_IMAGE_URL: "http://example.com/still",
CONF_USERNAME: "frenck",
CONF_VERIFY_SSL: False,
},
)
assert result.get("type") == RESULT_TYPE_CREATE_ENTRY
assert result.get("title") == "Imported Camera"
assert result.get("data") == {}
assert result.get("options") == {
CONF_AUTHENTICATION: HTTP_DIGEST_AUTHENTICATION,
CONF_MJPEG_URL: "http://example.com/mjpeg",
CONF_PASSWORD: "omgpuppies",
CONF_STILL_IMAGE_URL: "http://example.com/still",
CONF_USERNAME: "frenck",
CONF_VERIFY_SSL: False,
}
assert len(mock_setup_entry.mock_calls) == 1
assert mock_mjpeg_requests.call_count == 0
async def test_import_flow_already_configured(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_setup_entry: AsyncMock,
) -> None:
"""Test the import configuration flow for an already configured entry."""
mock_config_entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_IMPORT},
data={
CONF_AUTHENTICATION: HTTP_DIGEST_AUTHENTICATION,
CONF_MJPEG_URL: "https://example.com/mjpeg",
CONF_NAME: "Imported Camera",
CONF_PASSWORD: "omgpuppies",
CONF_STILL_IMAGE_URL: "https://example.com/still",
CONF_USERNAME: "frenck",
CONF_VERIFY_SSL: False,
},
)
assert result.get("type") == RESULT_TYPE_ABORT
assert result.get("reason") == "already_configured"
assert len(mock_setup_entry.mock_calls) == 0
async def test_options_flow(
hass: HomeAssistant,
mock_mjpeg_requests: Mocker,
init_integration: MockConfigEntry,
) -> None:
"""Test options config flow."""
result = await hass.config_entries.options.async_init(init_integration.entry_id)
assert result.get("type") == RESULT_TYPE_FORM
assert result.get("step_id") == "init"
assert "flow_id" in result
# Register a second camera
mock_mjpeg_requests.get("https://example.com/second_camera", text="resp")
mock_second_config_entry = MockConfigEntry(
title="Another Camera",
domain=DOMAIN,
data={},
options={
CONF_AUTHENTICATION: HTTP_BASIC_AUTHENTICATION,
CONF_MJPEG_URL: "https://example.com/second_camera",
CONF_PASSWORD: "",
CONF_STILL_IMAGE_URL: None,
CONF_USERNAME: None,
CONF_VERIFY_SSL: True,
},
)
mock_second_config_entry.add_to_hass(hass)
# Try updating options to already existing secondary camera
result2 = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={
CONF_MJPEG_URL: "https://example.com/second_camera",
},
)
assert result2.get("type") == RESULT_TYPE_FORM
assert result2.get("step_id") == "init"
assert result2.get("errors") == {"mjpeg_url": "already_configured"}
assert "flow_id" in result2
assert mock_mjpeg_requests.call_count == 1
# Test connectione error on MJPEG url
mock_mjpeg_requests.get(
"https://example.com/invalid_mjpeg", exc=requests.exceptions.ConnectionError
)
result3 = await hass.config_entries.options.async_configure(
result2["flow_id"],
user_input={
CONF_MJPEG_URL: "https://example.com/invalid_mjpeg",
CONF_STILL_IMAGE_URL: "https://example.com/still",
},
)
assert result3.get("type") == RESULT_TYPE_FORM
assert result3.get("step_id") == "init"
assert result3.get("errors") == {"mjpeg_url": "cannot_connect"}
assert "flow_id" in result3
assert mock_mjpeg_requests.call_count == 2
# Test connectione error on still url
mock_mjpeg_requests.get(
"https://example.com/invalid_still", exc=requests.exceptions.ConnectionError
)
result4 = await hass.config_entries.options.async_configure(
result3["flow_id"],
user_input={
CONF_MJPEG_URL: "https://example.com/mjpeg",
CONF_STILL_IMAGE_URL: "https://example.com/invalid_still",
},
)
assert result4.get("type") == RESULT_TYPE_FORM
assert result4.get("step_id") == "init"
assert result4.get("errors") == {"still_image_url": "cannot_connect"}
assert "flow_id" in result4
assert mock_mjpeg_requests.call_count == 4
# Invalid credentials
mock_mjpeg_requests.get(
"https://example.com/invalid_auth", text="Access Denied!", status_code=401
)
result5 = await hass.config_entries.options.async_configure(
result4["flow_id"],
user_input={
CONF_MJPEG_URL: "https://example.com/invalid_auth",
CONF_PASSWORD: "omgpuppies",
CONF_USERNAME: "frenck",
},
)
assert result5.get("type") == RESULT_TYPE_FORM
assert result5.get("step_id") == "init"
assert result5.get("errors") == {"username": "invalid_auth"}
assert "flow_id" in result5
assert mock_mjpeg_requests.call_count == 6
# Finish
result6 = await hass.config_entries.options.async_configure(
result5["flow_id"],
user_input={
CONF_MJPEG_URL: "https://example.com/mjpeg",
CONF_PASSWORD: "evenmorepuppies",
CONF_USERNAME: "newuser",
},
)
assert result6.get("type") == RESULT_TYPE_CREATE_ENTRY
assert result6.get("data") == {
CONF_AUTHENTICATION: HTTP_BASIC_AUTHENTICATION,
CONF_MJPEG_URL: "https://example.com/mjpeg",
CONF_PASSWORD: "evenmorepuppies",
CONF_STILL_IMAGE_URL: None,
CONF_USERNAME: "newuser",
CONF_VERIFY_SSL: True,
}
assert mock_mjpeg_requests.call_count == 7

View File

@ -0,0 +1,99 @@
"""Tests for the MJPEG IP Camera integration."""
from unittest.mock import AsyncMock, MagicMock
import pytest
from homeassistant.components.camera import DOMAIN as CAMERA_DOMAIN
from homeassistant.components.mjpeg.const import (
CONF_MJPEG_URL,
CONF_STILL_IMAGE_URL,
DOMAIN,
)
from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import (
CONF_AUTHENTICATION,
CONF_NAME,
CONF_PASSWORD,
CONF_USERNAME,
CONF_VERIFY_SSL,
HTTP_BASIC_AUTHENTICATION,
)
from homeassistant.core import HomeAssistant
from homeassistant.setup import async_setup_component
from tests.common import MockConfigEntry
async def test_load_unload_config_entry(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_mjpeg_requests: MagicMock,
) -> None:
"""Test the MJPEG IP Camera configuration entry loading/unloading."""
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert mock_config_entry.state is ConfigEntryState.LOADED
await hass.config_entries.async_unload(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert not hass.data.get(DOMAIN)
assert mock_config_entry.state is ConfigEntryState.NOT_LOADED
async def test_reload_config_entry(
hass: HomeAssistant,
mock_reload_entry: AsyncMock,
init_integration: MockConfigEntry,
) -> None:
"""Test the MJPEG IP Camera configuration entry is reloaded on change."""
assert len(mock_reload_entry.mock_calls) == 0
hass.config_entries.async_update_entry(
init_integration, options={"something": "else"}
)
assert len(mock_reload_entry.mock_calls) == 1
async def test_import_config(
hass: HomeAssistant,
mock_mjpeg_requests: MagicMock,
mock_setup_entry: AsyncMock,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test MJPEG IP Camera being set up from config via import."""
assert await async_setup_component(
hass,
CAMERA_DOMAIN,
{
CAMERA_DOMAIN: {
"platform": DOMAIN,
CONF_MJPEG_URL: "http://example.com/mjpeg",
CONF_NAME: "Random Camera",
CONF_PASSWORD: "supersecret",
CONF_STILL_IMAGE_URL: "http://example.com/still",
CONF_USERNAME: "frenck",
CONF_VERIFY_SSL: False,
}
},
)
await hass.async_block_till_done()
config_entries = hass.config_entries.async_entries(DOMAIN)
assert len(config_entries) == 1
assert "the MJPEG IP Camera platform in YAML is deprecated" in caplog.text
entry = config_entries[0]
assert entry.title == "Random Camera"
assert entry.unique_id is None
assert entry.data == {}
assert entry.options == {
CONF_AUTHENTICATION: HTTP_BASIC_AUTHENTICATION,
CONF_MJPEG_URL: "http://example.com/mjpeg",
CONF_PASSWORD: "supersecret",
CONF_STILL_IMAGE_URL: "http://example.com/still",
CONF_USERNAME: "frenck",
CONF_VERIFY_SSL: False,
}