TonBERRY/microdot
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Improved handling of 400 and 405 errors

Browse Source
This commit is contained in:
Miguel Grinberg
2022-07-31 11:28:44 +01:00
parent cd5b35d86f
commit 8177b9c7f1
4 changed files with 157 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

26
src/microdot.py
View File

@@ -527,7 +527,7 @@ class Response():
automatically from the file extension.
Security note: The filename is assumed to be trusted. Never pass
filenames provided by the user before validating and sanitizing them
filenames provided by the user without validating and sanitizing them
first.
"""
if content_type is None:
@@ -885,13 +885,15 @@ class Microdot():
self.shutdown_requested = True
def find_route(self, req):
f = None
f = 404
for route_methods, route_pattern, route_handler in self.url_map:
if req.method in route_methods:
req.url_args = route_pattern.match(req.path)
if req.url_args is not None:
req.url_args = route_pattern.match(req.path)
if req.url_args is not None:
if req.method in route_methods:
f = route_handler
break
else:
f = 405
return f
def handle_request(self, sock, addr):
@@ -934,7 +936,7 @@ class Microdot():
f = self.find_route(req)
try:
res = None
if f:
if callable(f):
for handler in self.before_request_handlers:
res = handler(req)
if res:
@@ -949,10 +951,10 @@ class Microdot():
res = handler(req, res) or res
for handler in req.after_request_handlers:
res = handler(req, res) or res
elif 404 in self.error_handlers:
res = self.error_handlers[404](req)
elif f in self.error_handlers:
res = self.error_handlers[f](req)
else:
res = 'Not found', 404
res = 'Not found', f
except Exception as exc:
print_exception(exc)
res = None
@@ -967,7 +969,11 @@ class Microdot():
else:
res = 'Internal server error', 500
else:
res = 'Bad request', 400
if 400 in self.error_handlers:
res = self.error_handlers[400](req)
else:
res = 'Bad request', 400
if isinstance(res, tuple):
res = Response(*res)
elif not isinstance(res, Response):

15
src/microdot_asyncio.py
View File

@@ -58,7 +58,7 @@ class Request(BaseRequest):
"""
# request line
line = (await Request._safe_readline(client_stream)).strip().decode()
if not line: # pragma: no cover
if not line:
return None
method, url, http_version = line.split()
http_version = http_version.split('/', 1)[1]
@@ -331,7 +331,7 @@ class Microdot(BaseMicrodot):
f = self.find_route(req)
try:
res = None
if f:
if callable(f):
for handler in self.before_request_handlers:
res = await self._invoke_handler(handler, req)
if res:
@@ -346,11 +346,11 @@ class Microdot(BaseMicrodot):
for handler in self.after_request_handlers:
res = await self._invoke_handler(
handler, req, res) or res
elif 404 in self.error_handlers:
elif f in self.error_handlers:
res = await self._invoke_handler(
self.error_handlers[404], req)
self.error_handlers[f], req)
else:
res = 'Not found', 404
res = 'Not found', f
except Exception as exc:
print_exception(exc)
res = None
@@ -367,7 +367,10 @@ class Microdot(BaseMicrodot):
else:
res = 'Internal server error', 500
else:
res = 'Bad request', 400
if 400 in self.error_handlers:
res = await self._invoke_handler(self.error_handlers[400], req)
else:
res = 'Bad request', 400
if isinstance(res, tuple):
res = Response(*res)
elif not isinstance(res, Response):

66
tests/microdot/test_microdot.py
View File

