implemented InputValidationShip and activated it for POST, PUT and DELETE requests. Created 10 unit tests to check for the functionality. Refactored some functions to avoid circular importing.
This commit is contained in:
parent
be58c44a2f
commit
4c1b230de9
@ -7,7 +7,7 @@ import json
|
||||
import logging
|
||||
|
||||
from BreCal.validators.input_validation import check_if_user_is_bsmd_type
|
||||
|
||||
from BreCal.validators.input_validation_ship import InputValidationShip
|
||||
|
||||
bp = Blueprint('ships', __name__)
|
||||
|
||||
@ -38,6 +38,10 @@ def PostShip():
|
||||
|
||||
content = request.get_json(force=True)
|
||||
loadedModel = model.ShipSchema().load(data=content, many=False, partial=True)
|
||||
|
||||
# validate the request data & user permissions
|
||||
InputValidationShip.evaluate_post_data(user_data, loadedModel, content)
|
||||
|
||||
except Exception as ex:
|
||||
logging.error(ex)
|
||||
print(ex)
|
||||
@ -54,14 +58,12 @@ def PutShip():
|
||||
# read the user data from the JWT token (set when login is performed)
|
||||
user_data = check_jwt()
|
||||
|
||||
# check, whether the user belongs to a participant, which is of type ParticipantType.BSMD
|
||||
# as ParticipantType is an IntFlag, a user belonging to multiple groups is properly evaluated.
|
||||
is_bsmd = check_if_user_is_bsmd_type(user_data)
|
||||
if not is_bsmd:
|
||||
raise ValidationError(f"current user does not belong to BSMD. Cannot post shipcalls. Found user data: {user_data}")
|
||||
|
||||
content = request.get_json(force=True)
|
||||
loadedModel = model.ShipSchema().load(data=content, many=False, partial=True, unknown=EXCLUDE)
|
||||
loadedModel = model.Ship().load(data=content, many=False, partial=True, unknown=EXCLUDE)
|
||||
|
||||
# validate the request data & user permissions
|
||||
InputValidationShip.evaluate_put_data(user_data, loadedModel, content)
|
||||
|
||||
except Exception as ex:
|
||||
logging.error(ex)
|
||||
print(ex)
|
||||
@ -77,18 +79,17 @@ def DeleteShip():
|
||||
try:
|
||||
# read the user data from the JWT token (set when login is performed)
|
||||
user_data = check_jwt()
|
||||
|
||||
# check, whether the user belongs to a participant, which is of type ParticipantType.BSMD
|
||||
# as ParticipantType is an IntFlag, a user belonging to multiple groups is properly evaluated.
|
||||
is_bsmd = check_if_user_is_bsmd_type(user_data)
|
||||
if not is_bsmd:
|
||||
raise ValidationError(f"current user does not belong to BSMD. Cannot post shipcalls. Found user data: {user_data}")
|
||||
ship_id = request.args.get("id")
|
||||
|
||||
if 'id' in request.args:
|
||||
options = {}
|
||||
options["id"] = request.args.get("id")
|
||||
else:
|
||||
return json.dumps("no id provided"), 400
|
||||
|
||||
# validate the request data & user permissions
|
||||
InputValidationShip.evaluate_delete_data(user_data, ship_id)
|
||||
|
||||
except Exception as ex:
|
||||
logging.error(ex)
|
||||
print(ex)
|
||||
|
||||
@ -10,6 +10,7 @@ from typing import List
|
||||
import json
|
||||
import datetime
|
||||
from BreCal.validators.time_logic import validate_time_exceeds_threshold
|
||||
from BreCal.validators.validation_base_utils import check_if_string_has_special_characters
|
||||
from BreCal.database.enums import ParticipantType, ParticipantFlag
|
||||
|
||||
# from BreCal. ... import check_if_user_is_bsmd_type
|
||||
@ -165,7 +166,7 @@ class Participant(Schema):
|
||||
|
||||
|
||||
@validates("flags")
|
||||
def validate_type(self, value):
|
||||
def validate_flags(self, value):
|
||||
# e.g., when an IntFlag has the values 1,2,4; the maximum valid value is 7
|
||||
max_int = sum([int(val) for val in list(ParticipantFlag._value2member_map_.values())])
|
||||
min_int = 0
|
||||
@ -472,8 +473,8 @@ class ShipSchema(Schema):
|
||||
imo = fields.Int(allow_none=False, metadata={'Required':True})
|
||||
callsign = fields.String(allow_none=True, metadata={'Required':False})
|
||||
participant_id = fields.Int(allow_none=True, metadata={'Required':False})
|
||||
length = fields.Float(allow_none=True, metadata={'Required':False})
|
||||
width = fields.Float(allow_none=True, metadata={'Required':False})
|
||||
length = fields.Float(allow_none=True, metadata={'Required':False}, validate=[validate.Range(min=0, max=1000, min_inclusive=False, max_inclusive=False)])
|
||||
width = fields.Float(allow_none=True, metadata={'Required':False}, validate=[validate.Range(min=0, max=100, min_inclusive=False, max_inclusive=False)])
|
||||
is_tug = fields.Bool(allow_none=True, metadata={'Required':False}, default=False)
|
||||
bollard_pull = fields.Int(allow_none=True, metadata={'Required':False})
|
||||
eni = fields.Int(allow_none=True, metadata={'Required':False})
|
||||
@ -481,6 +482,34 @@ class ShipSchema(Schema):
|
||||
modified = fields.DateTime(allow_none=True, metadata={'Required':False})
|
||||
deleted = fields.Bool(allow_none=True, metadata={'Required':False}, default=False)
|
||||
|
||||
@validates("name")
|
||||
def validate_name(self, value):
|
||||
character_length = len(str(value))
|
||||
if character_length>=64:
|
||||
raise ValidationError(f"'name' argument should have at max. 63 characters")
|
||||
|
||||
if check_if_string_has_special_characters(value):
|
||||
raise ValidationError(f"'name' argument should not have special characters.")
|
||||
return
|
||||
|
||||
@validates("imo")
|
||||
def validate_imo(self, value):
|
||||
imo_length = len(str(value))
|
||||
if imo_length != 7:
|
||||
raise ValidationError(f"'imo' should be a 7-digit number")
|
||||
return
|
||||
|
||||
@validates("callsign")
|
||||
def validate_callsign(self, value):
|
||||
if value is not None:
|
||||
callsign_length = len(str(value))
|
||||
if callsign_length>8:
|
||||
raise ValidationError(f"'callsign' argument should not have more than 8 characters")
|
||||
|
||||
if check_if_string_has_special_characters(value):
|
||||
raise ValidationError(f"'callsign' argument should not have special characters.")
|
||||
return
|
||||
|
||||
|
||||
class TimesId(Schema):
|
||||
pass
|
||||
|
||||
@ -36,3 +36,27 @@ def get_ship_simple():
|
||||
)
|
||||
return ship
|
||||
|
||||
def get_stub_valid_ship():
|
||||
post_data = {
|
||||
'name': 'BOTHNIABORG',
|
||||
'imo': 9267728,
|
||||
'callsign': "PBIO",
|
||||
'participant_id': None,
|
||||
'length': 153.05,
|
||||
'width': 21.8,
|
||||
'is_tug': 0,
|
||||
'bollard_pull': None,
|
||||
'eni': None,
|
||||
'created': '2023-10-04 11:52:32',
|
||||
'modified': None,
|
||||
'deleted': 0
|
||||
}
|
||||
return post_data
|
||||
|
||||
def get_stub_valid_ship_loaded_model(post_data=None):
|
||||
from BreCal.schemas import model
|
||||
if post_data is None:
|
||||
post_data = get_stub_valid_ship()
|
||||
loadedModel = model.ShipSchema().load(data=post_data, many=False, partial=True)
|
||||
return loadedModel
|
||||
|
||||
@ -15,7 +15,8 @@ from BreCal.impl.berths import GetBerths
|
||||
from BreCal.database.enums import ParticipantType
|
||||
|
||||
|
||||
from BreCal.validators.input_validation_utils import check_if_user_is_bsmd_type, get_participant_id_dictionary, check_if_ship_id_is_valid, check_if_berth_id_is_valid, check_if_participant_ids_are_valid, get_berth_id_dictionary, check_if_string_has_special_characters, get_ship_id_dictionary, check_if_int_is_valid_flag
|
||||
from BreCal.validators.input_validation_utils import check_if_user_is_bsmd_type, get_participant_id_dictionary, check_if_ship_id_is_valid, check_if_berth_id_is_valid, check_if_participant_ids_are_valid, get_berth_id_dictionary, get_ship_id_dictionary
|
||||
from BreCal.validators.validation_base_utils import check_if_string_has_special_characters
|
||||
|
||||
def validation_error_default_asserts(response):
|
||||
"""creates assertions, when the response does not fail as expected. This function is extensively used in the input validation pytests"""
|
||||
@ -24,7 +25,6 @@ def validation_error_default_asserts(response):
|
||||
return
|
||||
|
||||
|
||||
|
||||
def validate_posted_shipcall_data(user_data:dict, loadedModel:dict, content:dict):
|
||||
"""this function applies more complex validation functions to data, which is sent to a post-request of shipcalls"""
|
||||
# DEPRECATED: this function has been refactored into InputValidationShipcall (see methods for POST and PUT evaluation)
|
||||
@ -280,4 +280,4 @@ class ParticipantValidation(DataclassValidation):
|
||||
]
|
||||
]
|
||||
return all_rules
|
||||
|
||||
|
||||
|
||||
140
src/server/BreCal/validators/input_validation_ship.py
Normal file
140
src/server/BreCal/validators/input_validation_ship.py
Normal file
@ -0,0 +1,140 @@
|
||||
import typing
|
||||
import json
|
||||
import datetime
|
||||
from abc import ABC, abstractmethod
|
||||
from marshmallow import ValidationError
|
||||
from string import ascii_letters, digits
|
||||
|
||||
from BreCal.schemas.model import Ship, Shipcall, Berth, User, Participant, ShipcallType
|
||||
from BreCal.impl.participant import GetParticipant
|
||||
from BreCal.impl.ships import GetShips
|
||||
from BreCal.impl.berths import GetBerths
|
||||
|
||||
from BreCal.database.enums import ParticipantType, ParticipantFlag
|
||||
from BreCal.validators.input_validation_utils import check_if_user_is_bsmd_type, check_if_ship_id_is_valid, check_if_berth_id_is_valid, check_if_participant_ids_are_valid, check_if_participant_ids_and_types_are_valid, get_shipcall_id_dictionary, get_participant_type_from_user_data
|
||||
from BreCal.database.sql_handler import execute_sql_query_standalone
|
||||
from BreCal.validators.validation_base_utils import check_if_int_is_valid_flag
|
||||
from BreCal.validators.validation_base_utils import check_if_string_has_special_characters
|
||||
import werkzeug
|
||||
|
||||
class InputValidationShip():
|
||||
"""
|
||||
This class combines a complex set of individual input validation functions into a joint object.
|
||||
It uses static methods, so the object does not need to be instantiated, but functions can be called immediately.
|
||||
|
||||
Example:
|
||||
InputValidationShip.evaluate(user_data, loadedModel, content)
|
||||
|
||||
When the data violates one of the rules, a marshmallow.ValidationError is raised, which details the issues.
|
||||
"""
|
||||
def __init__(self) -> None:
|
||||
pass
|
||||
|
||||
@staticmethod
|
||||
def evaluate_post_data(user_data:dict, loadedModel:dict, content:dict):
|
||||
# 1.) Only users of type BSMD are allowed to POST
|
||||
InputValidationShip.check_user_is_bsmd_type(user_data)
|
||||
|
||||
# 2.) The ship IMOs are used as matching keys. They must be unique in the database.
|
||||
InputValidationShip.check_ship_imo_already_exists(loadedModel)
|
||||
|
||||
# 3.) Check for reasonable Values (see BreCal.schemas.model.ShipSchema)
|
||||
InputValidationShip.optionally_evaluate_bollard_pull_value(content)
|
||||
return
|
||||
|
||||
@staticmethod
|
||||
def evaluate_put_data(user_data:dict, loadedModel:dict, content:dict):
|
||||
# 1.) Only users of type BSMD are allowed to PUT
|
||||
InputValidationShip.check_user_is_bsmd_type(user_data)
|
||||
|
||||
# 2.) The IMO number field may not be changed
|
||||
InputValidationShip.put_content_may_not_contain_imo_number(content)
|
||||
|
||||
# 3.) Check for reasonable Values (see BreCal.schemas.model.ShipSchema)
|
||||
InputValidationShip.optionally_evaluate_bollard_pull_value(content)
|
||||
|
||||
# 4.) ID field is mandatory
|
||||
InputValidationShip.content_contains_ship_id(content)
|
||||
return
|
||||
|
||||
@staticmethod
|
||||
def evaluate_delete_data(user_data:dict, ship_id:int):
|
||||
# 1.) Only users of type BSMD are allowed to PUT
|
||||
InputValidationShip.check_user_is_bsmd_type(user_data)
|
||||
|
||||
# 2.) The dataset entry may not be deleted already
|
||||
InputValidationShip.check_if_entry_is_already_deleted(ship_id)
|
||||
return
|
||||
|
||||
@staticmethod
|
||||
def optionally_evaluate_bollard_pull_value(content:dict):
|
||||
bollard_pull = content.get("bollard_pull",None)
|
||||
is_tug = content.get("is_tug", None)
|
||||
|
||||
if bollard_pull is not None:
|
||||
if not is_tug:
|
||||
raise ValidationError(f"'bollard_pull' is only allowed, when a ship is a tug ('is_tug').")
|
||||
|
||||
if (not (0 < bollard_pull < 500)) & (is_tug):
|
||||
raise ValidationError(f"when a ship is a tug, the bollard pull must be 0 < value < 500. ")
|
||||
|
||||
@staticmethod
|
||||
def check_user_is_bsmd_type(user_data:dict):
|
||||
is_bsmd = check_if_user_is_bsmd_type(user_data)
|
||||
if not is_bsmd:
|
||||
raise ValidationError(f"current user does not belong to BSMD. Cannot post shipcalls. Found user data: {user_data}")
|
||||
|
||||
@staticmethod
|
||||
def check_ship_imo_already_exists(loadedModel:dict):
|
||||
# get the ships, convert them to a list of JSON dictionaries
|
||||
response, status_code, header = GetShips(token=None)
|
||||
ships = json.loads(response)
|
||||
|
||||
# extract only the 'imo' values
|
||||
ship_imos = [ship.get("imo") for ship in ships]
|
||||
|
||||
# check, if the imo in the POST-request already exists in the list
|
||||
imo_already_exists = loadedModel.get("imo") in ship_imos
|
||||
if imo_already_exists:
|
||||
raise ValidationError(f"the provided ship IMO {loadedModel.get('imo')} already exists. A ship may only be added, if there is no other ship with the same IMO number.")
|
||||
return
|
||||
|
||||
@staticmethod
|
||||
def put_content_may_not_contain_imo_number(content:dict):
|
||||
put_data_ship_imo = content.get("imo",None)
|
||||
if put_data_ship_imo is not None:
|
||||
raise ValidationError(f"The IMO number field may not be changed since it serves the purpose of a primary (matching) key.")
|
||||
return
|
||||
|
||||
@staticmethod
|
||||
def content_contains_ship_id(content:dict):
|
||||
put_data_ship_id = content.get('id',None)
|
||||
if put_data_ship_id is None:
|
||||
raise ValidationError(f"The id field is required.")
|
||||
return
|
||||
|
||||
@staticmethod
|
||||
def check_if_entry_is_already_deleted(ship_id:int):
|
||||
"""
|
||||
When calling a delete request for ships, the dataset may not be deleted already. This method
|
||||
makes sure, that the request contains and ID, has a matching entry in the database, and the
|
||||
database entry may not have a deletion state already.
|
||||
"""
|
||||
if ship_id is None:
|
||||
raise ValidationError(f"The ship_id must be provided.")
|
||||
|
||||
response, status_code, header = GetShips(token=None)
|
||||
ships = json.loads(response)
|
||||
existing_database_entries = [ship for ship in ships if ship.get("id")==ship_id]
|
||||
if len(existing_database_entries)==0:
|
||||
raise ValidationError(f"Could not find a ship with the specified ID. Selected: {ship_id}")
|
||||
|
||||
existing_database_entry = existing_database_entries[0]
|
||||
|
||||
deletion_state = existing_database_entry.get("deleted",None)
|
||||
if deletion_state:
|
||||
raise ValidationError(f"The selected ship entry is already deleted.")
|
||||
return
|
||||
|
||||
|
||||
|
||||
@ -11,8 +11,10 @@ from BreCal.impl.ships import GetShips
|
||||
from BreCal.impl.berths import GetBerths
|
||||
|
||||
from BreCal.database.enums import ParticipantType, ParticipantFlag
|
||||
from BreCal.validators.input_validation_utils import check_if_user_is_bsmd_type, check_if_ship_id_is_valid, check_if_berth_id_is_valid, check_if_participant_ids_are_valid, check_if_participant_ids_and_types_are_valid, check_if_string_has_special_characters, get_shipcall_id_dictionary, get_participant_type_from_user_data, check_if_int_is_valid_flag
|
||||
from BreCal.validators.input_validation_utils import check_if_user_is_bsmd_type, check_if_ship_id_is_valid, check_if_berth_id_is_valid, check_if_participant_ids_are_valid, check_if_participant_ids_and_types_are_valid, get_shipcall_id_dictionary, get_participant_type_from_user_data
|
||||
from BreCal.database.sql_handler import execute_sql_query_standalone
|
||||
from BreCal.validators.validation_base_utils import check_if_int_is_valid_flag
|
||||
from BreCal.validators.validation_base_utils import check_if_string_has_special_characters
|
||||
import werkzeug
|
||||
|
||||
class InputValidationShipcall():
|
||||
|
||||
@ -1,6 +1,5 @@
|
||||
import logging
|
||||
import json
|
||||
from string import ascii_letters, digits
|
||||
from collections import Counter
|
||||
|
||||
from BreCal.impl.participant import GetParticipant
|
||||
@ -159,21 +158,5 @@ def check_if_participant_ids_and_types_are_valid(participants:list[dict[str,int]
|
||||
|
||||
|
||||
|
||||
def check_if_string_has_special_characters(text:str):
|
||||
"""
|
||||
check, whether there are any characters within the provided string, which are not found in the ascii letters or digits
|
||||
ascii_letters: abcd (...) and ABCD (...)
|
||||
digits: 0123 (...)
|
||||
|
||||
Source: https://stackoverflow.com/questions/57062794/is-there-a-way-to-check-if-a-string-contains-special-characters
|
||||
User: https://stackoverflow.com/users/10035985/andrej-kesely
|
||||
returns bool
|
||||
"""
|
||||
return bool(set(text).difference(ascii_letters + digits))
|
||||
|
||||
def check_if_int_is_valid_flag(value, enum_object):
|
||||
# e.g., when an IntFlag has the values 1,2,4; the maximum valid value is 7
|
||||
max_int = sum([int(val) for val in list(enum_object._value2member_map_.values())])
|
||||
return 0 < value <= max_int
|
||||
|
||||
|
||||
|
||||
20
src/server/BreCal/validators/validation_base_utils.py
Normal file
20
src/server/BreCal/validators/validation_base_utils.py
Normal file
@ -0,0 +1,20 @@
|
||||
from string import ascii_letters, digits
|
||||
|
||||
|
||||
def check_if_string_has_special_characters(text:str):
|
||||
"""
|
||||
check, whether there are any characters within the provided string, which are not found in the ascii letters or digits
|
||||
ascii_letters: abcd (...) and ABCD (...)
|
||||
digits: 0123 (...)
|
||||
|
||||
Source: https://stackoverflow.com/questions/57062794/is-there-a-way-to-check-if-a-string-contains-special-characters
|
||||
User: https://stackoverflow.com/users/10035985/andrej-kesely
|
||||
returns bool
|
||||
"""
|
||||
return bool(set(text).difference(ascii_letters + digits))
|
||||
|
||||
|
||||
def check_if_int_is_valid_flag(value, enum_object):
|
||||
# e.g., when an IntFlag has the values 1,2,4; the maximum valid value is 7
|
||||
max_int = sum([int(val) for val in list(enum_object._value2member_map_.values())])
|
||||
return 0 < value <= max_int
|
||||
235
src/server/tests/validators/test_input_validation_ship.py
Normal file
235
src/server/tests/validators/test_input_validation_ship.py
Normal file
@ -0,0 +1,235 @@
|
||||
import pytest
|
||||
|
||||
import os
|
||||
import jwt
|
||||
import json
|
||||
import requests
|
||||
import datetime
|
||||
import werkzeug
|
||||
import re
|
||||
from marshmallow import ValidationError
|
||||
|
||||
from BreCal import local_db
|
||||
from BreCal.schemas import model
|
||||
|
||||
from BreCal.impl.ships import GetShips
|
||||
|
||||
from BreCal.schemas.model import Participant_Assignment, EvaluationType, ShipcallType
|
||||
from BreCal.stubs.ship import get_stub_valid_ship, get_stub_valid_ship_loaded_model
|
||||
from BreCal.validators.input_validation import validation_error_default_asserts
|
||||
from BreCal.schemas.model import ParticipantType
|
||||
from BreCal.validators.input_validation_ship import InputValidationShip
|
||||
|
||||
instance_path = os.path.join(os.path.expanduser('~'), "brecal", "src", "server", "instance", "instance")
|
||||
local_db.initPool(os.path.dirname(instance_path), connection_filename="connection_data_local.json")
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def get_stub_token():
|
||||
"""
|
||||
performs a login to the user 'maxm' and returns the respective url and the token. The token will be used in
|
||||
further requests in the following format (example of post-request):
|
||||
requests.post(f"{url}/shipcalls", headers={"Content-Type":"text", "Authorization":f"Bearer {token}"}, json=post_data)
|
||||
|
||||
"""
|
||||
port = 9013
|
||||
url = f"http://127.0.0.1:{port}"
|
||||
|
||||
# set the JWT key
|
||||
os.environ['SECRET_KEY'] = 'zdiTz8P3jXOc7jztIQAoelK4zztyuCpJ'
|
||||
|
||||
try:
|
||||
response = requests.post(f"{url}/login", json=jwt.decode("eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJ1c2VybmFtZSI6Im1heG0iLCJwYXNzd29yZCI6IlN0YXJ0MTIzNCJ9.uIrbz3g-IwwTLz6C1zXELRGtAtRJ_myYJ4J4x0ozjAI", key=os.environ.get("SECRET_KEY"), algorithms=["HS256"]))
|
||||
except requests.ConnectionError as err:
|
||||
raise AssertionError(f"could not establish a connection to the default url. Did you start an instance of the local database at port {port}? Looking for a connection to {url}")
|
||||
user = response.json()
|
||||
token = user.get("token")
|
||||
return locals()
|
||||
|
||||
def test_():
|
||||
ivs = InputValidationShip()
|
||||
return
|
||||
|
||||
# length: 0 < value < 1000
|
||||
# width: 0 < value < 100
|
||||
|
||||
|
||||
|
||||
def test_input_validation_ship_fails_when_length_is_incorrect():
|
||||
with pytest.raises(ValidationError, match=re.escape("Must be greater than 0 and less than 1000.")):
|
||||
post_data = get_stub_valid_ship()
|
||||
post_data["length"] = 0
|
||||
loadedModel = model.ShipSchema().load(data=post_data, many=False, partial=True)
|
||||
|
||||
with pytest.raises(ValidationError, match=re.escape("Must be greater than 0 and less than 1000.")):
|
||||
post_data = get_stub_valid_ship()
|
||||
post_data["length"] = 1000
|
||||
loadedModel = model.ShipSchema().load(data=post_data, many=False, partial=True)
|
||||
|
||||
# success
|
||||
post_data = get_stub_valid_ship()
|
||||
post_data["length"] = 123
|
||||
loadedModel = model.ShipSchema().load(data=post_data, many=False, partial=True)
|
||||
return
|
||||
|
||||
def test_input_validation_ship_fails_when_width_is_incorrect():
|
||||
with pytest.raises(ValidationError, match=re.escape("Must be greater than 0 and less than 100.")):
|
||||
post_data = get_stub_valid_ship()
|
||||
post_data["width"] = 0
|
||||
loadedModel = model.ShipSchema().load(data=post_data, many=False, partial=True)
|
||||
|
||||
with pytest.raises(ValidationError, match=re.escape("Must be greater than 0 and less than 100.")):
|
||||
post_data = get_stub_valid_ship()
|
||||
post_data["width"] = 100
|
||||
loadedModel = model.ShipSchema().load(data=post_data, many=False, partial=True)
|
||||
|
||||
# success
|
||||
post_data = get_stub_valid_ship()
|
||||
post_data["width"] = 12
|
||||
loadedModel = model.ShipSchema().load(data=post_data, many=False, partial=True)
|
||||
return
|
||||
|
||||
def test_input_validation_ship_fails_when_name_is_incorrect():
|
||||
with pytest.raises(ValidationError, match=re.escape("'name' argument should have at max. 63 characters")):
|
||||
post_data = get_stub_valid_ship()
|
||||
post_data["name"] = "0123456789012345678901234567890123456789012345678901234567890123"
|
||||
loadedModel = model.ShipSchema().load(data=post_data, many=False, partial=True)
|
||||
|
||||
with pytest.raises(ValidationError, match=re.escape("'name' argument should not have special characters.")):
|
||||
post_data = get_stub_valid_ship()
|
||||
post_data["name"] = '👽'
|
||||
loadedModel = model.ShipSchema().load(data=post_data, many=False, partial=True)
|
||||
|
||||
post_data = get_stub_valid_ship()
|
||||
post_data["name"] = "012345678901234567890123456789012345678901234567890123456789012"
|
||||
loadedModel = model.ShipSchema().load(data=post_data, many=False, partial=True)
|
||||
return
|
||||
|
||||
def test_input_validation_ship_fails_when_callsign_is_incorrect():
|
||||
with pytest.raises(ValidationError, match=re.escape("'callsign' argument should not have more than 8 characters")):
|
||||
post_data = get_stub_valid_ship()
|
||||
post_data["callsign"] = "123456789"
|
||||
loadedModel = model.ShipSchema().load(data=post_data, many=False, partial=True)
|
||||
|
||||
with pytest.raises(ValidationError, match=re.escape("'callsign' argument should not have special characters.")):
|
||||
post_data = get_stub_valid_ship()
|
||||
post_data["callsign"] = '👽'
|
||||
loadedModel = model.ShipSchema().load(data=post_data, many=False, partial=True)
|
||||
|
||||
# success
|
||||
post_data = get_stub_valid_ship()
|
||||
post_data["callsign"] = 'PBIO'
|
||||
loadedModel = model.ShipSchema().load(data=post_data, many=False, partial=True)
|
||||
|
||||
# success
|
||||
post_data = get_stub_valid_ship()
|
||||
post_data["callsign"] = None
|
||||
loadedModel = model.ShipSchema().load(data=post_data, many=False, partial=True)
|
||||
|
||||
return
|
||||
|
||||
def test_input_validation_ship_fails_when_imo_is_incorrect():
|
||||
# imo must have exactly 7 digits and can't be None
|
||||
with pytest.raises(ValidationError, match=re.escape("'imo' should be a 7-digit number")):
|
||||
post_data = get_stub_valid_ship()
|
||||
post_data["imo"] = 123456
|
||||
loadedModel = model.ShipSchema().load(data=post_data, many=False, partial=True)
|
||||
|
||||
with pytest.raises(ValidationError, match=re.escape("'imo' should be a 7-digit number")):
|
||||
post_data = get_stub_valid_ship()
|
||||
post_data["imo"] = 12345678
|
||||
loadedModel = model.ShipSchema().load(data=post_data, many=False, partial=True)
|
||||
|
||||
with pytest.raises(ValidationError, match=re.escape("Field may not be null.")):
|
||||
post_data = get_stub_valid_ship()
|
||||
post_data["imo"] = None
|
||||
loadedModel = model.ShipSchema().load(data=post_data, many=False, partial=True)
|
||||
|
||||
# success
|
||||
post_data = get_stub_valid_ship()
|
||||
post_data["imo"] = 1234567
|
||||
loadedModel = model.ShipSchema().load(data=post_data, many=False, partial=True)
|
||||
return
|
||||
|
||||
def test_input_validation_ship_fails_when_bollard_pull_and_tug_values_are_set():
|
||||
ivs = InputValidationShip()
|
||||
|
||||
with pytest.raises(ValidationError, match=re.escape("'bollard_pull' is only allowed, when a ship is a tug ('is_tug').")):
|
||||
content = {'is_tug':0, 'bollard_pull':230}
|
||||
ivs.optionally_evaluate_bollard_pull_value(content)
|
||||
|
||||
with pytest.raises(ValidationError, match=re.escape("'bollard_pull' is only allowed, when a ship is a tug ('is_tug').")):
|
||||
content = {'is_tug':None, 'bollard_pull':230}
|
||||
ivs.optionally_evaluate_bollard_pull_value(content)
|
||||
|
||||
content = {'is_tug':0, 'bollard_pull':None}
|
||||
ivs.optionally_evaluate_bollard_pull_value(content)
|
||||
|
||||
content = {'is_tug':1, 'bollard_pull':None}
|
||||
ivs.optionally_evaluate_bollard_pull_value(content)
|
||||
|
||||
content = {'is_tug':1, 'bollard_pull':125}
|
||||
ivs.optionally_evaluate_bollard_pull_value(content)
|
||||
|
||||
with pytest.raises(ValidationError, match=re.escape("when a ship is a tug, the bollard pull must be 0 < value < 500.")):
|
||||
content = {'is_tug':1, 'bollard_pull':-1}
|
||||
ivs.optionally_evaluate_bollard_pull_value(content)
|
||||
|
||||
with pytest.raises(ValidationError, match=re.escape("when a ship is a tug, the bollard pull must be 0 < value < 500.")):
|
||||
content = {'is_tug':1, 'bollard_pull':0}
|
||||
ivs.optionally_evaluate_bollard_pull_value(content)
|
||||
|
||||
with pytest.raises(ValidationError, match=re.escape("when a ship is a tug, the bollard pull must be 0 < value < 500.")):
|
||||
content = {'is_tug':1, 'bollard_pull':500}
|
||||
ivs.optionally_evaluate_bollard_pull_value(content)
|
||||
|
||||
with pytest.raises(ValidationError, match=re.escape("when a ship is a tug, the bollard pull must be 0 < value < 500.")):
|
||||
content = {'is_tug':1, 'bollard_pull':501}
|
||||
ivs.optionally_evaluate_bollard_pull_value(content)
|
||||
return
|
||||
|
||||
def test_input_validation_ship_post_request_fails_when_ship_imo_already_exists():
|
||||
# get the ships, convert them to a list of JSON dictionaries
|
||||
response, status_code, header = GetShips(token=None)
|
||||
ships = json.loads(response)
|
||||
|
||||
# extract only the 'imo' values
|
||||
ship_imos = [ship.get("imo") for ship in ships]
|
||||
|
||||
post_data = get_stub_valid_ship()
|
||||
post_data["imo"] = ship_imos[-1] # assign one of the IMOs, which already exist
|
||||
loadedModel = get_stub_valid_ship_loaded_model(post_data)
|
||||
content = post_data
|
||||
|
||||
with pytest.raises(ValidationError, match="the provided ship IMO 9186687 already exists. A ship may only be added, if there is no other ship with the same IMO number."):
|
||||
InputValidationShip.check_ship_imo_already_exists(loadedModel)
|
||||
return
|
||||
|
||||
|
||||
|
||||
def test_input_validation_ship_put_request_fails_when_ship_imo_should_be_changed():
|
||||
# get the ships, convert them to a list of JSON dictionaries
|
||||
response, status_code, header = GetShips(token=None)
|
||||
ships = json.loads(response)
|
||||
selected_ship = ships[-1] # select one of the ships; in this case the last one.
|
||||
|
||||
put_data = get_stub_valid_ship()
|
||||
put_data["imo"] = selected_ship.get("imo")+1 # assign one of the IMOs, which already exist
|
||||
|
||||
loadedModel = get_stub_valid_ship_loaded_model(put_data)
|
||||
content = put_data
|
||||
|
||||
with pytest.raises(ValidationError, match=re.escape("The IMO number field may not be changed since it serves the purpose of a primary (matching) key.")):
|
||||
InputValidationShip.put_content_may_not_contain_imo_number(content)
|
||||
return
|
||||
|
||||
|
||||
def test_input_validation_ship_put_request_fails_when_ship_id_is_missing():
|
||||
put_data = get_stub_valid_ship()
|
||||
put_data.pop("id",None) # make sure there is no ID within the put data for this test
|
||||
|
||||
loadedModel = get_stub_valid_ship_loaded_model(put_data)
|
||||
content = put_data
|
||||
|
||||
with pytest.raises(ValidationError, match="The id field is required."):
|
||||
InputValidationShip.content_contains_ship_id(content)
|
||||
return
|
||||
Reference in New Issue
Block a user