Class AsyncRequestFactory

django/test/client.py:674–782

Class that lets you create mock ASGI-like Request objects for use in testing. Usage: rf = AsyncRequestFactory() get_request = rf.get("/hello/") post_request = rf.post("/submit/", {"foo": "bar"}) Once you have a request object you can pass it to any view function, inclu

class AsyncRequestFactory(RequestFactory):
    """
    Class that lets you create mock ASGI-like Request objects for use in
    testing. Usage:

    rf = AsyncRequestFactory()
    get_request = rf.get("/hello/")
    post_request = rf.post("/submit/", {"foo": "bar"})

    Once you have a request object you can pass it to any view function,
    including synchronous ones. The reason we have a separate class here is:
    a) this makes ASGIRequest subclasses, and
    b) AsyncClient can subclass it.
    """

    def _base_scope(self, **request):
        """The base scope for a request."""
        # This is a minimal valid ASGI scope, plus:
        # - headers['cookie'] for cookie support,
        # - 'client' often useful, see #8551.
        scope = {
            "asgi": {"version": "3.0"},
            "type": "http",
            "http_version": "1.1",
            "client": ["127.0.0.1", 0],
            "server": ("testserver", "80"),
            "scheme": "http",
            "method": "GET",
            "headers": [],
            **self.defaults,
            **request,
        }
        scope["headers"].append(
            (
                b"cookie",
                b"; ".join(
                    sorted(
                        ("%s=%s" % (morsel.key, morsel.coded_value)).encode("ascii")
                        for morsel in self.cookies.values()
                    )
                ),
            )
        )
        return scope

    def request(self, **request):
        """Construct a generic request object."""
        # This is synchronous, which means all methods on this class are.
        # AsyncClient, however, has an async request function, which makes all
        # its methods async.
        if "_body_file" in request:
            body_file = request.pop("_body_file")
        else:
            body_file = FakePayload("")
        # Wrap FakePayload body_file to allow large read() in test environment.
        return ASGIRequest(
            self._base_scope(**request), LimitedStream(body_file, len(body_file))
        )