diff --git a/src/server/BreCal/validators/input_validation_times.py b/src/server/BreCal/validators/input_validation_times.py index 6dd1234..10db2e9 100644 --- a/src/server/BreCal/validators/input_validation_times.py +++ b/src/server/BreCal/validators/input_validation_times.py @@ -23,7 +23,7 @@ def build_post_data_type_dependent_required_fields_dict()->dict[ShipcallType,dic 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'. + 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:{ @@ -69,7 +69,7 @@ class InputValidationTimes(): Example: InputValidationTimes.evaluate(user_data, loadedModel, content) - When the data violates one of the rules, a marshmallow.ValidationError is raised, which details the issues. + When the data violates one of the rules, a marshmallow.ValidationError is raised, which details the issues. """ def __init__(self) -> None: pass @@ -91,7 +91,7 @@ class InputValidationTimes(): # 4.) Value checking InputValidationTimes.check_dataset_values(user_data, loadedModel, content) return - + @staticmethod def evaluate_put_data(user_data:dict, loadedModel:dict, content:dict): # 1.) Check for the presence of required fields @@ -106,7 +106,7 @@ class InputValidationTimes(): # 4.) Value checking InputValidationTimes.check_dataset_values(user_data, loadedModel, content) return - + @staticmethod def evaluate_delete_data(user_data:dict, times_id:typing.Optional[int]): # 0.) an ID reference must be provided and will be converted to int @@ -120,13 +120,13 @@ class InputValidationTimes(): # 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): """ 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. + 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. @@ -146,19 +146,19 @@ class InputValidationTimes(): if is_bsmd: raise ValidationError({"participant_type":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 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. + # validates the times dataset. # ensure loadedModel["participant_type"] is of type ParticipantType if not isinstance(loadedModel["participant_type"], ParticipantType): @@ -166,18 +166,18 @@ class InputValidationTimes(): if ParticipantType.BSMD in loadedModel["participant_type"]: raise ValidationError({"participant_type":f"current user belongs to BSMD. Cannot post times datasets. Found user data: {user_data}"}) - - if (loadedModel["etd_interval_end"] is not None) and (loadedModel["etd_berth"] is not None): + + if ("etd_interval_end" in loadedModel and loadedModel["etd_interval_end"] is not None) and ("etd_berth" in loadedModel and loadedModel["etd_berth"] is not None): time_end_after_time_start = loadedModel["etd_interval_end"] >= loadedModel["etd_berth"] if not time_end_after_time_start: raise ValidationError({"etd":f"The provided time interval for the estimated departure time is invalid. The interval end takes place before the interval start. Found interval data: {loadedModel['etd_berth']} to {loadedModel['etd_interval_end']}"}) - if (loadedModel["eta_interval_end"] is not None) and (loadedModel["eta_berth"] is not None): + if ("eta_interval_end" in loadedModel and loadedModel["eta_interval_end"] is not None) and ("eta_berth" in loadedModel and loadedModel["eta_berth"] is not None): time_end_after_time_start = loadedModel["eta_interval_end"] >= loadedModel["eta_berth"] if not time_end_after_time_start: raise ValidationError({"eta":f"The provided time interval for the estimated arrival time is invalid. The interval begin takes place after the interval end. Found interval data: {loadedModel['eta_berth']} to {loadedModel['eta_interval_end']}"}) return - + @staticmethod def check_dataset_references(content:dict): """ @@ -192,25 +192,25 @@ class InputValidationTimes(): valid_berth_id_reference = check_if_berth_id_is_valid(berth_id) if not valid_berth_id_reference: raise ValidationError({"berth_id":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({"shipcall_id":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, participant_type=None) if not valid_participant_id_reference: raise ValidationError({"participant_id":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. + 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. + 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"] @@ -240,14 +240,14 @@ class InputValidationTimes(): verbosity_tuple = [(field, missing) for field, missing in zip(required_fields, missing_required_fields) if missing] raise ValidationError({"required_fields":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({"id":f"A PUT-request requires an 'id' reference, which was not found."}) return - + @staticmethod def get_post_data_type_independent_fields()->list[str]: """ @@ -257,16 +257,16 @@ class InputValidationTimes(): "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'). + 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 + 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. @@ -282,7 +282,7 @@ class InputValidationTimes(): dependent_required_fields = dependent_required_fields_dict.get(shipcall_type,{}).get(participant_type,[]) dependent_required_fields = dependent_required_fields if dependent_required_fields is not None else [] return dependent_required_fields - + @staticmethod def check_if_user_fits_shipcall_participant_map(user_data:dict, loadedModel:dict, content:dict, spm_shipcall_data=None): """ @@ -290,15 +290,15 @@ class InputValidationTimes(): 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. + authorized to send the request. This method also checks for a special case: when an assigned AGENCY participant has the .BSMD flag enabled, a user of type BSMD may also post the times dataset. options: - spm_shipcall_data: + 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. + an optional argument to allow for much easier unit testing. """ ### TIMES DATASET (ShipcallParticipantMap) ### # identify shipcall_id @@ -310,13 +310,13 @@ class InputValidationTimes(): # 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? AND type=?type?)", + query = "SELECT participant_id, type FROM shipcall_participant_map WHERE (shipcall_id=?shipcall_id? AND type=?type?)", param={"shipcall_id":shipcall_id, "type":int(DATASET_participant_type)}, pooledConnection=None ) - + DATASET_participant_id = InputValidationTimes.get_participant_id_from_shipcall_participant_map(shipcall_id, participant_type=DATASET_participant_type, spm_shipcall_data=spm_shipcall_data) - + ### USER DATA (token) ### # identify user's participant_id & type (get all participants; then filter these for the {participant_id}) user_participant_id = user_data["participant_id"] #participants = get_participant_id_dictionary() #participant_type = ParticipantType(participants.get(participant_id,{}).get("type")) @@ -326,7 +326,7 @@ class InputValidationTimes(): if (special_case__bsmd_may_edit_agency_dataset): # when a BSMD user posts a dataset of an AGENCY with BSMD-flag, there is no violation return - + # check, if participant_id is assigned to the ShipcallParticipantMap matching_spm = [ spm @@ -337,7 +337,7 @@ class InputValidationTimes(): if not len(matching_spm)>0: raise ValidationError({"participant_id":f'The participant group with id {user_participant_id} is not assigned to the shipcall. Found ShipcallParticipantMap: {spm_shipcall_data}'}) # part of a pytest.raises 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""" @@ -356,7 +356,7 @@ class InputValidationTimes(): if participant_type_exists_already: raise ValidationError({"participant_type":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, pdata:typing.Optional[list[dict]]=None): """ @@ -371,7 +371,7 @@ class InputValidationTimes(): times_id is used to directly identify the matching times entry A special exception takes place, when a participant of type AGENCY is involved. In those times-entries, users with the - IS_BSMD-Flag may also edit the entry. + IS_BSMD-Flag may also edit the 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." @@ -384,7 +384,7 @@ class InputValidationTimes(): shipcall_id = loadedModel["shipcall_id"] participant_type = loadedModel["participant_type"] - # get the matching entry from the shipcall participant map. Raise an error, when there is no match. + # get the matching entry from the shipcall participant map. Raise an error, when there is no match. participant_id_of_times_dataset = InputValidationTimes.get_participant_id_from_shipcall_participant_map(shipcall_id, participant_type) # commonly used in the DELETE-request @@ -402,7 +402,7 @@ class InputValidationTimes(): participant_type = pdata[0].get("participant_type") shipcall_id = pdata[0].get("shipcall_id") - # get the matching entry from the shipcall participant map. Raise an error, when there is no match. + # get the matching entry from the shipcall participant map. Raise an error, when there is no match. participant_id_of_times_dataset = pdata[0].get("participant_id") # participant_id_of_times_dataset = InputValidationTimes.get_participant_id_from_shipcall_participant_map(shipcall_id, participant_type) @@ -415,7 +415,7 @@ class InputValidationTimes(): else: raise ValidationError({"user_participant_type":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 - + @staticmethod def get_participant_id_from_shipcall_participant_map(shipcall_id:int, participant_type:int, spm_shipcall_data=None)->int: """use shipcall_id and participant_type to identify the matching participant_id""" @@ -424,17 +424,17 @@ class InputValidationTimes(): if spm_shipcall_data is None: spm_shipcall_data = execute_sql_query_standalone( - query=SQLQuery.get_shipcall_participant_map_by_shipcall_id_and_type(), - param={"id":shipcall_id, "type":participant_type}, + query=SQLQuery.get_shipcall_participant_map_by_shipcall_id_and_type(), + param={"id":shipcall_id, "type":participant_type}, command_type="query") # returns a list of matches - + # raise an error when there are no matches if len(spm_shipcall_data)==0: raise ValidationError({"participant_type":f"Could not find a matching time dataset for the provided participant_type: {participant_type} at shipcall with id {shipcall_id}."}) - + participant_id_of_times_dataset = spm_shipcall_data[0].get("participant_id") return participant_id_of_times_dataset - + @staticmethod def check_if_bsmd_may_edit_agency_dataset(user_participant_id:int, participant_id_of_times_dataset:int, participant_type:ParticipantType)->bool: """ @@ -448,35 +448,35 @@ class InputValidationTimes(): args: user_participant_id: ID of the user, obtained from the jwt-token - participant_id_of_times_dataset: assigned participant of the shipcall, obtained from the ShipcallParticipantMap + participant_id_of_times_dataset: assigned participant of the shipcall, obtained from the ShipcallParticipantMap """ - # when the participant type of the dataset is not an AGENCY, this exception rule does not take place + # when the participant type of the dataset is not an AGENCY, this exception rule does not take place dataset_participant_type_is_agency = int(participant_type)==int(ParticipantType.AGENCY) if not dataset_participant_type_is_agency: return False - + ### TIMES ENTRY (ShipcallParticipantMap) ### # identify, whether the dataset's assigned participant has the BSMD flag agency_has_bsmd_flag = InputValidationTimes.check_if_participant_has_bsmd_flag(participant_id=participant_id_of_times_dataset) - + ### USER DATA (token) ### # determine, whether the user is of participant_type BSMD user_is_bsmd_type = check_if_user_is_bsmd_type(user_data={"participant_id":user_participant_id}) return (agency_has_bsmd_flag) & (user_is_bsmd_type) - + @staticmethod def check_if_participant_has_bsmd_flag(participant_id:int)->bool: """ Given a participant_id, this method checks, whether the participant with {participant_id} - has the .BSMD flag in the .flags field. + has the .BSMD flag in the .flags field. """ # get the dataset's assigned Participant, which matches the SPM entry participant = execute_sql_query_standalone( SQLQuery.get_participant_from_id(), - param={"participant_id":participant_id}, - command_type="single", + param={"participant_id":participant_id}, + command_type="single", pooledConnection=None) - + has_bsmd_flag = ParticipantFlag.BSMD in [ParticipantFlag(participant.get("flags"))] return has_bsmd_flag @@ -486,7 +486,7 @@ def deprecated_build_post_data_type_dependent_required_fields_dict()->dict[Shipc 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'. + 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:{