adapting shipcall, times and user to include ValidationError (marshmallow). Adjusting the Schemas for User, Times and Shipcall to be validated with additional input validators. Creating a set of tests for the input validations.

brecal.code-workspace

@@ -0,0 +1,8 @@
+{
+	"folders": [
+		{
+			"path": "."
+		}
+	],
+	"settings": {}
+}

src/server/BreCal/api/shipcalls.py

@@ -1,6 +1,6 @@
 from flask import Blueprint, request
 from webargs.flaskparser import parser
-from marshmallow import Schema, fields
+from marshmallow import Schema, fields, ValidationError
 from ..schemas import model
 from .. import impl
 from ..services.auth_guard import auth_guard
@@ -40,6 +40,12 @@ def PostShipcalls():
     try:
         content = request.get_json(force=True)
         loadedModel = model.ShipcallSchema().load(data=content, many=False, partial=True)
+        
+    except ValidationError as ex:
+        logging.error(ex)
+        print(ex)
+        return json.dumps(f"bad format. \nError Messages: {ex.messages}. \nValid Data: {ex.valid_data}"), 400
+        
     except Exception as ex:
         logging.error(ex)
         print(ex)
@@ -56,6 +62,12 @@ def PutShipcalls():
         content = request.get_json(force=True)
         logging.info(content)
         loadedModel = model.ShipcallSchema().load(data=content, many=False, partial=True)
+        
+    except ValidationError as ex:
+        logging.error(ex)
+        print(ex)
+        return json.dumps(f"bad format. \nError Messages: {ex.messages}. \nValid Data: {ex.valid_data}"), 400
+        
     except Exception as ex:
         logging.error(ex)
         print(ex)

src/server/BreCal/api/times.py

@@ -4,6 +4,7 @@ from .. import impl
 from ..services.auth_guard import auth_guard
 import json
 import logging
+from marshmallow import ValidationError
 
 bp = Blueprint('times', __name__)
 
@@ -29,6 +30,11 @@ def PostTimes():
         # print (content)
         # body = parser.parse(schema, request, location='json')
         loadedModel = model.TimesSchema().load(data=content, many=False, partial=True)
+        
+    except ValidationError as ex:
+        logging.error(ex)
+        print(ex)
+        return json.dumps(f"bad format. \nError Messages: {ex.messages}. \nValid Data: {ex.valid_data}"), 400
 
     except Exception as ex:
         logging.error(ex)
@@ -45,6 +51,11 @@ def PutTimes():
     try:
         content = request.get_json(force=True)
         loadedModel = model.TimesSchema().load(data=content, many=False, partial=True)
+        
+    except ValidationError as ex:
+        logging.error(ex)
+        print(ex)
+        return json.dumps(f"bad format. \nError Messages: {ex.messages}. \nValid Data: {ex.valid_data}"), 400
 
     except Exception as ex:
         logging.error(ex)

src/server/BreCal/api/user.py

@@ -4,6 +4,7 @@ from .. import impl
 from ..services.auth_guard import auth_guard
 import json
 import logging
+from marshmallow import ValidationError
 
 bp = Blueprint('user', __name__)
 
@@ -14,6 +15,11 @@ def PutUser():
     try:
         content = request.get_json(force=True)
         loadedModel = model.UserSchema().load(data=content, many=False, partial=True)
+        
+    except ValidationError as ex:
+        logging.error(ex)
+        print(ex)
+        return json.dumps(f"bad format. \nError Messages: {ex.messages}. \nValid Data: {ex.valid_data}"), 400
 
     except Exception as ex:
         logging.error(ex)

src/server/BreCal/schemas/model.py

@@ -1,5 +1,5 @@
 from dataclasses import field, dataclass
-from marshmallow import Schema, fields, post_load, INCLUDE, ValidationError
+from marshmallow import Schema, fields, INCLUDE, ValidationError, validate, validates
 from marshmallow.fields import Field
 from marshmallow_enum import EnumField
 from enum import IntEnum
@@ -9,6 +9,7 @@ from typing import List
 
 import json
 import datetime
+from BreCal.validators.time_logic import validate_time_exceeds_threshold
 
 def obj_dict(obj):
     if isinstance(obj, datetime.datetime):
@@ -152,34 +153,34 @@ class ParticipantList(Participant):
     pass
 
 class ParticipantAssignmentSchema(Schema):
-    participant_id = fields.Int()
+    participant_id = fields.Integer()
-    type = fields.Int()
+    type = fields.Integer()
 
 class ShipcallSchema(Schema):
     def __init__(self):
         super().__init__(unknown=None)
         pass
 
-    id = fields.Int()
+    id = fields.Integer()
-    ship_id = fields.Int()
+    ship_id = fields.Integer()
-    type = fields.Enum(ShipcallType,  required=True)
+    type = fields.Integer()
     eta = fields.DateTime(Required = False, allow_none=True)
