mirror of
https://github.com/firestar5683/StarPilot.git
synced 2026-09-15 03:53:58 +08:00
firestar5683
50 lines
1.5 KiB
Python
50 lines
1.5 KiB
Python
import json
|
|
|
|
import pytest
|
|
|
|
pytest.importorskip("libdatachannel", reason="the upstream WebRTC backend requires Python 3.12")
|
|
|
|
from openpilot.system.webrtc.webrtcd import ServerState, handle_get_schema, handle_get_stream, handle_post_notify, on_shutdown
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_schema():
|
|
status, body, content_type = await handle_get_schema(ServerState(), "carState")
|
|
|
|
assert status == 200
|
|
assert content_type.startswith("application/json")
|
|
assert "carState" in json.loads(body)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_schema_rejects_unknown_service():
|
|
with pytest.raises(AssertionError, match="Invalid service name"):
|
|
await handle_get_schema(ServerState(), "notARealService")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_stream_rejects_non_json_content_type():
|
|
response = await handle_get_stream(ServerState(), b"{}", "text/plain")
|
|
|
|
assert response == (415, b'{"error": "unsupported media type"}', "application/json; charset=utf-8")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_notify_and_shutdown_active_stream(mocker):
|
|
state = ServerState()
|
|
session = mocker.MagicMock()
|
|
session.stop = mocker.AsyncMock()
|
|
state.streams["test"] = session
|
|
|
|
status, body, content_type = await handle_post_notify(state, {"type": "ping"})
|
|
|
|
assert (status, body) == (200, b"OK")
|
|
assert content_type.startswith("text/plain")
|
|
channel = session.stream.get_messaging_channel.return_value
|
|
channel.send.assert_called_once_with(json.dumps({"type": "ping"}))
|
|
|
|
await on_shutdown(state)
|
|
|
|
session.stop.assert_awaited_once()
|
|
assert state.streams == {}
|