| 182 | pass |
| 183 | |
| 184 | def start(self, static_path: pathlib.Path = _dirname / "assets") -> None: |
| 185 | request_subscribers: Dict[str, asyncio.Future] = {} |
| 186 | auth: Dict[str, Tuple[str, str]] = {} |
| 187 | csp: Dict[str, str] = {} |
| 188 | routes: Dict[str, Callable[[TestServerRequest], Any]] = {} |
| 189 | gzip_routes: Set[str] = set() |
| 190 | self.request_subscribers = request_subscribers |
| 191 | self.auth = auth |
| 192 | self.csp = csp |
| 193 | self.routes = routes |
| 194 | self._ws_handlers: List[Callable[["WebSocketProtocol"], None]] = [] |
| 195 | self.gzip_routes = gzip_routes |
| 196 | self.static_path = static_path |
| 197 | factory = TestServerFactory() |
| 198 | factory.server_instance = self |
| 199 | |
| 200 | ws_factory = WebSocketServerFactory() |
| 201 | ws_factory.protocol = WebSocketProtocol |
| 202 | setattr(ws_factory, "server_instance", self) |
| 203 | self._ws_resource = WebSocketResource(ws_factory) |
| 204 | |
| 205 | self.listen(factory) |
| 206 | |
| 207 | async def wait_for_request(self, path: str) -> TestServerRequest: |
| 208 | if path in self.request_subscribers: |