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

Removes ThumbnailProxyView (#63940)

pull/63947/head
Christopher Bailey 2022-01-11 18:45:40 -05:00 committed by GitHub
parent 05ee5e0251
commit ce0b378e05
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
8 changed files with 3 additions and 419 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
homeassistant/components/unifiprotect/__init__.py
View File

@ -34,7 +34,6 @@ from .const import (
)
from .data import ProtectData
from .services import async_cleanup_services, async_setup_services
from .views import ThumbnailProxyView
_LOGGER = logging.getLogger(__name__)
@ -84,7 +83,6 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
hass.data.setdefault(DOMAIN, {})[entry.entry_id] = data_service
hass.config_entries.async_setup_platforms(entry, PLATFORMS)
async_setup_services(hass)
hass.http.register_view(ThumbnailProxyView(hass))
entry.async_on_unload(entry.add_update_listener(_async_options_updated))
entry.async_on_unload(

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

@ -7,7 +7,6 @@ from homeassistant.const import Platform
DOMAIN = "unifiprotect"
ATTR_EVENT_SCORE = "event_score"
ATTR_EVENT_THUMB = "event_thumbnail"
ATTR_WIDTH = "width"
ATTR_HEIGHT = "height"
ATTR_FPS = "fps"

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

@ -1,7 +1,6 @@
"""Base class for protect data."""
from __future__ import annotations
from collections import deque
from collections.abc import Generator, Iterable
from datetime import timedelta
import logging
@ -43,7 +42,6 @@ class ProtectData:
self._unsub_websocket: CALLBACK_TYPE | None = None
self.last_update_success = False
self.access_tokens: dict[str, deque] = {}
self.api = protect
@property
@ -177,10 +175,3 @@ class ProtectData:
_LOGGER.debug("Updating device: %s", device_id)
for update_callback in self._subscriptions[device_id]:
update_callback()
@callback
def async_get_or_create_access_tokens(self, entity_id: str) -> deque:
"""Wrap access_tokens to automatically create underlying data structure if missing."""
if entity_id not in self.access_tokens:
self.access_tokens[entity_id] = deque([], 2)
return self.access_tokens[entity_id]

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

@ -1,14 +1,9 @@
"""Shared Entity definition for UniFi Protect Integration."""
from __future__ import annotations
from collections import deque
from collections.abc import Sequence
from datetime import datetime, timedelta
import hashlib
import logging
from random import SystemRandom
from typing import Any, Final
from urllib.parse import urlencode
from typing import Any
from pyunifiprotect.data import (
Camera,
@ -26,21 +21,11 @@ from homeassistant.core import callback
import homeassistant.helpers.device_registry as dr
from homeassistant.helpers.entity import DeviceInfo, Entity, EntityDescription
from .const import (
ATTR_EVENT_SCORE,
ATTR_EVENT_THUMB,
DEFAULT_ATTRIBUTION,
DEFAULT_BRAND,
DOMAIN,
)
from .const import ATTR_EVENT_SCORE, DEFAULT_ATTRIBUTION, DEFAULT_BRAND, DOMAIN
from .data import ProtectData
from .models import ProtectRequiredKeysMixin
from .utils import get_nested_attr
from .views import ThumbnailProxyView
EVENT_UPDATE_TOKENS = "unifiprotect_update_tokens"
TOKEN_CHANGE_INTERVAL: Final = timedelta(minutes=1)
_RND: Final = SystemRandom()
_LOGGER = logging.getLogger(__name__)
@ -219,50 +204,7 @@ class ProtectNVREntity(ProtectDeviceEntity):
self._attr_available = self.data.last_update_success
class AccessTokenMixin(ProtectDeviceEntity):
"""Adds access_token attribute and provides access tokens for use for anonymous views."""
@property
def access_tokens(self) -> deque[str]:
"""Get valid access_tokens for current entity."""
return self.data.async_get_or_create_access_tokens(self.entity_id)
@callback
def _async_update_and_write_token(self, now: datetime) -> None:
_LOGGER.debug("Updating access tokens for %s", self.entity_id)
self.async_update_token()
self.async_write_ha_state()
@callback
def async_update_token(self) -> None:
"""Update the used token."""
self.access_tokens.append(
hashlib.sha256(_RND.getrandbits(256).to_bytes(32, "little")).hexdigest()
)
@callback
def async_cleanup_tokens(self) -> None:
"""Clean up any remaining tokens on removal."""
if self.entity_id in self.data.access_tokens:
del self.data.access_tokens[self.entity_id]
async def async_added_to_hass(self) -> None:
"""Run when entity about to be added to hass.
Injects callbacks to update access tokens automatically
"""
await super().async_added_to_hass()
self.async_update_token()
self.async_on_remove(
self.hass.helpers.event.async_track_time_interval(
self._async_update_and_write_token, TOKEN_CHANGE_INTERVAL
)
)
self.async_on_remove(self.async_cleanup_tokens)
class EventThumbnailMixin(AccessTokenMixin):
class EventThumbnailMixin(ProtectDeviceEntity):
"""Adds motion event attributes to sensor."""
def __init__(self, *args: Any, **kwarg: Any) -> None:
@ -283,21 +225,12 @@ class EventThumbnailMixin(AccessTokenMixin):
# Camera motion sensors with object detection
attrs: dict[str, Any] = {
ATTR_EVENT_SCORE: 0,
ATTR_EVENT_THUMB: None,
}
if self._event is None:
return attrs
attrs[ATTR_EVENT_SCORE] = self._event.score
if len(self.access_tokens) > 0:
params = urlencode(
{"entity_id": self.entity_id, "token": self.access_tokens[-1]}
)
attrs[ATTR_EVENT_THUMB] = (
ThumbnailProxyView.url.format(event_id=self._event.id) + f"?{params}"
)
return attrs
@callback

91
homeassistant/components/unifiprotect/views.py
View File

@ -1,91 +0,0 @@
"""UniFi Protect Integration views."""
from __future__ import annotations
import collections
from http import HTTPStatus
import logging
from typing import Any
from aiohttp import web
from pyunifiprotect.api import ProtectApiClient
from pyunifiprotect.exceptions import NvrError
from homeassistant.components.http import KEY_AUTHENTICATED, HomeAssistantView
from homeassistant.core import HomeAssistant
from .const import DOMAIN
from .data import ProtectData
_LOGGER = logging.getLogger(__name__)
def _404(message: Any) -> web.Response:
_LOGGER.error("Error on load thumbnail: %s", message)
return web.Response(status=HTTPStatus.NOT_FOUND)
class ThumbnailProxyView(HomeAssistantView):
"""View to proxy event thumbnails from UniFi Protect."""
requires_auth = False
url = "/api/ufp/thumbnail/{event_id}"
name = "api:ufp_thumbnail"
def __init__(self, hass: HomeAssistant) -> None:
"""Initialize a thumbnail proxy view."""
self.hass = hass
self.data = hass.data[DOMAIN]
def _get_access_tokens(
self, entity_id: str
) -> tuple[collections.deque, ProtectApiClient] | None:
entries: list[ProtectData] = list(self.data.values())
for entry in entries:
if entity_id in entry.access_tokens:
return entry.access_tokens[entity_id], entry.api
return None
async def get(self, request: web.Request, event_id: str) -> web.Response:
"""Start a get request."""
entity_id: str | None = request.query.get("entity_id")
width: int | str | None = request.query.get("w")
height: int | str | None = request.query.get("h")
token: str | None = request.query.get("token")
if width is not None:
try:
width = int(width)
except ValueError:
return _404("Invalid width param")
if height is not None:
try:
height = int(height)
except ValueError:
return _404("Invalid height param")
access_tokens: list[str] = []
if entity_id is not None:
items = self._get_access_tokens(entity_id)
if items is None:
return _404(f"Could not find entity with entity_id {entity_id}")
access_tokens = list(items[0])
instance = items[1]
authenticated = request[KEY_AUTHENTICATED] or token in access_tokens
if not authenticated:
raise web.HTTPUnauthorized()
try:
thumbnail = await instance.get_event_thumbnail(
event_id, width=width, height=height
)
except NvrError as err:
return _404(err)
if thumbnail is None:
return _404("Event thumbnail not found")
return web.Response(body=thumbnail, content_type="image/jpeg")

5
tests/components/unifiprotect/test_binary_sensor.py
View File

@ -17,7 +17,6 @@ from homeassistant.components.unifiprotect.binary_sensor import (
)
from homeassistant.components.unifiprotect.const import (
ATTR_EVENT_SCORE,
ATTR_EVENT_THUMB,
DEFAULT_ATTRIBUTION,
)
from homeassistant.const import (
@ -258,7 +257,6 @@ async def test_binary_sensor_setup_camera_all(
assert state.state == STATE_OFF
assert state.attributes[ATTR_ATTRIBUTION] == DEFAULT_ATTRIBUTION
assert state.attributes[ATTR_EVENT_SCORE] == 0
assert state.attributes[ATTR_EVENT_THUMB] is None
async def test_binary_sensor_setup_camera_none(
@ -350,6 +348,3 @@ async def test_binary_sensor_update_motion(
assert state.state == STATE_ON
assert state.attributes[ATTR_ATTRIBUTION] == DEFAULT_ATTRIBUTION
assert state.attributes[ATTR_EVENT_SCORE] == 100
assert state.attributes[ATTR_EVENT_THUMB].startswith(
f"/api/ufp/thumbnail/test_event_id?entity_id={entity_id}&token="
)

5
tests/components/unifiprotect/test_sensor.py
View File

@ -13,7 +13,6 @@ from pyunifiprotect.data.types import EventType, SmartDetectObjectType
from homeassistant.components.unifiprotect.const import (
ATTR_EVENT_SCORE,
ATTR_EVENT_THUMB,
DEFAULT_ATTRIBUTION,
)
from homeassistant.components.unifiprotect.sensor import (
@ -407,7 +406,6 @@ async def test_sensor_setup_camera(
assert state.state == DETECTED_OBJECT_NONE
assert state.attributes[ATTR_ATTRIBUTION] == DEFAULT_ATTRIBUTION
assert state.attributes[ATTR_EVENT_SCORE] == 0
assert state.attributes[ATTR_EVENT_THUMB] is None
async def test_sensor_update_motion(
@ -450,6 +448,3 @@ async def test_sensor_update_motion(
assert state.state == SmartDetectObjectType.PERSON.value
assert state.attributes[ATTR_ATTRIBUTION] == DEFAULT_ATTRIBUTION
assert state.attributes[ATTR_EVENT_SCORE] == 100
assert state.attributes[ATTR_EVENT_THUMB].startswith(
f"/api/ufp/thumbnail/test_event_id?entity_id={entity_id}&token="
)

236
tests/components/unifiprotect/test_views.py
View File

@ -1,236 +0,0 @@
"""Test UniFi Protect views."""
# pylint: disable=protected-access
from __future__ import annotations
from datetime import datetime, timedelta
from unittest.mock import AsyncMock
import pytest
from pyunifiprotect.data import Camera, Event, EventType
from pyunifiprotect.exceptions import NvrError
from homeassistant.components.unifiprotect.binary_sensor import MOTION_SENSORS
from homeassistant.components.unifiprotect.const import ATTR_EVENT_THUMB
from homeassistant.components.unifiprotect.entity import TOKEN_CHANGE_INTERVAL
from homeassistant.const import STATE_ON, Platform
from homeassistant.core import HomeAssistant
from .conftest import MockEntityFixture, ids_from_device_description, time_changed
@pytest.fixture(name="thumb_url")
async def thumb_url_fixture(
hass: HomeAssistant,
mock_entry: MockEntityFixture,
mock_camera: Camera,
now: datetime,
):
"""Fixture for a single camera for testing the binary_sensor 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.is_motion_detected = True
event = Event(
id="test_event_id",
type=EventType.MOTION,
start=now - timedelta(seconds=1),
end=None,
score=100,
smart_detect_types=[],
smart_detect_event_ids=[],
camera_id=camera_obj.id,
)
camera_obj.last_motion_event_id = event.id
mock_entry.api.bootstrap.reset_objects()
mock_entry.api.bootstrap.nvr.system_info.storage.devices = []
mock_entry.api.bootstrap.cameras = {
camera_obj.id: camera_obj,
}
mock_entry.api.bootstrap.events = {event.id: event}
await hass.config_entries.async_setup(mock_entry.entry.entry_id)
await hass.async_block_till_done()
_, entity_id = ids_from_device_description(
Platform.BINARY_SENSOR, camera_obj, MOTION_SENSORS[0]
)
# make sure access tokens are generated
await time_changed(hass, 1)
state = hass.states.get(entity_id)
assert state
assert state.state == STATE_ON
assert state.attributes[ATTR_EVENT_THUMB].startswith(
f"/api/ufp/thumbnail/test_event_id?entity_id={entity_id}&token="
)
yield state.attributes[ATTR_EVENT_THUMB]
Camera.__config__.validate_assignment = True
async def test_thumbnail_view_good(
thumb_url: str,
hass_client_no_auth,
mock_entry: MockEntityFixture,
):
"""Test good result from thumbnail view."""
mock_entry.api.get_event_thumbnail = AsyncMock()
client = await hass_client_no_auth()
response = await client.get(thumb_url)
assert response.status == 200
mock_entry.api.get_event_thumbnail.assert_called_once_with(
"test_event_id", width=None, height=None
)
async def test_thumbnail_view_good_args(
thumb_url: str,
hass_client_no_auth,
mock_entry: MockEntityFixture,
):
"""Test good result from thumbnail view."""
mock_entry.api.get_event_thumbnail = AsyncMock()
client = await hass_client_no_auth()
response = await client.get(thumb_url + "&w=200&h=200")
assert response.status == 200
mock_entry.api.get_event_thumbnail.assert_called_once_with(
"test_event_id", width=200, height=200
)
async def test_thumbnail_view_bad_width(
thumb_url: str,
hass_client_no_auth,
mock_entry: MockEntityFixture,
):
"""Test good result from thumbnail view."""
mock_entry.api.get_event_thumbnail = AsyncMock()
client = await hass_client_no_auth()
response = await client.get(thumb_url + "&w=safds&h=200")
assert response.status == 404
assert not mock_entry.api.get_event_thumbnail.called
async def test_thumbnail_view_bad_height(
thumb_url: str,
hass_client_no_auth,
mock_entry: MockEntityFixture,
):
"""Test good result from thumbnail view."""
mock_entry.api.get_event_thumbnail = AsyncMock()
client = await hass_client_no_auth()
response = await client.get(thumb_url + "&w=200&h=asda")
assert response.status == 404
assert not mock_entry.api.get_event_thumbnail.called
async def test_thumbnail_view_bad_entity_id(
thumb_url: str,
hass_client_no_auth,
mock_entry: MockEntityFixture,
):
"""Test good result from thumbnail view."""
mock_entry.api.get_event_thumbnail = AsyncMock()
client = await hass_client_no_auth()
response = await client.get("/api/ufp/thumbnail/test_event_id?entity_id=sdfsfd")
assert response.status == 404
assert not mock_entry.api.get_event_thumbnail.called
async def test_thumbnail_view_bad_access_token(
thumb_url: str,
hass_client_no_auth,
mock_entry: MockEntityFixture,
):
"""Test good result from thumbnail view."""
mock_entry.api.get_event_thumbnail = AsyncMock()
client = await hass_client_no_auth()
thumb_url = thumb_url[:-1]
response = await client.get(thumb_url)
assert response.status == 401
assert not mock_entry.api.get_event_thumbnail.called
async def test_thumbnail_view_upstream_error(
thumb_url: str,
hass_client_no_auth,
mock_entry: MockEntityFixture,
):
"""Test good result from thumbnail view."""
mock_entry.api.get_event_thumbnail = AsyncMock(side_effect=NvrError)
client = await hass_client_no_auth()
response = await client.get(thumb_url)
assert response.status == 404
async def test_thumbnail_view_no_thumb(
thumb_url: str,
hass_client_no_auth,
mock_entry: MockEntityFixture,
):
"""Test good result from thumbnail view."""
mock_entry.api.get_event_thumbnail = AsyncMock(return_value=None)
client = await hass_client_no_auth()
response = await client.get(thumb_url)
assert response.status == 404
async def test_thumbnail_view_expired_access_token(
hass: HomeAssistant,
thumb_url: str,
hass_client_no_auth,
mock_entry: MockEntityFixture,
):
"""Test good result from thumbnail view."""
mock_entry.api.get_event_thumbnail = AsyncMock()
await time_changed(hass, TOKEN_CHANGE_INTERVAL.total_seconds())
await time_changed(hass, TOKEN_CHANGE_INTERVAL.total_seconds())
client = await hass_client_no_auth()
response = await client.get(thumb_url)
assert response.status == 401