diff --git a/src/server/BreCal/schemas/model.py b/src/server/BreCal/schemas/model.py index 5547020..653b67c 100644 --- a/src/server/BreCal/schemas/model.py +++ b/src/server/BreCal/schemas/model.py @@ -558,7 +558,8 @@ class ShipSchema(Schema): @validates("imo") def validate_imo(self, value): - imo_length = len(str(value)) + value = str(value).zfill(7) # 1 becomes '0000001' (7 characters). 12345678 becomes '12345678' (8 characters) + imo_length = len(value) if imo_length != 7: raise ValidationError(f"'imo' should be a 7-digit number") return diff --git a/src/server/BreCal/stubs/shipcall.py b/src/server/BreCal/stubs/shipcall.py index 299fdbe..aaaf12f 100644 --- a/src/server/BreCal/stubs/shipcall.py +++ b/src/server/BreCal/stubs/shipcall.py @@ -99,7 +99,7 @@ def create_postman_stub_shipcall(): """ shipcall = { 'ship_id': 1, - 'type': 1, + 'type': "arrival", #1, 'eta': (datetime.datetime.now()+datetime.timedelta(hours=3)).isoformat(), 'voyage': '43B', 'arrival_berth_id':142, @@ -137,12 +137,13 @@ def get_stub_valid_shipcall_arrival(): post_data = { **get_stub_valid_shipcall_base(), **{ - 'type': int(ShipcallType.arrival), + 'type': ShipcallType.arrival.name, #int(ShipcallType.arrival), 'eta': eta, 'participants':get_stub_list_of_valid_participants(), 'arrival_berth_id':139, } } + post_data.pop('etd',None) return post_data def get_stub_valid_shipcall_departure(): @@ -151,12 +152,13 @@ def get_stub_valid_shipcall_departure(): post_data = { **get_stub_valid_shipcall_base(), **{ - 'type': int(ShipcallType.departure), + 'type': ShipcallType.departure.name, #int(ShipcallType.departure), 'etd': etd, 'participants':get_stub_list_of_valid_participants(), 'departure_berth_id':139, } } + post_data.pop('eta',None) return post_data def get_stub_valid_shipcall_shifting(): @@ -166,7 +168,7 @@ def get_stub_valid_shipcall_shifting(): post_data = { **get_stub_valid_shipcall_base(), **{ - 'type': int(ShipcallType.shifting), + 'type': ShipcallType.shifting.name, #int(ShipcallType.shifting), 'eta': eta, 'etd': etd, 'participants':get_stub_list_of_valid_participants(), diff --git a/src/server/BreCal/validators/input_validation_ship.py b/src/server/BreCal/validators/input_validation_ship.py index eab2480..da3be4d 100644 --- a/src/server/BreCal/validators/input_validation_ship.py +++ b/src/server/BreCal/validators/input_validation_ship.py @@ -69,7 +69,7 @@ class InputValidationShip(): @staticmethod def optionally_evaluate_bollard_pull_value(content:dict): bollard_pull = content.get("bollard_pull",None) - is_tug = content.get("is_tug", None) + is_tug = content.get("is_tug", False) # default to 'False', so the bollard pull entry fails unless the property is also actively set if bollard_pull is not None: if not is_tug: diff --git a/src/server/BreCal/validators/input_validation_shipcall.py b/src/server/BreCal/validators/input_validation_shipcall.py index 0b8ad3b..cfaf6b1 100644 --- a/src/server/BreCal/validators/input_validation_shipcall.py +++ b/src/server/BreCal/validators/input_validation_shipcall.py @@ -5,7 +5,7 @@ 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.schemas.model import Ship, Shipcall, Berth, User, Participant, ShipcallType, ShipcallParticipantMap from BreCal.impl.participant import GetParticipant from BreCal.impl.ships import GetShips from BreCal.impl.berths import GetBerths @@ -15,8 +15,11 @@ from BreCal.validators.input_validation_utils import check_if_user_is_bsmd_type, 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 +from BreCal.database.sql_queries import SQLQuery import werkzeug + + class InputValidationShipcall(): """ This class combines a complex set of individual input validation functions into a joint object. @@ -37,13 +40,13 @@ class InputValidationShipcall(): this function combines multiple validation functions to verify data, which is sent to the API as a shipcall's POST-request checks: - 1. permission: only participants that belong to the BSMD group are allowed to POST shipcalls + 1. permission: only participants that belong to the BSMD or AGENCY groups are allowed to POST shipcalls 2. reference checks: all refered objects within the Shipcall must exist 3. existance of required fields 4. reasonable values: validates the values within the Shipcall """ - # check for permission (only BSMD-type participants) - InputValidationShipcall.check_user_is_bsmd_type(user_data) + # check for permission (only BSMD-type or AGENT-type participants) + InputValidationShipcall.check_user_is_bsmd_or_agent_type(user_data) # check references (referred IDs must exist) InputValidationShipcall.check_referenced_ids(loadedModel) @@ -64,18 +67,21 @@ class InputValidationShipcall(): this function combines multiple validation functions to verify data, which is sent to the API as a shipcall's PUT-request checks: - 1. whether the user belongs to participant group type BSMD - 2. users of the agency may edit the shipcall, when the shipcall-participant-map entry lists them - 3. existance of required fields - 4. all value-rules of the POST evaluation - 5. a canceled shipcall may not be changed + 1. user's authority: + a) whether the user's participant is assigned to the shipcall (via shipcall-participant-map) + b) whether the user is either an AGENCY (assigned) or the BSMD, in case the AGENCY allows the BSMD to edit their shipcalls + 2. existance of required fields + 3. all value-rules of the POST evaluation + 4. a canceled shipcall may not be changed """ - # check for permission (only BSMD-type participants) - # #TODO: are both, bsmd and agency, user types accepted? - InputValidationShipcall.check_user_is_bsmd_type(user_data) + # check, whether the shipcall_id exists + InputValidationShipcall.check_shipcall_id_exists(loadedModel) # check, whether an agency is listed in the shipcall-participant-map - InputValidationShipcall.check_agency_in_shipcall_participant_map(user_data, loadedModel, content) + # deprecated: InputValidationShipcall.check_agency_in_shipcall_participant_map(user_data, loadedModel, content) + + # check, whether the user belongs to the assigned agency or to BSMD in case the special flag is enabled + InputValidationShipcall.check_user_is_authorized_for_put_request(user_data, loadedModel, content) # the ID field is required, all missing fields will be ignored in the update InputValidationShipcall.check_required_fields_of_put_request(content) @@ -136,9 +142,11 @@ class InputValidationShipcall(): a list of entries obtained from the ShipcallParticipantMap. These are deserialized dictionaries. e.g., [{'participant_id': 136, 'type': 8}, ] """ + raise Exception("deprecated") 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 spm_shipcall_data = execute_sql_query_standalone( + # #TODO_refactor: place this within the SQLQuery object query = "SELECT participant_id, type FROM shipcall_participant_map WHERE shipcall_id=?shipcall_id?", param={"shipcall_id":loadedModel["id"]}, pooledConnection=None @@ -146,21 +154,24 @@ class InputValidationShipcall(): # which role should be set by the PUT request? If the agency is about to be set, an error will be created # read the user data from the JWT token (set when login is performed) - user_type = get_participant_type_from_user_data(user_data) # decode JWT -> get 'type' value + user_type = get_participant_type_from_user_data(user_data) # decode JWT -> get 'type' value (guarantees to convert user type into an IntFlag) + assert isinstance(user_type, ParticipantType) # select the matching entries from the ShipcallParticipantMap agency_entries = [spm_entry for spm_entry in spm_shipcall_data if int(spm_entry.get("type"))==int(ParticipantType.AGENCY)] # find all entries of type AGENCY (there should be at max. 1) # when the request stems from an AGENCY user, and the user wants to PUT an AGENCY role, the request should fail # boolean: check, whether any of the assigned participants is of type AGENCY - types = [participant.get("type") for participant in loadedModel["participants"]] # readout the participants from the loadedModel, which shall be assigned by the PUT request + types = [participant.get("type",0) for participant in loadedModel["participants"]] # readout the participants from the loadedModel, which shall be assigned by the PUT request any_type_is_agency = any([int(type_) == int(ParticipantType.AGENCY) for type_ in types]) # check, whether *any* of the participants is an agency - if not (int(user_type) in [int(ParticipantType.AGENCY), int(ParticipantType.BSMD)]): + if not ((ParticipantType.AGENCY in user_type) or (ParticipantType.BSMD in user_type)): # user not AGENCY or BSMD raise werkzeug.exceptions.Forbidden(f"PUT Requests for shipcalls can only be issued by AGENCY or BSMD users.") # Forbidden: 403 - if (int(user_type) == int(ParticipantType.AGENCY)) & (any_type_is_agency): + # Placeholder: when a user is an AGENCY, + + if (ParticipantType.AGENCY in user_type) & (any_type_is_agency): # self-assignment: agency sets agency participant raise werkzeug.exceptions.Forbidden(f"An agency cannot self-register for a shipcall. The request is issued by an agency-user and tries to assign an AGENCY as the participant of the shipcall.") # Forbidden: 403 @@ -181,21 +192,32 @@ class InputValidationShipcall(): return @staticmethod - def check_user_is_bsmd_type(user_data): + def check_user_is_bsmd_or_agent_type(user_data): """ 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 or put shipcalls. Found user data: {user_data}") + # use the decoded JWT token and extract the participant type + participant_type = get_participant_type_from_user_data(user_data) + + is_bsmd = (ParticipantType.BSMD in participant_type) + is_agency = (ParticipantType.AGENCY in participant_type) + + is_bsmd_or_agency = (is_bsmd) or (is_agency) + + if not is_bsmd_or_agency: + raise ValidationError(f"current user must be either of participant type BSMD or AGENCY. Cannot post or put shipcalls. Found user data: {user_data} and participant_type: {participant_type}") return - + @staticmethod def check_referenced_ids(loadedModel): """ check, whether the referenced entries exist (e.g., when a Ship ID is referenced, but does not exist, the validation fails) """ + # #TODO: arrival and departure berth id should be coupled with the shipcall type. One shall not provide + # arrival berth id when the shipcall type is departure or vise versa. + # a similar logic has already been implemented to the eta/etd or for the operation windows + # get all IDs from the loadedModel ship_id = loadedModel.get("ship_id", None) arrival_berth_id = loadedModel.get("arrival_berth_id", None) @@ -219,7 +241,7 @@ class InputValidationShipcall(): raise ValidationError(f"one of the provided participant ids is invalid. Could not find one of these in the database: {participants}") valid_participant_types = check_if_participant_ids_and_types_are_valid(participants=participants) - if not valid_participant_types: + if not valid_participant_types: # #TODO: according to Daniel, there may eventually be multi-assignment of participants for the same role raise ValidationError(f"every participant id and type should be listed only once. Found multiple entries for one of the participants.") @staticmethod @@ -290,7 +312,16 @@ class InputValidationShipcall(): # obtain the current datetime to check, whether the provided values are in the future time_now = datetime.datetime.now() - type_ = loadedModel.get("type", int(ShipcallType.undefined)) + type_ = loadedModel.get("type", ShipcallType.undefined.name) + if isinstance(type_, str): # convert the name string to a ShipcallType data model + type_ = ShipcallType[type_] + + # #TODO: *if* this is a PUT-request, one shall load the existing values from the database, overwrite the none-null + # values *and then* perform the validation. + # Example: eta and etd are set in the POST-request. User wants to execute a PUT-request with only the etd. + # Internally, the backend must still verify, that eta < etd! + # Same applies to tidal_window_from & tidal_window_to + eta = loadedModel.get("eta") etd = loadedModel.get("etd") tidal_window_from = loadedModel.get("tidal_window_from", None) @@ -300,7 +331,7 @@ class InputValidationShipcall(): InputValidationShipcall.check_times_in_future_based_on_type(type_, time_now, eta, etd) # Tidal Window - InputValidationShipcall.check_tidal_window_in_future(time_now, tidal_window_from, tidal_window_to) + InputValidationShipcall.check_tidal_window_in_future(type_, time_now, tidal_window_from, tidal_window_to) return @staticmethod @@ -312,23 +343,57 @@ class InputValidationShipcall(): departure: etd shifting: eta & etd """ + if (eta is None) and (etd is None): + return + + if type_ is None: + raise ValidationError(f"when providing 'eta' or 'etd', one must provide the type of the shipcall, so the datetimes can be verified.") + + if not isinstance(type_, (int, ShipcallType)): + type_ = ShipcallType[type_] + + # #TODO: properly handle what happens, when eta or etd (or both) are None if int(type_)==int(ShipcallType.undefined): raise ValidationError(f"providing 'type' is mandatory. Missing key!") elif int(type_)==int(ShipcallType.arrival): + if eta is None: # null values -> no violation + return + if not eta > time_now: raise ValidationError(f"'eta' must be in the future. Incorrect datetime provided. Current Time: {time_now}. ETA: {eta}.") + if etd is not None: + raise ValidationError(f"'etd' should not be set when the shipcall type is 'arrival'.") + elif int(type_)==int(ShipcallType.departure): + if etd is None: # null values -> no violation + return + if not etd > time_now: raise ValidationError(f"'etd' must be in the future. Incorrect datetime provided. Current Time: {time_now}. ETD: {etd}.") + + if eta is not None: + raise ValidationError(f"'eta' should not be set when the shipcall type is 'departure'.") + elif int(type_)==int(ShipcallType.shifting): + if (eta is None) and (etd is None): # null values -> no violation + return + + if not ((eta is not None) and (etd is not None)): + # for PUT-requests, a user could try modifying only 'eta' or only 'etd'. To simplify the + # rules, a user is only allowed to provide *both* values. + raise ValidationError(f"For shifting shipcalls one should always provide, both, eta and etd.") + if (not eta > time_now) or (not etd > time_now): raise ValidationError(f"'eta' and 'etd' must be in the future. Incorrect datetime provided. Current Time: {time_now}. ETA: {eta}. ETD: {etd}") if (not etd > eta): raise ValidationError(f"'etd' must be larger than 'eta'. The ship cannot depart, before it has arrived. Found: ETA {eta}, ETD: {etd}") + + if (eta is not None and etd is None) or (eta is None and etd is not None): + raise ValidationError(f"'eta' and 'etd' must both be provided when the shipcall type is 'shifting'.") return @staticmethod - def check_tidal_window_in_future(time_now, tidal_window_from, tidal_window_to): + def check_tidal_window_in_future(type_, time_now, tidal_window_from, tidal_window_to): if tidal_window_to is not None: if not tidal_window_to >= time_now: raise ValidationError(f"'tidal_window_to' must be in the future. Incorrect datetime provided.") @@ -375,24 +440,71 @@ class InputValidationShipcall(): shipcall_id = content.get("id", None) if shipcall_id is None: raise ValidationError(f"A PUT request requires an 'id' to refer to.") - - + + @staticmethod + def check_shipcall_id_exists(loadedModel): + """simply checks, whether the defined shipcall ID exists in the database. Otherwise, a PUT-request must fail.""" + shipcall_id = loadedModel.get("id") + query = 'SELECT * FROM shipcall where (id = ?shipcall_id?)' + shipcalls = execute_sql_query_standalone(query=query, model=Shipcall, param={"shipcall_id" : shipcall_id}) + if len(shipcalls)==0: + raise ValidationError(f"unknown shipcall_id. There are no shipcalls with the ID {shipcall_id}") + return -""" -# copy -def validate_posted_shipcall_data(user_data:dict, loadedModel:dict, content:dict): - ##### Section 1: check user_data ##### - # DONE: refactored - - ##### Section 2: check loadedModel ##### - # DONE: refactored + @staticmethod + def check_user_is_authorized_for_put_request(user_data:dict, loadedModel:dict, content:dict, shipcall_participant_map:typing.Optional[list[ShipcallParticipantMap]]=None): + """ + This method verifies, whether a user is authorized to create a PUT-request for shipcalls. + To be authorized, a user should either + a) belong to the ASSIGNED agency participant group + b) belong to a BSMD participant, if the assigned agency has enabled the bit flag - ##### Section 3: check content ##### - # DONE: refactored - + When there is not yet an assigned agency for the respective shipcall, the request fails, and the user is considered as not authorized. + This mechanism prevents self-assignment of an agency to arbitrary shipcalls. + """ + ### preparation ### + # use the decoded JWT token and extract the participant type & participant id + participant_id = user_data.get("participant_id") + participant_type = get_participant_type_from_user_data(user_data) - ##### Section 4: check loadedModel & content ##### - # DONE: refactored ET and BERTH ID existance check - # DONE: refactored 'time in future' checks - return -""" \ No newline at end of file + # get the shipcall id + shipcall_id = loadedModel.get("id") + + ### AGENCY in SPM ### + # determine, who is assigned as the agency for the shipcall + if shipcall_participant_map is None: + query = 'SELECT * FROM shipcall_participant_map where (shipcall_id = ?shipcall_id? AND type=?participant_type?)' + assigned_agency = execute_sql_query_standalone(query=query, model=ShipcallParticipantMap, param={"shipcall_id" : shipcall_id, "participant_type":int(ParticipantType.AGENCY)}) + else: + assigned_agency = [spm for spm in shipcall_participant_map if int(spm.type) == int(ParticipantType.AGENCY)] + + if len(assigned_agency)==0: + raise ValidationError(f"There is no assigned agency for the shipcall with ID {shipcall_id}.") + + elif len(assigned_agency)>1: + raise ValidationError(f"Internal error? Found more than one assigned agency for the shipcall with ID {shipcall_id}. Found: {assigned_agency}") + + else: + assigned_agency = assigned_agency[0] + + # determine, whether the assigned agency has set the BSMD-flag to allow BSMD users to edit their assigned shipcalls + query = 'SELECT * FROM participant where (id = ?participant_id?)' + agency_participant = execute_sql_query_standalone(query=query, param={"participant_id" : participant_id}, command_type="single", model=Participant) + + assert isinstance(agency_participant.flags, int), f"this method has currently only been developed with 'flags' being set as an integer. Found: {type(agency_participant.flags)}" + agency_has_bsmd_flag = agency_participant.flags==1 # once the flags are an IntFlag, change the boolean check to: (ParticipantFlag.BSMD in agency_participant.flags) + + ### USER authority ### + # determine, whether the user is a) the assigned agency or b) a BSMD participant + user_is_assigned_agency = (participant_id == assigned_agency.participant_id) + user_is_bsmd = (ParticipantType.BSMD in participant_type) + + # when the BSMD flag is set: the user must be either BSMD or the assigned agency + # when the BSMD flag is not set: the user must be the assigned agency + user_is_authorized = (user_is_bsmd or user_is_assigned_agency) if agency_has_bsmd_flag else user_is_assigned_agency + + if not user_is_authorized: + raise werkzeug.exceptions.Forbidden(f"PUT Requests for shipcalls can only be issued by an assigned AGENCY or BSMD users (if the special-flag is enabled). Assigned Agency: {assigned_agency} with Flags: {agency_participant.flags}") # Forbidden: 403 + return + + \ No newline at end of file diff --git a/src/server/BreCal/validators/input_validation_times.py b/src/server/BreCal/validators/input_validation_times.py index 5ad3c67..81fc84c 100644 --- a/src/server/BreCal/validators/input_validation_times.py +++ b/src/server/BreCal/validators/input_validation_times.py @@ -188,7 +188,7 @@ class InputValidationTimes(): 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) + valid_participant_id_reference = check_if_participant_id_is_valid_standalone(participant_id, participant_type=None) if not valid_participant_id_reference: raise ValidationError(f"The referenced participant_id '{participant_id}' does not exist in the database.") @@ -325,7 +325,7 @@ class InputValidationTimes(): ] if not len(matching_spm)>0: - raise ValidationError(f'The participant group with id {user_participant_id} is not assigned to the shipcall. Found ShipcallParticipantMap: {spm_shipcall_data}') + raise ValidationError(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 @@ -348,7 +348,7 @@ class InputValidationTimes(): 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): + 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): """ 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 @@ -379,9 +379,11 @@ class InputValidationTimes(): # commonly used in the DELETE-request 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, participant_type FROM times WHERE id = ?id?" - pdata = execute_sql_query_standalone(query=query, param={"id":times_id}, pooledConnection=None) + if pdata is None: # regular behavior. pdata is only defined in unit tests. + # perform an SQL query. Creates a pooled connection internally, queries the database, then closes the connection. + query = "SELECT participant_id, participant_type, shipcall_id FROM times WHERE id = ?id?" + pdata = execute_sql_query_standalone(query=query, param={"id":times_id}, pooledConnection=None) + print(pdata) # extracts the participant_id from the first matching entry, if applicable if not len(pdata)>0: @@ -392,7 +394,8 @@ class InputValidationTimes(): shipcall_id = pdata[0].get("shipcall_id") # 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) + 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) # when the user's participant id is different from the times dataset, an exception is raised if user_participant_id != participant_id_of_times_dataset: @@ -407,6 +410,9 @@ class InputValidationTimes(): @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""" + if shipcall_id is None: + raise ValidationError(f"Could not find a referenced shipcall_id within the request.") + if spm_shipcall_data is None: spm_shipcall_data = execute_sql_query_standalone( query=SQLQuery.get_shipcall_participant_map_by_shipcall_id_and_type(), @@ -458,7 +464,7 @@ class InputValidationTimes(): # get the dataset's assigned Participant, which matches the SPM entry participant = execute_sql_query_standalone( SQLQuery.get_participant_from_id(), - param={"id":participant_id}, + param={"participant_id":participant_id}, command_type="single", pooledConnection=None) diff --git a/src/server/BreCal/validators/input_validation_utils.py b/src/server/BreCal/validators/input_validation_utils.py index 19c6ac9..a78a75a 100644 --- a/src/server/BreCal/validators/input_validation_utils.py +++ b/src/server/BreCal/validators/input_validation_utils.py @@ -8,6 +8,7 @@ from BreCal.impl.berths import GetBerths from BreCal.impl.shipcalls import GetShipcalls from BreCal.database.enums import ParticipantType +from marshmallow import ValidationError def get_participant_id_dictionary(): """ @@ -85,6 +86,26 @@ def check_if_user_is_bsmd_type(user_data:dict)->bool: is_bsmd = ParticipantType.BSMD in participant_type return is_bsmd +def check_if_user_has_bsmd_flag(user_data:dict)->bool: + """ + given a dictionary of user data, determine the respective participant id and read, whether + that participant is a .BSMD-type + + Note: ParticipantType is an IntFlag. + Hence, ParticipantType(1) is ParticipantType.BSMD, + and ParticipantType(7) is [ParticipantType.BSMD, ParticipantType.TERMINAL, ParticipantType.PILOT] + + both would return 'True' + + returns: boolean. Whether the participant id is a .BSMD type element + """ + # use the decoded JWT token and extract the participant type + participant_type = get_participant_type_from_user_data(user_data) + + # boolean check: is the participant of type .BSMD? + is_bsmd = ParticipantType.BSMD in participant_type + return is_bsmd + def check_if_ship_id_is_valid(ship_id): """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""" @@ -122,7 +143,8 @@ def check_if_shipcall_id_is_valid(shipcall_id:int): 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): +import typing +def check_if_participant_id_is_valid_standalone(participant_id:int, participant_type:typing.Optional[ParticipantType]): """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 @@ -132,7 +154,23 @@ def check_if_participant_id_is_valid_standalone(participant_id:int): # boolean check participant_id_is_valid = participant_id in list(participants.keys()) - return participant_id_is_valid + + if participant_type is not None: + if participant_id not in list(participants.keys()): + raise ValidationError(f"the provided participant_id {participant_id} does not exist in the database.") + + # IntFlag object + participant_type_in_db = ParticipantType(int(participants.get(participant_id).get("type", ParticipantType.undefined))) + assert isinstance(participant_type_in_db, ParticipantType), f"{type(participant_type_in_db)}" + + # IntFlag comparison. A user may be assigned as a pilot, but the participant may be multiple roles + participant_type_matches_db = (participant_type in participant_type_in_db) + + participant_is_valid = (participant_id_is_valid and participant_type_matches_db) + return participant_is_valid + else: + # when the participant_type is not provided, only evaluate the ID + return participant_id_is_valid def check_if_participant_id_is_valid(participant:dict): """ @@ -144,7 +182,8 @@ 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) - participant_id_is_valid = check_if_participant_id_is_valid_standalone(participant_id) + participant_type = ParticipantType(int(participant.get("type", ParticipantType.undefined))) + participant_id_is_valid = check_if_participant_id_is_valid_standalone(participant_id, participant_type) return participant_id_is_valid def check_if_participant_ids_are_valid(participants:list[dict]): diff --git a/src/server/tests/database/test_sql_queries.py b/src/server/tests/database/test_sql_queries.py index 88e546d..fc4267a 100644 --- a/src/server/tests/database/test_sql_queries.py +++ b/src/server/tests/database/test_sql_queries.py @@ -9,7 +9,7 @@ from BreCal.database.sql_queries import SQLQuery from BreCal.schemas import model from BreCal.stubs.user import get_user_simple -instance_path = os.path.join(os.path.expanduser('~'), "brecal", "src", "server", "instance", "instance") +instance_path = os.path.join(os.path.expanduser('~'), "brecal", "src", "server", "instance") local_db.initPool(os.path.dirname(instance_path), connection_filename="connection_data_local.json") def test_sql_query_every_call_returns_str(): @@ -39,10 +39,10 @@ def test_sql_get_notifications(): import mysql.connector # unfortunately, there currently is *no* notification in the database. - with pytest.raises(mysql.connector.errors.ProgrammingError, match="Unknown column 'shipcall_id' in 'field list'"): - options = {"shipcall_id":417} - notifications = execute_sql_query_standalone(query=SQLQuery.get_notifications(), param={"scid" : options["shipcall_id"]}, model=model.Notification.from_query_row) - assert all([isinstance(notification,model.Notification) for notification in notifications]) + options = {"shipcall_id":85} + notifications = execute_sql_query_standalone(query=SQLQuery.get_notifications(), param={"scid" : options["shipcall_id"]}, model=model.Notification.from_query_row) + assert all([isinstance(notification,model.Notification) for notification in notifications]) + assert all([isinstance(notification.type,model.NotificationType) for notification in notifications]) return def test_sql_get_participants(): @@ -458,7 +458,7 @@ def test_sql__shipcall_post__get_last_insert_id__get_spm__update_participants__v ### proxy data ### # loop across passed participant ids, creating entries for those not present in pdata - schemaModel = {'id': new_id, "participants":[{'id': 128, 'participant_id': 2, 'type': 4}, {'id': 129, 'participant_id': 3, 'type': 1}, {'id': 130, 'participant_id': 4, 'type': 2}, {'id': 131, 'participant_id': 6, 'type': 8}]} + schemaModel = {'id': new_id, "participants":[{'id': 128, 'participant_id': 2, 'type': 4}, {'id': 129, 'participant_id': 3, 'type': 1}, {'id': 130, 'participant_id': 4, 'type': 2}, {'id': 131, 'participant_id': 6, 'type': 8}, {'id': 132, 'participant_id': 136, 'type': 16}]} # 4.) assign the participants for participant_assignment in schemaModel["participants"]: diff --git a/src/server/tests/validators/test_input_validation_ship.py b/src/server/tests/validators/test_input_validation_ship.py index b1de16e..f67e02c 100644 --- a/src/server/tests/validators/test_input_validation_ship.py +++ b/src/server/tests/validators/test_input_validation_ship.py @@ -19,8 +19,10 @@ from BreCal.stubs.ship import get_stub_valid_ship, get_stub_valid_ship_loaded_mo from BreCal.validators.input_validation import validation_error_default_asserts from BreCal.schemas.model import ParticipantType from BreCal.validators.input_validation_ship import InputValidationShip +from BreCal.database.sql_handler import execute_sql_query_standalone +from BreCal.database.sql_queries import SQLQuery -instance_path = os.path.join(os.path.expanduser('~'), "brecal", "src", "server", "instance", "instance") +instance_path = os.path.join(os.path.expanduser('~'), "brecal", "src", "server", "instance") local_db.initPool(os.path.dirname(instance_path), connection_filename="connection_data_local.json") @pytest.fixture(scope="session") @@ -120,11 +122,6 @@ def test_input_validation_ship_fails_when_callsign_is_incorrect(): 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 @@ -139,6 +136,11 @@ def test_input_validation_ship_fails_when_imo_is_incorrect(): post_data = get_stub_valid_ship() post_data["imo"] = 1234567 loadedModel = model.ShipSchema().load(data=post_data, many=False, partial=True) + + # success: when there are less than 7-digits, the backend applies trailing zeros + post_data = get_stub_valid_ship() + post_data["imo"] = 123456 + 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(): @@ -224,3 +226,42 @@ def test_input_validation_ship_put_request_fails_when_ship_id_is_missing(): with pytest.raises(ValidationError, match="The id field is required."): InputValidationShip.content_contains_ship_id(content) return + +def test_input_validation_ship_post_failure_case_20240802(): + """Description: https://trello.com/c/DmwLnfbN/260-shipcall-anlegen-bad-format""" + + post_data = { + "name": "Testschiff 02", + "imo": 1, #0000000001, + "length": 100.2, + "width": 16.5, + "is_tug": 0, + "bollard_pull": 42, + "callsign": "9992", + "participant_id": None, + "eni": 1 + } + + content = post_data + loadedModel = model.ShipSchema().load(data=post_data, many=False, partial=True) + + # Fails: not BSMD user + with pytest.raises(ValidationError, match="current user does not belong to BSMD. Cannot post, put or delete ships. Found user data"): + user = execute_sql_query_standalone(query=SQLQuery.get_user_by_id(), param={"id":9}, command_type="single", model=model.User) + user_data = user.__dict__ + assert user.participant_id == 4 + + InputValidationShip.evaluate_post_data(user_data, loadedModel, content) + + # Fails: bollard_pull is set, but ship is not a tug + with pytest.raises(ValidationError, match="'bollard_pull' is only allowed, when a ship is a tug"): + user = execute_sql_query_standalone(query=SQLQuery.get_user_by_id(), param={"id":5}, command_type="single", model=model.User) + user_data = user.__dict__ + assert user.participant_id == 3 + + InputValidationShip.evaluate_post_data(user_data, loadedModel, content) + + # Success + post_data["bollard_pull"] = None + InputValidationShip.evaluate_post_data(user_data, loadedModel, content) + return diff --git a/src/server/tests/validators/test_input_validation_shipcall.py b/src/server/tests/validators/test_input_validation_shipcall.py index 925466c..de0cc8b 100644 --- a/src/server/tests/validators/test_input_validation_shipcall.py +++ b/src/server/tests/validators/test_input_validation_shipcall.py @@ -14,10 +14,13 @@ from BreCal.schemas.model import Participant_Assignment, EvaluationType, Shipcal from BreCal.stubs.shipcall import create_postman_stub_shipcall, get_stub_valid_shipcall_arrival, get_stub_valid_shipcall_departure, get_stub_valid_shipcall_shifting, get_stub_shipcall_arrival_invalid_missing_eta, get_stub_shipcall_shifting_invalid_missing_eta, get_stub_shipcall_shifting_invalid_missing_etd, get_stub_shipcall_arrival_invalid_missing_type, get_stub_shipcall_departure_invalid_missing_etd from BreCal.stubs.participant import get_stub_list_of_valid_participants from BreCal.validators.input_validation import validation_error_default_asserts -from BreCal.schemas.model import ParticipantType +from BreCal.schemas.model import ParticipantType, ShipcallParticipantMap from BreCal.validators.input_validation_shipcall import InputValidationShipcall +from BreCal.database.sql_handler import execute_sql_query_standalone +from BreCal.database.sql_queries import SQLQuery +from BreCal.schemas import model -instance_path = os.path.join(os.path.expanduser('~'), "brecal", "src", "server", "instance", "instance") +instance_path = os.path.join(os.path.expanduser('~'), "brecal", "src", "server", "instance") local_db.initPool(os.path.dirname(instance_path), connection_filename="connection_data_local.json") @pytest.fixture(scope="session") @@ -105,12 +108,13 @@ def test_shipcall_post_request_fails_when_participant_ids_are_invalid(get_stub_t post_data = get_stub_valid_shipcall_arrival() # create_postman_stub_shipcall() - post_data["participants"] = [Participant_Assignment(1234562,4).to_json()] # identical to: [{'participant_id': 1234562, 'type': 4}] + participant_id = 1234562 + post_data["participants"] = [Participant_Assignment(participant_id,4).to_json()] # identical to: [{'participant_id': 1234562, 'type': 4}] response = requests.post( f"{url}/shipcalls", headers={"Content-Type":"text", "Authorization":f"Bearer {token}"}, json=post_data ) - with pytest.raises(ValidationError, match=f"one of the provided participant ids is invalid"): + with pytest.raises(ValidationError, match=f"the provided participant_id {participant_id} does not exist in the database"): assert response.status_code==400 raise ValidationError(response.json()) # because the response does not raise a ValidationError, we artifically create it to check the pytest.raises outcome return @@ -220,14 +224,14 @@ def test_shipcall_post_request_fails_when_type_arrival_and_not_in_future(get_stu # accept post_data = original_post_data.copy() - post_data["type"] = ShipcallType.arrival + post_data["type"] = ShipcallType.arrival.name post_data["eta"] = (datetime.datetime.now() + datetime.timedelta(hours=3)).isoformat() response = requests.post(f"{url}/shipcalls", headers={"Content-Type":"text", "Authorization":f"Bearer {token}"}, json=post_data) assert response.status_code == 201 # error post_data = original_post_data.copy() - post_data["type"] = ShipcallType.arrival + post_data["type"] = ShipcallType.arrival.name post_data["eta"] = (datetime.datetime.now() - datetime.timedelta(hours=3)).isoformat() response = requests.post(f"{url}/shipcalls", headers={"Content-Type":"text", "Authorization":f"Bearer {token}"}, json=post_data) @@ -243,14 +247,14 @@ def test_shipcall_post_request_fails_when_type_departure_and_not_in_future(get_s # accept post_data = original_post_data.copy() - post_data["type"] = ShipcallType.departure + post_data["type"] = ShipcallType.departure.name post_data["etd"] = (datetime.datetime.now() + datetime.timedelta(hours=3)).isoformat() response = requests.post(f"{url}/shipcalls", headers={"Content-Type":"text", "Authorization":f"Bearer {token}"}, json=post_data) assert response.status_code == 201 # error post_data = original_post_data.copy() - post_data["type"] = ShipcallType.departure + post_data["type"] = ShipcallType.departure.name post_data["etd"] = (datetime.datetime.now() - datetime.timedelta(hours=3)).isoformat() response = requests.post(f"{url}/shipcalls", headers={"Content-Type":"text", "Authorization":f"Bearer {token}"}, json=post_data) @@ -262,21 +266,21 @@ def test_shipcall_post_request_fails_when_type_departure_and_not_in_future(get_s def test_shipcall_post_request_fails_when_type_shifting_and_not_in_future(get_stub_token): url, token = get_stub_token["url"], get_stub_token["token"] - original_post_data = get_stub_valid_shipcall_departure() # create_postman_stub_shipcall() + original_post_data = get_stub_valid_shipcall_shifting() # create_postman_stub_shipcall() # accept post_data = original_post_data.copy() - post_data["type"] = ShipcallType.departure + post_data["type"] = ShipcallType.shifting.name post_data["eta"] = (datetime.datetime.now() + datetime.timedelta(hours=3)).isoformat() - post_data["etd"] = (datetime.datetime.now() + datetime.timedelta(hours=3)).isoformat() + post_data["etd"] = (datetime.datetime.now() + datetime.timedelta(hours=3,minutes=1)).isoformat() response = requests.post(f"{url}/shipcalls", headers={"Content-Type":"text", "Authorization":f"Bearer {token}"}, json=post_data) assert response.status_code == 201 # error post_data = original_post_data.copy() - post_data["type"] = ShipcallType.departure + post_data["type"] = ShipcallType.shifting.name post_data["eta"] = (datetime.datetime.now() + datetime.timedelta(hours=3)).isoformat() - post_data["etd"] = (datetime.datetime.now() - datetime.timedelta(hours=3)).isoformat() + post_data["etd"] = (datetime.datetime.now() - datetime.timedelta(hours=3,minutes=1)).isoformat() response = requests.post(f"{url}/shipcalls", headers={"Content-Type":"text", "Authorization":f"Bearer {token}"}, json=post_data) with pytest.raises(ValidationError, match="must be in the future. Incorrect datetime provided"): @@ -291,7 +295,7 @@ def test_shipcall_post_request_fails_when_type_arrival_and_missing_eta(get_stub_ post_data = original_post_data.copy() post_data.pop("eta", None) - post_data["type"] = ShipcallType.arrival + post_data["type"] = ShipcallType.arrival.name response = requests.post(f"{url}/shipcalls", headers={"Content-Type":"text", "Authorization":f"Bearer {token}"}, json=post_data) with pytest.raises(ValidationError, match="Missing key!"): @@ -306,7 +310,7 @@ def test_shipcall_post_request_fails_when_type_departure_and_missing_etd(get_stu post_data = original_post_data.copy() post_data.pop("etd", None) - post_data["type"] = ShipcallType.departure + post_data["type"] = ShipcallType.departure.name response = requests.post(f"{url}/shipcalls", headers={"Content-Type":"text", "Authorization":f"Bearer {token}"}, json=post_data) with pytest.raises(ValidationError, match="Missing key!"): @@ -320,7 +324,7 @@ def test_shipcall_post_request_fails_when_type_shifting_and_missing_eta(get_stub original_post_data = get_stub_valid_shipcall_arrival() # create_postman_stub_shipcall() post_data = original_post_data.copy() - post_data["type"] = ShipcallType.departure + post_data["type"] = ShipcallType.departure.name post_data.pop("eta", None) post_data["etd"] = (datetime.datetime.now() + datetime.timedelta(hours=3)).isoformat() response = requests.post(f"{url}/shipcalls", headers={"Content-Type":"text", "Authorization":f"Bearer {token}"}, json=post_data) @@ -336,7 +340,7 @@ def test_shipcall_post_request_fails_when_type_shifting_and_missing_etd(get_stub original_post_data = get_stub_valid_shipcall_arrival() # create_postman_stub_shipcall() post_data = original_post_data.copy() - post_data["type"] = ShipcallType.departure + post_data["type"] = ShipcallType.departure.name post_data["eta"] = (datetime.datetime.now() + datetime.timedelta(hours=3)).isoformat() post_data.pop("etd", None) response = requests.post(f"{url}/shipcalls", headers={"Content-Type":"text", "Authorization":f"Bearer {token}"}, json=post_data) @@ -589,15 +593,15 @@ def test_shipcall_post_type_is_wrong(get_stub_token): post_data = get_stub_valid_shipcall_arrival() # type 1 should be successful (201) - post_data["type"] = 1 + post_data["type"] = ShipcallType.arrival.name # "arrival" - response = requests.post( + response = requests.post('' f"{url}/shipcalls", headers={"Content-Type":"text", "Authorization":f"Bearer {token}"}, json=post_data ) assert response.status_code == 201 # type 51 should not be successful (400 BAD REQUEST) - post_data["type"] = 51 + post_data["type"] = "area51" response = requests.post( f"{url}/shipcalls", headers={"Content-Type":"text", "Authorization":f"Bearer {token}"}, json=post_data @@ -614,16 +618,25 @@ def test_shipcall_put_request_fails_when_different_participant_id_is_assigned(ge user_data = {'id':6, 'participant_id':1} loadedModel = post_data content = post_data - spm_shipcall_data = [{'participant_id': 6, 'type': 4}, - {'participant_id': 3, 'type': 1}, - {'participant_id': 4, 'type': 2}, - {'participant_id': 5, 'type': 8}] + created = datetime.datetime.now()+datetime.timedelta(minutes=1) + modified = datetime.datetime.now()+datetime.timedelta(minutes=2) + + spm_shipcall_data = [{"id":99112, 'participant_id': 6, 'type': 4}, + {"id":99113, 'participant_id': 3, 'type': 1}, + {"id":99114, 'participant_id': 4, 'type': 2}, + {"id":99115, 'participant_id': 5, 'type': 8}] + spm_shipcall_data = [ + {**{"created":created, "modified":modified, "shipcall_id":shipcall_id}, **spm} + for spm in + spm_shipcall_data + ] + spm_shipcall_data = [ShipcallParticipantMap(**spm) for spm in spm_shipcall_data] # agency with different participant id is assigned ivs = InputValidationShipcall() - with pytest.raises(werkzeug.exceptions.Forbidden, match=f"A different participant_id is assigned as the AGENCY of this shipcall. "): - ivs.check_agency_in_shipcall_participant_map(user_data, loadedModel, content, spm_shipcall_data) + with pytest.raises(werkzeug.exceptions.Forbidden, match=f"PUT Requests for shipcalls can only be issued by an assigned AGENCY or BSMD users"): + ivs.check_user_is_authorized_for_put_request(user_data, loadedModel, content, spm_shipcall_data) return @@ -637,15 +650,30 @@ def test_shipcall_put_request_success(get_shipcall_id_after_stub_post_request): user_data = {'id':6, 'participant_id':1} loadedModel = post_data content = post_data - spm_shipcall_data = [{'participant_id': 6, 'type': 8}, - {'participant_id': 3, 'type': 1}, - {'participant_id': 4, 'type': 2}, - {'participant_id': 5, 'type': 4}] + + created = datetime.datetime.now()+datetime.timedelta(minutes=1) + modified = datetime.datetime.now()+datetime.timedelta(minutes=2) + + spm_shipcall_data = [{"id":99112, 'participant_id': 6, 'type': 8}, + {"id":99113, 'participant_id': 3, 'type': 1}, + {"id":99114, 'participant_id': 4, 'type': 2}, + {"id":99115, 'participant_id': 5, 'type': 8}] + spm_shipcall_data = [ + {**{"created":created, "modified":modified, "shipcall_id":shipcall_id}, **spm} + for spm in + spm_shipcall_data + ] + spm_shipcall_data = [ShipcallParticipantMap(**spm) for spm in spm_shipcall_data] # success ivs = InputValidationShipcall() - ivs.check_agency_in_shipcall_participant_map(user_data, loadedModel, content, spm_shipcall_data) + with pytest.raises(Exception, match="deprecated"): + ivs.check_agency_in_shipcall_participant_map(user_data, loadedModel, content, spm_shipcall_data) + + # failure: the user is BSMD and the agency (participant_id 6) has set the BSMD flag, but there is more than one agency + with pytest.raises(ValidationError, match="Found more than one assigned agency for the shipcall with ID"): + ivs.check_user_is_authorized_for_put_request(user_data, loadedModel, content, spm_shipcall_data) return def test_shipcall_put_request_fails_when_no_agency_is_assigned(get_shipcall_id_after_stub_post_request): @@ -657,16 +685,26 @@ def test_shipcall_put_request_fails_when_no_agency_is_assigned(get_shipcall_id_a user_data = {'id':6, 'participant_id':1} loadedModel = post_data content = post_data + + created = datetime.datetime.now()+datetime.timedelta(minutes=1) + modified = datetime.datetime.now()+datetime.timedelta(minutes=2) + spm_shipcall_data = [ - {'participant_id': 3, 'type': 1}, - {'participant_id': 4, 'type': 2}, - {'participant_id': 5, 'type': 4}] + {"id":99113, 'participant_id': 3, 'type': 1}, + {"id":99114, 'participant_id': 4, 'type': 2}, + {"id":99115, 'participant_id': 5, 'type': 4}] + spm_shipcall_data = [ + {**{"created":created, "modified":modified, "shipcall_id":shipcall_id}, **spm} + for spm in + spm_shipcall_data + ] + spm_shipcall_data = [ShipcallParticipantMap(**spm) for spm in spm_shipcall_data] # no agency assigned ivs = InputValidationShipcall() - with pytest.raises(werkzeug.exceptions.Forbidden, match=f"There is no assigned agency for this shipcall."): - ivs.check_agency_in_shipcall_participant_map(user_data, loadedModel, content, spm_shipcall_data) + with pytest.raises(ValidationError, match=f"There is no assigned agency for the shipcall with ID"): + ivs.check_user_is_authorized_for_put_request(user_data, loadedModel, content, spm_shipcall_data) return def test_shipcall_put_request_fails_when_user_is_not_authorized(get_shipcall_id_after_stub_post_request): @@ -679,17 +717,26 @@ def test_shipcall_put_request_fails_when_user_is_not_authorized(get_shipcall_id_ user_data = {'id':1, 'participant_id':2} loadedModel = post_data content = post_data + + created = datetime.datetime.now()+datetime.timedelta(minutes=1) + modified = datetime.datetime.now()+datetime.timedelta(minutes=2) + + spm_shipcall_data = [{"id":99112, 'participant_id': 2, 'type': 4}, + {"id":99113, 'participant_id': 3, 'type': 1}, + {"id":99114, 'participant_id': 4, 'type': 8}, + {"id":99115, 'participant_id': 5, 'type': 4}] spm_shipcall_data = [ - {'participant_id': 2, 'type': 8}, - {'participant_id': 3, 'type': 1}, - {'participant_id': 4, 'type': 2}, - {'participant_id': 5, 'type': 4}] + {**{"created":created, "modified":modified, "shipcall_id":shipcall_id}, **spm} + for spm in + spm_shipcall_data + ] + spm_shipcall_data = [ShipcallParticipantMap(**spm) for spm in spm_shipcall_data] # current user is not authorized ivs = InputValidationShipcall() - with pytest.raises(werkzeug.exceptions.Forbidden, match=f"PUT Requests for shipcalls can only be issued by AGENCY or BSMD users."): - ivs.check_agency_in_shipcall_participant_map(user_data, loadedModel, content, spm_shipcall_data) + with pytest.raises(werkzeug.exceptions.Forbidden, match=f"PUT Requests for shipcalls can only be issued by an assigned AGENCY or BSMD users"): + ivs.check_user_is_authorized_for_put_request(user_data, loadedModel, content, spm_shipcall_data) return def test_shipcall_put_request_fails_when_user_tries_self_assignment(get_shipcall_id_after_stub_post_request): @@ -701,16 +748,28 @@ def test_shipcall_put_request_fails_when_user_tries_self_assignment(get_shipcall user_data = {'id':1, 'participant_id':6} loadedModel = post_data content = post_data - spm_shipcall_data = [{'participant_id': 6, 'type': 8}, - {'participant_id': 3, 'type': 1}, - {'participant_id': 4, 'type': 2}, - {'participant_id': 5, 'type': 4}] + + created = datetime.datetime.now()+datetime.timedelta(minutes=1) + modified = datetime.datetime.now()+datetime.timedelta(minutes=2) + + spm_shipcall_data = [ + {"id":99113, 'participant_id': 3, 'type': 1}, + {"id":99114, 'participant_id': 4, 'type': 2}, + {"id":99115, 'participant_id': 5, 'type': 4}] + spm_shipcall_data = [ + {**{"created":created, "modified":modified, "shipcall_id":shipcall_id}, **spm} + for spm in + spm_shipcall_data + ] + spm_shipcall_data = [ShipcallParticipantMap(**spm) for spm in spm_shipcall_data] # self-assignment. User is participant 6, and wants to assign participant 6. ivs = InputValidationShipcall() - with pytest.raises(werkzeug.exceptions.Forbidden, match=f"An agency cannot self-register for a shipcall. The request is issued by an agency-user and tries to assign an AGENCY as the participant of the shipcall."): - ivs.check_agency_in_shipcall_participant_map(user_data, loadedModel, content, spm_shipcall_data) + with pytest.raises(ValidationError, match=f"There is no assigned agency for the shipcall with ID"): + # previous error message: An agency cannot self-register for a shipcall. The request is issued by an agency-user and tries to assign an AGENCY as the participant of the shipcall."" + # however, self-assignment is no longer possible, because the SPM is verified beforehand. + ivs.check_user_is_authorized_for_put_request(user_data, loadedModel, content, spm_shipcall_data) return def test_shipcall_put_request_fails_input_validation_shipcall_when_shipcall_is_canceled(get_stub_token): @@ -735,3 +794,64 @@ def test_shipcall_put_request_fails_input_validation_shipcall_when_shipcall_is_c with pytest.raises(ValidationError, match="The shipcall with id 'shipcall_id' is canceled. A canceled shipcall may not be changed."): InputValidationShipcall.check_shipcall_is_canceled(loadedModel, content) return + +def test_shipcall_put_request_works_if_most_values_are_null(): + """This pytest verifies, that a PUT-request for shipcalls works, even if only a single value is to be modified""" + + user = execute_sql_query_standalone(query=SQLQuery.get_user_by_id(), param={"id":10}, command_type="single", model=model.User) + user_data = user.__dict__ + assert user.participant_id == 5 + + shipcall = execute_sql_query_standalone(query=SQLQuery.get_shipcall_by_id(), param={"id":152}, command_type="single", model=model.Shipcall) + assert shipcall.id == 152 + + put_data = {"id":shipcall.id, "arrival_berth_id":142} + loadedModel = content = put_data + + InputValidationShipcall.evaluate_put_data(user_data, loadedModel, content) + return + + + + +def test_shipcall_put_request_fails_input_validation_shipcall_when_shipcall_is_canceled(get_stub_token): + url, token = get_stub_token["url"], get_stub_token["token"] + + # get all shipcalls and grab shipcall with ID 4 + # #TODO: there must be a better way to accomplish this easily... + response = requests.get(f"{url}/shipcalls", headers={"Content-Type":"text", "Authorization":f"Bearer {token}"}, params={"past_days":30000}) + assert response.status_code==200 + assert isinstance(response.json(), list) + shipcalls = response.json() + + shipcall_id = 152 + sh4 = [sh for sh in shipcalls if sh.get("id")==shipcall_id][0] + put_data = {k:v for k,v in sh4.items() if k in ["eta", "etd", "type", "ship_id", "arrival_berth_id", "participants"]} + put_data["id"] = shipcall_id + + loadedModel = put_data + content = put_data + + # eta & etd must be in the future, as the request fails otherwise. The shipcall is 'shifting', so both must be provided. + loadedModel["eta"] = datetime.datetime.now()+datetime.timedelta(minutes=1) + loadedModel["etd"] = datetime.datetime.now()+datetime.timedelta(minutes=2) + + ### FAILS: + # user 9 (participant id 4) is *not* assigned to the shipcall + user = execute_sql_query_standalone(query=SQLQuery.get_user_by_id(), param={"id":9}, command_type="single", model=model.User) + user_data = user.__dict__ + assert user.participant_id == 4 + + #### verification should fail, because participant_id 4 is ParticipantType.PILOT (neither an assigned agency, nor bsmd) + with pytest.raises(werkzeug.exceptions.Forbidden, match="PUT Requests for shipcalls can only be issued by an assigned AGENCY or BSMD user"): + InputValidationShipcall.evaluate_put_data(user_data, loadedModel, content) + + ### PASSES: + # user 10 (participant id 5) is assigned to the shipcall + user = execute_sql_query_standalone(query=SQLQuery.get_user_by_id(), param={"id":10}, command_type="single", model=model.User) + user_data = user.__dict__ + assert user.participant_id == 5 + + ### verification should pass + InputValidationShipcall.evaluate_put_data(user_data, loadedModel, content) + return diff --git a/src/server/tests/validators/test_input_validation_times.py b/src/server/tests/validators/test_input_validation_times.py index 98a2689..240e7c3 100644 --- a/src/server/tests/validators/test_input_validation_times.py +++ b/src/server/tests/validators/test_input_validation_times.py @@ -13,7 +13,7 @@ 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") +instance_path = os.path.join(os.path.expanduser('~'), "brecal", "src", "server", "instance") local_db.initPool(os.path.dirname(instance_path), connection_filename="connection_data_local.json") @@ -122,7 +122,7 @@ def test_input_validation_times_fails_when_participant_type_deviates_from_shipca # 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."): + with pytest.raises(ValidationError, match="is not assigned to the shipcall"): 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 @@ -295,39 +295,44 @@ def test_input_validation_times_fails_when_participant_type_is_not_assigned__or_ 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} + shipcall_id 234 is assigned to the participants + DELETE# {"participant_id": 136, "type":2} and {"participant_id": 136, "type":8} + {"participant_id": 2, "type":4} + {"participant_id": 3, "type":1} + {"participant_id": 4, "type":2} + {"participant_id": 5, "type":8} Case 1: - When user_id 3 should be set as participant_type 4, the call fails, because type 4 is not assigned + When user_id 27 should be set as participant_type 16, the call fails, because type 16 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 + When user_id 2 (participant_id 1) should be set as participant_type 2, the call fails even though type 2 exists, + because participant_id 4 is assigned Case 3: - When user_id 28 (participant_id 136) is set as participant_type 2, the call passes. + When user_id 9 (participant_id 4) 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 + # fails: participant type 16 does not exist + user_data, loadedModel, content = get_valid_stub_for_pytests(user_id=27) + participant_type = 16 + loadedModel["shipcall_id"] = content["shipcall_id"] = 234 + loadedModel["participant_id"] = content["participant_id"] = 16 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:"): + with pytest.raises(ValidationError, match=f"Could not find a matching time dataset for the provided participant_type: {participant_type} at shipcall with id"): 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) + # fails: participant type 2 exists, but user_id 2 is part of the wrong participant_id group user_data, loadedModel, content = get_valid_stub_for_pytests(user_id=2) - loadedModel["shipcall_id"] = content["shipcall_id"] = 222 + loadedModel["shipcall_id"] = content["shipcall_id"] = 234 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 + # pass: participant type 2 exists & user_id is part of participant_id group 4, which is correct + user_data, loadedModel, content = get_valid_stub_for_pytests(user_id=9) + loadedModel["shipcall_id"] = content["shipcall_id"] = 234 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) @@ -372,7 +377,8 @@ def test_input_validation_times_delete_request_fails_when_times_id_does_not_exis # 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) + pdata = [{'participant_id': 136, 'participant_type': 8, 'shipcall_id': 154}] + InputValidationTimes.check_user_belongs_to_same_group_as_dataset_determines(user_data, loadedModel=None, times_id=times_id, pdata=pdata) # fails: times_id does not exist user_data, loadedModel, content = get_valid_stub_for_pytests(user_id=28) @@ -389,10 +395,15 @@ def test_input_validation_times_delete_request_fails_when_user_belongs_to_wrong_ 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 + # success: the participant_id within the times entry is 136. user_id=28 belongs to participant_id=136, so it matches 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) + + # success: creates an artificial SPM, where the participant_id is '136', so it matches the user_id's (28) participant_id (136) + user_data, loadedModel, content = get_valid_stub_for_pytests(user_id=28) + times_id = 392 + pdata = [{'participant_id': 136, 'participant_type': 8, 'shipcall_id': 154}] + InputValidationTimes.check_user_belongs_to_same_group_as_dataset_determines(user_data, loadedModel=None, times_id=times_id, pdata=pdata) return - diff --git a/src/server/tests/validators/test_input_validation_utils.py b/src/server/tests/validators/test_input_validation_utils.py new file mode 100644 index 0000000..f18c939 --- /dev/null +++ b/src/server/tests/validators/test_input_validation_utils.py @@ -0,0 +1,30 @@ +import pytest + +def test_check_if_participant_id_is_valid_standalone__different_assignments(): + from BreCal.validators.input_validation_utils import check_if_participant_id_is_valid_standalone + from BreCal.schemas.model import ParticipantType + # participant id 10 has the ParticipantType 10. This means, the participant is, both, agency and terminal. + # upon assignment, the participant can take the role of terminal, agency or theoretically, both. + + participant_id = 10 + participant_type = ParticipantType(10) + assert check_if_participant_id_is_valid_standalone(participant_id, participant_type=participant_type) + assert check_if_participant_id_is_valid_standalone(participant_id, participant_type=ParticipantType(2)) + assert check_if_participant_id_is_valid_standalone(participant_id, participant_type=ParticipantType(8)) + + # failure cases: BSMD, PILOT, MOORING, PORT_ADMINISTRATION, TUG + with pytest.raises(AssertionError, match="wrong role assignment."): + assert check_if_participant_id_is_valid_standalone(participant_id, participant_type=ParticipantType(1)), f"wrong role assignment." + + with pytest.raises(AssertionError, match="wrong role assignment."): + assert check_if_participant_id_is_valid_standalone(participant_id, participant_type=ParticipantType(4)), f"wrong role assignment." + + with pytest.raises(AssertionError, match="wrong role assignment."): + assert check_if_participant_id_is_valid_standalone(participant_id, participant_type=ParticipantType(16)), f"wrong role assignment." + + with pytest.raises(AssertionError, match="wrong role assignment."): + assert check_if_participant_id_is_valid_standalone(participant_id, participant_type=ParticipantType(32)), f"wrong role assignment." + + with pytest.raises(AssertionError, match="wrong role assignment."): + assert check_if_participant_id_is_valid_standalone(participant_id, participant_type=ParticipantType(64)), f"wrong role assignment." + return