(
self,
side: str,
agreed_parameters: Dict[str, Any],
compression_options: Optional[Dict[str, Any]] = None,
)
| 999 | return options |
| 1000 | |
| 1001 | def _create_compressors( |
| 1002 | self, |
| 1003 | side: str, |
| 1004 | agreed_parameters: Dict[str, Any], |
| 1005 | compression_options: Optional[Dict[str, Any]] = None, |
| 1006 | ) -> None: |
| 1007 | # TODO: handle invalid parameters gracefully |
| 1008 | allowed_keys = set( |
| 1009 | [ |
| 1010 | "server_no_context_takeover", |
| 1011 | "client_no_context_takeover", |
| 1012 | "server_max_window_bits", |
| 1013 | "client_max_window_bits", |
| 1014 | ] |
| 1015 | ) |
| 1016 | for key in agreed_parameters: |
| 1017 | if key not in allowed_keys: |
| 1018 | raise ValueError("unsupported compression parameter %r" % key) |
| 1019 | other_side = "client" if (side == "server") else "server" |
| 1020 | self._compressor = _PerMessageDeflateCompressor( |
| 1021 | **self._get_compressor_options(side, agreed_parameters, compression_options) |
| 1022 | ) |
| 1023 | self._decompressor = _PerMessageDeflateDecompressor( |
| 1024 | max_message_size=self.params.max_message_size, |
| 1025 | **self._get_compressor_options( |
| 1026 | other_side, agreed_parameters, compression_options |
| 1027 | ) |
| 1028 | ) |
| 1029 | |
| 1030 | def _write_frame( |
| 1031 | self, fin: bool, opcode: int, data: bytes, flags: int = 0 |
no test coverage detected