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

Clean up camera and demo camera (#34058)

Browse Source 
* Clean up demo camera

* Complete test_motion_detection

* Clean up image reading

* Clean up camera integration async methods

* Fix and clean camera integration tests

* Fix image processing patch
This commit is contained in:
Martin Hjelmare 2020-04-12 14:56:19 +02:00 committed by GitHub
parent 4f519a2fe1
commit 20aa089243
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 63 additions and 63 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
homeassistant/components/camera/__init__.py
View File

@ -373,7 +373,7 @@ class Camera(Entity):
async def async_camera_image(self):
"""Return bytes of camera image."""
return await self.hass.async_add_job(self.camera_image)
return await self.hass.async_add_executor_job(self.camera_image)
async def handle_async_still_stream(self, request, interval):
"""Generate an HTTP MJPEG stream from camera images."""
@ -409,7 +409,7 @@ class Camera(Entity):
async def async_turn_off(self):
"""Turn off camera."""
await self.hass.async_add_job(self.turn_off)
await self.hass.async_add_executor_job(self.turn_off)
def turn_on(self):
"""Turn off camera."""
@ -417,25 +417,23 @@ class Camera(Entity):
async def async_turn_on(self):
"""Turn off camera."""
await self.hass.async_add_job(self.turn_on)
await self.hass.async_add_executor_job(self.turn_on)
def enable_motion_detection(self):
"""Enable motion detection in the camera."""
raise NotImplementedError()
@callback
def async_enable_motion_detection(self):
async def async_enable_motion_detection(self):
"""Call the job and enable motion detection."""
return self.hass.async_add_job(self.enable_motion_detection)
await self.hass.async_add_executor_job(self.enable_motion_detection)
def disable_motion_detection(self):
"""Disable motion detection in camera."""
raise NotImplementedError()
@callback
def async_disable_motion_detection(self):
async def async_disable_motion_detection(self):
"""Call the job and disable motion detection."""
return self.hass.async_add_job(self.disable_motion_detection)
await self.hass.async_add_executor_job(self.disable_motion_detection)
@property
def state_attributes(self):

30
homeassistant/components/demo/camera.py
View File

@ -1,6 +1,6 @@
"""Demo camera platform that has a fake camera."""
import logging
import os
from pathlib import Path
from homeassistant.components.camera import SUPPORT_ON_OFF, Camera
@ -28,16 +28,12 @@ class DemoCamera(Camera):
self.is_streaming = True
self._images_index = 0
def camera_image(self):
async def async_camera_image(self):
"""Return a faked still image response."""
self._images_index = (self._images_index + 1) % 4
image_path = Path(__file__).parent / f"demo_{self._images_index}.jpg"
image_path = os.path.join(
os.path.dirname(__file__), f"demo_{self._images_index}.jpg"
)
_LOGGER.debug("Loading camera_image: %s", image_path)
with open(image_path, "rb") as file:
return file.read()
return await self.hass.async_add_executor_job(image_path.read_bytes)
@property
def name(self):
@ -48,7 +44,7 @@ class DemoCamera(Camera):
def should_poll(self):
"""Demo camera doesn't need poll.
Need explicitly call schedule_update_ha_state() after state changed.
Need explicitly call async_write_ha_state() after state changed.
"""
return False
@ -67,22 +63,22 @@ class DemoCamera(Camera):
"""Camera Motion Detection Status."""
return self._motion_status
def enable_motion_detection(self):
async def async_enable_motion_detection(self):
"""Enable the Motion detection in base station (Arm)."""
self._motion_status = True
self.schedule_update_ha_state()
self.async_write_ha_state()
def disable_motion_detection(self):
async def async_disable_motion_detection(self):
"""Disable the motion detection in base station (Disarm)."""
self._motion_status = False
self.schedule_update_ha_state()
self.async_write_ha_state()
def turn_off(self):
async def async_turn_off(self):
"""Turn off camera."""
self.is_streaming = False
self.schedule_update_ha_state()
self.async_write_ha_state()
def turn_on(self):
async def async_turn_on(self):
"""Turn on camera."""
self.is_streaming = True
self.schedule_update_ha_state()
self.async_write_ha_state()

40
tests/components/camera/test_init.py
View File

@ -2,8 +2,9 @@
import asyncio
import base64
import io
from unittest.mock import PropertyMock, mock_open, patch
from unittest.mock import PropertyMock, mock_open
from asynctest import patch
import pytest
from homeassistant.components import camera
@ -14,40 +15,38 @@ from homeassistant.const import ATTR_ENTITY_ID, EVENT_HOMEASSISTANT_START
from homeassistant.exceptions import HomeAssistantError
from homeassistant.setup import async_setup_component
from tests.common import mock_coro
from tests.components.camera import common
@pytest.fixture
def mock_camera(hass):
@pytest.fixture(name="mock_camera")
def mock_camera_fixture(hass):
"""Initialize a demo camera platform."""
assert hass.loop.run_until_complete(
async_setup_component(hass, "camera", {camera.DOMAIN: {"platform": "demo"}})
)
with patch(
"homeassistant.components.demo.camera.DemoCamera.camera_image",
return_value=b"Test",
"homeassistant.components.demo.camera.Path.read_bytes", return_value=b"Test",
):
yield
@pytest.fixture
def mock_stream(hass):
@pytest.fixture(name="mock_stream")
def mock_stream_fixture(hass):
"""Initialize a demo camera platform with streaming."""
assert hass.loop.run_until_complete(
async_setup_component(hass, "stream", {"stream": {}})
)
@pytest.fixture
def setup_camera_prefs(hass):
@pytest.fixture(name="setup_camera_prefs")
def setup_camera_prefs_fixture(hass):
"""Initialize HTTP API."""
return common.mock_camera_prefs(hass, "camera.demo_camera")
@pytest.fixture
async def image_mock_url(hass):
@pytest.fixture(name="image_mock_url")
async def image_mock_url_fixture(hass):
"""Fixture for get_image tests."""
await async_setup_component(
hass, camera.DOMAIN, {camera.DOMAIN: {"platform": "demo"}}
@ -58,7 +57,7 @@ async def test_get_image_from_camera(hass, image_mock_url):
"""Grab an image from camera entity."""
with patch(
"homeassistant.components.demo.camera.DemoCamera.camera_image",
"homeassistant.components.demo.camera.Path.read_bytes",
autospec=True,
return_value=b"Test",
) as mock_camera:
@ -80,7 +79,7 @@ async def test_get_image_without_exists_camera(hass, image_mock_url):
async def test_get_image_with_timeout(hass, image_mock_url):
"""Try to get image with timeout."""
with patch(
"homeassistant.components.camera.Camera.async_camera_image",
"homeassistant.components.demo.camera.DemoCamera.async_camera_image",
side_effect=asyncio.TimeoutError,
), pytest.raises(HomeAssistantError):
await camera.async_get_image(hass, "camera.demo_camera")
@ -89,8 +88,8 @@ async def test_get_image_with_timeout(hass, image_mock_url):
async def test_get_image_fails(hass, image_mock_url):
"""Try to get image with timeout."""
with patch(
"homeassistant.components.camera.Camera.async_camera_image",
return_value=mock_coro(None),
"homeassistant.components.demo.camera.DemoCamera.async_camera_image",
return_value=None,
), pytest.raises(HomeAssistantError):
await camera.async_get_image(hass, "camera.demo_camera")
@ -169,7 +168,7 @@ async def test_websocket_camera_stream(hass, hass_ws_client, mock_camera, mock_s
return_value="http://home.assistant/playlist.m3u8",
) as mock_request_stream, patch(
"homeassistant.components.demo.camera.DemoCamera.stream_source",
return_value=mock_coro("http://example.com"),
return_value="http://example.com",
):
# Request playlist through WebSocket
client = await hass_ws_client(hass)
@ -248,7 +247,7 @@ async def test_handle_play_stream_service(hass, mock_camera, mock_stream):
"homeassistant.components.camera.request_stream"
) as mock_request_stream, patch(
"homeassistant.components.demo.camera.DemoCamera.stream_source",
return_value=mock_coro("http://example.com"),
return_value="http://example.com",
):
# Call service
await hass.services.async_call(
@ -294,7 +293,7 @@ async def test_preload_stream(hass, mock_stream):
return_value=demo_prefs,
), patch(
"homeassistant.components.demo.camera.DemoCamera.stream_source",
return_value=mock_coro("http://example.com"),
return_value="http://example.com",
):
await async_setup_component(hass, "camera", {DOMAIN: {"platform": "demo"}})
hass.bus.async_fire(EVENT_HOMEASSISTANT_START)
@ -323,10 +322,9 @@ async def test_record_service(hass, mock_camera, mock_stream):
"""Test record service."""
with patch(
"homeassistant.components.demo.camera.DemoCamera.stream_source",
return_value=mock_coro("http://example.com"),
return_value="http://example.com",
), patch(
"homeassistant.components.stream.async_handle_record_service",
return_value=mock_coro(),
) as mock_record_service, patch.object(
hass.config, "is_allowed_path", return_value=True
):

28
tests/components/demo/test_camera.py
View File

@ -1,10 +1,11 @@
"""The tests for local file camera component."""
from unittest.mock import mock_open, patch
from unittest.mock import patch
import pytest
from homeassistant.components.camera import (
DOMAIN as CAMERA_DOMAIN,
SERVICE_DISABLE_MOTION,
SERVICE_ENABLE_MOTION,
SERVICE_TURN_OFF,
SERVICE_TURN_ON,
@ -35,16 +36,11 @@ async def test_init_state_is_streaming(hass):
state = hass.states.get(ENTITY_CAMERA)
assert state.state == STATE_STREAMING
mock_on_img = mock_open(read_data=b"ON")
with patch("homeassistant.components.demo.camera.open", mock_on_img, create=True):
with patch(
"homeassistant.components.demo.camera.Path.read_bytes", return_value=b"ON"
) as mock_read_bytes:
image = await async_get_image(hass, ENTITY_CAMERA)
assert mock_on_img.called
assert mock_on_img.call_args_list[0][0][0][-6:] in [
"_0.jpg",
"_1.jpg",
"_2.jpg",
"_3.jpg",
]
assert mock_read_bytes.call_count == 1
assert image.content == b"ON"
@ -113,3 +109,15 @@ async def test_motion_detection(hass):
# Check if state has been updated.
state = hass.states.get(ENTITY_CAMERA)
assert state.attributes.get("motion_detection")
# Call service to turn off motion detection
await hass.services.async_call(
CAMERA_DOMAIN,
SERVICE_DISABLE_MOTION,
{ATTR_ENTITY_ID: ENTITY_CAMERA},
blocking=True,
)
# Check if state has been updated.
state = hass.states.get(ENTITY_CAMERA)
assert not state.attributes.get("motion_detection")

12
tests/components/image_processing/test_init.py
View File

@ -1,5 +1,7 @@
"""The tests for the image_processing component."""
from unittest.mock import PropertyMock, patch
from unittest.mock import PropertyMock
from asynctest import patch
import homeassistant.components.http as http
import homeassistant.components.image_processing as ip
@ -69,18 +71,16 @@ class TestImageProcessing:
self.hass.stop()
@patch(
"homeassistant.components.demo.camera.DemoCamera.camera_image",
autospec=True,
return_value=b"Test",
"homeassistant.components.demo.camera.Path.read_bytes", return_value=b"Test",
)
def test_get_image_from_camera(self, mock_camera):
def test_get_image_from_camera(self, mock_camera_read):
"""Grab an image from camera entity."""
common.scan(self.hass, entity_id="image_processing.test")
self.hass.block_till_done()
state = self.hass.states.get("image_processing.test")
assert mock_camera.called
assert mock_camera_read.called
assert state.state == "1"
assert state.attributes["image"] == b"Test"