starred/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-08-24 10:13:52 -05:00
Code Issues Packages Projects Releases Wiki Activity

Source UniFi Protect camera streams from the public API (#174369)

Browse Source
This commit is contained in:
Raphael Hehl
2026-06-21 15:44:26 -05:00
committed by GitHub
parent 8198aa263b
commit 5e6085f961
10 changed files with 868 additions and 905 deletions
Download Patch File Download Diff File Expand all files Collapse all files
homeassistant/components/unifiprotect/__init__.py
+19 -4
View File
@@ -159,12 +159,27 @@ async def _async_setup_entry(
await async_migrate_data(hass, entry, data_service.api, bootstrap)
data_service.async_setup()
# Prime the public bootstrap. The devices websocket subscription was already
# registered in async_setup() per library docs (subscribe first, then prime).
# Prime the public bootstrap (subscribe-then-prime, per library docs). Camera
# streams depend on it, so a failed prime retries instead of building
# streamless cameras.
try:
await data_service.api.update_public()
except Exception: # noqa: BLE001
_LOGGER.debug("Public API bootstrap update failed", exc_info=True)
except NotAuthorized as err:
# A public 401 means a bad/revoked API key (independent of the private
# session); route to reauth instead of retrying forever.
await data_service.async_stop()
raise ConfigEntryAuthFailed(
translation_domain=DOMAIN,
translation_key="api_key_required",
) from err
except (TimeoutError, ClientError, ServerDisconnectedError) as err:
# async_setup() already subscribed the websockets and started polling;
# tear them down so a setup retry does not leak another set.
await data_service.async_stop()
raise ConfigEntryNotReady(
translation_domain=DOMAIN,
translation_key="public_bootstrap_failed",
) from err
# Load PTZ patrol data before loading platforms
await data_service.async_load_ptz_patrols()
homeassistant/components/unifiprotect/camera.py
+84 -89
View File
@@ -1,6 +1,5 @@
"""Support for Ubiquiti's UniFi Protect NVR."""
from collections.abc import Generator
import logging
from uiprotect.data import (
@@ -36,38 +35,41 @@ PARALLEL_UPDATES = 0
@callback
def _create_rtsp_repair(
hass: HomeAssistant, entry: UFPConfigEntry, data: ProtectData, camera: UFPCamera
hass: HomeAssistant, entry: UFPConfigEntry, camera: UFPCamera
) -> None:
edit_key = "readonly"
if camera.can_write(data.api.bootstrap.auth_user):
edit_key = "writable"
translation_key = f"rtsp_disabled_{edit_key}"
issue_key = f"rtsp_disabled_{camera.id}"
ir.async_create_issue(
hass,
DOMAIN,
issue_key,
f"rtsp_disabled_{camera.id}",
is_fixable=True,
is_persistent=False,
learn_more_url="https://www.home-assistant.io/integrations/unifiprotect/#camera-streams",
severity=IssueSeverity.WARNING,
translation_key=translation_key,
translation_key="rtsp_disabled",
translation_placeholders={"camera": camera.display_name},
data={"entry_id": entry.entry_id, "camera_id": camera.id},
)
@callback
def _get_camera_channels(
def _async_camera_entities(
hass: HomeAssistant,
entry: UFPConfigEntry,
data: ProtectData,
ufp_device: UFPCamera | None = None,
) -> Generator[tuple[UFPCamera, CameraChannel, bool]]:
"""Get all the camera channels."""
) -> list[ProtectDeviceEntity]:
"""Create camera entities with stream URLs sourced from the public API.
One entity per *active* RTSPS quality (the first is enabled by default). The
package channel is a snapshot-first view and is always exposed (disabled by
default), streaming only when its quality is active. When no main quality is
active the first non-package channel is still created so snapshots work, and
a repair offers to activate its stream. RTSPS URLs come from the public API
(the authoritative per-camera host, so stacked consoles resolve correctly)
with SRTP stripped for go2rtc.
"""
disable_stream = data.disable_stream
entities: list[ProtectDeviceEntity] = []
cameras = data.get_cameras() if ufp_device is None else [ufp_device]
for camera in cameras:
if not camera.channels:
@@ -81,61 +83,48 @@ def _get_camera_channels(
data.async_add_pending_camera_id(camera.id)
continue
is_default = True
streams = data.get_rtsps_streams(camera.id)
active = set(streams.get_active_stream_qualities()) if streams else set()
issue_id = f"rtsp_disabled_{camera.id}"
has_stream = False
package_channel: CameraChannel | None = None
for channel in camera.channels:
if channel.is_package:
yield camera, channel, True
elif channel.is_rtsp_enabled:
yield camera, channel, is_default
is_default = False
# no RTSP enabled use first channel with no stream
if is_default and not camera.is_third_party_camera:
# Only create repair issue if RTSP is not disabled globally
if not data.disable_stream:
_create_rtsp_repair(hass, entry, data, camera)
else:
ir.async_delete_issue(hass, DOMAIN, f"rtsp_disabled_{camera.id}")
yield camera, camera.channels[0], True
else:
ir.async_delete_issue(hass, DOMAIN, f"rtsp_disabled_{camera.id}")
def _async_camera_entities(
hass: HomeAssistant,
entry: UFPConfigEntry,
data: ProtectData,
ufp_device: UFPCamera | None = None,
) -> list[ProtectDeviceEntity]:
disable_stream = data.disable_stream
entities: list[ProtectDeviceEntity] = []
for camera, channel, is_default in _get_camera_channels(
hass, entry, data, ufp_device
):
# do not enable streaming for package camera
# 2 FPS causes a lot of buffering
entities.append(
ProtectCamera(
data,
camera,
channel,
is_default,
True,
disable_stream or channel.is_package,
)
)
if channel.is_rtsp_enabled and not channel.is_package:
entities.append(
ProtectCamera(
data,
camera,
channel,
is_default,
False,
disable_stream,
package_channel = channel
continue
if channel.rtsps_quality in active:
entities.append(
ProtectCamera(data, camera, channel, not has_stream, disable_stream)
)
has_stream = True
# the package channel is a snapshot-first view (very low FPS); always
# expose it (disabled by default), streaming only when its quality is active
if package_channel is not None:
entities.append(
ProtectCamera(data, camera, package_channel, False, disable_stream)
)
if has_stream:
ir.async_delete_issue(hass, DOMAIN, issue_id)
continue
# no active main stream: expose the first non-package channel for snapshots
fallback = next((c for c in camera.channels if not c.is_package), None)
if fallback is None:
continue
entities.append(ProtectCamera(data, camera, fallback, True, disable_stream))
# no repair when the stream can't be enabled anyway: a disconnected
# camera is streamless because it is offline, not because it needs one
if (
disable_stream
or camera.is_third_party_camera
or camera.state is not StateType.CONNECTED
):
ir.async_delete_issue(hass, DOMAIN, issue_id)
else:
_create_rtsp_repair(hass, entry, camera)
return entities
@@ -186,26 +175,17 @@ class ProtectCamera(ProtectDeviceEntity, Camera):
camera: UFPCamera,
channel: CameraChannel,
is_default: bool,
secure: bool,
disable_stream: bool,
) -> None:
"""Initialize an UniFi camera."""
self.channel = channel
self._secure = secure
self._disable_stream = disable_stream
self._last_image: bytes | None = None
super().__init__(data, camera)
device = self.device
camera_name = get_camera_base_name(channel)
if self._secure:
self._attr_unique_id = f"{device.mac}_{channel.id}"
self._attr_name = camera_name
else:
self._attr_unique_id = f"{device.mac}_{channel.id}_insecure"
self._attr_name = f"{camera_name} (insecure)"
# only the default (first) channel is enabled by default
self._attr_entity_registry_enabled_default = is_default and secure
self._attr_unique_id = f"{self.device.mac}_{channel.id}"
self._attr_name = get_camera_base_name(channel)
# only the default (first active) channel is enabled by default
self._attr_entity_registry_enabled_default = is_default
# Set the stream source before finishing the init
# because async_added_to_hass is too late and camera
# integration uses async_internal_added_to_hass to access
@@ -214,12 +194,25 @@ class ProtectCamera(ProtectDeviceEntity, Camera):
@callback
def _async_set_stream_source(self) -> None:
channel = self.channel
enable_stream = not self._disable_stream and channel.is_rtsp_enabled
# SRTP disabled because go2rtc does not support it
# https://github.com/AlexxIT/go2rtc/#source-rtsp
rtsp_url = channel.rtsps_no_srtp_url if self._secure else channel.rtsp_url
source = rtsp_url if enable_stream else None
"""Set the public-API RTSPS stream URL (SRTP stripped for go2rtc)."""
quality = self.channel.rtsps_quality
streams = self.data.get_rtsps_streams(self.device.id)
if self._disable_stream or quality is None or streams is None:
source = None
if (
streams is None
and not self._disable_stream
and not self.channel.is_package
):
# online camera unexpectedly absent from the public bootstrap;
# log so this is distinguishable from an intentionally off stream
_LOGGER.debug(
"No public RTSPS data for camera %s (%s); using snapshots",
self.device.display_name,
self.device.id,
)
else:
source = streams.get_stream_url(quality, srtp=False)
self._attr_supported_features = _ENABLE_FEATURE if source else _DISABLE_FEATURE
self._stream_source = source
@@ -254,11 +247,13 @@ class ProtectCamera(ProtectDeviceEntity, Camera):
self, width: int | None = None, height: int | None = None
) -> bytes | None:
"""Return the Camera Image."""
if self.channel.is_package:
last_image = await self.device.get_package_snapshot(width, height)
else:
last_image = await self.device.get_public_api_snapshot()
self._last_image = last_image
# Without a stream the camera is rendered by rapidly polling snapshots;
# request low quality then to avoid hammering the console with large
# images. width/height are unused (the public endpoint has no resize).
high_quality = None if self._stream_source else False
self._last_image = await self.device.get_public_api_snapshot(
high_quality=high_quality, package=self.channel.is_package
)
return self._last_image
async def stream_source(self) -> str | None:
homeassistant/components/unifiprotect/data.py
+15
View File
@@ -9,6 +9,7 @@ import logging
from typing import TYPE_CHECKING, Any, cast
from uiprotect import ProtectApiClient
from uiprotect.api import RTSPSStreams
from uiprotect.data import (
NVR,
Camera,
@@ -112,6 +113,20 @@ class ProtectData:
"""Max number of events to load at once."""
return self._entry.options.get(CONF_MAX_MEDIA, DEFAULT_MAX_MEDIA) # type: ignore[no-any-return]
def get_rtsps_streams(self, camera_id: str) -> RTSPSStreams | None:
"""Return the library-owned public-API RTSPS streams for a camera.
The library primes ``PublicCamera.rtsps_streams`` during
``update_public()`` and keeps it fresh (reconnect refresh + create/delete
write-through), so the integration reads it synchronously and stores
nothing itself.
"""
api = self.api
if not api.has_public_bootstrap:
return None
camera = api.public_bootstrap.cameras.get(camera_id)
return camera.rtsps_streams if camera is not None else None
@callback
def async_subscribe_adopt(
self, add_callback: Callable[[ProtectAdoptableDeviceModel], None]
homeassistant/components/unifiprotect/migrate.py
+69
View File
@@ -109,6 +109,75 @@ async def async_migrate_data(
async_deprecate_hdr(hass, entry)
_LOGGER.debug("Completed Migrate: async_deprecate_hdr")
_LOGGER.debug("Start Migrate: async_migrate_insecure_cameras")
async_migrate_insecure_cameras(hass, entry)
_LOGGER.debug("Completed Migrate: async_migrate_insecure_cameras")
@callback
def async_migrate_insecure_cameras(hass: HomeAssistant, entry: UFPConfigEntry) -> None:
"""Migrate the legacy plain-RTSP "(insecure)" camera entities.
Streams now come from the public API, which is RTSPS-only, so the old
``{mac}_{channel}_insecure`` camera entities no longer exist. Redirect each
to its secure unique_id (``{mac}_{channel}``) so its history/customizations
carry over to the public stream; if the secure entity already exists, drop
the redundant insecure one (raising a repair first if it is still used).
Added in 2026.7.0
"""
registry = er.async_get(hass)
for entity in er.async_entries_for_config_entry(registry, entry.entry_id):
if entity.domain != Platform.CAMERA or not entity.unique_id.endswith(
"_insecure"
):
continue
secure_unique_id = entity.unique_id.removesuffix("_insecure")
secure_entity_id = registry.async_get_entity_id(
Platform.CAMERA, DOMAIN, secure_unique_id
)
if secure_entity_id is None:
registry.async_update_entity(
entity.entity_id, new_unique_id=secure_unique_id
)
continue
_async_repair_if_insecure_used(hass, entity, secure_entity_id)
registry.async_remove(entity.entity_id)
@callback
def _async_repair_if_insecure_used(
hass: HomeAssistant, insecure: er.RegistryEntry, replacement: str
) -> None:
"""Warn before removing a redundant insecure camera entity that is in use.
Removal cannot rewrite the user's automations/scripts, so a persistent repair
lists the affected ones and points to the surviving secure entity. Disabled
entities are skipped: they are not active in any automation.
"""
if insecure.disabled_by is not None:
return
items = sorted(
set(automations_with_entity(hass, insecure.entity_id))
| set(scripts_with_entity(hass, insecure.entity_id))
)
if not items:
return
ir.async_create_issue(
hass,
DOMAIN,
f"insecure_camera_removed_{insecure.unique_id}",
is_fixable=False,
is_persistent=True,
severity=IssueSeverity.WARNING,
translation_key="insecure_camera_removed",
translation_placeholders={
"entity_id": insecure.entity_id,
"replacement": replacement,
"items": "* `" + "`\n* `".join(items) + "`\n",
},
)
@callback
def async_deprecate_hdr(hass: HomeAssistant, entry: UFPConfigEntry) -> None:
homeassistant/components/unifiprotect/repairs.py
+29 -44
View File
@@ -1,9 +1,10 @@
"""unifiprotect.repairs."""
import logging
from typing import cast
from uiprotect import ProtectApiClient
from uiprotect.data import Bootstrap, Camera
from uiprotect.exceptions import ClientError
import voluptuous as vol
from homeassistant.components.repairs import (
@@ -17,6 +18,8 @@ from homeassistant.helpers import issue_registry as ir
from .data import UFPConfigEntry, async_get_data_for_entry_id
from .utils import async_create_api_client
_LOGGER = logging.getLogger(__name__)
class ProtectRepair(RepairsFlow):
"""Handler for an issue fixing flow."""
@@ -71,50 +74,25 @@ class CloudAccountRepair(ProtectRepair):
class RTSPRepair(ProtectRepair):
"""Handler for an issue fixing flow."""
"""Fix flow for a camera without an active RTSPS stream.
Verifies and creates the stream through the public API, so it also works in
a public-only setup. The camera name is carried by the issue placeholders.
"""
_camera_id: str
_camera: Camera | None
_bootstrap: Bootstrap | None
def __init__(
self,
*,
api: ProtectApiClient,
entry: UFPConfigEntry,
camera_id: str,
self, *, api: ProtectApiClient, entry: UFPConfigEntry, camera_id: str
) -> None:
"""Create flow."""
super().__init__(api=api, entry=entry)
self._camera_id = camera_id
self._bootstrap = None
self._camera = None
@callback
def _async_get_placeholders(self) -> dict[str, str]:
description_placeholders = super()._async_get_placeholders()
if self._camera is not None:
description_placeholders["camera"] = self._camera.display_name
return description_placeholders
async def _get_boostrap(self) -> Bootstrap:
if self._bootstrap is None:
self._bootstrap = await self._api.get_bootstrap()
return self._bootstrap
async def _get_camera(self) -> Camera:
if self._camera is None:
bootstrap = await self._get_boostrap()
self._camera = bootstrap.cameras.get(self._camera_id)
assert self._camera is not None
return self._camera
async def _enable_rtsp(self) -> None:
camera = await self._get_camera()
await camera.create_rtsps_streams(qualities="high")
async def _async_has_active_stream(self) -> bool:
streams = await self._api.get_camera_rtsps_streams(self._camera_id)
return bool(streams and streams.get_active_stream_qualities())
async def async_step_init(
self, user_input: dict[str, str] | None = None
@@ -129,21 +107,28 @@ class RTSPRepair(ProtectRepair):
"""Handle the first step of a fix flow."""
if user_input is None:
# make sure camera object is loaded for placeholders
await self._get_camera()
placeholders = self._async_get_placeholders()
return self.async_show_form(
step_id="start",
data_schema=vol.Schema({}),
description_placeholders=placeholders,
description_placeholders=self._async_get_placeholders(),
)
updated_camera = await self._api.get_camera(self._camera_id)
if not any(c.is_rtsp_enabled for c in updated_camera.channels):
await self._enable_rtsp()
# Creating a stream needs write permission; a NotAuthorized/ClientError
# routes to the confirm step (which explains the manual fallback)
# instead of raising out of the fix flow.
try:
active = await self._async_has_active_stream()
if not active:
await self._api.create_camera_rtsps_streams(self._camera_id, "high")
active = await self._async_has_active_stream()
except ClientError:
_LOGGER.debug(
"Auto-creating RTSPS stream failed; routing to manual fallback",
exc_info=True,
)
active = False
updated_camera = await self._api.get_camera(self._camera_id)
if any(c.is_rtsp_enabled for c in updated_camera.channels):
if active:
await self.hass.config_entries.async_reload(self._entry.entry_id)
return self.async_create_entry(data={})
return await self.async_step_confirm()
homeassistant/components/unifiprotect/strings.json
+12 -20
View File
@@ -706,6 +706,9 @@
"ptz_preset_not_found": {
"message": "Could not find PTZ preset with name {preset_name} on camera {camera_name}"
},
"public_bootstrap_failed": {
"message": "Could not load the public API bootstrap for camera streams"
},
"relay_not_available": {
"message": "Relay is no longer available"
},
@@ -744,35 +747,24 @@
"description": "UniFi Protect v3 added a new state for HDR (auto). As a result, the HDR Mode switch has been replaced with an HDR Mode select, and it is deprecated.\n\nBelow are the detected automations or scripts that use one or more of the deprecated entities:\n{items}\nThe above list may be incomplete and it does not include any template usages inside of dashboards. Please update any templates, automations or scripts accordingly.",
"title": "HDR Mode switch deprecated"
},
"rtsp_disabled_readonly": {
"fix_flow": {
"step": {
"confirm": {
"description": "Are you sure you want to leave RTSPS disabled for {camera}?",
"title": "[%key:component::unifiprotect::issues::rtsp_disabled_readonly::fix_flow::step::start::title%]"
},
"start": {
"description": "RTSPS is disabled on the camera {camera}. RTSPS is required to be able to live stream your camera within Home Assistant. If you do not enable RTSPS, it may create an additional load on your UniFi Protect NVR, as any live video players will default to rapidly pulling snapshots from the camera.\n\nPlease [enable RTSPS]({learn_more}) on the camera and then come back and confirm this repair.",
"title": "RTSPS is disabled on camera {camera}"
}
}
},
"title": "RTSPS is disabled on camera {camera}"
"insecure_camera_removed": {
"description": "The camera entity `{entity_id}` used plain RTSP, which is no longer available now that streams come from the UniFi Protect public API (RTSPS only). It has been removed; use `{replacement}` instead.\n\nUpdate the following automations and scripts:\n{items}",
"title": "Insecure camera entity removed"
},
"rtsp_disabled_writable": {
"rtsp_disabled": {
"fix_flow": {
"step": {
"confirm": {
"description": "[%key:component::unifiprotect::issues::rtsp_disabled_readonly::fix_flow::step::confirm::description%]",
"title": "[%key:component::unifiprotect::issues::rtsp_disabled_readonly::fix_flow::step::start::title%]"
"description": "Home Assistant could not enable an RTSPS stream for {camera}. You may need to enable a stream manually on the camera, or disable the camera entity if you only need snapshots.",
"title": "[%key:component::unifiprotect::issues::rtsp_disabled::fix_flow::step::start::title%]"
},
"start": {
"description": "RTSPS is disabled on the camera {camera}. RTSPS is required to live stream your camera within Home Assistant. If you do not enable RTSPS, it may create an additional load on your UniFi Protect NVR as any live video players will default to rapidly pulling snapshots from the camera.\n\nYou may manually [enable RTSPS]({learn_more}) on your selected camera quality channel or Home Assistant can automatically enable the highest quality channel for you. Confirm this repair once you have enabled the RTSPS channel or if you want Home Assistant to enable the highest quality automatically.",
"title": "[%key:component::unifiprotect::issues::rtsp_disabled_readonly::fix_flow::step::start::title%]"
"description": "No RTSPS stream is available for the camera {camera}, so it cannot live stream within Home Assistant. Until a stream is enabled, live views fall back to rapidly pulling snapshots (effectively a slideshow), which can put significant load on your UniFi Protect console. Confirm this repair and Home Assistant will create the highest quality RTSPS stream for you through the UniFi Protect public API.",
"title": "No stream is available for camera {camera}"
}
}
},
"title": "RTSPS is disabled on camera {camera}"
"title": "No stream is available for camera {camera}"
}
},
"options": {
tests/components/unifiprotect/conftest.py
+76 -4
View File
@@ -6,11 +6,13 @@ from functools import partial
from ipaddress import IPv4Address
from pathlib import Path
from tempfile import gettempdir
from types import SimpleNamespace
from typing import Any
from unittest.mock import AsyncMock, Mock, patch
import pytest
from uiprotect import ProtectApiClient
from uiprotect.api import RTSPSStreams
from uiprotect.data import (
NVR,
AiPort,
@@ -22,6 +24,7 @@ from uiprotect.data import (
Liveview,
Sensor,
SmartDetectObjectType,
StateType,
VideoMode,
Viewer,
WSSubscriptionMessage,
@@ -46,6 +49,22 @@ from .utils import MockUFPFixture
from tests.common import MockConfigEntry, load_json_object_fixture
def _public_rtsps_for(camera: Any) -> RTSPSStreams | None:
"""Build a camera's primed RTSPS streams from its RTSP-enabled channels.
Mirrors what the library writes onto ``PublicCamera.rtsps_streams`` during
``update_public()`` — only RTSP-enabled channels carry an active URL, and a
camera with none is left streamless (``None``).
"""
urls = {
channel.rtsps_quality: channel.rtsps_url
for channel in camera.channels
if channel.is_rtsp_enabled and channel.rtsps_quality is not None
}
return RTSPSStreams(**urls) if urls else None
MAC_ADDR = "aa:bb:cc:dd:ee:ff"
# Common test data constants
@@ -79,8 +98,17 @@ def mock_nvr():
NVR.model_config["validate_assignment"] = True
@pytest.fixture(name="ufp_options")
def mock_ufp_options(request: pytest.FixtureRequest) -> dict[str, Any]:
"""Options for the mock config entry (override per-test via indirect param)."""
options: dict[str, Any] = {}
if hasattr(request, "param"):
options.update(request.param)
return options
@pytest.fixture(name="ufp_config_entry")
def mock_ufp_config_entry():
def mock_ufp_config_entry(ufp_options: dict[str, Any]):
"""Mock the unifiprotect config entry."""
return MockConfigEntry(
@@ -94,6 +122,7 @@ def mock_ufp_config_entry():
CONF_PORT: DEFAULT_PORT,
CONF_VERIFY_SSL: DEFAULT_VERIFY_SSL,
},
options=ufp_options,
version=2,
unique_id="A1E00C826924",
)
@@ -150,8 +179,30 @@ def mock_ufp_client(bootstrap: Bootstrap):
client.get_nvr = get_nvr
client.get_bootstrap = AsyncMock(return_value=bootstrap)
client.update = AsyncMock(return_value=bootstrap)
client.update_public = AsyncMock()
client.async_disconnect_ws = AsyncMock()
client.has_public_bootstrap = False
client.has_public_bootstrap = True
# The library owns RTSPS streams on ``PublicCamera.rtsps_streams`` and primes
# them in ``update_public()``; the integration reads them synchronously. Start
# with empty collections; the ``update_public`` side effect (see ``mock_entry``)
# primes the cameras from the private bootstrap.
client.public_bootstrap = Mock()
client.public_bootstrap.cameras = {}
client.public_bootstrap.relays = {}
client.public_bootstrap.sirens = {}
client.public_bootstrap.arm_profiles = {}
client.public_bootstrap.arm_mode = None
async def get_camera_rtsps_streams(
camera_id: str, *args: Any, **kwargs: Any
) -> RTSPSStreams | None:
"""Fetch a camera's RTSPS streams (used by the repair flow)."""
camera = client.bootstrap.cameras.get(camera_id)
return _public_rtsps_for(camera) if camera is not None else None
client.get_camera_rtsps_streams = AsyncMock(side_effect=get_camera_rtsps_streams)
client.create_camera_rtsps_streams = AsyncMock(return_value=None)
return client
@@ -198,8 +249,29 @@ def mock_entry(
ufp_client.subscribe_websocket_state = subscribe_websocket_state
ufp_client.subscribe_devices_websocket = subscribe_devices_websocket
ufp_client.subscribe_devices_websocket_state = subscribe_devices_websocket_state
ufp_client.update_public = AsyncMock()
ufp_client.has_public_bootstrap = False
async def update_public() -> Any:
# Mirror the library prime: populate PublicCamera.rtsps_streams for
# every camera from the private bootstrap (connected cameras only, so
# a disconnected camera stays streamless), keyed by id.
pb = ufp_client.public_bootstrap
pb.cameras = {
camera.id: SimpleNamespace(
id=camera.id,
name=camera.display_name,
state=camera.state,
rtsps_streams=(
_public_rtsps_for(camera)
if camera.state is StateType.CONNECTED
else None
),
)
for camera in ufp_client.bootstrap.cameras.values()
}
return pb
ufp_client.update_public = AsyncMock(side_effect=update_public)
ufp_client.has_public_bootstrap = True
yield ufp
tests/components/unifiprotect/test_camera.py
+289 -553
View File
@@ -1,610 +1,348 @@
"""Test the UniFi Protect camera platform."""
from unittest.mock import AsyncMock, Mock
from types import SimpleNamespace
from typing import Any
from unittest.mock import AsyncMock
import pytest
from uiprotect.api import DEVICE_UPDATE_INTERVAL
from uiprotect.data import AiPort, Camera as ProtectCamera, CameraChannel, StateType
from uiprotect.exceptions import NvrError
from uiprotect.websocket import WebsocketState
from webrtc_models import RTCIceCandidateInit
from uiprotect.data import AiPort, Camera as ProtectCamera, StateType
from uiprotect.exceptions import ClientError, NotAuthorized
from homeassistant.components.camera import (
CameraCapabilities,
CameraEntityFeature,
CameraState,
CameraWebRTCProvider,
StreamType,
WebRTCSendMessage,
async_get_image,
async_get_stream_source,
async_register_webrtc_provider,
get_camera_from_entity_id,
)
from homeassistant.components.homeassistant import (
DOMAIN as HOMEASSISTANT_DOMAIN,
SERVICE_UPDATE_ENTITY,
)
from homeassistant.components.unifiprotect.const import (
ATTR_BITRATE,
ATTR_CHANNEL_ID,
ATTR_FPS,
ATTR_HEIGHT,
ATTR_WIDTH,
DEFAULT_ATTRIBUTION,
DOMAIN,
)
from homeassistant.components.unifiprotect.const import CONF_DISABLE_RTSP, DOMAIN
from homeassistant.components.unifiprotect.utils import get_camera_base_name
from homeassistant.const import (
ATTR_ATTRIBUTION,
ATTR_ENTITY_ID,
ATTR_SUPPORTED_FEATURES,
STATE_UNAVAILABLE,
Platform,
)
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import (
device_registry as dr,
entity_registry as er,
issue_registry as ir,
)
from homeassistant.setup import async_setup_component
from homeassistant.config_entries import SOURCE_REAUTH, ConfigEntryState
from homeassistant.const import ATTR_ENTITY_ID, Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er, issue_registry as ir
from . import patch_ufp_method
from .utils import (
Camera,
MockUFPFixture,
adopt_devices,
assert_entity_counts,
enable_entity,
init_entry,
remove_entities,
time_changed,
)
class MockWebRTCProvider(CameraWebRTCProvider):
"""WebRTC provider."""
@property
def domain(self) -> str:
"""Return the integration domain of the provider."""
return DOMAIN
@callback
def async_is_supported(self, stream_source: str) -> bool:
"""Return if this provider is supports the Camera as source."""
return True
async def async_handle_async_webrtc_offer(
self,
camera: Camera,
offer_sdp: str,
session_id: str,
send_message: WebRTCSendMessage,
) -> None:
"""Handle the WebRTC offer and return the answer via the provided callback."""
async def async_on_webrtc_candidate(
self, session_id: str, candidate: RTCIceCandidateInit
) -> None:
"""Handle the WebRTC candidate."""
@callback
def async_close_session(self, session_id: str) -> None:
"""Close the session."""
def _channel_entity_id(camera_obj: ProtectCamera, channel_id: int) -> str:
"""Return the entity_id for a camera channel."""
channel = camera_obj.channels[channel_id]
base_name = get_camera_base_name(channel)
return f"camera.{camera_obj.name}_{base_name}".replace(" ", "_").lower()
@pytest.fixture
async def web_rtc_provider(hass: HomeAssistant) -> None:
"""Fixture to enable WebRTC provider for camera entities."""
await async_setup_component(hass, "camera", {})
async_register_webrtc_provider(hass, MockWebRTCProvider())
def validate_default_camera_entity(
def _assert_entity(
hass: HomeAssistant,
camera_obj: ProtectCamera,
channel_id: int,
*,
enabled: bool,
) -> str:
"""Validate a camera entity."""
channel = camera_obj.channels[channel_id]
camera_name = get_camera_base_name(channel)
entity_name = f"{camera_obj.name} {camera_name}"
unique_id = f"{camera_obj.mac}_{channel.id}"
entity_id = f"camera.{entity_name.replace(' ', '_').lower()}"
"""Assert a camera entity exists with the secure unique_id and no insecure twin."""
entity_id = _channel_entity_id(camera_obj, channel_id)
entity_registry = er.async_get(hass)
entity = entity_registry.async_get(entity_id)
assert entity
assert entity.disabled is False
assert entity.unique_id == unique_id
device_registry = dr.async_get(hass)
device = device_registry.async_get(entity.device_id)
assert device
assert device.manufacturer == "Ubiquiti"
assert device.name == camera_obj.name
assert device.model == camera_obj.market_name or camera_obj.type
assert device.model_id == camera_obj.type
return entity_id
def validate_rtsps_camera_entity(
hass: HomeAssistant,
camera_obj: ProtectCamera,
channel_id: int,
) -> str:
"""Validate a disabled RTSPS camera entity."""
channel = camera_obj.channels[channel_id]
entity_name = f"{camera_obj.name} {channel.name} Resolution Channel"
unique_id = f"{camera_obj.mac}_{channel.id}"
entity_id = f"camera.{entity_name.replace(' ', '_').lower()}"
entity_registry = er.async_get(hass)
entity = entity_registry.async_get(entity_id)
assert entity
assert entity.disabled is True
assert entity.unique_id == unique_id
return entity_id
def validate_rtsp_camera_entity(
hass: HomeAssistant,
camera_obj: ProtectCamera,
channel_id: int,
) -> str:
"""Validate a disabled RTSP camera entity."""
channel = camera_obj.channels[channel_id]
entity_name = f"{camera_obj.name} {channel.name} Resolution Channel (Insecure)"
unique_id = f"{camera_obj.mac}_{channel.id}_insecure"
entity_id = (
"camera."
f"{entity_name.replace(' ', '_').replace('(', '').replace(')', '').lower()}"
assert entity.disabled is not enabled
assert entity.unique_id == f"{camera_obj.mac}_{camera_obj.channels[channel_id].id}"
# The legacy insecure private-path entity must not exist anymore.
assert (
entity_registry.async_get(f"{entity_id}_insecure".replace(" ", "_").lower())
is None
)
entity_registry = er.async_get(hass)
entity = entity_registry.async_get(entity_id)
assert entity
assert entity.disabled is True
assert entity.unique_id == unique_id
return entity_id
def validate_common_camera_state(
hass: HomeAssistant,
channel: CameraChannel,
entity_id: str,
features: int = CameraEntityFeature.STREAM,
):
"""Validate state that is common to all camera entity, regardless of type."""
entity_state = hass.states.get(entity_id)
assert entity_state
assert entity_state.attributes[ATTR_ATTRIBUTION] == DEFAULT_ATTRIBUTION
assert entity_state.attributes[ATTR_SUPPORTED_FEATURES] == features
assert entity_state.attributes[ATTR_WIDTH] == channel.width
assert entity_state.attributes[ATTR_HEIGHT] == channel.height
assert entity_state.attributes[ATTR_FPS] == channel.fps
assert entity_state.attributes[ATTR_BITRATE] == channel.bitrate
assert entity_state.attributes[ATTR_CHANNEL_ID] == channel.id
async def validate_rtsps_camera_state(
hass: HomeAssistant,
camera_obj: ProtectCamera,
channel_id: int,
entity_id: str,
features: int = CameraEntityFeature.STREAM,
):
"""Validate a camera's state."""
channel = camera_obj.channels[channel_id]
assert await async_get_stream_source(hass, entity_id) == channel.rtsps_no_srtp_url
validate_common_camera_state(hass, channel, entity_id, features)
async def validate_rtsp_camera_state(
hass: HomeAssistant,
camera_obj: ProtectCamera,
channel_id: int,
entity_id: str,
features: int = CameraEntityFeature.STREAM,
):
"""Validate a camera's state."""
channel = camera_obj.channels[channel_id]
assert await async_get_stream_source(hass, entity_id) == channel.rtsp_url
validate_common_camera_state(hass, channel, entity_id, features)
async def validate_no_stream_camera_state(
hass: HomeAssistant,
camera_obj: ProtectCamera,
channel_id: int,
entity_id: str,
features: int = CameraEntityFeature.STREAM,
):
"""Validate a camera's state."""
channel = camera_obj.channels[channel_id]
assert await async_get_stream_source(hass, entity_id) is None
validate_common_camera_state(hass, channel, entity_id, features)
async def test_basic_setup(
hass: HomeAssistant,
ufp: MockUFPFixture,
camera_all: ProtectCamera,
doorbell: ProtectCamera,
hass: HomeAssistant, ufp: MockUFPFixture, camera_all: ProtectCamera
) -> None:
"""Test working setup of unifiprotect entry."""
"""One enabled high entity plus disabled medium/low, all secure."""
await init_entry(hass, ufp, [camera_all])
camera_high_only = camera_all.model_copy()
camera_high_only.channels = [c.model_copy() for c in camera_all.channels]
camera_high_only.name = "Test Camera 1"
camera_high_only.channels[0].is_rtsp_enabled = True
camera_high_only.channels[1].is_rtsp_enabled = False
camera_high_only.channels[2].is_rtsp_enabled = False
# high (enabled) + medium + low (disabled); no insecure entities
assert_entity_counts(hass, Platform.CAMERA, 3, 1)
camera_medium_only = camera_all.model_copy()
camera_medium_only.channels = [c.model_copy() for c in camera_all.channels]
camera_medium_only.name = "Test Camera 2"
camera_medium_only.channels[0].is_rtsp_enabled = False
camera_medium_only.channels[1].is_rtsp_enabled = True
camera_medium_only.channels[2].is_rtsp_enabled = False
camera_all.name = "Test Camera 3"
camera_no_channels = camera_all.model_copy()
camera_no_channels.channels = [c.model_copy() for c in camera_all.channels]
camera_no_channels.name = "Test Camera 4"
camera_no_channels.channels[0].is_rtsp_enabled = False
camera_no_channels.channels[1].is_rtsp_enabled = False
camera_no_channels.channels[2].is_rtsp_enabled = False
doorbell.name = "Test Camera 5"
devices = [
camera_high_only,
camera_medium_only,
camera_all,
camera_no_channels,
doorbell,
]
await init_entry(hass, ufp, devices)
assert_entity_counts(hass, Platform.CAMERA, 14, 6)
# test camera 1
entity_id = validate_default_camera_entity(hass, camera_high_only, 0)
await validate_rtsps_camera_state(hass, camera_high_only, 0, entity_id)
entity_id = validate_rtsp_camera_entity(hass, camera_high_only, 0)
await enable_entity(hass, ufp.entry.entry_id, entity_id)
await validate_rtsp_camera_state(hass, camera_high_only, 0, entity_id)
# test camera 2
entity_id = validate_default_camera_entity(hass, camera_medium_only, 1)
await validate_rtsps_camera_state(hass, camera_medium_only, 1, entity_id)
entity_id = validate_rtsp_camera_entity(hass, camera_medium_only, 1)
await enable_entity(hass, ufp.entry.entry_id, entity_id)
await validate_rtsp_camera_state(hass, camera_medium_only, 1, entity_id)
# test camera 3
entity_id = validate_default_camera_entity(hass, camera_all, 0)
await validate_rtsps_camera_state(hass, camera_all, 0, entity_id)
entity_id = validate_rtsp_camera_entity(hass, camera_all, 0)
await enable_entity(hass, ufp.entry.entry_id, entity_id)
await validate_rtsp_camera_state(hass, camera_all, 0, entity_id)
entity_id = validate_rtsps_camera_entity(hass, camera_all, 1)
await enable_entity(hass, ufp.entry.entry_id, entity_id)
await validate_rtsps_camera_state(hass, camera_all, 1, entity_id)
entity_id = validate_rtsp_camera_entity(hass, camera_all, 1)
await enable_entity(hass, ufp.entry.entry_id, entity_id)
await validate_rtsp_camera_state(hass, camera_all, 1, entity_id)
entity_id = validate_rtsps_camera_entity(hass, camera_all, 2)
await enable_entity(hass, ufp.entry.entry_id, entity_id)
await validate_rtsps_camera_state(hass, camera_all, 2, entity_id)
entity_id = validate_rtsp_camera_entity(hass, camera_all, 2)
await enable_entity(hass, ufp.entry.entry_id, entity_id)
await validate_rtsp_camera_state(hass, camera_all, 2, entity_id)
# test camera 4
entity_id = validate_default_camera_entity(hass, camera_no_channels, 0)
await validate_no_stream_camera_state(
hass, camera_no_channels, 0, entity_id, features=0
high_id = _assert_entity(hass, camera_all, 0, enabled=True)
assert (
await async_get_stream_source(hass, high_id)
== camera_all.channels[0].rtsps_no_srtp_url
)
# test camera 5
entity_id = validate_default_camera_entity(hass, doorbell, 0)
await validate_rtsps_camera_state(hass, doorbell, 0, entity_id)
# medium starts disabled; once enabled it streams from the public API
medium_id = _assert_entity(hass, camera_all, 1, enabled=False)
await enable_entity(hass, ufp.entry.entry_id, medium_id)
assert (
await async_get_stream_source(hass, medium_id)
== camera_all.channels[1].rtsps_no_srtp_url
)
entity_id = validate_rtsp_camera_entity(hass, doorbell, 0)
await enable_entity(hass, ufp.entry.entry_id, entity_id)
await validate_rtsp_camera_state(hass, doorbell, 0, entity_id)
entity_id = validate_default_camera_entity(hass, doorbell, 3)
await validate_no_stream_camera_state(hass, doorbell, 3, entity_id, features=0)
_assert_entity(hass, camera_all, 2, enabled=False)
@pytest.mark.usefixtures("web_rtc_provider")
async def test_webrtc_support(
hass: HomeAssistant,
ufp: MockUFPFixture,
camera_all: ProtectCamera,
async def test_doorbell_setup(
hass: HomeAssistant, ufp: MockUFPFixture, doorbell: ProtectCamera
) -> None:
"""Test webrtc support is available."""
camera_high_only = camera_all.model_copy()
camera_high_only.channels = [c.model_copy() for c in camera_all.channels]
camera_high_only.name = "Test Camera 1"
camera_high_only.channels[0].is_rtsp_enabled = True
camera_high_only.channels[1].is_rtsp_enabled = False
camera_high_only.channels[2].is_rtsp_enabled = False
await init_entry(hass, ufp, [camera_high_only])
entity_id = validate_default_camera_entity(hass, camera_high_only, 0)
assert hass.states.get(entity_id)
camera_obj = get_camera_from_entity_id(hass, entity_id)
assert camera_obj.camera_capabilities == CameraCapabilities(
{StreamType.HLS, StreamType.WEB_RTC}
"""Doorbell exposes its active qualities; the package channel is just one of them."""
await init_entry(hass, ufp, [doorbell])
# high (enabled) + package (active too, disabled by default); both stream
assert_entity_counts(hass, Platform.CAMERA, 2, 1)
high_id = _assert_entity(hass, doorbell, 0, enabled=True)
assert (
await async_get_stream_source(hass, high_id)
== doorbell.channels[0].rtsps_no_srtp_url
)
# the package channel is not special-cased: it streams like any quality tier
package_id = _assert_entity(hass, doorbell, 3, enabled=False)
await enable_entity(hass, ufp.entry.entry_id, package_id)
assert (
await async_get_stream_source(hass, package_id)
== doorbell.channels[3].rtsps_no_srtp_url
)
async def test_adopt(
hass: HomeAssistant, ufp: MockUFPFixture, camera: ProtectCamera
async def test_first_active_quality_is_default(
hass: HomeAssistant,
ufp: MockUFPFixture,
camera_all: ProtectCamera,
issue_registry: ir.IssueRegistry,
) -> None:
"""Test setting up camera with no camera channels."""
"""The first active quality is the default; inactive ones get no entity."""
camera_all.channels = [c.model_copy() for c in camera_all.channels]
camera_all.channels[0].is_rtsp_enabled = False # high inactive
camera_all.channels[1].is_rtsp_enabled = True # medium active
camera_all.channels[2].is_rtsp_enabled = False # low inactive
camera1 = camera.model_copy()
camera1.channels = []
await init_entry(hass, ufp, [camera_all])
await init_entry(hass, ufp, [camera1])
assert_entity_counts(hass, Platform.CAMERA, 0, 0)
# only medium exists, enabled by default, with a working stream
assert_entity_counts(hass, Platform.CAMERA, 1, 1)
medium_id = _assert_entity(hass, camera_all, 1, enabled=True)
assert (
await async_get_stream_source(hass, medium_id)
== camera_all.channels[1].rtsps_no_srtp_url
)
await remove_entities(hass, ufp, [camera1])
assert_entity_counts(hass, Platform.CAMERA, 0, 0)
camera1.channels = []
await adopt_devices(hass, ufp, [camera1])
assert_entity_counts(hass, Platform.CAMERA, 0, 0)
entity_registry = er.async_get(hass)
assert entity_registry.async_get(_channel_entity_id(camera_all, 0)) is None
assert entity_registry.async_get(_channel_entity_id(camera_all, 2)) is None
assert (
issue_registry.async_get_issue(DOMAIN, f"rtsp_disabled_{camera_all.id}") is None
)
camera1.channels = camera.channels
for channel in camera1.channels:
channel._api = ufp.api
mock_msg = Mock()
mock_msg.changed_data = {"channels": camera.channels}
mock_msg.new_obj = camera1
ufp.ws_msg(mock_msg)
await hass.async_block_till_done()
assert_entity_counts(hass, Platform.CAMERA, 2, 1)
async def test_no_active_stream(
hass: HomeAssistant,
ufp: MockUFPFixture,
camera: ProtectCamera,
issue_registry: ir.IssueRegistry,
) -> None:
"""A camera with no active stream is exposed for snapshots and raises a repair."""
camera.channels = [c.model_copy() for c in camera.channels]
for channel in camera.channels:
channel.is_rtsp_enabled = False
await remove_entities(hass, ufp, [camera1])
assert_entity_counts(hass, Platform.CAMERA, 0, 0)
await adopt_devices(hass, ufp, [camera1])
assert_entity_counts(hass, Platform.CAMERA, 2, 1)
await init_entry(hass, ufp, [camera])
high_id = _assert_entity(hass, camera, 0, enabled=True)
assert await async_get_stream_source(hass, high_id) is None
state = hass.states.get(high_id)
assert state
assert state.attributes["supported_features"] == CameraEntityFeature(0)
assert (
issue_registry.async_get_issue(DOMAIN, f"rtsp_disabled_{camera.id}") is not None
)
async def test_offline_camera_no_repair(
hass: HomeAssistant,
ufp: MockUFPFixture,
camera: ProtectCamera,
issue_registry: ir.IssueRegistry,
) -> None:
"""An offline camera with no stream gets no repair (offline, not disabled)."""
camera.state = StateType.DISCONNECTED
await init_entry(hass, ufp, [camera])
_assert_entity(hass, camera, 0, enabled=True)
assert issue_registry.async_get_issue(DOMAIN, f"rtsp_disabled_{camera.id}") is None
@pytest.mark.parametrize("ufp_options", [{CONF_DISABLE_RTSP: True}], indirect=True)
async def test_disable_rtsp(
hass: HomeAssistant, ufp: MockUFPFixture, camera_all: ProtectCamera
) -> None:
"""Disabling RTSP globally removes the stream source."""
await init_entry(hass, ufp, [camera_all])
high_id = _assert_entity(hass, camera_all, 0, enabled=True)
assert await async_get_stream_source(hass, high_id) is None
async def test_camera_image(
hass: HomeAssistant, ufp: MockUFPFixture, camera: ProtectCamera
) -> None:
"""Test retrieving camera image."""
"""Main snapshot is fetched from the public API."""
await init_entry(hass, ufp, [camera])
assert_entity_counts(hass, Platform.CAMERA, 2, 1)
assert_entity_counts(hass, Platform.CAMERA, 1, 1)
ufp.api.get_public_api_camera_snapshot = AsyncMock()
await async_get_image(hass, "camera.test_camera_high_resolution_channel")
await async_get_image(hass, _channel_entity_id(camera, 0))
ufp.api.get_public_api_camera_snapshot.assert_called_once()
assert ufp.api.get_public_api_camera_snapshot.call_args.kwargs["package"] is False
async def test_package_camera_image(
hass: HomeAssistant, ufp: MockUFPFixture, doorbell: ProtectCamera
) -> None:
"""Test retrieving package camera image."""
"""Package snapshot is fetched from the public API with package=True."""
await init_entry(hass, ufp, [doorbell])
assert_entity_counts(hass, Platform.CAMERA, 2, 1)
package_id = _channel_entity_id(doorbell, 3)
await enable_entity(hass, ufp.entry.entry_id, package_id)
ufp.api.get_public_api_camera_snapshot = AsyncMock()
await async_get_image(hass, package_id)
ufp.api.get_public_api_camera_snapshot.assert_called_once()
assert ufp.api.get_public_api_camera_snapshot.call_args.kwargs["package"] is True
async def test_package_camera_without_stream(
hass: HomeAssistant, ufp: MockUFPFixture, doorbell: ProtectCamera
) -> None:
"""The package camera stays available for snapshots when its stream is off."""
doorbell.channels = [c.model_copy() for c in doorbell.channels]
doorbell.channels[3].is_rtsp_enabled = False # package stream off (the default)
await init_entry(hass, ufp, [doorbell])
assert_entity_counts(hass, Platform.CAMERA, 3, 2)
ufp.api.get_package_camera_snapshot = AsyncMock()
# high (enabled, streaming) + package (disabled, snapshot-only) still present
assert_entity_counts(hass, Platform.CAMERA, 2, 1)
await async_get_image(hass, "camera.test_camera_package_camera")
ufp.api.get_package_camera_snapshot.assert_called_once()
package_id = _assert_entity(hass, doorbell, 3, enabled=False)
await enable_entity(hass, ufp.entry.entry_id, package_id)
assert await async_get_stream_source(hass, package_id) is None
state = hass.states.get(package_id)
assert state
assert state.attributes["supported_features"] == CameraEntityFeature(0)
# snapshots still work, fetched with package=True
ufp.api.get_public_api_camera_snapshot = AsyncMock()
await async_get_image(hass, package_id)
assert ufp.api.get_public_api_camera_snapshot.call_args.kwargs["package"] is True
async def test_camera_generic_update(
hass: HomeAssistant, ufp: MockUFPFixture, camera: ProtectCamera
async def test_package_only_camera(
hass: HomeAssistant,
ufp: MockUFPFixture,
camera: ProtectCamera,
issue_registry: ir.IssueRegistry,
) -> None:
"""Tests generic entity update service."""
"""A camera with only a package channel still exposes a snapshot entity."""
package = camera.channels[0].model_copy()
package.id = 3
package.fps = 2
package.is_rtsp_enabled = False
camera.channels = [package]
await init_entry(hass, ufp, [camera])
assert_entity_counts(hass, Platform.CAMERA, 2, 1)
entity_id = "camera.test_camera_high_resolution_channel"
assert await async_setup_component(hass, HOMEASSISTANT_DOMAIN, {})
# only the package snapshot entity (disabled by default); no main fallback, no repair
assert_entity_counts(hass, Platform.CAMERA, 1, 0)
_assert_entity(hass, camera, 0, enabled=False)
assert issue_registry.async_get_issue(DOMAIN, f"rtsp_disabled_{camera.id}") is None
state = hass.states.get(entity_id)
assert state and state.state == "idle"
ufp.api.update = AsyncMock(return_value=None)
await hass.services.async_call(
HOMEASSISTANT_DOMAIN,
SERVICE_UPDATE_ENTITY,
{ATTR_ENTITY_ID: entity_id},
blocking=True,
async def test_no_channels(
hass: HomeAssistant, ufp: MockUFPFixture, camera: ProtectCamera
) -> None:
"""A camera without channels yet creates no entities."""
camera.channels = []
await init_entry(hass, ufp, [camera])
assert_entity_counts(hass, Platform.CAMERA, 0, 0)
async def test_streams_unavailable(
hass: HomeAssistant, ufp: MockUFPFixture, camera_all: ProtectCamera
) -> None:
"""A camera the library leaves unprimed (no streams) has no stream source."""
async def _prime_streamless() -> Any:
pb = ufp.api.public_bootstrap
pb.cameras = {
camera_all.id: SimpleNamespace(
id=camera_all.id, state=camera_all.state, rtsps_streams=None
)
}
return pb
ufp.api.update_public = AsyncMock(side_effect=_prime_streamless)
await init_entry(hass, ufp, [camera_all])
high_id = _channel_entity_id(camera_all, 0)
assert await async_get_stream_source(hass, high_id) is None
async def test_public_bootstrap_failure_not_ready(
hass: HomeAssistant, ufp: MockUFPFixture, camera: ProtectCamera
) -> None:
"""A failed public bootstrap prime leaves the entry in setup-retry.
The websockets/poll subscriptions opened during setup must be torn down so
a retry does not leak another set.
"""
ufp.api.update_public = AsyncMock(side_effect=ClientError("boom"))
await init_entry(hass, ufp, [camera])
assert ufp.entry.state is ConfigEntryState.SETUP_RETRY
ufp.api.async_disconnect_ws.assert_called()
async def test_public_bootstrap_revoked_key_reauth(
hass: HomeAssistant, ufp: MockUFPFixture, camera: ProtectCamera
) -> None:
"""A revoked API key (public 401) triggers reauth, not an endless retry."""
ufp.api.update_public = AsyncMock(side_effect=NotAuthorized("revoked"))
await init_entry(hass, ufp, [camera])
assert ufp.entry.state is ConfigEntryState.SETUP_ERROR
ufp.api.async_disconnect_ws.assert_called()
assert any(ufp.entry.async_get_active_flows(hass, {SOURCE_REAUTH}))
async def test_adopt(
hass: HomeAssistant, ufp: MockUFPFixture, camera_all: ProtectCamera
) -> None:
"""A camera adopted at runtime loads its public streams before entities."""
await init_entry(hass, ufp, [camera_all])
assert_entity_counts(hass, Platform.CAMERA, 3, 1)
await remove_entities(hass, ufp, [camera_all])
assert_entity_counts(hass, Platform.CAMERA, 0, 0)
await adopt_devices(hass, ufp, [camera_all])
assert_entity_counts(hass, Platform.CAMERA, 3, 1)
high_id = _assert_entity(hass, camera_all, 0, enabled=True)
assert (
await async_get_stream_source(hass, high_id)
== camera_all.channels[0].rtsps_no_srtp_url
)
state = hass.states.get(entity_id)
assert state and state.state == "idle"
async def test_camera_interval_update(
hass: HomeAssistant, ufp: MockUFPFixture, camera: ProtectCamera
) -> None:
"""Interval updates updates camera entity."""
await init_entry(hass, ufp, [camera])
assert_entity_counts(hass, Platform.CAMERA, 2, 1)
entity_id = "camera.test_camera_high_resolution_channel"
state = hass.states.get(entity_id)
assert state and state.state == "idle"
new_camera = camera.model_copy()
new_camera.is_recording = True
ufp.api.bootstrap.cameras = {new_camera.id: new_camera}
ufp.api.update = AsyncMock(return_value=ufp.api.bootstrap)
await time_changed(hass, DEVICE_UPDATE_INTERVAL)
state = hass.states.get(entity_id)
assert state and state.state == "recording"
async def test_camera_bad_interval_update(
hass: HomeAssistant, ufp: MockUFPFixture, camera: ProtectCamera
) -> None:
"""Interval updates marks camera unavailable."""
await init_entry(hass, ufp, [camera])
assert_entity_counts(hass, Platform.CAMERA, 2, 1)
entity_id = "camera.test_camera_high_resolution_channel"
state = hass.states.get(entity_id)
assert state and state.state == "idle"
# update fails
ufp.api.update = AsyncMock(side_effect=NvrError)
await time_changed(hass, DEVICE_UPDATE_INTERVAL)
state = hass.states.get(entity_id)
assert state and state.state == "unavailable"
# next update succeeds
ufp.api.update = AsyncMock(return_value=ufp.api.bootstrap)
await time_changed(hass, DEVICE_UPDATE_INTERVAL)
state = hass.states.get(entity_id)
assert state and state.state == "idle"
async def test_camera_websocket_disconnected(
hass: HomeAssistant, ufp: MockUFPFixture, camera: ProtectCamera
) -> None:
"""Test the websocket gets disconnected and reconnected."""
await init_entry(hass, ufp, [camera])
assert_entity_counts(hass, Platform.CAMERA, 2, 1)
entity_id = "camera.test_camera_high_resolution_channel"
state = hass.states.get(entity_id)
assert state and state.state == CameraState.IDLE
# websocket disconnects
ufp.ws_state_subscription(WebsocketState.DISCONNECTED)
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state and state.state == STATE_UNAVAILABLE
# websocket reconnects
ufp.ws_state_subscription(WebsocketState.CONNECTED)
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state and state.state == CameraState.IDLE
async def test_camera_ws_update(
hass: HomeAssistant, ufp: MockUFPFixture, camera: ProtectCamera
) -> None:
"""WS update updates camera entity."""
await init_entry(hass, ufp, [camera])
assert_entity_counts(hass, Platform.CAMERA, 2, 1)
entity_id = "camera.test_camera_high_resolution_channel"
state = hass.states.get(entity_id)
assert state and state.state == "idle"
new_camera = camera.model_copy()
new_camera.is_recording = True
no_camera = camera.model_copy()
no_camera.is_adopted = False
ufp.api.bootstrap.cameras = {new_camera.id: new_camera}
mock_msg = Mock()
mock_msg.changed_data = {}
mock_msg.new_obj = new_camera
ufp.ws_msg(mock_msg)
mock_msg = Mock()
mock_msg.changed_data = {}
mock_msg.new_obj = no_camera
ufp.ws_msg(mock_msg)
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state and state.state == "recording"
async def test_camera_ws_update_offline(
hass: HomeAssistant, ufp: MockUFPFixture, camera: ProtectCamera
) -> None:
"""WS updates marks camera unavailable."""
await init_entry(hass, ufp, [camera])
assert_entity_counts(hass, Platform.CAMERA, 2, 1)
entity_id = "camera.test_camera_high_resolution_channel"
state = hass.states.get(entity_id)
assert state and state.state == "idle"
# camera goes offline
new_camera = camera.model_copy()
new_camera.state = StateType.DISCONNECTED
mock_msg = Mock()
mock_msg.changed_data = {}
mock_msg.new_obj = new_camera
ufp.api.bootstrap.cameras = {new_camera.id: new_camera}
ufp.ws_msg(mock_msg)
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state and state.state == "unavailable"
# camera comes back online
new_camera.state = StateType.CONNECTED
mock_msg = Mock()
mock_msg.changed_data = {}
mock_msg.new_obj = new_camera
ufp.api.bootstrap.cameras = {new_camera.id: new_camera}
ufp.ws_msg(mock_msg)
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state and state.state == "idle"
@pytest.mark.parametrize(
("service", "expected_value"),
@@ -620,10 +358,10 @@ async def test_camera_motion_detection(
service: str,
expected_value: bool,
) -> None:
"""Test enabling/disabling motion detection on camera."""
"""Test enabling/disabling motion detection on a camera."""
await init_entry(hass, ufp, [camera])
assert_entity_counts(hass, Platform.CAMERA, 2, 1)
entity_id = "camera.test_camera_high_resolution_channel"
assert_entity_counts(hass, Platform.CAMERA, 1, 1)
entity_id = _channel_entity_id(camera, 0)
with patch_ufp_method(
camera, "set_motion_detection", new_callable=AsyncMock
@@ -634,42 +372,28 @@ async def test_camera_motion_detection(
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
mock_method.assert_called_once_with(expected_value)
async def test_aiport_no_camera_entities(
hass: HomeAssistant,
ufp: MockUFPFixture,
aiport: AiPort,
hass: HomeAssistant, ufp: MockUFPFixture, aiport: AiPort
) -> None:
"""Test that AI Port devices do not create camera entities."""
"""AI Port devices do not create camera entities."""
await init_entry(hass, ufp, [aiport])
# AI Port should not create any camera entities
assert_entity_counts(hass, Platform.CAMERA, 0, 0)
async def test_aiport_rtsp_issue_cleanup(
async def test_aiport_stream_issue_cleanup(
hass: HomeAssistant,
ufp: MockUFPFixture,
aiport: AiPort,
issue_registry: ir.IssueRegistry,
) -> None:
"""Test that RTSP disabled issues for AI Ports are cleaned up on setup."""
# Set up the integration with the AI Port first
# (init_entry regenerates IDs, so we need to get the new ID)
"""Stale public-stream issues for AI Ports are cleaned up on setup."""
await init_entry(hass, ufp, [aiport])
# Now get the actual AI Port ID after regeneration
actual_aiport_id = aiport.id
# Create an RTSP disabled issue for the AI Port
# (simulating an issue that might have been created by a previous buggy version)
issue_id = f"rtsp_disabled_{actual_aiport_id}"
# Get the issue registry and create the issue directly via internal method
# to avoid translation validation (as we're simulating a legacy issue)
issue_registry = ir.async_get(hass)
issue_id = f"rtsp_disabled_{aiport.id}"
# Simulate a legacy issue created directly (bypass translation validation).
issue_registry.issues[(DOMAIN, issue_id)] = ir.IssueEntry(
active=True,
breaks_in_ha_version=None,
@@ -686,16 +410,28 @@ async def test_aiport_rtsp_issue_cleanup(
translation_key="rtsp_disabled",
translation_placeholders=None,
)
# Verify the issue exists
assert issue_registry.async_get_issue(DOMAIN, issue_id) is not None
# Reload the integration - this should clean up the issue
await hass.config_entries.async_reload(ufp.entry.entry_id)
await hass.async_block_till_done()
# The issue should be cleaned up since AI Ports can't have RTSP
assert issue_registry.async_get_issue(DOMAIN, issue_id) is None
# Verify no camera entities were created
assert_entity_counts(hass, Platform.CAMERA, 0, 0)
async def test_snapshot_low_quality_without_stream(
hass: HomeAssistant, ufp: MockUFPFixture, camera: ProtectCamera
) -> None:
"""Without a live stream, snapshots are requested at low quality."""
camera.channels = [c.model_copy() for c in camera.channels]
for channel in camera.channels:
channel.is_rtsp_enabled = False
await init_entry(hass, ufp, [camera])
ufp.api.get_public_api_camera_snapshot = AsyncMock()
await async_get_image(hass, _channel_entity_id(camera, 0))
ufp.api.get_public_api_camera_snapshot.assert_called_once()
assert (
ufp.api.get_public_api_camera_snapshot.call_args.kwargs["high_quality"] is False
)
tests/components/unifiprotect/test_migrate.py
+125 -1
View File
@@ -9,7 +9,7 @@ from homeassistant.components.script import DOMAIN as SCRIPT_DOMAIN
from homeassistant.components.unifiprotect.const import DOMAIN
from homeassistant.const import SERVICE_RELOAD, Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from homeassistant.helpers import entity_registry as er, issue_registry as ir
from homeassistant.setup import async_setup_component
from .utils import MockUFPFixture, init_entry
@@ -218,3 +218,127 @@ async def test_deprecate_entity_script(
if i["issue_id"] == "deprecate_hdr_switch":
issue = i
assert issue is None
async def test_migrate_insecure_camera_redirected(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
ufp: MockUFPFixture,
doorbell: Camera,
) -> None:
"""A legacy insecure camera entity is redirected to the secure stream."""
insecure = entity_registry.async_get_or_create(
Platform.CAMERA,
DOMAIN,
f"{doorbell.mac}_0_insecure",
config_entry=ufp.entry,
)
await init_entry(hass, ufp, [doorbell], regenerate_ids=False)
# the insecure entity now carries the secure unique_id (history preserved)
migrated = entity_registry.async_get(insecure.entity_id)
assert migrated is not None
assert migrated.unique_id == f"{doorbell.mac}_0"
assert (
entity_registry.async_get_entity_id(
Platform.CAMERA, DOMAIN, f"{doorbell.mac}_0_insecure"
)
is None
)
async def test_migrate_insecure_camera_removed(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
issue_registry: ir.IssueRegistry,
ufp: MockUFPFixture,
doorbell: Camera,
) -> None:
"""A redundant, unused insecure entity is removed silently."""
entity_registry.async_get_or_create(
Platform.CAMERA, DOMAIN, f"{doorbell.mac}_0", config_entry=ufp.entry
)
insecure = entity_registry.async_get_or_create(
Platform.CAMERA,
DOMAIN,
f"{doorbell.mac}_0_insecure",
config_entry=ufp.entry,
)
await init_entry(hass, ufp, [doorbell], regenerate_ids=False)
assert entity_registry.async_get(insecure.entity_id) is None
assert (
entity_registry.async_get_entity_id(
Platform.CAMERA, DOMAIN, f"{doorbell.mac}_0"
)
is not None
)
assert (
issue_registry.async_get_issue(
DOMAIN, f"insecure_camera_removed_{doorbell.mac}_0_insecure"
)
is None
)
async def test_migrate_insecure_camera_removed_in_use(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
issue_registry: ir.IssueRegistry,
ufp: MockUFPFixture,
doorbell: Camera,
) -> None:
"""Removing an insecure entity that is still used raises an actionable repair."""
secure = entity_registry.async_get_or_create(
Platform.CAMERA, DOMAIN, f"{doorbell.mac}_0", config_entry=ufp.entry
)
insecure = entity_registry.async_get_or_create(
Platform.CAMERA,
DOMAIN,
f"{doorbell.mac}_0_insecure",
config_entry=ufp.entry,
)
await _load_automation(hass, insecure.entity_id)
await init_entry(hass, ufp, [doorbell], regenerate_ids=False)
assert entity_registry.async_get(insecure.entity_id) is None
issue = issue_registry.async_get_issue(
DOMAIN, f"insecure_camera_removed_{doorbell.mac}_0_insecure"
)
assert issue is not None
assert issue.translation_placeholders["entity_id"] == insecure.entity_id
assert issue.translation_placeholders["replacement"] == secure.entity_id
async def test_migrate_insecure_camera_removed_disabled_not_repaired(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
issue_registry: ir.IssueRegistry,
ufp: MockUFPFixture,
doorbell: Camera,
) -> None:
"""A disabled insecure entity is removed without a repair even if referenced."""
entity_registry.async_get_or_create(
Platform.CAMERA, DOMAIN, f"{doorbell.mac}_0", config_entry=ufp.entry
)
insecure = entity_registry.async_get_or_create(
Platform.CAMERA,
DOMAIN,
f"{doorbell.mac}_0_insecure",
config_entry=ufp.entry,
disabled_by=er.RegistryEntryDisabler.USER,
)
await _load_automation(hass, insecure.entity_id)
await init_entry(hass, ufp, [doorbell], regenerate_ids=False)
assert entity_registry.async_get(insecure.entity_id) is None
assert (
issue_registry.async_get_issue(
DOMAIN, f"insecure_camera_removed_{doorbell.mac}_0_insecure"
)
is None
)
tests/components/unifiprotect/test_repairs.py
+150 -190
View File
@@ -1,20 +1,29 @@
"""Test repairs for unifiprotect."""
from copy import deepcopy
from unittest.mock import AsyncMock
from aiohttp.test_utils import TestClient
import pytest
from uiprotect.api import RTSPSStreams
from uiprotect.data import Camera, CloudAccount, Version
from uiprotect.exceptions import NotAuthorized
from homeassistant.components.repairs import ConfirmRepairFlow
from homeassistant.components.unifiprotect.const import CONF_DISABLE_RTSP, DOMAIN
from homeassistant.components.unifiprotect.repairs import async_create_fix_flow
from homeassistant.config_entries import SOURCE_REAUTH
from homeassistant.core import HomeAssistant
from homeassistant.helpers import issue_registry as ir
from homeassistant.setup import async_setup_component
from .utils import MockUFPFixture, init_entry
from tests.components.repairs import process_repair_fix_flow, start_repair_fix_flow
from tests.typing import ClientSessionGenerator, WebSocketGenerator
_ACTIVE_STREAMS = RTSPSStreams(high="rtsps://example.test:7441/abc?enableSrtp")
_NO_STREAMS = RTSPSStreams()
async def test_cloud_user_fix(
hass: HomeAssistant,
@@ -30,6 +39,7 @@ async def test_cloud_user_fix(
user.cloud_account = cloud_account
ufp.api.bootstrap.users[ufp.api.bootstrap.auth_user_id] = user
await init_entry(hass, ufp, [])
assert await async_setup_component(hass, "repairs", {})
ws_client = await hass_ws_client(hass)
client = await hass_client()
@@ -56,240 +66,190 @@ async def test_cloud_user_fix(
assert any(ufp.entry.async_get_active_flows(hass, {SOURCE_REAUTH}))
async def test_rtsp_read_only_ignore(
async def _raise_and_assert_repair(
hass: HomeAssistant,
ufp: MockUFPFixture,
doorbell: Camera,
hass_client: ClientSessionGenerator,
hass_ws_client: WebSocketGenerator,
) -> tuple[TestClient, str]:
"""Set up a streamless camera, assert its repair exists, return (client, issue_id)."""
for channel in doorbell.channels:
channel.is_rtsp_enabled = False
await init_entry(hass, ufp, [doorbell])
assert await async_setup_component(hass, "repairs", {})
ws_client = await hass_ws_client(hass)
client = await hass_client()
issue_id = f"rtsp_disabled_{doorbell.id}"
await ws_client.send_json({"id": 1, "type": "repairs/list_issues"})
msg = await ws_client.receive_json()
assert msg["success"]
assert any(i["issue_id"] == issue_id for i in msg["result"]["issues"])
return client, issue_id
async def test_rtsp_repair_fix(
hass: HomeAssistant,
ufp: MockUFPFixture,
doorbell: Camera,
hass_client: ClientSessionGenerator,
hass_ws_client: WebSocketGenerator,
) -> None:
"""Test RTSP disabled warning if camera is read-only and it is ignored."""
"""A camera without an active stream raises a repair that creates one."""
client, issue_id = await _raise_and_assert_repair(
hass, ufp, doorbell, hass_client, hass_ws_client
)
for channel in doorbell.channels:
channel.is_rtsp_enabled = False
for user in ufp.api.bootstrap.users.values():
user.all_permissions = []
# Model the device: no stream until one is created (via the public API).
state = {"created": False}
ufp.api.get_camera = AsyncMock(return_value=doorbell)
ufp.api.create_camera_rtsps_streams = AsyncMock(return_value=None)
async def get_streams(
camera_id: str, *args: object, **kwargs: object
) -> RTSPSStreams:
return _ACTIVE_STREAMS if state["created"] else _NO_STREAMS
await init_entry(hass, ufp, [doorbell])
ws_client = await hass_ws_client(hass)
client = await hass_client()
async def create_streams(
camera_id: str, qualities: str, *args: object, **kwargs: object
) -> RTSPSStreams:
state["created"] = True
return _ACTIVE_STREAMS
issue_id = f"rtsp_disabled_{doorbell.id}"
await ws_client.send_json({"id": 1, "type": "repairs/list_issues"})
msg = await ws_client.receive_json()
assert msg["success"]
assert len(msg["result"]["issues"]) > 0
issue = None
for i in msg["result"]["issues"]:
if i["issue_id"] == issue_id:
issue = i
assert issue is not None
ufp.api.get_camera_rtsps_streams = AsyncMock(side_effect=get_streams)
ufp.api.create_camera_rtsps_streams = AsyncMock(side_effect=create_streams)
data = await start_repair_fix_flow(client, DOMAIN, issue_id)
flow_id = data["flow_id"]
assert data["step_id"] == "start"
data = await process_repair_fix_flow(client, flow_id)
flow_id = data["flow_id"]
assert data["step_id"] == "confirm"
data = await process_repair_fix_flow(client, flow_id)
assert data["type"] == "create_entry"
async def test_rtsp_read_only_fix(
hass: HomeAssistant,
ufp: MockUFPFixture,
doorbell: Camera,
hass_client: ClientSessionGenerator,
hass_ws_client: WebSocketGenerator,
) -> None:
"""Test RTSP disabled warning if camera is read-only and it is fixed."""
for channel in doorbell.channels:
channel.is_rtsp_enabled = False
for user in ufp.api.bootstrap.users.values():
user.all_permissions = []
await init_entry(hass, ufp, [doorbell])
ws_client = await hass_ws_client(hass)
client = await hass_client()
new_doorbell = deepcopy(doorbell)
new_doorbell.channels[1].is_rtsp_enabled = True
ufp.api.get_camera = AsyncMock(return_value=new_doorbell)
ufp.api.create_camera_rtsps_streams = AsyncMock(return_value=None)
issue_id = f"rtsp_disabled_{doorbell.id}"
await ws_client.send_json({"id": 1, "type": "repairs/list_issues"})
msg = await ws_client.receive_json()
assert msg["success"]
assert len(msg["result"]["issues"]) > 0
issue = None
for i in msg["result"]["issues"]:
if i["issue_id"] == issue_id:
issue = i
assert issue is not None
data = await start_repair_fix_flow(client, DOMAIN, issue_id)
flow_id = data["flow_id"]
assert data["step_id"] == "start"
data = await process_repair_fix_flow(client, flow_id)
assert data["type"] == "create_entry"
async def test_rtsp_writable_fix(
hass: HomeAssistant,
ufp: MockUFPFixture,
doorbell: Camera,
hass_client: ClientSessionGenerator,
hass_ws_client: WebSocketGenerator,
) -> None:
"""Test RTSP disabled warning if camera is writable and it is ignored."""
for channel in doorbell.channels:
channel.is_rtsp_enabled = False
await init_entry(hass, ufp, [doorbell])
ws_client = await hass_ws_client(hass)
client = await hass_client()
new_doorbell = deepcopy(doorbell)
new_doorbell.channels[0].is_rtsp_enabled = True
ufp.api.get_camera = AsyncMock(side_effect=[doorbell, new_doorbell])
ufp.api.create_camera_rtsps_streams = AsyncMock(return_value=None)
issue_id = f"rtsp_disabled_{doorbell.id}"
await ws_client.send_json({"id": 1, "type": "repairs/list_issues"})
msg = await ws_client.receive_json()
assert msg["success"]
assert len(msg["result"]["issues"]) > 0
issue = None
for i in msg["result"]["issues"]:
if i["issue_id"] == issue_id:
issue = i
assert issue is not None
data = await start_repair_fix_flow(client, DOMAIN, issue_id)
flow_id = data["flow_id"]
assert data["step_id"] == "start"
data = await process_repair_fix_flow(client, flow_id)
data = await process_repair_fix_flow(client, data["flow_id"])
assert data["type"] == "create_entry"
ufp.api.create_camera_rtsps_streams.assert_called_with(doorbell.id, "high")
async def test_rtsp_writable_fix_when_not_setup(
@pytest.mark.parametrize(
"create_side_effect",
[
pytest.param(None, id="unresolved"),
pytest.param(NotAuthorized("missing write permission"), id="no_permission"),
],
)
async def test_rtsp_repair_confirm_fallback(
hass: HomeAssistant,
ufp: MockUFPFixture,
doorbell: Camera,
hass_client: ClientSessionGenerator,
hass_ws_client: WebSocketGenerator,
create_side_effect: Exception | None,
) -> None:
"""When a stream cannot be created, the flow falls back to a confirm step.
Covers both an unresolved create (still no active stream) and a permission
error while creating; both route start -> confirm -> create_entry.
"""
client, issue_id = await _raise_and_assert_repair(
hass, ufp, doorbell, hass_client, hass_ws_client
)
ufp.api.get_camera_rtsps_streams = AsyncMock(return_value=_NO_STREAMS)
ufp.api.create_camera_rtsps_streams = AsyncMock(
return_value=_NO_STREAMS, side_effect=create_side_effect
)
data = await start_repair_fix_flow(client, DOMAIN, issue_id)
assert data["step_id"] == "start"
data = await process_repair_fix_flow(client, data["flow_id"])
assert data["step_id"] == "confirm"
data = await process_repair_fix_flow(client, data["flow_id"])
assert data["type"] == "create_entry"
async def test_create_fix_flow_unknown_issue(hass: HomeAssistant) -> None:
"""An issue without matching data falls back to a confirm-only flow."""
flow = await async_create_fix_flow(hass, "some_other_issue", None)
assert isinstance(flow, ConfirmRepairFlow)
async def test_rtsp_repair_when_not_setup(
hass: HomeAssistant,
ufp: MockUFPFixture,
doorbell: Camera,
hass_client: ClientSessionGenerator,
hass_ws_client: WebSocketGenerator,
) -> None:
"""Test RTSP disabled warning if the integration is no longer set up."""
"""The repair works while the entry is unloaded (fresh API client)."""
client, issue_id = await _raise_and_assert_repair(
hass, ufp, doorbell, hass_client, hass_ws_client
)
for channel in doorbell.channels:
channel.is_rtsp_enabled = False
await init_entry(hass, ufp, [doorbell])
ws_client = await hass_ws_client(hass)
client = await hass_client()
new_doorbell = deepcopy(doorbell)
new_doorbell.channels[0].is_rtsp_enabled = True
ufp.api.get_camera = AsyncMock(side_effect=[doorbell, new_doorbell])
ufp.api.create_camera_rtsps_streams = AsyncMock(return_value=None)
issue_id = f"rtsp_disabled_{doorbell.id}"
await ws_client.send_json({"id": 1, "type": "repairs/list_issues"})
msg = await ws_client.receive_json()
assert msg["success"]
assert len(msg["result"]["issues"]) > 0
issue = None
for i in msg["result"]["issues"]:
if i["issue_id"] == issue_id:
issue = i
assert issue is not None
# Unload the integration to ensure the fix flow still works
# if the integration is no longer set up
await hass.config_entries.async_unload(ufp.entry.entry_id)
await hass.async_block_till_done()
data = await start_repair_fix_flow(client, DOMAIN, issue_id)
state = {"created": False}
flow_id = data["flow_id"]
async def get_streams(
camera_id: str, *args: object, **kwargs: object
) -> RTSPSStreams:
return _ACTIVE_STREAMS if state["created"] else _NO_STREAMS
async def create_streams(
camera_id: str, qualities: str, *args: object, **kwargs: object
) -> RTSPSStreams:
state["created"] = True
return _ACTIVE_STREAMS
ufp.api.get_camera_rtsps_streams = AsyncMock(side_effect=get_streams)
ufp.api.create_camera_rtsps_streams = AsyncMock(side_effect=create_streams)
data = await start_repair_fix_flow(client, DOMAIN, issue_id)
assert data["step_id"] == "start"
data = await process_repair_fix_flow(client, flow_id)
data = await process_repair_fix_flow(client, data["flow_id"])
assert data["type"] == "create_entry"
ufp.api.create_camera_rtsps_streams.assert_called_with(doorbell.id, "high")
async def test_rtsp_no_fix_if_third_party(
hass: HomeAssistant,
ufp: MockUFPFixture,
doorbell: Camera,
hass_ws_client: WebSocketGenerator,
) -> None:
"""Test no RTSP disabled warning if camera is third-party."""
for channel in doorbell.channels:
channel.is_rtsp_enabled = False
for user in ufp.api.bootstrap.users.values():
user.all_permissions = []
ufp.api.get_camera = AsyncMock(return_value=doorbell)
doorbell.is_third_party_camera = True
await init_entry(hass, ufp, [doorbell])
ws_client = await hass_ws_client(hass)
await ws_client.send_json({"id": 1, "type": "repairs/list_issues"})
msg = await ws_client.receive_json()
assert msg["success"]
assert not msg["result"]["issues"]
async def test_rtsp_no_fix_if_globally_disabled(
async def test_rtsp_no_repair_if_active(
hass: HomeAssistant,
ufp: MockUFPFixture,
doorbell: Camera,
issue_registry: ir.IssueRegistry,
) -> None:
"""Test no RTSP disabled warning if RTSP is globally disabled on integration."""
"""No repair when the default high stream is active."""
await init_entry(hass, ufp, [doorbell])
assert (
issue_registry.async_get_issue(DOMAIN, f"rtsp_disabled_{doorbell.id}") is None
)
@pytest.mark.parametrize(
("third_party", "extra_options"),
[
pytest.param(True, {}, id="third_party"),
pytest.param(False, {CONF_DISABLE_RTSP: True}, id="globally_disabled"),
],
)
async def test_rtsp_no_repair_when_suppressed(
hass: HomeAssistant,
ufp: MockUFPFixture,
doorbell: Camera,
issue_registry: ir.IssueRegistry,
third_party: bool,
extra_options: dict[str, bool],
) -> None:
"""No repair for third-party cameras or when RTSP is globally disabled."""
for channel in doorbell.channels:
channel.is_rtsp_enabled = False
# Set RTSP globally disabled in config entry options
doorbell.is_third_party_camera = third_party
hass.config_entries.async_update_entry(
ufp.entry,
options={**ufp.entry.options, CONF_DISABLE_RTSP: True},
ufp.entry, options={**ufp.entry.options, **extra_options}
)
await init_entry(hass, ufp, [doorbell])
assert len(issue_registry.issues) == 0
assert (
issue_registry.async_get_issue(DOMAIN, f"rtsp_disabled_{doorbell.id}") is None
)