forked from s3lph/matemat
Added tests for HTTP POST requests.
This commit is contained in:
parent
78d347f26c
commit
67db4c654a
3 changed files with 93 additions and 16 deletions
|
@ -227,13 +227,8 @@ 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
|
||||||
try:
|
|
||||||
with open(filepath, 'rb') as f:
|
with open(filepath, 'rb') as f:
|
||||||
data = f.read()
|
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?
|
||||||
|
@ -292,22 +287,16 @@ class HttpHandler(BaseHTTPRequestHandler):
|
||||||
path, args = self._parse_args(self.path)
|
path, args = self._parse_args(self.path)
|
||||||
self._handle('GET', path, args)
|
self._handle('GET', path, args)
|
||||||
# Special handling for some errors
|
# Special handling for some errors
|
||||||
except PermissionError as e:
|
except PermissionError:
|
||||||
self.send_response(403, 'Forbidden')
|
self.send_response(403, 'Forbidden')
|
||||||
self.end_headers()
|
self.end_headers()
|
||||||
print(e)
|
except ValueError:
|
||||||
traceback.print_tb(e.__traceback__)
|
|
||||||
except ValueError as e:
|
|
||||||
self.send_response(400, 'Bad Request')
|
self.send_response(400, 'Bad Request')
|
||||||
self.end_headers()
|
self.end_headers()
|
||||||
print(e)
|
except BaseException:
|
||||||
traceback.print_tb(e.__traceback__)
|
|
||||||
except BaseException as e:
|
|
||||||
# Generic error handling
|
# Generic error handling
|
||||||
self.send_response(500, 'Internal Server Error')
|
self.send_response(500, 'Internal Server Error')
|
||||||
self.end_headers()
|
self.end_headers()
|
||||||
print(e)
|
|
||||||
traceback.print_tb(e.__traceback__)
|
|
||||||
|
|
||||||
# noinspection PyPep8Naming
|
# noinspection PyPep8Naming
|
||||||
def do_POST(self) -> None:
|
def do_POST(self) -> None:
|
||||||
|
|
77
matemat/webserver/test/test_post.py
Normal file
77
matemat/webserver/test/test_post.py
Normal file
|
@ -0,0 +1,77 @@
|
||||||
|
|
||||||
|
from typing import Any, Dict, List
|
||||||
|
|
||||||
|
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/post')
|
||||||
|
def post_test_pagelet(method: str,
|
||||||
|
path: str,
|
||||||
|
args: Dict[str, str],
|
||||||
|
session_vars: Dict[str, Any],
|
||||||
|
headers: Dict[str, str]):
|
||||||
|
dump: str = ''
|
||||||
|
for k, v in args.items():
|
||||||
|
dump += f'{k}: {v}\n'
|
||||||
|
return 200, dump
|
||||||
|
|
||||||
|
|
||||||
|
class TestPost(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')
|
||||||
|
|
||||||
|
def test_post_get_only_args(self):
|
||||||
|
self.client_sock.set_request(b'POST /just/testing/post?foo=bar&test=1 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()
|
||||||
|
lines: List[str] = packet.body.split('\n')[:-1]
|
||||||
|
kv: Dict[str, str] = dict()
|
||||||
|
for l in lines:
|
||||||
|
k, v = l.split(':', 1)
|
||||||
|
kv[k.strip()] = v.strip()
|
||||||
|
self.assertEqual('bar', kv['foo'])
|
||||||
|
self.assertEqual('1', kv['test'])
|
||||||
|
|
||||||
|
def test_post_post_only_args(self):
|
||||||
|
self.client_sock.set_request(b'POST /just/testing/post HTTP/1.1\r\n'
|
||||||
|
b'Content-Type: application/x-www-form-urlencoded\r\n'
|
||||||
|
b'Content-Length: 14\r\n\r\n'
|
||||||
|
b'foo=bar&test=1\r\n')
|
||||||
|
HttpHandler(self.client_sock, ('::1', 45678), self.server)
|
||||||
|
packet = self.client_sock.get_response()
|
||||||
|
lines: List[str] = packet.body.split('\n')[:-1]
|
||||||
|
kv: Dict[str, str] = dict()
|
||||||
|
for l in lines:
|
||||||
|
k, v = l.split(':', 1)
|
||||||
|
kv[k.strip()] = v.strip()
|
||||||
|
self.assertEqual('bar', kv['foo'])
|
||||||
|
self.assertEqual('1', kv['test'])
|
||||||
|
|
||||||
|
def test_post_mixed_args(self):
|
||||||
|
self.client_sock.set_request(b'POST /just/testing/post?gettest=1&foo=baz HTTP/1.1\r\n'
|
||||||
|
b'Content-Type: application/x-www-form-urlencoded\r\n'
|
||||||
|
b'Content-Length: 18\r\n\r\n'
|
||||||
|
b'foo=bar&posttest=2\r\n')
|
||||||
|
HttpHandler(self.client_sock, ('::1', 45678), self.server)
|
||||||
|
packet = self.client_sock.get_response()
|
||||||
|
lines: List[str] = packet.body.split('\n')[:-1]
|
||||||
|
kv: Dict[str, str] = dict()
|
||||||
|
for l in lines:
|
||||||
|
k, v = l.split(':', 1)
|
||||||
|
kv[k.strip()] = v.strip()
|
||||||
|
self.assertEqual('bar', kv['foo'])
|
||||||
|
self.assertEqual('1', kv['gettest'])
|
||||||
|
self.assertEqual('2', kv['posttest'])
|
||||||
|
|
|
@ -113,3 +113,14 @@ class TestServe(AbstractHttpdTest):
|
||||||
self.assertIsNone(packet.pagelet)
|
self.assertIsNone(packet.pagelet)
|
||||||
# Make sure a 404 header is served
|
# Make sure a 404 header is served
|
||||||
self.assertEqual(404, packet.statuscode)
|
self.assertEqual(404, packet.statuscode)
|
||||||
|
|
||||||
|
def test_static_post_not_allowed(self):
|
||||||
|
# Request a resource outside the webroot
|
||||||
|
self.client_sock.set_request(b'POST /iwanttouploadthis HTTP/1.1\r\n\r\nq=this%20should%20not%20be%20uploaded')
|
||||||
|
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 405 Method Not Allowed header is served
|
||||||
|
self.assertEqual(405, packet.statuscode)
|
||||||
|
|
Loading…
Reference in a new issue