Add UniFi Protect switch platform (#63177)

This commit is contained in:
Christopher Bailey 2022-01-01 16:23:10 -05:00 committed by GitHub
parent 817f0c9aae
commit e5b7eac411
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
11 changed files with 869 additions and 6 deletions

View file

@ -41,4 +41,10 @@ DEVICES_FOR_SUBSCRIBE = DEVICES_WITH_ENTITIES | {ModelType.EVENT}
MIN_REQUIRED_PROTECT_V = Version("1.20.0")
OUTDATED_LOG_MESSAGE = "You are running v%s of UniFi Protect. Minimum required version is v%s. Please upgrade UniFi Protect and then retry"
PLATFORMS = [Platform.BUTTON, Platform.CAMERA, Platform.LIGHT, Platform.MEDIA_PLAYER]
PLATFORMS = [
Platform.BUTTON,
Platform.CAMERA,
Platform.LIGHT,
Platform.MEDIA_PLAYER,
Platform.SWITCH,
]

View file

@ -1,7 +1,18 @@
"""Shared Entity definition for UniFi Protect Integration."""
from __future__ import annotations
from pyunifiprotect.data import ProtectAdoptableDeviceModel, StateType
from collections.abc import Sequence
import logging
from pyunifiprotect.data import (
Camera,
Light,
ModelType,
ProtectAdoptableDeviceModel,
Sensor,
StateType,
Viewer,
)
from homeassistant.core import callback
import homeassistant.helpers.device_registry as dr
@ -9,6 +20,72 @@ from homeassistant.helpers.entity import DeviceInfo, Entity, EntityDescription
from .const import DEFAULT_ATTRIBUTION, DEFAULT_BRAND, DOMAIN
from .data import ProtectData
from .models import ProtectRequiredKeysMixin
from .utils import get_nested_attr
_LOGGER = logging.getLogger(__name__)
@callback
def _async_device_entities(
data: ProtectData,
klass: type[ProtectDeviceEntity],
model_type: ModelType,
descs: Sequence[ProtectRequiredKeysMixin],
) -> list[ProtectDeviceEntity]:
if len(descs) == 0:
return []
entities: list[ProtectDeviceEntity] = []
for device in data.get_by_types({model_type}):
assert isinstance(device, (Camera, Light, Sensor, Viewer))
for description in descs:
assert isinstance(description, EntityDescription)
if description.ufp_required_field:
required_field = get_nested_attr(device, description.ufp_required_field)
if not required_field:
continue
entities.append(
klass(
data,
device=device,
description=description,
)
)
_LOGGER.debug(
"Adding %s entity %s for %s",
klass.__name__,
description.name,
device.name,
)
return entities
@callback
def async_all_device_entities(
data: ProtectData,
klass: type[ProtectDeviceEntity],
camera_descs: Sequence[ProtectRequiredKeysMixin] | None = None,
light_descs: Sequence[ProtectRequiredKeysMixin] | None = None,
sense_descs: Sequence[ProtectRequiredKeysMixin] | None = None,
viewport_descs: Sequence[ProtectRequiredKeysMixin] | None = None,
all_descs: Sequence[ProtectRequiredKeysMixin] | None = None,
) -> list[ProtectDeviceEntity]:
"""Generate a list of all the device entities."""
all_descs = list(all_descs or [])
camera_descs = list(camera_descs or []) + all_descs
light_descs = list(light_descs or []) + all_descs
sense_descs = list(sense_descs or []) + all_descs
viewport_descs = list(viewport_descs or []) + all_descs
return (
_async_device_entities(data, klass, ModelType.CAMERA, camera_descs)
+ _async_device_entities(data, klass, ModelType.LIGHT, light_descs)
+ _async_device_entities(data, klass, ModelType.SENSOR, sense_descs)
+ _async_device_entities(data, klass, ModelType.VIEWPORT, viewport_descs)
)
class ProtectDeviceEntity(Entity):

View file

@ -0,0 +1,12 @@
"""The unifiprotect integration models."""
from __future__ import annotations
from dataclasses import dataclass
@dataclass
class ProtectRequiredKeysMixin:
"""Mixin for required keys."""
ufp_required_field: str | None = None
ufp_value: str | None = None

View file

@ -0,0 +1,253 @@
"""This component provides Switches for UniFi Protect."""
from __future__ import annotations
from dataclasses import dataclass
import logging
from typing import Any
from pyunifiprotect.data import Camera, RecordingMode, VideoMode
from pyunifiprotect.data.base import ProtectAdoptableDeviceModel
from homeassistant.components.switch import SwitchEntity, SwitchEntityDescription
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity import EntityCategory
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .const import DOMAIN
from .data import ProtectData
from .entity import ProtectDeviceEntity, async_all_device_entities
from .models import ProtectRequiredKeysMixin
from .utils import get_nested_attr
_LOGGER = logging.getLogger(__name__)
@dataclass
class ProtectSwitchEntityDescription(ProtectRequiredKeysMixin, SwitchEntityDescription):
"""Describes UniFi Protect Switch entity."""
ufp_set_function: str | None = None
_KEY_STATUS_LIGHT = "status_light"
_KEY_HDR_MODE = "hdr_mode"
_KEY_HIGH_FPS = "high_fps"
_KEY_PRIVACY_MODE = "privacy_mode"
_KEY_SYSTEM_SOUNDS = "system_sounds"
_KEY_OSD_NAME = "osd_name"
_KEY_OSD_DATE = "osd_date"
_KEY_OSD_LOGO = "osd_logo"
_KEY_OSD_BITRATE = "osd_bitrate"
_KEY_SMART_PERSON = "smart_person"
_KEY_SMART_VEHICLE = "smart_vehicle"
_KEY_SSH = "ssh"
ALL_DEVICES_SWITCHES: tuple[ProtectSwitchEntityDescription, ...] = (
ProtectSwitchEntityDescription(
key=_KEY_SSH,
name="SSH Enabled",
icon="mdi:lock",
entity_registry_enabled_default=False,
entity_category=EntityCategory.CONFIG,
ufp_value="is_ssh_enabled",
ufp_set_function="set_ssh",
),
)
CAMERA_SWITCHES: tuple[ProtectSwitchEntityDescription, ...] = (
ProtectSwitchEntityDescription(
key=_KEY_STATUS_LIGHT,
name="Status Light On",
icon="mdi:led-on",
entity_category=EntityCategory.CONFIG,
ufp_required_field="feature_flags.has_led_status",
ufp_value="led_settings.is_enabled",
ufp_set_function="set_status_light",
),
ProtectSwitchEntityDescription(
key=_KEY_HDR_MODE,
name="HDR Mode",
icon="mdi:brightness-7",
entity_category=EntityCategory.CONFIG,
ufp_required_field="feature_flags.has_hdr",
ufp_value="hdr_mode",
ufp_set_function="set_hdr",
),
ProtectSwitchEntityDescription(
key=_KEY_HIGH_FPS,
name="High FPS",
icon="mdi:video-high-definition",
entity_category=EntityCategory.CONFIG,
ufp_required_field="feature_flags.has_highfps",
ufp_value="video_mode",
),
ProtectSwitchEntityDescription(
key=_KEY_PRIVACY_MODE,
name="Privacy Mode",
icon="mdi:eye-settings",
entity_category=None,
ufp_required_field="feature_flags.has_privacy_mask",
ufp_value="is_privacy_on",
),
ProtectSwitchEntityDescription(
key=_KEY_SYSTEM_SOUNDS,
name="System Sounds",
icon="mdi:speaker",
entity_category=EntityCategory.CONFIG,
ufp_required_field="feature_flags.has_speaker",
ufp_value="speaker_settings.are_system_sounds_enabled",
ufp_set_function="set_system_sounds",
),
ProtectSwitchEntityDescription(
key=_KEY_OSD_NAME,
name="Overlay: Show Name",
icon="mdi:fullscreen",
entity_category=EntityCategory.CONFIG,
ufp_value="osd_settings.is_name_enabled",
ufp_set_function="set_osd_name",
),
ProtectSwitchEntityDescription(
key=_KEY_OSD_DATE,
name="Overlay: Show Date",
icon="mdi:fullscreen",
entity_category=EntityCategory.CONFIG,
ufp_value="osd_settings.is_date_enabled",
ufp_set_function="set_osd_date",
),
ProtectSwitchEntityDescription(
key=_KEY_OSD_LOGO,
name="Overlay: Show Logo",
icon="mdi:fullscreen",
entity_category=EntityCategory.CONFIG,
ufp_value="osd_settings.is_logo_enabled",
ufp_set_function="set_osd_logo",
),
ProtectSwitchEntityDescription(
key=_KEY_OSD_BITRATE,
name="Overlay: Show Bitrate",
icon="mdi:fullscreen",
entity_category=EntityCategory.CONFIG,
ufp_value="osd_settings.is_debug_enabled",
ufp_set_function="set_osd_bitrate",
),
ProtectSwitchEntityDescription(
key=_KEY_SMART_PERSON,
name="Detections: Person",
icon="mdi:walk",
entity_category=EntityCategory.CONFIG,
ufp_required_field="feature_flags.has_smart_detect",
ufp_value="is_person_detection_on",
ufp_set_function="set_person_detection",
),
ProtectSwitchEntityDescription(
key=_KEY_SMART_VEHICLE,
name="Detections: Vehicle",
icon="mdi:car",
entity_category=EntityCategory.CONFIG,
ufp_required_field="feature_flags.has_smart_detect",
ufp_value="is_vehicle_detection_on",
ufp_set_function="set_vehicle_detection",
),
)
LIGHT_SWITCHES: tuple[ProtectSwitchEntityDescription, ...] = (
ProtectSwitchEntityDescription(
key=_KEY_STATUS_LIGHT,
name="Status Light On",
icon="mdi:led-on",
entity_category=EntityCategory.CONFIG,
ufp_value="light_device_settings.is_indicator_enabled",
ufp_set_function="set_status_light",
),
)
async def async_setup_entry(
hass: HomeAssistant,
entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up sensors for UniFi Protect integration."""
data: ProtectData = hass.data[DOMAIN][entry.entry_id]
entities: list[ProtectDeviceEntity] = async_all_device_entities(
data,
ProtectSwitch,
all_descs=ALL_DEVICES_SWITCHES,
camera_descs=CAMERA_SWITCHES,
light_descs=LIGHT_SWITCHES,
)
async_add_entities(entities)
class ProtectSwitch(ProtectDeviceEntity, SwitchEntity):
"""A UniFi Protect Switch."""
def __init__(
self,
data: ProtectData,
device: ProtectAdoptableDeviceModel,
description: ProtectSwitchEntityDescription,
) -> None:
"""Initialize an UniFi Protect Switch."""
self.entity_description: ProtectSwitchEntityDescription = description
super().__init__(data, device)
self._attr_name = f"{self.device.name} {self.entity_description.name}"
self._switch_type = self.entity_description.key
if not isinstance(self.device, Camera):
return
if self.entity_description.key == _KEY_PRIVACY_MODE:
if self.device.is_privacy_on:
self._previous_mic_level = 100
self._previous_record_mode = RecordingMode.ALWAYS
else:
self._previous_mic_level = self.device.mic_volume
self._previous_record_mode = self.device.recording_settings.mode
@property
def is_on(self) -> bool:
"""Return true if device is on."""
assert self.entity_description.ufp_value is not None
ufp_value = get_nested_attr(self.device, self.entity_description.ufp_value)
if self._switch_type == _KEY_HIGH_FPS:
return bool(ufp_value == VideoMode.HIGH_FPS)
return ufp_value is True
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn the device on."""
if self.entity_description.ufp_set_function is not None:
await getattr(self.device, self.entity_description.ufp_set_function)(True)
return
assert isinstance(self.device, Camera)
if self._switch_type == _KEY_HIGH_FPS:
_LOGGER.debug("Turning on High FPS mode")
await self.device.set_video_mode(VideoMode.HIGH_FPS)
return
if self._switch_type == _KEY_PRIVACY_MODE:
_LOGGER.debug("Turning Privacy Mode on for %s", self.device.name)
self._previous_mic_level = self.device.mic_volume
self._previous_record_mode = self.device.recording_settings.mode
await self.device.set_privacy(True, 0, RecordingMode.NEVER)
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn the device off."""
if self.entity_description.ufp_set_function is not None:
await getattr(self.device, self.entity_description.ufp_set_function)(False)
return
assert isinstance(self.device, Camera)
if self._switch_type == _KEY_HIGH_FPS:
_LOGGER.debug("Turning off High FPS mode")
await self.device.set_video_mode(VideoMode.DEFAULT)
elif self._switch_type == _KEY_PRIVACY_MODE:
_LOGGER.debug("Turning Privacy Mode off for %s", self.device.name)
await self.device.set_privacy(
False, self._previous_mic_level, self._previous_record_mode
)

View file

@ -0,0 +1,21 @@
"""UniFi Protect Integration utils."""
from __future__ import annotations
from enum import Enum
from typing import Any
def get_nested_attr(obj: Any, attr: str) -> Any:
"""Fetch a nested attribute."""
attrs = attr.split(".")
value = obj
for key in attrs:
if not hasattr(value, key):
return None
value = getattr(value, key)
if isinstance(value, Enum):
value = value.value
return value

View file

@ -11,11 +11,13 @@ from unittest.mock import AsyncMock, Mock, patch
import pytest
from pyunifiprotect.data import Camera, Light, Version, WSSubscriptionMessage
from pyunifiprotect.data.base import ProtectAdoptableDeviceModel
from homeassistant.components.unifiprotect.const import DOMAIN, MIN_REQUIRED_PROTECT_V
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant, split_entity_id
from homeassistant.helpers import entity_registry as er
from homeassistant.helpers.entity import EntityDescription
import homeassistant.util.dt as dt_util
from tests.common import MockConfigEntry, async_fire_time_changed
@ -176,3 +178,21 @@ def assert_entity_counts(
assert len(entities) == total
assert len(hass.states.async_all(platform.value)) == enabled
def ids_from_device_description(
platform: Platform,
device: ProtectAdoptableDeviceModel,
description: EntityDescription,
) -> tuple[str, str]:
"""Return expected unique_id and entity_id for a give platform/device/description combination."""
entity_name = device.name.lower().replace(":", "").replace(" ", "_")
description_entity_name = (
description.name.lower().replace(":", "").replace(" ", "_")
)
unique_id = f"{device.id}_{description.key}"
entity_id = f"{platform.value}.{entity_name}_{description_entity_name}"
return unique_id, entity_id

View file

@ -19,7 +19,7 @@ from .conftest import MockEntityFixture, assert_entity_counts, enable_entity
async def camera_fixture(
hass: HomeAssistant, mock_entry: MockEntityFixture, mock_camera: Camera
):
"""Fixture for a single camera with only the button platform active, no extra setup."""
"""Fixture for a single camera for testing the button platform."""
camera_obj = mock_camera.copy(deep=True)
camera_obj._api = mock_entry.api

View file

@ -48,7 +48,7 @@ from .conftest import (
async def camera_fixture(
hass: HomeAssistant, mock_entry: MockEntityFixture, mock_camera: Camera
):
"""Fixture for a single camera, no extra setup."""
"""Fixture for a single camera for testing the camera platform."""
camera_obj = mock_camera.copy(deep=True)
camera_obj._api = mock_entry.api

View file

@ -27,7 +27,7 @@ from .conftest import MockEntityFixture, assert_entity_counts
async def light_fixture(
hass: HomeAssistant, mock_entry: MockEntityFixture, mock_light: Light
):
"""Fixture for a single light with only the button platform active, no extra setup."""
"""Fixture for a single light for testing the light platform."""
# disable pydantic validation so mocking can happen
Light.__config__.validate_assignment = False

View file

@ -33,7 +33,7 @@ from .conftest import MockEntityFixture, assert_entity_counts
async def camera_fixture(
hass: HomeAssistant, mock_entry: MockEntityFixture, mock_camera: Camera
):
"""Fixture for a single camera with only the media_player platform active, camera has speaker."""
"""Fixture for a single camera for testing the media_player platform."""
# disable pydantic validation so mocking can happen
Camera.__config__.validate_assignment = False

View file

@ -0,0 +1,474 @@
"""Test the UniFi Protect light platform."""
# pylint: disable=protected-access
from __future__ import annotations
from unittest.mock import AsyncMock, Mock
import pytest
from pyunifiprotect.data import Camera, Light
from pyunifiprotect.data.types import RecordingMode, VideoMode
from homeassistant.components.unifiprotect.const import DEFAULT_ATTRIBUTION
from homeassistant.components.unifiprotect.switch import (
ALL_DEVICES_SWITCHES,
CAMERA_SWITCHES,
LIGHT_SWITCHES,
ProtectSwitchEntityDescription,
)
from homeassistant.const import ATTR_ATTRIBUTION, ATTR_ENTITY_ID, STATE_OFF, Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from .conftest import (
MockEntityFixture,
assert_entity_counts,
enable_entity,
ids_from_device_description,
)
@pytest.fixture(name="light")
async def light_fixture(
hass: HomeAssistant, mock_entry: MockEntityFixture, mock_light: Light
):
"""Fixture for a single light for testing the switch platform."""
# disable pydantic validation so mocking can happen
Light.__config__.validate_assignment = False
light_obj = mock_light.copy(deep=True)
light_obj._api = mock_entry.api
light_obj.name = "Test Light"
light_obj.is_ssh_enabled = False
light_obj.light_device_settings.is_indicator_enabled = False
mock_entry.api.bootstrap.cameras = {}
mock_entry.api.bootstrap.lights = {
light_obj.id: light_obj,
}
await hass.config_entries.async_setup(mock_entry.entry.entry_id)
await hass.async_block_till_done()
assert_entity_counts(hass, Platform.SWITCH, 2, 1)
yield light_obj
Light.__config__.validate_assignment = True
@pytest.fixture(name="camera")
async def camera_fixture(
hass: HomeAssistant, mock_entry: MockEntityFixture, mock_camera: Camera
):
"""Fixture for a single camera for testing the switch platform."""
# disable pydantic validation so mocking can happen
Camera.__config__.validate_assignment = False
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"
camera_obj.recording_settings.mode = RecordingMode.DETECTIONS
camera_obj.feature_flags.has_led_status = True
camera_obj.feature_flags.has_hdr = True
camera_obj.feature_flags.video_modes = [VideoMode.DEFAULT, VideoMode.HIGH_FPS]
camera_obj.feature_flags.has_privacy_mask = True
camera_obj.feature_flags.has_speaker = True
camera_obj.feature_flags.has_smart_detect = True
camera_obj.is_ssh_enabled = False
camera_obj.led_settings.is_enabled = False
camera_obj.hdr_mode = False
camera_obj.video_mode = VideoMode.DEFAULT
camera_obj.remove_privacy_zone()
camera_obj.speaker_settings.are_system_sounds_enabled = False
camera_obj.osd_settings.is_name_enabled = False
camera_obj.osd_settings.is_date_enabled = False
camera_obj.osd_settings.is_logo_enabled = False
camera_obj.osd_settings.is_debug_enabled = False
camera_obj.smart_detect_settings.object_types = []
mock_entry.api.bootstrap.lights = {}
mock_entry.api.bootstrap.cameras = {
camera_obj.id: camera_obj,
}
await hass.config_entries.async_setup(mock_entry.entry.entry_id)
await hass.async_block_till_done()
assert_entity_counts(hass, Platform.SWITCH, 12, 11)
yield camera_obj
Camera.__config__.validate_assignment = True
@pytest.fixture(name="camera_none")
async def camera_none_fixture(
hass: HomeAssistant, mock_entry: MockEntityFixture, mock_camera: Camera
):
"""Fixture for a single camera for testing the switch platform."""
# disable pydantic validation so mocking can happen
Camera.__config__.validate_assignment = False
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"
camera_obj.recording_settings.mode = RecordingMode.DETECTIONS
camera_obj.feature_flags.has_led_status = False
camera_obj.feature_flags.has_hdr = False
camera_obj.feature_flags.video_modes = [VideoMode.DEFAULT]
camera_obj.feature_flags.has_privacy_mask = False
camera_obj.feature_flags.has_speaker = False
camera_obj.feature_flags.has_smart_detect = False
camera_obj.is_ssh_enabled = False
camera_obj.osd_settings.is_name_enabled = False
camera_obj.osd_settings.is_date_enabled = False
camera_obj.osd_settings.is_logo_enabled = False
camera_obj.osd_settings.is_debug_enabled = False
mock_entry.api.bootstrap.lights = {}
mock_entry.api.bootstrap.cameras = {
camera_obj.id: camera_obj,
}
await hass.config_entries.async_setup(mock_entry.entry.entry_id)
await hass.async_block_till_done()
assert_entity_counts(hass, Platform.SWITCH, 5, 4)
yield camera_obj
Camera.__config__.validate_assignment = True
@pytest.fixture(name="camera_privacy")
async def camera_privacy_fixture(
hass: HomeAssistant, mock_entry: MockEntityFixture, mock_camera: Camera
):
"""Fixture for a single camera for testing the switch platform."""
# disable pydantic validation so mocking can happen
Camera.__config__.validate_assignment = False
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"
camera_obj.recording_settings.mode = RecordingMode.NEVER
camera_obj.feature_flags.has_led_status = False
camera_obj.feature_flags.has_hdr = False
camera_obj.feature_flags.video_modes = [VideoMode.DEFAULT]
camera_obj.feature_flags.has_privacy_mask = True
camera_obj.feature_flags.has_speaker = False
camera_obj.feature_flags.has_smart_detect = False
camera_obj.add_privacy_zone()
camera_obj.is_ssh_enabled = False
camera_obj.osd_settings.is_name_enabled = False
camera_obj.osd_settings.is_date_enabled = False
camera_obj.osd_settings.is_logo_enabled = False
camera_obj.osd_settings.is_debug_enabled = False
mock_entry.api.bootstrap.lights = {}
mock_entry.api.bootstrap.cameras = {
camera_obj.id: camera_obj,
}
await hass.config_entries.async_setup(mock_entry.entry.entry_id)
await hass.async_block_till_done()
assert_entity_counts(hass, Platform.SWITCH, 6, 5)
yield camera_obj
Camera.__config__.validate_assignment = True
async def test_switch_setup_light(
hass: HomeAssistant,
mock_entry: MockEntityFixture,
light: Light,
):
"""Test switch entity setup for light devices."""
entity_registry = er.async_get(hass)
description = LIGHT_SWITCHES[0]
unique_id, entity_id = ids_from_device_description(
Platform.SWITCH, light, description
)
entity = entity_registry.async_get(entity_id)
assert entity
assert entity.unique_id == unique_id
state = hass.states.get(entity_id)
assert state
assert state.state == STATE_OFF
assert state.attributes[ATTR_ATTRIBUTION] == DEFAULT_ATTRIBUTION
description = ALL_DEVICES_SWITCHES[0]
unique_id = f"{light.id}_{description.key}"
entity_id = f"switch.test_light_{description.name.lower().replace(' ', '_')}"
entity = entity_registry.async_get(entity_id)
assert entity
assert entity.disabled is True
assert entity.unique_id == unique_id
await enable_entity(hass, mock_entry.entry.entry_id, entity_id)
state = hass.states.get(entity_id)
assert state
assert state.state == STATE_OFF
assert state.attributes[ATTR_ATTRIBUTION] == DEFAULT_ATTRIBUTION
async def test_switch_setup_camera_all(
hass: HomeAssistant,
mock_entry: MockEntityFixture,
camera: Camera,
):
"""Test switch entity setup for camera devices (all enabled feature flags)."""
entity_registry = er.async_get(hass)
for description in CAMERA_SWITCHES:
unique_id, entity_id = ids_from_device_description(
Platform.SWITCH, camera, description
)
entity = entity_registry.async_get(entity_id)
assert entity
assert entity.unique_id == unique_id
state = hass.states.get(entity_id)
assert state
assert state.state == STATE_OFF
assert state.attributes[ATTR_ATTRIBUTION] == DEFAULT_ATTRIBUTION
description = ALL_DEVICES_SWITCHES[0]
description_entity_name = (
description.name.lower().replace(":", "").replace(" ", "_")
)
unique_id = f"{camera.id}_{description.key}"
entity_id = f"switch.test_camera_{description_entity_name}"
entity = entity_registry.async_get(entity_id)
assert entity
assert entity.disabled is True
assert entity.unique_id == unique_id
await enable_entity(hass, mock_entry.entry.entry_id, entity_id)
state = hass.states.get(entity_id)
assert state
assert state.state == STATE_OFF
assert state.attributes[ATTR_ATTRIBUTION] == DEFAULT_ATTRIBUTION
async def test_switch_setup_camera_none(
hass: HomeAssistant,
mock_entry: MockEntityFixture,
camera_none: Camera,
):
"""Test switch entity setup for camera devices (no enabled feature flags)."""
entity_registry = er.async_get(hass)
for description in CAMERA_SWITCHES:
if description.ufp_required_field is not None:
continue
unique_id, entity_id = ids_from_device_description(
Platform.SWITCH, camera_none, description
)
entity = entity_registry.async_get(entity_id)
assert entity
assert entity.unique_id == unique_id
state = hass.states.get(entity_id)
assert state
assert state.state == STATE_OFF
assert state.attributes[ATTR_ATTRIBUTION] == DEFAULT_ATTRIBUTION
description = ALL_DEVICES_SWITCHES[0]
description_entity_name = (
description.name.lower().replace(":", "").replace(" ", "_")
)
unique_id = f"{camera_none.id}_{description.key}"
entity_id = f"switch.test_camera_{description_entity_name}"
entity = entity_registry.async_get(entity_id)
assert entity
assert entity.disabled is True
assert entity.unique_id == unique_id
await enable_entity(hass, mock_entry.entry.entry_id, entity_id)
state = hass.states.get(entity_id)
assert state
assert state.state == STATE_OFF
assert state.attributes[ATTR_ATTRIBUTION] == DEFAULT_ATTRIBUTION
async def test_switch_light_status(hass: HomeAssistant, light: Light):
"""Tests status light switch for lights."""
description = LIGHT_SWITCHES[0]
light.__fields__["set_status_light"] = Mock()
light.set_status_light = AsyncMock()
_, entity_id = ids_from_device_description(Platform.SWITCH, light, description)
await hass.services.async_call(
"switch", "turn_on", {ATTR_ENTITY_ID: entity_id}, blocking=True
)
light.set_status_light.assert_called_once_with(True)
await hass.services.async_call(
"switch", "turn_off", {ATTR_ENTITY_ID: entity_id}, blocking=True
)
light.set_status_light.assert_called_with(False)
async def test_switch_camera_ssh(
hass: HomeAssistant, camera: Camera, mock_entry: MockEntityFixture
):
"""Tests SSH switch for cameras."""
description = ALL_DEVICES_SWITCHES[0]
camera.__fields__["set_ssh"] = Mock()
camera.set_ssh = AsyncMock()
_, entity_id = ids_from_device_description(Platform.SWITCH, camera, description)
await enable_entity(hass, mock_entry.entry.entry_id, entity_id)
await hass.services.async_call(
"switch", "turn_on", {ATTR_ENTITY_ID: entity_id}, blocking=True
)
camera.set_ssh.assert_called_once_with(True)
await hass.services.async_call(
"switch", "turn_off", {ATTR_ENTITY_ID: entity_id}, blocking=True
)
camera.set_ssh.assert_called_with(False)
@pytest.mark.parametrize("description", CAMERA_SWITCHES)
async def test_switch_camera_simple(
hass: HomeAssistant, camera: Camera, description: ProtectSwitchEntityDescription
):
"""Tests all simple switches for cameras."""
if description.name in ("High FPS", "Privacy Mode"):
return
assert description.ufp_set_function is not None
camera.__fields__[description.ufp_set_function] = Mock()
setattr(camera, description.ufp_set_function, AsyncMock())
set_method = getattr(camera, description.ufp_set_function)
_, entity_id = ids_from_device_description(Platform.SWITCH, camera, description)
await hass.services.async_call(
"switch", "turn_on", {ATTR_ENTITY_ID: entity_id}, blocking=True
)
set_method.assert_called_once_with(True)
await hass.services.async_call(
"switch", "turn_off", {ATTR_ENTITY_ID: entity_id}, blocking=True
)
set_method.assert_called_with(False)
async def test_switch_camera_highfps(hass: HomeAssistant, camera: Camera):
"""Tests High FPS switch for cameras."""
description = CAMERA_SWITCHES[2]
camera.__fields__["set_video_mode"] = Mock()
camera.set_video_mode = AsyncMock()
_, entity_id = ids_from_device_description(Platform.SWITCH, camera, description)
await hass.services.async_call(
"switch", "turn_on", {ATTR_ENTITY_ID: entity_id}, blocking=True
)
camera.set_video_mode.assert_called_once_with(VideoMode.HIGH_FPS)
await hass.services.async_call(
"switch", "turn_off", {ATTR_ENTITY_ID: entity_id}, blocking=True
)
camera.set_video_mode.assert_called_with(VideoMode.DEFAULT)
async def test_switch_camera_privacy(hass: HomeAssistant, camera: Camera):
"""Tests Privacy Mode switch for cameras."""
description = CAMERA_SWITCHES[3]
camera.__fields__["set_privacy"] = Mock()
camera.set_privacy = AsyncMock()
_, entity_id = ids_from_device_description(Platform.SWITCH, camera, description)
await hass.services.async_call(
"switch", "turn_on", {ATTR_ENTITY_ID: entity_id}, blocking=True
)
camera.set_privacy.assert_called_once_with(True, 0, RecordingMode.NEVER)
await hass.services.async_call(
"switch", "turn_off", {ATTR_ENTITY_ID: entity_id}, blocking=True
)
camera.set_privacy.assert_called_with(
False, camera.mic_volume, camera.recording_settings.mode
)
async def test_switch_camera_privacy_already_on(
hass: HomeAssistant, camera_privacy: Camera
):
"""Tests Privacy Mode switch for cameras with privacy mode defaulted on."""
description = CAMERA_SWITCHES[3]
camera_privacy.__fields__["set_privacy"] = Mock()
camera_privacy.set_privacy = AsyncMock()
_, entity_id = ids_from_device_description(
Platform.SWITCH, camera_privacy, description
)
await hass.services.async_call(
"switch", "turn_off", {ATTR_ENTITY_ID: entity_id}, blocking=True
)
camera_privacy.set_privacy.assert_called_once_with(False, 100, RecordingMode.ALWAYS)