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

Allow new shipcalls up to 1 day in the past for serverside validation

Browse Source
This commit is contained in:
Daniel Schick 2024-11-06 16:22:34 +01:00
parent b5a8a3d31c
commit d6e3ae20c1
3 changed files with 77 additions and 76 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

88
src/server/BreCal/validators/input_validation.py
View File

@ -28,7 +28,7 @@ def validation_error_default_asserts(response):
def validate_posted_shipcall_data(user_data:dict, loadedModel:dict, content:dict): def validate_posted_shipcall_data(user_data:dict, loadedModel:dict, content:dict):
"""this function applies more complex validation functions to data, which is sent to a post-request of shipcalls""" """this function applies more complex validation functions to data, which is sent to a post-request of shipcalls"""
# DEPRECATED: this function has been refactored into InputValidationShipcall (see methods for POST and PUT evaluation) # DEPRECATED: this function has been refactored into InputValidationShipcall (see methods for POST and PUT evaluation)
# #TODO_refactor: this function is pretty complex. One may instead build an object, which calls the methods separately. # #TODO_refactor: this function is pretty complex. One may instead build an object, which calls the methods separately.
##### Section 1: check user_data ##### ##### Section 1: check user_data #####
# check, whether the user belongs to a participant, which is of type ParticipantType.BSMD # check, whether the user belongs to a participant, which is of type ParticipantType.BSMD
@ -36,23 +36,23 @@ def validate_posted_shipcall_data(user_data:dict, loadedModel:dict, content:dict
is_bsmd = check_if_user_is_bsmd_type(user_data) is_bsmd = check_if_user_is_bsmd_type(user_data)
if not is_bsmd: if not is_bsmd:
raise ValidationError({"user_participant_type":f"current user does not belong to BSMD. Cannot post shipcalls. Found user data: {user_data}"}) raise ValidationError({"user_participant_type":f"current user does not belong to BSMD. Cannot post shipcalls. Found user data: {user_data}"})
##### Section 2: check loadedModel ##### ##### Section 2: check loadedModel #####
valid_ship_id = check_if_ship_id_is_valid(ship_id=loadedModel.get("ship_id", None)) valid_ship_id = check_if_ship_id_is_valid(ship_id=loadedModel.get("ship_id", None))
if not valid_ship_id: if not valid_ship_id:
raise ValidationError({"ship_id":f"provided an invalid ship id, which is not found in the database: {loadedModel.get('ship_id', None)}"}) raise ValidationError({"ship_id":f"provided an invalid ship id, which is not found in the database: {loadedModel.get('ship_id', None)}"})
valid_arrival_berth_id = check_if_berth_id_is_valid(berth_id=loadedModel.get("arrival_berth_id", None)) valid_arrival_berth_id = check_if_berth_id_is_valid(berth_id=loadedModel.get("arrival_berth_id", None))
if not valid_arrival_berth_id: if not valid_arrival_berth_id:
raise ValidationError({"arrival_berth_id":f"provided an invalid arrival berth id, which is not found in the database: {loadedModel.get('arrival_berth_id', None)}"}) raise ValidationError({"arrival_berth_id":f"provided an invalid arrival berth id, which is not found in the database: {loadedModel.get('arrival_berth_id', None)}"})
valid_departure_berth_id = check_if_berth_id_is_valid(berth_id=loadedModel.get("departure_berth_id", None)) valid_departure_berth_id = check_if_berth_id_is_valid(berth_id=loadedModel.get("departure_berth_id", None))
if not valid_departure_berth_id: if not valid_departure_berth_id:
raise ValidationError({"departure_berth_id":f"provided an invalid departure berth id, which is not found in the database: {loadedModel.get('departure_berth_id', None)}"}) raise ValidationError({"departure_berth_id":f"provided an invalid departure berth id, which is not found in the database: {loadedModel.get('departure_berth_id', None)}"})
valid_participant_ids = check_if_participant_ids_are_valid(participants=loadedModel.get("participants",[])) valid_participant_ids = check_if_participant_ids_are_valid(participants=loadedModel.get("participants",[]))
if not valid_participant_ids: if not valid_participant_ids:
raise ValidationError({"participants":f"one of the provided participant ids is invalid. Could not find one of these in the database: {loadedModel.get('participants', None)}"}) raise ValidationError({"participants":f"one of the provided participant ids is invalid. Could not find one of these in the database: {loadedModel.get('participants', None)}"})
##### Section 3: check content ##### ##### Section 3: check content #####
@ -63,18 +63,18 @@ def validate_posted_shipcall_data(user_data:dict, loadedModel:dict, content:dict
value = content.get(forbidden_key, None) value = content.get(forbidden_key, None)
if value is not None: if value is not None:
raise ValidationError({"forbidden_key":f"'{forbidden_key}' may not be set on POST. Found: {value}"}) raise ValidationError({"forbidden_key":f"'{forbidden_key}' may not be set on POST. Found: {value}"})
voyage_str_is_invalid = check_if_string_has_special_characters(text=content.get("voyage","")) voyage_str_is_invalid = check_if_string_has_special_characters(text=content.get("voyage",""))
if voyage_str_is_invalid: if voyage_str_is_invalid:
raise ValidationError({"voyage":f"there are invalid characters in the 'voyage'-string. Please use only digits and ASCII letters. Allowed: {ascii_letters+digits}. Found: {content.get('voyage')}"}) raise ValidationError({"voyage":f"there are invalid characters in the 'voyage'-string. Please use only digits and ASCII letters. Allowed: {ascii_letters+digits}. Found: {content.get('voyage')}"})
##### Section 4: check loadedModel & content ##### ##### Section 4: check loadedModel & content #####
# #TODO_refactor: these methods should be placed in separate locations # #TODO_refactor: these methods should be placed in separate locations
# existance checks in content # existance checks in content
# datetime checks in loadedModel (datetime.datetime objects). Dates should be in the future. # datetime checks in loadedModel (datetime.datetime objects). Dates should be in the future.
time_now = datetime.datetime.now() time_now = datetime.datetime.now() - datetime.timedelta(days=1)
type_ = loadedModel.get("type", int(ShipcallType.undefined)) type_ = loadedModel.get("type", int(ShipcallType.undefined))
if int(type_)==int(ShipcallType.undefined): if int(type_)==int(ShipcallType.undefined):
raise ValidationError({"type":f"providing 'type' is mandatory. Missing key!"}) raise ValidationError({"type":f"providing 'type' is mandatory. Missing key!"})
@ -85,7 +85,7 @@ def validate_posted_shipcall_data(user_data:dict, loadedModel:dict, content:dict
if content.get("arrival_berth_id", None) is None: if content.get("arrival_berth_id", None) is None:
raise ValidationError({"arrival_berth_id":f"providing 'arrival_berth_id' is mandatory. Missing key!"}) raise ValidationError({"arrival_berth_id":f"providing 'arrival_berth_id' is mandatory. Missing key!"})
if not eta >= time_now: if not eta >= time_now:
raise ValidationError({"eta":f"'eta' must be in the future. Incorrect datetime provided."}) raise ValidationError({"eta":f"'eta' is too far in the past. Incorrect datetime provided."})
elif int(type_)==int(ShipcallType.departure): elif int(type_)==int(ShipcallType.departure):
etd = loadedModel.get("etd") etd = loadedModel.get("etd")
if (content.get("etd", None) is None): if (content.get("etd", None) is None):
@ -93,7 +93,7 @@ def validate_posted_shipcall_data(user_data:dict, loadedModel:dict, content:dict
if content.get("departure_berth_id", None) is None: if content.get("departure_berth_id", None) is None:
raise ValidationError({"departure_berth_id":f"providing 'departure_berth_id' is mandatory. Missing key!"}) raise ValidationError({"departure_berth_id":f"providing 'departure_berth_id' is mandatory. Missing key!"})
if not etd >= time_now: if not etd >= time_now:
raise ValidationError({"etd":f"'etd' must be in the future. Incorrect datetime provided."}) raise ValidationError({"etd":f"'etd' is too far in the past. Incorrect datetime provided."})
elif int(type_)==int(ShipcallType.shifting): elif int(type_)==int(ShipcallType.shifting):
eta = loadedModel.get("eta") eta = loadedModel.get("eta")
etd = loadedModel.get("etd") etd = loadedModel.get("etd")
@ -103,20 +103,20 @@ def validate_posted_shipcall_data(user_data:dict, loadedModel:dict, content:dict
if (content.get("arrival_berth_id", None) is None) or (content.get("departure_berth_id", None) is None): if (content.get("arrival_berth_id", None) is None) or (content.get("departure_berth_id", None) is None):
raise ValidationError({"arrival_berth_id_or_departure_berth_id":f"providing 'arrival_berth_id' & 'departure_berth_id' is mandatory. Missing key!"}) raise ValidationError({"arrival_berth_id_or_departure_berth_id":f"providing 'arrival_berth_id' & 'departure_berth_id' is mandatory. Missing key!"})
if (not eta >= time_now) or (not etd >= time_now) or (not eta >= etd): if (not eta >= time_now) or (not etd >= time_now) or (not eta >= etd):
raise ValidationError({"eta_or_etd":f"'eta' and 'etd' must be in the future. Incorrect datetime provided."}) raise ValidationError({"eta_or_etd":f"'eta' and 'etd' are too far in the past. Incorrect datetime provided."})
tidal_window_from = loadedModel.get("tidal_window_from", None) tidal_window_from = loadedModel.get("tidal_window_from", None)
tidal_window_to = loadedModel.get("tidal_window_to", None) tidal_window_to = loadedModel.get("tidal_window_to", None)
if tidal_window_to is not None: if tidal_window_to is not None:
if not tidal_window_to >= time_now: if not tidal_window_to >= time_now:
raise ValidationError({"tidal_window_to":f"'tidal_window_to' must be in the future. Incorrect datetime provided."}) raise ValidationError({"tidal_window_to":f"'tidal_window_to' is too far in the past. Incorrect datetime provided."})
if tidal_window_from is not None: if tidal_window_from is not None:
if not tidal_window_from >= time_now: if not tidal_window_from >= time_now:
raise ValidationError({"tidal_window_from":f"'tidal_window_from' must be in the future. Incorrect datetime provided."}) raise ValidationError({"tidal_window_from":f"'tidal_window_from' is too far in the past. Incorrect datetime provided."})
# #TODO: assert tidal_window_from > tidal_window_to # #TODO: assert tidal_window_from > tidal_window_to
# #TODO: len of participants > 0, if agency # #TODO: len of participants > 0, if agency
# * assigned participant for agency # * assigned participant for agency
return return
@ -126,21 +126,21 @@ class InputValidation():
def __init__(self): def __init__(self):
self.build_supported_models_dictionary() self.build_supported_models_dictionary()
return return
def build_supported_models_dictionary(self): def build_supported_models_dictionary(self):
self.supported_models = { self.supported_models = {
Ship:ShipValidation(), Ship:ShipValidation(),
Shipcall:ShipcallValidation(), Shipcall:ShipcallValidation(),
Berth:BerthValidation(), Berth:BerthValidation(),
User:UserValidation(), User:UserValidation(),
Participant:ParticipantValidation(), Participant:ParticipantValidation(),
} }
return return
def assert_if_not_supported(self, dataclass_object): def assert_if_not_supported(self, dataclass_object):
assert type(dataclass_object) in self.supported_models, f"unsupported model. Found: {type(dataclass_object)}" assert type(dataclass_object) in self.supported_models, f"unsupported model. Found: {type(dataclass_object)}"
return return
def verify(self, dataclass_object): def verify(self, dataclass_object):
self.assert_if_not_supported(dataclass_object) self.assert_if_not_supported(dataclass_object)
@ -157,17 +157,17 @@ class DataclassValidation(ABC):
"""parent class of dataclas validators, which determines the outline of every object""" """parent class of dataclas validators, which determines the outline of every object"""
def __init__(self): def __init__(self):
return return
def check(self, dataclass_object) -> (list, bool): def check(self, dataclass_object) -> (list, bool):
""" """
the 'check' method provides a default style, how each dataclass object is validated. It returns a list of violations the 'check' method provides a default style, how each dataclass object is validated. It returns a list of violations
and a boolean, which determines, whether the check is passed successfully and a boolean, which determines, whether the check is passed successfully
""" """
all_rules = self.apply_all_rules(dataclass_object) all_rules = self.apply_all_rules(dataclass_object)
violations = self.filter_violations(all_rules) violations = self.filter_violations(all_rules)
input_validation_state = self.evaluate(violations) input_validation_state = self.evaluate(violations)
return (violations, input_validation_state) return (violations, input_validation_state)
@abstractmethod @abstractmethod
def apply_all_rules(self, dataclass_object) -> list: def apply_all_rules(self, dataclass_object) -> list:
""" """
@ -176,13 +176,13 @@ class DataclassValidation(ABC):
""" """
all_rules = [(True, 'blank_validation_rule')] all_rules = [(True, 'blank_validation_rule')]
return all_rules return all_rules
def filter_violations(self, all_rules): def filter_violations(self, all_rules):
"""input: all_rules, a list of tuples, where each element is (output, validation_name), which are (bool, str). """ """input: all_rules, a list of tuples, where each element is (output, validation_name), which are (bool, str). """
# if output is False, a violation is observed # if output is False, a violation is observed
violations = [result[1] for result in all_rules if not result[0]] violations = [result[1] for result in all_rules if not result[0]]
return violations return violations
def evaluate(self, violations) -> bool: def evaluate(self, violations) -> bool:
input_validation_state = len(violations)==0 input_validation_state = len(violations)==0
return input_validation_state return input_validation_state
@ -194,7 +194,7 @@ class ShipcallValidation(DataclassValidation):
def __init__(self): def __init__(self):
super().__init__() super().__init__()
return return
def apply_all_rules(self, dataclass_object) -> list: def apply_all_rules(self, dataclass_object) -> list:
"""apply all input validation rules to determine, whether there are violations. returns a list of tuples (output, validation_name)""" """apply all input validation rules to determine, whether there are violations. returns a list of tuples (output, validation_name)"""
raise NotImplementedError() raise NotImplementedError()
@ -202,16 +202,16 @@ class ShipcallValidation(DataclassValidation):
from BreCal.validators.schema_validation import ship_bollard_pull_is_defined_or_is_not_tug, ship_bollard_pull_is_none_or_in_range, ship_callsign_len_is_seven_at_maximum, ship_eni_len_is_eight, ship_imo_len_is_seven, ship_length_in_range, ship_participant_id_is_defined_or_is_not_tug, ship_participant_id_is_none_or_int, ship_width_in_range from BreCal.validators.schema_validation import ship_bollard_pull_is_defined_or_is_not_tug, ship_bollard_pull_is_none_or_in_range, ship_callsign_len_is_seven_at_maximum, ship_eni_len_is_eight, ship_imo_len_is_seven, ship_length_in_range, ship_participant_id_is_defined_or_is_not_tug, ship_participant_id_is_none_or_int, ship_width_in_range
# skip: ship_max_draft_is_defined_or_is_not_tug, ship_max_draft_is_none_or_in_range, # skip: ship_max_draft_is_defined_or_is_not_tug, ship_max_draft_is_none_or_in_range,
class ShipValidation(DataclassValidation): class ShipValidation(DataclassValidation):
"""an object that validates a Ship dataclass object""" """an object that validates a Ship dataclass object"""
def __init__(self): def __init__(self):
super().__init__() super().__init__()
return return
def apply_all_rules(self, dataclass_object) -> list: def apply_all_rules(self, dataclass_object) -> list:
"""apply all input validation rules to determine, whether there are violations. returns a list of tuples (output, validation_name)""" """apply all input validation rules to determine, whether there are violations. returns a list of tuples (output, validation_name)"""
# skip: ship_max_draft_is_defined_or_is_not_tug, ship_max_draft_is_none_or_in_range, # skip: ship_max_draft_is_defined_or_is_not_tug, ship_max_draft_is_none_or_in_range,
""" """
#TODO_ship_max_draft #TODO_ship_max_draft
with pytest.raises(AttributeError, match="'Ship' object has no attribute 'max_draft'"): with pytest.raises(AttributeError, match="'Ship' object has no attribute 'max_draft'"):
@ -225,14 +225,14 @@ class ShipValidation(DataclassValidation):
check_rule(dataclass_object) check_rule(dataclass_object)
for check_rule in [ for check_rule in [
ship_bollard_pull_is_defined_or_is_not_tug, ship_bollard_pull_is_defined_or_is_not_tug,
ship_bollard_pull_is_none_or_in_range, ship_bollard_pull_is_none_or_in_range,
ship_callsign_len_is_seven_at_maximum, ship_callsign_len_is_seven_at_maximum,
ship_eni_len_is_eight, ship_eni_len_is_eight,
ship_imo_len_is_seven, ship_imo_len_is_seven,
ship_length_in_range, ship_length_in_range,
ship_participant_id_is_defined_or_is_not_tug, ship_participant_id_is_defined_or_is_not_tug,
ship_participant_id_is_none_or_int, ship_participant_id_is_none_or_int,
ship_width_in_range ship_width_in_range
] ]
] ]
@ -243,7 +243,7 @@ class BerthValidation(DataclassValidation):
def __init__(self): def __init__(self):
super().__init__() super().__init__()
return return
def apply_all_rules(self, dataclass_object) -> list: def apply_all_rules(self, dataclass_object) -> list:
"""apply all input validation rules to determine, whether there are violations. returns a list of tuples (output, validation_name)""" """apply all input validation rules to determine, whether there are violations. returns a list of tuples (output, validation_name)"""
raise NotImplementedError() raise NotImplementedError()
@ -254,7 +254,7 @@ class UserValidation(DataclassValidation):
def __init__(self): def __init__(self):
super().__init__() super().__init__()
return return
def apply_all_rules(self, dataclass_object) -> list: def apply_all_rules(self, dataclass_object) -> list:
"""apply all input validation rules to determine, whether there are violations. returns a list of tuples (output, validation_name)""" """apply all input validation rules to determine, whether there are violations. returns a list of tuples (output, validation_name)"""
raise NotImplementedError() raise NotImplementedError()
@ -266,7 +266,7 @@ class ParticipantValidation(DataclassValidation):
def __init__(self): def __init__(self):
super().__init__() super().__init__()
return return
def apply_all_rules(self, dataclass_object) -> list: def apply_all_rules(self, dataclass_object) -> list:
"""apply all input validation rules to determine, whether there are violations. returns a list of tuples (output, validation_name)""" """apply all input validation rules to determine, whether there are violations. returns a list of tuples (output, validation_name)"""

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

@ -335,8 +335,8 @@ class InputValidationShipcall():
Dates should be in the future. Depending on the ShipcallType, specific values should be checked Dates should be in the future. Depending on the ShipcallType, specific values should be checked
Perfornms datetime checks in the loadedModel (datetime.datetime objects). Perfornms datetime checks in the loadedModel (datetime.datetime objects).
""" """
# obtain the current datetime to check, whether the provided values are in the future # obtain the current datetime to check, whether the provided values are after ref time
time_now = datetime.datetime.now() time_ref = datetime.datetime.now() - datetime.timedelta(days=1)
type_ = loadedModel.get("type", ShipcallType.undefined.name) type_ = loadedModel.get("type", ShipcallType.undefined.name)
if isinstance(type_, str): # convert the name string to a ShipcallType data model if isinstance(type_, str): # convert the name string to a ShipcallType data model
@ -354,14 +354,14 @@ class InputValidationShipcall():
tidal_window_to = loadedModel.get("tidal_window_to", None) tidal_window_to = loadedModel.get("tidal_window_to", None)
# Estimated arrival or departure times # Estimated arrival or departure times
InputValidationShipcall.check_times_in_future_based_on_type(type_, time_now, eta, etd) InputValidationShipcall.check_times_in_future_based_on_type(type_, time_ref, eta, etd)
# Tidal Window # Tidal Window
InputValidationShipcall.check_tidal_window_in_future(type_, time_now, tidal_window_from, tidal_window_to) InputValidationShipcall.check_tidal_window_in_future(type_, time_ref, tidal_window_from, tidal_window_to)
return return
@staticmethod @staticmethod
def check_times_in_future_based_on_type(type_, time_now, eta, etd): def check_times_in_future_based_on_type(type_, time_ref, eta, etd):
""" """
checks, whether the ETA & ETD times are in the future. checks, whether the ETA & ETD times are in the future.
based on the type, this function checks: based on the type, this function checks:
@ -387,8 +387,8 @@ class InputValidationShipcall():
if eta is None: # null values -> no violation if eta is None: # null values -> no violation
return return
if not eta > time_now: if not eta > time_ref:
raise ValidationError({"eta":f"'eta' must be in the future. Incorrect datetime provided. Current Time: {time_now}. ETA: {eta}."}) raise ValidationError({"eta":f"'eta' is too far in the past. Incorrect datetime provided. Current Time: {time_ref}. ETA: {eta}."})
if etd is not None: if etd is not None:
raise ValidationError({"etd":f"'etd' should not be set when the shipcall type is 'arrival'."}) raise ValidationError({"etd":f"'etd' should not be set when the shipcall type is 'arrival'."})
if eta > time_in_a_year: if eta > time_in_a_year:
@ -398,8 +398,9 @@ class InputValidationShipcall():
if etd is None: # null values -> no violation if etd is None: # null values -> no violation
return return
if not etd > time_now: if not etd > time_ref:
raise ValidationError({"etd":f"'etd' must be in the future. Incorrect datetime provided. Current Time: {time_now}. ETD: {etd}."}) raise ValidationError({"etd":f"'etd' is too far in the past. Incorrect datetime provided. Current Time: {time_ref}. ETD: {etd}."})
if eta is not None: if eta is not None:
raise ValidationError({"eta":f"'eta' should not be set when the shipcall type is 'departure'."}) raise ValidationError({"eta":f"'eta' should not be set when the shipcall type is 'departure'."})
if etd > time_in_a_year: if etd > time_in_a_year:
@ -414,8 +415,8 @@ class InputValidationShipcall():
# rules, a user is only allowed to provide *both* values. # rules, a user is only allowed to provide *both* values.
raise ValidationError({"eta_or_etd":f"For shifting shipcalls one should always provide, both, eta and etd."}) raise ValidationError({"eta_or_etd":f"For shifting shipcalls one should always provide, both, eta and etd."})
if (not eta > time_now) or (not etd > time_now): if (not eta > time_ref) or (not etd > time_ref):
raise ValidationError({"eta_or_etd":f"'eta' and 'etd' must be in the future. Incorrect datetime provided. Current Time: {time_now}. ETA: {eta}. ETD: {etd}"}) raise ValidationError({"eta_or_etd":f"'eta' and 'etd' is too far in the past. Incorrect datetime provided. Current Time: {time_ref}. ETA: {eta}. ETD: {etd}"})
if (not etd < eta): if (not etd < eta):
raise ValidationError({"eta_or_etd":f"The estimated time of departure ('etd') must take place *before the estimated time of arrival ('eta'). The ship cannot arrive, before it has departed. Found: ETD: {etd}, ETA: {eta}"}) raise ValidationError({"eta_or_etd":f"The estimated time of departure ('etd') must take place *before the estimated time of arrival ('eta'). The ship cannot arrive, before it has departed. Found: ETD: {etd}, ETA: {eta}"})

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

@ -207,7 +207,7 @@ def test_shipcall_post_request_fails_when_voyage_string_is_invalid(get_stub_toke
with pytest.raises(ValidationError, match="Longer than maximum length 16"): with pytest.raises(ValidationError, match="Longer than maximum length 16"):
assert response.status_code==400 assert response.status_code==400
raise ValidationError(response.json()) raise ValidationError(response.json())
# Fail: special characters # Fail: special characters
post_data = original_post_data.copy() post_data = original_post_data.copy()
post_data["voyage"] = '👽' post_data["voyage"] = '👽'
@ -226,17 +226,17 @@ def test_shipcall_post_request_fails_when_type_arrival_and_not_in_future(get_stu
# accept # accept
post_data = original_post_data.copy() post_data = original_post_data.copy()
post_data["type"] = ShipcallType.arrival.name post_data["type"] = ShipcallType.arrival.name
post_data["eta"] = (datetime.datetime.now() + datetime.timedelta(hours=3)).isoformat() post_data["eta"] = (datetime.datetime.now() + datetime.timedelta(days=2)).isoformat()
response = requests.post(f"{url}/shipcalls", headers={"Content-Type":"text", "Authorization":f"Bearer {token}"}, json=post_data) response = requests.post(f"{url}/shipcalls", headers={"Content-Type":"text", "Authorization":f"Bearer {token}"}, json=post_data)
assert response.status_code == 201 assert response.status_code == 201
# error # error
post_data = original_post_data.copy() post_data = original_post_data.copy()
post_data["type"] = ShipcallType.arrival.name post_data["type"] = ShipcallType.arrival.name
post_data["eta"] = (datetime.datetime.now() - datetime.timedelta(hours=3)).isoformat() post_data["eta"] = (datetime.datetime.now() - datetime.timedelta(days=2)).isoformat()
response = requests.post(f"{url}/shipcalls", headers={"Content-Type":"text", "Authorization":f"Bearer {token}"}, json=post_data) 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"): with pytest.raises(ValidationError, match="is too far in the past. Incorrect datetime provided"):
assert response.status_code==400 assert response.status_code==400
raise ValidationError(response.json()) raise ValidationError(response.json())
return return
@ -256,10 +256,10 @@ def test_shipcall_post_request_fails_when_type_departure_and_not_in_future(get_s
# error # error
post_data = original_post_data.copy() post_data = original_post_data.copy()
post_data["type"] = ShipcallType.departure.name post_data["type"] = ShipcallType.departure.name
post_data["etd"] = (datetime.datetime.now() - datetime.timedelta(hours=3)).isoformat() post_data["etd"] = (datetime.datetime.now() - datetime.timedelta(days=3)).isoformat()
response = requests.post(f"{url}/shipcalls", headers={"Content-Type":"text", "Authorization":f"Bearer {token}"}, json=post_data) 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"): with pytest.raises(ValidationError, match="is too far in the past. Incorrect datetime provided"):
assert response.status_code==400 assert response.status_code==400
raise ValidationError(response.json()) raise ValidationError(response.json())
return return
@ -280,11 +280,11 @@ def test_shipcall_post_request_fails_when_type_shifting_and_not_in_future(get_st
# error # error
post_data = original_post_data.copy() post_data = original_post_data.copy()
post_data["type"] = ShipcallType.shifting.name post_data["type"] = ShipcallType.shifting.name
post_data["etd"] = (datetime.datetime.now() - datetime.timedelta(hours=3)).isoformat() post_data["etd"] = (datetime.datetime.now() - datetime.timedelta(days=3)).isoformat()
post_data["eta"] = (datetime.datetime.now() + datetime.timedelta(hours=3,minutes=1)).isoformat() post_data["eta"] = (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) 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"): with pytest.raises(ValidationError, match="is too far in the past. Incorrect datetime provided"):
assert response.status_code==400 assert response.status_code==400
raise ValidationError(response.json()) raise ValidationError(response.json())
return return
@ -629,7 +629,7 @@ def test_shipcall_put_request_fails_when_different_participant_id_is_assigned(ge
{"id":99115, 'participant_id': 5, 'type': 8}] {"id":99115, 'participant_id': 5, 'type': 8}]
spm_shipcall_data = [ spm_shipcall_data = [
{**{"created":created, "modified":modified, "shipcall_id":shipcall_id}, **spm} {**{"created":created, "modified":modified, "shipcall_id":shipcall_id}, **spm}
for spm in for spm in
spm_shipcall_data spm_shipcall_data
] ]
spm_shipcall_data = [ShipcallParticipantMap(**spm) for spm in spm_shipcall_data] spm_shipcall_data = [ShipcallParticipantMap(**spm) for spm in spm_shipcall_data]
@ -652,7 +652,7 @@ def test_shipcall_put_request_success(get_shipcall_id_after_stub_post_request):
user_data = {'id':6, 'participant_id':1} user_data = {'id':6, 'participant_id':1}
loadedModel = post_data loadedModel = post_data
content = post_data content = post_data
created = datetime.datetime.now()+datetime.timedelta(minutes=1) created = datetime.datetime.now()+datetime.timedelta(minutes=1)
modified = datetime.datetime.now()+datetime.timedelta(minutes=2) modified = datetime.datetime.now()+datetime.timedelta(minutes=2)
@ -662,7 +662,7 @@ def test_shipcall_put_request_success(get_shipcall_id_after_stub_post_request):
{"id":99115, 'participant_id': 5, 'type': 8}] {"id":99115, 'participant_id': 5, 'type': 8}]
spm_shipcall_data = [ spm_shipcall_data = [
{**{"created":created, "modified":modified, "shipcall_id":shipcall_id}, **spm} {**{"created":created, "modified":modified, "shipcall_id":shipcall_id}, **spm}
for spm in for spm in
spm_shipcall_data spm_shipcall_data
] ]
spm_shipcall_data = [ShipcallParticipantMap(**spm) for spm in spm_shipcall_data] spm_shipcall_data = [ShipcallParticipantMap(**spm) for spm in spm_shipcall_data]
@ -687,7 +687,7 @@ def test_shipcall_put_request_fails_when_no_agency_is_assigned(get_shipcall_id_a
user_data = {'id':6, 'participant_id':2} # participant_id 2 is not BSMD and is not authorized. user_data = {'id':6, 'participant_id':2} # participant_id 2 is not BSMD and is not authorized.
loadedModel = post_data loadedModel = post_data
content = post_data content = post_data
created = datetime.datetime.now()+datetime.timedelta(minutes=1) created = datetime.datetime.now()+datetime.timedelta(minutes=1)
modified = datetime.datetime.now()+datetime.timedelta(minutes=2) modified = datetime.datetime.now()+datetime.timedelta(minutes=2)
@ -697,7 +697,7 @@ def test_shipcall_put_request_fails_when_no_agency_is_assigned(get_shipcall_id_a
{"id":99115, 'participant_id': 5, 'type': 4}] {"id":99115, 'participant_id': 5, 'type': 4}]
spm_shipcall_data = [ spm_shipcall_data = [
{**{"created":created, "modified":modified, "shipcall_id":shipcall_id}, **spm} {**{"created":created, "modified":modified, "shipcall_id":shipcall_id}, **spm}
for spm in for spm in
spm_shipcall_data spm_shipcall_data
] ]
spm_shipcall_data = [ShipcallParticipantMap(**spm) for spm in spm_shipcall_data] spm_shipcall_data = [ShipcallParticipantMap(**spm) for spm in spm_shipcall_data]
@ -705,7 +705,7 @@ def test_shipcall_put_request_fails_when_no_agency_is_assigned(get_shipcall_id_a
# no agency assigned # no agency assigned
ivs = InputValidationShipcall() ivs = InputValidationShipcall()
with pytest.raises(werkzeug.exceptions.Forbidden, match=re.escape(f"PUT Requests for shipcalls can only be issued by an assigned AGENCY or BSMD users (if the special-flag is enabled). There is no assigned agency yet, so only BSMD users can change datasets.")): with pytest.raises(werkzeug.exceptions.Forbidden, match=re.escape(f"PUT Requests for shipcalls can only be issued by an assigned AGENCY or BSMD users (if the special-flag is enabled). There is no assigned agency yet, so only BSMD users can change datasets.")):
ivs.check_user_is_authorized_for_put_request(user_data, loadedModel, content, spm_shipcall_data) ivs.check_user_is_authorized_for_put_request(user_data, loadedModel, content, spm_shipcall_data)
return return
@ -730,7 +730,7 @@ def test_shipcall_put_request_fails_when_user_is_not_authorized(get_shipcall_id_
{"id":99115, 'participant_id': 5, 'type': 4}] {"id":99115, 'participant_id': 5, 'type': 4}]
spm_shipcall_data = [ spm_shipcall_data = [
{**{"created":created, "modified":modified, "shipcall_id":shipcall_id}, **spm} {**{"created":created, "modified":modified, "shipcall_id":shipcall_id}, **spm}
for spm in for spm in
spm_shipcall_data spm_shipcall_data
] ]
spm_shipcall_data = [ShipcallParticipantMap(**spm) for spm in spm_shipcall_data] spm_shipcall_data = [ShipcallParticipantMap(**spm) for spm in spm_shipcall_data]
@ -754,20 +754,20 @@ def test_shipcall_put_request_fails_when_user_tries_self_assignment(get_shipcall
created = datetime.datetime.now()+datetime.timedelta(minutes=1) created = datetime.datetime.now()+datetime.timedelta(minutes=1)
modified = datetime.datetime.now()+datetime.timedelta(minutes=2) modified = datetime.datetime.now()+datetime.timedelta(minutes=2)
spm_shipcall_data = [ spm_shipcall_data = [
{"id":99113, 'participant_id': 3, 'type': 1}, {"id":99113, 'participant_id': 3, 'type': 1},
{"id":99114, 'participant_id': 4, 'type': 2}, {"id":99114, 'participant_id': 4, 'type': 2},
{"id":99115, 'participant_id': 5, 'type': 4}] {"id":99115, 'participant_id': 5, 'type': 4}]
spm_shipcall_data = [ spm_shipcall_data = [
{**{"created":created, "modified":modified, "shipcall_id":shipcall_id}, **spm} {**{"created":created, "modified":modified, "shipcall_id":shipcall_id}, **spm}
for spm in for spm in
spm_shipcall_data spm_shipcall_data
] ]
spm_shipcall_data = [ShipcallParticipantMap(**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. # self-assignment. User is participant 6, and wants to assign participant 6.
ivs = InputValidationShipcall() ivs = InputValidationShipcall()
with pytest.raises(werkzeug.exceptions.Forbidden, match=re.escape("PUT Requests for shipcalls can only be issued by an assigned AGENCY or BSMD users (if the special-flag is enabled). There is no assigned agency yet, so only BSMD users can change datasets.")): with pytest.raises(werkzeug.exceptions.Forbidden, match=re.escape("PUT Requests for shipcalls can only be issued by an assigned AGENCY or BSMD users (if the special-flag is enabled). There is no assigned agency yet, so only BSMD users can change datasets.")):
# 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."" # 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.""
@ -813,7 +813,7 @@ def test_shipcall_put_request_works_if_most_values_are_null():
InputValidationShipcall.evaluate_put_data(user_data, loadedModel, content) InputValidationShipcall.evaluate_put_data(user_data, loadedModel, content)
return return
@ -860,10 +860,10 @@ def test_shipcall_put_request_fails_input_validation_shipcall_when_shipcall_is_c
return return
def test_post_data_with_valid_data(get_stub_token): def test_post_data_with_valid_data(get_stub_token):
"""This unit test uses the input data from """This unit test uses the input data from
# https://trello.com/c/VXVSLTF4/267-shipcall-anlegen-shifting-erh%C3%A4lt-fehler-aufgrund-fr%C3%BCherem-etd-als-eta # https://trello.com/c/VXVSLTF4/267-shipcall-anlegen-shifting-erh%C3%A4lt-fehler-aufgrund-fr%C3%BCherem-etd-als-eta
to make sure, the failure case no longer appears. to make sure, the failure case no longer appears.
""" """
url, token = get_stub_token["url"], get_stub_token["token"] url, token = get_stub_token["url"], get_stub_token["token"]