/
test_exceptions_handler.py
245 lines (178 loc) · 7.54 KB
/
test_exceptions_handler.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
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
import asyncio
import logging
import pytest
from unittest.mock import Mock
from bs4 import BeautifulSoup
from sanic import Sanic, handlers
from sanic.exceptions import Forbidden, InvalidUsage, NotFound, ServerError
from sanic.handlers import ErrorHandler
from sanic.response import stream, text
async def sample_streaming_fn(response):
await response.write("foo,")
await asyncio.sleep(0.001)
await response.write("bar")
class ErrorWithRequestCtx(ServerError):
pass
@pytest.fixture
def exception_handler_app():
exception_handler_app = Sanic("test_exception_handler")
@exception_handler_app.route("/1", error_format="html")
def handler_1(request):
raise InvalidUsage("OK")
@exception_handler_app.route("/2", error_format="html")
def handler_2(request):
raise ServerError("OK")
@exception_handler_app.route("/3", error_format="html")
def handler_3(request):
raise NotFound("OK")
@exception_handler_app.route("/4", error_format="html")
def handler_4(request):
foo = bar # noqa -- F821
return text(foo)
@exception_handler_app.route("/5", error_format="html")
def handler_5(request):
class CustomServerError(ServerError):
pass
raise CustomServerError("Custom server error")
@exception_handler_app.route("/6/<arg:int>", error_format="html")
def handler_6(request, arg):
try:
foo = 1 / arg
except Exception as e:
raise e from ValueError(f"{arg}")
return text(foo)
@exception_handler_app.route("/7", error_format="html")
def handler_7(request):
raise Forbidden("go away!")
@exception_handler_app.route("/8", error_format="html")
def handler_8(request):
raise ErrorWithRequestCtx("OK")
@exception_handler_app.exception(ErrorWithRequestCtx, NotFound)
def handler_exception_with_ctx(request, exception):
return text(request.ctx.middleware_ran)
@exception_handler_app.exception(ServerError)
def handler_exception(request, exception):
return text("OK")
@exception_handler_app.exception(Forbidden)
async def async_handler_exception(request, exception):
return stream(
sample_streaming_fn,
content_type="text/csv",
)
@exception_handler_app.middleware
async def some_request_middleware(request):
request.ctx.middleware_ran = "Done."
return exception_handler_app
def test_invalid_usage_exception_handler(exception_handler_app):
request, response = exception_handler_app.test_client.get("/1")
assert response.status == 400
def test_server_error_exception_handler(exception_handler_app):
request, response = exception_handler_app.test_client.get("/2")
assert response.status == 200
assert response.text == "OK"
def test_not_found_exception_handler(exception_handler_app):
request, response = exception_handler_app.test_client.get("/3")
assert response.status == 200
def test_text_exception__handler(exception_handler_app):
request, response = exception_handler_app.test_client.get("/random")
assert response.status == 200
assert response.text == "Done."
def test_async_exception_handler(exception_handler_app):
request, response = exception_handler_app.test_client.get("/7")
assert response.status == 200
assert response.text == "foo,bar"
def test_html_traceback_output_in_debug_mode(exception_handler_app):
request, response = exception_handler_app.test_client.get("/4", debug=True)
assert response.status == 500
soup = BeautifulSoup(response.body, "html.parser")
html = str(soup)
assert "handler_4" in html
assert "foo = bar" in html
summary_text = " ".join(soup.select(".summary")[0].text.split())
assert (
"NameError: name 'bar' is not defined while handling path /4"
) == summary_text
def test_inherited_exception_handler(exception_handler_app):
request, response = exception_handler_app.test_client.get("/5")
assert response.status == 200
def test_chained_exception_handler(exception_handler_app):
request, response = exception_handler_app.test_client.get(
"/6/0", debug=True
)
assert response.status == 500
soup = BeautifulSoup(response.body, "html.parser")
html = str(soup)
assert "handler_6" in html
assert "foo = 1 / arg" in html
assert "ValueError" in html
summary_text = " ".join(soup.select(".summary")[0].text.split())
assert (
"ZeroDivisionError: division by zero while handling path /6/0"
) == summary_text
def test_exception_handler_lookup(exception_handler_app):
class CustomError(Exception):
pass
class CustomServerError(ServerError):
pass
def custom_error_handler():
pass
def server_error_handler():
pass
def import_error_handler():
pass
try:
ModuleNotFoundError
except Exception:
class ModuleNotFoundError(ImportError):
pass
handler = ErrorHandler("auto")
handler.add(ImportError, import_error_handler)
handler.add(CustomError, custom_error_handler)
handler.add(ServerError, server_error_handler)
assert handler.lookup(ImportError(), None) == import_error_handler
assert handler.lookup(ModuleNotFoundError(), None) == import_error_handler
assert handler.lookup(CustomError(), None) == custom_error_handler
assert handler.lookup(ServerError("Error"), None) == server_error_handler
assert (
handler.lookup(CustomServerError("Error"), None)
== server_error_handler
)
# once again to ensure there is no caching bug
assert handler.lookup(ImportError(), None) == import_error_handler
assert handler.lookup(ModuleNotFoundError(), None) == import_error_handler
assert handler.lookup(CustomError(), None) == custom_error_handler
assert handler.lookup(ServerError("Error"), None) == server_error_handler
assert (
handler.lookup(CustomServerError("Error"), None)
== server_error_handler
)
def test_exception_handler_processed_request_middleware(exception_handler_app):
request, response = exception_handler_app.test_client.get("/8")
assert response.status == 200
assert response.text == "Done."
def test_single_arg_exception_handler_notice(exception_handler_app, caplog):
class CustomErrorHandler(ErrorHandler):
def lookup(self, exception):
return super().lookup(exception, None)
exception_handler_app.error_handler = CustomErrorHandler()
with caplog.at_level(logging.WARNING):
_, response = exception_handler_app.test_client.get("/1")
assert caplog.records[0].message == (
"You are using a deprecated error handler. The lookup method should "
"accept two positional parameters: (exception, route_name: "
"Optional[str]). Until you upgrade your ErrorHandler.lookup, "
"Blueprint specific exceptions will not work properly. Beginning in "
"v22.3, the legacy style lookup method will not work at all."
)
assert response.status == 400
def test_error_handler_noisy_log(exception_handler_app, monkeypatch):
err_logger = Mock()
monkeypatch.setattr(handlers, "error_logger", err_logger)
exception_handler_app.config["NOISY_EXCEPTIONS"] = False
exception_handler_app.test_client.get("/1")
err_logger.exception.assert_not_called()
exception_handler_app.config["NOISY_EXCEPTIONS"] = True
request, _ = exception_handler_app.test_client.get("/1")
err_logger.exception.assert_called_with(
"Exception occurred while handling uri: %s", repr(request.url)
)