From 1dc254789b054facace14fb03fb19cfe07d40d08 Mon Sep 17 00:00:00 2001 From: Raphael Hehl <7577984+RaHehl@users.noreply.github.com> Date: Wed, 30 Sep 2026 12:44:19 +0200 Subject: [PATCH] Add UniFi Protect selects to the API key only mode (#182744) --- .../unifiprotect/alarm_control_panel.py | 24 +- .../components/unifiprotect/const.py | 1 + .../components/unifiprotect/entity.py | 12 + .../components/unifiprotect/select.py | 43 +- tests/components/unifiprotect/test_init.py | 20 + tests/components/unifiprotect/test_select.py | 381 +++++++++++++++++- 6 files changed, 448 insertions(+), 33 deletions(-) diff --git a/homeassistant/components/unifiprotect/alarm_control_panel.py b/homeassistant/components/unifiprotect/alarm_control_panel.py index 472aa43a4c75..979c81457adf 100644 --- a/homeassistant/components/unifiprotect/alarm_control_panel.py +++ b/homeassistant/components/unifiprotect/alarm_control_panel.py @@ -12,15 +12,13 @@ from homeassistant.components.alarm_control_panel import ( ) from homeassistant.core import HomeAssistant, callback from homeassistant.exceptions import HomeAssistantError -from homeassistant.helpers import device_registry as dr -from homeassistant.helpers.device_registry import DeviceInfo from homeassistant.helpers.entity import EntityDescription from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback -from .const import DEFAULT_BRAND, DOMAIN +from .const import DOMAIN from .data import ProtectData, ProtectDeviceType, UFPConfigEntry from .entity import ProtectNVREntity -from .utils import _async_unifi_mac_from_hass, async_ufp_instance_command +from .utils import async_ufp_instance_command PARALLEL_UPDATES = 0 @@ -77,24 +75,6 @@ class ProtectNVRAlarmControlPanel(ProtectNVREntity, AlarmControlPanelEntity): super().__init__(data, device, EntityDescription(key="alarm")) self._refresh_alarm_state() - @callback - @override - def _async_set_device_info(self) -> None: - if not self.data.api.is_public_only: - super()._async_set_device_info() - return - # Degraded: no market name or console URL, and ``type`` only on - # newer firmware. The mac is backfilled by the library, matching the - # device created at setup. - mac = _async_unifi_mac_from_hass(self.device.mac) - self._attr_device_info = DeviceInfo( - connections={(dr.CONNECTION_NETWORK_MAC, mac)}, - identifiers={(DOMAIN, mac)}, - manufacturer=DEFAULT_BRAND, - name=self.device.display_name, - model=self.device.type, - ) - @callback def _refresh_alarm_state(self) -> None: """Update _attr_alarm_state from the public bootstrap cache.""" diff --git a/homeassistant/components/unifiprotect/const.py b/homeassistant/components/unifiprotect/const.py index 9c9fc734ac81..de73115c819b 100644 --- a/homeassistant/components/unifiprotect/const.py +++ b/homeassistant/components/unifiprotect/const.py @@ -90,6 +90,7 @@ PUBLIC_ONLY_PLATFORMS = [ Platform.EVENT, Platform.LIGHT, Platform.NUMBER, + Platform.SELECT, Platform.SENSOR, Platform.SIREN, Platform.SWITCH, diff --git a/homeassistant/components/unifiprotect/entity.py b/homeassistant/components/unifiprotect/entity.py index b081503bce87..f6118dcb77a2 100644 --- a/homeassistant/components/unifiprotect/entity.py +++ b/homeassistant/components/unifiprotect/entity.py @@ -49,6 +49,7 @@ from .const import ( DOMAIN, ) from .data import ProtectData, ProtectDeviceType +from .utils import _async_unifi_mac_from_hass _LOGGER = logging.getLogger(__name__) @@ -551,6 +552,17 @@ class ProtectNVREntity(BaseProtectEntity): @callback @override def _async_set_device_info(self) -> None: + if self.data.api.is_public_only: + # The public NVR carries no market name, version or console URL. + mac = _async_unifi_mac_from_hass(self.device.mac) + self._attr_device_info = DeviceInfo( + connections={(dr.CONNECTION_NETWORK_MAC, mac)}, + identifiers={(DOMAIN, mac)}, + manufacturer=DEFAULT_BRAND, + name=self.device.display_name or None, + model=self.device.type, + ) + return self._attr_device_info = DeviceInfo( connections={(dr.CONNECTION_NETWORK_MAC, self.device.mac)}, identifiers={(DOMAIN, self.device.mac)}, diff --git a/homeassistant/components/unifiprotect/select.py b/homeassistant/components/unifiprotect/select.py index 3686158ff44a..ad64b2df0ecd 100644 --- a/homeassistant/components/unifiprotect/select.py +++ b/homeassistant/components/unifiprotect/select.py @@ -4,7 +4,7 @@ from collections.abc import Callable, Sequence from dataclasses import dataclass from enum import Enum import logging -from typing import Any, override +from typing import Any, cast, override from uiprotect.api import ProtectApiClient from uiprotect.data import ( @@ -27,6 +27,7 @@ from uiprotect.data import ( ) from uiprotect.data.public_devices import ( PublicCamera, + PublicDeviceModel, PublicLight, SensorFeatureCapability, ) @@ -36,6 +37,7 @@ from homeassistant.components.select import SelectEntity, SelectEntityDescriptio from homeassistant.const import EntityCategory, Platform from homeassistant.core import HomeAssistant, callback from homeassistant.exceptions import HomeAssistantError +from homeassistant.helpers.dispatcher import async_dispatcher_connect from homeassistant.helpers.entity import EntityDescription from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback @@ -380,7 +382,21 @@ async def async_setup_entry( entities.append(ProtectPTZPatrolSelect(data, device, patrols)) async_add_entities(entities) + @callback + def _add_new_public_device(device: PublicDeviceModel) -> None: + async_add_entities( + async_all_device_entities( + data, + ProtectSelects, + model_descriptions=_MODEL_DESCRIPTIONS, + public_device=device, + ) + ) + data.async_subscribe_adopt(_add_new_device) + entry.async_on_unload( + async_dispatcher_connect(hass, data.public_add_signal, _add_new_public_device) + ) entities = list( async_all_device_entities( @@ -388,18 +404,26 @@ async def async_setup_entry( ) ) - for camera in data.api.bootstrap.cameras.values(): - if camera.feature_flags.is_ptz and camera.is_adopted_by_us: - patrols = data.ptz_patrols.get(camera.id, []) - entities.append(ProtectPTZPatrolSelect(data, camera, patrols)) - api = data.api + if not api.is_public_only: + # PTZ patrols are read from the private bootstrap. + for camera in api.bootstrap.cameras.values(): + if camera.feature_flags.is_ptz and camera.is_adopted_by_us: + patrols = data.ptz_patrols.get(camera.id, []) + entities.append(ProtectPTZPatrolSelect(data, camera, patrols)) + if ( api.has_public_bootstrap and api.public_bootstrap.arm_mode is not None and api.public_bootstrap.arm_profiles ): - entities.append(ProtectNVRArmProfileSelect(data, device=api.bootstrap.nvr)) + # Without a private bootstrap the NVR is the public one, as for the alarm panel. + nvr = ( + cast(NVR, api.public_bootstrap.nvr) + if api.is_public_only + else api.bootstrap.nvr + ) + entities.append(ProtectNVRArmProfileSelect(data, device=nvr)) async_add_entities(entities) @@ -407,14 +431,13 @@ async def async_setup_entry( class ProtectSelects(ProtectDeviceEntity, SelectEntity): """A UniFi Protect Select Entity.""" - device: Camera | Light | Viewer entity_description: ProtectSelectEntityDescription _state_attrs = ("_attr_available", "_attr_options", "_attr_current_option") def __init__( self, data: ProtectData, - device: Camera | Light | Viewer, + device: ProtectDeviceType, description: ProtectSelectEntityDescription, ) -> None: """Initialize the unifi protect select entity.""" @@ -572,6 +595,8 @@ class ProtectNVRArmProfileSelect(ProtectNVREntity, SelectEntity): self._attr_current_option = ( self._id_to_name.get(profile_id) if profile_id else None ) + # Arm data comes over the public WS, so availability follows it, not the private one. + self._attr_available = self.data.last_public_update_success @callback @override diff --git a/tests/components/unifiprotect/test_init.py b/tests/components/unifiprotect/test_init.py index 1f1f0e54528a..b5ab5bd15759 100644 --- a/tests/components/unifiprotect/test_init.py +++ b/tests/components/unifiprotect/test_init.py @@ -708,6 +708,26 @@ async def test_public_only_setup( assert state.state == AlarmControlPanelState.DISARMED +async def test_public_only_setup_nameless_nvr( + hass: HomeAssistant, + device_registry: dr.DeviceRegistry, + ufp_public_only: MockUFPFixture, + setup_public_only: Callable[[], Coroutine[Any, Any, None]], +) -> None: + """A public NVR without name or type keeps a nameless device, not an empty name.""" + nvr = ufp_public_only.api.public_bootstrap.nvr + nvr.name = nvr.device_type = nvr.type = None + nvr.display_name = "" + + await setup_public_only() + + device = device_registry.async_get_device_by_identifier( + (DOMAIN, UNIFI_MAC), ufp_public_only.entry.entry_id + ) + assert device is not None + assert device.name is None + + async def test_public_only_forwards_only_public_platforms( hass: HomeAssistant, setup_public_only: Callable[[], Coroutine[Any, Any, None]], diff --git a/tests/components/unifiprotect/test_select.py b/tests/components/unifiprotect/test_select.py index b92133b5a527..bbe6deadec5f 100644 --- a/tests/components/unifiprotect/test_select.py +++ b/tests/components/unifiprotect/test_select.py @@ -1,6 +1,8 @@ """Test the UniFi Protect select platform.""" +from collections.abc import Callable, Coroutine from copy import copy +from functools import partial from typing import Any from unittest.mock import AsyncMock, Mock, patch @@ -19,16 +21,21 @@ from uiprotect.data import ( Liveview, NvrArmMode, NvrArmModeStatus, + ProtectAdoptableDeviceModel, PTZPatrol, PublicHdrMode, RecordingMode, + Sensor, Viewer, + WSAction, ) from uiprotect.data.nvr import DoorbellMessage +from uiprotect.data.public_devices import SensorFeatureCapability from uiprotect.exceptions import GlobalAlarmManagerError +from uiprotect.websocket import WebsocketState from homeassistant.components.select import ATTR_OPTIONS -from homeassistant.components.unifiprotect.const import DEFAULT_ATTRIBUTION +from homeassistant.components.unifiprotect.const import DEFAULT_ATTRIBUTION, DOMAIN from homeassistant.components.unifiprotect.select import ( CAMERA_SELECTS, LIGHT_MODE_OFF, @@ -36,6 +43,7 @@ from homeassistant.components.unifiprotect.select import ( PTZ_PATROL_STOP, VIEWER_SELECTS, ) +from homeassistant.config_entries import ConfigEntryState from homeassistant.const import ( ATTR_ATTRIBUTION, ATTR_ENTITY_ID, @@ -46,9 +54,10 @@ from homeassistant.const import ( ) from homeassistant.core import HomeAssistant from homeassistant.exceptions import HomeAssistantError -from homeassistant.helpers import entity_registry as er +from homeassistant.helpers import device_registry as dr, entity_registry as er from . import patch_ufp_method +from .conftest import UNIFI_MAC from .utils import ( MockUFPFixture, adopt_devices, @@ -58,6 +67,7 @@ from .utils import ( make_public_bootstrap, make_public_camera, make_public_light, + make_public_sensor, public_device_ws_message, remove_entities, setup_public_camera, @@ -1220,3 +1230,370 @@ async def test_select_nvr_arm_profile_ws_update( state = hass.states.get(ARM_PROFILE_ENTITY_ID) assert state is not None assert state.state == "Away (p2)" + + +def _select_keys(entity_registry: er.EntityRegistry, mac: str) -> set[str]: + """Return the description keys of the selects registered for a device.""" + prefix = f"{mac}_" + return { + entry.unique_id.removeprefix(prefix) + for entry in entity_registry.entities.values() + if entry.domain == Platform.SELECT and entry.unique_id.startswith(prefix) + } + + +def _make_streamless_public_camera(camera: Camera, **kwargs: Any) -> Mock: + """Build a public camera without RTSPS streams (snapshot-only).""" + public = make_public_camera(camera, **kwargs) + public.rtsps_streams = None + return public + + +@pytest.mark.parametrize( + ( + "fixture_name", + "make", + "key", + "value", + "option", + "setter", + "setter_call", + ), + [ + pytest.param( + "doorbell", + partial(_make_streamless_public_camera, hdr_type=PublicHdrMode.AUTO), + "hdr_mode", + "auto", + "always", + "set_hdr_mode", + ((PublicHdrMode.ON,), {}), + id="camera", + ), + pytest.param( + "light", + partial( + make_public_light, + light_mode=LightModeType.WHEN_DARK, + light_mode_enable_at=LightModeEnableType.DARK, + ), + "light_motion", + "when_dark", + "motion", + "set_light_mode", + ((LightModeType.MOTION,), {"enable_at": LightModeEnableType.ALWAYS}), + id="light", + ), + ], +) +async def test_public_only_select_end_to_end( + hass: HomeAssistant, + request: pytest.FixtureRequest, + device_registry: dr.DeviceRegistry, + entity_registry: er.EntityRegistry, + ufp_public_only: MockUFPFixture, + setup_public_only: Callable[[], Coroutine[Any, Any, None]], + fixture_name: str, + make: Callable[[ProtectAdoptableDeviceModel], Mock], + key: str, + value: str, + option: str, + setter: str, + setter_call: tuple[tuple[Any, ...], dict[str, Any]], +) -> None: + """A public-only entry builds the migrated selects from the public object. + + Private-only selects are absent, the device is registered from public + identity and a chosen option goes to the public setter. + """ + device = request.getfixturevalue(fixture_name) + public = make(device) + store = getattr(ufp_public_only.api.public_bootstrap, f"{device.model.value}s") + store[device.id] = public + + await setup_public_only() + + assert ufp_public_only.entry.state is ConfigEntryState.LOADED + assert _select_keys(entity_registry, device.mac) == {key} + + entity_id = entity_registry.async_get_entity_id( + Platform.SELECT, DOMAIN, f"{device.mac}_{key}" + ) + assert entity_id + assert hass.states.get(entity_id).state == value + + entry = entity_registry.async_get(entity_id) + assert entry + device_entry = device_registry.async_get(entry.device_id) + assert device_entry + assert device_entry.model == public.type + nvr_device = device_registry.async_get_device_by_identifier( + (DOMAIN, UNIFI_MAC), ufp_public_only.entry.entry_id + ) + assert nvr_device + assert device_entry.via_device_id == nvr_device.id + + await hass.services.async_call( + "select", + "select_option", + {ATTR_ENTITY_ID: entity_id, ATTR_OPTION: option}, + blocking=True, + ) + args, kwargs = setter_call + getattr(public, setter).assert_awaited_once_with(*args, **kwargs) + + +@pytest.mark.parametrize( + ("fixture_name", "make", "updated", "key", "before", "after"), + [ + pytest.param( + "doorbell", + partial(_make_streamless_public_camera, hdr_type=PublicHdrMode.AUTO), + partial(_make_streamless_public_camera, hdr_type=PublicHdrMode.OFF), + "hdr_mode", + "auto", + "off", + id="camera", + ), + pytest.param( + "light", + partial( + make_public_light, + light_mode=LightModeType.MOTION, + light_mode_enable_at=LightModeEnableType.ALWAYS, + ), + partial( + make_public_light, + light_mode=LightModeType.MOTION, + light_mode_enable_at=LightModeEnableType.DARK, + ), + "light_motion", + "motion", + "motion_dark", + id="light", + ), + ], +) +async def test_public_only_select_follows_public_ws( + hass: HomeAssistant, + request: pytest.FixtureRequest, + entity_registry: er.EntityRegistry, + ufp_public_only: MockUFPFixture, + setup_public_only: Callable[[], Coroutine[Any, Any, None]], + fixture_name: str, + make: Callable[[ProtectAdoptableDeviceModel], Mock], + updated: Callable[[ProtectAdoptableDeviceModel], Mock], + key: str, + before: str, + after: str, +) -> None: + """A public devices websocket frame updates the select.""" + device = request.getfixturevalue(fixture_name) + store = getattr(ufp_public_only.api.public_bootstrap, f"{device.model.value}s") + store[device.id] = make(device) + + await setup_public_only() + + entity_id = entity_registry.async_get_entity_id( + Platform.SELECT, DOMAIN, f"{device.mac}_{key}" + ) + assert entity_id + assert hass.states.get(entity_id).state == before + + store[device.id] = public = updated(device) + ufp_public_only.devices_ws_subscription(public_device_ws_message(public)) + await hass.async_block_till_done() + + assert hass.states.get(entity_id).state == after + + +async def test_public_only_select_sensor_has_none( + entity_registry: er.EntityRegistry, + sensor_all: Sensor, + ufp_public_only: MockUFPFixture, + setup_public_only: Callable[[], Coroutine[Any, Any, None]], +) -> None: + """The sense selects are private-only, so a public sensor yields none.""" + ufp_public_only.api.public_bootstrap.sensors[sensor_all.id] = make_public_sensor( + sensor_all, capabilities={SensorFeatureCapability.OPEN} + ) + + await setup_public_only() + + assert _select_keys(entity_registry, sensor_all.mac) == set() + + +async def test_public_only_select_added_after_setup( + hass: HomeAssistant, + entity_registry: er.EntityRegistry, + light: Light, + ufp_public_only: MockUFPFixture, + setup_public_only: Callable[[], Coroutine[Any, Any, None]], + caplog: pytest.LogCaptureFixture, +) -> None: + """In public-only mode a light added later gets its select from its add frame. + + The public devices websocket ``add`` frame is the only discovery signal + without a local user; a re-delivered frame must not add a second time. + """ + await setup_public_only() + assert_entity_counts(hass, Platform.SELECT, 0, 0) + + public = make_public_light(light) + ufp_public_only.api.public_bootstrap.lights[light.id] = public + msg = public_device_ws_message(public) + msg.action = WSAction.ADD + ufp_public_only.devices_ws_subscription(msg) + await hass.async_block_till_done() + + assert _select_keys(entity_registry, light.mac) == {"light_motion"} + count = len(hass.states.async_entity_ids(Platform.SELECT.value)) + + ufp_public_only.devices_ws_subscription(msg) + await hass.async_block_till_done() + + assert len(hass.states.async_entity_ids(Platform.SELECT.value)) == count + assert "already exists" not in caplog.text + + +async def test_public_only_select_sense_registry_cleanup( + entity_registry: er.EntityRegistry, + sensor_all: Sensor, + ufp_public_only: MockUFPFixture, + setup_public_only: Callable[[], Coroutine[Any, Any, None]], +) -> None: + """The capability cleanup runs without a private bootstrap.""" + stale = entity_registry.async_get_or_create( + Platform.SELECT, + DOMAIN, + f"{sensor_all.mac}_mount_type", + config_entry=ufp_public_only.entry, + ) + ufp_public_only.api.public_bootstrap.sensors[sensor_all.id] = make_public_sensor( + sensor_all, capabilities={SensorFeatureCapability.TEMPERATURE} + ) + + await setup_public_only() + + assert entity_registry.async_get(stale.entity_id) is None + + +async def test_public_only_select_nvr_arm_profile( + hass: HomeAssistant, + entity_registry: er.EntityRegistry, + ufp_public_only: MockUFPFixture, + setup_public_only: Callable[[], Coroutine[Any, Any, None]], +) -> None: + """The arm profile select is built from the public NVR without a private one.""" + pb = ufp_public_only.api.public_bootstrap + pb.arm_profiles = { + "p1": _make_arm_profile("p1", "Home"), + "p2": _make_arm_profile("p2", "Away"), + } + pb.arm_mode = _make_nvr_arm_mode(profile_id="p2") + ufp_public_only.api.set_current_arm_profile_public = AsyncMock() + + await setup_public_only() + + entity_id = entity_registry.async_get_entity_id( + Platform.SELECT, DOMAIN, f"{UNIFI_MAC}_nvr_arm_profile" + ) + assert entity_id + state = hass.states.get(entity_id) + assert state + assert state.state == "Away (p2)" + assert set(state.attributes[ATTR_OPTIONS]) == {"Home (p1)", "Away (p2)"} + + await hass.services.async_call( + "select", + "select_option", + {ATTR_ENTITY_ID: entity_id, ATTR_OPTION: "Home (p1)"}, + blocking=True, + ) + ufp_public_only.api.set_current_arm_profile_public.assert_awaited_once_with("p1") + + +async def test_public_only_select_nvr_arm_profile_ws_update( + hass: HomeAssistant, + entity_registry: er.EntityRegistry, + ufp_public_only: MockUFPFixture, + setup_public_only: Callable[[], Coroutine[Any, Any, None]], +) -> None: + """An NVR devices websocket frame updates the arm profile select.""" + pb = ufp_public_only.api.public_bootstrap + pb.arm_profiles = {"p1": _make_arm_profile("p1", "Home")} + pb.arm_mode = _make_nvr_arm_mode(profile_id="p1") + + await setup_public_only() + + entity_id = entity_registry.async_get_entity_id( + Platform.SELECT, DOMAIN, f"{UNIFI_MAC}_nvr_arm_profile" + ) + assert entity_id + assert hass.states.get(entity_id).state == "Home (p1)" + + pb.arm_profiles = { + "p1": _make_arm_profile("p1", "Home"), + "p2": _make_arm_profile("p2", "Away"), + } + pb.arm_mode = _make_nvr_arm_mode(profile_id="p2") + msg = Mock() + msg.new_obj = pb.nvr + msg.old_obj = None + ufp_public_only.devices_ws_subscription(msg) + await hass.async_block_till_done() + + state = hass.states.get(entity_id) + assert state.state == "Away (p2)" + assert set(state.attributes[ATTR_OPTIONS]) == {"Home (p1)", "Away (p2)"} + + +async def test_public_only_select_nvr_arm_profile_follows_public_ws( + hass: HomeAssistant, + entity_registry: er.EntityRegistry, + ufp_public_only: MockUFPFixture, + setup_public_only: Callable[[], Coroutine[Any, Any, None]], +) -> None: + """The arm profile select goes unavailable with the public devices websocket.""" + pb = ufp_public_only.api.public_bootstrap + pb.arm_profiles = {"p1": _make_arm_profile("p1", "Home")} + pb.arm_mode = _make_nvr_arm_mode(profile_id="p1") + + await setup_public_only() + + entity_id = entity_registry.async_get_entity_id( + Platform.SELECT, DOMAIN, f"{UNIFI_MAC}_nvr_arm_profile" + ) + assert entity_id + assert hass.states.get(entity_id).state == "Home (p1)" + + ufp_public_only.devices_ws_state_subscription(WebsocketState.DISCONNECTED) + await hass.async_block_till_done() + assert hass.states.get(entity_id).state == STATE_UNAVAILABLE + + ufp_public_only.devices_ws_state_subscription(WebsocketState.CONNECTED) + await hass.async_block_till_done() + assert hass.states.get(entity_id).state == "Home (p1)" + + +async def test_select_nvr_arm_profile_decoupled_from_private_websocket( + hass: HomeAssistant, ufp: MockUFPFixture +) -> None: + """Arm profile availability follows the public WS only: private loss is a no-op.""" + profiles = {"p1": _make_arm_profile("p1", "Home")} + ufp.api.has_public_bootstrap = True + ufp.api.public_bootstrap = _make_public_bootstrap( + arm_mode=_make_nvr_arm_mode(profile_id="p1"), profiles=profiles + ) + + await init_entry(hass, ufp, []) + assert hass.states.get(ARM_PROFILE_ENTITY_ID).state == "Home (p1)" + + assert ufp.ws_state_subscription is not None + ufp.ws_state_subscription(WebsocketState.DISCONNECTED) + await hass.async_block_till_done() + assert hass.states.get(ARM_PROFILE_ENTITY_ID).state == "Home (p1)" + + ufp.devices_ws_state_subscription(WebsocketState.DISCONNECTED) + await hass.async_block_till_done() + assert hass.states.get(ARM_PROFILE_ENTITY_ID).state == STATE_UNAVAILABLE