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

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