core/tests/components/generic/test_camera.py

518 lines
16 KiB
Python
Raw Normal View History

"""The tests for generic camera component."""
import asyncio
from http import HTTPStatus
2021-01-01 21:31:56 +00:00
from unittest.mock import patch
2021-11-04 15:07:50 +00:00
import aiohttp
import httpx
2021-11-04 15:07:50 +00:00
import pytest
import respx
from homeassistant import config as hass_config
from homeassistant.components.generic import DOMAIN
from homeassistant.components.websocket_api.const import TYPE_RESULT
from homeassistant.const import SERVICE_RELOAD
from homeassistant.setup import async_setup_component
from tests.common import AsyncMock, Mock, get_fixture_path
2021-11-02 03:47:05 +00:00
@respx.mock
async def test_fetching_url(hass, hass_client, fakeimgbytes_png):
"""Test that it fetches the given url."""
respx.get("http://example.com").respond(stream=fakeimgbytes_png)
2019-07-31 19:25:30 +00:00
await async_setup_component(
2019-07-31 19:25:30 +00:00
hass,
"camera",
{
"camera": {
"name": "config_test",
"platform": "generic",
"still_image_url": "http://example.com",
"username": "user",
"password": "pass",
}
},
)
await hass.async_block_till_done()
client = await hass_client()
resp = await client.get("/api/camera_proxy/camera.config_test")
assert resp.status == HTTPStatus.OK
assert respx.calls.call_count == 1
body = await resp.read()
assert body == fakeimgbytes_png
resp = await client.get("/api/camera_proxy/camera.config_test")
assert respx.calls.call_count == 2
@respx.mock
async def test_fetching_without_verify_ssl(hass, hass_client, fakeimgbytes_png):
"""Test that it fetches the given url when ssl verify is off."""
respx.get("https://example.com").respond(stream=fakeimgbytes_png)
2019-07-31 19:25:30 +00:00
await async_setup_component(
2019-07-31 19:25:30 +00:00
hass,
"camera",
{
"camera": {
"name": "config_test",
"platform": "generic",
"still_image_url": "https://example.com",
"username": "user",
"password": "pass",
"verify_ssl": "false",
}
},
)
await hass.async_block_till_done()
client = await hass_client()
resp = await client.get("/api/camera_proxy/camera.config_test")
assert resp.status == HTTPStatus.OK
@respx.mock
async def test_fetching_url_with_verify_ssl(hass, hass_client, fakeimgbytes_png):
"""Test that it fetches the given url when ssl verify is explicitly on."""
respx.get("https://example.com").respond(stream=fakeimgbytes_png)
2019-07-31 19:25:30 +00:00
await async_setup_component(
2019-07-31 19:25:30 +00:00
hass,
"camera",
{
"camera": {
"name": "config_test",
"platform": "generic",
"still_image_url": "https://example.com",
"username": "user",
"password": "pass",
"verify_ssl": "true",
}
},
)
await hass.async_block_till_done()
client = await hass_client()
resp = await client.get("/api/camera_proxy/camera.config_test")
assert resp.status == HTTPStatus.OK
@respx.mock
async def test_limit_refetch(hass, hass_client, fakeimgbytes_png, fakeimgbytes_jpg):
"""Test that it fetches the given url."""
respx.get("http://example.com/5a").respond(stream=fakeimgbytes_png)
respx.get("http://example.com/10a").respond(stream=fakeimgbytes_png)
respx.get("http://example.com/15a").respond(stream=fakeimgbytes_jpg)
respx.get("http://example.com/20a").respond(status_code=HTTPStatus.NOT_FOUND)
2019-07-31 19:25:30 +00:00
await async_setup_component(
2019-07-31 19:25:30 +00:00
hass,
"camera",
{
"camera": {
"name": "config_test",
"platform": "generic",
"still_image_url": 'http://example.com/{{ states.sensor.temp.state + "a" }}',
"limit_refetch_to_url_change": True,
}
},
)
await hass.async_block_till_done()
client = await hass_client()
resp = await client.get("/api/camera_proxy/camera.config_test")
2019-07-31 19:25:30 +00:00
hass.states.async_set("sensor.temp", "5")
2021-11-04 15:07:50 +00:00
with pytest.raises(aiohttp.ServerTimeoutError), patch(
"async_timeout.timeout", side_effect=asyncio.TimeoutError()
):
resp = await client.get("/api/camera_proxy/camera.config_test")
2021-11-04 15:07:50 +00:00
assert respx.calls.call_count == 0
assert resp.status == HTTPStatus.INTERNAL_SERVER_ERROR
2019-07-31 19:25:30 +00:00
hass.states.async_set("sensor.temp", "10")
resp = await client.get("/api/camera_proxy/camera.config_test")
assert respx.calls.call_count == 1
assert resp.status == HTTPStatus.OK
body = await resp.read()
assert body == fakeimgbytes_png
resp = await client.get("/api/camera_proxy/camera.config_test")
assert respx.calls.call_count == 1
assert resp.status == HTTPStatus.OK
body = await resp.read()
assert body == fakeimgbytes_png
2019-07-31 19:25:30 +00:00
hass.states.async_set("sensor.temp", "15")
# Url change = fetch new image
resp = await client.get("/api/camera_proxy/camera.config_test")
assert respx.calls.call_count == 2
assert resp.status == HTTPStatus.OK
body = await resp.read()
assert body == fakeimgbytes_jpg
# Cause a template render error
2019-07-31 19:25:30 +00:00
hass.states.async_remove("sensor.temp")
resp = await client.get("/api/camera_proxy/camera.config_test")
assert respx.calls.call_count == 2
assert resp.status == HTTPStatus.OK
body = await resp.read()
assert body == fakeimgbytes_jpg
async def test_stream_source(hass, hass_client, hass_ws_client, fakeimgbytes_png):
"""Test that the stream source is rendered."""
respx.get("http://example.com").respond(stream=fakeimgbytes_png)
assert await async_setup_component(
hass,
"camera",
{
"camera": {
"name": "config_test",
"platform": "generic",
"still_image_url": "https://example.com",
"stream_source": 'http://example.com/{{ states.sensor.temp.state + "a" }}',
"limit_refetch_to_url_change": True,
},
},
)
assert await async_setup_component(hass, "stream", {})
await hass.async_block_till_done()
hass.states.async_set("sensor.temp", "5")
with patch(
"homeassistant.components.camera.Stream.endpoint_url",
return_value="http://home.assistant/playlist.m3u8",
) as mock_stream_url:
# Request playlist through WebSocket
client = await hass_ws_client(hass)
await client.send_json(
{"id": 1, "type": "camera/stream", "entity_id": "camera.config_test"}
)
msg = await client.receive_json()
# Assert WebSocket response
assert mock_stream_url.call_count == 1
assert msg["id"] == 1
assert msg["type"] == TYPE_RESULT
assert msg["success"]
assert msg["result"]["url"][-13:] == "playlist.m3u8"
async def test_stream_source_error(hass, hass_client, hass_ws_client, fakeimgbytes_png):
"""Test that the stream source has an error."""
respx.get("http://example.com").respond(stream=fakeimgbytes_png)
assert await async_setup_component(
hass,
"camera",
{
"camera": {
"name": "config_test",
"platform": "generic",
"still_image_url": "https://example.com",
# Does not exist
"stream_source": 'http://example.com/{{ states.sensor.temp.state + "a" }}',
"limit_refetch_to_url_change": True,
},
},
)
assert await async_setup_component(hass, "stream", {})
await hass.async_block_till_done()
with patch(
"homeassistant.components.camera.Stream.endpoint_url",
return_value="http://home.assistant/playlist.m3u8",
) as mock_stream_url:
# Request playlist through WebSocket
client = await hass_ws_client(hass)
await client.send_json(
{"id": 1, "type": "camera/stream", "entity_id": "camera.config_test"}
)
msg = await client.receive_json()
# Assert WebSocket response
assert mock_stream_url.call_count == 0
assert msg["id"] == 1
assert msg["type"] == TYPE_RESULT
assert msg["success"] is False
assert msg["error"] == {
"code": "start_stream_failed",
"message": "camera.config_test does not support play stream service",
}
async def test_setup_alternative_options(hass, hass_ws_client):
"""Test that the stream source is setup with different config options."""
assert await async_setup_component(
hass,
"camera",
{
"camera": {
"name": "config_test",
"platform": "generic",
"still_image_url": "https://example.com",
"authentication": "digest",
"username": "user",
"password": "pass",
"stream_source": "rtsp://example.com:554/rtsp/",
"rtsp_transport": "udp",
},
},
)
await hass.async_block_till_done()
assert hass.data["camera"].get_entity("camera.config_test")
async def test_no_stream_source(hass, hass_client, hass_ws_client, fakeimgbytes_png):
"""Test a stream request without stream source option set."""
respx.get("http://example.com").respond(stream=fakeimgbytes_png)
assert await async_setup_component(
hass,
"camera",
{
"camera": {
"name": "config_test",
"platform": "generic",
"still_image_url": "https://example.com",
"limit_refetch_to_url_change": True,
}
},
)
await hass.async_block_till_done()
with patch(
"homeassistant.components.camera.Stream.endpoint_url",
return_value="http://home.assistant/playlist.m3u8",
) as mock_request_stream:
# Request playlist through WebSocket
client = await hass_ws_client(hass)
await client.send_json(
{"id": 3, "type": "camera/stream", "entity_id": "camera.config_test"}
)
msg = await client.receive_json()
# Assert the websocket error message
assert mock_request_stream.call_count == 0
assert msg["id"] == 3
assert msg["type"] == TYPE_RESULT
assert msg["success"] is False
assert msg["error"] == {
"code": "start_stream_failed",
"message": "camera.config_test does not support play stream service",
}
@respx.mock
async def test_camera_content_type(
hass, hass_client, fakeimgbytes_svg, fakeimgbytes_jpg
):
"""Test generic camera with custom content_type."""
2019-07-31 19:25:30 +00:00
urlsvg = "https://upload.wikimedia.org/wikipedia/commons/0/02/SVG_logo.svg"
respx.get(urlsvg).respond(stream=fakeimgbytes_svg)
urljpg = "https://upload.wikimedia.org/wikipedia/commons/0/0e/Felis_silvestris_silvestris.jpg"
respx.get(urljpg).respond(stream=fakeimgbytes_jpg)
cam_config_svg = {
2019-07-31 19:25:30 +00:00
"name": "config_test_svg",
"platform": "generic",
"still_image_url": urlsvg,
"content_type": "image/svg+xml",
}
cam_config_jpg = {
"name": "config_test_jpg",
"platform": "generic",
"still_image_url": urljpg,
"content_type": "image/jpeg",
}
await async_setup_component(
hass, "camera", {"camera": [cam_config_svg, cam_config_jpg]}
2019-07-31 19:25:30 +00:00
)
await hass.async_block_till_done()
client = await hass_client()
resp_1 = await client.get("/api/camera_proxy/camera.config_test_svg")
assert respx.calls.call_count == 1
assert resp_1.status == HTTPStatus.OK
2019-07-31 19:25:30 +00:00
assert resp_1.content_type == "image/svg+xml"
body = await resp_1.read()
assert body == fakeimgbytes_svg
resp_2 = await client.get("/api/camera_proxy/camera.config_test_jpg")
assert respx.calls.call_count == 2
assert resp_2.status == HTTPStatus.OK
2019-07-31 19:25:30 +00:00
assert resp_2.content_type == "image/jpeg"
body = await resp_2.read()
assert body == fakeimgbytes_jpg
@respx.mock
async def test_reloading(hass, hass_client):
"""Test we can cleanly reload."""
respx.get("http://example.com").respond(text="hello world")
await async_setup_component(
hass,
"camera",
{
"camera": {
"name": "config_test",
"platform": "generic",
"still_image_url": "http://example.com",
"username": "user",
"password": "pass",
}
},
)
await hass.async_block_till_done()
client = await hass_client()
resp = await client.get("/api/camera_proxy/camera.config_test")
assert resp.status == HTTPStatus.OK
assert respx.calls.call_count == 1
body = await resp.text()
assert body == "hello world"
2021-11-02 03:47:05 +00:00
yaml_path = get_fixture_path("configuration.yaml", "generic")
with patch.object(hass_config, "YAML_CONFIG_FILE", yaml_path):
await hass.services.async_call(
2020-08-28 15:33:08 +00:00
DOMAIN,
SERVICE_RELOAD,
{},
blocking=True,
)
await hass.async_block_till_done()
assert len(hass.states.async_all()) == 1
resp = await client.get("/api/camera_proxy/camera.config_test")
assert resp.status == HTTPStatus.NOT_FOUND
resp = await client.get("/api/camera_proxy/camera.reload")
assert resp.status == HTTPStatus.OK
assert respx.calls.call_count == 2
body = await resp.text()
assert body == "hello world"
@respx.mock
async def test_timeout_cancelled(hass, hass_client, fakeimgbytes_png, fakeimgbytes_jpg):
"""Test that timeouts and cancellations return last image."""
respx.get("http://example.com").respond(stream=fakeimgbytes_png)
await async_setup_component(
hass,
"camera",
{
"camera": {
"name": "config_test",
"platform": "generic",
"still_image_url": "http://example.com",
"username": "user",
"password": "pass",
}
},
)
await hass.async_block_till_done()
client = await hass_client()
resp = await client.get("/api/camera_proxy/camera.config_test")
assert resp.status == HTTPStatus.OK
assert respx.calls.call_count == 1
assert await resp.read() == fakeimgbytes_png
respx.get("http://example.com").respond(stream=fakeimgbytes_jpg)
with patch(
"homeassistant.components.generic.camera.GenericCamera.async_camera_image",
side_effect=asyncio.CancelledError(),
):
resp = await client.get("/api/camera_proxy/camera.config_test")
assert respx.calls.call_count == 1
assert resp.status == HTTPStatus.INTERNAL_SERVER_ERROR
respx.get("http://example.com").side_effect = [
httpx.RequestError,
httpx.TimeoutException,
]
for total_calls in range(2, 3):
resp = await client.get("/api/camera_proxy/camera.config_test")
assert respx.calls.call_count == total_calls
assert resp.status == HTTPStatus.OK
assert await resp.read() == fakeimgbytes_png
async def test_no_still_image_url(hass, hass_client):
"""Test that the component can grab images from stream with no still_image_url."""
assert await async_setup_component(
hass,
"camera",
{
"camera": {
"name": "config_test",
"platform": "generic",
"stream_source": "rtsp://example.com:554/rtsp/",
},
},
)
await hass.async_block_till_done()
client = await hass_client()
with patch(
"homeassistant.components.generic.camera.GenericCamera.stream_source",
return_value=None,
) as mock_stream_source:
# First test when there is no stream_source should fail
resp = await client.get("/api/camera_proxy/camera.config_test")
await hass.async_block_till_done()
mock_stream_source.assert_called_once()
assert resp.status == HTTPStatus.INTERNAL_SERVER_ERROR
with patch("homeassistant.components.camera.create_stream") as mock_create_stream:
# Now test when creating the stream succeeds
mock_stream = Mock()
mock_stream.async_get_image = AsyncMock()
mock_stream.async_get_image.return_value = b"stream_keyframe_image"
mock_create_stream.return_value = mock_stream
# should start the stream and get the image
resp = await client.get("/api/camera_proxy/camera.config_test")
await hass.async_block_till_done()
mock_create_stream.assert_called_once()
mock_stream.async_get_image.assert_called_once()
assert resp.status == HTTPStatus.OK
assert await resp.read() == b"stream_keyframe_image"