From bbe58091a8614f0688688c11bbf2cf3d80f670e9 Mon Sep 17 00:00:00 2001 From: Dermot Duffy Date: Fri, 23 Apr 2021 23:00:28 -0700 Subject: [PATCH] Create a motionEye integration (#48239) --- CODEOWNERS | 1 + .../components/motioneye/__init__.py | 258 ++++++++++++++ homeassistant/components/motioneye/camera.py | 208 ++++++++++++ .../components/motioneye/config_flow.py | 127 +++++++ homeassistant/components/motioneye/const.py | 20 ++ .../components/motioneye/manifest.json | 13 + .../components/motioneye/strings.json | 25 ++ homeassistant/generated/config_flows.py | 1 + requirements_all.txt | 3 + requirements_test_all.txt | 3 + tests/components/motioneye/__init__.py | 180 ++++++++++ tests/components/motioneye/test_camera.py | 315 ++++++++++++++++++ .../components/motioneye/test_config_flow.py | 233 +++++++++++++ 13 files changed, 1387 insertions(+) create mode 100644 homeassistant/components/motioneye/__init__.py create mode 100644 homeassistant/components/motioneye/camera.py create mode 100644 homeassistant/components/motioneye/config_flow.py create mode 100644 homeassistant/components/motioneye/const.py create mode 100644 homeassistant/components/motioneye/manifest.json create mode 100644 homeassistant/components/motioneye/strings.json create mode 100644 tests/components/motioneye/__init__.py create mode 100644 tests/components/motioneye/test_camera.py create mode 100644 tests/components/motioneye/test_config_flow.py diff --git a/CODEOWNERS b/CODEOWNERS index 6d044f4d06b..d6226c08a5d 100644 --- a/CODEOWNERS +++ b/CODEOWNERS @@ -294,6 +294,7 @@ homeassistant/components/modbus/* @adamchengtkc @janiversen @vzahradnik homeassistant/components/monoprice/* @etsinko @OnFreund homeassistant/components/moon/* @fabaff homeassistant/components/motion_blinds/* @starkillerOG +homeassistant/components/motioneye/* @dermotduffy homeassistant/components/mpd/* @fabaff homeassistant/components/mqtt/* @emontnemery homeassistant/components/msteams/* @peroyvind diff --git a/homeassistant/components/motioneye/__init__.py b/homeassistant/components/motioneye/__init__.py new file mode 100644 index 00000000000..61e7a7d12f3 --- /dev/null +++ b/homeassistant/components/motioneye/__init__.py @@ -0,0 +1,258 @@ +"""The motionEye integration.""" +from __future__ import annotations + +import asyncio +import logging +from typing import Any, Callable + +from motioneye_client.client import ( + MotionEyeClient, + MotionEyeClientError, + MotionEyeClientInvalidAuthError, +) +from motioneye_client.const import KEY_CAMERAS, KEY_ID, KEY_NAME + +from homeassistant.components.camera.const import DOMAIN as CAMERA_DOMAIN +from homeassistant.config_entries import SOURCE_REAUTH, ConfigEntry +from homeassistant.const import CONF_SOURCE, CONF_URL +from homeassistant.core import HomeAssistant, callback +from homeassistant.exceptions import ConfigEntryNotReady +from homeassistant.helpers import device_registry as dr +from homeassistant.helpers.dispatcher import ( + async_dispatcher_connect, + async_dispatcher_send, +) +from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed + +from .const import ( + CONF_ADMIN_PASSWORD, + CONF_ADMIN_USERNAME, + CONF_CLIENT, + CONF_CONFIG_ENTRY, + CONF_COORDINATOR, + CONF_SURVEILLANCE_PASSWORD, + CONF_SURVEILLANCE_USERNAME, + DEFAULT_SCAN_INTERVAL, + DOMAIN, + MOTIONEYE_MANUFACTURER, + SIGNAL_CAMERA_ADD, +) + +_LOGGER = logging.getLogger(__name__) + +PLATFORMS = [CAMERA_DOMAIN] + + +def create_motioneye_client( + *args: Any, + **kwargs: Any, +) -> MotionEyeClient: + """Create a MotionEyeClient.""" + return MotionEyeClient(*args, **kwargs) + + +def get_motioneye_device_identifier( + config_entry_id: str, camera_id: int +) -> tuple[str, str, int]: + """Get the identifiers for a motionEye device.""" + return (DOMAIN, config_entry_id, camera_id) + + +def get_motioneye_entity_unique_id( + config_entry_id: str, camera_id: int, entity_type: str +) -> str: + """Get the unique_id for a motionEye entity.""" + return f"{config_entry_id}_{camera_id}_{entity_type}" + + +def get_camera_from_cameras( + camera_id: int, data: dict[str, Any] +) -> dict[str, Any] | None: + """Get an individual camera dict from a multiple cameras data response.""" + for camera in data.get(KEY_CAMERAS) or []: + if camera.get(KEY_ID) == camera_id: + val: dict[str, Any] = camera + return val + return None + + +def is_acceptable_camera(camera: dict[str, Any] | None) -> bool: + """Determine if a camera dict is acceptable.""" + return bool(camera and KEY_ID in camera and KEY_NAME in camera) + + +@callback +def listen_for_new_cameras( + hass: HomeAssistant, + entry: ConfigEntry, + add_func: Callable, +) -> None: + """Listen for new cameras.""" + + entry.async_on_unload( + async_dispatcher_connect( + hass, + SIGNAL_CAMERA_ADD.format(entry.entry_id), + add_func, + ) + ) + + +async def _create_reauth_flow( + hass: HomeAssistant, + config_entry: ConfigEntry, +) -> None: + hass.async_create_task( + hass.config_entries.flow.async_init( + DOMAIN, + context={ + CONF_SOURCE: SOURCE_REAUTH, + CONF_CONFIG_ENTRY: config_entry, + }, + data=config_entry.data, + ) + ) + + +@callback +def _add_camera( + hass: HomeAssistant, + device_registry: dr.DeviceRegistry, + client: MotionEyeClient, + entry: ConfigEntry, + camera_id: int, + camera: dict[str, Any], + device_identifier: tuple[str, str, int], +) -> None: + """Add a motionEye camera to hass.""" + + device_registry.async_get_or_create( + config_entry_id=entry.entry_id, + identifiers={device_identifier}, + manufacturer=MOTIONEYE_MANUFACTURER, + model=MOTIONEYE_MANUFACTURER, + name=camera[KEY_NAME], + ) + + async_dispatcher_send( + hass, + SIGNAL_CAMERA_ADD.format(entry.entry_id), + camera, + ) + + +async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: + """Set up motionEye from a config entry.""" + hass.data.setdefault(DOMAIN, {}) + + client = create_motioneye_client( + entry.data[CONF_URL], + admin_username=entry.data.get(CONF_ADMIN_USERNAME), + admin_password=entry.data.get(CONF_ADMIN_PASSWORD), + surveillance_username=entry.data.get(CONF_SURVEILLANCE_USERNAME), + surveillance_password=entry.data.get(CONF_SURVEILLANCE_PASSWORD), + ) + + try: + await client.async_client_login() + except MotionEyeClientInvalidAuthError: + await client.async_client_close() + await _create_reauth_flow(hass, entry) + return False + except MotionEyeClientError as exc: + await client.async_client_close() + raise ConfigEntryNotReady from exc + + @callback + async def async_update_data() -> dict[str, Any] | None: + try: + return await client.async_get_cameras() + except MotionEyeClientError as exc: + raise UpdateFailed("Error communicating with API") from exc + + coordinator = DataUpdateCoordinator( + hass, + _LOGGER, + name=DOMAIN, + update_method=async_update_data, + update_interval=DEFAULT_SCAN_INTERVAL, + ) + hass.data[DOMAIN][entry.entry_id] = { + CONF_CLIENT: client, + CONF_COORDINATOR: coordinator, + } + + current_cameras: set[tuple[str, str, int]] = set() + device_registry = await dr.async_get_registry(hass) + + @callback + def _async_process_motioneye_cameras() -> None: + """Process motionEye camera additions and removals.""" + inbound_camera: set[tuple[str, str, int]] = set() + if KEY_CAMERAS not in coordinator.data: + return + + for camera in coordinator.data[KEY_CAMERAS]: + if not is_acceptable_camera(camera): + return + camera_id = camera[KEY_ID] + device_identifier = get_motioneye_device_identifier( + entry.entry_id, camera_id + ) + inbound_camera.add(device_identifier) + + if device_identifier in current_cameras: + continue + current_cameras.add(device_identifier) + _add_camera( + hass, + device_registry, + client, + entry, + camera_id, + camera, + device_identifier, + ) + + # Ensure every device associated with this config entry is still in the list of + # motionEye cameras, otherwise remove the device (and thus entities). + for device_entry in dr.async_entries_for_config_entry( + device_registry, entry.entry_id + ): + for identifier in device_entry.identifiers: + if identifier in inbound_camera: + break + else: + device_registry.async_remove_device(device_entry.id) + + async def setup_then_listen() -> None: + await asyncio.gather( + *[ + hass.config_entries.async_forward_entry_setup(entry, platform) + for platform in PLATFORMS + ] + ) + entry.async_on_unload( + coordinator.async_add_listener(_async_process_motioneye_cameras) + ) + await coordinator.async_refresh() + + hass.async_create_task(setup_then_listen()) + return True + + +async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: + """Unload a config entry.""" + unload_ok = all( + await asyncio.gather( + *[ + hass.config_entries.async_forward_entry_unload(entry, platform) + for platform in PLATFORMS + ] + ) + ) + if unload_ok: + config_data = hass.data[DOMAIN].pop(entry.entry_id) + await config_data[CONF_CLIENT].async_client_close() + + return unload_ok diff --git a/homeassistant/components/motioneye/camera.py b/homeassistant/components/motioneye/camera.py new file mode 100644 index 00000000000..58df22198bf --- /dev/null +++ b/homeassistant/components/motioneye/camera.py @@ -0,0 +1,208 @@ +"""The motionEye integration.""" +from __future__ import annotations + +import logging +from typing import Any, Callable + +import aiohttp +from motioneye_client.client import MotionEyeClient +from motioneye_client.const import ( + DEFAULT_SURVEILLANCE_USERNAME, + KEY_ID, + KEY_MOTION_DETECTION, + KEY_NAME, + KEY_STREAMING_AUTH_MODE, +) + +from homeassistant.components.mjpeg.camera import ( + CONF_MJPEG_URL, + CONF_STILL_IMAGE_URL, + CONF_VERIFY_SSL, + MjpegCamera, +) +from homeassistant.config_entries import ConfigEntry +from homeassistant.const import ( + CONF_AUTHENTICATION, + CONF_NAME, + CONF_PASSWORD, + CONF_USERNAME, + HTTP_BASIC_AUTHENTICATION, + HTTP_DIGEST_AUTHENTICATION, +) +from homeassistant.core import HomeAssistant, callback +from homeassistant.helpers.update_coordinator import ( + CoordinatorEntity, + DataUpdateCoordinator, +) + +from . import ( + get_camera_from_cameras, + get_motioneye_device_identifier, + get_motioneye_entity_unique_id, + is_acceptable_camera, + listen_for_new_cameras, +) +from .const import ( + CONF_CLIENT, + CONF_COORDINATOR, + CONF_SURVEILLANCE_PASSWORD, + CONF_SURVEILLANCE_USERNAME, + DOMAIN, + MOTIONEYE_MANUFACTURER, + TYPE_MOTIONEYE_MJPEG_CAMERA, +) + +_LOGGER = logging.getLogger(__name__) + +PLATFORMS = ["camera"] + + +async def async_setup_entry( + hass: HomeAssistant, entry: ConfigEntry, async_add_entities: Callable +) -> bool: + """Set up motionEye from a config entry.""" + entry_data = hass.data[DOMAIN][entry.entry_id] + + @callback + def camera_add(camera: dict[str, Any]) -> None: + """Add a new motionEye camera.""" + async_add_entities( + [ + MotionEyeMjpegCamera( + entry.entry_id, + entry.data.get( + CONF_SURVEILLANCE_USERNAME, DEFAULT_SURVEILLANCE_USERNAME + ), + entry.data.get(CONF_SURVEILLANCE_PASSWORD, ""), + camera, + entry_data[CONF_CLIENT], + entry_data[CONF_COORDINATOR], + ) + ] + ) + + listen_for_new_cameras(hass, entry, camera_add) + return True + + +class MotionEyeMjpegCamera(MjpegCamera, CoordinatorEntity): + """motionEye mjpeg camera.""" + + def __init__( + self, + config_entry_id: str, + username: str, + password: str, + camera: dict[str, Any], + client: MotionEyeClient, + coordinator: DataUpdateCoordinator, + ): + """Initialize a MJPEG camera.""" + self._surveillance_username = username + self._surveillance_password = password + self._client = client + self._camera_id = camera[KEY_ID] + self._device_identifier = get_motioneye_device_identifier( + config_entry_id, self._camera_id + ) + self._unique_id = get_motioneye_entity_unique_id( + config_entry_id, self._camera_id, TYPE_MOTIONEYE_MJPEG_CAMERA + ) + self._motion_detection_enabled: bool = camera.get(KEY_MOTION_DETECTION, False) + self._available = MotionEyeMjpegCamera._is_acceptable_streaming_camera(camera) + + # motionEye cameras are always streaming or unavailable. + self.is_streaming = True + + MjpegCamera.__init__( + self, + { + CONF_VERIFY_SSL: False, + **self._get_mjpeg_camera_properties_for_camera(camera), + }, + ) + CoordinatorEntity.__init__(self, coordinator) + + @callback + def _get_mjpeg_camera_properties_for_camera( + self, camera: dict[str, Any] + ) -> dict[str, Any]: + """Convert a motionEye camera to MjpegCamera internal properties.""" + auth = None + if camera.get(KEY_STREAMING_AUTH_MODE) in [ + HTTP_BASIC_AUTHENTICATION, + HTTP_DIGEST_AUTHENTICATION, + ]: + auth = camera[KEY_STREAMING_AUTH_MODE] + + return { + CONF_NAME: camera[KEY_NAME], + CONF_USERNAME: self._surveillance_username if auth is not None else None, + CONF_PASSWORD: self._surveillance_password if auth is not None else None, + CONF_MJPEG_URL: self._client.get_camera_stream_url(camera) or "", + CONF_STILL_IMAGE_URL: self._client.get_camera_snapshot_url(camera), + CONF_AUTHENTICATION: auth, + } + + @callback + def _set_mjpeg_camera_state_for_camera(self, camera: dict[str, Any]) -> None: + """Set the internal state to match the given camera.""" + + # Sets the state of the underlying (inherited) MjpegCamera based on the updated + # MotionEye camera dictionary. + properties = self._get_mjpeg_camera_properties_for_camera(camera) + self._name = properties[CONF_NAME] + self._username = properties[CONF_USERNAME] + self._password = properties[CONF_PASSWORD] + self._mjpeg_url = properties[CONF_MJPEG_URL] + self._still_image_url = properties[CONF_STILL_IMAGE_URL] + self._authentication = properties[CONF_AUTHENTICATION] + + if self._authentication == HTTP_BASIC_AUTHENTICATION: + self._auth = aiohttp.BasicAuth(self._username, password=self._password) + + @property + def unique_id(self) -> str: + """Return a unique id for this instance.""" + return self._unique_id + + @classmethod + def _is_acceptable_streaming_camera(cls, camera: dict[str, Any] | None) -> bool: + """Determine if a camera is streaming/usable.""" + return is_acceptable_camera(camera) and MotionEyeClient.is_camera_streaming( + camera + ) + + @property + def available(self) -> bool: + """Return if entity is available.""" + return self._available + + @callback + def _handle_coordinator_update(self) -> None: + """Handle updated data from the coordinator.""" + available = False + if self.coordinator.last_update_success: + camera = get_camera_from_cameras(self._camera_id, self.coordinator.data) + if MotionEyeMjpegCamera._is_acceptable_streaming_camera(camera): + assert camera + self._set_mjpeg_camera_state_for_camera(camera) + self._motion_detection_enabled = camera.get(KEY_MOTION_DETECTION, False) + available = True + self._available = available + CoordinatorEntity._handle_coordinator_update(self) + + @property + def brand(self) -> str: + """Return the camera brand.""" + return MOTIONEYE_MANUFACTURER + + @property + def motion_detection_enabled(self) -> bool: + """Return the camera motion detection status.""" + return self._motion_detection_enabled + + @property + def device_info(self) -> dict[str, Any]: + """Return the device information.""" + return {"identifiers": {self._device_identifier}} diff --git a/homeassistant/components/motioneye/config_flow.py b/homeassistant/components/motioneye/config_flow.py new file mode 100644 index 00000000000..45da759e91b --- /dev/null +++ b/homeassistant/components/motioneye/config_flow.py @@ -0,0 +1,127 @@ +"""Config flow for motionEye integration.""" +from __future__ import annotations + +import logging +from typing import Any + +from motioneye_client.client import ( + MotionEyeClientConnectionError, + MotionEyeClientInvalidAuthError, + MotionEyeClientRequestError, +) +import voluptuous as vol + +from homeassistant.config_entries import ( + CONN_CLASS_LOCAL_POLL, + SOURCE_REAUTH, + ConfigFlow, +) +from homeassistant.const import CONF_SOURCE, CONF_URL +from homeassistant.helpers import config_validation as cv +from homeassistant.helpers.typing import ConfigType + +from . import create_motioneye_client +from .const import ( # pylint:disable=unused-import + CONF_ADMIN_PASSWORD, + CONF_ADMIN_USERNAME, + CONF_CONFIG_ENTRY, + CONF_SURVEILLANCE_PASSWORD, + CONF_SURVEILLANCE_USERNAME, + DOMAIN, +) + +_LOGGER = logging.getLogger(__name__) + + +class MotionEyeConfigFlow(ConfigFlow, domain=DOMAIN): + """Handle a config flow for motionEye.""" + + VERSION = 1 + CONNECTION_CLASS = CONN_CLASS_LOCAL_POLL + + async def async_step_user( + self, user_input: ConfigType | None = None + ) -> dict[str, Any]: + """Handle the initial step.""" + out: dict[str, Any] = {} + errors = {} + if user_input is None: + entry = self.context.get(CONF_CONFIG_ENTRY) + user_input = entry.data if entry else {} + else: + try: + # Cannot use cv.url validation in the schema itself, so + # apply extra validation here. + cv.url(user_input[CONF_URL]) + except vol.Invalid: + errors["base"] = "invalid_url" + else: + client = create_motioneye_client( + user_input[CONF_URL], + admin_username=user_input.get(CONF_ADMIN_USERNAME), + admin_password=user_input.get(CONF_ADMIN_PASSWORD), + surveillance_username=user_input.get(CONF_SURVEILLANCE_USERNAME), + surveillance_password=user_input.get(CONF_SURVEILLANCE_PASSWORD), + ) + + try: + await client.async_client_login() + except MotionEyeClientConnectionError: + errors["base"] = "cannot_connect" + except MotionEyeClientInvalidAuthError: + errors["base"] = "invalid_auth" + except MotionEyeClientRequestError: + errors["base"] = "unknown" + else: + entry = self.context.get(CONF_CONFIG_ENTRY) + if ( + self.context.get(CONF_SOURCE) == SOURCE_REAUTH + and entry is not None + ): + self.hass.config_entries.async_update_entry( + entry, data=user_input + ) + # Need to manually reload, as the listener won't have been + # installed because the initial load did not succeed (the reauth + # flow will not be initiated if the load succeeds). + await self.hass.config_entries.async_reload(entry.entry_id) + out = self.async_abort(reason="reauth_successful") + return out + + out = self.async_create_entry( + title=f"{user_input[CONF_URL]}", + data=user_input, + ) + return out + + out = self.async_show_form( + step_id="user", + data_schema=vol.Schema( + { + vol.Required(CONF_URL, default=user_input.get(CONF_URL, "")): str, + vol.Optional( + CONF_ADMIN_USERNAME, default=user_input.get(CONF_ADMIN_USERNAME) + ): str, + vol.Optional( + CONF_ADMIN_PASSWORD, default=user_input.get(CONF_ADMIN_PASSWORD) + ): str, + vol.Optional( + CONF_SURVEILLANCE_USERNAME, + default=user_input.get(CONF_SURVEILLANCE_USERNAME), + ): str, + vol.Optional( + CONF_SURVEILLANCE_PASSWORD, + default=user_input.get(CONF_SURVEILLANCE_PASSWORD), + ): str, + } + ), + errors=errors, + ) + return out + + async def async_step_reauth( + self, + config_data: ConfigType | None = None, + ) -> dict[str, Any]: + """Handle a reauthentication flow.""" + return await self.async_step_user(config_data) diff --git a/homeassistant/components/motioneye/const.py b/homeassistant/components/motioneye/const.py new file mode 100644 index 00000000000..a76053b2854 --- /dev/null +++ b/homeassistant/components/motioneye/const.py @@ -0,0 +1,20 @@ +"""Constants for the motionEye integration.""" +from datetime import timedelta + +DOMAIN = "motioneye" + +CONF_CONFIG_ENTRY = "config_entry" +CONF_CLIENT = "client" +CONF_COORDINATOR = "coordinator" +CONF_ADMIN_PASSWORD = "admin_password" +CONF_ADMIN_USERNAME = "admin_username" +CONF_SURVEILLANCE_USERNAME = "surveillance_username" +CONF_SURVEILLANCE_PASSWORD = "surveillance_password" +DEFAULT_SCAN_INTERVAL = timedelta(seconds=30) + +MOTIONEYE_MANUFACTURER = "motionEye" + +SIGNAL_CAMERA_ADD = f"{DOMAIN}_camera_add_signal." "{}" +SIGNAL_CAMERA_REMOVE = f"{DOMAIN}_camera_remove_signal." "{}" + +TYPE_MOTIONEYE_MJPEG_CAMERA = "motioneye_mjpeg_camera" diff --git a/homeassistant/components/motioneye/manifest.json b/homeassistant/components/motioneye/manifest.json new file mode 100644 index 00000000000..a4a1e028d53 --- /dev/null +++ b/homeassistant/components/motioneye/manifest.json @@ -0,0 +1,13 @@ +{ + "domain": "motioneye", + "name": "motionEye", + "documentation": "https://www.home-assistant.io/integrations/motioneye", + "config_flow": true, + "requirements": [ + "motioneye-client==0.3.2" + ], + "codeowners": [ + "@dermotduffy" + ], + "iot_class": "local_polling" +} diff --git a/homeassistant/components/motioneye/strings.json b/homeassistant/components/motioneye/strings.json new file mode 100644 index 00000000000..d365ba272ea --- /dev/null +++ b/homeassistant/components/motioneye/strings.json @@ -0,0 +1,25 @@ +{ + "config": { + "step": { + "user": { + "data": { + "url": "[%key:common::config_flow::data::url%]", + "admin_username": "Admin [%key:common::config_flow::data::username%]", + "admin_password": "Admin [%key:common::config_flow::data::password%]", + "surveillance_username": "Surveillance [%key:common::config_flow::data::username%]", + "surveillance_password": "Surveillance [%key:common::config_flow::data::password%]" + } + } + }, + "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%]", + "invalid_url": "Invalid URL" + }, + "abort": { + "already_configured": "[%key:common::config_flow::abort::already_configured_service%]", + "reauth_successful": "[%key:common::config_flow::abort::reauth_successful%]" + } + } +} diff --git a/homeassistant/generated/config_flows.py b/homeassistant/generated/config_flows.py index f4bb23d698c..764ce9e594b 100644 --- a/homeassistant/generated/config_flows.py +++ b/homeassistant/generated/config_flows.py @@ -151,6 +151,7 @@ FLOWS = [ "mobile_app", "monoprice", "motion_blinds", + "motioneye", "mqtt", "mullvad", "myq", diff --git a/requirements_all.txt b/requirements_all.txt index 9d139759fcc..bfb926195ac 100644 --- a/requirements_all.txt +++ b/requirements_all.txt @@ -953,6 +953,9 @@ mitemp_bt==0.0.3 # homeassistant.components.motion_blinds motionblinds==0.4.10 +# homeassistant.components.motioneye +motioneye-client==0.3.2 + # homeassistant.components.mullvad mullvad-api==1.0.0 diff --git a/requirements_test_all.txt b/requirements_test_all.txt index 075d3be686f..50f0d3b70ca 100644 --- a/requirements_test_all.txt +++ b/requirements_test_all.txt @@ -513,6 +513,9 @@ minio==4.0.9 # homeassistant.components.motion_blinds motionblinds==0.4.10 +# homeassistant.components.motioneye +motioneye-client==0.3.2 + # homeassistant.components.mullvad mullvad-api==1.0.0 diff --git a/tests/components/motioneye/__init__.py b/tests/components/motioneye/__init__.py new file mode 100644 index 00000000000..a462d083038 --- /dev/null +++ b/tests/components/motioneye/__init__.py @@ -0,0 +1,180 @@ +"""Tests for the motionEye integration.""" +from __future__ import annotations + +from typing import Any +from unittest.mock import AsyncMock, Mock, patch + +from motioneye_client.const import DEFAULT_PORT + +from homeassistant.components.motioneye.const import DOMAIN +from homeassistant.config_entries import ConfigEntry +from homeassistant.const import CONF_URL +from homeassistant.core import HomeAssistant + +from tests.common import MockConfigEntry + +TEST_CONFIG_ENTRY_ID = "74565ad414754616000674c87bdc876c" +TEST_URL = f"http://test:{DEFAULT_PORT+1}" +TEST_CAMERA_ID = 100 +TEST_CAMERA_NAME = "Test Camera" +TEST_CAMERA_ENTITY_ID = "camera.test_camera" +TEST_CAMERA_DEVICE_IDENTIFIER = (DOMAIN, TEST_CONFIG_ENTRY_ID, TEST_CAMERA_ID) +TEST_CAMERA = { + "show_frame_changes": False, + "framerate": 25, + "actions": [], + "preserve_movies": 0, + "auto_threshold_tuning": True, + "recording_mode": "motion-triggered", + "monday_to": "", + "streaming_resolution": 100, + "light_switch_detect": 0, + "command_end_notifications_enabled": False, + "smb_shares": False, + "upload_server": "", + "monday_from": "", + "movie_passthrough": False, + "auto_brightness": False, + "frame_change_threshold": 3.0, + "name": TEST_CAMERA_NAME, + "movie_format": "mp4:h264_omx", + "network_username": "", + "preserve_pictures": 0, + "event_gap": 30, + "enabled": True, + "upload_movie": True, + "video_streaming": True, + "upload_location": "", + "max_movie_length": 0, + "movie_file_name": "%Y-%m-%d/%H-%M-%S", + "upload_authorization_key": "", + "still_images": False, + "upload_method": "post", + "max_frame_change_threshold": 0, + "device_url": "rtsp://localhost/live", + "text_overlay": False, + "right_text": "timestamp", + "upload_picture": True, + "email_notifications_enabled": False, + "working_schedule_type": "during", + "movie_quality": 75, + "disk_total": 44527655808, + "upload_service": "ftp", + "upload_password": "", + "wednesday_to": "", + "mask_type": "smart", + "command_storage_enabled": False, + "disk_used": 11419704992, + "streaming_motion": 0, + "manual_snapshots": True, + "noise_level": 12, + "mask_lines": [], + "upload_enabled": False, + "root_directory": f"/var/lib/motioneye/{TEST_CAMERA_NAME}", + "clean_cloud_enabled": False, + "working_schedule": False, + "pre_capture": 1, + "command_notifications_enabled": False, + "streaming_framerate": 25, + "email_notifications_picture_time_span": 0, + "thursday_to": "", + "streaming_server_resize": False, + "upload_subfolders": True, + "sunday_to": "", + "left_text": "", + "image_file_name": "%Y-%m-%d/%H-%M-%S", + "rotation": 0, + "capture_mode": "manual", + "movies": False, + "motion_detection": True, + "text_scale": 1, + "upload_username": "", + "upload_port": "", + "available_disks": [], + "network_smb_ver": "1.0", + "streaming_auth_mode": "basic", + "despeckle_filter": "", + "snapshot_interval": 0, + "minimum_motion_frames": 20, + "auto_noise_detect": True, + "network_share_name": "", + "sunday_from": "", + "friday_from": "", + "web_hook_storage_enabled": False, + "custom_left_text": "", + "streaming_port": 8081, + "id": TEST_CAMERA_ID, + "post_capture": 1, + "streaming_quality": 75, + "wednesday_from": "", + "proto": "netcam", + "extra_options": [], + "image_quality": 85, + "create_debug_media": False, + "friday_to": "", + "custom_right_text": "", + "web_hook_notifications_enabled": False, + "saturday_from": "", + "available_resolutions": [ + "1600x1200", + "1920x1080", + ], + "tuesday_from": "", + "network_password": "", + "saturday_to": "", + "network_server": "", + "smart_mask_sluggishness": 5, + "mask": False, + "tuesday_to": "", + "thursday_from": "", + "storage_device": "custom-path", + "resolution": "1920x1080", +} +TEST_CAMERAS = {"cameras": [TEST_CAMERA]} +TEST_SURVEILLANCE_USERNAME = "surveillance_username" + + +def create_mock_motioneye_client() -> AsyncMock: + """Create mock motionEye client.""" + mock_client = AsyncMock() + mock_client.async_client_login = AsyncMock(return_value={}) + mock_client.async_get_cameras = AsyncMock(return_value=TEST_CAMERAS) + mock_client.async_client_close = AsyncMock(return_value=True) + mock_client.get_camera_snapshot_url = Mock(return_value="") + mock_client.get_camera_stream_url = Mock(return_value="") + return mock_client + + +def create_mock_motioneye_config_entry( + hass: HomeAssistant, + data: dict[str, Any] | None = None, + options: dict[str, Any] | None = None, +) -> ConfigEntry: + """Add a test config entry.""" + config_entry: MockConfigEntry = MockConfigEntry( # type: ignore[no-untyped-call] + entry_id=TEST_CONFIG_ENTRY_ID, + domain=DOMAIN, + data=data or {CONF_URL: TEST_URL}, + title=f"{TEST_URL}", + options=options or {}, + ) + config_entry.add_to_hass(hass) # type: ignore[no-untyped-call] + return config_entry + + +async def setup_mock_motioneye_config_entry( + hass: HomeAssistant, + config_entry: ConfigEntry | None = None, + client: Mock | None = None, +) -> ConfigEntry: + """Add a mock MotionEye config entry to hass.""" + config_entry = config_entry or create_mock_motioneye_config_entry(hass) + client = client or create_mock_motioneye_client() + + with patch( + "homeassistant.components.motioneye.MotionEyeClient", + return_value=client, + ): + await hass.config_entries.async_setup(config_entry.entry_id) + await hass.async_block_till_done() + return config_entry diff --git a/tests/components/motioneye/test_camera.py b/tests/components/motioneye/test_camera.py new file mode 100644 index 00000000000..921dc9df920 --- /dev/null +++ b/tests/components/motioneye/test_camera.py @@ -0,0 +1,315 @@ +"""Test the motionEye camera.""" +import copy +import logging +from typing import Any +from unittest.mock import AsyncMock, Mock + +from aiohttp import web # type: ignore +from aiohttp.web_exceptions import HTTPBadGateway +from motioneye_client.client import ( + MotionEyeClientError, + MotionEyeClientInvalidAuthError, +) +from motioneye_client.const import ( + KEY_CAMERAS, + KEY_MOTION_DETECTION, + KEY_NAME, + KEY_VIDEO_STREAMING, +) +import pytest + +from homeassistant.components.camera import async_get_image, async_get_mjpeg_stream +from homeassistant.components.motioneye import get_motioneye_device_identifier +from homeassistant.components.motioneye.const import ( + CONF_SURVEILLANCE_USERNAME, + DEFAULT_SCAN_INTERVAL, + DOMAIN, + MOTIONEYE_MANUFACTURER, +) +from homeassistant.const import CONF_URL +from homeassistant.core import HomeAssistant +from homeassistant.exceptions import HomeAssistantError +from homeassistant.helpers import device_registry as dr, entity_registry as er +from homeassistant.helpers.device_registry import async_get_registry +import homeassistant.util.dt as dt_util + +from . import ( + TEST_CAMERA_DEVICE_IDENTIFIER, + TEST_CAMERA_ENTITY_ID, + TEST_CAMERA_ID, + TEST_CAMERA_NAME, + TEST_CAMERAS, + TEST_CONFIG_ENTRY_ID, + TEST_SURVEILLANCE_USERNAME, + create_mock_motioneye_client, + create_mock_motioneye_config_entry, + setup_mock_motioneye_config_entry, +) + +from tests.common import async_fire_time_changed + +_LOGGER = logging.getLogger(__name__) + + +async def test_setup_camera(hass: HomeAssistant) -> None: + """Test a basic camera.""" + client = create_mock_motioneye_client() + await setup_mock_motioneye_config_entry(hass, client=client) + + entity_state = hass.states.get(TEST_CAMERA_ENTITY_ID) + assert entity_state + assert entity_state.state == "idle" + assert entity_state.attributes.get("friendly_name") == TEST_CAMERA_NAME + + +async def test_setup_camera_auth_fail(hass: HomeAssistant) -> None: + """Test a successful camera.""" + client = create_mock_motioneye_client() + client.async_client_login = AsyncMock(side_effect=MotionEyeClientInvalidAuthError) + await setup_mock_motioneye_config_entry(hass, client=client) + assert not hass.states.get(TEST_CAMERA_ENTITY_ID) + + +async def test_setup_camera_client_error(hass: HomeAssistant) -> None: + """Test a successful camera.""" + client = create_mock_motioneye_client() + client.async_client_login = AsyncMock(side_effect=MotionEyeClientError) + await setup_mock_motioneye_config_entry(hass, client=client) + assert not hass.states.get(TEST_CAMERA_ENTITY_ID) + + +async def test_setup_camera_empty_data(hass: HomeAssistant) -> None: + """Test a successful camera.""" + client = create_mock_motioneye_client() + client.async_get_cameras = AsyncMock(return_value={}) + await setup_mock_motioneye_config_entry(hass, client=client) + assert not hass.states.get(TEST_CAMERA_ENTITY_ID) + + +async def test_setup_camera_bad_data(hass: HomeAssistant) -> None: + """Test bad camera data.""" + client = create_mock_motioneye_client() + cameras = copy.deepcopy(TEST_CAMERAS) + del cameras[KEY_CAMERAS][0][KEY_NAME] + + client.async_get_cameras = AsyncMock(return_value=cameras) + await setup_mock_motioneye_config_entry(hass, client=client) + assert not hass.states.get(TEST_CAMERA_ENTITY_ID) + + +async def test_setup_camera_without_streaming(hass: HomeAssistant) -> None: + """Test a camera without streaming enabled.""" + client = create_mock_motioneye_client() + cameras = copy.deepcopy(TEST_CAMERAS) + cameras[KEY_CAMERAS][0][KEY_VIDEO_STREAMING] = False + + client.async_get_cameras = AsyncMock(return_value=cameras) + await setup_mock_motioneye_config_entry(hass, client=client) + entity_state = hass.states.get(TEST_CAMERA_ENTITY_ID) + assert entity_state + assert entity_state.state == "unavailable" + + +async def test_setup_camera_new_data_same(hass: HomeAssistant) -> None: + """Test a data refresh with the same data.""" + client = create_mock_motioneye_client() + await setup_mock_motioneye_config_entry(hass, client=client) + async_fire_time_changed(hass, dt_util.utcnow() + DEFAULT_SCAN_INTERVAL) + await hass.async_block_till_done() + assert hass.states.get(TEST_CAMERA_ENTITY_ID) + + +async def test_setup_camera_new_data_camera_removed(hass: HomeAssistant) -> None: + """Test a data refresh with a removed camera.""" + device_registry = await async_get_registry(hass) + entity_registry = await er.async_get_registry(hass) + + client = create_mock_motioneye_client() + config_entry = await setup_mock_motioneye_config_entry(hass, client=client) + + # Create some random old devices/entity_ids and ensure they get cleaned up. + old_device_id = "old-device-id" + old_entity_unique_id = "old-entity-unique_id" + old_device = device_registry.async_get_or_create( + config_entry_id=config_entry.entry_id, identifiers={(DOMAIN, old_device_id)} + ) + entity_registry.async_get_or_create( + domain=DOMAIN, + platform="camera", + unique_id=old_entity_unique_id, + config_entry=config_entry, + device_id=old_device.id, + ) + + await hass.async_block_till_done() + assert hass.states.get(TEST_CAMERA_ENTITY_ID) + assert device_registry.async_get_device({TEST_CAMERA_DEVICE_IDENTIFIER}) + + client.async_get_cameras = AsyncMock(return_value={KEY_CAMERAS: []}) + async_fire_time_changed(hass, dt_util.utcnow() + DEFAULT_SCAN_INTERVAL) + await hass.async_block_till_done() + assert not hass.states.get(TEST_CAMERA_ENTITY_ID) + assert not device_registry.async_get_device({TEST_CAMERA_DEVICE_IDENTIFIER}) + assert not device_registry.async_get_device({(DOMAIN, old_device_id)}) + assert not entity_registry.async_get_entity_id( + DOMAIN, "camera", old_entity_unique_id + ) + + +async def test_setup_camera_new_data_error(hass: HomeAssistant) -> None: + """Test a data refresh that fails.""" + client = create_mock_motioneye_client() + await setup_mock_motioneye_config_entry(hass, client=client) + assert hass.states.get(TEST_CAMERA_ENTITY_ID) + client.async_get_cameras = AsyncMock(side_effect=MotionEyeClientError) + async_fire_time_changed(hass, dt_util.utcnow() + DEFAULT_SCAN_INTERVAL) + await hass.async_block_till_done() + entity_state = hass.states.get(TEST_CAMERA_ENTITY_ID) + assert entity_state.state == "unavailable" + + +async def test_setup_camera_new_data_without_streaming(hass: HomeAssistant) -> None: + """Test a data refresh without streaming.""" + client = create_mock_motioneye_client() + await setup_mock_motioneye_config_entry(hass, client=client) + entity_state = hass.states.get(TEST_CAMERA_ENTITY_ID) + assert entity_state.state == "idle" + + cameras = copy.deepcopy(TEST_CAMERAS) + cameras[KEY_CAMERAS][0][KEY_VIDEO_STREAMING] = False + client.async_get_cameras = AsyncMock(return_value=cameras) + async_fire_time_changed(hass, dt_util.utcnow() + DEFAULT_SCAN_INTERVAL) + await hass.async_block_till_done() + entity_state = hass.states.get(TEST_CAMERA_ENTITY_ID) + assert entity_state.state == "unavailable" + + +async def test_unload_camera(hass: HomeAssistant) -> None: + """Test unloading camera.""" + client = create_mock_motioneye_client() + entry = await setup_mock_motioneye_config_entry(hass, client=client) + assert hass.states.get(TEST_CAMERA_ENTITY_ID) + assert not client.async_client_close.called + await hass.config_entries.async_unload(entry.entry_id) + assert client.async_client_close.called + + +async def test_get_still_image_from_camera( + aiohttp_server: Any, hass: HomeAssistant +) -> None: + """Test getting a still image.""" + + image_handler = Mock(return_value="") + + app = web.Application() + app.add_routes( + [ + web.get( + "/foo", + image_handler, + ) + ] + ) + + server = await aiohttp_server(app) + client = create_mock_motioneye_client() + client.get_camera_snapshot_url = Mock( + return_value=f"http://localhost:{server.port}/foo" + ) + config_entry = create_mock_motioneye_config_entry( + hass, + data={ + CONF_URL: f"http://localhost:{server.port}", + CONF_SURVEILLANCE_USERNAME: TEST_SURVEILLANCE_USERNAME, + }, + ) + + await setup_mock_motioneye_config_entry( + hass, config_entry=config_entry, client=client + ) + await hass.async_block_till_done() + + # It won't actually get a stream from the dummy handler, so just catch + # the expected exception, then verify the right handler was called. + with pytest.raises(HomeAssistantError): + await async_get_image(hass, TEST_CAMERA_ENTITY_ID, timeout=None) # type: ignore[no-untyped-call] + assert image_handler.called + + +async def test_get_stream_from_camera(aiohttp_server: Any, hass: HomeAssistant) -> None: + """Test getting a stream.""" + + stream_handler = Mock(return_value="") + app = web.Application() + app.add_routes([web.get("/", stream_handler)]) + stream_server = await aiohttp_server(app) + + client = create_mock_motioneye_client() + client.get_camera_stream_url = Mock( + return_value=f"http://localhost:{stream_server.port}/" + ) + config_entry = create_mock_motioneye_config_entry( + hass, + data={ + CONF_URL: f"http://localhost:{stream_server.port}", + # The port won't be used as the client is a mock. + CONF_SURVEILLANCE_USERNAME: TEST_SURVEILLANCE_USERNAME, + }, + ) + cameras = copy.deepcopy(TEST_CAMERAS) + client.async_get_cameras = AsyncMock(return_value=cameras) + await setup_mock_motioneye_config_entry( + hass, config_entry=config_entry, client=client + ) + await hass.async_block_till_done() + + # It won't actually get a stream from the dummy handler, so just catch + # the expected exception, then verify the right handler was called. + with pytest.raises(HTTPBadGateway): + await async_get_mjpeg_stream(hass, None, TEST_CAMERA_ENTITY_ID) # type: ignore[no-untyped-call] + assert stream_handler.called + + +async def test_state_attributes(hass: HomeAssistant) -> None: + """Test state attributes are set correctly.""" + client = create_mock_motioneye_client() + await setup_mock_motioneye_config_entry(hass, client=client) + + entity_state = hass.states.get(TEST_CAMERA_ENTITY_ID) + assert entity_state + assert entity_state.attributes.get("brand") == MOTIONEYE_MANUFACTURER + assert entity_state.attributes.get("motion_detection") + + cameras = copy.deepcopy(TEST_CAMERAS) + cameras[KEY_CAMERAS][0][KEY_MOTION_DETECTION] = False + client.async_get_cameras = AsyncMock(return_value=cameras) + async_fire_time_changed(hass, dt_util.utcnow() + DEFAULT_SCAN_INTERVAL) + await hass.async_block_till_done() + + entity_state = hass.states.get(TEST_CAMERA_ENTITY_ID) + assert entity_state + assert not entity_state.attributes.get("motion_detection") + + +async def test_device_info(hass: HomeAssistant) -> None: + """Verify device information includes expected details.""" + client = create_mock_motioneye_client() + entry = await setup_mock_motioneye_config_entry(hass, client=client) + + device_identifier = get_motioneye_device_identifier(entry.entry_id, TEST_CAMERA_ID) + device_registry = dr.async_get(hass) + + device = device_registry.async_get_device({device_identifier}) + assert device + assert device.config_entries == {TEST_CONFIG_ENTRY_ID} + assert device.identifiers == {device_identifier} + assert device.manufacturer == MOTIONEYE_MANUFACTURER + assert device.model == MOTIONEYE_MANUFACTURER + assert device.name == TEST_CAMERA_NAME + + entity_registry = await er.async_get_registry(hass) + entities_from_device = [ + entry.entity_id + for entry in er.async_entries_for_device(entity_registry, device.id) + ] + assert TEST_CAMERA_ENTITY_ID in entities_from_device diff --git a/tests/components/motioneye/test_config_flow.py b/tests/components/motioneye/test_config_flow.py new file mode 100644 index 00000000000..2c16aea14be --- /dev/null +++ b/tests/components/motioneye/test_config_flow.py @@ -0,0 +1,233 @@ +"""Test the motionEye config flow.""" +import logging +from unittest.mock import AsyncMock, patch + +from motioneye_client.client import ( + MotionEyeClientConnectionError, + MotionEyeClientInvalidAuthError, + MotionEyeClientRequestError, +) + +from homeassistant import config_entries, data_entry_flow, setup +from homeassistant.components.motioneye.const import ( + CONF_ADMIN_PASSWORD, + CONF_ADMIN_USERNAME, + CONF_CONFIG_ENTRY, + CONF_SURVEILLANCE_PASSWORD, + CONF_SURVEILLANCE_USERNAME, + DOMAIN, +) +from homeassistant.const import CONF_URL +from homeassistant.core import HomeAssistant + +from . import TEST_URL, create_mock_motioneye_client, create_mock_motioneye_config_entry + +_LOGGER = logging.getLogger(__name__) + + +async def test_user_success(hass: HomeAssistant) -> None: + """Test successful user flow.""" + await setup.async_setup_component(hass, "persistent_notification", {}) + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": config_entries.SOURCE_USER} + ) + assert result["type"] == "form" + assert result["errors"] == {} + + mock_client = create_mock_motioneye_client() + + with patch( + "homeassistant.components.motioneye.MotionEyeClient", + return_value=mock_client, + ), patch( + "homeassistant.components.motioneye.async_setup_entry", + return_value=True, + ) as mock_setup_entry: + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + { + CONF_URL: TEST_URL, + CONF_ADMIN_USERNAME: "admin-username", + CONF_ADMIN_PASSWORD: "admin-password", + CONF_SURVEILLANCE_USERNAME: "surveillance-username", + CONF_SURVEILLANCE_PASSWORD: "surveillance-password", + }, + ) + await hass.async_block_till_done() + + assert result["type"] == "create_entry" + assert result["title"] == f"{TEST_URL}" + assert result["data"] == { + CONF_URL: TEST_URL, + CONF_ADMIN_USERNAME: "admin-username", + CONF_ADMIN_PASSWORD: "admin-password", + CONF_SURVEILLANCE_USERNAME: "surveillance-username", + CONF_SURVEILLANCE_PASSWORD: "surveillance-password", + } + assert len(mock_setup_entry.mock_calls) == 1 + + +async def test_user_invalid_auth(hass: HomeAssistant) -> None: + """Test invalid auth is handled correctly.""" + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": config_entries.SOURCE_USER} + ) + + mock_client = create_mock_motioneye_client() + mock_client.async_client_login = AsyncMock( + side_effect=MotionEyeClientInvalidAuthError + ) + + with patch( + "homeassistant.components.motioneye.MotionEyeClient", + return_value=mock_client, + ): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + { + CONF_URL: TEST_URL, + CONF_ADMIN_USERNAME: "admin-username", + CONF_ADMIN_PASSWORD: "admin-password", + CONF_SURVEILLANCE_USERNAME: "surveillance-username", + CONF_SURVEILLANCE_PASSWORD: "surveillance-password", + }, + ) + await mock_client.async_client_close() + + assert result["type"] == "form" + assert result["errors"] == {"base": "invalid_auth"} + + +async def test_user_invalid_url(hass: HomeAssistant) -> None: + """Test invalid url is handled correctly.""" + await setup.async_setup_component(hass, "persistent_notification", {}) + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": config_entries.SOURCE_USER} + ) + + with patch( + "homeassistant.components.motioneye.MotionEyeClient", + return_value=create_mock_motioneye_client(), + ): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + { + CONF_URL: "not a url", + CONF_ADMIN_USERNAME: "admin-username", + CONF_ADMIN_PASSWORD: "admin-password", + CONF_SURVEILLANCE_USERNAME: "surveillance-username", + CONF_SURVEILLANCE_PASSWORD: "surveillance-password", + }, + ) + + assert result["type"] == "form" + assert result["errors"] == {"base": "invalid_url"} + + +async def test_user_cannot_connect(hass: HomeAssistant) -> None: + """Test connection failure is handled correctly.""" + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": config_entries.SOURCE_USER} + ) + + mock_client = create_mock_motioneye_client() + mock_client.async_client_login = AsyncMock( + side_effect=MotionEyeClientConnectionError, + ) + + with patch( + "homeassistant.components.motioneye.MotionEyeClient", + return_value=mock_client, + ): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + { + CONF_URL: TEST_URL, + CONF_ADMIN_USERNAME: "admin-username", + CONF_ADMIN_PASSWORD: "admin-password", + CONF_SURVEILLANCE_USERNAME: "surveillance-username", + CONF_SURVEILLANCE_PASSWORD: "surveillance-password", + }, + ) + await mock_client.async_client_close() + + assert result["type"] == "form" + assert result["errors"] == {"base": "cannot_connect"} + + +async def test_user_request_error(hass: HomeAssistant) -> None: + """Test a request error is handled correctly.""" + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": config_entries.SOURCE_USER} + ) + + mock_client = create_mock_motioneye_client() + mock_client.async_client_login = AsyncMock(side_effect=MotionEyeClientRequestError) + + with patch( + "homeassistant.components.motioneye.MotionEyeClient", + return_value=mock_client, + ): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + { + CONF_URL: TEST_URL, + CONF_ADMIN_USERNAME: "admin-username", + CONF_ADMIN_PASSWORD: "admin-password", + CONF_SURVEILLANCE_USERNAME: "surveillance-username", + CONF_SURVEILLANCE_PASSWORD: "surveillance-password", + }, + ) + await mock_client.async_client_close() + + assert result["type"] == "form" + assert result["errors"] == {"base": "unknown"} + + +async def test_reauth(hass: HomeAssistant) -> None: + """Test a reauth.""" + config_data = { + CONF_URL: TEST_URL, + } + + config_entry = create_mock_motioneye_config_entry(hass, data=config_data) + + await setup.async_setup_component(hass, "persistent_notification", {}) + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={ + "source": config_entries.SOURCE_REAUTH, + CONF_CONFIG_ENTRY: config_entry, + }, + ) + assert result["type"] == "form" + assert result["errors"] == {} + + mock_client = create_mock_motioneye_client() + + new_data = { + CONF_URL: TEST_URL, + CONF_ADMIN_USERNAME: "admin-username", + CONF_ADMIN_PASSWORD: "admin-password", + CONF_SURVEILLANCE_USERNAME: "surveillance-username", + CONF_SURVEILLANCE_PASSWORD: "surveillance-password", + } + + with patch( + "homeassistant.components.motioneye.MotionEyeClient", + return_value=mock_client, + ), patch( + "homeassistant.components.motioneye.async_setup_entry", + return_value=True, + ) as mock_setup_entry: + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + new_data, + ) + await hass.async_block_till_done() + + assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT + assert result["reason"] == "reauth_successful" + assert config_entry.data == new_data + + assert len(mock_setup_entry.mock_calls) == 1