Add UniFi Protect media_player (#62895)
This commit is contained in:
Christopher Bailey
GitHub
parent
4025ce8f97
commit
490d76e01e
8 changed files with 470 additions and 83 deletions
|
|
@ -1,9 +1,11 @@
|
|||
"""Test the UniFi Protect camera platform."""
|
||||
# pylint: disable=protected-access
|
||||
from __future__ import annotations
|
||||
|
||||
from copy import copy
|
||||
from unittest.mock import AsyncMock, Mock
|
||||
from unittest.mock import AsyncMock, Mock, patch
|
||||
|
||||
import pytest
|
||||
from pyunifiprotect.data import Camera as ProtectCamera
|
||||
from pyunifiprotect.data.devices import CameraChannel
|
||||
from pyunifiprotect.exceptions import NvrError
|
||||
|
|
@ -27,6 +29,7 @@ from homeassistant.const import (
|
|||
ATTR_ATTRIBUTION,
|
||||
ATTR_ENTITY_ID,
|
||||
ATTR_SUPPORTED_FEATURES,
|
||||
Platform,
|
||||
)
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers import entity_registry as er
|
||||
|
|
@ -35,17 +38,50 @@ from homeassistant.setup import async_setup_component
|
|||
from .conftest import MockEntityFixture, enable_entity, time_changed
|
||||
|
||||
|
||||
@pytest.fixture(name="camera")
|
||||
async def camera_fixture(
|
||||
hass: HomeAssistant, mock_entry: MockEntityFixture, mock_camera: Camera
|
||||
):
|
||||
"""Fixture for a single camera, no extra setup."""
|
||||
|
||||
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.channels[0].is_rtsp_enabled = True
|
||||
camera_obj.channels[0].name = "High"
|
||||
camera_obj.channels[1].is_rtsp_enabled = False
|
||||
camera_obj.channels[2].is_rtsp_enabled = False
|
||||
|
||||
mock_entry.api.bootstrap.cameras = {
|
||||
camera_obj.id: camera_obj,
|
||||
}
|
||||
|
||||
with patch("homeassistant.components.unifiprotect.PLATFORMS", [Platform.CAMERA]):
|
||||
await hass.config_entries.async_setup(mock_entry.entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
entity_registry = er.async_get(hass)
|
||||
|
||||
assert len(hass.states.async_all()) == 1
|
||||
assert len(entity_registry.entities) == 2
|
||||
|
||||
yield (camera_obj, "camera.test_camera_high")
|
||||
|
||||
|
||||
def validate_default_camera_entity(
|
||||
hass: HomeAssistant,
|
||||
camera: ProtectCamera,
|
||||
camera_obj: ProtectCamera,
|
||||
channel_id: int,
|
||||
) -> str:
|
||||
"""Validate a camera entity."""
|
||||
|
||||
channel = camera.channels[channel_id]
|
||||
channel = camera_obj.channels[channel_id]
|
||||
|
||||
entity_name = f"{camera.name} {channel.name}"
|
||||
unique_id = f"{camera.id}_{channel.id}"
|
||||
entity_name = f"{camera_obj.name} {channel.name}"
|
||||
unique_id = f"{camera_obj.id}_{channel.id}"
|
||||
entity_id = f"camera.{entity_name.replace(' ', '_').lower()}"
|
||||
|
||||
entity_registry = er.async_get(hass)
|
||||
|
|
@ -59,15 +95,15 @@ def validate_default_camera_entity(
|
|||
|
||||
def validate_rtsps_camera_entity(
|
||||
hass: HomeAssistant,
|
||||
camera: ProtectCamera,
|
||||
camera_obj: ProtectCamera,
|
||||
channel_id: int,
|
||||
) -> str:
|
||||
"""Validate a disabled RTSPS camera entity."""
|
||||
|
||||
channel = camera.channels[channel_id]
|
||||
channel = camera_obj.channels[channel_id]
|
||||
|
||||
entity_name = f"{camera.name} {channel.name}"
|
||||
unique_id = f"{camera.id}_{channel.id}"
|
||||
entity_name = f"{camera_obj.name} {channel.name}"
|
||||
unique_id = f"{camera_obj.id}_{channel.id}"
|
||||
entity_id = f"camera.{entity_name.replace(' ', '_').lower()}"
|
||||
|
||||
entity_registry = er.async_get(hass)
|
||||
|
|
@ -81,15 +117,15 @@ def validate_rtsps_camera_entity(
|
|||
|
||||
def validate_rtsp_camera_entity(
|
||||
hass: HomeAssistant,
|
||||
camera: ProtectCamera,
|
||||
camera_obj: ProtectCamera,
|
||||
channel_id: int,
|
||||
) -> str:
|
||||
"""Validate a disabled RTSP camera entity."""
|
||||
|
||||
channel = camera.channels[channel_id]
|
||||
channel = camera_obj.channels[channel_id]
|
||||
|
||||
entity_name = f"{camera.name} {channel.name} Insecure"
|
||||
unique_id = f"{camera.id}_{channel.id}_insecure"
|
||||
entity_name = f"{camera_obj.name} {channel.name} Insecure"
|
||||
unique_id = f"{camera_obj.id}_{channel.id}_insecure"
|
||||
entity_id = f"camera.{entity_name.replace(' ', '_').lower()}"
|
||||
|
||||
entity_registry = er.async_get(hass)
|
||||
|
|
@ -121,13 +157,13 @@ def validate_common_camera_state(
|
|||
|
||||
async def validate_rtsps_camera_state(
|
||||
hass: HomeAssistant,
|
||||
camera: ProtectCamera,
|
||||
camera_obj: ProtectCamera,
|
||||
channel_id: int,
|
||||
entity_id: str,
|
||||
features: int = SUPPORT_STREAM,
|
||||
):
|
||||
"""Validate a camera's state."""
|
||||
channel = camera.channels[channel_id]
|
||||
channel = camera_obj.channels[channel_id]
|
||||
|
||||
assert await async_get_stream_source(hass, entity_id) == channel.rtsps_url
|
||||
validate_common_camera_state(hass, channel, entity_id, features)
|
||||
|
|
@ -135,13 +171,13 @@ async def validate_rtsps_camera_state(
|
|||
|
||||
async def validate_rtsp_camera_state(
|
||||
hass: HomeAssistant,
|
||||
camera: ProtectCamera,
|
||||
camera_obj: ProtectCamera,
|
||||
channel_id: int,
|
||||
entity_id: str,
|
||||
features: int = SUPPORT_STREAM,
|
||||
):
|
||||
"""Validate a camera's state."""
|
||||
channel = camera.channels[channel_id]
|
||||
channel = camera_obj.channels[channel_id]
|
||||
|
||||
assert await async_get_stream_source(hass, entity_id) == channel.rtsp_url
|
||||
validate_common_camera_state(hass, channel, entity_id, features)
|
||||
|
|
@ -149,13 +185,13 @@ async def validate_rtsp_camera_state(
|
|||
|
||||
async def validate_no_stream_camera_state(
|
||||
hass: HomeAssistant,
|
||||
camera: ProtectCamera,
|
||||
camera_obj: ProtectCamera,
|
||||
channel_id: int,
|
||||
entity_id: str,
|
||||
features: int = SUPPORT_STREAM,
|
||||
):
|
||||
"""Validate a camera's state."""
|
||||
channel = camera.channels[channel_id]
|
||||
channel = camera_obj.channels[channel_id]
|
||||
|
||||
assert await async_get_stream_source(hass, entity_id) is None
|
||||
validate_common_camera_state(hass, channel, entity_id, features)
|
||||
|
|
@ -228,8 +264,9 @@ async def test_basic_setup(
|
|||
camera_no_channels.id: camera_no_channels,
|
||||
}
|
||||
|
||||
await hass.config_entries.async_setup(mock_entry.entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
with patch("homeassistant.components.unifiprotect.PLATFORMS", [Platform.CAMERA]):
|
||||
await hass.config_entries.async_setup(mock_entry.entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
entity_registry = er.async_get(hass)
|
||||
|
||||
|
|
@ -305,52 +342,52 @@ async def test_missing_channels(
|
|||
async def test_camera_image(
|
||||
hass: HomeAssistant,
|
||||
mock_entry: MockEntityFixture,
|
||||
simple_camera: tuple[Camera, str],
|
||||
camera: tuple[Camera, str],
|
||||
):
|
||||
"""Test retrieving camera image."""
|
||||
|
||||
mock_entry.api.get_camera_snapshot = AsyncMock()
|
||||
|
||||
await async_get_image(hass, simple_camera[1])
|
||||
await async_get_image(hass, camera[1])
|
||||
mock_entry.api.get_camera_snapshot.assert_called_once()
|
||||
|
||||
|
||||
async def test_camera_generic_update(
|
||||
hass: HomeAssistant,
|
||||
mock_entry: MockEntityFixture,
|
||||
simple_camera: tuple[ProtectCamera, str],
|
||||
camera: tuple[ProtectCamera, str],
|
||||
):
|
||||
"""Tests generic entity update service."""
|
||||
|
||||
assert await async_setup_component(hass, "homeassistant", {})
|
||||
|
||||
state = hass.states.get(simple_camera[1])
|
||||
state = hass.states.get(camera[1])
|
||||
assert state and state.state == "idle"
|
||||
|
||||
mock_entry.api.update = AsyncMock(return_value=None)
|
||||
await hass.services.async_call(
|
||||
"homeassistant",
|
||||
"update_entity",
|
||||
{ATTR_ENTITY_ID: simple_camera[1]},
|
||||
{ATTR_ENTITY_ID: camera[1]},
|
||||
blocking=True,
|
||||
)
|
||||
|
||||
state = hass.states.get(simple_camera[1])
|
||||
state = hass.states.get(camera[1])
|
||||
assert state and state.state == "idle"
|
||||
|
||||
|
||||
async def test_camera_interval_update(
|
||||
hass: HomeAssistant,
|
||||
mock_entry: MockEntityFixture,
|
||||
simple_camera: tuple[ProtectCamera, str],
|
||||
camera: tuple[ProtectCamera, str],
|
||||
):
|
||||
"""Interval updates updates camera entity."""
|
||||
|
||||
state = hass.states.get(simple_camera[1])
|
||||
state = hass.states.get(camera[1])
|
||||
assert state and state.state == "idle"
|
||||
|
||||
new_bootstrap = copy(mock_entry.api.bootstrap)
|
||||
new_camera = simple_camera[0].copy()
|
||||
new_camera = camera[0].copy()
|
||||
new_camera.is_recording = True
|
||||
|
||||
new_bootstrap.cameras = {new_camera.id: new_camera}
|
||||
|
|
@ -358,47 +395,47 @@ async def test_camera_interval_update(
|
|||
mock_entry.api.bootstrap = new_bootstrap
|
||||
await time_changed(hass, DEFAULT_SCAN_INTERVAL)
|
||||
|
||||
state = hass.states.get(simple_camera[1])
|
||||
state = hass.states.get(camera[1])
|
||||
assert state and state.state == "recording"
|
||||
|
||||
|
||||
async def test_camera_bad_interval_update(
|
||||
hass: HomeAssistant,
|
||||
mock_entry: MockEntityFixture,
|
||||
simple_camera: tuple[Camera, str],
|
||||
camera: tuple[Camera, str],
|
||||
):
|
||||
"""Interval updates marks camera unavailable."""
|
||||
|
||||
state = hass.states.get(simple_camera[1])
|
||||
state = hass.states.get(camera[1])
|
||||
assert state and state.state == "idle"
|
||||
|
||||
# update fails
|
||||
mock_entry.api.update = AsyncMock(side_effect=NvrError)
|
||||
await time_changed(hass, DEFAULT_SCAN_INTERVAL)
|
||||
|
||||
state = hass.states.get(simple_camera[1])
|
||||
state = hass.states.get(camera[1])
|
||||
assert state and state.state == "unavailable"
|
||||
|
||||
# next update succeeds
|
||||
mock_entry.api.update = AsyncMock(return_value=mock_entry.api.bootstrap)
|
||||
await time_changed(hass, DEFAULT_SCAN_INTERVAL)
|
||||
|
||||
state = hass.states.get(simple_camera[1])
|
||||
state = hass.states.get(camera[1])
|
||||
assert state and state.state == "idle"
|
||||
|
||||
|
||||
async def test_camera_ws_update(
|
||||
hass: HomeAssistant,
|
||||
mock_entry: MockEntityFixture,
|
||||
simple_camera: tuple[ProtectCamera, str],
|
||||
camera: tuple[ProtectCamera, str],
|
||||
):
|
||||
"""WS update updates camera entity."""
|
||||
|
||||
state = hass.states.get(simple_camera[1])
|
||||
state = hass.states.get(camera[1])
|
||||
assert state and state.state == "idle"
|
||||
|
||||
new_bootstrap = copy(mock_entry.api.bootstrap)
|
||||
new_camera = simple_camera[0].copy()
|
||||
new_camera = camera[0].copy()
|
||||
new_camera.is_recording = True
|
||||
|
||||
mock_msg = Mock()
|
||||
|
|
@ -409,23 +446,23 @@ async def test_camera_ws_update(
|
|||
mock_entry.api.ws_subscription(mock_msg)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
state = hass.states.get(simple_camera[1])
|
||||
state = hass.states.get(camera[1])
|
||||
assert state and state.state == "recording"
|
||||
|
||||
|
||||
async def test_camera_ws_update_offline(
|
||||
hass: HomeAssistant,
|
||||
mock_entry: MockEntityFixture,
|
||||
simple_camera: tuple[ProtectCamera, str],
|
||||
camera: tuple[ProtectCamera, str],
|
||||
):
|
||||
"""WS updates marks camera unavailable."""
|
||||
|
||||
state = hass.states.get(simple_camera[1])
|
||||
state = hass.states.get(camera[1])
|
||||
assert state and state.state == "idle"
|
||||
|
||||
# camera goes offline
|
||||
new_bootstrap = copy(mock_entry.api.bootstrap)
|
||||
new_camera = simple_camera[0].copy()
|
||||
new_camera = camera[0].copy()
|
||||
new_camera.is_connected = False
|
||||
|
||||
mock_msg = Mock()
|
||||
|
|
@ -436,7 +473,7 @@ async def test_camera_ws_update_offline(
|
|||
mock_entry.api.ws_subscription(mock_msg)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
state = hass.states.get(simple_camera[1])
|
||||
state = hass.states.get(camera[1])
|
||||
assert state and state.state == "unavailable"
|
||||
|
||||
# camera comes back online
|
||||
|
|
@ -450,5 +487,5 @@ async def test_camera_ws_update_offline(
|
|||
mock_entry.api.ws_subscription(mock_msg)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
state = hass.states.get(simple_camera[1])
|
||||
state = hass.states.get(camera[1])
|
||||
assert state and state.state == "idle"
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue