Method _raise_on_close

starlette/testclient.py:150–163

(self, message: Message)

            await anyio.sleep_forever()

    def _raise_on_close(self, message: Message) -> None:
        if message[class="st">"type"] == class="st">"websocket.close":
            raise WebSocketDisconnect(code=message.get(class="st">"code", 1000), reason=message.get(class="st">"reason", class="st">""))
        elif message[class="st">"type"] == class="st">"websocket.http.response.start":
            status_code: int = message[class="st">"status"]
            headers: list[tuple[bytes, bytes]] = message[class="st">"headers"]
            body: list[bytes] = []
            while True:
                message = self.receive()
                assert message[class="st">"type"] == class="st">"websocket.http.response.body"
                body.append(message[class="st">"body"])
                if not message.get(class="st">"more_body", False):
                    break
            raise WebSocketDenialResponse(status_code=status_code, headers=headers, content=bclass="st">"".join(body))

    def send(self, message: Message) -> None:
        self.portal.call(self._receive_tx.send, message)