Add UniFi Protect key fob (USL-FOB) support (#175630)

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Steven Beshensky
2026-09-08 17:49:12 +02:00
committed by GitHub
co-authored by Claude Opus 4.8
parent 11d6fb6198
commit 48155c888f
15 changed files with 1095 additions and 20 deletions
@@ -1,12 +1,13 @@
"""Component providing binary sensors for UniFi Protect."""
from collections.abc import Sequence
from collections.abc import Callable, Sequence
import dataclasses
import operator
from typing import cast, override
from uiprotect.data import (
NVR,
Fob,
ModelType,
MountType,
ProtectAdoptableDeviceModel,
@@ -26,6 +27,7 @@ from homeassistant.components.binary_sensor import (
)
from homeassistant.const import EntityCategory, Platform
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .data import ProtectData, ProtectDeviceType, UFPConfigEntry
@@ -36,6 +38,7 @@ from .entity import (
ProtectDeviceEntity,
ProtectEntityDescription,
ProtectEventMixin,
ProtectFobEntity,
ProtectIsOnEntity,
ProtectNVREntity,
async_all_device_entities,
@@ -727,6 +730,53 @@ def _async_nvr_entities(
]
def _fob_battery_low(fob: Fob) -> bool | None:
"""Return whether the key fob battery is low, if it has been reported."""
if (battery := fob.wireless_connection_state.battery_status) is not None:
return battery.is_low
return None
@dataclasses.dataclass(frozen=True, kw_only=True)
class ProtectFobBinaryEntityDescription(BinarySensorEntityDescription):
"""Describes a UniFi Protect key fob binary sensor entity."""
value_fn: Callable[[Fob], bool | None]
FOB_BINARY_SENSORS: tuple[ProtectFobBinaryEntityDescription, ...] = (
ProtectFobBinaryEntityDescription(
key="battery_low",
device_class=BinarySensorDeviceClass.BATTERY,
entity_category=EntityCategory.DIAGNOSTIC,
value_fn=_fob_battery_low,
),
)
class ProtectFobBinarySensor(ProtectFobEntity, BinarySensorEntity):
"""A binary sensor entity for a UniFi Protect key fob (Public API)."""
entity_description: ProtectFobBinaryEntityDescription
_fob_state_attrs = ("_attr_available", "_attr_is_on")
def __init__(
self,
data: ProtectData,
fob: Fob,
description: ProtectFobBinaryEntityDescription,
) -> None:
"""Initialize the key fob binary sensor."""
self.entity_description = description
self._attr_unique_id = f"{fob.mac}_{description.key}"
super().__init__(data, fob)
@callback
@override
def _async_update_from_fob(self, fob: Fob) -> None:
self._attr_is_on = self.entity_description.value_fn(fob)
async def async_setup_entry(
hass: HomeAssistant,
entry: UFPConfigEntry,
@@ -734,6 +784,34 @@ async def async_setup_entry(
) -> None:
"""Set up binary sensors for UniFi Protect integration."""
data = entry.runtime_data
@callback
def _add_new_public_device(device: PublicDeviceModel) -> None:
if isinstance(device, Fob):
async_add_entities(
ProtectFobBinarySensor(data, device, description)
for description in FOB_BINARY_SENSORS
)
entry.async_on_unload(
async_dispatcher_connect(hass, data.public_add_signal, _add_new_public_device)
)
# 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(
ProtectFobBinarySensor(data, fob, description)
for fob in api.public_bootstrap.fobs.values()
for description in FOB_BINARY_SENSORS
)
# Everything below is driven by the private bootstrap, which public-only
# entries do not have.
if api.is_public_only:
return
async_remove_unsupported_sense_entities(
hass, Platform.BINARY_SENSOR, data, (*SENSE_SENSORS, *MOUNTABLE_SENSE_SENSORS)
)
@@ -85,8 +85,11 @@ PLATFORMS = [
# rest enumerate from the private bootstrap, which is absent in this mode.
PUBLIC_ONLY_PLATFORMS = [
Platform.ALARM_CONTROL_PANEL,
Platform.BINARY_SENSOR,
Platform.CAMERA,
Platform.EVENT,
Platform.LIGHT,
Platform.SENSOR,
]
# Stored local-user credentials do not imply the mode: they are kept on a
@@ -14,6 +14,7 @@ from uiprotect.data import (
NVR,
DeviceState,
Event,
Fob,
ModelType,
ProtectAdoptableDeviceModel,
PublicDeviceModel,
@@ -443,6 +444,91 @@ class ProtectNVREntity(BaseProtectEntity):
)
class ProtectFobEntity(Entity):
"""Base class for UniFi Protect key fob (Public API) entities.
A key fob is a public-only device: it lives in
``ProtectApiClient.public_bootstrap.fobs`` and is refreshed over the public
devices websocket, so it does not use the private-device machinery in
:class:`BaseProtectEntity`. Availability follows the public websocket health
and the fob's presence in the bootstrap, mirroring the relay switch.
Subclasses fed by the events websocket set ``_ufp_requires_events_ws`` so
they also go unavailable when that stream drops.
"""
_attr_should_poll = False
_attr_attribution = DEFAULT_ATTRIBUTION
_attr_has_entity_name = True
_ufp_requires_events_ws: bool = False
_fob_state_attrs: tuple[str, ...] = ("_attr_available",)
def __init__(self, data: ProtectData, fob: Fob) -> None:
"""Initialize the fob entity and prime its state from the bootstrap."""
self.data = data
self._fob_id = fob.id
self._fob_mac = fob.mac
self._attr_device_info = DeviceInfo(
connections={(dr.CONNECTION_NETWORK_MAC, fob.mac)},
identifiers={(DOMAIN, fob.mac)},
manufacturer=DEFAULT_BRAND,
# A freshly-paired, unnamed fob reports ``name=None``; fall back to a
# stable default so the device is never registered nameless.
name=fob.name or f"Key Fob {fob.mac}",
model="Key Fob",
via_device_id=data.nvr_device_id,
)
self._attr_available = self._async_public_available()
self._async_update_from_fob(fob)
@property
def _fob(self) -> Fob | None:
"""Return the cached fob from the public bootstrap, if still present."""
api = self.data.api
if not api.has_public_bootstrap:
return None
return api.public_bootstrap.fobs.get(self._fob_id)
@callback
def _async_public_available(self) -> bool:
"""Return whether the streams backing this entity are healthy."""
data = self.data
return data.last_public_update_success and (
not self._ufp_requires_events_ws or data.last_events_update_success
)
@callback
def _async_update_from_fob(self, fob: Fob) -> None:
"""Refresh entity state from the fob. Overridden by subclasses."""
@callback
def _async_updated(self, _obj: PublicDeviceModel | None) -> None:
"""Handle a public devices WS update for this fob.
The state is always re-read from the public bootstrap: the library
merges WS updates into it before dispatching, and ``None`` (a websocket
state change or a delete) carries no object to read.
"""
prev = [getattr(self, attr, None) for attr in self._fob_state_attrs]
if (fob := self._fob) is None:
self._attr_available = False
else:
self._attr_available = self._async_public_available()
self._async_update_from_fob(fob)
if [getattr(self, attr, None) for attr in self._fob_state_attrs] != prev:
self.async_write_ha_state()
@override
async def async_added_to_hass(self) -> None:
"""Subscribe to public devices WS updates dispatched by ProtectData."""
await super().async_added_to_hass()
self.async_on_remove(
self.data.async_subscribe_public(self._fob_mac, self._async_updated)
)
# Refresh from the bootstrap: an update or delete that landed between
# construction and this subscription would otherwise be missed.
self._async_updated(None)
class EventEntityMixin(ProtectDeviceEntity):
"""Adds motion event attributes to sensor."""
+108 -18
View File
@@ -5,8 +5,9 @@ import re
from typing import Any, override
from uiprotect import ProtectEvent
from uiprotect.data import ModelType, SmartDetectObjectType
from uiprotect.data import Fob, ModelType, PublicDeviceModel, SmartDetectObjectType
from uiprotect.data.nvr import Event, EventDetectedThumbnail
from uiprotect.data.types import EventButtonType
from homeassistant.components.event import (
DoorbellEventType,
@@ -15,6 +16,7 @@ from homeassistant.components.event import (
EventEntityDescription,
)
from homeassistant.core import CALLBACK_TYPE, HomeAssistant, callback
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.helpers.event import async_call_at
@@ -41,7 +43,12 @@ from .data import (
ProtectDeviceType,
UFPConfigEntry,
)
from .entity import EventEntityMixin, ProtectDeviceEntity, ProtectEventMixin
from .entity import (
EventEntityMixin,
ProtectDeviceEntity,
ProtectEventMixin,
ProtectFobEntity,
)
PARALLEL_UPDATES = 0
@@ -102,26 +109,17 @@ class ProtectDetectionEventEntityDescription(ProtectEventEntityDescription):
include_event_source: bool = False
class ProtectDevicePublicEventEntity(
EventEntityMixin, ProtectDeviceEntity, EventEntity
):
"""Base for entities driven by the public events WS.
class ProtectFireOnceMixin(EventEntity):
"""Dedup mixin for entities fired from the public events WS.
A detection type can surface at the event start, on a later update, or only
as the event ends, and every non-eviction change is dispatched — so firing is
deduped per ``(event id, object type, event source)``.
Availability follows the public API (device present and connected) plus the
events websocket, which is the only channel these entities fire from.
as the event ends, and every non-eviction change is dispatched, so firing is
deduped per ``(event id, surfaced type, Protect event type)``.
"""
_ufp_uses_public = True
_ufp_requires_events_ws = True
entity_description: ProtectEventEntityDescription
# A camera can run two overlapping events of the same category whose
# dispatches interleave, so dedup tracks fired object/source pairs per
# recent event id (bounded), not just the current one.
# A device can run two overlapping events of the same category whose
# dispatches interleave, so dedup tracks fired type pairs per recent event
# id (bounded), not just the current one.
_fired: dict[str, frozenset[tuple[str, EventType]]] | None = None
@callback
@@ -148,6 +146,21 @@ class ProtectDevicePublicEventEntity(
self.async_write_ha_state()
class ProtectDevicePublicEventEntity(
ProtectFireOnceMixin, EventEntityMixin, ProtectDeviceEntity, EventEntity
):
"""Base for entities driven by the public events WS.
Availability follows the public API (device present and connected) plus the
events websocket, which is the only channel these entities fire from.
"""
_ufp_uses_public = True
_ufp_requires_events_ws = True
entity_description: ProtectEventEntityDescription
class ProtectDeviceRingEventEntity(ProtectDevicePublicEventEntity):
"""A UniFi Protect doorbell ring event entity driven by the public events WS."""
@@ -565,6 +578,60 @@ class ProtectDeviceMotionEventEntity(ProtectDeviceDetectionEventEntity):
self._fire_once(event, EventType.MOTION.value, {ATTR_EVENT_ID: event.id})
# Real hardware reports an empty ``feature_flags.buttons``, so the whole
# vocabulary is declared. Sourced from the enum matched against
# ``metadata.button`` below so the two cannot drift apart.
_FOB_EVENT_TYPES: list[str] = [
button.name.lower()
for button in EventButtonType
if button is not EventButtonType.UNKNOWN
]
class ProtectFobButtonEventEntity(ProtectFireOnceMixin, ProtectFobEntity, EventEntity):
"""A UniFi Protect key fob button-press event entity.
Each fob exposes one event entity that fires the pressed button (from a
public ``sensorButtonPressed`` event's ``metadata.button``) as its event
type.
"""
_attr_translation_key = "keyfob"
_attr_event_types = _FOB_EVENT_TYPES
# Presses arrive only on the events websocket, so its health gates
# availability on top of the devices websocket.
_ufp_requires_events_ws = True
def __init__(self, data: ProtectData, fob: Fob) -> None:
"""Initialize the key fob button event entity."""
self._attr_unique_id = f"{fob.mac}_keyfob"
super().__init__(data, fob)
@override
async def async_added_to_hass(self) -> None:
"""Subscribe to public key-fob button-press events."""
await super().async_added_to_hass()
# A press arrives as a ``sensorButtonPressed`` event whose ``device`` is
# the fob and whose ``metadata.button`` is the pressed button.
self.async_on_remove(
self.data.async_subscribe_public_event(
self._fob_id, EventType.SENSOR_BUTTON_PRESSED, self._async_button_event
)
)
@callback
def _async_button_event(self, event: ProtectEvent) -> None:
if (metadata := event.metadata) is None or (button := metadata.button) is None:
return
# Skip a button added by newer firmware that coerces to
# ``EventButtonType.UNKNOWN`` (not among the declared event types).
if (button_type := button.name.lower()) not in self.event_types:
return
# A press is dispatched on every non-eviction change to its event, so
# the same press can arrive more than once.
self._fire_once(event, button_type, {ATTR_EVENT_ID: event.id})
EVENT_DESCRIPTIONS: tuple[ProtectEventEntityDescription, ...] = (
ProtectEventEntityDescription(
key="doorbell",
@@ -658,6 +725,29 @@ async def async_setup_entry(
"""Set up event entities for UniFi Protect integration."""
data = entry.runtime_data
@callback
def _add_new_public_device(device: PublicDeviceModel) -> None:
if isinstance(device, Fob):
async_add_entities([ProtectFobButtonEventEntity(data, device)])
entry.async_on_unload(
async_dispatcher_connect(hass, data.public_add_signal, _add_new_public_device)
)
# 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(
ProtectFobButtonEventEntity(data, fob)
for fob in api.public_bootstrap.fobs.values()
)
# Everything below is driven by the private bootstrap, which public-only
# entries do not have.
if api.is_public_only:
return
@callback
def _add_new_device(device: ProtectAdoptableDeviceModel) -> None:
# AiPort inherits from Camera but should not create camera-specific entities
@@ -181,6 +181,9 @@
"fingerprint": {
"default": "mdi:fingerprint"
},
"keyfob": {
"default": "mdi:remote"
},
"motion_detection": {
"default": "mdi:motion-sensor"
},
@@ -274,6 +277,9 @@
"doorbell_text": {
"default": "mdi:card-text"
},
"fob_status": {
"default": "mdi:remote"
},
"infrared_mode": {
"default": "mdi:circle-opacity"
},
@@ -10,6 +10,8 @@ from typing import Any, cast, override
from uiprotect.data import (
NVR,
Camera,
Fob,
FobAwayState,
Light,
ModelType,
ProtectAdoptableDeviceModel,
@@ -42,6 +44,7 @@ from homeassistant.const import (
UnitOfTime,
)
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .data import ProtectData, ProtectDeviceType, UFPConfigEntry
@@ -52,6 +55,7 @@ from .entity import (
ProtectDeviceEntity,
ProtectEntityDescription,
ProtectEventMixin,
ProtectFobEntity,
ProtectNVREntity,
T,
async_all_device_entities,
@@ -591,6 +595,91 @@ _MODEL_DESCRIPTIONS: dict[ModelType, Sequence[ProtectEntityDescription]] = {
}
def _fob_battery_level(fob: Fob) -> int | None:
"""Return the key fob battery percentage, if it has been reported."""
if (battery := fob.wireless_connection_state.battery_status) is not None:
return battery.percentage
return None
def _fob_signal_strength(fob: Fob) -> int | None:
"""Return the key fob Bluetooth signal strength, if it has been reported."""
if (signal := fob.wireless_connection_state.signal_state) is not None:
return signal.signal_strength
return None
def _fob_status(fob: Fob) -> str | None:
"""Return the key fob presence state.
``FobAwayState`` carries an ``UNKNOWN`` member that the library coerces
unrecognized wire values into; map it to ``None`` (unknown state) rather
than a value the enum sensor's ``options`` do not list.
"""
if (away_state := fob.away_state) is FobAwayState.UNKNOWN:
return None
return away_state.value.lower()
@dataclass(frozen=True, kw_only=True)
class ProtectFobSensorEntityDescription(SensorEntityDescription):
"""Describes a UniFi Protect key fob sensor entity."""
value_fn: Callable[[Fob], int | str | None]
FOB_SENSORS: tuple[ProtectFobSensorEntityDescription, ...] = (
ProtectFobSensorEntityDescription(
key="battery_level",
device_class=SensorDeviceClass.BATTERY,
native_unit_of_measurement=PERCENTAGE,
entity_category=EntityCategory.DIAGNOSTIC,
state_class=SensorStateClass.MEASUREMENT,
value_fn=_fob_battery_level,
),
ProtectFobSensorEntityDescription(
key="signal_strength",
device_class=SensorDeviceClass.SIGNAL_STRENGTH,
native_unit_of_measurement=SIGNAL_STRENGTH_DECIBELS_MILLIWATT,
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
state_class=SensorStateClass.MEASUREMENT,
value_fn=_fob_signal_strength,
),
ProtectFobSensorEntityDescription(
key="status",
translation_key="fob_status",
device_class=SensorDeviceClass.ENUM,
entity_category=EntityCategory.DIAGNOSTIC,
options=["online", "recently_seen", "no_recent_heartbeat", "device_lost"],
value_fn=_fob_status,
),
)
class ProtectFobSensor(ProtectFobEntity, SensorEntity):
"""A sensor entity for a UniFi Protect key fob (Public API)."""
entity_description: ProtectFobSensorEntityDescription
_fob_state_attrs = ("_attr_available", "_attr_native_value")
def __init__(
self,
data: ProtectData,
fob: Fob,
description: ProtectFobSensorEntityDescription,
) -> None:
"""Initialize the key fob sensor."""
self.entity_description = description
self._attr_unique_id = f"{fob.mac}_{description.key}"
super().__init__(data, fob)
@callback
@override
def _async_update_from_fob(self, fob: Fob) -> None:
self._attr_native_value = self.entity_description.value_fn(fob)
async def async_setup_entry(
hass: HomeAssistant,
entry: UFPConfigEntry,
@@ -598,6 +687,34 @@ async def async_setup_entry(
) -> None:
"""Set up sensors for UniFi Protect integration."""
data = entry.runtime_data
@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
)
entry.async_on_unload(
async_dispatcher_connect(hass, data.public_add_signal, _add_new_public_device)
)
# 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
)
# Everything below is driven by the private bootstrap, which public-only
# entries do not have.
if api.is_public_only:
return
async_remove_unsupported_sense_entities(hass, Platform.SENSOR, data, SENSE_SENSORS)
@callback
@@ -365,6 +365,26 @@
}
}
},
"keyfob": {
"name": "Button",
"state_attributes": {
"event_type": {
"state": {
"alarm_hub_button": "Alarm hub button",
"arm": "Arm",
"disarm": "Disarm",
"function": "Function",
"input1": "Input 1",
"input2": "Input 2",
"left": "Left",
"main": "Main",
"night": "Night",
"panic": "Panic",
"right": "Right"
}
}
}
},
"motion_detection": {
"name": "Motion detection",
"state_attributes": {
@@ -586,6 +606,15 @@
"doorbell_text": {
"name": "[%key:component::unifiprotect::entity::select::doorbell_text::name%]"
},
"fob_status": {
"name": "Status",
"state": {
"device_lost": "Device lost",
"no_recent_heartbeat": "No recent heartbeat",
"online": "Online",
"recently_seen": "Recently seen"
}
},
"infrared_mode": {
"name": "[%key:component::unifiprotect::entity::select::infrared_mode::name%]"
},
+8 -1
View File
@@ -199,6 +199,7 @@ def mock_ufp_client(bootstrap: Bootstrap):
client.public_bootstrap.lights = {}
client.public_bootstrap.relays = {}
client.public_bootstrap.sirens = {}
client.public_bootstrap.fobs = {}
client.public_bootstrap.arm_profiles = {}
client.public_bootstrap.arm_mode = None
client.public_bootstrap.nvr = Mock()
@@ -229,6 +230,7 @@ def mock_ufp_client(bootstrap: Bootstrap):
yield from pb.lights.values()
yield from pb.relays.values()
yield from pb.sirens.values()
yield from pb.fobs.values()
client.public_bootstrap.all_devices = _public_all_devices
@@ -652,7 +654,12 @@ def mock_ufp_public_only_client() -> Mock:
# them, so both helpers below read the attribute at call time.
pb.cameras = {}
pb.lights = {}
device_maps = {ModelType.CAMERA: "cameras", ModelType.LIGHT: "lights"}
pb.fobs = {}
device_maps = {
ModelType.CAMERA: "cameras",
ModelType.LIGHT: "lights",
ModelType.FOB: "fobs",
}
def _all_devices(*, include_nvr: bool = False) -> Iterator[Mock]:
if include_nvr and pb.nvr is not None:
@@ -0,0 +1,16 @@
# serializer version: 1
# name: test_fob_button_event_types
list([
'function',
'alarm_hub_button',
'arm',
'disarm',
'night',
'panic',
'left',
'right',
'input1',
'input2',
'main',
])
# ---
@@ -46,6 +46,7 @@ def _make_public_bootstrap(arm_mode: Mock | None) -> Mock:
pb.arm_profiles = {}
pb.relays = {}
pb.sirens = {}
pb.fobs = {}
return pb
+636
View File
@@ -0,0 +1,636 @@
"""Tests for the UniFi Protect key fob (Public API) entities."""
from datetime import datetime
from unittest.mock import AsyncMock, Mock
import pytest
from syrupy.assertion import SnapshotAssertion
from uiprotect import EventChange, ProtectEvent, ProtectEventChannel
from uiprotect.data import (
DeviceState,
EventType,
Fob,
FobAwayState,
FobButton,
ModelType,
PublicBootstrap,
PublicFobFeatureFlags,
WSAction,
)
from uiprotect.data.public_devices import (
PublicSignalState,
PublicWirelessBatteryStatus,
PublicWirelessConnectionState,
)
from uiprotect.data.public_event import PublicEventMetadata
from uiprotect.data.types import EventButtonType
from uiprotect.websocket import WebsocketState
from homeassistant.components.unifiprotect.const import ATTR_EVENT_ID
from homeassistant.const import ATTR_ATTRIBUTION, STATE_UNAVAILABLE
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
FOB_ID = "fob-id-1"
FOB_MAC = "AA:BB:CC:DD:EE:F0"
FOB_NAME = "Front Door Fob"
BATTERY_SENSOR = "sensor.front_door_fob_battery"
SIGNAL_SENSOR = "sensor.front_door_fob_signal_strength"
STATUS_SENSOR = "sensor.front_door_fob_status"
BATTERY_LOW_BINARY = "binary_sensor.front_door_fob_battery"
BUTTON_EVENT = "event.front_door_fob_button"
def _make_fob(
*,
buttons: list[FobButton] | None = None,
away_state: FobAwayState = FobAwayState.ONLINE,
percentage: int | None = 80,
is_low: bool = False,
signal_strength: int | None = -55,
state: DeviceState = DeviceState.CONNECTED,
name: str | None = FOB_NAME,
) -> Mock:
"""Build a mock :class:`Fob` backed by real public sub-models."""
fob = Mock(spec=Fob)
fob.id = FOB_ID
fob.mac = FOB_MAC
fob.name = name
fob.model = ModelType.FOB
fob.state = state
fob.away_state = away_state
# Real USL-FOB hardware reports an empty featureFlags.buttons.
fob.feature_flags = PublicFobFeatureFlags(buttons=buttons or [])
fob.wireless_connection_state = PublicWirelessConnectionState(
battery_status=PublicWirelessBatteryStatus(
percentage=percentage, is_low=is_low
),
signal_state=PublicSignalState(
signal_strength=signal_strength, signal_quality=None
),
)
return 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
@pytest.fixture(name="ufp_with_fob")
def _ufp_with_fob(ufp: MockUFPFixture) -> tuple[MockUFPFixture, Mock]:
"""Configure the ufp fixture with a single key fob on the public API."""
fob = _make_fob()
ufp.api.has_public_bootstrap = True
ufp.api.public_bootstrap = _make_public_bootstrap(fob)
return ufp, fob
def _button_event(
fob: Mock,
*,
event_id: str = "evt-1",
button: EventButtonType | None = EventButtonType.ARM,
now: datetime | None = None,
) -> ProtectEvent:
"""Build a button-press ``ProtectEvent`` matching a real USL-FOB capture.
A real press is born-closed (``start == end``) and carries the fob itself as
the event ``device`` with the pressed button in ``metadata.button``. Pass
``button=None`` to omit metadata entirely.
"""
when = now or datetime(2026, 1, 1)
return ProtectEvent(
id=event_id,
type=EventType.SENSOR_BUTTON_PRESSED,
channel=ProtectEventChannel.SENSOR,
device_id=fob.id,
device_mac=fob.mac,
start=when,
end=when,
metadata=None if button is None else PublicEventMetadata(button=button),
)
async def test_fob_not_created_without_public_bootstrap(
hass: HomeAssistant, ufp: MockUFPFixture
) -> None:
"""No fob entities are created when the public bootstrap is unavailable."""
ufp.api.has_public_bootstrap = False
await init_entry(hass, ufp, [])
assert hass.states.get(BATTERY_SENSOR) is None
assert hass.states.get(BUTTON_EVENT) is None
async def test_fob_entities_created(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
device_registry: dr.DeviceRegistry,
ufp_with_fob: tuple[MockUFPFixture, Mock],
) -> None:
"""The fob device and its entities are created from the public bootstrap."""
ufp, _fob = ufp_with_fob
await init_entry(hass, ufp, [])
device = device_registry.async_get_device_by_connection(
(dr.CONNECTION_NETWORK_MAC, FOB_MAC), ufp.entry.entry_id
)
assert device is not None
assert device.name == FOB_NAME
battery = entity_registry.async_get(BATTERY_SENSOR)
assert battery is not None
assert battery.unique_id == f"{FOB_MAC}_battery_level"
state = hass.states.get(BATTERY_SENSOR)
assert state is not None
assert state.state == "80"
assert state.attributes[ATTR_ATTRIBUTION]
# Signal strength is a diagnostic sensor, disabled by default.
assert hass.states.get(SIGNAL_SENSOR) is None
signal = entity_registry.async_get(SIGNAL_SENSOR)
assert signal is not None
assert signal.disabled
status = hass.states.get(STATUS_SENSOR)
assert status is not None
assert status.state == "online"
battery_low = hass.states.get(BATTERY_LOW_BINARY)
assert battery_low is not None
assert battery_low.state == "off"
async def test_fob_button_event_types(
hass: HomeAssistant,
ufp_with_fob: tuple[MockUFPFixture, Mock],
snapshot: SnapshotAssertion,
) -> None:
"""The button event entity declares the full button vocabulary.
Real hardware reports an empty ``feature_flags.buttons``, so the entity
cannot derive its types from the device and declares them all.
"""
ufp, _fob = ufp_with_fob
await init_entry(hass, ufp, [])
state = hass.states.get(BUTTON_EVENT)
assert state is not None
assert state.attributes["event_types"] == snapshot
@pytest.mark.parametrize(
("button", "expected_event_type"),
[
pytest.param(EventButtonType.ARM, "arm", id="arm"),
pytest.param(EventButtonType.DISARM, "disarm", id="disarm"),
pytest.param(EventButtonType.PANIC, "panic", id="panic"),
pytest.param(
EventButtonType.ALARM_HUB_BUTTON,
"alarm_hub_button",
id="camelcase_maps_to_snake_case",
),
],
)
async def test_fob_button_press_fires_event(
hass: HomeAssistant,
ufp_with_fob: tuple[MockUFPFixture, Mock],
button: EventButtonType,
expected_event_type: str,
) -> None:
"""A sensor button-press fires an event of the matching snake_case type."""
ufp, fob = ufp_with_fob
await init_entry(hass, ufp, [])
events: list[HAEvent] = []
@callback
def _capture(event: HAEvent) -> None:
events.append(event)
unsub = async_track_state_change_event(hass, BUTTON_EVENT, _capture)
ufp.events_msg(_button_event(fob, button=button), EventChange.STARTED)
await hass.async_block_till_done()
assert len(events) == 1
new_state = events[0].data["new_state"]
assert new_state.attributes["event_type"] == expected_event_type
assert new_state.attributes[ATTR_EVENT_ID] == "evt-1"
unsub()
@pytest.mark.parametrize(
"button",
[
pytest.param(EventButtonType.UNKNOWN, id="unknown_button"),
pytest.param(None, id="no_metadata"),
],
)
async def test_fob_button_press_ignored(
hass: HomeAssistant,
ufp_with_fob: tuple[MockUFPFixture, Mock],
button: EventButtonType | None,
) -> None:
"""A press for an unknown button, or one with no metadata, does not fire."""
ufp, fob = ufp_with_fob
await init_entry(hass, ufp, [])
events: list[HAEvent] = []
@callback
def _capture(event: HAEvent) -> None:
events.append(event)
unsub = async_track_state_change_event(hass, BUTTON_EVENT, _capture)
ufp.events_msg(_button_event(fob, button=button), EventChange.STARTED)
await hass.async_block_till_done()
assert len(events) == 0
unsub()
async def test_fob_battery_updates_from_public_ws(
hass: HomeAssistant,
ufp_with_fob: tuple[MockUFPFixture, Mock],
) -> None:
"""A public devices WS update for the fob refreshes the battery sensor."""
ufp, fob = ufp_with_fob
await init_entry(hass, ufp, [])
assert hass.states.get(BATTERY_SENSOR).state == "80"
fob.wireless_connection_state = PublicWirelessConnectionState(
battery_status=PublicWirelessBatteryStatus(percentage=42, is_low=False),
signal_state=PublicSignalState(signal_strength=-55, signal_quality=None),
)
mock_msg = Mock()
mock_msg.changed_data = {}
mock_msg.old_obj = fob
mock_msg.new_obj = fob
assert ufp.devices_ws_subscription is not None
ufp.devices_ws_subscription(mock_msg)
await hass.async_block_till_done()
assert hass.states.get(BATTERY_SENSOR).state == "42"
async def test_fob_unavailable_on_public_ws_disconnect(
hass: HomeAssistant,
ufp_with_fob: tuple[MockUFPFixture, Mock],
) -> None:
"""Fob entities go unavailable when the public websocket disconnects."""
ufp, _fob = ufp_with_fob
await init_entry(hass, ufp, [])
assert hass.states.get(BATTERY_SENSOR).state == "80"
assert ufp.devices_ws_state_subscription is not None
ufp.devices_ws_state_subscription(WebsocketState.DISCONNECTED)
await hass.async_block_till_done()
assert hass.states.get(BATTERY_SENSOR).state == STATE_UNAVAILABLE
async def test_fob_present_at_startup_not_added_twice(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
ufp: MockUFPFixture,
caplog: pytest.LogCaptureFixture,
) -> None:
"""A fob known at setup is not added a second time by a later add frame.
The startup baseline is taken from ``all_devices()``; a fob missing from it
is treated as new when any add frame for it arrives (a re-delivered frame,
or a re-pair), which would clash on unique_id.
"""
fob = _make_fob()
ufp.api.is_public_only = True
ufp.api.has_public_bootstrap = True
pb = _make_public_bootstrap(fob)
ufp.api.public_bootstrap = pb
ufp.api.update_public = AsyncMock(return_value=pb)
await init_entry(hass, ufp, [])
assert hass.states.get(BATTERY_SENSOR).state == "80"
msg = public_device_ws_message(fob)
msg.action = WSAction.ADD
assert ufp.devices_ws_subscription is not None
ufp.devices_ws_subscription(msg)
await hass.async_block_till_done()
assert "already exists" not in caplog.text
assert (
len(
[
entry
for entry in entity_registry.entities.values()
if entry.unique_id.startswith(FOB_MAC)
]
)
== 5
)
async def test_fob_button_dedup_across_dispatches(
hass: HomeAssistant,
ufp_with_fob: tuple[MockUFPFixture, Mock],
) -> None:
"""A press fires once even though every change to its event is dispatched."""
ufp, fob = ufp_with_fob
await init_entry(hass, ufp, [])
events: list[HAEvent] = []
@callback
def _capture(event: HAEvent) -> None:
events.append(event)
unsub = async_track_state_change_event(hass, BUTTON_EVENT, _capture)
for change in (EventChange.STARTED, EventChange.UPDATED, EventChange.ENDED):
ufp.events_msg(_button_event(fob), change)
await hass.async_block_till_done()
unsub()
assert len(events) == 1
# A genuinely new press on the same button still fires.
unsub = async_track_state_change_event(hass, BUTTON_EVENT, _capture)
ufp.events_msg(_button_event(fob, event_id="evt-2"), EventChange.STARTED)
await hass.async_block_till_done()
unsub()
assert len(events) == 2
async def test_fob_button_unavailable_on_events_ws_disconnect(
hass: HomeAssistant,
ufp_with_fob: tuple[MockUFPFixture, Mock],
) -> None:
"""Only the button entity follows the events websocket.
Presses arrive solely on the events stream, so the event entity must not
stay available while it is down; the sensors are fed by the devices
stream and are unaffected.
"""
ufp, _fob = ufp_with_fob
await init_entry(hass, ufp, [])
assert hass.states.get(BUTTON_EVENT).state != STATE_UNAVAILABLE
assert ufp.events_ws_state_subscription is not None
ufp.events_ws_state_subscription(WebsocketState.DISCONNECTED)
await hass.async_block_till_done()
assert hass.states.get(BUTTON_EVENT).state == STATE_UNAVAILABLE
assert hass.states.get(BATTERY_SENSOR).state == "80"
ufp.events_ws_state_subscription(WebsocketState.CONNECTED)
await hass.async_block_till_done()
assert hass.states.get(BUTTON_EVENT).state != STATE_UNAVAILABLE
async def test_fob_battery_none_is_unknown(
hass: HomeAssistant,
ufp: MockUFPFixture,
) -> None:
"""A freshly-paired fob reporting null battery yields an unknown state."""
fob = _make_fob(percentage=None)
ufp.api.has_public_bootstrap = True
ufp.api.public_bootstrap = _make_public_bootstrap(fob)
await init_entry(hass, ufp, [])
state = hass.states.get(BATTERY_SENSOR)
assert state is not None
assert state.state == "unknown"
async def test_fob_signal_sensor_when_enabled(
hass: HomeAssistant,
ufp_with_fob: tuple[MockUFPFixture, Mock],
) -> None:
"""Enabling the signal-strength sensor exposes the fob's signal value."""
ufp, _fob = ufp_with_fob
await init_entry(hass, ufp, [])
await enable_entity(hass, ufp.entry.entry_id, SIGNAL_SENSOR)
state = hass.states.get(SIGNAL_SENSOR)
assert state is not None
assert state.state == "-55"
async def test_fob_event_entity_created_with_empty_feature_flags(
hass: HomeAssistant,
ufp: MockUFPFixture,
) -> None:
"""A fob with empty feature_flags (real hardware) still gets an event entity."""
fob = _make_fob(buttons=[])
ufp.api.has_public_bootstrap = True
ufp.api.public_bootstrap = _make_public_bootstrap(fob)
await init_entry(hass, ufp, [])
state = hass.states.get(BUTTON_EVENT)
assert state is not None
assert "arm" in state.attributes["event_types"]
async def test_fob_status_reflects_away_state(
hass: HomeAssistant,
ufp: MockUFPFixture,
) -> None:
"""The status sensor reflects a lost fob's away state."""
fob = _make_fob(away_state=FobAwayState.DEVICE_LOST)
ufp.api.has_public_bootstrap = True
ufp.api.public_bootstrap = _make_public_bootstrap(fob)
await init_entry(hass, ufp, [])
state = hass.states.get(STATUS_SENSOR)
assert state is not None
assert state.state == "device_lost"
async def test_fob_unknown_away_state_is_unknown(
hass: HomeAssistant,
ufp: MockUFPFixture,
) -> None:
"""An unrecognized away state maps to an unknown status, not a ValueError."""
fob = _make_fob(away_state=FobAwayState.UNKNOWN)
ufp.api.has_public_bootstrap = True
ufp.api.public_bootstrap = _make_public_bootstrap(fob)
await init_entry(hass, ufp, [])
state = hass.states.get(STATUS_SENSOR)
assert state is not None
assert state.state == "unknown"
async def test_fob_unavailable_when_removed_from_bootstrap(
hass: HomeAssistant,
ufp_with_fob: tuple[MockUFPFixture, Mock],
) -> None:
"""A fob deleted from the public bootstrap marks its entities unavailable."""
ufp, fob = ufp_with_fob
await init_entry(hass, ufp, [])
assert hass.states.get(BATTERY_SENSOR).state == "80"
# Delete event: the library removes the object before dispatching ``None``.
del ufp.api.public_bootstrap.fobs[fob.id]
mock_msg = Mock()
mock_msg.old_obj = fob
mock_msg.new_obj = None
assert ufp.devices_ws_subscription is not None
ufp.devices_ws_subscription(mock_msg)
await hass.async_block_till_done()
state = hass.states.get(BATTERY_SENSOR)
assert state is not None
assert state.state == STATE_UNAVAILABLE
async def test_fob_without_wireless_data_is_unknown(
hass: HomeAssistant,
ufp: MockUFPFixture,
) -> None:
"""A freshly-paired fob with no battery/signal reported reads as unknown."""
fob = _make_fob()
fob.wireless_connection_state = PublicWirelessConnectionState(
battery_status=None, signal_state=None
)
ufp.api.has_public_bootstrap = True
ufp.api.public_bootstrap = _make_public_bootstrap(fob)
await init_entry(hass, ufp, [])
assert hass.states.get(BATTERY_SENSOR).state == "unknown"
assert hass.states.get(BATTERY_LOW_BINARY).state == "unknown"
await enable_entity(hass, ufp.entry.entry_id, SIGNAL_SENSOR)
assert hass.states.get(SIGNAL_SENSOR).state == "unknown"
async def test_fob_unavailable_when_public_bootstrap_lost(
hass: HomeAssistant,
ufp_with_fob: tuple[MockUFPFixture, Mock],
) -> None:
"""Losing the public bootstrap marks fob entities unavailable."""
ufp, fob = ufp_with_fob
await init_entry(hass, ufp, [])
assert hass.states.get(BATTERY_SENSOR).state == "80"
ufp.api.has_public_bootstrap = False
mock_msg = Mock()
mock_msg.changed_data = {}
mock_msg.old_obj = fob
mock_msg.new_obj = fob
assert ufp.devices_ws_subscription is not None
ufp.devices_ws_subscription(mock_msg)
await hass.async_block_till_done()
state = hass.states.get(BATTERY_SENSOR)
assert state is not None
assert state.state == STATE_UNAVAILABLE
async def test_fob_entity_counts(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
ufp_with_fob: tuple[MockUFPFixture, Mock],
) -> None:
"""Exactly the expected fob entities are created across platforms."""
ufp, _fob = ufp_with_fob
await init_entry(hass, ufp, [])
fob_entities = [
entry
for entry in entity_registry.entities.values()
if entry.unique_id.startswith(FOB_MAC)
]
# battery sensor, signal sensor, status sensor, battery-low binary, button event
assert len(fob_entities) == 5
assert sum(not entry.disabled for entry in fob_entities) == 4
async def test_fob_added_at_runtime(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
ufp: MockUFPFixture,
caplog: pytest.LogCaptureFixture,
) -> None:
"""A fob paired after setup is discovered from its public add frame."""
ufp.api.is_public_only = True
ufp.api.has_public_bootstrap = True
pb = _make_public_bootstrap(None)
ufp.api.public_bootstrap = pb
ufp.api.update_public = AsyncMock(return_value=pb)
await init_entry(hass, ufp, [])
assert hass.states.get(BATTERY_SENSOR) is None
fob = _make_fob()
pb.fobs = {fob.id: fob}
msg = public_device_ws_message(fob)
msg.action = WSAction.ADD
assert ufp.devices_ws_subscription is not None
ufp.devices_ws_subscription(msg)
await hass.async_block_till_done()
assert hass.states.get(BATTERY_SENSOR).state == "80"
assert hass.states.get(BATTERY_LOW_BINARY).state == "off"
assert hass.states.get(BUTTON_EVENT) is not None
# A re-delivered add frame is deduped before the platforms see it, so no
# duplicate unique_id error is logged.
msg = public_device_ws_message(fob)
msg.action = WSAction.ADD
ufp.devices_ws_subscription(msg)
await hass.async_block_till_done()
assert "already exists" not in caplog.text
assert (
len(
[
entry
for entry in entity_registry.entities.values()
if entry.unique_id.startswith(FOB_MAC)
]
)
== 5
)
@@ -80,6 +80,7 @@ def _make_public_bootstrap(relay: Mock | None) -> Mock:
pb.arm_mode = None
pb.arm_profiles = {}
pb.sirens = {}
pb.fobs = {}
return pb
@@ -1024,6 +1024,7 @@ def _make_public_bootstrap(arm_mode: Mock | None, profiles: dict[str, Mock]) ->
pb.arm_profiles = profiles
pb.relays = {}
pb.sirens = {}
pb.fobs = {}
return pb
@@ -73,6 +73,7 @@ def _make_public_bootstrap(siren: Mock | None) -> Mock:
pb.relays = {}
pb.arm_mode = None
pb.arm_profiles = {}
pb.fobs = {}
return pb
+3
View File
@@ -588,6 +588,7 @@ def setup_public_sensor(
pb.sensors = public_bootstrap.sensors
pb.relays = {}
pb.sirens = {}
pb.fobs = {}
pb.arm_mode = None
pb.arm_profiles = {}
@@ -618,6 +619,7 @@ def setup_public_light(ufp: MockUFPFixture) -> None:
pb.lights = public_bootstrap.lights
pb.relays = {}
pb.sirens = {}
pb.fobs = {}
pb.arm_mode = None
pb.arm_profiles = {}
@@ -648,6 +650,7 @@ def setup_public_camera(ufp: MockUFPFixture) -> None:
pb.cameras = public_bootstrap.cameras
pb.relays = {}
pb.sirens = {}
pb.fobs = {}
pb.arm_mode = None
pb.arm_profiles = {}