(self, sid, namespace, eio_sid=None)
| 369 | }, namespace=self.admin_namespace) |
| 370 | |
| 371 | def serialize_socket(self, sid, namespace, eio_sid=None): |
| 372 | if eio_sid is None: # pragma: no cover |
| 373 | eio_sid = self.sio.manager.eio_sid_from_sid(sid) |
| 374 | socket = self.sio.eio._get_socket(eio_sid) |
| 375 | environ = self.sio.environ.get(eio_sid, {}) |
| 376 | tm = self.sio.manager._timestamps[sid] if sid in \ |
| 377 | self.sio.manager._timestamps else 0 |
| 378 | return { |
| 379 | 'id': sid, |
| 380 | 'clientId': eio_sid, |
| 381 | 'transport': 'websocket' if socket.upgraded else 'polling', |
| 382 | 'nsp': namespace, |
| 383 | 'data': {}, |
| 384 | 'handshake': { |
| 385 | 'address': environ.get('REMOTE_ADDR', ''), |
| 386 | 'headers': {k[5:].lower(): v for k, v in environ.items() |
| 387 | if k.startswith('HTTP_')}, |
| 388 | 'query': {k: v[0] if len(v) == 1 else v for k, v in parse_qs( |
| 389 | environ.get('QUERY_STRING', '')).items()}, |
| 390 | 'secure': environ.get('wsgi.url_scheme', '') == 'https', |
| 391 | 'url': environ.get('PATH_INFO', ''), |
| 392 | 'issued': tm * 1000, |
| 393 | 'time': datetime.fromtimestamp(tm, timezone.utc).isoformat() |
| 394 | if tm else '', |
| 395 | }, |
| 396 | 'rooms': self.sio.manager.get_rooms(sid, namespace), |
| 397 | } |
no test coverage detected