starred/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-10-06 22:38:02 -04:00
Code Issues Packages Projects Releases Wiki Activity

Add UniFi Protect sensors to the API key only mode (#182748)

Browse Source
This commit is contained in:
Raphael Hehl
2026-09-26 20:44:17 +02:00
committed by GitHub
parent 3b9bdca957
commit fc229b336f
3 changed files with 193 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files
homeassistant/components/unifiprotect/sensor.py
+46 -24
View File
@@ -41,6 +41,7 @@ from homeassistant.const import (
)
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity import Entity
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .data import ProtectData, ProtectDeviceType, UFPConfigEntry
@@ -671,6 +672,25 @@ class ProtectFobSensor(ProtectFobEntity, SensorEntity):
self._attr_native_value = self.entity_description.value_fn(fob)
@callback
def _async_public_entities(
data: ProtectData, device: PublicDeviceModel
) -> list[Entity]:
"""Return the sensors for one public device."""
if isinstance(device, Fob):
return [
ProtectFobSensor(data, device, description) for description in FOB_SENSORS
]
return list(
async_all_device_entities(
data,
ProtectDeviceSensor,
model_descriptions=_MODEL_DESCRIPTIONS,
public_device=device,
)
)
async def async_setup_entry(
hass: HomeAssistant,
entry: UFPConfigEntry,
@@ -681,36 +701,36 @@ async def async_setup_entry(
@callback
def _add_new_public_device(device: PublicDeviceModel) -> None:
if isinstance(device, Fob):
async_add_entities(
ProtectFobSensor(data, device, description)
for description in FOB_SENSORS
)
async_add_entities(_async_public_entities(data, device))
entry.async_on_unload(
async_dispatcher_connect(hass, data.public_add_signal, _add_new_public_device)
)
async_remove_unsupported_sense_entities(hass, Platform.SENSOR, data, SENSE_SENSORS)
entities: list[Entity] = []
# The public bootstrap is primed only with an API key and supported NVR
# firmware; without it there are no fobs to expose.
api = data.api
if api.has_public_bootstrap:
async_add_entities(
ProtectFobSensor(data, fob, description)
for fob in api.public_bootstrap.fobs.values()
for description in FOB_SENSORS
)
for fob in api.public_bootstrap.fobs.values():
entities.extend(_async_public_entities(data, fob))
# Everything below is driven by the private bootstrap, which public-only
# entries do not have.
if api.is_public_only:
# The remaining sensors read the private bootstrap; the migrated ones
# are built from the public devices instead.
entities.extend(
async_all_device_entities(
data, ProtectDeviceSensor, model_descriptions=_MODEL_DESCRIPTIONS
)
)
async_add_entities(entities)
return
async_remove_unsupported_sense_entities(hass, Platform.SENSOR, data, SENSE_SENSORS)
@callback
def _add_new_device(device: ProtectAdoptableDeviceModel) -> None:
entities = async_all_device_entities(
device_entities = async_all_device_entities(
data,
ProtectDeviceSensor,
all_descs=ALL_DEVICES_SENSORS,
@@ -723,18 +743,20 @@ async def async_setup_entry(
and isinstance(device, Camera)
and device.is_adopted_by_us
):
entities += _async_event_entities(data, ufp_device=device)
async_add_entities(entities)
device_entities += _async_event_entities(data, ufp_device=device)
async_add_entities(device_entities)
data.async_subscribe_adopt(_add_new_device)
entities = async_all_device_entities(
data,
ProtectDeviceSensor,
all_descs=ALL_DEVICES_SENSORS,
model_descriptions=_MODEL_DESCRIPTIONS,
entities.extend(
async_all_device_entities(
data,
ProtectDeviceSensor,
all_descs=ALL_DEVICES_SENSORS,
model_descriptions=_MODEL_DESCRIPTIONS,
)
)
entities += _async_event_entities(data)
entities += _async_nvr_entities(data)
entities.extend(_async_event_entities(data))
entities.extend(_async_nvr_entities(data))
async_add_entities(entities)
tests/components/unifiprotect/test_fob.py
+12 -25
View File
@@ -13,7 +13,6 @@ from uiprotect.data import (
FobAwayState,
FobButton,
ModelType,
PublicBootstrap,
PublicFobFeatureFlags,
WSAction,
)
@@ -32,7 +31,13 @@ from homeassistant.core import Event as HAEvent, HomeAssistant, callback
from homeassistant.helpers import device_registry as dr, entity_registry as er
from homeassistant.helpers.event import async_track_state_change_event
from .utils import MockUFPFixture, enable_entity, init_entry, public_device_ws_message
from .utils import (
MockUFPFixture,
enable_entity,
init_entry,
make_public_bootstrap,
public_device_ws_message,
)
FOB_ID = "fob-id-1"
FOB_MAC = "AA:BB:CC:DD:EE:F0"
@@ -78,29 +83,11 @@ def _make_fob(
def _make_public_bootstrap(fob: Mock | None) -> Mock:
"""Build a public bootstrap mock holding the given fob."""
pb = Mock(spec=PublicBootstrap)
pb.fobs = {fob.id: fob} if fob is not None else {}
pb.cameras = {}
pb.lights = {}
pb.relays = {}
pb.sirens = {}
pb.arm_mode = None
pb.arm_profiles = {}
pb.nvr = Mock()
pb.nvr.mac = "aa:bb:cc:dd:ee:ff"
pb.nvr.name = "Test NVR"
pb.nvr.display_name = "Test NVR"
pb.nvr.device_type = None
pb.nvr.type = None
# The baseline and reconnect resync enumerate all_devices(); a fob missing
# from it would be redispatched as new on every reconnect.
def _all_devices(*, include_nvr: bool = False) -> list[Mock]:
devices = list(pb.fobs.values())
return [pb.nvr, *devices] if include_nvr else devices
pb.all_devices = _all_devices
return pb
nvr = Mock(
mac="aa:bb:cc:dd:ee:ff", display_name="Test NVR", device_type=None, type=None
)
nvr.name = "Test NVR"
return make_public_bootstrap(fobs={fob.id: fob} if fob is not None else {}, nvr=nvr)
@pytest.fixture(name="ufp_with_fob")
tests/components/unifiprotect/test_sensor.py
+135 -2
View File
@@ -1,6 +1,8 @@
"""Test the UniFi Protect sensor platform."""
from collections.abc import Callable, Coroutine
from datetime import datetime, timedelta
from typing import Any
from unittest.mock import Mock
import pytest
@@ -14,13 +16,14 @@ from uiprotect.data import (
Light,
ModelType,
Sensor,
WSAction,
)
from uiprotect.data.nvr import EventMetadata
from uiprotect.data.public_devices import SensorFeatureCapability
from uiprotect.utils import convert_to_datetime
from uiprotect.utils import convert_to_datetime, to_js_time
from uiprotect.websocket import WebsocketState
from homeassistant.components.unifiprotect.const import DEFAULT_ATTRIBUTION
from homeassistant.components.unifiprotect.const import DEFAULT_ATTRIBUTION, DOMAIN
from homeassistant.components.unifiprotect.sensor import (
ALL_DEVICES_SENSORS,
CAMERA_DISABLED_SENSORS,
@@ -32,6 +35,7 @@ from homeassistant.components.unifiprotect.sensor import (
SENSE_SENSORS,
ProtectSensorEntityDescription,
)
from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import (
ATTR_ATTRIBUTION,
STATE_UNAVAILABLE,
@@ -40,6 +44,7 @@ from homeassistant.const import (
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from homeassistant.util.dt import utcnow
from .utils import (
MockUFPFixture,
@@ -48,6 +53,7 @@ from .utils import (
enable_entity,
ids_from_device_description,
init_entry,
make_public_camera,
make_public_light,
make_public_sensor,
public_device_ws_message,
@@ -824,3 +830,130 @@ async def test_sensor_light_last_motion_unavailable_without_public(
await enable_entity(hass, ufp.entry.entry_id, entity_id)
assert hass.states.get(entity_id).state == STATE_UNAVAILABLE
def _sensor_keys(entity_registry: er.EntityRegistry, mac: str) -> set[str]:
"""Return the description keys of the sensors registered for a device."""
prefix = f"{mac}_"
return {
entry.unique_id.removeprefix(prefix)
for entry in entity_registry.entities.values()
if entry.domain == Platform.SENSOR and entry.unique_id.startswith(prefix)
}
async def test_public_only_sensor_sense_end_to_end(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
sensor_all: Sensor,
ufp_public_only: MockUFPFixture,
setup_public_only: Callable[[], Coroutine[Any, Any, None]],
) -> None:
"""A public-only entry builds the migrated sense sensors from the public object.
The readings and trip timestamps follow the capability map; the private-only
sensors (alarm sound, sensitivity, mount type, paired camera) are absent.
"""
public = make_public_sensor(
sensor_all,
percentage=42,
temperature_value=21.5,
capabilities={
SensorFeatureCapability.TEMPERATURE,
SensorFeatureCapability.MOTION,
},
)
ufp_public_only.api.public_bootstrap.sensors[sensor_all.id] = public
await setup_public_only()
assert ufp_public_only.entry.state is ConfigEntryState.LOADED
keys = _sensor_keys(entity_registry, sensor_all.mac)
assert {"battery_level", "temperature_level", "motion_last_trip_time"} <= keys
assert not keys & {"alarm_sound", "sensitivity", "mount_type", "paired_camera"}
assert "humidity_level" not in keys
entity_id = entity_registry.async_get_entity_id(
Platform.SENSOR, DOMAIN, f"{sensor_all.mac}_battery_level"
)
assert entity_id
assert hass.states.get(entity_id).state == "42"
async def test_public_only_sensor_light_end_to_end(
entity_registry: er.EntityRegistry,
light: Light,
ufp_public_only: MockUFPFixture,
setup_public_only: Callable[[], Coroutine[Any, Any, None]],
) -> None:
"""A public-only entry builds the migrated floodlight sensor.
``paired_camera`` reads the private bootstrap, so it stays absent. The trip
timestamp is disabled by default, like its private counterpart.
"""
public = make_public_light(light, last_motion_ms=to_js_time(utcnow()))
ufp_public_only.api.public_bootstrap.lights[light.id] = public
await setup_public_only()
keys = _sensor_keys(entity_registry, light.mac)
assert "motion_last_trip_time" in keys
assert "paired_camera" not in keys
entity_id = entity_registry.async_get_entity_id(
Platform.SENSOR, DOMAIN, f"{light.mac}_motion_last_trip_time"
)
assert entity_id
assert (
entity_registry.async_get(entity_id).disabled_by
is er.RegistryEntryDisabler.INTEGRATION
)
async def test_public_only_sensor_camera_has_none(
entity_registry: er.EntityRegistry,
camera: Camera,
ufp_public_only: MockUFPFixture,
setup_public_only: Callable[[], Coroutine[Any, Any, None]],
) -> None:
"""The camera sensors all read the private bootstrap, so none are built."""
public = make_public_camera(camera)
public.rtsps_streams = None
ufp_public_only.api.public_bootstrap.cameras[camera.id] = public
await setup_public_only()
assert _sensor_keys(entity_registry, camera.mac) == set()
async def test_public_only_sensor_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:
"""A light added later gets its sensors from its public 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.SENSOR, 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 "motion_last_trip_time" in _sensor_keys(entity_registry, light.mac)
count = len(hass.states.async_entity_ids(Platform.SENSOR.value))
ufp_public_only.devices_ws_subscription(msg)
await hass.async_block_till_done()
assert len(hass.states.async_entity_ids(Platform.SENSOR.value)) == count
assert "already exists" not in caplog.text