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
9 class TestsSansDB(TestCase):
10 """Module tests not requiring DB setup."""
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)
18 def test_Process_legal_ID(self) -> None:
19 """Test Process cannot be instantiated with id_=0."""
20 with self.assertRaises(HandledException):
24 class TestsWithDB(TestCaseWithDB):
25 """Mdule tests not requiring DB setup."""
27 def setUp(self) -> None:
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)
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_)
43 proc5.save(self.db_conn)
44 self.assertEqual(proc5.id_,
45 Process.by_id(self.db_conn, 5, create=False).id_)
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,
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_),
75 step_3 = ProcessStepsNode(self.proc3, None, False, {}, True)
76 p_1_dict[2].steps[4] = ProcessStepsNode(self.proc2, step_2.id_, True,
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),
86 self.assertEqual(self.proc1.used_as_step_by(self.db_conn),
88 self.assertEqual(self.proc2.used_as_step_by(self.db_conn),
90 self.assertEqual(self.proc3.used_as_step_by(self.db_conn),
91 [self.proc1, self.proc2])
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)
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])
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_)
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)))
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)
137 def test_Process_singularity(self) -> None:
138 """Test pointers made for single object keep pointing to it."""
139 assert isinstance(self.proc1.id_, int)
140 assert isinstance(self.proc2.id_, int)
141 self.proc1.set_steps(self.db_conn, [(None, self.proc2.id_, None)])
142 p_retrieved = Process.by_id(self.db_conn, self.proc1.id_)
143 self.assertEqual(self.proc1.explicit_steps, p_retrieved.explicit_steps)
145 def test_Process_versioned_attributes_singularity(self) -> None:
146 """Test behavior of VersionedAttributes on saving (with .title)."""
147 assert isinstance(self.proc1.id_, int)
148 self.proc1.title.set('named')
149 p_loaded = Process.by_id(self.db_conn, self.proc1.id_)
150 self.assertEqual(self.proc1.title.history, p_loaded.title.history)
153 class TestsWithServer(TestCaseWithServer):
154 """Module tests against our HTTP server/handler (and database)."""
156 def test_do_POST_process(self) -> None:
157 """Test POST /process and its effect on the database."""
158 self.assertEqual(0, len(Process.all(self.db_conn)))
159 form_data = {'title': 'foo', 'description': 'foo', 'effort': 1.1}
160 self.check_post(form_data, '/process?id=', 302, '/')
161 self.assertEqual(1, len(Process.all(self.db_conn)))
162 self.check_post(form_data, '/process?id=FOO', 400)
163 form_data['effort'] = 'foo'
164 self.check_post(form_data, '/process?id=', 400)
165 self.check_post({}, '/process?id=', 400)
166 form_data = {'title': '', 'description': ''}
167 self.check_post(form_data, '/process?id=', 400)
168 form_data = {'title': '', 'effort': 1.1}
169 self.check_post(form_data, '/process?id=', 400)
170 form_data = {'description': '', 'effort': 1.0}
171 self.check_post(form_data, '/process?id=', 400)
172 self.assertEqual(1, len(Process.all(self.db_conn)))
173 form_data = {'title': 'foo', 'description': 'foo', 'effort': 1.0,
175 self.check_post(form_data, '/process?id=', 302, '/')
176 form_data['condition'] = [1]
177 self.check_post(form_data, '/process?id=', 404)
178 form_data_cond = {'title': 'foo', 'description': 'foo'}
179 self.check_post(form_data_cond, '/condition', 302, '/')
180 self.check_post(form_data, '/process?id=', 302, '/')
181 form_data['disables'] = [1]
182 self.check_post(form_data, '/process?id=', 302, '/')
183 form_data['enables'] = [1]
184 self.check_post(form_data, '/process?id=', 302, '/')
186 def test_do_GET(self) -> None:
187 """Test /process and /processes response codes."""
188 self.check_get('/process', 200)
189 self.check_get('/process?id=', 200)
190 self.check_get('/process?id=0', 500)
191 self.check_get('/process?id=FOO', 400)
192 self.check_get('/processes', 200)