Add tests for ring camera platform for 100% coverage (#122197)

homeassistant/components/ring/camera.py

@@ -4,7 +4,7 @@ from __future__ import annotations
 
 from datetime import timedelta
 import logging
-from typing import Any
+from typing import TYPE_CHECKING, Any
 
 from aiohttp import web
 from haffmpeg.camera import CameraMjpeg
@@ -145,8 +145,9 @@ class RingCam(RingEntity[RingDoorBell], Camera):
             self._attr_motion_detection_enabled = self._device.motion_detection
             self.async_write_ha_state()
 
-        if self._last_event is None:
-            return
+        if TYPE_CHECKING:
+            # _last_event is set before calling update so will never be None
+            assert self._last_event
 
         if self._last_event["recording"]["status"] != "ready":
             return
@@ -165,8 +166,9 @@ class RingCam(RingEntity[RingDoorBell], Camera):
 
     @exception_wrap
     def _get_video(self) -> str | None:
-        if self._last_event is None:
-            return None
+        if TYPE_CHECKING:
+            # _last_event is set before calling update so will never be None
+            assert self._last_event
         event_id = self._last_event.get("id")
         assert event_id and isinstance(event_id, int)
         return self._device.recording_url(event_id)

tests/components/ring/device_mocks.py

@@ -142,6 +142,9 @@ def _mocked_ring_device(device_dict, device_family, device_class, capabilities):
             DOORBOT_HISTORY if device_family != "other" else INTERCOM_HISTORY
         )
 
