starred/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-08-24 10:13:52 -05:00
Code Issues Packages Projects Releases Wiki Activity

Remove just go2rtc sessions of the failing camera (#179534)

Browse Source 
Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>
This commit is contained in:
Robert Resch
2026-08-21 21:54:07 +02:00
committed by GitHub
co-authored by Copilot Autofix powered by AI
parent 770da0afde
commit 99262963c9
4 changed files with 260 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files
homeassistant/components/go2rtc/__init__.py
+33 -11
View File
@@ -261,6 +261,14 @@ async def _get_binary(hass: HomeAssistant) -> str | None:
return await hass.async_add_executor_job(shutil.which, "go2rtc")
@dataclass(frozen=True)
class _SessionInfo:
"""Session info."""
ws_client: Go2RtcWsClient
camera: Camera
class WebRTCProvider(CameraWebRTCProvider):
"""WebRTC provider."""
@@ -276,7 +284,7 @@ class WebRTCProvider(CameraWebRTCProvider):
self._url = url
self._session = session
self._rest_client = rest_client
self._sessions: dict[str, Go2RtcWsClient] = {}
self._sessions: dict[str, _SessionInfo] = {}
self._supported_schemes: set[str] = set()
@property
@@ -310,9 +318,13 @@ class WebRTCProvider(CameraWebRTCProvider):
send_message(WebRTCError("go2rtc_webrtc_offer_failed", str(err)))
return
self._sessions[session_id] = ws_client = Go2RtcWsClient(
ws_client = Go2RtcWsClient(
self._session, self._url, source=get_camera_identifier(camera)
)
self._sessions[session_id] = _SessionInfo(
ws_client=ws_client,
camera=camera,
)
@callback
def on_messages(message: ReceiveMessages) -> None:
@@ -338,8 +350,8 @@ class WebRTCProvider(CameraWebRTCProvider):
) -> None:
"""Handle the WebRTC candidate."""
if ws_client := self._sessions.get(session_id):
await ws_client.send(WebRTCCandidate(candidate.candidate))
if session_info := self._sessions.get(session_id):
await session_info.ws_client.send(WebRTCCandidate(candidate.candidate))
else:
_LOGGER.debug("Unknown session %s. Ignoring candidate", session_id)
@@ -347,8 +359,8 @@ class WebRTCProvider(CameraWebRTCProvider):
@override
def async_close_session(self, session_id: str) -> None:
"""Close the session."""
ws_client = self._sessions.pop(session_id)
self._hass.async_create_task(ws_client.close())
if session_info := self._sessions.pop(session_id, None):
self._hass.async_create_task(session_info.ws_client.close())
@override
async def async_get_image(
@@ -366,7 +378,7 @@ class WebRTCProvider(CameraWebRTCProvider):
async def _update_stream_source(self, camera: Camera) -> None:
"""Update the stream source in go2rtc config if needed."""
if not (stream_source := await camera.stream_source()):
await self.teardown()
await self._close_camera_sessions(camera)
raise HomeAssistantError("Camera has no stream source")
if camera.platform.platform_name == "generic":
@@ -376,7 +388,7 @@ class WebRTCProvider(CameraWebRTCProvider):
stream_source = "ffmpeg:" + stream_source
if not self.async_is_supported(stream_source):
await self.teardown()
await self._close_camera_sessions(camera)
raise HomeAssistantError("Stream source is not supported by go2rtc")
camera_prefs = await get_dynamic_camera_stream_settings(
@@ -440,11 +452,20 @@ class WebRTCProvider(CameraWebRTCProvider):
else:
await self._rest_client.preload.disable(identifier)
async def _close_camera_sessions(self, camera: Camera) -> None:
for session_id in list(self._sessions):
session_info = self._sessions.get(session_id)
if session_info is None or session_info.camera != camera:
continue
# Unregister before closing, as closing yields to the event loop
del self._sessions[session_id]
await session_info.ws_client.close()
async def teardown(self) -> None:
"""Tear down the provider."""
for ws_client in self._sessions.values():
await ws_client.close()
self._sessions.clear()
while self._sessions:
_, session_info = self._sessions.popitem()
await session_info.ws_client.close()
@override
async def async_register_camera(
@@ -460,6 +481,7 @@ class WebRTCProvider(CameraWebRTCProvider):
camera: Camera,
) -> None:
"""Will be called when the provider is unregistered for a camera."""
await self._close_camera_sessions(camera)
identifier = get_camera_identifier(camera)
if identifier in await self._rest_client.preload.list():
await self._rest_client.preload.disable(identifier)
tests/components/go2rtc/__init__.py
+2 -2
View File
@@ -6,14 +6,14 @@ from homeassistant.components.camera import Camera, CameraEntityFeature
class MockCamera(Camera):
"""Mock Camera Entity."""
_attr_name = "Test"
_attr_supported_features: CameraEntityFeature = CameraEntityFeature.STREAM
def __init__(self, unique_id: str | None) -> None:
def __init__(self, unique_id: str | None, name: str = "Test") -> None:
"""Initialize the mock entity."""
super().__init__()
self._stream_source: str | None = "rtsp://stream"
self._attr_unique_id = unique_id
self._attr_name = name
def set_stream_source(self, stream_source: str | None) -> None:
"""Set the stream source."""
tests/components/go2rtc/conftest.py
+46 -11
View File
@@ -2,7 +2,8 @@
from collections.abc import Generator
from pathlib import Path
from unittest.mock import AsyncMock, Mock, patch
from typing import Any
from unittest.mock import AsyncMock, Mock, create_autospec, patch
from awesomeversion import AwesomeVersion
from go2rtc_client.rest import (
@@ -11,6 +12,7 @@ from go2rtc_client.rest import (
_StreamClient,
_WebRTCClient,
)
from go2rtc_client.ws import Go2RtcWsClient
import pytest
from homeassistant.components.camera import DOMAIN as CAMERA_DOMAIN
@@ -82,6 +84,19 @@ def ws_client() -> Generator[Mock]:
yield ws_client_mock.return_value
@pytest.fixture
def ws_clients() -> Generator[list[Mock]]:
"""Mock go2rtc websocket clients with a separate mock per created client."""
clients: list[Mock] = []
def create_client(*args: Any, **kwargs: Any) -> Mock:
clients.append(client := create_autospec(Go2RtcWsClient, instance=True))
return client
with patch(f"{GO2RTC_PATH}.Go2RtcWsClient", side_effect=create_client):
yield clients
@pytest.fixture
def server_stdout() -> list[str]:
"""Server stdout lines."""
@@ -198,13 +213,12 @@ def camera_unique_id() -> str | None:
return "camera_unique_id"
@pytest.fixture
async def init_test_integration(
async def _setup_test_integration(
hass: HomeAssistant,
integration_config_entry: ConfigEntry,
camera_unique_id: str | None,
) -> MockCamera:
"""Initialize components."""
cameras: list[MockCamera],
) -> None:
"""Set up the test integration with the given cameras."""
async def async_setup_entry_init(
hass: HomeAssistant, config_entry: ConfigEntry
@@ -232,17 +246,38 @@ async def init_test_integration(
async_unload_entry=async_unload_entry_init,
),
)
test_camera = MockCamera(camera_unique_id)
setup_test_component_platform(
hass, CAMERA_DOMAIN, [test_camera], from_config_entry=True
)
setup_test_component_platform(hass, CAMERA_DOMAIN, cameras, from_config_entry=True)
mock_platform(hass, f"{TEST_DOMAIN}.config_flow", Mock())
with mock_config_flow(TEST_DOMAIN, ConfigFlow):
assert await hass.config_entries.async_setup(integration_config_entry.entry_id)
await hass.async_block_till_done()
return test_camera
@pytest.fixture
async def init_test_integration(
hass: HomeAssistant,
integration_config_entry: ConfigEntry,
camera_unique_id: str | None,
) -> MockCamera:
"""Initialize components."""
camera = MockCamera(camera_unique_id)
await _setup_test_integration(hass, integration_config_entry, [camera])
return camera
@pytest.fixture
async def init_test_integration_two_cameras(
hass: HomeAssistant,
integration_config_entry: ConfigEntry,
) -> tuple[MockCamera, MockCamera]:
"""Initialize components with two cameras."""
cameras = (
MockCamera("camera_unique_id_1"),
MockCamera("camera_unique_id_2", "Test 2"),
)
await _setup_test_integration(hass, integration_config_entry, list(cameras))
return cameras
@pytest.fixture
tests/components/go2rtc/test_init.py
+179 -7
View File
@@ -1,5 +1,6 @@
"""The tests for the go2rtc component."""
import asyncio
from collections.abc import Awaitable, Callable
import logging
from pathlib import Path
@@ -194,14 +195,16 @@ async def _test_setup_and_signaling(
receive_message_callback.assert_called_once_with(
WebRTCError("go2rtc_webrtc_offer_failed", "Camera has no stream source")
)
teardown.assert_called_once()
# Only the sessions of the failing camera are closed, the provider stays up
teardown.assert_not_called()
# We use one ws_client mock for all sessions
assert ws_client.close.call_count == len(sessions)
assert not provider._sessions
await hass.config_entries.async_unload(config_entry.entry_id)
await hass.async_block_till_done()
assert config_entry.state is ConfigEntryState.NOT_LOADED
assert teardown.call_count == 2
teardown.assert_called_once()
@pytest.mark.usefixtures(
@@ -466,8 +469,7 @@ async def test_close_session(
session_id = "session_id"
# Session doesn't exist
with pytest.raises(KeyError):
camera.close_webrtc_session(session_id)
camera.close_webrtc_session(session_id)
ws_client.close.assert_not_called()
# Store session
@@ -485,13 +487,183 @@ async def test_close_session(
camera.close_webrtc_session(session_id)
ws_client.close.assert_called_once()
# Close again should raise an error
# Closing an already closed session is a no-op
ws_client.reset_mock()
with pytest.raises(KeyError):
camera.close_webrtc_session(session_id)
camera.close_webrtc_session(session_id)
ws_client.close.assert_not_called()
async def _fail_with_offer(hass: HomeAssistant, camera: MockCamera, error: str) -> None:
"""Update the stream source via a new WebRTC offer, expecting an error."""
send_message = Mock(spec_set=WebRTCSendMessage)
await camera.async_handle_async_webrtc_offer(OFFER_SDP, "new_session", send_message)
send_message.assert_called_once_with(
WebRTCError("go2rtc_webrtc_offer_failed", error)
)
async def _fail_with_image_request(
hass: HomeAssistant, camera: MockCamera, error: str
) -> None:
"""Update the stream source via a snapshot request, expecting an error."""
with pytest.raises(HomeAssistantError, match=error):
await async_get_image(hass, camera.entity_id)
@pytest.mark.parametrize(
("stream_source", "error"),
[
(
None,
"Camera has no stream source",
),
(
"invalid://not_supported",
"Stream source is not supported by go2rtc",
),
],
ids=["no_stream_source", "unsupported_stream_source"],
)
@pytest.mark.parametrize(
"trigger",
[
_fail_with_offer,
_fail_with_image_request,
],
ids=["offer", "image_request"],
)
@pytest.mark.usefixtures("init_integration")
async def test_invalid_stream_source_closes_only_sessions_of_that_camera(
hass: HomeAssistant,
ws_clients: list[Mock],
init_test_integration_two_cameras: tuple[MockCamera, MockCamera],
caplog: pytest.LogCaptureFixture,
trigger: Callable[[HomeAssistant, MockCamera, str], Awaitable[None]],
stream_source: str | None,
error: str,
) -> None:
"""Test an invalid stream source only closes the sessions of that camera."""
camera_1, camera_2 = init_test_integration_two_cameras
await camera_1.async_handle_async_webrtc_offer(OFFER_SDP, "session_1", Mock())
await camera_2.async_handle_async_webrtc_offer(OFFER_SDP, "session_2", Mock())
ws_client_1, ws_client_2 = ws_clients
ws_client_1.reset_mock()
ws_client_2.reset_mock()
caplog.clear()
camera_1.set_stream_source(stream_source)
await trigger(hass, camera_1, error)
ws_client_1.close.assert_called_once()
ws_client_2.close.assert_not_called()
# The session of camera 1 is gone
await camera_1.async_on_webrtc_candidate(
"session_1", RTCIceCandidateInit("candidate")
)
assert (
"homeassistant.components.go2rtc",
logging.DEBUG,
"Unknown session session_1. Ignoring candidate",
) in caplog.record_tuples
ws_client_1.send.assert_not_called()
# Closing the already closed session, e.g. by the frontend, is a no-op
camera_1.close_webrtc_session("session_1")
ws_client_1.close.assert_called_once()
# The session of camera 2 is untouched
await camera_2.async_on_webrtc_candidate(
"session_2", RTCIceCandidateInit("candidate")
)
ws_client_2.send.assert_called_once_with(WebRTCCandidate("candidate"))
camera_2.close_webrtc_session("session_2")
ws_client_2.close.assert_called_once()
@pytest.mark.usefixtures("init_integration")
async def test_unregister_camera_closes_only_sessions_of_that_camera(
ws_clients: list[Mock],
init_test_integration_two_cameras: tuple[MockCamera, MockCamera],
) -> None:
"""Test removing a camera closes only the sessions of that camera."""
camera_1, camera_2 = init_test_integration_two_cameras
await camera_1.async_handle_async_webrtc_offer(OFFER_SDP, "session_1", Mock())
await camera_2.async_handle_async_webrtc_offer(OFFER_SDP, "session_2", Mock())
ws_client_1, ws_client_2 = ws_clients
ws_client_1.reset_mock()
ws_client_2.reset_mock()
await camera_1.async_remove()
ws_client_1.close.assert_called_once()
ws_client_2.close.assert_not_called()
# The session of camera 2 is untouched
await camera_2.async_on_webrtc_candidate(
"session_2", RTCIceCandidateInit("candidate")
)
ws_client_2.send.assert_called_once_with(WebRTCCandidate("candidate"))
@pytest.mark.usefixtures("init_integration")
async def test_teardown_while_a_camera_is_removed(
ws_clients: list[Mock],
init_test_integration_two_cameras: tuple[MockCamera, MockCamera],
) -> None:
"""Test tearing down the provider while a camera is removed."""
camera_1, camera_2 = init_test_integration_two_cameras
await camera_1.async_handle_async_webrtc_offer(OFFER_SDP, "session_1", Mock())
await camera_2.async_handle_async_webrtc_offer(OFFER_SDP, "session_2", Mock())
ws_client_1, ws_client_2 = ws_clients
assert isinstance(camera_1.webrtc_provider, WebRTCProvider)
provider = camera_1.webrtc_provider
async def yield_control() -> None:
"""Let the camera removal run while the teardown is in progress."""
await asyncio.sleep(0)
ws_client_1.close.side_effect = yield_control
ws_client_2.close.side_effect = yield_control
await asyncio.gather(provider.teardown(), camera_2.async_remove())
ws_client_1.close.assert_called_once()
ws_client_2.close.assert_called_once()
assert not provider._sessions
@pytest.mark.usefixtures("init_integration")
async def test_camera_removed_while_a_snapshot_fails(
hass: HomeAssistant,
ws_clients: list[Mock],
init_test_integration: MockCamera,
) -> None:
"""Test a camera being removed while a snapshot closes the same session."""
camera = init_test_integration
await camera.async_handle_async_webrtc_offer(OFFER_SDP, "session_1", Mock())
(ws_client,) = ws_clients
async def yield_control() -> None:
"""Let the camera removal run while the snapshot is still failing."""
await asyncio.sleep(0)
ws_client.close.side_effect = yield_control
camera.set_stream_source(None)
async def failing_snapshot() -> None:
with pytest.raises(HomeAssistantError, match="Camera has no stream source"):
await async_get_image(hass, camera.entity_id)
await asyncio.gather(failing_snapshot(), camera.async_remove())
ws_client.close.assert_called_once()
ERR_BINARY_NOT_FOUND = "Could not find go2rtc docker binary"
ERR_CONNECT = "Could not connect to go2rtc instance"
ERR_CONNECT_RETRY = (