From 4c1b230de95a43241948d538983c930d7237b291 Mon Sep 17 00:00:00 2001 From: Max Metz Date: Mon, 27 May 2024 18:23:07 +0200 Subject: [PATCH] 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. --- src/server/BreCal/api/ships.py | 29 +-- src/server/BreCal/schemas/model.py | 35 ++- src/server/BreCal/stubs/ship.py | 24 ++ .../BreCal/validators/input_validation.py | 6 +- .../validators/input_validation_ship.py | 140 +++++++++++ .../validators/input_validation_shipcall.py | 4 +- .../validators/input_validation_utils.py | 17 -- .../validators/validation_base_utils.py | 20 ++ .../validators/test_input_validation_ship.py | 235 ++++++++++++++++++ 9 files changed, 472 insertions(+), 38 deletions(-) create mode 100644 src/server/BreCal/validators/input_validation_ship.py create mode 100644 src/server/BreCal/validators/validation_base_utils.py create mode 100644 src/server/tests/validators/test_input_validation_ship.py diff --git a/src/server/BreCal/api/ships.py b/src/server/BreCal/api/ships.py index 5fc3c50..96bceb2 100644 --- a/src/server/BreCal/api/ships.py +++ b/src/server/BreCal/api/ships.py @@ -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) diff --git a/src/server/BreCal/schemas/model.py b/src/server/BreCal/schemas/model.py index 9158aed..55ee12c 100644 --- a/src/server/BreCal/schemas/model.py +++ b/src/server/BreCal/schemas/model.py @@ -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 diff --git a/src/server/BreCal/stubs/ship.py b/src/server/BreCal/stubs/ship.py index 4ab288e..77e6e45 100644 --- a/src/server/BreCal/stubs/ship.py +++ b/src/server/BreCal/stubs/ship.py @@ -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 + \ No newline at end of file diff --git a/src/server/BreCal/validators/input_validation.py b/src/server/BreCal/validators/input_validation.py index 6e25bda..1fb9649 100644 --- a/src/server/BreCal/validators/input_validation.py +++ b/src/server/BreCal/validators/input_validation.py @@ -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 - + diff --git a/src/server/BreCal/validators/input_validation_ship.py b/src/server/BreCal/validators/input_validation_ship.py new file mode 100644 index 0000000..bfe0fcb --- /dev/null +++ b/src/server/BreCal/validators/input_validation_ship.py @@ -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 + + + diff --git a/src/server/BreCal/validators/input_validation_shipcall.py b/src/server/BreCal/validators/input_validation_shipcall.py index 59bb5ce..899d037 100644 --- a/src/server/BreCal/validators/input_validation_shipcall.py +++ b/src/server/BreCal/validators/input_validation_shipcall.py @@ -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(): diff --git a/src/server/BreCal/validators/input_validation_utils.py b/src/server/BreCal/validators/input_validation_utils.py index bf0e479..cbe4e43 100644 --- a/src/server/BreCal/validators/input_validation_utils.py +++ b/src/server/BreCal/validators/input_validation_utils.py @@ -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 diff --git a/src/server/BreCal/validators/validation_base_utils.py b/src/server/BreCal/validators/validation_base_utils.py new file mode 100644 index 0000000..82ced1f --- /dev/null +++ b/src/server/BreCal/validators/validation_base_utils.py @@ -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 \ No newline at end of file diff --git a/src/server/tests/validators/test_input_validation_ship.py b/src/server/tests/validators/test_input_validation_ship.py new file mode 100644 index 0000000..d0c4671 --- /dev/null +++ b/src/server/tests/validators/test_input_validation_ship.py @@ -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