Class AsyncPathMapDispatch

tests/clients/asgi_helper.py:38–67

class AsyncPathMapDispatch:
    def __init__(self, path_maps, side_effects=None):
        self.path_maps = path_maps
        self.side_effects = side_effects or dict()

    async def __call__(self, scope, receive, send):
        request = ASGIRequest(scope, receive=receive)

        side_effect = self.side_effects.get(request.url.path)
        if side_effect is not None:
            side_effect(request)

        rv = self.path_maps[request.url.path]
        status_code = rv.get("status_code", 200)
        body = rv.get("body")
        headers = rv.get("headers", {})
        if isinstance(body, dict):
            body = json.dumps(body).encode()
            headers["Content-Type"] = "application/json"
        else:
            if isinstance(body, str):
                body = body.encode()
            headers["Content-Type"] = "application/x-www-form-urlencoded"

        response = ASGIResponse(
            status_code=status_code,
            content=body,
            headers=headers,
        )
        await response(scope, receive, send)