Validate forms on backend, TODO: unittests & cleanup

A lot of extra, unecessary try & except blocks in inventory.py - and a lot of testcases to be made.

BookingSystem/__init__.py

@@ -16,6 +16,7 @@
 LABEL_SERVER = os.getenv('LABEL_SERVER')
 KIOSK_FQDN = os.getenv('KIOSK_FQDN')
 API_TOKEN = os.getenv('API_TOKEN')
+REGEX_ITEM = r'^(?:(?![\s])[ÆØÅæøåa-zA-Z0-9_\s\-]*[ÆØÅæøåa-zA-Z0-9_\-]+)$'
 
 # Logger setup
 logger = Logger(__name__)

BookingSystem/api.py

@@ -16,6 +16,7 @@
 import user
 from __init__ import DATABASE, LABEL_SERVER
 from inventory import Item
+from sanitizer import VALIDATORS, MINMAX, sanitize, handle_api_exception
 from utils import login_required, next_july
 
 api = flask.Blueprint('api', __name__)
@@ -62,46 +63,50 @@ def get_items_by_userid(userid: str) -> flask.Response:
 
 @api.route('/items', methods=['POST'])
 @login_required(admin_only=True)
+@handle_api_exception
 def add_item() -> flask.Response:
     """Add an item to the database."""
-    form = flask.request.form
-    item = {key: form.get(key) for key in form.keys() if key in Item.__annotations__}
+    # START: Validation
+    validation_map = {
+        'id': VALIDATORS.UNIQUE_ID,
+        'name': VALIDATORS.NAME,
+        'category': VALIDATORS.CATEGORY,
+    }
+    item_dict = sanitize(validation_map, flask.request.form)
+    # END: Validation
+
+    item = {key: val for key, val in item_dict.items()}
     item = Item(**item)
-    print_label_count = int(form.get('print_label_count'))
-    print_label_type = form.get('print_label_type')
-    try:
-        inventory.add(item)
-    except ValueError as e:
-        return flask.Response(str(e), status=400)
-    if print_label_count > 0:
-        url = f'{LABEL_SERVER}/print?count={print_label_count}&variant={print_label_type}&id={item.id}&name={item.name}&category={item.category}'
-        print(url)
+    inventory.add(item)
     return flask.Response(f'La til {item.id} i databasen.', status=201)
 
 
 @api.route('/items/<item_id>', methods=['PUT'])
 @login_required(admin_only=True)
+@handle_api_exception
 def edit_item(item_id: str) -> flask.Response:
     """Edit an item in the database."""
-    form = flask.request.form
-    item = {key: form.get(key) for key in form.keys() if key in Item.__annotations__}
-    item = Item(**item)
+    # START: Validation
+    validation_map = {
+        'id': VALIDATORS.UNIQUE_OR_SAME_ID,
+        'name': VALIDATORS.NAME,
+        'category': VALIDATORS.CATEGORY,
+    }
+    item_dict = sanitize(validation_map, flask.request.form, {'id': item_id})
+    # END: Validation
 
-    try:
-        inventory.edit(item_id, item)
-    except ValueError as e:
-        return flask.Response(str(e), status=400)
+    item = {key: val for key, val in item_dict.items()}
+    item = Item(**item)
+    inventory.edit(item_id, item)
     return flask.Response(f'Redigerte {item_id} i databasen.', status=200)
 
 
 @api.route('/items/<item_id>', methods=['DELETE'])
 @login_required(admin_only=True)
+@handle_api_exception
 def delete_item(item_id: str) -> flask.Response:
     """Delete an item from the database."""
-    try:
-        inventory.delete(item_id)
-    except ValueError as e:
-        return flask.Response(str(e), status=400)
+    inventory.delete(item_id)
     return flask.Response(f'Slettet {item_id} fra databasen.', status=200)
 
 
@@ -116,41 +121,56 @@ def get_label_preview(item_id: str, variant: str = 'qr') -> flask.Response:
 
 @api.route('/items/<item_id>/label/print', methods=['POST'])
 @login_required(admin_only=True)
