| 1 | import asyncio |
| 2 | import contextlib |
| 3 | import socket |
| 4 | from typing import Any, AsyncIterator |
| 5 | |
| 6 | import pytest |
| 7 | |
| 8 | |
| 9 | @contextlib.asynccontextmanager |
| 10 | async def _run_asgi_app(app: Any) -> AsyncIterator[str]: |
| 11 | import uvicorn |
| 12 | |
| 13 | sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) |
| 14 | sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) |
| 15 | sock.bind(("127.0.0.1", 0)) |
| 16 | sock.listen(128) |
| 17 | |
| 18 | port = sock.getsockname()[1] |
| 19 | |
| 20 | config = uvicorn.Config( |
| 21 | app, |
| 22 | host="127.0.0.1", |
| 23 | port=port, |
| 24 | log_level="warning", |
| 25 | access_log=False, |
| 26 | lifespan="off", |
| 27 | ) |
| 28 | server = uvicorn.Server(config) |
| 29 | server.install_signal_handlers = lambda: None # type: ignore[method-assign] |
| 30 | |
| 31 | task = asyncio.create_task(server.serve(sockets=[sock])) |
| 32 | try: |
| 33 | while not server.started: |
| 34 | await asyncio.sleep(0.01) |
| 35 | yield f"http://127.0.0.1:{port}" |
| 36 | finally: |
| 37 | server.should_exit = True |
| 38 | try: |
| 39 | await asyncio.wait_for(task, timeout=5) |
| 40 | finally: |
| 41 | sock.close() |
| 42 | |
| 43 | |
| 44 | @pytest.mark.asyncio |
| 45 | async def test_unknown_namespace_connect_error_can_be_made_deterministic() -> None: |
| 46 | """ |
| 47 | Library-semantics test: demonstrate a deterministic connect_error payload shape for |
| 48 | unknown namespaces using a server-side allowlist gatekeeper. |
| 49 | """ |
| 50 | |
| 51 | import socketio |
| 52 | from socketio import packet |
| 53 | |
| 54 | sio = socketio.AsyncServer(async_mode="asgi", cors_allowed_origins="*", namespaces="*") |
| 55 | |
| 56 | allowed_namespaces = {"/known", "/"} |
| 57 | |
| 58 | original_handle_connect = sio._handle_connect |
| 59 | |
| 60 | async def _gatekeeper_handle_connect(eio_sid: str, namespace: str | None, data: Any) -> None: |
| 61 | namespace = namespace or "/" |
| 62 | if namespace not in allowed_namespaces: |
| 63 | await sio._send_packet( |
| 64 | eio_sid, |
| 65 | sio.packet_class( |
| 66 | packet.CONNECT_ERROR, |
| 67 | data={ |
| 68 | "message": "UNKNOWN_NAMESPACE", |
| 69 | "data": {"code": "UNKNOWN_NAMESPACE", "namespace": namespace}, |
| 70 | }, |
| 71 | namespace=namespace, |
| 72 | ), |
| 73 | ) |
| 74 | return |
| 75 | |
| 76 | await original_handle_connect(eio_sid, namespace, data) |
| 77 | |
| 78 | sio._handle_connect = _gatekeeper_handle_connect # type: ignore[assignment] |
| 79 | |
| 80 | app = socketio.ASGIApp(sio) |
| 81 | |
| 82 | async with _run_asgi_app(app) as base_url: |
| 83 | client = socketio.AsyncClient() |
| 84 | connect_error_fut: asyncio.Future[Any] = asyncio.get_running_loop().create_future() |
| 85 | |
| 86 | async def _on_connect_error(data: Any) -> None: |
| 87 | if not connect_error_fut.done(): |
| 88 | connect_error_fut.set_result(data) |
| 89 | |
| 90 | client.on("connect_error", _on_connect_error, namespace="/unknown") |
| 91 | |
| 92 | try: |
| 93 | with pytest.raises(socketio.exceptions.ConnectionError): |
| 94 | await client.connect(base_url, namespaces=["/unknown"]) |
| 95 | |
| 96 | err = await asyncio.wait_for(connect_error_fut, timeout=2) |
| 97 | assert err["message"] == "UNKNOWN_NAMESPACE" |
| 98 | assert err["data"] == {"code": "UNKNOWN_NAMESPACE", "namespace": "/unknown"} |
| 99 | finally: |
| 100 | try: |
| 101 | await client.disconnect() |
| 102 | except Exception: |
| 103 | pass |