Fix webrtc provider interface and tests (#129488)

* Fix webrtc provider tests

* Remove future code

* Add a test of the optional provider interface
pull/129436/head^2
Martin Hjelmare 2024-10-30 13:24:23 +01:00 committed by GitHub
parent c8594045df
commit 24829bc44f
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
5 changed files with 78 additions and 38 deletions

View File

@ -135,6 +135,7 @@ class CameraWebRTCProvider(Protocol):
@callback
def async_close_session(self, session_id: str) -> None:
"""Close the session."""
return ## This is an optional method so we need a default here.
class CameraWebRTCLegacyProvider(Protocol):

View File

@ -128,6 +128,7 @@ class WebRTCProvider(CameraWebRTCProvider):
self._rest_client = Go2RtcRestClient(self._session, url)
self._sessions: dict[str, Go2RtcWsClient] = {}
@callback
def async_is_supported(self, stream_source: str) -> bool:
"""Return if this provider is supports the Camera as source."""
return stream_source.partition(":")[0] in _SUPPORTED_STREAMS

View File

@ -6,13 +6,6 @@ components. Instead call the service directly.
from unittest.mock import Mock
from homeassistant.components.camera import Camera
from homeassistant.components.camera.webrtc import (
CameraWebRTCProvider,
async_register_webrtc_provider,
)
from homeassistant.core import HomeAssistant
EMPTY_8_6_JPEG = b"empty_8_6"
WEBRTC_ANSWER = "a=sendonly"
STREAM_SOURCE = "rtsp://127.0.0.1/stream"
@ -30,25 +23,3 @@ def mock_turbo_jpeg(
mocked_turbo_jpeg.scale_with_quality.return_value = EMPTY_8_6_JPEG
mocked_turbo_jpeg.encode.return_value = EMPTY_8_6_JPEG
return mocked_turbo_jpeg
async def add_webrtc_provider(hass: HomeAssistant) -> CameraWebRTCProvider:
"""Add test WebRTC provider."""
class SomeTestProvider(CameraWebRTCProvider):
"""Test provider."""
async def async_is_supported(self, stream_source: str) -> bool:
"""Determine if the provider supports the stream source."""
return True
async def async_handle_web_rtc_offer(
self, camera: Camera, offer_sdp: str
) -> str | None:
"""Handle the WebRTC offer and return an answer."""
return "answer"
provider = SomeTestProvider()
async_register_webrtc_provider(hass, provider)
await hass.async_block_till_done()
return provider

View File

@ -9,6 +9,13 @@ import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components import camera
from homeassistant.components.camera import (
Camera,
CameraWebRTCProvider,
WebRTCAnswer,
WebRTCSendMessage,
async_register_webrtc_provider,
)
from homeassistant.components.camera.const import (
DOMAIN,
PREF_ORIENTATION,
@ -23,20 +30,14 @@ from homeassistant.const import (
EVENT_HOMEASSISTANT_STARTED,
STATE_UNAVAILABLE,
)
from homeassistant.core import HomeAssistant
from homeassistant.core import HomeAssistant, callback
from homeassistant.core_config import async_process_ha_core_config
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import entity_registry as er, issue_registry as ir
from homeassistant.setup import async_setup_component
from homeassistant.util import dt as dt_util
from .common import (
EMPTY_8_6_JPEG,
STREAM_SOURCE,
WEBRTC_ANSWER,
add_webrtc_provider,
mock_turbo_jpeg,
)
from .common import EMPTY_8_6_JPEG, STREAM_SOURCE, WEBRTC_ANSWER, mock_turbo_jpeg
from tests.common import (
MockConfigEntry,
@ -933,7 +934,33 @@ async def _test_capabilities(
await test(expected_stream_types)
# Test with WebRTC provider
await add_webrtc_provider(hass)
class SomeTestProvider(CameraWebRTCProvider):
"""Test provider."""
@callback
def async_is_supported(self, stream_source: str) -> bool:
"""Determine if the provider supports the stream source."""
return True
async def async_handle_async_webrtc_offer(
self,
camera: Camera,
offer_sdp: str,
session_id: str,
send_message: WebRTCSendMessage,
) -> None:
"""Handle the WebRTC offer and return the answer via the provided callback."""
send_message(WebRTCAnswer("answer"))
async def async_on_webrtc_candidate(
self, session_id: str, candidate: str
) -> None:
"""Handle the WebRTC candidate."""
provider = SomeTestProvider()
async_register_webrtc_provider(hass, provider)
await hass.async_block_till_done()
await test(expected_stream_types_with_webrtc_provider)

View File

@ -56,6 +56,7 @@ class TestProvider(CameraWebRTCProvider):
"""Initialize the provider."""
self._is_supported = True
@callback
def async_is_supported(self, stream_source: str) -> bool:
"""Determine if the provider supports the stream source."""
return self._is_supported
@ -1085,3 +1086,42 @@ async def test_ws_webrtc_candidate_invalid_stream_type(
"code": "webrtc_candidate_failed",
"message": "Camera does not support WebRTC, frontend_stream_type=hls",
}
async def test_webrtc_provider_optional_interface(hass: HomeAssistant) -> None:
"""Test optional interface for WebRTC provider."""
class OnlyRequiredInterfaceProvider(CameraWebRTCProvider):
"""Test provider."""
@callback
def async_is_supported(self, stream_source: str) -> bool:
"""Determine if the provider supports the stream source."""
return True
async def async_handle_async_webrtc_offer(
self,
camera: Camera,
offer_sdp: str,
session_id: str,
send_message: WebRTCSendMessage,
) -> None:
"""Handle the WebRTC offer and return the answer via the provided callback.
Return value determines if the offer was handled successfully.
"""
send_message(WebRTCAnswer(answer="answer"))
async def async_on_webrtc_candidate(
self, session_id: str, candidate: str
) -> None:
"""Handle the WebRTC candidate."""
provider = OnlyRequiredInterfaceProvider()
# Call all interface methods
assert provider.async_is_supported("stream_source") is True
await provider.async_handle_async_webrtc_offer(
Mock(), "offer_sdp", "session_id", Mock()
)
await provider.async_on_webrtc_candidate("session_id", "candidate")
provider.async_close_session("session_id")