1 """Shared test utilities."""
2 from unittest import TestCase
3 from threading import Thread
4 from http.client import HTTPConnection
5 from urllib.parse import urlencode
6 from datetime import datetime
7 from os import remove as remove_file
8 from plomtask.db import DatabaseFile, DatabaseConnection
9 from plomtask.http import TaskHandler, TaskServer
12 class TestCaseWithDB(TestCase):
13 """Module tests not requiring DB setup."""
15 def setUp(self) -> None:
16 timestamp = datetime.now().timestamp()
17 self.db_file = DatabaseFile(f'test_db:{timestamp}')
19 self.db_conn = DatabaseConnection(self.db_file)
21 def tearDown(self) -> None:
23 remove_file(self.db_file.path)
26 class TestCaseWithServer(TestCaseWithDB):
27 """Module tests against our HTTP server/handler (and database)."""
29 def setUp(self) -> None:
31 self.httpd = TaskServer(self.db_file, ('localhost', 0), TaskHandler)
32 self.server_thread = Thread(target=self.httpd.serve_forever)
33 self.server_thread.daemon = True
34 self.server_thread.start()
35 self.conn = HTTPConnection(str(self.httpd.server_address[0]),
36 self.httpd.server_address[1])
38 def tearDown(self) -> None:
40 self.httpd.server_close()
41 self.server_thread.join()
44 def post_to(self, data: dict[str, object], target: str) -> None:
45 """Post form data to target URL."""
46 encoded_form_data = urlencode(data).encode('utf-8')
47 headers = {'Content-Type': 'application/x-www-form-urlencoded',
48 'Content-Length': str(len(encoded_form_data))}
49 self.conn.request('POST', target,
50 body=encoded_form_data, headers=headers)
52 def check_redirect(self, target: str) -> None:
53 """Check that self.conn answers with a 302 redirect to target."""
54 response = self.conn.getresponse()
55 self.assertEqual(response.status, 302)
56 self.assertEqual(response.getheader('Location'), target)