async_asgi_testclient/testing.py
async def open(
        self,
        path: str,
        *,
        method: str = "GET",
        headers: Optional[Union[dict, CIMultiDict]] = None,
        data: Any = None,
        form: Optional[dict] = None,
        files: Optional[dict] = None,
        query_string: Optional[dict] = None,
        json: Any = sentinel,
        scheme: str = "http",
        cookies: Optional[dict] = None,
        stream: bool = False,
        allow_redirects: bool = True,
    ):
        """Open a request to the app associated with this client.

        Arguments:
            path
                The path to request. If the query_string argument is not
                defined this argument will be partitioned on a '?' with the
                following part being considered the query_string.

            method
                The method to make the request with, defaults to 'GET'.

            headers
                Headers to include in the request.

            data
                Raw data to send in the request body or async generator

            form
                Data to send form encoded in the request body.

            files
                Data to send as multipart in the request body.

            query_string
                To send as a dictionary, alternatively the query_string can be
                determined from the path.

            json
                Data to send json encoded in the request body.

            scheme
                The scheme to use in the request, default http.

            cookies
                Cookies to send in the request instead of cookies in
                TestClient.cookie_jar

            stream
                Return the response in streaming instead of buffering

            allow_redirects
                If set to True follows redirects

        Returns:
            The response from the app handling the request.
        """
        input_queue: asyncio.Queue[dict] = asyncio.Queue()
        output_queue: asyncio.Queue[dict] = asyncio.Queue()

        if not headers:
            headers = {}
        merged_headers = self.headers.copy()
        merged_headers.update(headers)
        headers, path, query_string_bytes = make_test_headers_path_and_query_string(
            self.application, path, merged_headers, query_string
        )

        if [
            json is not sentinel,
            form is not None,
            data is not None,
            files is not None,
        ].count(True) > 1:
            raise ValueError(
                "Test args 'json', 'form', 'files' and 'data' are mutually exclusive"
            )

        request_data = b""

        if isinstance(data, str):
            request_data = data.encode("utf-8")
        elif isinstance(data, bytes):
            request_data = data

        if json is not sentinel:
            request_data = dumps(json).encode("utf-8")
            headers["Content-Type"] = "application/json"

        if form is not None:
            request_data = urlencode(form).encode("utf-8")
            headers["Content-Type"] = "application/x-www-form-urlencoded"

        if files is not None:
            request_data, content_type = encode_multipart_formdata(files)
            headers["Content-Type"] = content_type

        if cookies is None:  # use TestClient.cookie_jar
            cookie_jar = self.cookie_jar
        else:
            cookie_jar = SimpleCookie(cookies)

        if cookie_jar and cookie_jar.output(header=""):
            headers.add("Cookie", cookie_jar.output(header=""))

        scope = {
            "type": "http",
            "http_version": "1.1",
            "asgi": {"version": "3.0"},
            "method": method,
            "scheme": scheme,
            "path": path,
            "query_string": query_string_bytes,
            "root_path": "",
            "headers": flatten_headers(headers),
        }
        scope.update(self._scope)

        create_monitored_task(
            self.application(scope, input_queue.get, output_queue.put),
            output_queue.put_nowait,
        )

        send = input_queue.put_nowait
        receive_or_fail = partial(receive, output_queue, timeout=self.timeout)

        # Send request
        if inspect.isasyncgen(data):
            async for is_last, body in is_last_one(data):
                send({"type": "http.request", "body": body, "more_body": not is_last})
        else:
            send({"type": "http.request", "body": request_data})

        response = Response(stream, receive_or_fail, send)

        # Receive response start
        message = await self.wait_response(receive_or_fail, "http.response.start")
        response.status_code = message["status"]
        response.headers = CIMultiDict(
            [(k.decode("utf8"), v.decode("utf8")) for k, v in message["headers"]]
        )

        # Receive initial response body
        message = await self.wait_response(receive_or_fail, "http.response.body")
        response.raw.write(message["body"])
        response._more_body = message.get("more_body", False)

        # Consume the remaining response if not in stream
        if not stream:
            bytes_io = BytesRW()
            bytes_io.write(response.raw.read())
            async for chunk in response:
                bytes_io.write(chunk)
            response.raw = bytes_io
            response._content = bytes_io.read()
            response._content_consumed = True

        if cookie_jar is not None:
            cookies = SimpleCookie()
            for c in response.headers.getall("Set-Cookie", ""):
                cookies.load(c)
            response.cookies = requests.cookies.RequestsCookieJar()
            response.cookies.update(cookies)
            cookie_jar.update(cookies)

        if allow_redirects and response.is_redirect:
            path = response.headers["location"]
            return await self.get(path)
        else:
            return response
