(self, event: h11.Request)
| 282 | break |
| 283 | |
| 284 | def handle_websocket_upgrade(self, event: h11.Request) -> None: |
| 285 | if self.logger.level <= TRACE_LOG_LEVEL: # pragma: full coverage |
| 286 | prefix = "%s:%d - " % self.client if self.client else "" |
| 287 | self.logger.log(TRACE_LOG_LEVEL, "%sUpgrading to WebSocket", prefix) |
| 288 | |
| 289 | self.connections.discard(self) |
| 290 | output = [event.method, b" ", event.target, b" HTTP/1.1\r\n"] |
| 291 | for name, value in self.headers: |
| 292 | output += [name, b": ", value, b"\r\n"] |
| 293 | output.append(b"\r\n") |
| 294 | protocol = self.ws_protocol_class( # type: ignore[call-arg, misc] |
| 295 | config=self.config, |
| 296 | server_state=self.server_state, |
| 297 | app_state=self.app_state, |
| 298 | ) |
| 299 | protocol.connection_made(self.transport) |
| 300 | protocol.data_received(b"".join(output)) |
| 301 | self.transport.set_protocol(protocol) |
| 302 | |
| 303 | def send_400_response(self, msg: str) -> None: |
| 304 | reason = STATUS_PHRASES[400] |
no test coverage detected