1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
|
import json
from starlette.requests import Request as ASGIRequest
from starlette.responses import Response as ASGIResponse
class AsyncMockDispatch:
def __init__(self, body=b"", status_code=200, headers=None, assert_func=None):
if headers is None:
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"
self.body = body
self.status_code = status_code
self.headers = headers
self.assert_func = assert_func
async def __call__(self, scope, receive, send):
request = ASGIRequest(scope, receive=receive)
if self.assert_func:
await self.assert_func(request)
response = ASGIResponse(
status_code=self.status_code,
content=self.body,
headers=self.headers,
)
await response(scope, receive, send)
class AsyncPathMapDispatch:
def __init__(self, path_maps):
self.path_maps = path_maps
async def __call__(self, scope, receive, send):
request = ASGIRequest(scope, receive=receive)
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)
|
