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

Various camera test improvements (#34022)

pull/34054/head
Franck Nijhof 2020-04-11 21:10:15 +02:00 committed by GitHub
parent ea9f1376a6
commit 123ae941a9
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 90 additions and 101 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

58
tests/components/camera/common.py
View File

@ -3,63 +3,13 @@
All containing methods are legacy helpers that should not be used by new
components. Instead call the service directly.
"""
from homeassistant.components.camera import (
ATTR_FILENAME,
SERVICE_ENABLE_MOTION,
SERVICE_SNAPSHOT,
)
from homeassistant.components.camera.const import (
DATA_CAMERA_PREFS,
DOMAIN,
PREF_PRELOAD_STREAM,
)
from homeassistant.const import (
ATTR_ENTITY_ID,
ENTITY_MATCH_ALL,
SERVICE_TURN_OFF,
SERVICE_TURN_ON,
)
from homeassistant.core import callback
from homeassistant.loader import bind_hass
from homeassistant.components.camera.const import DATA_CAMERA_PREFS, PREF_PRELOAD_STREAM
@bind_hass
async def async_turn_off(hass, entity_id=ENTITY_MATCH_ALL):
"""Turn off camera."""
data = {ATTR_ENTITY_ID: entity_id} if entity_id else {}
await hass.services.async_call(DOMAIN, SERVICE_TURN_OFF, data)
@bind_hass
async def async_turn_on(hass, entity_id=ENTITY_MATCH_ALL):
"""Turn on camera, and set operation mode."""
data = {}
if entity_id is not None:
data[ATTR_ENTITY_ID] = entity_id
await hass.services.async_call(DOMAIN, SERVICE_TURN_ON, data)
@bind_hass
def enable_motion_detection(hass, entity_id=ENTITY_MATCH_ALL):
"""Enable Motion Detection."""
data = {ATTR_ENTITY_ID: entity_id} if entity_id else None
hass.async_add_job(hass.services.async_call(DOMAIN, SERVICE_ENABLE_MOTION, data))
@bind_hass
@callback
def async_snapshot(hass, filename, entity_id=ENTITY_MATCH_ALL):
"""Make a snapshot from a camera."""
data = {ATTR_ENTITY_ID: entity_id} if entity_id else {}
data[ATTR_FILENAME] = filename
hass.async_add_job(hass.services.async_call(DOMAIN, SERVICE_SNAPSHOT, data))
def mock_camera_prefs(hass, entity_id, prefs={}):
def mock_camera_prefs(hass, entity_id, prefs=None):
"""Fixture for cloud component."""
prefs_to_set = {PREF_PRELOAD_STREAM: True}
prefs_to_set.update(prefs)
if prefs is not None:
prefs_to_set.update(prefs)
hass.data[DATA_CAMERA_PREFS]._prefs[entity_id] = prefs_to_set
return prefs_to_set

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

@ -102,8 +102,15 @@ async def test_snapshot_service(hass, mock_camera):
with patch(
"homeassistant.components.camera.open", mopen, create=True
), patch.object(hass.config, "is_allowed_path", return_value=True):
common.async_snapshot(hass, "/test/snapshot.jpg")
await hass.async_block_till_done()
await hass.services.async_call(
camera.DOMAIN,
camera.SERVICE_SNAPSHOT,
{
ATTR_ENTITY_ID: "camera.demo_camera",
camera.ATTR_FILENAME: "/test/snapshot.jpg",
},
blocking=True,
)
mock_write = mopen().write
@ -237,10 +244,6 @@ async def test_play_stream_service_no_source(hass, mock_camera, mock_stream):
async def test_handle_play_stream_service(hass, mock_camera, mock_stream):
"""Test camera play_stream service."""
await async_setup_component(hass, "media_player", {})
data = {
ATTR_ENTITY_ID: "camera.demo_camera",
camera.ATTR_MEDIA_PLAYER: "media_player.test",
}
with patch(
"homeassistant.components.camera.request_stream"
) as mock_request_stream, patch(
@ -249,7 +252,13 @@ async def test_handle_play_stream_service(hass, mock_camera, mock_stream):
):
# Call service
await hass.services.async_call(
camera.DOMAIN, camera.SERVICE_PLAY_STREAM, data, blocking=True
camera.DOMAIN,
camera.SERVICE_PLAY_STREAM,
{
ATTR_ENTITY_ID: "camera.demo_camera",
camera.ATTR_MEDIA_PLAYER: "media_player.test",
},
blocking=True,
)
# So long as we request the stream, the rest should be covered
# by the play_media service tests.
@ -295,23 +304,23 @@ async def test_preload_stream(hass, mock_stream):
async def test_record_service_invalid_path(hass, mock_camera):
"""Test record service with invalid path."""
data = {
ATTR_ENTITY_ID: "camera.demo_camera",
camera.CONF_FILENAME: "/my/invalid/path",
}
with patch.object(
hass.config, "is_allowed_path", return_value=False
), pytest.raises(HomeAssistantError):
# Call service
await hass.services.async_call(
camera.DOMAIN, camera.SERVICE_RECORD, data, blocking=True
camera.DOMAIN,
camera.SERVICE_RECORD,
{
ATTR_ENTITY_ID: "camera.demo_camera",
camera.CONF_FILENAME: "/my/invalid/path",
},
blocking=True,
)
async def test_record_service(hass, mock_camera, mock_stream):
"""Test record service."""
data = {ATTR_ENTITY_ID: "camera.demo_camera", camera.CONF_FILENAME: "/my/path"}
with patch(
"homeassistant.components.demo.camera.DemoCamera.stream_source",
return_value=mock_coro("http://example.com"),
@ -323,7 +332,10 @@ async def test_record_service(hass, mock_camera, mock_stream):
):
# Call service
await hass.services.async_call(
camera.DOMAIN, camera.SERVICE_RECORD, data, blocking=True
camera.DOMAIN,
camera.SERVICE_RECORD,
{ATTR_ENTITY_ID: "camera.demo_camera", camera.CONF_FILENAME: "/my/path"},
blocking=True,
)
# So long as we call stream.record, the rest should be covered
# by those tests.

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

@ -3,30 +3,41 @@ from unittest.mock import mock_open, patch
import pytest
from homeassistant.components import camera
from homeassistant.components.camera import STATE_IDLE, STATE_STREAMING
from homeassistant.components.camera import (
DOMAIN as CAMERA_DOMAIN,
SERVICE_ENABLE_MOTION,
SERVICE_TURN_OFF,
SERVICE_TURN_ON,
STATE_IDLE,
STATE_STREAMING,
async_get_image,
)
from homeassistant.components.demo import DOMAIN
from homeassistant.const import ATTR_ENTITY_ID
from homeassistant.exceptions import HomeAssistantError
from homeassistant.setup import async_setup_component
from tests.components.camera import common
ENTITY_CAMERA = "camera.demo_camera"
@pytest.fixture
@pytest.fixture(autouse=True)
def demo_camera(hass):
"""Initialize a demo camera platform."""
hass.loop.run_until_complete(
async_setup_component(hass, "camera", {camera.DOMAIN: {"platform": "demo"}})
async_setup_component(
hass, CAMERA_DOMAIN, {CAMERA_DOMAIN: {"platform": DOMAIN}}
)
)
return hass.data["camera"].get_entity("camera.demo_camera")
async def test_init_state_is_streaming(hass, demo_camera):
async def test_init_state_is_streaming(hass):
"""Demo camera initialize as streaming."""
assert demo_camera.state == STATE_STREAMING
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):
image = await camera.async_get_image(hass, demo_camera.entity_id)
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",
@ -37,52 +48,68 @@ async def test_init_state_is_streaming(hass, demo_camera):
assert image.content == b"ON"
async def test_turn_on_state_back_to_streaming(hass, demo_camera):
async def test_turn_on_state_back_to_streaming(hass):
"""After turn on state back to streaming."""
assert demo_camera.state == STATE_STREAMING
await common.async_turn_off(hass, demo_camera.entity_id)
await hass.async_block_till_done()
state = hass.states.get(ENTITY_CAMERA)
assert state.state == STATE_STREAMING
assert demo_camera.state == STATE_IDLE
await hass.services.async_call(
CAMERA_DOMAIN, SERVICE_TURN_OFF, {ATTR_ENTITY_ID: ENTITY_CAMERA}, blocking=True
)
await common.async_turn_on(hass, demo_camera.entity_id)
await hass.async_block_till_done()
state = hass.states.get(ENTITY_CAMERA)
assert state.state == STATE_IDLE
assert demo_camera.state == STATE_STREAMING
await hass.services.async_call(
CAMERA_DOMAIN, SERVICE_TURN_ON, {ATTR_ENTITY_ID: ENTITY_CAMERA}, blocking=True
)
state = hass.states.get(ENTITY_CAMERA)
assert state.state == STATE_STREAMING
async def test_turn_off_image(hass, demo_camera):
async def test_turn_off_image(hass):
"""After turn off, Demo camera raise error."""
await common.async_turn_off(hass, demo_camera.entity_id)
await hass.async_block_till_done()
await hass.services.async_call(
CAMERA_DOMAIN, SERVICE_TURN_OFF, {ATTR_ENTITY_ID: ENTITY_CAMERA}, blocking=True
)
with pytest.raises(HomeAssistantError) as error:
await camera.async_get_image(hass, demo_camera.entity_id)
await async_get_image(hass, ENTITY_CAMERA)
assert error.args[0] == "Camera is off"
async def test_turn_off_invalid_camera(hass, demo_camera):
async def test_turn_off_invalid_camera(hass):
"""Turn off non-exist camera should quietly fail."""
assert demo_camera.state == STATE_STREAMING
await common.async_turn_off(hass, "camera.invalid_camera")
await hass.async_block_till_done()
state = hass.states.get(ENTITY_CAMERA)
assert state.state == STATE_STREAMING
assert demo_camera.state == STATE_STREAMING
await hass.services.async_call(
CAMERA_DOMAIN,
SERVICE_TURN_OFF,
{ATTR_ENTITY_ID: "camera.invalid_camera"},
blocking=True,
)
state = hass.states.get(ENTITY_CAMERA)
assert state.state == STATE_STREAMING
async def test_motion_detection(hass):
"""Test motion detection services."""
# Setup platform
await async_setup_component(hass, "camera", {"camera": {"platform": "demo"}})
# Fetch state and check motion detection attribute
state = hass.states.get("camera.demo_camera")
state = hass.states.get(ENTITY_CAMERA)
assert not state.attributes.get("motion_detection")
# Call service to turn on motion detection
common.enable_motion_detection(hass, "camera.demo_camera")
await hass.async_block_till_done()
await hass.services.async_call(
CAMERA_DOMAIN,
SERVICE_ENABLE_MOTION,
{ATTR_ENTITY_ID: ENTITY_CAMERA},
blocking=True,
)
# Check if state has been updated.
state = hass.states.get("camera.demo_camera")
state = hass.states.get(ENTITY_CAMERA)
assert state.attributes.get("motion_detection")