music-assistant-server

1"""Tests for remote access feature."""
2
3import asyncio
4import base64
5import hashlib
6import json
7import logging
8from collections.abc import AsyncIterator, Callable, Coroutine
9from types import SimpleNamespace
10from typing import Any, cast
11from unittest.mock import AsyncMock, MagicMock, Mock, patch
12from urllib.parse import urlparse
13
14import aiohttp
15import pytest
16from aiolibdatachannel import DataChannel, IceServer, LogLevel, PeerConnection, RTCConfiguration
17from cryptography.hazmat.primitives import serialization
18
19from music_assistant.constants import VERBOSE_LOG_LEVEL
20from music_assistant.controllers.webserver.remote_access import (
21    STARTUP_DELAY,
22    TASK_ID_START_GATEWAY,
23    RemoteAccessInfo,
24    RemoteAccessManager,
25)
26from music_assistant.controllers.webserver.remote_access.gateway import (
27    DATA_CHANNEL_CHUNK_SIZE,
28    WebRTCGateway,
29    WebRTCSession,
30    _is_usable_ice_url,
31)
32from music_assistant.helpers.webrtc_certificate import (
33    _generate_certificate,
34    _remote_id_from_certificate,
35)
36
37
38async def test_remote_id_from_certificate() -> None:
39    """Test deterministic remote ID generation from a certificate."""
40    _, cert = _generate_certificate()
41    remote_id = _remote_id_from_certificate(cert)
42
43    # Should be base32 encoded, uppercase, no padding
44    assert remote_id.isalnum()
45    assert remote_id == remote_id.upper()
46    # 128 bits = 16 bytes -> 26 base32 chars (without padding)
47    assert len(remote_id) == 26
48    # deterministic: the same certificate always yields the same id
49    assert _remote_id_from_certificate(cert) == remote_id
50
51
52async def test_remote_access_info_dataclass() -> None:
53    """Test RemoteAccessInfo dataclass."""
54    info = RemoteAccessInfo(
55        enabled=True,
56        running=True,
57        connected=False,
58        remote_id="VVPN3TLP34YMGIZDINCEKQKSIR",
59        using_ha_cloud=False,
60        signaling_url="wss://signaling.music-assistant.io/ws",
61    )
62
63    assert info.enabled is True
64    assert info.running is True
65    assert info.connected is False
66    assert info.remote_id == "VVPN3TLP34YMGIZDINCEKQKSIR"
67    assert info.using_ha_cloud is False
68    assert info.signaling_url == "wss://signaling.music-assistant.io/ws"
69
70
71def _create_remote_access_manager() -> RemoteAccessManager:
72    """Create an enabled remote access manager with mocked dependencies."""
73    webserver = Mock()
74    webserver.mass = Mock()
75    webserver.logger = Mock()
76    manager = RemoteAccessManager(webserver)
77    manager._enabled = True
78    return manager
79
80
81def test_remote_access_debounces_restart_without_dropping_gateway() -> None:
82    """Keep the active gateway connected while a replacement is being debounced."""
83    manager = _create_remote_access_manager()
84    gateway = Mock()
85    gateway.stop = AsyncMock()
86    manager.gateway = gateway
87
88    manager._schedule_start()
89
90    gateway.stop.assert_not_awaited()
91    cast("Mock", manager.mass.cancel_timer).assert_called_once_with(TASK_ID_START_GATEWAY)
92    cast("Mock", manager.mass.call_later).assert_called_once_with(
93        STARTUP_DELAY,
94        manager._restart_gateway,
95        task_id=TASK_ID_START_GATEWAY,
96    )
97
98
99async def test_remote_access_stop_cancels_pending_restart() -> None:
100    """Cancel a restart after its debounce timer has promoted it to a task."""
101    manager = _create_remote_access_manager()
102    gateway = Mock()
103    gateway.stop = AsyncMock()
104    manager.gateway = gateway
105
106    await manager.stop()
107
108    cast("Mock", manager.mass.cancel_timer).assert_called_once_with(TASK_ID_START_GATEWAY)
109    cast("Mock", manager.mass.cancel_task).assert_called_once_with(TASK_ID_START_GATEWAY)
110    gateway.stop.assert_awaited_once_with()
111    assert manager.gateway is None
112
113
114async def test_remote_access_serializes_concurrent_starts() -> None:
115    """Create only one gateway when immediate start requests overlap."""
116    manager = _create_remote_access_manager()
117    start_entered = asyncio.Event()
118    allow_start = asyncio.Event()
119    gateway = Mock()
120    gateway.is_running = True
121
122    async def start_gateway() -> None:
123        start_entered.set()
124        await allow_start.wait()
125        manager.gateway = gateway
126
127    with patch.object(
128        manager,
129        "_start_gateway_locked",
130        new=AsyncMock(side_effect=start_gateway),
131    ) as start_gateway_locked:
132        first_start = asyncio.create_task(manager._start_gateway())
133        await start_entered.wait()
134        second_start = asyncio.create_task(manager._start_gateway())
135        await asyncio.sleep(0)
136        allow_start.set()
137        await asyncio.gather(first_start, second_start)
138
139    start_gateway_locked.assert_awaited_once()
140
141
142async def test_remote_access_provider_update_storm_schedules_one_restart() -> None:
143    """Schedule one restart when concurrent provider updates detect the same mode change."""
144    manager = _create_remote_access_manager()
145    ice_servers = [{"urls": "turn:example.com"}]
146
147    with (
148        patch.object(
149            manager,
150            "_get_ha_cloud_status",
151            new=AsyncMock(return_value=(True, ice_servers)),
152        ),
153        patch.object(manager, "_schedule_start") as schedule_start,
154    ):
155        await asyncio.gather(*(manager._on_providers_updated(Mock()) for _ in range(10)))
156
157    schedule_start.assert_called_once_with()
158    assert manager._target_using_ha_cloud is True
159
160
161async def test_remote_access_skips_restart_after_mode_flap_settles() -> None:
162    """Keep the current gateway when a fresh status check matches its active mode."""
163    manager = _create_remote_access_manager()
164    gateway = Mock()
165    gateway.is_running = True
166    manager.gateway = gateway
167    manager._target_using_ha_cloud = True
168
169    with (
170        patch.object(
171            manager,
172            "_get_ha_cloud_status",
173            new=AsyncMock(return_value=(False, None)),
174        ),
175        patch.object(
176            manager,
177            "_stop_gateway_locked",
178            new=AsyncMock(),
179        ) as stop_gateway,
180        patch.object(
181            manager,
182            "_start_gateway_locked",
183            new=AsyncMock(),
184        ) as start_gateway,
185    ):
186        await manager._restart_gateway()
187
188    stop_gateway.assert_not_awaited()
189    start_gateway.assert_not_awaited()
190    assert manager._target_using_ha_cloud is False
191
192
193async def test_remote_access_gateway_uses_internal_sendspin_url(
194    cert_pems: tuple[str, str],
195) -> None:
196    """Bridge the Sendspin data channel to the locally reachable Sendspin server."""
197    internal_url = "ws://127.0.0.1:8927/sendspin"
198    manager = _create_remote_access_manager()
199    manager._remote_id = "TEST-REMOTE-ID"
200    cast("Mock", manager.webserver).internal_base_url = "http://127.0.0.1:8095"
201    cast("Mock", manager.webserver).internal_sendspin_url = internal_url
202
203    with (
204        patch(
205            "music_assistant.controllers.webserver.remote_access"
206            ".get_or_create_webrtc_certificate_pems",
207            return_value=cert_pems,
208        ),
209        patch.object(manager, "_get_ha_cloud_status", new=AsyncMock(return_value=(False, None))),
210        patch.object(WebRTCGateway, "start", new=AsyncMock()),
211    ):
212        await manager._start_gateway_locked()
213
214    assert manager.gateway is not None
215    assert manager.gateway.sendspin_url == internal_url
216
217
218@pytest.mark.parametrize(
219    ("internal_base_url", "expected"),
220    [
221        ("http://127.0.0.1:8095", "ws://127.0.0.1:8095/ws"),
222        ("https://127.0.0.1:8095", "wss://127.0.0.1:8095/ws"),
223    ],
224)
225async def test_remote_access_gateway_uses_internal_base_url(
226    cert_pems: tuple[str, str],
227    internal_base_url: str,
228    expected: str,
229) -> None:
230    """Bridge the API data channel to the locally reachable webserver."""
231    manager = _create_remote_access_manager()
232    manager._remote_id = "TEST-REMOTE-ID"
233    # an external base URL must never be dialed back into this host
234    cast("Mock", manager.mass).webserver.base_url = "https://ma.example.com"
235    cast("Mock", manager.webserver).internal_base_url = internal_base_url
236
237    with (
238        patch(
239            "music_assistant.controllers.webserver.remote_access"
240            ".get_or_create_webrtc_certificate_pems",
241            return_value=cert_pems,
242        ),
243        patch.object(manager, "_get_ha_cloud_status", new=AsyncMock(return_value=(False, None))),
244        patch.object(WebRTCGateway, "start", new=AsyncMock()),
245    ):
246        await manager._start_gateway_locked()
247
248    assert manager.gateway is not None
249    assert manager.gateway.local_ws_url == expected
250
251
252async def test_webrtc_gateway_initialization(cert_pems: tuple[str, str]) -> None:
253    """Test WebRTCGateway initializes correctly."""
254    cert_pem, key_pem = cert_pems
255    mock_session = Mock()
256    gateway = WebRTCGateway(
257        http_session=mock_session,
258        remote_id="TEST-REMOTE-ID",
259        cert_pem=cert_pem,
260        key_pem=key_pem,
261        signaling_url="wss://test.example.com/ws",
262        local_ws_url="ws://localhost:8095/ws",
263    )
264
265    assert gateway._remote_id == "TEST-REMOTE-ID"
266    assert gateway.signaling_url == "wss://test.example.com/ws"
267    assert gateway.local_ws_url == "ws://localhost:8095/ws"
268    assert gateway.is_running is False
269    assert gateway.is_connected is False
270    assert len(gateway.ice_servers) > 0
271
272
273async def test_webrtc_gateway_custom_ice_servers(cert_pems: tuple[str, str]) -> None:
274    """Test WebRTCGateway accepts custom ICE servers."""
275    cert_pem, key_pem = cert_pems
276    mock_session = Mock()
277    custom_ice_servers = [
278        {"urls": "stun:custom.stun.server:3478"},
279        {"urls": "turn:custom.turn.server:3478", "username": "user", "credential": "pass"},
280    ]
281
282    gateway = WebRTCGateway(
283        http_session=mock_session,
284        remote_id="TEST-REMOTE-ID",
285        cert_pem=cert_pem,
286        key_pem=key_pem,
287        ice_servers=custom_ice_servers,
288    )
289
290    assert gateway.ice_servers == custom_ice_servers
291
292
293async def test_webrtc_gateway_local_ice_servers_skip_non_udp_turn(
294    cert_pems: tuple[str, str],
295) -> None:
296    """Test the local peer connection only gets ICE servers libjuice can use."""
297    cert_pem, key_pem = cert_pems
298    cred = {"username": "u", "credential": "p"}
299    # shape of what HA Cloud hands us: one UDP TURN url plus TCP/TLS variants
300    ha_cloud_ice_servers = [
301        {"urls": "stun:stun.cloudflare.com:3478"},
302        {"urls": "turn:turn.cloudflare.com:3478?transport=udp", **cred},
303        {"urls": "turn:turn.cloudflare.com:53", **cred},
304        {"urls": "turn:turn.cloudflare.com:3478?transport=tcp", **cred},
305        {"urls": "turns:turn.cloudflare.com:5349?transport=tcp", **cred},
306        {"urls": "turns:turn.cloudflare.com:443?transport=tcp", **cred},
307    ]
308
309    gateway = WebRTCGateway(
310        http_session=Mock(),
311        remote_id="TEST-REMOTE-ID",
312        cert_pem=cert_pem,
313        key_pem=key_pem,
314        ice_servers=ha_cloud_ice_servers,
315    )
316
317    assert [server.url for server in gateway._build_ice_servers(ha_cloud_ice_servers)] == [
318        "stun:stun.cloudflare.com:3478",
319        "turn:turn.cloudflare.com:3478?transport=udp",
320        "turn:turn.cloudflare.com:53",
321    ]
322    # remote clients still get the unfiltered list, since browsers do support TCP/TLS TURN
323    assert gateway.ice_servers == ha_cloud_ice_servers
324
325
326@pytest.mark.parametrize(
327    ("url", "usable"),
328    [
329        ("stun:stun.example.com:3478", True),
330        ("turn:turn.example.com:3478", True),
331        ("turn:turn.example.com:3478?transport=udp", True),
332        # the transport parameter wins over the scheme, matching rtc::IceServer
333        ("turns:turn.example.com:5349?transport=udp", True),
334        ("turn:turn.example.com:3478?transport=tcp", False),
335        ("turns:turn.example.com:5349", False),
336        ("turn:turn.example.com:3478?transport=tls", False),
337        ("https://turn.example.com", False),
338    ],
339)
340def test_is_usable_ice_url(url: str, usable: bool) -> None:
341    """Test only ICE server urls libjuice can actually use are kept."""
342    assert _is_usable_ice_url(url) is usable
343
344
345async def test_webrtc_gateway_start_stop(cert_pems: tuple[str, str]) -> None:
346    """Test WebRTCGateway start and stop."""
347    cert_pem, key_pem = cert_pems
348    mock_session = Mock()
349    gateway = WebRTCGateway(
350        http_session=mock_session,
351        remote_id="TEST-REMOTE-ID",
352        cert_pem=cert_pem,
353        key_pem=key_pem,
354    )
355
356    # Mock the _run method to avoid actual connection
357    with patch.object(gateway, "_run", new_callable=AsyncMock):
358        await gateway.start()
359        assert gateway.is_running is True
360        assert gateway._run_task is not None
361
362        await gateway.stop()
363        assert gateway.is_running is False
364
365
366@pytest.mark.parametrize(
367    ("logger_level", "expected_rtc_level"),
368    [
369        (VERBOSE_LOG_LEVEL, LogLevel.VERBOSE),
370        (logging.DEBUG, LogLevel.WARNING),
371        (logging.INFO, LogLevel.ERROR),
372        (logging.WARNING, LogLevel.ERROR),
373    ],
374)
375async def test_webrtc_gateway_native_log_level(
376    cert_pems: tuple[str, str], logger_level: int, expected_rtc_level: LogLevel
377) -> None:
378    """Test native libdatachannel logging is capped at ERROR unless debugging."""
379    cert_pem, key_pem = cert_pems
380    gateway = WebRTCGateway(
381        http_session=Mock(),
382        remote_id="TEST-REMOTE-ID",
383        cert_pem=cert_pem,
384        key_pem=key_pem,
385    )
386    gateway.logger = logging.getLogger("test_webrtc_native_log_level")
387    gateway.logger.setLevel(logger_level)
388
389    with (
390        patch.object(gateway, "_run", new_callable=AsyncMock),
391        patch(
392            "music_assistant.controllers.webserver.remote_access.gateway.install_python_logger"
393        ) as install_logger,
394    ):
395        await gateway.start()
396        await gateway.stop()
397
398    install_logger.assert_called_once_with(gateway.logger, level=expected_rtc_level)
399    assert not gateway.logger.filters
400
401
402async def test_webrtc_gateway_drops_benign_turn_warning_at_debug(
403    cert_pems: tuple[str, str], caplog: pytest.LogCaptureFixture
404) -> None:
405    """Test the benign Cloudflare CreatePermission warning is dropped at DEBUG level."""
406    cert_pem, key_pem = cert_pems
407    gateway = WebRTCGateway(
408        http_session=Mock(),
409        remote_id="TEST-REMOTE-ID",
410        cert_pem=cert_pem,
411        key_pem=key_pem,
412    )
413    gateway.logger = logging.getLogger("test_webrtc_benign_turn_warning")
414    gateway.logger.setLevel(logging.DEBUG)
415
416    with (
417        patch.object(gateway, "_run", new_callable=AsyncMock),
418        patch("music_assistant.controllers.webserver.remote_access.gateway.install_python_logger"),
419    ):
420        await gateway.start()
421        with caplog.at_level(logging.DEBUG, logger="test_webrtc_benign_turn_warning"):
422            gateway.logger.warning(
423                "rtc::impl::IceTransport::LogCallback@390: "
424                "juice: Got TURN CreatePermission error response, code=0"
425            )
426            gateway.logger.warning("juice: Lost connectivity")
427        await gateway.stop()
428
429    messages = [record.getMessage() for record in caplog.records if "juice" in record.getMessage()]
430    assert messages == ["juice: Lost connectivity"]
431    # stop() must remove the filter again
432    assert not gateway.logger.filters
433
434
435async def test_webrtc_gateway_handle_registration_message(cert_pems: tuple[str, str]) -> None:
436    """Test WebRTCGateway handles registration confirmation."""
437    cert_pem, key_pem = cert_pems
438    mock_session = Mock()
439    gateway = WebRTCGateway(
440        http_session=mock_session,
441        remote_id="TEST-REMOTE-ID",
442        cert_pem=cert_pem,
443        key_pem=key_pem,
444    )
445
446    # Mock signaling WebSocket
447    gateway._signaling_ws = Mock()
448
449    message = {"type": "registered", "remoteId": "TEST-REMOTE-ID"}
450    await gateway._handle_signaling_message(message)
451
452    # Should log but not crash
453
454
455async def test_webrtc_gateway_handle_error_message(cert_pems: tuple[str, str]) -> None:
456    """Test WebRTCGateway handles error messages."""
457    cert_pem, key_pem = cert_pems
458    mock_session = Mock()
459    gateway = WebRTCGateway(
460        http_session=mock_session,
461        remote_id="TEST-REMOTE-ID",
462        cert_pem=cert_pem,
463        key_pem=key_pem,
464    )
465
466    message = {"type": "error", "message": "Test error"}
467    # Should log error but not crash
468    await gateway._handle_signaling_message(message)
469
470
471async def test_webrtc_gateway_create_session(cert_pems: tuple[str, str]) -> None:
472    """Test WebRTCGateway creates sessions for clients."""
473    cert_pem, key_pem = cert_pems
474    mock_session = Mock()
475    gateway = WebRTCGateway(
476        http_session=mock_session,
477        remote_id="TEST-REMOTE-ID",
478        cert_pem=cert_pem,
479        key_pem=key_pem,
480    )
481
482    session_id = "test-session-123"
483    with patch.object(gateway, "_get_fresh_ice_servers", AsyncMock(return_value=[])):
484        await gateway._create_session(session_id)
485
486        assert session_id in gateway.sessions
487        assert gateway.sessions[session_id].session_id == session_id
488        assert gateway.sessions[session_id].pc is not None
489
490        # Cleanup
491        await gateway._close_session(session_id)
492
493
494async def test_webrtc_gateway_close_session(cert_pems: tuple[str, str]) -> None:
495    """Test WebRTCGateway closes sessions properly."""
496    cert_pem, key_pem = cert_pems
497    mock_session = Mock()
498    gateway = WebRTCGateway(
499        http_session=mock_session,
500        remote_id="TEST-REMOTE-ID",
501        cert_pem=cert_pem,
502        key_pem=key_pem,
503    )
504
505    session_id = "test-session-456"
506    with patch.object(gateway, "_get_fresh_ice_servers", AsyncMock(return_value=[])):
507        await gateway._create_session(session_id)
508        assert session_id in gateway.sessions
509
510        await gateway._close_session(session_id)
511        assert session_id not in gateway.sessions
512
513
514async def test_webrtc_gateway_close_nonexistent_session(cert_pems: tuple[str, str]) -> None:
515    """Test WebRTCGateway handles closing non-existent session gracefully."""
516    cert_pem, key_pem = cert_pems
517    mock_session = Mock()
518    gateway = WebRTCGateway(
519        http_session=mock_session,
520        remote_id="TEST-REMOTE-ID",
521        cert_pem=cert_pem,
522        key_pem=key_pem,
523    )
524
525    # Should not raise an error
526    await gateway._close_session("nonexistent-session")
527
528
529async def test_webrtc_gateway_default_ice_servers(cert_pems: tuple[str, str]) -> None:
530    """Test WebRTCGateway uses default ICE servers."""
531    cert_pem, key_pem = cert_pems
532    mock_session = Mock()
533    gateway = WebRTCGateway(
534        http_session=mock_session,
535        remote_id="TEST-REMOTE-ID",
536        cert_pem=cert_pem,
537        key_pem=key_pem,
538    )
539
540    assert len(gateway.ice_servers) > 0
541    # Should have at least one STUN server
542    assert any("stun:" in server["urls"] for server in gateway.ice_servers)
543
544
545async def test_webrtc_gateway_handle_client_connected(cert_pems: tuple[str, str]) -> None:
546    """Test WebRTCGateway handles client-connected message."""
547    cert_pem, key_pem = cert_pems
548    mock_session = Mock()
549    gateway = WebRTCGateway(
550        http_session=mock_session,
551        remote_id="TEST-REMOTE-ID",
552        cert_pem=cert_pem,
553        key_pem=key_pem,
554    )
555
556    with patch.object(gateway, "_get_fresh_ice_servers", AsyncMock(return_value=[])):
557        message = {"type": "client-connected", "sessionId": "test-session"}
558        await gateway._handle_signaling_message(message)
559
560        # Session should be created
561        assert "test-session" in gateway.sessions
562
563        # Cleanup
564        await gateway._close_session("test-session")
565
566
567async def test_webrtc_gateway_handle_client_disconnected(cert_pems: tuple[str, str]) -> None:
568    """Test WebRTCGateway handles client-disconnected message."""
569    cert_pem, key_pem = cert_pems
570    mock_session = Mock()
571    gateway = WebRTCGateway(
572        http_session=mock_session,
573        remote_id="TEST-REMOTE-ID",
574        cert_pem=cert_pem,
575        key_pem=key_pem,
576    )
577
578    with patch.object(gateway, "_get_fresh_ice_servers", AsyncMock(return_value=[])):
579        # Create a session first
580        session_id = "test-disconnect-session"
581        await gateway._create_session(session_id)
582        assert session_id in gateway.sessions
583
584        # Handle disconnect
585        message = {"type": "client-disconnected", "sessionId": session_id}
586        await gateway._handle_signaling_message(message)
587
588        # Session should be removed
589        assert session_id not in gateway.sessions
590
591
592async def test_webrtc_gateway_reconnection_logic(cert_pems: tuple[str, str]) -> None:
593    """Test WebRTCGateway has proper reconnection backoff."""
594    cert_pem, key_pem = cert_pems
595    mock_session = Mock()
596    gateway = WebRTCGateway(
597        http_session=mock_session,
598        remote_id="TEST-REMOTE-ID",
599        cert_pem=cert_pem,
600        key_pem=key_pem,
601    )
602
603    # Check initial reconnect delay
604    assert gateway._current_reconnect_delay == 10
605
606    # Simulate multiple failed connections (without actually connecting)
607    initial_delay = gateway._current_reconnect_delay
608    gateway._current_reconnect_delay = min(
609        gateway._current_reconnect_delay * 2, gateway._max_reconnect_delay
610    )
611
612    assert gateway._current_reconnect_delay == initial_delay * 2
613
614    # Should not exceed max
615    for _ in range(10):
616        gateway._current_reconnect_delay = min(
617            gateway._current_reconnect_delay * 2, gateway._max_reconnect_delay
618        )
619
620    assert gateway._current_reconnect_delay <= gateway._max_reconnect_delay
621
622
623async def test_webrtc_gateway_handle_offer_without_session(cert_pems: tuple[str, str]) -> None:
624    """Test WebRTCGateway handles offer for non-existent session gracefully."""
625    cert_pem, key_pem = cert_pems
626    mock_session = Mock()
627    gateway = WebRTCGateway(
628        http_session=mock_session,
629        remote_id="TEST-REMOTE-ID",
630        cert_pem=cert_pem,
631        key_pem=key_pem,
632    )
633
634    # Try to handle offer for non-existent session
635    offer_data = {"sdp": "test-sdp", "type": "offer"}
636    await gateway._handle_offer("nonexistent-session", offer_data)
637
638    # Should not crash
639
640
641async def test_webrtc_gateway_handle_ice_candidate_without_session(
642    cert_pems: tuple[str, str],
643) -> None:
644    """Test WebRTCGateway handles ICE candidate for non-existent session gracefully."""
645    cert_pem, key_pem = cert_pems
646    mock_session = Mock()
647    gateway = WebRTCGateway(
648        http_session=mock_session,
649        remote_id="TEST-REMOTE-ID",
650        cert_pem=cert_pem,
651        key_pem=key_pem,
652    )
653
654    # Try to handle ICE candidate for non-existent session
655    candidate_data = {
656        "candidate": "candidate:1 1 UDP 1234 192.168.1.1 12345 typ host",
657        "sdpMid": "0",
658        "sdpMLineIndex": 0,
659    }
660    await gateway._handle_ice_candidate("nonexistent-session", candidate_data)
661
662    # Should not crash
663
664
665@pytest.mark.parametrize(
666    "malicious_path",
667    [
668        "@evil.com",  # netloc becomes basic-auth creds, evil.com becomes the host
669        "//evil.com",  # protocol-relative URL pointing at another host
670        "@evil.com/foo",
671        "//evil.com/foo",
672    ],
673)
674async def test_http_proxy_request_cannot_change_host(
675    cert_pems: tuple[str, str], malicious_path: str
676) -> None:
677    """An attacker-controlled proxy path must never change the target host (SSRF guard)."""
678    cert_pem, key_pem = cert_pems
679    mock_session = Mock()
680    captured_url: dict[str, str] = {}
681
682    def fake_request(_method: str, url: str, **_kwargs: object) -> AsyncMock:
683        captured_url["url"] = url
684        response = AsyncMock()
685        response.status = 200
686        response.headers = {}
687        response.read = AsyncMock(return_value=b"")
688        ctx = AsyncMock()
689        ctx.__aenter__ = AsyncMock(return_value=response)
690        ctx.__aexit__ = AsyncMock(return_value=False)
691        return ctx
692
693    mock_session.request = fake_request
694
695    gateway = WebRTCGateway(
696        http_session=mock_session,
697        remote_id="TEST-REMOTE-ID",
698        cert_pem=cert_pem,
699        key_pem=key_pem,
700        local_ws_url="ws://localhost:8095/ws",
701    )
702
703    await gateway._handle_http_proxy_request(
704        None, {"id": "1", "method": "GET", "path": malicious_path}
705    )
706
707    parsed = urlparse(captured_url["url"])
708    assert parsed.hostname == "localhost"
709    assert parsed.port == 8095
710    assert parsed.username is None
711    assert "evil.com" not in (parsed.netloc or "")
712
713
714async def test_http_proxy_request_keeps_the_unverified_dial_on_this_host(
715    cert_pems: tuple[str, str],
716) -> None:
717    """The proxy must not carry its unverified TLS dial off this host."""
718    cert_pem, key_pem = cert_pems
719    mock_session = Mock()
720    captured_kwargs: dict[str, object] = {}
721
722    def fake_request(_method: str, _url: str, **kwargs: object) -> AsyncMock:
723        captured_kwargs.update(kwargs)
724        response = AsyncMock()
725        response.status = 200
726        response.headers = {}
727        response.read = AsyncMock(return_value=b"")
728        ctx = AsyncMock()
729        ctx.__aenter__ = AsyncMock(return_value=response)
730        ctx.__aexit__ = AsyncMock(return_value=False)
731        return ctx
732
733    mock_session.request = fake_request
734
735    gateway = WebRTCGateway(
736        http_session=mock_session,
737        remote_id="TEST-REMOTE-ID",
738        cert_pem=cert_pem,
739        key_pem=key_pem,
740        local_ws_url="wss://127.0.0.1:8095/ws",
741    )
742
743    await gateway._handle_http_proxy_request(None, {"id": "1", "method": "GET", "path": "/info"})
744
745    assert captured_kwargs["ssl"] is False
746    assert captured_kwargs["allow_redirects"] is False
747
748
749async def test_local_websocket_dial_skips_certificate_verification(
750    cert_pems: tuple[str, str],
751) -> None:
752    """Reach the local API on the bind address, which no certificate is issued for."""
753    cert_pem, key_pem = cert_pems
754    mock_session = Mock()
755    captured_kwargs: dict[str, object] = {}
756
757    async def fake_ws_connect(_url: str, **kwargs: object) -> AsyncMock:
758        captured_kwargs.update(kwargs)
759        return AsyncMock()
760
761    mock_session.ws_connect = fake_ws_connect
762
763    gateway = WebRTCGateway(
764        http_session=mock_session,
765        remote_id="TEST-REMOTE-ID",
766        cert_pem=cert_pem,
767        key_pem=key_pem,
768        local_ws_url="wss://127.0.0.1:8095/ws",
769    )
770    session = WebRTCSession(session_id="s1", pc=Mock())
771    channel = MagicMock()
772    channel.wait_open = AsyncMock()
773
774    with patch.object(gateway, "_schedule_close"):
775        await gateway._bridge_ma_api(session, channel)
776
777    assert captured_kwargs["ssl"] is False
778
779
780# ---- aiolibdatachannel loopback tests --------------------------------------
781#
782# These exercise the migrated gateway against real loopback PeerConnections
783# (offerer = browser role, answerer = gateway role) rather than mocking the
784# WebRTC layer. ICE servers are stubbed to [] so gathering completes on host
785# candidates only (fast, offline).
786
787
788@pytest.fixture(scope="session")
789def cert_pems() -> tuple[str, str]:
790    """Generate a throwaway DTLS cert/key as PEM strings for the gateway."""
791    private_key, cert = _generate_certificate()
792    cert_pem = cert.public_bytes(serialization.Encoding.PEM).decode()
793    key_pem = private_key.private_bytes(
794        encoding=serialization.Encoding.PEM,
795        format=serialization.PrivateFormat.PKCS8,
796        encryption_algorithm=serialization.NoEncryption(),
797    ).decode()
798    return cert_pem, key_pem
799
800
801def _sha256_fingerprint(cert_pem: str) -> str:
802    """Compute the uppercase colon-separated SHA-256 fingerprint of a PEM certificate."""
803    body = "".join(line for line in cert_pem.splitlines() if line and not line.startswith("-----"))
804    der = base64.b64decode(body)
805    digest = hashlib.sha256(der).hexdigest().upper()
806    return ":".join(digest[i : i + 2] for i in range(0, len(digest), 2))
807
808
809class _FakeSignaling:
810    """Signaling WebSocket stand-in that captures outbound JSON messages."""
811
812    def __init__(self) -> None:
813        self.messages: list[dict[str, Any]] = []
814
815    async def send_json(self, data: dict[str, Any]) -> None:
816        self.messages.append(data)
817
818    @property
819    def answers(self) -> list[dict[str, Any]]:
820        return [m for m in self.messages if m.get("type") == "answer"]
821
822
823class _FakeLocalWS:
824    """Minimal aiohttp WebSocket stand-in for channel-bridging tests."""
825
826    def __init__(self) -> None:
827        self.closed = False
828        self.sent: list[str | bytes] = []
829        self._incoming: asyncio.Queue[SimpleNamespace | None] = asyncio.Queue()
830
831    async def send_str(self, data: str) -> None:
832        self.sent.append(data)
833
834    async def send_bytes(self, data: bytes) -> None:
835        self.sent.append(data)
836
837    async def close(self) -> None:
838        self.closed = True
839        self._incoming.put_nowait(None)
840
841    def feed_text(self, data: str) -> None:
842        """Queue a text message as if the local server sent it."""
843        self._incoming.put_nowait(SimpleNamespace(type=aiohttp.WSMsgType.TEXT, data=data))
844
845    def feed_bytes(self, data: bytes) -> None:
846        """Queue a binary message as if the local server sent it."""
847        self._incoming.put_nowait(SimpleNamespace(type=aiohttp.WSMsgType.BINARY, data=data))
848
849    def __aiter__(self) -> AsyncIterator[SimpleNamespace]:
850        return self
851
852    async def __anext__(self) -> SimpleNamespace:
853        msg = await self._incoming.get()
854        if msg is None:
855            raise StopAsyncIteration
856        return msg
857
858
859class _FakeBidiChannel:
860    """
861    Async-iterable data-channel stand-in for bridging tests.
862
863    Drives the gateway's channel->local pump without a real WebRTC handshake: feed
864    inbound messages with :meth:`feed`, end the stream with :meth:`close`, and read
865    what the gateway sent back on ``sent``.
866    """
867
868    def __init__(self, label: str = "ma-api", max_message_size: int = 256 * 1024) -> None:
869        self.label = label
870        self.is_open = True
871        self.is_closed = False
872        self.closed = False
873        self.max_message_size = max_message_size
874        self.sent: list[str | bytes] = []
875        self._inbound: asyncio.Queue[str | bytes | None] = asyncio.Queue()
876
877    async def wait_open(self) -> None:
878        return
879
880    async def send(self, data: str | bytes) -> None:
881        # a real send always suspends, which is what lets concurrent senders interleave
882        await asyncio.sleep(0)
883        self.sent.append(data)
884
885    def feed(self, message: str | bytes) -> None:
886        """Queue an inbound message as if the browser sent it over the channel."""
887        self._inbound.put_nowait(message)
888
889    def close(self) -> None:
890        self.closed = True
891        self.is_closed = True
892        self.is_open = False
893        self._inbound.put_nowait(None)
894
895    async def aclose(self) -> None:
896        self.close()
897
898    def __aiter__(self) -> AsyncIterator[str | bytes]:
899        return self
900
901    async def __anext__(self) -> str | bytes:
902        message = await self._inbound.get()
903        if message is None:
904            raise StopAsyncIteration
905        return message
906
907
908class _FakeHttpSession:
909    """ClientSession stand-in handing out one fake WebSocket per dialed url."""
910
911    def __init__(self) -> None:
912        self.dialed: list[str] = []
913        self.dial_kwargs: list[dict[str, Any]] = []
914        self.websockets: dict[str, _FakeLocalWS] = {}
915        self.requested: list[str] = []
916        self.response_body = b""
917        self.bodies: dict[str, bytes] = {}
918
919    async def ws_connect(self, url: str, **kwargs: Any) -> _FakeLocalWS:
920        self.dialed.append(url)
921        self.dial_kwargs.append(kwargs)
922        local_ws = _FakeLocalWS()
923        self.websockets[url] = local_ws
924        return local_ws
925
926    def request(self, _method: str, url: str, **_kwargs: Any) -> AsyncMock:
927        """Serve this url's entry in ``bodies``, falling back to ``response_body``."""
928        self.requested.append(url)
929        response = AsyncMock()
930        response.status = 200
931        response.headers = {"Content-Type": "image/jpeg"}
932        response.read = AsyncMock(return_value=self.bodies.get(url, self.response_body))
933        ctx = AsyncMock()
934        ctx.__aenter__ = AsyncMock(return_value=response)
935        ctx.__aexit__ = AsyncMock(return_value=False)
936        return ctx
937
938
939class _FakePeerConnection:
940    """PeerConnection stand-in that runs gateway-spawned pumps as asyncio tasks."""
941
942    def __init__(self) -> None:
943        self._tasks: list[asyncio.Task[None]] = []
944        self._incoming: asyncio.Queue[_FakeBidiChannel] = asyncio.Queue()
945
946    def spawn_task(self, coro: Coroutine[Any, Any, None]) -> None:
947        self._tasks.append(asyncio.ensure_future(coro))
948
949    def offer_channel(self, channel: _FakeBidiChannel) -> None:
950        """Offer a data channel as if the browser had opened it."""
951        self._incoming.put_nowait(channel)
952
953    async def incoming_data_channels(self) -> AsyncIterator[DataChannel]:
954        while True:
955            yield cast("DataChannel", await self._incoming.get())
956
957    async def aclose(self) -> None:
958        for task in self._tasks:
959            task.cancel()
960        # await cancellation so no pump task lingers past teardown
961        await asyncio.gather(*self._tasks, return_exceptions=True)
962
963
964def _register_bridge_session(
965    gateway: WebRTCGateway, session_id: str, channel: _FakeBidiChannel
966) -> WebRTCSession:
967    """Register a session backed by a fake PeerConnection and ma-api channel."""
968    session = WebRTCSession(
969        session_id=session_id,
970        pc=cast("PeerConnection", _FakePeerConnection()),
971        data_channel=cast("DataChannel", channel),
972    )
973    gateway.sessions[session_id] = session
974    return session
975
976
977def _register_routed_session(
978    gateway: WebRTCGateway, session_id: str
979) -> tuple[WebRTCSession, _FakePeerConnection]:
980    """Register a session that routes the data channels offered to its PeerConnection."""
981    pc = _FakePeerConnection()
982    session = WebRTCSession(session_id=session_id, pc=cast("PeerConnection", pc))
983    gateway.sessions[session_id] = session
984    pc.spawn_task(gateway._accept_channels(session))
985    return session, pc
986
987
988async def _wait_for(predicate: Callable[[], bool], timeout: float = 15.0) -> None:
989    """Poll ``predicate`` until it is true or the timeout elapses."""
990    loop = asyncio.get_running_loop()
991    deadline = loop.time() + timeout
992    while loop.time() < deadline:
993        if predicate():
994            return
995        await asyncio.sleep(0.02)
996    raise AssertionError("condition not met within timeout")
997
998
999async def test_handle_offer_answers_with_pinned_fingerprint(cert_pems: tuple[str, str]) -> None:
1000    """The answer SDP carries the DTLS fingerprint of the configured certificate (pinning)."""
1001    cert_pem, key_pem = cert_pems
1002    gateway = WebRTCGateway(
1003        http_session=Mock(),
1004        remote_id="TEST-REMOTE-ID",
1005        cert_pem=cert_pem,
1006        key_pem=key_pem,
1007    )
1008    signaling = _FakeSignaling()
1009    gateway._signaling_ws = cast("aiohttp.ClientWebSocketResponse", signaling)
1010    offerer = PeerConnection(RTCConfiguration())
1011    session_id = "fingerprint-session"
1012    try:
1013        await offerer.create_data_channel("ma-api")
1014        offer = await offerer.create_offer()
1015        with patch.object(gateway, "_get_fresh_ice_servers", AsyncMock(return_value=[])):
1016            await gateway._create_session(session_id)
1017            await asyncio.wait_for(
1018                gateway._handle_offer(session_id, {"sdp": offer.sdp, "type": "offer"}),
1019                timeout=15,
1020            )
1021
1022        assert len(signaling.answers) == 1
1023        answer = signaling.answers[0]
1024        assert answer["sessionId"] == session_id
1025        assert answer["data"]["type"] == "answer"
1026        fingerprint_line = next(
1027            line for line in answer["data"]["sdp"].splitlines() if line.startswith("a=fingerprint:")
1028        )
1029        assert _sha256_fingerprint(cert_pem) in fingerprint_line
1030    finally:
1031        await gateway._close_session(session_id)
1032        await offerer.aclose()
1033
1034
1035async def test_ma_api_channel_bridges_to_local_ws(cert_pems: tuple[str, str]) -> None:
1036    """Messages flow both ways across the ma-api data channel and the local WebSocket."""
1037    cert_pem, key_pem = cert_pems
1038    fake_ws = _FakeLocalWS()
1039    http_session = Mock()
1040    http_session.ws_connect = AsyncMock(
1041        return_value=cast("aiohttp.ClientWebSocketResponse", fake_ws)
1042    )
1043    gateway = WebRTCGateway(
1044        http_session=http_session,
1045        remote_id="TEST-REMOTE-ID",
1046        cert_pem=cert_pem,
1047        key_pem=key_pem,
1048    )
1049    channel = _FakeBidiChannel()
1050    session = _register_bridge_session(gateway, "bridge-session", channel)
1051    bridge = asyncio.ensure_future(gateway._bridge_ma_api(session, cast("DataChannel", channel)))
1052    try:
1053        await _wait_for(lambda: session.local_ws is not None)
1054
1055        # browser -> local WebSocket
1056        channel.feed("from browser")
1057        await _wait_for(lambda: fake_ws.sent == ["from browser"])
1058
1059        # local WebSocket -> browser
1060        fake_ws.feed_text("from local")
1061        await _wait_for(lambda: channel.sent == ["from local"])
1062    finally:
1063        channel.close()
1064        await asyncio.wait_for(bridge, timeout=5)
1065        await _wait_for(lambda: "bridge-session" not in gateway.sessions)
1066
1067
1068def test_build_ice_servers_maps_dicts() -> None:
1069    """ICE server dicts map to one IceServer per url, preserving TURN credentials."""
1070    gateway = WebRTCGateway(
1071        http_session=Mock(),
1072        remote_id="TEST-REMOTE-ID",
1073        cert_pem="cert",
1074        key_pem="key",
1075    )
1076    servers: list[dict[str, Any]] = [
1077        {"urls": "stun:stun.example.com:3478"},
1078        {"urls": "turn:turn.example.com:3478", "username": "user", "credential": "pass"},
1079        {
1080            "urls": ["stun:a.example.com:3478", "turn:b.example.com:3478"],
1081            "username": "u2",
1082            "credential": "c2",
1083        },
1084    ]
1085
1086    result = gateway._build_ice_servers(servers)
1087
1088    assert all(isinstance(server, IceServer) for server in result)
1089    # the two-url entry fans out, so 1 + 1 + 2 = 4 IceServers
1090    assert len(result) == 4
1091    assert result[0] == IceServer(url="stun:stun.example.com:3478")
1092    assert result[1] == IceServer(
1093        url="turn:turn.example.com:3478", username="user", credential="pass"
1094    )
1095    # to_url() inlines the TURN credentials for libdatachannel
1096    assert result[1].to_url() == "turn:user:[email protected]:3478"
1097    # list urls share the entry's credentials
1098    assert result[2] == IceServer(url="stun:a.example.com:3478", username="u2", credential="c2")
1099    assert result[3] == IceServer(url="turn:b.example.com:3478", username="u2", credential="c2")
1100
1101
1102async def test_session_closes_when_ma_api_channel_closes(cert_pems: tuple[str, str]) -> None:
1103    """Closing the browser ma-api channel tears down the whole gateway session."""
1104    cert_pem, key_pem = cert_pems
1105    fake_ws = _FakeLocalWS()
1106    http_session = Mock()
1107    http_session.ws_connect = AsyncMock(
1108        return_value=cast("aiohttp.ClientWebSocketResponse", fake_ws)
1109    )
1110    gateway = WebRTCGateway(
1111        http_session=http_session,
1112        remote_id="TEST-REMOTE-ID",
1113        cert_pem=cert_pem,
1114        key_pem=key_pem,
1115    )
1116    channel = _FakeBidiChannel()
1117    session = _register_bridge_session(gateway, "channel-close-session", channel)
1118    bridge = asyncio.ensure_future(gateway._bridge_ma_api(session, cast("DataChannel", channel)))
1119    await _wait_for(lambda: session.local_ws is not None)
1120
1121    # the browser closes the ma-api channel -> the whole session is torn down
1122    channel.close()
1123
1124    await _wait_for(lambda: "channel-close-session" not in gateway.sessions)
1125    assert "channel-close-session" not in gateway.sessions
1126    await asyncio.wait_for(bridge, timeout=5)
1127
1128
1129# ---- channel routing -------------------------------------------------------
1130
1131LOCAL_WS_URL = "ws://127.0.0.1:8095/ws"
1132SENDSPIN_URL = "ws://127.0.0.1:8927/sendspin"
1133
1134
1135def _routing_gateway(
1136    cert_pems: tuple[str, str],
1137    http_session: _FakeHttpSession,
1138    local_ws_url: str = LOCAL_WS_URL,
1139    set_sendspin_player_callback: Callable[[str, str], None] | None = None,
1140) -> WebRTCGateway:
1141    """Create a gateway whose local WebSockets are all served by the fake HTTP session."""
1142    cert_pem, key_pem = cert_pems
1143    return WebRTCGateway(
1144        http_session=cast("aiohttp.ClientSession", http_session),
1145        remote_id="TEST-REMOTE-ID",
1146        cert_pem=cert_pem,
1147        key_pem=key_pem,
1148        local_ws_url=local_ws_url,
1149        sendspin_url=SENDSPIN_URL,
1150        set_sendspin_player_callback=set_sendspin_player_callback,
1151    )
1152
1153
1154async def test_sendspin_channel_bridges_to_the_sendspin_server(cert_pems: tuple[str, str]) -> None:
1155    """A sendspin channel reaches the internal sendspin server, web player id and all."""
1156    http_session = _FakeHttpSession()
1157    announced_players: list[tuple[str, str]] = []
1158    gateway = _routing_gateway(
1159        cert_pems,
1160        http_session,
1161        set_sendspin_player_callback=lambda session_id, player_id: announced_players.append(
1162            (session_id, player_id)
1163        ),
1164    )
1165    session, pc = _register_routed_session(gateway, "sendspin-session")
1166    channel = _FakeBidiChannel(label="sendspin")
1167    pc.offer_channel(channel)
1168    try:
1169        await _wait_for(lambda: SENDSPIN_URL in http_session.websockets)
1170        local_ws = http_session.websockets[SENDSPIN_URL]
1171
1172        # the first message announces the web player, and is forwarded verbatim
1173        auth = json.dumps({"type": "auth", "token": "t", "client_id": "web-player-1"})
1174        channel.feed(auth)
1175        await _wait_for(lambda: local_ws.sent == [auth])
1176        assert announced_players == [("sendspin-session", "web-player-1")]
1177        assert session.sendspin_player_id == "web-player-1"
1178
1179        # audio keeps flowing in both directions, text and binary alike
1180        channel.feed(b"\x01\x02")
1181        await _wait_for(lambda: local_ws.sent == [auth, b"\x01\x02"])
1182        local_ws.feed_text('{"type":"hello"}')
1183        local_ws.feed_bytes(b"\x03\x04")
1184        await _wait_for(lambda: channel.sent == ['{"type":"hello"}', b"\x03\x04"])
1185    finally:
1186        await gateway._close_session("sendspin-session")
1187
1188
1189@pytest.mark.parametrize(
1190    ("local_ws_url", "expected_url"),
1191    [
1192        ("ws://127.0.0.1:8095/ws", "ws://127.0.0.1:8095/live_announcement"),
1193        # an https webserver is still dialed on its bind address, which no cert covers
1194        ("wss://127.0.0.1:8095/ws", "wss://127.0.0.1:8095/live_announcement"),
1195    ],
1196)
1197async def test_live_announcement_channel_bridges_to_the_webserver(
1198    cert_pems: tuple[str, str], local_ws_url: str, expected_url: str
1199) -> None:
1200    """A live announcement channel reaches the webserver route that takes the audio."""
1201    http_session = _FakeHttpSession()
1202    gateway = _routing_gateway(cert_pems, http_session, local_ws_url=local_ws_url)
1203    session, pc = _register_routed_session(gateway, "announce-session")
1204    channel = _FakeBidiChannel(label="live_announcement")
1205    pc.offer_channel(channel)
1206    try:
1207        await _wait_for(lambda: expected_url in http_session.websockets)
1208        local_ws = http_session.websockets[expected_url]
1209        assert http_session.dial_kwargs == [{"ssl": False}]
1210
1211        # the client authenticates on the route itself, so its handshake passes through
1212        handshake = [
1213            json.dumps({"type": "auth", "token": "t"}),
1214            json.dumps({"type": "start", "player_id": "player1", "sample_rate": 16000}),
1215        ]
1216        for message in handshake:
1217            channel.feed(message)
1218        await _wait_for(lambda: local_ws.sent == handshake)
1219        # the sendspin snoop belongs to the sendspin bridge only
1220        assert session.sendspin_player_id is None
1221
1222        # spoken audio goes up, the route's replies come back
1223        channel.feed(b"\x00\x01")
1224        await _wait_for(lambda: local_ws.sent == [*handshake, b"\x00\x01"])
1225        local_ws.feed_text('{"type":"started"}')
1226        await _wait_for(lambda: channel.sent == ['{"type":"started"}'])
1227    finally:
1228        await gateway._close_session("announce-session")
1229
1230
1231@pytest.mark.parametrize("closed_by", ["browser", "local"])
1232async def test_closing_a_bridged_channel_leaves_the_api_session_up(
1233    cert_pems: tuple[str, str], closed_by: str
1234) -> None:
1235    """Losing a bridged WebSocket tears down that bridge only, never the API session."""
1236    http_session = _FakeHttpSession()
1237    gateway = _routing_gateway(cert_pems, http_session)
1238    session, pc = _register_routed_session(gateway, "mixed-session")
1239    api_channel = _FakeBidiChannel()
1240    sendspin_channel = _FakeBidiChannel(label="sendspin")
1241    pc.offer_channel(api_channel)
1242    pc.offer_channel(sendspin_channel)
1243    try:
1244        await _wait_for(
1245            lambda: session.local_ws is not None and SENDSPIN_URL in http_session.dialed
1246        )
1247        sendspin_ws = http_session.websockets[SENDSPIN_URL]
1248
1249        if closed_by == "browser":
1250            sendspin_channel.close()
1251        else:
1252            await sendspin_ws.close()
1253        await _wait_for(lambda: not session.channels)
1254
1255        assert sendspin_ws.closed is True
1256        assert sendspin_channel.closed is True
1257        # the API session is untouched: still registered, still bridged, channel still open
1258        assert "mixed-session" in gateway.sessions
1259        assert session.local_ws is not None
1260        assert cast("_FakeLocalWS", session.local_ws).closed is False
1261        assert api_channel.closed is False
1262    finally:
1263        await gateway._close_session("mixed-session")
1264
1265
1266async def test_the_first_channel_is_the_api_channel_whatever_its_label(
1267    cert_pems: tuple[str, str],
1268) -> None:
1269    """Clients may label their API channel freely, so the first channel bridges to the API."""
1270    http_session = _FakeHttpSession()
1271    gateway = _routing_gateway(cert_pems, http_session)
1272    session, pc = _register_routed_session(gateway, "labelled-session")
1273    channel = _FakeBidiChannel(label="ma-api-v2")
1274    pc.offer_channel(channel)
1275    try:
1276        await _wait_for(lambda: session.local_ws is not None)
1277        assert session.data_channel is cast("DataChannel", channel)
1278        assert http_session.dialed == [f"{LOCAL_WS_URL}?webrtc_session_id=labelled-session"]
1279    finally:
1280        await gateway._close_session("labelled-session")
1281
1282
1283async def test_unknown_channel_label_cannot_replace_the_api_channel(
1284    cert_pems: tuple[str, str],
1285) -> None:
1286    """A channel this server has no route for is refused instead of hijacking the session."""
1287    http_session = _FakeHttpSession()
1288    gateway = _routing_gateway(cert_pems, http_session)
1289    session, pc = _register_routed_session(gateway, "unknown-session")
1290    api_channel = _FakeBidiChannel()
1291    unknown_channel = _FakeBidiChannel(label="channel-from-the-future")
1292    pc.offer_channel(api_channel)
1293    try:
1294        await _wait_for(lambda: session.local_ws is not None)
1295        pc.offer_channel(unknown_channel)
1296        await _wait_for(lambda: unknown_channel.closed)
1297
1298        assert session.data_channel is cast("DataChannel", api_channel)
1299        assert "unknown-session" in gateway.sessions
1300        # only the API channel was ever bridged
1301        assert http_session.dialed == [f"{LOCAL_WS_URL}?webrtc_session_id=unknown-session"]
1302    finally:
1303        await gateway._close_session("unknown-session")
1304
1305
1306def _proxy_request(request_id: str, path: str) -> str:
1307    """Build the http-proxy-request message a client sends for a proxied path."""
1308    return json.dumps(
1309        {"type": "http-proxy-request", "id": request_id, "method": "GET", "path": path}
1310    )
1311
1312
1313def _body_delivered(sent: list[str | bytes], size: int) -> bool:
1314    """Return whether the binary frames sent so far add up to a whole body of ``size``."""
1315    return sum(len(m) for m in sent if isinstance(m, bytes)) >= size
1316
1317
1318def _read_proxy_response(sent: list[str | bytes]) -> tuple[dict[str, Any], bytes]:
1319    """
1320    Read the binary-framed response a client would reassemble from the proxy channel.
1321
1322    :param sent: Messages the gateway sent, starting at the response's JSON header.
1323    """
1324    header = json.loads(cast("str", sent[0]))
1325    body = b""
1326    for message in sent[1:]:
1327        assert isinstance(message, bytes), f"expected a binary body frame, got {message!r}"
1328        body += message
1329        if len(body) >= header["size"]:
1330            break
1331    assert len(body) == header["size"]
1332    return header, body
1333
1334
1335async def test_http_proxy_channel_answers_on_its_own_channel(cert_pems: tuple[str, str]) -> None:
1336    """A proxied request on the http proxy channel is answered there, not on the API channel."""
1337    http_session = _FakeHttpSession()
1338    http_session.response_body = b"\xff\xd8jpeg-bytes"
1339    gateway = _routing_gateway(cert_pems, http_session)
1340    session, pc = _register_routed_session(gateway, "proxy-session")
1341    api_channel = _FakeBidiChannel()
1342    proxy_channel = _FakeBidiChannel(label="http_proxy")
1343    pc.offer_channel(api_channel)
1344    pc.offer_channel(proxy_channel)
1345    try:
1346        await _wait_for(lambda: session.local_ws is not None)
1347
1348        proxy_channel.feed(_proxy_request("img-1", "/imageproxy/abc"))
1349        await _wait_for(lambda: len(proxy_channel.sent) >= 2)
1350
1351        assert http_session.requested == ["http://127.0.0.1:8095/imageproxy/abc"]
1352        response, body = _read_proxy_response(proxy_channel.sent)
1353        assert response["type"] == "http-proxy-response"
1354        assert response["id"] == "img-1"
1355        assert response["status"] == 200
1356        # the body rides as raw binary, so it costs its own size on the wire and no more
1357        assert body == b"\xff\xd8jpeg-bytes"
1358        assert "body" not in response
1359        # the image never touches the API channel, nor the local API WebSocket
1360        assert api_channel.sent == []
1361        assert cast("_FakeLocalWS", session.local_ws).sent == []
1362    finally:
1363        await gateway._close_session("proxy-session")
1364
1365
1366async def test_http_proxy_request_on_the_api_channel_is_still_answered(
1367    cert_pems: tuple[str, str],
1368) -> None:
1369    """Clients that predate the http proxy channel keep proxying over the API channel."""
1370    http_session = _FakeHttpSession()
1371    http_session.response_body = b"legacy-body"
1372    gateway = _routing_gateway(cert_pems, http_session)
1373    session, pc = _register_routed_session(gateway, "legacy-session")
1374    api_channel = _FakeBidiChannel()
1375    pc.offer_channel(api_channel)
1376    try:
1377        await _wait_for(lambda: session.local_ws is not None)
1378
1379        api_channel.feed(_proxy_request("img-2", "/imageproxy/def"))
1380        await _wait_for(lambda: bool(api_channel.sent))
1381
1382        assert http_session.requested == ["http://127.0.0.1:8095/imageproxy/def"]
1383        response = json.loads(cast("str", api_channel.sent[0]))
1384        assert response["type"] == "http-proxy-response"
1385        assert response["id"] == "img-2"
1386        assert bytes.fromhex(response["body"]) == b"legacy-body"
1387        # the proxy request is served here, never forwarded to the local API WebSocket
1388        assert cast("_FakeLocalWS", session.local_ws).sent == []
1389    finally:
1390        await gateway._close_session("legacy-session")
1391
1392
1393async def test_http_proxy_channel_reports_a_failed_fetch_on_its_own_channel(
1394    cert_pems: tuple[str, str],
1395) -> None:
1396    """A fetch that raises still answers the client, on the channel it asked over."""
1397    http_session = _FakeHttpSession()
1398    http_session.request = Mock(side_effect=RuntimeError("boom"))  # type: ignore[method-assign]
1399    gateway = _routing_gateway(cert_pems, http_session)
1400    session, pc = _register_routed_session(gateway, "proxy-error-session")
1401    api_channel = _FakeBidiChannel()
1402    proxy_channel = _FakeBidiChannel(label="http_proxy")
1403    pc.offer_channel(api_channel)
1404    pc.offer_channel(proxy_channel)
1405    try:
1406        await _wait_for(lambda: session.local_ws is not None)
1407
1408        proxy_channel.feed(_proxy_request("img-3", "/imageproxy/boom"))
1409        await _wait_for(lambda: len(proxy_channel.sent) >= 2)
1410
1411        response, body = _read_proxy_response(proxy_channel.sent)
1412        assert response["id"] == "img-3"
1413        assert response["status"] == 500
1414        assert b"boom" in body
1415        assert api_channel.sent == []
1416    finally:
1417        await gateway._close_session("proxy-error-session")
1418
1419
1420async def test_http_proxy_channel_splits_a_large_body_into_binary_frames(
1421    cert_pems: tuple[str, str],
1422) -> None:
1423    """A body past the channel's message limit arrives as raw frames that concatenate back."""
1424    http_session = _FakeHttpSession()
1425    http_session.response_body = bytes(range(256)) * 2048  # 512 KiB
1426    gateway = _routing_gateway(cert_pems, http_session)
1427    session, pc = _register_routed_session(gateway, "proxy-large-session")
1428    proxy_channel = _FakeBidiChannel(label="http_proxy")
1429    pc.offer_channel(proxy_channel)
1430    try:
1431        await _wait_for(lambda: "http_proxy" in session.channels)
1432
1433        proxy_channel.feed(_proxy_request("img-5", "/imageproxy/large"))
1434        await _wait_for(
1435            lambda: _body_delivered(proxy_channel.sent, len(http_session.response_body))
1436        )
1437
1438        response, body = _read_proxy_response(proxy_channel.sent)
1439        assert response["size"] == len(http_session.response_body)
1440        assert body == http_session.response_body
1441        frames = proxy_channel.sent[1:]
1442        assert all(len(frame) <= proxy_channel.max_message_size for frame in frames)
1443        # the wire cost is the body itself, not the ~2.7x a hex-in-base64 response took
1444        assert sum(len(frame) for frame in frames) == len(http_session.response_body)
1445    finally:
1446        await gateway._close_session("proxy-large-session")
1447
1448
1449async def test_http_proxy_channel_honours_the_negotiated_message_limit(
1450    cert_pems: tuple[str, str],
1451) -> None:
1452    """A peer that advertises a small limit gets frames it can actually accept."""
1453    http_session = _FakeHttpSession()
1454    http_session.response_body = b"x" * (200 * 1024)
1455    gateway = _routing_gateway(cert_pems, http_session)
1456    session, pc = _register_routed_session(gateway, "proxy-small-frames-session")
1457    # what a peer that advertises no a=max-message-size in its SDP is assumed to accept
1458    proxy_channel = _FakeBidiChannel(label="http_proxy", max_message_size=64 * 1024)
1459    pc.offer_channel(proxy_channel)
1460    try:
1461        await _wait_for(lambda: "http_proxy" in session.channels)
1462
1463        proxy_channel.feed(_proxy_request("img-6", "/imageproxy/small-frames"))
1464        await _wait_for(
1465            lambda: _body_delivered(proxy_channel.sent, len(http_session.response_body))
1466        )
1467
1468        _, body = _read_proxy_response(proxy_channel.sent)
1469        assert body == http_session.response_body
1470        assert all(len(frame) <= 64 * 1024 for frame in proxy_channel.sent[1:])
1471    finally:
1472        await gateway._close_session("proxy-small-frames-session")
1473
1474
1475async def test_http_proxy_channel_never_interleaves_two_responses(
1476    cert_pems: tuple[str, str],
1477) -> None:
1478    """Body frames carry no request id, so each response must reach the client in one run."""
1479    http_session = _FakeHttpSession()
1480    http_session.bodies = {
1481        "http://127.0.0.1:8095/imageproxy/one": b"1" * (300 * 1024),
1482        "http://127.0.0.1:8095/imageproxy/two": b"2" * (300 * 1024),
1483    }
1484    gateway = _routing_gateway(cert_pems, http_session)
1485    session, pc = _register_routed_session(gateway, "proxy-concurrent-session")
1486    proxy_channel = _FakeBidiChannel(label="http_proxy")
1487    pc.offer_channel(proxy_channel)
1488    try:
1489        await _wait_for(lambda: "http_proxy" in session.channels)
1490
1491        proxy_channel.feed(_proxy_request("img-one", "/imageproxy/one"))
1492        proxy_channel.feed(_proxy_request("img-two", "/imageproxy/two"))
1493        await _wait_for(lambda: sum(isinstance(m, str) for m in proxy_channel.sent) == 2)
1494        await _wait_for(lambda: len(proxy_channel.sent) >= 6)
1495
1496        # split at the second header: each response owns an unbroken run of body frames
1497        second = next(i for i, m in enumerate(proxy_channel.sent) if i and isinstance(m, str))
1498        first_header, first_body = _read_proxy_response(proxy_channel.sent[:second])
1499        second_header, second_body = _read_proxy_response(proxy_channel.sent[second:])
1500        assert {first_header["id"], second_header["id"]} == {"img-one", "img-two"}
1501        # each body is one repeated byte, so any interleaving shows up as a mixed run
1502        filler = {"img-one": ord("1"), "img-two": ord("2")}
1503        assert set(first_body) == {filler[first_header["id"]]}
1504        assert set(second_body) == {filler[second_header["id"]]}
1505        assert len(first_body) == len(second_body) == 300 * 1024
1506    finally:
1507        await gateway._close_session("proxy-concurrent-session")
1508
1509
1510@pytest.mark.parametrize(
1511    "junk",
1512    [
1513        "not json at all",
1514        json.dumps(["not", "a", "dict"]),
1515        json.dumps({"type": "something-else"}),
1516        b"\x00\x01binary",
1517    ],
1518)
1519async def test_http_proxy_channel_survives_junk(
1520    cert_pems: tuple[str, str], junk: str | bytes
1521) -> None:
1522    """Anything that is not a proxy request is ignored without killing the channel."""
1523    http_session = _FakeHttpSession()
1524    http_session.response_body = b"still-here"
1525    gateway = _routing_gateway(cert_pems, http_session)
1526    session, pc = _register_routed_session(gateway, "proxy-junk-session")
1527    proxy_channel = _FakeBidiChannel(label="http_proxy")
1528    pc.offer_channel(proxy_channel)
1529    try:
1530        await _wait_for(lambda: "http_proxy" in session.channels)
1531
1532        proxy_channel.feed(junk)
1533        proxy_channel.feed(_proxy_request("img-4", "/imageproxy/ok"))
1534        await _wait_for(lambda: len(proxy_channel.sent) >= 2)
1535
1536        response, body = _read_proxy_response(proxy_channel.sent)
1537        assert response["id"] == "img-4"
1538        assert body == b"still-here"
1539    finally:
1540        await gateway._close_session("proxy-junk-session")
1541
1542
1543async def test_closing_the_http_proxy_channel_leaves_the_api_session_up(
1544    cert_pems: tuple[str, str],
1545) -> None:
1546    """Losing the http proxy channel tears down that channel only, never the API session."""
1547    http_session = _FakeHttpSession()
1548    gateway = _routing_gateway(cert_pems, http_session)
1549    session, pc = _register_routed_session(gateway, "proxy-close-session")
1550    api_channel = _FakeBidiChannel()
1551    proxy_channel = _FakeBidiChannel(label="http_proxy")
1552    pc.offer_channel(api_channel)
1553    pc.offer_channel(proxy_channel)
1554    try:
1555        await _wait_for(lambda: "http_proxy" in session.channels)
1556
1557        proxy_channel.close()
1558        await _wait_for(lambda: not session.channels)
1559
1560        assert "proxy-close-session" in gateway.sessions
1561        assert session.local_ws is not None
1562        assert api_channel.closed is False
1563    finally:
1564        await gateway._close_session("proxy-close-session")
1565
1566
1567async def test_a_second_http_proxy_channel_is_refused(cert_pems: tuple[str, str]) -> None:
1568    """A duplicate label is refused so the running handler is never left untracked."""
1569    http_session = _FakeHttpSession()
1570    gateway = _routing_gateway(cert_pems, http_session)
1571    session, pc = _register_routed_session(gateway, "duplicate-session")
1572    first = _FakeBidiChannel(label="http_proxy")
1573    second = _FakeBidiChannel(label="http_proxy")
1574    pc.offer_channel(first)
1575    try:
1576        await _wait_for(lambda: "http_proxy" in session.channels)
1577        pc.offer_channel(second)
1578        await _wait_for(lambda: second.closed)
1579
1580        assert first.closed is False
1581        assert session.channels["http_proxy"].channel is cast("DataChannel", first)
1582    finally:
1583        await gateway._close_session("duplicate-session")
1584
1585
1586class _FakeDataChannel:
1587    """
1588    Data channel stand-in that captures outbound messages for proxy tests.
1589
1590    :param close_after: Close the channel once this many messages have been sent.
1591    """
1592
1593    def __init__(self, max_message_size: int = 256 * 1024, close_after: int | None = None) -> None:
1594        self.is_closed = False
1595        self.max_message_size = max_message_size
1596        self.sent: list[str] = []
1597        # the gateway consults the channel once per outbound message, so this counts how far
1598        # a send loop got even when the messages themselves are discarded
1599        self.open_checks = 0
1600        self._is_open = True
1601        self._close_after = close_after
1602
1603    @property
1604    def is_open(self) -> bool:
1605        self.open_checks += 1
1606        return self._is_open
1607
1608    async def send(self, data: str) -> None:
1609        self.sent.append(data)
1610        if self._close_after is not None and len(self.sent) >= self._close_after:
1611            self._is_open = False
1612            self.is_closed = True
1613
1614
1615def _proxy_gateway(cert_pems: tuple[str, str]) -> WebRTCGateway:
1616    cert_pem, key_pem = cert_pems
1617    return WebRTCGateway(
1618        http_session=Mock(),
1619        remote_id="TEST-REMOTE-ID",
1620        cert_pem=cert_pem,
1621        key_pem=key_pem,
1622    )
1623
1624
1625def _reassemble_chunks(frames: list[str]) -> str:
1626    """Reassemble __chunk__ frames back into the original text (base64 -> bytes -> utf-8)."""
1627    parsed = [json.loads(f) for f in frames]
1628    assert all(f["type"] == "__chunk__" for f in parsed)
1629    assert len({f["id"] for f in parsed}) == 1
1630    count = parsed[0]["count"]
1631    parts: list[bytes] = [b""] * count
1632    for frame in parsed:
1633        parts[frame["seq"]] = base64.b64decode(frame["b64"])
1634    return b"".join(parts).decode()
1635
1636
1637async def test_http_proxy_response_small_body_single_message(
1638    cert_pems: tuple[str, str],
1639) -> None:
1640    """A body within the chunk size is sent as one http-proxy-response message."""
1641    gateway = _proxy_gateway(cert_pems)
1642    channel = _FakeDataChannel()
1643    body = b"\x00\x01\x02small-body"
1644
1645    await gateway._send_http_proxy_response(
1646        cast("DataChannel", channel), "req-small", 200, {"X-Test": "y"}, body
1647    )
1648
1649    assert len(channel.sent) == 1
1650    msg = json.loads(channel.sent[0])
1651    assert msg["type"] == "http-proxy-response"
1652    assert msg["id"] == "req-small"
1653    assert msg["status"] == 200
1654    assert msg["headers"] == {"X-Test": "y"}
1655    assert bytes.fromhex(msg["body"]) == body
1656
1657
1658async def test_http_proxy_response_large_body_chunked(cert_pems: tuple[str, str]) -> None:
1659    """A large HTTP-proxy response is split into base64 chunk frames the client reassembles."""
1660    gateway = _proxy_gateway(cert_pems)
1661    channel = _FakeDataChannel()
1662    # big body -> big JSON message
1663    body = bytes(range(256)) * ((DATA_CHANNEL_CHUNK_SIZE * 5) // 512)
1664
1665    await gateway._send_http_proxy_response(cast("DataChannel", channel), "req-big", 200, {}, body)
1666
1667    assert len(channel.sent) > 1
1668    assert all(json.loads(m)["type"] == "__chunk__" for m in channel.sent)
1669    # every serialized frame must stay under the negotiated 256 KiB data-channel limit
1670    assert all(len(m.encode()) < 256 * 1024 for m in channel.sent)
1671
1672    reassembled = json.loads(_reassemble_chunks(channel.sent))
1673    assert reassembled["type"] == "http-proxy-response"
1674    assert reassembled["id"] == "req-big"
1675    assert reassembled["status"] == 200
1676    assert bytes.fromhex(reassembled["body"]) == body
1677
1678
1679async def test_send_chunked_small_message_passthrough(cert_pems: tuple[str, str]) -> None:
1680    """A message within the limit is sent verbatim, not chunked."""
1681    gateway = _proxy_gateway(cert_pems)
1682    channel = _FakeDataChannel()
1683    await gateway._send_chunked(cast("DataChannel", channel), '{"event":"player_updated"}')
1684    assert channel.sent == ['{"event":"player_updated"}']
1685
1686
1687async def test_send_chunked_large_message_chunked(cert_pems: tuple[str, str]) -> None:
1688    """A large message is chunked and reassembles byte-identically (multibyte-safe)."""
1689    gateway = _proxy_gateway(cert_pems)
1690    channel = _FakeDataChannel()
1691    # multibyte payload so chunk boundaries fall mid-character, exercising the byte-level split
1692    text = '{"data":"' + "音楽" * DATA_CHANNEL_CHUNK_SIZE + '"}'
1693
1694    await gateway._send_chunked(cast("DataChannel", channel), text)
1695
1696    assert len(channel.sent) > 1
1697    assert all(len(m.encode()) < 256 * 1024 for m in channel.sent)
1698    assert _reassemble_chunks(channel.sent) == text
1699
1700
1701async def test_send_chunked_keeps_the_framing_released_clients_expect(
1702    cert_pems: tuple[str, str],
1703) -> None:
1704    """Against the 256 KiB limit every browser advertises, the frames must not move."""
1705    gateway = _proxy_gateway(cert_pems)
1706    channel = _FakeDataChannel()
1707    text = '{"event":"' + "x" * (DATA_CHANNEL_CHUNK_SIZE * 2 + 5000) + '"}'
1708    data = text.encode()
1709
1710    await gateway._send_chunked(cast("DataChannel", channel), text)
1711
1712    # the exact wire format the bundled frontend and the mobile app reassemble: 64 KiB pieces,
1713    # numbered from zero within a group id that counts up per message
1714    assert channel.sent == [
1715        json.dumps(
1716            {
1717                "type": "__chunk__",
1718                "id": 1,
1719                "seq": seq,
1720                "count": 3,
1721                "b64": base64.b64encode(
1722                    data[seq * DATA_CHANNEL_CHUNK_SIZE : (seq + 1) * DATA_CHANNEL_CHUNK_SIZE]
1723                ).decode(),
1724            }
1725        )
1726        for seq in range(3)
1727    ]
1728    # a full piece has always serialised well past 64 KiB, which only a peer advertising no
1729    # limit of its own would reject
1730    assert len(channel.sent[0].encode()) == 87447
1731
1732
1733async def test_send_chunked_honours_the_negotiated_message_limit(
1734    cert_pems: tuple[str, str],
1735) -> None:
1736    """A peer that advertises no limit in its SDP gets frames it can actually accept."""
1737    gateway = _proxy_gateway(cert_pems)
1738    # what libdatachannel assumes when the peer advertises no a=max-message-size
1739    channel = _FakeDataChannel(max_message_size=64 * 1024)
1740    text = '{"data":"' + "音楽" * DATA_CHANNEL_CHUNK_SIZE + '"}'
1741
1742    await gateway._send_chunked(cast("DataChannel", channel), text)
1743
1744    assert len(channel.sent) > 1
1745    assert all(len(m.encode()) <= channel.max_message_size for m in channel.sent)
1746    assert _reassemble_chunks(channel.sent) == text
1747
1748
1749async def test_send_chunked_passthrough_stops_at_the_negotiated_limit(
1750    cert_pems: tuple[str, str],
1751) -> None:
1752    """A message past a small peer's limit is chunked rather than sent whole."""
1753    gateway = _proxy_gateway(cert_pems)
1754    channel = _FakeDataChannel(max_message_size=16 * 1024)
1755    text = '{"data":"' + "x" * (32 * 1024) + '"}'
1756
1757    await gateway._send_chunked(cast("DataChannel", channel), text)
1758
1759    assert len(channel.sent) > 1
1760    assert all(len(m.encode()) <= channel.max_message_size for m in channel.sent)
1761    assert _reassemble_chunks(channel.sent) == text
1762
1763
1764async def test_send_chunked_stops_framing_once_the_channel_closes(
1765    cert_pems: tuple[str, str],
1766) -> None:
1767    """A channel that goes away mid-message stops the loop instead of encoding the rest."""
1768    gateway = _proxy_gateway(cert_pems)
1769    channel = _FakeDataChannel(close_after=1)
1770    text = '{"data":"' + "x" * (DATA_CHANNEL_CHUNK_SIZE * 10) + '"}'
1771
1772    await gateway._send_chunked(cast("DataChannel", channel), text)
1773
1774    assert len(channel.sent) == 1
1775    # the guard and the send of the first piece, then the guard again: the other ten pieces
1776    # are never framed
1777    assert channel.open_checks == 3
1778
1779
1780async def test_ma_api_channel_chunks_within_the_negotiated_limit(
1781    cert_pems: tuple[str, str],
1782) -> None:
1783    """A large API event reaches a peer that advertises no limit instead of being dropped."""
1784    cert_pem, key_pem = cert_pems
1785    fake_ws = _FakeLocalWS()
1786    http_session = Mock()
1787    http_session.ws_connect = AsyncMock(
1788        return_value=cast("aiohttp.ClientWebSocketResponse", fake_ws)
1789    )
1790    gateway = WebRTCGateway(
1791        http_session=http_session,
1792        remote_id="TEST-REMOTE-ID",
1793        cert_pem=cert_pem,
1794        key_pem=key_pem,
1795    )
1796    # what libdatachannel assumes when the peer advertises no a=max-message-size
1797    channel = _FakeBidiChannel(max_message_size=64 * 1024)
1798    session = _register_bridge_session(gateway, "small-limit-session", channel)
1799    bridge = asyncio.ensure_future(gateway._bridge_ma_api(session, cast("DataChannel", channel)))
1800    try:
1801        await _wait_for(lambda: session.local_ws is not None)
1802
1803        event = json.dumps({"event": "queue_updated", "data": "x" * (200 * 1024)})
1804        fake_ws.feed_text(event)
1805        await _wait_for(
1806            lambda: bool(channel.sent) and len(channel.sent) == json.loads(channel.sent[0])["count"]
1807        )
1808
1809        frames = cast("list[str]", channel.sent)
1810        assert all(len(frame.encode()) <= channel.max_message_size for frame in frames)
1811        assert _reassemble_chunks(frames) == event
1812    finally:
1813        channel.close()
1814        await asyncio.wait_for(bridge, timeout=5)
1815        await _wait_for(lambda: "small-limit-session" not in gateway.sessions)
1816