(self, message: Message)
| 110 | await self.send({"type": "websocket.accept", "subprotocol": subprotocol, "headers": headers}) |
| 111 | |
| 112 | def _raise_on_disconnect(self, message: Message) -> None: |
| 113 | if message["type"] == "websocket.disconnect": |
| 114 | raise WebSocketDisconnect(message["code"], message.get("reason")) |
| 115 | |
| 116 | async def receive_text(self) -> str: |
| 117 | if self.application_state != WebSocketState.CONNECTED: |
no test coverage detected