Make homekit camera snapshots HAP spec compliant (#35299)

tests/components/homekit/common.py

@@ -1,5 +1,7 @@
 """Collection of fixtures and functions for the HomeKit tests."""
-from tests.async_mock import patch
+from tests.async_mock import Mock, patch
+
+EMPTY_8_6_JPEG = b"empty_8_6"
 
 
 def patch_debounce():
@@ -8,3 +10,16 @@ def patch_debounce():
         "homeassistant.components.homekit.accessories.debounce",
         lambda f: lambda *args, **kwargs: f(*args, **kwargs),
     )
+
+
+def mock_turbo_jpeg(
+    first_width=None, second_width=None, first_height=None, second_height=None
+):
+    """Mock a TurboJPEG instance."""
+    mocked_turbo_jpeg = Mock()
+    mocked_turbo_jpeg.decode_header.side_effect = [
+        (first_width, first_height, 0, 0),
+        (second_width, second_height, 0, 0),
+    ]
+    mocked_turbo_jpeg.scale_with_quality.return_value = EMPTY_8_6_JPEG
+    return mocked_turbo_jpeg

tests/components/homekit/test_img_util.py

@@ -0,0 +1,62 @@
+"""Test HomeKit img_util module."""
+from homeassistant.components.camera import Image
+from homeassistant.components.homekit.img_util import (
+    TurboJPEGSingleton,
+    scale_jpeg_camera_image,
+)
+
+from .common import EMPTY_8_6_JPEG, mock_turbo_jpeg
+
+from tests.async_mock import patch
+
+EMPTY_16_12_JPEG = b"empty_16_12"
+
+
+def test_turbojpeg_singleton():
+    """Verify the instance always gives back the same."""
+    assert TurboJPEGSingleton.instance() == TurboJPEGSingleton.instance()
+
+
+def test_scale_jpeg_camera_image():
+    """Test we can scale a jpeg image."""
+
+    camera_image = Image("image/jpeg", EMPTY_16_12_JPEG)
+
+    turbo_jpeg = mock_turbo_jpeg(first_width=16, first_height=12)
+    with patch(
+        "homeassistant.components.homekit.img_util.TurboJPEG", return_value=False
+    ):
+        TurboJPEGSingleton()
+        assert scale_jpeg_camera_image(camera_image, 16, 12) == camera_image.content
+
+    turbo_jpeg = mock_turbo_jpeg(first_width=16, first_height=12)
+    with patch(
+        "homeassistant.components.homekit.img_util.TurboJPEG", return_value=turbo_jpeg
+    ):
+        TurboJPEGSingleton()
+        assert scale_jpeg_camera_image(camera_image, 16, 12) == EMPTY_16_12_JPEG
+
+    turbo_jpeg = mock_turbo_jpeg(
+        first_width=16, first_height=12, second_width=8, second_height=6
+    )
+    with patch(
+        "homeassistant.components.homekit.img_util.TurboJPEG", return_value=turbo_jpeg
+    ):
+        TurboJPEGSingleton()
+        jpeg_bytes = scale_jpeg_camera_image(camera_image, 8, 6)
+
+    assert jpeg_bytes == EMPTY_8_6_JPEG
+
+
+def test_turbojpeg_load_failure():
+    """Handle libjpegturbo not being installed."""
+
+    with patch(
+        "homeassistant.components.homekit.img_util.TurboJPEG", side_effect=Exception
+    ):
+        TurboJPEGSingleton()
+        assert TurboJPEGSingleton.instance() is False
+
+    with patch("homeassistant.components.homekit.img_util.TurboJPEG"):
+        TurboJPEGSingleton()
+        assert TurboJPEGSingleton.instance()

tests/components/homekit/test_type_cameras.py

@@ -15,11 +15,14 @@ from homeassistant.components.homekit.const import (
     CONF_VIDEO_CODEC,
     VIDEO_CODEC_COPY,
 )
+from homeassistant.components.homekit.img_util import TurboJPEGSingleton
 from homeassistant.components.homekit.type_cameras import Camera
 from homeassistant.components.homekit.type_switches import Switch
 from homeassistant.exceptions import HomeAssistantError
 from homeassistant.setup import async_setup_component
 
+from .common import mock_turbo_jpeg
+
 from tests.async_mock import AsyncMock, MagicMock, patch
 
 MOCK_START_STREAM_TLV = "ARUCAQEBEDMD1QMXzEaatnKSQ2pxovYCNAEBAAIJAQECAgECAwEAAwsBAgAFAgLQAgMBHgQXAQFjAgQ768/RAwIrAQQEAAAAPwUCYgUDLAEBAwIMAQEBAgEAAwECBAEUAxYBAW4CBCzq28sDAhgABAQAAKBABgENBAEA"
@@ -135,15 +138,30 @@ async def test_camera_stream_source_configured(hass, run_driver, events):
         await acc.stop_stream(session_info)
         await hass.async_block_till_done()
 
-    assert await hass.async_add_executor_job(acc.get_snapshot, 1024)
+    turbo_jpeg = mock_turbo_jpeg(
+        first_width=16, first_height=12, second_width=300, second_height=200
+    )
+    with patch(
+        "homeassistant.components.homekit.img_util.TurboJPEG", return_value=turbo_jpeg
+    ):
+        TurboJPEGSingleton()
+        assert await hass.async_add_executor_job(
+            acc.get_snapshot, {"aid": 2, "image-width": 300, "image-height": 200}
+        )
+        # Verify the bridge only forwards get_snapshot for
+        # cameras and valid accessory ids
+        assert await hass.async_add_executor_job(
+            bridge.get_snapshot, {"aid": 2, "image-width": 300, "image-height": 200}
+        )
 
-    # Verify the bridge only forwards get_snapshot for
-    # cameras and valid accessory ids
-    assert await hass.async_add_executor_job(bridge.get_snapshot, {"aid": 2})
     with pytest.raises(ValueError):
-        assert await hass.async_add_executor_job(bridge.get_snapshot, {"aid": 3})
+        assert await hass.async_add_executor_job(
+            bridge.get_snapshot, {"aid": 3, "image-width": 300, "image-height": 200}
+        )
     with pytest.raises(ValueError):
-        assert await hass.async_add_executor_job(bridge.get_snapshot, {"aid": 4})
+        assert await hass.async_add_executor_job(
+            bridge.get_snapshot, {"aid": 4, "image-width": 300, "image-height": 200}
+        )
 
 
 async def test_camera_stream_source_configured_with_failing_ffmpeg(
@@ -289,7 +307,9 @@ async def test_camera_with_no_stream(hass, run_driver, events):
     await hass.async_block_till_done()
 
     with pytest.raises(HomeAssistantError):
-        await hass.async_add_executor_job(acc.get_snapshot, 1024)
+        await hass.async_add_executor_job(
+            acc.get_snapshot, {"aid": 2, "image-width": 300, "image-height": 200}
+        )
 
 
 async def test_camera_stream_source_configured_and_copy_codec(hass, run_driver, events):