(
self,
side: str,
agreed_parameters: Dict[str, Any],
compression_options: Optional[Dict[str, Any]] = None,
)
| 1022 | return options |
| 1023 | |
| 1024 | def _create_compressors( |
| 1025 | self, |
| 1026 | side: str, |
| 1027 | agreed_parameters: Dict[str, Any], |
| 1028 | compression_options: Optional[Dict[str, Any]] = None, |
| 1029 | ) -> None: |
| 1030 | # TODO: handle invalid parameters gracefully |
| 1031 | allowed_keys = { |
| 1032 | "server_no_context_takeover", |
| 1033 | "client_no_context_takeover", |
| 1034 | "server_max_window_bits", |
| 1035 | "client_max_window_bits", |
| 1036 | } |
| 1037 | for key in agreed_parameters: |
| 1038 | if key not in allowed_keys: |
| 1039 | raise ValueError("unsupported compression parameter %r" % key) |
| 1040 | other_side = "client" if (side == "server") else "server" |
| 1041 | self._compressor = _PerMessageDeflateCompressor( |
| 1042 | **self._get_compressor_options(side, agreed_parameters, compression_options) |
| 1043 | ) |
| 1044 | self._decompressor = _PerMessageDeflateDecompressor( |
| 1045 | max_message_size=self.params.max_message_size, |
| 1046 | **self._get_compressor_options( |
| 1047 | other_side, agreed_parameters, compression_options |
| 1048 | ), |
| 1049 | ) |
| 1050 | |
| 1051 | def _write_frame( |
| 1052 | self, fin: bool, opcode: int, data: bytes, flags: int = 0 |
no test coverage detected