Migrate UniFi Protect FloodLight to the public API (#174650)

Co-authored-by: Joost Lekkerkerker <joostlek@outlook.com>
This commit is contained in:
Raphael Hehl
2026-07-15 15:49:45 +02:00
committed by GitHub
co-authored by Joost Lekkerkerker
parent 577d344b56
commit 4b24077c8c
14 changed files with 397 additions and 92 deletions
@@ -302,18 +302,18 @@ LIGHT_SENSORS: tuple[ProtectBinaryEntityDescription, ...] = (
ProtectBinaryEntityDescription(
key="dark",
translation_key="is_dark",
ufp_value="is_dark",
ufp_public_value="is_dark",
),
ProtectBinaryEntityDescription(
key="motion",
device_class=BinarySensorDeviceClass.MOTION,
ufp_value="is_pir_motion_detected",
ufp_public_value="is_pir_motion_detected",
),
ProtectBinaryEntityDescription(
key="light",
translation_key="flood_light",
entity_category=EntityCategory.DIAGNOSTIC,
ufp_value="is_light_on",
ufp_public_value="is_light_on",
ufp_perm=PermRequired.NO_WRITE,
),
ProtectBinaryEntityDescription(
@@ -328,7 +328,7 @@ LIGHT_SENSORS: tuple[ProtectBinaryEntityDescription, ...] = (
key="status_light",
translation_key="status_light",
entity_category=EntityCategory.DIAGNOSTIC,
ufp_value="light_device_settings.is_indicator_enabled",
ufp_public_value="light_device_settings.is_indicator_enabled",
ufp_perm=PermRequired.NO_WRITE,
),
)
+21 -5
View File
@@ -1,10 +1,11 @@
"""Component providing Lights for UniFi Protect."""
import logging
from typing import Any, override
from typing import Any, cast, override
from uiprotect.data import Light, ModelType, ProtectAdoptableDeviceModel
from uiprotect.data.devices import LightDeviceSettings
from uiprotect.data.public_devices import PublicLight
from homeassistant.components.light import ATTR_BRIGHTNESS, ColorMode, LightEntity
from homeassistant.core import HomeAssistant, callback
@@ -61,14 +62,29 @@ class ProtectLight(ProtectDeviceEntity, LightEntity):
_attr_supported_color_modes = {ColorMode.BRIGHTNESS}
_state_attrs = ("_attr_available", "_attr_is_on", "_attr_brightness")
@override
async def async_added_to_hass(self) -> None:
"""Read state from the public API (primed before the first update)."""
self._ufp_uses_public = True
self._ufp_public_obj = self.data.async_get_public_device(self.device)
self.async_on_remove(
self.data.async_subscribe_public(
self.device.mac, self._async_public_updated
)
)
await super().async_added_to_hass()
@callback
@override
def _async_update_device_from_protect(self, device: ProtectDeviceType) -> None:
super()._async_update_device_from_protect(device)
updated_device = self.device
self._attr_is_on = updated_device.is_light_on
self._attr_brightness = unifi_brightness_to_hass(
updated_device.light_device_settings.led_level
if (public := self._ufp_public_obj) is None:
return
light = cast(PublicLight, public)
self._attr_is_on = light.is_light_on
led_level = light.light_device_settings.led_level
self._attr_brightness = (
None if led_level is None else unifi_brightness_to_hass(led_level)
)
@async_ufp_instance_command
@@ -173,8 +173,8 @@ LIGHT_NUMBERS: tuple[ProtectNumberEntityDescription, ...] = (
ufp_min=0,
ufp_max=100,
ufp_step=1,
ufp_value="light_device_settings.pir_sensitivity",
ufp_set_method="set_sensitivity",
ufp_public_value="light_device_settings.pir_sensitivity",
ufp_set_method="set_sensitivity_public",
ufp_perm=PermRequired.WRITE,
),
ProtectNumberEntityDescription[Light](
@@ -51,7 +51,7 @@ from .entity import (
async_all_device_entities,
async_remove_unsupported_sense_entities,
)
from .utils import async_get_light_motion_current, async_ufp_instance_command
from .utils import async_get_light_motion_current_public, async_ufp_instance_command
_LOGGER = logging.getLogger(__name__)
_KEY_LIGHT_MOTION = "light_motion"
@@ -173,7 +173,7 @@ def _get_doorbell_current(obj: Camera) -> str | None:
async def _set_light_mode(obj: Light, mode: str) -> None:
lightmode, timing = LIGHT_MODE_TO_SETTINGS[mode]
await obj.set_light_settings(
await obj.set_light_mode_public(
LightModeType(lightmode),
enable_at=None if timing is None else LightModeEnableType(timing),
)
@@ -308,7 +308,7 @@ LIGHT_SELECTS: tuple[ProtectSelectEntityDescription, ...] = (
translation_key="light_mode",
entity_category=EntityCategory.CONFIG,
ufp_options=MOTION_MODE_TO_LIGHT_MODE,
ufp_value_fn=async_get_light_motion_current,
ufp_public_value_fn=async_get_light_motion_current_public,
ufp_set_method_fn=_set_light_mode,
ufp_perm=PermRequired.WRITE,
),
@@ -5,7 +5,7 @@ from dataclasses import dataclass
from datetime import datetime
from functools import partial
import logging
from typing import Any, override
from typing import Any, cast, override
from uiprotect.data import (
NVR,
@@ -16,7 +16,12 @@ from uiprotect.data import (
ProtectDeviceModel,
Sensor,
)
from uiprotect.data.public_devices import SensorFeatureCapability
from uiprotect.data.public_devices import (
PublicDeviceModel,
PublicLight,
SensorFeatureCapability,
)
from uiprotect.utils import convert_to_datetime
from homeassistant.components.sensor import (
SensorDeviceClass,
@@ -52,7 +57,7 @@ from .entity import (
async_all_device_entities,
async_remove_unsupported_sense_entities,
)
from .utils import async_get_light_motion_current
from .utils import async_get_light_motion_current_public
_LOGGER = logging.getLogger(__name__)
OBJECT_TYPE_NONE = "none"
@@ -90,6 +95,11 @@ class ProtectSensorEventEntityDescription(
"""Describes UniFi Protect Sensor entity."""
def _get_last_motion_public(obj: PublicDeviceModel) -> datetime | None:
# Public API reports last motion as a JS epoch (ms); private side a datetime.
return convert_to_datetime(cast(PublicLight, obj).last_motion)
def _get_uptime(obj: ProtectDeviceModel) -> datetime | None:
if obj.up_since is None:
return None
@@ -508,7 +518,7 @@ LIGHT_SENSORS: tuple[ProtectSensorEntityDescription, ...] = (
key="motion_last_trip_time",
translation_key="last_motion_detected",
device_class=SensorDeviceClass.TIMESTAMP,
ufp_value="last_motion",
ufp_public_value_fn=_get_last_motion_public,
entity_registry_enabled_default=False,
),
ProtectSensorEntityDescription(
@@ -516,14 +526,14 @@ LIGHT_SENSORS: tuple[ProtectSensorEntityDescription, ...] = (
translation_key="motion_sensitivity",
native_unit_of_measurement=PERCENTAGE,
entity_category=EntityCategory.DIAGNOSTIC,
ufp_value="light_device_settings.pir_sensitivity",
ufp_public_value="light_device_settings.pir_sensitivity",
ufp_perm=PermRequired.NO_WRITE,
),
ProtectSensorEntityDescription[Light](
key="light_motion",
translation_key="light_mode",
entity_category=EntityCategory.DIAGNOSTIC,
ufp_value_fn=async_get_light_motion_current,
ufp_public_value_fn=async_get_light_motion_current_public,
ufp_perm=PermRequired.NO_WRITE,
),
ProtectSensorEntityDescription(
@@ -387,8 +387,8 @@ LIGHT_SWITCHES: tuple[ProtectSwitchEntityDescription, ...] = (
key="status_light",
translation_key="status_light",
entity_category=EntityCategory.CONFIG,
ufp_value="light_device_settings.is_indicator_enabled",
ufp_set_method="set_status_light",
ufp_public_value="light_device_settings.is_indicator_enabled",
ufp_set_method="set_status_light_public",
ufp_perm=PermRequired.WRITE,
),
)
+9 -10
View File
@@ -5,18 +5,18 @@ import contextlib
from functools import wraps
from pathlib import Path
import socket
from typing import TYPE_CHECKING, Any, Concatenate
from typing import TYPE_CHECKING, Any, Concatenate, cast
from aiohttp import CookieJar
from uiprotect import ProtectApiClient
from uiprotect.data import (
Bootstrap,
ChannelQuality,
Light,
LightModeEnableType,
LightModeType,
ProtectAdoptableDeviceModel,
)
from uiprotect.data.public_devices import PublicDeviceModel, PublicLight
from uiprotect.exceptions import ClientError, NotAuthorized
from homeassistant.const import (
@@ -95,15 +95,14 @@ def async_get_devices(
@callback
def async_get_light_motion_current(obj: Light) -> str:
"""Get light motion mode for Flood Light."""
if (
obj.light_mode_settings.mode is LightModeType.MOTION
and obj.light_mode_settings.enable_at is LightModeEnableType.DARK
):
def async_get_light_motion_current_public(obj: PublicDeviceModel) -> str | None:
"""Get light motion mode for a Flood Light from the public API."""
settings = cast(PublicLight, obj).light_mode_settings
if (mode := settings.mode) is None:
return None
if mode is LightModeType.MOTION and settings.enable_at is LightModeEnableType.DARK:
return f"{LightModeType.MOTION.value}_dark"
return obj.light_mode_settings.mode.value
return mode.value
@callback
@@ -15,7 +15,6 @@ from uiprotect.data import (
Sensor,
SmartDetectObjectType,
)
from uiprotect.data.nvr import EventMetadata
from uiprotect.data.public_devices import SensorFeatureCapability
from uiprotect.websocket import WebsocketState
@@ -51,9 +50,11 @@ from .utils import (
assert_entity_counts,
ids_from_device_description,
init_entry,
make_public_light,
make_public_sensor,
public_device_ws_message,
remove_entities,
setup_public_light,
setup_public_sensor,
)
@@ -118,6 +119,7 @@ async def test_binary_sensor_setup_light(
) -> None:
"""Test binary_sensor entity setup for light devices."""
setup_public_light(ufp)
await init_entry(hass, ufp, [light])
assert_entity_counts(hass, Platform.BINARY_SENSOR, 8, 8)
@@ -729,47 +731,38 @@ async def test_binary_sensor_update_motion(
async def test_binary_sensor_update_light_motion(
hass: HomeAssistant, ufp: MockUFPFixture, light: Light, fixed_now: datetime
hass: HomeAssistant, ufp: MockUFPFixture, light: Light
) -> None:
"""Test binary_sensor motion entity."""
"""Test the light motion binary_sensor reads PIR motion from the public API."""
setup_public_light(ufp)
await init_entry(hass, ufp, [light])
assert_entity_counts(hass, Platform.BINARY_SENSOR, 8, 8)
_, entity_id = await ids_from_device_description(
hass, Platform.BINARY_SENSOR, light, LIGHT_SENSOR_WRITE[1]
)
assert hass.states.get(entity_id).state == STATE_OFF
event_metadata = EventMetadata(light_id=light.id)
event = Event(
model=ModelType.EVENT,
id="test_event_id",
type=EventType.MOTION_LIGHT,
start=fixed_now - timedelta(seconds=1),
end=None,
score=100,
smart_detect_types=[],
smart_detect_event_ids=[],
metadata=event_metadata,
api=ufp.api,
)
new_light = light.model_copy()
new_light.is_pir_motion_detected = True
new_light.last_motion_event_id = event.id
mock_msg = Mock()
mock_msg.changed_data = {}
mock_msg.new_obj = event
ufp.api.bootstrap.lights = {new_light.id: new_light}
ufp.api.bootstrap.events = {event.id: event}
ufp.ws_msg(mock_msg)
public = make_public_light(light, is_pir_motion_detected=True)
ufp.devices_ws_subscription(public_device_ws_message(public))
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state
assert state.state == STATE_ON
assert hass.states.get(entity_id).state == STATE_ON
async def test_binary_sensor_light_unavailable_without_public(
hass: HomeAssistant, ufp: MockUFPFixture, light: Light
) -> None:
"""The migrated light binary_sensors are unavailable without a public object."""
await init_entry(hass, ufp, [light])
for description in LIGHT_SENSOR_WRITE:
_, entity_id = await ids_from_device_description(
hass, Platform.BINARY_SENSOR, light, description
)
assert hass.states.get(entity_id).state == STATE_UNAVAILABLE
async def test_binary_sensor_update_mount_type_window(
+65 -14
View File
@@ -1,9 +1,8 @@
"""Test the UniFi Protect light platform."""
from unittest.mock import AsyncMock, Mock
from unittest.mock import AsyncMock
from uiprotect.data import Light
from uiprotect.data.types import LEDLevel
from uiprotect.data import DeviceState, Light
from homeassistant.components.light import ATTR_BRIGHTNESS
from homeassistant.components.unifiprotect.const import DEFAULT_ATTRIBUTION
@@ -12,6 +11,7 @@ from homeassistant.const import (
ATTR_ENTITY_ID,
STATE_OFF,
STATE_ON,
STATE_UNAVAILABLE,
Platform,
)
from homeassistant.core import HomeAssistant
@@ -22,7 +22,10 @@ from .utils import (
adopt_devices,
assert_entity_counts,
init_entry,
make_public_light,
public_device_ws_message,
remove_entities,
setup_public_light,
)
@@ -48,6 +51,7 @@ async def test_light_setup(
) -> None:
"""Test light entity setup."""
setup_public_light(ufp)
await init_entry(hass, ufp, [light, unadopted_light])
assert_entity_counts(hass, Platform.LIGHT, 1, 1)
@@ -67,21 +71,15 @@ async def test_light_setup(
async def test_light_update(
hass: HomeAssistant, ufp: MockUFPFixture, light: Light, unadopted_light: Light
) -> None:
"""Test light entity update."""
"""Test the light reads on/off and brightness from a public WS update."""
setup_public_light(ufp)
await init_entry(hass, ufp, [light, unadopted_light])
assert_entity_counts(hass, Platform.LIGHT, 1, 1)
new_light = light.model_copy()
new_light.is_light_on = True
new_light.light_device_settings.led_level = LEDLevel(3)
mock_msg = Mock()
mock_msg.changed_data = {}
mock_msg.new_obj = new_light
ufp.api.bootstrap.lights = {new_light.id: new_light}
ufp.ws_msg(mock_msg)
# Divergent public values (on, led_level 3 -> 128) prove the read path.
public = make_public_light(light, is_light_on=True, led_level=3)
ufp.devices_ws_subscription(public_device_ws_message(public))
await hass.async_block_till_done()
state = hass.states.get("light.test_light")
@@ -90,6 +88,56 @@ async def test_light_update(
assert state.attributes[ATTR_BRIGHTNESS] == 128
async def test_light_unavailable_without_public(
hass: HomeAssistant, ufp: MockUFPFixture, light: Light, unadopted_light: Light
) -> None:
"""The light is unavailable without a public object."""
await init_entry(hass, ufp, [light, unadopted_light])
assert_entity_counts(hass, Platform.LIGHT, 1, 1)
state = hass.states.get("light.test_light")
assert state
assert state.state == STATE_UNAVAILABLE
async def test_light_unavailable_on_public_disconnect(
hass: HomeAssistant, ufp: MockUFPFixture, light: Light, unadopted_light: Light
) -> None:
"""Light availability follows the public object's connection state."""
setup_public_light(ufp)
await init_entry(hass, ufp, [light, unadopted_light])
entity_id = "light.test_light"
assert hass.states.get(entity_id).state != STATE_UNAVAILABLE
public = make_public_light(light, state=DeviceState.DISCONNECTED)
ufp.devices_ws_subscription(public_device_ws_message(public))
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_UNAVAILABLE
async def test_light_brightness_none(
hass: HomeAssistant, ufp: MockUFPFixture, light: Light, unadopted_light: Light
) -> None:
"""A light without a public LED level reports no brightness."""
setup_public_light(ufp)
await init_entry(hass, ufp, [light, unadopted_light])
public = make_public_light(light, is_light_on=True)
public.light_device_settings.led_level = None
ufp.devices_ws_subscription(public_device_ws_message(public))
await hass.async_block_till_done()
state = hass.states.get("light.test_light")
assert state
assert state.state == STATE_ON
assert state.attributes[ATTR_BRIGHTNESS] is None
async def test_light_turn_on(
hass: HomeAssistant, ufp: MockUFPFixture, light: Light, unadopted_light: Light
) -> None:
@@ -98,6 +146,7 @@ async def test_light_turn_on(
light._api = ufp.api
light.api.update_light_public = AsyncMock()
setup_public_light(ufp)
await init_entry(hass, ufp, [light, unadopted_light])
assert_entity_counts(hass, Platform.LIGHT, 1, 1)
@@ -120,6 +169,7 @@ async def test_light_turn_on_with_brightness(
light._api = ufp.api
light.api.update_light_public = AsyncMock()
setup_public_light(ufp)
await init_entry(hass, ufp, [light, unadopted_light])
assert_entity_counts(hass, Platform.LIGHT, 1, 1)
@@ -146,6 +196,7 @@ async def test_light_turn_off(
light._api = ufp.api
light.api.update_light_public = AsyncMock()
setup_public_light(ufp)
await init_entry(hass, ufp, [light, unadopted_light])
assert_entity_counts(hass, Platform.LIGHT, 1, 1)
+36 -2
View File
@@ -167,8 +167,9 @@ async def test_number_setup_camera_missing_attr(
async def test_number_light_sensitivity(
hass: HomeAssistant, ufp: MockUFPFixture, light: Light
) -> None:
"""Test sensitivity number entity for lights."""
"""Test sensitivity number entity for lights (public API)."""
setup_public_light(ufp)
await init_entry(hass, ufp, [light])
assert_entity_counts(hass, Platform.NUMBER, 2, 2)
@@ -180,7 +181,7 @@ async def test_number_light_sensitivity(
)
with patch_ufp_method(
light, "set_sensitivity", new_callable=AsyncMock
light, "set_sensitivity_public", new_callable=AsyncMock
) as mock_method:
await hass.services.async_call(
"number",
@@ -192,6 +193,39 @@ async def test_number_light_sensitivity(
mock_method.assert_called_once_with(15.0)
async def test_number_light_sensitivity_public_value(
hass: HomeAssistant, ufp: MockUFPFixture, light: Light
) -> None:
"""Sensitivity reads from the public object and refreshes on a public WS update."""
setup_public_light(ufp)
await init_entry(hass, ufp, [light])
_, entity_id = await ids_from_device_description(
hass, Platform.NUMBER, light, LIGHT_NUMBERS[0]
)
# A value the private fixture (45) would not produce proves the public source.
public = make_public_light(light, pir_sensitivity=30)
ufp.devices_ws_subscription(public_device_ws_message(public))
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == "30"
async def test_number_light_sensitivity_unavailable_without_public(
hass: HomeAssistant, ufp: MockUFPFixture, light: Light
) -> None:
"""The migrated sensitivity number is unavailable without a public object."""
await init_entry(hass, ufp, [light])
_, entity_id = await ids_from_device_description(
hass, Platform.NUMBER, light, LIGHT_NUMBERS[0]
)
assert hass.states.get(entity_id).state == STATE_UNAVAILABLE
async def test_number_light_duration(
hass: HomeAssistant, ufp: MockUFPFixture, light: Light
) -> None:
+65 -2
View File
@@ -42,6 +42,7 @@ from homeassistant.const import (
ATTR_ENTITY_ID,
ATTR_OPTION,
STATE_UNAVAILABLE,
STATE_UNKNOWN,
Platform,
)
from homeassistant.core import HomeAssistant
@@ -56,9 +57,11 @@ from .utils import (
ids_from_device_description,
init_entry,
make_public_camera,
make_public_light,
public_device_ws_message,
remove_entities,
setup_public_camera,
setup_public_light,
)
@@ -113,6 +116,7 @@ async def test_select_setup_light(
"""Test select entity setup for light devices."""
light.light_mode_settings.enable_at = LightModeEnableType.DARK
setup_public_light(ufp)
await init_entry(hass, ufp, [light])
assert_entity_counts(hass, Platform.SELECT, 2, 2)
@@ -415,8 +419,9 @@ async def test_select_update_doorbell_message(
async def test_select_set_option_light_motion(
hass: HomeAssistant, ufp: MockUFPFixture, light: Light
) -> None:
"""Test Light Mode select."""
"""Test Light Mode select (public API)."""
setup_public_light(ufp)
await init_entry(hass, ufp, [light])
assert_entity_counts(hass, Platform.SELECT, 2, 2)
@@ -425,7 +430,7 @@ async def test_select_set_option_light_motion(
)
with patch_ufp_method(
light, "set_light_settings", new_callable=AsyncMock
light, "set_light_mode_public", new_callable=AsyncMock
) as mock_method:
await hass.services.async_call(
"select",
@@ -437,6 +442,64 @@ async def test_select_set_option_light_motion(
mock_method.assert_called_once_with(LightModeType.MANUAL, enable_at=None)
async def test_select_light_motion_public_value(
hass: HomeAssistant, ufp: MockUFPFixture, light: Light
) -> None:
"""Light Mode select reads from the public object and refreshes on a WS update."""
setup_public_light(ufp)
await init_entry(hass, ufp, [light])
_, entity_id = await ids_from_device_description(
hass, Platform.SELECT, light, LIGHT_SELECTS[0]
)
assert hass.states.get(entity_id).state == "motion"
# The private fixture is full-time motion; when_dark proves the public source.
public = make_public_light(
light,
light_mode=LightModeType.WHEN_DARK,
light_mode_enable_at=LightModeEnableType.DARK,
)
ufp.devices_ws_subscription(public_device_ws_message(public))
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == "when_dark"
async def test_select_light_motion_unavailable_without_public(
hass: HomeAssistant, ufp: MockUFPFixture, light: Light
) -> None:
"""The migrated light motion select is unavailable without a public object."""
await init_entry(hass, ufp, [light])
_, entity_id = await ids_from_device_description(
hass, Platform.SELECT, light, LIGHT_SELECTS[0]
)
assert hass.states.get(entity_id).state == STATE_UNAVAILABLE
async def test_select_light_motion_none(
hass: HomeAssistant, ufp: MockUFPFixture, light: Light
) -> None:
"""A light that does not report a public mode leaves the select unknown."""
setup_public_light(ufp)
await init_entry(hass, ufp, [light])
_, entity_id = await ids_from_device_description(
hass, Platform.SELECT, light, LIGHT_SELECTS[0]
)
public = make_public_light(light)
public.light_mode_settings.mode = None
ufp.devices_ws_subscription(public_device_ws_message(public))
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_UNKNOWN
async def test_select_set_option_light_camera(
hass: HomeAssistant, ufp: MockUFPFixture, light: Light, camera: Camera
) -> None:
@@ -11,11 +11,13 @@ from uiprotect.data import (
DeviceState,
Event,
EventType,
Light,
ModelType,
Sensor,
)
from uiprotect.data.nvr import EventMetadata
from uiprotect.data.public_devices import SensorFeatureCapability
from uiprotect.utils import convert_to_datetime
from uiprotect.websocket import WebsocketState
from homeassistant.components.unifiprotect.const import DEFAULT_ATTRIBUTION
@@ -23,6 +25,7 @@ from homeassistant.components.unifiprotect.sensor import (
ALL_DEVICES_SENSORS,
CAMERA_DISABLED_SENSORS,
CAMERA_SENSORS,
LIGHT_SENSORS,
MOTION_TRIP_SENSORS,
NVR_DISABLED_SENSORS,
NVR_SENSORS,
@@ -45,10 +48,12 @@ from .utils import (
enable_entity,
ids_from_device_description,
init_entry,
make_public_light,
make_public_sensor,
public_device_ws_message,
remove_entities,
reset_objects,
setup_public_light,
setup_public_sensor,
time_changed,
)
@@ -704,6 +709,13 @@ async def test_aiport_no_sensor_entities(
entities = er.async_entries_for_config_entry(entity_registry, ufp.entry.entry_id)
assert not [e for e in entities if e.unique_id.startswith(f"{aiport.mac}_")]
# Check no camera-specific sensors like motion detection exist
for entity in entities:
if entity.domain == Platform.SENSOR:
# Camera-specific sensors should not exist for AI Port
assert "detected_object" not in entity.unique_id
assert "last_motion" not in entity.unique_id
async def test_aiport_no_sensor_entities_on_runtime_adopt(
hass: HomeAssistant,
@@ -721,3 +733,43 @@ async def test_aiport_no_sensor_entities_on_runtime_adopt(
entities = er.async_entries_for_config_entry(entity_registry, ufp.entry.entry_id)
assert not [e for e in entities if e.unique_id.startswith(f"{aiport.mac}_")]
async def test_sensor_light_last_motion_public(
hass: HomeAssistant, ufp: MockUFPFixture, light: Light
) -> None:
"""The light's last-motion timestamp reads from the public API."""
setup_public_light(ufp)
await init_entry(hass, ufp, [light])
_, entity_id = await ids_from_device_description(
hass, Platform.SENSOR, light, LIGHT_SENSORS[0]
)
await enable_entity(hass, ufp.entry.entry_id, entity_id)
# A value the private fixture would not produce proves the public source.
last_motion_ms = 1700000000000
public = make_public_light(light, last_motion_ms=last_motion_ms)
ufp.devices_ws_subscription(public_device_ws_message(public))
await hass.async_block_till_done()
assert (
hass.states.get(entity_id).state
== convert_to_datetime(last_motion_ms).isoformat()
)
async def test_sensor_light_last_motion_unavailable_without_public(
hass: HomeAssistant, ufp: MockUFPFixture, light: Light
) -> None:
"""The migrated last-motion sensor is unavailable without a public object."""
await init_entry(hass, ufp, [light])
_, entity_id = await ids_from_device_description(
hass, Platform.SENSOR, light, LIGHT_SENSORS[0]
)
await enable_entity(hass, ufp.entry.entry_id, entity_id)
assert hass.states.get(entity_id).state == STATE_UNAVAILABLE
+52 -4
View File
@@ -24,7 +24,14 @@ from homeassistant.components.unifiprotect.switch import (
PRIVACY_MODE_SWITCH,
ProtectSwitchEntityDescription,
)
from homeassistant.const import ATTR_ATTRIBUTION, ATTR_ENTITY_ID, STATE_OFF, Platform
from homeassistant.const import (
ATTR_ATTRIBUTION,
ATTR_ENTITY_ID,
STATE_OFF,
STATE_ON,
STATE_UNAVAILABLE,
Platform,
)
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import entity_registry as er
@@ -37,7 +44,10 @@ from .utils import (
enable_entity,
ids_from_device_description,
init_entry,
make_public_light,
public_device_ws_message,
remove_entities,
setup_public_light,
)
CAMERA_SWITCHES_BASIC = [
@@ -139,6 +149,7 @@ async def test_switch_setup_light(
) -> None:
"""Test switch entity setup for light devices."""
setup_public_light(ufp)
await init_entry(hass, ufp, [light])
assert_entity_counts(hass, Platform.SWITCH, 4, 3)
@@ -269,6 +280,7 @@ async def test_switch_light_status(
) -> None:
"""Tests status light switch for lights."""
setup_public_light(ufp)
await init_entry(hass, ufp, [light])
assert_entity_counts(hass, Platform.SWITCH, 4, 3)
@@ -279,7 +291,7 @@ async def test_switch_light_status(
)
with patch_ufp_method(
light, "set_status_light", new_callable=AsyncMock
light, "set_status_light_public", new_callable=AsyncMock
) as mock_method:
await hass.services.async_call(
"switch", "turn_on", {ATTR_ENTITY_ID: entity_id}, blocking=True
@@ -294,6 +306,40 @@ async def test_switch_light_status(
mock_method.assert_called_with(False)
async def test_switch_light_status_public_value(
hass: HomeAssistant, ufp: MockUFPFixture, light: Light
) -> None:
"""Status light switch reads from the public object and refreshes on a WS update."""
setup_public_light(ufp)
await init_entry(hass, ufp, [light])
_, entity_id = await ids_from_device_description(
hass, Platform.SWITCH, light, LIGHT_SWITCHES[1]
)
assert hass.states.get(entity_id).state == STATE_OFF
# The private fixture has the indicator disabled; the public ON proves the source.
public = make_public_light(light, is_indicator_enabled=True)
ufp.devices_ws_subscription(public_device_ws_message(public))
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_ON
async def test_switch_light_status_unavailable_without_public(
hass: HomeAssistant, ufp: MockUFPFixture, light: Light
) -> None:
"""The migrated status light switch is unavailable without a public object."""
await init_entry(hass, ufp, [light])
_, entity_id = await ids_from_device_description(
hass, Platform.SWITCH, light, LIGHT_SWITCHES[1]
)
assert hass.states.get(entity_id).state == STATE_UNAVAILABLE
async def test_switch_camera_ssh(
hass: HomeAssistant, ufp: MockUFPFixture, doorbell: Camera
) -> None:
@@ -569,6 +615,7 @@ async def test_switch_turn_on_client_error(
) -> None:
"""Test switch turn on with ClientError raises HomeAssistantError."""
setup_public_light(ufp)
await init_entry(hass, ufp, [light])
description = LIGHT_SWITCHES[1]
@@ -580,7 +627,7 @@ async def test_switch_turn_on_client_error(
with (
patch_ufp_method(
light,
"set_status_light",
"set_status_light_public",
new_callable=AsyncMock,
side_effect=ClientError("Test error"),
),
@@ -596,6 +643,7 @@ async def test_switch_turn_on_not_authorized(
) -> None:
"""Test switch turn on with NotAuthorized raises HomeAssistantError."""
setup_public_light(ufp)
await init_entry(hass, ufp, [light])
description = LIGHT_SWITCHES[1]
@@ -607,7 +655,7 @@ async def test_switch_turn_on_not_authorized(
with (
patch_ufp_method(
light,
"set_status_light",
"set_status_light_public",
new_callable=AsyncMock,
side_effect=NotAuthorized("Not authorized"),
),
+46 -7
View File
@@ -15,6 +15,8 @@ from uiprotect.data import (
Event,
EventType,
Light,
LightModeEnableType,
LightModeType,
ModelType,
MountType,
ProtectAdoptableDeviceModel,
@@ -29,6 +31,7 @@ from uiprotect.data.public_devices import (
PublicHdrMode,
PublicLight,
PublicLightDeviceSettings,
PublicLightModeSettings,
PublicSensor,
PublicSensorLeakSettings,
PublicSensorMotionSettingsRead,
@@ -335,29 +338,65 @@ def make_public_light(
light: Light,
*,
state: DeviceState | None = None,
is_light_on: bool | None = None,
is_dark: bool | None = None,
is_pir_motion_detected: bool | None = None,
last_motion_ms: int | None = None,
led_level: int | None = None,
pir_duration_ms: int | None = None,
pir_sensitivity: int | None = None,
is_indicator_enabled: bool | None = None,
light_mode: LightModeType | None = None,
light_mode_enable_at: LightModeEnableType | None = None,
) -> Mock:
"""Build a public-API light for the migrated PIR auto-shutoff duration number.
"""Build a public-API light mirroring the private fixture's migrated fields.
``light_device_settings`` mirrors the private fixture (the public API reports
``pir_duration`` in milliseconds); ``pir_duration_ms`` overrides it so a test
can assert a value the private object would not produce.
Every field the FloodLight entities read over the public API is mirrored from
the private light; each ``*`` override lets a test set a value the private
object would not produce, proving the entity reads the public source. The
public API reports ``pir_duration`` and ``last_motion`` in milliseconds.
"""
lds = light.light_device_settings
lms = light.light_mode_settings
public = Mock(spec=PublicLight)
public.id = light.id
public.mac = light.mac
public.model = ModelType.LIGHT
public.state = DeviceState[light.state.name] if state is None else state
public.is_light_on = light.is_light_on if is_light_on is None else is_light_on
public.is_dark = light.is_dark if is_dark is None else is_dark
public.is_pir_motion_detected = (
light.is_pir_motion_detected
if is_pir_motion_detected is None
else is_pir_motion_detected
)
if last_motion_ms is not None:
public.last_motion = last_motion_ms
elif light.last_motion is not None:
public.last_motion = round(light.last_motion.timestamp() * 1000)
else:
public.last_motion = None
public.light_mode_settings = PublicLightModeSettings(
mode=lms.mode if light_mode is None else light_mode,
enable_at=(
lms.enable_at if light_mode_enable_at is None else light_mode_enable_at
),
)
public.light_device_settings = PublicLightDeviceSettings(
is_indicator_enabled=lds.is_indicator_enabled,
led_level=lds.led_level,
is_indicator_enabled=(
lds.is_indicator_enabled
if is_indicator_enabled is None
else is_indicator_enabled
),
led_level=lds.led_level if led_level is None else led_level,
pir_duration=(
round(lds.pir_duration.total_seconds() * 1000)
if pir_duration_ms is None
else pir_duration_ms
),
pir_sensitivity=lds.pir_sensitivity,
pir_sensitivity=(
lds.pir_sensitivity if pir_sensitivity is None else pir_sensitivity
),
)
return public