diff --git a/homeassistant/components/tplink/__init__.py b/homeassistant/components/tplink/__init__.py index a7ffce686be..e2a2f99517f 100644 --- a/homeassistant/components/tplink/__init__.py +++ b/homeassistant/components/tplink/__init__.py @@ -47,10 +47,12 @@ from homeassistant.helpers.typing import ConfigType from .const import ( CONF_AES_KEYS, + CONF_CAMERA_CREDENTIALS, CONF_CONFIG_ENTRY_MINOR_VERSION, CONF_CONNECTION_PARAMETERS, CONF_CREDENTIALS_HASH, CONF_DEVICE_CONFIG, + CONF_LIVE_VIEW, CONF_USES_HTTP, CONNECT_TIMEOUT, DISCOVERY_TIMEOUT, @@ -226,7 +228,16 @@ async def async_setup_entry(hass: HomeAssistant, entry: TPLinkConfigEntry) -> bo for child in device.children ] - entry.runtime_data = TPLinkData(parent_coordinator, child_coordinators) + camera_creds: Credentials | None = None + if camera_creds_dict := entry.data.get(CONF_CAMERA_CREDENTIALS): + camera_creds = Credentials( + camera_creds_dict[CONF_USERNAME], camera_creds_dict[CONF_PASSWORD] + ) + live_view = entry.data.get(CONF_LIVE_VIEW) + + entry.runtime_data = TPLinkData( + parent_coordinator, child_coordinators, camera_creds, live_view + ) await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS) return True diff --git a/homeassistant/components/tplink/camera.py b/homeassistant/components/tplink/camera.py new file mode 100644 index 00000000000..5ed279909d6 --- /dev/null +++ b/homeassistant/components/tplink/camera.py @@ -0,0 +1,220 @@ +"""Support for TPLink camera entities.""" + +import asyncio +from dataclasses import dataclass +import logging +import time + +from aiohttp import web +from haffmpeg.camera import CameraMjpeg +from kasa import Credentials, Device, Module +from kasa.smartcam.modules import Camera as CameraModule + +from homeassistant.components import ffmpeg, stream +from homeassistant.components.camera import ( + Camera, + CameraEntityDescription, + CameraEntityFeature, +) +from homeassistant.config_entries import ConfigFlowContext +from homeassistant.core import HomeAssistant, callback +from homeassistant.helpers.aiohttp_client import async_aiohttp_proxy_stream +from homeassistant.helpers.entity_platform import AddEntitiesCallback + +from . import TPLinkConfigEntry, legacy_device_id +from .const import CONF_CAMERA_CREDENTIALS +from .coordinator import TPLinkDataUpdateCoordinator +from .entity import CoordinatedTPLinkEntity, TPLinkModuleEntityDescription + +_LOGGER = logging.getLogger(__name__) + + +@dataclass(frozen=True, kw_only=True) +class TPLinkCameraEntityDescription( + CameraEntityDescription, TPLinkModuleEntityDescription +): + """Base class for camera entity description.""" + + +CAMERA_DESCRIPTIONS: tuple[TPLinkCameraEntityDescription, ...] = ( + TPLinkCameraEntityDescription( + key="live_view", + translation_key="live_view", + ), +) + + +async def async_setup_entry( + hass: HomeAssistant, + config_entry: TPLinkConfigEntry, + async_add_entities: AddEntitiesCallback, +) -> None: + """Set up camera entities.""" + data = config_entry.runtime_data + parent_coordinator = data.parent_coordinator + device = parent_coordinator.device + camera_credentials = data.camera_credentials + live_view = data.live_view + ffmpeg_manager = ffmpeg.get_ffmpeg_manager(hass) + + async_add_entities( + TPLinkCameraEntity( + device, + parent_coordinator, + description, + camera_module=camera_module, + parent=None, + ffmpeg_manager=ffmpeg_manager, + camera_credentials=camera_credentials, + ) + for description in CAMERA_DESCRIPTIONS + if (camera_module := device.modules.get(Module.Camera)) and live_view + ) + + +class TPLinkCameraEntity(CoordinatedTPLinkEntity, Camera): + """Representation of a TPLink camera.""" + + IMAGE_INTERVAL = 5 * 60 + + _attr_supported_features = CameraEntityFeature.STREAM | CameraEntityFeature.ON_OFF + + entity_description: TPLinkCameraEntityDescription + + def __init__( + self, + device: Device, + coordinator: TPLinkDataUpdateCoordinator, + description: TPLinkCameraEntityDescription, + *, + camera_module: CameraModule, + parent: Device | None = None, + ffmpeg_manager: ffmpeg.FFmpegManager, + camera_credentials: Credentials | None, + ) -> None: + """Initialize a TPlink camera.""" + self.entity_description = description + self._camera_module = camera_module + self._video_url = camera_module.stream_rtsp_url(camera_credentials) + self._image: bytes | None = None + super().__init__(device, coordinator, parent=parent) + Camera.__init__(self) + self._ffmpeg_manager = ffmpeg_manager + self._image_lock = asyncio.Lock() + self._last_update: float = 0 + self._camera_credentials = camera_credentials + self._can_stream = True + self._http_mpeg_stream_running = False + + def _get_unique_id(self) -> str: + """Return unique ID for the entity.""" + return f"{legacy_device_id(self._device)}-{self.entity_description}" + + @callback + def _async_update_attrs(self) -> None: + """Update the entity's attributes.""" + self._attr_is_on = self._camera_module.is_on + + async def stream_source(self) -> str | None: + """Return the source of the stream.""" + return self._video_url + + async def _async_check_stream_auth(self, video_url: str) -> None: + """Check for an auth error and start reauth flow.""" + try: + await stream.async_check_stream_client_error(self.hass, video_url) + except stream.StreamOpenClientError as ex: + if ex.stream_client_error is stream.StreamClientError.Unauthorized: + _LOGGER.debug( + "Camera stream failed authentication for %s", + self._device.host, + ) + self._can_stream = False + self.coordinator.config_entry.async_start_reauth( + self.hass, + ConfigFlowContext( + reauth_source=CONF_CAMERA_CREDENTIALS, # type: ignore[typeddict-unknown-key] + ), + {"device": self._device}, + ) + + async def async_camera_image( + self, width: int | None = None, height: int | None = None + ) -> bytes | None: + """Return a still image response from the camera.""" + now = time.monotonic() + + if self._image and now - self._last_update < self.IMAGE_INTERVAL: + return self._image + + # Don't try to capture a new image if a stream is running + if (self.stream and self.stream.available) or self._http_mpeg_stream_running: + return self._image + + if self._can_stream and (video_url := self._video_url): + # Sometimes the front end makes multiple image requests + async with self._image_lock: + if self._image and (now - self._last_update) < self.IMAGE_INTERVAL: + return self._image + + _LOGGER.debug("Updating camera image for %s", self._device.host) + image = await ffmpeg.async_get_image( + self.hass, + video_url, + width=width, + height=height, + ) + if image: + self._image = image + self._last_update = now + _LOGGER.debug("Updated camera image for %s", self._device.host) + # This coroutine is called by camera with an asyncio.timeout + # so image could be None whereas an auth issue returns b'' + elif image == b"": + _LOGGER.debug( + "Empty camera image returned for %s", self._device.host + ) + # image could be empty if a stream is running so check for explicit auth error + await self._async_check_stream_auth(video_url) + else: + _LOGGER.debug( + "None camera image returned for %s", self._device.host + ) + + return self._image + + async def handle_async_mjpeg_stream( + self, request: web.Request + ) -> web.StreamResponse | None: + """Generate an HTTP MJPEG stream from the camera. + + The frontend falls back to calling this method if the HLS + stream fails. + """ + _LOGGER.debug("Starting http mjpeg stream for %s", self._device.host) + if self._video_url is None or self._can_stream is False: + return None + + mjpeg_stream = CameraMjpeg(self._ffmpeg_manager.binary) + await mjpeg_stream.open_camera(self._video_url) + self._http_mpeg_stream_running = True + try: + stream_reader = await mjpeg_stream.get_reader() + return await async_aiohttp_proxy_stream( + self.hass, + request, + stream_reader, + self._ffmpeg_manager.ffmpeg_stream_content_type, + ) + finally: + self._http_mpeg_stream_running = False + await mjpeg_stream.close() + _LOGGER.debug("Stopped http mjpeg stream for %s", self._device.host) + + async def async_turn_on(self) -> None: + """Turn on camera.""" + await self._camera_module.set_state(True) + + async def async_turn_off(self) -> None: + """Turn off camera.""" + await self._camera_module.set_state(False) diff --git a/homeassistant/components/tplink/config_flow.py b/homeassistant/components/tplink/config_flow.py index 63f1b4e125b..db6f9a58ba5 100644 --- a/homeassistant/components/tplink/config_flow.py +++ b/homeassistant/components/tplink/config_flow.py @@ -4,7 +4,7 @@ from __future__ import annotations from collections.abc import Mapping import logging -from typing import TYPE_CHECKING, Any, Self +from typing import TYPE_CHECKING, Any, Self, cast from kasa import ( AuthenticationError, @@ -13,13 +13,15 @@ from kasa import ( DeviceConfig, Discover, KasaException, + Module, TimeoutError, ) import voluptuous as vol -from homeassistant.components import dhcp +from homeassistant.components import dhcp, ffmpeg, stream from homeassistant.config_entries import ( SOURCE_REAUTH, + SOURCE_RECONFIGURE, ConfigEntry, ConfigEntryState, ConfigFlow, @@ -31,6 +33,7 @@ from homeassistant.const import ( CONF_HOST, CONF_MAC, CONF_MODEL, + CONF_NAME, CONF_PASSWORD, CONF_PORT, CONF_USERNAME, @@ -48,9 +51,11 @@ from . import ( ) from .const import ( CONF_AES_KEYS, + CONF_CAMERA_CREDENTIALS, CONF_CONFIG_ENTRY_MINOR_VERSION, CONF_CONNECTION_PARAMETERS, CONF_CREDENTIALS_HASH, + CONF_LIVE_VIEW, CONF_USES_HTTP, CONNECT_TIMEOUT, DOMAIN, @@ -62,6 +67,16 @@ STEP_AUTH_DATA_SCHEMA = vol.Schema( {vol.Required(CONF_USERNAME): str, vol.Required(CONF_PASSWORD): str} ) +STEP_RECONFIGURE_DATA_SCHEMA = vol.Schema({vol.Required(CONF_HOST): str}) + +STEP_CAMERA_AUTH_DATA_SCHEMA = vol.Schema( + { + vol.Required(CONF_LIVE_VIEW): bool, + vol.Optional(CONF_USERNAME): str, + vol.Optional(CONF_PASSWORD): str, + } +) + class TPLinkConfigFlow(ConfigFlow, domain=DOMAIN): """Handle a config flow for tplink.""" @@ -227,7 +242,12 @@ class TPLinkConfigFlow(ConfigFlow, domain=DOMAIN): self.hass.async_create_task( self._async_reload_requires_auth_entries(), eager_start=False ) - return self._async_create_entry_from_device(self._discovered_device) + if self._async_supports_camera_credentials(device): + return await self.async_step_camera_auth_confirm() + + return self._async_create_or_update_entry_from_device( + self._discovered_device + ) self.context["title_placeholders"] = placeholders return self.async_show_form( @@ -253,7 +273,12 @@ class TPLinkConfigFlow(ConfigFlow, domain=DOMAIN): """Confirm discovery.""" assert self._discovered_device is not None if user_input is not None: - return self._async_create_entry_from_device(self._discovered_device) + if self._async_supports_camera_credentials(self._discovered_device): + return await self.async_step_camera_auth_confirm() + + return self._async_create_or_update_entry_from_device( + self._discovered_device + ) self._set_confirm_only() placeholders = self._async_make_placeholders_from_discovery() @@ -282,6 +307,13 @@ class TPLinkConfigFlow(ConfigFlow, domain=DOMAIN): return host, port + def _async_supports_camera_credentials(self, device: Device) -> bool: + """Return True if device could have separate camera credentials.""" + if camera_module := device.modules.get(Module.Camera): + self._discovered_device = device + return bool(camera_module.stream_rtsp_url()) + return False + async def async_step_user( self, user_input: dict[str, Any] | None = None ) -> ConfigFlowResult: @@ -324,7 +356,11 @@ class TPLinkConfigFlow(ConfigFlow, domain=DOMAIN): else: if not device: return await self.async_step_user_auth_confirm() - return self._async_create_entry_from_device(device) + + if self._async_supports_camera_credentials(device): + return await self.async_step_camera_auth_confirm() + + return self._async_create_or_update_entry_from_device(device) return self.async_show_form( step_id="user", @@ -375,7 +411,10 @@ class TPLinkConfigFlow(ConfigFlow, domain=DOMAIN): self.hass.async_create_task( self._async_reload_requires_auth_entries(), eager_start=False ) - return self._async_create_entry_from_device(device) + if self._async_supports_camera_credentials(device): + return await self.async_step_camera_auth_confirm() + + return self._async_create_or_update_entry_from_device(device) return self.async_show_form( step_id="user_auth_confirm", @@ -384,6 +423,104 @@ class TPLinkConfigFlow(ConfigFlow, domain=DOMAIN): description_placeholders=placeholders, ) + def _create_camera_entry( + self, device: Device, un: str, pw: str + ) -> ConfigFlowResult: + entry_data: dict[str, bool | dict[str, str]] = {CONF_LIVE_VIEW: True} + entry_data[CONF_CAMERA_CREDENTIALS] = { + CONF_USERNAME: un, + CONF_PASSWORD: pw, + } + _LOGGER.debug("Creating camera account entry for device %s", device.host) + return self._async_create_or_update_entry_from_device( + device, camera_data=entry_data + ) + + async def async_step_camera_auth_confirm( + self, user_input: dict[str, Any] | None = None + ) -> ConfigFlowResult: + """Dialog that gives the user option to set camera credentials.""" + errors: dict[str, str] = {} + placeholders: dict[str, str] = {} + device = self._discovered_device + assert device + + if user_input: + live_view = user_input[CONF_LIVE_VIEW] + if not live_view: + return self._async_create_or_update_entry_from_device( + device, camera_data={CONF_LIVE_VIEW: False} + ) + + un = user_input.get(CONF_USERNAME) + pw = user_input.get(CONF_PASSWORD) + + if user_input and un and pw: + camera_creds = Credentials(un, cast(str, pw)) + + camera_module = device.modules[Module.Camera] + rtsp_url = camera_module.stream_rtsp_url(camera_creds) + assert rtsp_url + + # If camera fails to create HLS stream via 'stream' then try + # ffmpeg.async_get_image as some cameras do not work with HLS + # and the frontend will fallback to mpeg on error + try: + await stream.async_check_stream_client_error(self.hass, rtsp_url) + except stream.StreamOpenClientError as ex: + if ex.stream_client_error is stream.StreamClientError.Unauthorized: + errors["base"] = "invalid_camera_auth" + else: + _LOGGER.debug( + "Device %s client error checking stream: %s", device.host, ex + ) + if await ffmpeg.async_get_image(self.hass, rtsp_url): + return self._create_camera_entry(device, un, pw) + + errors["base"] = "cannot_connect_camera" + placeholders["error"] = str(ex) + except Exception as ex: # noqa: BLE001 + _LOGGER.debug("Device %s error checking stream: %s", device.host, ex) + if await ffmpeg.async_get_image(self.hass, rtsp_url): + return self._create_camera_entry(device, un, pw) + + errors["base"] = "cannot_connect_camera" + placeholders["error"] = str(ex) + else: + return self._create_camera_entry(device, un, pw) + + elif user_input: + errors["base"] = "camera_creds" + + entry = None + if self.source == SOURCE_RECONFIGURE: + entry = self._get_reconfigure_entry() + elif self.source == SOURCE_REAUTH: + entry = self._get_reauth_entry() + + if entry: + placeholders[CONF_NAME] = entry.data[CONF_ALIAS] + placeholders[CONF_MODEL] = entry.data[CONF_MODEL] + placeholders[CONF_HOST] = entry.data[CONF_HOST] + + if user_input: + form_data = {**user_input} + elif entry: + form_data = {**entry.data.get(CONF_CAMERA_CREDENTIALS, {})} + form_data[CONF_LIVE_VIEW] = entry.data.get(CONF_LIVE_VIEW, False) + else: + form_data = {} + + self.context["title_placeholders"] = placeholders + return self.async_show_form( + step_id="camera_auth_confirm", + data_schema=self.add_suggested_values_to_schema( + STEP_CAMERA_AUTH_DATA_SCHEMA, form_data + ), + errors=errors, + description_placeholders=placeholders, + ) + async def async_step_pick_device( self, user_input: dict[str, Any] | None = None ) -> ConfigFlowResult: @@ -403,7 +540,11 @@ class TPLinkConfigFlow(ConfigFlow, domain=DOMAIN): return await self.async_step_user_auth_confirm() except KasaException: return self.async_abort(reason="cannot_connect") - return self._async_create_entry_from_device(device) + + if self._async_supports_camera_credentials(device): + return await self.async_step_camera_auth_confirm() + + return self._async_create_or_update_entry_from_device(device) configured_devices = { entry.unique_id for entry in self._async_current_entries() @@ -444,11 +585,19 @@ class TPLinkConfigFlow(ConfigFlow, domain=DOMAIN): _config_entries.flow.async_abort(flow["flow_id"]) @callback - def _async_create_entry_from_device(self, device: Device) -> ConfigFlowResult: + def _async_create_or_update_entry_from_device( + self, device: Device, *, camera_data: dict | None = None + ) -> ConfigFlowResult: """Create a config entry from a smart device.""" - # This is only ever called after a successful device update so we know that - # the credential_hash is correct and should be saved. - self._abort_if_unique_id_configured(updates={CONF_HOST: device.host}) + entry = None + if self.source == SOURCE_RECONFIGURE: + entry = self._get_reconfigure_entry() + elif self.source == SOURCE_REAUTH: + entry = self._get_reauth_entry() + + if not entry: + self._abort_if_unique_id_configured(updates={CONF_HOST: device.host}) + data: dict[str, Any] = { CONF_HOST: device.host, CONF_ALIAS: device.alias, @@ -456,16 +605,28 @@ class TPLinkConfigFlow(ConfigFlow, domain=DOMAIN): CONF_CONNECTION_PARAMETERS: device.config.connection_type.to_dict(), CONF_USES_HTTP: device.config.uses_http, } + if camera_data is not None: + data[CONF_LIVE_VIEW] = camera_data[CONF_LIVE_VIEW] + if camera_creds := camera_data.get(CONF_CAMERA_CREDENTIALS): + data[CONF_CAMERA_CREDENTIALS] = camera_creds + if device.config.aes_keys: data[CONF_AES_KEYS] = device.config.aes_keys + + # This is only ever called after a successful device update so we know that + # the credential_hash is correct and should be saved. if device.credentials_hash: data[CONF_CREDENTIALS_HASH] = device.credentials_hash if port := device.config.port_override: data[CONF_PORT] = port - return self.async_create_entry( - title=f"{device.alias} {device.model}", - data=data, - ) + + if not entry: + return self.async_create_entry( + title=f"{device.alias} {device.model}", + data=data, + ) + + return self.async_update_reload_and_abort(entry, data=data) async def _async_try_connect_all( self, @@ -546,7 +707,8 @@ class TPLinkConfigFlow(ConfigFlow, domain=DOMAIN): credentials: Credentials | None, ) -> Device: """Try to connect.""" - self._async_abort_entries_match({CONF_HOST: discovered_device.host}) + if self.source not in {SOURCE_RECONFIGURE, SOURCE_REAUTH}: + self._async_abort_entries_match({CONF_HOST: discovered_device.host}) config = discovered_device.config if credentials: @@ -566,6 +728,10 @@ class TPLinkConfigFlow(ConfigFlow, domain=DOMAIN): self, entry_data: Mapping[str, Any] ) -> ConfigFlowResult: """Start the reauthentication flow if the device needs updated credentials.""" + if self.context.get("reauth_source") == CONF_CAMERA_CREDENTIALS: + self._discovered_device = entry_data["device"] + return await self.async_step_camera_auth_confirm() + return await self.async_step_reauth_confirm() async def async_step_reauth_confirm( @@ -634,3 +800,62 @@ class TPLinkConfigFlow(ConfigFlow, domain=DOMAIN): errors=errors, description_placeholders=placeholders, ) + + async def async_step_reconfigure( + self, user_input: dict[str, Any] | None = None + ) -> ConfigFlowResult: + """Trigger a reconfiguration flow.""" + errors: dict[str, str] = {} + placeholders: dict[str, str] = {} + + reconfigure_entry = self._get_reconfigure_entry() + assert reconfigure_entry.unique_id + await self.async_set_unique_id(reconfigure_entry.unique_id) + + host = reconfigure_entry.data[CONF_HOST] + port = reconfigure_entry.data.get(CONF_PORT) + + if user_input is not None: + host, port = self._async_get_host_port(host) + + self.host = host + credentials = await get_credentials(self.hass) + try: + device = await self._async_try_discover_and_update( + host, + credentials, + raise_on_progress=False, + raise_on_timeout=False, + port=port, + ) or await self._async_try_connect_all( + host, + credentials=credentials, + raise_on_progress=False, + port=port, + ) + except AuthenticationError: # Error from the update() + return await self.async_step_user_auth_confirm() + except KasaException as ex: + errors["base"] = "cannot_connect" + placeholders["error"] = str(ex) + else: + if not device: + return await self.async_step_user_auth_confirm() + + if self._async_supports_camera_credentials(device): + return await self.async_step_camera_auth_confirm() + + return self._async_create_or_update_entry_from_device(device) + + return self.async_show_form( + step_id="reconfigure", + data_schema=self.add_suggested_values_to_schema( + STEP_RECONFIGURE_DATA_SCHEMA, + {CONF_HOST: f"{host}:{port}" if port else host}, + ), + errors=errors, + description_placeholders={ + **placeholders, + CONF_MAC: reconfigure_entry.unique_id, + }, + ) diff --git a/homeassistant/components/tplink/const.py b/homeassistant/components/tplink/const.py index 28e4b04bcf9..61c1bf1cb9b 100644 --- a/homeassistant/components/tplink/const.py +++ b/homeassistant/components/tplink/const.py @@ -24,12 +24,15 @@ CONF_CREDENTIALS_HASH: Final = "credentials_hash" CONF_CONNECTION_PARAMETERS: Final = "connection_parameters" CONF_USES_HTTP: Final = "uses_http" CONF_AES_KEYS: Final = "aes_keys" +CONF_CAMERA_CREDENTIALS = "camera_credentials" +CONF_LIVE_VIEW = "live_view" CONF_CONFIG_ENTRY_MINOR_VERSION: Final = 5 PLATFORMS: Final = [ Platform.BINARY_SENSOR, Platform.BUTTON, + Platform.CAMERA, Platform.CLIMATE, Platform.FAN, Platform.LIGHT, diff --git a/homeassistant/components/tplink/entity.py b/homeassistant/components/tplink/entity.py index ef9e2ad5eee..60d066012a2 100644 --- a/homeassistant/components/tplink/entity.py +++ b/homeassistant/components/tplink/entity.py @@ -73,6 +73,13 @@ EXCLUDED_FEATURES = { "check_latest_firmware", # siren "alarm", + # camera + "pan_left", + "pan_right", + "pan_step", + "tilt_up", + "tilt_down", + "tilt_step", } @@ -91,6 +98,13 @@ class TPLinkFeatureEntityDescription(EntityDescription): deprecated_info: DeprecatedInfo | None = None +@dataclass(frozen=True, kw_only=True) +class TPLinkModuleEntityDescription(EntityDescription): + """Base class for a TPLink module based entity description.""" + + deprecated_info: DeprecatedInfo | None = None + + def async_refresh_after[_T: CoordinatedTPLinkEntity, **_P]( func: Callable[Concatenate[_T, _P], Awaitable[None]], ) -> Callable[Concatenate[_T, _P], Coroutine[Any, Any, None]]: diff --git a/homeassistant/components/tplink/manifest.json b/homeassistant/components/tplink/manifest.json index 65061882027..7797f0a36a3 100644 --- a/homeassistant/components/tplink/manifest.json +++ b/homeassistant/components/tplink/manifest.json @@ -3,7 +3,7 @@ "name": "TP-Link Smart Home", "codeowners": ["@rytilahti", "@bdraco", "@sdb9696"], "config_flow": true, - "dependencies": ["network"], + "dependencies": ["network", "ffmpeg", "stream"], "dhcp": [ { "registered_devices": true diff --git a/homeassistant/components/tplink/models.py b/homeassistant/components/tplink/models.py index ced58d3d21f..389260a388b 100644 --- a/homeassistant/components/tplink/models.py +++ b/homeassistant/components/tplink/models.py @@ -4,6 +4,8 @@ from __future__ import annotations from dataclasses import dataclass +from kasa import Credentials + from .coordinator import TPLinkDataUpdateCoordinator @@ -13,3 +15,5 @@ class TPLinkData: parent_coordinator: TPLinkDataUpdateCoordinator children_coordinators: list[TPLinkDataUpdateCoordinator] + camera_credentials: Credentials | None + live_view: bool | None diff --git a/homeassistant/components/tplink/strings.json b/homeassistant/components/tplink/strings.json index 8e5118c2720..7443636c3c0 100644 --- a/homeassistant/components/tplink/strings.json +++ b/homeassistant/components/tplink/strings.json @@ -42,16 +42,36 @@ "username": "[%key:common::config_flow::data::username%]", "password": "[%key:common::config_flow::data::password%]" } + }, + "reconfigure": { + "title": "Reconfigure TPLink entry", + "description": "Update your configuration for device {mac}", + "data": { + "host": "[%key:common::config_flow::data::host%]" + } + }, + "camera_auth_confirm": { + "title": "Set camera account credentials", + "description": "Input device camera account credentials. Leave blank if they are the same as your TPLink cloud credentials.", + "data": { + "live_view": "Enable camera live view", + "username": "[%key:common::config_flow::data::username%]", + "password": "[%key:common::config_flow::data::password%]" + } } }, "error": { "cannot_connect": "Connection error: {error}", - "invalid_auth": "Invalid authentication: {error}" + "invalid_auth": "Unable to authenticate: {error}", + "invalid_camera_auth": "Camera stream authentication failed", + "cannot_connect_camera": "Unable to access the camera stream, verify that you have set up the camera account: {error}", + "camera_creds": "You have to set both username and password" }, "abort": { "already_configured": "[%key:common::config_flow::abort::already_configured_device%]", "no_devices_found": "[%key:common::config_flow::abort::no_devices_found%]", "reauth_successful": "[%key:common::config_flow::abort::reauth_successful%]", + "reconfigure_successful": "[%key:common::config_flow::abort::reconfigure_successful%]", "cannot_connect": "[%key:common::config_flow::error::cannot_connect%]" } }, @@ -102,6 +122,11 @@ "name": "Stop alarm" } }, + "camera": { + "live_view": { + "name": "Live view" + } + }, "select": { "light_preset": { "name": "Light preset" diff --git a/tests/components/tplink/__init__.py b/tests/components/tplink/__init__.py index fdef5c35bfa..e322cf9f5de 100644 --- a/tests/components/tplink/__init__.py +++ b/tests/components/tplink/__init__.py @@ -1,6 +1,7 @@ """Tests for the TP-Link component.""" from collections import namedtuple +from dataclasses import replace from datetime import datetime from typing import Any from unittest.mock import AsyncMock, MagicMock, patch @@ -19,15 +20,18 @@ from kasa import ( ) from kasa.interfaces import Fan, Light, LightEffect, LightState from kasa.smart.modules.alarm import Alarm +from kasa.smartcam.modules.camera import LOCAL_STREAMING_PORT, Camera from syrupy import SnapshotAssertion from homeassistant.components.automation import DOMAIN as AUTOMATION_DOMAIN from homeassistant.components.tplink import ( CONF_AES_KEYS, CONF_ALIAS, + CONF_CAMERA_CREDENTIALS, CONF_CONNECTION_PARAMETERS, CONF_CREDENTIALS_HASH, CONF_HOST, + CONF_LIVE_VIEW, CONF_MODEL, CONF_USES_HTTP, Credentials, @@ -49,14 +53,19 @@ MODULE = "homeassistant.components.tplink" MODULE_CONFIG_FLOW = "homeassistant.components.tplink.config_flow" IP_ADDRESS = "127.0.0.1" IP_ADDRESS2 = "127.0.0.2" +IP_ADDRESS3 = "127.0.0.3" ALIAS = "My Bulb" +ALIAS_CAMERA = "My Camera" MODEL = "HS100" +MODEL_CAMERA = "C210" MAC_ADDRESS = "aa:bb:cc:dd:ee:ff" DEVICE_ID = "123456789ABCDEFGH" DEVICE_ID_MAC = "AA:BB:CC:DD:EE:FF" DHCP_FORMATTED_MAC_ADDRESS = MAC_ADDRESS.replace(":", "") MAC_ADDRESS2 = "11:22:33:44:55:66" +MAC_ADDRESS3 = "66:55:44:33:22:11" DEFAULT_ENTRY_TITLE = f"{ALIAS} {MODEL}" +DEFAULT_ENTRY_TITLE_CAMERA = f"{ALIAS_CAMERA} {MODEL_CAMERA}" CREDENTIALS_HASH_LEGACY = "" CONN_PARAMS_LEGACY = DeviceConnectionParameters( DeviceFamily.IotSmartPlugSwitch, DeviceEncryptionType.Xor @@ -80,7 +89,26 @@ DEVICE_CONFIG_KLAP = DeviceConfig( CONN_PARAMS_AES = DeviceConnectionParameters( DeviceFamily.SmartTapoPlug, DeviceEncryptionType.Aes ) -AES_KEYS = {"private": "foo", "public": "bar"} +_test_privkey = ( + "MIICdwIBADANBgkqhkiG9w0BAQEFAASCAmEwggJdAgEAAoGBAKLJKmBWGj6WYo9sewI8vkqar" + "Ed5H1JUr8Jj/LEWLTtV6+Mm4mfyEk6YKFHSmIG4AGgrVsGK/EbEkTZk9CwtixNQpBVc36oN2R" + "vuWWV38YnP4vI63mNxTA/gQonCsahjN4HfwE87pM7O5z39aeunoYm6Be663t33DbJH1ZUbZjm" + "tAgMBAAECgYB1Bn1KaFvRprcQOIJt51E9vNghQbf8rhj0fIEKpdC6mVhNIoUdCO+URNqnh+hP" + "SQIx4QYreUlHbsSeABFxOQSDJm6/kqyQsp59nCVDo/bXTtlvcSJ/sU3riqJNxYqEU1iJ0xMvU" + "N1VKKTmik89J8e5sN9R0AFfUSJIk7MpdOoD2QJBANTbV27nenyvbqee/ul4frdt2rrPGcGpcV" + "QmY87qbbrZgqgL5LMHHD7T/v/I8D1wRog1sBz/AiZGcnv/ox8dHKsCQQDDx8DCGPySSVqKVua" + "yUkBNpglN83wiCXZjyEtWIt+aB1A2n5ektE/o8oHnnOuvMdooxvtid7Mdapi2VLHV7VMHAkAE" + "d0GjWwnv2cJpk+VnQpbuBEkFiFjS/loZWODZM4Pv2qZqHi3DL9AA5XPBLBcWQufH7dBvG06RP" + "QMj5N4oRfUXAkEAuJJkVliqHNvM4OkGewzyFII4+WVYHNqg43dcFuuvtA27AJQ6qYtYXrvp3k" + "phI3yzOIhHTNCea1goepSkR5ODFwJBAJCTRbB+P47aEr/xA51ZFHE6VefDBJG9yg6yK4jcOxg" + "5ficXEpx8442okNtlzwa+QHpm/L3JOFrHwiEeVqXtiqY=" +) +_test_pubkey = ( + "MIGfMA0GCSqGSIb3DQEBAQUAA4GNADCBiQKBgQCiySpgVho+lmKPbHsCPL5KmqxHeR9SVK/CY" + "/yxFi07VevjJuJn8hJOmChR0piBuABoK1bBivxGxJE2ZPQsLYsTUKQVXN+qDdkb7llld/GJz+" + "LyOt5jcUwP4EKJwrGoYzeB38BPO6TOzuc9/Wnrp6GJugXuut7d9w2yR9WVG2Y5rQIDAQAB" +) +AES_KEYS = {"private": _test_privkey, "public": _test_pubkey} DEVICE_CONFIG_AES = DeviceConfig( IP_ADDRESS2, credentials=CREDENTIALS, @@ -88,6 +116,16 @@ DEVICE_CONFIG_AES = DeviceConfig( uses_http=True, aes_keys=AES_KEYS, ) +CONN_PARAMS_AES_CAMERA = DeviceConnectionParameters( + DeviceFamily.SmartIpCamera, DeviceEncryptionType.Aes, https=True, login_version=2 +) +DEVICE_CONFIG_AES_CAMERA = DeviceConfig( + IP_ADDRESS3, + credentials=CREDENTIALS, + connection_type=CONN_PARAMS_AES_CAMERA, + uses_http=True, +) + DEVICE_CONFIG_DICT_KLAP = { k: v for k, v in DEVICE_CONFIG_KLAP.to_dict().items() if k != "credentials" } @@ -119,6 +157,22 @@ CREATE_ENTRY_DATA_AES = { CONF_USES_HTTP: True, CONF_AES_KEYS: AES_KEYS, } +CREATE_ENTRY_DATA_AES_CAMERA = { + CONF_HOST: IP_ADDRESS3, + CONF_ALIAS: ALIAS_CAMERA, + CONF_MODEL: MODEL_CAMERA, + CONF_CREDENTIALS_HASH: CREDENTIALS_HASH_AES, + CONF_CONNECTION_PARAMETERS: CONN_PARAMS_AES_CAMERA.to_dict(), + CONF_USES_HTTP: True, + CONF_LIVE_VIEW: True, + CONF_CAMERA_CREDENTIALS: {"username": "camuser", "password": "campass"}, +} +SMALLEST_VALID_JPEG = ( + "ffd8ffe000104a46494600010101004800480000ffdb00430003020202020203020202030303030406040404040408060" + "6050609080a0a090809090a0c0f0c0a0b0e0b09090d110d0e0f101011100a0c12131210130f101010ffc9000b08000100" + "0101011100ffcc000600101005ffda0008010100003f00d2cf20ffd9" +) +SMALLEST_VALID_JPEG_BYTES = bytes.fromhex(SMALLEST_VALID_JPEG) def _load_feature_fixtures(): @@ -245,6 +299,9 @@ def _mocked_device( device.modules = {} device.features = {} + # replace device_config to prevent changes affecting between tests + device_config = replace(device_config) + if not ip_address: ip_address = IP_ADDRESS else: @@ -429,6 +486,17 @@ def _mocked_alarm_module(device): return alarm +def _mocked_camera_module(device): + camera = MagicMock(auto_spec=Camera, name="Mocked camera") + camera.is_on = True + camera.set_state = AsyncMock() + camera.stream_rtsp_url.return_value = ( + f"rtsp://user:pass@{device.host}:{LOCAL_STREAMING_PORT}/stream1" + ) + + return camera + + def _mocked_strip_children(features=None, alias=None) -> list[Device]: plug0 = _mocked_device( alias="Plug0" if alias is None else alias, @@ -496,6 +564,7 @@ MODULE_TO_MOCK_GEN = { Module.LightEffect: _mocked_light_effect_module, Module.Fan: _mocked_fan_module, Module.Alarm: _mocked_alarm_module, + Module.Camera: _mocked_camera_module, } diff --git a/tests/components/tplink/conftest.py b/tests/components/tplink/conftest.py index 25a4bd20270..f1bbb80b80c 100644 --- a/tests/components/tplink/conftest.py +++ b/tests/components/tplink/conftest.py @@ -1,30 +1,73 @@ """tplink conftest.""" from collections.abc import Generator +from contextlib import contextmanager from unittest.mock import DEFAULT, AsyncMock, patch -from kasa import DeviceConfig +from kasa import DeviceConfig, Module import pytest from homeassistant.components.tplink import DOMAIN from homeassistant.core import HomeAssistant from . import ( + ALIAS_CAMERA, + CREATE_ENTRY_DATA_AES_CAMERA, CREATE_ENTRY_DATA_LEGACY, CREDENTIALS_HASH_AES, CREDENTIALS_HASH_KLAP, DEVICE_CONFIG_AES, + DEVICE_CONFIG_AES_CAMERA, DEVICE_CONFIG_KLAP, IP_ADDRESS, IP_ADDRESS2, + IP_ADDRESS3, MAC_ADDRESS, MAC_ADDRESS2, + MAC_ADDRESS3, + MODEL_CAMERA, _mocked_device, ) from tests.common import MockConfigEntry +@contextmanager +def override_side_effect(mock: AsyncMock, effect): + """Temporarily override a mock side effect and replace afterwards.""" + try: + default_side_effect = mock.side_effect + mock.side_effect = effect + yield mock + finally: + mock.side_effect = default_side_effect + + +def _get_mock_devices(): + return { + IP_ADDRESS: _mocked_device( + device_config=DeviceConfig.from_dict(DEVICE_CONFIG_KLAP.to_dict()), + credentials_hash=CREDENTIALS_HASH_KLAP, + ip_address=IP_ADDRESS, + ), + IP_ADDRESS2: _mocked_device( + device_config=DeviceConfig.from_dict(DEVICE_CONFIG_AES.to_dict()), + credentials_hash=CREDENTIALS_HASH_AES, + mac=MAC_ADDRESS2, + ip_address=IP_ADDRESS2, + ), + IP_ADDRESS3: _mocked_device( + device_config=DeviceConfig.from_dict(DEVICE_CONFIG_AES_CAMERA.to_dict()), + credentials_hash=CREDENTIALS_HASH_AES, + mac=MAC_ADDRESS3, + ip_address=IP_ADDRESS3, + modules=[Module.Camera], + alias=ALIAS_CAMERA, + model=MODEL_CAMERA, + ), + } + + @pytest.fixture def mock_discovery(): """Mock python-kasa discovery.""" @@ -34,22 +77,15 @@ def mock_discovery(): discover_single=DEFAULT, try_connect_all=DEFAULT, ) as mock_discovery: - device = _mocked_device( - device_config=DeviceConfig.from_dict(DEVICE_CONFIG_KLAP.to_dict()), - credentials_hash=CREDENTIALS_HASH_KLAP, - alias="My Bulb", - ) - devices = { - "127.0.0.1": _mocked_device( - device_config=DeviceConfig.from_dict(DEVICE_CONFIG_KLAP.to_dict()), - credentials_hash=CREDENTIALS_HASH_KLAP, - alias=None, - ) - } + devices = _get_mock_devices() + + def get_device(host, **kwargs): + return devices[host] + mock_discovery["discover"].return_value = devices - mock_discovery["discover_single"].return_value = device - mock_discovery["try_connect_all"].return_value = device - mock_discovery["mock_device"] = device + mock_discovery["discover_single"].side_effect = get_device + mock_discovery["try_connect_all"].side_effect = get_device + mock_discovery["mock_devices"] = devices yield mock_discovery @@ -57,22 +93,9 @@ def mock_discovery(): def mock_connect(): """Mock python-kasa connect.""" with patch("homeassistant.components.tplink.Device.connect") as mock_connect: - devices = { - IP_ADDRESS: _mocked_device( - device_config=DeviceConfig.from_dict(DEVICE_CONFIG_KLAP.to_dict()), - credentials_hash=CREDENTIALS_HASH_KLAP, - ip_address=IP_ADDRESS, - ), - IP_ADDRESS2: _mocked_device( - device_config=DeviceConfig.from_dict(DEVICE_CONFIG_AES.to_dict()), - credentials_hash=CREDENTIALS_HASH_AES, - mac=MAC_ADDRESS2, - ip_address=IP_ADDRESS2, - ), - } + devices = _get_mock_devices() def get_device(config): - nonlocal devices return devices[config.host] mock_connect.side_effect = get_device @@ -117,6 +140,17 @@ def mock_config_entry() -> MockConfigEntry: ) +@pytest.fixture +def mock_camera_config_entry() -> MockConfigEntry: + """Mock camera ConfigEntry.""" + return MockConfigEntry( + title="TPLink", + domain=DOMAIN, + data={**CREATE_ENTRY_DATA_AES_CAMERA}, + unique_id=MAC_ADDRESS3, + ) + + @pytest.fixture async def mock_added_config_entry( hass: HomeAssistant, diff --git a/tests/components/tplink/fixtures/features.json b/tests/components/tplink/fixtures/features.json index d822bfc9b57..a54edf56c62 100644 --- a/tests/components/tplink/fixtures/features.json +++ b/tests/components/tplink/fixtures/features.json @@ -320,5 +320,35 @@ "type": "Sensor", "category": "Info", "value": "2024-06-24 10:03:11.046643+01:00" + }, + "pan_left": { + "value": "", + "type": "Action", + "category": "Config" + }, + "pan_right": { + "value": "", + "type": "Action", + "category": "Config" + }, + "pan_step": { + "value": 10, + "type": "Number", + "category": "Config" + }, + "tilt_up": { + "value": "", + "type": "Action", + "category": "Config" + }, + "tilt_down": { + "value": "", + "type": "Action", + "category": "Config" + }, + "tilt_step": { + "value": 10, + "type": "Number", + "category": "Config" } } diff --git a/tests/components/tplink/snapshots/test_camera.ambr b/tests/components/tplink/snapshots/test_camera.ambr new file mode 100644 index 00000000000..4ce1813d704 --- /dev/null +++ b/tests/components/tplink/snapshots/test_camera.ambr @@ -0,0 +1,87 @@ +# serializer version: 1 +# name: test_states[camera.my_camera_live_view-entry] + EntityRegistryEntrySnapshot({ + 'aliases': set({ + }), + 'area_id': None, + 'capabilities': None, + 'config_entry_id': , + 'device_class': None, + 'device_id': , + 'disabled_by': None, + 'domain': 'camera', + 'entity_category': None, + 'entity_id': 'camera.my_camera_live_view', + 'has_entity_name': True, + 'hidden_by': None, + 'icon': None, + 'id': , + 'labels': set({ + }), + 'name': None, + 'options': dict({ + }), + 'original_device_class': None, + 'original_icon': None, + 'original_name': 'Live view', + 'platform': 'tplink', + 'previous_unique_id': None, + 'supported_features': , + 'translation_key': 'live_view', + 'unique_id': "123456789ABCDEFGH-TPLinkCameraEntityDescription(key='live_view', device_class=None, entity_category=None, entity_registry_enabled_default=True, entity_registry_visible_default=True, force_update=False, icon=None, has_entity_name=False, name=, translation_key='live_view', translation_placeholders=None, unit_of_measurement=None, deprecated_info=None)", + 'unit_of_measurement': None, + }) +# --- +# name: test_states[camera.my_camera_live_view-state] + StateSnapshot({ + 'attributes': ReadOnlyDict({ + 'access_token': '1caab5c3b3', + 'entity_picture': '/api/camera_proxy/camera.my_camera_live_view?token=1caab5c3b3', + 'friendly_name': 'my_camera Live view', + 'frontend_stream_type': , + 'supported_features': , + }), + 'context': , + 'entity_id': 'camera.my_camera_live_view', + 'last_changed': , + 'last_reported': , + 'last_updated': , + 'state': 'idle', + }) +# --- +# name: test_states[my_camera-entry] + DeviceRegistryEntrySnapshot({ + 'area_id': None, + 'config_entries': , + 'configuration_url': None, + 'connections': set({ + tuple( + 'mac', + '66:55:44:33:22:11', + ), + }), + 'disabled_by': None, + 'entry_type': None, + 'hw_version': '1.0.0', + 'id': , + 'identifiers': set({ + tuple( + 'tplink', + '123456789ABCDEFGH', + ), + }), + 'is_new': False, + 'labels': set({ + }), + 'manufacturer': 'TP-Link', + 'model': 'HS100', + 'model_id': None, + 'name': 'my_camera', + 'name_by_user': None, + 'primary_config_entry': , + 'serial_number': None, + 'suggested_area': None, + 'sw_version': '1.0.0', + 'via_device_id': None, + }) +# --- diff --git a/tests/components/tplink/test_camera.py b/tests/components/tplink/test_camera.py new file mode 100644 index 00000000000..d8b0f82e32a --- /dev/null +++ b/tests/components/tplink/test_camera.py @@ -0,0 +1,431 @@ +"""The tests for the tplink camera platform.""" + +import asyncio +from unittest.mock import AsyncMock, patch + +from aiohttp.test_utils import make_mocked_request +from freezegun.api import FrozenDateTimeFactory +from kasa import Module +import pytest +from syrupy.assertion import SnapshotAssertion + +from homeassistant.components import stream +from homeassistant.components.camera import ( + CameraEntityFeature, + StreamType, + async_get_image, + async_get_mjpeg_stream, + get_camera_from_entity_id, +) +from homeassistant.components.tplink.camera import TPLinkCameraEntity +from homeassistant.components.websocket_api import TYPE_RESULT +from homeassistant.const import Platform +from homeassistant.core import HomeAssistant, HomeAssistantError +from homeassistant.helpers import device_registry as dr, entity_registry as er + +from . import ( + IP_ADDRESS3, + MAC_ADDRESS3, + SMALLEST_VALID_JPEG_BYTES, + _mocked_device, + setup_platform_for_device, + snapshot_platform, +) + +from tests.common import MockConfigEntry, async_fire_time_changed +from tests.typing import WebSocketGenerator + + +async def test_states( + hass: HomeAssistant, + mock_camera_config_entry: MockConfigEntry, + entity_registry: er.EntityRegistry, + device_registry: dr.DeviceRegistry, + snapshot: SnapshotAssertion, +) -> None: + """Test states.""" + mock_camera_config_entry.add_to_hass(hass) + + mock_device = _mocked_device( + modules=[Module.Camera], + alias="my_camera", + ip_address=IP_ADDRESS3, + mac=MAC_ADDRESS3, + ) + + # Patch getrandbits so the access_token doesn't change on camera attributes + with patch("random.SystemRandom.getrandbits", return_value=123123123123): + await setup_platform_for_device( + hass, mock_camera_config_entry, Platform.CAMERA, mock_device + ) + + await snapshot_platform( + hass, + entity_registry, + device_registry, + snapshot, + mock_camera_config_entry.entry_id, + ) + + +async def test_handle_mjpeg_stream( + hass: HomeAssistant, + mock_camera_config_entry: MockConfigEntry, + freezer: FrozenDateTimeFactory, +) -> None: + """Test handle_async_mjpeg_stream.""" + mock_device = _mocked_device( + modules=[Module.Camera], + alias="my_camera", + ip_address=IP_ADDRESS3, + mac=MAC_ADDRESS3, + ) + + await setup_platform_for_device( + hass, mock_camera_config_entry, Platform.CAMERA, mock_device + ) + + state = hass.states.get("camera.my_camera_live_view") + assert state is not None + + mock_request = make_mocked_request("GET", "/", headers={"token": "x"}) + stream = await async_get_mjpeg_stream( + hass, mock_request, "camera.my_camera_live_view" + ) + assert stream is not None + + +async def test_handle_mjpeg_stream_not_supported( + hass: HomeAssistant, + mock_camera_config_entry: MockConfigEntry, + freezer: FrozenDateTimeFactory, +) -> None: + """Test handle_async_mjpeg_stream.""" + mock_device = _mocked_device( + modules=[Module.Camera], + alias="my_camera", + ip_address=IP_ADDRESS3, + mac=MAC_ADDRESS3, + ) + mock_camera = mock_device.modules[Module.Camera] + + mock_camera.stream_rtsp_url.return_value = None + + await setup_platform_for_device( + hass, mock_camera_config_entry, Platform.CAMERA, mock_device + ) + + mock_request = make_mocked_request("GET", "/", headers={"token": "x"}) + stream = await async_get_mjpeg_stream( + hass, mock_request, "camera.my_camera_live_view" + ) + assert stream is None + + +async def test_camera_image( + hass: HomeAssistant, + mock_camera_config_entry: MockConfigEntry, + freezer: FrozenDateTimeFactory, + caplog: pytest.LogCaptureFixture, +) -> None: + """Test async_get_image.""" + mock_device = _mocked_device( + modules=[Module.Camera], + alias="my_camera", + ip_address=IP_ADDRESS3, + mac=MAC_ADDRESS3, + ) + + await setup_platform_for_device( + hass, mock_camera_config_entry, Platform.CAMERA, mock_device + ) + + state = hass.states.get("camera.my_camera_live_view") + assert state is not None + + with patch( + "homeassistant.components.ffmpeg.async_get_image", + return_value=SMALLEST_VALID_JPEG_BYTES, + ) as mock_get_image: + image = await async_get_image(hass, "camera.my_camera_live_view") + assert image + assert image.content == SMALLEST_VALID_JPEG_BYTES + mock_get_image.assert_called_once() + + mock_get_image.reset_mock() + image = await async_get_image(hass, "camera.my_camera_live_view") + mock_get_image.assert_not_called() + + freezer.tick(TPLinkCameraEntity.IMAGE_INTERVAL) + async_fire_time_changed(hass) + await hass.async_block_till_done(wait_background_tasks=True) + + mock_get_image.reset_mock() + image = await async_get_image(hass, "camera.my_camera_live_view") + mock_get_image.assert_called_once() + + freezer.tick(TPLinkCameraEntity.IMAGE_INTERVAL) + async_fire_time_changed(hass) + await hass.async_block_till_done(wait_background_tasks=True) + + # Test image returns None + with patch( + "homeassistant.components.ffmpeg.async_get_image", + return_value=None, + ) as mock_get_image: + msg = f"None camera image returned for {IP_ADDRESS3}" + assert msg not in caplog.text + + mock_get_image.reset_mock() + image = await async_get_image(hass, "camera.my_camera_live_view") + mock_get_image.assert_called_once() + + assert msg in caplog.text + + +async def test_no_camera_image_when_streaming( + hass: HomeAssistant, + mock_camera_config_entry: MockConfigEntry, + freezer: FrozenDateTimeFactory, +) -> None: + """Test async_get_image.""" + mock_device = _mocked_device( + modules=[Module.Camera], + alias="my_camera", + ip_address=IP_ADDRESS3, + mac=MAC_ADDRESS3, + ) + + await setup_platform_for_device( + hass, mock_camera_config_entry, Platform.CAMERA, mock_device + ) + + state = hass.states.get("camera.my_camera_live_view") + assert state is not None + + with patch( + "homeassistant.components.ffmpeg.async_get_image", + return_value=SMALLEST_VALID_JPEG_BYTES, + ) as mock_get_image: + await async_get_image(hass, "camera.my_camera_live_view") + mock_get_image.assert_called_once() + + freezer.tick(TPLinkCameraEntity.IMAGE_INTERVAL) + async_fire_time_changed(hass) + await hass.async_block_till_done(wait_background_tasks=True) + + start_event = asyncio.Event() + finish_event = asyncio.Event() + + async def _waiter(*_, **__): + start_event.set() + await finish_event.wait() + + async def _get_stream(): + mock_request = make_mocked_request("GET", "/", headers={"token": "x"}) + await async_get_mjpeg_stream( + hass, mock_request, "camera.my_camera_live_view" + ) + + mock_get_image.reset_mock() + with patch( + "homeassistant.components.tplink.camera.async_aiohttp_proxy_stream", + new=_waiter, + ): + task = asyncio.create_task(_get_stream()) + await start_event.wait() + await async_get_image(hass, "camera.my_camera_live_view") + finish_event.set() + await task + + mock_get_image.assert_not_called() + + +async def test_no_concurrent_camera_image( + hass: HomeAssistant, + mock_camera_config_entry: MockConfigEntry, + freezer: FrozenDateTimeFactory, +) -> None: + """Test async_get_image.""" + mock_device = _mocked_device( + modules=[Module.Camera], + alias="my_camera", + ip_address=IP_ADDRESS3, + mac=MAC_ADDRESS3, + ) + + await setup_platform_for_device( + hass, mock_camera_config_entry, Platform.CAMERA, mock_device + ) + + state = hass.states.get("camera.my_camera_live_view") + assert state is not None + + finish_event = asyncio.Event() + call_count = 0 + + async def _waiter(*_, **__): + nonlocal call_count + call_count += 1 + await finish_event.wait() + return SMALLEST_VALID_JPEG_BYTES + + with patch( + "homeassistant.components.ffmpeg.async_get_image", + new=_waiter, + ): + tasks = asyncio.gather( + async_get_image(hass, "camera.my_camera_live_view"), + async_get_image(hass, "camera.my_camera_live_view"), + ) + # Sleep to give both tasks chance to get to th asyncio.Lock() + await asyncio.sleep(0) + finish_event.set() + results = await tasks + assert len(results) == 2 + assert all(img and img.content == SMALLEST_VALID_JPEG_BYTES for img in results) + assert call_count == 1 + + +async def test_camera_image_auth_error( + hass: HomeAssistant, + mock_camera_config_entry: MockConfigEntry, + mock_connect: AsyncMock, + mock_discovery: AsyncMock, +) -> None: + """Test async_get_image.""" + mock_device = _mocked_device( + modules=[Module.Camera], + alias="my_camera", + ip_address=IP_ADDRESS3, + mac=MAC_ADDRESS3, + ) + + await setup_platform_for_device( + hass, mock_camera_config_entry, Platform.CAMERA, mock_device + ) + + state = hass.states.get("camera.my_camera_live_view") + assert state is not None + flows = hass.config_entries.flow.async_progress() + assert len(flows) == 0 + + with ( + patch( + "homeassistant.components.ffmpeg.async_get_image", + return_value=b"", + ), + patch( + "homeassistant.components.stream.async_check_stream_client_error", + side_effect=stream.StreamOpenClientError( + stream_client_error=stream.StreamClientError.Unauthorized + ), + ), + pytest.raises(HomeAssistantError), + ): + await async_get_image(hass, "camera.my_camera_live_view") + await hass.async_block_till_done() + + flows = hass.config_entries.flow.async_progress() + assert len(flows) == 1 + [result] = flows + + assert result["step_id"] == "camera_auth_confirm" + + +async def test_camera_stream_source( + hass: HomeAssistant, + mock_camera_config_entry: MockConfigEntry, + hass_ws_client: WebSocketGenerator, +) -> None: + """Test async_get_image. + + This test would fail if the integration didn't properly + put stream in the dependencies. + """ + mock_device = _mocked_device( + modules=[Module.Camera], + alias="my_camera", + ip_address=IP_ADDRESS3, + mac=MAC_ADDRESS3, + ) + + await setup_platform_for_device( + hass, mock_camera_config_entry, Platform.CAMERA, mock_device + ) + + state = hass.states.get("camera.my_camera_live_view") + assert state is not None + + client = await hass_ws_client(hass) + await client.send_json_auto_id( + {"type": "camera/stream", "entity_id": "camera.my_camera_live_view"} + ) + msg = await client.receive_json() + + # Assert WebSocket response + assert msg["type"] == TYPE_RESULT + assert msg["success"] + assert "url" in msg["result"] + + +async def test_camera_stream_attributes( + hass: HomeAssistant, + mock_camera_config_entry: MockConfigEntry, +) -> None: + """Test stream attributes.""" + mock_device = _mocked_device( + modules=[Module.Camera], + alias="my_camera", + ip_address=IP_ADDRESS3, + mac=MAC_ADDRESS3, + ) + + await setup_platform_for_device( + hass, mock_camera_config_entry, Platform.CAMERA, mock_device + ) + + state = hass.states.get("camera.my_camera_live_view") + assert state is not None + + supported_features = state.attributes.get("supported_features") + assert supported_features is CameraEntityFeature.STREAM | CameraEntityFeature.ON_OFF + camera = get_camera_from_entity_id(hass, "camera.my_camera_live_view") + assert camera.camera_capabilities.frontend_stream_types == {StreamType.HLS} + + +async def test_camera_turn_on_off( + hass: HomeAssistant, + mock_camera_config_entry: MockConfigEntry, +) -> None: + """Test camera turn on and off.""" + mock_device = _mocked_device( + modules=[Module.Camera], + alias="my_camera", + ip_address=IP_ADDRESS3, + mac=MAC_ADDRESS3, + ) + mock_camera = mock_device.modules[Module.Camera] + + await setup_platform_for_device( + hass, mock_camera_config_entry, Platform.CAMERA, mock_device + ) + + state = hass.states.get("camera.my_camera_live_view") + assert state is not None + + await hass.services.async_call( + "camera", + "turn_on", + {"entity_id": "camera.my_camera_live_view"}, + blocking=True, + ) + mock_camera.set_state.assert_called_with(True) + + await hass.services.async_call( + "camera", + "turn_off", + {"entity_id": "camera.my_camera_live_view"}, + blocking=True, + ) + mock_camera.set_state.assert_called_with(False) diff --git a/tests/components/tplink/test_config_flow.py b/tests/components/tplink/test_config_flow.py index 2697696c667..980fd0a3f51 100644 --- a/tests/components/tplink/test_config_flow.py +++ b/tests/components/tplink/test_config_flow.py @@ -1,14 +1,13 @@ """Test the tplink config flow.""" -from contextlib import contextmanager import logging from unittest.mock import ANY, AsyncMock, patch -from kasa import TimeoutError +from kasa import Module, TimeoutError import pytest from homeassistant import config_entries -from homeassistant.components import dhcp +from homeassistant.components import dhcp, stream from homeassistant.components.tplink import ( DOMAIN, AuthenticationError, @@ -19,9 +18,11 @@ from homeassistant.components.tplink import ( ) from homeassistant.components.tplink.config_flow import TPLinkConfigFlow from homeassistant.components.tplink.const import ( + CONF_CAMERA_CREDENTIALS, CONF_CONNECTION_PARAMETERS, CONF_CREDENTIALS_HASH, CONF_DEVICE_CONFIG, + CONF_LIVE_VIEW, ) from homeassistant.config_entries import SOURCE_REAUTH, ConfigEntryState from homeassistant.const import ( @@ -39,44 +40,43 @@ from homeassistant.data_entry_flow import FlowResultType from . import ( AES_KEYS, ALIAS, + ALIAS_CAMERA, CONN_PARAMS_AES, CONN_PARAMS_KLAP, CONN_PARAMS_LEGACY, CREATE_ENTRY_DATA_AES, + CREATE_ENTRY_DATA_AES_CAMERA, CREATE_ENTRY_DATA_KLAP, CREATE_ENTRY_DATA_LEGACY, CREDENTIALS_HASH_AES, CREDENTIALS_HASH_KLAP, DEFAULT_ENTRY_TITLE, + DEFAULT_ENTRY_TITLE_CAMERA, DEVICE_CONFIG_AES, + DEVICE_CONFIG_AES_CAMERA, DEVICE_CONFIG_DICT_KLAP, DEVICE_CONFIG_KLAP, DEVICE_CONFIG_LEGACY, DHCP_FORMATTED_MAC_ADDRESS, IP_ADDRESS, + IP_ADDRESS2, + IP_ADDRESS3, MAC_ADDRESS, MAC_ADDRESS2, + MAC_ADDRESS3, + MODEL_CAMERA, MODULE, + SMALLEST_VALID_JPEG_BYTES, _mocked_device, _patch_connect, _patch_discovery, _patch_single_discovery, ) +from .conftest import override_side_effect from tests.common import MockConfigEntry -@contextmanager -def override_side_effect(mock: AsyncMock, effect): - """Temporarily override a mock side effect and replace afterwards.""" - try: - default_side_effect = mock.side_effect - mock.side_effect = effect - yield mock - finally: - mock.side_effect = default_side_effect - - @pytest.mark.parametrize( ("device_config", "expected_entry_data", "credentials_hash"), [ @@ -98,6 +98,7 @@ async def test_discovery( device_config=device_config, credentials_hash=credentials_hash, ip_address=ip_address, + mac=MAC_ADDRESS, ) with ( _patch_discovery(device, ip_address=ip_address), @@ -143,7 +144,7 @@ async def test_discovery( result["flow_id"], {CONF_DEVICE: MAC_ADDRESS}, ) - await hass.async_block_till_done() + await hass.async_block_till_done(wait_background_tasks=True) assert result3["type"] is FlowResultType.CREATE_ENTRY assert result3["title"] == DEFAULT_ENTRY_TITLE @@ -167,13 +168,142 @@ async def test_discovery( assert result2["reason"] == "no_devices_found" +async def test_discovery_camera( + hass: HomeAssistant, mock_discovery: AsyncMock, mock_connect: AsyncMock, mock_init +) -> None: + """Test authenticated discovery for camera with stream.""" + mock_device = _mocked_device( + alias=ALIAS_CAMERA, + ip_address=IP_ADDRESS3, + mac=MAC_ADDRESS3, + model=MODEL_CAMERA, + device_config=DEVICE_CONFIG_AES_CAMERA, + credentials_hash=CREDENTIALS_HASH_AES, + modules=[Module.Camera], + ) + + with override_side_effect(mock_connect["connect"], lambda *_, **__: mock_device): + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": config_entries.SOURCE_INTEGRATION_DISCOVERY}, + data={ + CONF_HOST: IP_ADDRESS3, + CONF_MAC: MAC_ADDRESS3, + CONF_ALIAS: ALIAS, + CONF_DEVICE: mock_device, + }, + ) + await hass.async_block_till_done() + assert result["type"] is FlowResultType.FORM + assert result["step_id"] == "discovery_confirm" + assert not result["errors"] + + with override_side_effect(mock_connect["connect"], lambda *_, **__: mock_device): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={}, + ) + await hass.async_block_till_done() + assert result["type"] is FlowResultType.FORM + assert result["step_id"] == "camera_auth_confirm" + assert not result["errors"] + + with patch( + "homeassistant.components.stream.async_check_stream_client_error", + return_value=None, + ): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={ + CONF_LIVE_VIEW: True, + CONF_USERNAME: "camuser", + CONF_PASSWORD: "campass", + }, + ) + await hass.async_block_till_done() + + assert result["type"] is FlowResultType.CREATE_ENTRY + assert result["title"] == DEFAULT_ENTRY_TITLE_CAMERA + assert result["data"] == CREATE_ENTRY_DATA_AES_CAMERA + assert result["context"]["unique_id"] == MAC_ADDRESS3 + + +async def test_discovery_pick_device_camera( + hass: HomeAssistant, mock_discovery: AsyncMock, mock_connect: AsyncMock, mock_init +) -> None: + """Test authenticated discovery for camera with stream.""" + mock_device = _mocked_device( + alias=ALIAS_CAMERA, + ip_address=IP_ADDRESS3, + mac=MAC_ADDRESS3, + model=MODEL_CAMERA, + device_config=DEVICE_CONFIG_AES_CAMERA, + credentials_hash=CREDENTIALS_HASH_AES, + modules=[Module.Camera], + ) + + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": config_entries.SOURCE_USER} + ) + await hass.async_block_till_done() + assert result["type"] is FlowResultType.FORM + assert result["step_id"] == "user" + assert not result["errors"] + + with override_side_effect( + mock_discovery["discover"], lambda *_, **__: {IP_ADDRESS3: mock_device} + ): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={}, + ) + await hass.async_block_till_done() + assert result["type"] is FlowResultType.FORM + assert result["step_id"] == "pick_device" + assert not result["errors"] + + with override_side_effect(mock_connect["connect"], lambda *_, **__: mock_device): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={CONF_DEVICE: MAC_ADDRESS3}, + ) + await hass.async_block_till_done() + + assert result["type"] is FlowResultType.FORM + assert result["step_id"] == "camera_auth_confirm" + assert not result["errors"] + + with patch( + "homeassistant.components.stream.async_check_stream_client_error", + return_value=None, + ): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={ + CONF_LIVE_VIEW: True, + CONF_USERNAME: "camuser", + CONF_PASSWORD: "campass", + }, + ) + await hass.async_block_till_done() + + assert result["type"] is FlowResultType.CREATE_ENTRY + assert result["title"] == DEFAULT_ENTRY_TITLE_CAMERA + assert result["data"] == CREATE_ENTRY_DATA_AES_CAMERA + assert result["context"]["unique_id"] == MAC_ADDRESS3 + + async def test_discovery_auth( hass: HomeAssistant, mock_discovery: AsyncMock, mock_connect: AsyncMock, mock_init ) -> None: """Test authenticated discovery.""" - - mock_device = mock_connect["mock_devices"][IP_ADDRESS] - assert mock_device.config == DEVICE_CONFIG_KLAP + mock_device = _mocked_device( + alias=ALIAS, + ip_address=IP_ADDRESS, + mac=MAC_ADDRESS, + device_config=DEVICE_CONFIG_KLAP, + credentials_hash=CREDENTIALS_HASH_KLAP, + ) with override_side_effect(mock_connect["connect"], AuthenticationError): result = await hass.config_entries.flow.async_init( @@ -191,13 +321,14 @@ async def test_discovery_auth( assert result["step_id"] == "discovery_auth_confirm" assert not result["errors"] - result2 = await hass.config_entries.flow.async_configure( - result["flow_id"], - user_input={ - CONF_USERNAME: "fake_username", - CONF_PASSWORD: "fake_password", - }, - ) + with override_side_effect(mock_connect["connect"], lambda *_, **__: mock_device): + result2 = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={ + CONF_USERNAME: "fake_username", + CONF_PASSWORD: "fake_password", + }, + ) assert result2["type"] is FlowResultType.CREATE_ENTRY assert result2["title"] == DEFAULT_ENTRY_TITLE @@ -205,6 +336,69 @@ async def test_discovery_auth( assert result2["context"]["unique_id"] == MAC_ADDRESS +async def test_discovery_auth_camera( + hass: HomeAssistant, mock_discovery: AsyncMock, mock_connect: AsyncMock, mock_init +) -> None: + """Test authenticated discovery for camera with stream.""" + mock_device = _mocked_device( + alias=ALIAS_CAMERA, + ip_address=IP_ADDRESS3, + mac=MAC_ADDRESS3, + model=MODEL_CAMERA, + device_config=DEVICE_CONFIG_AES_CAMERA, + credentials_hash=CREDENTIALS_HASH_AES, + modules=[Module.Camera], + ) + + with override_side_effect(mock_connect["connect"], AuthenticationError): + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": config_entries.SOURCE_INTEGRATION_DISCOVERY}, + data={ + CONF_HOST: IP_ADDRESS3, + CONF_MAC: MAC_ADDRESS3, + CONF_ALIAS: ALIAS, + CONF_DEVICE: mock_device, + }, + ) + await hass.async_block_till_done() + assert result["type"] is FlowResultType.FORM + assert result["step_id"] == "discovery_auth_confirm" + assert not result["errors"] + + with override_side_effect(mock_connect["connect"], lambda *_, **__: mock_device): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={ + CONF_USERNAME: "fake_username", + CONF_PASSWORD: "fake_password", + }, + ) + await hass.async_block_till_done() + assert result["type"] is FlowResultType.FORM + assert result["step_id"] == "camera_auth_confirm" + assert not result["errors"] + + with patch( + "homeassistant.components.stream.async_check_stream_client_error", + return_value=None, + ): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={ + CONF_LIVE_VIEW: True, + CONF_USERNAME: "camuser", + CONF_PASSWORD: "campass", + }, + ) + await hass.async_block_till_done() + + assert result["type"] is FlowResultType.CREATE_ENTRY + assert result["title"] == DEFAULT_ENTRY_TITLE_CAMERA + assert result["data"] == CREATE_ENTRY_DATA_AES_CAMERA + assert result["context"]["unique_id"] == MAC_ADDRESS3 + + @pytest.mark.parametrize( ("error_type", "errors_msg", "error_placement"), [ @@ -385,7 +579,7 @@ async def test_discovery_new_credentials_invalid( async def test_discovery_with_existing_device_present(hass: HomeAssistant) -> None: """Test setting up discovery.""" config_entry = MockConfigEntry( - domain=DOMAIN, data={CONF_HOST: "127.0.0.2"}, unique_id="dd:dd:dd:dd:dd:dd" + domain=DOMAIN, data={CONF_HOST: IP_ADDRESS2}, unique_id="dd:dd:dd:dd:dd:dd" ) config_entry.add_to_hass(hass) @@ -535,6 +729,227 @@ async def test_manual(hass: HomeAssistant) -> None: assert result2["reason"] == "already_configured" +async def test_manual_camera( + hass: HomeAssistant, + mock_discovery: AsyncMock, + mock_connect: AsyncMock, +) -> None: + """Test manual camera.""" + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": config_entries.SOURCE_USER} + ) + assert result["type"] is FlowResultType.FORM + assert result["step_id"] == "user" + assert not result["errors"] + + result = await hass.config_entries.flow.async_configure( + result["flow_id"], {CONF_HOST: IP_ADDRESS3} + ) + await hass.async_block_till_done() + + assert result["type"] is FlowResultType.FORM + assert result["step_id"] == "camera_auth_confirm" + + # Test no username or pass + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={ + CONF_LIVE_VIEW: True, + CONF_USERNAME: "camuser", + }, + ) + await hass.async_block_till_done() + assert result["type"] is FlowResultType.FORM + assert result["step_id"] == "camera_auth_confirm" + assert result["errors"] == {"base": "camera_creds"} + + # Test unknown error + with ( + patch( + "homeassistant.components.stream.async_check_stream_client_error", + side_effect=stream.StreamOpenClientError( + stream_client_error=stream.StreamClientError.NotFound + ), + ), + ): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={ + CONF_LIVE_VIEW: True, + CONF_USERNAME: "camuser", + CONF_PASSWORD: "campass", + }, + ) + await hass.async_block_till_done() + assert result["type"] is FlowResultType.FORM + assert result["step_id"] == "camera_auth_confirm" + assert result["errors"] == {"base": "cannot_connect_camera"} + assert "error" in result["description_placeholders"] + + # Test unknown error + with ( + patch( + "homeassistant.components.stream.async_check_stream_client_error", + side_effect=stream.StreamOpenClientError( + stream_client_error=stream.StreamClientError.Unauthorized + ), + ), + ): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={ + CONF_LIVE_VIEW: True, + CONF_USERNAME: "camuser", + CONF_PASSWORD: "campass", + }, + ) + await hass.async_block_till_done() + assert result["type"] is FlowResultType.FORM + assert result["step_id"] == "camera_auth_confirm" + assert result["errors"] == {"base": "invalid_camera_auth"} + + with patch( + "homeassistant.components.stream.async_check_stream_client_error", + return_value=None, + ): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={ + CONF_LIVE_VIEW: True, + CONF_USERNAME: "camuser", + CONF_PASSWORD: "campass", + }, + ) + await hass.async_block_till_done() + + assert result["type"] is FlowResultType.CREATE_ENTRY + assert result["data"][CONF_CAMERA_CREDENTIALS] == { + CONF_USERNAME: "camuser", + CONF_PASSWORD: "campass", + } + assert result["data"][CONF_LIVE_VIEW] is True + + +@pytest.mark.parametrize( + "exception", + [ + pytest.param( + stream.StreamOpenClientError( + stream_client_error=stream.StreamClientError.NotFound + ), + id="open_client_error", + ), + pytest.param(Exception(), id="other_error"), + ], +) +async def test_manual_camera_no_hls( + hass: HomeAssistant, + mock_discovery: AsyncMock, + mock_connect: AsyncMock, + exception: Exception, +) -> None: + """Test manual camera when hls stream fails but mpeg stream works.""" + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": config_entries.SOURCE_USER} + ) + assert result["type"] is FlowResultType.FORM + assert result["step_id"] == "user" + assert not result["errors"] + + result = await hass.config_entries.flow.async_configure( + result["flow_id"], {CONF_HOST: IP_ADDRESS3} + ) + await hass.async_block_till_done() + + assert result["type"] is FlowResultType.FORM + assert result["step_id"] == "camera_auth_confirm" + + # Test stream error + with ( + patch( + "homeassistant.components.stream.async_check_stream_client_error", + side_effect=exception, + ), + patch("homeassistant.components.ffmpeg.async_get_image", return_value=None), + ): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={ + CONF_LIVE_VIEW: True, + CONF_USERNAME: "camuser", + CONF_PASSWORD: "campass", + }, + ) + await hass.async_block_till_done() + assert result["type"] is FlowResultType.FORM + assert result["step_id"] == "camera_auth_confirm" + assert result["errors"] == {"base": "cannot_connect_camera"} + assert "error" in result["description_placeholders"] + + # async_get_image will succeed + with ( + patch( + "homeassistant.components.stream.async_check_stream_client_error", + side_effect=exception, + ), + patch( + "homeassistant.components.ffmpeg.async_get_image", + return_value=SMALLEST_VALID_JPEG_BYTES, + ), + ): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={ + CONF_LIVE_VIEW: True, + CONF_USERNAME: "camuser", + CONF_PASSWORD: "campass", + }, + ) + await hass.async_block_till_done() + + assert result["type"] is FlowResultType.CREATE_ENTRY + assert result["data"][CONF_CAMERA_CREDENTIALS] == { + CONF_USERNAME: "camuser", + CONF_PASSWORD: "campass", + } + assert result["data"][CONF_LIVE_VIEW] is True + + +async def test_manual_camera_no_live_view( + hass: HomeAssistant, + mock_discovery: AsyncMock, + mock_connect: AsyncMock, +) -> None: + """Test manual camera.""" + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": config_entries.SOURCE_USER} + ) + assert result["type"] is FlowResultType.FORM + assert result["step_id"] == "user" + assert not result["errors"] + + result = await hass.config_entries.flow.async_configure( + result["flow_id"], {CONF_HOST: IP_ADDRESS3} + ) + await hass.async_block_till_done() + + assert result["type"] is FlowResultType.FORM + assert result["step_id"] == "camera_auth_confirm" + + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={ + CONF_LIVE_VIEW: False, + CONF_USERNAME: "camuser", + CONF_PASSWORD: "campass", + }, + ) + + assert result["type"] is FlowResultType.CREATE_ENTRY + assert CONF_CAMERA_CREDENTIALS not in result["data"] + assert result["data"][CONF_LIVE_VIEW] is False + + async def test_manual_no_capabilities(hass: HomeAssistant) -> None: """Test manually setup without successful get_capabilities.""" result = await hass.config_entries.flow.async_init( @@ -575,7 +990,7 @@ async def test_manual_auth( assert result["step_id"] == "user" assert not result["errors"] - mock_discovery["mock_device"].update.side_effect = AuthenticationError + mock_discovery["mock_devices"][IP_ADDRESS].update.side_effect = AuthenticationError result2 = await hass.config_entries.flow.async_configure( result["flow_id"], user_input={CONF_HOST: IP_ADDRESS} @@ -586,7 +1001,7 @@ async def test_manual_auth( assert result2["step_id"] == "user_auth_confirm" assert not result2["errors"] - mock_discovery["mock_device"].update.reset_mock(side_effect=True) + mock_discovery["mock_devices"][IP_ADDRESS].update.reset_mock(side_effect=True) result3 = await hass.config_entries.flow.async_configure( result2["flow_id"], @@ -602,6 +1017,63 @@ async def test_manual_auth( assert result3["context"]["unique_id"] == MAC_ADDRESS +async def test_manual_auth_camera( + hass: HomeAssistant, + mock_discovery: AsyncMock, + mock_connect: AsyncMock, +) -> None: + """Test manual camera.""" + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": config_entries.SOURCE_USER} + ) + assert result["type"] is FlowResultType.FORM + assert result["step_id"] == "user" + assert not result["errors"] + + with override_side_effect( + mock_discovery["mock_devices"][IP_ADDRESS3].update, AuthenticationError + ): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], {CONF_HOST: IP_ADDRESS3} + ) + await hass.async_block_till_done() + + assert result["type"] is FlowResultType.FORM + assert result["step_id"] == "user_auth_confirm" + + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={ + CONF_USERNAME: "foobar", + CONF_PASSWORD: "foobar", + }, + ) + await hass.async_block_till_done() + assert result["type"] is FlowResultType.FORM + assert result["step_id"] == "camera_auth_confirm" + + with patch( + "homeassistant.components.stream.async_check_stream_client_error", + return_value=None, + ): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={ + CONF_LIVE_VIEW: True, + CONF_USERNAME: "camuser", + CONF_PASSWORD: "campass", + }, + ) + await hass.async_block_till_done() + + assert result["type"] is FlowResultType.CREATE_ENTRY + assert result["data"][CONF_CAMERA_CREDENTIALS] == { + CONF_USERNAME: "camuser", + CONF_PASSWORD: "campass", + } + assert result["data"][CONF_LIVE_VIEW] is True + + @pytest.mark.parametrize( ("error_type", "errors_msg", "error_placement"), [ @@ -627,7 +1099,7 @@ async def test_manual_auth_errors( assert result["step_id"] == "user" assert not result["errors"] - mock_discovery["mock_device"].update.side_effect = AuthenticationError + mock_discovery["mock_devices"][IP_ADDRESS].update.side_effect = AuthenticationError with override_side_effect(mock_connect["connect"], error_type): result2 = await hass.config_entries.flow.async_configure( @@ -682,11 +1154,27 @@ async def test_manual_port_override( port, ) -> None: """Test manually setup.""" - mock_discovery["mock_device"].config.port_override = port - mock_discovery["mock_device"].host = host - result = await hass.config_entries.flow.async_init( - DOMAIN, context={"source": config_entries.SOURCE_USER} + config = DeviceConfig( + host, + credentials=None, + port_override=port, + uses_http=True, + connection_type=CONN_PARAMS_KLAP, ) + mock_device = _mocked_device( + alias=ALIAS, + ip_address=host, + mac=MAC_ADDRESS, + device_config=config, + credentials_hash=CREDENTIALS_HASH_KLAP, + ) + + with override_side_effect( + mock_discovery["try_connect_all"], lambda *_, **__: mock_device + ): + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": config_entries.SOURCE_USER} + ) assert result["type"] is FlowResultType.FORM assert result["step_id"] == "user" assert not result["errors"] @@ -696,23 +1184,29 @@ async def test_manual_port_override( mock_discovery["discover_single"].side_effect = TimeoutError mock_connect["connect"].side_effect = AuthenticationError - result2 = await hass.config_entries.flow.async_configure( - result["flow_id"], {CONF_HOST: host_str} - ) - await hass.async_block_till_done() + with override_side_effect( + mock_discovery["try_connect_all"], lambda *_, **__: mock_device + ): + result2 = await hass.config_entries.flow.async_configure( + result["flow_id"], {CONF_HOST: host_str} + ) + await hass.async_block_till_done() assert result2["type"] is FlowResultType.FORM assert result2["step_id"] == "user_auth_confirm" assert not result2["errors"] creds = Credentials("fake_username", "fake_password") - result3 = await hass.config_entries.flow.async_configure( - result2["flow_id"], - user_input={ - CONF_USERNAME: "fake_username", - CONF_PASSWORD: "fake_password", - }, - ) + with override_side_effect( + mock_discovery["try_connect_all"], lambda *_, **__: mock_device + ): + result3 = await hass.config_entries.flow.async_configure( + result2["flow_id"], + user_input={ + CONF_USERNAME: "fake_username", + CONF_PASSWORD: "fake_password", + }, + ) await hass.async_block_till_done() mock_discovery["try_connect_all"].assert_called_once_with( host, credentials=creds, port=port, http_client=ANY @@ -744,7 +1238,7 @@ async def test_manual_port_override_invalid( await hass.async_block_till_done() mock_discovery["discover_single"].assert_called_once_with( - "127.0.0.1", credentials=None, port=None + IP_ADDRESS, credentials=None, port=None ) assert result2["type"] is FlowResultType.CREATE_ENTRY @@ -941,7 +1435,7 @@ async def test_integration_discovery_with_ip_change( mock_config_entry.data[CONF_CONNECTION_PARAMETERS] == CONN_PARAMS_LEGACY.to_dict() ) - assert mock_config_entry.data[CONF_HOST] == "127.0.0.1" + assert mock_config_entry.data[CONF_HOST] == IP_ADDRESS mocked_device = _mocked_device(device_config=DEVICE_CONFIG_KLAP) with override_side_effect(mock_connect["connect"], lambda *_, **__: mocked_device): @@ -949,7 +1443,7 @@ async def test_integration_discovery_with_ip_change( DOMAIN, context={"source": config_entries.SOURCE_INTEGRATION_DISCOVERY}, data={ - CONF_HOST: "127.0.0.2", + CONF_HOST: IP_ADDRESS2, CONF_MAC: MAC_ADDRESS, CONF_ALIAS: ALIAS, CONF_DEVICE: mocked_device, @@ -961,7 +1455,7 @@ async def test_integration_discovery_with_ip_change( assert ( mock_config_entry.data[CONF_CONNECTION_PARAMETERS] == CONN_PARAMS_KLAP.to_dict() ) - assert mock_config_entry.data[CONF_HOST] == "127.0.0.2" + assert mock_config_entry.data[CONF_HOST] == IP_ADDRESS2 config = DeviceConfig.from_dict(DEVICE_CONFIG_DICT_KLAP) @@ -984,8 +1478,8 @@ async def test_integration_discovery_with_ip_change( await hass.async_block_till_done() assert mock_config_entry.state is ConfigEntryState.LOADED # Check that init set the new host correctly before calling connect - assert config.host == "127.0.0.1" - config.host = "127.0.0.2" + assert config.host == IP_ADDRESS + config.host = IP_ADDRESS2 config.uses_http = False # Not passed in to new config class config.http_client = "Foo" mock_connect["connect"].assert_awaited_once_with(config=config) @@ -1024,7 +1518,7 @@ async def test_integration_discovery_with_connection_change( ) == 0 ) - assert mock_config_entry.data[CONF_HOST] == "127.0.0.2" + assert mock_config_entry.data[CONF_HOST] == IP_ADDRESS2 assert ( mock_config_entry.data[CONF_CONNECTION_PARAMETERS] == CONN_PARAMS_AES.to_dict() ) @@ -1034,7 +1528,7 @@ async def test_integration_discovery_with_connection_change( NEW_DEVICE_CONFIG = { **DEVICE_CONFIG_DICT_KLAP, "connection_type": CONN_PARAMS_KLAP.to_dict(), - CONF_HOST: "127.0.0.2", + CONF_HOST: IP_ADDRESS2, } config = DeviceConfig.from_dict(NEW_DEVICE_CONFIG) # Reset the connect mock so when the config flow reloads the entry it succeeds @@ -1055,7 +1549,7 @@ async def test_integration_discovery_with_connection_change( DOMAIN, context={"source": config_entries.SOURCE_INTEGRATION_DISCOVERY}, data={ - CONF_HOST: "127.0.0.2", + CONF_HOST: IP_ADDRESS2, CONF_MAC: MAC_ADDRESS2, CONF_ALIAS: ALIAS, CONF_DEVICE: bulb, @@ -1067,12 +1561,12 @@ async def test_integration_discovery_with_connection_change( assert ( mock_config_entry.data[CONF_CONNECTION_PARAMETERS] == CONN_PARAMS_KLAP.to_dict() ) - assert mock_config_entry.data[CONF_HOST] == "127.0.0.2" + assert mock_config_entry.data[CONF_HOST] == IP_ADDRESS2 assert CREDENTIALS_HASH_AES not in mock_config_entry.data assert mock_config_entry.state is ConfigEntryState.LOADED - config.host = "127.0.0.2" + config.host = IP_ADDRESS2 config.uses_http = False # Not passed in to new config class config.http_client = "Foo" config.aes_keys = AES_KEYS @@ -1097,18 +1591,18 @@ async def test_dhcp_discovery_with_ip_change( flows = hass.config_entries.flow.async_progress() assert len(flows) == 0 - assert mock_config_entry.data[CONF_HOST] == "127.0.0.1" + assert mock_config_entry.data[CONF_HOST] == IP_ADDRESS discovery_result = await hass.config_entries.flow.async_init( DOMAIN, context={"source": config_entries.SOURCE_DHCP}, data=dhcp.DhcpServiceInfo( - ip="127.0.0.2", macaddress=DHCP_FORMATTED_MAC_ADDRESS, hostname=ALIAS + ip=IP_ADDRESS2, macaddress=DHCP_FORMATTED_MAC_ADDRESS, hostname=ALIAS ), ) assert discovery_result["type"] is FlowResultType.ABORT assert discovery_result["reason"] == "already_configured" - assert mock_config_entry.data[CONF_HOST] == "127.0.0.2" + assert mock_config_entry.data[CONF_HOST] == IP_ADDRESS2 async def test_dhcp_discovery_discover_fail( @@ -1121,14 +1615,14 @@ async def test_dhcp_discovery_discover_fail( flows = hass.config_entries.flow.async_progress() assert len(flows) == 0 - assert mock_config_entry.data[CONF_HOST] == "127.0.0.1" + assert mock_config_entry.data[CONF_HOST] == IP_ADDRESS with override_side_effect(mock_discovery["discover_single"], TimeoutError): discovery_result = await hass.config_entries.flow.async_init( DOMAIN, context={"source": config_entries.SOURCE_DHCP}, data=dhcp.DhcpServiceInfo( - ip="127.0.0.2", macaddress=DHCP_FORMATTED_MAC_ADDRESS, hostname=ALIAS + ip=IP_ADDRESS2, macaddress=DHCP_FORMATTED_MAC_ADDRESS, hostname=ALIAS ), ) assert discovery_result["type"] is FlowResultType.ABORT @@ -1160,15 +1654,58 @@ async def test_reauth( ) credentials = Credentials("fake_username", "fake_password") mock_discovery["discover_single"].assert_called_once_with( - "127.0.0.1", credentials=credentials, port=None + IP_ADDRESS, credentials=credentials, port=None ) - mock_discovery["mock_device"].update.assert_called_once_with() + mock_discovery["mock_devices"][IP_ADDRESS].update.assert_called_once_with() assert result2["type"] is FlowResultType.ABORT assert result2["reason"] == "reauth_successful" await hass.async_block_till_done() +async def test_reauth_camera( + hass: HomeAssistant, + mock_camera_config_entry: MockConfigEntry, + mock_discovery: AsyncMock, + mock_connect: AsyncMock, +) -> None: + """Test async_get_image.""" + mock_device = mock_connect["mock_devices"][IP_ADDRESS3] + mock_camera_config_entry.add_to_hass(hass) + mock_camera_config_entry.async_start_reauth( + hass, + config_entries.ConfigFlowContext( + reauth_source=CONF_CAMERA_CREDENTIALS, # type: ignore[typeddict-unknown-key] + ), + {"device": mock_device}, + ) + await hass.async_block_till_done() + flows = hass.config_entries.flow.async_progress() + assert len(flows) == 1 + [result] = flows + + assert result["step_id"] == "camera_auth_confirm" + + with patch( + "homeassistant.components.stream.async_check_stream_client_error", + return_value=None, + ): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={ + CONF_LIVE_VIEW: True, + CONF_USERNAME: "camuser2", + CONF_PASSWORD: "campass2", + }, + ) + assert result["type"] is FlowResultType.ABORT + assert result["reason"] == "reauth_successful" + assert dict(mock_camera_config_entry.data) == { + **CREATE_ENTRY_DATA_AES_CAMERA, + CONF_CAMERA_CREDENTIALS: {CONF_USERNAME: "camuser2", CONF_PASSWORD: "campass2"}, + } + + async def test_reauth_try_connect_all( hass: HomeAssistant, mock_added_config_entry: MockConfigEntry, @@ -1195,7 +1732,7 @@ async def test_reauth_try_connect_all( ) credentials = Credentials("fake_username", "fake_password") mock_discovery["discover_single"].assert_called_once_with( - "127.0.0.1", credentials=credentials, port=None + IP_ADDRESS, credentials=credentials, port=None ) mock_discovery["try_connect_all"].assert_called_once() assert result2["type"] is FlowResultType.ABORT @@ -1233,7 +1770,7 @@ async def test_reauth_try_connect_all_fail( ) credentials = Credentials("fake_username", "fake_password") mock_discovery["discover_single"].assert_called_once_with( - "127.0.0.1", credentials=credentials, port=None + IP_ADDRESS, credentials=credentials, port=None ) mock_discovery["try_connect_all"].assert_called_once() assert result2["errors"] == {"base": "cannot_connect"} @@ -1278,40 +1815,48 @@ async def test_reauth_update_with_encryption_change( assert CONF_CREDENTIALS_HASH not in mock_config_entry.data new_config = DeviceConfig( - "127.0.0.2", + IP_ADDRESS2, credentials=None, connection_type=Device.ConnectionParameters( Device.Family.SmartTapoPlug, Device.EncryptionType.Klap ), uses_http=True, ) - mock_discovery["mock_device"].host = "127.0.0.2" - mock_discovery["mock_device"].config = new_config - mock_discovery["mock_device"].credentials_hash = None - mock_connect["mock_devices"]["127.0.0.2"].config = new_config - mock_connect["mock_devices"]["127.0.0.2"].credentials_hash = CREDENTIALS_HASH_KLAP - - result2 = await hass.config_entries.flow.async_configure( - result["flow_id"], - user_input={ - CONF_USERNAME: "fake_username", - CONF_PASSWORD: "fake_password", - }, + mock_device = _mocked_device( + alias="my_device", + ip_address=IP_ADDRESS2, + mac=MAC_ADDRESS2, + device_config=new_config, + credentials_hash=CREDENTIALS_HASH_KLAP, ) - await hass.async_block_till_done(wait_background_tasks=True) + + with ( + override_side_effect( + mock_discovery["discover_single"], lambda *_, **__: mock_device + ), + override_side_effect(mock_connect["connect"], lambda *_, **__: mock_device), + ): + result2 = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={ + CONF_USERNAME: "fake_username", + CONF_PASSWORD: "fake_password", + }, + ) + await hass.async_block_till_done(wait_background_tasks=True) assert "Connection type changed for 127.0.0.2" in caplog.text credentials = Credentials("fake_username", "fake_password") mock_discovery["discover_single"].assert_called_once_with( - "127.0.0.2", credentials=credentials, port=None + IP_ADDRESS2, credentials=credentials, port=None ) - mock_discovery["mock_device"].update.assert_called_once_with() + mock_device.update.assert_called_once_with() assert result2["type"] is FlowResultType.ABORT assert result2["reason"] == "reauth_successful" assert mock_config_entry.state is ConfigEntryState.LOADED assert ( mock_config_entry.data[CONF_CONNECTION_PARAMETERS] == CONN_PARAMS_KLAP.to_dict() ) - assert mock_config_entry.data[CONF_HOST] == "127.0.0.2" + assert mock_config_entry.data[CONF_HOST] == IP_ADDRESS2 assert mock_config_entry.data[CONF_CREDENTIALS_HASH] == CREDENTIALS_HASH_KLAP @@ -1398,7 +1943,7 @@ async def test_reauth_update_from_discovery_with_ip_change( DOMAIN, context={"source": config_entries.SOURCE_INTEGRATION_DISCOVERY}, data={ - CONF_HOST: "127.0.0.2", + CONF_HOST: IP_ADDRESS2, CONF_MAC: MAC_ADDRESS, CONF_ALIAS: ALIAS, CONF_DEVICE: device, @@ -1410,7 +1955,7 @@ async def test_reauth_update_from_discovery_with_ip_change( assert ( mock_config_entry.data[CONF_CONNECTION_PARAMETERS] == CONN_PARAMS_KLAP.to_dict() ) - assert mock_config_entry.data[CONF_HOST] == "127.0.0.2" + assert mock_config_entry.data[CONF_HOST] == IP_ADDRESS2 async def test_reauth_no_update_if_config_and_ip_the_same( @@ -1493,26 +2038,27 @@ async def test_reauth_errors( [result] = flows assert result["step_id"] == "reauth_confirm" - mock_discovery["mock_device"].update.side_effect = error_type - result2 = await hass.config_entries.flow.async_configure( - result["flow_id"], - user_input={ - CONF_USERNAME: "fake_username", - CONF_PASSWORD: "fake_password", - }, - ) + mock_device = mock_discovery["mock_devices"][IP_ADDRESS] + with override_side_effect(mock_device.update, error_type): + result2 = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={ + CONF_USERNAME: "fake_username", + CONF_PASSWORD: "fake_password", + }, + ) credentials = Credentials("fake_username", "fake_password") mock_discovery["discover_single"].assert_called_once_with( - "127.0.0.1", credentials=credentials, port=None + IP_ADDRESS, credentials=credentials, port=None ) - mock_discovery["mock_device"].update.assert_called_once_with() + mock_device.update.assert_called_once_with() assert result2["type"] is FlowResultType.FORM assert result2["errors"] == {error_placement: errors_msg} assert result2["description_placeholders"]["error"] == str(error_type) mock_discovery["discover_single"].reset_mock() - mock_discovery["mock_device"].update.reset_mock(side_effect=True) + mock_device.update.reset_mock(side_effect=True) result3 = await hass.config_entries.flow.async_configure( result2["flow_id"], user_input={ @@ -1522,9 +2068,9 @@ async def test_reauth_errors( ) mock_discovery["discover_single"].assert_called_once_with( - "127.0.0.1", credentials=credentials, port=None + IP_ADDRESS, credentials=credentials, port=None ) - mock_discovery["mock_device"].update.assert_called_once_with() + mock_device.update.assert_called_once_with() assert result3["type"] is FlowResultType.ABORT assert result3["reason"] == "reauth_successful" @@ -1731,12 +2277,169 @@ async def test_reauth_update_other_flows( ) credentials = Credentials("fake_username", "fake_password") mock_discovery["discover_single"].assert_called_once_with( - "127.0.0.1", credentials=credentials, port=None + IP_ADDRESS, credentials=credentials, port=None ) - mock_discovery["mock_device"].update.assert_called_once_with() + mock_discovery["mock_devices"][IP_ADDRESS].update.assert_called_once_with() assert result2["type"] is FlowResultType.ABORT assert result2["reason"] == "reauth_successful" await hass.async_block_till_done() flows = hass.config_entries.flow.async_progress() assert len(flows) == 0 + + +async def test_reconfigure( + hass: HomeAssistant, + mock_added_config_entry: MockConfigEntry, + mock_discovery: AsyncMock, + mock_connect: AsyncMock, +) -> None: + """Test reconfigure flow.""" + result = await mock_added_config_entry.start_reconfigure_flow(hass) + + assert result["type"] is FlowResultType.FORM + assert result["step_id"] == "reconfigure" + + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={ + CONF_HOST: IP_ADDRESS, + }, + ) + + assert result["type"] is FlowResultType.ABORT + assert result["reason"] == "reconfigure_successful" + + +async def test_reconfigure_auth_discovered( + hass: HomeAssistant, + mock_added_config_entry: MockConfigEntry, + mock_discovery: AsyncMock, + mock_connect: AsyncMock, +) -> None: + """Test reconfigure auth flow for device that's discovered.""" + result = await mock_added_config_entry.start_reconfigure_flow(hass) + + assert result["type"] is FlowResultType.FORM + assert result["step_id"] == "reconfigure" + + # Simulate a bad host + with ( + override_side_effect( + mock_discovery["mock_devices"][IP_ADDRESS].update, KasaException + ), + ): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={ + CONF_HOST: "WRONG_IP", + }, + ) + + assert result["type"] is FlowResultType.FORM + assert result["step_id"] == "reconfigure" + assert result["errors"] == {"base": "cannot_connect"} + assert "error" in result["description_placeholders"] + + with ( + override_side_effect( + mock_discovery["mock_devices"][IP_ADDRESS].update, AuthenticationError + ), + ): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={ + CONF_HOST: IP_ADDRESS, + }, + ) + + assert result["type"] is FlowResultType.FORM + assert result["step_id"] == "user_auth_confirm" + + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={ + CONF_USERNAME: "user", + CONF_PASSWORD: "pass", + }, + ) + + assert result["type"] is FlowResultType.ABORT + assert result["reason"] == "reconfigure_successful" + + +async def test_reconfigure_auth_try_connect_all( + hass: HomeAssistant, + mock_added_config_entry: MockConfigEntry, + mock_discovery: AsyncMock, + mock_connect: AsyncMock, +) -> None: + """Test reconfigure auth flow for device that's not discovered.""" + result = await mock_added_config_entry.start_reconfigure_flow(hass) + + assert result["type"] is FlowResultType.FORM + assert result["step_id"] == "reconfigure" + + with ( + override_side_effect(mock_discovery["discover_single"], TimeoutError), + override_side_effect(mock_connect["connect"], KasaException), + ): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={ + CONF_HOST: IP_ADDRESS, + }, + ) + + assert result["type"] is FlowResultType.FORM + assert result["step_id"] == "user_auth_confirm" + + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={ + CONF_USERNAME: "user", + CONF_PASSWORD: "pass", + }, + ) + + assert result["type"] is FlowResultType.ABORT + assert result["reason"] == "reconfigure_successful" + + +async def test_reconfigure_camera( + hass: HomeAssistant, + mock_camera_config_entry: MockConfigEntry, + mock_discovery: AsyncMock, + mock_connect: AsyncMock, +) -> None: + """Test reconfigure flow.""" + mock_camera_config_entry.add_to_hass(hass) + result = await mock_camera_config_entry.start_reconfigure_flow(hass) + + assert result["type"] is FlowResultType.FORM + assert result["step_id"] == "reconfigure" + + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={ + CONF_HOST: IP_ADDRESS3, + }, + ) + + assert result["type"] is FlowResultType.FORM + assert result["step_id"] == "camera_auth_confirm" + + with patch( + "homeassistant.components.stream.async_check_stream_client_error", + return_value=None, + ): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={ + CONF_LIVE_VIEW: True, + CONF_USERNAME: "camuser", + CONF_PASSWORD: "campass", + }, + ) + assert result["type"] is FlowResultType.ABORT + assert result["reason"] == "reconfigure_successful" diff --git a/tests/components/tplink/test_init.py b/tests/components/tplink/test_init.py index dd967e0e0d6..8dad8881b9b 100644 --- a/tests/components/tplink/test_init.py +++ b/tests/components/tplink/test_init.py @@ -59,6 +59,7 @@ from . import ( _patch_discovery, _patch_single_discovery, ) +from .conftest import override_side_effect from tests.common import MockConfigEntry, async_fire_time_changed @@ -70,6 +71,7 @@ async def test_configuring_tplink_causes_discovery( with ( patch("homeassistant.components.tplink.Discover.discover") as discover, patch("homeassistant.components.tplink.Discover.discover_single"), + patch("homeassistant.components.tplink.Device.connect"), ): discover.return_value = {MagicMock(): MagicMock()} await async_setup_component(hass, tplink.DOMAIN, {tplink.DOMAIN: {}}) @@ -221,8 +223,12 @@ async def test_config_entry_with_stored_credentials( hass.data.setdefault(DOMAIN, {})[CONF_AUTHENTICATION] = auth mock_config_entry.add_to_hass(hass) - with patch( - "homeassistant.components.tplink.async_create_clientsession", return_value="Foo" + with ( + patch( + "homeassistant.components.tplink.async_create_clientsession", + return_value="Foo", + ), + override_side_effect(mock_discovery["discover"], lambda *_, **__: {}), ): await hass.config_entries.async_setup(mock_config_entry.entry_id) await hass.async_block_till_done()