@@ -164,6 +164,36 @@ class TestMicrodot(unittest.TestCase):
self.assertIn(b'Content-Type: text/plain\r\n', fd.response)
self.assertTrue(fd.response.endswith(b'\r\n\r\nbaz'))
def test_400(self):
app = Microdot()
mock_socket.clear_requests()
fd = mock_socket.FakeStream(b'\n')
mock_socket._requests.append(fd)
self._add_shutdown(app)
app.run()
self.assertTrue(fd.response.startswith(b'HTTP/1.0 400 N/A\r\n'))
self.assertIn(b'Content-Length: 11\r\n', fd.response)
self.assertIn(b'Content-Type: text/plain\r\n', fd.response)
self.assertTrue(fd.response.endswith(b'\r\n\r\nBad request'))
def test_400_handler(self):
app = Microdot()
@app.errorhandler(400)
def handle_400(req):
return '400'
mock_socket.clear_requests()
fd = mock_socket.FakeStream(b'\n')
mock_socket._requests.append(fd)
self._add_shutdown(app)
app.run()
self.assertTrue(fd.response.startswith(b'HTTP/1.0 200 OK\r\n'))
self.assertIn(b'Content-Length: 3\r\n', fd.response)
self.assertIn(b'Content-Type: text/plain\r\n', fd.response)
self.assertTrue(fd.response.endswith(b'\r\n\r\n400'))
def test_404(self):
app = Microdot()
@@ -200,6 +230,42 @@ class TestMicrodot(unittest.TestCase):
self.assertIn(b'Content-Type: text/plain\r\n', fd.response)
self.assertTrue(fd.response.endswith(b'\r\n\r\n404'))
def test_405(self):
app = Microdot()
@app.route('/foo')
def index(req):
return 'foo'
mock_socket.clear_requests()
fd = mock_socket.add_request('POST', '/foo')
self._add_shutdown(app)
app.run()
self.assertTrue(fd.response.startswith(b'HTTP/1.0 405 N/A\r\n'))
self.assertIn(b'Content-Length: 9\r\n', fd.response)
self.assertIn(b'Content-Type: text/plain\r\n', fd.response)
self.assertTrue(fd.response.endswith(b'\r\n\r\nNot found'))
def test_405_handler(self):
app = Microdot()
@app.route('/foo')
def index(req):
return 'foo'
@app.errorhandler(405)
def handle_404(req):
return '405'
mock_socket.clear_requests()
fd = mock_socket.add_request('POST', '/foo')
self._add_shutdown(app)
app.run()
self.assertTrue(fd.response.startswith(b'HTTP/1.0 200 OK\r\n'))
self.assertIn(b'Content-Length: 3\r\n', fd.response)
self.assertIn(b'Content-Type: text/plain\r\n', fd.response)
self.assertTrue(fd.response.endswith(b'\r\n\r\n405'))
def test_413(self):
app = Microdot()

66
tests/microdot_asyncio/test_microdot_asyncio.py
View File

@@ -137,6 +137,36 @@ class TestMicrodotAsync(unittest.TestCase):
self.assertIn(b'Content-Type: text/plain\r\n', fd.response)
self.assertTrue(fd.response.endswith(b'\r\n\r\nbaz'))
def test_400(self):
app = Microdot()
mock_socket.clear_requests()
fd = mock_socket.FakeStream(b'\n')
mock_socket._requests.append(fd)
self._add_shutdown(app)
app.run()
self.assertTrue(fd.response.startswith(b'HTTP/1.0 400 N/A\r\n'))
self.assertIn(b'Content-Length: 11\r\n', fd.response)
self.assertIn(b'Content-Type: text/plain\r\n', fd.response)
self.assertTrue(fd.response.endswith(b'\r\n\r\nBad request'))
def test_400_handler(self):
app = Microdot()
@app.errorhandler(400)
async def handle_404(req):
return '400'
mock_socket.clear_requests()
fd = mock_socket.FakeStream(b'\n')
mock_socket._requests.append(fd)
self._add_shutdown(app)
app.run()
self.assertTrue(fd.response.startswith(b'HTTP/1.0 200 OK\r\n'))
self.assertIn(b'Content-Length: 3\r\n', fd.response)
self.assertIn(b'Content-Type: text/plain\r\n', fd.response)
self.assertTrue(fd.response.endswith(b'\r\n\r\n400'))
def test_404(self):
app = Microdot()
@@ -173,6 +203,42 @@ class TestMicrodotAsync(unittest.TestCase):
self.assertIn(b'Content-Type: text/plain\r\n', fd.response)
self.assertTrue(fd.response.endswith(b'\r\n\r\n404'))
def test_405(self):
app = Microdot()
@app.route('/foo')
def index(req):
return 'foo'
mock_socket.clear_requests()
fd = mock_socket.add_request('POST', '/foo')
self._add_shutdown(app)
app.run()
self.assertTrue(fd.response.startswith(b'HTTP/1.0 405 N/A\r\n'))
self.assertIn(b'Content-Length: 9\r\n', fd.response)
self.assertIn(b'Content-Type: text/plain\r\n', fd.response)
self.assertTrue(fd.response.endswith(b'\r\n\r\nNot found'))
def test_405_handler(self):
app = Microdot()
@app.route('/foo')
def index(req):
return 'foo'
@app.errorhandler(405)
async def handle_405(req):
return '405'
mock_socket.clear_requests()
fd = mock_socket.add_request('POST', '/foo')
self._add_shutdown(app)
app.run()
self.assertTrue(fd.response.startswith(b'HTTP/1.0 200 OK\r\n'))
self.assertIn(b'Content-Length: 3\r\n', fd.response)
self.assertIn(b'Content-Type: text/plain\r\n', fd.response)
self.assertTrue(fd.response.endswith(b'\r\n\r\n405'))
def test_413(self):
app = Microdot()