jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-04-24 09:17:53 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add camera platform to tplink integration (#129180)

Browse Source 
Co-authored-by: Teemu R. <tpr@iki.fi>
This commit is contained in:
Steven B. 2024-12-22 18:56:33 +00:00 committed by GitHub
parent 475f19c140
commit b1f6563fb2
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
15 changed files with 2012 additions and 150 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
homeassistant/components/tplink/__init__.py
View File

@ -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

220
homeassistant/components/tplink/camera.py Normal file
View File

@ -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)

257
homeassistant/components/tplink/config_flow.py
View File

@ -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,
},
)

3
homeassistant/components/tplink/const.py
View File

@ -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,

14
homeassistant/components/tplink/entity.py
View File

@ -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]]:

2
homeassistant/components/tplink/manifest.json
View File

@ -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

4
homeassistant/components/tplink/models.py
View File

@ -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

27
homeassistant/components/tplink/strings.json
View File

@ -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"

71
tests/components/tplink/__init__.py
View File

@ -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,
}

94
tests/components/tplink/conftest.py
View File

@ -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,

30
tests/components/tplink/fixtures/features.json
View File

@ -320,5 +320,35 @@
"type": "Sensor",
"category": "Info",
"value": "2024-06-24 10:03:11.046643+01:00"
},
"pan_left": {
"value": "<Action>",
"type": "Action",
"category": "Config"
},
"pan_right": {
"value": "<Action>",
"type": "Action",
"category": "Config"
},
"pan_step": {
"value": 10,
"type": "Number",
"category": "Config"
},
"tilt_up": {
"value": "<Action>",
"type": "Action",
"category": "Config"
},
"tilt_down": {
"value": "<Action>",
"type": "Action",
"category": "Config"
},
"tilt_step": {
"value": 10,
"type": "Number",
"category": "Config"
}
}

87
tests/components/tplink/snapshots/test_camera.ambr Normal file
View File

@ -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': <ANY>,
'device_class': None,
'device_id': <ANY>,
'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': <ANY>,
'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': <CameraEntityFeature: 3>,
'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=<UndefinedType._singleton: 0>, 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': <StreamType.HLS: 'hls'>,
'supported_features': <CameraEntityFeature: 3>,
}),
'context': <ANY>,
'entity_id': 'camera.my_camera_live_view',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'idle',
})
# ---
# name: test_states[my_camera-entry]
DeviceRegistryEntrySnapshot({
'area_id': None,
'config_entries': <ANY>,
'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': <ANY>,
'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': <ANY>,
'serial_number': None,
'suggested_area': None,
'sw_version': '1.0.0',
'via_device_id': None,
})
# ---

431
tests/components/tplink/test_camera.py Normal file
View File

@ -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)

899
tests/components/tplink/test_config_flow.py
View File

File diff suppressed because it is too large Load Diff

10
tests/components/tplink/test_init.py
View File

@ -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()