-    voyage = fields.Str(allow_none=True, metadata={'Required':False}) # Solving: RemovedInMarshmallow4Warning: Passing field metadata as keyword arguments is deprecated. Use the explicit `metadata=...` argument instead. Additional metadata: {'Required': False}
+    voyage = fields.String(allow_none=True, metadata={'Required':False}, validate=[validate.Length(max=16)]) # Solving: RemovedInMarshmallow4Warning: Passing field metadata as keyword arguments is deprecated. Use the explicit `metadata=...` argument instead. Additional metadata: {'Required': False}
     etd = fields.DateTime(Required = False, allow_none=True)
-    arrival_berth_id = fields.Int(Required = False, allow_none=True)
+    arrival_berth_id = fields.Integer(Required = False, allow_none=True)
-    departure_berth_id = fields.Int(Required = False, allow_none=True)
+    departure_berth_id = fields.Integer(Required = False, allow_none=True)
     tug_required = fields.Bool(Required = False, allow_none=True)
     pilot_required = fields.Bool(Required = False, allow_none=True)
-    flags = fields.Int(Required = False, allow_none=True)
+    flags = fields.Integer(Required = False, allow_none=True)
     pier_side = fields.Bool(Required = False, allow_none=True)
     bunkering = fields.Bool(Required = False, allow_none=True)
     replenishing_terminal = fields.Bool(Required = False, allow_none=True)
     replenishing_lock = fields.Bool(Required = False, allow_none=True)
-    draft = fields.Float(Required = False, allow_none=True)
+    draft = fields.Float(Required = False, allow_none=True, validate=[validate.Range(min=0, max=20, min_inclusive=False, max_inclusive=True)])
     tidal_window_from = fields.DateTime(Required = False, allow_none=True)
     tidal_window_to = fields.DateTime(Required = False, allow_none=True)
     rain_sensitive_cargo = fields.Bool(Required = False, allow_none=True)
-    recommended_tugs = fields.Int(Required = False, allow_none=True)
+    recommended_tugs = fields.Integer(Required = False, allow_none=True)
     anchored = fields.Bool(Required = False, allow_none=True)
     moored_lock = fields.Bool(Required = False, allow_none=True)
     canceled = fields.Bool(Required = False, allow_none=True)
@@ -297,12 +298,13 @@ class ShipcallId(Schema):
 
 # this is the way!
 
+
 class TimesSchema(Schema):
     def __init__(self):
         super().__init__(unknown=None)
         pass
 
-    id = fields.Int(Required=False)
+    id = fields.Integer(Required=False)
     eta_berth = fields.DateTime(Required = False, allow_none=True)
     eta_berth_fixed = fields.Bool(Required = False, allow_none=True)
     etd_berth = fields.DateTime(Required = False, allow_none=True)
@@ -313,13 +315,13 @@ class TimesSchema(Schema):
     zone_entry_fixed = fields.Bool(Required = False, allow_none=True)
     operations_start = fields.DateTime(Required = False, allow_none=True)
     operations_end  = fields.DateTime(Required = False, allow_none=True)
-    remarks = fields.String(Required = False, allow_none=True)
+    remarks = fields.String(Required = False, allow_none=True, validate=[validate.Length(max=256)])
-    participant_id = fields.Int(Required = True)
+    participant_id = fields.Integer(Required = True)
-    berth_id = fields.Int(Required = False, allow_none = True)
+    berth_id = fields.Integer(Required = False, allow_none = True)
-    berth_info = fields.String(Required = False, allow_none=True)
+    berth_info = fields.String(Required = False, allow_none=True, validate=[validate.Length(max=256)])
     pier_side = fields.Bool(Required = False, allow_none = True)
-    shipcall_id = fields.Int(Required = True)
+    shipcall_id = fields.Integer(Required = True)
-    participant_type = fields.Int(Required = False, allow_none=True)
+    participant_type = fields.Integer(Required = False, allow_none=True)
     ata = fields.DateTime(Required = False, allow_none=True)
     atd = fields.DateTime(Required = False, allow_none=True)
     eta_interval_end = fields.DateTime(Required = False, allow_none=True)
@@ -327,19 +329,38 @@ class TimesSchema(Schema):
     created = fields.DateTime(Required = False, allow_none=True)
     modified = fields.DateTime(Required = False, allow_none=True)
 
+    @validates("eta_berth")
+    def validate_eta_berth(self, value):
+        threshold_exceeded = validate_time_exceeds_threshold(value, months=12)
+        print(threshold_exceeded, value)
+        if threshold_exceeded:
+            raise ValidationError(f"the provided time exceeds the twelve month threshold.")
+
 # deserialize PUT object target
 
 class UserSchema(Schema):
     def __init__(self):
         super().__init__(unknown=None)
         pass