Open a request to the app associated with this client.

        Arguments:
            path
                The path to request. If the query_string argument is not
                defined this argument will be partitioned on a '?' with the
                following part being considered the query_string.

            method
                The method to make the request with, defaults to 'GET'.

            headers
                Headers to include in the request.

            data
                Raw data to send in the request body or async generator

            form
                Data to send form encoded in the request body.

            files
                Data to send as multipart in the request body.

            query_string
                To send as a dictionary, alternatively the query_string can be
                determined from the path.

            json
                Data to send json encoded in the request body.

            scheme
                The scheme to use in the request, default http.

            cookies
                Cookies to send in the request instead of cookies in
                TestClient.cookie_jar

            stream
                Return the response in streaming instead of buffering

            allow_redirects
                If set to True follows redirects

        Returns:
            The response from the app handling the request.
        
Open a request to the app associated with this client.
Arguments:
path
The path to request. If the query_string argument is not
defined this argument will be partitioned on a '?' with the
following part being considered the query_string.

method
The method to make the request with, defaults to 'GET'.

headers
Headers to include in the request.

data
Raw data to send in the request body or async generator

form
Data to send form encoded in the request body.

files
Data to send as multipart in the request body.

query_string
To send as a dictionary, alternatively the query_string can be
determined from the path.

json
Data to send json encoded in the request body.

scheme
The scheme to use in the request, default http.

cookies
Cookies to send in the request instead of cookies in
TestClient.cookie_jar

stream
Return the response in streaming instead of buffering

allow_redirects
If set to True follows redirects

