Method process_response

src/websockets/client.py:131–174

Check a handshake response. Args: request: WebSocket handshake response received from the server. Raises: InvalidHandshake: If the handshake response is invalid.

(self, response: Response)

        return Request(self.uri.resource_name, headers)

    def process_response(self, response: Response) -> None:
        """
        Check a handshake response.

        Args:
            request: WebSocket handshake response received from the server.

        Raises:
            InvalidHandshake: If the handshake response is invalid.

        """

        if response.status_code != 101:
            raise InvalidStatus(response)

        headers = response.headers

        connection: list[ConnectionOption] = sum(
            [parse_connection(value) for value in headers.get_all("Connection")], []
        )
        if not any(value.lower() == "upgrade" for value in connection):
            raise InvalidUpgrade(
                "Connection", ", ".join(connection) if connection else None
            )

        upgrade: list[UpgradeProtocol] = sum(
            [parse_upgrade(value) for value in headers.get_all("Upgrade")], []
        )
        # For compatibility with non-strict implementations, ignore case when
        # checking the Upgrade header. It's supposed to be 'WebSocket'.
        if not (len(upgrade) == 1 and upgrade[0].lower() == "websocket"):
            raise InvalidUpgrade("Upgrade", ", ".join(upgrade) if upgrade else None)

        try:
            s_w_accept = headers["Sec-WebSocket-Accept"]
        except KeyError:
            raise InvalidHeader("Sec-WebSocket-Accept") from None
        except MultipleValuesError:
            raise InvalidHeader("Sec-WebSocket-Accept", "multiple values") from None
        if s_w_accept != accept_key(self.key):
            raise InvalidHeaderValue("Sec-WebSocket-Accept", s_w_accept)

        self.extensions = self.process_extensions(headers)
        self.subprotocol = self.process_subprotocol(headers)

    def process_extensions(self, headers: Headers) -> list[Extension]:
        """