Add UniFi Protect selects to the API key only mode (#182744)

This commit is contained in:
Raphael Hehl
2026-09-30 12:44:19 +02:00
committed by GitHub
parent 70591e17d8
commit 1dc254789b
6 changed files with 448 additions and 33 deletions
@@ -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."""
@@ -90,6 +90,7 @@ PUBLIC_ONLY_PLATFORMS = [
Platform.EVENT,
Platform.LIGHT,
Platform.NUMBER,
Platform.SELECT,
Platform.SENSOR,
Platform.SIREN,
Platform.SWITCH,
@@ -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)},
@@ -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
@@ -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]],
+379 -2
View File
@@ -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