jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-04-24 09:17:53 +00:00
Code Issues Packages Projects Releases Wiki Activity

Unifiprotect Add user information retrieval for NFC and fingerprint events (#132604)

Browse Source 
Co-authored-by: J. Nick Koston <nick@koston.org>
This commit is contained in:
Raphael Hehl 2024-12-22 21:50:30 +01:00 committed by GitHub
parent 368e958457
commit 0f18f128fd
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 417 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

62
homeassistant/components/unifiprotect/event.py
View File

@ -4,8 +4,6 @@ from __future__ import annotations
import dataclasses
from uiprotect.data import Camera, EventType, ProtectAdoptableDeviceModel
from homeassistant.components.event import (
EventDeviceClass,
EventEntity,
@ -14,17 +12,43 @@ from homeassistant.components.event import (
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from . import Bootstrap
from .const import (
ATTR_EVENT_ID,
EVENT_TYPE_DOORBELL_RING,
EVENT_TYPE_FINGERPRINT_IDENTIFIED,
EVENT_TYPE_FINGERPRINT_NOT_IDENTIFIED,
EVENT_TYPE_NFC_SCANNED,
KEYRINGS_KEY_TYPE_ID_NFC,
KEYRINGS_ULP_ID,
KEYRINGS_USER_FULL_NAME,
KEYRINGS_USER_STATUS,
)
from .data import (
Camera,
EventType,
ProtectAdoptableDeviceModel,
ProtectData,
ProtectDeviceType,
UFPConfigEntry,
)
from .data import ProtectData, ProtectDeviceType, UFPConfigEntry
from .entity import EventEntityMixin, ProtectDeviceEntity, ProtectEventMixin
def _add_ulp_user_infos(
bootstrap: Bootstrap, event_data: dict[str, str], ulp_id: str
) -> None:
"""Add ULP user information to the event data."""
if ulp_usr := bootstrap.ulp_users.by_ulp_id(ulp_id):
event_data.update(
{
KEYRINGS_ULP_ID: ulp_usr.ulp_id,
KEYRINGS_USER_FULL_NAME: ulp_usr.full_name,
KEYRINGS_USER_STATUS: ulp_usr.status,
}
)
@dataclasses.dataclass(frozen=True, kw_only=True)
class ProtectEventEntityDescription(ProtectEventMixin, EventEntityDescription):
"""Describes UniFi Protect event entity."""
@ -78,9 +102,22 @@ class ProtectDeviceNFCEventEntity(EventEntityMixin, ProtectDeviceEntity, EventEn
and not self._event_already_ended(prev_event, prev_event_end)
and event.type is EventType.NFC_CARD_SCANNED
):
event_data = {ATTR_EVENT_ID: event.id}
event_data = {
ATTR_EVENT_ID: event.id,
KEYRINGS_USER_FULL_NAME: "",
KEYRINGS_ULP_ID: "",
KEYRINGS_USER_STATUS: "",
KEYRINGS_KEY_TYPE_ID_NFC: "",
}
if event.metadata and event.metadata.nfc and event.metadata.nfc.nfc_id:
event_data["nfc_id"] = event.metadata.nfc.nfc_id
nfc_id = event.metadata.nfc.nfc_id
event_data[KEYRINGS_KEY_TYPE_ID_NFC] = nfc_id
keyring = self.data.api.bootstrap.keyrings.by_registry_id(nfc_id)
if keyring and keyring.ulp_user:
_add_ulp_user_infos(
self.data.api.bootstrap, event_data, keyring.ulp_user
)
self._trigger_event(EVENT_TYPE_NFC_SCANNED, event_data)
self.async_write_ha_state()
@ -109,17 +146,22 @@ class ProtectDeviceFingerprintEventEntity(
and not self._event_already_ended(prev_event, prev_event_end)
and event.type is EventType.FINGERPRINT_IDENTIFIED
):
event_data = {ATTR_EVENT_ID: event.id}
event_data = {
ATTR_EVENT_ID: event.id,
KEYRINGS_USER_FULL_NAME: "",
KEYRINGS_ULP_ID: "",
}
event_identified = EVENT_TYPE_FINGERPRINT_NOT_IDENTIFIED
if (
event.metadata
and event.metadata.fingerprint
and event.metadata.fingerprint.ulp_id
):
event_data["ulp_id"] = event.metadata.fingerprint.ulp_id
event_identified = EVENT_TYPE_FINGERPRINT_IDENTIFIED
else:
event_data["ulp_id"] = ""
event_identified = EVENT_TYPE_FINGERPRINT_NOT_IDENTIFIED
ulp_id = event.metadata.fingerprint.ulp_id
if ulp_id:
event_data[KEYRINGS_ULP_ID] = ulp_id
_add_ulp_user_infos(self.data.api.bootstrap, event_data, ulp_id)
self._trigger_event(event_identified, event_data)
self.async_write_ha_state()

368
tests/components/unifiprotect/test_event.py
View File

@ -175,6 +175,10 @@ async def test_doorbell_nfc_scanned(
Platform.EVENT, doorbell, EVENT_DESCRIPTIONS[1]
)
ulp_id = "ulp_id"
test_user_full_name = "Test User"
test_nfc_id = "test_nfc_id"
unsub = async_track_state_change_event(hass, entity_id, _capture_event)
event = Event(
model=ModelType.EVENT,
@ -187,7 +191,224 @@ async def test_doorbell_nfc_scanned(
smart_detect_event_ids=[],
camera_id=doorbell.id,
api=ufp.api,
metadata={"nfc": {"nfc_id": "test_nfc_id", "user_id": "test_user_id"}},
metadata={"nfc": {"nfc_id": test_nfc_id, "user_id": "test_user_id"}},
)
new_camera = doorbell.copy()
new_camera.last_nfc_card_scanned_event_id = "test_event_id"
ufp.api.bootstrap.cameras = {new_camera.id: new_camera}
ufp.api.bootstrap.events = {event.id: event}
mock_keyring = Mock()
mock_keyring.registry_id = test_nfc_id
mock_keyring.registry_type = "nfc"
mock_keyring.ulp_user = ulp_id
ufp.api.bootstrap.keyrings.add(mock_keyring)
mock_ulp_user = Mock()
mock_ulp_user.ulp_id = ulp_id
mock_ulp_user.full_name = test_user_full_name
mock_ulp_user.status = "ACTIVE"
ufp.api.bootstrap.ulp_users.add(mock_ulp_user)
mock_msg = Mock()
mock_msg.changed_data = {}
mock_msg.new_obj = event
ufp.ws_msg(mock_msg)
await hass.async_block_till_done()
assert len(events) == 1
state = events[0].data["new_state"]
assert state
assert state.attributes[ATTR_ATTRIBUTION] == DEFAULT_ATTRIBUTION
assert state.attributes[ATTR_EVENT_ID] == "test_event_id"
assert state.attributes["nfc_id"] == "test_nfc_id"
assert state.attributes["full_name"] == test_user_full_name
unsub()
async def test_doorbell_nfc_scanned_ulpusr_deactivated(
hass: HomeAssistant,
ufp: MockUFPFixture,
doorbell: Camera,
unadopted_camera: Camera,
fixed_now: datetime,
) -> None:
"""Test a doorbell NFC scanned event."""
await init_entry(hass, ufp, [doorbell, unadopted_camera])
assert_entity_counts(hass, Platform.EVENT, 3, 3)
events: list[HAEvent] = []
@callback
def _capture_event(event: HAEvent) -> None:
events.append(event)
_, entity_id = ids_from_device_description(
Platform.EVENT, doorbell, EVENT_DESCRIPTIONS[1]
)
ulp_id = "ulp_id"
test_user_full_name = "Test User"
test_nfc_id = "test_nfc_id"
unsub = async_track_state_change_event(hass, entity_id, _capture_event)
event = Event(
model=ModelType.EVENT,
id="test_event_id",
type=EventType.NFC_CARD_SCANNED,
start=fixed_now - timedelta(seconds=1),
end=None,
score=100,
smart_detect_types=[],
smart_detect_event_ids=[],
camera_id=doorbell.id,
api=ufp.api,
metadata={"nfc": {"nfc_id": test_nfc_id, "user_id": "test_user_id"}},
)
new_camera = doorbell.copy()
new_camera.last_nfc_card_scanned_event_id = "test_event_id"
ufp.api.bootstrap.cameras = {new_camera.id: new_camera}
ufp.api.bootstrap.events = {event.id: event}
mock_keyring = Mock()
mock_keyring.registry_id = test_nfc_id
mock_keyring.registry_type = "nfc"
mock_keyring.ulp_user = ulp_id
ufp.api.bootstrap.keyrings.add(mock_keyring)
mock_ulp_user = Mock()
mock_ulp_user.ulp_id = ulp_id
mock_ulp_user.full_name = test_user_full_name
mock_ulp_user.status = "DEACTIVATED"
ufp.api.bootstrap.ulp_users.add(mock_ulp_user)
mock_msg = Mock()
mock_msg.changed_data = {}
mock_msg.new_obj = event
ufp.ws_msg(mock_msg)
await hass.async_block_till_done()
assert len(events) == 1
state = events[0].data["new_state"]
assert state
assert state.attributes[ATTR_ATTRIBUTION] == DEFAULT_ATTRIBUTION
assert state.attributes[ATTR_EVENT_ID] == "test_event_id"
assert state.attributes["nfc_id"] == "test_nfc_id"
assert state.attributes["full_name"] == "Test User"
assert state.attributes["user_status"] == "DEACTIVATED"
unsub()
async def test_doorbell_nfc_scanned_no_ulpusr(
hass: HomeAssistant,
ufp: MockUFPFixture,
doorbell: Camera,
unadopted_camera: Camera,
fixed_now: datetime,
) -> None:
"""Test a doorbell NFC scanned event."""
await init_entry(hass, ufp, [doorbell, unadopted_camera])
assert_entity_counts(hass, Platform.EVENT, 3, 3)
events: list[HAEvent] = []
@callback
def _capture_event(event: HAEvent) -> None:
events.append(event)
_, entity_id = ids_from_device_description(
Platform.EVENT, doorbell, EVENT_DESCRIPTIONS[1]
)
ulp_id = "ulp_id"
test_nfc_id = "test_nfc_id"
unsub = async_track_state_change_event(hass, entity_id, _capture_event)
event = Event(
model=ModelType.EVENT,
id="test_event_id",
type=EventType.NFC_CARD_SCANNED,
start=fixed_now - timedelta(seconds=1),
end=None,
score=100,
smart_detect_types=[],
smart_detect_event_ids=[],
camera_id=doorbell.id,
api=ufp.api,
metadata={"nfc": {"nfc_id": test_nfc_id, "user_id": "test_user_id"}},
)
new_camera = doorbell.copy()
new_camera.last_nfc_card_scanned_event_id = "test_event_id"
ufp.api.bootstrap.cameras = {new_camera.id: new_camera}
ufp.api.bootstrap.events = {event.id: event}
mock_keyring = Mock()
mock_keyring.registry_id = test_nfc_id
mock_keyring.registry_type = "nfc"
mock_keyring.ulp_user = ulp_id
ufp.api.bootstrap.keyrings.add(mock_keyring)
mock_msg = Mock()
mock_msg.changed_data = {}
mock_msg.new_obj = event
ufp.ws_msg(mock_msg)
await hass.async_block_till_done()
assert len(events) == 1
state = events[0].data["new_state"]
assert state
assert state.attributes[ATTR_ATTRIBUTION] == DEFAULT_ATTRIBUTION
assert state.attributes[ATTR_EVENT_ID] == "test_event_id"
assert state.attributes["nfc_id"] == "test_nfc_id"
assert state.attributes["full_name"] == ""
unsub()
async def test_doorbell_nfc_scanned_no_keyring(
hass: HomeAssistant,
ufp: MockUFPFixture,
doorbell: Camera,
unadopted_camera: Camera,
fixed_now: datetime,
) -> None:
"""Test a doorbell NFC scanned event."""
await init_entry(hass, ufp, [doorbell, unadopted_camera])
assert_entity_counts(hass, Platform.EVENT, 3, 3)
events: list[HAEvent] = []
@callback
def _capture_event(event: HAEvent) -> None:
events.append(event)
_, entity_id = ids_from_device_description(
Platform.EVENT, doorbell, EVENT_DESCRIPTIONS[1]
)
test_nfc_id = "test_nfc_id"
unsub = async_track_state_change_event(hass, entity_id, _capture_event)
event = Event(
model=ModelType.EVENT,
id="test_event_id",
type=EventType.NFC_CARD_SCANNED,
start=fixed_now - timedelta(seconds=1),
end=None,
score=100,
smart_detect_types=[],
smart_detect_event_ids=[],
camera_id=doorbell.id,
api=ufp.api,
metadata={"nfc": {"nfc_id": test_nfc_id, "user_id": "test_user_id"}},
)
new_camera = doorbell.model_copy()
@ -208,6 +429,7 @@ async def test_doorbell_nfc_scanned(
assert state.attributes[ATTR_ATTRIBUTION] == DEFAULT_ATTRIBUTION
assert state.attributes[ATTR_EVENT_ID] == "test_event_id"
assert state.attributes["nfc_id"] == "test_nfc_id"
assert state.attributes["full_name"] == ""
unsub()
@ -233,6 +455,9 @@ async def test_doorbell_fingerprint_identified(
Platform.EVENT, doorbell, EVENT_DESCRIPTIONS[2]
)
ulp_id = "ulp_id"
test_user_full_name = "Test User"
unsub = async_track_state_change_event(hass, entity_id, _capture_event)
event = Event(
model=ModelType.EVENT,
@ -245,7 +470,143 @@ async def test_doorbell_fingerprint_identified(
smart_detect_event_ids=[],
camera_id=doorbell.id,
api=ufp.api,
metadata={"fingerprint": {"ulp_id": "test_ulp_id"}},
metadata={"fingerprint": {"ulp_id": ulp_id}},
)
new_camera = doorbell.copy()
new_camera.last_fingerprint_identified_event_id = "test_event_id"
ufp.api.bootstrap.cameras = {new_camera.id: new_camera}
ufp.api.bootstrap.events = {event.id: event}
mock_ulp_user = Mock()
mock_ulp_user.ulp_id = ulp_id
mock_ulp_user.full_name = test_user_full_name
mock_ulp_user.status = "ACTIVE"
ufp.api.bootstrap.ulp_users.add(mock_ulp_user)
mock_msg = Mock()
mock_msg.changed_data = {}
mock_msg.new_obj = event
ufp.ws_msg(mock_msg)
await hass.async_block_till_done()
assert len(events) == 1
state = events[0].data["new_state"]
assert state
assert state.attributes[ATTR_ATTRIBUTION] == DEFAULT_ATTRIBUTION
assert state.attributes[ATTR_EVENT_ID] == "test_event_id"
assert state.attributes["ulp_id"] == ulp_id
assert state.attributes["full_name"] == test_user_full_name
unsub()
async def test_doorbell_fingerprint_identified_user_deactivated(
hass: HomeAssistant,
ufp: MockUFPFixture,
doorbell: Camera,
unadopted_camera: Camera,
fixed_now: datetime,
) -> None:
"""Test a doorbell fingerprint identified event."""
await init_entry(hass, ufp, [doorbell, unadopted_camera])
assert_entity_counts(hass, Platform.EVENT, 3, 3)
events: list[HAEvent] = []
@callback
def _capture_event(event: HAEvent) -> None:
events.append(event)
_, entity_id = ids_from_device_description(
Platform.EVENT, doorbell, EVENT_DESCRIPTIONS[2]
)
ulp_id = "ulp_id"
test_user_full_name = "Test User"
unsub = async_track_state_change_event(hass, entity_id, _capture_event)
event = Event(
model=ModelType.EVENT,
id="test_event_id",
type=EventType.FINGERPRINT_IDENTIFIED,
start=fixed_now - timedelta(seconds=1),
end=None,
score=100,
smart_detect_types=[],
smart_detect_event_ids=[],
camera_id=doorbell.id,
api=ufp.api,
metadata={"fingerprint": {"ulp_id": ulp_id}},
)
new_camera = doorbell.copy()
new_camera.last_fingerprint_identified_event_id = "test_event_id"
ufp.api.bootstrap.cameras = {new_camera.id: new_camera}
ufp.api.bootstrap.events = {event.id: event}
mock_ulp_user = Mock()
mock_ulp_user.ulp_id = ulp_id
mock_ulp_user.full_name = test_user_full_name
mock_ulp_user.status = "DEACTIVATED"
ufp.api.bootstrap.ulp_users.add(mock_ulp_user)
mock_msg = Mock()
mock_msg.changed_data = {}
mock_msg.new_obj = event
ufp.ws_msg(mock_msg)
await hass.async_block_till_done()
assert len(events) == 1
state = events[0].data["new_state"]
assert state
assert state.attributes[ATTR_ATTRIBUTION] == DEFAULT_ATTRIBUTION
assert state.attributes[ATTR_EVENT_ID] == "test_event_id"
assert state.attributes["ulp_id"] == ulp_id
assert state.attributes["full_name"] == "Test User"
assert state.attributes["user_status"] == "DEACTIVATED"
unsub()
async def test_doorbell_fingerprint_identified_no_user(
hass: HomeAssistant,
ufp: MockUFPFixture,
doorbell: Camera,
unadopted_camera: Camera,
fixed_now: datetime,
) -> None:
"""Test a doorbell fingerprint identified event."""
await init_entry(hass, ufp, [doorbell, unadopted_camera])
assert_entity_counts(hass, Platform.EVENT, 3, 3)
events: list[HAEvent] = []
@callback
def _capture_event(event: HAEvent) -> None:
events.append(event)
_, entity_id = ids_from_device_description(
Platform.EVENT, doorbell, EVENT_DESCRIPTIONS[2]
)
ulp_id = "ulp_id"
unsub = async_track_state_change_event(hass, entity_id, _capture_event)
event = Event(
model=ModelType.EVENT,
id="test_event_id",
type=EventType.FINGERPRINT_IDENTIFIED,
start=fixed_now - timedelta(seconds=1),
end=None,
score=100,
smart_detect_types=[],
smart_detect_event_ids=[],
camera_id=doorbell.id,
api=ufp.api,
metadata={"fingerprint": {"ulp_id": ulp_id}},
)
new_camera = doorbell.model_copy()
@ -265,7 +626,8 @@ async def test_doorbell_fingerprint_identified(
assert state
assert state.attributes[ATTR_ATTRIBUTION] == DEFAULT_ATTRIBUTION
assert state.attributes[ATTR_EVENT_ID] == "test_event_id"
assert state.attributes["ulp_id"] == "test_ulp_id"
assert state.attributes["ulp_id"] == ulp_id
assert state.attributes["full_name"] == ""
unsub()