Slightly improve and re-organize Condition tests.

[plomtask] / tests / processes.py

diff --git a/tests/processes.py b/tests/processes.py
index 70090d15c5f9e8e49c5f64187f750d29d9d2d638..1b20e217d077d826765f5a83c9a2b3250de38ba2 100644 (file)
--- a/tests/processes.py
+++ b/tests/processes.py
@@ -1,4 +1,5 @@
 """Test Processes module."""
 """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
 from tests.utils import TestCaseWithDB, TestCaseWithServer, TestCaseSansDB
 from plomtask.processes import Process, ProcessStep, ProcessStepsNode
 from plomtask.conditions import Condition


@@ -9,7 +10,6 @@ from plomtask.todos import Todo
 class TestsSansDB(TestCaseSansDB):
     """Module tests not requiring DB setup."""
     checked_class = Process
 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}
 
     versioned_defaults_to_test = {'title': 'UNNAMED', 'description': '',
                                   'effort': 1.0}
 


@@ -17,7 +17,6 @@ class TestsSansDB(TestCaseSansDB):
 class TestsSansDBProcessStep(TestCaseSansDB):
     """Module tests not requiring DB setup."""
     checked_class = ProcessStep
 class TestsSansDBProcessStep(TestCaseSansDB):
     """Module tests not requiring DB setup."""
     checked_class = ProcessStep

-    do_id_test = True

     default_init_args = [2, 3, 4]
 
 
     default_init_args = [2, 3, 4]
 
 


@@ -58,17 +57,15 @@ class TestsWithDB(TestCaseWithDB):
     def test_Process_conditions_saving(self) -> None:
         """Test .save/.save_core."""
         p, set1, set2, set3 = self.p_of_conditions()
     def test_Process_conditions_saving(self) -> None:
         """Test .save/.save_core."""
         p, set1, set2, set3 = self.p_of_conditions()

+        assert p.id_ is not None

         r = Process.by_id(self.db_conn, p.id_)
         self.assertEqual(sorted(r.conditions), sorted(set1))
         self.assertEqual(sorted(r.enables), sorted(set2))
         self.assertEqual(sorted(r.disables), sorted(set3))
 
     def test_from_table_row(self) -> None:
         r = Process.by_id(self.db_conn, p.id_)
         self.assertEqual(sorted(r.conditions), sorted(set1))
         self.assertEqual(sorted(r.enables), sorted(set2))
         self.assertEqual(sorted(r.disables), sorted(set3))
 
     def test_from_table_row(self) -> None:

-        """Test .from_table_row() properly reads in class from DB"""
+        """Test .from_table_row() properly reads in class from DB."""

         super().test_from_table_row()
         super().test_from_table_row()

-        self.check_versioned_from_table_row('title', str)
-        self.check_versioned_from_table_row('description', str)
-        self.check_versioned_from_table_row('effort', float)

         p, set1, set2, set3 = self.p_of_conditions()
         p.save(self.db_conn)
         assert isinstance(p.id_, int)
         p, set1, set2, set3 = self.p_of_conditions()
         p.save(self.db_conn)
         assert isinstance(p.id_, int)


@@ -182,21 +179,9 @@ class TestsWithDB(TestCaseWithDB):
             method(self.db_conn, [c1.id_, c2.id_])
             self.assertEqual(getattr(p, target), [c1, c2])
 
             method(self.db_conn, [c1.id_, c2.id_])
             self.assertEqual(getattr(p, target), [c1, c2])
 

-    def test_Process_by_id(self) -> None:
-        """Test .by_id(), including creation"""
-        self.check_by_id()
-
-    def test_Process_all(self) -> None:
-        """Test .all()."""
-        self.check_all()
-
-    def test_Process_versioned_attributes_singularity(self) -> None:
-        """Test behavior of VersionedAttributes on saving (with .title)."""
-        self.check_versioned_singularity()
-
-    def test_Process_removal(self) -> None:
+    def test_remove(self) -> None:

         """Test removal of Processes and ProcessSteps."""
         """Test removal of Processes and ProcessSteps."""

-        self.check_remove()
+        super().test_remove()

         p1, p2, p3 = self.three_processes()
         assert isinstance(p1.id_, int)
         assert isinstance(p2.id_, int)
         p1, p2, p3 = self.three_processes()
         assert isinstance(p1.id_, int)
         assert isinstance(p2.id_, int)


@@ -208,13 +193,15 @@ class TestsWithDB(TestCaseWithDB):
             p1.remove(self.db_conn)
         p2.set_steps(self.db_conn, [])
         with self.assertRaises(NotFoundException):
             p1.remove(self.db_conn)
         p2.set_steps(self.db_conn, [])
         with self.assertRaises(NotFoundException):

+            assert step_id is not None

             ProcessStep.by_id(self.db_conn, step_id)
         p1.remove(self.db_conn)
         step = ProcessStep(None, p2.id_, p3.id_, None)
             ProcessStep.by_id(self.db_conn, step_id)
         p1.remove(self.db_conn)
         step = ProcessStep(None, p2.id_, p3.id_, None)

-        step_id = step.id_

         p2.set_steps(self.db_conn, [step])
         p2.set_steps(self.db_conn, [step])

+        step_id = step.id_

         p2.remove(self.db_conn)
         with self.assertRaises(NotFoundException):
         p2.remove(self.db_conn)
         with self.assertRaises(NotFoundException):

+            assert step_id is not None

             ProcessStep.by_id(self.db_conn, step_id)
         todo = Todo(None, p3, False, '2024-01-01')
         todo.save(self.db_conn)
             ProcessStep.by_id(self.db_conn, step_id)
         todo = Todo(None, p3, False, '2024-01-01')
         todo.save(self.db_conn)


@@ -227,33 +214,25 @@ class TestsWithDB(TestCaseWithDB):
 class TestsWithDBForProcessStep(TestCaseWithDB):
     """Module tests requiring DB setup."""
     checked_class = ProcessStep
 class TestsWithDBForProcessStep(TestCaseWithDB):
     """Module tests requiring DB setup."""
     checked_class = ProcessStep

-    default_init_kwargs = {'owner_id': 2, 'step_process_id': 3,
-                           'parent_step_id': 4}
+    default_init_kwargs = {'owner_id': 1, 'step_process_id': 2,
+                           'parent_step_id': 3}

 
     def setUp(self) -> None:
         super().setUp()
 
     def setUp(self) -> None:
         super().setUp()

-        p = Process(1)
-        p.save(self.db_conn)
-        p = Process(2)
-        p.save(self.db_conn)
+        self.p1 = Process(1)
+        self.p1.save(self.db_conn)

 
 

-    def test_saving_and_caching(self) -> None:
-        """Test storage and initialization of instances and attributes."""
-        self.check_saving_and_caching(id_=1, **self.default_init_kwargs)
-
-    def test_ProcessStep_remove(self) -> None:
+    def test_remove(self) -> None:

         """Test .remove and unsetting of owner's .explicit_steps entry."""
         """Test .remove and unsetting of owner's .explicit_steps entry."""

