Improve tests for generic camera (#63197)
* Improve tests for generic camera * Fix black error * Code review: Move common code to fixtures * Remove unnecessary patches from tests. * Address review comments * Code review: swap more patches for respx * Code review: use _attr for frame interval.pull/62789/head
parent
4099d84fa4
commit
89895c6c04
|
@ -88,7 +88,7 @@ class GenericCamera(Camera):
|
|||
if self._stream_source is not None:
|
||||
self._stream_source.hass = hass
|
||||
self._limit_refetch = device_info[CONF_LIMIT_REFETCH_TO_URL_CHANGE]
|
||||
self._frame_interval = 1 / device_info[CONF_FRAMERATE]
|
||||
self._attr_frames_interval = 1 / device_info[CONF_FRAMERATE]
|
||||
self._supported_features = SUPPORT_STREAM if self._stream_source else 0
|
||||
self.content_type = device_info[CONF_CONTENT_TYPE]
|
||||
self.verify_ssl = device_info[CONF_VERIFY_SSL]
|
||||
|
@ -116,11 +116,6 @@ class GenericCamera(Camera):
|
|||
"""Return supported features for this camera."""
|
||||
return self._supported_features
|
||||
|
||||
@property
|
||||
def frame_interval(self):
|
||||
"""Return the interval between frames of the mjpeg stream."""
|
||||
return self._frame_interval
|
||||
|
||||
async def async_camera_image(
|
||||
self, width: int | None = None, height: int | None = None
|
||||
) -> bytes | None:
|
||||
|
|
|
@ -0,0 +1,31 @@
|
|||
"""Test fixtures for the generic component."""
|
||||
|
||||
from io import BytesIO
|
||||
|
||||
from PIL import Image
|
||||
import pytest
|
||||
|
||||
|
||||
@pytest.fixture(scope="package")
|
||||
def fakeimgbytes_png():
|
||||
"""Fake image in RAM for testing."""
|
||||
buf = BytesIO()
|
||||
Image.new("RGB", (1, 1)).save(buf, format="PNG")
|
||||
yield bytes(buf.getbuffer())
|
||||
|
||||
|
||||
@pytest.fixture(scope="package")
|
||||
def fakeimgbytes_jpg():
|
||||
"""Fake image in RAM for testing."""
|
||||
buf = BytesIO() # fake image in ram for testing.
|
||||
Image.new("RGB", (1, 1)).save(buf, format="jpeg")
|
||||
yield bytes(buf.getbuffer())
|
||||
|
||||
|
||||
@pytest.fixture(scope="package")
|
||||
def fakeimgbytes_svg():
|
||||
"""Fake image in RAM for testing."""
|
||||
yield bytes(
|
||||
'<svg xmlns="http://www.w3.org/2000/svg"><circle r="50"/></svg>',
|
||||
encoding="utf-8",
|
||||
)
|
|
@ -18,9 +18,9 @@ from tests.common import AsyncMock, Mock, get_fixture_path
|
|||
|
||||
|
||||
@respx.mock
|
||||
async def test_fetching_url(hass, hass_client):
|
||||
async def test_fetching_url(hass, hass_client, fakeimgbytes_png):
|
||||
"""Test that it fetches the given url."""
|
||||
respx.get("http://example.com").respond(text="hello world")
|
||||
respx.get("http://example.com").respond(stream=fakeimgbytes_png)
|
||||
|
||||
await async_setup_component(
|
||||
hass,
|
||||
|
@ -43,17 +43,17 @@ async def test_fetching_url(hass, hass_client):
|
|||
|
||||
assert resp.status == HTTPStatus.OK
|
||||
assert respx.calls.call_count == 1
|
||||
body = await resp.text()
|
||||
assert body == "hello world"
|
||||
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):
|
||||
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(text="hello world")
|
||||
respx.get("https://example.com").respond(stream=fakeimgbytes_png)
|
||||
|
||||
await async_setup_component(
|
||||
hass,
|
||||
|
@ -79,9 +79,9 @@ async def test_fetching_without_verify_ssl(hass, hass_client):
|
|||
|
||||
|
||||
@respx.mock
|
||||
async def test_fetching_url_with_verify_ssl(hass, hass_client):
|
||||
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(text="hello world")
|
||||
respx.get("https://example.com").respond(stream=fakeimgbytes_png)
|
||||
|
||||
await async_setup_component(
|
||||
hass,
|
||||
|
@ -107,11 +107,11 @@ async def test_fetching_url_with_verify_ssl(hass, hass_client):
|
|||
|
||||
|
||||
@respx.mock
|
||||
async def test_limit_refetch(hass, hass_client):
|
||||
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(text="hello world")
|
||||
respx.get("http://example.com/10a").respond(text="hello world")
|
||||
respx.get("http://example.com/15a").respond(text="hello planet")
|
||||
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)
|
||||
|
||||
await async_setup_component(
|
||||
|
@ -147,14 +147,14 @@ async def test_limit_refetch(hass, hass_client):
|
|||
resp = await client.get("/api/camera_proxy/camera.config_test")
|
||||
assert respx.calls.call_count == 1
|
||||
assert resp.status == HTTPStatus.OK
|
||||
body = await resp.text()
|
||||
assert body == "hello world"
|
||||
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.text()
|
||||
assert body == "hello world"
|
||||
body = await resp.read()
|
||||
assert body == fakeimgbytes_png
|
||||
|
||||
hass.states.async_set("sensor.temp", "15")
|
||||
|
||||
|
@ -162,20 +162,22 @@ async def test_limit_refetch(hass, hass_client):
|
|||
resp = await client.get("/api/camera_proxy/camera.config_test")
|
||||
assert respx.calls.call_count == 2
|
||||
assert resp.status == HTTPStatus.OK
|
||||
body = await resp.text()
|
||||
assert body == "hello planet"
|
||||
body = await resp.read()
|
||||
assert body == fakeimgbytes_jpg
|
||||
|
||||
# Cause a template render error
|
||||
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.text()
|
||||
assert body == "hello planet"
|
||||
body = await resp.read()
|
||||
assert body == fakeimgbytes_jpg
|
||||
|
||||
|
||||
async def test_stream_source(hass, hass_client, hass_ws_client):
|
||||
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",
|
||||
|
@ -214,8 +216,10 @@ async def test_stream_source(hass, hass_client, hass_ws_client):
|
|||
assert msg["result"]["url"][-13:] == "playlist.m3u8"
|
||||
|
||||
|
||||
async def test_stream_source_error(hass, hass_client, hass_ws_client):
|
||||
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",
|
||||
|
@ -278,8 +282,10 @@ async def test_setup_alternative_options(hass, hass_ws_client):
|
|||
assert hass.data["camera"].get_entity("camera.config_test")
|
||||
|
||||
|
||||
async def test_no_stream_source(hass, hass_client, hass_ws_client):
|
||||
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",
|
||||
|
@ -318,24 +324,29 @@ async def test_no_stream_source(hass, hass_client, hass_ws_client):
|
|||
|
||||
|
||||
@respx.mock
|
||||
async def test_camera_content_type(hass, hass_client):
|
||||
async def test_camera_content_type(
|
||||
hass, hass_client, fakeimgbytes_svg, fakeimgbytes_jpg
|
||||
):
|
||||
"""Test generic camera with custom content_type."""
|
||||
svg_image = "<some image>"
|
||||
urlsvg = "https://upload.wikimedia.org/wikipedia/commons/0/02/SVG_logo.svg"
|
||||
respx.get(urlsvg).respond(text=svg_image)
|
||||
|
||||
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 = {
|
||||
"name": "config_test_svg",
|
||||
"platform": "generic",
|
||||
"still_image_url": urlsvg,
|
||||
"content_type": "image/svg+xml",
|
||||
}
|
||||
cam_config_normal = cam_config_svg.copy()
|
||||
cam_config_normal.pop("content_type")
|
||||
cam_config_normal["name"] = "config_test_jpg"
|
||||
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_normal]}
|
||||
hass, "camera", {"camera": [cam_config_svg, cam_config_jpg]}
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
|
@ -345,15 +356,15 @@ async def test_camera_content_type(hass, hass_client):
|
|||
assert respx.calls.call_count == 1
|
||||
assert resp_1.status == HTTPStatus.OK
|
||||
assert resp_1.content_type == "image/svg+xml"
|
||||
body = await resp_1.text()
|
||||
assert body == svg_image
|
||||
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
|
||||
assert resp_2.content_type == "image/jpeg"
|
||||
body = await resp_2.text()
|
||||
assert body == svg_image
|
||||
body = await resp_2.read()
|
||||
assert body == fakeimgbytes_jpg
|
||||
|
||||
|
||||
@respx.mock
|
||||
|
@ -411,10 +422,10 @@ async def test_reloading(hass, hass_client):
|
|||
|
||||
|
||||
@respx.mock
|
||||
async def test_timeout_cancelled(hass, hass_client):
|
||||
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(text="hello world")
|
||||
respx.get("http://example.com").respond(stream=fakeimgbytes_png)
|
||||
|
||||
await async_setup_component(
|
||||
hass,
|
||||
|
@ -437,9 +448,9 @@ async def test_timeout_cancelled(hass, hass_client):
|
|||
|
||||
assert resp.status == HTTPStatus.OK
|
||||
assert respx.calls.call_count == 1
|
||||
assert await resp.text() == "hello world"
|
||||
assert await resp.read() == fakeimgbytes_png
|
||||
|
||||
respx.get("http://example.com").respond(text="not hello world")
|
||||
respx.get("http://example.com").respond(stream=fakeimgbytes_jpg)
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.generic.camera.GenericCamera.async_camera_image",
|
||||
|
@ -454,11 +465,11 @@ async def test_timeout_cancelled(hass, hass_client):
|
|||
httpx.TimeoutException,
|
||||
]
|
||||
|
||||
for total_calls in range(2, 4):
|
||||
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.text() == "hello world"
|
||||
assert await resp.read() == fakeimgbytes_png
|
||||
|
||||
|
||||
async def test_no_still_image_url(hass, hass_client):
|
||||
|
|
Loading…
Reference in New Issue