home · contact · privacy
More tests refactoring.
[plomtask] / tests / utils.py
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 typing import Mapping, Any
9 from plomtask.db import DatabaseFile, DatabaseConnection
10 from plomtask.http import TaskHandler, TaskServer
11 from plomtask.processes import Process, ProcessStep
12 from plomtask.conditions import Condition
13 from plomtask.days import Day
14 from plomtask.todos import Todo
15 from plomtask.exceptions import NotFoundException, HandledException
16
17
18 class TestCaseSansDB(TestCase):
19     """Tests requiring no DB setup."""
20     checked_class: Any
21     do_id_test: bool = False
22     default_init_args: list[Any] = []
23     versioned_defaults_to_test: dict[str, str | float] = {}
24
25     def test_id_setting(self) -> None:
26         """Test .id_ being set and its legal range being enforced."""
27         if not self.do_id_test:
28             return
29         with self.assertRaises(HandledException):
30             self.checked_class(0, *self.default_init_args)
31         obj = self.checked_class(5, *self.default_init_args)
32         self.assertEqual(obj.id_, 5)
33
34     def test_versioned_defaults(self) -> None:
35         """Test defaults of VersionedAttributes."""
36         if len(self.versioned_defaults_to_test) == 0:
37             return
38         obj = self.checked_class(1, *self.default_init_args)
39         for k, v in self.versioned_defaults_to_test.items():
40             self.assertEqual(getattr(obj, k).newest, v)
41
42
43 class TestCaseWithDB(TestCase):
44     """Module tests not requiring DB setup."""
45     checked_class: Any
46     default_ids: tuple[int | str, int | str, int | str] = (1, 2, 3)
47     default_init_kwargs: dict[str, Any] = {}
48     test_versioneds: dict[str, type] = {}
49
50     def setUp(self) -> None:
51         Condition.empty_cache()
52         Day.empty_cache()
53         Process.empty_cache()
54         ProcessStep.empty_cache()
55         Todo.empty_cache()
56         timestamp = datetime.now().timestamp()
57         self.db_file = DatabaseFile.create_at(f'test_db:{timestamp}')
58         self.db_conn = DatabaseConnection(self.db_file)
59
60     def tearDown(self) -> None:
61         self.db_conn.close()
62         remove_file(self.db_file.path)
63
64     def test_saving_and_caching(self) -> None:
65         """Test storage and initialization of instances and attributes."""
66         if not hasattr(self, 'checked_class'):
67             return
68         self.check_saving_and_caching(id_=1, **self.default_init_kwargs)
69         obj = self.checked_class(None, **self.default_init_kwargs)
70         obj.save(self.db_conn)
71         self.assertEqual(obj.id_, 2)
72         for k, v in self.test_versioneds.items():
73             self.check_saving_of_versioned(k, v)
74
75     def check_storage(self, content: list[Any]) -> None:
76         """Test cache and DB equal content."""
77         expected_cache = {}
78         for item in content:
79             expected_cache[item.id_] = item
80         self.assertEqual(self.checked_class.get_cache(), expected_cache)
81         db_found: list[Any] = []
82         for item in content:
83             assert isinstance(item.id_, type(self.default_ids[0]))
84             for row in self.db_conn.row_where(self.checked_class.table_name,
85                                               'id', item.id_):
86                 db_found += [self.checked_class.from_table_row(self.db_conn,
87                                                                row)]
88         self.assertEqual(sorted(content), sorted(db_found))
89
90     def check_saving_and_caching(self, **kwargs: Any) -> None:
91         """Test instance.save in its core without relations."""
92         obj = self.checked_class(**kwargs)  # pylint: disable=not-callable
93         # check object init itself doesn't store anything yet
94         self.check_storage([])
95         # check saving stores in cache and DB
96         obj.save(self.db_conn)
97         self.check_storage([obj])
98         # check core attributes set properly (and not unset by saving)
99         for key, value in kwargs.items():
100             self.assertEqual(getattr(obj, key), value)
101
102     def check_saving_of_versioned(self, attr_name: str, type_: type) -> None:
103         """Test owner's versioned attributes."""
104         owner = self.checked_class(None)
105         vals: list[Any] = ['t1', 't2'] if type_ == str else [0.9, 1.1]
106         attr = getattr(owner, attr_name)
107         attr.set(vals[0])
108         attr.set(vals[1])
109         owner.save(self.db_conn)
110         owner.uncache()
111         retrieved = owner.__class__.by_id(self.db_conn, owner.id_)
112         attr = getattr(retrieved, attr_name)
113         self.assertEqual(sorted(attr.history.values()), vals)
114
115     def check_by_id(self) -> None:
116         """Test .by_id(), including creation."""
117         # check failure if not yet saved
118         id1, id2 = self.default_ids[0], self.default_ids[1]
119         obj = self.checked_class(id1)  # pylint: disable=not-callable
120         with self.assertRaises(NotFoundException):
121             self.checked_class.by_id(self.db_conn, id1)
122         # check identity of saved and retrieved
123         obj.save(self.db_conn)
124         self.assertEqual(obj, self.checked_class.by_id(self.db_conn, id1))
125         # check create=True acts like normal instantiation (sans saving)
126         by_id_created = self.checked_class.by_id(self.db_conn, id2,
127                                                  create=True)
128         # pylint: disable=not-callable
129         self.assertEqual(self.checked_class(id2), by_id_created)
130         self.check_storage([obj])
131
132     def check_from_table_row(self, *args: Any) -> None:
133         """Test .from_table_row() properly reads in class from DB"""
134         id_ = self.default_ids[0]
135         obj = self.checked_class(id_, *args)  # pylint: disable=not-callable
136         obj.save(self.db_conn)
137         assert isinstance(obj.id_, type(self.default_ids[0]))
138         for row in self.db_conn.row_where(self.checked_class.table_name,
139                                           'id', obj.id_):
140             retrieved = self.checked_class.from_table_row(self.db_conn, row)
141             self.assertEqual(obj, retrieved)
142             self.assertEqual({obj.id_: obj}, self.checked_class.get_cache())
143
144     def check_versioned_from_table_row(self, attr_name: str,
145                                        type_: type) -> None:
146         """Test .from_table_row() reads versioned attributes from DB."""
147         owner = self.checked_class(None)
148         vals: list[Any] = ['t1', 't2'] if type_ == str else [0.9, 1.1]
149         attr = getattr(owner, attr_name)
150         attr.set(vals[0])
151         attr.set(vals[1])
152         owner.save(self.db_conn)
153         for row in self.db_conn.row_where(owner.table_name, 'id', owner.id_):
154             retrieved = owner.__class__.from_table_row(self.db_conn, row)
155             attr = getattr(retrieved, attr_name)
156             self.assertEqual(sorted(attr.history.values()), vals)
157
158     def check_all(self) -> tuple[Any, Any, Any]:
159         """Test .all()."""
160         # pylint: disable=not-callable
161         item1 = self.checked_class(self.default_ids[0])
162         item2 = self.checked_class(self.default_ids[1])
163         item3 = self.checked_class(self.default_ids[2])
164         # check pre-save .all() returns empty list
165         self.assertEqual(self.checked_class.all(self.db_conn), [])
166         # check that all() shows all saved, but no unsaved items
167         item1.save(self.db_conn)
168         item3.save(self.db_conn)
169         self.assertEqual(sorted(self.checked_class.all(self.db_conn)),
170                          sorted([item1, item3]))
171         item2.save(self.db_conn)
172         self.assertEqual(sorted(self.checked_class.all(self.db_conn)),
173                          sorted([item1, item2, item3]))
174         return item1, item2, item3
175
176     def check_singularity(self, defaulting_field: str,
177                           non_default_value: Any, *args: Any) -> None:
178         """Test pointers made for single object keep pointing to it."""
179         id1 = self.default_ids[0]
180         obj = self.checked_class(id1, *args)  # pylint: disable=not-callable
181         obj.save(self.db_conn)
182         setattr(obj, defaulting_field, non_default_value)
183         retrieved = self.checked_class.by_id(self.db_conn, id1)
184         self.assertEqual(non_default_value,
185                          getattr(retrieved, defaulting_field))
186
187     def check_versioned_singularity(self) -> None:
188         """Test singularity of VersionedAttributes on saving (with .title)."""
189         obj = self.checked_class(None)  # pylint: disable=not-callable
190         obj.save(self.db_conn)
191         assert isinstance(obj.id_, int)
192         obj.title.set('named')
193         retrieved = self.checked_class.by_id(self.db_conn, obj.id_)
194         self.assertEqual(obj.title.history, retrieved.title.history)
195
196     def check_remove(self, *args: Any) -> None:
197         """Test .remove() effects on DB and cache."""
198         id_ = self.default_ids[0]
199         obj = self.checked_class(id_, *args)  # pylint: disable=not-callable
200         with self.assertRaises(HandledException):
201             obj.remove(self.db_conn)
202         obj.save(self.db_conn)
203         obj.remove(self.db_conn)
204         self.check_storage([])
205
206
207 class TestCaseWithServer(TestCaseWithDB):
208     """Module tests against our HTTP server/handler (and database)."""
209
210     def setUp(self) -> None:
211         super().setUp()
212         self.httpd = TaskServer(self.db_file, ('localhost', 0), TaskHandler)
213         self.server_thread = Thread(target=self.httpd.serve_forever)
214         self.server_thread.daemon = True
215         self.server_thread.start()
216         self.conn = HTTPConnection(str(self.httpd.server_address[0]),
217                                    self.httpd.server_address[1])
218
219     def tearDown(self) -> None:
220         self.httpd.shutdown()
221         self.httpd.server_close()
222         self.server_thread.join()
223         super().tearDown()
224
225     def check_redirect(self, target: str) -> None:
226         """Check that self.conn answers with a 302 redirect to target."""
227         response = self.conn.getresponse()
228         self.assertEqual(response.status, 302)
229         self.assertEqual(response.getheader('Location'), target)
230
231     def check_get(self, target: str, expected_code: int) -> None:
232         """Check that a GET to target yields expected_code."""
233         self.conn.request('GET', target)
234         self.assertEqual(self.conn.getresponse().status, expected_code)
235
236     def check_post(self, data: Mapping[str, object], target: str,
237                    expected_code: int, redirect_location: str = '') -> None:
238         """Check that POST of data to target yields expected_code."""
239         encoded_form_data = urlencode(data, doseq=True).encode('utf-8')
240         headers = {'Content-Type': 'application/x-www-form-urlencoded',
241                    'Content-Length': str(len(encoded_form_data))}
242         self.conn.request('POST', target,
243                           body=encoded_form_data, headers=headers)
244         if 302 == expected_code:
245             if redirect_location == '':
246                 redirect_location = target
247             self.check_redirect(redirect_location)
248         else:
249             self.assertEqual(self.conn.getresponse().status, expected_code)
250
251     def check_get_defaults(self, path: str) -> None:
252         """Some standard model paths to test."""
253         self.check_get(path, 200)
254         self.check_get(f'{path}?id=', 200)
255         self.check_get(f'{path}?id=foo', 400)
256         self.check_get(f'/{path}?id=0', 500)
257         self.check_get(f'{path}?id=1', 200)
258
259     def post_process(self, id_: int = 1,
260                      form_data: dict[str, Any] | None = None
261                      ) -> dict[str, Any]:
262         """POST basic Process."""
263         if not form_data:
264             form_data = {'title': 'foo', 'description': 'foo', 'effort': 1.1}
265         self.check_post(form_data, '/process?id=', 302, f'/process?id={id_}')
266         return form_data