diff --git a/tests/components/camera/test_go2rtc_hls.py b/tests/components/camera/test_go2rtc_hls.py new file mode 100644 index 00000000000..c46b50355e5 --- /dev/null +++ b/tests/components/camera/test_go2rtc_hls.py @@ -0,0 +1,165 @@ +"""Test go2rtc HLS integration with camera component.""" + +from unittest.mock import AsyncMock, Mock, patch + +import pytest + +from homeassistant.components.camera import _async_stream_endpoint_url +from homeassistant.core import HomeAssistant +from homeassistant.exceptions import HomeAssistantError + +from tests.components.camera.common import mock_turbo_jpeg +from tests.common import MockConfigEntry + + +@pytest.fixture(autouse=True) +def mock_turbo() -> None: + """Mock TurboJPEG.""" + mock_turbo_jpeg() + + +async def test_stream_endpoint_prefers_go2rtc_hls(hass: HomeAssistant) -> None: + """Test that camera streaming prefers go2rtc HLS when available.""" + # Mock camera + camera = Mock() + camera.entity_id = "camera.test" + camera.stream_source = AsyncMock(return_value="rtsp://example.com/stream") + camera.async_create_stream = AsyncMock(return_value=None) + + # Mock go2rtc HLS provider + hls_provider = Mock() + hls_provider.async_is_supported = Mock(return_value=True) + hls_provider.async_get_stream_url = AsyncMock(return_value="/api/go2rtc_hls/camera.test/playlist.m3u8") + + # Set up hass.data with go2rtc HLS provider + hass.data["go2rtc"] = {"hls_provider": hls_provider} + + # Test that go2rtc HLS is used + url = await _async_stream_endpoint_url(hass, camera, "hls") + assert url == "/api/go2rtc_hls/camera.test/playlist.m3u8" + + # Verify go2rtc methods were called + hls_provider.async_is_supported.assert_called_once_with("rtsp://example.com/stream") + hls_provider.async_get_stream_url.assert_called_once_with(camera) + + # Verify stream integration was not used + camera.async_create_stream.assert_not_called() + + +async def test_stream_endpoint_fallback_to_stream_integration(hass: HomeAssistant) -> None: + """Test that camera streaming falls back to stream integration when go2rtc is not available.""" + # Mock camera + camera = Mock() + camera.entity_id = "camera.test" + camera.stream_source = AsyncMock(return_value="rtsp://example.com/stream") + + # Mock stream integration + stream = Mock() + stream.add_provider = Mock() + stream.start = AsyncMock() + stream.endpoint_url = Mock(return_value="/api/hls/token/master_playlist.m3u8") + camera.async_create_stream = AsyncMock(return_value=stream) + + # No go2rtc provider available + hass.data.clear() + + # Test that stream integration is used + url = await _async_stream_endpoint_url(hass, camera, "hls") + assert url == "/api/hls/token/master_playlist.m3u8" + + # Verify stream integration methods were called + camera.async_create_stream.assert_called_once() + stream.add_provider.assert_called_once_with("hls") + stream.start.assert_called_once() + stream.endpoint_url.assert_called_once_with("hls") + + +async def test_stream_endpoint_fallback_when_go2rtc_unsupported(hass: HomeAssistant) -> None: + """Test fallback to stream integration when go2rtc doesn't support the camera.""" + # Mock camera with unsupported stream source + camera = Mock() + camera.entity_id = "camera.test" + camera.stream_source = AsyncMock(return_value="unsupported://stream") + + # Mock stream integration + stream = Mock() + stream.add_provider = Mock() + stream.start = AsyncMock() + stream.endpoint_url = Mock(return_value="/api/hls/token/master_playlist.m3u8") + camera.async_create_stream = AsyncMock(return_value=stream) + + # Mock go2rtc HLS provider that doesn't support this stream + hls_provider = Mock() + hls_provider.async_is_supported = Mock(return_value=False) + hass.data["go2rtc"] = {"hls_provider": hls_provider} + + # Test that stream integration is used + url = await _async_stream_endpoint_url(hass, camera, "hls") + assert url == "/api/hls/token/master_playlist.m3u8" + + # Verify go2rtc was checked but rejected + hls_provider.async_is_supported.assert_called_once_with("unsupported://stream") + + # Verify stream integration was used as fallback + camera.async_create_stream.assert_called_once() + + +async def test_stream_endpoint_fallback_when_go2rtc_fails(hass: HomeAssistant) -> None: + """Test fallback to stream integration when go2rtc fails.""" + # Mock camera + camera = Mock() + camera.entity_id = "camera.test" + camera.stream_source = AsyncMock(return_value="rtsp://example.com/stream") + + # Mock stream integration + stream = Mock() + stream.add_provider = Mock() + stream.start = AsyncMock() + stream.endpoint_url = Mock(return_value="/api/hls/token/master_playlist.m3u8") + camera.async_create_stream = AsyncMock(return_value=stream) + + # Mock go2rtc HLS provider that fails + hls_provider = Mock() + hls_provider.async_is_supported = Mock(return_value=True) + hls_provider.async_get_stream_url = AsyncMock(side_effect=Exception("go2rtc failed")) + hass.data["go2rtc"] = {"hls_provider": hls_provider} + + # Test that stream integration is used as fallback + url = await _async_stream_endpoint_url(hass, camera, "hls") + assert url == "/api/hls/token/master_playlist.m3u8" + + # Verify go2rtc was attempted but failed + hls_provider.async_get_stream_url.assert_called_once_with(camera) + + # Verify stream integration was used as fallback + camera.async_create_stream.assert_called_once() + + +async def test_stream_endpoint_non_hls_format_uses_stream_integration(hass: HomeAssistant) -> None: + """Test that non-HLS formats always use stream integration.""" + # Mock camera + camera = Mock() + camera.entity_id = "camera.test" + camera.stream_source = AsyncMock(return_value="rtsp://example.com/stream") + + # Mock stream integration + stream = Mock() + stream.add_provider = Mock() + stream.start = AsyncMock() + stream.endpoint_url = Mock(return_value="/api/recorder/token/recording.mp4") + camera.async_create_stream = AsyncMock(return_value=stream) + + # Mock go2rtc HLS provider (should not be used for non-HLS) + hls_provider = Mock() + hass.data["go2rtc"] = {"hls_provider": hls_provider} + + # Test with recorder format + url = await _async_stream_endpoint_url(hass, camera, "recorder") + assert url == "/api/recorder/token/recording.mp4" + + # Verify go2rtc was not used + assert not hls_provider.method_calls + + # Verify stream integration was used + camera.async_create_stream.assert_called_once() + stream.add_provider.assert_called_once_with("recorder") \ No newline at end of file diff --git a/tests/components/go2rtc/test_init.py b/tests/components/go2rtc/test_init.py index e77e61346b6..316de75d5e4 100644 --- a/tests/components/go2rtc/test_init.py +++ b/tests/components/go2rtc/test_init.py @@ -30,7 +30,7 @@ from homeassistant.components.camera import ( async_get_image, ) from homeassistant.components.default_config import DOMAIN as DEFAULT_CONFIG_DOMAIN -from homeassistant.components.go2rtc import HomeAssistant, WebRTCProvider +from homeassistant.components.go2rtc import HomeAssistant, WebRTCProvider, Go2RtcHlsProvider from homeassistant.components.go2rtc.const import ( CONF_DEBUG_UI, DEBUG_UI_URL_MESSAGE, @@ -696,3 +696,51 @@ async def test_generic_workaround( f"ffmpeg:{camera.entity_id}#audio=opus#query=log_level=debug", ], ) + + +@pytest.mark.usefixtures("init_integration") +async def test_hls_provider_setup( + hass: HomeAssistant, + init_test_integration: MockCamera, +) -> None: + """Test HLS provider is properly set up.""" + from homeassistant.components.go2rtc.const import DOMAIN as GO2RTC_DOMAIN + + # Check that HLS provider is available in hass.data + assert GO2RTC_DOMAIN in hass.data + assert "hls_provider" in hass.data[GO2RTC_DOMAIN] + + hls_provider = hass.data[GO2RTC_DOMAIN]["hls_provider"] + assert hls_provider is not None + + # Test that the provider supports common stream sources + assert hls_provider.async_is_supported("rtsp://example.com/stream") + assert hls_provider.async_is_supported("http://example.com/stream.m3u8") + assert not hls_provider.async_is_supported("invalid://stream") + + +@pytest.mark.usefixtures("init_integration") +async def test_hls_stream_url_generation( + hass: HomeAssistant, + init_test_integration: MockCamera, + rest_client: AsyncMock, +) -> None: + """Test HLS stream URL generation.""" + from homeassistant.components.go2rtc.const import DOMAIN as GO2RTC_DOMAIN + + camera = init_test_integration + hls_provider = hass.data[GO2RTC_DOMAIN]["hls_provider"] + + # Test URL generation + url = await hls_provider.async_get_stream_url(camera) + expected_url = f"/api/go2rtc_hls/{camera.entity_id}/playlist.m3u8" + assert url == expected_url + + # Verify stream was configured in go2rtc + rest_client.streams.add.assert_called_with( + camera.entity_id, + [ + "rtsp://stream", + f"ffmpeg:{camera.entity_id}#audio=opus#query=log_level=debug", + ], + )