"""Test Processes module."""
+from typing import Any
from tests.utils import TestCaseWithDB, TestCaseWithServer, TestCaseSansDB
from plomtask.processes import Process, ProcessStep, ProcessStepsNode
from plomtask.conditions import Condition
class TestsSansDB(TestCaseSansDB):
"""Module tests not requiring DB setup."""
checked_class = Process
- do_id_test = True
- versioned_defaults_to_test = {'title': 'UNNAMED', 'description': '',
- 'effort': 1.0}
class TestsSansDBProcessStep(TestCaseSansDB):
"""Module tests not requiring DB setup."""
checked_class = ProcessStep
- do_id_test = True
- default_init_args = [2, 3, 4]
+ default_init_kwargs = {'owner_id': 2, 'step_process_id': 3,
+ 'parent_step_id': 4}
class TestsWithDB(TestCaseWithDB):
"""Module tests requiring DB setup."""
checked_class = Process
- test_versioneds = {'title': str, 'description': str, 'effort': float}
def three_processes(self) -> tuple[Process, Process, Process]:
"""Return three saved processes."""
set_1 = [c1, c2]
set_2 = [c2, c3]
set_3 = [c1, c3]
- p.set_conditions(self.db_conn, [c.id_ for c in set_1
- if isinstance(c.id_, int)])
- p.set_enables(self.db_conn, [c.id_ for c in set_2
- if isinstance(c.id_, int)])
- p.set_disables(self.db_conn, [c.id_ for c in set_3
- if isinstance(c.id_, int)])
+ conds = [c.id_ for c in set_1 if isinstance(c.id_, int)]
+ enables = [c.id_ for c in set_2 if isinstance(c.id_, int)]
+ disables = [c.id_ for c in set_3 if isinstance(c.id_, int)]
+ p.set_condition_relations(self.db_conn, conds, [], enables, disables)
p.save(self.db_conn)
return p, set_1, set_2, set_3
"""Test setting Process.conditions/enables/disables."""
p = Process(None)
p.save(self.db_conn)
- for target in ('conditions', 'enables', 'disables'):
- method = getattr(p, f'set_{target}')
+ targets = ['conditions', 'blockers', 'enables', 'disables']
+ for i, target in enumerate(targets):
c1, c2 = Condition(None), Condition(None)
c1.save(self.db_conn)
c2.save(self.db_conn)
assert isinstance(c1.id_, int)
assert isinstance(c2.id_, int)
- method(self.db_conn, [])
+ args: list[list[int]] = [[], [], [], []]
+ args[i] = []
+ p.set_condition_relations(self.db_conn, *args)
self.assertEqual(getattr(p, target), [])
- method(self.db_conn, [c1.id_])
+ args[i] = [c1.id_]
+ p.set_condition_relations(self.db_conn, *args)
self.assertEqual(getattr(p, target), [c1])
- method(self.db_conn, [c2.id_])
+ args[i] = [c2.id_]
+ p.set_condition_relations(self.db_conn, *args)
self.assertEqual(getattr(p, target), [c2])
- method(self.db_conn, [c1.id_, c2.id_])
+ args[i] = [c1.id_, c2.id_]
+ p.set_condition_relations(self.db_conn, *args)
self.assertEqual(getattr(p, target), [c1, c2])
def test_remove(self) -> None:
'/process?id=', 400)
self.assertEqual(1, len(Process.all(self.db_conn)))
form_data = {'title': 'foo', 'description': 'foo', 'effort': 1.0}
- self.post_process(2, form_data | {'condition': []})
- self.check_post(form_data | {'condition': [1]}, '/process?id=', 404)
+ self.post_process(2, form_data | {'conditions': []})
+ self.check_post(form_data | {'conditions': [1]}, '/process?id=', 404)
self.check_post({'title': 'foo', 'description': 'foo',
'is_active': False},
'/condition', 302, '/condition?id=1')
- self.post_process(3, form_data | {'condition': [1]})
+ self.post_process(3, form_data | {'conditions': [1]})
self.post_process(4, form_data | {'disables': [1]})
self.post_process(5, form_data | {'enables': [1]})
form_data['delete'] = ''
self.post_process(1, form_data_1)
retrieved_process = Process.by_id(self.db_conn, 1)
self.assertEqual(len(retrieved_process.explicit_steps), 2)
- retrieved_step_0 = retrieved_process.explicit_steps[0]
+ retrieved_step_0 = retrieved_process.explicit_steps[1]
self.assertEqual(retrieved_step_0.step_process_id, 3)
self.assertEqual(retrieved_step_0.owner_id, 1)
self.assertEqual(retrieved_step_0.parent_step_id, None)
- retrieved_step_1 = retrieved_process.explicit_steps[1]
+ retrieved_step_1 = retrieved_process.explicit_steps[0]
self.assertEqual(retrieved_step_1.step_process_id, 2)
self.assertEqual(retrieved_step_1.owner_id, 1)
self.assertEqual(retrieved_step_1.parent_step_id, None)
self.post_process(1, form_data_1)
retrieved_process = Process.by_id(self.db_conn, 1)
self.assertEqual(len(retrieved_process.explicit_steps), 3)
- retrieved_step_0 = retrieved_process.explicit_steps[0]
+ retrieved_step_0 = retrieved_process.explicit_steps[1]
self.assertEqual(retrieved_step_0.step_process_id, 2)
self.assertEqual(retrieved_step_0.owner_id, 1)
self.assertEqual(retrieved_step_0.parent_step_id, None)
- retrieved_step_1 = retrieved_process.explicit_steps[1]
+ retrieved_step_1 = retrieved_process.explicit_steps[0]
self.assertEqual(retrieved_step_1.step_process_id, 3)
self.assertEqual(retrieved_step_1.owner_id, 1)
self.assertEqual(retrieved_step_1.parent_step_id, None)
def test_do_GET(self) -> None:
"""Test /process and /processes response codes."""
+ self.check_get('/process', 200)
+ self.check_get('/process?id=', 200)
+ self.check_get('/process?id=1', 200)
self.check_get_defaults('/process')
self.check_get('/processes', 200)
+
+ def test_fail_GET_process(self) -> None:
+ """Test invalid GET /process params."""
+ # check for invalid IDs
+ self.check_get('/process?id=foo', 400)
+ self.check_get('/process?id=0', 500)
+ # check we catch invalid base64
+ self.check_get('/process?title_b64=foo', 400)
+ # check failure on references to unknown processes; we create Process
+ # of ID=1 here so we know the 404 comes from step_to=2 etc. (that tie
+ # the Process displayed by /process to others), not from not finding
+ # the main Process itself
+ self.post_process(1)
+ self.check_get('/process?id=1&step_to=2', 404)
+ self.check_get('/process?id=1&has_step=2', 404)
+
+ @classmethod
+ def GET_processes_dict(cls, procs: list[dict[str, object]]
+ ) -> dict[str, object]:
+ """Return JSON of GET /processes to expect."""
+ library = {'Process': cls.as_refs(procs)} if procs else {}
+ d: dict[str, object] = {'processes': cls.as_id_list(procs),
+ 'sort_by': 'title',
+ 'pattern': '',
+ '_library': library}
+ return d
+
+ def test_GET_processes(self) -> None:
+ """Test GET /processes."""
+ # pylint: disable=too-many-statements
+ # test empty result on empty DB, default-settings on empty params
+ expected = self.GET_processes_dict([])
+ self.check_json_get('/processes', expected)
+ # test on meaningless non-empty params (incl. entirely un-used key),
+ # that 'sort_by' default to 'title' (even if set to something else, as
+ # long as without handler) and 'pattern' get preserved
+ expected['pattern'] = 'bar' # preserved despite zero effect!
+ url = '/processes?sort_by=foo&pattern=bar&foo=x'
+ self.check_json_get(url, expected)
+ # test non-empty result, automatic (positive) sorting by title
+ post1: dict[str, Any]
+ post2: dict[str, Any]
+ post3: dict[str, Any]
+ post1 = {'title': 'foo', 'description': 'oof', 'effort': 1.0}
+ post2 = {'title': 'bar', 'description': 'rab', 'effort': 1.1}
+ post2['new_top_step'] = 1
+ post3 = {'title': 'baz', 'description': 'zab', 'effort': 0.9}
+ post3['new_top_step'] = 1
+ self.post_process(1, post1)
+ self.post_process(2, post2)
+ self.post_process(3, post3)
+ post3['new_top_step'] = 2
+ post3['keep_step'] = 2
+ post3['steps'] = [2]
+ post3['step_2_process_id'] = 1
+ self.post_process(3, post3)
+ proc1 = self.proc_as_dict(1, post1['title'],
+ post1['description'], post1['effort'])
+ proc2 = self.proc_as_dict(2, post2['title'],
+ post2['description'], post2['effort'])
+ proc3 = self.proc_as_dict(3, post3['title'],
+ post3['description'], post3['effort'])
+ proc2['explicit_steps'] = [1]
+ proc3['explicit_steps'] = [2, 3]
+ step1 = self.procstep_as_dict(1, 2, 1)
+ step2 = self.procstep_as_dict(2, 3, 1)
+ step3 = self.procstep_as_dict(3, 3, 2)
+ expected = self.GET_processes_dict([proc2, proc3, proc1])
+ assert isinstance(expected['_library'], dict)
+ expected['_library']['ProcessStep'] = self.as_refs([step1, step2,
+ step3])
+ self.check_json_get('/processes', expected)
+ # test other sortings
+ expected['sort_by'] = '-title'
+ expected['processes'] = self.as_id_list([proc1, proc3, proc2])
+ self.check_json_get('/processes?sort_by=-title', expected)
+ expected['sort_by'] = 'effort'
+ expected['processes'] = self.as_id_list([proc3, proc1, proc2])
+ self.check_json_get('/processes?sort_by=effort', expected)
+ expected['sort_by'] = '-effort'
+ expected['processes'] = self.as_id_list([proc2, proc1, proc3])
+ self.check_json_get('/processes?sort_by=-effort', expected)
+ expected['sort_by'] = 'steps'
+ expected['processes'] = self.as_id_list([proc1, proc2, proc3])
+ self.check_json_get('/processes?sort_by=steps', expected)
+ expected['sort_by'] = '-steps'
+ expected['processes'] = self.as_id_list([proc3, proc2, proc1])
+ self.check_json_get('/processes?sort_by=-steps', expected)
+ expected['sort_by'] = 'owners'
+ expected['processes'] = self.as_id_list([proc3, proc2, proc1])
+ self.check_json_get('/processes?sort_by=owners', expected)
+ expected['sort_by'] = '-owners'
+ expected['processes'] = self.as_id_list([proc1, proc2, proc3])
+ self.check_json_get('/processes?sort_by=-owners', expected)
+ # test pattern matching on title
+ expected = self.GET_processes_dict([proc2, proc3])
+ assert isinstance(expected['_library'], dict)
+ expected['pattern'] = 'ba'
+ expected['_library']['ProcessStep'] = self.as_refs([step1, step2,
+ step3])
+ self.check_json_get('/processes?pattern=ba', expected)
+ # test pattern matching on description
+ expected['processes'] = self.as_id_list([proc1])
+ expected['_library'] = {'Process': self.as_refs([proc1])}
+ expected['pattern'] = 'of'
+ self.check_json_get('/processes?pattern=of', expected)