home · contact · privacy
Set 'is_new' even when provided not-yet-existing ID, adapt and fix tests.
authorChristian Heller <c.heller@plomlompom.de>
Mon, 12 Aug 2024 09:11:54 +0000 (11:11 +0200)
committerChristian Heller <c.heller@plomlompom.de>
Mon, 12 Aug 2024 09:11:54 +0000 (11:11 +0200)
plomtask/db.py
plomtask/http.py
tests/conditions.py
tests/processes.py
tests/todos.py
tests/utils.py

index ee5f3b99f04e66460ef4e0a90c5d10d600210881..1fdd3e1c21d96804b37611d13664b89ace333535 100644 (file)
@@ -376,7 +376,8 @@ class BaseModel(Generic[BaseModelId]):
 
     @classmethod
     def _get_cached(cls: type[BaseModelInstance],
-                    id_: BaseModelId) -> BaseModelInstance | None:
+                    id_: BaseModelId
+                    ) -> BaseModelInstance | None:
         """Get object of id_ from class's cache, or None if not found."""
         cache = cls.get_cache()
         if id_ in cache:
@@ -449,7 +450,7 @@ class BaseModel(Generic[BaseModelId]):
     def by_id_or_create(cls, db_conn: DatabaseConnection,
                         id_: BaseModelId | None
                         ) -> Self:
-        """Wrapper around .by_id, creating (not caching/saving) if not find."""
+        """Wrapper around .by_id, creating (not caching/saving) if no find."""
         if not cls.can_create_by_id:
             raise HandledException('Class cannot .by_id_or_create.')
         if id_ is None:
index e307f1429f93227f7b960d2cad3d55b95e3d737c..4426bbab1288623b01385c8b4f798eb2a019ed0e 100644 (file)
@@ -1,5 +1,6 @@
 """Web server stuff."""
 from __future__ import annotations
+from inspect import signature
 from typing import Any, Callable
 from base64 import b64encode, b64decode
 from binascii import Error as binascii_Exception
@@ -309,6 +310,9 @@ class TaskHandler(BaseHTTPRequestHandler):
                     item = target_class.by_id_or_create(self._conn, id_)
                 else:
                     item = target_class.by_id(self._conn, id_)
+                if 'exists' in signature(f).parameters:
+                    exists = id_ is not None and target_class._get_cached(id_)
+                    return f(self, item, exists)
                 return f(self, item)
             return wrapper
         return decorator
@@ -474,10 +478,14 @@ class TaskHandler(BaseHTTPRequestHandler):
                 'pattern': pattern}
 
     @_get_item(Condition)
-    def do_GET_condition(self, c: Condition) -> dict[str, object]:
+    def do_GET_condition(self,
+                         c: Condition,
+                         exists: bool
+                         ) -> dict[str, object]:
         """Show Condition of ?id=."""
         ps = Process.all(self._conn)