-        p1 = Process(None)
-        p2 = Process(None)
-        p1.save(self.db_conn)
+        p2 = Process(2)

         p2.save(self.db_conn)
         p2.save(self.db_conn)

-        assert isinstance(p1.id_, int)
+        assert isinstance(self.p1.id_, int)

         assert isinstance(p2.id_, int)
         assert isinstance(p2.id_, int)

-        step = ProcessStep(None, p1.id_, p2.id_, None)
-        p1.set_steps(self.db_conn, [step])
+        step = ProcessStep(None, self.p1.id_, p2.id_, None)
+        self.p1.set_steps(self.db_conn, [step])

         step.remove(self.db_conn)
         step.remove(self.db_conn)

-        self.assertEqual(p1.explicit_steps, [])
-        self.check_storage([])
+        self.assertEqual(self.p1.explicit_steps, [])
+        self.check_identity_with_cache_and_db([])

 
 
 class TestsWithServer(TestCaseWithServer):
 
 
 class TestsWithServer(TestCaseWithServer):


@@ -273,12 +252,12 @@ class TestsWithServer(TestCaseWithServer):
                         '/process?id=', 400)
         self.assertEqual(1, len(Process.all(self.db_conn)))
         form_data = {'title': 'foo', 'description': 'foo', 'effort': 1.0}
                         '/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.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(4, form_data | {'disables': [1]})
         self.post_process(5, form_data | {'enables': [1]})
         form_data['delete'] = ''


@@ -308,6 +287,7 @@ class TestsWithServer(TestCaseWithServer):
         self.post_process(1, form_data_1)
         retrieved_process = Process.by_id(self.db_conn, 1)
         self.assertEqual(retrieved_process.explicit_steps, [])
         self.post_process(1, form_data_1)
         retrieved_process = Process.by_id(self.db_conn, 1)
         self.assertEqual(retrieved_process.explicit_steps, [])

+        assert retrieved_step_id is not None

         with self.assertRaises(NotFoundException):
             ProcessStep.by_id(self.db_conn, retrieved_step_id)
         # post new first (top_level) step of process 3 to process 1
         with self.assertRaises(NotFoundException):
             ProcessStep.by_id(self.db_conn, retrieved_step_id)
         # post new first (top_level) step of process 3 to process 1


@@ -358,11 +338,11 @@ class TestsWithServer(TestCaseWithServer):
         self.post_process(1, form_data_1)
         retrieved_process = Process.by_id(self.db_conn, 1)
         self.assertEqual(len(retrieved_process.explicit_steps), 2)
         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)
         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.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)


@@ -390,11 +370,11 @@ class TestsWithServer(TestCaseWithServer):
         self.post_process(1, form_data_1)
         retrieved_process = Process.by_id(self.db_conn, 1)
         self.assertEqual(len(retrieved_process.explicit_steps), 3)
         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)
         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)
         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)


@@ -405,5 +385,126 @@ class TestsWithServer(TestCaseWithServer):
 
     def test_do_GET(self) -> None:
         """Test /process and /processes response codes."""
 
     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)
         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
+
+    @staticmethod
+    def procstep_as_dict(id_: int,
+                         owner_id: int,
+                         step_process_id: int,
+                         parent_step_id: int | None = None
+                         ) -> dict[str, object]:
+        """Return JSON of Process to expect."""
+        return {'id': id_,
+                'owner_id': owner_id,
+                'step_process_id': step_process_id,
+                'parent_step_id': parent_step_id}
+
+    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)