The response from the app handling the request.
Open a request to the app associated with this client . Arguments : path The path to request . If the query_string argument is not defined this argument will be partitioned on a ' ? ' with the following part being considered the query_string . method The method to make the request with defaults to ' GET ' . headers Headers to include in the request . data Raw data to send in the request body or async generator form Data to send form encoded in the request body . files Data to send as multipart in the request body . query_string To send as a dictionary alternatively the query_string can be determined from the path . json Data to send json encoded in the request body . scheme The scheme to use in the request default http . cookies Cookies to send in the request instead of cookies in TestClient . cookie_jar stream Return the response in streaming instead of buffering allow_redirects If set to True follows redirects The response from the app handling the request .
async def open(
        self,
        path: str,
        *,
        method: str = "GET",
        headers: Optional[Union[dict, CIMultiDict]] = None,
        data: Any = None,
        form: Optional[dict] = None,
        files: Optional[dict] = None,
        query_string: Optional[dict] = None,
        json: Any = sentinel,
        scheme: str = "http",
        cookies: Optional[dict] = None,
        stream: bool = False,
        allow_redirects: bool = True,
    ):
        input_queue: asyncio.Queue[dict] = asyncio.Queue()
        output_queue: asyncio.Queue[dict] = asyncio.Queue()
        if not headers:
            headers = {}
        merged_headers = self.headers.copy()
        merged_headers.update(headers)
        headers, path, query_string_bytes = make_test_headers_path_and_query_string(
            self.application, path, merged_headers, query_string
        )
        if [
            json is not sentinel,
            form is not None,
            data is not None,
            files is not None,
        ].count(True) > 1:
            raise ValueError(
                "Test args 'json', 'form', 'files' and 'data' are mutually exclusive"
            )
        request_data = b""
        if isinstance(data, str):
            request_data = data.encode("utf-8")
        elif isinstance(data, bytes):
            request_data = data
        if json is not sentinel:
            request_data = dumps(json).encode("utf-8")
            headers["Content-Type"] = "application/json"
        if form is not None:
            request_data = urlencode(form).encode("utf-8")
            headers["Content-Type"] = "application/x-www-form-urlencoded"
        if files is not None:
            request_data, content_type = encode_multipart_formdata(files)
            headers["Content-Type"] = content_type
        if cookies is None:  
            cookie_jar = self.cookie_jar
        else:
            cookie_jar = SimpleCookie(cookies)
        if cookie_jar and cookie_jar.output(header=""):
            headers.add("Cookie", cookie_jar.output(header=""))
        scope = {
            "type": "http",
            "http_version": "1.1",
            "asgi": {"version": "3.0"},
            "method": method,
            "scheme": scheme,
            "path": path,
            "query_string": query_string_bytes,
            "root_path": "",
            "headers": flatten_headers(headers),
        }
        scope.update(self._scope)
        create_monitored_task(
            self.application(scope, input_queue.get, output_queue.put),
            output_queue.put_nowait,
        )
        send = input_queue.put_nowait
        receive_or_fail = partial(receive, output_queue, timeout=self.timeout)
        if inspect.isasyncgen(data):
            async for is_last, body in is_last_one(data):
                send({"type": "http.request", "body": body, "more_body": not is_last})
        else:
            send({"type": "http.request", "body": request_data})
        response = Response(stream, receive_or_fail, send)
        message = await self.wait_response(receive_or_fail, "http.response.start")
        response.status_code = message["status"]
        response.headers = CIMultiDict(
            [(k.decode("utf8"), v.decode("utf8")) for k, v in message["headers"]]
        )
        message = await self.wait_response(receive_or_fail, "http.response.body")
        response.raw.write(message["body"])
        response._more_body = message.get("more_body", False)
        if not stream:
            bytes_io = BytesRW()
            bytes_io.write(response.raw.read())
            async for chunk in response:
                bytes_io.write(chunk)
            response.raw = bytes_io
            response._content = bytes_io.read()
            response._content_consumed = True
        if cookie_jar is not None:
            cookies = SimpleCookie()
            for c in response.headers.getall("Set-Cookie", ""):
                cookies.load(c)
            response.cookies = requests.cookies.RequestsCookieJar()
            response.cookies.update(cookies)
            cookie_jar.update(cookies)
        if allow_redirects and response.is_redirect:
            path = response.headers["location"]
            return await self.get(path)
        else:
            return response