+    if has_capability(RingCapability.VIDEO):
+        mock_device.recording_url = MagicMock(return_value="http://dummy.url")
+
     if has_capability(RingCapability.MOTION_DETECTION):
         mock_device.configure_mock(
             motion_detection=device_dict["settings"].get("motion_detection_enabled"),

tests/components/ring/test_camera.py

@@ -1,18 +1,33 @@
 """The tests for the Ring switch platform."""
 
-from unittest.mock import PropertyMock
+from unittest.mock import AsyncMock, MagicMock, PropertyMock, patch
 
+from aiohttp.test_utils import make_mocked_request
+from freezegun.api import FrozenDateTimeFactory
 import pytest
 import ring_doorbell
 
+from homeassistant.components import camera
+from homeassistant.components.ring.camera import FORCE_REFRESH_INTERVAL
+from homeassistant.components.ring.const import SCAN_INTERVAL
 from homeassistant.config_entries import SOURCE_REAUTH
 from homeassistant.const import Platform
 from homeassistant.core import HomeAssistant
 from homeassistant.exceptions import HomeAssistantError
 from homeassistant.helpers import entity_registry as er
+from homeassistant.util.aiohttp import MockStreamReader
 
 from .common import setup_platform
 
+from tests.common import async_fire_time_changed
+
+SMALLEST_VALID_JPEG = (
+    "ffd8ffe000104a46494600010101004800480000ffdb00430003020202020203020202030303030406040404040408060"
+    "6050609080a0a090809090a0c0f0c0a0b0e0b09090d110d0e0f101011100a0c12131210130f101010ffc9000b08000100"
+    "0101011100ffcc000600101005ffda0008010100003f00d2cf20ffd9"
+)
+SMALLEST_VALID_JPEG_BYTES = bytes.fromhex(SMALLEST_VALID_JPEG)
+
 
 async def test_entity_registry(
     hass: HomeAssistant,
@@ -52,7 +67,7 @@ async def test_camera_motion_detection_state_reports_correctly(
     assert state.attributes.get("friendly_name") == friendly_name
 
 
-async def test_camera_motion_detection_can_be_turned_on(
+async def test_camera_motion_detection_can_be_turned_on_and_off(
     hass: HomeAssistant, mock_ring_client
 ) -> None:
     """Tests the siren turns on correctly."""
@@ -73,6 +88,55 @@ async def test_camera_motion_detection_can_be_turned_on(
     state = hass.states.get("camera.front")
     assert state.attributes.get("motion_detection") is True
 
+    await hass.services.async_call(
+        "camera",
+        "disable_motion_detection",
+        {"entity_id": "camera.front"},
+        blocking=True,
+    )
+
+    await hass.async_block_till_done()
+
+    state = hass.states.get("camera.front")
+    assert state.attributes.get("motion_detection") is None
+
+
+async def test_camera_motion_detection_not_supported(
+    hass: HomeAssistant,
+    mock_ring_client,
+    mock_ring_devices,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """Tests the siren turns on correctly."""
+    front_camera_mock = mock_ring_devices.get_device(765432)
+    has_capability = front_camera_mock.has_capability.side_effect
+
+    def _has_capability(capability):
+        if capability == "motion_detection":
+            return False
+        return has_capability(capability)
+
+    front_camera_mock.has_capability.side_effect = _has_capability
+
+    await setup_platform(hass, Platform.CAMERA)
+
+    state = hass.states.get("camera.front")
+    assert state.attributes.get("motion_detection") is None
+
+    await hass.services.async_call(
+        "camera",
+        "enable_motion_detection",
+        {"entity_id": "camera.front"},
+        blocking=True,
+    )
+
+    await hass.async_block_till_done()
+    state = hass.states.get("camera.front")
+    assert state.attributes.get("motion_detection") is None
+    assert (
+        "Entity camera.front does not have motion detection capability" in caplog.text
+    )
+
 
 async def test_updates_work(
     hass: HomeAssistant, mock_ring_client, mock_ring_devices
@@ -136,3 +200,117 @@ async def test_motion_detection_errors_when_turned_on(
         )
         == reauth_expected
     )
+
+
+async def test_camera_handle_mjpeg_stream(
+    hass: HomeAssistant,
+    mock_ring_client,
+    mock_ring_devices,
+    freezer: FrozenDateTimeFactory,
+) -> None:
+    """Test camera returns handle mjpeg stream when available."""
+    await setup_platform(hass, Platform.CAMERA)
+
+    front_camera_mock = mock_ring_devices.get_device(765432)
+    front_camera_mock.recording_url.return_value = None
+
+    state = hass.states.get("camera.front")
+    assert state is not None
+
+    mock_request = make_mocked_request("GET", "/", headers={"token": "x"})
+
+    # history not updated yet
+    front_camera_mock.history.assert_not_called()
+    front_camera_mock.recording_url.assert_not_called()
+    stream = await camera.async_get_mjpeg_stream(hass, mock_request, "camera.front")
+    assert stream is None
+
+    # Video url will be none so no  stream
+    freezer.tick(SCAN_INTERVAL)
+    async_fire_time_changed(hass)
+    await hass.async_block_till_done(wait_background_tasks=True)
+    front_camera_mock.history.assert_called_once()
+    front_camera_mock.recording_url.assert_called_once()
+
+    stream = await camera.async_get_mjpeg_stream(hass, mock_request, "camera.front")
+    assert stream is None
+
+    # Stop the history updating so we can update the values manually
+    front_camera_mock.history = MagicMock()
+    front_camera_mock.last_history[0]["recording"]["status"] = "not ready"
+    freezer.tick(SCAN_INTERVAL)
+    async_fire_time_changed(hass)
+    await hass.async_block_till_done(wait_background_tasks=True)
+    front_camera_mock.recording_url.assert_called_once()
+    stream = await camera.async_get_mjpeg_stream(hass, mock_request, "camera.front")
+    assert stream is None
+
+    # If the history id hasn't changed the camera will not check again for the video url
+    # until the FORCE_REFRESH_INTERVAL has passed
+    front_camera_mock.last_history[0]["recording"]["status"] = "ready"
+    front_camera_mock.recording_url = MagicMock(return_value="http://dummy.url")
+    freezer.tick(SCAN_INTERVAL)
+    async_fire_time_changed(hass)
+    await hass.async_block_till_done(wait_background_tasks=True)
+    front_camera_mock.recording_url.assert_not_called()
+
+    stream = await camera.async_get_mjpeg_stream(hass, mock_request, "camera.front")
+    assert stream is None
+
+    freezer.tick(FORCE_REFRESH_INTERVAL)
+    async_fire_time_changed(hass)
+    await hass.async_block_till_done(wait_background_tasks=True)
+    front_camera_mock.recording_url.assert_called_once()
+
+    # Now the stream should be returned
+    stream_reader = MockStreamReader(SMALLEST_VALID_JPEG_BYTES)
+    with patch("homeassistant.components.ring.camera.CameraMjpeg") as mock_camera:
+        mock_camera.return_value.get_reader = AsyncMock(return_value=stream_reader)
+        mock_camera.return_value.open_camera = AsyncMock()
+        mock_camera.return_value.close = AsyncMock()
+
+        stream = await camera.async_get_mjpeg_stream(hass, mock_request, "camera.front")
+        assert stream is not None
+        # Check the stream has been read
+        assert not await stream_reader.read(-1)
+
+
+async def test_camera_image(
+    hass: HomeAssistant,
+    mock_ring_client,
+    mock_ring_devices,
+    freezer: FrozenDateTimeFactory,
+) -> None:
+    """Test camera will return still image when available."""
+    await setup_platform(hass, Platform.CAMERA)
+
+    front_camera_mock = mock_ring_devices.get_device(765432)
+
+    state = hass.states.get("camera.front")
+    assert state is not None
+
+    # history not updated yet
+    front_camera_mock.history.assert_not_called()
+    front_camera_mock.recording_url.assert_not_called()
+    with (
+        patch(
+            "homeassistant.components.ring.camera.ffmpeg.async_get_image",
+            return_value=SMALLEST_VALID_JPEG_BYTES,
+        ),
+        pytest.raises(HomeAssistantError),
+    ):
+        image = await camera.async_get_image(hass, "camera.front")
+
+    freezer.tick(SCAN_INTERVAL)
+    async_fire_time_changed(hass)
+    await hass.async_block_till_done(wait_background_tasks=True)
+    # history updated so image available
+    front_camera_mock.history.assert_called_once()
+    front_camera_mock.recording_url.assert_called_once()
+
+    with patch(
+        "homeassistant.components.ring.camera.ffmpeg.async_get_image",
+        return_value=SMALLEST_VALID_JPEG_BYTES,
+    ):
+        image = await camera.async_get_image(hass, "camera.front")
+        assert image.content == SMALLEST_VALID_JPEG_BYTES