weva/matemat
1
0
Fork 0
forked from s3lph/matemat
Code Pull requests Activity

Increased coverage in httpd, also some code deduplication.

Browse source
This commit is contained in:
s3lph 2018-08-28 21:07:32 +02:00
parent 4c2c454f12
commit e71817cf8e
3 changed files with 171 additions and 25 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

46
matemat/webserver/httpd.py
View file

@ -455,15 +455,9 @@ class HttpHandler(BaseHTTPRequestHandler):
self.send_response(404)
self.end_headers()
# noinspection PyPep8Naming
def do_GET(self) -> None:
"""
Called by BasicHTTPRequestHandler for GET requests.
"""
def _handle_request(self, method: str, path: str, args: RequestArguments):
try:
# Parse the request and hand it to the handle function
path, args = parse_args(self.path)
self._handle('GET', path, args)
self._handle(method, path, args)
# Special handling for some errors
except HttpException as e:
self.send_error(e.status, e.title, e.message)
@ -483,6 +477,19 @@ class HttpHandler(BaseHTTPRequestHandler):
self.send_error(500, 'Internal Server Error')
self.server.logger.exception('', e.args, e)
# noinspection PyPep8Naming
def do_GET(self) -> None:
"""
Called by BasicHTTPRequestHandler for GET requests.
"""
# Parse the request and hand it to the handle function
try:
path, args = parse_args(self.path)
self._handle_request('GET', path, args)
except ValueError as e:
self.send_error(400, 'Bad Request')
self.server.logger.debug('', exc_info=e)
# noinspection PyPep8Naming
def do_POST(self) -> None:
"""
@ -492,29 +499,18 @@ class HttpHandler(BaseHTTPRequestHandler):
# Read the POST body, if it exists, and its MIME type is application/x-www-form-urlencoded
clen: int = int(str(self.headers.get('Content-Length', failobj='0')))
if clen > _MAX_POST:
raise ValueError('Request too big')
# Return a 413 error page if the request size exceeds boundaries
self.send_error(413, 'Payload Too Large')
self.server.logger.debug('', exc_info=HttpException(413, 'Payload Too Large'))
return
ctype: str = self.headers.get('Content-Type', failobj='application/octet-stream')
post: bytes = self.rfile.read(clen)
path, args = parse_args(self.path, postbody=post, enctype=ctype)
# Parse the request and hand it to the handle function
self._handle('POST', path, args)
# Special handling for some errors
except HttpException as e:
if 500 <= e.status < 600:
self.send_error(500, 'Internal Server Error')
self.server.logger.exception('', exc_info=e)
else:
self.server.logger.debug('', exc_info=e)
except PermissionError as e:
self.send_error(403, 'Forbidden')
self.server.logger.debug('', exc_info=e)
path, args = parse_args(self.path, postbody=post, enctype=ctype)
self._handle_request('POST', path, args)
except ValueError as e:
self.send_error(400, 'Bad Request')
self.server.logger.debug('', exc_info=e)
except BaseException as e:
# Generic error handling
self.send_error(500, 'Internal Server Error')
self.server.logger.exception('', e.args, e)
@property
def session_vars(self) -> Dict[str, Any]:

108
matemat/webserver/test/test_httpd.py Normal file
View file

@ -0,0 +1,108 @@
from typing import Any, Dict, Union
from matemat.exceptions import HttpException
from matemat.webserver import HttpHandler, RequestArguments, PageletResponse
from matemat.webserver.test.abstract_httpd_test import AbstractHttpdTest, test_pagelet
@test_pagelet('/just/testing/http_exception')
def test_pagelet_http_exception(method: str,
path: str,
args: RequestArguments,
session_vars: Dict[str, Any],
headers: Dict[str, str],
pagelet_variables: Dict[str, str]) -> Union[bytes, str, PageletResponse]:
raise HttpException(int(str(args.exc)), 'Test Exception')
@test_pagelet('/just/testing/value_error')
def test_pagelet_value_error(method: str,
path: str,
args: RequestArguments,
session_vars: Dict[str, Any],
headers: Dict[str, str],
pagelet_variables: Dict[str, str]) -> Union[bytes, str, PageletResponse]:
raise ValueError('test')
@test_pagelet('/just/testing/permission_error')
def test_pagelet_permission_error(method: str,
path: str,
args: RequestArguments,
session_vars: Dict[str, Any],
headers: Dict[str, str],
pagelet_variables: Dict[str, str]) -> Union[bytes, str, PageletResponse]:
raise PermissionError('test')
@test_pagelet('/just/testing/other_error')
def test_pagelet_other_error(method: str,
path: str,
args: RequestArguments,
session_vars: Dict[str, Any],
headers: Dict[str, str],
pagelet_variables: Dict[str, str]) -> Union[bytes, str, PageletResponse]:
raise TypeError('test')
class TestHttpd(AbstractHttpdTest):
def test_httpd_get_illegal_path(self):
self.client_sock.set_request(b'GET /foo?bar?baz HTTP/1.1\r\n\r\n')
HttpHandler(self.client_sock, ('::1', 45678), self.server)
packet = self.client_sock.get_response()
self.assertEqual(400, packet.statuscode)
def test_httpd_post_illegal_path(self):
self.client_sock.set_request(b'POST /foo?bar?baz HTTP/1.1\r\n'
b'Content-Length: 0\r\n'
b'Content-Type: application/x-www-form-urlencoded\r\n\r\n')
HttpHandler(self.client_sock, ('::1', 45678), self.server)
packet = self.client_sock.get_response()
self.assertEqual(400, packet.statuscode)
def test_httpd_post_illegal_header(self):
self.client_sock.set_request(b'POST /foo?bar=baz HTTP/1.1\r\n'
b'Content-Length: 0\r\n'
b'Content-Type: application/octet-stream\r\n\r\n')
HttpHandler(self.client_sock, ('::1', 45678), self.server)
packet = self.client_sock.get_response()
self.assertEqual(400, packet.statuscode)
def test_httpd_post_request_too_big(self):
self.client_sock.set_request(b'POST /foo?bar=baz HTTP/1.1\r\n'
b'Content-Length: 1000001\r\n'
b'Content-Type: application/octet-stream\r\n\r\n')
HttpHandler(self.client_sock, ('::1', 45678), self.server)
packet = self.client_sock.get_response()
self.assertEqual(413, packet.statuscode)
def test_httpd_exception_http_400(self):
self.client_sock.set_request(b'GET /just/testing/http_exception?exc=400 HTTP/1.1\r\n\r\n')
HttpHandler(self.client_sock, ('::1', 45678), self.server)
packet = self.client_sock.get_response()
self.assertEqual(400, packet.statuscode)
def test_httpd_exception_http_500(self):
self.client_sock.set_request(b'GET /just/testing/http_exception?exc=500 HTTP/1.1\r\n\r\n')
HttpHandler(self.client_sock, ('::1', 45678), self.server)
packet = self.client_sock.get_response()
self.assertEqual(500, packet.statuscode)
def test_httpd_exception_value_error(self):
self.client_sock.set_request(b'GET /just/testing/value_error HTTP/1.1\r\n\r\n')
HttpHandler(self.client_sock, ('::1', 45678), self.server)
packet = self.client_sock.get_response()
self.assertEqual(400, packet.statuscode)
def test_httpd_exception_permission_error(self):
self.client_sock.set_request(b'GET /just/testing/permission_error HTTP/1.1\r\n\r\n')
HttpHandler(self.client_sock, ('::1', 45678), self.server)
packet = self.client_sock.get_response()
self.assertEqual(403, packet.statuscode)
def test_httpd_exception_other_error(self):
self.client_sock.set_request(b'GET /just/testing/other_error HTTP/1.1\r\n\r\n')
HttpHandler(self.client_sock, ('::1', 45678), self.server)
packet = self.client_sock.get_response()
self.assertEqual(500, packet.statuscode)

42
matemat/webserver/test/test_serve.py
View file

@ -63,6 +63,28 @@ def serve_test_pagelet_fail(method: str,
raise HttpException(599, 'Error expected during unit testing')
# noinspection PyTypeChecker
@test_pagelet('/just/testing/serve_pagelet_empty')
def serve_test_pagelet_empty(method: str,
path: str,
args: RequestArguments,
session_vars: Dict[str, Any],
headers: Dict[str, str],
pagelet_variables: Dict[str, str]) -> Union[bytes, str, PageletResponse]:
return PageletResponse()
# noinspection PyTypeChecker
@test_pagelet('/just/testing/serve_pagelet_type_error')
def serve_test_pagelet_fail(method: str,
path: str,
args: RequestArguments,
session_vars: Dict[str, Any],
headers: Dict[str, str],
pagelet_variables: Dict[str, str]) -> Union[bytes, str, PageletResponse]:
return 42
class TestServe(AbstractHttpdTest):
"""
Test cases for the content serving of the web server.
@ -147,6 +169,18 @@ class TestServe(AbstractHttpdTest):
# Make sure the response body was rendered correctly by the templating engine
self.assertEqual(b'Hello, World!', packet.body)
def test_serve_pagelet_empty(self):
# Call the test pagelet that redirects to another path
self.client_sock.set_request(b'GET /just/testing/serve_pagelet_empty HTTP/1.1\r\n\r\n')
HttpHandler(self.client_sock, ('::1', 45678), self.server)
packet = self.client_sock.get_response()
# Make sure the correct pagelet was called
self.assertEqual('serve_test_pagelet_empty', packet.pagelet)
self.assertEqual(200, packet.statuscode)
# Make sure the response body was rendered correctly by the templating engine
self.assertEqual(b'', packet.body)
def test_serve_static_ok(self):
# Request a static resource
self.client_sock.set_request(b'GET /static_resource.txt HTTP/1.1\r\n\r\n')
@ -249,6 +283,14 @@ class TestServe(AbstractHttpdTest):
packet = self.client_sock.get_response()
self.assertEqual('application/x-foo-bar', packet.headers['Content-Type'])
def test_serve_pagelet_type_error(self):
# A 500 error should be returned if a pagelet returns an invalid type
self.client_sock.set_request(b'GET /just/testing/serve_pagelet_type_error HTTP/1.1\r\n\r\n')
HttpHandler(self.client_sock, ('::1', 45678), self.server)
packet = self.client_sock.get_response()
# Make sure a 500 header is served
self.assertEqual(500, packet.statuscode)
def test_serve_static_mime_extension(self):
# The correct Content-Type should be guessed by file extension primarily
self.client_sock.set_request(b'GET /teststyle.css HTTP/1.1\r\n\r\n')