From 277e28c518620f78ad9d181dea2b2d0d36072dd4 Mon Sep 17 00:00:00 2001 From: Max Metz Date: Tue, 28 May 2024 14:18:59 +0200 Subject: [PATCH] added InputValidationTimes for POST, PUT and DELETE requests. Created unit tests to check the individual functions. The input validation has been activated in BreCal.api.times. There is now a total of 188 unit tests, all passing. Created some utility functions and stub objects to ease the unit testing. --- src/server/BreCal/api/times.py | 24 +- src/server/BreCal/database/sql_utils.py | 14 + src/server/BreCal/schemas/model.py | 61 ++- src/server/BreCal/services/jwt_handler.py | 17 +- src/server/BreCal/stubs/times_full.py | 112 +++-- .../validators/input_validation_ship.py | 2 +- .../validators/input_validation_shipcall.py | 2 +- .../validators/input_validation_times.py | 384 +++++++++++++++-- .../validators/input_validation_utils.py | 38 +- src/server/BreCal/validators/time_logic.py | 33 ++ .../validators/test_input_validation_ship.py | 9 - .../validators/test_input_validation_times.py | 398 ++++++++++++++++++ 12 files changed, 999 insertions(+), 95 deletions(-) create mode 100644 src/server/BreCal/database/sql_utils.py create mode 100644 src/server/tests/validators/test_input_validation_times.py diff --git a/src/server/BreCal/api/times.py b/src/server/BreCal/api/times.py index a333064..e1b3786 100644 --- a/src/server/BreCal/api/times.py +++ b/src/server/BreCal/api/times.py @@ -1,10 +1,11 @@ from flask import Blueprint, request from ..schemas import model from .. import impl -from ..services.auth_guard import auth_guard +from ..services.auth_guard import auth_guard, check_jwt import json import logging from marshmallow import ValidationError +from BreCal.validators.input_validation_times import InputValidationTimes bp = Blueprint('times', __name__) @@ -30,6 +31,12 @@ def PostTimes(): # print (content) # body = parser.parse(schema, request, location='json') loadedModel = model.TimesSchema().load(data=content, many=False, partial=True) + + # read the user data from the JWT token (set when login is performed) + user_data = check_jwt() + + # validate the request + InputValidationTimes.evaluate_post_data(user_data, loadedModel, content) except ValidationError as ex: logging.error(ex) @@ -51,7 +58,13 @@ def PutTimes(): try: content = request.get_json(force=True) loadedModel = model.TimesSchema().load(data=content, many=False, partial=True) + + # read the user data from the JWT token (set when login is performed) + user_data = check_jwt() + # validate the request + InputValidationTimes.evaluate_put_data(user_data, loadedModel, content) + except ValidationError as ex: logging.error(ex) print(ex) @@ -69,11 +82,16 @@ def PutTimes(): @auth_guard() # no restriction by role def DeleteTimes(): - # TODO check if I am allowd to delete this thing by deriving the participant from the bearer token - if 'id' in request.args: options = {} options["id"] = request.args.get("id") + + # read the user data from the JWT token (set when login is performed) + user_data = check_jwt() + + # validate the request + InputValidationTimes.evaluate_delete_data(user_data, times_id = request.args.get("id")) + return impl.times.DeleteTimes(options) else: logging.warning("Times delete missing id argument") diff --git a/src/server/BreCal/database/sql_utils.py b/src/server/BreCal/database/sql_utils.py new file mode 100644 index 0000000..5780ace --- /dev/null +++ b/src/server/BreCal/database/sql_utils.py @@ -0,0 +1,14 @@ +from BreCal.database.sql_handler import execute_sql_query_standalone +import datetime + +def get_user_data_for_id(user_id:int, expiration_time:int=90): + """debugging function, which is useful to pull user_data from the database, which may be used to create stub data and unit tests""" + query = "SELECT * FROM user where id = ?id?" + pdata = execute_sql_query_standalone(query=query, param={"id":user_id}) + pdata = pdata[0] if len(pdata)>0 else None + assert pdata is not None, f"could not find user with id {user_id}" + + user_data = {k:v for k,v in pdata.items() if k in ['id','participant_id','first_name','last_name','user_name','user_phone','user_email']} + user_data["exp"] = (datetime.datetime.now()+datetime.timedelta(minutes=expiration_time)).timestamp() + return user_data + diff --git a/src/server/BreCal/schemas/model.py b/src/server/BreCal/schemas/model.py index 55ee12c..d6683a7 100644 --- a/src/server/BreCal/schemas/model.py +++ b/src/server/BreCal/schemas/model.py @@ -9,7 +9,7 @@ from typing import List import json import datetime -from BreCal.validators.time_logic import validate_time_exceeds_threshold +from BreCal.validators.time_logic import validate_time_is_in_not_too_distant_future from BreCal.validators.validation_base_utils import check_if_string_has_special_characters from BreCal.database.enums import ParticipantType, ParticipantFlag @@ -354,10 +354,10 @@ class TimesSchema(Schema): zone_entry_fixed = fields.Bool(metadata={'required':False}, allow_none=True) operations_start = fields.DateTime(metadata={'required':False}, allow_none=True) operations_end = fields.DateTime(metadata={'required':False}, allow_none=True) - remarks = fields.String(metadata={'required':False}, allow_none=True, validate=[validate.Length(max=256)]) + remarks = fields.String(metadata={'required':False}, allow_none=True, validate=[validate.Length(max=512)]) participant_id = fields.Integer(metadata={'required':True}) berth_id = fields.Integer(metadata={'required':False}, allow_none = True) - berth_info = fields.String(metadata={'required':False}, allow_none=True, validate=[validate.Length(max=256)]) + berth_info = fields.String(metadata={'required':False}, allow_none=True, validate=[validate.Length(max=512)]) pier_side = fields.Bool(metadata={'required':False}, allow_none = True) shipcall_id = fields.Integer(metadata={'required':True}) participant_type = fields.Integer(Required = False, allow_none=True)# TODO: could become Enum. fields.Enum(ParticipantType, metadata={'required':False}, allow_none=True, default=ParticipantType.undefined) #fields.Integer(metadata={'required':False}, allow_none=True) @@ -368,11 +368,60 @@ class TimesSchema(Schema): created = fields.DateTime(metadata={'required':False}, allow_none=True) modified = fields.DateTime(metadata={'required':False}, allow_none=True) + @validates("participant_type") + def validate_participant_type(self, value): + # #TODO: it may also make sense to block multi-assignments, whereas a value could be BSMD+AGENCY + # while the validation fails when one of those multi-assignments is BSMD, it passes in cases, + # such as AGENCY+PILOT + + # a participant type should not be .BSMD + if not isinstance(value, ParticipantType): + value = ParticipantType(value) + + if ParticipantType.BSMD in value: + raise ValidationError(f"the participant_type must not be .BSMD") + @validates("eta_berth") def validate_eta_berth(self, value): - threshold_exceeded = validate_time_exceeds_threshold(value, months=12) - if threshold_exceeded: - raise ValidationError(f"the provided time exceeds the twelve month threshold.") + # violation when time is not in the future, but also does not exceed a threshold for the 'reasonable' future + # when 'value' is 'None', a ValidationError is not issued. + valid_time = validate_time_is_in_not_too_distant_future(raise_validation_error=True, value=value, months=12) + return + + @validates("etd_berth") + def validate_etd_berth(self, value): + # violation when time is not in the future, but also does not exceed a threshold for the 'reasonable' future + # when 'value' is 'None', a ValidationError is not issued. + valid_time = validate_time_is_in_not_too_distant_future(raise_validation_error=True, value=value, months=12) + return + + @validates("lock_time") + def validate_lock_time(self, value): + # violation when time is not in the future, but also does not exceed a threshold for the 'reasonable' future + # when 'value' is 'None', a ValidationError is not issued. + valid_time = validate_time_is_in_not_too_distant_future(raise_validation_error=True, value=value, months=12) + return + + @validates("zone_entry") + def validate_zone_entry(self, value): + # violation when time is not in the future, but also does not exceed a threshold for the 'reasonable' future + # when 'value' is 'None', a ValidationError is not issued. + valid_time = validate_time_is_in_not_too_distant_future(raise_validation_error=True, value=value, months=12) + return + + @validates("operations_start") + def validate_operations_start(self, value): + # violation when time is not in the future, but also does not exceed a threshold for the 'reasonable' future + # when 'value' is 'None', a ValidationError is not issued. + valid_time = validate_time_is_in_not_too_distant_future(raise_validation_error=True, value=value, months=12) + return + + @validates("operations_end") + def validate_operations_end(self, value): + # violation when time is not in the future, but also does not exceed a threshold for the 'reasonable' future + # when 'value' is 'None', a ValidationError is not issued. + valid_time = validate_time_is_in_not_too_distant_future(raise_validation_error=True, value=value, months=12) + return # deserialize PUT object target diff --git a/src/server/BreCal/services/jwt_handler.py b/src/server/BreCal/services/jwt_handler.py index b3c80e7..d1e7ad6 100644 --- a/src/server/BreCal/services/jwt_handler.py +++ b/src/server/BreCal/services/jwt_handler.py @@ -26,7 +26,22 @@ def generate_jwt(payload, lifetime=None): return jwt.encode(payload, os.environ.get('SECRET_KEY'), algorithm="HS256") def decode_jwt(token): - """this function reverts the {generate_jwt} function. An encoded JWT token is decoded into a JSON dictionary.""" + """ + this function reverts the {generate_jwt} function. An encoded JWT token is decoded into a JSON dictionary. + The function is commonly used to decode a login-token and obtain a 'user_data' variable, which is a dictionary. + + Example of 'user_data': + { + 'id': 1, + 'participant_id': 1, + 'first_name': 'Firstname', + 'last_name': 'Lastname', + 'user_name': 'xUsername01', + 'user_phone': '+01 123 456 7890', + 'user_email': 'firstname.lastname@internet.com', + 'exp': 1716881626.056438 # expiration timestamp + } + """ return jwt.decode(token, os.environ.get('SECRET_KEY'), algorithms=["HS256"]) diff --git a/src/server/BreCal/stubs/times_full.py b/src/server/BreCal/stubs/times_full.py index 774cc9f..b88fdb1 100644 --- a/src/server/BreCal/stubs/times_full.py +++ b/src/server/BreCal/stubs/times_full.py @@ -5,10 +5,12 @@ users will thereby be able to modify these values import datetime from BreCal.stubs import generate_uuid1_int -from BreCal.schemas.model import Times +from BreCal.schemas.model import Times, ParticipantType +from BreCal.database.sql_utils import get_user_data_for_id -def get_times_full_simple(): + +def get_times_full_simple(return_dataclass=True): # only used for the stub base_time = datetime.datetime.now() @@ -49,32 +51,86 @@ def get_times_full_simple(): created = datetime.datetime.now() modified = created+datetime.timedelta(seconds=10) - times = Times( - id=times_id, - eta_berth=eta_berth, - eta_berth_fixed=eta_berth_fixed, - etd_berth=etd_berth, - etd_berth_fixed=etd_berth_fixed, - lock_time=lock_time, - lock_time_fixed=lock_time_fixed, - zone_entry=zone_entry, - zone_entry_fixed=zone_entry_fixed, - operations_start=operations_start, - operations_end=operations_end, - remarks=remarks, - participant_id=participant_id, - berth_id=berth_id, - berth_info=berth_info, - pier_side=pier_side, - participant_type=participant_type, - shipcall_id=shipcall_id, - ata=ata, - atd=atd, - eta_interval_end=eta_interval_end, - etd_interval_end=etd_interval_end, - created=created, - modified=modified, - ) + if return_dataclass: + times = Times( + id=times_id, + eta_berth=eta_berth, + eta_berth_fixed=eta_berth_fixed, + etd_berth=etd_berth, + etd_berth_fixed=etd_berth_fixed, + lock_time=lock_time, + lock_time_fixed=lock_time_fixed, + zone_entry=zone_entry, + zone_entry_fixed=zone_entry_fixed, + operations_start=operations_start, + operations_end=operations_end, + remarks=remarks, + participant_id=participant_id, + berth_id=berth_id, + berth_info=berth_info, + pier_side=pier_side, + participant_type=participant_type, + shipcall_id=shipcall_id, + ata=ata, + atd=atd, + eta_interval_end=eta_interval_end, + etd_interval_end=etd_interval_end, + created=created, + modified=modified, + ) + else: + times = dict( + id=times_id, + eta_berth=eta_berth, + eta_berth_fixed=eta_berth_fixed, + etd_berth=etd_berth, + etd_berth_fixed=etd_berth_fixed, + lock_time=lock_time, + lock_time_fixed=lock_time_fixed, + zone_entry=zone_entry, + zone_entry_fixed=zone_entry_fixed, + operations_start=operations_start, + operations_end=operations_end, + remarks=remarks, + participant_id=participant_id, + berth_id=berth_id, + berth_info=berth_info, + pier_side=pier_side, + participant_type=participant_type, + shipcall_id=shipcall_id, + ata=ata, + atd=atd, + eta_interval_end=eta_interval_end, + etd_interval_end=etd_interval_end, + created=created, + modified=modified,) + times = {k:v.isoformat() if isinstance(v, datetime.datetime) else v for k,v in times.items()} return times +def get_valid_stub_times(): + """create a stub entry for a times dataset, which is valid""" + times_entry = get_times_full_simple(return_dataclass=False) + times_entry.pop('id',None) + times_entry["participant_id"] = 136 + times_entry["participant_type"] = int(ParticipantType.PILOT) + times_entry["shipcall_id"] = 222 + times_entry["berth_id"] = 143 + times_entry["remarks"] = "stub entry." + return times_entry + + +def get_valid_stub_times_loaded_model(post_data=None): + from BreCal.schemas import model + if post_data is None: + post_data = get_valid_stub_times() + loadedModel = model.TimesSchema().load(data=post_data, many=False, partial=True) + return loadedModel + +def get_valid_stub_for_pytests(user_id:int=3): + user_data = get_user_data_for_id(user_id=user_id) + post_data = get_valid_stub_times() + + content = post_data + loadedModel = get_valid_stub_times_loaded_model(post_data=post_data) + return user_data, loadedModel, content diff --git a/src/server/BreCal/validators/input_validation_ship.py b/src/server/BreCal/validators/input_validation_ship.py index bfe0fcb..eab2480 100644 --- a/src/server/BreCal/validators/input_validation_ship.py +++ b/src/server/BreCal/validators/input_validation_ship.py @@ -82,7 +82,7 @@ class InputValidationShip(): 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}") + raise ValidationError(f"current user does not belong to BSMD. Cannot post, put or delete ships. Found user data: {user_data}") @staticmethod def check_ship_imo_already_exists(loadedModel:dict): diff --git a/src/server/BreCal/validators/input_validation_shipcall.py b/src/server/BreCal/validators/input_validation_shipcall.py index 899d037..0b8ad3b 100644 --- a/src/server/BreCal/validators/input_validation_shipcall.py +++ b/src/server/BreCal/validators/input_validation_shipcall.py @@ -188,7 +188,7 @@ class InputValidationShipcall(): """ 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}") + raise ValidationError(f"current user does not belong to BSMD. Cannot post or put shipcalls. Found user data: {user_data}") return @staticmethod diff --git a/src/server/BreCal/validators/input_validation_times.py b/src/server/BreCal/validators/input_validation_times.py index ec79732..6b8ab6a 100644 --- a/src/server/BreCal/validators/input_validation_times.py +++ b/src/server/BreCal/validators/input_validation_times.py @@ -12,11 +12,54 @@ from BreCal.impl.berths import GetBerths from BreCal.impl.times import GetTimes 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.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_shipcall_id_is_valid, get_shipcall_id_dictionary, get_participant_type_from_user_data, get_participant_id_dictionary, check_if_participant_id_is_valid_standalone from BreCal.database.sql_handler import execute_sql_query_standalone from BreCal.validators.validation_base_utils import check_if_int_is_valid_flag, check_if_string_has_special_characters import werkzeug +def build_post_data_type_dependent_required_fields_dict()->dict[ShipcallType,dict[ParticipantType,typing.Optional[list[str]]]]: + """ + The required fields of a POST-request depend on ShipcallType and ParticipantType. This function creates + a dictionary, which maps those types to a list of required fields. + + The participant types 'undefined' and 'bsmd' should not be used in POST-requests. They return 'None'. + """ + post_data_type_dependent_required_fields_dict = { + ShipcallType.arrival:{ + ParticipantType.undefined:None, # should not be set in POST requests + ParticipantType.BSMD:None, # should not be set in POST requests + ParticipantType.TERMINAL:["operations_start"], + ParticipantType.AGENCY:["eta_berth"], + ParticipantType.MOORING:["eta_berth"], + ParticipantType.PILOT:["eta_berth"], + ParticipantType.PORT_ADMINISTRATION:["eta_berth"], + ParticipantType.TUG:["eta_berth"], + }, + ShipcallType.departure:{ + ParticipantType.undefined:None, # should not be set in POST requests + ParticipantType.BSMD:None, # should not be set in POST requests + ParticipantType.TERMINAL:["operations_end"], + ParticipantType.AGENCY:["etd_berth"], + ParticipantType.MOORING:["etd_berth"], + ParticipantType.PILOT:["etd_berth"], + ParticipantType.PORT_ADMINISTRATION:["etd_berth"], + ParticipantType.TUG:["etd_berth"], + }, + ShipcallType.shifting:{ + ParticipantType.undefined:None, # should not be set in POST requests + ParticipantType.BSMD:None, # should not be set in POST requests + ParticipantType.TERMINAL:["operations_start", "operations_end"], + ParticipantType.AGENCY:["eta_berth", "etd_berth"], + ParticipantType.MOORING:["eta_berth", "etd_berth"], + ParticipantType.PILOT:["eta_berth", "etd_berth"], + ParticipantType.PORT_ADMINISTRATION:["eta_berth", "etd_berth"], + ParticipantType.TUG:["eta_berth", "etd_berth"], + }, + } + return post_data_type_dependent_required_fields_dict + + + class InputValidationTimes(): """ This class combines a complex set of individual input validation functions into a joint object. @@ -32,60 +75,327 @@ class InputValidationTimes(): @staticmethod def evaluate_post_data(user_data:dict, loadedModel:dict, content:dict): - raise NotImplementedError("skeleton") + # 0.) Check for the presence of required fields + InputValidationTimes.check_times_required_fields_post_data(loadedModel, content) + + # 1.) datasets may only be created, if the current user fits the appropriate type in the ShipcallParticipantMap + InputValidationTimes.check_if_user_fits_shipcall_participant_map(user_data, loadedModel, content) + + # 2.) datasets may only be created, if the respective participant type did not already create one. + InputValidationTimes.check_if_entry_already_exists_for_participant_type(user_data, loadedModel, content) + + # 3.) only users who are *not* of type BSMD may post times datasets. + InputValidationTimes.check_user_is_not_bsmd_type(user_data) + + # 4.) Reference checking + InputValidationTimes.check_dataset_references(content) + + # 5.) Value checking + InputValidationTimes.check_dataset_values(user_data, loadedModel, content) return @staticmethod def evaluate_put_data(user_data:dict, loadedModel:dict, content:dict): - raise NotImplementedError("skeleton") - # 1.) Only users of the same participant_id, which the times dataset refers to, can delete the entry - # (same as for .evaluate_delete_data) + # 1.) Check for the presence of required fields + InputValidationTimes.check_times_required_fields_put_data(content) - # 2.) Reference checking - # (same as for .evaluate_post_data) + # 2.) Only users of the same participant_id, which the times dataset refers to, can update the entry + InputValidationTimes.check_user_belongs_to_same_group_as_dataset_determines(user_data, loadedModel=loadedModel, times_id=None) - # 3.) Value checking - # (same as for .evaluate_post_data) - # participant type should be dynamically checked for POST / PUT. All other values can be validated within the Schema + # 3.) Reference checking + InputValidationTimes.check_dataset_references(content) + + # 4.) Value checking + InputValidationTimes.check_dataset_values(user_data, loadedModel, content) return @staticmethod def evaluate_delete_data(user_data:dict, times_id:int): - raise NotImplementedError("skeleton") - # 1.) Only users of the same participant_id, which the times dataset refers to, can delete the entry - # (same as for .evaluate_put_data) + # #TODO_determine: is times_id always an int or does the request.args call provide a string? + times_id = int(times_id) if not isinstance(times_id, int) else times_id - # 2.) The dataset entry may not be deleted already - shipcall_id is not Defined - InputValidationTimes.check_if_entry_is_already_deleted(times_id, shipcall_id) + # 1.) The dataset entry may not be deleted already + InputValidationTimes.check_if_entry_is_already_deleted(times_id) + + # 2.) Only users of the same participant_id, which the times dataset refers to, can delete the entry + InputValidationTimes.check_user_belongs_to_same_group_as_dataset_determines(user_data, loadedModel=None, times_id=times_id) return @staticmethod - def check_if_entry_is_already_deleted(times_id:int, shipcall_id:int): + def check_if_entry_is_already_deleted(times_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. + When calling a delete request for times, the dataset may not be deleted already. This method + makes sure, that the request contains and ID, has a matching entry in the database. + When a times dataset is deleted, it is directly removed from the database. + + To identify deleted entries, query from the database and check, whether there is a match for the times id. + """ - raise NotImplementedError("skeleton. fully untested pseudo-code.") - if times_id is None: - raise ValidationError(f"The times_id must be provided.") - - # options["shipcall_id"] - assert 'shipcall_id' in options.keys() + # perform an SQL query. Creates a pooled connection internally, queries the database, then closes the connection. + query = "SELECT shipcall_id FROM times WHERE id = ?id?" + pdata = execute_sql_query_standalone(query=query, param={"id":times_id}, pooledConnection=None) - response, status_code, header = GetTimes(options) - times = json.loads(response) - existing_database_entries = [time_ for time_ in times if time_.get("id")==times_id] - if len(existing_database_entries)==0: - raise ValidationError(f"Could not find a times entry with the specified ID. Selected: {times_id}") - - existing_database_entry = existing_database_entries[0] - - deletion_state = existing_database_entry.get("deleted",None) - if deletion_state: - raise ValidationError(f"The selected time entry is already deleted.") + if len(pdata)==0: + raise ValidationError(f"The selected time entry is already deleted. ID: {times_id}") return + + @staticmethod + def check_user_is_not_bsmd_type(user_data:dict): + """a new dataset may only be created by a user who is *not* belonging to participant group BSMD""" + is_bsmd = check_if_user_is_bsmd_type(user_data) + if is_bsmd: + raise ValidationError(f"current user belongs to BSMD. Cannot post 'times' datasets. Found user data: {user_data}") + return + + @staticmethod + def check_dataset_values(user_data:dict, loadedModel:dict, content:dict): + """ + this method validates POST and PUT data. Most of the dataset arguments are validated directly in the + BreCal.schemas.model.TimesSchema, using @validates. This is exclusive for 'simple' validation rules. + + This applies to: + "remarks" & "berth_info" + "eta_berth", "etd_berth", "lock_time", "zone_entry", "operations_start", "operations_end" + """ + # while InputValidationTimes.check_user_is_not_bsmd_type already validates a user, this method + # validates the times dataset. + + # ensure loadedModel["participant_type"] is of type ParticipantType + if not isinstance(loadedModel["participant_type"], ParticipantType): + loadedModel["participant_type"] = ParticipantType(loadedModel["participant_type"]) + + if ParticipantType.BSMD in loadedModel["participant_type"]: + raise ValidationError(f"current user belongs to BSMD. Cannot post times datasets. Found user data: {user_data}") + return + + @staticmethod + def check_dataset_references(content:dict): + """ + When IDs are referenced, they must exist in the database. This method individually validates the existance of referred + berth ID, participant IDs and shipcall ID. + + Note: whenever an ID is 'None', there is no exception, because a different method is supposed to capture non-existant mandatory fields. + """ + # extract the IDs + berth_id, participant_id, shipcall_id = content.get("berth_id"), content.get("participant_id"), content.get("shipcall_id") + + valid_berth_id_reference = check_if_berth_id_is_valid(berth_id) + if not valid_berth_id_reference: + raise ValidationError(f"The referenced berth_id '{berth_id}' does not exist in the database.") + + valid_shipcall_id_reference = check_if_shipcall_id_is_valid(shipcall_id) + if not valid_shipcall_id_reference: + raise ValidationError(f"The referenced shipcall_id '{shipcall_id}' does not exist in the database.") + + valid_participant_id_reference = check_if_participant_id_is_valid_standalone(participant_id) + if not valid_participant_id_reference: + raise ValidationError(f"The referenced participant_id '{participant_id}' does not exist in the database.") + + return + + @staticmethod + def check_times_required_fields_post_data(loadedModel:dict, content:dict): + """ + Depending on ShipcallType and ParticipantType, there is a rather complex set of required fields. + Independent of those types, any POST request for times should always include the default fields. + + The dependent and independent fields are validated by checking, whether the respective value in 'content' + is undefined (returns None). When any of these fields is undefined, a ValidationError is raised. + """ + participant_type = loadedModel["participant_type"] + shipcall_id = loadedModel["shipcall_id"] + + # build a dictionary of id:item pairs, so one can select the respective participant + # must look-up the shipcall_type based on the shipcall_id + shipcalls = get_shipcall_id_dictionary() + shipcall_type = ShipcallType[shipcalls.get(shipcall_id,{}).get("type",ShipcallType.undefined.name)] + + if (participant_type is None) or (int(shipcall_type) == int(ShipcallType.undefined)): + raise ValidationError(f"At least one of the required fields is missing. Missing: 'participant_type' or 'shipcall_type'") + + + # build a list of required fields based on shipcall and participant type, as well as type-independent fields + independent_required_fields = InputValidationTimes.get_post_data_type_independent_fields() + dependent_required_fields = InputValidationTimes.get_post_data_type_dependent_fields(shipcall_type, participant_type) + + required_fields = independent_required_fields + dependent_required_fields + + # generate a list of booleans, where each element shows, whether one of the required fields is missing. + missing_required_fields = [ + content.get(field,None) is None for field in required_fields + ] + + if any(missing_required_fields): + # create a tuple of (field_key, bool) to describe to a user, which one of the fields may be missing + verbosity_tuple = [(field, missing) for field, missing in zip(required_fields, missing_required_fields) if missing] + raise ValidationError(f"At least one of the required fields is missing. Missing: {verbosity_tuple}") + return + + @staticmethod + def check_times_required_fields_put_data(content:dict): + """in a PUT request, only the 'id' is a required field. All other fields are simply ignored, when they are not provided.""" + if content.get("id") is None: + raise ValidationError(f"A PUT-request requires an 'id' reference, which was not found.") + return + + @staticmethod + def get_post_data_type_independent_fields()->list[str]: + """ + Independent of the ShipcallType and ParticipantType, any POST request for times should always include the default fields. + """ + independent_required_fields = [ + "shipcall_id", "participant_id", "participant_type" + ] + return independent_required_fields + + @staticmethod + def get_post_data_type_dependent_fields(shipcall_type:typing.Union[int, ShipcallType], participant_type:typing.Union[int, ParticipantType]): + """ + Depending on ShipcallType and ParticipantType, there is a rather complex set of required fields. + + Arriving shipcalls need arrival times (e.g., 'eta'), Departing shipcalls need departure times (e.g., 'etd') and + Shifting shipcalls need both times (e.g., 'eta' and 'etd'). + + Further, the ParticipantType determines the set of relevant times. In particular, the terminal uses + 'operations_start' and 'operations_end', while other users use 'eta_berth' or 'etd_berth'. + """ + # ensure that both, shipcall_type and participant_type, refer to the enumerators, as opposed to integers. + if not isinstance(shipcall_type, ShipcallType): + shipcall_type = ShipcallType(shipcall_type) + if not isinstance(participant_type, ParticipantType): + participant_type = ParticipantType(participant_type) + + # build a dictionary, which maps shipcall type and participant type to a list of fields + dependent_required_fields_dict = build_post_data_type_dependent_required_fields_dict() + + # select shipcall type & participant type + dependent_required_fields = dependent_required_fields_dict.get(shipcall_type,{}).get(participant_type,None) + return dependent_required_fields + + @staticmethod + def check_if_user_fits_shipcall_participant_map(user_data:dict, loadedModel:dict, content:dict, spm_shipcall_data=None): + """ + a new dataset may only be created, if the user belongs to the participant group (participant_id), + which is assigned to the shipcall within the ShipcallParticipantMap + + This method does not validate, what the POST-request contains, but it validates, whether the *user* is + authorized to send the request. + + options: + spm_shipcall_data: + data from the ShipcallParticipantMap, which refers to the respective shipcall ID. The SPM can be + an optional argument to allow for much easier unit testing. + """ + + # identify shipcall_id + shipcall_id = loadedModel["shipcall_id"] + + # identify user's participant_id & type (get all participants; then filter these for the {participant_id}) + participant_id = user_data["participant_id"] #participants = get_participant_id_dictionary() #participant_type = ParticipantType(participants.get(participant_id,{}).get("type")) + participant_type = ParticipantType(loadedModel["participant_type"]) if not isinstance(loadedModel["participant_type"],ParticipantType) else loadedModel["participant_type"] + + # get ShipcallParticipantMap for the shipcall_id + if spm_shipcall_data is None: + # read the ShipcallParticipantMap entry of the current shipcall_id. This is used within the input validation of a PUT request + # creates a list of {'participant_id: ..., 'type': ...} elements + spm_shipcall_data = execute_sql_query_standalone( + query = "SELECT participant_id, type FROM shipcall_participant_map WHERE shipcall_id=?shipcall_id?", + param={"shipcall_id":shipcall_id}, + pooledConnection=None + ) + + # check, if participant_id is assigned to the ShipcallParticipantMap + matching_spm = [ + spm + for spm in spm_shipcall_data + if spm.get("participant_id")==participant_id + ] + + if not len(matching_spm)>0: + raise ValidationError(f'The participant group with id {participant_id} is not assigned to the shipcall. Found ShipcallParticipantMap: {spm_shipcall_data}') + + # check, if the assigned participant_id is assigned with the same role + matching_spm_element = matching_spm[0] + matching_spm_element_participant_type = ParticipantType(matching_spm_element.get("type")) + if not matching_spm_element_participant_type in participant_type: + raise ValidationError(f'The participant group with id {participant_id} is assigned to the shipcall in a different role. Request Role: {participant_type}, ShipcallParticipantMap Role Assignment: {matching_spm_element_participant_type}') + return + + @staticmethod + def check_if_entry_already_exists_for_participant_type(user_data:dict, loadedModel:dict, content:dict): + """determines, whether a dataset for the participant type is already present""" + # determine participant_type and shipcall_id from the loadedModel + participant_type = loadedModel["participant_type"] + if not isinstance(participant_type, ParticipantType): # ensure the correct data type + participant_type = ParticipantType(participant_type) + shipcall_id = loadedModel["shipcall_id"] + + # get all times entries of the shipcall_id from the database + times, status_code, headers = GetTimes(options={"shipcall_id":shipcall_id}) + times = json.loads(times) + + # check, if there is already a dataset for the participant type + participant_type_exists_already = any([ParticipantType(time_.get("participant_type",0)) in participant_type for time_ in times]) + if participant_type_exists_already: + raise ValidationError(f"A dataset for the participant type is already present. Participant Type: {participant_type}. Times Datasets: {times}") + return + + @staticmethod + def check_user_belongs_to_same_group_as_dataset_determines(user_data:dict, loadedModel:typing.Optional[dict]=None, times_id:typing.Optional[int]=None): + """ + This method checks, whether a user belongs to the same participant_id, as the dataset entry refers to. + It is used in, both, PUT requests and DELETE requests, but uses different arguments to determine the matching + time dataset entry. + + PUT: + loadedModel is unbundled to identify the matching times entry by the shipcall id + + DELETE: + times_id is used to directly identify the matching times entry + """ + assert not ((loadedModel is None) and (times_id is None)), f"must provide either loadedModel OR times_id. Both are 'None'" + assert (loadedModel is None) or (times_id is None), f"must provide either loadedModel OR times_id. Both are defined." + + # identify the user's participant id + user_participant_id = user_data["participant_id"] + + if loadedModel is not None: + shipcall_id = loadedModel["shipcall_id"] + participant_type = loadedModel["participant_type"] + + # get all times entries of the shipcall_id from the database as a list of {'participant_id':..., 'participant_type':...} elements + query = "SELECT participant_id, participant_type FROM times WHERE shipcall_id = ?shipcall_id?" + times = execute_sql_query_standalone(query=query, param={"shipcall_id":shipcall_id}, pooledConnection=None) + + # get the matching datasets, where the participant id is identical + time_datasets_of_participant_type = [time_ for time_ in times if time_.get("participant_type")==participant_type] + + # when there are no matching participants, raise a ValidationError + if not len(time_datasets_of_participant_type)>0: + raise ValidationError(f"Could not find a matching time dataset for the provided participant_type: {participant_type}. Found Time Datasets: {times}") + + # take the first match. There should always be only one match. + time_datasets_of_participant_type = time_datasets_of_participant_type[0] + participant_id_of_times_dataset = time_datasets_of_participant_type.get("participant_id") + + if times_id is not None: + # perform an SQL query. Creates a pooled connection internally, queries the database, then closes the connection. + query = "SELECT participant_id FROM times WHERE id = ?id?" + pdata = execute_sql_query_standalone(query=query, param={"id":times_id}, pooledConnection=None) + + # extracts the participant_id from the first matching entry, if applicable + if not len(pdata)>0: + # this case is usually covered by the InputValidationTimes.check_if_entry_is_already_deleted method already + raise ValidationError(f"Unknown times_id. Could not find a matching entry for ID: {times_id}") + else: + participant_id_of_times_dataset = pdata[0].get("participant_id") + + if user_participant_id != participant_id_of_times_dataset: + raise ValidationError(f"The dataset may only be changed by a user belonging to the same participant group as the times dataset is referring to. User participant_id: {user_participant_id}; Dataset participant_id: {participant_id_of_times_dataset}") + return + + + diff --git a/src/server/BreCal/validators/input_validation_utils.py b/src/server/BreCal/validators/input_validation_utils.py index cbe4e43..325e3fb 100644 --- a/src/server/BreCal/validators/input_validation_utils.py +++ b/src/server/BreCal/validators/input_validation_utils.py @@ -105,6 +105,30 @@ def check_if_berth_id_is_valid(berth_id): berth_id_is_valid = berth_id in list(berths.keys()) return berth_id_is_valid +def check_if_shipcall_id_is_valid(shipcall_id:int): + """check, whether the provided ID is valid. If it is 'None', it will be considered valid. This is, because a request, may not have to include all IDs at once""" + if shipcall_id is None: + return True + + # build a dictionary of id:item pairs, so one can select the respective participant + shipcalls = get_shipcall_id_dictionary() + + # boolean check + shipcall_id_is_valid = shipcall_id in list(shipcalls.keys()) + return shipcall_id_is_valid + +def check_if_participant_id_is_valid_standalone(participant_id:int): + """check, whether the provided ID is valid. If it is 'None', it will be considered valid. This is, because a request, may not have to include all IDs at once""" + if participant_id is None: + return True + + # build a dictionary of id:item pairs, so one can select the respective participant + participants = get_participant_id_dictionary() + + # boolean check + participant_id_is_valid = participant_id in list(participants.keys()) + return participant_id_is_valid + def check_if_participant_id_is_valid(participant:dict): """ check, whether the provided ID is valid. If it is 'None', it will be considered valid. This is, because a shipcall POST-request, does not have to include all IDs at once @@ -115,15 +139,7 @@ def check_if_participant_id_is_valid(participant:dict): """ # #TODO1: Daniel Schick: 'types may only appear once and must not include type "BSMD"' participant_id = participant.get("participant_id", None) - - if participant_id is None: - return True - - # build a dictionary of id:item pairs, so one can select the respective participant - participants = get_participant_id_dictionary() - - # boolean check - participant_id_is_valid = participant_id in list(participants.keys()) + participant_id_is_valid = check_if_participant_id_is_valid_standalone(participant_id) return participant_id_is_valid def check_if_participant_ids_are_valid(participants:list[dict]): @@ -135,6 +151,10 @@ def check_if_participant_ids_are_valid(participants:list[dict]): 'participant_id' : int 'type' : ParticipantType """ + # empty list -> invalid + if participants is None: + return False + # check each participant id individually valid_participant_ids = [check_if_participant_id_is_valid(participant) for participant in participants] diff --git a/src/server/BreCal/validators/time_logic.py b/src/server/BreCal/validators/time_logic.py index 3c096a1..8ebad56 100644 --- a/src/server/BreCal/validators/time_logic.py +++ b/src/server/BreCal/validators/time_logic.py @@ -2,6 +2,8 @@ import datetime import numpy as np import pandas as pd +from marshmallow import ValidationError + def validate_time_exceeds_threshold(value:datetime.datetime, seconds:int=60, minutes:int=60, hours:int=24, days:int=30, months:int=12)->bool: """returns a boolean when the input value is very distant in the future. The parameters provide the threshold""" # time difference in seconds. Positive: in the future, Negative: in the past @@ -10,6 +12,37 @@ def validate_time_exceeds_threshold(value:datetime.datetime, seconds:int=60, min threshold = seconds*minutes*hours*days*months return time_>=threshold +def validate_time_is_in_future(value:datetime.datetime): + """returns a boolean when the input value is in the future.""" + current_time = datetime.datetime.now() + return value >= current_time + +def validate_time_is_in_not_too_distant_future(raise_validation_error:bool, value:datetime.datetime, seconds:int=60, minutes:int=60, hours:int=24, days:int=30, months:int=12)->bool: + """ + combines two boolean operations. Returns True when both conditions are met. + a) value is in the future + b) value is not too distant (e.g., at max. 1 year in the future) + + When the value is 'None', the validation will be skipped. A ValidationError is never issued, but the method returns 'False'. + + options: + raise_validation_error: boolean. If set to True, this method issues a marshmallow.ValidationError, when the conditions fail. + """ + if value is None: + return False + + is_in_future = validate_time_is_in_future(value) + is_too_distant = validate_time_exceeds_threshold(value, seconds, minutes, hours, days, months) + + if raise_validation_error: + if not is_in_future: + raise ValidationError(f"The provided value must be in the future. Current Time: {datetime.datetime.now()}, Value: {value}") + + if is_too_distant: + raise ValidationError(f"The provided value is in the too distant future and exceeds a threshold for 'reasonable' entries. Found: {value}") + + return is_in_future & (not is_too_distant) + class TimeLogic(): def __init__(self): return diff --git a/src/server/tests/validators/test_input_validation_ship.py b/src/server/tests/validators/test_input_validation_ship.py index d0c4671..b1de16e 100644 --- a/src/server/tests/validators/test_input_validation_ship.py +++ b/src/server/tests/validators/test_input_validation_ship.py @@ -45,15 +45,6 @@ def get_stub_token(): 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() diff --git a/src/server/tests/validators/test_input_validation_times.py b/src/server/tests/validators/test_input_validation_times.py new file mode 100644 index 0000000..98a2689 --- /dev/null +++ b/src/server/tests/validators/test_input_validation_times.py @@ -0,0 +1,398 @@ +import pytest + +import os +import random +import datetime +from marshmallow import ValidationError + +from BreCal import local_db +from BreCal.schemas import model + +from BreCal.schemas.model import ParticipantType +from BreCal.validators.input_validation_times import InputValidationTimes + +from BreCal.stubs.times_full import get_valid_stub_times, get_valid_stub_for_pytests + +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") + + +def test_input_validation_times_fails_when_berth_info_exceeds_length_limit(): + # success + post_data = get_valid_stub_times() + post_data["berth_info"] = "a"*512 # 512 characters + model.TimesSchema().load(data=post_data, many=False, partial=True) + + post_data["berth_info"] = "" # 0 characters + model.TimesSchema().load(data=post_data, many=False, partial=True) + + # failure + with pytest.raises(ValidationError, match="Longer than maximum length 512."): + post_data["berth_info"] = "a"*513 # 513 characters + model.TimesSchema().load(data=post_data, many=False, partial=True) + return + +def test_input_validation_times_fails_when_remarks_exceeds_length_limit(): + # success + post_data = get_valid_stub_times() + post_data["remarks"] = "a"*512 # 512 characters + model.TimesSchema().load(data=post_data, many=False, partial=True) + + post_data["remarks"] = "" # 0 characters + model.TimesSchema().load(data=post_data, many=False, partial=True) + + # failure + with pytest.raises(ValidationError, match="Longer than maximum length 512."): + post_data["remarks"] = "a"*513 # 513 characters + model.TimesSchema().load(data=post_data, many=False, partial=True) + return + +def test_input_validation_times_fails_when_participant_type_is_bsmd(): + # BSMD -> Failure + post_data = get_valid_stub_times() + post_data["participant_type"] = int(ParticipantType.BSMD) + with pytest.raises(ValidationError, match="the participant_type must not be .BSMD"): + model.TimesSchema().load(data=post_data, many=False, partial=True) + + # IntFlag property: BSMD & AGENCY -> Failure + post_data = get_valid_stub_times() + post_data["participant_type"] = int(ParticipantType(ParticipantType.BSMD+ParticipantType.AGENCY)) + with pytest.raises(ValidationError, match="the participant_type must not be .BSMD"): + model.TimesSchema().load(data=post_data, many=False, partial=True) + return + +def test_input_validation_times_fails_when_time_key_is_not_reasonable(): + """ + every time key (e.g., 'eta_berth' or 'zone_entry') must be reasonable. The validation expects + these values to be 'in the future' (larger than datetime.datetime.now()) and not 'in the too distant future' + (e.g., more than one year from now.) + """ + for time_key in ["eta_berth", "etd_berth", "lock_time", "zone_entry", "operations_start", "operations_end"]: + post_data = get_valid_stub_times() + + # success + post_data[time_key] = (datetime.datetime.now() + datetime.timedelta(minutes=11)).isoformat() + model.TimesSchema().load(data=post_data, many=False, partial=True) + + # fails + with pytest.raises(ValidationError, match="The provided value must be in the future."): + post_data[time_key] = (datetime.datetime.now() - datetime.timedelta(minutes=11)).isoformat() + model.TimesSchema().load(data=post_data, many=False, partial=True) + + # fails + with pytest.raises(ValidationError, match="The provided value is in the too distant future and exceeds a threshold for 'reasonable' entries."): + post_data[time_key] = (datetime.datetime.now() + datetime.timedelta(days=367)).isoformat() + model.TimesSchema().load(data=post_data, many=False, partial=True) + return + +def test_input_validation_times_fails_when_user_is_bsmd_user(): + # create stub-data for a POST request + from BreCal.services.jwt_handler import decode_jwt + from BreCal.database.sql_utils import get_user_data_for_id + import re + + # user 4 is a BSMD user -> fails + user_data, loadedModel, content = get_valid_stub_for_pytests(user_id=4) + + with pytest.raises(ValidationError, match=re.escape("current user belongs to BSMD. Cannot post 'times' datasets.")): + InputValidationTimes.check_user_is_not_bsmd_type(user_data) + + # user 13 is not a BSMD user -> passes + user_data, loadedModel, content = get_valid_stub_for_pytests(user_id=13) + + # success + InputValidationTimes.check_user_is_not_bsmd_type(user_data) + return + +def test_input_validation_times_fails_when_participant_type_entry_already_exists(): + # the participant type already has an entry -> fails + with pytest.raises(ValidationError, match="A dataset for the participant type is already present. Participant Type:"): + user_data, loadedModel, content = get_valid_stub_for_pytests(user_id=3) + loadedModel["participant_type"] = int(ParticipantType.AGENCY) + + # 2.) datasets may only be created, if the respective participant type did not already create one. + InputValidationTimes.check_if_entry_already_exists_for_participant_type(user_data, loadedModel, content) + return + +def test_input_validation_times_fails_when_participant_type_deviates_from_shipcall_participant_map(): + # success + # user id 3 is assigned as participant_type=4, but the stub assigns participant_type=4 + user_data, loadedModel, content = get_valid_stub_for_pytests(user_id=3) + InputValidationTimes.check_if_user_fits_shipcall_participant_map(user_data, loadedModel, content) + + # fails + # user id 4 is assigned as participant_type=1, but the stub assigns participant_type=4 + with pytest.raises(ValidationError, match="is assigned to the shipcall in a different role."): + user_data, loadedModel, content = get_valid_stub_for_pytests(user_id=4) + InputValidationTimes.check_if_user_fits_shipcall_participant_map(user_data, loadedModel, content) + return + +def test_input_validation_times_fails_when_id_references_do_not_exist(): + # success: all IDs exist + user_data, loadedModel, content = get_valid_stub_for_pytests(user_id=3) + InputValidationTimes.check_dataset_references(content) + + # fails: IDs do not exist + # iterates once for each, berth_id, shipcall_id, participant_id and generates an artificial, non-existing ID + for key in ["berth_id", "shipcall_id", "participant_id"]: + user_data, loadedModel, content = get_valid_stub_for_pytests(user_id=3) + content[key] = loadedModel[key] = 9912737 + + with pytest.raises(ValidationError, match=f"The referenced {key} '{content[key]}' does not exist in the database."): + InputValidationTimes.check_dataset_references(content) + return + +from BreCal.schemas.model import ParticipantType + + +def test_input_validation_times_fails_when_missing_required_fields_arrival(): + """ + evaluates every individual combination of arriving shipcalls, where one of the required values is arbitrarily missing + randomly selects one of the non-terminal ParticipantTypes, which are reasonable (not .BSMD), and validates. This makes sure, + that over time, every possible combination has been tested. + """ + # arrival + not-terminal + non_terminal_list = [ParticipantType.AGENCY, ParticipantType.MOORING, ParticipantType.PILOT, ParticipantType.PORT_ADMINISTRATION, ParticipantType.TUG] + for key in ["eta_berth"]+InputValidationTimes.get_post_data_type_independent_fields(): + random_participant_type_for_unit_test = random.sample(non_terminal_list,k=1)[0] + + # pass: all required fields exist for the current shipcall type (arrival/incoming) + user_data, loadedModel, content = get_valid_stub_for_pytests(user_id=3) + loadedModel["shipcall_id"] = content["shipcall_id"] = 222 + loadedModel["participant_type"] = random_participant_type_for_unit_test + content["participant_type"] = int(random_participant_type_for_unit_test) + InputValidationTimes.check_times_required_fields_post_data(loadedModel, content) + + # fails: iteratively creates stubs, where one of the required keys is missing + user_data, loadedModel, content = get_valid_stub_for_pytests(user_id=3) + loadedModel["shipcall_id"] = content["shipcall_id"] = 222 + loadedModel["participant_type"] = random_participant_type_for_unit_test + content["participant_type"] = int(random_participant_type_for_unit_test) + with pytest.raises(ValidationError, match="At least one of the required fields is missing. Missing:"): + loadedModel[key] = content[key] = None + InputValidationTimes.check_times_required_fields_post_data(loadedModel, content) + + # arrival + terminal + for key in ["operations_start"]+InputValidationTimes.get_post_data_type_independent_fields(): + # pass: all required fields exist for the current shipcall type (arrival/incoming) + user_data, loadedModel, content = get_valid_stub_for_pytests(user_id=3) + loadedModel["shipcall_id"] = content["shipcall_id"] = 222 + loadedModel["participant_type"] = ParticipantType.TERMINAL + content["participant_type"] = int(ParticipantType.TERMINAL) + InputValidationTimes.check_times_required_fields_post_data(loadedModel, content) + + # fails: iteratively creates stubs, where one of the required keys is missing + user_data, loadedModel, content = get_valid_stub_for_pytests(user_id=3) + loadedModel["shipcall_id"] = content["shipcall_id"] = 222 + loadedModel["participant_type"] = ParticipantType.TERMINAL + content["participant_type"] = int(ParticipantType.TERMINAL) + + with pytest.raises(ValidationError, match="At least one of the required fields is missing. Missing:"): + loadedModel[key] = content[key] = None + InputValidationTimes.check_times_required_fields_post_data(loadedModel, content) + return + +def test_input_validation_times_fails_when_missing_required_fields_departure(): + """ + evaluates every individual combination of departing shipcalls, where one of the required values is arbitrarily missing + randomly selects one of the non-terminal ParticipantTypes, which are reasonable (not .BSMD), and validates. This makes sure, + that over time, every possible combination has been tested. + """ + # departure + not-terminal + non_terminal_list = [ParticipantType.AGENCY, ParticipantType.MOORING, ParticipantType.PILOT, ParticipantType.PORT_ADMINISTRATION, ParticipantType.TUG] + + for key in ["etd_berth"]+InputValidationTimes.get_post_data_type_independent_fields(): + # select a *random* particiipant type, which is reasonable and *not* TERMINAL, and validate the function. + random_participant_type_for_unit_test = random.sample(non_terminal_list,k=1)[0] + + # pass + user_data, loadedModel, content = get_valid_stub_for_pytests(user_id=3) + loadedModel["shipcall_id"] = content["shipcall_id"] = 241 + loadedModel["participant_type"] = random_participant_type_for_unit_test + content["participant_type"] = int(random_participant_type_for_unit_test) + InputValidationTimes.check_times_required_fields_post_data(loadedModel, content) + + # fails: iteratively creates stubs, where one of the required keys is missing + user_data, loadedModel, content = get_valid_stub_for_pytests(user_id=3) + loadedModel["shipcall_id"] = content["shipcall_id"] = 241 + loadedModel["participant_type"] = random_participant_type_for_unit_test + content["participant_type"] = int(random_participant_type_for_unit_test) + with pytest.raises(ValidationError, match="At least one of the required fields is missing. Missing:"): + loadedModel[key] = content[key] = None + InputValidationTimes.check_times_required_fields_post_data(loadedModel, content) + + # departure + terminal + for key in ["operations_end"]+InputValidationTimes.get_post_data_type_independent_fields(): + # pass + user_data, loadedModel, content = get_valid_stub_for_pytests(user_id=3) + loadedModel["shipcall_id"] = content["shipcall_id"] = 241 + loadedModel["participant_type"] = ParticipantType.TERMINAL + content["participant_type"] = int(ParticipantType.TERMINAL) + InputValidationTimes.check_times_required_fields_post_data(loadedModel, content) + + # fails: iteratively creates stubs, where one of the required keys is missing + user_data, loadedModel, content = get_valid_stub_for_pytests(user_id=3) + loadedModel["shipcall_id"] = content["shipcall_id"] = 241 + loadedModel["participant_type"] = ParticipantType.TERMINAL + content["participant_type"] = int(ParticipantType.TERMINAL) + with pytest.raises(ValidationError, match="At least one of the required fields is missing. Missing:"): + loadedModel[key] = content[key] = None + InputValidationTimes.check_times_required_fields_post_data(loadedModel, content) + return + +def test_input_validation_times_fails_when_missing_required_fields_shifting(): + """ + evaluates every individual combination of shifting shipcalls, where one of the required values is arbitrarily missing + randomly selects one of the non-terminal ParticipantTypes, which are reasonable (not .BSMD), and validates. This makes sure, + that over time, every possible combination has been tested. + """ + # shifting + not-terminal + non_terminal_list = [ParticipantType.AGENCY, ParticipantType.MOORING, ParticipantType.PILOT, ParticipantType.PORT_ADMINISTRATION, ParticipantType.TUG] + for key in ["eta_berth", "etd_berth"]+InputValidationTimes.get_post_data_type_independent_fields(): + # select a *random* particiipant type, which is reasonable and *not* TERMINAL, and validate the function. + random_participant_type_for_unit_test = random.sample(non_terminal_list,k=1)[0] + # pass: all required fields exist for the current shipcall type (arrival/incoming) + user_data, loadedModel, content = get_valid_stub_for_pytests(user_id=3) + loadedModel["shipcall_id"] = content["shipcall_id"] = 189 + loadedModel["participant_type"] =random_participant_type_for_unit_test + content["participant_type"] = int(random_participant_type_for_unit_test) + InputValidationTimes.check_times_required_fields_post_data(loadedModel, content) + + # fails: iteratively creates stubs, where one of the required keys is missing + user_data, loadedModel, content = get_valid_stub_for_pytests(user_id=3) + loadedModel["shipcall_id"] = content["shipcall_id"] = 189 + loadedModel["participant_type"] = random_participant_type_for_unit_test + content["participant_type"] = int(random_participant_type_for_unit_test) + with pytest.raises(ValidationError, match="At least one of the required fields is missing. Missing:"): + loadedModel[key] = content[key] = None + InputValidationTimes.check_times_required_fields_post_data(loadedModel, content) + + # shifting + terminal + for key in ["operations_start", "operations_end"]+InputValidationTimes.get_post_data_type_independent_fields(): + # pass: all required fields exist for the current shipcall type (arrival/incoming) + user_data, loadedModel, content = get_valid_stub_for_pytests(user_id=3) + loadedModel["shipcall_id"] = content["shipcall_id"] = 189 + loadedModel["participant_type"] = ParticipantType.TERMINAL + content["participant_type"] = int(ParticipantType.TERMINAL) + InputValidationTimes.check_times_required_fields_post_data(loadedModel, content) + + # fails: iteratively creates stubs, where one of the required keys is missing + user_data, loadedModel, content = get_valid_stub_for_pytests(user_id=3) + loadedModel["shipcall_id"] = content["shipcall_id"] = 189 + loadedModel["participant_type"] = ParticipantType.TERMINAL + content["participant_type"] = int(ParticipantType.TERMINAL) + with pytest.raises(ValidationError, match="At least one of the required fields is missing. Missing:"): + loadedModel[key] = content[key] = None + InputValidationTimes.check_times_required_fields_post_data(loadedModel, content) + return + + + +def test_input_validation_times_fails_when_participant_type_is_not_assigned__or__user_does_not_belong_to_the_same_participant_id(): + """ + There are two failure cases in InputValidationTimes.check_user_belongs_to_same_group_as_dataset_determines + 1.) when the participant type is simply not assigned + 2.) when the participant type matches to the user, but the participant_id is not assigned + + Test case: + shipcall_id 222 is assigned to the participants {"participant_id": 136, "type":2} and {"participant_id": 136, "type":8} + + Case 1: + When user_id 3 should be set as participant_type 4, the call fails, because type 4 is not assigned + + Case 2: + When user_id 2 (participant_id 2) should be set as participant_type 2, the call fails even though type 2 exists, + because participant_id 136 is assigned + + Case 3: + When user_id 28 (participant_id 136) is set as participant_type 2, the call passes. + """ + # fails: participant type 4 does not exist + user_data, loadedModel, content = get_valid_stub_for_pytests(user_id=3) + participant_type = 4 + loadedModel["shipcall_id"] = content["shipcall_id"] = 222 + loadedModel["participant_id"] = content["participant_id"] = 2 + loadedModel["participant_type"] = content["participant_type"] = participant_type + + with pytest.raises(ValidationError, match=f"Could not find a matching time dataset for the provided participant_type: {participant_type}. Found Time Datasets:"): + InputValidationTimes.check_user_belongs_to_same_group_as_dataset_determines(user_data, loadedModel=loadedModel, times_id=None) + + # fails: participant type 2 exists, but user_id 2 is part of the wrong participant_id group (user_id 28 or 29 would be) + user_data, loadedModel, content = get_valid_stub_for_pytests(user_id=2) + loadedModel["shipcall_id"] = content["shipcall_id"] = 222 + participant_type = 2 + loadedModel["participant_type"] = content["participant_type"] = participant_type + with pytest.raises(ValidationError, match="The dataset may only be changed by a user belonging to the same participant group as the times dataset is referring to. User participant_id:"): + InputValidationTimes.check_user_belongs_to_same_group_as_dataset_determines(user_data, loadedModel=loadedModel, times_id=None) + + # pass: participant type 2 exists & user_id is part of participant_id group 136, which is correct + user_data, loadedModel, content = get_valid_stub_for_pytests(user_id=28) + loadedModel["shipcall_id"] = content["shipcall_id"] = 222 + participant_type = 2 + loadedModel["participant_type"] = content["participant_type"] = participant_type + InputValidationTimes.check_user_belongs_to_same_group_as_dataset_determines(user_data, loadedModel=loadedModel, times_id=None) + return + + +def test_input_validation_times_put_request_fails_when_id_field_is_missing(): + """used within PUT-requests. When 'id' is missing, a ValidationError is issued""" + # passes: as an 'id' is provided + user_data, loadedModel, content = get_valid_stub_for_pytests(user_id=3) + content["id"] = 379 + InputValidationTimes.check_times_required_fields_put_data(content) + + # fails: 'id' field is missing + user_data, loadedModel, content = get_valid_stub_for_pytests(user_id=3) + content.pop("id",None) + with pytest.raises(ValidationError, match="A PUT-request requires an 'id' reference, which was not found."): + InputValidationTimes.check_times_required_fields_put_data(content) + return + +def test_input_validation_times_delete_request_fails_when_times_id_is_deleted_already(): + # passes: id exists + times_id = 379 + InputValidationTimes.check_if_entry_is_already_deleted(times_id) + + # passes: id exists + times_id = 391 + InputValidationTimes.check_if_entry_is_already_deleted(times_id) + + # fails + times_id = 11 + with pytest.raises(ValidationError, match=f"The selected time entry is already deleted. ID: {times_id}"): + InputValidationTimes.check_if_entry_is_already_deleted(times_id) + + # fails + times_id = 4 + with pytest.raises(ValidationError, match=f"The selected time entry is already deleted. ID: {times_id}"): + InputValidationTimes.check_if_entry_is_already_deleted(times_id) + return + +def test_input_validation_times_delete_request_fails_when_times_id_does_not_exist_(): + # passes: times_id exists + user_data, loadedModel, content = get_valid_stub_for_pytests(user_id=28) + times_id = 392 + InputValidationTimes.check_user_belongs_to_same_group_as_dataset_determines(user_data, loadedModel=None, times_id=times_id) + + # fails: times_id does not exist + user_data, loadedModel, content = get_valid_stub_for_pytests(user_id=28) + times_id = 4 + with pytest.raises(ValidationError, match=f"Unknown times_id. Could not find a matching entry for ID: {times_id}"): + InputValidationTimes.check_user_belongs_to_same_group_as_dataset_determines(user_data, loadedModel=None, times_id=times_id) + return + +def test_input_validation_times_delete_request_fails_when_user_belongs_to_wrong_participant_id(): + # fails: participant_id should be 136, but user_id=3 belongs to participant_id=2 + user_data, loadedModel, content = get_valid_stub_for_pytests(user_id=3) + times_id = 392 + + with pytest.raises(ValidationError, match=f"The dataset may only be changed by a user belonging to the same participant group as the times dataset is referring to. User participant_id:"): + InputValidationTimes.check_user_belongs_to_same_group_as_dataset_determines(user_data, loadedModel=None, times_id=times_id) + + # passes: participant_id should be 136, and user_id=28 belongs to participant_id=2 + user_data, loadedModel, content = get_valid_stub_for_pytests(user_id=28) + times_id = 392 + InputValidationTimes.check_user_belongs_to_same_group_as_dataset_determines(user_data, loadedModel=None, times_id=times_id) + return + +