Added unit tests for pagelet and static resource serving.

This commit is contained in:
s3lph 2018-06-19 21:34:48 +02:00
parent 2f12403e1f
commit 0bfd4efab9
3 changed files with 131 additions and 6 deletions

View file

@ -227,8 +227,13 @@ class HttpHandler(BaseHTTPRequestHandler):
# Make sure the file is actually inside the webroot directory and that it exists # Make sure the file is actually inside the webroot directory and that it exists
if os.path.commonpath([filepath, self.server.webroot]) == self.server.webroot and os.path.exists(filepath): if os.path.commonpath([filepath, self.server.webroot]) == self.server.webroot and os.path.exists(filepath):
# Open and read the file # Open and read the file
with open(filepath, 'rb') as f: try:
data = f.read() with open(filepath, 'rb') as f:
data = f.read()
except PermissionError:
self.send_error(403)
self.end_headers()
return
# File read successfully, send 'OK' header # File read successfully, send 'OK' header
self.send_response(200) self.send_response(200)
# TODO: Guess the MIME type. Unfortunately this call solely relies on the file extension, not ideal? # TODO: Guess the MIME type. Unfortunately this call solely relies on the file extension, not ideal?
@ -236,8 +241,9 @@ class HttpHandler(BaseHTTPRequestHandler):
# Fall back to octet-stream type, if unknown # Fall back to octet-stream type, if unknown
if mimetype is None: if mimetype is None:
mimetype = 'application/octet-stream' mimetype = 'application/octet-stream'
# Send content type header # Send content type and length header
self.send_header('Content-Type', mimetype) self.send_header('Content-Type', mimetype)
self.send_header('Content-Length', len(data))
self.end_headers() self.end_headers()
# Send the requested resource as response body # Send the requested resource as response body
self.wfile.write(data) self.wfile.write(data)

View file

@ -3,6 +3,7 @@ from typing import Any, Callable, Dict, Tuple, Union
import unittest.mock import unittest.mock
from io import BytesIO from io import BytesIO
from tempfile import TemporaryDirectory
from abc import ABC from abc import ABC
from datetime import datetime from datetime import datetime
@ -45,6 +46,8 @@ class HttpResponse:
""" """
Parse a new fragment of data. This function does nothing if the parsed HTTP response is already complete. Parse a new fragment of data. This function does nothing if the parsed HTTP response is already complete.
DO NOT USE THIS OUTSIDE UNIT TESTING!
:param fragment: The data fragment to parse. :param fragment: The data fragment to parse.
""" """
# response packet complete, nothing to do # response packet complete, nothing to do
@ -178,10 +181,12 @@ class AbstractHttpdTest(ABC, unittest.TestCase):
self.client_sock.set_request(b'GET /just/testing/sessions HTTP/1.1\r\n\r\n') self.client_sock.set_request(b'GET /just/testing/sessions HTTP/1.1\r\n\r\n')
handler = HttpHandler(self.client_sock, ('::1', 45678), self.server) handler = HttpHandler(self.client_sock, ('::1', 45678), self.server)
packet = self.client_sock.get_response() packet = self.client_sock.get_response()
TODO(s3lph): This could probably go here instead.
""" """
def setUp(self) -> None: def setUp(self) -> None:
self.server: HTTPServer = MockServer() self.tempdir: TemporaryDirectory = TemporaryDirectory(prefix='matemat.', dir='/tmp/')
self.server: HTTPServer = MockServer(webroot=self.tempdir.name)
self.client_sock: MockSocket = MockSocket() self.client_sock: MockSocket = MockSocket()
def tearDown(self):
self.tempdir.cleanup()

View file

@ -0,0 +1,114 @@
from typing import Any, Dict
import os
import os.path
from matemat.webserver.httpd import HttpHandler
from matemat.webserver.test.abstract_httpd_test import AbstractHttpdTest, test_pagelet
@test_pagelet('/just/testing/serve_pagelet_ok')
def serve_test_pagelet_ok(method: str,
path: str,
args: Dict[str, str],
session_vars: Dict[str, Any],
headers: Dict[str, str]):
headers['Content-Type'] = 'text/plain'
return 200, 'serve test pagelet ok'
@test_pagelet('/just/testing/serve_pagelet_fail')
def serve_test_pagelet_fail(method: str,
path: str,
args: Dict[str, str],
session_vars: Dict[str, Any],
headers: Dict[str, str]):
session_vars['test'] = 'hello, world!'
headers['Content-Type'] = 'text/plain'
return 500, 'serve test pagelet fail'
class TestServe(AbstractHttpdTest):
"""
Test cases for the content serving of the web server.
"""
def setUp(self):
super().setUp()
# Create a static resource in the temp dir
with open(os.path.join(self.tempdir.name, 'static_resource.txt'), 'w') as f:
f.write('static resource test')
# Create a second static resource chmodded to 0000, to test 403 Forbidden error
forbidden: str = os.path.join(self.tempdir.name, 'forbidden_static_resource.txt')
with open(forbidden, 'w') as f:
f.write('This should not be readable')
os.chmod(forbidden, 0)
def test_serve_pagelet_ok(self):
# Call the test pagelet that produces a 200 OK result
self.client_sock.set_request(b'GET /just/testing/serve_pagelet_ok 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_ok', packet.pagelet)
# Make sure the expected content is served
self.assertEqual(200, packet.statuscode)
self.assertEqual('serve test pagelet ok', packet.body)
def test_serve_pagelet_fail(self):
# Call the test pagelet that produces a 500 Internal Server Error result
self.client_sock.set_request(b'GET /just/testing/serve_pagelet_fail 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_fail', packet.pagelet)
# Make sure the expected content is served
self.assertEqual(500, packet.statuscode)
self.assertEqual('serve test pagelet fail', 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')
HttpHandler(self.client_sock, ('::1', 45678), self.server)
packet = self.client_sock.get_response()
# Make sure that no pagelet was called
self.assertIsNone(packet.pagelet)
# Make sure the expected content is served
self.assertEqual(200, packet.statuscode)
self.assertEqual('static resource test', packet.body)
def test_serve_static_forbidden(self):
# Request a static resource with lacking permissions
self.client_sock.set_request(b'GET /forbidden_static_resource.txt HTTP/1.1\r\n\r\n')
HttpHandler(self.client_sock, ('::1', 45678), self.server)
packet = self.client_sock.get_response()
# Make sure that no pagelet was called
self.assertIsNone(packet.pagelet)
# Make sure a 403 header is served
self.assertEqual(403, packet.statuscode)
def test_serve_not_found(self):
# Request a nonexistent resource
self.client_sock.set_request(b'GET /nonexistent HTTP/1.1\r\n\r\n')
HttpHandler(self.client_sock, ('::1', 45678), self.server)
packet = self.client_sock.get_response()
# Make sure that no pagelet was called
self.assertIsNone(packet.pagelet)
# Make sure a 404 header is served
self.assertEqual(404, packet.statuscode)
def test_serve_directory_traversal(self):
# Request a resource outside the webroot
self.client_sock.set_request(b'GET /../../../../../etc/passwd HTTP/1.1\r\n\r\n')
HttpHandler(self.client_sock, ('::1', 45678), self.server)
packet = self.client_sock.get_response()
# Make sure that no pagelet was called
self.assertIsNone(packet.pagelet)
# Make sure a 404 header is served
self.assertEqual(404, packet.statuscode)