home · contact · privacy
Adapt web server to plomlib.web.
authorChristian Heller <c.heller@plomlompom.de>
Sat, 18 Jan 2025 02:27:37 +0000 (03:27 +0100)
committerChristian Heller <c.heller@plomlompom.de>
Sat, 18 Jan 2025 02:27:37 +0000 (03:27 +0100)
plomlib
plomtask/http.py
tests/misc.py

diff --git a/plomlib b/plomlib
index 743dbe0d493ddeb47eca981fa5be6d78e4d754c9..e7202fcfd78c6a60bd90da789a68c8ec4baf7b1a 160000 (submodule)
--- a/plomlib
+++ b/plomlib
@@ -1 +1 @@
-Subproject commit 743dbe0d493ddeb47eca981fa5be6d78e4d754c9
+Subproject commit e7202fcfd78c6a60bd90da789a68c8ec4baf7b1a
index a4d2ed42d63ff3af34921e9b19050e1d39cd226e..3d969cfa6a8a8151a65482781948c0f99aff4749 100644 (file)
@@ -1,14 +1,11 @@
 """Web server stuff."""
 from __future__ import annotations
+from pathlib import Path
 from inspect import signature
 from typing import Any, Callable
 from base64 import b64encode, b64decode
 from binascii import Error as binascii_Exception
-from http.server import HTTPServer, BaseHTTPRequestHandler
-from urllib.parse import urlparse, parse_qs
 from json import dumps as json_dumps
-from os.path import split as path_split
-from jinja2 import Environment as JinjaEnv, FileSystemLoader as JinjaFSLoader
 from plomtask.dating import (
         days_n_from_dt_date, dt_date_from_str, date_in_n_days)
 from plomtask.days import Day
@@ -19,32 +16,26 @@ from plomtask.processes import Process, ProcessStep, ProcessStepsNode
 from plomtask.conditions import Condition
 from plomtask.todos import Todo, TodoOrProcStepNode
 from plomtask.misc import DictableNode
+from plomlib.web import PlomHttpServer, PlomHttpHandler, PlomQueryMap
 
-TEMPLATES_DIR = 'templates'
+TEMPLATES_DIR = Path('templates')
 
 
-class TaskServer(HTTPServer):
-    """Variant of HTTPServer that knows .jinja as Jinja Environment."""
+class TaskServer(PlomHttpServer):
+    """Extends parent by DatabaseFile .db and .render_mode='html'."""
 
-    def __init__(self, db_file: DatabaseFile,
-                 *args: Any, **kwargs: Any) -> None:
-        super().__init__(*args, **kwargs)
+    def __init__(self, db_file: DatabaseFile, *args, **kwargs) -> None:
+        super().__init__(TEMPLATES_DIR, *args, **kwargs)
         self.db = db_file
         self.render_mode = 'html'
-        self.jinja = JinjaEnv(loader=JinjaFSLoader(TEMPLATES_DIR))
 
 
-class InputsParser:
+class InputsParser(PlomQueryMap):
     """Wrapper for validating and retrieving dict-like HTTP inputs."""
 
-    def __init__(self, dict_: dict[str, list[str]]) -> None:
-        self.inputs = dict_
-
     def get_all_str(self, key: str) -> list[str]:
         """Retrieve list of string values at key (empty if no key)."""
-        if key not in self.inputs.keys():
-            return []
-        return self.inputs[key]
+        return self.all(key) or []
 
     def get_all_int(self, key: str, fail_on_empty: bool = False) -> list[int]:
         """Retrieve list of int values at key."""
@@ -57,10 +48,8 @@ class InputsParser:
 
     def get_str(self, key: str, default: str | None = None) -> str | None:
         """Retrieve single/first string value of key, or default."""
-        vals = self.get_all_str(key)
-        if vals:
-            return vals[0]
-        return default
+        first = self.first(key)
+        return default if first is None else first
 
     def get_str_or_fail(self, key: str, default: str | None = None) -> str:
         """Retrieve first string value of key, if none: fail or default."""
@@ -89,8 +78,8 @@ class InputsParser:
     def get_all_of_key_prefixed(self, key_prefix: str) -> dict[str, list[str]]:
         """Retrieve dict of strings at keys starting with key_prefix."""
         ret = {}
