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()
62 r = Process.by_id(self.db_conn, p.id_)
63 self.assertEqual(sorted(r.conditions), sorted(set1))
64 self.assertEqual(sorted(r.enables), sorted(set2))
65 self.assertEqual(sorted(r.disables), sorted(set3))
67 def test_Process_from_table_row(self) -> None:
68 """Test .from_table_row() properly reads in class from DB"""
69 self.check_from_table_row()
70 self.check_versioned_from_table_row('title', str)
71 self.check_versioned_from_table_row('description', str)
72 self.check_versioned_from_table_row('effort', float)
73 p, set1, set2, set3 = self.p_of_conditions()
75 assert isinstance(p.id_, int)
76 for row in self.db_conn.row_where(self.checked_class.table_name,
78 # pylint: disable=no-member
79 r = Process.from_table_row(self.db_conn, row)
80 self.assertEqual(sorted(r.conditions), sorted(set1))
81 self.assertEqual(sorted(r.enables), sorted(set2))
82 self.assertEqual(sorted(r.disables), sorted(set3))
84 def test_Process_steps(self) -> None:
85 """Test addition, nesting, and non-recursion of ProcessSteps"""
86 # pylint: disable=too-many-locals
87 p1, p2, p3 = self.three_processes()
88 assert isinstance(p1.id_, int)
89 assert isinstance(p2.id_, int)
90 assert isinstance(p3.id_, int)
91 steps_p1: list[ProcessStep] = []
92 # add step of process p2 as first (top-level) step to p1
93 s_p2_to_p1 = ProcessStep(None, p1.id_, p2.id_, None)
94 steps_p1 += [s_p2_to_p1]
95 p1.set_steps(self.db_conn, steps_p1)
96 p1_dict: dict[int, ProcessStepsNode] = {}
97 p1_dict[1] = ProcessStepsNode(p2, None, True, {}, False)
98 self.assertEqual(p1.get_steps(self.db_conn, None), p1_dict)
99 # add step of process p3 as second (top-level) step to p1
100 s_p3_to_p1 = ProcessStep(None, p1.id_, p3.id_, None)
101 steps_p1 += [s_p3_to_p1]
102 p1.set_steps(self.db_conn, steps_p1)
103 p1_dict[2] = ProcessStepsNode(p3, None, True, {}, False)
104 self.assertEqual(p1.get_steps(self.db_conn, None), p1_dict)
105 # add step of process p3 as first (top-level) step to p2,
106 steps_p2: list[ProcessStep] = []
107 s_p3_to_p2 = ProcessStep(None, p2.id_, p3.id_, None)
108 steps_p2 += [s_p3_to_p2]
109 p2.set_steps(self.db_conn, steps_p2)
110 # expect it as implicit sub-step of p1's second (p3) step
111 p2_dict = {3: ProcessStepsNode(p3, None, False, {}, False)}
112 p1_dict[1].steps[3] = p2_dict[3]
113 self.assertEqual(p1.get_steps(self.db_conn, None), p1_dict)
114 # add step of process p2 as explicit sub-step to p1's second sub-step
115 s_p2_to_p1_first = ProcessStep(None, p1.id_, p2.id_, s_p3_to_p1.id_)
116 steps_p1 += [s_p2_to_p1_first]
117 p1.set_steps(self.db_conn, steps_p1)
118 seen_3 = ProcessStepsNode(p3, None, False, {}, 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, {}, False)
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, {}, False)
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, {}, False)}
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, {},
154 p1_dict[2].steps[4].steps[3].steps[7] = ProcessStepsNode(p3, 3, False,
156 self.assertEqual(p1.get_steps(self.db_conn, None), p1_dict)
158 def test_Process_conditions(self) -> None:
159 """Test setting Process.conditions/enables/disables."""
162 for target in ('conditions', 'enables', 'disables'):
163 method = getattr(p, f'set_{target}')
164 c1, c2 = Condition(None), Condition(None)
165 c1.save(self.db_conn)
166 c2.save(self.db_conn)
167 assert isinstance(c1.id_, int)
168 assert isinstance(c2.id_, int)
169 method(self.db_conn, [])
170 self.assertEqual(getattr(p, target), [])
171 method(self.db_conn, [c1.id_])
172 self.assertEqual(getattr(p, target), [c1])
173 method(self.db_conn, [c2.id_])
174 self.assertEqual(getattr(p, target), [c2])
175 method(self.db_conn, [c1.id_, c2.id_])
176 self.assertEqual(getattr(p, target), [c1, c2])
178 def test_Process_by_id(self) -> None:
179 """Test .by_id(), including creation"""
182 def test_Process_all(self) -> None:
186 def test_Process_singularity(self) -> None:
187 """Test pointers made for single object keep pointing to it."""
188 self.check_singularity('conditions', [Condition(None)])
190 def test_Process_versioned_attributes_singularity(self) -> None:
191 """Test behavior of VersionedAttributes on saving (with .title)."""
192 self.check_versioned_singularity()
194 def test_Process_removal(self) -> None:
195 """Test removal of Processes and ProcessSteps."""
197 p1, p2, p3 = self.three_processes()
198 assert isinstance(p1.id_, int)
199 assert isinstance(p2.id_, int)
200 assert isinstance(p3.id_, int)
201 step = ProcessStep(None, p2.id_, p1.id_, None)
202 p2.set_steps(self.db_conn, [step])
203 with self.assertRaises(HandledException):
204 p1.remove(self.db_conn)
205 p2.set_steps(self.db_conn, [])
206 with self.assertRaises(NotFoundException):
207 ProcessStep.by_id(self.db_conn, step.id_)
208 p1.remove(self.db_conn)
209 step = ProcessStep(None, p2.id_, p3.id_, None)
210 p2.set_steps(self.db_conn, [step])
211 p2.remove(self.db_conn)
212 with self.assertRaises(NotFoundException):
213 ProcessStep.by_id(self.db_conn, step.id_)
214 todo = Todo(None, p3, False, '2024-01-01')
215 todo.save(self.db_conn)
216 with self.assertRaises(HandledException):
217 p3.remove(self.db_conn)
218 todo.remove(self.db_conn)
219 p3.remove(self.db_conn)
222 class TestsWithDBForProcessStep(TestCaseWithDB):
223 """Module tests requiring DB setup."""
224 checked_class = ProcessStep
225 default_init_kwargs = {'owner_id': 2, 'step_process_id': 3,
228 def test_ProcessStep_from_table_row(self) -> None:
229 """Test .from_table_row() properly reads in class from DB"""
230 self.check_from_table_row(2, 3, None)
232 def test_ProcessStep_singularity(self) -> None:
233 """Test pointers made for single object keep pointing to it."""
234 self.check_singularity('parent_step_id', 1, 2, 3, None)
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_storage([])
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'},
271 '/condition', 302, '/condition?id=1')
272 self.post_process(3, form_data | {'condition': [1]})
273 self.post_process(4, form_data | {'disables': [1]})
274 self.post_process(5, form_data | {'enables': [1]})
275 form_data['delete'] = ''
276 self.check_post(form_data, '/process?id=', 404)
277 self.check_post(form_data, '/process?id=6', 404)
278 self.check_post(form_data, '/process?id=5', 302, '/processes')
280 def test_do_POST_process_steps(self) -> None:
281 """Test behavior of ProcessStep posting."""
282 # pylint: disable=too-many-statements
283 form_data_1 = self.post_process(1)
286 # post first (top-level) step of process 2 to process 1
287 form_data_1['new_top_step'] = [2]
288 self.post_process(1, form_data_1)
289 retrieved_process = Process.by_id(self.db_conn, 1)
290 self.assertEqual(len(retrieved_process.explicit_steps), 1)
291 retrieved_step = retrieved_process.explicit_steps[0]
292 self.assertEqual(retrieved_step.step_process_id, 2)
293 self.assertEqual(retrieved_step.owner_id, 1)
294 self.assertEqual(retrieved_step.parent_step_id, None)
295 # post empty steps list to process, expect clean slate, and old step to
296 # completely disappear
297 form_data_1['new_top_step'] = []
298 self.post_process(1, form_data_1)
299 retrieved_process = Process.by_id(self.db_conn, 1)
300 self.assertEqual(retrieved_process.explicit_steps, [])
301 with self.assertRaises(NotFoundException):
302 ProcessStep.by_id(self.db_conn, retrieved_step.id_)
303 # post new first (top_level) step of process 3 to process 1
304 form_data_1['new_top_step'] = [3]
305 self.post_process(1, form_data_1)
306 retrieved_process = Process.by_id(self.db_conn, 1)
307 retrieved_step = retrieved_process.explicit_steps[0]
308 self.assertEqual(retrieved_step.step_process_id, 3)
309 self.assertEqual(retrieved_step.owner_id, 1)
310 self.assertEqual(retrieved_step.parent_step_id, None)
311 # post to process steps list without keeps, expect clean slate
312 form_data_1['new_top_step'] = []
313 form_data_1['steps'] = [retrieved_step.id_]
314 self.post_process(1, form_data_1)
315 retrieved_process = Process.by_id(self.db_conn, 1)
316 self.assertEqual(retrieved_process.explicit_steps, [])
317 # post to process empty steps list but keep, expect 400
318 form_data_1['steps'] = []
319 form_data_1['keep_step'] = [retrieved_step.id_]
320 self.check_post(form_data_1, '/process?id=1', 400, '/process?id=1')
321 # post to process steps list with keep on non-created step, expect 400
322 form_data_1['steps'] = [retrieved_step.id_]
323 form_data_1['keep_step'] = [retrieved_step.id_]
324 self.check_post(form_data_1, '/process?id=1', 400, '/process?id=1')
325 # post to process steps list with keep and process ID, expect 200
326 form_data_1[f'step_{retrieved_step.id_}_process_id'] = [2]
327 self.post_process(1, form_data_1)
328 retrieved_process = Process.by_id(self.db_conn, 1)
329 self.assertEqual(len(retrieved_process.explicit_steps), 1)
330 retrieved_step = retrieved_process.explicit_steps[0]
331 self.assertEqual(retrieved_step.step_process_id, 2)
332 self.assertEqual(retrieved_step.owner_id, 1)
333 self.assertEqual(retrieved_step.parent_step_id, None)
334 # post nonsensical new_top_step id and otherwise zero'd steps, expect
335 # 400 and preservation of previous state
336 form_data_1['new_top_step'] = ['foo']
337 form_data_1['steps'] = []
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, 2)
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, 3)
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 form_data_1[f'step_{retrieved_step_1.id_}_process_id'] = [3]
380 # post sub-step to step
381 form_data_1[f'new_step_to_{retrieved_step_1.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)