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

   1 """Test Processes module."""
   2 from unittest import TestCase
   3 from tests.utils import TestCaseWithDB, TestCaseWithServer
   4 from plomtask.processes import Process, ProcessStep, ProcessStepsNode
   5 from plomtask.conditions import Condition
   6 from plomtask.exceptions import NotFoundException, HandledException
   7
   8
   9 class TestsSansDB(TestCase):
  10     """Module tests not requiring DB setup."""
  11
  12     def test_Process_versioned_defaults(self) -> None:
  13         """Test defaults of Process' VersionedAttributes."""
  14         self.assertEqual(Process(None).title.newest, 'UNNAMED')
  15         self.assertEqual(Process(None).description.newest, '')
  16         self.assertEqual(Process(None).effort.newest, 1.0)
  17
  18     def test_Process_legal_ID(self) -> None:
  19         """Test Process cannot be instantiated with id_=0."""
  20         with self.assertRaises(HandledException):
  21             Process(0)
  22
  23
  24 class TestsWithDB(TestCaseWithDB):
  25     """Mdule tests not requiring DB setup."""
  26
  27     def setUp(self) -> None:
  28         super().setUp()
  29         self.proc1 = Process(None)
  30         self.proc1.save(self.db_conn)
  31         self.proc2 = Process(None)
  32         self.proc2.save(self.db_conn)
  33         self.proc3 = Process(None)
  34         self.proc3.save(self.db_conn)
  35
  36     def test_Process_ids(self) -> None:
  37         """Test Process.save() re Process.id_."""
  38         self.assertEqual(self.proc1.id_,
  39                          Process.by_id(self.db_conn, 1, create=False).id_)
  40         self.assertEqual(self.proc2.id_,
  41                          Process.by_id(self.db_conn, 2, create=False).id_)
  42         proc5 = Process(5)
  43         proc5.save(self.db_conn)
  44         self.assertEqual(proc5.id_,
  45                          Process.by_id(self.db_conn, 5, create=False).id_)
  46
  47     def test_Process_steps(self) -> None:
  48         """Test addition, nesting, and non-recursion of ProcessSteps"""
  49         def add_step(proc: Process,
  50                      steps_proc: list[tuple[int | None, int, int | None]],
  51                      step_tuple: tuple[int | None, int, int | None],
  52                      expected_id: int) -> None:
  53             steps_proc += [step_tuple]
  54             proc.set_steps(self.db_conn, steps_proc)
  55             steps_proc[-1] = (expected_id, step_tuple[1], step_tuple[2])
  56         assert isinstance(self.proc2.id_, int)
  57         assert isinstance(self.proc3.id_, int)
  58         steps_proc1: list[tuple[int | None, int, int | None]] = []
  59         add_step(self.proc1, steps_proc1, (None, self.proc2.id_, None), 1)
  60         p_1_dict: dict[int, ProcessStepsNode] = {}
  61         p_1_dict[1] = ProcessStepsNode(self.proc2, None, True, {}, False)
  62         self.assertEqual(self.proc1.get_steps(self.db_conn, None), p_1_dict)
  63         add_step(self.proc1, steps_proc1, (None, self.proc3.id_, None), 2)
  64         step_2 = self.proc1.explicit_steps[-1]
  65         assert isinstance(step_2.id_, int)
  66         p_1_dict[2] = ProcessStepsNode(self.proc3, None, True, {}, False)
  67         self.assertEqual(self.proc1.get_steps(self.db_conn, None), p_1_dict)
  68         steps_proc2: list[tuple[int | None, int, int | None]] = []
  69         add_step(self.proc2, steps_proc2, (None, self.proc3.id_, None), 3)
  70         p_1_dict[1].steps[3] = ProcessStepsNode(self.proc3, None,
  71                                                 False, {}, False)
  72         self.assertEqual(self.proc1.get_steps(self.db_conn, None), p_1_dict)
  73         add_step(self.proc1, steps_proc1, (None, self.proc2.id_, step_2.id_),
  74                  4)
  75         step_3 = ProcessStepsNode(self.proc3, None, False, {}, True)
  76         p_1_dict[2].steps[4] = ProcessStepsNode(self.proc2, step_2.id_, True,
  77                                                 {3: step_3}, False)
  78         self.assertEqual(self.proc1.get_steps(self.db_conn, None), p_1_dict)
  79         add_step(self.proc1, steps_proc1, (None, self.proc3.id_, 999), 5)
  80         p_1_dict[5] = ProcessStepsNode(self.proc3, None, True, {}, False)
  81         self.assertEqual(self.proc1.get_steps(self.db_conn, None), p_1_dict)
  82         add_step(self.proc1, steps_proc1, (None, self.proc3.id_, 3), 6)
  83         p_1_dict[6] = ProcessStepsNode(self.proc3, None, True, {}, False)
  84         self.assertEqual(self.proc1.get_steps(self.db_conn, None),
  85                          p_1_dict)
  86         self.assertEqual(self.proc1.used_as_step_by(self.db_conn),
  87                          [])
  88         self.assertEqual(self.proc2.used_as_step_by(self.db_conn),
  89                          [self.proc1])
  90         self.assertEqual(self.proc3.used_as_step_by(self.db_conn),
  91                          [self.proc1, self.proc2])
  92
  93     def test_Process_conditions(self) -> None:
  94         """Test setting Process.conditions/enables/disables."""
  95         for target in ('conditions', 'enables', 'disables'):
  96             c1 = Condition(None, False)
  97             c1.save(self.db_conn)
  98             assert isinstance(c1.id_, int)
  99             c2 = Condition(None, False)
 100             c2.save(self.db_conn)
 101             assert isinstance(c2.id_, int)
 102             self.proc1.set_conditions(self.db_conn, [], target)
 103             self.assertEqual(getattr(self.proc1, target), [])
 104             self.proc1.set_conditions(self.db_conn, [c1.id_], target)
 105             self.assertEqual(getattr(self.proc1, target), [c1])
 106             self.proc1.set_conditions(self.db_conn, [c2.id_], target)
 107             self.assertEqual(getattr(self.proc1, target), [c2])
 108             self.proc1.set_conditions(self.db_conn, [c1.id_, c2.id_], target)
 109             self.assertEqual(getattr(self.proc1, target), [c1, c2])
 110
 111     def test_Process_by_id(self) -> None:
 112         """Test Process.by_id()."""
 113         with self.assertRaises(NotFoundException):
 114             Process.by_id(self.db_conn, None, create=False)
 115         with self.assertRaises(NotFoundException):
 116             Process.by_id(self.db_conn, 0, create=False)
 117         self.assertNotEqual(self.proc1.id_,
 118                             Process.by_id(self.db_conn, None, create=True).id_)
 119         self.assertEqual(Process(2).id_,
 120                          Process.by_id(self.db_conn, 2, create=True).id_)
 121
 122     def test_Process_all(self) -> None:
 123         """Test Process.all()."""
 124         self.assertEqual({self.proc1.id_, self.proc2.id_, self.proc3.id_},
 125                          set(p.id_ for p in Process.all(self.db_conn)))
 126
 127     def test_ProcessStep_singularity(self) -> None:
 128         """Test pointers made for single object keep pointing to it."""
 129         assert isinstance(self.proc2.id_, int)
 130         self.proc1.set_steps(self.db_conn, [(None, self.proc2.id_, None)])
 131         step = self.proc1.explicit_steps[-1]
 132         assert isinstance(step.id_, int)
 133         step_retrieved = ProcessStep.by_id(self.db_conn, step.id_)
 134         step.parent_step_id = 99
 135         self.assertEqual(step.parent_step_id, step_retrieved.parent_step_id)
 136
 137     def test_Process_singularity(self) -> None:
 138         """Test pointers made for single object keep pointing to it, and
 139         subsequent retrievals don't overload relations."""
 140         assert isinstance(self.proc1.id_, int)
 141         assert isinstance(self.proc2.id_, int)
 142         c1 = Condition(None, False)
 143         c1.save(self.db_conn)
 144         assert isinstance(c1.id_, int)
 145         self.proc1.set_conditions(self.db_conn, [c1.id_])
 146         self.proc1.set_steps(self.db_conn, [(None, self.proc2.id_, None)])
 147         self.proc1.save(self.db_conn)
 148         p_retrieved = Process.by_id(self.db_conn, self.proc1.id_)
 149         self.assertEqual(self.proc1.explicit_steps, p_retrieved.explicit_steps)
 150         self.assertEqual(self.proc1.conditions, p_retrieved.conditions)
 151         self.proc1.save(self.db_conn)
 152
 153     def test_Process_versioned_attributes_singularity(self) -> None:
 154         """Test behavior of VersionedAttributes on saving (with .title)."""
 155         assert isinstance(self.proc1.id_, int)
 156         self.proc1.title.set('named')
 157         p_loaded = Process.by_id(self.db_conn, self.proc1.id_)
 158         self.assertEqual(self.proc1.title.history, p_loaded.title.history)
 159
 160
 161 class TestsWithServer(TestCaseWithServer):
 162     """Module tests against our HTTP server/handler (and database)."""
 163
 164     def test_do_POST_process(self) -> None:
 165         """Test POST /process and its effect on the database."""
 166         self.assertEqual(0, len(Process.all(self.db_conn)))
 167         form_data = {'title': 'foo', 'description': 'foo', 'effort': 1.1}
 168         self.check_post(form_data, '/process?id=', 302, '/')
 169         self.assertEqual(1, len(Process.all(self.db_conn)))
 170         self.check_post(form_data, '/process?id=FOO', 400)
 171         form_data['effort'] = 'foo'
 172         self.check_post(form_data, '/process?id=', 400)
 173         self.check_post({}, '/process?id=', 400)
 174         form_data = {'title': '', 'description': ''}
 175         self.check_post(form_data, '/process?id=', 400)
 176         form_data = {'title': '', 'effort': 1.1}
 177         self.check_post(form_data, '/process?id=', 400)
 178         form_data = {'description': '', 'effort': 1.0}
 179         self.check_post(form_data, '/process?id=', 400)
 180         self.assertEqual(1, len(Process.all(self.db_conn)))
 181         form_data = {'title': 'foo', 'description': 'foo', 'effort': 1.0,
 182                      'condition': []}
 183         self.check_post(form_data, '/process?id=', 302, '/')
 184         form_data['condition'] = [1]
 185         self.check_post(form_data, '/process?id=', 404)
 186         form_data_cond = {'title': 'foo', 'description': 'foo'}
 187         self.check_post(form_data_cond, '/condition', 302, '/')
 188         self.check_post(form_data, '/process?id=', 302, '/')
 189         form_data['disables'] = [1]
 190         self.check_post(form_data, '/process?id=', 302, '/')
 191         form_data['enables'] = [1]
 192         self.check_post(form_data, '/process?id=', 302, '/')
 193
 194     def test_do_GET(self) -> None:
 195         """Test /process and /processes response codes."""
 196         self.check_get('/process', 200)
 197         self.check_get('/process?id=', 200)
 198         self.check_get('/process?id=0', 500)
 199         self.check_get('/process?id=FOO', 400)
 200         self.check_get('/processes', 200)