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
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
|
import http
import pytest
import falcon
from falcon.http_status import HTTPStatus
import falcon.testing as testing
@pytest.fixture()
def client(asgi, util):
app = util.create_app(asgi)
app.add_route('/status', TestStatusResource())
return testing.TestClient(app)
@pytest.fixture()
def hook_test_client(asgi, util):
app = util.create_app(asgi)
app.add_route('/status', TestHookResource())
return testing.TestClient(app)
def before_hook(req, resp, resource, params):
raise HTTPStatus(falcon.HTTP_200, headers={'X-Failed': 'False'}, text='Pass')
def after_hook(req, resp, resource):
resp.status = falcon.HTTP_200
resp.set_header('X-Failed', 'False')
resp.text = 'Pass'
def noop_after_hook(req, resp, resource):
pass
class TestStatusResource:
@falcon.before(before_hook)
def on_get(self, req, resp):
resp.status = falcon.HTTP_500
resp.set_header('X-Failed', 'True')
resp.text = 'Fail'
def on_post(self, req, resp):
resp.status = falcon.HTTP_500
resp.set_header('X-Failed', 'True')
resp.text = 'Fail'
raise HTTPStatus(falcon.HTTP_200, headers={'X-Failed': 'False'}, text='Pass')
@falcon.after(after_hook)
def on_put(self, req, resp):
# NOTE(kgriffs): Test that passing a unicode status string
# works just fine.
resp.status = '500 Internal Server Error'
resp.set_header('X-Failed', 'True')
resp.text = 'Fail'
def on_patch(self, req, resp):
raise HTTPStatus(falcon.HTTP_200, text=None)
@falcon.after(noop_after_hook)
def on_delete(self, req, resp):
raise HTTPStatus(201, headers={'X-Failed': 'False'}, text='Pass')
class TestHookResource:
def on_get(self, req, resp):
resp.status_code = 500
resp.set_header('X-Failed', 'True')
resp.text = 'Fail'
def on_patch(self, req, resp):
raise HTTPStatus(200, text=None)
class TestHTTPStatus:
def test_raise_status_in_before_hook(self, client):
"""Make sure we get the 200 raised by before hook"""
response = client.simulate_request(path='/status', method='GET')
assert response.status == falcon.HTTP_200
assert response.status_code == 200
assert response.headers['x-failed'] == 'False'
assert response.text == 'Pass'
def test_raise_status_in_responder(self, client):
"""Make sure we get the 200 raised by responder"""
response = client.simulate_request(path='/status', method='POST')
assert response.status == falcon.HTTP_200
assert response.status_code == 200
assert response.headers['x-failed'] == 'False'
assert response.text == 'Pass'
def test_raise_status_runs_after_hooks(self, client):
"""Make sure after hooks still run"""
response = client.simulate_request(path='/status', method='PUT')
assert response.status == falcon.HTTP_200
assert response.status_code == 200
assert response.headers['x-failed'] == 'False'
assert response.text == 'Pass'
def test_raise_status_survives_after_hooks(self, client):
"""Make sure after hook doesn't overwrite our status"""
response = client.simulate_request(path='/status', method='DELETE')
assert response.status == falcon.HTTP_201
assert response.status_code == 201
assert response.headers['x-failed'] == 'False'
assert response.text == 'Pass'
def test_raise_status_empty_body(self, client):
"""Make sure passing None to body results in empty body"""
response = client.simulate_request(path='/status', method='PATCH')
assert response.text == ''
class TestHTTPStatusWithMiddleware:
def test_raise_status_in_process_request(self, hook_test_client):
"""Make sure we can raise status from middleware process request"""
client = hook_test_client
class TestMiddleware:
def process_request(self, req, resp):
raise HTTPStatus(
falcon.HTTP_200, headers={'X-Failed': 'False'}, text='Pass'
)
# NOTE(kgriffs): Test the side-by-side support for dual WSGI and
# ASGI compatibility.
async def process_request_async(self, req, resp):
self.process_request(req, resp)
client.app.add_middleware(TestMiddleware())
response = client.simulate_request(path='/status', method='GET')
assert response.status_code == 200
assert response.headers['x-failed'] == 'False'
assert response.text == 'Pass'
def test_raise_status_in_process_resource(self, hook_test_client):
"""Make sure we can raise status from middleware process resource"""
client = hook_test_client
class TestMiddleware:
def process_resource(self, req, resp, resource, params):
raise HTTPStatus(
falcon.HTTP_200, headers={'X-Failed': 'False'}, text='Pass'
)
async def process_resource_async(self, *args):
self.process_resource(*args)
# NOTE(kgriffs): Pass a list to test that add_middleware can handle it
client.app.add_middleware([TestMiddleware()])
response = client.simulate_request(path='/status', method='GET')
assert response.status == falcon.HTTP_200
assert response.headers['x-failed'] == 'False'
assert response.text == 'Pass'
def test_raise_status_runs_process_response(self, hook_test_client):
"""Make sure process_response still runs"""
client = hook_test_client
class TestMiddleware:
def process_response(self, req, resp, resource, req_succeeded):
resp.status = falcon.HTTP_200
resp.set_header('X-Failed', 'False')
resp.text = 'Pass'
async def process_response_async(self, *args):
self.process_response(*args)
# NOTE(kgriffs): Pass a generic iterable to test that add_middleware
# can handle it.
client.app.add_middleware(iter([TestMiddleware()]))
response = client.simulate_request(path='/status', method='GET')
assert response.status == falcon.HTTP_200
assert response.headers['x-failed'] == 'False'
assert response.text == 'Pass'
class NoBodyResource:
def on_get(self, req, res):
res.data = b'foo'
http_status = HTTPStatus(745)
assert http_status.status_code == 745
raise http_status
def on_post(self, req, res):
res.media = {'a': 1}
http_status = HTTPStatus(falcon.HTTP_725)
assert http_status.status_code == 725
raise http_status
def on_put(self, req, res):
res.text = 'foo'
raise HTTPStatus(falcon.HTTP_719)
@pytest.fixture()
def body_client(asgi, util):
app = util.create_app(asgi)
app.add_route('/status', NoBodyResource())
return testing.TestClient(app)
class TestNoBodyWithStatus:
def test_data_is_set(self, body_client):
res = body_client.simulate_get('/status')
assert res.status == falcon.HTTP_745
assert res.status_code == 745
assert res.content == b''
def test_media_is_set(self, body_client):
res = body_client.simulate_post('/status')
assert res.status == falcon.HTTP_725
assert res.status_code == 725
assert res.content == b''
def test_body_is_set(self, body_client):
res = body_client.simulate_put('/status')
assert res.status == falcon.HTTP_719
assert res.status_code == 719
assert res.content == b''
@pytest.fixture()
def custom_status_client(asgi, util):
def client(status):
class Resource:
def on_get(self, req, resp):
resp.content_type = falcon.MEDIA_TEXT
resp.data = b'Hello, World!'
resp.status = status
app = util.create_app(asgi)
app.add_route('/status', Resource())
return testing.TestClient(app)
return client
@pytest.mark.parametrize(
'status,expected_code',
[
(http.HTTPStatus(200), 200),
(http.HTTPStatus(202), 202),
(http.HTTPStatus(403), 403),
(http.HTTPStatus(500), 500),
(http.HTTPStatus.OK, 200),
(http.HTTPStatus.USE_PROXY, 305),
(http.HTTPStatus.NOT_FOUND, 404),
(http.HTTPStatus.NOT_IMPLEMENTED, 501),
(200, 200),
(307, 307),
(500, 500),
(702, 702),
(b'200 OK', 200),
(b'702 Emacs', 702),
],
)
def test_non_string_status(custom_status_client, status, expected_code):
client = custom_status_client(status)
resp = client.simulate_get('/status')
assert resp.text == 'Hello, World!'
assert resp.status_code == expected_code
def test_deprecated_body():
with pytest.raises(TypeError) as type_error:
sts = HTTPStatus(falcon.HTTP_701, body='foo')
assert 'unexpected keyword argument' in str(type_error.value)
sts = HTTPStatus(falcon.HTTP_701, text='foo')
assert sts.text == 'foo'
|
