Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/websockets/extensions/base.py: 76%
21 statements
« prev ^ index » next coverage.py v7.3.1, created at 2023-09-25 06:20 +0000
1from __future__ import annotations
3from typing import List, Optional, Sequence, Tuple
5from .. import frames
6from ..typing import ExtensionName, ExtensionParameter
9__all__ = ["Extension", "ClientExtensionFactory", "ServerExtensionFactory"]
12class Extension:
13 """
14 Base class for extensions.
16 """
18 name: ExtensionName
19 """Extension identifier."""
21 def decode(
22 self,
23 frame: frames.Frame,
24 *,
25 max_size: Optional[int] = None,
26 ) -> frames.Frame:
27 """
28 Decode an incoming frame.
30 Args:
31 frame (Frame): incoming frame.
32 max_size: maximum payload size in bytes.
34 Returns:
35 Frame: Decoded frame.
37 Raises:
38 PayloadTooBig: if decoding the payload exceeds ``max_size``.
40 """
41 raise NotImplementedError
43 def encode(self, frame: frames.Frame) -> frames.Frame:
44 """
45 Encode an outgoing frame.
47 Args:
48 frame (Frame): outgoing frame.
50 Returns:
51 Frame: Encoded frame.
53 """
54 raise NotImplementedError
57class ClientExtensionFactory:
58 """
59 Base class for client-side extension factories.
61 """
63 name: ExtensionName
64 """Extension identifier."""
66 def get_request_params(self) -> List[ExtensionParameter]:
67 """
68 Build parameters to send to the server for this extension.
70 Returns:
71 List[ExtensionParameter]: Parameters to send to the server.
73 """
74 raise NotImplementedError
76 def process_response_params(
77 self,
78 params: Sequence[ExtensionParameter],
79 accepted_extensions: Sequence[Extension],
80 ) -> Extension:
81 """
82 Process parameters received from the server.
84 Args:
85 params (Sequence[ExtensionParameter]): parameters received from
86 the server for this extension.
87 accepted_extensions (Sequence[Extension]): list of previously
88 accepted extensions.
90 Returns:
91 Extension: An extension instance.
93 Raises:
94 NegotiationError: if parameters aren't acceptable.
96 """
97 raise NotImplementedError
100class ServerExtensionFactory:
101 """
102 Base class for server-side extension factories.
104 """
106 name: ExtensionName
107 """Extension identifier."""
109 def process_request_params(
110 self,
111 params: Sequence[ExtensionParameter],
112 accepted_extensions: Sequence[Extension],
113 ) -> Tuple[List[ExtensionParameter], Extension]:
114 """
115 Process parameters received from the client.
117 Args:
118 params (Sequence[ExtensionParameter]): parameters received from
119 the client for this extension.
120 accepted_extensions (Sequence[Extension]): list of previously
121 accepted extensions.
123 Returns:
124 Tuple[List[ExtensionParameter], Extension]: To accept the offer,
125 parameters to send to the client for this extension and an
126 extension instance.
128 Raises:
129 NegotiationError: to reject the offer, if parameters received from
130 the client aren't acceptable.
132 """
133 raise NotImplementedError