plomlompom.com Git - plomtask/blob - tests/processes.py

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