From 9414b19dc21a59c1a65c522cbd5e093b00bcb2df Mon Sep 17 00:00:00 2001 From: s3lph Date: Tue, 19 Jun 2018 00:03:30 +0200 Subject: [PATCH] More unit tests for the webserver. --- matemat/webserver/test/abstract_httpd_test.py | 30 +++++++++- matemat/webserver/test/test_session.py | 59 +++++++++++++++++-- 2 files changed, 80 insertions(+), 9 deletions(-) diff --git a/matemat/webserver/test/abstract_httpd_test.py b/matemat/webserver/test/abstract_httpd_test.py index 806a312..a931628 100644 --- a/matemat/webserver/test/abstract_httpd_test.py +++ b/matemat/webserver/test/abstract_httpd_test.py @@ -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 diff --git a/matemat/webserver/test/test_session.py b/matemat/webserver/test/test_session.py index af51a75..6697da3 100644 --- a/matemat/webserver/test/test_session.py +++ b/matemat/webserver/test/test_session.py @@ -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'])