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

Respond to HEAD and OPTIONS requests

Browse Source
This commit is contained in:
Miguel Grinberg
2023-03-21 23:54:19 +00:00
parent eaf2ef62d1
commit 6a31f89673
5 changed files with 134 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

52
src/microdot.py
View File

@@ -548,6 +548,7 @@ class Response():
else:
# this applies to bytes, file-like objects or generators
self.body = body
self.is_head = False
def set_cookie(self, cookie, value, path=None, domain=None, expires=None,
max_age=None, secure=False, http_only=False):
@@ -612,19 +613,20 @@ class Response():
stream.write(b'\r\n')
# body
can_flush = hasattr(stream, 'flush')
try:
for body in self.body_iter():
if isinstance(body, str): # pragma: no cover
body = body.encode()
stream.write(body)
if can_flush: # pragma: no cover
stream.flush()
except OSError as exc: # pragma: no cover
if exc.errno in MUTED_SOCKET_ERRORS:
pass
else:
raise
if not self.is_head:
can_flush = hasattr(stream, 'flush')
try:
for body in self.body_iter():
if isinstance(body, str): # pragma: no cover
body = body.encode()
stream.write(body)
if can_flush: # pragma: no cover
stream.flush()
except OSError as exc: # pragma: no cover
if exc.errno in MUTED_SOCKET_ERRORS:
pass
else:
raise
def body_iter(self):
if self.body:
@@ -793,6 +795,7 @@ class Microdot():
self.after_error_request_handlers = []
self.error_handlers = {}
self.shutdown_requested = False
self.options_handler = self.default_options_handler
self.debug = False
self.server = None
@@ -828,7 +831,8 @@ class Microdot():
"""
def decorated(f):
self.url_map.append(
(methods or ['GET'], URLPattern(url_pattern), f))
([m.upper() for m in (methods or ['GET'])],
URLPattern(url_pattern), f))
return f
return decorated
@@ -1114,17 +1118,32 @@ class Microdot():
self.shutdown_requested = True
def find_route(self, req):
method = req.method.upper()
if method == 'OPTIONS' and self.options_handler:
return self.options_handler(req)
if method == 'HEAD':
method = 'GET'
f = 404
for route_methods, route_pattern, route_handler in self.url_map:
req.url_args = route_pattern.match(req.path)
if req.url_args is not None:
if req.method in route_methods:
if method in route_methods:
f = route_handler
break
else:
f = 405
return f
def default_options_handler(self, req):
allow = []
for route_methods, route_pattern, route_handler in self.url_map:
if route_pattern.match(req.path) is not None:
allow.extend(route_methods)
if 'GET' in allow:
allow.append('HEAD')
allow.append('OPTIONS')
return {'Allow': ', '.join(allow)}
def handle_request(self, sock, addr):
if Request.socket_read_timeout and \
hasattr(sock, 'settimeout'): # pragma: no cover
@@ -1199,6 +1218,8 @@ class Microdot():
for handler in req.after_request_handlers:
res = handler(req, res) or res
after_request_handled = True
elif isinstance(f, dict):
res = Response(headers=f)
elif f in self.error_handlers:
res = self.error_handlers[f](req)
else:
@@ -1242,6 +1263,7 @@ class Microdot():
if not after_request_handled:
for handler in self.after_error_request_handlers:
res = handler(req, res) or res
res.is_head = (req and req.method == 'HEAD')
return res

12
src/microdot_asyncio.py
View File

@@ -151,10 +151,11 @@ class Response(BaseResponse):
await stream.awrite(b'\r\n')
# body
async for body in self.body_iter():
if isinstance(body, str): # pragma: no cover
body = body.encode()
await stream.awrite(body)
if not self.is_head:
async for body in self.body_iter():
if isinstance(body, str): # pragma: no cover
body = body.encode()
await stream.awrite(body)
except OSError as exc: # pragma: no cover
if exc.errno in MUTED_SOCKET_ERRORS or \
exc.args[0] == 'Connection lost':
@@ -385,6 +386,8 @@ class Microdot(BaseMicrodot):
res = await self._invoke_handler(
handler, req, res) or res
after_request_handled = True
elif isinstance(f, dict):
res = Response(headers=f)
elif f in self.error_handlers:
res = await self._invoke_handler(
self.error_handlers[f], req)
@@ -431,6 +434,7 @@ class Microdot(BaseMicrodot):
for handler in self.after_error_request_handlers:
res = await self._invoke_handler(
handler, req, res) or res
res.is_head = (req and req.method == 'HEAD')
return res
async def _invoke_handler(self, f_or_coro, *args, **kwargs):

1
src/microdot_test_client.py
View File

@@ -58,6 +58,7 @@ class TestResponse:
test_res._initialize_body(res)
test_res._process_text_body()
test_res._process_json_body()
test_res.is_head = res.is_head
return test_res

46
tests/test_microdot.py
View File

@@ -63,6 +63,52 @@ class TestMicrodot(unittest.TestCase):
self.assertEqual(res.headers['Content-Length'], '3')
self.assertEqual(res.text, 'bar')
def test_head_request(self):
self._mock()
app = Microdot()
@app.route('/foo')
def index(req):
return 'foo'
mock_socket.clear_requests()
fd = mock_socket.add_request('HEAD', '/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; charset=UTF-8\r\n',
fd.response)
self.assertTrue(fd.response.endswith(b'\r\n\r\n'))
self._unmock()
def test_options_request(self):
app = Microdot()
@app.route('/', methods=['GET', 'DELETE'])
def index(req):
return 'foo'
@app.post('/')
def index_post(req):
return 'bar'
@app.route('/foo', methods=['POST', 'PUT'])
def foo(req):
return 'baz'
client = TestClient(app)
res = client.request('OPTIONS', '/')
self.assertEqual(res.status_code, 200)
self.assertEqual(res.headers['Allow'],
'GET, DELETE, POST, HEAD, OPTIONS')
res = client.request('OPTIONS', '/foo')
self.assertEqual(res.status_code, 200)
self.assertEqual(res.headers['Allow'], 'POST, PUT, OPTIONS')
def test_empty_request(self):
self._mock()

42
tests/test_microdot_asyncio.py
View File

@@ -101,6 +101,48 @@ class TestMicrodotAsync(unittest.TestCase):
self.assertEqual(res.body, b'bar-async')
self.assertEqual(res.json, None)
def test_head_request(self):
app = Microdot()
@app.route('/foo')
def index(req):
return 'foo'
mock_socket.clear_requests()
fd = mock_socket.add_request('HEAD', '/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; charset=UTF-8\r\n',
fd.response)
self.assertTrue(fd.response.endswith(b'\r\n\r\n'))
def test_options_request(self):
app = Microdot()
@app.route('/', methods=['GET', 'DELETE'])
async def index(req):
return 'foo'
@app.post('/')
async def index_post(req):
return 'bar'
@app.route('/foo', methods=['POST', 'PUT'])
async def foo(req):
return 'baz'
client = TestClient(app)
res = self._run(client.request('OPTIONS', '/'))
self.assertEqual(res.status_code, 200)
self.assertEqual(res.headers['Allow'],
'GET, DELETE, POST, HEAD, OPTIONS')
res = self._run(client.request('OPTIONS', '/foo'))
self.assertEqual(res.status_code, 200)
self.assertEqual(res.headers['Allow'], 'POST, PUT, OPTIONS')
def test_empty_request(self):
app = Microdot()