From cb8868ce0af63d71539abbecd15d4aa1df480bcb Mon Sep 17 00:00:00 2001 From: Robert Resch Date: Wed, 20 Aug 2025 10:09:32 +0200 Subject: [PATCH] Add tests --- homeassistant/components/camera/helper.py | 2 + tests/components/camera/test_helper.py | 76 ++++++++ tests/components/go2rtc/test_init.py | 216 +++++++++++++++++++++- 3 files changed, 293 insertions(+), 1 deletion(-) create mode 100644 tests/components/camera/test_helper.py diff --git a/homeassistant/components/camera/helper.py b/homeassistant/components/camera/helper.py index 334d74b7a88..8fd6d6cf20c 100644 --- a/homeassistant/components/camera/helper.py +++ b/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) diff --git a/tests/components/camera/test_helper.py b/tests/components/camera/test_helper.py new file mode 100644 index 00000000000..5fb9f914a81 --- /dev/null +++ b/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 diff --git a/tests/components/go2rtc/test_init.py b/tests/components/go2rtc/test_init.py index e77e61346b6..58472d020b9 100644 --- a/tests/components/go2rtc/test_init.py +++ b/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", + )