1 """Test Processes module."""
2 from tests.utils import TestCaseWithDB, TestCaseWithServer, TestCaseSansDB
3 from plomtask.processes import Process, ProcessStep, ProcessStepsNode
4 from plomtask.conditions import Condition
5 from plomtask.exceptions import HandledException, NotFoundException
6 from plomtask.todos import Todo
9 class TestsSansDB(TestCaseSansDB):
10 """Module tests not requiring DB setup."""
11 checked_class = Process
13 versioned_defaults_to_test = {'title': 'UNNAMED', 'description': '',
17 class TestsSansDBProcessStep(TestCaseSansDB):
18 """Module tests not requiring DB setup."""
19 checked_class = ProcessStep
21 default_init_args = [2, 3, 4]
24 class TestsWithDB(TestCaseWithDB):
25 """Module tests requiring DB setup."""
26 checked_class = Process
27 test_versioneds = {'title': str, 'description': str, 'effort': float}
29 def three_processes(self) -> tuple[Process, Process, Process]:
30 """Return three saved processes."""
31 p1, p2, p3 = Process(None), Process(None), Process(None)
32 for p in [p1, p2, p3]:
36 def p_of_conditions(self) -> tuple[Process, list[Condition],
37 list[Condition], list[Condition]]:
38 """Return Process and its three Condition sets."""
40 c1, c2, c3 = Condition(None), Condition(None), Condition(None)
41 for c in [c1, c2, c3]:
43 assert isinstance(c1.id_, int)
44 assert isinstance(c2.id_, int)
45 assert isinstance(c3.id_, int)
49 p.set_conditions(self.db_conn, [c.id_ for c in set_1
50 if isinstance(c.id_, int)])
51 p.set_enables(self.db_conn, [c.id_ for c in set_2
52 if isinstance(c.id_, int)])
53 p.set_disables(self.db_conn, [c.id_ for c in set_3
54 if isinstance(c.id_, int)])
56 return p, set_1, set_2, set_3
58 def test_Process_conditions_saving(self) -> None:
59 """Test .save/.save_core."""
60 p, set1, set2, set3 = self.p_of_conditions()
61 r = Process.by_id(self.db_conn, p.id_)
62 self.assertEqual(sorted(r.conditions), sorted(set1))
63 self.assertEqual(sorted(r.enables), sorted(set2))
64 self.assertEqual(sorted(r.disables), sorted(set3))
66 def test_from_table_row(self) -> None:
67 """Test .from_table_row() properly reads in class from DB"""
68 super().test_from_table_row()
69 self.check_versioned_from_table_row('title', str)
70 self.check_versioned_from_table_row('description', str)
71 self.check_versioned_from_table_row('effort', float)
72 p, set1, set2, set3 = self.p_of_conditions()
74 assert isinstance(p.id_, int)
75 for row in self.db_conn.row_where(self.checked_class.table_name,
77 r = Process.from_table_row(self.db_conn, row)
78 self.assertEqual(sorted(r.conditions), sorted(set1))
79 self.assertEqual(sorted(r.enables), sorted(set2))
80 self.assertEqual(sorted(r.disables), sorted(set3))
82 def test_Process_steps(self) -> None:
83 """Test addition, nesting, and non-recursion of ProcessSteps"""
84 # pylint: disable=too-many-locals
85 # pylint: disable=too-many-statements
86 p1, p2, p3 = self.three_processes()
87 assert isinstance(p1.id_, int)
88 assert isinstance(p2.id_, int)
89 assert isinstance(p3.id_, int)
90 steps_p1: list[ProcessStep] = []
91 # add step of process p2 as first (top-level) step to p1
92 s_p2_to_p1 = ProcessStep(None, p1.id_, p2.id_, None)
93 steps_p1 += [s_p2_to_p1]
94 p1.set_steps(self.db_conn, steps_p1)
95 p1_dict: dict[int, ProcessStepsNode] = {}
96 p1_dict[1] = ProcessStepsNode(p2, None, True, {})
97 self.assertEqual(p1.get_steps(self.db_conn, None), p1_dict)
98 # add step of process p3 as second (top-level) step to p1
99 s_p3_to_p1 = ProcessStep(None, p1.id_, p3.id_, None)
100 steps_p1 += [s_p3_to_p1]
101 p1.set_steps(self.db_conn, steps_p1)
102 p1_dict[2] = ProcessStepsNode(p3, None, True, {})
103 self.assertEqual(p1.get_steps(self.db_conn, None), p1_dict)
104 # add step of process p3 as first (top-level) step to p2,
105 steps_p2: list[ProcessStep] = []
106 s_p3_to_p2 = ProcessStep(None, p2.id_, p3.id_, None)
107 steps_p2 += [s_p3_to_p2]
108 p2.set_steps(self.db_conn, steps_p2)
109 # expect it as implicit sub-step of p1's second (p3) step
110 p2_dict = {3: ProcessStepsNode(p3, None, False, {})}
111 p1_dict[1].steps[3] = p2_dict[3]
112 self.assertEqual(p1.get_steps(self.db_conn, None), p1_dict)
113 # add step of process p2 as explicit sub-step to p1's second sub-step
114 s_p2_to_p1_first = ProcessStep(None, p1.id_, p2.id_, s_p3_to_p1.id_)
115 steps_p1 += [s_p2_to_p1_first]
116 p1.set_steps(self.db_conn, steps_p1)
117 seen_3 = ProcessStepsNode(p3, None, False, {}, False)
118 p1_dict[1].steps[3].seen = True
119 p1_dict[2].steps[4] = ProcessStepsNode(p2, s_p3_to_p1.id_, True,
121 self.assertEqual(p1.get_steps(self.db_conn, None), p1_dict)
122 # add step of process p3 as explicit sub-step to non-existing p1
123 # sub-step (of id=999), expect it to become another p1 top-level step
124 s_p3_to_p1_999 = ProcessStep(None, p1.id_, p3.id_, 999)
125 steps_p1 += [s_p3_to_p1_999]
126 p1.set_steps(self.db_conn, steps_p1)
127 p1_dict[5] = ProcessStepsNode(p3, None, True, {})
128 self.assertEqual(p1.get_steps(self.db_conn, None), p1_dict)
129 # add step of process p3 as explicit sub-step to p1's implicit p3
130 # sub-step, expect it to become another p1 top-level step
131 s_p3_to_p1_impl_p3 = ProcessStep(None, p1.id_, p3.id_, s_p3_to_p2.id_)
132 steps_p1 += [s_p3_to_p1_impl_p3]
133 p1.set_steps(self.db_conn, steps_p1)
134 p1_dict[6] = ProcessStepsNode(p3, None, True, {})
135 self.assertEqual(p1.get_steps(self.db_conn, None), p1_dict)
136 self.assertEqual(p1.used_as_step_by(self.db_conn), [])
137 self.assertEqual(p2.used_as_step_by(self.db_conn), [p1])
138 self.assertEqual(p3.used_as_step_by(self.db_conn), [p1, p2])
139 # # add step of process p3 as explicit sub-step to p1's first sub-step,
140 # # expect it to eliminate implicit p3 sub-step
141 # s_p3_to_p1_first_explicit = ProcessStep(None, p1.id_, p3.id_,
143 # p1_dict[1].steps = {7: ProcessStepsNode(p3, 1, True, {})}
144 # p1_dict[2].steps[4].steps[3].seen = False
145 # steps_p1 += [s_p3_to_p1_first_explicit]
146 # p1.set_steps(self.db_conn, steps_p1)
147 # self.assertEqual(p1.get_steps(self.db_conn, None), p1_dict)
148 # ensure implicit steps non-top explicit steps are shown
149 s_p3_to_p2_first = ProcessStep(None, p2.id_, p3.id_, s_p3_to_p2.id_)
150 steps_p2 += [s_p3_to_p2_first]
151 p2.set_steps(self.db_conn, steps_p2)
152 p1_dict[1].steps[3].steps[7] = ProcessStepsNode(p3, 3, False, {}, True)
153 p1_dict[2].steps[4].steps[3].steps[7] = ProcessStepsNode(p3, 3, False,
155 self.assertEqual(p1.get_steps(self.db_conn, None), p1_dict)
156 # ensure suppressed step nodes are hidden
157 assert isinstance(s_p3_to_p2.id_, int)
158 p1.set_step_suppressions(self.db_conn, [s_p3_to_p2.id_])
159 p1_dict[1].steps[3].steps = {}
160 p1_dict[1].steps[3].is_suppressed = True
161 p1_dict[2].steps[4].steps[3].steps = {}
162 p1_dict[2].steps[4].steps[3].is_suppressed = True
163 self.assertEqual(p1.get_steps(self.db_conn), p1_dict)
165 def test_Process_conditions(self) -> None:
166 """Test setting Process.conditions/enables/disables."""
169 for target in ('conditions', 'enables', 'disables'):
170 method = getattr(p, f'set_{target}')
171 c1, c2 = Condition(None), Condition(None)
172 c1.save(self.db_conn)
173 c2.save(self.db_conn)
174 assert isinstance(c1.id_, int)
175 assert isinstance(c2.id_, int)
176 method(self.db_conn, [])
177 self.assertEqual(getattr(p, target), [])
178 method(self.db_conn, [c1.id_])
179 self.assertEqual(getattr(p, target), [c1])
180 method(self.db_conn, [c2.id_])
181 self.assertEqual(getattr(p, target), [c2])
182 method(self.db_conn, [c1.id_, c2.id_])
183 self.assertEqual(getattr(p, target), [c1, c2])
185 def test_Process_versioned_attributes_singularity(self) -> None:
186 """Test behavior of VersionedAttributes on saving (with .title)."""
187 self.check_versioned_singularity()
189 def test_Process_removal(self) -> None:
190 """Test removal of Processes and ProcessSteps."""
192 p1, p2, p3 = self.three_processes()
193 assert isinstance(p1.id_, int)
194 assert isinstance(p2.id_, int)
195 assert isinstance(p3.id_, int)
196 step = ProcessStep(None, p2.id_, p1.id_, None)
197 p2.set_steps(self.db_conn, [step])
199 with self.assertRaises(HandledException):
200 p1.remove(self.db_conn)
201 p2.set_steps(self.db_conn, [])
202 with self.assertRaises(NotFoundException):
203 ProcessStep.by_id(self.db_conn, step_id)
204 p1.remove(self.db_conn)
205 step = ProcessStep(None, p2.id_, p3.id_, None)
207 p2.set_steps(self.db_conn, [step])
208 p2.remove(self.db_conn)
209 with self.assertRaises(NotFoundException):
210 ProcessStep.by_id(self.db_conn, step_id)
211 todo = Todo(None, p3, False, '2024-01-01')
212 todo.save(self.db_conn)
213 with self.assertRaises(HandledException):
214 p3.remove(self.db_conn)
215 todo.remove(self.db_conn)
216 p3.remove(self.db_conn)
219 class TestsWithDBForProcessStep(TestCaseWithDB):
220 """Module tests requiring DB setup."""
221 checked_class = ProcessStep
222 default_init_kwargs = {'owner_id': 2, 'step_process_id': 3,
225 def setUp(self) -> None:
232 def test_saving_and_caching(self) -> None:
233 """Test storage and initialization of instances and attributes."""
234 self.check_saving_and_caching(id_=1, **self.default_init_kwargs)
236 def test_ProcessStep_remove(self) -> None:
237 """Test .remove and unsetting of owner's .explicit_steps entry."""
240 p1.save(self.db_conn)
241 p2.save(self.db_conn)
242 assert isinstance(p1.id_, int)
243 assert isinstance(p2.id_, int)
244 step = ProcessStep(None, p1.id_, p2.id_, None)
245 p1.set_steps(self.db_conn, [step])
246 step.remove(self.db_conn)
247 self.assertEqual(p1.explicit_steps, [])
248 self.check_identity_with_cache_and_db([])
251 class TestsWithServer(TestCaseWithServer):
252 """Module tests against our HTTP server/handler (and database)."""
254 def test_do_POST_process(self) -> None:
255 """Test POST /process and its effect on the database."""
256 self.assertEqual(0, len(Process.all(self.db_conn)))
257 form_data = self.post_process()
258 self.assertEqual(1, len(Process.all(self.db_conn)))
259 self.check_post(form_data, '/process?id=FOO', 400)
260 self.check_post(form_data | {'effort': 'foo'}, '/process?id=', 400)
261 self.check_post({}, '/process?id=', 400)
262 self.check_post({'title': '', 'description': ''}, '/process?id=', 400)
263 self.check_post({'title': '', 'effort': 1.1}, '/process?id=', 400)
264 self.check_post({'description': '', 'effort': 1.0},
266 self.assertEqual(1, len(Process.all(self.db_conn)))
267 form_data = {'title': 'foo', 'description': 'foo', 'effort': 1.0}
268 self.post_process(2, form_data | {'condition': []})
269 self.check_post(form_data | {'condition': [1]}, '/process?id=', 404)
270 self.check_post({'title': 'foo', 'description': 'foo',
272 '/condition', 302, '/condition?id=1')
273 self.post_process(3, form_data | {'condition': [1]})
274 self.post_process(4, form_data | {'disables': [1]})
275 self.post_process(5, form_data | {'enables': [1]})
276 form_data['delete'] = ''
277 self.check_post(form_data, '/process?id=', 404)
278 self.check_post(form_data, '/process?id=6', 404)
279 self.check_post(form_data, '/process?id=5', 302, '/processes')
281 def test_do_POST_process_steps(self) -> None:
282 """Test behavior of ProcessStep posting."""
283 # pylint: disable=too-many-statements
284 form_data_1 = self.post_process(1)
287 # post first (top-level) step of process 2 to process 1
288 form_data_1['new_top_step'] = [2]
289 self.post_process(1, form_data_1)
290 retrieved_process = Process.by_id(self.db_conn, 1)
291 self.assertEqual(len(retrieved_process.explicit_steps), 1)
292 retrieved_step = retrieved_process.explicit_steps[0]
293 retrieved_step_id = retrieved_step.id_
294 self.assertEqual(retrieved_step.step_process_id, 2)
295 self.assertEqual(retrieved_step.owner_id, 1)
296 self.assertEqual(retrieved_step.parent_step_id, None)
297 # post empty steps list to process, expect clean slate, and old step to
298 # completely disappear
299 form_data_1['new_top_step'] = []
300 self.post_process(1, form_data_1)
301 retrieved_process = Process.by_id(self.db_conn, 1)
302 self.assertEqual(retrieved_process.explicit_steps, [])
303 with self.assertRaises(NotFoundException):
304 ProcessStep.by_id(self.db_conn, retrieved_step_id)
305 # post new first (top_level) step of process 3 to process 1
306 form_data_1['new_top_step'] = [3]
307 self.post_process(1, form_data_1)
308 retrieved_process = Process.by_id(self.db_conn, 1)
309 retrieved_step = retrieved_process.explicit_steps[0]
310 self.assertEqual(retrieved_step.step_process_id, 3)
311 self.assertEqual(retrieved_step.owner_id, 1)
312 self.assertEqual(retrieved_step.parent_step_id, None)
313 # post to process steps list without keeps, expect clean slate
314 form_data_1['new_top_step'] = []
315 form_data_1['steps'] = [retrieved_step.id_]
316 self.post_process(1, form_data_1)
317 retrieved_process = Process.by_id(self.db_conn, 1)
318 self.assertEqual(retrieved_process.explicit_steps, [])
319 # post to process empty steps list but keep, expect 400
320 form_data_1['steps'] = []
321 form_data_1['keep_step'] = [retrieved_step_id]
322 self.check_post(form_data_1, '/process?id=1', 400, '/process?id=1')
323 # post to process steps list with keep on non-created step, expect 400
324 form_data_1['steps'] = [retrieved_step_id]
325 form_data_1['keep_step'] = [retrieved_step_id]
326 self.check_post(form_data_1, '/process?id=1', 400, '/process?id=1')
327 # post to process steps list with keep and process ID, expect 200
328 form_data_1[f'step_{retrieved_step_id}_process_id'] = [2]
329 self.post_process(1, form_data_1)
330 retrieved_process = Process.by_id(self.db_conn, 1)
331 self.assertEqual(len(retrieved_process.explicit_steps), 1)
332 retrieved_step = retrieved_process.explicit_steps[0]
333 self.assertEqual(retrieved_step.step_process_id, 2)
334 self.assertEqual(retrieved_step.owner_id, 1)
335 self.assertEqual(retrieved_step.parent_step_id, None)
336 # post nonsense, expect 400 and preservation of previous state
337 form_data_1['steps'] = ['foo']
338 form_data_1['keep_step'] = []
339 self.check_post(form_data_1, '/process?id=1', 400, '/process?id=1')
340 retrieved_process = Process.by_id(self.db_conn, 1)
341 self.assertEqual(len(retrieved_process.explicit_steps), 1)
342 retrieved_step = retrieved_process.explicit_steps[0]
343 self.assertEqual(retrieved_step.step_process_id, 2)
344 self.assertEqual(retrieved_step.owner_id, 1)
345 self.assertEqual(retrieved_step.parent_step_id, None)
346 # post to process steps list with keep and process ID, expect 200
347 form_data_1['new_top_step'] = [3]
348 form_data_1['steps'] = [retrieved_step.id_]
349 form_data_1['keep_step'] = [retrieved_step.id_]
350 self.post_process(1, form_data_1)
351 retrieved_process = Process.by_id(self.db_conn, 1)
352 self.assertEqual(len(retrieved_process.explicit_steps), 2)
353 retrieved_step_0 = retrieved_process.explicit_steps[0]
354 self.assertEqual(retrieved_step_0.step_process_id, 3)
355 self.assertEqual(retrieved_step_0.owner_id, 1)
356 self.assertEqual(retrieved_step_0.parent_step_id, None)
357 retrieved_step_1 = retrieved_process.explicit_steps[1]
358 self.assertEqual(retrieved_step_1.step_process_id, 2)
359 self.assertEqual(retrieved_step_1.owner_id, 1)
360 self.assertEqual(retrieved_step_1.parent_step_id, None)
361 # post to process steps list with keeps etc., but trigger recursion
362 form_data_1['new_top_step'] = []
363 form_data_1['steps'] = [retrieved_step_0.id_, retrieved_step_1.id_]
364 form_data_1['keep_step'] = [retrieved_step_0.id_, retrieved_step_1.id_]
365 form_data_1[f'step_{retrieved_step_0.id_}_process_id'] = [2]
366 form_data_1[f'step_{retrieved_step_1.id_}_process_id'] = [1]
367 self.check_post(form_data_1, '/process?id=1', 400, '/process?id=1')
368 # check previous status preserved despite failed steps setting
369 retrieved_process = Process.by_id(self.db_conn, 1)
370 self.assertEqual(len(retrieved_process.explicit_steps), 2)
371 retrieved_step_0 = retrieved_process.explicit_steps[0]
372 self.assertEqual(retrieved_step_0.step_process_id, 2)
373 self.assertEqual(retrieved_step_0.owner_id, 1)
374 self.assertEqual(retrieved_step_0.parent_step_id, None)
375 retrieved_step_1 = retrieved_process.explicit_steps[1]
376 self.assertEqual(retrieved_step_1.step_process_id, 3)
377 self.assertEqual(retrieved_step_1.owner_id, 1)
378 self.assertEqual(retrieved_step_1.parent_step_id, None)
379 # post sub-step to step
380 form_data_1[f'step_{retrieved_step_0.id_}_process_id'] = [3]
381 form_data_1[f'new_step_to_{retrieved_step_0.id_}'] = [3]
382 self.post_process(1, form_data_1)
383 retrieved_process = Process.by_id(self.db_conn, 1)
384 self.assertEqual(len(retrieved_process.explicit_steps), 3)
385 retrieved_step_0 = retrieved_process.explicit_steps[0]
386 self.assertEqual(retrieved_step_0.step_process_id, 2)
387 self.assertEqual(retrieved_step_0.owner_id, 1)
388 self.assertEqual(retrieved_step_0.parent_step_id, None)
389 retrieved_step_1 = retrieved_process.explicit_steps[1]
390 self.assertEqual(retrieved_step_1.step_process_id, 3)
391 self.assertEqual(retrieved_step_1.owner_id, 1)
392 self.assertEqual(retrieved_step_1.parent_step_id, None)
393 retrieved_step_2 = retrieved_process.explicit_steps[2]
394 self.assertEqual(retrieved_step_2.step_process_id, 3)
395 self.assertEqual(retrieved_step_2.owner_id, 1)
396 self.assertEqual(retrieved_step_2.parent_step_id, retrieved_step_1.id_)
398 def test_do_GET(self) -> None:
399 """Test /process and /processes response codes."""
400 self.check_get_defaults('/process')
401 self.check_get('/processes', 200)