/
utils.py
60 lines (51 loc) · 1.74 KB
/
utils.py
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
import json
from httpx import (
Dispatcher,
Response,
)
class MockDispatch(Dispatcher):
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 send(self, request, verify=None, cert=None, timeout=None):
if self.assert_func:
self.assert_func(request)
return Response(
self.status_code,
content=self.body,
headers=self.headers,
request=request,
)
class PathMapDispatch(Dispatcher):
def __init__(self, path_maps):
self.path_maps = path_maps
async def send(self, request, verify=None, cert=None, timeout=None):
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'
return Response(
status_code,
content=body,
headers=headers,
request=request,
)