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

More unit tests for the webserver.

Browse source
This commit is contained in:
s3lph 2018-06-19 00:03:30 +02:00
parent 2f2f6b73e9
commit 9414b19dc2
2 changed files with 80 additions and 9 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

30
matemat/webserver/test/abstract_httpd_test.py
View file

@ -1,5 +1,5 @@
from typing import Any, Dict, Tuple
from typing import Any, Callable, Dict, Tuple, Union
import unittest.mock
from io import BytesIO
@ -8,6 +8,8 @@ from abc import ABC
from datetime import datetime
from http.server import HTTPServer
from matemat.webserver.httpd import pagelet
class HttpResponse:
"""
@ -27,6 +29,7 @@ class HttpResponse:
self.headers: Dict[str, str] = {
'Content-Length': 0
}
self.pagelet: str = None
# The response body. Only UTF-8 strings are supported
self.body: str = ''
# Parsing phase, one of 'begin', 'hdr', 'body' or 'done'
@ -34,6 +37,10 @@ class HttpResponse:
# Buffer for uncompleted lines
self.buffer: bytes = bytes()
def __finalize(self):
self.parse_phase = 'done'
self.pagelet = self.headers['X-Test-Pagelet']
def parse(self, fragment: bytes) -> None:
"""
Parse a new fragment of data. This function does nothing if the parsed HTTP response is already complete.
@ -47,7 +54,7 @@ class HttpResponse:
elif self.parse_phase == 'body':
self.body += fragment.decode('utf-8')
if len(self.body) >= int(self.headers['Content-Length']):
self.parse_phase = 'done'
self.__finalize()
return
if b'\r\n' not in fragment:
# If the fragment does not contain a CR-LF, add it to the buffer, we only want to parse whole lines
@ -82,7 +89,7 @@ class HttpResponse:
# if there is a remainder in the data packet, it is (part of) the body, add to body string
self.body += line
if len(self.body) >= int(self.headers['Content-Length']):
self.parse_phase = 'done'
self.__finalize()
class MockServer:
@ -144,6 +151,23 @@ class MockSocket(bytes):
return self.__packet
def test_pagelet(path: str):
def with_testing_headers(fun: Callable[[str, str, Dict[str, str], Dict[str, Any], Dict[str, str]],
Tuple[int, Union[bytes, str]]]):
@pagelet(path)
def testing_wrapper(method: str,
path: str,
args: Dict[str, str],
session_vars: Dict[str, Any],
headers: Dict[str, str]):
status, body = fun(method, path, args, session_vars, headers)
headers['X-Test-Pagelet'] = fun.__name__
return status, body
return testing_wrapper
return with_testing_headers
class AbstractHttpdTest(ABC, unittest.TestCase):
"""
An abstract test case that can be inherited by test case classes that want to test part of the webserver's core

59
matemat/webserver/test/test_session.py
View file

@ -2,13 +2,18 @@
from typing import Any, Dict
from datetime import datetime, timedelta
from time import sleep
from matemat.webserver.httpd import HttpHandler, pagelet
from matemat.webserver.test.abstract_httpd_test import AbstractHttpdTest
from matemat.webserver.httpd import HttpHandler
from matemat.webserver.test.abstract_httpd_test import AbstractHttpdTest, test_pagelet
@pagelet('/just/testing/sessions')
def test_pagelet(method: str, path: str, args: Dict[str, str], session_vars: Dict[str, Any], headers: Dict[str, str]):
@test_pagelet('/just/testing/sessions')
def session_test_pagelet(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 200, 'session test'
@ -21,7 +26,7 @@ class TestSession(AbstractHttpdTest):
def test_create_new_session(self):
# Reference date to make sure the session expiry lies in the future
refdate = datetime.utcnow() + timedelta(seconds=3500)
refdate: datetime = datetime.utcnow() + timedelta(seconds=3500)
# Send a mock GET request for '/just/testing/sessions'
self.client_sock.set_request(b'GET /just/testing/sessions HTTP/1.1\r\n\r\n')
# Trigger request handling
@ -31,7 +36,7 @@ class TestSession(AbstractHttpdTest):
# Make sure a full HTTP response was parsed
self.assertEqual('done', packet.parse_phase)
# Make sure the request was served by the test pagelet
self.assertEqual('session test', packet.body)
self.assertEqual('session_test_pagelet', packet.pagelet)
self.assertEqual(200, packet.statuscode)
session_id: str = list(handler.server.session_vars.keys())[0]
@ -51,3 +56,45 @@ class TestSession(AbstractHttpdTest):
# Make sure the session exists on the server
self.assertIn('test', handler.session_vars)
self.assertEqual('hello, world!', handler.session_vars['test'])
def test_resume_session(self):
# Test session expiry date
refdate: datetime = datetime.utcnow() + timedelta(hours=1)
# Session ID for testing
session_id: str = 'testsessionid'
# Insert test session
self.server.session_vars[session_id] = refdate, {'test': 'bar'}
sleep(2)
# Send a mock GET request for '/just/testing/sessions' with a matemat session cookie
self.client_sock.set_request(
f'GET /just/testing/sessions HTTP/1.1\r\nCookie: matemat_session_id={session_id}\r\n'.encode('utf-8'))
# Trigger request handling
handler = HttpHandler(self.client_sock, ('::1', 45678), self.server)
# Fetch the parsed response
packet = self.client_sock.get_response()
# Make sure a full HTTP response was parsed
self.assertEqual('done', packet.parse_phase)
# Make sure the request was served by the test pagelet
self.assertEqual('session_test_pagelet', packet.pagelet)
self.assertEqual(200, packet.statuscode)
response_session_id: str = list(handler.server.session_vars.keys())[0]
# Make sure a cookie was set - assuming that only one was set
self.assertIn('Set-Cookie', packet.headers)
# Split into the cookie itself
cookie, expiry = packet.headers['Set-Cookie'].split(';')
cookie: str = cookie.strip()
expiry: str = expiry.strip()
# Make sure the 'matemat_session_id' cookie was set to the session ID string
self.assertEqual(f'matemat_session_id={response_session_id}', cookie)
# Make sure the session ID matches the one we sent along
self.assertEqual(session_id, response_session_id)
# Make sure the session timeout was postponed
self.assertTrue(expiry.startswith('expires='))
_, expdatestr = expiry.split('=', 1)
expdate = datetime.strptime(expdatestr, '%a, %d %b %Y %H:%M:%S GMT')
self.assertTrue(expdate > refdate)
# Make sure the session exists on the server
self.assertIn('test', handler.session_vars)
self.assertEqual('hello, world!', handler.session_vars['test'])