(self, sid, environ, client_auth)
| 135 | Socket._send_ping = Socket.__send_ping |
| 136 | |
| 137 | def admin_connect(self, sid, environ, client_auth): |
| 138 | if self.auth: |
| 139 | authenticated = False |
| 140 | if isinstance(self.auth, dict): |
| 141 | authenticated = client_auth == self.auth |
| 142 | elif isinstance(self.auth, list): |
| 143 | authenticated = client_auth in self.auth |
| 144 | else: |
| 145 | authenticated = self.auth(client_auth) |
| 146 | if not authenticated: |
| 147 | raise ConnectionRefusedError('authentication failed') |
| 148 | |
| 149 | def config(sid): |
| 150 | self.sio.sleep(0.1) |
| 151 | |
| 152 | # supported features |
| 153 | features = ['AGGREGATED_EVENTS'] |
| 154 | if not self.read_only: |
| 155 | features += ['EMIT', 'JOIN', 'LEAVE', 'DISCONNECT', 'MJOIN', |
| 156 | 'MLEAVE', 'MDISCONNECT'] |
| 157 | if self.mode == 'development': |
| 158 | features.append('ALL_EVENTS') |
| 159 | self.sio.emit('config', {'supportedFeatures': features}, |
| 160 | to=sid, namespace=self.admin_namespace) |
| 161 | |
| 162 | # send current sockets |
| 163 | if self.mode == 'development': |
| 164 | all_sockets = [] |
| 165 | for nsp in self.sio.manager.get_namespaces(): |
| 166 | for sid, eio_sid in self.sio.manager.get_participants( |
| 167 | nsp, None): |
| 168 | all_sockets.append( |
| 169 | self.serialize_socket(sid, nsp, eio_sid)) |
| 170 | self.sio.emit('all_sockets', all_sockets, to=sid, |
| 171 | namespace=self.admin_namespace) |
| 172 | |
| 173 | self.sio.start_background_task(config, sid) |
| 174 | |
| 175 | def admin_emit(self, _, namespace, room_filter, event, *data): |
| 176 | self.sio.emit(event, data, to=room_filter, namespace=namespace) |
nothing calls this directly
no test coverage detected