Add tests

homeassistant/components/camera/helper.py

@@ -33,4 +33,6 @@ async def get_dynamic_camera_stream_settings(
     hass: HomeAssistant, entity_id: str
 ) -> DynamicStreamSettings:
     """Get dynamic stream settings for a camera entity."""
+    if DATA_CAMERA_PREFS not in hass.data:
+        raise HomeAssistantError("Camera integration not set up")
     return await hass.data[DATA_CAMERA_PREFS].get_dynamic_stream_settings(entity_id)

tests/components/camera/test_helper.py

@@ -0,0 +1,76 @@
+"""Test camera helper functions."""
+
+import pytest
+
+from homeassistant.components.camera.const import DATA_CAMERA_PREFS
+from homeassistant.components.camera.helper import get_dynamic_camera_stream_settings
+from homeassistant.components.camera.prefs import (
+    CameraPreferences,
+    DynamicStreamSettings,
+)
+from homeassistant.components.stream import Orientation
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import HomeAssistantError
+
+
+async def test_get_dynamic_camera_stream_settings_missing_prefs(
+    hass: HomeAssistant,
+) -> None:
+    """Test get_dynamic_camera_stream_settings when camera prefs are not set up."""
+    with pytest.raises(HomeAssistantError, match="Camera integration not set up"):
+        await get_dynamic_camera_stream_settings(hass, "camera.test")
+
+
+async def test_get_dynamic_camera_stream_settings_success(hass: HomeAssistant) -> None:
+    """Test successful retrieval of dynamic camera stream settings."""
+    # Set up camera preferences
+    prefs = CameraPreferences(hass)
+    await prefs.async_load()
+    hass.data[DATA_CAMERA_PREFS] = prefs
+
+    # Test with default settings
+    settings = await get_dynamic_camera_stream_settings(hass, "camera.test")
+    assert settings.orientation == Orientation.NO_TRANSFORM
+    assert settings.preload_stream is False
+
+
+async def test_get_dynamic_camera_stream_settings_with_custom_orientation(
+    hass: HomeAssistant,
+) -> None:
+    """Test get_dynamic_camera_stream_settings with custom orientation set."""
+    # Set up camera preferences
+    prefs = CameraPreferences(hass)
+    await prefs.async_load()
+    hass.data[DATA_CAMERA_PREFS] = prefs
+
+    # Set custom orientation - this requires entity registry
+    # For this test, we'll directly manipulate the internal state
+    # since entity registry setup is complex for a unit test
+    test_settings = DynamicStreamSettings(
+        orientation=Orientation.ROTATE_LEFT, preload_stream=False
+    )
+    prefs._dynamic_stream_settings_by_entity_id["camera.test"] = test_settings
+
+    settings = await get_dynamic_camera_stream_settings(hass, "camera.test")
+    assert settings.orientation == Orientation.ROTATE_LEFT
+    assert settings.preload_stream is False
+
+
+async def test_get_dynamic_camera_stream_settings_with_preload_stream(
+    hass: HomeAssistant,
+) -> None:
+    """Test get_dynamic_camera_stream_settings with preload stream enabled."""
+    # Set up camera preferences
+    prefs = CameraPreferences(hass)
+    await prefs.async_load()
+    hass.data[DATA_CAMERA_PREFS] = prefs
+
+    # Set preload stream by directly setting the dynamic stream settings
+    test_settings = DynamicStreamSettings(
+        orientation=Orientation.NO_TRANSFORM, preload_stream=True
+    )
+    prefs._dynamic_stream_settings_by_entity_id["camera.test"] = test_settings
+
+    settings = await get_dynamic_camera_stream_settings(hass, "camera.test")
+    assert settings.orientation == Orientation.NO_TRANSFORM
+    assert settings.preload_stream is True

tests/components/go2rtc/test_init.py

@@ -1,6 +1,6 @@
 """The tests for the go2rtc component."""
 
-from collections.abc import Callable
+from collections.abc import Awaitable, Callable
 import logging
 from typing import NamedTuple
 from unittest.mock import AsyncMock, Mock, patch
@@ -29,6 +29,11 @@ from homeassistant.components.camera import (
     WebRTCSendMessage,
     async_get_image,
 )
