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

Migrate UniFi Protect camera enumeration to the public API (#176267)

Browse Source
This commit is contained in:
Raphael Hehl
2026-07-13 12:23:53 +02:00
committed by GitHub
parent 9684558931
commit ef9ef35d7d
9 changed files with 1207 additions and 180 deletions
Download Patch File Download Diff File Expand all files Collapse all files
homeassistant/components/unifiprotect/camera.py
+307 -97
View File
@@ -1,19 +1,30 @@
"""Support for Ubiquiti's UniFi Protect NVR."""
from collections.abc import Iterable
import logging
from typing import override
from typing import cast, override
from uiprotect.data import (
Camera as UFPCamera,
CameraChannel,
ChannelQuality,
DeviceState,
ModelType,
ProtectAdoptableDeviceModel,
PublicDeviceModel,
StateType,
channel_id_for_quality,
)
from uiprotect.data.public_devices import PublicCamera
from homeassistant.components.camera import Camera, CameraEntityFeature
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import issue_registry as ir
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import (
device_registry as dr,
entity_platform,
issue_registry as ir,
)
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.helpers.issue_registry import IssueSeverity
@@ -24,6 +35,7 @@ from .const import (
ATTR_FPS,
ATTR_HEIGHT,
ATTR_WIDTH,
DEFAULT_BRAND,
DOMAIN,
)
from .data import ProtectData, ProtectDeviceType, UFPConfigEntry
@@ -33,22 +45,31 @@ from .utils import async_ufp_instance_command, get_camera_base_name
_LOGGER = logging.getLogger(__name__)
PARALLEL_UPDATES = 0
# Main (non-package) RTSPS quality tiers, in default-preference order.
_MAIN_QUALITIES = (
ChannelQuality.HIGH,
ChannelQuality.MEDIUM,
ChannelQuality.LOW,
)
@callback
def _create_rtsp_repair(
hass: HomeAssistant, entry: UFPConfigEntry, camera: UFPCamera
hass: HomeAssistant, entry: UFPConfigEntry, public: PublicCamera
) -> None:
# Keyed on the public camera: the fix flow verifies and creates the stream
# through the public API, so it works without a private session too.
ir.async_create_issue(
hass,
DOMAIN,
f"rtsp_disabled_{camera.id}",
f"rtsp_disabled_{public.id}",
is_fixable=True,
is_persistent=False,
learn_more_url="https://www.home-assistant.io/integrations/unifiprotect/#camera-streams",
severity=IssueSeverity.WARNING,
translation_key="rtsp_disabled",
translation_placeholders={"camera": camera.display_name},
data={"entry_id": entry.entry_id, "camera_id": camera.id},
translation_placeholders={"camera": public.display_name},
data={"entry_id": entry.entry_id, "camera_id": public.id},
)
@@ -58,74 +79,115 @@ def _async_camera_entities(
entry: UFPConfigEntry,
data: ProtectData,
ufp_device: UFPCamera | None = None,
public_device: PublicCamera | None = None,
) -> list[ProtectDeviceEntity]:
"""Create camera entities with stream URLs sourced from the public API.
"""Create camera entities, enumerated public-master from ``PublicCamera``.
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.
Stream URLs come from the public API because it carries the authoritative
per-camera host (stacked consoles resolve correctly), 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:
if ufp_device is None:
# only warn on startup
_LOGGER.warning(
"Camera does not have any channels: %s (id: %s)",
# Public-master enumeration: iterate the public camera list; the private
# camera is paired by shared id (fill) and is None in public-only mode.
pairs: Iterable[tuple[PublicCamera | None, UFPCamera | None]]
if public_device is not None:
private = (
None
if data.api.is_public_only
else data.api.bootstrap.cameras.get(public_device.id)
)
# mirror the startup enumeration's adopted filter
if private is not None and not private.is_adopted_by_us:
return entities
pairs = [(public_device, private)]
elif ufp_device is None:
pairs = data.get_public_cameras()
else:
adopted = data.async_get_public_device(ufp_device)
pairs = [(adopted if isinstance(adopted, PublicCamera) else None, ufp_device)]
for public, camera in pairs:
# A just-adopted camera not yet mirrored into the public bootstrap is
# deferred and picked up when enumeration re-runs.
if public is None:
if camera is not None:
_LOGGER.debug(
"Deferring camera %s until its public mirror arrives",
camera.display_name,
camera.id,
)
data.async_add_pending_camera_id(camera.id)
data.async_add_pending_camera_id(camera.id)
continue
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}"
# Hybrid: a camera not yet in the private bootstrap (adopt race) is
# skipped rather than built private-less — the adopt dispatch creates
# it with its private fill, which would otherwise collide on unique_id.
if camera is None and not data.api.is_public_only:
_LOGGER.debug(
"Deferring camera %s until its private object is adopted",
public.display_name,
)
continue
streams = data.get_rtsps_streams(public.id)
issue_id = f"rtsp_disabled_{public.id}"
tiers = public.hardware_stream_qualities()
main_qualities = [q for q in _MAIN_QUALITIES if q in tiers]
has_package = ChannelQuality.PACKAGE in tiers
if not main_qualities:
# The library guarantees the three main tiers; a camera without any
# is a broken contract — surface it loudly, but do not let one
# camera abort enumeration for the rest.
_LOGGER.warning(
"Camera %s reports no main stream tiers (%s); skipping",
public.display_name,
tiers,
)
continue
# Active stream tiers come from the public ``rtsps_streams`` object.
active = set(streams.get_active_stream_qualities()) if streams else set()
has_stream = False
package_channel: CameraChannel | None = None
for channel in camera.channels:
if channel.is_package:
package_channel = channel
continue
if channel.rtsps_quality in active:
for quality in main_qualities:
if quality in active:
entities.append(
ProtectCamera(data, camera, channel, not has_stream, disable_stream)
ProtectCamera(
data, public, camera, quality, 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:
if has_package:
entities.append(
ProtectCamera(data, camera, package_channel, False, disable_stream)
ProtectCamera(
data, public, camera, ChannelQuality.PACKAGE, 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 active main stream: expose the first main tier for snapshots
entities.append(
ProtectCamera(data, public, camera, main_qualities[0], 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
# camera is streamless because it is offline, not because it needs one.
# The fix flow runs entirely on the public API, so public-only cameras
# get the repair too; third-party is only knowable with a private fill.
if (
disable_stream
or camera.is_third_party_camera
or camera.state is not StateType.CONNECTED
or public.state is not DeviceState.CONNECTED
or (camera is not None and camera.is_third_party_camera)
):
ir.async_delete_issue(hass, DOMAIN, issue_id)
else:
_create_rtsp_repair(hass, entry, camera)
_create_rtsp_repair(hass, entry, public)
return entities
@@ -136,13 +198,22 @@ async def async_setup_entry(
) -> None:
"""Discover cameras on a UniFi Protect NVR."""
data = entry.runtime_data
platform = entity_platform.async_get_current_platform()
@callback
def _add_new_device(device: ProtectAdoptableDeviceModel) -> None:
# AiPort inherits from Camera but should not create camera entities
if not isinstance(device, UFPCamera) or device.model is ModelType.AIPORT:
return
async_add_entities(_async_camera_entities(hass, entry, data, ufp_device=device))
def _add_new_device(device: ProtectAdoptableDeviceModel | PublicCamera) -> None:
if isinstance(device, PublicCamera):
entities = _async_camera_entities(hass, entry, data, public_device=device)
else:
# AiPort inherits from Camera but should not create camera entities
if not isinstance(device, UFPCamera) or device.model is ModelType.AIPORT:
return
entities = _async_camera_entities(hass, entry, data, ufp_device=device)
# A re-enumeration (deferred mirror, RTSPS prime) overlaps entities
# that already exist; the platform errors on live duplicates rather
# than deduplicating, so add only the missing ones.
live = {e.unique_id for e in platform.entities.values()}
async_add_entities([e for e in entities if e.unique_id not in live])
data.async_subscribe_adopt(_add_new_device)
entry.async_on_unload(
@@ -164,24 +235,38 @@ class ProtectCamera(ProtectDeviceEntity, Camera):
"_attr_available",
"_attr_is_recording",
"_attr_motion_detection_enabled",
# flips with the stream source (an RTSPS prime can be the only change)
"_attr_supported_features",
)
def __init__(
self,
data: ProtectData,
camera: UFPCamera,
channel: CameraChannel,
public: PublicCamera,
private: UFPCamera | None,
quality: ChannelQuality,
is_default: bool,
disable_stream: bool,
) -> None:
"""Initialize an UniFi camera."""
self.channel = channel
"""Initialize an UniFi camera.
The public camera is the master; the private camera fills gaps the
public API does not cover and is ``None`` in public-only mode.
"""
self._public = public
self._public_missing = False
self._private = private
self._quality = quality
self._is_package = quality is ChannelQuality.PACKAGE
self._channel_id = channel_id_for_quality(quality)
self._disable_stream = disable_stream
self._last_image: bytes | None = None
super().__init__(data, camera)
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
# The base tracks the private device in hybrid (unchanged behaviour) and
# the public device in public-only, so it always has a mac to key on.
super().__init__(data, cast(ProtectDeviceType, private or public))
self._attr_unique_id = f"{self.device.mac}_{self._channel_id}"
self._attr_name = get_camera_base_name(quality)
# only the default (first active) quality 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
@@ -192,21 +277,17 @@ class ProtectCamera(ProtectDeviceEntity, Camera):
@callback
def _async_set_stream_source(self) -> 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:
quality = self._quality
streams = self.data.get_rtsps_streams(self._public.id)
if self._disable_stream or streams is None:
source = None
if (
streams is None
and not self._disable_stream
and not self.channel.is_package
):
if streams is None and not self._disable_stream and not self._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,
self._public.name,
self._public.id,
)
else:
source = streams.get_stream_url(quality, srtp=False)
@@ -215,43 +296,162 @@ class ProtectCamera(ProtectDeviceEntity, Camera):
@callback
@override
def _async_update_device_from_protect(self, device: ProtectDeviceType) -> None:
super()._async_update_device_from_protect(device)
updated_device = self.device
channel = updated_device.channels[self.channel.id]
self.channel = channel
motion_enabled = updated_device.recording_settings.enable_motion_detection
self._attr_motion_detection_enabled = (
motion_enabled if motion_enabled is not None else True
def _async_set_device_info(self) -> None:
if self._private is not None:
super()._async_set_device_info()
return
# public-only: no market_name/firmware_version/protect_url, and
# ``type`` only on newer firmware, so device identity is limited. The
# NVR link is omitted — an API-key-only client has no private
# bootstrap to read the NVR mac from, and resolving it publicly is
# async; the public-only config mode wires it at setup instead.
public = self._public
self._attr_device_info = DeviceInfo(
name=public.display_name,
model=public.type,
manufacturer=DEFAULT_BRAND,
connections={(dr.CONNECTION_NETWORK_MAC, public.mac)},
)
state_type_is_connected = updated_device.state is StateType.CONNECTED
self._attr_is_recording = (
state_type_is_connected and updated_device.is_recording
)
is_connected = self.data.last_update_success and state_type_is_connected
# some cameras have detachable lens that could cause the camera to be offline
self._attr_available = is_connected and updated_device.is_video_ready
@callback
@override
def _async_update_device_from_protect(self, device: ProtectDeviceType) -> None:
if self._private is not None:
super()._async_update_device_from_protect(device)
updated_device = self.device
# A poll/resync can replace the bootstrap objects; follow them so
# commands and reads never act on a detached model.
self._private = updated_device
if isinstance(
public := self.data.async_get_public_device(updated_device),
PublicCamera,
):
self._public = public
else:
# keep the last object for identity, but log so a vanished
# public mirror is observable rather than a silent no-op
_LOGGER.debug(
"Camera %s has no public mirror; keeping the last known one",
updated_device.display_name,
)
channel_id = self._channel_id
channel = (
updated_device.channels[channel_id]
if channel_id is not None and channel_id < len(updated_device.channels)
else None
)
if channel is None:
# A tier without its private channel blanks the diagnostics;
# log so a camera reconfiguration (or a quality that maps to no
# channel) is distinguishable from a bug.
_LOGGER.debug(
"Camera %s has no private channel %s; diagnostic attributes"
" unavailable",
updated_device.display_name,
channel_id,
)
motion_enabled = updated_device.recording_settings.enable_motion_detection
self._attr_motion_detection_enabled = (
motion_enabled if motion_enabled is not None else True
)
state_type_is_connected = updated_device.state is StateType.CONNECTED
self._attr_is_recording = (
state_type_is_connected and updated_device.is_recording
)
is_connected = self.data.last_update_success and state_type_is_connected
# some cameras have detachable lens that could make them offline
self._attr_available = is_connected and updated_device.is_video_ready
self._async_set_stream_source()
self._attr_extra_state_attributes = {
ATTR_WIDTH: channel.width if channel else None,
ATTR_HEIGHT: channel.height if channel else None,
ATTR_FPS: channel.fps if channel else None,
ATTR_BITRATE: channel.bitrate if channel else None,
ATTR_CHANNEL_ID: channel_id,
}
return
# public-only: recording/motion state and the per-stream diagnostics
# have no public equivalent and degrade; availability tracks the public
# devices websocket health and the public camera state.
public = self._public
self._attr_motion_detection_enabled = False
self._attr_is_recording = False
self._attr_available = (
self.data.last_public_update_success
and not self._public_missing
and public.state is DeviceState.CONNECTED
)
self._async_set_stream_source()
self._attr_extra_state_attributes = {
ATTR_WIDTH: channel.width,
ATTR_HEIGHT: channel.height,
ATTR_FPS: channel.fps,
ATTR_BITRATE: channel.bitrate,
ATTR_CHANNEL_ID: channel.id,
ATTR_WIDTH: None,
ATTR_HEIGHT: None,
ATTR_FPS: None,
ATTR_BITRATE: None,
ATTR_CHANNEL_ID: self._channel_id,
}
@callback
def _async_public_camera_updated(self, obj: PublicDeviceModel | None) -> None:
"""Handle a public devices websocket update for this camera.
``obj`` is the refreshed public object, or ``None`` for a websocket
state change or an unmergeable frame, in which case it is re-read from
the public bootstrap. A camera missing from the bootstrap on re-read
has been removed and reads as unavailable until it reappears.
"""
if obj is None:
obj = self.data.async_get_public_device(self._public)
if isinstance(obj, PublicCamera):
self._public = obj
self._public_missing = False
else:
self._public_missing = True
device = (
self._private
if self._private is not None
else cast(ProtectDeviceType, self._public)
)
self._async_updated_event(device)
@override
async def async_added_to_hass(self) -> None:
"""When entity is added to hass."""
await super().async_added_to_hass()
# The stream URLs live on the public camera and change outside the
# private websocket (a background RTSPS prime announces itself on the
# public channel), so every camera tracks its public mirror; in
# public-only mode this is also the only state source.
self.async_on_remove(
self.data.async_subscribe_public(
self._public.mac, self._async_public_camera_updated
)
)
# A public update or delete can land between entity construction and
# this subscription; re-read so the entity does not start stale.
self._async_public_camera_updated(None)
@override
async def async_camera_image(
self, width: int | None = None, height: int | None = None
) -> bytes | None:
"""Return the Camera 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 the Camera Image.
While snapshot-polling (no stream) request low quality to avoid
hammering the console. width/height are unused (the public endpoint
has no resize).
"""
# Inlines the library's device-level default (support_full_hd_snapshot
# when streaming, low otherwise) since public-only has no private
# device object; the resolved value is unchanged.
high_quality = bool(
self._stream_source and self._public.feature_flags.support_full_hd_snapshot
)
self._last_image = await self.data.api.get_public_api_camera_snapshot(
camera_id=self._public.id,
high_quality=high_quality,
package=self._is_package,
)
return self._last_image
@@ -264,10 +464,20 @@ class ProtectCamera(ProtectDeviceEntity, Camera):
@override
async def async_enable_motion_detection(self) -> None:
"""Call the job and enable motion detection."""
await self.device.set_motion_detection(True)
await self._async_set_motion_detection(True)
@async_ufp_instance_command
@override
async def async_disable_motion_detection(self) -> None:
"""Call the job and disable motion detection."""
await self.device.set_motion_detection(False)
await self._async_set_motion_detection(False)
async def _async_set_motion_detection(self, enabled: bool) -> None:
# the public API has no motion-detection setter; without a private
# session the command cannot be sent and must not report success.
if (private := self._private) is None:
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="motion_detection_public_only",
)
await private.set_motion_detection(enabled)
homeassistant/components/unifiprotect/data.py
+103 -3
View File
@@ -8,6 +8,7 @@ from functools import partial
import logging
from typing import TYPE_CHECKING, Any, cast
from aiohttp.client_exceptions import ServerDisconnectedError
from uiprotect import EventChange, ProtectApiClient, ProtectEvent
from uiprotect.api import RTSPSStreams
from uiprotect.data import (
@@ -19,8 +20,10 @@ from uiprotect.data import (
ProtectAdoptableDeviceModel,
PTZPatrol,
PublicDeviceModel,
WSAction,
WSSubscriptionMessage,
)
from uiprotect.data.public_devices import PublicCamera
from uiprotect.exceptions import ClientError, NotAuthorized
from uiprotect.utils import log_event
from uiprotect.websocket import WebsocketState
@@ -152,6 +155,37 @@ class ProtectData:
Generator[Camera], self.get_by_types({ModelType.CAMERA}, ignore_unadopted)
)
def get_public_cameras(
self,
) -> Generator[tuple[PublicCamera | None, Camera | None]]:
"""Iterate cameras public-master with private-fill.
The public bootstrap is the master list; the matching private camera is
paired by shared id when present (hybrid) and ``None`` in public-only
mode. An adopted private camera not (yet) mirrored into the public
bootstrap is yielded as ``(None, private)`` so the caller can defer it.
Adopted-filtering mirrors ``get_cameras`` whenever a private object is
available.
"""
api = self.api
if not api.has_public_bootstrap:
return
# An API-key-only client never initializes the private bootstrap;
# accessing it would raise.
private_cameras: dict[str, Camera] = (
{} if api.is_public_only else api.bootstrap.cameras
)
public_cameras = api.public_bootstrap.cameras
for camera_id, public in public_cameras.items():
private = private_cameras.get(camera_id)
if private is not None and not private.is_adopted_by_us:
continue
yield public, private
for camera_id, private in private_cameras.items():
if camera_id in public_cameras or not private.is_adopted_by_us:
continue
yield None, private
async def async_load_ptz_patrols(self) -> None:
"""Load PTZ patrols for all PTZ cameras."""
await asyncio.gather(
@@ -232,11 +266,44 @@ class ProtectData:
self._async_signal_public_update(old_obj.mac, None)
return
if new_obj.model is ModelType.NVR:
self._async_signal_device_update(self.api.bootstrap.nvr)
# An API-key-only client has no private NVR (reading it would raise).
if not self.api.is_public_only:
self._async_signal_device_update(self.api.bootstrap.nvr)
return
if isinstance(new_obj, PublicDeviceModel):
if new_obj.model is ModelType.CAMERA:
self._async_reenumerate_camera_on_public_change(new_obj, message)
self._async_signal_public_update(new_obj.mac, new_obj)
@callback
def _async_reenumerate_camera_on_public_change(
self, new_obj: PublicDeviceModel, message: WSSubscriptionMessage
) -> None:
"""Re-run camera enumeration when a public frame can add entities.
Three cases dispatch the public camera to the channels signal:
- A camera deferred at enumeration because its public mirror had not
arrived yet (the private channels-update path cannot be relied on to
fire again).
- A camera whose RTSPS streams the library primes in the background
after it comes online or is added, announced by an ``rtsps_streams``
change: the quality tiers that just became active still need their
entities.
- In public-only mode, a newly added camera — there is no private
adopt path that could discover it.
The platform adds only entities that do not exist yet, so overlapping
re-enumerations are safe.
"""
if new_obj.id in self._pending_camera_ids:
self._pending_camera_ids.remove(new_obj.id)
elif "rtsps_streams" not in message.changed_data and not (
self.api.is_public_only and message.action is WSAction.ADD
):
return
async_dispatcher_send(self._hass, self.channels_signal, new_obj)
@callback
def _async_process_public_event(
self, event: ProtectEvent, change: EventChange
@@ -274,6 +341,37 @@ class ProtectData:
return
self.last_public_update_success = success
self._async_process_public_updates()
if success:
# The library resyncs its public bootstrap on reconnect, but the
# resync applies silently and races this callback, so the re-read
# above may see the pre-disconnect cache. Refresh again behind a
# guaranteed-fresh snapshot (``update_public`` is serialized) so a
# change from the disconnect gap cannot stay stale.
self._entry.async_create_background_task(
self._hass,
self._async_resignal_after_public_resync(),
"unifiprotect public reconnect refresh",
)
async def _async_resignal_after_public_resync(self) -> None:
"""Re-signal public entities once a fresh public snapshot is applied."""
try:
await self.api.update_public()
except NotAuthorized:
# A revoked API key cannot self-recover.
self._entry.async_start_reauth(self._hass)
return
except (TimeoutError, ClientError, ServerDisconnectedError) as err:
# Transport errors retry on the next reconnect.
_LOGGER.debug("Public refresh after reconnect failed: %s", err)
return
self._async_process_public_updates()
# Existing subscriptions are refreshed above, but a camera that
# appeared (or gained streams) during the gap still needs its
# entities; the platform adds only the missing ones.
if self.api.has_public_bootstrap:
for public in list(self.api.public_bootstrap.cameras.values()):
async_dispatcher_send(self._hass, self.channels_signal, public)
@callback
def _async_process_public_updates(self) -> None:
@@ -282,7 +380,9 @@ class ProtectData:
if not api.has_public_bootstrap:
return
# The NVR alarm panel reads the public arm_mode, so refresh it too.
self._async_signal_device_update(api.bootstrap.nvr)
# An API-key-only client has no private NVR (reading it would raise).
if not api.is_public_only:
self._async_signal_device_update(api.bootstrap.nvr)
# Subscribers recompute from the public bootstrap on ``None``.
for subscriptions in self._public_subscriptions.values():
for update_callback in subscriptions:
@@ -529,7 +629,7 @@ class ProtectData:
@callback
def async_get_public_device(
self, device: ProtectDeviceType
self, device: ProtectDeviceType | PublicDeviceModel
) -> PublicDeviceModel | None:
"""Return the public-API object matching a device, if available."""
api = self.api
homeassistant/components/unifiprotect/strings.json
+3
View File
@@ -692,6 +692,9 @@
"global_alarm_manager": {
"message": "The alarm manager on this UniFi Protect NVR is set to Global mode and cannot be controlled locally."
},
"motion_detection_public_only": {
"message": "Motion detection cannot be changed over the public API; configure it in the UniFi Protect app"
},
"no_users_found": {
"message": "No users found, please check Protect permissions"
},
homeassistant/components/unifiprotect/utils.py
+6 -8
View File
@@ -11,7 +11,7 @@ from aiohttp import CookieJar
from uiprotect import ProtectApiClient
from uiprotect.data import (
Bootstrap,
CameraChannel,
ChannelQuality,
Light,
LightModeEnableType,
LightModeType,
@@ -134,14 +134,12 @@ def async_create_api_client(
@callback
def get_camera_base_name(channel: CameraChannel) -> str:
"""Get base name for cameras channel."""
def get_camera_base_name(quality: ChannelQuality) -> str:
"""Get base name for a camera's RTSPS quality channel."""
camera_name = channel.name
if channel.name != "Package Camera":
camera_name = f"{channel.name} resolution channel"
return camera_name
if quality is ChannelQuality.PACKAGE:
return "Package Camera"
return f"{quality.value.title()} resolution channel"
def async_ufp_instance_command[_EntityT, **_P](
tests/components/unifiprotect/conftest.py
+29 -36
View File
@@ -6,7 +6,6 @@ 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
@@ -22,6 +21,8 @@ from uiprotect.data import (
CloudAccount,
Light,
Liveview,
ModelType,
ProtectModelWithId,
Sensor,
SmartDetectObjectType,
StateType,
@@ -45,26 +46,10 @@ from homeassistant.core import HomeAssistant
from homeassistant.util import dt as dt_util
from . import _patch_discovery
from .utils import MockUFPFixture
from .utils import MockUFPFixture, make_public_camera, public_rtsps_for
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
@@ -182,6 +167,7 @@ def mock_ufp_client(bootstrap: Bootstrap):
client.update_public = AsyncMock()
client.async_disconnect_ws = AsyncMock()
client.has_public_bootstrap = True
client.is_public_only = False
# The library owns RTSPS streams on ``PublicCamera.rtsps_streams`` and primes
# them in ``update_public()``; the integration reads them synchronously. Start
@@ -193,15 +179,25 @@ def mock_ufp_client(bootstrap: Bootstrap):
client.public_bootstrap.sirens = {}
client.public_bootstrap.arm_profiles = {}
client.public_bootstrap.arm_mode = None
# No paired public device by default; tests opt in via setup_public_* helpers.
client.public_bootstrap.get = Mock(return_value=None)
# Cameras resolve to their primed public model (see ``update_public`` in
# ``mock_entry``); other device types opt in via the ``setup_public_*``
# helpers, so they default to no paired public object.
def _public_bootstrap_get(
model: ModelType, obj_id: str
) -> ProtectModelWithId | None:
if model is ModelType.CAMERA:
return client.public_bootstrap.cameras.get(obj_id)
return None
client.public_bootstrap.get = Mock(side_effect=_public_bootstrap_get)
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
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)
@@ -265,23 +261,20 @@ def mock_entry(
ufp_client.subscribe_devices_websocket_state = subscribe_devices_websocket_state
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.
# Mirror the library prime: build each camera's public model from the
# private bootstrap and attach its RTSPS streams (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
),
cameras: dict[str, Any] = {}
for camera in ufp_client.bootstrap.cameras.values():
public = make_public_camera(camera)
public.rtsps_streams = (
public_rtsps_for(camera)
if camera.state is StateType.CONNECTED
else None
)
for camera in ufp_client.bootstrap.cameras.values()
}
cameras[camera.id] = public
pb.cameras = cameras
return pb
ufp_client.update_public = AsyncMock(side_effect=update_public)
tests/components/unifiprotect/test_camera.py
+714 -36
View File
@@ -1,12 +1,21 @@
"""Test the UniFi Protect camera platform."""
from types import SimpleNamespace
from typing import Any
from unittest.mock import AsyncMock
from unittest.mock import AsyncMock, Mock, patch
from aiohttp.client_exceptions import ServerDisconnectedError
import pytest
from uiprotect.data import AiPort, Camera as ProtectCamera, StateType
from uiprotect.data import (
AiPort,
Camera as ProtectCamera,
ChannelQuality,
DeviceState,
ModelType,
StateType,
WSAction,
)
from uiprotect.exceptions import ClientError, NotAuthorized
from uiprotect.websocket import WebsocketState
from homeassistant.components.camera import (
CameraEntityFeature,
@@ -16,9 +25,14 @@ from homeassistant.components.camera import (
from homeassistant.components.unifiprotect.const import CONF_DISABLE_RTSP, DOMAIN
from homeassistant.components.unifiprotect.utils import get_camera_base_name
from homeassistant.config_entries import SOURCE_REAUTH, ConfigEntryState
from homeassistant.const import ATTR_ENTITY_ID, Platform
from homeassistant.const import ATTR_ENTITY_ID, STATE_UNAVAILABLE, Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er, issue_registry as ir
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import (
device_registry as dr,
entity_registry as er,
issue_registry as ir,
)
from . import patch_ufp_method
from .utils import (
@@ -27,14 +41,18 @@ from .utils import (
assert_entity_counts,
enable_entity,
init_entry,
make_public_camera,
public_device_ws_message,
public_rtsps_for,
remove_entities,
)
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)
quality = camera_obj.channels[channel_id].rtsps_quality
assert quality is not None
base_name = get_camera_base_name(quality)
return f"camera.{camera_obj.name}_{base_name}".replace(" ", "_").lower()
@@ -242,36 +260,29 @@ async def test_package_camera_without_stream(
assert ufp.api.get_public_api_camera_snapshot.call_args.kwargs["package"] is True
async def test_package_only_camera(
hass: HomeAssistant,
ufp: MockUFPFixture,
camera: ProtectCamera,
issue_registry: ir.IssueRegistry,
) -> None:
"""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])
# 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
async def test_no_channels(
async def test_camera_not_in_public_bootstrap(
hass: HomeAssistant, ufp: MockUFPFixture, camera: ProtectCamera
) -> None:
"""A camera without channels yet creates no entities."""
camera.channels = []
"""A camera not yet mirrored into the public bootstrap is deferred."""
async def _prime_without_camera() -> Any:
pb = ufp.api.public_bootstrap
pb.cameras = {}
return pb
ufp.api.update_public = AsyncMock(side_effect=_prime_without_camera)
await init_entry(hass, ufp, [camera])
assert_entity_counts(hass, Platform.CAMERA, 0, 0)
# the public mirror arriving on the devices websocket creates the entity
public = make_public_camera(camera)
public.rtsps_streams = public_rtsps_for(camera)
ufp.api.public_bootstrap.cameras = {camera.id: public}
ufp.devices_ws_subscription(public_device_ws_message(public))
await hass.async_block_till_done()
assert_entity_counts(hass, Platform.CAMERA, 1, 1)
async def test_streams_unavailable(
hass: HomeAssistant, ufp: MockUFPFixture, camera_all: ProtectCamera
@@ -280,11 +291,9 @@ async def test_streams_unavailable(
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
)
}
public = make_public_camera(camera_all)
public.rtsps_streams = None
pb.cameras = {camera_all.id: public}
return pb
ufp.api.update_public = AsyncMock(side_effect=_prime_streamless)
@@ -383,6 +392,263 @@ async def test_aiport_no_camera_entities(
assert_entity_counts(hass, Platform.CAMERA, 0, 0)
async def test_public_only_camera(
hass: HomeAssistant, ufp: MockUFPFixture, camera: ProtectCamera
) -> None:
"""A public-only camera builds a working entity with degraded diagnostics."""
# This camera is intentionally kept out of the private bootstrap; wire the
# channel api so its public RTSPS URLs resolve, as add_device would.
for channel in camera.channels:
channel._api = ufp.api
public = make_public_camera(camera)
public.rtsps_streams = public_rtsps_for(camera)
async def _prime_public_only() -> Any:
pb = ufp.api.public_bootstrap
pb.cameras = {camera.id: public}
return pb
ufp.api.update_public = AsyncMock(side_effect=_prime_public_only)
ufp.api.is_public_only = True
# No private cameras in the bootstrap: the public object is the only source.
await init_entry(hass, ufp, [])
assert_entity_counts(hass, Platform.CAMERA, 1, 1)
entity_id = _channel_entity_id(camera, 0)
state = hass.states.get(entity_id)
assert state
assert state.state != STATE_UNAVAILABLE
# diagnostics have no public equivalent and degrade to None
assert state.attributes["fps"] is None
# device identity degrades to name-only; the NVR link is omitted (resolving
# the NVR identity publicly is wired with the config-mode setup)
device_registry = dr.async_get(hass)
device = device_registry.async_get_device(
connections={(dr.CONNECTION_NETWORK_MAC, public.mac)}
)
assert device is not None
assert device.via_device_id is None
assert device.name == camera.display_name
assert device.model == camera.type
assert (
await async_get_stream_source(hass, entity_id)
== camera.channels[0].rtsps_no_srtp_url
)
ufp.api.get_public_api_camera_snapshot = AsyncMock()
await async_get_image(hass, entity_id)
ufp.api.get_public_api_camera_snapshot.assert_called_once()
# a frame without a merged object re-reads the public bootstrap and stays up
none_msg = Mock()
none_msg.changed_data = {}
none_msg.new_obj = None
none_msg.old_obj = public
ufp.devices_ws_subscription(none_msg)
await hass.async_block_till_done()
assert hass.states.get(entity_id).state != STATE_UNAVAILABLE
# a public devices websocket update drives availability
public.state = DeviceState.DISCONNECTED
ufp.devices_ws_subscription(public_device_ws_message(public))
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_UNAVAILABLE
async def test_adopt_before_public_bootstrap(
hass: HomeAssistant, ufp: MockUFPFixture, camera: ProtectCamera
) -> None:
"""A camera adopted before the public bootstrap mirrors it is deferred."""
async def _prime_empty() -> Any:
pb = ufp.api.public_bootstrap
pb.cameras = {}
return pb
ufp.api.update_public = AsyncMock(side_effect=_prime_empty)
await init_entry(hass, ufp, [])
assert_entity_counts(hass, Platform.CAMERA, 0, 0)
# adopt the camera while it is still absent from the public bootstrap
camera._api = ufp.api
await adopt_devices(hass, ufp, [camera], fully_adopt=True)
assert_entity_counts(hass, Platform.CAMERA, 0, 0)
# the public mirror arriving on the devices websocket creates the entity
for channel in camera.channels:
channel._api = ufp.api
public = make_public_camera(camera)
public.rtsps_streams = public_rtsps_for(camera)
ufp.api.public_bootstrap.cameras = {camera.id: public}
ufp.devices_ws_subscription(public_device_ws_message(public))
await hass.async_block_till_done()
assert_entity_counts(hass, Platform.CAMERA, 1, 1)
async def test_public_only_camera_removed(
hass: HomeAssistant, ufp: MockUFPFixture, camera: ProtectCamera
) -> None:
"""A public-only camera removed from the public bootstrap goes unavailable."""
for channel in camera.channels:
channel._api = ufp.api
public = make_public_camera(camera)
public.rtsps_streams = public_rtsps_for(camera)
async def _prime_public_only() -> Any:
pb = ufp.api.public_bootstrap
pb.cameras = {camera.id: public}
return pb
ufp.api.update_public = AsyncMock(side_effect=_prime_public_only)
ufp.api.is_public_only = True
await init_entry(hass, ufp, [])
entity_id = _channel_entity_id(camera, 0)
assert hass.states.get(entity_id).state != STATE_UNAVAILABLE
# on a delete the library has already dropped the object; the re-read
# comes up empty and the entity goes unavailable
ufp.api.public_bootstrap.cameras = {}
delete_msg = Mock()
delete_msg.changed_data = {}
delete_msg.new_obj = None
delete_msg.old_obj = public
ufp.devices_ws_subscription(delete_msg)
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_UNAVAILABLE
# the camera reappearing on the websocket recovers it
ufp.api.public_bootstrap.cameras = {camera.id: public}
ufp.devices_ws_subscription(public_device_ws_message(public))
await hass.async_block_till_done()
assert hass.states.get(entity_id).state != STATE_UNAVAILABLE
async def test_public_only_camera_ws_state_availability(
hass: HomeAssistant, ufp: MockUFPFixture, camera: ProtectCamera
) -> None:
"""Public devices websocket state drives a public-only camera without private reads."""
for channel in camera.channels:
channel._api = ufp.api
public = make_public_camera(camera)
public.rtsps_streams = public_rtsps_for(camera)
async def _prime_public_only() -> Any:
pb = ufp.api.public_bootstrap
pb.cameras = {camera.id: public}
return pb
ufp.api.update_public = AsyncMock(side_effect=_prime_public_only)
ufp.api.is_public_only = True
await init_entry(hass, ufp, [])
entity_id = _channel_entity_id(camera, 0)
assert hass.states.get(entity_id).state != STATE_UNAVAILABLE
# tripwire: from here on nothing may read the private bootstrap
ufp.api.bootstrap = None
# a public NVR frame has no private NVR to signal and is ignored
nvr_msg = Mock()
nvr_msg.changed_data = {}
nvr_msg.old_obj = None
nvr_msg.new_obj = Mock(model=ModelType.NVR)
ufp.devices_ws_subscription(nvr_msg)
await hass.async_block_till_done()
assert hass.states.get(entity_id).state != STATE_UNAVAILABLE
# a websocket drop marks the camera unavailable, a reconnect recovers it
ufp.devices_ws_state_subscription(WebsocketState.DISCONNECTED)
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_UNAVAILABLE
ufp.devices_ws_state_subscription(WebsocketState.CONNECTED)
await hass.async_block_till_done()
assert hass.states.get(entity_id).state != STATE_UNAVAILABLE
async def test_camera_motion_detection_uses_replaced_device(
hass: HomeAssistant, ufp: MockUFPFixture, camera: ProtectCamera
) -> None:
"""Motion commands act on the refreshed device after a bootstrap replacement."""
await init_entry(hass, ufp, [camera])
entity_id = _channel_entity_id(camera, 0)
new_camera = camera.model_copy()
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)
await hass.async_block_till_done()
with patch_ufp_method(
new_camera, "set_motion_detection", new_callable=AsyncMock
) as mock_method:
await hass.services.async_call(
"camera",
"enable_motion_detection",
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
mock_method.assert_called_once_with(True)
@pytest.mark.parametrize(
"service", ["enable_motion_detection", "disable_motion_detection"]
)
async def test_public_only_camera_motion_detection_raises(
hass: HomeAssistant, ufp: MockUFPFixture, camera: ProtectCamera, service: str
) -> None:
"""Motion detection cannot be changed without a private session."""
for channel in camera.channels:
channel._api = ufp.api
public = make_public_camera(camera)
public.rtsps_streams = public_rtsps_for(camera)
async def _prime_public_only() -> Any:
pb = ufp.api.public_bootstrap
pb.cameras = {camera.id: public}
return pb
ufp.api.update_public = AsyncMock(side_effect=_prime_public_only)
ufp.api.is_public_only = True
await init_entry(hass, ufp, [])
entity_id = _channel_entity_id(camera, 0)
with pytest.raises(HomeAssistantError, match="public API"):
await hass.services.async_call(
"camera",
service,
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
async def test_hybrid_public_camera_without_private_deferred(
hass: HomeAssistant, ufp: MockUFPFixture, camera: ProtectCamera
) -> None:
"""Hybrid: a public camera without its private twin is left to the adopt flow."""
public = make_public_camera(camera)
async def _prime_public_only() -> Any:
pb = ufp.api.public_bootstrap
pb.cameras = {camera.id: public}
return pb
ufp.api.update_public = AsyncMock(side_effect=_prime_public_only)
# Not public-only mode (conftest default): the entity must not be built
# private-less; the later adopt dispatch creates it with its private fill.
await init_entry(hass, ufp, [])
assert_entity_counts(hass, Platform.CAMERA, 0, 0)
async def test_snapshot_low_quality_without_stream(
hass: HomeAssistant, ufp: MockUFPFixture, camera: ProtectCamera
) -> None:
@@ -399,3 +665,415 @@ async def test_snapshot_low_quality_without_stream(
assert (
ufp.api.get_public_api_camera_snapshot.call_args.kwargs["high_quality"] is False
)
async def test_private_enumeration_upgrade_keeps_entities(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
ufp: MockUFPFixture,
camera_all: ProtectCamera,
) -> None:
"""Old private-enumeration registry entries survive the public enumeration.
Same unique_ids, entity_ids, and enabled split — including across a reload.
"""
seeded: dict[str, str] = {}
for channel_id, disabled_by in (
(0, None),
(1, er.RegistryEntryDisabler.INTEGRATION),
(2, er.RegistryEntryDisabler.INTEGRATION),
):
entry = entity_registry.async_get_or_create(
Platform.CAMERA,
DOMAIN,
f"{camera_all.mac}_{channel_id}",
config_entry=ufp.entry,
suggested_object_id=f"my_renamed_cam_{channel_id}",
disabled_by=disabled_by,
)
seeded[entry.unique_id] = entry.entity_id
await init_entry(hass, ufp, [camera_all], regenerate_ids=False)
# Same totals as a fresh setup: nothing duplicated, nothing orphaned.
assert_entity_counts(hass, Platform.CAMERA, 3, 1)
for unique_id, entity_id in seeded.items():
entry = entity_registry.async_get(entity_id)
assert entry is not None
assert entry.unique_id == unique_id
# The customized (enabled) entity is live and streams from the public API.
high_id = seeded[f"{camera_all.mac}_0"]
assert hass.states.get(high_id) is not None
assert (
await async_get_stream_source(hass, high_id)
== camera_all.channels[0].rtsps_no_srtp_url
)
await hass.config_entries.async_reload(ufp.entry.entry_id)
await hass.async_block_till_done()
assert_entity_counts(hass, Platform.CAMERA, 3, 1)
for unique_id, entity_id in seeded.items():
entry = entity_registry.async_get(entity_id)
assert entry is not None
assert entry.unique_id == unique_id
assert hass.states.get(high_id) is not None
async def test_streamless_camera_reenumerated_on_rtsps_prime(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
ufp: MockUFPFixture,
camera_all: ProtectCamera,
caplog: pytest.LogCaptureFixture,
) -> None:
"""A camera primed after enumeration gains its active-quality entities.
A disconnected camera enumerates streamless (only the snapshot fallback).
When it comes online the library primes its RTSPS streams in the background
and announces the change with an ``rtsps_streams`` devices-WS frame; the
integration must re-enumerate so the now-active tiers get their entities,
without re-adding the live fallback entity, and the existing entity must
pick up its now-available stream URL.
"""
camera_all.state = StateType.DISCONNECTED
await init_entry(hass, ufp, [camera_all])
# Streamless: only the snapshot fallback (high) exists, without a stream.
assert_entity_counts(hass, Platform.CAMERA, 1, 1)
high_id = _channel_entity_id(camera_all, 0)
assert await async_get_stream_source(hass, high_id) is None
assert entity_registry.async_get(_channel_entity_id(camera_all, 1)) is None
assert entity_registry.async_get(_channel_entity_id(camera_all, 2)) is None
# The camera comes online and the library primes its streams, announced by
# an rtsps_streams change on the public devices websocket (sent twice: the
# re-enumeration must not attempt to re-add live entities).
camera_all.state = StateType.CONNECTED
public = ufp.api.public_bootstrap.cameras[camera_all.id]
public.state = DeviceState.CONNECTED
public.rtsps_streams = public_rtsps_for(camera_all)
msg = public_device_ws_message(public)
msg.changed_data = {"rtsps_streams": public.rtsps_streams}
ufp.devices_ws_subscription(msg)
await hass.async_block_till_done()
ufp.devices_ws_subscription(msg)
await hass.async_block_till_done()
# The active medium/low tiers now have entities, the existing high entity
# streams, and no duplicate-add errors were logged.
assert_entity_counts(hass, Platform.CAMERA, 3, 1)
_assert_entity(hass, camera_all, 0, enabled=True)
_assert_entity(hass, camera_all, 1, enabled=False)
_assert_entity(hass, camera_all, 2, enabled=False)
assert (
await async_get_stream_source(hass, high_id)
== camera_all.channels[0].rtsps_no_srtp_url
)
assert "does not generate unique IDs" not in caplog.text
async def test_public_only_camera_added_after_setup(
hass: HomeAssistant, ufp: MockUFPFixture, camera: ProtectCamera
) -> None:
"""In public-only mode a camera added later is discovered from its frame.
There is no private adopt path without a local user, so the public devices
websocket ``add`` frame is the only discovery signal.
"""
async def _prime_public_only() -> Any:
pb = ufp.api.public_bootstrap
pb.cameras = {}
return pb
ufp.api.update_public = AsyncMock(side_effect=_prime_public_only)
ufp.api.is_public_only = True
await init_entry(hass, ufp, [])
assert_entity_counts(hass, Platform.CAMERA, 0, 0)
# A new camera appears on the public devices websocket.
for channel in camera.channels:
channel._api = ufp.api
public = make_public_camera(camera)
public.rtsps_streams = public_rtsps_for(camera)
ufp.api.public_bootstrap.cameras = {camera.id: public}
msg = public_device_ws_message(public)
msg.action = WSAction.ADD
ufp.devices_ws_subscription(msg)
await hass.async_block_till_done()
assert_entity_counts(hass, Platform.CAMERA, 1, 1)
state = hass.states.get(_channel_entity_id(camera, 0))
assert state
assert state.state != STATE_UNAVAILABLE
async def test_public_only_streamless_camera_gets_repair(
hass: HomeAssistant,
ufp: MockUFPFixture,
camera: ProtectCamera,
issue_registry: ir.IssueRegistry,
) -> None:
"""A streamless public-only camera raises the RTSP repair.
The fix flow verifies and creates the stream entirely through the public
API, so it works without a private session.
"""
for channel in camera.channels:
channel._api = ufp.api
public = make_public_camera(camera)
public.rtsps_streams = None
async def _prime_public_only() -> Any:
pb = ufp.api.public_bootstrap
pb.cameras = {camera.id: public}
return pb
ufp.api.update_public = AsyncMock(side_effect=_prime_public_only)
ufp.api.is_public_only = True
await init_entry(hass, ufp, [])
assert_entity_counts(hass, Platform.CAMERA, 1, 1)
assert (
issue_registry.async_get_issue(DOMAIN, f"rtsp_disabled_{camera.id}") is not None
)
async def test_stream_capability_published_on_prime(
hass: HomeAssistant, ufp: MockUFPFixture, camera_all: ProtectCamera
) -> None:
"""Gaining a stream publishes the STREAM capability when nothing else changes."""
camera_all.channels = [c.model_copy() for c in camera_all.channels]
for channel in camera_all.channels:
channel.is_rtsp_enabled = False
await init_entry(hass, ufp, [camera_all])
high_id = _channel_entity_id(camera_all, 0)
state = hass.states.get(high_id)
assert state
assert state.attributes["supported_features"] == CameraEntityFeature(0)
# The library primes the streams; availability, recording, and motion are
# unchanged, so the capability flip is the only observable difference.
camera_all.channels[0].is_rtsp_enabled = True
public = ufp.api.public_bootstrap.cameras[camera_all.id]
public.rtsps_streams = public_rtsps_for(camera_all)
msg = public_device_ws_message(public)
msg.changed_data = {"rtsps_streams": public.rtsps_streams}
ufp.devices_ws_subscription(msg)
await hass.async_block_till_done()
state = hass.states.get(high_id)
assert state
assert state.attributes["supported_features"] == CameraEntityFeature.STREAM
async def test_camera_without_main_tiers_skipped_with_warning(
hass: HomeAssistant,
ufp: MockUFPFixture,
camera: ProtectCamera,
caplog: pytest.LogCaptureFixture,
) -> None:
"""A camera violating the main-tier contract is skipped loudly, not fatally."""
for channel in camera.channels:
channel._api = ufp.api
healthy = make_public_camera(camera)
healthy.rtsps_streams = public_rtsps_for(camera)
broken = make_public_camera(camera)
broken.id = "broken-camera"
broken.mac = "FFEEDDCCBBAA"
broken.display_name = "Broken"
broken.rtsps_streams = None
broken.hardware_stream_qualities.return_value = [ChannelQuality.PACKAGE]
async def _prime_public_only() -> Any:
pb = ufp.api.public_bootstrap
pb.cameras = {camera.id: healthy, broken.id: broken}
return pb
ufp.api.update_public = AsyncMock(side_effect=_prime_public_only)
ufp.api.is_public_only = True
await init_entry(hass, ufp, [])
# The healthy camera enumerates; the broken one is skipped with a warning
# instead of aborting the platform setup.
assert "reports no main stream tiers" in caplog.text
assert_entity_counts(hass, Platform.CAMERA, 1, 1)
assert hass.states.get(_channel_entity_id(camera, 0)) is not None
async def test_public_only_camera_deleted_during_gap(
hass: HomeAssistant, ufp: MockUFPFixture, camera: ProtectCamera
) -> None:
"""A camera deleted while the websocket was down reads as unavailable.
The library resyncs its public bootstrap on reconnect but applies the
snapshot silently; the integration must re-read behind a fresh snapshot
rather than resurrect the entity from the pre-disconnect cache.
"""
for channel in camera.channels:
channel._api = ufp.api
public = make_public_camera(camera)
public.rtsps_streams = public_rtsps_for(camera)
async def _prime_public_only() -> Any:
pb = ufp.api.public_bootstrap
pb.cameras = {camera.id: public}
return pb
ufp.api.update_public = AsyncMock(side_effect=_prime_public_only)
ufp.api.is_public_only = True
await init_entry(hass, ufp, [])
entity_id = _channel_entity_id(camera, 0)
assert hass.states.get(entity_id).state != STATE_UNAVAILABLE
# The websocket drops; the camera is deleted during the gap, so the
# reconnect resync returns a snapshot without it.
async def _prime_empty() -> Any:
pb = ufp.api.public_bootstrap
pb.cameras = {}
return pb
ufp.api.update_public = AsyncMock(side_effect=_prime_empty)
ufp.devices_ws_state_subscription(WebsocketState.DISCONNECTED)
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_UNAVAILABLE
ufp.devices_ws_state_subscription(WebsocketState.CONNECTED)
await hass.async_block_till_done()
# Not resurrected from the stale cache: the fresh snapshot has no camera.
assert hass.states.get(entity_id).state == STATE_UNAVAILABLE
@pytest.mark.parametrize(
("error", "expect_reauth"),
[
pytest.param(NotAuthorized("revoked"), True, id="revoked_key"),
pytest.param(ServerDisconnectedError(), False, id="transport_error"),
],
)
async def test_reconnect_refresh_failures(
hass: HomeAssistant,
ufp: MockUFPFixture,
camera: ProtectCamera,
error: Exception,
expect_reauth: bool,
) -> None:
"""A failed reconnect refresh starts reauth on 401 and retries on transport."""
for channel in camera.channels:
channel._api = ufp.api
public = make_public_camera(camera)
public.rtsps_streams = public_rtsps_for(camera)
async def _prime_public_only() -> Any:
pb = ufp.api.public_bootstrap
pb.cameras = {camera.id: public}
return pb
ufp.api.update_public = AsyncMock(side_effect=_prime_public_only)
ufp.api.is_public_only = True
await init_entry(hass, ufp, [])
ufp.api.update_public = AsyncMock(side_effect=error)
ufp.devices_ws_state_subscription(WebsocketState.DISCONNECTED)
await hass.async_block_till_done()
with patch.object(ufp.entry, "async_start_reauth") as mock_reauth:
ufp.devices_ws_state_subscription(WebsocketState.CONNECTED)
await hass.async_block_till_done()
assert mock_reauth.called is expect_reauth
async def test_hybrid_camera_lost_public_mirror_logs(
hass: HomeAssistant,
ufp: MockUFPFixture,
camera: ProtectCamera,
caplog: pytest.LogCaptureFixture,
) -> None:
"""A vanished public mirror is observable instead of a silent no-op."""
await init_entry(hass, ufp, [camera])
ufp.api.public_bootstrap.cameras = {}
mock_msg = Mock()
mock_msg.changed_data = {}
mock_msg.new_obj = camera
ufp.ws_msg(mock_msg)
await hass.async_block_till_done()
assert "has no public mirror" in caplog.text
async def test_public_only_camera_added_during_gap(
hass: HomeAssistant, ufp: MockUFPFixture, camera_all: ProtectCamera
) -> None:
"""A camera added while the websocket was down enumerates on reconnect.
The resync snapshot is applied silently and no add frame ever arrives for
a camera that appeared during the gap, so the reconnect refresh must
dispatch it for enumeration itself.
"""
for channel in camera_all.channels:
channel._api = ufp.api
first = make_public_camera(camera_all)
first.rtsps_streams = public_rtsps_for(camera_all)
async def _prime_one() -> Any:
pb = ufp.api.public_bootstrap
pb.cameras = {camera_all.id: first}
return pb
ufp.api.update_public = AsyncMock(side_effect=_prime_one)
ufp.api.is_public_only = True
await init_entry(hass, ufp, [])
assert_entity_counts(hass, Platform.CAMERA, 3, 1)
# A second camera appears during the gap; the reconnect resync includes it.
second = make_public_camera(camera_all)
second.id = "gap-camera"
second.mac = "FFEEDDCCBB01"
second.name = "Gap Camera"
second.display_name = "Gap Camera"
second.rtsps_streams = first.rtsps_streams
async def _prime_two() -> Any:
pb = ufp.api.public_bootstrap
pb.cameras = {camera_all.id: first, second.id: second}
return pb
ufp.api.update_public = AsyncMock(side_effect=_prime_two)
ufp.devices_ws_state_subscription(WebsocketState.DISCONNECTED)
await hass.async_block_till_done()
ufp.devices_ws_state_subscription(WebsocketState.CONNECTED)
await hass.async_block_till_done()
# Three tiers each for both cameras; no duplicates for the first one.
assert_entity_counts(hass, Platform.CAMERA, 6, 2)
assert hass.states.get("camera.gap_camera_high_resolution_channel") is not None
async def test_unadopted_camera_not_enumerated_from_public_frame(
hass: HomeAssistant, ufp: MockUFPFixture, camera: ProtectCamera
) -> None:
"""A public frame cannot create entities for an unadopted camera."""
camera.is_adopted = False
await init_entry(hass, ufp, [camera])
assert_entity_counts(hass, Platform.CAMERA, 0, 0)
public = ufp.api.public_bootstrap.cameras[camera.id]
msg = public_device_ws_message(public)
msg.changed_data = {"rtsps_streams": public.rtsps_streams}
ufp.devices_ws_subscription(msg)
await hass.async_block_till_done()
# still excluded, exactly like the startup enumeration
assert_entity_counts(hass, Platform.CAMERA, 0, 0)
tests/components/unifiprotect/test_number.py
+8
View File
@@ -349,6 +349,14 @@ async def test_number_camera_mic_volume_unavailable_without_public(
) -> None:
"""The migrated mic volume number is unavailable without a public object."""
# The default fixture mirrors every camera into the public bootstrap;
# prime it empty to model a camera the public API does not know yet.
async def _prime_empty() -> Mock:
pb = ufp.api.public_bootstrap
pb.cameras = {}
return pb
ufp.api.update_public = AsyncMock(side_effect=_prime_empty)
await init_entry(hass, ufp, [camera])
_, entity_id = await ids_from_device_description(
tests/components/unifiprotect/test_select.py
+8
View File
@@ -1,6 +1,7 @@
"""Test the UniFi Protect select platform."""
from copy import copy
from typing import Any
from unittest.mock import AsyncMock, Mock
import pytest
@@ -227,6 +228,13 @@ async def test_select_camera_hdr_mode_unavailable_without_public(
) -> None:
"""The migrated HDR mode select is unavailable without a public object."""
async def _prime_without_camera() -> Any:
pb = ufp.api.public_bootstrap
pb.cameras = {}
return pb
ufp.api.update_public = AsyncMock(side_effect=_prime_without_camera)
await init_entry(hass, ufp, [doorbell])
description = next(d for d in CAMERA_SELECTS if d.key == "hdr_mode")
tests/components/unifiprotect/utils.py
+29
View File
@@ -6,9 +6,11 @@ from datetime import timedelta
from unittest.mock import Mock
from uiprotect import EventChange, ProtectApiClient, ProtectEvent
from uiprotect.api import RTSPSStreams
from uiprotect.data import (
Bootstrap,
Camera,
ChannelQuality,
DeviceState,
Event,
EventType,
@@ -229,6 +231,21 @@ async def init_entry(
await hass.async_block_till_done()
def public_rtsps_for(camera: Camera) -> 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
def make_public_sensor(
sensor: Sensor,
*,
@@ -368,6 +385,9 @@ def make_public_camera(
public = Mock(spec=PublicCamera)
public.id = camera.id
public.mac = camera.mac
public.name = camera.name
public.display_name = camera.display_name
public.type = camera.type
public.model = ModelType.CAMERA
public.state = DeviceState[camera.state.name] if state is None else state
public.mic_volume = camera.mic_volume if mic_volume is None else mic_volume
@@ -376,6 +396,15 @@ def make_public_camera(
if hdr_type is None
else hdr_type
)
public.has_package_camera = camera.feature_flags.has_package_camera
public.feature_flags = Mock()
public.feature_flags.support_full_hd_snapshot = (
camera.feature_flags.support_full_hd_snapshot
)
qualities = [ChannelQuality.HIGH, ChannelQuality.MEDIUM, ChannelQuality.LOW]
if public.has_package_camera:
qualities.append(ChannelQuality.PACKAGE)
public.hardware_stream_qualities.return_value = qualities
return public