-        return {'condition': c, 'is_new': c.id_ is None,
+        return {'condition': c,
+                'is_new': not exists,
                 'enabled_processes': [p for p in ps if c in p.conditions],
                 'disabled_processes': [p for p in ps if c in p.blockers],
                 'enabling_processes': [p for p in ps if c in p.enables],
@@ -494,7 +502,10 @@ class TaskHandler(BaseHTTPRequestHandler):
         return {'condition': c}
 
     @_get_item(Process)
-    def do_GET_process(self, process: Process) -> dict[str, object]:
+    def do_GET_process(self,
+                       process: Process,
+                       exists: bool
+                       ) -> dict[str, object]:
         """Show Process of ?id=."""
         owner_ids = self._params.get_all_int('step_to')
         owned_ids = self._params.get_all_int('has_step')
@@ -516,7 +527,8 @@ class TaskHandler(BaseHTTPRequestHandler):
         for process_id in owned_ids:
             Process.by_id(self._conn, process_id)  # to ensure ID exists
             preset_top_step = process_id
-        return {'process': process, 'is_new': process.id_ is None,
+        return {'process': process,
+                'is_new': not exists,
                 'preset_top_step': preset_top_step,
                 'steps': process.get_steps(self._conn),
                 'owners': owners,
index 6feda94fb8bc9007773bb0f47f08ed6dfd49d545..a9b28bbb9fd59a7b124ffb740ca45bdafd48c0e2 100644 (file)
@@ -51,9 +51,10 @@ class ExpectedGetConditions(Expected):
 
 class ExpectedGetCondition(Expected):
     """Builder of expectations for GET /condition."""
+    _default_dict = {'is_new': False}
     _on_empty_make_temp = ('Condition', 'cond_as_dict')
 
-    def __init__(self, id_: int, *args: Any, **kwargs: Any) -> None:
+    def __init__(self, id_: int | None, *args: Any, **kwargs: Any) -> None:
         self._fields = {'condition': id_}
         super().__init__(*args, **kwargs)
 
@@ -67,7 +68,6 @@ class ExpectedGetCondition(Expected):
             self._fields[c_field] = self.as_ids([
                 p for p in self.lib_all('Process')
                 if self._fields['condition'] in p[p_field]])
-        self._fields['is_new'] = False
 
 
 class TestsWithServer(TestCaseWithServer):
@@ -79,51 +79,55 @@ class TestsWithServer(TestCaseWithServer):
         url = '/condition'
         self.check_post({}, url, 400)
         self.check_post({'title': ''}, url, 400)
-        self.check_post({'title': '', 'is_active': 0}, url, 400)
-        self.check_post({'description': '', 'is_active': 0}, url, 400)
+        self.check_post({'description': ''}, url, 400)
         # check valid POST payload on bad paths
         valid_payload = {'title': '', 'description': '', 'is_active': 0}
-        self.check_post(valid_payload, '/condition?id=foo', 400)
+        self.check_post(valid_payload, f'{url}?id=foo', 400)
 
     def test_POST_condition(self) -> None:
         """Test (valid) POST /condition and its effect on GET /condition[s]."""
-        exp_single = ExpectedGetCondition(1)
-        exp_all = ExpectedGetConditions()
+        url_single, url_all = '/condition?id=1', '/conditions'
+        exp_single, exp_all = ExpectedGetCondition(1), ExpectedGetConditions()
         all_exps = [exp_single, exp_all]
         # test valid POST's effect on single /condition and full /conditions
-        post = {'title': 'foo', 'description': 'oof', 'is_active': 0}
-        self.post_exp_cond(all_exps, 1, post, '', '?id=1')
-        self.check_json_get('/condition?id=1', exp_single)
-        self.check_json_get('/conditions', exp_all)
+        self.post_exp_cond(all_exps, {'title': 'foo', 'description': 'oof'},
+                           post_to_id=False)
+        self.check_json_get(url_single, exp_single)
+        self.check_json_get(url_all, exp_all)
         # test (no) effect of invalid POST to existing Condition on /condition
-        self.check_post({}, '/condition?id=1', 400)
-        self.check_json_get('/condition?id=1', exp_single)
+        self.check_post({}, url_single, 400)
+        self.check_json_get(url_single, exp_single)
         # test effect of POST changing title and activeness
-        post = {'title': 'bar', 'description': 'oof', 'is_active': 1}
-        self.post_exp_cond(all_exps, 1, post, '?id=1', '?id=1')
-        self.check_json_get('/condition?id=1', exp_single)
-        self.check_json_get('/conditions', exp_all)
+        self.post_exp_cond(all_exps, {'title': 'bar', 'description': 'oof',
+                                      'is_active': 1})
+        self.check_json_get(url_single, exp_single)
+        self.check_json_get(url_all, exp_all)
         # test deletion POST's effect, both to return id=1 into empty single,
         # full /conditions into empty list
-        self.post_exp_cond(all_exps, 1, {'delete': ''}, '?id=1', 's')
-        self.check_json_get('/condition?id=1', exp_single)
-        self.check_json_get('/conditions', exp_all)
+        self.post_exp_cond(all_exps, {'delete': ''}, redir_to_id=False)
+        exp_single.set('is_new', True)
+        self.check_json_get(url_single, exp_single)
+        self.check_json_get(url_all, exp_all)
 
     def test_GET_condition(self) -> None:
         """More GET /condition testing, especially for Process relations."""
         # check expected default status codes
         self.check_get_defaults('/condition')
+        # check 'is_new' set if id= absent or pointing to not-yet-existing ID
+        exp = ExpectedGetCondition(None)
+        exp.set('is_new', True)
+        self.check_json_get('/condition', exp)
+        exp = ExpectedGetCondition(1)
+        exp.set('is_new', True)
+        self.check_json_get('/condition?id=1', exp)
         # make Condition and two Processes that among them establish all
         # possible ConditionsRelations to it, check /condition displays all
         exp = ExpectedGetCondition(1)
-        cond_post = {'title': 'foo', 'description': 'oof', 'is_active': 0}
-        self.post_exp_cond([exp], 1, cond_post, '', '?id=1')
-        proc1_post = {'title': 'A', 'description': '', 'effort': 1.1,
-                      'conditions': [1], 'disables': [1]}
-        proc2_post = {'title': 'B', 'description': '', 'effort': 0.9,
-                      'enables': [1], 'blockers': [1]}
-        self.post_exp_process([exp], proc1_post, 1)
-        self.post_exp_process([exp], proc2_post, 2)
+        self.post_exp_cond([exp], {'title': 'foo', 'description': 'oof'},
+                           post_to_id=False)
+        for i, p in enumerate([('conditions', 'disables'),
+                               ('enables', 'blockers')]):
+            self.post_exp_process([exp], {k: [1] for k in p}, i+1)
         self.check_json_get('/condition?id=1', exp)
 
     def test_GET_conditions(self) -> None:
@@ -131,19 +135,17 @@ class TestsWithServer(TestCaseWithServer):
         # test empty result on empty DB, default-settings on empty params
         exp = ExpectedGetConditions()
         self.check_json_get('/conditions', exp)
-        # test ignorance of meaningless non-empty params (incl. unknown key),
-        # that 'sort_by' default to 'title' (even if set to something else, as
+        # test 'sort_by' default to 'title' (even if set to something else, as
         # long as without handler) and 'pattern' get preserved
         exp.set('pattern', 'bar')
-        exp.set('sort_by', 'title')  # for clarity (already default)
         self.check_json_get('/conditions?sort_by=foo&pattern=bar&foo=x', exp)
-        # test non-empty result, automatic (positive) sorting by title
         exp.set('pattern', '')
+        # test non-empty result, automatic (positive) sorting by title
         post_cond1 = {'is_active': 0, 'title': 'foo', 'description': 'oof'}
         post_cond2 = {'is_active': 0, 'title': 'bar', 'description': 'rab'}
         post_cond3 = {'is_active': 1, 'title': 'baz', 'description': 'zab'}
         for i, post in enumerate([post_cond1, post_cond2, post_cond3]):
-            self.post_exp_cond([exp], i+1, post, '', f'?id={i+1}')
+            self.post_exp_cond([exp], post, i+1, post_to_id=False)
         self.check_filter(exp, 'conditions', 'sort_by', 'title', [2, 3, 1])
         # test other sortings
         self.check_filter(exp, 'conditions', 'sort_by', '-title', [1, 3, 2])
index 422c28324a9b04c79c5ee96cb50ec1032f7ead96..24a62bd02795e7cb34710238370b47c6dd89b2a5 100644 (file)
@@ -319,10 +319,11 @@ class TestsWithServer(TestCaseWithServer):
         # check on un-saved
         exp = ExpectedGetProcess(1)
         exp.force('process_candidates', [])
+        exp.set('is_new', True)
         self.check_json_get('/process?id=1', exp)
         # check on minimal payload post
+        exp = ExpectedGetProcess(1)
         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
@@ -345,6 +346,7 @@ class TestsWithServer(TestCaseWithServer):
         exp.set_proc_from_post(3, valid_post)
         exp.lib_set('ProcessStep', [exp.procstep_as_dict(1, 3, 2)])
         exp.force('process_candidates', [1, 2, 3])
+        exp.set('is_new', True)
         self.check_json_get('/process?id=4', exp)
 
     def test_POST_process_steps(self) -> None:
index 2ecf3b845d4f0128712dfdd63510a683c2d9aeaf..f048d468eee981db2f3ae68759b4ff6066898b41 100644 (file)
@@ -288,10 +288,10 @@ class TestsWithServer(TestCaseWithServer):
         for prefix in ['make_', '']:
             for suffix in ['', 'x', '1.1']:
                 self.check_post({'step_filler_to_1': [f'{prefix}{suffix}']},
-                                 '/todo?id=1', 400, '/todo')
+                                '/todo?id=1', 400, '/todo')
         for suffix in ['', 'x', '1.1']:
-                self.check_post({'step_filler_to_{suffix}': ['1']},
-                                 '/todo?id=1', 400, '/todo')
+            self.check_post({'step_filler_to_{suffix}': ['1']},
+                            '/todo?id=1', 400, '/todo')
 
     def test_basic_POST_todo(self) -> None:
         """Test basic POST /todo manipulations."""
@@ -318,8 +318,8 @@ class TestsWithServer(TestCaseWithServer):
         # test Condition posts
         c1_post = {'title': 'foo', 'description': 'oof', 'is_active': 0}
         c2_post = {'title': 'bar', 'description': 'rab', 'is_active': 1}
-        self.post_exp_cond([exp], 1, c1_post, '?id=1', '?id=1')
-        self.post_exp_cond([exp], 2, c2_post, '?id=2', '?id=2')
+        self.post_exp_cond([exp], c1_post, 1)
+        self.post_exp_cond([exp], c2_post, 2)
         self.check_json_get('/todo?id=1', exp)
         todo_post = {'conditions': [1], 'disables': [1],
                      'blockers': [2], 'enables': [2]}
index d1b6eac52ea08c196afc7501d10224de138879ec..7945f61fc113b169aaf39827474df2fdc8466211 100644 (file)
@@ -569,8 +569,7 @@ class Expected:
             id_ = self._fields[category.lower()]
             make_temp = not bool(self.lib_get(category, id_))
             if make_temp:
-                f = getattr(self, dicter)
-                self.lib_set(category, [f(id_)])
+                self.lib_set(category, [getattr(self, dicter)(id_)])
         self.recalc()
         d = {'_library': self._lib}
         for k, v in self._fields.items():
@@ -587,6 +586,7 @@ class Expected:
             d[k] = v
         if make_temp:
             json = json_dumps(d)
+            id_ = id_ if id_ is not None else '?'
             self.lib_del(category, id_)
             d = json_loads(json)
         return d
@@ -640,7 +640,8 @@ class Expected:
         """Return dictionary of items by their 'id' fields."""
         refs = {}
         for item in items:
-            refs[str(item['id'])] = item
+            id_ = str(item['id']) if item['id'] is not None else '?'
+            refs[id_] = item
         return refs
 
     @staticmethod
@@ -703,7 +704,8 @@ class Expected:
             return
         cond = self.lib_get('Condition', id_)
         if cond:
-            cond['is_active'] = d['is_active']
+            if 'is_active' in d:
+                cond['is_active'] = d['is_active']
             for category in ['title', 'description']:
                 history = cond['_versioned'][category]
                 if len(history) > 0:
@@ -713,8 +715,7 @@ class Expected:
                 else:
                     history['0'] = d[category]
         else:
-            cond = self.cond_as_dict(
-                    id_, d['is_active'], d['title'], d['description'])
+            cond = self.cond_as_dict(id_, **d)
         self.lib_set('Condition', [cond])
 
     @staticmethod
@@ -755,11 +756,11 @@ class Expected:
         for k, v in d.items():
             if k.startswith('step_filler_to_'):
                 continue
-            elif 'adopt' == k:
+            if 'adopt' == k:
                 new_children = v if isinstance(v, list) else [v]
                 corrected_kwargs['children'] += new_children
                 continue
-            elif k in {'is_done', 'calendarize'}:
+            if k in {'is_done', 'calendarize'}:
                 v = v in VALID_TRUES
             corrected_kwargs[k] = v
         todo = self.lib_get('Todo', id_)
@@ -866,16 +867,16 @@ class TestCaseWithServer(TestCaseWithDB):
 
     def post_exp_cond(self,
                       exps: list[Expected],
-                      id_: int,
                       payload: dict[str, object],
-                      path_suffix: str = '',
-                      redir_suffix: str = ''
+                      id_: int = 1,
+                      post_to_id: bool = True,
+                      redir_to_id: bool = True
                       ) -> None:
         """POST /condition(s), appropriately update Expecteds."""
         # pylint: disable=too-many-arguments
-        path = f'/condition{path_suffix}'
-        redir = f'/condition{redir_suffix}'
-        self.check_post(payload, path, redir=redir)
+        target = f'/condition?id={id_}' if post_to_id else '/condition'
+        redir = f'/condition?id={id_}' if redir_to_id else '/conditions'
+        self.check_post(payload, target, redir=redir)
         for exp in exps:
             exp.set_cond_from_post(id_, payload)