Reporting to Socket.IO Admin UI (#1164)

.github/workflows/tests.yml

@@ -26,7 +26,7 @@ jobs:
         exclude:
           # pypy3 currently fails to run on Windows
           - os: windows-latest
-            python: pypy-3.8
+            python: pypy-3.9
       fail-fast: false
     runs-on: ${{ matrix.os }}
     steps:

docs/server.rst

@@ -617,6 +617,41 @@ callbacks when emitting. When the external process needs to receive callbacks,
 using a client to connect to the server with read and write support is a better
 option than a write-only client manager.
 
+Monitoring and Administration
+-----------------------------
+
+The Socket.IO server can be configured to accept connections from the official
+`Socket.IO Admin UI <https://socket.io/docs/v4/admin-ui/>`_. This tool provides
+real-time information about currently connected clients, rooms in use and
+events being emitted. It also allows an administrator to manually emit events,
+change room assignments and disconnect clients. The hosted version of this tool
+is available at `https://admin.socket.io <https://admin.socket.io>`_. 
+
+Given that enabling this feature can affect the performance of the server, it
+is disabled by default. To enable it, call the
+:func:`instrument() <socketio.Server.instrument>` method. For example::
+
+    import os
+    import socketio
+
+    sio = socketio.Server(cors_allowed_origins=[
+        'http://localhost:5000',
+        'https://admin.socket.io',
+    ])
+    sio.instrument(auth={
+        'username': 'admin',
+        'password': os.environ['ADMIN_PASSWORD'],
+    })
+
+This configures the server to accept connections from the hosted Admin UI
+client. Administrators can then open https://admin.socket.io in their web
+browsers and log in with username ``admin`` and the password given by the
+``ADMIN_PASSWORD`` environment variable. To ensure the Admin UI front end is
+allowed to connect, CORS is also configured.
+
+Consult the reference documentation to learn about additional configuration
+options that are available.
+
 Debugging and Troubleshooting
 -----------------------------
 

examples/server/asgi/app.py

@@ -1,9 +1,25 @@
 #!/usr/bin/env python
-import uvicorn
 
+# set instrument to `True` to accept connections from the official Socket.IO
+# Admin UI hosted at https://admin.socket.io
+instrument = False
+admin_login = {
+    'username': 'admin',
+    'password': 'python',  # change this to a strong secret for production use!
+}
+
+import uvicorn
 import socketio
 
-sio = socketio.AsyncServer(async_mode='asgi')
+sio = socketio.AsyncServer(
+    async_mode='asgi',
+    cors_allowed_origins=None if not instrument else [
+        'http://localhost:5000',
+        'https://admin.socket.io',  # edit the allowed origins if necessary
+    ])
+if instrument:
+    sio.instrument(auth=admin_login)
+
 app = socketio.ASGIApp(sio, static_files={
     '/': 'app.html',
 })

examples/server/wsgi/app.py

@@ -3,10 +3,26 @@
 # installed
 async_mode = None
 
+# set instrument to `True` to accept connections from the official Socket.IO
+# Admin UI hosted at https://admin.socket.io
+instrument = False
+admin_login = {
+    'username': 'admin',
+    'password': 'python',  # change this to a strong secret for production use!
+}
+
 from flask import Flask, render_template
 import socketio
 
-sio = socketio.Server(logger=True, async_mode=async_mode)
+sio = socketio.Server(
+    async_mode=async_mode,
+    cors_allowed_origins=None if not instrument else [
+        'http://localhost:5000',
+        'https://admin.socket.io',  # edit the allowed origins if necessary
+    ])
+if instrument:
+    sio.instrument(auth=admin_login)
+
 app = Flask(__name__)
 app.wsgi_app = socketio.WSGIApp(sio, app.wsgi_app)
 app.config['SECRET_KEY'] = 'secret!'

examples/server/wsgi/templates/index.html

@@ -6,7 +6,7 @@
     <script type="text/javascript" src="//cdnjs.cloudflare.com/ajax/libs/socket.io/4.7.2/socket.io.min.js"></script>
     <script type="text/javascript" charset="utf-8">
         $(document).ready(function(){
-            var socket = io.connect({transports: ['websocket']});
+            var socket = io.connect();
 
             socket.on('connect', function() {
                 socket.emit('my_event', {data: 'I\'m connected!'});

src/socketio/admin.py

@@ -0,0 +1,405 @@
+from datetime import datetime
+import functools
+import os
+import socket
+import time
+from urllib.parse import parse_qs
+from .exceptions import ConnectionRefusedError
+
+HOSTNAME = socket.gethostname()
+PID = os.getpid()
+
+
+class EventBuffer:
+    def __init__(self):
+        self.buffer = {}
+
+    def push(self, type, count=1):
+        timestamp = int(time.time()) * 1000
+        key = '{};{}'.format(timestamp, type)
+        if key not in self.buffer:
+            self.buffer[key] = {
+                'timestamp': timestamp,
+                'type': type,
+                'count': count,
+            }
+        else:
+            self.buffer[key]['count'] += count
+
+    def get_and_clear(self):
+        buffer = self.buffer
+        self.buffer = {}
+        return [value for value in buffer.values()]
+
+
+class InstrumentedServer:
+    def __init__(self, sio, auth=None, mode='development', read_only=False,
+                 server_id=None, namespace='/admin', server_stats_interval=2):
+        """Instrument the Socket.IO server for monitoring with the `Socket.IO
+        Admin UI <https://socket.io/docs/v4/admin-ui/>`_.
+        """
+        if auth is None:
+            raise ValueError('auth must be specified')
+        self.sio = sio
+        self.auth = auth
+        self.admin_namespace = namespace
+        self.read_only = read_only
+        self.server_id = server_id or (
+            self.sio.manager.host_id if hasattr(self.sio.manager, 'host_id')
+            else HOSTNAME
+        )
+        self.mode = mode
+        self.server_stats_interval = server_stats_interval
+        self.event_buffer = EventBuffer()
+
+        # task that emits "server_stats" every 2 seconds
+        self.stop_stats_event = None
+        self.stats_task = None
+
+        # monkey-patch the server to report metrics to the admin UI
+        self.instrument()
+
+    def instrument(self):
+        self.sio.on('connect', self.admin_connect,
+                    namespace=self.admin_namespace)
+
+        if self.mode == 'development':
+            if not self.read_only:  # pragma: no branch
+                self.sio.on('emit', self.admin_emit,
+                            namespace=self.admin_namespace)
+                self.sio.on('join', self.admin_enter_room,
+                            namespace=self.admin_namespace)
+                self.sio.on('leave', self.admin_leave_room,
+                            namespace=self.admin_namespace)
+                self.sio.on('_disconnect', self.admin_disconnect,
+                            namespace=self.admin_namespace)
+
+            # track socket connection times
+            self.sio.manager._timestamps = {}
+
+            # report socket.io connections
+            self.sio.manager.__connect = self.sio.manager.connect
+            self.sio.manager.connect = self._connect
+
+            # report socket.io disconnection
+            self.sio.manager.__disconnect = self.sio.manager.disconnect
+            self.sio.manager.disconnect = self._disconnect
+
+            # report join rooms
+            self.sio.manager.__basic_enter_room = \
+                self.sio.manager.basic_enter_room
+            self.sio.manager.basic_enter_room = self._basic_enter_room
+
+            # report leave rooms
+            self.sio.manager.__basic_leave_room = \
+                self.sio.manager.basic_leave_room
+            self.sio.manager.basic_leave_room = self._basic_leave_room
+
+            # report emit events
+            self.sio.manager.__emit = self.sio.manager.emit
+            self.sio.manager.emit = self._emit
+
+            # report receive events
+            self.sio.__handle_event_internal = self.sio._handle_event_internal
+            self.sio._handle_event_internal = self._handle_event_internal
+
+        # report engine.io connections
+        self.sio.eio.on('connect', self._handle_eio_connect)
+        self.sio.eio.on('disconnect', self._handle_eio_disconnect)
+
+        # report polling packets
+        from engineio.socket import Socket
+        self.sio.eio.__ok = self.sio.eio._ok
+        self.sio.eio._ok = self._eio_http_response
+        Socket.__handle_post_request = Socket.handle_post_request
+        Socket.handle_post_request = functools.partialmethod(
+            self.__class__._eio_handle_post_request, self)
+
+        # report websocket packets
+        Socket.__websocket_handler = Socket._websocket_handler
+        Socket._websocket_handler = functools.partialmethod(
+            self.__class__._eio_websocket_handler, self)
+
+        # report connected sockets with each ping
+        if self.mode == 'development':
+            Socket.__send_ping = Socket._send_ping
+            Socket._send_ping = functools.partialmethod(
+                self.__class__._eio_send_ping, self)
+
+    def uninstrument(self):  # pragma: no cover
+        if self.mode == 'development':
+            self.sio.manager.connect = self.sio.manager.__connect
+            self.sio.manager.disconnect = self.sio.manager.__disconnect
+            self.sio.manager.basic_enter_room = \
+                self.sio.manager.__basic_enter_room
+            self.sio.manager.basic_leave_room = \
+                self.sio.manager.__basic_leave_room
+            self.sio.manager.emit = self.sio.manager.__emit
+            self.sio._handle_event_internal = self.sio.__handle_event_internal
+        self.sio.eio._ok = self.sio.eio.__ok
+
+        from engineio.socket import Socket
+        Socket.handle_post_request = Socket.__handle_post_request
+        Socket._websocket_handler = Socket.__websocket_handler
+        if self.mode == 'development':
+            Socket._send_ping = Socket.__send_ping
+
+    def admin_connect(self, sid, environ, client_auth):
+        if self.auth:
+            authenticated = False
+            if isinstance(self.auth, dict):
+                authenticated = client_auth == self.auth
+            elif isinstance(self.auth, list):
+                authenticated = client_auth in self.auth
+            else:
+                authenticated = self.auth(client_auth)
+            if not authenticated:
+                raise ConnectionRefusedError('authentication failed')
+
+        def config(sid):
+            self.sio.sleep(0.1)
+
+            # supported features
+            features = ['AGGREGATED_EVENTS']
+            if not self.read_only:
+                features += ['EMIT', 'JOIN', 'LEAVE', 'DISCONNECT', 'MJOIN',
+                             'MLEAVE', 'MDISCONNECT']
+            if self.mode == 'development':
+                features.append('ALL_EVENTS')
+            self.sio.emit('config', {'supportedFeatures': features},
+                          to=sid, namespace=self.admin_namespace)
+
+            # send current sockets
+            if self.mode == 'development':
+                all_sockets = []
+                for nsp in self.sio.manager.get_namespaces():
+                    for sid, eio_sid in self.sio.manager.get_participants(
+                            nsp, None):
+                        all_sockets.append(
+                            self.serialize_socket(sid, nsp, eio_sid))
+                self.sio.emit('all_sockets', all_sockets, to=sid,
+                              namespace=self.admin_namespace)
+
+        self.sio.start_background_task(config, sid)
+
+    def admin_emit(self, _, namespace, room_filter, event, *data):
+        self.sio.emit(event, data, to=room_filter, namespace=namespace)
+
+    def admin_enter_room(self, _, namespace, room, room_filter=None):
+        for sid, _ in self.sio.manager.get_participants(
+                namespace, room_filter):
+            self.sio.enter_room(sid, room, namespace=namespace)
+
+    def admin_leave_room(self, _, namespace, room, room_filter=None):
+        for sid, _ in self.sio.manager.get_participants(
+                namespace, room_filter):
+            self.sio.leave_room(sid, room, namespace=namespace)
+
+    def admin_disconnect(self, _, namespace, close, room_filter=None):
+        for sid, _ in self.sio.manager.get_participants(
+                namespace, room_filter):
+            self.sio.disconnect(sid, namespace=namespace)
+
+    def shutdown(self):
+        if self.stats_task:  # pragma: no branch
+            self.stop_stats_event.set()
+            self.stats_task.join()
+
+    def _connect(self, eio_sid, namespace):
+        sid = self.sio.manager.__connect(eio_sid, namespace)
+        t = time.time()
+        self.sio.manager._timestamps[sid] = t
+        serialized_socket = self.serialize_socket(sid, namespace, eio_sid)
+        self.sio.emit('socket_connected', (
+            serialized_socket,
+            datetime.utcfromtimestamp(t).isoformat() + 'Z',
+        ), namespace=self.admin_namespace)
+        return sid
+
+    def _disconnect(self, sid, namespace, **kwargs):
+        del self.sio.manager._timestamps[sid]
+        self.sio.emit('socket_disconnected', (
+            namespace,
+            sid,
+            'N/A',
+            datetime.utcnow().isoformat() + 'Z',
+        ), namespace=self.admin_namespace)
+        return self.sio.manager.__disconnect(sid, namespace, **kwargs)
+
+    def _check_for_upgrade(self, eio_sid, sid, namespace):  # pragma: no cover
+        for _ in range(5):
+            self.sio.sleep(5)
+            try:
+                if self.sio.eio._get_socket(eio_sid).upgraded:
+                    self.sio.emit('socket_updated', {
+                        'id': sid,
+                        'nsp': namespace,
+                        'transport': 'websocket',
+                    }, namespace=self.admin_namespace)
+                    break
+            except KeyError:
+                pass
+
+    def _basic_enter_room(self, sid, namespace, room, eio_sid=None):
+        ret = self.sio.manager.__basic_enter_room(sid, namespace, room,
+                                                  eio_sid)
+        if room:
+            self.sio.emit('room_joined', (
+                namespace,
+                room,
+                sid,
+                datetime.utcnow().isoformat() + 'Z',
+            ), namespace=self.admin_namespace)
+        return ret
+
+    def _basic_leave_room(self, sid, namespace, room):
+        if room:
+            self.sio.emit('room_left', (
+                namespace,
+                room,
+                sid,
+                datetime.utcnow().isoformat() + 'Z',
+            ), namespace=self.admin_namespace)
+        return self.sio.manager.__basic_leave_room(sid, namespace, room)
+
+    def _emit(self, event, data, namespace, room=None, skip_sid=None,
+              callback=None, **kwargs):
+        ret = self.sio.manager.__emit(event, data, namespace, room=room,
+                                      skip_sid=skip_sid, callback=callback,
+                                      **kwargs)
+        if namespace != self.admin_namespace:
+            event_data = [event] + list(data) if isinstance(data, tuple) \
+                else [data]
+            if not isinstance(skip_sid, list):  # pragma: no branch
+                skip_sid = [skip_sid]
+            for sid, _ in self.sio.manager.get_participants(namespace, room):
+                if sid not in skip_sid:
+                    self.sio.emit('event_sent', (
+                        namespace,
+                        sid,
+                        event_data,
+                        datetime.utcnow().isoformat() + 'Z',
+                    ), namespace=self.admin_namespace)
+        return ret
+
+    def _handle_event_internal(self, server, sid, eio_sid, data, namespace,
+                               id):
+        ret = self.sio.__handle_event_internal(server, sid, eio_sid, data,
+                                               namespace, id)
+        self.sio.emit('event_received', (
+            namespace,
+            sid,
+            data,
+            datetime.utcnow().isoformat() + 'Z',
+        ), namespace=self.admin_namespace)
+        return ret
+
+    def _handle_eio_connect(self, eio_sid, environ):
+        if self.stop_stats_event is None:
+            self.stop_stats_event = self.sio.eio.create_event()
+            self.stats_task = self.sio.start_background_task(
+                self._emit_server_stats)
+
+        self.event_buffer.push('rawConnection')
+        return self.sio._handle_eio_connect(eio_sid, environ)
+
+    def _handle_eio_disconnect(self, eio_sid):
+        self.event_buffer.push('rawDisconnection')
+        return self.sio._handle_eio_disconnect(eio_sid)
+
+    def _eio_http_response(self, packets=None, headers=None, jsonp_index=None):
+        ret = self.sio.eio.__ok(packets=packets, headers=headers,
+                                jsonp_index=jsonp_index)
+        self.event_buffer.push('packetsOut')
+        self.event_buffer.push('bytesOut', len(ret['response']))
+        return ret
+
+    def _eio_handle_post_request(socket, self, environ):
+        ret = socket.__handle_post_request(environ)
+        self.event_buffer.push('packetsIn')
+        self.event_buffer.push(
+            'bytesIn', int(environ.get('CONTENT_LENGTH', 0)))
+        return ret
+
+    def _eio_websocket_handler(socket, self, ws):
+        def _send(ws, data, *args, **kwargs):
+            self.event_buffer.push('packetsOut')
+            self.event_buffer.push('bytesOut', len(data))
+            return ws.__send(data, *args, **kwargs)
+
+        def _wait(ws):
+            ret = ws.__wait()
+            self.event_buffer.push('packetsIn')
+            self.event_buffer.push('bytesIn', len(ret or ''))
+            return ret
+
+        ws.__send = ws.send
+        ws.send = functools.partial(_send, ws)
+        ws.__wait = ws.wait
+        ws.wait = functools.partial(_wait, ws)
+        return socket.__websocket_handler(ws)
+
+    def _eio_send_ping(socket, self):  # pragma: no cover
+        eio_sid = socket.sid
+        t = time.time()
+        for namespace in self.sio.manager.get_namespaces():
+            sid = self.sio.manager.sid_from_eio_sid(eio_sid, namespace)
+            if sid:
+                serialized_socket = self.serialize_socket(sid, namespace,
+                                                          eio_sid)
+                self.sio.emit('socket_connected', (
+                    serialized_socket,
+                    datetime.utcfromtimestamp(t).isoformat() + 'Z',
+                ), namespace=self.admin_namespace)
+        return socket.__send_ping()
+
+    def _emit_server_stats(self):
+        start_time = time.time()
+        namespaces = list(self.sio.handlers.keys())
+        namespaces.sort()
+        while not self.stop_stats_event.is_set():
+            self.sio.sleep(self.server_stats_interval)
+            self.sio.emit('server_stats', {
+                'serverId': self.server_id,
+                'hostname': HOSTNAME,
+                'pid': PID,
+                'uptime': time.time() - start_time,
+                'clientsCount': len(self.sio.eio.sockets),
+                'pollingClientsCount': len(
+                    [s for s in self.sio.eio.sockets.values()
+                     if not s.upgraded]),
+                'aggregatedEvents': self.event_buffer.get_and_clear(),
+                'namespaces': [{
+                    'name': nsp,
+                    'socketsCount': len(self.sio.manager.rooms.get(
+                        nsp, {None: []}).get(None, []))
+                } for nsp in namespaces],
+            }, namespace=self.admin_namespace)
+
+    def serialize_socket(self, sid, namespace, eio_sid=None):
+        if eio_sid is None:  # pragma: no cover
+            eio_sid = self.sio.manager.eio_sid_from_sid(sid)
+        socket = self.sio.eio._get_socket(eio_sid)
+        environ = self.sio.environ.get(eio_sid, {})
+        tm = self.sio.manager._timestamps[sid] if sid in \
+            self.sio.manager._timestamps else 0
+        return {
+            'id': sid,
+            'clientId': eio_sid,
+            'transport': 'websocket' if socket.upgraded else 'polling',
+            'nsp': namespace,
+            'data': {},
+            'handshake': {
+                'address': environ.get('REMOTE_ADDR', ''),
+                'headers': {k[5:].lower(): v for k, v in environ.items()
+                            if k.startswith('HTTP_')},
+                'query': {k: v[0] if len(v) == 1 else v for k, v in parse_qs(
+                    environ.get('QUERY_STRING', '')).items()},
+                'secure': environ.get('wsgi.url_scheme', '') == 'https',
+                'url': environ.get('PATH_INFO', ''),
+                'issued': tm * 1000,
+                'time': datetime.utcfromtimestamp(tm).isoformat() + 'Z'
+                if tm else '',
+            },
+            'rooms': self.sio.manager.get_rooms(sid, namespace),
+        }

src/socketio/async_admin.py

@@ -0,0 +1,398 @@
+import asyncio
+from datetime import datetime
+import functools
+import os
+import socket
+import time
+from urllib.parse import parse_qs
+from .admin import EventBuffer
+from .exceptions import ConnectionRefusedError
+
+HOSTNAME = socket.gethostname()
+PID = os.getpid()
+
+
+class InstrumentedAsyncServer:
+    def __init__(self, sio, auth=None, namespace='/admin', read_only=False,
+                 server_id=None, mode='development', server_stats_interval=2):
+        """Instrument the Socket.IO server for monitoring with the `Socket.IO
+        Admin UI <https://socket.io/docs/v4/admin-ui/>`_.
+        """
+        if auth is None:
+            raise ValueError('auth must be specified')
+        self.sio = sio
+        self.auth = auth
+        self.admin_namespace = namespace
+        self.read_only = read_only
+        self.server_id = server_id or (
+            self.sio.manager.host_id if hasattr(self.sio.manager, 'host_id')
+            else HOSTNAME
+        )
+        self.mode = mode
+        self.server_stats_interval = server_stats_interval
+        self.admin_queue = []
+        self.event_buffer = EventBuffer()
+
+        # task that emits "server_stats" every 2 seconds
+        self.stop_stats_event = None
+        self.stats_task = None
+
+        # monkey-patch the server to report metrics to the admin UI
+        self.instrument()
+
+    def instrument(self):
+        self.sio.on('connect', self.admin_connect,
+                    namespace=self.admin_namespace)
+
+        if self.mode == 'development':
+            if not self.read_only:  # pragma: no branch
+                self.sio.on('emit', self.admin_emit,
+                            namespace=self.admin_namespace)
+                self.sio.on('join', self.admin_enter_room,
+                            namespace=self.admin_namespace)
+                self.sio.on('leave', self.admin_leave_room,
+                            namespace=self.admin_namespace)
+                self.sio.on('_disconnect', self.admin_disconnect,
+                            namespace=self.admin_namespace)
+
+            # track socket connection times
+            self.sio.manager._timestamps = {}
+
+            # report socket.io connections
+            self.sio.manager.__connect = self.sio.manager.connect
+            self.sio.manager.connect = self._connect
+
+            # report socket.io disconnection
+            self.sio.manager.__disconnect = self.sio.manager.disconnect
+            self.sio.manager.disconnect = self._disconnect
+
+            # report join rooms
+            self.sio.manager.__basic_enter_room = \
+                self.sio.manager.basic_enter_room
+            self.sio.manager.basic_enter_room = self._basic_enter_room
+
+            # report leave rooms
+            self.sio.manager.__basic_leave_room = \
+                self.sio.manager.basic_leave_room
+            self.sio.manager.basic_leave_room = self._basic_leave_room
+
+            # report emit events
+            self.sio.manager.__emit = self.sio.manager.emit
+            self.sio.manager.emit = self._emit
+
+            # report receive events
+            self.sio.__handle_event_internal = self.sio._handle_event_internal
+            self.sio._handle_event_internal = self._handle_event_internal
+
+        # report engine.io connections
+        self.sio.eio.on('connect', self._handle_eio_connect)
+        self.sio.eio.on('disconnect', self._handle_eio_disconnect)
+
+        # report polling packets
+        from engineio.async_socket import AsyncSocket
+        self.sio.eio.__ok = self.sio.eio._ok
+        self.sio.eio._ok = self._eio_http_response
+        AsyncSocket.__handle_post_request = AsyncSocket.handle_post_request
+        AsyncSocket.handle_post_request = functools.partialmethod(
+            self.__class__._eio_handle_post_request, self)
+
+        # report websocket packets
+        AsyncSocket.__websocket_handler = AsyncSocket._websocket_handler
+        AsyncSocket._websocket_handler = functools.partialmethod(
+            self.__class__._eio_websocket_handler, self)
+
+        # report connected sockets with each ping
+        if self.mode == 'development':
+            AsyncSocket.__send_ping = AsyncSocket._send_ping
+            AsyncSocket._send_ping = functools.partialmethod(
+                self.__class__._eio_send_ping, self)
+
+    def uninstrument(self):  # pragma: no cover
+        if self.mode == 'development':
+            self.sio.manager.connect = self.sio.manager.__connect
+            self.sio.manager.disconnect = self.sio.manager.__disconnect
+            self.sio.manager.basic_enter_room = \
+                self.sio.manager.__basic_enter_room
+            self.sio.manager.basic_leave_room = \
+                self.sio.manager.__basic_leave_room
+            self.sio.manager.emit = self.sio.manager.__emit
+            self.sio._handle_event_internal = self.sio.__handle_event_internal
+        self.sio.eio._ok = self.sio.eio.__ok
+
+        from engineio.async_socket import AsyncSocket
+        AsyncSocket.handle_post_request = AsyncSocket.__handle_post_request
+        AsyncSocket._websocket_handler = AsyncSocket.__websocket_handler
+        if self.mode == 'development':
+            AsyncSocket._send_ping = AsyncSocket.__send_ping
+
+    async def admin_connect(self, sid, environ, client_auth):
+        authenticated = True
+        if self.auth:
+            authenticated = False
+            if isinstance(self.auth, dict):
+                authenticated = client_auth == self.auth
+            elif isinstance(self.auth, list):
+                authenticated = client_auth in self.auth
+            else:
+                if asyncio.iscoroutinefunction(self.auth):
+                    authenticated = await self.auth(client_auth)
+                else:
+                    authenticated = self.auth(client_auth)
+            if not authenticated:
+                raise ConnectionRefusedError('authentication failed')
+
+        async def config(sid):
+            await self.sio.sleep(0.1)
+
+            # supported features
+            features = ['AGGREGATED_EVENTS']
+            if not self.read_only:
+                features += ['EMIT', 'JOIN', 'LEAVE', 'DISCONNECT', 'MJOIN',
+                             'MLEAVE', 'MDISCONNECT']
+            if self.mode == 'development':
+                features.append('ALL_EVENTS')
+            await self.sio.emit('config', {'supportedFeatures': features},
+                                to=sid, namespace=self.admin_namespace)
+
+            # send current sockets
+            if self.mode == 'development':
+                all_sockets = []
+                for nsp in self.sio.manager.get_namespaces():
+                    for sid, eio_sid in self.sio.manager.get_participants(
+                            nsp, None):
+                        all_sockets.append(
+                            self.serialize_socket(sid, nsp, eio_sid))
+                await self.sio.emit('all_sockets', all_sockets, to=sid,
+                                    namespace=self.admin_namespace)
+
+        self.sio.start_background_task(config, sid)
+        self.stop_stats_event = self.sio.eio.create_event()
+        self.stats_task = self.sio.start_background_task(
+            self._emit_server_stats)
+
+    async def admin_emit(self, _, namespace, room_filter, event, *data):
+        await self.sio.emit(event, data, to=room_filter, namespace=namespace)
+
+    async def admin_enter_room(self, _, namespace, room, room_filter=None):
+        for sid, _ in self.sio.manager.get_participants(
+                namespace, room_filter):
+            await self.sio.enter_room(sid, room, namespace=namespace)
+
+    async def admin_leave_room(self, _, namespace, room, room_filter=None):
+        for sid, _ in self.sio.manager.get_participants(
+                namespace, room_filter):
+            await self.sio.leave_room(sid, room, namespace=namespace)
+
+    async def admin_disconnect(self, _, namespace, close, room_filter=None):
+        for sid, _ in self.sio.manager.get_participants(
+                namespace, room_filter):
+            await self.sio.disconnect(sid, namespace=namespace)
+
+    async def shutdown(self):
+        if self.stats_task:  # pragma: no branch
+            self.stop_stats_event.set()
+            await asyncio.gather(self.stats_task)
+
+    async def _connect(self, eio_sid, namespace):
+        sid = await self.sio.manager.__connect(eio_sid, namespace)
+        t = time.time()
+        self.sio.manager._timestamps[sid] = t
+        serialized_socket = self.serialize_socket(sid, namespace, eio_sid)
+        await self.sio.emit('socket_connected', (
+            serialized_socket,
+            datetime.utcfromtimestamp(t).isoformat() + 'Z',
+        ), namespace=self.admin_namespace)
+        return sid
+
+    async def _disconnect(self, sid, namespace, **kwargs):
+        del self.sio.manager._timestamps[sid]
+        await self.sio.emit('socket_disconnected', (
+            namespace,
+            sid,
+            'N/A',
+            datetime.utcnow().isoformat() + 'Z',
+        ), namespace=self.admin_namespace)
+        return await self.sio.manager.__disconnect(sid, namespace, **kwargs)
+
+    async def _check_for_upgrade(self, eio_sid, sid,
+                                 namespace):  # pragma: no cover
+        for _ in range(5):
+            await self.sio.sleep(5)
+            try:
+                if self.sio.eio._get_socket(eio_sid).upgraded:
+                    await self.sio.emit('socket_updated', {
+                        'id': sid,
+                        'nsp': namespace,
+                        'transport': 'websocket',
+                    }, namespace=self.admin_namespace)
+                    break
+            except KeyError:
+                pass
+
+    def _basic_enter_room(self, sid, namespace, room, eio_sid=None):
+        ret = self.sio.manager.__basic_enter_room(sid, namespace, room,
+                                                  eio_sid)
+        if room:
+            self.admin_queue.append(('room_joined', (
+                namespace,
+                room,
+                sid,
+                datetime.utcnow().isoformat() + 'Z',
+            )))
+        return ret
+
+    def _basic_leave_room(self, sid, namespace, room):
+        if room:
+            self.admin_queue.append(('room_left', (
+                namespace,
+                room,
+                sid,
+                datetime.utcnow().isoformat() + 'Z',
+            )))
+        return self.sio.manager.__basic_leave_room(sid, namespace, room)
+
+    async def _emit(self, event, data, namespace, room=None, skip_sid=None,
+                    callback=None, **kwargs):
+        ret = await self.sio.manager.__emit(
+            event, data, namespace, room=room, skip_sid=skip_sid,
+            callback=callback, **kwargs)
+        if namespace != self.admin_namespace:
+            event_data = [event] + list(data) if isinstance(data, tuple) \
+                else [data]
+            if not isinstance(skip_sid, list):  # pragma: no branch
+                skip_sid = [skip_sid]
+            for sid, _ in self.sio.manager.get_participants(namespace, room):
+                if sid not in skip_sid:
+                    await self.sio.emit('event_sent', (
+                        namespace,
+                        sid,
+                        event_data,
+                        datetime.utcnow().isoformat() + 'Z',
+                    ), namespace=self.admin_namespace)
+        return ret
+
+    async def _handle_event_internal(self, server, sid, eio_sid, data,
+                                     namespace, id):
+        ret = await self.sio.__handle_event_internal(server, sid, eio_sid,
+                                                     data, namespace, id)
+        await self.sio.emit('event_received', (
+            namespace,
+            sid,
+            data,
+            datetime.utcnow().isoformat() + 'Z',
+        ), namespace=self.admin_namespace)
+        return ret
+
+    async def _handle_eio_connect(self, eio_sid, environ):
+        if self.stop_stats_event is None:
+            self.stop_stats_event = self.sio.eio.create_event()
+            self.stats_task = self.sio.start_background_task(
+                self._emit_server_stats)
+
+        self.event_buffer.push('rawConnection')
+        return await self.sio._handle_eio_connect(eio_sid, environ)
+
+    async def _handle_eio_disconnect(self, eio_sid):
+        self.event_buffer.push('rawDisconnection')
+        return await self.sio._handle_eio_disconnect(eio_sid)
+
+    def _eio_http_response(self, packets=None, headers=None, jsonp_index=None):
+        ret = self.sio.eio.__ok(packets=packets, headers=headers,
+                                jsonp_index=jsonp_index)
+        self.event_buffer.push('packetsOut')
+        self.event_buffer.push('bytesOut', len(ret['response']))
+        return ret
+
+    async def _eio_handle_post_request(socket, self, environ):
+        ret = await socket.__handle_post_request(environ)
+        self.event_buffer.push('packetsIn')
+        self.event_buffer.push(
+            'bytesIn', int(environ.get('CONTENT_LENGTH', 0)))
+        return ret
+
+    async def _eio_websocket_handler(socket, self, ws):
+        async def _send(ws, data):
+            self.event_buffer.push('packetsOut')
+            self.event_buffer.push('bytesOut', len(data))
+            return await ws.__send(data)
+
+        async def _wait(ws):
+            ret = await ws.__wait()
+            self.event_buffer.push('packetsIn')
+            self.event_buffer.push('bytesIn', len(ret or ''))
+            return ret
+
+        ws.__send = ws.send
+        ws.send = functools.partial(_send, ws)
+        ws.__wait = ws.wait
+        ws.wait = functools.partial(_wait, ws)
+        return await socket.__websocket_handler(ws)
+
+    async def _eio_send_ping(socket, self):  # pragma: no cover
+        eio_sid = socket.sid
+        t = time.time()
+        for namespace in self.sio.manager.get_namespaces():
+            sid = self.sio.manager.sid_from_eio_sid(eio_sid, namespace)
+            if sid:
+                serialized_socket = self.serialize_socket(sid, namespace,
+                                                          eio_sid)
+                await self.sio.emit('socket_connected', (
+                    serialized_socket,
+                    datetime.utcfromtimestamp(t).isoformat() + 'Z',
+                ), namespace=self.admin_namespace)
+        return await socket.__send_ping()
+
+    async def _emit_server_stats(self):
+        start_time = time.time()
+        namespaces = list(self.sio.handlers.keys())
+        namespaces.sort()
+        while not self.stop_stats_event.is_set():
+            await self.sio.sleep(self.server_stats_interval)
+            await self.sio.emit('server_stats', {
+                'serverId': self.server_id,
+                'hostname': HOSTNAME,
+                'pid': PID,
+                'uptime': time.time() - start_time,
+                'clientsCount': len(self.sio.eio.sockets),
+                'pollingClientsCount': len(
+                    [s for s in self.sio.eio.sockets.values()
+                     if not s.upgraded]),
+                'aggregatedEvents': self.event_buffer.get_and_clear(),
+                'namespaces': [{
+                    'name': nsp,
+                    'socketsCount': len(self.sio.manager.rooms.get(
+                        nsp, {None: []}).get(None, []))
+                } for nsp in namespaces],
+            }, namespace=self.admin_namespace)
+            while self.admin_queue:
+                event, args = self.admin_queue.pop(0)
+                await self.sio.emit(event, args,
+                                    namespace=self.admin_namespace)
+
+    def serialize_socket(self, sid, namespace, eio_sid=None):
+        if eio_sid is None:  # pragma: no cover
+            eio_sid = self.sio.manager.eio_sid_from_sid(sid)
+        socket = self.sio.eio._get_socket(eio_sid)
+        environ = self.sio.environ.get(eio_sid, {})
+        tm = self.sio.manager._timestamps[sid] if sid in \
+            self.sio.manager._timestamps else 0
+        return {
+            'id': sid,
+            'clientId': eio_sid,
+            'transport': 'websocket' if socket.upgraded else 'polling',
+            'nsp': namespace,
+            'data': {},
+            'handshake': {
+                'address': environ.get('REMOTE_ADDR', ''),
+                'headers': {k[5:].lower(): v for k, v in environ.items()
+                            if k.startswith('HTTP_')},
+                'query': {k: v[0] if len(v) == 1 else v for k, v in parse_qs(
+                    environ.get('QUERY_STRING', '')).items()},
+                'secure': environ.get('wsgi.url_scheme', '') == 'https',
+                'url': environ.get('PATH_INFO', ''),
+                'issued': tm * 1000,
+                'time': datetime.utcfromtimestamp(tm).isoformat() + 'Z'
+                if tm else '',
+            },
+            'rooms': self.sio.manager.get_rooms(sid, namespace),
+        }

src/socketio/async_manager.py

@@ -62,6 +62,13 @@ class AsyncManager(BaseManager):
             return
         await asyncio.wait(tasks)
 
+    async def connect(self, eio_sid, namespace):
+        """Register a client connection to a namespace.
+
+        Note: this method is a coroutine.
+        """
+        return super().connect(eio_sid, namespace)
+
     async def disconnect(self, sid, namespace, **kwargs):
         """Disconnect a client.
 

src/socketio/async_server.py

@@ -467,6 +467,45 @@ class AsyncServer(base_server.BaseServer):
         """
         return await self.eio.sleep(seconds)
 
+    def instrument(self, auth=None, mode='development', read_only=False,
+                   server_id=None, namespace='/admin',
+                   server_stats_interval=2):
+        """Instrument the Socket.IO server for monitoring with the `Socket.IO
+        Admin UI <https://socket.io/docs/v4/admin-ui/>`_.
+
+        :param auth: Authentication credentials for Admin UI access. Set to a
+                     dictionary with the expected login (usually ``username``
+                     and ``password``) or a list of dictionaries if more than
+                     one set of credentials need to be available. For more
+                     complex authentication methods, set to a callable that
+                     receives the authentication dictionary as an argument and
+                     returns ``True`` if the user is allowed or ``False``
+                     otherwise. To disable authentication, set this argument to
+                     ``False`` (not recommended, never do this on a production
+                     server).
+        :param mode: The reporting mode. The default is ``'development'``,
+                     which is best used while debugging, as it may have a
+                     significant performance effect. Set to ``'production'`` to
+                     reduce the amount of information that is reported to the
+                     admin UI.
+        :param read_only: If set to ``True``, the admin interface will be
+                          read-only, with no option to modify room assignments
+                          or disconnect clients. The default is ``False``.
+        :param server_id: The server name to use for this server. If this
+                          argument is omitted, the server generates its own
+                          name.
+        :param namespace: The Socket.IO namespace to use for the admin
+                          interface. The default is ``/admin``.
+        :param server_stats_interval: The interval in seconds at which the
+                                      server emits a summary of it stats to all
+                                      connected admins.
+        """
+        from .async_admin import InstrumentedAsyncServer
+        return InstrumentedAsyncServer(
+            self, auth=auth, mode=mode, read_only=read_only,
+            server_id=server_id, namespace=namespace,
+            server_stats_interval=server_stats_interval)
+
     async def _send_packet(self, eio_sid, pkt):
         """Send a Socket.IO packet to a client."""
         encoded_packet = pkt.encode()
@@ -486,7 +525,7 @@ class AsyncServer(base_server.BaseServer):
         sid = None
         if namespace in self.handlers or namespace in self.namespace_handlers \
                 or self.namespaces == '*' or namespace in self.namespaces:
-            sid = self.manager.connect(eio_sid, namespace)
+            sid = await self.manager.connect(eio_sid, namespace)
         if sid is None:
             await self._send_packet(eio_sid, self.packet_class(
                 packet.CONNECT_ERROR, data='Unable to connect',

src/socketio/async_simple_client.py

@@ -23,7 +23,8 @@ class AsyncSimpleClient:
         self.input_buffer = []
 
     async def connect(self, url, headers={}, auth=None, transports=None,
-                      namespace='/', socketio_path='socket.io'):
+                      namespace='/', socketio_path='socket.io',
+                      wait_timeout=5):
         """Connect to a Socket.IO server.
 
         :param url: The URL of the Socket.IO server. It can include custom
@@ -49,6 +50,8 @@ class AsyncSimpleClient:
         :param socketio_path: The endpoint where the Socket.IO server is
                               installed. The default value is appropriate for
                               most cases.
+        :param wait_timeout: How long the client should wait for the
+                             connection. The default is 5 seconds.
 
         Note: this method is a coroutine.
         """
@@ -80,7 +83,8 @@ class AsyncSimpleClient:
 
         await self.client.connect(
             url, headers=headers, auth=auth, transports=transports,
-            namespaces=[namespace], socketio_path=socketio_path)
+            namespaces=[namespace], socketio_path=socketio_path,
+            wait_timeout=wait_timeout)
 
     @property
     def sid(self):
@@ -89,7 +93,7 @@ class AsyncSimpleClient:
         The session ID is not guaranteed to remain constant throughout the life
         of the connection, as reconnections can cause it to change.
         """
-        return self.client.sid if self.client else None
+        return self.client.get_sid(self.namespace) if self.client else None
 
     @property
     def transport(self):

src/socketio/base_manager.py

@@ -30,7 +30,7 @@ class BaseManager:
 
     def get_participants(self, namespace, room):
         """Return an iterable with the active participants in a room."""
-        ns = self.rooms[namespace]
+        ns = self.rooms.get(namespace, {})
         if hasattr(room, '__len__') and not isinstance(room, str):
             participants = ns[room[0]]._fwdm.copy() if room[0] in ns else {}
             for r in room[1:]:
@@ -126,7 +126,7 @@ class BaseManager:
         try:
             for sid, _ in self.get_participants(namespace, room):
                 self.basic_leave_room(sid, namespace, room)
-        except KeyError:
+        except KeyError:  # pragma: no cover
             pass
 
     def get_rooms(self, sid, namespace):

src/socketio/server.py

@@ -454,6 +454,45 @@ class Server(base_server.BaseServer):
         """
         return self.eio.sleep(seconds)
 
+    def instrument(self, auth=None, mode='development', read_only=False,
+                   server_id=None, namespace='/admin',
+                   server_stats_interval=2):
+        """Instrument the Socket.IO server for monitoring with the `Socket.IO
+        Admin UI <https://socket.io/docs/v4/admin-ui/>`_.
+
+        :param auth: Authentication credentials for Admin UI access. Set to a
+                     dictionary with the expected login (usually ``username``
+                     and ``password``) or a list of dictionaries if more than
+                     one set of credentials need to be available. For more
+                     complex authentication methods, set to a callable that
+                     receives the authentication dictionary as an argument and
+                     returns ``True`` if the user is allowed or ``False``
+                     otherwise. To disable authentication, set this argument to
+                     ``False`` (not recommended, never do this on a production
+                     server).
+        :param mode: The reporting mode. The default is ``'development'``,
+                     which is best used while debugging, as it may have a
+                     significant performance effect. Set to ``'production'`` to
+                     reduce the amount of information that is reported to the
+                     admin UI.
+        :param read_only: If set to ``True``, the admin interface will be
+                          read-only, with no option to modify room assignments
+                          or disconnect clients. The default is ``False``.
+        :param server_id: The server name to use for this server. If this
+                          argument is omitted, the server generates its own
+                          name.
+        :param namespace: The Socket.IO namespace to use for the admin
+                          interface. The default is ``/admin``.
+        :param server_stats_interval: The interval in seconds at which the
+                                      server emits a summary of it stats to all
+                                      connected admins.
+        """
+        from .admin import InstrumentedServer
+        return InstrumentedServer(
+            self, auth=auth, mode=mode, read_only=read_only,
+            server_id=server_id, namespace=namespace,
+            server_stats_interval=server_stats_interval)
+
     def _send_packet(self, eio_sid, pkt):
         """Send a Socket.IO packet to a client."""
         encoded_packet = pkt.encode()

src/socketio/simple_client.py

@@ -23,7 +23,7 @@ class SimpleClient:
         self.input_buffer = []
 
     def connect(self, url, headers={}, auth=None, transports=None,
-                namespace='/', socketio_path='socket.io'):
+                namespace='/', socketio_path='socket.io', wait_timeout=5):
         """Connect to a Socket.IO server.
 
         :param url: The URL of the Socket.IO server. It can include custom
@@ -49,6 +49,9 @@ class SimpleClient:
         :param socketio_path: The endpoint where the Socket.IO server is
                               installed. The default value is appropriate for
                               most cases.
+        :param wait_timeout: How long the client should wait for the
+                             connection to be established. The default is 5
+                             seconds.
         """
         if self.connected:
             raise RuntimeError('Already connected')
@@ -78,7 +81,8 @@ class SimpleClient:
 
         self.client.connect(url, headers=headers, auth=auth,
                             transports=transports, namespaces=[namespace],
-                            socketio_path=socketio_path)
+                            socketio_path=socketio_path,
+                            wait_timeout=wait_timeout)
 
     @property
     def sid(self):
@@ -87,7 +91,7 @@ class SimpleClient:
         The session ID is not guaranteed to remain constant throughout the life
         of the connection, as reconnections can cause it to change.
         """
-        return self.client.sid if self.client else None
+        return self.client.get_sid(self.namespace) if self.client else None
 
     @property
     def transport(self):

tests/async/test_admin.py

@@ -0,0 +1,311 @@
+from functools import wraps
+import threading
+import time
+from unittest import mock
+import unittest
+import pytest
+try:
+    from engineio.async_socket import AsyncSocket as EngineIOSocket
+except ImportError:
+    from engineio.asyncio_socket import AsyncSocket as EngineIOSocket
+import socketio
+from socketio.exceptions import ConnectionError
+from tests.asyncio_web_server import SocketIOWebServer
+from .helpers import AsyncMock
+
+
+def with_instrumented_server(auth=False, **ikwargs):
+    """This decorator can be applied to test functions or methods so that they
+    run with a Socket.IO server that has been instrumented for the official
+    Admin UI project. The arguments passed to the decorator are passed directly
+    to the ``instrument()`` method of the server.
+    """
+    def decorator(f):
+        @wraps(f)
+        def wrapped(self, *args, **kwargs):
+            sio = socketio.AsyncServer(async_mode='asgi')
+
+            @sio.event
+            async def enter_room(sid, data):
+                await sio.enter_room(sid, data)
+
+            @sio.event
+            async def emit(sid, event):
+                await sio.emit(event, skip_sid=sid)
+
+            @sio.event(namespace='/foo')
+            def connect(sid, environ, auth):
+                pass
+
+            async def shutdown():
+                await instrumented_server.shutdown()
+                await sio.shutdown()
+
+            if 'server_stats_interval' not in ikwargs:
+                ikwargs['server_stats_interval'] = 0.25
+
+            instrumented_server = sio.instrument(auth=auth, **ikwargs)
+            server = SocketIOWebServer(sio, on_shutdown=shutdown)
+            server.start()
+
+            # import logging
+            # logging.getLogger('engineio.client').setLevel(logging.DEBUG)
+            # logging.getLogger('socketio.client').setLevel(logging.DEBUG)
+
+            original_schedule_ping = EngineIOSocket.schedule_ping
+            EngineIOSocket.schedule_ping = mock.MagicMock()
+
+            try:
+                ret = f(self, instrumented_server, *args, **kwargs)
+            finally:
+                server.stop()
+                instrumented_server.uninstrument()
+
+            EngineIOSocket.schedule_ping = original_schedule_ping
+
+            # import logging
+            # logging.getLogger('engineio.client').setLevel(logging.NOTSET)
+            # logging.getLogger('socketio.client').setLevel(logging.NOTSET)
+
+            return ret
+        return wrapped
+    return decorator
+
+
+def _custom_auth(auth):
+    return auth == {'foo': 'bar'}
+
+
+async def _async_custom_auth(auth):
+    return auth == {'foo': 'bar'}
+
+
+class TestAsyncAdmin(unittest.TestCase):
+    def setUp(self):
+        print('threads at start:', threading.enumerate())
+        self.thread_count = threading.active_count()
+
+    def tearDown(self):
+        print('threads at end:', threading.enumerate())
+        assert self.thread_count == threading.active_count()
+
+    def test_missing_auth(self):
+        sio = socketio.AsyncServer(async_mode='asgi')
+        with pytest.raises(ValueError):
+            sio.instrument()
+
+    @with_instrumented_server(auth=False)
+    def test_admin_connect_with_no_auth(self, isvr):
+        with socketio.SimpleClient() as admin_client:
+            admin_client.connect('http://localhost:8900', namespace='/admin')
+        with socketio.SimpleClient() as admin_client:
+            admin_client.connect('http://localhost:8900', namespace='/admin',
+                                 auth={'foo': 'bar'})
+
+    @with_instrumented_server(auth={'foo': 'bar'})
+    def test_admin_connect_with_dict_auth(self, isvr):
+        with socketio.SimpleClient() as admin_client:
+            admin_client.connect('http://localhost:8900', namespace='/admin',
+                                 auth={'foo': 'bar'})
+        with socketio.SimpleClient() as admin_client:
+            with pytest.raises(ConnectionError):
+                admin_client.connect(
+                    'http://localhost:8900', namespace='/admin',
+                    auth={'foo': 'baz'})
+        with socketio.SimpleClient() as admin_client:
+            with pytest.raises(ConnectionError):
+                admin_client.connect(
+                    'http://localhost:8900', namespace='/admin')
+
+    @with_instrumented_server(auth=[{'foo': 'bar'},
+                                    {'u': 'admin', 'p': 'secret'}])
+    def test_admin_connect_with_list_auth(self, isvr):
+        with socketio.SimpleClient() as admin_client:
+            admin_client.connect('http://localhost:8900', namespace='/admin',
+                                 auth={'foo': 'bar'})
+        with socketio.SimpleClient() as admin_client:
+            admin_client.connect('http://localhost:8900', namespace='/admin',
+                                 auth={'u': 'admin', 'p': 'secret'})
+        with socketio.SimpleClient() as admin_client:
+            with pytest.raises(ConnectionError):
+                admin_client.connect('http://localhost:8900',
+                                     namespace='/admin', auth={'foo': 'baz'})
+        with socketio.SimpleClient() as admin_client:
+            with pytest.raises(ConnectionError):
+                admin_client.connect('http://localhost:8900',
+                                     namespace='/admin')
+
+    @with_instrumented_server(auth=_custom_auth)
+    def test_admin_connect_with_function_auth(self, isvr):
+        with socketio.SimpleClient() as admin_client:
+            admin_client.connect('http://localhost:8900', namespace='/admin',
+                                 auth={'foo': 'bar'})
+        with socketio.SimpleClient() as admin_client:
+            with pytest.raises(ConnectionError):
+                admin_client.connect('http://localhost:8900',
+                                     namespace='/admin', auth={'foo': 'baz'})
+        with socketio.SimpleClient() as admin_client:
+            with pytest.raises(ConnectionError):
+                admin_client.connect('http://localhost:8900',
+                                     namespace='/admin')
+
+    @with_instrumented_server(auth=_async_custom_auth)
+    def test_admin_connect_with_async_function_auth(self, isvr):
+        with socketio.SimpleClient() as admin_client:
+            admin_client.connect('http://localhost:8900', namespace='/admin',
+                                 auth={'foo': 'bar'})
+        with socketio.SimpleClient() as admin_client:
+            with pytest.raises(ConnectionError):
+                admin_client.connect('http://localhost:8900',
+                                     namespace='/admin', auth={'foo': 'baz'})
+        with socketio.SimpleClient() as admin_client:
+            with pytest.raises(ConnectionError):
+                admin_client.connect('http://localhost:8900',
+                                     namespace='/admin')
+
+    @with_instrumented_server()
+    def test_admin_connect_only_admin(self, isvr):
+        with socketio.SimpleClient() as admin_client:
+            admin_client.connect('http://localhost:8900', namespace='/admin')
+            sid = admin_client.sid
+            expected = ['config', 'all_sockets', 'server_stats']
+            events = {}
+            while expected:
+                data = admin_client.receive(timeout=5)
+                if data[0] in expected:
+                    events[data[0]] = data[1]
+                    expected.remove(data[0])
+
+        assert 'supportedFeatures' in events['config']
+        assert 'ALL_EVENTS' in events['config']['supportedFeatures']
+        assert 'AGGREGATED_EVENTS' in events['config']['supportedFeatures']
+        assert 'EMIT' in events['config']['supportedFeatures']
+        assert len(events['all_sockets']) == 1
+        assert events['all_sockets'][0]['id'] == sid
+        assert events['all_sockets'][0]['rooms'] == [sid]
+        assert events['server_stats']['clientsCount'] == 1
+        assert events['server_stats']['pollingClientsCount'] == 0
+        assert len(events['server_stats']['namespaces']) == 3
+        assert {'name': '/', 'socketsCount': 0} in \
+            events['server_stats']['namespaces']
+        assert {'name': '/foo', 'socketsCount': 0} in \
+            events['server_stats']['namespaces']
+        assert {'name': '/admin', 'socketsCount': 1} in \
+            events['server_stats']['namespaces']
+
+    @with_instrumented_server()
+    def test_admin_connect_with_others(self, isvr):
+        with socketio.SimpleClient() as client1, \
+                socketio.SimpleClient() as client2, \
+                socketio.SimpleClient() as client3, \
+                socketio.SimpleClient() as admin_client:
+            client1.connect('http://localhost:8900')
+            client1.emit('enter_room', 'room')
+            sid1 = client1.sid
+
+            saved_check_for_upgrade = isvr._check_for_upgrade
+            isvr._check_for_upgrade = AsyncMock()
+            client2.connect('http://localhost:8900', namespace='/foo',
+                            transports=['polling'])
+            sid2 = client2.sid
+            isvr._check_for_upgrade = saved_check_for_upgrade
+
+            client3.connect('http://localhost:8900', namespace='/admin')
+            sid3 = client3.sid
+
+            admin_client.connect('http://localhost:8900', namespace='/admin')
+            sid = admin_client.sid
+            expected = ['config', 'all_sockets', 'server_stats']
+            events = {}
+            while expected:
+                data = admin_client.receive(timeout=5)
+                if data[0] in expected:
+                    events[data[0]] = data[1]
+                    expected.remove(data[0])
+
+        assert 'supportedFeatures' in events['config']
+        assert 'ALL_EVENTS' in events['config']['supportedFeatures']
+        assert 'AGGREGATED_EVENTS' in events['config']['supportedFeatures']
+        assert 'EMIT' in events['config']['supportedFeatures']
+        assert len(events['all_sockets']) == 4
+        assert events['server_stats']['clientsCount'] == 4
+        assert events['server_stats']['pollingClientsCount'] == 1
+        assert len(events['server_stats']['namespaces']) == 3
+        assert {'name': '/', 'socketsCount': 1} in \
+            events['server_stats']['namespaces']
+        assert {'name': '/foo', 'socketsCount': 1} in \
+            events['server_stats']['namespaces']
+        assert {'name': '/admin', 'socketsCount': 2} in \
+            events['server_stats']['namespaces']
+
+        for socket in events['all_sockets']:
+            if socket['id'] == sid:
+                assert socket['rooms'] == [sid]
+            elif socket['id'] == sid1:
+                assert socket['rooms'] == [sid1, 'room']
+            elif socket['id'] == sid2:
+                assert socket['rooms'] == [sid2]
+            elif socket['id'] == sid3:
+                assert socket['rooms'] == [sid3]
+
+    @with_instrumented_server(mode='production', read_only=True)
+    def test_admin_connect_production(self, isvr):
+        with socketio.SimpleClient() as admin_client:
+            admin_client.connect('http://localhost:8900', namespace='/admin')
+            expected = ['config', 'server_stats']
+            events = {}
+            while expected:
+                data = admin_client.receive(timeout=5)
+                if data[0] in expected:
+                    events[data[0]] = data[1]
+                    expected.remove(data[0])
+
+        assert 'supportedFeatures' in events['config']
+        assert 'ALL_EVENTS' not in events['config']['supportedFeatures']
+        assert 'AGGREGATED_EVENTS' in events['config']['supportedFeatures']
+        assert 'EMIT' not in events['config']['supportedFeatures']
+        assert events['server_stats']['clientsCount'] == 1
+        assert events['server_stats']['pollingClientsCount'] == 0
+        assert len(events['server_stats']['namespaces']) == 3
+        assert {'name': '/', 'socketsCount': 0} in \
+            events['server_stats']['namespaces']
+        assert {'name': '/foo', 'socketsCount': 0} in \
+            events['server_stats']['namespaces']
+        assert {'name': '/admin', 'socketsCount': 1} in \
+            events['server_stats']['namespaces']
+
+    @with_instrumented_server()
+    def test_admin_features(self, isvr):
+        with socketio.SimpleClient() as client1, \
+                socketio.SimpleClient() as client2, \
+                socketio.SimpleClient() as admin_client:
+            client1.connect('http://localhost:8900')
+            client2.connect('http://localhost:8900')
+            admin_client.connect('http://localhost:8900', namespace='/admin')
+
+            # emit from admin
+            admin_client.emit(
+                'emit', ('/', client1.sid, 'foo', {'bar': 'baz'}, 'extra'))
+            data = client1.receive(timeout=5)
+            assert data == ['foo', {'bar': 'baz'}, 'extra']
+
+            # emit from regular client
+            client1.emit('emit', 'foo')
+            data = client2.receive(timeout=5)
+            assert data == ['foo']
+
+            # join and leave
+            admin_client.emit('join', ('/', 'room', client1.sid))
+            admin_client.emit(
+                'emit', ('/', 'room', 'foo', {'bar': 'baz'}))
+            data = client1.receive(timeout=5)
+            assert data == ['foo', {'bar': 'baz'}]
+            admin_client.emit('leave', ('/', 'room'))
+
+            # disconnect
+            admin_client.emit('_disconnect', ('/', False, client1.sid))
+            for _ in range(10):
+                if not client1.connected:
+                    break
+                time.sleep(0.2)
+            assert not client1.connected

tests/async/test_manager.py

@@ -27,7 +27,7 @@ class TestAsyncManager(unittest.TestCase):
         self.bm.initialize()
 
     def test_connect(self):
-        sid = self.bm.connect('123', '/foo')
+        sid = _run(self.bm.connect('123', '/foo'))
         assert None in self.bm.rooms['/foo']
         assert sid in self.bm.rooms['/foo']
         assert sid in self.bm.rooms['/foo'][None]
@@ -37,8 +37,8 @@ class TestAsyncManager(unittest.TestCase):
         assert self.bm.sid_from_eio_sid('123', '/foo') == sid
 
     def test_pre_disconnect(self):
-        sid1 = self.bm.connect('123', '/foo')
-        sid2 = self.bm.connect('456', '/foo')
+        sid1 = _run(self.bm.connect('123', '/foo'))
+        sid2 = _run(self.bm.connect('456', '/foo'))
         assert self.bm.is_connected(sid1, '/foo')
         assert self.bm.pre_disconnect(sid1, '/foo') == '123'
         assert self.bm.pending_disconnect == {'/foo': [sid1]}
@@ -52,8 +52,8 @@ class TestAsyncManager(unittest.TestCase):
         assert self.bm.pending_disconnect == {}
 
     def test_disconnect(self):
-        sid1 = self.bm.connect('123', '/foo')
-        sid2 = self.bm.connect('456', '/foo')
+        sid1 = _run(self.bm.connect('123', '/foo'))
+        sid2 = _run(self.bm.connect('456', '/foo'))
         _run(self.bm.enter_room(sid1, '/foo', 'bar'))
         _run(self.bm.enter_room(sid2, '/foo', 'baz'))
         _run(self.bm.disconnect(sid1, '/foo'))
@@ -62,10 +62,10 @@ class TestAsyncManager(unittest.TestCase):
         assert dict(self.bm.rooms['/foo']['baz']) == {sid2: '456'}
 
     def test_disconnect_default_namespace(self):
-        sid1 = self.bm.connect('123', '/')
-        sid2 = self.bm.connect('123', '/foo')
-        sid3 = self.bm.connect('456', '/')
-        sid4 = self.bm.connect('456', '/foo')
+        sid1 = _run(self.bm.connect('123', '/'))
+        sid2 = _run(self.bm.connect('123', '/foo'))
+        sid3 = _run(self.bm.connect('456', '/'))
+        sid4 = _run(self.bm.connect('456', '/foo'))
         assert self.bm.is_connected(sid1, '/')
         assert self.bm.is_connected(sid2, '/foo')
         assert not self.bm.is_connected(sid2, '/')
@@ -81,10 +81,10 @@ class TestAsyncManager(unittest.TestCase):
         assert dict(self.bm.rooms['/foo'][sid4]) == {sid4: '456'}
 
     def test_disconnect_twice(self):
-        sid1 = self.bm.connect('123', '/')
-        sid2 = self.bm.connect('123', '/foo')
-        sid3 = self.bm.connect('456', '/')
-        sid4 = self.bm.connect('456', '/foo')
+        sid1 = _run(self.bm.connect('123', '/'))
+        sid2 = _run(self.bm.connect('123', '/foo'))
+        sid3 = _run(self.bm.connect('456', '/'))
+        sid4 = _run(self.bm.connect('456', '/foo'))
         _run(self.bm.disconnect(sid1, '/'))
         _run(self.bm.disconnect(sid2, '/foo'))
         _run(self.bm.disconnect(sid1, '/'))
@@ -95,8 +95,8 @@ class TestAsyncManager(unittest.TestCase):
         assert dict(self.bm.rooms['/foo'][sid4]) == {sid4: '456'}
 
     def test_disconnect_all(self):
-        sid1 = self.bm.connect('123', '/foo')
-        sid2 = self.bm.connect('456', '/foo')
+        sid1 = _run(self.bm.connect('123', '/foo'))
+        sid2 = _run(self.bm.connect('456', '/foo'))
         _run(self.bm.enter_room(sid1, '/foo', 'bar'))
         _run(self.bm.enter_room(sid2, '/foo', 'baz'))
         _run(self.bm.disconnect(sid1, '/foo'))
@@ -104,9 +104,9 @@ class TestAsyncManager(unittest.TestCase):
         assert self.bm.rooms == {}
 
     def test_disconnect_with_callbacks(self):
-        sid1 = self.bm.connect('123', '/')
-        sid2 = self.bm.connect('123', '/foo')
-        sid3 = self.bm.connect('456', '/foo')
+        sid1 = _run(self.bm.connect('123', '/'))
+        sid2 = _run(self.bm.connect('123', '/foo'))
+        sid3 = _run(self.bm.connect('456', '/foo'))
         self.bm._generate_ack_id(sid1, 'f')
         self.bm._generate_ack_id(sid2, 'g')
         self.bm._generate_ack_id(sid3, 'h')
@@ -117,8 +117,8 @@ class TestAsyncManager(unittest.TestCase):
         assert sid3 in self.bm.callbacks
 
     def test_trigger_sync_callback(self):
-        sid1 = self.bm.connect('123', '/')
-        sid2 = self.bm.connect('123', '/foo')
+        sid1 = _run(self.bm.connect('123', '/'))
+        sid2 = _run(self.bm.connect('123', '/foo'))
         cb = mock.MagicMock()
         id1 = self.bm._generate_ack_id(sid1, cb)
         id2 = self.bm._generate_ack_id(sid2, cb)
@@ -129,8 +129,8 @@ class TestAsyncManager(unittest.TestCase):
         cb.assert_any_call('bar', 'baz')
 
     def test_trigger_async_callback(self):
-        sid1 = self.bm.connect('123', '/')
-        sid2 = self.bm.connect('123', '/foo')
+        sid1 = _run(self.bm.connect('123', '/'))
+        sid2 = _run(self.bm.connect('123', '/foo'))
         cb = AsyncMock()
         id1 = self.bm._generate_ack_id(sid1, cb)
         id2 = self.bm._generate_ack_id(sid2, cb)
@@ -141,7 +141,7 @@ class TestAsyncManager(unittest.TestCase):
         cb.mock.assert_any_call('bar', 'baz')
 
     def test_invalid_callback(self):
-        sid = self.bm.connect('123', '/')
+        sid = _run(self.bm.connect('123', '/'))
         cb = mock.MagicMock()
         id = self.bm._generate_ack_id(sid, cb)
 
@@ -152,17 +152,17 @@ class TestAsyncManager(unittest.TestCase):
 
     def test_get_namespaces(self):
         assert list(self.bm.get_namespaces()) == []
-        self.bm.connect('123', '/')
-        self.bm.connect('123', '/foo')
+        _run(self.bm.connect('123', '/'))
+        _run(self.bm.connect('123', '/foo'))
         namespaces = list(self.bm.get_namespaces())
         assert len(namespaces) == 2
         assert '/' in namespaces
         assert '/foo' in namespaces
 
     def test_get_participants(self):
-        sid1 = self.bm.connect('123', '/')
-        sid2 = self.bm.connect('456', '/')
-        sid3 = self.bm.connect('789', '/')
+        sid1 = _run(self.bm.connect('123', '/'))
+        sid2 = _run(self.bm.connect('456', '/'))
+        sid3 = _run(self.bm.connect('789', '/'))
         _run(self.bm.disconnect(sid3, '/'))
         assert sid3 not in self.bm.rooms['/'][None]
         participants = list(self.bm.get_participants('/', None))
@@ -172,7 +172,7 @@ class TestAsyncManager(unittest.TestCase):
         assert (sid3, '789') not in participants
 
     def test_leave_invalid_room(self):
-        sid = self.bm.connect('123', '/foo')
+        sid = _run(self.bm.connect('123', '/foo'))
         _run(self.bm.leave_room(sid, '/foo', 'baz'))
         _run(self.bm.leave_room(sid, '/bar', 'baz'))
 
@@ -181,9 +181,9 @@ class TestAsyncManager(unittest.TestCase):
         assert [] == rooms
 
     def test_close_room(self):
-        sid = self.bm.connect('123', '/foo')
-        self.bm.connect('456', '/foo')
-        self.bm.connect('789', '/foo')
+        sid = _run(self.bm.connect('123', '/foo'))
+        _run(self.bm.connect('456', '/foo'))
+        _run(self.bm.connect('789', '/foo'))
         _run(self.bm.enter_room(sid, '/foo', 'bar'))
         _run(self.bm.enter_room(sid, '/foo', 'bar'))
         _run(self.bm.close_room('bar', '/foo'))
@@ -195,7 +195,7 @@ class TestAsyncManager(unittest.TestCase):
         self.bm.close_room('bar', '/foo')
 
     def test_rooms(self):
-        sid = self.bm.connect('123', '/foo')
+        sid = _run(self.bm.connect('123', '/foo'))
         _run(self.bm.enter_room(sid, '/foo', 'bar'))
         r = self.bm.get_rooms(sid, '/foo')
         assert len(r) == 2
@@ -203,8 +203,8 @@ class TestAsyncManager(unittest.TestCase):
         assert 'bar' in r
 
     def test_emit_to_sid(self):
-        sid = self.bm.connect('123', '/foo')
-        self.bm.connect('456', '/foo')
+        sid = _run(self.bm.connect('123', '/foo'))
+        _run(self.bm.connect('456', '/foo'))
         _run(
             self.bm.emit(
                 'my event', {'foo': 'bar'}, namespace='/foo', room=sid
@@ -217,11 +217,11 @@ class TestAsyncManager(unittest.TestCase):
         assert pkt.encode() == '42/foo,["my event",{"foo":"bar"}]'
 
     def test_emit_to_room(self):
-        sid1 = self.bm.connect('123', '/foo')
+        sid1 = _run(self.bm.connect('123', '/foo'))
         _run(self.bm.enter_room(sid1, '/foo', 'bar'))
-        sid2 = self.bm.connect('456', '/foo')
+        sid2 = _run(self.bm.connect('456', '/foo'))
         _run(self.bm.enter_room(sid2, '/foo', 'bar'))
-        self.bm.connect('789', '/foo')
+        _run(self.bm.connect('789', '/foo'))
         _run(
             self.bm.emit(
                 'my event', {'foo': 'bar'}, namespace='/foo', room='bar'
@@ -238,12 +238,12 @@ class TestAsyncManager(unittest.TestCase):
         assert pkt.encode() == '42/foo,["my event",{"foo":"bar"}]'
 
     def test_emit_to_rooms(self):
-        sid1 = self.bm.connect('123', '/foo')
+        sid1 = _run(self.bm.connect('123', '/foo'))
         _run(self.bm.enter_room(sid1, '/foo', 'bar'))
-        sid2 = self.bm.connect('456', '/foo')
+        sid2 = _run(self.bm.connect('456', '/foo'))
         _run(self.bm.enter_room(sid2, '/foo', 'bar'))
         _run(self.bm.enter_room(sid2, '/foo', 'baz'))
-        sid3 = self.bm.connect('789', '/foo')
+        sid3 = _run(self.bm.connect('789', '/foo'))
         _run(self.bm.enter_room(sid3, '/foo', 'baz'))
         _run(
             self.bm.emit('my event', {'foo': 'bar'}, namespace='/foo',
@@ -264,12 +264,12 @@ class TestAsyncManager(unittest.TestCase):
         assert pkt.encode() == '42/foo,["my event",{"foo":"bar"}]'
 
     def test_emit_to_all(self):
-        sid1 = self.bm.connect('123', '/foo')
+        sid1 = _run(self.bm.connect('123', '/foo'))
         _run(self.bm.enter_room(sid1, '/foo', 'bar'))
-        sid2 = self.bm.connect('456', '/foo')
+        sid2 = _run(self.bm.connect('456', '/foo'))
         _run(self.bm.enter_room(sid2, '/foo', 'bar'))
-        self.bm.connect('789', '/foo')
-        self.bm.connect('abc', '/bar')
+        _run(self.bm.connect('789', '/foo'))
+        _run(self.bm.connect('abc', '/bar'))
         _run(self.bm.emit('my event', {'foo': 'bar'}, namespace='/foo'))
         assert self.bm.server._send_eio_packet.mock.call_count == 3
         assert self.bm.server._send_eio_packet.mock.call_args_list[0][0][0] \
@@ -286,12 +286,12 @@ class TestAsyncManager(unittest.TestCase):
         assert pkt.encode() == '42/foo,["my event",{"foo":"bar"}]'
 
     def test_emit_to_all_skip_one(self):
-        sid1 = self.bm.connect('123', '/foo')
+        sid1 = _run(self.bm.connect('123', '/foo'))
         _run(self.bm.enter_room(sid1, '/foo', 'bar'))
-        sid2 = self.bm.connect('456', '/foo')
+        sid2 = _run(self.bm.connect('456', '/foo'))
         _run(self.bm.enter_room(sid2, '/foo', 'bar'))
-        self.bm.connect('789', '/foo')
-        self.bm.connect('abc', '/bar')
+        _run(self.bm.connect('789', '/foo'))
+        _run(self.bm.connect('abc', '/bar'))
         _run(
             self.bm.emit(
                 'my event', {'foo': 'bar'}, namespace='/foo', skip_sid=sid2
@@ -308,12 +308,12 @@ class TestAsyncManager(unittest.TestCase):
         assert pkt.encode() == '42/foo,["my event",{"foo":"bar"}]'
 
     def test_emit_to_all_skip_two(self):
-        sid1 = self.bm.connect('123', '/foo')
+        sid1 = _run(self.bm.connect('123', '/foo'))
         _run(self.bm.enter_room(sid1, '/foo', 'bar'))
-        sid2 = self.bm.connect('456', '/foo')
+        sid2 = _run(self.bm.connect('456', '/foo'))
         _run(self.bm.enter_room(sid2, '/foo', 'bar'))
-        sid3 = self.bm.connect('789', '/foo')
-        self.bm.connect('abc', '/bar')
+        sid3 = _run(self.bm.connect('789', '/foo'))
+        _run(self.bm.connect('abc', '/bar'))
         _run(
             self.bm.emit(
                 'my event',
@@ -329,7 +329,7 @@ class TestAsyncManager(unittest.TestCase):
         assert pkt.encode() == '42/foo,["my event",{"foo":"bar"}]'
 
     def test_emit_with_callback(self):
-        sid = self.bm.connect('123', '/foo')
+        sid = _run(self.bm.connect('123', '/foo'))
         self.bm._generate_ack_id = mock.MagicMock()
         self.bm._generate_ack_id.return_value = 11
         _run(
@@ -353,7 +353,7 @@ class TestAsyncManager(unittest.TestCase):
         _run(self.bm.emit('my event', {'foo': 'bar'}, namespace='/foo'))
 
     def test_emit_with_tuple(self):
-        sid = self.bm.connect('123', '/foo')
+        sid = _run(self.bm.connect('123', '/foo'))
         _run(
             self.bm.emit(
                 'my event', ('foo', 'bar'), namespace='/foo', room=sid
@@ -366,7 +366,7 @@ class TestAsyncManager(unittest.TestCase):
         assert pkt.encode() == '42/foo,["my event","foo","bar"]'
 
     def test_emit_with_list(self):
-        sid = self.bm.connect('123', '/foo')
+        sid = _run(self.bm.connect('123', '/foo'))
         _run(
             self.bm.emit(
                 'my event', ['foo', 'bar'], namespace='/foo', room=sid
@@ -379,7 +379,7 @@ class TestAsyncManager(unittest.TestCase):
         assert pkt.encode() == '42/foo,["my event",["foo","bar"]]'
 
     def test_emit_with_none(self):
-        sid = self.bm.connect('123', '/foo')
+        sid = _run(self.bm.connect('123', '/foo'))
         _run(
             self.bm.emit(
                 'my event', None, namespace='/foo', room=sid
@@ -392,7 +392,7 @@ class TestAsyncManager(unittest.TestCase):
         assert pkt.encode() == '42/foo,["my event"]'
 
     def test_emit_binary(self):
-        sid = self.bm.connect('123', '/')
+        sid = _run(self.bm.connect('123', '/'))
         _run(
             self.bm.emit(
                 u'my event', b'my binary data', namespace='/', room=sid

tests/async/test_pubsub_manager.py

@@ -145,7 +145,7 @@ class TestAsyncPubSubManager(unittest.TestCase):
                 _run(self.pm.emit('foo', 'bar', callback='cb'))
 
     def test_emit_with_ignore_queue(self):
-        sid = self.pm.connect('123', '/')
+        sid = _run(self.pm.connect('123', '/'))
         _run(
             self.pm.emit(
                 'foo', 'bar', room=sid, namespace='/', ignore_queue=True
@@ -159,7 +159,7 @@ class TestAsyncPubSubManager(unittest.TestCase):
         assert pkt.encode() == '42["foo","bar"]'
 
     def test_can_disconnect(self):
-        sid = self.pm.connect('123', '/')
+        sid = _run(self.pm.connect('123', '/'))
         assert _run(self.pm.can_disconnect(sid, '/')) is True
         _run(self.pm.can_disconnect(sid, '/foo'))
         self.pm._publish.mock.assert_called_once_with(
@@ -175,14 +175,14 @@ class TestAsyncPubSubManager(unittest.TestCase):
         )
 
     def test_disconnect_ignore_queue(self):
-        sid = self.pm.connect('123', '/')
+        sid = _run(self.pm.connect('123', '/'))
         self.pm.pre_disconnect(sid, '/')
         _run(self.pm.disconnect(sid, '/', ignore_queue=True))
         self.pm._publish.mock.assert_not_called()
         assert self.pm.is_connected(sid, '/') is False
 
     def test_enter_room(self):
-        sid = self.pm.connect('123', '/')
+        sid = _run(self.pm.connect('123', '/'))
         _run(self.pm.enter_room(sid, '/', 'foo'))
         _run(self.pm.enter_room('456', '/', 'foo'))
         assert sid in self.pm.rooms['/']['foo']
@@ -193,7 +193,7 @@ class TestAsyncPubSubManager(unittest.TestCase):
         )
 
     def test_leave_room(self):
-        sid = self.pm.connect('123', '/')
+        sid = _run(self.pm.connect('123', '/'))
         _run(self.pm.leave_room(sid, '/', 'foo'))
         _run(self.pm.leave_room('456', '/', 'foo'))
         assert 'foo' not in self.pm.rooms['/']
@@ -435,7 +435,7 @@ class TestAsyncPubSubManager(unittest.TestCase):
         )
 
     def test_handle_enter_room(self):
-        sid = self.pm.connect('123', '/')
+        sid = _run(self.pm.connect('123', '/'))
         with mock.patch.object(
             async_manager.AsyncManager, 'enter_room', new=AsyncMock()
         ) as super_enter_room:
@@ -456,7 +456,7 @@ class TestAsyncPubSubManager(unittest.TestCase):
             )
 
     def test_handle_leave_room(self):
-        sid = self.pm.connect('123', '/')
+        sid = _run(self.pm.connect('123', '/'))
         with mock.patch.object(
             async_manager.AsyncManager, 'leave_room', new=AsyncMock()
         ) as super_leave_room:

tests/async/test_server.py

@@ -596,7 +596,7 @@ class TestAsyncServer(unittest.TestCase):
     def test_handle_event(self, eio):
         eio.return_value.send = AsyncMock()
         s = async_server.AsyncServer(async_handlers=False)
-        sid = s.manager.connect('123', '/')
+        sid = _run(s.manager.connect('123', '/'))
         handler = AsyncMock()
         catchall_handler = AsyncMock()
         s.on('msg', handler)
@@ -610,7 +610,7 @@ class TestAsyncServer(unittest.TestCase):
     def test_handle_event_with_namespace(self, eio):
         eio.return_value.send = AsyncMock()
         s = async_server.AsyncServer(async_handlers=False)
-        sid = s.manager.connect('123', '/foo')
+        sid = _run(s.manager.connect('123', '/foo'))
         handler = mock.MagicMock()
         catchall_handler = mock.MagicMock()
         s.on('msg', handler, namespace='/foo')
@@ -624,7 +624,7 @@ class TestAsyncServer(unittest.TestCase):
     def test_handle_event_with_disconnected_namespace(self, eio):
         eio.return_value.send = AsyncMock()
         s = async_server.AsyncServer(async_handlers=False)
-        s.manager.connect('123', '/foo')
+        _run(s.manager.connect('123', '/foo'))
         handler = mock.MagicMock()
         s.on('my message', handler, namespace='/bar')
         _run(s._handle_eio_message('123', '2/bar,["my message","a","b","c"]'))
@@ -633,7 +633,7 @@ class TestAsyncServer(unittest.TestCase):
     def test_handle_event_binary(self, eio):
         eio.return_value.send = AsyncMock()
         s = async_server.AsyncServer(async_handlers=False)
-        sid = s.manager.connect('123', '/')
+        sid = _run(s.manager.connect('123', '/'))
         handler = mock.MagicMock()
         s.on('my message', handler)
         _run(
@@ -652,7 +652,7 @@ class TestAsyncServer(unittest.TestCase):
         eio.return_value.send = AsyncMock()
         s = async_server.AsyncServer(async_handlers=False)
         s.manager.trigger_callback = AsyncMock()
-        sid = s.manager.connect('123', '/')
+        sid = _run(s.manager.connect('123', '/'))
         _run(
             s._handle_eio_message(
                 '123',
@@ -667,7 +667,7 @@ class TestAsyncServer(unittest.TestCase):
     def test_handle_event_with_ack(self, eio):
         eio.return_value.send = AsyncMock()
         s = async_server.AsyncServer(async_handlers=False)
-        sid = s.manager.connect('123', '/')
+        sid = _run(s.manager.connect('123', '/'))
         handler = mock.MagicMock(return_value='foo')
         s.on('my message', handler)
         _run(s._handle_eio_message('123', '21000["my message","foo"]'))
@@ -679,7 +679,7 @@ class TestAsyncServer(unittest.TestCase):
     def test_handle_unknown_event_with_ack(self, eio):
         eio.return_value.send = AsyncMock()
         s = async_server.AsyncServer(async_handlers=False)
-        s.manager.connect('123', '/')
+        _run(s.manager.connect('123', '/'))
         handler = mock.MagicMock(return_value='foo')
         s.on('my message', handler)
         _run(s._handle_eio_message('123', '21000["another message","foo"]'))
@@ -688,7 +688,7 @@ class TestAsyncServer(unittest.TestCase):
     def test_handle_event_with_ack_none(self, eio):
         eio.return_value.send = AsyncMock()
         s = async_server.AsyncServer(async_handlers=False)
-        sid = s.manager.connect('123', '/')
+        sid = _run(s.manager.connect('123', '/'))
         handler = mock.MagicMock(return_value=None)
         s.on('my message', handler)
         _run(s._handle_eio_message('123', '21000["my message","foo"]'))
@@ -698,7 +698,7 @@ class TestAsyncServer(unittest.TestCase):
     def test_handle_event_with_ack_tuple(self, eio):
         eio.return_value.send = AsyncMock()
         s = async_server.AsyncServer(async_handlers=False)
-        sid = s.manager.connect('123', '/')
+        sid = _run(s.manager.connect('123', '/'))
         handler = mock.MagicMock(return_value=(1, '2', True))
         s.on('my message', handler)
         _run(s._handle_eio_message('123', '21000["my message","a","b","c"]'))
@@ -710,7 +710,7 @@ class TestAsyncServer(unittest.TestCase):
     def test_handle_event_with_ack_list(self, eio):
         eio.return_value.send = AsyncMock()
         s = async_server.AsyncServer(async_handlers=False)
-        sid = s.manager.connect('123', '/')
+        sid = _run(s.manager.connect('123', '/'))
         handler = mock.MagicMock(return_value=[1, '2', True])
         s.on('my message', handler)
         _run(s._handle_eio_message('123', '21000["my message","a","b","c"]'))
@@ -722,7 +722,7 @@ class TestAsyncServer(unittest.TestCase):
     def test_handle_event_with_ack_binary(self, eio):
         eio.return_value.send = AsyncMock()
         s = async_server.AsyncServer(async_handlers=False)
-        sid = s.manager.connect('123', '/')
+        sid = _run(s.manager.connect('123', '/'))
         handler = mock.MagicMock(return_value=b'foo')
         s.on('my message', handler)
         _run(s._handle_eio_message('123', '21000["my message","foo"]'))
@@ -973,7 +973,7 @@ class TestAsyncServer(unittest.TestCase):
 
     def test_async_handlers(self, eio):
         s = async_server.AsyncServer(async_handlers=True)
-        s.manager.connect('123', '/')
+        _run(s.manager.connect('123', '/'))
         _run(s._handle_eio_message('123', '2["my message","a","b","c"]'))
         s.eio.start_background_task.assert_called_once_with(
             s._handle_event_internal,

tests/async/test_simple_client.py

@@ -24,12 +24,13 @@ class TestAsyncAsyncSimpleClient(unittest.TestCase):
             mock_client.return_value.connect = AsyncMock()
 
             _run(client.connect('url', headers='h', auth='a', transports='t',
-                                namespace='n', socketio_path='s'))
+                                namespace='n', socketio_path='s',
+                                wait_timeout='w'))
             mock_client.assert_called_once_with(123, a='b')
             assert client.client == mock_client()
             mock_client().connect.mock.assert_called_once_with(
                 'url', headers='h', auth='a', transports='t',
-                namespaces=['n'], socketio_path='s')
+                namespaces=['n'], socketio_path='s', wait_timeout='w')
             mock_client().event.call_count == 3
             mock_client().on.called_once_with('*')
             assert client.namespace == 'n'
@@ -44,12 +45,12 @@ class TestAsyncAsyncSimpleClient(unittest.TestCase):
 
                     await client.connect('url', headers='h', auth='a',
                                          transports='t', namespace='n',
-                                         socketio_path='s')
+                                         socketio_path='s', wait_timeout='w')
                     mock_client.assert_called_once_with(123, a='b')
                     assert client.client == mock_client()
                     mock_client().connect.mock.assert_called_once_with(
                         'url', headers='h', auth='a', transports='t',
-                        namespaces=['n'], socketio_path='s')
+                        namespaces=['n'], socketio_path='s', wait_timeout='w')
                     mock_client().event.call_count == 3
                     mock_client().on.called_once_with('*')
                     assert client.namespace == 'n'
@@ -67,7 +68,8 @@ class TestAsyncAsyncSimpleClient(unittest.TestCase):
 
     def test_properties(self):
         client = AsyncSimpleClient()
-        client.client = mock.MagicMock(sid='sid', transport='websocket')
+        client.client = mock.MagicMock(transport='websocket')
+        client.client.get_sid.return_value = 'sid'
         client.connected_event.set()
         client.connected = True
 

tests/asyncio_web_server.py

@@ -0,0 +1,57 @@
+import requests
+import threading
+import time
+import uvicorn
+import socketio
+
+
+class SocketIOWebServer:
+    """A simple web server used for running Socket.IO servers in tests.
+
+    :param sio: a Socket.IO server instance.
+
+    Note 1: This class is not production-ready and is intended for testing.
+    Note 2: This class only supports the "asgi" async_mode.
+    """
+    def __init__(self, sio, on_shutdown=None):
+        if sio.async_mode != 'asgi':
+            raise ValueError('The async_mode must be "asgi"')
+
+        async def http_app(scope, receive, send):
+            await send({'type': 'http.response.start',
+                        'status': 200,
+                        'headers': [('Content-Type', 'text/plain')]})
+            await send({'type': 'http.response.body',
+                        'body': b'OK'})
+
+        self.sio = sio
+        self.app = socketio.ASGIApp(sio, http_app, on_shutdown=on_shutdown)
+        self.httpd = None
+        self.thread = None
+
+    def start(self, port=8900):
+        """Start the web server.
+
+        :param port: the port to listen on. Defaults to 8900.
+
+        The server is started in a background thread.
+        """
+        self.httpd = uvicorn.Server(config=uvicorn.Config(self.app, port=port))
+        self.thread = threading.Thread(target=self.httpd.run)
+        self.thread.start()
+
+        # wait for the server to start
+        while True:
+            try:
+                r = requests.get(f'http://localhost:{port}/')
+                r.raise_for_status()
+                if r.text == 'OK':
+                    break
+            except:
+                time.sleep(0.1)
+
+    def stop(self):
+        """Stop the web server."""
+        self.httpd.should_exit = True
+        self.thread.join()
+        self.thread = None

tests/common/test_admin.py

@@ -0,0 +1,286 @@
+from functools import wraps
+import threading
+import time
+from unittest import mock
+import unittest
+import pytest
+from engineio.socket import Socket as EngineIOSocket
+import socketio
+from socketio.exceptions import ConnectionError
+from tests.web_server import SocketIOWebServer
+
+
+def with_instrumented_server(auth=False, **ikwargs):
+    """This decorator can be applied to test functions or methods so that they
+    run with a Socket.IO server that has been instrumented for the official
+    Admin UI project. The arguments passed to the decorator are passed directly
+    to the ``instrument()`` method of the server.
+    """
+    def decorator(f):
+        @wraps(f)
+        def wrapped(self, *args, **kwargs):
+            sio = socketio.Server(async_mode='threading')
+
+            @sio.event
+            def enter_room(sid, data):
+                sio.enter_room(sid, data)
+
+            @sio.event
+            def emit(sid, event):
+                sio.emit(event, skip_sid=sid)
+
+            @sio.event(namespace='/foo')
+            def connect(sid, environ, auth):
+                pass
+
+            if 'server_stats_interval' not in ikwargs:
+                ikwargs['server_stats_interval'] = 0.25
+
+            instrumented_server = sio.instrument(auth=auth, **ikwargs)
+            server = SocketIOWebServer(sio)
+            server.start()
+
+            # import logging
+            # logging.getLogger('engineio.client').setLevel(logging.DEBUG)
+            # logging.getLogger('socketio.client').setLevel(logging.DEBUG)
+
+            original_schedule_ping = EngineIOSocket.schedule_ping
+            EngineIOSocket.schedule_ping = mock.MagicMock()
+
+            try:
+                ret = f(self, instrumented_server, *args, **kwargs)
+            finally:
+                server.stop()
+                instrumented_server.shutdown()
+                instrumented_server.uninstrument()
+
+            EngineIOSocket.schedule_ping = original_schedule_ping
+
+            # import logging
+            # logging.getLogger('engineio.client').setLevel(logging.NOTSET)
+            # logging.getLogger('socketio.client').setLevel(logging.NOTSET)
+
+            return ret
+        return wrapped
+    return decorator
+
+
+def _custom_auth(auth):
+    return auth == {'foo': 'bar'}
+
+
+class TestAdmin(unittest.TestCase):
+    def setUp(self):
+        print('threads at start:', threading.enumerate())
+        self.thread_count = threading.active_count()
+
+    def tearDown(self):
+        print('threads at end:', threading.enumerate())
+        assert self.thread_count == threading.active_count()
+
+    def test_missing_auth(self):
+        sio = socketio.Server(async_mode='threading')
+        with pytest.raises(ValueError):
+            sio.instrument()
+
+    @with_instrumented_server(auth=False)
+    def test_admin_connect_with_no_auth(self, isvr):
+        with socketio.SimpleClient() as admin_client:
+            admin_client.connect('http://localhost:8900', namespace='/admin')
+        with socketio.SimpleClient() as admin_client:
+            admin_client.connect('http://localhost:8900', namespace='/admin',
+                                 auth={'foo': 'bar'})
+
+    @with_instrumented_server(auth={'foo': 'bar'})
+    def test_admin_connect_with_dict_auth(self, isvr):
+        with socketio.SimpleClient() as admin_client:
+            admin_client.connect('http://localhost:8900', namespace='/admin',
+                                 auth={'foo': 'bar'})
+        with socketio.SimpleClient() as admin_client:
+            with pytest.raises(ConnectionError):
+                admin_client.connect(
+                    'http://localhost:8900', namespace='/admin',
+                    auth={'foo': 'baz'})
+        with socketio.SimpleClient() as admin_client:
+            with pytest.raises(ConnectionError):
+                admin_client.connect(
+                    'http://localhost:8900', namespace='/admin')
+
+    @with_instrumented_server(auth=[{'foo': 'bar'},
+                                    {'u': 'admin', 'p': 'secret'}])
+    def test_admin_connect_with_list_auth(self, isvr):
+        with socketio.SimpleClient() as admin_client:
+            admin_client.connect('http://localhost:8900', namespace='/admin',
+                                 auth={'foo': 'bar'})
+        with socketio.SimpleClient() as admin_client:
+            admin_client.connect('http://localhost:8900', namespace='/admin',
+                                 auth={'u': 'admin', 'p': 'secret'})
+        with socketio.SimpleClient() as admin_client:
+            with pytest.raises(ConnectionError):
+                admin_client.connect('http://localhost:8900',
+                                     namespace='/admin', auth={'foo': 'baz'})
+        with socketio.SimpleClient() as admin_client:
+            with pytest.raises(ConnectionError):
+                admin_client.connect('http://localhost:8900',
+                                     namespace='/admin')
+
+    @with_instrumented_server(auth=_custom_auth)
+    def test_admin_connect_with_function_auth(self, isvr):
+        with socketio.SimpleClient() as admin_client:
+            admin_client.connect('http://localhost:8900', namespace='/admin',
+                                 auth={'foo': 'bar'})
+        with socketio.SimpleClient() as admin_client:
+            with pytest.raises(ConnectionError):
+                admin_client.connect('http://localhost:8900',
+                                     namespace='/admin', auth={'foo': 'baz'})
+        with socketio.SimpleClient() as admin_client:
+            with pytest.raises(ConnectionError):
+                admin_client.connect('http://localhost:8900',
+                                     namespace='/admin')
+
+    @with_instrumented_server()
+    def test_admin_connect_only_admin(self, isvr):
+        with socketio.SimpleClient() as admin_client:
+            admin_client.connect('http://localhost:8900', namespace='/admin')
+            sid = admin_client.sid
+            expected = ['config', 'all_sockets', 'server_stats']
+            events = {}
+            while expected:
+                data = admin_client.receive(timeout=5)
+                if data[0] in expected:
+                    events[data[0]] = data[1]
+                    expected.remove(data[0])
+
+        assert 'supportedFeatures' in events['config']
+        assert 'ALL_EVENTS' in events['config']['supportedFeatures']
+        assert 'AGGREGATED_EVENTS' in events['config']['supportedFeatures']
+        assert 'EMIT' in events['config']['supportedFeatures']
+        assert len(events['all_sockets']) == 1
+        assert events['all_sockets'][0]['id'] == sid
+        assert events['all_sockets'][0]['rooms'] == [sid]
+        assert events['server_stats']['clientsCount'] == 1
+        assert events['server_stats']['pollingClientsCount'] == 0
+        assert len(events['server_stats']['namespaces']) == 3
+        assert {'name': '/', 'socketsCount': 0} in \
+            events['server_stats']['namespaces']
+        assert {'name': '/foo', 'socketsCount': 0} in \
+            events['server_stats']['namespaces']
+        assert {'name': '/admin', 'socketsCount': 1} in \
+            events['server_stats']['namespaces']
+
+    @with_instrumented_server()
+    def test_admin_connect_with_others(self, isvr):
+        with socketio.SimpleClient() as client1, \
+                socketio.SimpleClient() as client2, \
+                socketio.SimpleClient() as client3, \
+                socketio.SimpleClient() as admin_client:
+            client1.connect('http://localhost:8900')
+            client1.emit('enter_room', 'room')
+            sid1 = client1.sid
+
+            saved_check_for_upgrade = isvr._check_for_upgrade
+            isvr._check_for_upgrade = mock.MagicMock()
+            client2.connect('http://localhost:8900', namespace='/foo',
+                            transports=['polling'])
+            sid2 = client2.sid
+            isvr._check_for_upgrade = saved_check_for_upgrade
+
+            client3.connect('http://localhost:8900', namespace='/admin')
+            sid3 = client3.sid
+
+            admin_client.connect('http://localhost:8900', namespace='/admin')
+            sid = admin_client.sid
+            expected = ['config', 'all_sockets', 'server_stats']
+            events = {}
+            while expected:
+                data = admin_client.receive(timeout=5)
+                if data[0] in expected:
+                    events[data[0]] = data[1]
+                    expected.remove(data[0])
+
+        assert 'supportedFeatures' in events['config']
+        assert 'ALL_EVENTS' in events['config']['supportedFeatures']
+        assert 'AGGREGATED_EVENTS' in events['config']['supportedFeatures']
+        assert 'EMIT' in events['config']['supportedFeatures']
+        assert len(events['all_sockets']) == 4
+        assert events['server_stats']['clientsCount'] == 4
+        assert events['server_stats']['pollingClientsCount'] == 1
+        assert len(events['server_stats']['namespaces']) == 3
+        assert {'name': '/', 'socketsCount': 1} in \
+            events['server_stats']['namespaces']
+        assert {'name': '/foo', 'socketsCount': 1} in \
+            events['server_stats']['namespaces']
+        assert {'name': '/admin', 'socketsCount': 2} in \
+            events['server_stats']['namespaces']
+
+        for socket in events['all_sockets']:
+            if socket['id'] == sid:
+                assert socket['rooms'] == [sid]
+            elif socket['id'] == sid1:
+                assert socket['rooms'] == [sid1, 'room']
+            elif socket['id'] == sid2:
+                assert socket['rooms'] == [sid2]
+            elif socket['id'] == sid3:
+                assert socket['rooms'] == [sid3]
+
+    @with_instrumented_server(mode='production', read_only=True)
+    def test_admin_connect_production(self, isvr):
+        with socketio.SimpleClient() as admin_client:
+            admin_client.connect('http://localhost:8900', namespace='/admin')
+            expected = ['config', 'server_stats']
+            events = {}
+            while expected:
+                data = admin_client.receive(timeout=5)
+                if data[0] in expected:
+                    events[data[0]] = data[1]
+                    expected.remove(data[0])
+
+        assert 'supportedFeatures' in events['config']
+        assert 'ALL_EVENTS' not in events['config']['supportedFeatures']
+        assert 'AGGREGATED_EVENTS' in events['config']['supportedFeatures']
+        assert 'EMIT' not in events['config']['supportedFeatures']
+        assert events['server_stats']['clientsCount'] == 1
+        assert events['server_stats']['pollingClientsCount'] == 0
+        assert len(events['server_stats']['namespaces']) == 3
+        assert {'name': '/', 'socketsCount': 0} in \
+            events['server_stats']['namespaces']
+        assert {'name': '/foo', 'socketsCount': 0} in \
+            events['server_stats']['namespaces']
+        assert {'name': '/admin', 'socketsCount': 1} in \
+            events['server_stats']['namespaces']
+
+    @with_instrumented_server()
+    def test_admin_features(self, isvr):
+        with socketio.SimpleClient() as client1, \
+                socketio.SimpleClient() as client2, \
+                socketio.SimpleClient() as admin_client:
+            client1.connect('http://localhost:8900')
+            client2.connect('http://localhost:8900')
+            admin_client.connect('http://localhost:8900', namespace='/admin')
+
+            # emit from admin
+            admin_client.emit(
+                'emit', ('/', client1.sid, 'foo', {'bar': 'baz'}, 'extra'))
+            data = client1.receive(timeout=5)
+            assert data == ['foo', {'bar': 'baz'}, 'extra']
+
+            # emit from regular client
+            client1.emit('emit', 'foo')
+            data = client2.receive(timeout=5)
+            assert data == ['foo']
+
+            # join and leave
+            admin_client.emit('join', ('/', 'room', client1.sid))
+            admin_client.emit(
+                'emit', ('/', 'room', 'foo', {'bar': 'baz'}))
+            data = client1.receive(timeout=5)
+            assert data == ['foo', {'bar': 'baz'}]
+            admin_client.emit('leave', ('/', 'room'))
+
+            # disconnect
+            admin_client.emit('_disconnect', ('/', False, client1.sid))
+            for _ in range(10):
+                if not client1.connected:
+                    break
+                time.sleep(0.2)
+            assert not client1.connected

tests/common/test_simple_client.py

@@ -18,12 +18,12 @@ class TestSimpleClient(unittest.TestCase):
         client = SimpleClient(123, a='b')
         with mock.patch('socketio.simple_client.Client') as mock_client:
             client.connect('url', headers='h', auth='a', transports='t',
-                           namespace='n', socketio_path='s')
+                           namespace='n', socketio_path='s', wait_timeout='w')
             mock_client.assert_called_once_with(123, a='b')
             assert client.client == mock_client()
             mock_client().connect.assert_called_once_with(
                 'url', headers='h', auth='a', transports='t',
-                namespaces=['n'], socketio_path='s')
+                namespaces=['n'], socketio_path='s', wait_timeout='w')
             mock_client().event.call_count == 3
             mock_client().on.called_once_with('*')
             assert client.namespace == 'n'
@@ -33,12 +33,13 @@ class TestSimpleClient(unittest.TestCase):
         with SimpleClient(123, a='b') as client:
             with mock.patch('socketio.simple_client.Client') as mock_client:
                 client.connect('url', headers='h', auth='a', transports='t',
-                               namespace='n', socketio_path='s')
+                               namespace='n', socketio_path='s',
+                               wait_timeout='w')
                 mock_client.assert_called_once_with(123, a='b')
                 assert client.client == mock_client()
                 mock_client().connect.assert_called_once_with(
                     'url', headers='h', auth='a', transports='t',
-                    namespaces=['n'], socketio_path='s')
+                    namespaces=['n'], socketio_path='s', wait_timeout='w')
                 mock_client().event.call_count == 3
                 mock_client().on.called_once_with('*')
                 assert client.namespace == 'n'
@@ -54,7 +55,8 @@ class TestSimpleClient(unittest.TestCase):
 
     def test_properties(self):
         client = SimpleClient()
-        client.client = mock.MagicMock(sid='sid', transport='websocket')
+        client.client = mock.MagicMock(transport='websocket')
+        client.client.get_sid.return_value = 'sid'
         client.connected_event.set()
         client.connected = True
 

tests/web_server.py

@@ -0,0 +1,81 @@
+import threading
+import time
+from socketserver import ThreadingMixIn
+from wsgiref.simple_server import make_server, WSGIServer, WSGIRequestHandler
+import requests
+import socketio
+
+
+class SocketIOWebServer:
+    """A simple web server used for running Socket.IO servers in tests.
+
+    :param sio: a Socket.IO server instance.
+
+    Note 1: This class is not production-ready and is intended for testing.
+    Note 2: This class only supports the "threading" async_mode, with WebSocket
+    support provided by the simple-websocket package.
+    """
+    def __init__(self, sio):
+        if sio.async_mode != 'threading':
+            raise ValueError('The async_mode must be "threading"')
+
+        def http_app(environ, start_response):
+            start_response('200 OK', [('Content-Type', 'text/plain')])
+            return [b'OK']
+
+        self.sio = sio
+        self.app = socketio.WSGIApp(sio, http_app)
+        self.httpd = None
+        self.thread = None
+
+    def start(self, port=8900):
+        """Start the web server.
+
+        :param port: the port to listen on. Defaults to 8900.
+
+        The server is started in a background thread.
+        """
+        class ThreadingWSGIServer(ThreadingMixIn, WSGIServer):
+            pass
+
+        class WebSocketRequestHandler(WSGIRequestHandler):
+            def get_environ(self):
+                env = super().get_environ()
+
+                # pass the raw socket to the WSGI app so that it can be used
+                # by WebSocket connections (hack copied from gunicorn)
+                env['gunicorn.socket'] = self.connection
+                return env
+
+        self.httpd = make_server('', port, self._app_wrapper,
+                                 ThreadingWSGIServer, WebSocketRequestHandler)
+        self.thread = threading.Thread(target=self.httpd.serve_forever)
+        self.thread.start()
+
+        # wait for the server to start
+        while True:
+            try:
+                r = requests.get(f'http://localhost:{port}/')
+                r.raise_for_status()
+                if r.text == 'OK':
+                    break
+            except:
+                time.sleep(0.1)
+
+    def stop(self):
+        """Stop the web server."""
+        self.sio.shutdown()
+        self.httpd.shutdown()
+        self.httpd.server_close()
+        self.thread.join()
+        self.httpd = None
+        self.thread = None
+
+    def _app_wrapper(self, environ, start_response):
+        try:
+            return self.app(environ, start_response)
+        except StopIteration:
+            # end the WebSocket request without sending a response
+            # (this is a hack that was copied from gunicorn's threaded worker)
+            start_response('200 OK', [])
+            return []

tox.ini

@@ -14,10 +14,16 @@ python =
 [testenv]
 commands=
     pip install -e .
-    pytest -p no:logging --cov=socketio --cov-branch --cov-report=term-missing --cov-report=xml
+    pytest -p no:logging --timeout=60 --cov=socketio --cov-branch --cov-report=term-missing --cov-report=xml
 deps=
+    simple-websocket
+    uvicorn
+    requests
+    websocket-client
+    aiohttp
     msgpack
     pytest
+    pytest-timeout
     pytest-cov
 
 [testenv:flake8]