+from homeassistant.components.camera.const import DATA_CAMERA_PREFS
+from homeassistant.components.camera.prefs import (
+    CameraPreferences,
+    DynamicStreamSettings,
+)
 from homeassistant.components.default_config import DOMAIN as DEFAULT_CONFIG_DOMAIN
 from homeassistant.components.go2rtc import HomeAssistant, WebRTCProvider
 from homeassistant.components.go2rtc.const import (
@@ -37,6 +42,7 @@ from homeassistant.components.go2rtc.const import (
     DOMAIN,
     RECOMMENDED_VERSION,
 )
+from homeassistant.components.stream import Orientation
 from homeassistant.config_entries import ConfigEntryState
 from homeassistant.const import CONF_URL
 from homeassistant.exceptions import HomeAssistantError
@@ -696,3 +702,211 @@ async def test_generic_workaround(
             f"ffmpeg:{camera.entity_id}#audio=opus#query=log_level=debug",
         ],
     )
+
+
+async def _test_camera_orientation(
+    hass: HomeAssistant,
+    camera: MockCamera,
+    orientation: Orientation,
+    rest_client: AsyncMock,
+    expected_stream_source: str,
+    camera_fn: Callable[[HomeAssistant, MockCamera], Awaitable[None]],
+) -> None:
+    """Test camera orientation handling in go2rtc provider."""
+    # Ensure go2rtc provider is initialized
+    assert isinstance(camera._webrtc_provider, WebRTCProvider)
+
+    prefs = CameraPreferences(hass)
+    await prefs.async_load()
+    hass.data[DATA_CAMERA_PREFS] = prefs
+
+    # Set the specific orientation for this test by directly setting the dynamic stream settings
+    test_settings = DynamicStreamSettings(orientation=orientation, preload_stream=False)
+    prefs._dynamic_stream_settings_by_entity_id[camera.entity_id] = test_settings
+
+    # Call the camera function that should trigger stream update
+    await camera_fn(hass, camera)
+
+    # Verify the stream was configured correctly
+    rest_client.streams.add.assert_called_once_with(
+        camera.entity_id,
+        [
+            expected_stream_source,
+            f"ffmpeg:{camera.entity_id}#audio=opus#query=log_level=debug",
+        ],
+    )
+
+
+async def _test_camera_orientation_webrtc(
+    hass: HomeAssistant,
+    camera: MockCamera,
+    orientation: Orientation,
+    rest_client: AsyncMock,
+    expected_stream_source: str,
+) -> None:
+    """Test camera orientation handling in go2rtc provider on WebRTC stream."""
+
+    # Ensure go2rtc provider is initialized
+    async def camera_fn(hass: HomeAssistant, camera: MockCamera) -> None:
+        """Mock function to simulate WebRTC offer handling."""
+        receive_message_callback = Mock()
+        await camera.async_handle_async_webrtc_offer(
+            OFFER_SDP, "test_session", receive_message_callback
+        )
+
+    await _test_camera_orientation(
+        hass,
+        camera,
+        orientation,
+        rest_client,
+        expected_stream_source,
+        camera_fn,
+    )
+
+
+async def _test_camera_orientation_get_image(
+    hass: HomeAssistant,
+    camera: MockCamera,
+    orientation: Orientation,
+    rest_client: AsyncMock,
+    expected_stream_source: str,
+) -> None:
+    """Test camera orientation handling in go2rtc provider on get_image."""
+
+    async def camera_fn(hass: HomeAssistant, camera: MockCamera) -> None:
+        """Mock function to simulate get_image handling."""
+        rest_client.get_jpeg_snapshot.return_value = b"image_bytes"
+        # Get image which should trigger stream update with orientation
+        await async_get_image(hass, camera.entity_id)
+
+    await _test_camera_orientation(
+        hass,
+        camera,
+        orientation,
+        rest_client,
+        expected_stream_source,
+        camera_fn,
+    )
+
+
+@pytest.mark.usefixtures("init_integration", "ws_client")
+@pytest.mark.parametrize(
+    ("orientation", "expected_suffix"),
+    [
+        (Orientation.MIRROR, "#video=h264#audio=copy#raw=-vf hflip"),
+        (Orientation.ROTATE_180, "#video=h264#audio=copy#rotate=180"),
+        (Orientation.FLIP, "#video=h264#audio=copy#raw=-vf vflip"),
+        (
+            Orientation.ROTATE_LEFT_AND_FLIP,
+            "#video=h264#audio=copy#raw=-vf transpose=2,vflip",
+        ),
+        (Orientation.ROTATE_LEFT, "#video=h264#audio=copy#rotate=-90"),
+        (
+            Orientation.ROTATE_RIGHT_AND_FLIP,
+            "#video=h264#audio=copy#raw=-vf transpose=1,vflip",
+        ),
+        (Orientation.ROTATE_RIGHT, "#video=h264#audio=copy#rotate=90"),
+    ],
+)
+@pytest.mark.parametrize(
+    "test_fn",
+    [
+        _test_camera_orientation_webrtc,
+        _test_camera_orientation_get_image,
+    ],
+)
+async def test_stream_orientation(
+    hass: HomeAssistant,
+    rest_client: AsyncMock,
+    init_test_integration: MockCamera,
+    orientation: Orientation,
+    expected_suffix: str,
+    test_fn: Callable[
+        [HomeAssistant, MockCamera, Orientation, AsyncMock, str], Awaitable[None]
+    ],
+) -> None:
+    """Test WebRTC provider applies correct orientation filters."""
+    camera = init_test_integration
+    expected_stream_source = f"ffmpeg:rtsp://stream{expected_suffix}"
+
+    await test_fn(
+        hass,
+        camera,
+        orientation,
+        rest_client,
+        expected_stream_source,
+    )
+
+
+@pytest.mark.usefixtures("init_integration", "ws_client")
+@pytest.mark.parametrize(
+    ("orientation", "stream_source", "expected_go2rtc_stream_source"),
+    [
+        (Orientation.NO_TRANSFORM, "rtsp://stream", "rtsp://stream"),
+        (
+            Orientation.ROTATE_LEFT,
+            "ffmpeg:rtsp://test.stream",
+            "ffmpeg:rtsp://test.stream#video=h264#audio=copy#rotate=-90",
+        ),
+    ],
+)
+@pytest.mark.parametrize(
+    "test_fn",
+    [
+        _test_camera_orientation_webrtc,
+        _test_camera_orientation_get_image,
+    ],
+)
+async def test_stream_orientation_other_cases(
+    hass: HomeAssistant,
+    rest_client: AsyncMock,
+    init_test_integration: MockCamera,
+    orientation: Orientation,
+    stream_source: str,
+    expected_go2rtc_stream_source: str,
+    test_fn: Callable[
+        [HomeAssistant, MockCamera, Orientation, AsyncMock, str], Awaitable[None]
+    ],
+) -> None:
+    """Test WebRTC provider with no orientation transform."""
+    camera = init_test_integration
+    camera.set_stream_source(stream_source)
+
+    await test_fn(
+        hass,
+        camera,
+        orientation,
+        rest_client,
+        expected_go2rtc_stream_source,
+    )
+
+
+@pytest.mark.usefixtures("init_integration", "ws_client")
+@pytest.mark.parametrize(
+    "test_fn",
+    [
+        _test_camera_orientation_webrtc,
+        _test_camera_orientation_get_image,
+    ],
+)
+async def test_stream_orientation_with_generic_camera(
+    hass: HomeAssistant,
+    rest_client: AsyncMock,
+    init_test_integration: MockCamera,
+    test_fn: Callable[
+        [HomeAssistant, MockCamera, Orientation, AsyncMock, str], Awaitable[None]
+    ],
+) -> None:
+    """Test WebRTC provider with orientation and generic camera platform."""
+    camera = init_test_integration
+    camera.set_stream_source("https://test.stream/video.m3u8")
+
+    # Test WebRTC offer handling with generic platform
+    with patch.object(camera.platform.platform_data, "platform_name", "generic"):
+        await test_fn(
+            hass,
+            camera,
+            Orientation.FLIP,
+            rest_client,
+            "ffmpeg:https://test.stream/video.m3u8#video=h264#audio=copy#raw=-vf vflip",
+        )