-    id = fields.Int(required=True)
+    id = fields.Integer(required=True)
-    first_name = fields.Str(allow_none=True, metadata={'Required':False})
+    first_name = fields.String(allow_none=True, metadata={'Required':False}, validate=[validate.Length(max=64)])
-    last_name = fields.Str(allow_none=True, metadata={'Required':False})
+    last_name = fields.String(allow_none=True, metadata={'Required':False}, validate=[validate.Length(max=64)])
-    user_phone = fields.Str(allow_none=True, metadata={'Required':False})
+    user_phone = fields.String(allow_none=True, metadata={'Required':False})
-    user_email = fields.Str(allow_none=True, metadata={'Required':False})
+    user_email = fields.String(allow_none=True, metadata={'Required':False}, validate=[validate.Length(max=64)])
-    old_password = fields.Str(allow_none=True, metadata={'Required':False})
+    old_password = fields.String(allow_none=True, metadata={'Required':False}, validate=[validate.Length(max=128)])
-    new_password = fields.Str(allow_none=True, metadata={'Required':False})
+    new_password = fields.String(allow_none=True, metadata={'Required':False}, validate=[validate.Length(min=6, max=128)])
+
+    @validates("user_phone")
+    def validate_user_phone(self, value):
+        valid_characters = list(map(str,range(0,10)))+["+", " "]
+        if not all([v in valid_characters for v in value]):
+            raise ValidationError(f"one of the phone number values is not valid.")
+        
+    @validates("user_email")
+    def validate_user_email(self, value):
+        if not "@" in value:
+            raise ValidationError(f"invalid email address")
+    
 
 @dataclass
 class Times:

src/server/tests/schemas/test_model.py

@@ -0,0 +1,80 @@
+from marshmallow import ValidationError
+import pytest
+from BreCal.schemas.model import ShipcallSchema
+
+
+@pytest.fixture(scope="function") # function: destroy fixture at the end of each test
+def prepare_shipcall_content():
+    import datetime
+    from BreCal.stubs.shipcall import get_shipcall_simple
+    shipcall_stub = get_shipcall_simple()
+    content = shipcall_stub.__dict__
+    content["participants"] = []
+    content = {k:v.isoformat() if isinstance(v, datetime.datetime) else v for k,v in content.items()}
+    return locals()
+
+def test_shipcall_input_validation_draft(prepare_shipcall_content):
+    content = prepare_shipcall_content["content"]
+    content["draft"] = 24.11
+
+    schemaModel = ShipcallSchema()
+    with pytest.raises(ValidationError, match="Must be greater than 0 and less than or equal to 20."):
+        loadedModel = schemaModel.load(data=content, many=False, partial=True)
+    return
+
+def test_shipcall_input_validation_voyage(prepare_shipcall_content):
+    content = prepare_shipcall_content["content"]
+    content["voyage"] = "".join(list(map(str,list(range(0,24))))) # 38 characters
+
+    schemaModel = ShipcallSchema()
+    with pytest.raises(ValidationError, match="Longer than maximum length "):
+        loadedModel = schemaModel.load(data=content, many=False, partial=True)
+    return
+
+
+@pytest.fixture(scope="function") # function: destroy fixture at the end of each test
+def prepare_user_content():
+    import datetime
+    from BreCal.stubs.user import get_user_simple
+    from BreCal.schemas.model import UserSchema
+    schemaModel = UserSchema()
+
+    user_stub = get_user_simple()
+    content = user_stub.__dict__
+    content = {k:v.isoformat() if isinstance(v, datetime.datetime) else v for k,v in content.items()}
+    content = {k:v for k,v in content.items() if k in list(schemaModel.fields.keys())}
+    content["old_password"] = "myfavoritedog123"
+    content["new_password"] = "SecuRepassW0rd!"
+    return locals()
+
+
+def test_input_validation_berth_phone_number_is_valid(prepare_user_content):
+    content, schemaModel = prepare_user_content["content"], prepare_user_content["schemaModel"]
+    content["user_phone"] = "+49123 45678912" # whitespace and + are valid
+
+    loadedModel = schemaModel.load(data=content, many=False, partial=True)
+    return
+
+def test_input_validation_berth_phone_number_is_invalid(prepare_user_content):
+    content, schemaModel = prepare_user_content["content"], prepare_user_content["schemaModel"]
+    content["user_phone"] = "+49123 45678912!" # ! is invalid
+
+    with pytest.raises(ValidationError, match="one of the phone number values is not valid."):
+        loadedModel = schemaModel.load(data=content, many=False, partial=True)
+    return
+
+def test_input_validation_new_password_too_short(prepare_user_content):
+    content, schemaModel = prepare_user_content["content"], prepare_user_content["schemaModel"]
+    content["new_password"] = "1234" # must have between 6 and 128 characters
+
+    with pytest.raises(ValidationError, match="Length must be between 6 and 128."):
+        loadedModel = schemaModel.load(data=content, many=False, partial=True)
+    return
+
+def test_input_validation_user_email_invalid(prepare_user_content):
+    content, schemaModel = prepare_user_content["content"], prepare_user_content["schemaModel"]
+    content["user_email"] = "userbrecal.com" # forgot @ -> invalid
+
+    with pytest.raises(ValidationError, match="invalid email address"):
+        loadedModel = schemaModel.load(data=content, many=False, partial=True)
+    return