From: Christian Heller <c.heller@plomlompom.de>
Date: Mon, 5 Aug 2024 01:53:48 +0000 (+0200)
Subject: Refactor and extend tests, especially for Processes.
X-Git-Url: https://plomlompom.com/repos/%22https:/validator.w3.org/do_todos?a=commitdiff_plain;h=2f2543afed5c83d1759e02c1464c14de13f25ecf;p=plomtask

Refactor and extend tests, especially for Processes.
---

diff --git a/tests/processes.py b/tests/processes.py
index 8ce3312..45497a6 100644
--- a/tests/processes.py
+++ b/tests/processes.py
@@ -239,6 +239,42 @@ class TestsWithDBForProcessStep(TestCaseWithDB):
         self.check_identity_with_cache_and_db([])
 
 
+class ExpectedGetProcess(Expected):
+    """Builder of expectations for GET /processes."""
+    _default_dict = {'is_new': False, 'preset_top_step': None, 'n_todos': 0}
+    _on_empty_make_temp = ('Process', 'proc_as_dict')
+
+    def __init__(self,
+                 proc_id: int,
+                 *args: Any, **kwargs: Any) -> None:
+        self._fields = {'process': proc_id, 'steps': []}
+        super().__init__(*args, **kwargs)
+
+    @staticmethod
+    def stepnode_as_dict(step_id: int,
+                         seen: bool = False,
+                         steps: None | list[dict[str, object]] = None,
+                         is_explicit: bool = True,
+                         is_suppressed: bool = False) -> dict[str, object]:
+        """Return JSON of ProcessStepNode to expect."""
+        return {'step': step_id,
+                'seen': seen,
+                'steps': steps if steps else [],
+                'is_explicit': is_explicit,
+                'is_suppressed': is_suppressed}
+
+    def recalc(self) -> None:
+        """Update internal dictionary by subclass-specific rules."""
+        super().recalc()
+        self._fields['process_candidates'] = self.as_ids(
+                self.lib_all('Process'))
+        self._fields['condition_candidates'] = self.as_ids(
+                self.lib_all('Condition'))
+        self._fields['owners'] = [
+                s['owner_id'] for s in self.lib_all('ProcessStep')
+                if s['step_process_id'] == self._fields['process']]
+
+
 class ExpectedGetProcesses(Expected):
     """Builder of expectations for GET /processes."""
     _default_dict = {'sort_by': 'title', 'pattern': ''}
@@ -262,152 +298,122 @@ class TestsWithServer(TestCaseWithServer):
                         redir=f'/process?id={id_}')
         return form_data
 
-    def test_do_POST_process(self) -> None:
+    def test_fail_POST_process(self) -> None:
         """Test POST /process and its effect on the database."""
