Function wrapped

tests/async/test_admin.py:23–69

(self, *args, **kwargs)

    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 self.isvr.shutdown()
                await sio.shutdown()

            if 'server_stats_interval' not in ikwargs:
                ikwargs['server_stats_interval'] = 0.25

            self.isvr = 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, *args, **kwargs)
            finally:
                server.stop()
                self.isvr.uninstrument()
                self.isvr = None

            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