dschick/git_brcal
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

corrected open issues of the API validation functions. Made some validation errors more verbose, improved robustness, refactored some of the methods and adapted many unit tests to the novel format.

Browse Source
This commit is contained in:
Max Metz 2024-08-06 20:46:40 +02:00
parent 18719f15c1
commit d54fed9160
11 changed files with 504 additions and 142 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
src/server/BreCal/schemas/model.py
View File

@ -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

10
src/server/BreCal/stubs/shipcall.py
View File

@ -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(),

2
src/server/BreCal/validators/input_validation_ship.py
View File

@ -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:

200
src/server/BreCal/validators/input_validation_shipcall.py
View File

@ -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
"""
# 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

22
src/server/BreCal/validators/input_validation_times.py
View File

@ -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)

45
src/server/BreCal/validators/input_validation_utils.py
View File

@ -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]):

12
src/server/tests/database/test_sql_queries.py
View File

@ -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"]:

53
src/server/tests/validators/test_input_validation_ship.py
View File

@ -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

216
src/server/tests/validators/test_input_validation_shipcall.py
View File

@ -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

53
src/server/tests/validators/test_input_validation_times.py
View File

@ -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

30
src/server/tests/validators/test_input_validation_utils.py Normal file
View File

@ -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