-
Notifications
You must be signed in to change notification settings - Fork 421
/
Copy pathtest_openapi_responses.py
167 lines (122 loc) · 5.17 KB
/
test_openapi_responses.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
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
from secrets import randbelow
from typing import Union
from pydantic import BaseModel
from aws_lambda_powertools.event_handler import APIGatewayRestResolver, Response
def test_openapi_default_response():
app = APIGatewayRestResolver(enable_validation=True)
@app.get("/")
def handler():
pass
schema = app.get_openapi_schema()
responses = schema.paths["/"].get.responses
assert 200 in responses.keys()
assert responses[200].description == "Successful Response"
assert 422 in responses.keys()
assert responses[422].description == "Validation Error"
def test_openapi_200_response_with_description():
app = APIGatewayRestResolver(enable_validation=True)
@app.get("/", response_description="Custom response")
def handler():
return {"message": "hello world"}
schema = app.get_openapi_schema()
responses = schema.paths["/"].get.responses
assert 200 in responses.keys()
assert responses[200].description == "Custom response"
assert 422 in responses.keys()
assert responses[422].description == "Validation Error"
def test_openapi_200_custom_response():
app = APIGatewayRestResolver(enable_validation=True)
@app.get("/", responses={202: {"description": "Custom response"}})
def handler():
return {"message": "hello world"}
schema = app.get_openapi_schema()
responses = schema.paths["/"].get.responses
assert 202 in responses.keys()
assert responses[202].description == "Custom response"
assert 200 not in responses.keys() # 200 was not added due to custom responses
assert 422 in responses.keys() # 422 is always added due to potential data validation errors
def test_openapi_422_default_response():
app = APIGatewayRestResolver(enable_validation=True)
@app.get("/")
def handler():
return {"message": "hello world"}
schema = app.get_openapi_schema()
responses = schema.paths["/"].get.responses
assert 422 in responses.keys()
assert responses[422].description == "Validation Error"
def test_openapi_422_custom_response():
app = APIGatewayRestResolver(enable_validation=True)
@app.get("/", responses={422: {"description": "Custom validation response"}})
def handler():
return {"message": "hello world"}
schema = app.get_openapi_schema()
responses = schema.paths["/"].get.responses
assert 422 in responses.keys()
assert responses[422].description == "Custom validation response"
def test_openapi_200_custom_schema():
app = APIGatewayRestResolver(enable_validation=True)
class User(BaseModel):
pass
@app.get(
"/",
responses={200: {"description": "Custom response", "content": {"application/json": {"schema": User.schema()}}}},
)
def handler():
return {"message": "hello world"}
schema = app.get_openapi_schema()
responses = schema.paths["/"].get.responses
assert 200 in responses.keys()
assert responses[200].description == "Custom response"
assert responses[200].content["application/json"].schema_.title == "User"
def test_openapi_union_response():
app = APIGatewayRestResolver(enable_validation=True)
class User(BaseModel):
pass
class Order(BaseModel):
pass
@app.get(
"/",
responses={
200: {"description": "200 Response", "content": {"application/json": {"model": User}}},
202: {"description": "202 Response", "content": {"application/json": {"model": Order}}},
},
)
def handler() -> Response[Union[User, Order]]:
if randbelow(2) > 0:
return Response(status_code=200, body=User())
else:
return Response(status_code=202, body=Order())
schema = app.get_openapi_schema()
responses = schema.paths["/"].get.responses
assert 200 in responses.keys()
assert responses[200].description == "200 Response"
assert responses[200].content["application/json"].schema_.ref == "#/components/schemas/User"
assert 202 in responses.keys()
assert responses[202].description == "202 Response"
assert responses[202].content["application/json"].schema_.ref == "#/components/schemas/Order"
def test_openapi_union_partial_response():
app = APIGatewayRestResolver(enable_validation=True)
class User(BaseModel):
pass
class Order(BaseModel):
pass
@app.get(
"/",
responses={
200: {"description": "200 Response"},
202: {"description": "202 Response", "content": {"application/json": {"model": Order}}},
},
)
def handler() -> Response[Union[User, Order]]:
if randbelow(2) > 0:
return Response(status_code=200, body=User())
else:
return Response(status_code=202, body=Order())
schema = app.get_openapi_schema()
responses = schema.paths["/"].get.responses
assert 200 in responses.keys()
assert responses[200].description == "200 Response"
assert responses[200].content["application/json"].schema_.anyOf is not None
assert 202 in responses.keys()
assert responses[202].description == "202 Response"
assert responses[202].content["application/json"].schema_.ref == "#/components/schemas/Order"