jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-26 22:57:17 +00:00
Code Issues Packages Projects Releases Wiki Activity

Remove RTSPtoWebRTC (#144328)

Browse Source
This commit is contained in:
Robert Resch 2025-05-08 17:38:13 +02:00 committed by GitHub
parent 014c5dc764
commit 9e94e94075
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
17 changed files with 0 additions and 977 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.strict-typing
View File

@ -434,7 +434,6 @@ homeassistant.components.roku.*
homeassistant.components.romy.*
homeassistant.components.rpi_power.*
homeassistant.components.rss_feed_template.*
homeassistant.components.rtsp_to_webrtc.*
homeassistant.components.russound_rio.*
homeassistant.components.ruuvi_gateway.*
homeassistant.components.ruuvitag_ble.*

2
CODEOWNERS generated
View File

@ -1307,8 +1307,6 @@ build.json @home-assistant/supervisor
/tests/components/rpi_power/ @shenxn @swetoast
/homeassistant/components/rss_feed_template/ @home-assistant/core
/tests/components/rss_feed_template/ @home-assistant/core
/homeassistant/components/rtsp_to_webrtc/ @allenporter
/tests/components/rtsp_to_webrtc/ @allenporter
/homeassistant/components/ruckus_unleashed/ @lanrat @ms264556 @gabe565
/tests/components/ruckus_unleashed/ @lanrat @ms264556 @gabe565
/homeassistant/components/russound_rio/ @noahhusby

123
homeassistant/components/rtsp_to_webrtc/__init__.py
View File

@ -1,123 +0,0 @@
"""RTSPtoWebRTC integration with an external RTSPToWebRTC Server.
WebRTC uses a direct communication from the client (e.g. a web browser) to a
camera device. Home Assistant acts as the signal path for initial set up,
passing through the client offer and returning a camera answer, then the client
and camera communicate directly.
However, not all cameras natively support WebRTC. This integration is a shim
for camera devices that support RTSP streams only, relying on an external
server RTSPToWebRTC that is a proxy. Home Assistant does not participate in
the offer/answer SDP protocol, other than as a signal path pass through.
Other integrations may use this integration with these steps:
- Check if this integration is loaded
- Call is_supported_stream_source for compatibility
- Call async_offer_for_stream_source to get back an answer for a client offer
"""
from __future__ import annotations
import asyncio
import logging
from rtsp_to_webrtc.client import get_adaptive_client
from rtsp_to_webrtc.exceptions import ClientError, ResponseError
from rtsp_to_webrtc.interface import WebRTCClientInterface
from webrtc_models import RTCIceServer
from homeassistant.components import camera
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import ConfigEntryNotReady, HomeAssistantError
from homeassistant.helpers import issue_registry as ir
from homeassistant.helpers.aiohttp_client import async_get_clientsession
_LOGGER = logging.getLogger(__name__)
DOMAIN = "rtsp_to_webrtc"
DATA_SERVER_URL = "server_url"
DATA_UNSUB = "unsub"
TIMEOUT = 10
CONF_STUN_SERVER = "stun_server"
_DEPRECATED = "deprecated"
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up RTSPtoWebRTC from a config entry."""
hass.data.setdefault(DOMAIN, {})
ir.async_create_issue(
hass,
DOMAIN,
_DEPRECATED,
breaks_in_ha_version="2025.6.0",
is_fixable=False,
severity=ir.IssueSeverity.WARNING,
translation_key=_DEPRECATED,
translation_placeholders={
"go2rtc": "[go2rtc](https://www.home-assistant.io/integrations/go2rtc/)",
},
)
client: WebRTCClientInterface
try:
async with asyncio.timeout(TIMEOUT):
client = await get_adaptive_client(
async_get_clientsession(hass), entry.data[DATA_SERVER_URL]
)
except ResponseError as err:
raise ConfigEntryNotReady from err
except (TimeoutError, ClientError) as err:
raise ConfigEntryNotReady from err
hass.data[DOMAIN][CONF_STUN_SERVER] = entry.options.get(CONF_STUN_SERVER)
if server := entry.options.get(CONF_STUN_SERVER):
@callback
def get_servers() -> list[RTCIceServer]:
return [RTCIceServer(urls=[server])]
entry.async_on_unload(camera.async_register_ice_servers(hass, get_servers))
async def async_offer_for_stream_source(
stream_source: str,
offer_sdp: str,
stream_id: str,
) -> str:
"""Handle the signal path for a WebRTC stream.
This signal path is used to route the offer created by the client to the
proxy server that translates a stream to WebRTC. The communication for
the stream itself happens directly between the client and proxy.
"""
try:
async with asyncio.timeout(TIMEOUT):
return await client.offer_stream_id(stream_id, offer_sdp, stream_source)
except TimeoutError as err:
raise HomeAssistantError("Timeout talking to RTSPtoWebRTC server") from err
except ClientError as err:
raise HomeAssistantError(str(err)) from err
entry.async_on_unload(
camera.async_register_rtsp_to_web_rtc_provider(
hass, DOMAIN, async_offer_for_stream_source
)
)
entry.async_on_unload(entry.add_update_listener(async_reload_entry))
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry."""
if DOMAIN in hass.data:
del hass.data[DOMAIN]
ir.async_delete_issue(hass, DOMAIN, _DEPRECATED)
return True
async def async_reload_entry(hass: HomeAssistant, entry: ConfigEntry) -> None:
"""Reload config entry when options change."""
if hass.data[DOMAIN][CONF_STUN_SERVER] != entry.options.get(CONF_STUN_SERVER):
await hass.config_entries.async_reload(entry.entry_id)

149
homeassistant/components/rtsp_to_webrtc/config_flow.py
View File

@ -1,149 +0,0 @@
"""Config flow for RTSPtoWebRTC."""
from __future__ import annotations
import logging
from typing import Any
from urllib.parse import urlparse
import rtsp_to_webrtc
import voluptuous as vol
from homeassistant.config_entries import (
ConfigEntry,
ConfigFlow,
ConfigFlowResult,
OptionsFlow,
)
from homeassistant.const import CONF_HOST, CONF_PORT
from homeassistant.core import callback
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.service_info.hassio import HassioServiceInfo
from . import CONF_STUN_SERVER, DATA_SERVER_URL, DOMAIN
_LOGGER = logging.getLogger(__name__)
DATA_SCHEMA = vol.Schema({vol.Required(DATA_SERVER_URL): str})
class RTSPToWebRTCConfigFlow(ConfigFlow, domain=DOMAIN):
"""RTSPtoWebRTC config flow."""
_hassio_discovery: dict[str, Any]
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Configure the RTSPtoWebRTC server url."""
if self._async_current_entries():
return self.async_abort(reason="single_instance_allowed")
if user_input is None:
return self.async_show_form(step_id="user", data_schema=DATA_SCHEMA)
url = user_input[DATA_SERVER_URL]
result = urlparse(url)
if not all([result.scheme, result.netloc]):
return self.async_show_form(
step_id="user",
data_schema=DATA_SCHEMA,
errors={DATA_SERVER_URL: "invalid_url"},
)
if error_code := await self._test_connection(url):
return self.async_show_form(
step_id="user",
data_schema=DATA_SCHEMA,
errors={"base": error_code},
)
await self.async_set_unique_id(DOMAIN)
return self.async_create_entry(
title=url,
data={DATA_SERVER_URL: url},
)
async def _test_connection(self, url: str) -> str | None:
"""Test the connection and return any relevant errors."""
client = rtsp_to_webrtc.client.Client(async_get_clientsession(self.hass), url)
try:
await client.heartbeat()
except rtsp_to_webrtc.exceptions.ResponseError as err:
_LOGGER.error("RTSPtoWebRTC server failure: %s", str(err))
return "server_failure"
except rtsp_to_webrtc.exceptions.ClientError as err:
_LOGGER.error("RTSPtoWebRTC communication failure: %s", str(err))
return "server_unreachable"
return None
async def async_step_hassio(
self, discovery_info: HassioServiceInfo
) -> ConfigFlowResult:
"""Prepare configuration for the RTSPtoWebRTC server add-on discovery."""
if self._async_current_entries():
return self.async_abort(reason="single_instance_allowed")
self._hassio_discovery = discovery_info.config
return await self.async_step_hassio_confirm()
async def async_step_hassio_confirm(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Confirm Add-on discovery."""
errors = None
if user_input is not None:
# Validate server connection once user has confirmed
host = self._hassio_discovery[CONF_HOST]
port = self._hassio_discovery[CONF_PORT]
url = f"http://{host}:{port}"
if error_code := await self._test_connection(url):
return self.async_abort(reason=error_code)
if user_input is None or errors:
# Show initial confirmation or errors from server validation
return self.async_show_form(
step_id="hassio_confirm",
description_placeholders={"addon": self._hassio_discovery["addon"]},
errors=errors,
)
return self.async_create_entry(
title=self._hassio_discovery["addon"],
data={DATA_SERVER_URL: url},
)
@staticmethod
@callback
def async_get_options_flow(
config_entry: ConfigEntry,
) -> OptionsFlow:
"""Create an options flow."""
return OptionsFlowHandler()
class OptionsFlowHandler(OptionsFlow):
"""RTSPtoWeb Options flow."""
async def async_step_init(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Manage the options."""
if user_input is not None:
return self.async_create_entry(title="", data=user_input)
return self.async_show_form(
step_id="init",
data_schema=vol.Schema(
{
vol.Optional(
CONF_STUN_SERVER,
description={
"suggested_value": self.config_entry.options.get(
CONF_STUN_SERVER
),
},
): str,
}
),
)

17
homeassistant/components/rtsp_to_webrtc/diagnostics.py
View File

@ -1,17 +0,0 @@
"""Diagnostics support for Nest."""
from __future__ import annotations
from typing import Any
from rtsp_to_webrtc import client
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
async def async_get_config_entry_diagnostics(
hass: HomeAssistant, config_entry: ConfigEntry
) -> dict[str, Any]:
"""Return diagnostics for a config entry."""
return dict(client.get_diagnostics())

11
homeassistant/components/rtsp_to_webrtc/manifest.json
View File

@ -1,11 +0,0 @@
{
"domain": "rtsp_to_webrtc",
"name": "RTSPtoWebRTC",
"codeowners": ["@allenporter"],
"config_flow": true,
"dependencies": ["camera"],
"documentation": "https://www.home-assistant.io/integrations/rtsp_to_webrtc",
"iot_class": "local_push",
"loggers": ["rtsp_to_webrtc"],
"requirements": ["rtsp-to-webrtc==0.5.1"]
}

42
homeassistant/components/rtsp_to_webrtc/strings.json
View File

@ -1,42 +0,0 @@
{
"config": {
"step": {
"user": {
"title": "Configure RTSPtoWebRTC",
"description": "The RTSPtoWebRTC integration requires a server to translate RTSP streams into WebRTC. Enter the URL to the RTSPtoWebRTC server.",
"data": {
"server_url": "RTSPtoWebRTC server URL e.g. https://example.com"
}
},
"hassio_confirm": {
"title": "RTSPtoWebRTC via Home Assistant add-on",
"description": "Do you want to configure Home Assistant to connect to the RTSPtoWebRTC server provided by the add-on: {addon}?"
}
},
"error": {
"invalid_url": "Must be a valid RTSPtoWebRTC server URL e.g. https://example.com",
"server_failure": "RTSPtoWebRTC server returned an error. Check logs for more information.",
"server_unreachable": "Unable to communicate with RTSPtoWebRTC server. Check logs for more information."
},
"abort": {
"single_instance_allowed": "[%key:common::config_flow::abort::single_instance_allowed%]",
"server_failure": "[%key:component::rtsp_to_webrtc::config::error::server_failure%]",
"server_unreachable": "[%key:component::rtsp_to_webrtc::config::error::server_unreachable%]"
}
},
"issues": {
"deprecated": {
"title": "The RTSPtoWebRTC integration is deprecated",
"description": "The RTSPtoWebRTC integration is deprecated and will be removed. Please use the {go2rtc} integration instead, which is enabled by default and provides a better experience. You only need to remove the RTSPtoWebRTC config entry."
}
},
"options": {
"step": {
"init": {
"data": {
"stun_server": "Stun server address (host:port)"
}
}
}
}
}

1
homeassistant/generated/config_flows.py generated
View File

@ -536,7 +536,6 @@ FLOWS = {
"roon",
"rova",
"rpi_power",
"rtsp_to_webrtc",
"ruckus_unleashed",
"russound_rio",
"ruuvi_gateway",

6
homeassistant/generated/integrations.json
View File

@ -5576,12 +5576,6 @@
"config_flow": false,
"iot_class": "local_polling"
},
"rtsp_to_webrtc": {
"name": "RTSPtoWebRTC",
"integration_type": "hub",
"config_flow": true,
"iot_class": "local_push"
},
"ruckus_unleashed": {
"name": "Ruckus",
"integration_type": "hub",

10
mypy.ini generated
View File

@ -4096,16 +4096,6 @@ disallow_untyped_defs = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.rtsp_to_webrtc.*]
check_untyped_defs = true
disallow_incomplete_defs = true
disallow_subclassing_any = true
disallow_untyped_calls = true
disallow_untyped_decorators = true
disallow_untyped_defs = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.russound_rio.*]
check_untyped_defs = true
disallow_incomplete_defs = true

3
requirements_all.txt generated
View File

@ -2675,9 +2675,6 @@ rova==0.4.1
# homeassistant.components.rpi_power
rpi-bad-power==0.1.0
# homeassistant.components.rtsp_to_webrtc
rtsp-to-webrtc==0.5.1
# homeassistant.components.russound_rnet
russound==0.2.0

3
requirements_test_all.txt generated
View File

@ -2170,9 +2170,6 @@ rova==0.4.1
# homeassistant.components.rpi_power
rpi-bad-power==0.1.0
# homeassistant.components.rtsp_to_webrtc
rtsp-to-webrtc==0.5.1
# homeassistant.components.ruuvitag_ble
ruuvitag-ble==0.1.2

1
tests/components/rtsp_to_webrtc/__init__.py
View File

@ -1 +0,0 @@
"""Tests for the RTSPtoWebRTC integration."""

108
tests/components/rtsp_to_webrtc/conftest.py
View File

@ -1,108 +0,0 @@
"""Tests for RTSPtoWebRTC initialization."""
from __future__ import annotations
from collections.abc import AsyncGenerator, Awaitable, Callable
from typing import Any
from unittest.mock import patch
import pytest
import rtsp_to_webrtc
from homeassistant.components import camera
from homeassistant.components.rtsp_to_webrtc import DOMAIN
from homeassistant.config_entries import ConfigEntryState
from homeassistant.core import HomeAssistant
from homeassistant.setup import async_setup_component
from tests.common import MockConfigEntry
STREAM_SOURCE = "rtsp://example.com"
SERVER_URL = "http://127.0.0.1:8083"
CONFIG_ENTRY_DATA = {"server_url": SERVER_URL}
# Typing helpers
type ComponentSetup = Callable[[], Awaitable[None]]
type AsyncYieldFixture[_T] = AsyncGenerator[_T]
@pytest.fixture(autouse=True)
async def webrtc_server() -> None:
"""Patch client library to force usage of RTSPtoWebRTC server."""
with patch(
"rtsp_to_webrtc.client.WebClient.heartbeat",
side_effect=rtsp_to_webrtc.exceptions.ResponseError(),
):
yield
@pytest.fixture
async def mock_camera(hass: HomeAssistant) -> AsyncGenerator[None]:
"""Initialize a demo camera platform."""
assert await async_setup_component(
hass, "camera", {camera.DOMAIN: {"platform": "demo"}}
)
await hass.async_block_till_done()
with (
patch(
"homeassistant.components.demo.camera.Path.read_bytes",
return_value=b"Test",
),
patch(
"homeassistant.components.camera.Camera.stream_source",
return_value=STREAM_SOURCE,
),
):
yield
@pytest.fixture
async def config_entry_data() -> dict[str, Any]:
"""Fixture for MockConfigEntry data."""
return CONFIG_ENTRY_DATA
@pytest.fixture
def config_entry_options() -> dict[str, Any] | None:
"""Fixture to set initial config entry options."""
return None
@pytest.fixture
async def config_entry(
config_entry_data: dict[str, Any],
config_entry_options: dict[str, Any] | None,
) -> MockConfigEntry:
"""Fixture for MockConfigEntry."""
return MockConfigEntry(
domain=DOMAIN, data=config_entry_data, options=config_entry_options
)
@pytest.fixture
async def rtsp_to_webrtc_client() -> None:
"""Fixture for mock rtsp_to_webrtc client."""
with patch("rtsp_to_webrtc.client.Client.heartbeat"):
yield
@pytest.fixture
async def setup_integration(
hass: HomeAssistant, config_entry: MockConfigEntry
) -> AsyncYieldFixture[ComponentSetup]:
"""Fixture for setting up the component."""
config_entry.add_to_hass(hass)
async def func() -> None:
assert await async_setup_component(hass, DOMAIN, {})
await hass.async_block_till_done()
yield func
entries = hass.config_entries.async_entries(DOMAIN)
assert len(entries) == 1
await hass.config_entries.async_unload(entries[0].entry_id)
await hass.async_block_till_done()
assert not hass.data.get(DOMAIN)
assert entries[0].state is ConfigEntryState.NOT_LOADED

274
tests/components/rtsp_to_webrtc/test_config_flow.py
View File

@ -1,274 +0,0 @@
"""Test the RTSPtoWebRTC config flow."""
from __future__ import annotations
from unittest.mock import patch
import rtsp_to_webrtc
from homeassistant import config_entries
from homeassistant.components.rtsp_to_webrtc import DOMAIN
from homeassistant.config_entries import ConfigEntryState
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from homeassistant.helpers.service_info.hassio import HassioServiceInfo
from .conftest import ComponentSetup
from tests.common import MockConfigEntry
async def test_web_full_flow(hass: HomeAssistant) -> None:
"""Check full flow."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result.get("type") is FlowResultType.FORM
assert result.get("step_id") == "user"
assert result.get("data_schema").schema.get("server_url") is str
assert not result.get("errors")
with (
patch("rtsp_to_webrtc.client.Client.heartbeat"),
patch(
"homeassistant.components.rtsp_to_webrtc.async_setup_entry",
return_value=True,
) as mock_setup,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {"server_url": "https://example.com"}
)
assert result.get("type") is FlowResultType.CREATE_ENTRY
assert result.get("title") == "https://example.com"
assert "result" in result
assert result["result"].data == {"server_url": "https://example.com"}
assert len(mock_setup.mock_calls) == 1
async def test_single_config_entry(hass: HomeAssistant) -> None:
"""Test that only a single config entry is allowed."""
old_entry = MockConfigEntry(domain=DOMAIN, data={"example": True})
old_entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result.get("type") is FlowResultType.ABORT
assert result.get("reason") == "single_instance_allowed"
async def test_invalid_url(hass: HomeAssistant) -> None:
"""Check full flow."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result.get("type") is FlowResultType.FORM
assert result.get("step_id") == "user"
assert result.get("data_schema").schema.get("server_url") is str
assert not result.get("errors")
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {"server_url": "not-a-url"}
)
assert result.get("type") is FlowResultType.FORM
assert result.get("step_id") == "user"
assert result.get("errors") == {"server_url": "invalid_url"}
async def test_server_unreachable(hass: HomeAssistant) -> None:
"""Exercise case where the server is unreachable."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result.get("type") is FlowResultType.FORM
assert result.get("step_id") == "user"
assert not result.get("errors")
with patch(
"rtsp_to_webrtc.client.Client.heartbeat",
side_effect=rtsp_to_webrtc.exceptions.ClientError(),
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {"server_url": "https://example.com"}
)
assert result.get("type") is FlowResultType.FORM
assert result.get("step_id") == "user"
assert result.get("errors") == {"base": "server_unreachable"}
async def test_server_failure(hass: HomeAssistant) -> None:
"""Exercise case where server returns a failure."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result.get("type") is FlowResultType.FORM
assert result.get("step_id") == "user"
assert not result.get("errors")
with patch(
"rtsp_to_webrtc.client.Client.heartbeat",
side_effect=rtsp_to_webrtc.exceptions.ResponseError(),
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {"server_url": "https://example.com"}
)
assert result.get("type") is FlowResultType.FORM
assert result.get("step_id") == "user"
assert result.get("errors") == {"base": "server_failure"}
async def test_hassio_discovery(hass: HomeAssistant) -> None:
"""Test supervisor add-on discovery."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
data=HassioServiceInfo(
config={
"addon": "RTSPtoWebRTC",
"host": "fake-server",
"port": 8083,
},
name="RTSPtoWebRTC",
slug="rtsp-to-webrtc",
uuid="1234",
),
context={"source": config_entries.SOURCE_HASSIO},
)
assert result.get("type") is FlowResultType.FORM
assert result.get("step_id") == "hassio_confirm"
assert result.get("description_placeholders") == {"addon": "RTSPtoWebRTC"}
with (
patch("rtsp_to_webrtc.client.Client.heartbeat"),
patch(
"homeassistant.components.rtsp_to_webrtc.async_setup_entry",
return_value=True,
) as mock_setup,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={}
)
await hass.async_block_till_done()
assert result.get("type") is FlowResultType.CREATE_ENTRY
assert result.get("title") == "RTSPtoWebRTC"
assert "result" in result
assert result["result"].data == {"server_url": "http://fake-server:8083"}
assert len(mock_setup.mock_calls) == 1
async def test_hassio_single_config_entry(hass: HomeAssistant) -> None:
"""Test supervisor add-on discovery only allows a single entry."""
old_entry = MockConfigEntry(domain=DOMAIN, data={"example": True})
old_entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN,
data=HassioServiceInfo(
config={
"addon": "RTSPtoWebRTC",
"host": "fake-server",
"port": 8083,
},
name="RTSPtoWebRTC",
slug="rtsp-to-webrtc",
uuid="1234",
),
context={"source": config_entries.SOURCE_HASSIO},
)
assert result.get("type") is FlowResultType.ABORT
assert result.get("reason") == "single_instance_allowed"
async def test_hassio_ignored(hass: HomeAssistant) -> None:
"""Test ignoring superversor add-on discovery."""
old_entry = MockConfigEntry(domain=DOMAIN, source=config_entries.SOURCE_IGNORE)
old_entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN,
data=HassioServiceInfo(
config={
"addon": "RTSPtoWebRTC",
"host": "fake-server",
"port": 8083,
},
name="RTSPtoWebRTC",
slug="rtsp-to-webrtc",
uuid="1234",
),
context={"source": config_entries.SOURCE_HASSIO},
)
assert result.get("type") is FlowResultType.ABORT
assert result.get("reason") == "single_instance_allowed"
async def test_hassio_discovery_server_failure(hass: HomeAssistant) -> None:
"""Test server failure during supvervisor add-on discovery shows an error."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
data=HassioServiceInfo(
config={
"addon": "RTSPtoWebRTC",
"host": "fake-server",
"port": 8083,
},
name="RTSPtoWebRTC",
slug="rtsp-to-webrtc",
uuid="1234",
),
context={"source": config_entries.SOURCE_HASSIO},
)
assert result.get("type") is FlowResultType.FORM
assert result.get("step_id") == "hassio_confirm"
assert not result.get("errors")
with patch(
"rtsp_to_webrtc.client.Client.heartbeat",
side_effect=rtsp_to_webrtc.exceptions.ResponseError(),
):
result = await hass.config_entries.flow.async_configure(result["flow_id"], {})
assert result.get("type") is FlowResultType.ABORT
assert result.get("reason") == "server_failure"
async def test_options_flow(
hass: HomeAssistant,
config_entry: MockConfigEntry,
setup_integration: ComponentSetup,
) -> None:
"""Test setting stun server in options flow."""
with patch(
"homeassistant.components.rtsp_to_webrtc.async_setup_entry",
return_value=True,
):
await setup_integration()
assert config_entry.state is ConfigEntryState.LOADED
assert not config_entry.options
result = await hass.config_entries.options.async_init(config_entry.entry_id)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "init"
data_schema = result["data_schema"].schema
assert set(data_schema) == {"stun_server"}
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={
"stun_server": "example.com:1234",
},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
await hass.async_block_till_done()
assert config_entry.options == {"stun_server": "example.com:1234"}
# Clear the value
result = await hass.config_entries.options.async_init(config_entry.entry_id)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "init"
result = await hass.config_entries.options.async_configure(
result["flow_id"], user_input={}
)
assert result["type"] is FlowResultType.CREATE_ENTRY
await hass.async_block_till_done()
assert config_entry.options == {}

27
tests/components/rtsp_to_webrtc/test_diagnostics.py
View File

@ -1,27 +0,0 @@
"""Test nest diagnostics."""
from typing import Any
from homeassistant.core import HomeAssistant
from .conftest import ComponentSetup
from tests.common import MockConfigEntry
from tests.components.diagnostics import get_diagnostics_for_config_entry
from tests.typing import ClientSessionGenerator
THERMOSTAT_TYPE = "sdm.devices.types.THERMOSTAT"
async def test_entry_diagnostics(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
config_entry: MockConfigEntry,
rtsp_to_webrtc_client: Any,
setup_integration: ComponentSetup,
) -> None:
"""Test config entry diagnostics."""
await setup_integration()
result = await get_diagnostics_for_config_entry(hass, hass_client, config_entry)
assert "webrtc" in result

199
tests/components/rtsp_to_webrtc/test_init.py
View File

@ -1,199 +0,0 @@
"""Tests for RTSPtoWebRTC initialization."""
from __future__ import annotations
import base64
from typing import Any
from unittest.mock import patch
import aiohttp
import pytest
import rtsp_to_webrtc
from homeassistant.components.rtsp_to_webrtc import DOMAIN
from homeassistant.components.websocket_api import TYPE_RESULT
from homeassistant.config_entries import ConfigEntryState
from homeassistant.core import HomeAssistant
from homeassistant.helpers import issue_registry as ir
from homeassistant.setup import async_setup_component
from .conftest import SERVER_URL, STREAM_SOURCE, ComponentSetup
from tests.common import MockConfigEntry
from tests.test_util.aiohttp import AiohttpClientMocker
from tests.typing import WebSocketGenerator
# The webrtc component does not inspect the details of the offer and answer,
# and is only a pass through.
OFFER_SDP = "v=0\r\no=carol 28908764872 28908764872 IN IP4 100.3.6.6\r\n..."
ANSWER_SDP = "v=0\r\no=bob 2890844730 2890844730 IN IP4 host.example.com\r\n..."
@pytest.fixture(autouse=True)
async def setup_homeassistant(hass: HomeAssistant):
"""Set up the homeassistant integration."""
await async_setup_component(hass, "homeassistant", {})
@pytest.mark.usefixtures("rtsp_to_webrtc_client")
async def test_setup_success(
hass: HomeAssistant,
config_entry: MockConfigEntry,
issue_registry: ir.IssueRegistry,
) -> None:
"""Test successful setup and unload."""
config_entry.add_to_hass(hass)
assert await async_setup_component(hass, DOMAIN, {})
await hass.async_block_till_done()
assert issue_registry.async_get_issue(DOMAIN, "deprecated")
entries = hass.config_entries.async_entries(DOMAIN)
assert len(entries) == 1
assert entries[0].state is ConfigEntryState.LOADED
await hass.config_entries.async_unload(entries[0].entry_id)
await hass.async_block_till_done()
assert not hass.data.get(DOMAIN)
assert entries[0].state is ConfigEntryState.NOT_LOADED
assert not issue_registry.async_get_issue(DOMAIN, "deprecated")
@pytest.mark.parametrize("config_entry_data", [{}])
async def test_invalid_config_entry(
hass: HomeAssistant, rtsp_to_webrtc_client: Any, setup_integration: ComponentSetup
) -> None:
"""Test a config entry with missing required fields."""
await setup_integration()
entries = hass.config_entries.async_entries(DOMAIN)
assert len(entries) == 1
assert entries[0].state is ConfigEntryState.SETUP_ERROR
async def test_setup_server_failure(
hass: HomeAssistant, setup_integration: ComponentSetup
) -> None:
"""Test server responds with a failure on startup."""
with patch(
"rtsp_to_webrtc.client.Client.heartbeat",
side_effect=rtsp_to_webrtc.exceptions.ResponseError(),
):
await setup_integration()
entries = hass.config_entries.async_entries(DOMAIN)
assert len(entries) == 1
assert entries[0].state is ConfigEntryState.SETUP_RETRY
async def test_setup_communication_failure(
hass: HomeAssistant, setup_integration: ComponentSetup
) -> None:
"""Test unable to talk to server on startup."""
with patch(
"rtsp_to_webrtc.client.Client.heartbeat",
side_effect=rtsp_to_webrtc.exceptions.ClientError(),
):
await setup_integration()
entries = hass.config_entries.async_entries(DOMAIN)
assert len(entries) == 1
assert entries[0].state is ConfigEntryState.SETUP_RETRY
@pytest.mark.usefixtures("mock_camera", "rtsp_to_webrtc_client")
async def test_offer_for_stream_source(
hass: HomeAssistant,
aioclient_mock: AiohttpClientMocker,
hass_ws_client: WebSocketGenerator,
setup_integration: ComponentSetup,
) -> None:
"""Test successful response from RTSPtoWebRTC server."""
await setup_integration()
aioclient_mock.post(
f"{SERVER_URL}/stream",
json={"sdp64": base64.b64encode(ANSWER_SDP.encode("utf-8")).decode("utf-8")},
)
client = await hass_ws_client(hass)
await client.send_json_auto_id(
{
"type": "camera/webrtc/offer",
"entity_id": "camera.demo_camera",
"offer": OFFER_SDP,
}
)
response = await client.receive_json()
assert response["type"] == TYPE_RESULT
assert response["success"]
subscription_id = response["id"]
# Session id
response = await client.receive_json()
assert response["id"] == subscription_id
assert response["type"] == "event"
assert response["event"]["type"] == "session"
# Answer
response = await client.receive_json()
assert response["id"] == subscription_id
assert response["type"] == "event"
assert response["event"] == {
"type": "answer",
"answer": ANSWER_SDP,
}
# Validate request parameters were sent correctly
assert len(aioclient_mock.mock_calls) == 1
assert aioclient_mock.mock_calls[-1][2] == {
"sdp64": base64.b64encode(OFFER_SDP.encode("utf-8")).decode("utf-8"),
"url": STREAM_SOURCE,
}
@pytest.mark.usefixtures("mock_camera", "rtsp_to_webrtc_client")
async def test_offer_failure(
hass: HomeAssistant,
aioclient_mock: AiohttpClientMocker,
hass_ws_client: WebSocketGenerator,
setup_integration: ComponentSetup,
) -> None:
"""Test a transient failure talking to RTSPtoWebRTC server."""
await setup_integration()
aioclient_mock.post(
f"{SERVER_URL}/stream",
exc=aiohttp.ClientError,
)
client = await hass_ws_client(hass)
await client.send_json_auto_id(
{
"type": "camera/webrtc/offer",
"entity_id": "camera.demo_camera",
"offer": OFFER_SDP,
}
)
response = await client.receive_json()
assert response["type"] == TYPE_RESULT
assert response["success"]
subscription_id = response["id"]
# Session id
response = await client.receive_json()
assert response["id"] == subscription_id
assert response["type"] == "event"
assert response["event"]["type"] == "session"
# Answer
response = await client.receive_json()
assert response["id"] == subscription_id
assert response["type"] == "event"
assert response["event"] == {
"type": "error",
"code": "webrtc_offer_failed",
"message": "RTSPtoWebRTC server communication failure: ",
}