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 onceThe 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.