-        for key in [k for k in self.inputs.keys() if k.startswith(key_prefix)]:
-            ret[key[len(key_prefix):]] = self.inputs[key]
+        for key in self.keys_prefixed(key_prefix):
+            ret[key[len(key_prefix):]] = self.as_dict[key]
         return ret
 
     def get_float_or_fail(self, key: str) -> float:
@@ -117,14 +106,15 @@ class InputsParser:
         return ret
 
 
-class TaskHandler(BaseHTTPRequestHandler):
+class TaskHandler(PlomHttpHandler):
     """Handles single HTTP request."""
     # pylint: disable=too-many-public-methods
     server: TaskServer
+    params: InputsParser
+    postvars: InputsParser
+    mapper = InputsParser
     _conn: DatabaseConnection
     _site: str
-    _form: InputsParser
-    _params: InputsParser
 
     def _send_page(
             self, ctx: dict[str, Any], tmpl_name: str, code: int = 200
@@ -134,19 +124,12 @@ class TaskHandler(BaseHTTPRequestHandler):
         The differentiation by .server.render_mode serves to allow easily
         comparable JSON responses for automatic testing.
         """
-        body: str
-        headers: list[tuple[str, str]] = []
         if 'html' == self.server.render_mode:
-            tmpl = self.server.jinja.get_template(f'{tmpl_name}.html')
-            body = tmpl.render(ctx)
+            self.send_rendered(Path(f'{tmpl_name}.html'), ctx, code)
         else:
-            body = self._ctx_to_json(ctx)
-            headers += [('Content-Type', 'application/json')]
-        self.send_response(code)
-        for header_tuple in headers:
-            self.send_header(*header_tuple)
-        self.end_headers()
-        self.wfile.write(bytes(body, 'utf-8'))
+            self.send_http(self._ctx_to_json(ctx).encode(),
+                           [('Content-Type', 'application/json')],
+                           code)
 
     def _ctx_to_json(self, ctx: dict[str, object]) -> str:
         """Render ctx into JSON string.
@@ -237,24 +220,16 @@ class TaskHandler(BaseHTTPRequestHandler):
                 # method to self with respective access privileges)
                 try:
                     self._conn = DatabaseConnection(self.server.db)
-                    parsed_url = urlparse(self.path)
-                    self._site = path_split(parsed_url.path)[1]
-                    params = parse_qs(parsed_url.query,
-                                      keep_blank_values=True,
-                                      strict_parsing=True)
-                    self._params = InputsParser(params)
-                    handler_name = f'do_{http_method}_{self._site}'
+                    handler_name = f'do_{http_method}_{self.pagename}'
                     if hasattr(self, handler_name):
                         handler = getattr(self, handler_name)
                         redir_target = f(self, handler)
                         if 'POST' == http_method:
                             clear_caches()
                         if redir_target:
-                            self.send_response(302)
-                            self.send_header('Location', redir_target)
-                            self.end_headers()
+                            self.redirect(Path(redir_target))
                     else:
-                        msg = f'{not_found_msg}: {self._site}'
+                        msg = f'{not_found_msg}: {self.pagename}'
                         raise NotFoundException(msg)
                 except HandledException as error:
                     if 'POST' == http_method:
@@ -270,7 +245,7 @@ class TaskHandler(BaseHTTPRequestHandler):
     def do_GET(self, handler: Callable[[], str | dict[str, object]]
                ) -> str | None:
         """Render page with result of handler, or redirect if result is str."""
-        tmpl_name = f'{self._site}'
+        tmpl_name = f'{self.pagename}'
         ctx_or_redir_target = handler()
         if isinstance(ctx_or_redir_target, str):
             return ctx_or_redir_target
@@ -280,10 +255,6 @@ class TaskHandler(BaseHTTPRequestHandler):
     @_request_wrapper('POST', 'Unknown POST target')
     def do_POST(self, handler: Callable[[], str]) -> str:
         """Handle POST with handler, prepare redirection to result."""
-        length = int(self.headers['content-length'])
-        postvars = parse_qs(self.rfile.read(length).decode(),
-                            keep_blank_values=True)
-        self._form = InputsParser(postvars)
         redir_target = handler()
         self._conn.commit()
         return redir_target
@@ -301,7 +272,7 @@ class TaskHandler(BaseHTTPRequestHandler):
                 # (because pylint here fails to detect the use of wrapper as a
                 # method to self with respective access privileges)
                 id_ = None
-                for val in self._params.get_all_int('id', fail_on_empty=True):
+                for val in self.params.get_all_int('id', fail_on_empty=True):
                     id_ = val
                 if target_class.can_create_by_id:
                     item = target_class.by_id_or_create(self._conn, id_)
@@ -325,8 +296,8 @@ class TaskHandler(BaseHTTPRequestHandler):
         same, the only difference being the HTML template they are rendered to,
         which .do_GET selects from their method name.
         """
-        start = self._params.get_str_or_fail('start', '')
-        end = self._params.get_str_or_fail('end', '')
+        start = self.params.get_str_or_fail('start', '')
+        end = self.params.get_str_or_fail('end', '')
         dt_start = dt_date_from_str(start if start else date_in_n_days(-1))
         dt_end = dt_date_from_str(end if end else date_in_n_days(366))
         days = Day.with_filled_gaps(self._conn, dt_start, dt_end)
@@ -344,8 +315,8 @@ class TaskHandler(BaseHTTPRequestHandler):
 
     def do_GET_day(self) -> dict[str, object]:
         """Show single Day of ?date=."""
-        date = self._params.get_str('date', date_in_n_days(0))
-        make_type = self._params.get_str_or_fail('make_type', 'full')
+        date = self.params.get_str('date', date_in_n_days(0))
+        make_type = self.params.get_str_or_fail('make_type', 'full')
         #
         assert isinstance(date, str)
         day = Day.by_id_or_create(self._conn,
@@ -448,11 +419,11 @@ class TaskHandler(BaseHTTPRequestHandler):
 
     def do_GET_todos(self) -> dict[str, object]:
         """Show Todos from ?start= to ?end=, of ?process=, ?comment= pattern"""
-        sort_by = self._params.get_str_or_fail('sort_by', 'title')
-        start = self._params.get_str_or_fail('start', '')
-        end = self._params.get_str_or_fail('end', '')
-        process_id = self._params.get_int_or_none('process_id')
-        comment_pattern = self._params.get_str_or_fail('comment_pattern', '')
+        sort_by = self.params.get_str_or_fail('sort_by', 'title')
+        start = self.params.get_str_or_fail('start', '')
+        end = self.params.get_str_or_fail('end', '')
+        process_id = self.params.get_int_or_none('process_id')
+        comment_pattern = self.params.get_str_or_fail('comment_pattern', '')
         #
         ret = Todo.by_date_range_with_limits(self._conn, (start, end))
         todos_by_date_range, start, end = ret
@@ -466,8 +437,8 @@ class TaskHandler(BaseHTTPRequestHandler):
 
     def do_GET_conditions(self) -> dict[str, object]:
         """Show all Conditions."""
-        pattern = self._params.get_str_or_fail('pattern', '')
-        sort_by = self._params.get_str_or_fail('sort_by', 'title')
+        pattern = self.params.get_str_or_fail('pattern', '')
+        sort_by = self.params.get_str_or_fail('sort_by', 'title')
         #
         conditions = Condition.matching(self._conn, pattern)
         sort_by = Condition.sort_by(conditions, sort_by)
@@ -505,9 +476,9 @@ class TaskHandler(BaseHTTPRequestHandler):
                        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')
-        title_64 = self._params.get_str('title_b64')
+        owner_ids = self.params.get_all_int('step_to')
+        owned_ids = self.params.get_all_int('has_step')
+        title_64 = self.params.get_str('title_b64')
         title_new = None
         if title_64:
             try:
@@ -551,8 +522,8 @@ class TaskHandler(BaseHTTPRequestHandler):
 
     def do_GET_processes(self) -> dict[str, object]:
         """Show all Processes."""
-        pattern = self._params.get_str_or_fail('pattern', '')
-        sort_by = self._params.get_str_or_fail('sort_by', 'title')
+        pattern = self.params.get_str_or_fail('pattern', '')
+        sort_by = self.params.get_str_or_fail('sort_by', 'title')
         #
         processes = Process.matching(self._conn, pattern)
         sort_by = Process.sort_by(processes, sort_by)
@@ -569,8 +540,8 @@ class TaskHandler(BaseHTTPRequestHandler):
                 # pylint: disable=protected-access
                 # (because pylint here fails to detect the use of wrapper as a
                 # method to self with respective access privileges)
-                id_ = self._params.get_int_or_none('id')
-                for _ in self._form.get_all_str('delete'):
+                id_ = self.params.get_int_or_none('id')
+                for _ in self.postvars.get_all_str('delete'):
                     if id_ is None:
                         msg = 'trying to delete non-saved ' +\
                                 f'{target_class.__name__}'
@@ -588,10 +559,10 @@ class TaskHandler(BaseHTTPRequestHandler):
 
     def _change_versioned_timestamps(self, cls: Any, attr_name: str) -> str:
         """Update history timestamps for VersionedAttribute."""
-        id_ = self._params.get_int_or_none('id')
+        id_ = self.params.get_int_or_none('id')
         item = cls.by_id(self._conn, id_)
         attr = getattr(item, attr_name)
-        for k, vals in self._form.get_all_of_key_prefixed('at:').items():
+        for k, vals in self.postvars.get_all_of_key_prefixed('at:').items():
             if k[19:] != vals[0]:
                 attr.reset_timestamp(k, f'{vals[0]}.0')
         attr.save(self._conn)
@@ -600,14 +571,14 @@ class TaskHandler(BaseHTTPRequestHandler):
     def do_POST_day(self) -> str:
         """Update or insert Day of date and Todos mapped to it."""
         # pylint: disable=too-many-locals
-        date = self._params.get_str_or_fail('date')
-        day_comment = self._form.get_str_or_fail('day_comment')
-        make_type = self._form.get_str_or_fail('make_type')
-        old_todos = self._form.get_all_int('todo_id')
-        new_todos_by_process = self._form.get_all_int('new_todo')
-        comments = self._form.get_all_str('comment')
-        efforts = self._form.get_all_floats_or_nones('effort')
-        done_todos = self._form.get_all_int('done')
+        date = self.params.get_str_or_fail('date')
+        day_comment = self.postvars.get_str_or_fail('day_comment')
+        make_type = self.postvars.get_str_or_fail('make_type')
+        old_todos = self.postvars.get_all_int('todo_id')
+        new_todos_by_process = self.postvars.get_all_int('new_todo')
+        comments = self.postvars.get_all_str('comment')
+        efforts = self.postvars.get_all_floats_or_nones('effort')
+        done_todos = self.postvars.get_all_int('done')
         is_done = [t_id in done_todos for t_id in old_todos]
         if not (len(old_todos) == len(is_done) == len(comments)
                 == len(efforts)):
@@ -645,19 +616,21 @@ class TaskHandler(BaseHTTPRequestHandler):
         # pylint: disable=too-many-branches
         # pylint: disable=too-many-statements
         assert isinstance(todo.id_, int)
-        adoptees = [(id_, todo.id_) for id_ in self._form.get_all_int('adopt')]
-        to_make = {'full': [(id_, todo.id_)
-                            for id_ in self._form.get_all_int('make_full')],
-                   'empty': [(id_, todo.id_)
-                             for id_ in self._form.get_all_int('make_empty')]}
-        step_fillers_to = self._form.get_all_of_key_prefixed('step_filler_to_')
+        adoptees = [(id_, todo.id_) for id_
+                    in self.postvars.get_all_int('adopt')]
+        to_make = {'full': [(id_, todo.id_) for id_
+                            in self.postvars.get_all_int('make_full')],
+                   'empty': [(id_, todo.id_) for id_
+                             in self.postvars.get_all_int('make_empty')]}
+        step_fillers_to = self.postvars.get_all_of_key_prefixed(
+                'step_filler_to_')
         to_update: dict[str, Any] = {
-            'comment': self._form.get_str_or_fail('comment', ''),
-            'is_done': self._form.get_bool('is_done'),
-            'calendarize': self._form.get_bool('calendarize')}
-        cond_rels = [self._form.get_all_int(name) for name in
+            'comment': self.postvars.get_str_or_fail('comment', ''),
+            'is_done': self.postvars.get_bool('is_done'),
+            'calendarize': self.postvars.get_bool('calendarize')}
+        cond_rels = [self.postvars.get_all_int(name) for name in
                      ['conditions', 'blockers', 'enables', 'disables']]
-        effort_or_not = self._form.get_str('effort')
+        effort_or_not = self.postvars.get_str('effort')
         if effort_or_not is not None:
             if effort_or_not == '':
                 to_update['effort'] = None
@@ -742,19 +715,20 @@ class TaskHandler(BaseHTTPRequestHandler):
                     title = id_or_title
             return title, l_ids
 
-        versioned = {'title': self._form.get_str_or_fail('title'),
-                     'description': self._form.get_str_or_fail('description'),
-                     'effort': self._form.get_float_or_fail('effort')}
-        cond_rels = [self._form.get_all_int(s) for s
+        versioned = {
+                'title': self.postvars.get_str_or_fail('title'),
+                'description': self.postvars.get_str_or_fail('description'),
+                'effort': self.postvars.get_float_or_fail('effort')}
+        cond_rels = [self.postvars.get_all_int(s) for s
                      in ['conditions', 'blockers', 'enables', 'disables']]
-        calendarize = self._form.get_bool('calendarize')
-        step_of = self._form.get_all_str('step_of')
-        suppressions = self._form.get_all_int('suppressed_steps')
-        kept_steps = self._form.get_all_int('kept_steps')
-        new_top_step_procs = self._form.get_all_str('new_top_step')
+        calendarize = self.postvars.get_bool('calendarize')
+        step_of = self.postvars.get_all_str('step_of')
+        suppressions = self.postvars.get_all_int('suppressed_steps')
+        kept_steps = self.postvars.get_all_int('kept_steps')
+        new_top_step_procs = self.postvars.get_all_str('new_top_step')
         new_steps_to = {
-                int(k): [int(n) for n in v] for (k, v)
-                in self._form.get_all_of_key_prefixed('new_step_to_').items()}
+            int(k): [int(n) for n in v] for (k, v)
+            in self.postvars.get_all_of_key_prefixed('new_step_to_').items()}
         new_owner_title, owners_to_set = id_or_title(step_of)
         new_step_title, new_top_step_proc_ids = id_or_title(new_top_step_procs)
         #
@@ -797,9 +771,9 @@ class TaskHandler(BaseHTTPRequestHandler):
     @_delete_or_post(Condition, '/conditions')
     def do_POST_condition(self, condition: Condition) -> str:
         """Update/insert Condition of ?id= and fields defined in postvars."""
-        title = self._form.get_str_or_fail('title')
-        description = self._form.get_str_or_fail('description')
-        is_active = self._form.get_bool('is_active')
+        title = self.postvars.get_str_or_fail('title')
+        description = self.postvars.get_str_or_fail('description')
+        is_active = self.postvars.get_bool('is_active')
         condition.is_active = is_active
         #
         condition.title.set(title)
index a6df2e56481bb41deebe4fd4182ca96ac41c07fe..c26e83d96762204206cf25ceebd9c94975a4dab3 100644 (file)
@@ -1,4 +1,5 @@
 """Miscellaneous tests."""
+from typing import Callable
 from unittest import TestCase
 from tests.utils import TestCaseWithServer
 from plomtask.http import InputsParser
@@ -8,145 +9,110 @@ from plomtask.exceptions import BadFormatException
 class TestsSansServer(TestCase):
     """Tests that do not require DB setup or a server."""
 
+    def _test_parser(self,
+                     method: Callable,
+                     serialized: str,
+                     expected: object,
+                     method_args: list[object],
+                     fails: bool = False
+                     ) -> None:
+        # pylint: disable=too-many-arguments
+        parser = InputsParser(serialized)
+        if fails:
+            with self.assertRaises(BadFormatException):
+                method(parser, *method_args)
+        else:
+            self.assertEqual(expected, method(parser, *method_args))
+
     def test_InputsParser_get_str_or_fail(self) -> None:
         """Test InputsParser.get_str."""
-        parser = InputsParser({})
-        with self.assertRaises(BadFormatException):
-            parser.get_str_or_fail('foo')
-        self.assertEqual('bar', parser.get_str_or_fail('foo', 'bar'))
-        parser = InputsParser({'foo': []})
-        with self.assertRaises(BadFormatException):
-            parser.get_str_or_fail('foo')
-        self.assertEqual('bar', parser.get_str_or_fail('foo', 'bar'))
-        parser = InputsParser({'foo': ['baz']})
-        self.assertEqual('baz', parser.get_str_or_fail('foo', 'bar'))
-        parser = InputsParser({'foo': ['baz', 'quux']})
-        self.assertEqual('baz', parser.get_str_or_fail('foo', 'bar'))
+        m = InputsParser.get_str_or_fail
+        self._test_parser(m, '', 0, ['foo'], fails=True)
+        self._test_parser(m, '', 'bar', ['foo', 'bar'])
+        self._test_parser(m, 'foo=', '', ['foo'])
+        self._test_parser(m, 'foo=', '', ['foo', 'bar'])
+        self._test_parser(m, 'foo=baz', 'baz', ['foo', 'bar'])
+        self._test_parser(m, 'foo=baz&foo=quux', 'baz', ['foo', 'bar'])
+        self._test_parser(m, 'foo=baz,quux', 'baz,quux', ['foo', 'bar'])
 
     def test_InputsParser_get_str(self) -> None:
         """Test InputsParser.get_str."""
-        parser = InputsParser({})
-        self.assertEqual(None, parser.get_str('foo'))
-        self.assertEqual('bar', parser.get_str('foo', 'bar'))
-        parser = InputsParser({'foo': []})
-        self.assertEqual(None, parser.get_str('foo'))
-        self.assertEqual('bar', parser.get_str('foo', 'bar'))
-        parser = InputsParser({'foo': ['baz']})
-        self.assertEqual('baz', parser.get_str('foo', 'bar'))
-        parser = InputsParser({'foo': ['baz', 'quux']})
-        self.assertEqual('baz', parser.get_str('foo', 'bar'))
+        m = InputsParser.get_str
+        self._test_parser(m, '', None, ['foo'])
+        self._test_parser(m, '', 'bar', ['foo', 'bar'])
+        self._test_parser(m, 'foo=', '', ['foo'])
+        self._test_parser(m, 'foo=', '', ['foo', 'bar'])
+        self._test_parser(m, 'foo=baz', 'baz', ['foo', 'bar'])
+        self._test_parser(m, 'foo=baz&foo=quux', 'baz', ['foo', 'bar'])
+        self._test_parser(m, 'foo=baz,quux', 'baz,quux', ['foo', 'bar'])
 
     def test_InputsParser_get_all_of_key_prefixed(self) -> None:
         """Test InputsParser.get_all_of_key_prefixed."""
-        parser = InputsParser({})
-        self.assertEqual({},
-                         parser.get_all_of_key_prefixed(''))
-        self.assertEqual({},
-                         parser.get_all_of_key_prefixed('foo'))
-        parser = InputsParser({'foo': ['bar']})
-        self.assertEqual({'foo': ['bar']},
-                         parser.get_all_of_key_prefixed(''))
-        parser = InputsParser({'x': ['y', 'z']})
-        self.assertEqual({'': ['y', 'z']},
-                         parser.get_all_of_key_prefixed('x'))
-        parser = InputsParser({'xx': ['y', 'Z']})
-        self.assertEqual({'x': ['y', 'Z']},
-                         parser.get_all_of_key_prefixed('x'))
-        parser = InputsParser({'xx': ['y']})
-        self.assertEqual({},
-                         parser.get_all_of_key_prefixed('xxx'))
-        parser = InputsParser({'xxx': ['x'], 'xxy': ['y'], 'xyy': ['z']})
-        self.assertEqual({'x': ['x'], 'y': ['y']},
-                         parser.get_all_of_key_prefixed('xx'))
-        parser = InputsParser({'xxx': ['x', 'y'], 'xxy': ['y', 'z']})
-        self.assertEqual({'x': ['x', 'y'], 'y': ['y', 'z']},
-                         parser.get_all_of_key_prefixed('xx'))
+        m = InputsParser.get_all_of_key_prefixed
+        self._test_parser(m, '', {}, [''])
+        self._test_parser(m, '', {}, ['foo'])
+        self._test_parser(m, 'foo=bar', {'foo': ['bar']}, [''])
+        self._test_parser(m, 'x=y&x=z', {'': ['y', 'z']}, ['x'])
+        self._test_parser(m, 'xx=y&xx=Z', {'x': ['y', 'Z']}, ['x'])
+        self._test_parser(m, 'xx=y', {}, ['xxx'])
+        self._test_parser(m, 'xxx=x&xxy=y&xyy=z', {'x': ['x'], 'y': ['y']},
+                          ['xx'])
 
     def test_InputsParser_get_int_or_none(self) -> None:
         """Test InputsParser.get_int_or_none."""
-        parser = InputsParser({})
-        self.assertEqual(None, parser.get_int_or_none('foo'))
-        parser = InputsParser({'foo': []})
-        self.assertEqual(None, parser.get_int_or_none('foo'))
-        parser = InputsParser({'foo': ['']})
-        self.assertEqual(None, parser.get_int_or_none('foo'))
-        parser = InputsParser({'foo': ['0']})
-        self.assertEqual(0, parser.get_int_or_none('foo'))
-        with self.assertRaises(BadFormatException):
-            InputsParser({'foo': ['None']}).get_int_or_none('foo')
-        with self.assertRaises(BadFormatException):
-            InputsParser({'foo': ['0.1']}).get_int_or_none('foo')
-        parser = InputsParser({'foo': ['23']})
-        self.assertEqual(23, parser.get_int_or_none('foo'))
+        m = InputsParser.get_int_or_none
+        self._test_parser(m, '', None, ['foo'])
+        self._test_parser(m, 'foo=', None, ['foo'])
+        self._test_parser(m, 'foo=0', 0, ['foo'])
+        self._test_parser(m, 'foo=None', 0, ['foo'], fails=True)
+        self._test_parser(m, 'foo=0.1', 0, ['foo'], fails=True)
+        self._test_parser(m, 'foo=23', 23, ['foo'])
 
     def test_InputsParser_get_float_or_fail(self) -> None:
         """Test InputsParser.get_float_or_fail."""
-        with self.assertRaises(BadFormatException):
-            InputsParser({}).get_float_or_fail('foo')
-        with self.assertRaises(BadFormatException):
-            InputsParser({'foo': ['']}).get_float_or_fail('foo')
-        with self.assertRaises(BadFormatException):
-            InputsParser({'foo': ['bar']}).get_float_or_fail('foo')
-        parser = InputsParser({'foo': ['0']})
-        self.assertEqual(0, parser.get_float_or_fail('foo'))
-        parser = InputsParser({'foo': ['0.1']})
-        self.assertEqual(0.1, parser.get_float_or_fail('foo'))
-        parser = InputsParser({'foo': ['1.23', '456']})
-        self.assertEqual(1.23, parser.get_float_or_fail('foo'))
-        with self.assertRaises(BadFormatException):
-            InputsParser({}).get_float_or_fail('foo')
-        with self.assertRaises(BadFormatException):
-            InputsParser({'foo': []}).get_float_or_fail('foo')
+        m = InputsParser.get_float_or_fail
+        self._test_parser(m, '', 0, ['foo'], fails=True)
+        self._test_parser(m, 'foo=', 0, ['foo'], fails=True)
+        self._test_parser(m, 'foo=bar', 0, ['foo'], fails=True)
+        self._test_parser(m, 'foo=0', 0, ['foo'])
+        self._test_parser(m, 'foo=0.1', 0.1, ['foo'])
+        self._test_parser(m, 'foo=1.23&foo=456', 1.23, ['foo'])
 
     def test_InputsParser_get_bool(self) -> None:
         """Test InputsParser.get_bool."""
-        self.assertEqual(0, InputsParser({}).get_bool('foo'))
-        self.assertEqual(0, InputsParser({'val': ['foo']}).get_bool('foo'))
-        self.assertEqual(0, InputsParser({'val': ['True']}).get_bool('foo'))
-        self.assertEqual(0, InputsParser({'foo': []}).get_bool('foo'))
-        self.assertEqual(0, InputsParser({'foo': ['None']}).get_bool('foo'))
-        self.assertEqual(0, InputsParser({'foo': ['0']}).get_bool('foo'))
-        self.assertEqual(0, InputsParser({'foo': ['']}).get_bool('foo'))
-        self.assertEqual(0, InputsParser({'foo': ['bar']}).get_bool('foo'))
-        self.assertEqual(0, InputsParser({'foo': ['bar',
-                                                  'baz']}).get_bool('foo'))
-        self.assertEqual(0, InputsParser({'foo': ['False']}).get_bool('foo'))
-        self.assertEqual(1, InputsParser({'foo': ['true']}).get_bool('foo'))
-        self.assertEqual(1, InputsParser({'foo': ['True']}).get_bool('foo'))
-        self.assertEqual(1, InputsParser({'foo': ['1']}).get_bool('foo'))
-        self.assertEqual(1, InputsParser({'foo': ['on']}).get_bool('foo'))
+        m = InputsParser.get_bool
+        self._test_parser(m, '', 0, ['foo'])
+        self._test_parser(m, 'val=foo', 0, ['foo'])
+        self._test_parser(m, 'val=True', 0, ['foo'])
+        self._test_parser(m, 'foo=', 0, ['foo'])
+        self._test_parser(m, 'foo=None', 0, ['foo'])
+        self._test_parser(m, 'foo=0', 0, ['foo'])
+        self._test_parser(m, 'foo=bar', 0, ['foo'])
+        self._test_parser(m, 'foo=bar&foo=baz', 0, ['foo'])
+        self._test_parser(m, 'foo=False', 0, ['foo'])
+        self._test_parser(m, 'foo=true', 1, ['foo'])
+        self._test_parser(m, 'foo=True', 1, ['foo'])
+        self._test_parser(m, 'foo=1', 1, ['foo'])
+        self._test_parser(m, 'foo=on', 1, ['foo'])
 
     def test_InputsParser_get_all_str(self) -> None:
         """Test InputsParser.get_all_str."""
-        parser = InputsParser({})
-        self.assertEqual([], parser.get_all_str('foo'))
-        parser = InputsParser({'foo': []})
-        self.assertEqual([], parser.get_all_str('foo'))
-        parser = InputsParser({'foo': ['bar']})
-        self.assertEqual(['bar'], parser.get_all_str('foo'))
-        parser = InputsParser({'foo': ['bar', 'baz']})
-        self.assertEqual(['bar', 'baz'], parser.get_all_str('foo'))
+        m = InputsParser.get_all_str
+        self._test_parser(m, '', [], ['foo'])
+        self._test_parser(m, 'foo=', [''], ['foo'])
+        self._test_parser(m, 'foo=bar', ['bar'], ['foo'])
+        self._test_parser(m, 'foo=bar&foo=baz', ['bar', 'baz'], ['foo'])
 
     def test_InputsParser_get_all_int(self) -> None:
         """Test InputsParser.get_all_int."""
-        parser = InputsParser({})
-        self.assertEqual([], parser.get_all_int('foo'))
-        parser = InputsParser({'foo': []})
-        self.assertEqual([], parser.get_all_int('foo'))
-        parser = InputsParser({'foo': ['']})
-        parser.get_all_int('foo')
-        with self.assertRaises(BadFormatException):
-            parser.get_all_int('foo', fail_on_empty=True)
-        parser = InputsParser({'foo': ['0']})
-        self.assertEqual([0], parser.get_all_int('foo'))
-        parser = InputsParser({'foo': ['0', '17']})
-        self.assertEqual([0, 17], parser.get_all_int('foo'))
-        parser = InputsParser({'foo': ['0.1', '17']})
-        with self.assertRaises(BadFormatException):
-            parser.get_all_int('foo')
-        parser = InputsParser({'foo': ['None', '17']})
-        with self.assertRaises(BadFormatException):
-            parser.get_all_int('foo')
+        m = InputsParser.get_all_int
+        self._test_parser(m, '', [], ['foo'])
+        self._test_parser(m, 'foo=', [], ['foo'])
+        self._test_parser(m, 'foo=', 0, ['foo', True], fails=True)
+        self._test_parser(m, 'foo=0', [0], ['foo'])
+        self._test_parser(m, 'foo=0&foo=17', [0, 17], ['foo'])
+        self._test_parser(m, 'foo=0.1&foo=17', 0, ['foo'], fails=True)
+        self._test_parser(m, 'foo=None&foo=17', 0, ['foo'], fails=True)
 
 
 class TestsWithServer(TestCaseWithServer):