+@handle_api_exception
 def print_label(item_id: str) -> flask.Response:
-    form = flask.request.form
+    # START: Validation
+    validation_map = {
+        'print_label_count': VALIDATORS.INT,
+        'print_label_count_minmax': MINMAX(0, 9),
+        'print_label_type': VALIDATORS.LABEL_TYPE,
+    }
+    form = sanitize(validation_map, flask.request.form)
+    # END: Validation
+
     variant = form.get('print_label_type', 'qr')
     count = int(form.get('print_label_count', '1'))
     item = inventory.get(item_id)
     url = f'{LABEL_SERVER}/print?id={item.id}&name={item.name}&variant={variant}&count={count}'
     try:
         response = requests.post(url)
-    except Exception as e:
+    except requests.exceptions.RequestException as e:
         return flask.Response(str(e), status=500)
     return flask.Response(response.text, status=response.status_code)
 
 
 @api.route('/book/out', methods=['POST'])
 @login_required(admin_only=True)
+@handle_api_exception
 def book_equipment() -> flask.Response:
     """Book out equipment for a user."""
-    form = flask.request.form
+    # START: Validation
+    validation_map = {
+        'user': VALIDATORS.UNIQUE_ID,
+        'days': VALIDATORS.INT,
+        'days_minmax': MINMAX(1, 90),
+        'equipment': VALIDATORS.ITEM_LIST_EXISTS,
+    }
+    form = sanitize(validation_map, flask.request.form)
+    # END: Validation
     userid = form.get('user')
     days = form.get('days')
-    item_ids = form.getlist('equipment')
 
-    for item in item_ids:
+    for item in form.get('equipment'):
         inventory.register_out(item_id=item, userid=userid, days=days)
     return flask.Response(f'Utstyret ble utlevert til {user.get(userid).get("name")}.', status=200)
 
 
 @api.route('/return/<item_id>', methods=['POST'])
 @login_required(admin_only=True)
+@handle_api_exception
 def return_equipment(item_id: str) -> flask.Response:
     """Return equipment from a user."""
-    try:
-        inventory.register_in(item_id=item_id)
-    except ValueError as e:
-        return flask.Response(str(e), status=400)
+    inventory.register_in(item_id=item_id)
     return flask.Response('Utstyr ble innlevert.', status=200)
 
 

BookingSystem/app.py

@@ -13,7 +13,7 @@
 import mail
 import routes
 import user
-from __init__ import logger
+from __init__ import logger, REGEX_ITEM
 from db import init_db
 from flask_session import Session
 
@@ -51,7 +51,7 @@ def _jinja2_filter_split(string, split_char=',') -> list:
 
     @app.context_processor
     def context_processor() -> dict:
-        return dict(regex_item=r'^(?:(?![\s])[ÆØÅæøåa-zA-Z0-9_\s\-]*[ÆØÅæøåa-zA-Z0-9_\-]+)$',
+        return dict(regex_item=REGEX_ITEM,
                     groups=groups.get_all(),
                     categories=inventory.all_categories(),
                     emails=mail.get_all_emails(),

BookingSystem/inventory.py

@@ -8,6 +8,7 @@
 import user
 from __init__ import logger, DATABASE
 from db import read_sql_query
+from sanitizer import APIException
 
 """
 Schema for items and functions to interact with the database.
@@ -110,15 +111,15 @@ def add(item: Item) -> None:
     con = sqlite3.connect(DATABASE)
     if item.exists:
         logger.error(f'{item.id} er allerede i bruk.')
-        raise ValueError(f'{item.id} er allerede i bruk.')
+        raise APIException(f'{item.id} er allerede i bruk.')
     try:
         con.execute(read_sql_query('add_item.sql'), item.__dict__)
         con.commit()
         logger.info(f'La til {item.id}.')
         logger.debug(f'La til {item.id} med verdier: {item.__dict__}')
     except sqlite3.IntegrityError:
         logger.error(f'Ukjent feil ved innlegging av {item.id}.')
-        raise ValueError(f'Ukjent feil ved innlegging av {item.id}.')
+        raise APIException(f'Ukjent feil ved innlegging av {item.id}.')
     finally:
         con.close()
     _update_last_seen(item.id)
@@ -131,7 +132,7 @@ def edit(old_item_id: str, new_item: Item) -> None:
     con = sqlite3.connect(DATABASE)
     if old_item_id.lower() != new_item.id.lower() and new_item.exists:
         logger.error(f'{new_item.id} er allerede i bruk.')
-        raise ValueError(f'{new_item.id} er allerede i bruk.')
+        raise APIException(f'{new_item.id} er allerede i bruk.')
     try:
         sql = 'UPDATE inventory SET id=:id, name=:name, category=:category, included_batteries=:included_batteries WHERE id=:old_item_id'
         con.execute(sql, {**new_item.__dict__, 'old_item_id': old_item_id})
@@ -140,7 +141,7 @@ def edit(old_item_id: str, new_item: Item) -> None:
         logger.debug(f'Redigerte utstyr {old_item_id}, differanse: {new_item.__dict__}')
     except sqlite3.IntegrityError:
         logger.error(f'Ukjent feil ved redigering av {old_item_id}.')
-        raise ValueError(f'Ukjent feil ved redigering av {old_item_id}.')
+        raise APIException(f'Ukjent feil ved redigering av {old_item_id}.')
     finally:
         con.close()
     _update_last_seen(new_item.id)
@@ -160,7 +161,7 @@ def delete(item_id: str) -> None:
         logger.info(f'Slettet utstyr {item_id}.')
     except sqlite3.IntegrityError:
         logger.error(f'{item_id} eksisterer ikke.')
-        raise ValueError(f'{item_id} eksisterer ikke.')
+        raise APIException(f'{item_id} eksisterer ikke.')
     finally:
         con.close()
     audits.audit('ITEM_REM', f'{item_id} ble slettet.')
@@ -171,6 +172,9 @@ def get(item_id: str) -> Item:
     con = sqlite3.connect(DATABASE)
     item = Item(*con.execute(read_sql_query('get_item.sql'), {'id': item_id}).fetchone())
     con.close()
+    if not item:
+        logger.error(f'{item_id} eksisterer ikke.')
+        raise APIException(f'{item_id} eksisterer ikke.')
     return item
 
 
@@ -210,7 +214,7 @@ def _update_last_seen(item_id: str) -> None:
         con.execute(sql, {'id': item_id})
         con.commit()
     except sqlite3.IntegrityError:
-        print(f'{item_id} eksisterer ikke.')
+        logger.error(f'{item_id} eksisterer ikke.')
     finally:
         con.close()
 
@@ -231,7 +235,7 @@ def register_out(item_id: str, userid: str, days: str = 1) -> None:
         logger.info(f'{item_id} er ikke lenger tilgjengelig.')
     except sqlite3.IntegrityError:
         logger.error(f'{item_id} eksisterer ikke.')
-        raise ValueError(f'{item_id} eksisterer ikke.')
+        raise APIException(f'{item_id} eksisterer ikke.')
     finally:
         con.close()
     _update_last_seen(item_id)
@@ -245,7 +249,7 @@ def register_in(item_id: str) -> None:
     try:
         get(item_id)
     except TypeError:
-        raise ValueError(f'{item_id} eksisterer ikke i databasen.')
+        raise APIException(f'{item_id} eksisterer ikke.')
 
     con = sqlite3.connect(DATABASE)
     try:
@@ -255,7 +259,7 @@ def register_in(item_id: str) -> None:
         logger.info(f'{item_id} er nå tilgjengelig.')
     except sqlite3.IntegrityError:
         logger.error(f'{item_id} eksisterer ikke.')
-        raise ValueError(f'{item_id} eksisterer ikke.')
+        raise APIException(f'{item_id} eksisterer ikke.')
     finally:
         con.close()
     _update_last_seen(item_id)