-        self.assertEqual(0, len(Process.all(self.db_conn)))
-        form_data = self._post_process()
-        self.assertEqual(1, len(Process.all(self.db_conn)))
-        self.check_post(form_data, '/process?id=FOO', 400)
-        self.check_post(form_data | {'effort': 'foo'}, '/process?id=', 400)
-        self.check_post({}, '/process?id=', 400)
-        self.check_post({'title': '', 'description': ''}, '/process?id=', 400)
-        self.check_post({'title': '', 'effort': 1.1}, '/process?id=', 400)
-        self.check_post({'description': '', '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 | {'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 | {'conditions': [1]})
-        self._post_process(4, form_data | {'disables': [1]})
-        self._post_process(5, form_data | {'enables': [1]})
-        form_data['delete'] = ''
-        self.check_post(form_data, '/process?id=', 404)
-        self.check_post(form_data, '/process?id=6', 404)
-        self.check_post(form_data, '/process?id=5', 302, '/processes')
-
-    def test_do_POST_process_steps(self) -> None:
+        valid_post = {'title': '', 'description': '', 'effort': 1.0}
+        # check payloads lacking minimum expecteds
+        self.check_post({}, '/process', 400)
+        self.check_post({'title': '', 'description': ''}, '/process', 400)
+        self.check_post({'title': '', 'effort': 1}, '/process', 400)
+        self.check_post({'description': '', 'effort': 1}, '/process', 400)
+        # check payloads of bad data types
+        self.check_post(valid_post | {'effort': ''}, '/process', 400)
+        # check references to non-existant items
+        self.check_post(valid_post | {'conditions': [1]}, '/process', 404)
+        self.check_post(valid_post | {'disables': [1]}, '/process', 404)
+        self.check_post(valid_post | {'blockers': [1]}, '/process', 404)
+        self.check_post(valid_post | {'enables': [1]}, '/process', 404)
+        self.check_post(valid_post | {'new_top_step': 2}, '/process', 404)
+        # check deletion of non-existant
+        self.check_post({'delete': ''}, '/process?id=1', 404)
+
+    def test_basic_POST_process(self) -> None:
+        """Test basic GET/POST /process operations."""
+        # check on un-saved
+        exp = ExpectedGetProcess(1)
+        exp.force('process_candidates', [])
+        self.check_json_get('/process?id=1', exp)
+        # check on minimal payload post
+        valid_post = {'title': 'foo', 'description': 'oof', 'effort': 2.3}
+        exp.unforce('process_candidates')
+        self.post_exp_process([exp], valid_post, 1)
+        self.check_json_get('/process?id=1', exp)
+        # check n_todos field
+        self.post_exp_day([], {'new_todo': ['1']}, '2024-01-01')
+        self.post_exp_day([], {'new_todo': ['1']}, '2024-01-02')
+        exp.set('n_todos', 2)
+        self.check_json_get('/process?id=1', exp)
+        # check cannot delete if Todos to Process
+        self.check_post({'delete': ''}, '/process?id=1', 500)
+        # check successful deletion
+        self.post_exp_process([], valid_post, 2)
+        self.check_post({'delete': ''}, '/process?id=2', 302, '/processes')
+        exp = ExpectedGetProcess(2)
+        exp.set_proc_from_post(1, valid_post)
+        exp.force('process_candidates', [1])
+        self.check_json_get('/process?id=2', exp)
+
+    def test_POST_process_steps(self) -> None:
         """Test behavior of ProcessStep posting."""
         # pylint: disable=too-many-statements
-        form_data_1 = self._post_process(1)
-        self._post_process(2)
-        self._post_process(3)
-        # post first (top-level) step of process 2 to process 1
-        form_data_1['new_top_step'] = [2]
-        self._post_process(1, form_data_1)
-        retrieved_process = Process.by_id(self.db_conn, 1)
-        self.assertEqual(len(retrieved_process.explicit_steps), 1)
-        retrieved_step = retrieved_process.explicit_steps[0]
-        retrieved_step_id = retrieved_step.id_
-        self.assertEqual(retrieved_step.step_process_id, 2)
-        self.assertEqual(retrieved_step.owner_id, 1)
-        self.assertEqual(retrieved_step.parent_step_id, None)
-        # post empty steps list to process, expect clean slate, and old step to
-        # completely disappear
-        form_data_1['new_top_step'] = []
-        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
-        form_data_1['new_top_step'] = [3]
-        self._post_process(1, form_data_1)
-        retrieved_process = Process.by_id(self.db_conn, 1)
-        retrieved_step = retrieved_process.explicit_steps[0]
-        self.assertEqual(retrieved_step.step_process_id, 3)
-        self.assertEqual(retrieved_step.owner_id, 1)
-        self.assertEqual(retrieved_step.parent_step_id, None)
-        # post to process steps list without keeps, expect clean slate
-        form_data_1['new_top_step'] = []
-        form_data_1['steps'] = [retrieved_step.id_]
-        self._post_process(1, form_data_1)
-        retrieved_process = Process.by_id(self.db_conn, 1)
-        self.assertEqual(retrieved_process.explicit_steps, [])
-        # post to process empty steps list but keep, expect 400
-        form_data_1['steps'] = []
-        form_data_1['keep_step'] = [retrieved_step_id]
-        self.check_post(form_data_1, '/process?id=1', 400, '/process?id=1')
-        # post to process steps list with keep on non-created step, expect 400
-        form_data_1['steps'] = [retrieved_step_id]
-        form_data_1['keep_step'] = [retrieved_step_id]
-        self.check_post(form_data_1, '/process?id=1', 400, '/process?id=1')
-        # post to process steps list with keep and process ID, expect 200
-        form_data_1[f'step_{retrieved_step_id}_process_id'] = [2]
-        self._post_process(1, form_data_1)
-        retrieved_process = Process.by_id(self.db_conn, 1)
-        self.assertEqual(len(retrieved_process.explicit_steps), 1)
-        retrieved_step = retrieved_process.explicit_steps[0]
-        self.assertEqual(retrieved_step.step_process_id, 2)
-        self.assertEqual(retrieved_step.owner_id, 1)
-        self.assertEqual(retrieved_step.parent_step_id, None)
-        # post nonsense, expect 400 and preservation of previous state
-        form_data_1['steps'] = ['foo']
-        form_data_1['keep_step'] = []
-        self.check_post(form_data_1, '/process?id=1', 400, '/process?id=1')
-        retrieved_process = Process.by_id(self.db_conn, 1)
-        self.assertEqual(len(retrieved_process.explicit_steps), 1)
-        retrieved_step = retrieved_process.explicit_steps[0]
-        self.assertEqual(retrieved_step.step_process_id, 2)
-        self.assertEqual(retrieved_step.owner_id, 1)
-        self.assertEqual(retrieved_step.parent_step_id, None)
-        # post to process steps list with keep and process ID, expect 200
-        form_data_1['new_top_step'] = [3]
-        form_data_1['steps'] = [retrieved_step.id_]
-        form_data_1['keep_step'] = [retrieved_step.id_]
-        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[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[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)
-        # post to process steps list with keeps etc., but trigger recursion
-        form_data_1['new_top_step'] = []
-        form_data_1['steps'] = [retrieved_step_0.id_, retrieved_step_1.id_]
-        form_data_1['keep_step'] = [retrieved_step_0.id_, retrieved_step_1.id_]
-        form_data_1[f'step_{retrieved_step_0.id_}_process_id'] = [2]
-        form_data_1[f'step_{retrieved_step_1.id_}_process_id'] = [1]
-        self.check_post(form_data_1, '/process?id=1', 400, '/process?id=1')
-        # check previous status preserved despite failed steps setting
-        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]
-        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]
-        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)
-        # post sub-step to step
-        form_data_1[f'step_{retrieved_step_0.id_}_process_id'] = [3]
-        form_data_1[f'new_step_to_{retrieved_step_0.id_}'] = [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[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[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)
-        retrieved_step_2 = retrieved_process.explicit_steps[2]
-        self.assertEqual(retrieved_step_2.step_process_id, 3)
-        self.assertEqual(retrieved_step_2.owner_id, 1)
-        self.assertEqual(retrieved_step_2.parent_step_id, retrieved_step_1.id_)
-
-    def test_do_GET(self) -> None:
+        exp = ExpectedGetProcess(1)
+        self.post_exp_process([exp], {}, 1)
+        # post first (top-level) step of proc 2 to proc 1 by 'new_top_step'
+        self.post_exp_process([exp], {}, 2)
+        self.post_exp_process([exp], {'new_top_step': 2}, 1)
+        exp.lib_set('ProcessStep', [exp.procstep_as_dict(1, 1, 2)])
+        exp.set('steps', [exp.stepnode_as_dict(1)])
+        self.check_json_get('/process?id=1', exp)
+        # post empty/absent steps list to process, expect clean slate, and old
+        # step to completely disappear
+        self.post_exp_process([exp], {}, 1)
+        exp.lib_wipe('ProcessStep')
+        exp.set('steps', [])
+        self.check_json_get('/process?id=1', exp)
+        # post new step of proc2 to proc1 by 'keep_step', 'steps', and its deps
+        post = {'step_1_process_id': 2, 'step_1_parent_id': '', 'steps': [1]}
+        self.post_exp_process([exp], post | {'keep_step': [1]}, 1)
+        exp.lib_set('ProcessStep', [exp.procstep_as_dict(1, 1, 2)])
+        exp.set('steps', [exp.stepnode_as_dict(1)])
+        self.check_json_get('/process?id=1', exp)
+        # fail with invalid or missing 'steps' dependencies
+        p_min = {'title': '', 'description': '', 'effort': 0}
+        p = p_min | {'step_1_process_id': 2, 'steps': [1], 'keep_step': [1]}
+        self.check_post(p | {'step_1_parent_id': 'foo'}, '/process?id=1', 400)
+        p = p_min | {'step_1_parent_id': '', 'steps': [1], 'keep_step': [1]}
+        self.check_post(p | {'step_1_process_id': 'foo'}, '/process?id=1', 400)
+        self.check_post(p, '/process?id=1', 400)
+        # treat valid steps data without 'keep_step' like posting emptiness
+        p = {'step_1_process_id': 2, 'step_1_parent_id': '', 'steps': [1]}
+        self.post_exp_process([exp], post, 1)
+        exp.lib_wipe('ProcessStep')
+        exp.set('steps', [])
+        self.check_json_get('/process?id=1', exp)
+        # fail when refering in 'keep_step' to step not in 'steps'
+        p = p_min | p
+        self.check_post(p | {'keep_step': [2]}, '/process?id=1', 400)
+        # expect failure independently of what parent/proc_id keys exist
+        p['step_2_process_id'], p['step_2_parent_id'] = 3, None
+        self.check_post(p | {'keep_step': [2]}, '/process?id=1', 400)
+        # fail on single- and multi-step recursion
+        self.check_post(p | {'new_top_step': 1}, '/process?id=1', 400)
+        self.post_exp_process([exp], {'new_top_step': 1}, 2)
+        self.check_post(p | {'new_top_step': 2}, '/process?id=1', 400)
+        # post sibling steps
+        self.post_exp_process([exp], {}, 3)
+        p = {'step_1_process_id': 3, 'steps': [1], 'keep_step': [1]}
+        self.post_exp_process([exp], p | {'new_top_step': 3}, 1)
+        exp.lib_set('ProcessStep', [exp.procstep_as_dict(1, 1, 3),
+                                    exp.procstep_as_dict(2, 1, 3)])
+        exp.set('steps', [exp.stepnode_as_dict(1), exp.stepnode_as_dict(2)])
+        self.check_json_get('/process?id=1', exp)
+        # post sub-step chain
+        p = {'step_1_process_id': 3, 'step_2_process_id': 3,
+             'steps': [1, 2], 'keep_step': [1, 2]}
+        self.post_exp_process([exp], p | {'new_step_to_2': 3}, 1)
+        exp.lib_set('ProcessStep', [exp.procstep_as_dict(3, 1, 3, 2)])
+        exp.set('steps', [exp.stepnode_as_dict(1),
+                          exp.stepnode_as_dict(2, steps=[
+                              exp.stepnode_as_dict(3)])])
+        self.check_json_get('/process?id=1', exp)
+        p['steps'], p['keep_step'] = [1, 2, 3], [1, 2, 3]
+        p['step_3_process_id'] = 3
+        # fail post sub-step that would cause recursion
+        self.post_exp_process([exp], {'new_top_step': 1}, 2)
+        exp.lib_set('ProcessStep', [exp.procstep_as_dict(4, 2, 1)])
+        self.check_post(p_min | p | {'new_step_to_2': 2}, '/process?id=1', 400)
+
+    def test_GET(self) -> None:
         """Test /process and /processes response codes."""
         self.check_get('/process', 200)
         self.check_get('/process?id=', 200)
diff --git a/tests/utils.py b/tests/utils.py
index f256345..3d86163 100644
--- a/tests/utils.py
+++ b/tests/utils.py
@@ -1,4 +1,5 @@
 """Shared test utilities."""
+# pylint: disable=too-many-lines
 from __future__ import annotations
 from unittest import TestCase
 from typing import Mapping, Any, Callable
@@ -342,7 +343,7 @@ class TestCaseWithDB(TestCaseAugmented):
         obj2.save(self.db_conn)
         self.assertEqual(self.checked_class.get_cache(), {id1: obj2})
         # NB: we'll only compare hashes because obj2 itself disappears on
-        # .from_table_row-trioggered database reload
+        # .from_table_row-triggered database reload
         obj2_hash = hash(obj2)
         found_in_db += self._load_from_db(id1)
         self.assertEqual([hash(o) for o in found_in_db], [obj2_hash])
@@ -628,6 +629,10 @@ class Expected:
         """Set ._forced field to ensure value in .as_dict."""
         self._forced[field_name] = value
 
+    def unforce(self, field_name: str) -> None:
+        """Unset ._forced field."""
+        del self._forced[field_name]
+
     @staticmethod
     def _as_refs(items: list[dict[str, object]]
                  ) -> dict[str, dict[str, object]]:
@@ -699,14 +704,13 @@ class Expected:
         if cond:
             cond['is_active'] = d['is_active']
             for category in ['title', 'description']:
-                if category in cond['_versioned']:
-                    history = cond['_versioned'][category]
-                    if len(history) > 0:
-                        last_i = sorted([int(k) for k in history.keys()])[-1]
-                        if d[category] != history[str(last_i)]:
-                            history[str(last_i + 1)] = d[category]
-                        continue
-                cond['_versioned'][category]['0'] = d[category]
+                history = cond['_versioned'][category]
+                if len(history) > 0:
+                    last_i = sorted([int(k) for k in history.keys()])[-1]
+                    if d[category] != history[str(last_i)]:
+                        history[str(last_i + 1)] = d[category]
+                else:
+                    history['0'] = d[category]
         else:
             cond = self.cond_as_dict(
                     id_, d['is_active'], d['title'], d['description'])
@@ -776,9 +780,9 @@ class Expected:
 
     @staticmethod
     def proc_as_dict(id_: int = 1,
-                     title: str = 'A',
-                     description: str = '',
-                     effort: float = 1.0,
+                     title: None | str = None,
+                     description: None | str = None,
+                     effort: None | float = None,
                      conditions: None | list[int] = None,
                      disables: None | list[int] = None,
                      blockers: None | list[int] = None,
@@ -787,14 +791,19 @@ class Expected:
                      ) -> dict[str, object]:
         """Return JSON of Process to expect."""
         # pylint: disable=too-many-arguments
+        versioned: dict[str, dict[str, object]]
+        versioned = {'title': {}, 'description': {}, 'effort': {}}
+        if title is not None:
+            versioned['title']['0'] = title
+        if description is not None:
+            versioned['description']['0'] = description
+        if effort is not None:
+            versioned['effort']['0'] = effort
         d = {'id': id_,
              'calendarize': False,
              'suppressed_steps': [],
              'explicit_steps': explicit_steps if explicit_steps else [],
-             '_versioned': {
-                 'title': {'0': title},
-                 'description': {'0': description},
-                 'effort': {'0': effort}},
+             '_versioned': versioned,
              'conditions': conditions if conditions else [],
              'disables': disables if disables else [],
              'enables': enables if enables else [],
@@ -805,17 +814,22 @@ class Expected:
         """Set Process of id_ in library based on POST dict d."""
         proc = self.lib_get('Process', id_)
         if proc:
-            for k in ['title', 'description', 'effort']:
-                last_i = sorted(proc['_versioned'][k].keys())[-1]
-                if d[k] != proc['_versioned'][k][last_i]:
-                    proc['_versioned'][k][last_i + 1] = d[k]
+            for category in ['title', 'description', 'effort']:
+                history = proc['_versioned'][category]
+                if len(history) > 0:
+                    last_i = sorted([int(k) for k in history.keys()])[-1]
+                    if d[category] != history[str(last_i)]:
+                        history[str(last_i + 1)] = d[category]
+                else:
+                    history['0'] = d[category]
         else:
             proc = self.proc_as_dict(id_,
                                      d['title'], d['description'], d['effort'])
         ignore = {'title', 'description', 'effort', 'new_top_step', 'step_of',
                   'keep_step', 'steps'}
         for k, v in d.items():
-            if k in ignore or k.startswith('step_'):
+            if k in ignore\
+                    or k.startswith('step_') or k.startswith('new_step_to'):
                 continue
             if k in {'calendarize'}:
                 v = True
@@ -932,8 +946,8 @@ class TestCaseWithServer(TestCaseWithDB):
         """Compare JSON on GET path with expected.
 
         To simplify comparison of VersionedAttribute histories, transforms
-        timestamp keys of VersionedAttribute history keys into integers
-        counting chronologically forward from 0.
+        timestamp keys of VersionedAttribute history keys into (strings of)
+        integers counting chronologically forward from 0.
         """
 
         def rewrite_history_keys_in(item: Any) -> Any: