Method handle_request

starlette/testclient.py:225–373

(self, request: httpx.Request)

        self.client = client

    def handle_request(self, request: httpx.Request) -> httpx.Response:
        scheme = request.url.scheme
        netloc = request.url.netloc.decode(encoding=class="st">"ascii")
        path = request.url.path
        raw_path = request.url.raw_path
        query = request.url.query.decode(encoding=class="st">"ascii")

        default_port = {class="st">"http": 80, class="st">"ws": 80, class="st">"https": 443, class="st">"wss": 443}[scheme]

        if class="st">":" in netloc:
            host, port_string = netloc.split(class="st">":", 1)
            port = int(port_string)
        else:
            host = netloc
            port = default_port

        class="cm"># Include the class="st">'host' header.
        if class="st">"host" in request.headers:
            headers: list[tuple[bytes, bytes]] = []
        elif port == default_port:  class="cm"># pragma: no cover
            headers = [(bclass="st">"host", host.encode())]
        else:  class="cm"># pragma: no cover
            headers = [(bclass="st">"host", (fclass="st">"{host}:{port}").encode())]

        class="cm"># Include other request headers.
        headers += [(key.lower().encode(), value.encode()) for key, value in request.headers.multi_items()]

        scope: dict[str, Any]

        if scheme in {class="st">"ws", class="st">"wss"}:
            subprotocol = request.headers.get(class="st">"sec-websocket-protocol", None)
            if subprotocol is None:
                subprotocols: Sequence[str] = []
            else:
                subprotocols = [value.strip() for value in subprotocol.split(class="st">",")]
            scope = {
                class="st">"type": class="st">"websocket",
                class="st">"path": unquote(path),
                class="st">"raw_path": raw_path.split(bclass="st">"?", 1)[0],
                class="st">"root_path": self.root_path,
                class="st">"scheme": scheme,
                class="st">"query_string": query.encode(),
                class="st">"headers": headers,
                class="st">"client": self.client,
                class="st">"server": [host, port],
                class="st">"subprotocols": subprotocols,
                class="st">"state": self.app_state.copy(),
                class="st">"extensions": {class="st">"websocket.http.response": {}},
            }
            session = WebSocketTestSession(self.app, scope, self.portal_factory)
            raise _Upgrade(session)

        scope = {
            class="st">"type": class="st">"http",
            class="st">"http_version": class="st">"1.1",
            class="st">"method": request.method,
            class="st">"path": unquote(path),
            class="st">"raw_path": raw_path.split(bclass="st">"?", 1)[0],