home · contact · privacy
Refactor Process/ProcessStep setting and saving.
[plomtask] / 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, BadFormatException
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(BadFormatException):
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 self.proc2.id_ is not None
58         assert self.proc3.id_ is not None
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         p_1_dict[2] = {
69             'process': self.proc3, 'parent_id': None,
70             'is_explicit': True, 'steps': {}, 'seen': False
71         }
72         self.assertEqual(self.proc1.get_steps(self.db_conn, None), p_1_dict)
73         steps_proc2: list[tuple[int | None, int, int | None]] = []
74         add_step(self.proc2, steps_proc2, (None, self.proc3.id_, None), 3)
75         p_1_dict[1]['steps'] = {3: {
76             'process': self.proc3, 'parent_id': None,
77             'is_explicit': False, 'steps': {}, 'seen': False
78         }}
79         self.assertEqual(self.proc1.get_steps(self.db_conn, None), p_1_dict)
80         add_step(self.proc1, steps_proc1, (None, self.proc2.id_, step_2.id_),
81                  4)
82         p_1_dict[2]['steps'][4] = {
83             'process': self.proc2, 'parent_id': step_2.id_, 'seen': False,
84             'is_explicit': True, 'steps': {3: {
85                 'process': self.proc3, 'parent_id': None,
86                 'is_explicit': False, 'steps': {}, 'seen': True
87                 }}}
88         self.assertEqual(self.proc1.get_steps(self.db_conn, None), p_1_dict)
89         add_step(self.proc1, steps_proc1, (None, self.proc3.id_, 999), 5)
90         p_1_dict[5] = {
91             'process': self.proc3, 'parent_id': None,
92             'is_explicit': True, 'steps': {}, 'seen': False
93         }
94         self.assertEqual(self.proc1.get_steps(self.db_conn, None), p_1_dict)
95         add_step(self.proc1, steps_proc1, (None, self.proc3.id_, 3), 6)
96         p_1_dict[6] = {
97             'process': self.proc3, 'parent_id': None,
98             'is_explicit': True, 'steps': {}, 'seen': False
99         }
100         self.assertEqual(self.proc1.get_steps(self.db_conn, None),
101                          p_1_dict)
102         self.assertEqual(self.proc1.used_as_step_by(self.db_conn),
103                          [])
104         self.assertEqual(self.proc2.used_as_step_by(self.db_conn),
105                          [self.proc1])
106         self.assertEqual(self.proc3.used_as_step_by(self.db_conn),
107                          [self.proc1, self.proc2])
108
109     def test_Process_conditions(self) -> None:
110         """Test setting Process.conditions/fulfills/undoes."""
111         for target in ('conditions', 'fulfills', 'undoes'):
112             c1 = Condition(None, False)
113             c1.save(self.db_conn)
114             assert c1.id_ is not None
115             c2 = Condition(None, False)
116             c2.save(self.db_conn)
117             assert c2.id_ is not None
118             self.proc1.set_conditions(self.db_conn, [], target)
119             self.assertEqual(getattr(self.proc1, target), [])
120             self.proc1.set_conditions(self.db_conn, [c1.id_], target)
121             self.assertEqual(getattr(self.proc1, target), [c1])
122             self.proc1.set_conditions(self.db_conn, [c2.id_], target)
123             self.assertEqual(getattr(self.proc1, target), [c2])
124             self.proc1.set_conditions(self.db_conn, [c1.id_, c2.id_], target)
125             self.assertEqual(getattr(self.proc1, target), [c1, c2])
126
127     def test_Process_by_id(self) -> None:
128         """Test Process.by_id()."""
129         with self.assertRaises(NotFoundException):
130             Process.by_id(self.db_conn, None, create=False)
131         with self.assertRaises(NotFoundException):
132             Process.by_id(self.db_conn, 0, create=False)
133         self.assertNotEqual(self.proc1.id_,
134                             Process.by_id(self.db_conn, None, create=True).id_)
135         self.assertEqual(Process(2).id_,
136                          Process.by_id(self.db_conn, 2, create=True).id_)
137
138     def test_Process_all(self) -> None:
139         """Test Process.all()."""
140         self.assertEqual({self.proc1.id_, self.proc2.id_, self.proc3.id_},
141                          set(p.id_ for p in Process.all(self.db_conn)))
142
143     def test_ProcessStep_singularity(self) -> None:
144         """Test pointers made for single object keep pointing to it."""
145         assert self.proc2.id_ is not None
146         self.proc1.set_steps(self.db_conn, [(None, self.proc2.id_, None)])
147         step = self.proc1.explicit_steps[-1]
148         assert step.id_ is not None
149         step_retrieved = ProcessStep.by_id(self.db_conn, step.id_)
150         step.parent_step_id = 99
151         self.assertEqual(step.parent_step_id, step_retrieved.parent_step_id)
152
153     def test_Process_singularity(self) -> None:
154         """Test pointers made for single object keep pointing to it."""
155         assert self.proc2.id_ is not None
156         self.proc1.set_steps(self.db_conn, [(None, self.proc2.id_, None)])
157         p_retrieved = Process.by_id(self.db_conn, self.proc1.id_)
158         self.assertEqual(self.proc1.explicit_steps, p_retrieved.explicit_steps)
159
160     def test_Process_versioned_attributes_singularity(self) -> None:
161         """Test behavior of VersionedAttributes on saving (with .title)."""
162         self.proc1.title.set('named')
163         p_loaded = Process.by_id(self.db_conn, self.proc1.id_)
164         self.assertEqual(self.proc1.title.history, p_loaded.title.history)
165
166
167 class TestsWithServer(TestCaseWithServer):
168     """Module tests against our HTTP server/handler (and database)."""
169
170     def test_do_POST_process(self) -> None:
171         """Test POST /process and its effect on the database."""
172         self.assertEqual(0, len(Process.all(self.db_conn)))
173         form_data = {'title': 'foo', 'description': 'foo', 'effort': 1.1}
174         self.check_post(form_data, '/process?id=', 302, '/')
175         self.assertEqual(1, len(Process.all(self.db_conn)))
176         self.check_post(form_data, '/process?id=FOO', 400)
177         form_data['effort'] = 'foo'
178         self.check_post(form_data, '/process?id=', 400)
179         self.check_post({}, '/process?id=', 400)
180         form_data = {'title': '', 'description': ''}
181         self.check_post(form_data, '/process?id=', 400)
182         form_data = {'title': '', 'effort': 1.1}
183         self.check_post(form_data, '/process?id=', 400)
184         form_data = {'description': '', 'effort': 1.0}
185         self.check_post(form_data, '/process?id=', 400)
186         self.assertEqual(1, len(Process.all(self.db_conn)))
187         form_data = {'title': 'foo', 'description': 'foo', 'effort': 1.0,
188                      'condition': []}
189         self.check_post(form_data, '/process?id=', 302, '/')
190         form_data['condition'] = [1]
191         self.check_post(form_data, '/process?id=', 404)
192         form_data_cond = {'title': 'foo', 'description': 'foo'}
193         self.check_post(form_data_cond, '/condition', 302, '/')
194         self.check_post(form_data, '/process?id=', 302, '/')
195         form_data['undoes'] = [1]
196         self.check_post(form_data, '/process?id=', 302, '/')
197         form_data['fulfills'] = [1]
198         self.check_post(form_data, '/process?id=', 302, '/')
199
200     def test_do_GET(self) -> None:
201         """Test /process and /processes response codes."""
202         self.check_get('/process', 200)
203         self.check_get('/process?id=', 200)
204         self.check_get('/process?id=0', 400)
205         self.check_get('/process?id=FOO', 400)
206         self.check_get('/processes', 200)