"""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
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
'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],
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')
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,
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)
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):
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:
# 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])
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."""
# 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]}
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():
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
"""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
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:
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
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_)
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)
projects / plomtask / commitdiff