MCPcopy Create free account
hub / github.com/miguelgrinberg/python-socketio / wrapped

Function wrapped

tests/common/test_admin.py:20–63  ·  view source on GitHub ↗
(self, *args, **kwargs)

Source from the content-addressed store, hash-verified

18 def decorator(f):
19 @wraps(f)
20 def wrapped(self, *args, **kwargs):
21 sio = socketio.Server(async_mode='threading')
22
23 @sio.event
24 def enter_room(sid, data):
25 sio.enter_room(sid, data)
26
27 @sio.event
28 def emit(sid, event):
29 sio.emit(event, skip_sid=sid)
30
31 @sio.event(namespace='/foo')
32 def connect(sid, environ, auth):
33 pass
34
35 if 'server_stats_interval' not in ikwargs:
36 ikwargs['server_stats_interval'] = 0.25
37
38 self.isvr = sio.instrument(auth=auth, **ikwargs)
39 server = SocketIOWebServer(sio)
40 server.start()
41
42 # import logging
43 # logging.getLogger('engineio.client').setLevel(logging.DEBUG)
44 # logging.getLogger('socketio.client').setLevel(logging.DEBUG)
45
46 original_schedule_ping = EngineIOSocket.schedule_ping
47 EngineIOSocket.schedule_ping = mock.MagicMock()
48
49 try:
50 ret = f(self, *args, **kwargs)
51 finally:
52 server.stop()
53 self.isvr.shutdown()
54 self.isvr.uninstrument()
55 self.isvr = None
56
57 EngineIOSocket.schedule_ping = original_schedule_ping
58
59 # import logging
60 # logging.getLogger('engineio.client').setLevel(logging.NOTSET)
61 # logging.getLogger('socketio.client').setLevel(logging.NOTSET)
62
63 return ret
64 return wrapped
65 return decorator
66

Callers

nothing calls this directly

Calls 6

instrumentMethod · 0.95
startMethod · 0.95
stopMethod · 0.95
SocketIOWebServerClass · 0.90
shutdownMethod · 0.45
uninstrumentMethod · 0.45

Tested by

no test coverage detected

Used in the wild real call sites across dependent graphs

searching dependent graphs…