testing server enter_room, receiving unexpected number of rooms #1442
-
I'm testing part of my application that uses the Here is a minimal, reproducible example. import asyncio
import typing as t
import pytest
import pytest_asyncio
import socketio
import uvicorn
class UvicornTestServer(uvicorn.Server):
def __init__(self, app, host: str = "127.0.0.1", port: int = 8000):
self._startup_done = asyncio.Event()
self._serve_task: t.Awaitable[t.Any] | None = None
super().__init__(
config=uvicorn.Config(app, host=host, port=port, loop="asyncio")
)
async def startup(self, sockets: list[t.Any] | None = None) -> None:
"""Override uvicorn startup."""
await super().startup(sockets=sockets)
self.config.setup_event_loop()
self._startup_done.set()
async def up(self) -> None:
"""Start up server asynchronously."""
self._serve_task = asyncio.create_task(self.serve())
await self._startup_done.wait()
async def down(self) -> None:
"""Shut down server asynchronously."""
self.should_exit = True
if self._serve_task:
await self._serve_task
@pytest_asyncio.fixture
async def sio() -> socketio.AsyncServer:
"""Create a Socket.IO server."""
sio = socketio.AsyncServer(
async_mode="asgi",
monitor_clients=False,
logger=True,
engineio_logger=True,
)
return sio
@pytest_asyncio.fixture()
async def startup_and_shutdown_server(
sio: socketio.AsyncServer,
) -> t.AsyncIterator[None]:
"""Start server as test fixture and tear down after test."""
server = UvicornTestServer(socketio.ASGIApp(sio))
await server.up()
yield
await server.down()
tasks = [
task
for task in asyncio.all_tasks(loop=asyncio.get_running_loop())
if not task.done()
]
for task in tasks:
task.cancel()
try:
await asyncio.wait(tasks)
except asyncio.CancelledError:
# If the task was cancelled, it will raise a CancelledError
# We can ignore this error as it is expected
pass
@pytest_asyncio.fixture
async def client() -> t.AsyncIterator[socketio.AsyncClient]:
"""Create a Socket.IO client."""
client = socketio.AsyncClient(logger=True, engineio_logger=True)
await client.connect("http://localhost:8000")
yield client
await client.disconnect()
@pytest.mark.asyncio
@pytest.mark.usefixtures("startup_and_shutdown_server")
async def test_join_room(
sio: socketio.AsyncServer, client: socketio.AsyncClient
) -> None:
"""Test that the client can join a room."""
sid = t.cast(str, client.get_sid())
await sio.enter_room(sid, "test")
rooms = sio.rooms(sid)
assert "test" in rooms
assert len(rooms) == 1 # this assertion fails, but I'm not sure why as I've only called enter_room once The final assertion is what I'm confused about. Here are the installed dependencies:
I appreciate any insight! |
Beta Was this translation helpful? Give feedback.
Replies: 1 comment 1 reply
-
Each client gets assigned a personal room when they connect. The name of this default room is the same as the connection's |
Beta Was this translation helpful? Give feedback.
Each client gets assigned a personal room when they connect. The name of this default room is the same as the connection's
sid
. If you add the client to one room, then you will get two rooms when you ask for the list.