| 205 | |
| 206 | |
| 207 | class _TestClientTransport(httpx.BaseTransport): |
| 208 | def __init__( |
| 209 | self, |
| 210 | app: ASGI3App, |
| 211 | portal_factory: _PortalFactoryType, |
| 212 | raise_server_exceptions: bool = True, |
| 213 | root_path: str = "", |
| 214 | *, |
| 215 | client: tuple[str, int], |
| 216 | app_state: dict[str, Any], |
| 217 | ) -> None: |
| 218 | self.app = app |
| 219 | self.raise_server_exceptions = raise_server_exceptions |
| 220 | self.root_path = root_path |
| 221 | self.portal_factory = portal_factory |
| 222 | self.app_state = app_state |
| 223 | self.client = client |
| 224 | |
| 225 | def handle_request(self, request: httpx.Request) -> httpx.Response: |
| 226 | scheme = request.url.scheme |
| 227 | netloc = request.url.netloc.decode(encoding="ascii") |
| 228 | path = request.url.path |
| 229 | raw_path = request.url.raw_path |
| 230 | query = request.url.query.decode(encoding="ascii") |
| 231 | |
| 232 | default_port = {"http": 80, "ws": 80, "https": 443, "wss": 443}[scheme] |
| 233 | |
| 234 | if ":" in netloc: |
| 235 | host, port_string = netloc.split(":", 1) |
| 236 | port = int(port_string) |
| 237 | else: |
| 238 | host = netloc |
| 239 | port = default_port |
| 240 | |
| 241 | # Include the 'host' header. |
| 242 | if "host" in request.headers: |
| 243 | headers: list[tuple[bytes, bytes]] = [] |
| 244 | elif port == default_port: # pragma: no cover |
| 245 | headers = [(b"host", host.encode())] |
| 246 | else: # pragma: no cover |
| 247 | headers = [(b"host", (f"{host}:{port}").encode())] |
| 248 | |
| 249 | # Include other request headers. |
| 250 | headers += [(key.lower().encode(), value.encode()) for key, value in request.headers.multi_items()] |
| 251 | |
| 252 | scope: dict[str, Any] |
| 253 | |
| 254 | if scheme in {"ws", "wss"}: |
| 255 | subprotocol = request.headers.get("sec-websocket-protocol", None) |
| 256 | if subprotocol is None: |
| 257 | subprotocols: Sequence[str] = [] |
| 258 | else: |
| 259 | subprotocols = [value.strip() for value in subprotocol.split(",")] |
| 260 | scope = { |
| 261 | "type": "websocket", |
| 262 | "path": unquote(path), |
| 263 | "raw_path": raw_path.split(b"?", 1)[0], |
| 264 | "root_path": self.root_path, |