async def open ( self , path : str , * , method : str = "GET" , headers : Optional [ Union [ dict , CIMultiDict ] ] = None , data : Any = None , form : Optional [ dict ] = None , files : Optional [ dict ] = None , query_string : Optional [ dict ] = None , json : Any = sentinel , scheme : str = "http" , cookies : Optional [ dict ] = None , stream : bool = False , allow_redirects : bool = True , ) : input_queue : asyncio . Queue [ dict ] = asyncio . Queue ( ) output_queue : asyncio . Queue [ dict ] = asyncio . Queue ( ) if not headers : headers = { } merged_headers = self . headers . copy ( ) merged_headers . update ( headers ) headers , path , query_string_bytes = make_test_headers_path_and_query_string ( self . application , path , merged_headers , query_string ) if [ json is not sentinel , form is not None , data is not None , files is not None , ] . count ( True ) > 1 : raise ValueError ( "Test args 'json', 'form', 'files' and 'data' are mutually exclusive" ) request_data = b"" if isinstance ( data , str ) : request_data = data . encode ( "utf-8" ) elif isinstance ( data , bytes ) : request_data = data if json is not sentinel : request_data = dumps ( json ) . encode ( "utf-8" ) headers [ "Content-Type" ] = "application/json" if form is not None : request_data = urlencode ( form ) . encode ( "utf-8" ) headers [ "Content-Type" ] = "application/x-www-form-urlencoded" if files is not None : request_data , content_type = encode_multipart_formdata ( files ) headers [ "Content-Type" ] = content_type if cookies is None : cookie_jar = self . cookie_jar else : cookie_jar = SimpleCookie ( cookies ) if cookie_jar and cookie_jar . output ( header = "" ) : headers . add ( "Cookie" , cookie_jar . output ( header = "" ) ) scope = { "type" : "http" , "http_version" : "1.1" , "asgi" : { "version" : "3.0" } , "method" : method , "scheme" : scheme , "path" : path , "query_string" : query_string_bytes , "root_path" : "" , "headers" : flatten_headers ( headers ) , } scope . update ( self . _scope ) create_monitored_task ( self . application ( scope , input_queue . get , output_queue . put ) , output_queue . put_nowait , ) send = input_queue . put_nowait receive_or_fail = partial ( receive , output_queue , timeout = self . timeout ) if inspect . isasyncgen ( data ) : async for is_last , body in is_last_one ( data ) : send ( { "type" : "http.request" , "body" : body , "more_body" : not is_last } ) else : send ( { "type" : "http.request" , "body" : request_data } ) response = Response ( stream , receive_or_fail , send ) message = await self . wait_response ( receive_or_fail , "http.response.start" ) response . status_code = message [ "status" ] response . headers = CIMultiDict ( [ ( k . decode ( "utf8" ) , v . decode ( "utf8" ) ) for k , v in message [ "headers" ] ] ) message = await self . wait_response ( receive_or_fail , "http.response.body" ) response . raw . write ( message [ "body" ] ) response . _more_body = message . get ( "more_body" , False ) if not stream : bytes_io = BytesRW ( ) bytes_io . write ( response . raw . read ( ) ) async for chunk in response : bytes_io . write ( chunk ) response . raw = bytes_io response . _content = bytes_io . read ( ) response . _content_consumed = True if cookie_jar is not None : cookies = SimpleCookie ( ) for c in response . headers . getall ( "Set-Cookie" , "" ) : cookies . load ( c ) response . cookies = requests . cookies . RequestsCookieJar ( ) response . cookies . update ( cookies ) cookie_jar . update ( cookies ) if allow_redirects and response . is_redirect : path = response . headers [ "location" ] return await self . get ( path ) else : return response
Open a request to the app associated with this client.
Open a request to the app associated with this client .
"""Open a request to the app associated with this client.

        Arguments:
            path
                The path to request. If the query_string argument is not
                defined this argument will be partitioned on a '?' with the
                following part being considered the query_string.

            method
                The method to make the request with, defaults to 'GET'.

            headers
                Headers to include in the request.

            data
                Raw data to send in the request body or async generator

            form
                Data to send form encoded in the request body.

            files
                Data to send as multipart in the request body.

            query_string
                To send as a dictionary, alternatively the query_string can be
                determined from the path.

            json
                Data to send json encoded in the request body.

            scheme
                The scheme to use in the request, default http.

            cookies
                Cookies to send in the request instead of cookies in
                TestClient.cookie_jar

            stream
                Return the response in streaming instead of buffering

            allow_redirects
                If set to True follows redirects

        Returns:
            The response from the app handling the request.
        """ # use TestClient.cookie_jar # Send request # Receive response start # Receive initial response body # Consume the remaining response if not in stream
self
path
str
method
str
headers
Optional[Union[dict, CIMultiDict]]
data
Any
form
Optional[dict]
files
Optional[dict]
query_string
Optional[dict]
json
Any
scheme
str
cookies
Optional[dict]
stream
bool
allow_redirects
bool