-
Notifications
You must be signed in to change notification settings - Fork 0
/
__init__.py
68 lines (55 loc) · 1.8 KB
/
__init__.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
61
62
63
64
65
66
67
68
from dataclasses import dataclass
from typing import Any, Callable, Dict, Mapping, Optional
import schemas
from fastapi import FastAPI
from httpx import AsyncClient, Response
from starlette.datastructures import URLPath
from utils import create_access_token
DEFAULT_USER = schemas.UserWithoutPassword(
id=1,
name="default",
email="default@gmail.com",
password="default",
)
@dataclass
class RequestBody:
url: URLPath
body: Dict[str, Any]
@dataclass
class ResponseBody:
status_code: int
body: Dict[str, Any]
class AssertRequest:
def __init__(self):
claims = {**DEFAULT_USER.dict(), "exp": 123}
self.header = {"Authorization": f"Bearer {create_access_token(claims=claims)}"}
async def __call__(
self,
app: FastAPI,
method: str,
req_body: RequestBody,
resp_body: ResponseBody,
claims: Optional[Dict[str, Any]] = None,
assert_func: Callable = None,
data: Optional[Mapping[str, Any]] = None,
*args,
**kwargs,
):
header = self.header
if claims is not None:
header = {"Authorization": f"Bearer {create_access_token(claims=claims)}"}
async with AsyncClient(app=app, base_url="https://localhost") as ac:
resp: Response = await ac.request(
method=method,
url=req_body.url,
json=req_body.body,
data=data,
headers=header,
)
# If assert_func is not None, use assert_func to assert
if assert_func is not None:
assert_func(resp, resp_body, *args, **kwargs)
else:
assert resp.status_code == resp_body.status_code
assert resp.json() == resp_body.body
assert_request = AssertRequest()