(self, request: httpx.Request)
| 221 | self.client = client |
| 222 | |
| 223 | def handle_request(self, request: httpx.Request) -> httpx.Response: |
| 224 | scheme = request.url.scheme |
| 225 | host = request.url.raw_host.decode(encoding="ascii") |
| 226 | path = request.url.path |
| 227 | raw_path = request.url.raw_path |
| 228 | query = request.url.query.decode(encoding="ascii") |
| 229 | |
| 230 | default_port = {"http": 80, "ws": 80, "https": 443, "wss": 443}[scheme] |
| 231 | port = request.url.port |
| 232 | if port is None: |
| 233 | port = default_port |
| 234 | |
| 235 | # Include the 'host' header. |
| 236 | if "host" in request.headers: |
| 237 | headers: list[tuple[bytes, bytes]] = [] |
| 238 | else: # pragma: no cover |
| 239 | headers = [(b"host", request.url.netloc)] |
| 240 | |
| 241 | # Include other request headers. |
| 242 | headers += [(key.lower().encode(), value.encode()) for key, value in request.headers.multi_items()] |
| 243 | |
| 244 | scope: dict[str, Any] |
| 245 | |
| 246 | if scheme in {"ws", "wss"}: |
| 247 | subprotocol = request.headers.get("sec-websocket-protocol", None) |
| 248 | if subprotocol is None: |
| 249 | subprotocols: Sequence[str] = [] |
| 250 | else: |
| 251 | subprotocols = [value.strip() for value in subprotocol.split(",")] |
| 252 | scope = { |
| 253 | "type": "websocket", |
| 254 | "path": unquote(path), |
| 255 | "raw_path": raw_path.split(b"?", 1)[0], |
| 256 | "root_path": self.root_path, |
| 257 | "scheme": scheme, |
| 258 | "query_string": query.encode(), |
| 259 | "headers": headers, |
| 260 | "client": self.client, |
| 261 | "server": [host, port], |
| 262 | "subprotocols": subprotocols, |
| 263 | "state": self.app_state.copy(), |
| 264 | "extensions": {"websocket.http.response": {}}, |
| 265 | } |
| 266 | session = WebSocketTestSession(self.app, scope, self.portal_factory) |
| 267 | raise _Upgrade(session) |
| 268 | |
| 269 | scope = { |
| 270 | "type": "http", |
| 271 | "http_version": "1.1", |
| 272 | "method": request.method, |
| 273 | "path": unquote(path), |
| 274 | "raw_path": raw_path.split(b"?", 1)[0], |
| 275 | "root_path": self.root_path, |
| 276 | "scheme": scheme, |
| 277 | "query_string": query.encode(), |
| 278 | "headers": headers, |
| 279 | "client": self.client, |
| 280 | "server": [host, port], |
nothing calls this directly
no test coverage detected