Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Adds UP Chime support for UniFi Protect (#71874) 

Co-authored-by: J. Nick Koston <nick@koston.org>
pull/71497/head^2
Christopher Bailey 2022-05-20 16:16:01 -04:00 committed by GitHub
parent ad5dbae425
commit 267266c7c3
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
13 changed files with 398 additions and 81 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
homeassistant/components/unifiprotect/button.py
View File

@ -43,6 +43,22 @@ ALL_DEVICE_BUTTONS: tuple[ProtectButtonEntityDescription, ...] = (
),
)
CHIME_BUTTONS: tuple[ProtectButtonEntityDescription, ...] = (
ProtectButtonEntityDescription(
key="play",
name="Play Chime",
device_class=DEVICE_CLASS_CHIME_BUTTON,
icon="mdi:play",
ufp_press="play",
),
ProtectButtonEntityDescription(
key="play_buzzer",
name="Play Buzzer",
icon="mdi:play",
ufp_press="play_buzzer",
),
)
async def async_setup_entry(
hass: HomeAssistant,
@ -53,7 +69,7 @@ async def async_setup_entry(
data: ProtectData = hass.data[DOMAIN][entry.entry_id]
entities: list[ProtectDeviceEntity] = async_all_device_entities(
data, ProtectButton, all_descs=ALL_DEVICE_BUTTONS
data, ProtectButton, all_descs=ALL_DEVICE_BUTTONS, chime_descs=CHIME_BUTTONS
)
async_add_entities(entities)

1
homeassistant/components/unifiprotect/const.py
View File

@ -38,6 +38,7 @@ DEVICES_THAT_ADOPT = {
ModelType.VIEWPORT,
ModelType.SENSOR,
ModelType.DOORLOCK,
ModelType.CHIME,
}
DEVICES_WITH_ENTITIES = DEVICES_THAT_ADOPT | {ModelType.NVR}
DEVICES_FOR_SUBSCRIBE = DEVICES_WITH_ENTITIES | {ModelType.EVENT}

3
homeassistant/components/unifiprotect/data.py
View File

@ -12,10 +12,9 @@ from pyunifiprotect.data import (
Event,
Liveview,
ModelType,
ProtectAdoptableDeviceModel,
ProtectDeviceModel,
WSSubscriptionMessage,
)
from pyunifiprotect.data.base import ProtectAdoptableDeviceModel, ProtectDeviceModel
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import CALLBACK_TYPE, HomeAssistant, callback

6
homeassistant/components/unifiprotect/entity.py
View File

@ -8,6 +8,7 @@ from typing import Any
from pyunifiprotect.data import (
NVR,
Camera,
Chime,
Doorlock,
Event,
Light,
@ -42,7 +43,7 @@ def _async_device_entities(
entities: list[ProtectDeviceEntity] = []
for device in data.get_by_types({model_type}):
assert isinstance(device, (Camera, Light, Sensor, Viewer, Doorlock))
assert isinstance(device, (Camera, Light, Sensor, Viewer, Doorlock, Chime))
for description in descs:
if description.ufp_required_field:
required_field = get_nested_attr(device, description.ufp_required_field)
@ -75,6 +76,7 @@ def async_all_device_entities(
sense_descs: Sequence[ProtectRequiredKeysMixin] | None = None,
viewer_descs: Sequence[ProtectRequiredKeysMixin] | None = None,
lock_descs: Sequence[ProtectRequiredKeysMixin] | None = None,
chime_descs: Sequence[ProtectRequiredKeysMixin] | None = None,
all_descs: Sequence[ProtectRequiredKeysMixin] | None = None,
) -> list[ProtectDeviceEntity]:
"""Generate a list of all the device entities."""
@ -84,6 +86,7 @@ def async_all_device_entities(
sense_descs = list(sense_descs or []) + all_descs
viewer_descs = list(viewer_descs or []) + all_descs
lock_descs = list(lock_descs or []) + all_descs
chime_descs = list(chime_descs or []) + all_descs
return (
_async_device_entities(data, klass, ModelType.CAMERA, camera_descs)
@ -91,6 +94,7 @@ def async_all_device_entities(
+ _async_device_entities(data, klass, ModelType.SENSOR, sense_descs)
+ _async_device_entities(data, klass, ModelType.VIEWPORT, viewer_descs)
+ _async_device_entities(data, klass, ModelType.DOORLOCK, lock_descs)
+ _async_device_entities(data, klass, ModelType.CHIME, chime_descs)
)

15
homeassistant/components/unifiprotect/number.py
View File

@ -149,6 +149,20 @@ DOORLOCK_NUMBERS: tuple[ProtectNumberEntityDescription, ...] = (
),
)
CHIME_NUMBERS: tuple[ProtectNumberEntityDescription, ...] = (
ProtectNumberEntityDescription(
key="volume",
name="Volume",
icon="mdi:speaker",
entity_category=EntityCategory.CONFIG,
ufp_min=0,
ufp_max=100,
ufp_step=1,
ufp_value="volume",
ufp_set_method="set_volume",
),
)
async def async_setup_entry(
hass: HomeAssistant,
@ -164,6 +178,7 @@ async def async_setup_entry(
light_descs=LIGHT_NUMBERS,
sense_descs=SENSE_NUMBERS,
lock_descs=DOORLOCK_NUMBERS,
chime_descs=CHIME_NUMBERS,
)
async_add_entities(entities)

11
homeassistant/components/unifiprotect/sensor.py
View File

@ -450,6 +450,16 @@ MOTION_TRIP_SENSORS: tuple[ProtectSensorEntityDescription, ...] = (
),
)
CHIME_SENSORS: tuple[ProtectSensorEntityDescription, ...] = (
ProtectSensorEntityDescription(
key="last_ring",
name="Last Ring",
device_class=SensorDeviceClass.TIMESTAMP,
icon="mdi:bell",
ufp_value="last_ring",
),
)
async def async_setup_entry(
hass: HomeAssistant,
@ -466,6 +476,7 @@ async def async_setup_entry(
sense_descs=SENSE_SENSORS,
light_descs=LIGHT_SENSORS,
lock_descs=DOORLOCK_SENSORS,
chime_descs=CHIME_SENSORS,
)
entities += _async_motion_entities(data)
entities += _async_nvr_entities(data)

147
homeassistant/components/unifiprotect/services.py
View File

@ -10,25 +10,32 @@ from pyunifiprotect.api import ProtectApiClient
from pyunifiprotect.exceptions import BadRequest
import voluptuous as vol
from homeassistant.components.binary_sensor import BinarySensorDeviceClass
from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import ATTR_DEVICE_ID
from homeassistant.const import ATTR_DEVICE_ID, Platform
from homeassistant.core import HomeAssistant, ServiceCall, callback
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import config_validation as cv, device_registry as dr
from homeassistant.helpers import (
config_validation as cv,
device_registry as dr,
entity_registry as er,
)
from homeassistant.helpers.service import async_extract_referenced_entity_ids
from homeassistant.util.read_only_dict import ReadOnlyDict
from .const import ATTR_MESSAGE, DOMAIN
from .data import ProtectData
from .utils import _async_unifi_mac_from_hass
SERVICE_ADD_DOORBELL_TEXT = "add_doorbell_text"
SERVICE_REMOVE_DOORBELL_TEXT = "remove_doorbell_text"
SERVICE_SET_DEFAULT_DOORBELL_TEXT = "set_default_doorbell_text"
SERVICE_SET_CHIME_PAIRED = "set_chime_paired_doorbells"
ALL_GLOBAL_SERIVCES = [
SERVICE_ADD_DOORBELL_TEXT,
SERVICE_REMOVE_DOORBELL_TEXT,
SERVICE_SET_DEFAULT_DOORBELL_TEXT,
SERVICE_SET_CHIME_PAIRED,
]
DOORBELL_TEXT_SCHEMA = vol.All(
@ -41,70 +48,68 @@ DOORBELL_TEXT_SCHEMA = vol.All(
cv.has_at_least_one_key(ATTR_DEVICE_ID),
)
CHIME_PAIRED_SCHEMA = vol.All(
vol.Schema(
{
**cv.ENTITY_SERVICE_FIELDS,
"doorbells": cv.TARGET_SERVICE_FIELDS,
},
),
cv.has_at_least_one_key(ATTR_DEVICE_ID),
)
def _async_all_ufp_instances(hass: HomeAssistant) -> list[ProtectApiClient]:
"""All active UFP instances."""
return [
data.api for data in hass.data[DOMAIN].values() if isinstance(data, ProtectData)
]
def _async_ufp_instance_for_config_entry_ids(
hass: HomeAssistant, config_entry_ids: set[str]
) -> ProtectApiClient | None:
"""Find the UFP instance for the config entry ids."""
domain_data = hass.data[DOMAIN]
for config_entry_id in config_entry_ids:
if config_entry_id in domain_data:
protect_data: ProtectData = domain_data[config_entry_id]
return protect_data.api
return None
@callback
def _async_get_macs_for_device(device_entry: dr.DeviceEntry) -> list[str]:
return [
_async_unifi_mac_from_hass(cval)
for ctype, cval in device_entry.connections
if ctype == dr.CONNECTION_NETWORK_MAC
]
@callback
def _async_get_ufp_instances(
hass: HomeAssistant, device_id: str
) -> tuple[dr.DeviceEntry, ProtectApiClient]:
def _async_get_ufp_instance(hass: HomeAssistant, device_id: str) -> ProtectApiClient:
device_registry = dr.async_get(hass)
if not (device_entry := device_registry.async_get(device_id)):
raise HomeAssistantError(f"No device found for device id: {device_id}")
if device_entry.via_device_id is not None:
return _async_get_ufp_instances(hass, device_entry.via_device_id)
return _async_get_ufp_instance(hass, device_entry.via_device_id)
macs = _async_get_macs_for_device(device_entry)
ufp_instances = [
i for i in _async_all_ufp_instances(hass) if i.bootstrap.nvr.mac in macs
]
config_entry_ids = device_entry.config_entries
if ufp_instance := _async_ufp_instance_for_config_entry_ids(hass, config_entry_ids):
return ufp_instance
if not ufp_instances:
# should not be possible unless user manually enters a bad device ID
raise HomeAssistantError( # pragma: no cover
f"No UniFi Protect NVR found for device ID: {device_id}"
)
return device_entry, ufp_instances[0]
raise HomeAssistantError(f"No device found for device id: {device_id}")
@callback
def _async_get_protect_from_call(
hass: HomeAssistant, call: ServiceCall
) -> list[tuple[dr.DeviceEntry, ProtectApiClient]]:
referenced = async_extract_referenced_entity_ids(hass, call)
instances: list[tuple[dr.DeviceEntry, ProtectApiClient]] = []
for device_id in referenced.referenced_devices:
instances.append(_async_get_ufp_instances(hass, device_id))
return instances
) -> set[ProtectApiClient]:
return {
_async_get_ufp_instance(hass, device_id)
for device_id in async_extract_referenced_entity_ids(
hass, call
).referenced_devices
}
async def _async_call_nvr(
instances: list[tuple[dr.DeviceEntry, ProtectApiClient]],
async def _async_service_call_nvr(
hass: HomeAssistant,
call: ServiceCall,
method: str,
*args: Any,
**kwargs: Any,
) -> None:
instances = _async_get_protect_from_call(hass, call)
try:
await asyncio.gather(
*(getattr(i.bootstrap.nvr, method)(*args, **kwargs) for _, i in instances)
*(getattr(i.bootstrap.nvr, method)(*args, **kwargs) for i in instances)
)
except (BadRequest, ValidationError) as err:
raise HomeAssistantError(str(err)) from err
@ -113,22 +118,61 @@ async def _async_call_nvr(
async def add_doorbell_text(hass: HomeAssistant, call: ServiceCall) -> None:
"""Add a custom doorbell text message."""
message: str = call.data[ATTR_MESSAGE]
instances = _async_get_protect_from_call(hass, call)
await _async_call_nvr(instances, "add_custom_doorbell_message", message)
await _async_service_call_nvr(hass, call, "add_custom_doorbell_message", message)
async def remove_doorbell_text(hass: HomeAssistant, call: ServiceCall) -> None:
"""Remove a custom doorbell text message."""
message: str = call.data[ATTR_MESSAGE]
instances = _async_get_protect_from_call(hass, call)
await _async_call_nvr(instances, "remove_custom_doorbell_message", message)
await _async_service_call_nvr(hass, call, "remove_custom_doorbell_message", message)
async def set_default_doorbell_text(hass: HomeAssistant, call: ServiceCall) -> None:
"""Set the default doorbell text message."""
message: str = call.data[ATTR_MESSAGE]
instances = _async_get_protect_from_call(hass, call)
await _async_call_nvr(instances, "set_default_doorbell_message", message)
await _async_service_call_nvr(hass, call, "set_default_doorbell_message", message)
@callback
def _async_unique_id_to_ufp_device_id(unique_id: str) -> str:
"""Extract the UFP device id from the registry entry unique id."""
return unique_id.split("_")[0]
async def set_chime_paired_doorbells(hass: HomeAssistant, call: ServiceCall) -> None:
"""Set paired doorbells on chime."""
ref = async_extract_referenced_entity_ids(hass, call)
entity_registry = er.async_get(hass)
entity_id = ref.indirectly_referenced.pop()
chime_button = entity_registry.async_get(entity_id)
assert chime_button is not None
assert chime_button.device_id is not None
chime_ufp_device_id = _async_unique_id_to_ufp_device_id(chime_button.unique_id)
instance = _async_get_ufp_instance(hass, chime_button.device_id)
chime = instance.bootstrap.chimes[chime_ufp_device_id]
call.data = ReadOnlyDict(call.data.get("doorbells") or {})
doorbell_refs = async_extract_referenced_entity_ids(hass, call)
doorbell_ids: set[str] = set()
for camera_id in doorbell_refs.referenced | doorbell_refs.indirectly_referenced:
doorbell_sensor = entity_registry.async_get(camera_id)
assert doorbell_sensor is not None
if (
doorbell_sensor.platform != DOMAIN
or doorbell_sensor.domain != Platform.BINARY_SENSOR
or doorbell_sensor.original_device_class
!= BinarySensorDeviceClass.OCCUPANCY
):
continue
doorbell_ufp_device_id = _async_unique_id_to_ufp_device_id(
doorbell_sensor.unique_id
)
camera = instance.bootstrap.cameras[doorbell_ufp_device_id]
doorbell_ids.add(camera.id)
chime.camera_ids = sorted(doorbell_ids)
await chime.save_device()
def async_setup_services(hass: HomeAssistant) -> None:
@ -149,6 +193,11 @@ def async_setup_services(hass: HomeAssistant) -> None:
functools.partial(set_default_doorbell_text, hass),
DOORBELL_TEXT_SCHEMA,
),
(
SERVICE_SET_CHIME_PAIRED,
functools.partial(set_chime_paired_doorbells, hass),
CHIME_PAIRED_SCHEMA,
),
]
for name, method, schema in services:
if hass.services.has_service(DOMAIN, name):

25
homeassistant/components/unifiprotect/services.yaml
View File

@ -84,3 +84,28 @@ set_doorbell_message:
step: 1
mode: slider
unit_of_measurement: minutes
set_chime_paired_doorbells:
name: Set Chime Paired Doorbells
description: >
Use to set the paired doorbell(s) with a smart chime.
fields:
device_id:
name: Chime
description: The Chimes to link to the doorbells to
required: true
selector:
device:
integration: unifiprotect
entity:
device_class: unifiprotect__chime_button
doorbells:
name: Doorbells
description: The Doorbells to link to the chime
example: "binary_sensor.front_doorbell_doorbell"
required: false
selector:
target:
entity:
integration: unifiprotect
domain: binary_sensor
device_class: occupancy

12
tests/components/unifiprotect/conftest.py
View File

@ -14,6 +14,7 @@ import pytest
from pyunifiprotect.data import (
NVR,
Camera,
Chime,
Doorlock,
Light,
Liveview,
@ -49,6 +50,7 @@ class MockBootstrap:
liveviews: dict[str, Any]
events: dict[str, Any]
doorlocks: dict[str, Any]
chimes: dict[str, Any]
def reset_objects(self) -> None:
"""Reset all devices on bootstrap for tests."""
@ -59,6 +61,7 @@ class MockBootstrap:
self.liveviews = {}
self.events = {}
self.doorlocks = {}
self.chimes = {}
def process_ws_packet(self, msg: WSSubscriptionMessage) -> None:
"""Fake process method for tests."""
@ -127,6 +130,7 @@ def mock_bootstrap_fixture(mock_nvr: NVR):
liveviews={},
events={},
doorlocks={},
chimes={},
)
@ -220,6 +224,14 @@ def mock_doorlock():
return Doorlock.from_unifi_dict(**data)
@pytest.fixture
def mock_chime():
"""Mock UniFi Protect Chime device."""
data = json.loads(load_fixture("sample_chime.json", integration=DOMAIN))
return Chime.from_unifi_dict(**data)
@pytest.fixture
def now():
"""Return datetime object that will be consistent throughout test."""

48
tests/components/unifiprotect/fixtures/sample_chime.json Normal file
View File

@ -0,0 +1,48 @@
{
"mac": "BEEEE2FBE413",
"host": "192.168.144.146",
"connectionHost": "192.168.234.27",
"type": "UP Chime",
"name": "Xaorvu Tvsv",
"upSince": 1651882870009,
"uptime": 567870,
"lastSeen": 1652450740009,
"connectedSince": 1652448904587,
"state": "CONNECTED",
"hardwareRevision": null,
"firmwareVersion": "1.3.4",
"latestFirmwareVersion": "1.3.4",
"firmwareBuild": "58bd350.220401.1859",
"isUpdating": false,
"isAdopting": false,
"isAdopted": true,
"isAdoptedByOther": false,
"isProvisioned": false,
"isRebooting": false,
"isSshEnabled": true,
"canAdopt": false,
"isAttemptingToConnect": false,
"volume": 100,
"isProbingForWifi": false,
"apMac": null,
"apRssi": null,
"elementInfo": null,
"lastRing": 1652116059940,
"isWirelessUplinkEnabled": true,
"wiredConnectionState": {
"phyRate": null
},
"wifiConnectionState": {
"channel": null,
"frequency": null,
"phyRate": null,
"signalQuality": 100,
"signalStrength": -44,
"ssid": null
},
"cameraIds": [],
"id": "cf1a330397c08f919d02bd7c",
"isConnected": true,
"marketName": "UP Chime",
"modelKey": "chime"
}

61
tests/components/unifiprotect/test_button.py
View File

@ -5,7 +5,7 @@ from __future__ import annotations
from unittest.mock import AsyncMock
import pytest
from pyunifiprotect.data import Camera
from pyunifiprotect.data.devices import Chime
from homeassistant.components.unifiprotect.const import DEFAULT_ATTRIBUTION
from homeassistant.const import ATTR_ATTRIBUTION, ATTR_ENTITY_ID, Platform
@ -15,42 +15,39 @@ from homeassistant.helpers import entity_registry as er
from .conftest import MockEntityFixture, assert_entity_counts, enable_entity
@pytest.fixture(name="camera")
async def camera_fixture(
hass: HomeAssistant, mock_entry: MockEntityFixture, mock_camera: Camera
@pytest.fixture(name="chime")
async def chime_fixture(
hass: HomeAssistant, mock_entry: MockEntityFixture, mock_chime: Chime
):
"""Fixture for a single camera for testing the button platform."""
camera_obj = mock_camera.copy(deep=True)
camera_obj._api = mock_entry.api
camera_obj.channels[0]._api = mock_entry.api
camera_obj.channels[1]._api = mock_entry.api
camera_obj.channels[2]._api = mock_entry.api
camera_obj.name = "Test Camera"
chime_obj = mock_chime.copy(deep=True)
chime_obj._api = mock_entry.api
chime_obj.name = "Test Chime"
mock_entry.api.bootstrap.cameras = {
camera_obj.id: camera_obj,
mock_entry.api.bootstrap.chimes = {
chime_obj.id: chime_obj,
}
await hass.config_entries.async_setup(mock_entry.entry.entry_id)
await hass.async_block_till_done()
assert_entity_counts(hass, Platform.BUTTON, 1, 0)
assert_entity_counts(hass, Platform.BUTTON, 3, 2)
return (camera_obj, "button.test_camera_reboot_device")
return chime_obj
async def test_button(
async def test_reboot_button(
hass: HomeAssistant,
mock_entry: MockEntityFixture,
camera: tuple[Camera, str],
chime: Chime,
):
"""Test button entity."""
mock_entry.api.reboot_device = AsyncMock()
unique_id = f"{camera[0].id}_reboot"
entity_id = camera[1]
unique_id = f"{chime.id}_reboot"
entity_id = "button.test_chime_reboot_device"
entity_registry = er.async_get(hass)
entity = entity_registry.async_get(entity_id)
@ -67,3 +64,31 @@ async def test_button(
"button", "press", {ATTR_ENTITY_ID: entity_id}, blocking=True
)
mock_entry.api.reboot_device.assert_called_once()
async def test_chime_button(
hass: HomeAssistant,
mock_entry: MockEntityFixture,
chime: Chime,
):
"""Test button entity."""
mock_entry.api.play_speaker = AsyncMock()
unique_id = f"{chime.id}_play"
entity_id = "button.test_chime_play_chime"
entity_registry = er.async_get(hass)
entity = entity_registry.async_get(entity_id)
assert entity
assert not entity.disabled
assert entity.unique_id == unique_id
state = hass.states.get(entity_id)
assert state
assert state.attributes[ATTR_ATTRIBUTION] == DEFAULT_ATTRIBUTION
await hass.services.async_call(
"button", "press", {ATTR_ENTITY_ID: entity_id}, blocking=True
)
mock_entry.api.play_speaker.assert_called_once()

57
tests/components/unifiprotect/test_init.py
View File

@ -202,11 +202,11 @@ async def test_migrate_reboot_button(
registry = er.async_get(hass)
registry.async_get_or_create(
Platform.BUTTON, Platform.BUTTON, light1.id, config_entry=mock_entry.entry
Platform.BUTTON, DOMAIN, light1.id, config_entry=mock_entry.entry
)
registry.async_get_or_create(
Platform.BUTTON,
Platform.BUTTON,
DOMAIN,
f"{light2.id}_reboot",
config_entry=mock_entry.entry,
)
@ -218,24 +218,67 @@ async def test_migrate_reboot_button(
assert mock_entry.api.update.called
assert mock_entry.entry.unique_id == mock_entry.api.bootstrap.nvr.mac
buttons = []
for entity in er.async_entries_for_config_entry(
registry, mock_entry.entry.entry_id
):
if entity.domain == Platform.BUTTON.value:
buttons.append(entity)
print(entity.entity_id)
assert len(buttons) == 2
assert registry.async_get(f"{Platform.BUTTON}.test_light_1_reboot_device") is None
assert registry.async_get(f"{Platform.BUTTON}.test_light_1_reboot_device_2") is None
light = registry.async_get(f"{Platform.BUTTON}.test_light_1_reboot_device")
light = registry.async_get(f"{Platform.BUTTON}.unifiprotect_lightid1")
assert light is not None
assert light.unique_id == f"{light1.id}_reboot"
assert registry.async_get(f"{Platform.BUTTON}.test_light_2_reboot_device") is None
assert registry.async_get(f"{Platform.BUTTON}.test_light_2_reboot_device_2") is None
light = registry.async_get(f"{Platform.BUTTON}.test_light_2_reboot_device")
light = registry.async_get(f"{Platform.BUTTON}.unifiprotect_lightid2_reboot")
assert light is not None
assert light.unique_id == f"{light2.id}_reboot"
async def test_migrate_reboot_button_no_device(
hass: HomeAssistant, mock_entry: MockEntityFixture, mock_light: Light
):
"""Test migrating unique ID of reboot button if UniFi Protect device ID changed."""
light1 = mock_light.copy()
light1._api = mock_entry.api
light1.name = "Test Light 1"
light1.id = "lightid1"
mock_entry.api.bootstrap.lights = {
light1.id: light1,
}
mock_entry.api.get_bootstrap = AsyncMock(return_value=mock_entry.api.bootstrap)
registry = er.async_get(hass)
registry.async_get_or_create(
Platform.BUTTON, DOMAIN, "lightid2", config_entry=mock_entry.entry
)
await hass.config_entries.async_setup(mock_entry.entry.entry_id)
await hass.async_block_till_done()
assert mock_entry.entry.state == ConfigEntryState.LOADED
assert mock_entry.api.update.called
assert mock_entry.entry.unique_id == mock_entry.api.bootstrap.nvr.mac
buttons = []
for entity in er.async_entries_for_config_entry(
registry, mock_entry.entry.entry_id
):
if entity.platform == Platform.BUTTON.value:
if entity.domain == Platform.BUTTON.value:
buttons.append(entity)
assert len(buttons) == 2
light = registry.async_get(f"{Platform.BUTTON}.unifiprotect_lightid2")
assert light is not None
assert light.unique_id == "lightid2"
async def test_migrate_reboot_button_fail(
hass: HomeAssistant, mock_entry: MockEntityFixture, mock_light: Light
@ -255,14 +298,14 @@ async def test_migrate_reboot_button_fail(
registry = er.async_get(hass)
registry.async_get_or_create(
Platform.BUTTON,
Platform.BUTTON,
DOMAIN,
light1.id,
config_entry=mock_entry.entry,
suggested_object_id=light1.name,
)
registry.async_get_or_create(
Platform.BUTTON,
Platform.BUTTON,
DOMAIN,
f"{light1.id}_reboot",
config_entry=mock_entry.entry,
suggested_object_id=light1.name,

75
tests/components/unifiprotect/test_services.py
View File

@ -5,19 +5,21 @@ from __future__ import annotations
from unittest.mock import AsyncMock, Mock
import pytest
from pyunifiprotect.data import Light
from pyunifiprotect.data import Camera, Light, ModelType
from pyunifiprotect.data.devices import Chime
from pyunifiprotect.exceptions import BadRequest
from homeassistant.components.unifiprotect.const import ATTR_MESSAGE, DOMAIN
from homeassistant.components.unifiprotect.services import (
SERVICE_ADD_DOORBELL_TEXT,
SERVICE_REMOVE_DOORBELL_TEXT,
SERVICE_SET_CHIME_PAIRED,
SERVICE_SET_DEFAULT_DOORBELL_TEXT,
)
from homeassistant.const import ATTR_DEVICE_ID
from homeassistant.const import ATTR_DEVICE_ID, ATTR_ENTITY_ID
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers import device_registry as dr, entity_registry as er
from .conftest import MockEntityFixture
@ -143,3 +145,70 @@ async def test_set_default_doorbell_text(
blocking=True,
)
nvr.set_default_doorbell_message.assert_called_once_with("Test Message")
async def test_set_chime_paired_doorbells(
hass: HomeAssistant,
mock_entry: MockEntityFixture,
mock_chime: Chime,
mock_camera: Camera,
):
"""Test set_chime_paired_doorbells."""
mock_entry.api.update_device = AsyncMock()
mock_chime._api = mock_entry.api
mock_chime.name = "Test Chime"
mock_chime._initial_data = mock_chime.dict()
mock_entry.api.bootstrap.chimes = {
mock_chime.id: mock_chime,
}
camera1 = mock_camera.copy()
camera1.id = "cameraid1"
camera1.name = "Test Camera 1"
camera1._api = mock_entry.api
camera1.channels[0]._api = mock_entry.api
camera1.channels[1]._api = mock_entry.api
camera1.channels[2]._api = mock_entry.api
camera1.feature_flags.has_chime = True
camera2 = mock_camera.copy()
camera2.id = "cameraid2"
camera2.name = "Test Camera 2"
camera2._api = mock_entry.api
camera2.channels[0]._api = mock_entry.api
camera2.channels[1]._api = mock_entry.api
camera2.channels[2]._api = mock_entry.api
camera2.feature_flags.has_chime = True
mock_entry.api.bootstrap.cameras = {
camera1.id: camera1,
camera2.id: camera2,
}
await hass.config_entries.async_setup(mock_entry.entry.entry_id)
await hass.async_block_till_done()
registry = er.async_get(hass)
chime_entry = registry.async_get("button.test_chime_play_chime")
camera_entry = registry.async_get("binary_sensor.test_camera_2_doorbell")
assert chime_entry is not None
assert camera_entry is not None
await hass.services.async_call(
DOMAIN,
SERVICE_SET_CHIME_PAIRED,
{
ATTR_DEVICE_ID: chime_entry.device_id,
"doorbells": {
ATTR_ENTITY_ID: ["binary_sensor.test_camera_1_doorbell"],
ATTR_DEVICE_ID: [camera_entry.device_id],
},
},
blocking=True,
)
mock_entry.api.update_device.assert_called_once_with(
ModelType.CHIME, mock_chime.id, {"cameraIds": [camera1.id, camera2.id]}
)