mirror of
https://github.com/home-assistant/core.git
synced 2025-12-20 19:09:45 +00:00
251 lines
7.5 KiB
Python
251 lines
7.5 KiB
Python
"""Test the WebRTC integration."""
|
|
|
|
from webrtc_models import RTCIceServer
|
|
|
|
from homeassistant.components.web_rtc import (
|
|
async_get_ice_servers,
|
|
async_register_ice_servers,
|
|
)
|
|
from homeassistant.core import HomeAssistant, callback
|
|
from homeassistant.core_config import async_process_ha_core_config
|
|
from homeassistant.setup import async_setup_component
|
|
|
|
from tests.typing import WebSocketGenerator
|
|
|
|
|
|
async def test_async_setup(hass: HomeAssistant) -> None:
|
|
"""Test setting up the web_rtc integration."""
|
|
assert await async_setup_component(hass, "web_rtc", {})
|
|
await hass.async_block_till_done()
|
|
|
|
# Verify default ICE servers are registered
|
|
ice_servers = async_get_ice_servers(hass)
|
|
assert len(ice_servers) == 1
|
|
assert ice_servers[0].urls == [
|
|
"stun:stun.home-assistant.io:3478",
|
|
"stun:stun.home-assistant.io:80",
|
|
]
|
|
|
|
|
|
async def test_async_setup_custom_ice_servers_core(hass: HomeAssistant) -> None:
|
|
"""Test setting up web_rtc with custom ICE servers in config."""
|
|
await async_process_ha_core_config(
|
|
hass,
|
|
{"webrtc": {"ice_servers": [{"url": "stun:custom_stun_server:3478"}]}},
|
|
)
|
|
|
|
assert await async_setup_component(hass, "web_rtc", {})
|
|
await hass.async_block_till_done()
|
|
|
|
ice_servers = async_get_ice_servers(hass)
|
|
assert len(ice_servers) == 1
|
|
assert ice_servers[0].urls == ["stun:custom_stun_server:3478"]
|
|
|
|
|
|
async def test_async_setup_custom_ice_servers_integration(hass: HomeAssistant) -> None:
|
|
"""Test setting up web_rtc with custom ICE servers in config."""
|
|
assert await async_setup_component(
|
|
hass,
|
|
"web_rtc",
|
|
{
|
|
"web_rtc": {
|
|
"ice_servers": [
|
|
{"url": "stun:custom_stun_server:3478"},
|
|
{
|
|
"url": "stun:custom_stun_server:3478",
|
|
"credential": "mock-credential",
|
|
},
|
|
{
|
|
"url": "stun:custom_stun_server:3478",
|
|
"username": "mock-username",
|
|
},
|
|
{
|
|
"url": "stun:custom_stun_server:3478",
|
|
"credential": "mock-credential",
|
|
"username": "mock-username",
|
|
},
|
|
]
|
|
}
|
|
},
|
|
)
|
|
await hass.async_block_till_done()
|
|
|
|
ice_servers = async_get_ice_servers(hass)
|
|
assert ice_servers == [
|
|
RTCIceServer(
|
|
urls=["stun:custom_stun_server:3478"],
|
|
),
|
|
RTCIceServer(
|
|
urls=["stun:custom_stun_server:3478"],
|
|
credential="mock-credential",
|
|
),
|
|
RTCIceServer(
|
|
urls=["stun:custom_stun_server:3478"],
|
|
username="mock-username",
|
|
),
|
|
RTCIceServer(
|
|
urls=["stun:custom_stun_server:3478"],
|
|
username="mock-username",
|
|
credential="mock-credential",
|
|
),
|
|
]
|
|
|
|
|
|
async def test_async_setup_custom_ice_servers_core_and_integration(
|
|
hass: HomeAssistant,
|
|
) -> None:
|
|
"""Test setting up web_rtc with custom ICE servers in config."""
|
|
await async_process_ha_core_config(
|
|
hass,
|
|
{"webrtc": {"ice_servers": [{"url": "stun:custom_stun_server_core:3478"}]}},
|
|
)
|
|
|
|
assert await async_setup_component(
|
|
hass,
|
|
"web_rtc",
|
|
{
|
|
"web_rtc": {
|
|
"ice_servers": [{"url": "stun:custom_stun_server_integration:3478"}]
|
|
}
|
|
},
|
|
)
|
|
await hass.async_block_till_done()
|
|
|
|
ice_servers = async_get_ice_servers(hass)
|
|
assert ice_servers == [
|
|
RTCIceServer(
|
|
urls=["stun:custom_stun_server_core:3478"],
|
|
),
|
|
RTCIceServer(
|
|
urls=["stun:custom_stun_server_integration:3478"],
|
|
),
|
|
]
|
|
|
|
|
|
async def test_async_register_ice_servers(hass: HomeAssistant) -> None:
|
|
"""Test registering ICE servers."""
|
|
assert await async_setup_component(hass, "web_rtc", {})
|
|
await hass.async_block_till_done()
|
|
default_servers = async_get_ice_servers(hass)
|
|
|
|
called = 0
|
|
|
|
@callback
|
|
def get_ice_servers() -> list[RTCIceServer]:
|
|
nonlocal called
|
|
called += 1
|
|
return [
|
|
RTCIceServer(urls="stun:example.com"),
|
|
RTCIceServer(urls="turn:example.com"),
|
|
]
|
|
|
|
unregister = async_register_ice_servers(hass, get_ice_servers)
|
|
assert called == 0
|
|
|
|
# Getting ice servers should call the callback
|
|
ice_servers = async_get_ice_servers(hass)
|
|
assert called == 1
|
|
assert ice_servers == [
|
|
*default_servers,
|
|
RTCIceServer(urls="stun:example.com"),
|
|
RTCIceServer(urls="turn:example.com"),
|
|
]
|
|
|
|
# Unregister and verify servers are removed
|
|
unregister()
|
|
ice_servers = async_get_ice_servers(hass)
|
|
assert ice_servers == default_servers
|
|
|
|
|
|
async def test_multiple_ice_server_registrations(hass: HomeAssistant) -> None:
|
|
"""Test registering multiple ICE server providers."""
|
|
assert await async_setup_component(hass, "web_rtc", {})
|
|
await hass.async_block_till_done()
|
|
default_servers = async_get_ice_servers(hass)
|
|
|
|
@callback
|
|
def get_ice_servers_1() -> list[RTCIceServer]:
|
|
return [RTCIceServer(urls="stun:server1.com")]
|
|
|
|
@callback
|
|
def get_ice_servers_2() -> list[RTCIceServer]:
|
|
return [
|
|
RTCIceServer(
|
|
urls=["stun:server2.com", "turn:server2.com"],
|
|
username="user",
|
|
credential="pass",
|
|
)
|
|
]
|
|
|
|
unregister_1 = async_register_ice_servers(hass, get_ice_servers_1)
|
|
unregister_2 = async_register_ice_servers(hass, get_ice_servers_2)
|
|
|
|
ice_servers = async_get_ice_servers(hass)
|
|
assert ice_servers == [
|
|
*default_servers,
|
|
RTCIceServer(urls="stun:server1.com"),
|
|
RTCIceServer(
|
|
urls=["stun:server2.com", "turn:server2.com"],
|
|
username="user",
|
|
credential="pass",
|
|
),
|
|
]
|
|
|
|
# Unregister first provider
|
|
unregister_1()
|
|
ice_servers = async_get_ice_servers(hass)
|
|
assert ice_servers == [
|
|
*default_servers,
|
|
RTCIceServer(
|
|
urls=["stun:server2.com", "turn:server2.com"],
|
|
username="user",
|
|
credential="pass",
|
|
),
|
|
]
|
|
|
|
# Unregister second provider
|
|
unregister_2()
|
|
ice_servers = async_get_ice_servers(hass)
|
|
assert ice_servers == default_servers
|
|
|
|
|
|
async def test_ws_ice_servers_with_registered_servers(
|
|
hass: HomeAssistant, hass_ws_client: WebSocketGenerator
|
|
) -> None:
|
|
"""Test WebSocket ICE servers endpoint with registered servers."""
|
|
assert await async_setup_component(hass, "web_rtc", {})
|
|
await hass.async_block_till_done()
|
|
|
|
@callback
|
|
def get_ice_server() -> list[RTCIceServer]:
|
|
return [
|
|
RTCIceServer(
|
|
urls=["stun:example2.com", "turn:example2.com"],
|
|
username="user",
|
|
credential="pass",
|
|
)
|
|
]
|
|
|
|
async_register_ice_servers(hass, get_ice_server)
|
|
|
|
client = await hass_ws_client(hass)
|
|
await client.send_json_auto_id({"type": "web_rtc/ice_servers"})
|
|
msg = await client.receive_json()
|
|
|
|
# Assert WebSocket response includes registered ICE servers
|
|
assert msg["type"] == "result"
|
|
assert msg["success"]
|
|
assert msg["result"] == [
|
|
{
|
|
"urls": [
|
|
"stun:stun.home-assistant.io:3478",
|
|
"stun:stun.home-assistant.io:80",
|
|
]
|
|
},
|
|
{
|
|
"urls": ["stun:example2.com", "turn:example2.com"],
|
|
"username": "user",
|
|
"credential": "pass",
|
|
},
|
|
]
|