added an optional argument 'instance_path' to the create_app function. This does not change the current behaviour, but allows for easier local testing and development. Added roughly 40 unit tests to verify the input validation of shipcalls (mostly for POST requests). Updated pytests to support the 'ignore_terminal_flag', which was enabled in version 1.2. The flag is tested to ensure proper behaviour. Included most of the shipcall input validation rules for POST and PUT requests
This commit is contained in:
Max Metz
parent
98b8845015
commit
7c8cd3763a
@ -34,7 +34,7 @@ from BreCal.stubs.df_times import get_df_times
|
||||
|
||||
from BreCal.services.schedule_routines import setup_schedule, run_schedule_permanently_in_background
|
||||
|
||||
def create_app(test_config=None):
|
||||
def create_app(test_config=None, instance_path=None):
|
||||
|
||||
app = Flask(__name__, instance_relative_config=True)
|
||||
app.config.from_mapping(
|
||||
@ -45,6 +45,9 @@ def create_app(test_config=None):
|
||||
else:
|
||||
app.config.from_mapping(test_config)
|
||||
|
||||
if instance_path is not None:
|
||||
app.instance_path = instance_path
|
||||
|
||||
try:
|
||||
import os
|
||||
print(f'Instance path = {app.instance_path}')
|
||||
|
||||
@ -10,6 +10,8 @@ from ..services.auth_guard import check_jwt
|
||||
from BreCal.database.update_database import evaluate_shipcall_state
|
||||
from BreCal.database.sql_queries import create_sql_query_shipcall_get, create_sql_query_shipcall_post, create_sql_query_shipcall_put, create_sql_query_history_post, create_sql_query_history_put
|
||||
|
||||
from marshmallow import Schema, fields, ValidationError
|
||||
|
||||
def GetShipcalls(options):
|
||||
"""
|
||||
No parameters, gets all entries
|
||||
|
||||
@ -30,3 +30,8 @@ def get_participant_simple():
|
||||
deleted
|
||||
)
|
||||
return participant
|
||||
|
||||
|
||||
def get_stub_list_of_valid_participants():
|
||||
participants = [{'participant_id': 2, 'type': 4}, {'participant_id': 3, 'type': 1}, {'participant_id': 4, 'type': 2}, {'participant_id': 5, 'type': 8}]
|
||||
return participants
|
||||
|
||||
@ -3,6 +3,11 @@ from BreCal.stubs import generate_uuid1_int
|
||||
from BreCal.schemas.model import Shipcall
|
||||
from dataclasses import field
|
||||
|
||||
import json
|
||||
import datetime
|
||||
from BreCal.schemas.model import ShipcallType
|
||||
from BreCal.stubs.participant import get_stub_list_of_valid_participants
|
||||
|
||||
def get_shipcall_simple():
|
||||
# only used for the stub
|
||||
base_time = datetime.datetime.now()
|
||||
@ -107,3 +112,92 @@ def create_postman_stub_shipcall():
|
||||
}
|
||||
return shipcall
|
||||
|
||||
|
||||
def get_stub_valid_shipcall_base():
|
||||
tidal_window_from = (datetime.datetime.now()+datetime.timedelta(minutes=15)).isoformat()
|
||||
tidal_window_to = (datetime.datetime.now()+datetime.timedelta(minutes=115)).isoformat()
|
||||
|
||||
shipcall_base = {
|
||||
'ship_id': 1,
|
||||
'voyage': '43B',
|
||||
'tug_required': False,
|
||||
'pilot_required': True,
|
||||
'flags': 0,
|
||||
'pier_side': False,
|
||||
'bunkering': True,
|
||||
'recommended_tugs': 2,
|
||||
'tidal_window_from' : tidal_window_from,
|
||||
'tidal_window_to' : tidal_window_to
|
||||
}
|
||||
return shipcall_base
|
||||
|
||||
def get_stub_valid_shipcall_arrival():
|
||||
eta = (datetime.datetime.now()+datetime.timedelta(minutes=45)).isoformat()
|
||||
|
||||
post_data = {
|
||||
**get_stub_valid_shipcall_base(),
|
||||
**{
|
||||
'type': int(ShipcallType.arrival),
|
||||
'eta': eta,
|
||||
'participants':get_stub_list_of_valid_participants(),
|
||||
'arrival_berth_id':139,
|
||||
}
|
||||
}
|
||||
return post_data
|
||||
|
||||
def get_stub_valid_shipcall_departure():
|
||||
etd = (datetime.datetime.now()+datetime.timedelta(minutes=45)).isoformat()
|
||||
|
||||
post_data = {
|
||||
**get_stub_valid_shipcall_base(),
|
||||
**{
|
||||
'type': int(ShipcallType.departure),
|
||||
'etd': etd,
|
||||
'participants':get_stub_list_of_valid_participants(),
|
||||
'departure_berth_id':139,
|
||||
}
|
||||
}
|
||||
return post_data
|
||||
|
||||
def get_stub_valid_shipcall_shifting():
|
||||
eta = (datetime.datetime.now()+datetime.timedelta(minutes=45)).isoformat()
|
||||
etd = (datetime.datetime.now()+datetime.timedelta(minutes=60)).isoformat()
|
||||
|
||||
post_data = {
|
||||
**get_stub_valid_shipcall_base(),
|
||||
**{
|
||||
'type': int(ShipcallType.shifting),
|
||||
'eta': eta,
|
||||
'etd': etd,
|
||||
'participants':get_stub_list_of_valid_participants(),
|
||||
'arrival_berth_id':139,
|
||||
'departure_berth_id':139,
|
||||
}
|
||||
}
|
||||
return post_data
|
||||
|
||||
def get_stub_shipcall_arrival_invalid_missing_eta():
|
||||
post_data = get_stub_valid_shipcall_arrival()
|
||||
post_data.pop("eta", None)
|
||||
return post_data
|
||||
|
||||
def get_stub_shipcall_departure_invalid_missing_etd():
|
||||
post_data = get_stub_valid_shipcall_departure()
|
||||
post_data.pop("etd", None)
|
||||
return post_data
|
||||
|
||||
def get_stub_shipcall_shifting_invalid_missing_eta():
|
||||
post_data = get_stub_valid_shipcall_shifting()
|
||||
post_data.pop("eta", None)
|
||||
return post_data
|
||||
|
||||
def get_stub_shipcall_shifting_invalid_missing_etd():
|
||||
post_data = get_stub_valid_shipcall_shifting()
|
||||
post_data.pop("etd", None)
|
||||
return post_data
|
||||
|
||||
def get_stub_shipcall_arrival_invalid_missing_type():
|
||||
post_data = get_stub_valid_shipcall_arrival()
|
||||
post_data.pop("type", None)
|
||||
return post_data
|
||||
|
||||
|
||||
@ -17,6 +17,11 @@ from BreCal.database.enums import ParticipantType
|
||||
|
||||
from BreCal.validators.input_validation_utils import check_if_user_is_bsmd_type, get_participant_id_dictionary, check_if_ship_id_is_valid, check_if_berth_id_is_valid, check_if_participant_ids_are_valid, get_berth_id_dictionary, check_if_string_has_special_characters, get_ship_id_dictionary, check_if_int_is_valid_flag
|
||||
|
||||
def validation_error_default_asserts(response):
|
||||
"""creates assertions, when the response does not fail as expected. This function is extensively used in the input validation pytests"""
|
||||
assert response.status_code == 400
|
||||
assert 'message' in list(response.json().keys())
|
||||
return
|
||||
|
||||
|
||||
|
||||
|
||||
@ -152,7 +152,7 @@ class InputValidationShipcall():
|
||||
|
||||
valid_participant_ids = check_if_participant_ids_are_valid(participants=participants)
|
||||
if not valid_participant_ids:
|
||||
raise ValidationError(f"one of the provided participant ids are invalid. Could not find one of these in the database: {participants}")
|
||||
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:
|
||||
|
||||
@ -14,7 +14,8 @@ def test_create_app():
|
||||
|
||||
from BreCal import create_app
|
||||
os.chdir(os.path.join(lib_location,"BreCal")) # set the current directory to ~/brecal/src/server/BreCal, so the config is found
|
||||
application = create_app()
|
||||
instance_path = os.path.join(os.path.expanduser('~'), "brecal", "src", "server", "instance", "instance")
|
||||
application = create_app(test_config=None, instance_path=instance_path)
|
||||
return
|
||||
|
||||
if __name__=="__main__":
|
||||
|
||||
586
src/server/tests/validators/test_input_validation_shipcall.py
Normal file
586
src/server/tests/validators/test_input_validation_shipcall.py
Normal file
@ -0,0 +1,586 @@
|
||||
import pytest
|
||||
|
||||
import os
|
||||
import jwt
|
||||
import json
|
||||
import requests
|
||||
import datetime
|
||||
from marshmallow import ValidationError
|
||||
|
||||
from BreCal.schemas.model import Participant_Assignment, EvaluationType, ShipcallType
|
||||
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
|
||||
|
||||
@pytest.fixture
|
||||
def get_stub_token(scope="session"):
|
||||
"""
|
||||
performs a login to the user 'maxm' and returns the respective url and the token. The token will be used in
|
||||
further requests in the following format (example of post-request):
|
||||
requests.post(f"{url}/shipcalls", headers={"Content-Type":"text", "Authorization":f"Bearer {token}"}, json=post_data)
|
||||
|
||||
"""
|
||||
port = 9013
|
||||
url = f"http://127.0.0.1:{port}"
|
||||
|
||||
# set the JWT key
|
||||
os.environ['SECRET_KEY'] = 'zdiTz8P3jXOc7jztIQAoelK4zztyuCpJ'
|
||||
|
||||
try:
|
||||
response = requests.post(f"{url}/login", json=jwt.decode("eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJ1c2VybmFtZSI6Im1heG0iLCJwYXNzd29yZCI6IlN0YXJ0MTIzNCJ9.uIrbz3g-IwwTLz6C1zXELRGtAtRJ_myYJ4J4x0ozjAI", key=os.environ.get("SECRET_KEY"), algorithms=["HS256"]))
|
||||
except requests.ConnectionError as err:
|
||||
raise AssertionError(f"could not establish a connection to the default url. Did you start an instance of the local database at port {port}? Looking for a connection to {url}")
|
||||
user = response.json()
|
||||
token = user.get("token")
|
||||
return locals()
|
||||
|
||||
def test_shipcall_post_request_fails_when_ship_id_is_invalid(get_stub_token):
|
||||
url, token = get_stub_token["url"], get_stub_token["token"]
|
||||
|
||||
post_data = get_stub_valid_shipcall_arrival() # create_postman_stub_shipcall()
|
||||
|
||||
post_data["ship_id"] = 1234562
|
||||
response = requests.post(
|
||||
f"{url}/shipcalls", headers={"Content-Type":"text", "Authorization":f"Bearer {token}"}, json=post_data
|
||||
)
|
||||
|
||||
with pytest.raises(ValidationError, match=f"provided an invalid ship id"):
|
||||
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
|
||||
|
||||
def test_shipcall_post_request_fails_when_arrival_berth_id_is_invalid(get_stub_token):
|
||||
url, token = get_stub_token["url"], get_stub_token["token"]
|
||||
|
||||
post_data = get_stub_valid_shipcall_arrival() # create_postman_stub_shipcall()
|
||||
|
||||
post_data["arrival_berth_id"] = 1234562
|
||||
response = requests.post(
|
||||
f"{url}/shipcalls", headers={"Content-Type":"text", "Authorization":f"Bearer {token}"}, json=post_data
|
||||
)
|
||||
|
||||
with pytest.raises(ValidationError, match=f"provided an invalid arrival berth id"):
|
||||
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
|
||||
|
||||
def test_shipcall_post_request_fails_when_departure_berth_id_is_invalid(get_stub_token):
|
||||
url, token = get_stub_token["url"], get_stub_token["token"]
|
||||
|
||||
post_data = get_stub_valid_shipcall_arrival() # create_postman_stub_shipcall()
|
||||
|
||||
post_data["departure_berth_id"] = 1234562
|
||||
response = requests.post(
|
||||
f"{url}/shipcalls", headers={"Content-Type":"text", "Authorization":f"Bearer {token}"}, json=post_data
|
||||
)
|
||||
|
||||
with pytest.raises(ValidationError, match=f"provided an invalid departure berth id"):
|
||||
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
|
||||
|
||||
def test_shipcall_post_request_fails_when_participant_ids_are_invalid(get_stub_token):
|
||||
url, token = get_stub_token["url"], get_stub_token["token"]
|
||||
|
||||
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}]
|
||||
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"):
|
||||
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
|
||||
|
||||
def test_shipcall_post_request_fails_when_forbidden_keys_are_set(get_stub_token):
|
||||
url, token = get_stub_token["url"], get_stub_token["token"]
|
||||
|
||||
original_post_data = get_stub_valid_shipcall_arrival() # create_postman_stub_shipcall()
|
||||
|
||||
for forbidden_key, forbidden_value in zip(["canceled", "evaluation", "evaluation_message"], [1, EvaluationType.red.name, "random error message"]):
|
||||
post_data = original_post_data.copy()
|
||||
post_data[forbidden_key] = forbidden_value
|
||||
response = requests.post(
|
||||
f"{url}/shipcalls", headers={"Content-Type":"text", "Authorization":f"Bearer {token}"}, json=post_data
|
||||
)
|
||||
assert response.status_code==400
|
||||
with pytest.raises(ValidationError, match=f"may not be set on POST. "):
|
||||
raise ValidationError(response.json()) # because the response does not raise a ValidationError, we artifically create it to check the pytest.raises outcome
|
||||
return
|
||||
|
||||
def test_shipcall_post_request_fails_when_draft_is_out_of_range(get_stub_token):
|
||||
url, token = get_stub_token["url"], get_stub_token["token"]
|
||||
|
||||
post_data = get_stub_valid_shipcall_arrival() # create_postman_stub_shipcall()
|
||||
|
||||
post_data["draft"] = 0
|
||||
|
||||
response = requests.post(f"{url}/shipcalls", headers={"Content-Type":"text", "Authorization":f"Bearer {token}"}, json=post_data)
|
||||
with pytest.raises(ValidationError, match=f"Must be greater than 0 and less than or equal to "):
|
||||
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
|
||||
|
||||
post_data = get_stub_valid_shipcall_arrival() # create_postman_stub_shipcall()
|
||||
|
||||
post_data["draft"] = 21
|
||||
|
||||
response = requests.post(f"{url}/shipcalls", headers={"Content-Type":"text", "Authorization":f"Bearer {token}"}, json=post_data)
|
||||
with pytest.raises(ValidationError, match=f"Must be greater than 0 and less than or equal to "):
|
||||
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
|
||||
|
||||
post_data = get_stub_valid_shipcall_arrival() # create_postman_stub_shipcall()
|
||||
|
||||
post_data["draft"] = 20
|
||||
response = requests.post(f"{url}/shipcalls", headers={"Content-Type":"text", "Authorization":f"Bearer {token}"}, json=post_data)
|
||||
assert response.status_code==201, f"the request should accept 20.0 as a valid 'draft' value"
|
||||
return
|
||||
|
||||
def test_shipcall_post_request_fails_when_recommended_tugs_is_out_of_range(get_stub_token):
|
||||
url, token = get_stub_token["url"], get_stub_token["token"]
|
||||
|
||||
original_post_data = get_stub_valid_shipcall_arrival() # create_postman_stub_shipcall()
|
||||
|
||||
post_data = original_post_data.copy()
|
||||
post_data["recommended_tugs"] = 10
|
||||
response = requests.post(f"{url}/shipcalls", headers={"Content-Type":"text", "Authorization":f"Bearer {token}"}, json=post_data)
|
||||
assert response.status_code == 201
|
||||
|
||||
post_data = original_post_data.copy()
|
||||
post_data["recommended_tugs"] = 0
|
||||
response = requests.post(f"{url}/shipcalls", headers={"Content-Type":"text", "Authorization":f"Bearer {token}"}, json=post_data)
|
||||
assert response.status_code == 201
|
||||
|
||||
post_data = original_post_data.copy()
|
||||
post_data["recommended_tugs"] = 11
|
||||
response = requests.post(f"{url}/shipcalls", headers={"Content-Type":"text", "Authorization":f"Bearer {token}"}, json=post_data)
|
||||
|
||||
with pytest.raises(ValidationError, match=f"Must be greater than or equal to 0 and less than or equal to"):
|
||||
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
|
||||
|
||||
|
||||
def test_shipcall_post_request_fails_when_voyage_string_is_invalid(get_stub_token):
|
||||
url, token = get_stub_token["url"], get_stub_token["token"]
|
||||
|
||||
original_post_data = get_stub_valid_shipcall_arrival() # create_postman_stub_shipcall()
|
||||
|
||||
# Accept
|
||||
post_data = original_post_data.copy()
|
||||
post_data["voyage"] = "abcdefghijklmnop"
|
||||
response = requests.post(f"{url}/shipcalls", headers={"Content-Type":"text", "Authorization":f"Bearer {token}"}, json=post_data)
|
||||
assert response.status_code==201
|
||||
|
||||
# Fail: too long string
|
||||
post_data = original_post_data.copy()
|
||||
post_data["voyage"] = "abcdefghijklmnopq"
|
||||
response = requests.post(f"{url}/shipcalls", headers={"Content-Type":"text", "Authorization":f"Bearer {token}"}, json=post_data)
|
||||
|
||||
with pytest.raises(ValidationError, match="Longer than maximum length 16"):
|
||||
assert response.status_code==400
|
||||
raise ValidationError(response.json())
|
||||
|
||||
# Fail: special characters
|
||||
post_data = original_post_data.copy()
|
||||
post_data["voyage"] = '👽'
|
||||
response = requests.post(f"{url}/shipcalls", headers={"Content-Type":"text", "Authorization":f"Bearer {token}"}, json=post_data)
|
||||
|
||||
with pytest.raises(ValidationError, match="Please use only digits and ASCII letters."):
|
||||
assert response.status_code==400
|
||||
raise ValidationError(response.json())
|
||||
return
|
||||
|
||||
def test_shipcall_post_request_fails_when_type_arrival_and_not_in_future(get_stub_token):
|
||||
url, token = get_stub_token["url"], get_stub_token["token"]
|
||||
|
||||
original_post_data = get_stub_valid_shipcall_arrival() # create_postman_stub_shipcall()
|
||||
|
||||
# accept
|
||||
post_data = original_post_data.copy()
|
||||
post_data["type"] = ShipcallType.arrival
|
||||
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["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)
|
||||
|
||||
with pytest.raises(ValidationError, match="must be in the future. Incorrect datetime provided"):
|
||||
assert response.status_code==400
|
||||
raise ValidationError(response.json())
|
||||
return
|
||||
|
||||
def test_shipcall_post_request_fails_when_type_departure_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()
|
||||
|
||||
# accept
|
||||
post_data = original_post_data.copy()
|
||||
post_data["type"] = ShipcallType.departure
|
||||
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["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)
|
||||
|
||||
with pytest.raises(ValidationError, match="must be in the future. Incorrect datetime provided"):
|
||||
assert response.status_code==400
|
||||
raise ValidationError(response.json())
|
||||
return
|
||||
|
||||
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()
|
||||
|
||||
# accept
|
||||
post_data = original_post_data.copy()
|
||||
post_data["type"] = ShipcallType.departure
|
||||
post_data["eta"] = (datetime.datetime.now() + datetime.timedelta(hours=3)).isoformat()
|
||||
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["eta"] = (datetime.datetime.now() + datetime.timedelta(hours=3)).isoformat()
|
||||
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)
|
||||
|
||||
with pytest.raises(ValidationError, match="must be in the future. Incorrect datetime provided"):
|
||||
assert response.status_code==400
|
||||
raise ValidationError(response.json())
|
||||
return
|
||||
|
||||
def test_shipcall_post_request_fails_when_type_arrival_and_missing_eta(get_stub_token):
|
||||
url, token = get_stub_token["url"], get_stub_token["token"]
|
||||
|
||||
original_post_data = get_stub_valid_shipcall_arrival() # create_postman_stub_shipcall()
|
||||
|
||||
post_data = original_post_data.copy()
|
||||
post_data.pop("eta", None)
|
||||
post_data["type"] = ShipcallType.arrival
|
||||
response = requests.post(f"{url}/shipcalls", headers={"Content-Type":"text", "Authorization":f"Bearer {token}"}, json=post_data)
|
||||
|
||||
with pytest.raises(ValidationError, match="Missing key!"):
|
||||
assert response.status_code==400
|
||||
raise ValidationError(response.json())
|
||||
return
|
||||
|
||||
def test_shipcall_post_request_fails_when_type_departure_and_missing_etd(get_stub_token):
|
||||
url, token = get_stub_token["url"], get_stub_token["token"]
|
||||
|
||||
original_post_data = get_stub_valid_shipcall_arrival() # create_postman_stub_shipcall()
|
||||
|
||||
post_data = original_post_data.copy()
|
||||
post_data.pop("etd", None)
|
||||
post_data["type"] = ShipcallType.departure
|
||||
response = requests.post(f"{url}/shipcalls", headers={"Content-Type":"text", "Authorization":f"Bearer {token}"}, json=post_data)
|
||||
|
||||
with pytest.raises(ValidationError, match="Missing key!"):
|
||||
assert response.status_code==400
|
||||
raise ValidationError(response.json())
|
||||
return
|
||||
|
||||
def test_shipcall_post_request_fails_when_type_shifting_and_missing_eta(get_stub_token):
|
||||
url, token = get_stub_token["url"], get_stub_token["token"]
|
||||
|
||||
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.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)
|
||||
|
||||
with pytest.raises(ValidationError, match="Missing key!"):
|
||||
assert response.status_code==400
|
||||
raise ValidationError(response.json())
|
||||
return
|
||||
|
||||
def test_shipcall_post_request_fails_when_type_shifting_and_missing_etd(get_stub_token):
|
||||
url, token = get_stub_token["url"], get_stub_token["token"]
|
||||
|
||||
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["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)
|
||||
|
||||
with pytest.raises(ValidationError, match="Missing key!"):
|
||||
assert response.status_code==400
|
||||
raise ValidationError(response.json())
|
||||
return
|
||||
|
||||
|
||||
|
||||
def test_shipcall_post_invalid_tidal_window_to_smaller_than_tidal_window_from(get_stub_token):
|
||||
url, token = get_stub_token["url"], get_stub_token["token"]
|
||||
|
||||
post_data = get_stub_valid_shipcall_shifting()
|
||||
post_data["tidal_window_to"] = (datetime.datetime.fromisoformat(post_data["tidal_window_from"])-datetime.timedelta(minutes=1)).isoformat()
|
||||
|
||||
response = requests.post(
|
||||
f"{url}/shipcalls", headers={"Content-Type":"text", "Authorization":f"Bearer {token}"}, json=post_data
|
||||
)
|
||||
validation_error_default_asserts(response)
|
||||
assert "\'tidal_window_to\' must take place after \'tidal_window_from\'" in response.json().get("message","")
|
||||
return
|
||||
|
||||
def test_shipcall_post_invalid_tidal_windows_must_be_in_future(get_stub_token):
|
||||
url, token = get_stub_token["url"], get_stub_token["token"]
|
||||
|
||||
post_data = get_stub_valid_shipcall_shifting()
|
||||
post_data["tidal_window_from"] = (datetime.datetime.now()-datetime.timedelta(minutes=1)).isoformat()
|
||||
|
||||
response = requests.post(
|
||||
f"{url}/shipcalls", headers={"Content-Type":"text", "Authorization":f"Bearer {token}"}, json=post_data
|
||||
)
|
||||
validation_error_default_asserts(response)
|
||||
assert "\'tidal_window_from\' must be in the future. " in response.json().get("message","")
|
||||
|
||||
post_data = get_stub_valid_shipcall_shifting()
|
||||
post_data["tidal_window_to"] = (datetime.datetime.now()-datetime.timedelta(minutes=1)).isoformat()
|
||||
|
||||
response = requests.post(
|
||||
f"{url}/shipcalls", headers={"Content-Type":"text", "Authorization":f"Bearer {token}"}, json=post_data
|
||||
)
|
||||
validation_error_default_asserts(response)
|
||||
assert "\'tidal_window_to\' must be in the future. " in response.json().get("message","")
|
||||
return
|
||||
|
||||
def test_shipcall_post_invalid_canceled_must_not_be_set(get_stub_token):
|
||||
url, token = get_stub_token["url"], get_stub_token["token"]
|
||||
stubs = [("canceled", 1), ("evaluation", "green"), ("evaluation_message", "this is an error message")]
|
||||
|
||||
for key, value in stubs:
|
||||
post_data = get_stub_valid_shipcall_shifting()
|
||||
post_data[key] = value
|
||||
|
||||
response = requests.post(
|
||||
f"{url}/shipcalls", headers={"Content-Type":"text", "Authorization":f"Bearer {token}"}, json=post_data
|
||||
)
|
||||
validation_error_default_asserts(response)
|
||||
assert f"\'{key}\' may not be set on POST. Found:" in response.json().get("message","")
|
||||
return
|
||||
|
||||
def test_shipcall_post_invalid_participant_type_listed_multiple_times(get_stub_token):
|
||||
url, token = get_stub_token["url"], get_stub_token["token"]
|
||||
participants = get_stub_list_of_valid_participants()
|
||||
|
||||
# create double entry for 'id' and 'type'. Either of the two should raise an exception.
|
||||
participants.append(participants[0])
|
||||
|
||||
post_data = get_stub_valid_shipcall_shifting()
|
||||
post_data["participants"] = participants
|
||||
|
||||
response = requests.post(
|
||||
f"{url}/shipcalls", headers={"Content-Type":"text", "Authorization":f"Bearer {token}"}, json=post_data
|
||||
)
|
||||
validation_error_default_asserts(response)
|
||||
assert f"every participant id and type should be listed only once. Found multiple entries for one of the participants." in response.json().get("message","")
|
||||
return
|
||||
|
||||
def test_shipcall_post_invalid_participants_missing_agency(get_stub_token):
|
||||
url, token = get_stub_token["url"], get_stub_token["token"]
|
||||
|
||||
post_data = {}
|
||||
response = requests.get(
|
||||
f"{url}/participants", headers={"Content-Type":"text", "Authorization":f"Bearer {token}"}, json=post_data
|
||||
)
|
||||
participants = response.json()
|
||||
participant_id_dict = {item.get("id"):{"participant_id":item.get("id"), "type":item.get("type")} for item in response.json()}
|
||||
|
||||
# e.g., [{'participant_id': 2, 'type': 4}, {'participant_id': 3, 'type': 1}, {'participant_id': 4, 'type': 2}]
|
||||
participants = [participant_id_dict[2], participant_id_dict[3], participant_id_dict[4]]
|
||||
|
||||
post_data = get_stub_valid_shipcall_shifting()
|
||||
post_data["participants"] = participants
|
||||
|
||||
response = requests.post(
|
||||
f"{url}/shipcalls", headers={"Content-Type":"text", "Authorization":f"Bearer {token}"}, json=post_data
|
||||
)
|
||||
validation_error_default_asserts(response)
|
||||
assert "One of the assigned participants *must* be of type \'ParticipantType.AGENCY\'" in response.json().get("message","")
|
||||
return
|
||||
|
||||
def test_shipcall_post_invalid_etd_smaller_than_eta(get_stub_token):
|
||||
url, token = get_stub_token["url"], get_stub_token["token"]
|
||||
|
||||
post_data = get_stub_valid_shipcall_shifting()
|
||||
post_data["etd"] = (datetime.datetime.fromisoformat(post_data["eta"])-datetime.timedelta(minutes=1)).isoformat()
|
||||
|
||||
response = requests.post(
|
||||
f"{url}/shipcalls", headers={"Content-Type":"text", "Authorization":f"Bearer {token}"}, json=post_data
|
||||
)
|
||||
validation_error_default_asserts(response)
|
||||
assert "\'etd\' must be larger than \'eta\'. " in response.json().get("message","")
|
||||
return
|
||||
|
||||
def test_shipcall_post_invalid_eta_and_etd_must_be_in_future(get_stub_token):
|
||||
url, token = get_stub_token["url"], get_stub_token["token"]
|
||||
|
||||
post_data = get_stub_valid_shipcall_shifting()
|
||||
post_data["etd"] = (datetime.datetime.now()-datetime.timedelta(minutes=1)).isoformat()
|
||||
|
||||
response = requests.post(
|
||||
f"{url}/shipcalls", headers={"Content-Type":"text", "Authorization":f"Bearer {token}"}, json=post_data
|
||||
)
|
||||
validation_error_default_asserts(response)
|
||||
assert "\'eta\' and \'etd\' must be in the future. " in response.json().get("message","")
|
||||
|
||||
post_data = get_stub_valid_shipcall_shifting()
|
||||
post_data["eta"] = (datetime.datetime.now()-datetime.timedelta(minutes=1)).isoformat()
|
||||
|
||||
response = requests.post(
|
||||
f"{url}/shipcalls", headers={"Content-Type":"text", "Authorization":f"Bearer {token}"}, json=post_data
|
||||
)
|
||||
validation_error_default_asserts(response)
|
||||
assert "\'eta\' and \'etd\' must be in the future. " in response.json().get("message","")
|
||||
return
|
||||
|
||||
def test_shipcall_post_request_missing_mandatory_keys(get_stub_token): # fixture: some sort of local API start in the background
|
||||
"""
|
||||
creates a valid shipcall entry and modifies it, by dropping one of the mandatory keys. This test ensures,
|
||||
that each mandatory key raises a ValidationError, when the key is missing.
|
||||
"""
|
||||
url = get_stub_token.get("url")
|
||||
token = get_stub_token.get("token")
|
||||
|
||||
post_data = get_stub_shipcall_arrival_invalid_missing_eta()
|
||||
response = requests.post(
|
||||
f"{url}/shipcalls", headers={"Content-Type":"text", "Authorization":f"Bearer {token}"}, json=post_data
|
||||
)
|
||||
validation_error_default_asserts(response)
|
||||
assert "providing \'eta\' is mandatory." in response.json().get("message","")
|
||||
|
||||
post_data = get_stub_shipcall_departure_invalid_missing_etd()
|
||||
response = requests.post(
|
||||
f"{url}/shipcalls", headers={"Content-Type":"text", "Authorization":f"Bearer {token}"}, json=post_data
|
||||
)
|
||||
validation_error_default_asserts(response)
|
||||
assert "providing \'etd\' is mandatory." in response.json().get("message","")
|
||||
|
||||
post_data = get_stub_shipcall_shifting_invalid_missing_eta()
|
||||
response = requests.post(
|
||||
f"{url}/shipcalls", headers={"Content-Type":"text", "Authorization":f"Bearer {token}"}, json=post_data
|
||||
)
|
||||
validation_error_default_asserts(response)
|
||||
assert "providing \'eta\' and \'etd\' is mandatory." in response.json().get("message","")
|
||||
|
||||
post_data = get_stub_shipcall_shifting_invalid_missing_etd()
|
||||
response = requests.post(
|
||||
f"{url}/shipcalls", headers={"Content-Type":"text", "Authorization":f"Bearer {token}"}, json=post_data
|
||||
)
|
||||
validation_error_default_asserts(response)
|
||||
assert "providing \'eta\' and \'etd\' is mandatory." in response.json().get("message","")
|
||||
|
||||
# the following keys all share the same logic and will be tested in sequence
|
||||
for KEY in ["eta", "arrival_berth_id", "type", "ship_id"]:
|
||||
|
||||
# artificially remove the KEY from a valid shipcall entry
|
||||
post_data = get_stub_valid_shipcall_arrival()
|
||||
post_data.pop(KEY,None)
|
||||
|
||||
response = requests.post(
|
||||
f"{url}/shipcalls", headers={"Content-Type":"text", "Authorization":f"Bearer {token}"}, json=post_data
|
||||
)
|
||||
validation_error_default_asserts(response)
|
||||
assert f"providing \'{KEY}\' is mandatory." in response.json().get("message","")
|
||||
|
||||
# BERTH ID (arrival or departure, based on type)
|
||||
post_data = get_stub_valid_shipcall_arrival()
|
||||
post_data.pop("arrival_berth_id",None)
|
||||
|
||||
response = requests.post(
|
||||
f"{url}/shipcalls", headers={"Content-Type":"text", "Authorization":f"Bearer {token}"}, json=post_data
|
||||
)
|
||||
validation_error_default_asserts(response)
|
||||
assert "providing \'arrival_berth_id\' is mandatory." in response.json().get("message","")
|
||||
|
||||
|
||||
post_data = get_stub_valid_shipcall_departure()
|
||||
post_data.pop("departure_berth_id",None)
|
||||
|
||||
response = requests.post(
|
||||
f"{url}/shipcalls", headers={"Content-Type":"text", "Authorization":f"Bearer {token}"}, json=post_data
|
||||
)
|
||||
validation_error_default_asserts(response)
|
||||
assert "providing \'departure_berth_id\' is mandatory." in response.json().get("message","")
|
||||
|
||||
|
||||
post_data = get_stub_valid_shipcall_shifting()
|
||||
post_data.pop("arrival_berth_id",None)
|
||||
|
||||
response = requests.post(
|
||||
f"{url}/shipcalls", headers={"Content-Type":"text", "Authorization":f"Bearer {token}"}, json=post_data
|
||||
)
|
||||
validation_error_default_asserts(response)
|
||||
assert "providing \'arrival_berth_id\' & \'departure_berth_id\' is mandatory." in response.json().get("message","")
|
||||
|
||||
|
||||
post_data = get_stub_valid_shipcall_shifting()
|
||||
post_data.pop("departure_berth_id",None)
|
||||
|
||||
response = requests.post(
|
||||
f"{url}/shipcalls", headers={"Content-Type":"text", "Authorization":f"Bearer {token}"}, json=post_data
|
||||
)
|
||||
validation_error_default_asserts(response)
|
||||
assert "providing \'arrival_berth_id\' & \'departure_berth_id\' is mandatory." in response.json().get("message","")
|
||||
|
||||
return
|
||||
|
||||
|
||||
def test_shipcall_post_invalid_agency_missing_participant_list(get_stub_token):
|
||||
url, token = get_stub_token["url"], get_stub_token["token"]
|
||||
|
||||
post_data = get_stub_valid_shipcall_shifting()
|
||||
|
||||
# keep all participants, but drop the agency
|
||||
post_data["participants"] = [
|
||||
participant for participant in post_data.get("participants")
|
||||
if not int(participant.get("type")) == int(ParticipantType.AGENCY)
|
||||
]
|
||||
|
||||
response = requests.post(
|
||||
f"{url}/shipcalls", headers={"Content-Type":"text", "Authorization":f"Bearer {token}"}, json=post_data
|
||||
)
|
||||
validation_error_default_asserts(response)
|
||||
assert "One of the assigned participants *must* be of type \'ParticipantType.AGENCY\'" in response.json().get("message","")
|
||||
return
|
||||
|
||||
def test_shipcall_post_type_is_wrong(get_stub_token):
|
||||
url, token = get_stub_token["url"], get_stub_token["token"]
|
||||
|
||||
post_data = get_stub_valid_shipcall_arrival()
|
||||
|
||||
# type 1 should be successful (201)
|
||||
post_data["type"] = 1
|
||||
|
||||
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
|
||||
|
||||
response = requests.post(
|
||||
f"{url}/shipcalls", headers={"Content-Type":"text", "Authorization":f"Bearer {token}"}, json=post_data
|
||||
)
|
||||
validation_error_default_asserts(response)
|
||||
return
|
||||
@ -664,6 +664,9 @@ def test_validation_rule_fct_missing_time_tug_berth_etd__shipcall_soon_but_parti
|
||||
def test_validation_rule_fct_missing_time_terminal_berth_eta__shipcall_soon_but_participant_estimated_time_undefined(build_sql_proxy_connection):
|
||||
"""0001-L validation_rule_fct_missing_time_terminal_berth_eta"""
|
||||
vr = build_sql_proxy_connection['vr']
|
||||
import copy
|
||||
reset_to_default = copy.deepcopy(vr.ignore_terminal_flag)
|
||||
vr.ignore_terminal_flag = False
|
||||
|
||||
shipcall = get_shipcall_simple()
|
||||
df_times = get_df_times(shipcall)
|
||||
@ -694,6 +697,46 @@ def test_validation_rule_fct_missing_time_terminal_berth_eta__shipcall_soon_but_
|
||||
|
||||
# expectation: green state, no msg
|
||||
assert state==StatusFlags.YELLOW, f"function should return 'yellow', because the participant did not provide a time and the shipcall takes place soon (according to the agency)"
|
||||
vr.ignore_terminal_flag = reset_to_default
|
||||
return
|
||||
|
||||
def test_validation_rule_fct_missing_time_terminal_berth_eta__shipcall_soon_but_participant_estimated_time_undefined__no_violation_because_terminal_flag_is_active(build_sql_proxy_connection):
|
||||
"""0001-L validation_rule_fct_missing_time_terminal_berth_eta"""
|
||||
vr = build_sql_proxy_connection['vr']
|
||||
import copy
|
||||
reset_to_default = copy.deepcopy(vr.ignore_terminal_flag)
|
||||
vr.ignore_terminal_flag = True
|
||||
|
||||
shipcall = get_shipcall_simple()
|
||||
df_times = get_df_times(shipcall)
|
||||
|
||||
# according to the agency, a shipcall takes place soon (ETA/ETD)
|
||||
df_times.loc[df_times["participant_type"]==ParticipantType.AGENCY.value, "eta_berth"] = datetime.datetime.now() + datetime.timedelta(minutes=ParticipantwiseTimeDelta.TERMINAL-10)
|
||||
|
||||
# set times agency to be undetermined
|
||||
df_times.loc[df_times["participant_type"]==ParticipantType.TERMINAL.value, "operations_start"] = None # previously: eta_berth, which does not exist in times_terminal
|
||||
|
||||
# must adapt the shipcall_participant_map, so it suits the test
|
||||
agency_participant_id = df_times.loc[df_times["participant_type"]==ParticipantType.AGENCY.value, "participant_id"].iloc[0]
|
||||
terminal_participant_id = df_times.loc[df_times["participant_type"]==ParticipantType.TERMINAL.value, "participant_id"].iloc[0]
|
||||
|
||||
spm = vr.sql_handler.df_dict["shipcall_participant_map"]
|
||||
df = pd.DataFrame(
|
||||
[
|
||||
{"id":10001, "shipcall_id":shipcall.id, "participant_id":agency_participant_id, "type":ParticipantType.AGENCY.value, "created":pd.Timestamp(datetime.datetime.now().isoformat()), "modified":None},
|
||||
{"id":10002, "shipcall_id":shipcall.id, "participant_id":terminal_participant_id, "type":ParticipantType.TERMINAL.value, "created":pd.Timestamp(datetime.datetime.now().isoformat()), "modified":None}
|
||||
]
|
||||
)
|
||||
df.set_index("id", inplace=True)
|
||||
spm = pd.concat([spm, df], axis=0, ignore_index=True)
|
||||
vr.sql_handler.df_dict["shipcall_participant_map"] = spm
|
||||
|
||||
# apply the validation rule
|
||||
(state, msg) = vr.validation_rule_fct_missing_time_terminal_berth_eta(shipcall=shipcall, df_times=df_times)
|
||||
|
||||
# expectation: green state, no msg
|
||||
assert state==StatusFlags.GREEN, f"function should return 'green', becaues the ignore terminal flag is active. Hence, terminal validation rules are ignored."
|
||||
vr.ignore_terminal_flag = reset_to_default
|
||||
return
|
||||
|
||||
|
||||
@ -701,6 +744,9 @@ def test_validation_rule_fct_missing_time_terminal_berth_eta__shipcall_soon_but_
|
||||
def test_validation_rule_fct_missing_time_terminal_berth_etd__shipcall_soon_but_participant_estimated_time_undefined(build_sql_proxy_connection):
|
||||
"""0001-M validation_rule_fct_missing_time_terminal_berth_etd"""
|
||||
vr = build_sql_proxy_connection['vr']
|
||||
import copy
|
||||
reset_to_default = copy.deepcopy(vr.ignore_terminal_flag)
|
||||
vr.ignore_terminal_flag = False
|
||||
|
||||
shipcall = get_shipcall_simple()
|
||||
shipcall.type = ShipcallType.OUTGOING.value
|
||||
@ -733,6 +779,48 @@ def test_validation_rule_fct_missing_time_terminal_berth_etd__shipcall_soon_but_
|
||||
|
||||
# expectation: green state, no msg
|
||||
assert state==StatusFlags.YELLOW, f"function should return 'yellow', because the participant did not provide a time and the shipcall takes place soon (according to the agency)"
|
||||
vr.ignore_terminal_flag = reset_to_default
|
||||
return
|
||||
|
||||
def test_validation_rule_fct_missing_time_terminal_berth_etd__shipcall_soon_but_participant_estimated_time_undefined__no_violation_because_terminal_flag_is_active(build_sql_proxy_connection):
|
||||
"""0001-M validation_rule_fct_missing_time_terminal_berth_etd"""
|
||||
vr = build_sql_proxy_connection['vr']
|
||||
import copy
|
||||
reset_to_default = copy.deepcopy(vr.ignore_terminal_flag)
|
||||
vr.ignore_terminal_flag = True
|
||||
|
||||
shipcall = get_shipcall_simple()
|
||||
shipcall.type = ShipcallType.OUTGOING.value
|
||||
df_times = get_df_times(shipcall)
|
||||
|
||||
# according to the agency, a shipcall takes place soon (ETA/ETD)
|
||||
df_times.loc[df_times["participant_type"]==ParticipantType.AGENCY.value, "etd_berth"] = datetime.datetime.now() + datetime.timedelta(minutes=ParticipantwiseTimeDelta.TERMINAL-10)
|
||||
|
||||
# set times agency to be undetermined
|
||||
df_times.loc[df_times["participant_type"]==ParticipantType.TERMINAL.value, "operations_end"] = None # previously: etd_berth, which does not exist in times_terminal
|
||||
|
||||
|
||||
# must adapt the shipcall_participant_map, so it suits the test
|
||||
agency_participant_id = df_times.loc[df_times["participant_type"]==ParticipantType.AGENCY.value, "participant_id"].iloc[0]
|
||||
terminal_participant_id = df_times.loc[df_times["participant_type"]==ParticipantType.TERMINAL.value, "participant_id"].iloc[0]
|
||||
|
||||
spm = vr.sql_handler.df_dict["shipcall_participant_map"]
|
||||
df = pd.DataFrame(
|
||||
[
|
||||
{"id":10001, "shipcall_id":shipcall.id, "participant_id":agency_participant_id, "type":ParticipantType.AGENCY.value, "created":pd.Timestamp(datetime.datetime.now().isoformat()), "modified":None},
|
||||
{"id":10002, "shipcall_id":shipcall.id, "participant_id":terminal_participant_id, "type":ParticipantType.TERMINAL.value, "created":pd.Timestamp(datetime.datetime.now().isoformat()), "modified":None}
|
||||
]
|
||||
)
|
||||
df.set_index("id", inplace=True)
|
||||
spm = pd.concat([spm, df], axis=0, ignore_index=True)
|
||||
vr.sql_handler.df_dict["shipcall_participant_map"] = spm
|
||||
|
||||
# apply the validation rule
|
||||
(state, msg) = vr.validation_rule_fct_missing_time_terminal_berth_etd(shipcall=shipcall, df_times=df_times)
|
||||
|
||||
# expectation: green state, no msg
|
||||
assert state==StatusFlags.GREEN, f"function should return 'green', becaues the ignore terminal flag is active. Hence, terminal validation rules are ignored."
|
||||
vr.ignore_terminal_flag = reset_to_default
|
||||
return
|
||||
|
||||
|
||||
@ -922,6 +1010,10 @@ def test_validation_rule_fct_shipcall_shifting_participants_disagree_on_eta_or_e
|
||||
def test_validation_rule_fct_eta_time_not_in_operation_window__times_dont_match(build_sql_proxy_connection):
|
||||
"""0003-A validation_rule_fct_eta_time_not_in_operation_window"""
|
||||
vr = build_sql_proxy_connection['vr']
|
||||
import copy
|
||||
reset_to_default = copy.deepcopy(vr.ignore_terminal_flag)
|
||||
vr.ignore_terminal_flag = False
|
||||
|
||||
shipcall = get_shipcall_simple()
|
||||
df_times = get_df_times(shipcall)
|
||||
|
||||
@ -933,11 +1025,37 @@ def test_validation_rule_fct_eta_time_not_in_operation_window__times_dont_match(
|
||||
|
||||
(code, msg) = vr.validation_rule_fct_eta_time_not_in_operation_window(shipcall, df_times)
|
||||
assert code==StatusFlags.RED, f"status flag should be 'red', because the planned operations start is BEFORE the estimated time of arrival for the shipcall"
|
||||
vr.ignore_terminal_flag = reset_to_default
|
||||
return
|
||||
|
||||
def test_validation_rule_fct_eta_time_not_in_operation_window__times_dont_match__no_violation_because_terminal_flag_is_active(build_sql_proxy_connection):
|
||||
"""0003-A validation_rule_fct_eta_time_not_in_operation_window"""
|
||||
vr = build_sql_proxy_connection['vr']
|
||||
import copy
|
||||
reset_to_default = copy.deepcopy(vr.ignore_terminal_flag)
|
||||
vr.ignore_terminal_flag = True
|
||||
|
||||
shipcall = get_shipcall_simple()
|
||||
df_times = get_df_times(shipcall)
|
||||
|
||||
t0_time = datetime.datetime.now() # reference time for easier readability
|
||||
|
||||
# the planned operations_start is before eta_berth (by one minute in this case)
|
||||
df_times.loc[df_times["participant_type"]==ParticipantType.AGENCY.value, "eta_berth"] = t0_time + datetime.timedelta(minutes=1)
|
||||
df_times.loc[df_times["participant_type"]==ParticipantType.TERMINAL.value, "operations_start"] = t0_time + datetime.timedelta(minutes=0)
|
||||
|
||||
(code, msg) = vr.validation_rule_fct_eta_time_not_in_operation_window(shipcall, df_times)
|
||||
assert code==StatusFlags.GREEN, f"the ignore terminal flag is active, so this validation rule is ignored. There should not be a violation"
|
||||
vr.ignore_terminal_flag = reset_to_default
|
||||
return
|
||||
|
||||
def test_validation_rule_fct_etd_time_not_in_operation_window__times_dont_match(build_sql_proxy_connection):
|
||||
"""0003-B validation_rule_fct_etd_time_not_in_operation_window"""
|
||||
vr = build_sql_proxy_connection['vr']
|
||||
import copy
|
||||
reset_to_default = copy.deepcopy(vr.ignore_terminal_flag)
|
||||
vr.ignore_terminal_flag = False
|
||||
|
||||
shipcall = get_shipcall_simple()
|
||||
shipcall.type = ShipcallType.SHIFTING.value
|
||||
df_times = get_df_times(shipcall)
|
||||
@ -951,6 +1069,30 @@ def test_validation_rule_fct_etd_time_not_in_operation_window__times_dont_match(
|
||||
|
||||
(code, msg) = vr.validation_rule_fct_etd_time_not_in_operation_window(shipcall, df_times)
|
||||
assert code==StatusFlags.RED, f"status flag should be 'red', because the planned operations end is AFTER the estimated time of departure for the shipcall"
|
||||
vr.ignore_terminal_flag = reset_to_default
|
||||
return
|
||||
|
||||
def test_validation_rule_fct_etd_time_not_in_operation_window__times_dont_match__no_violation_because_terminal_flag_is_active(build_sql_proxy_connection):
|
||||
"""0003-B validation_rule_fct_etd_time_not_in_operation_window"""
|
||||
vr = build_sql_proxy_connection['vr']
|
||||
import copy
|
||||
reset_to_default = copy.deepcopy(vr.ignore_terminal_flag)
|
||||
vr.ignore_terminal_flag = True
|
||||
|
||||
shipcall = get_shipcall_simple()
|
||||
shipcall.type = ShipcallType.SHIFTING.value
|
||||
df_times = get_df_times(shipcall)
|
||||
|
||||
t0_time = datetime.datetime.now() # reference time for easier readability
|
||||
|
||||
# the planned operations_end is after etd_berth (by one minute in this case)
|
||||
df_times.loc[df_times["participant_type"]==ParticipantType.AGENCY.value, "etd_berth"] = t0_time + datetime.timedelta(hours=1)
|
||||
df_times.loc[df_times["participant_type"]==ParticipantType.TERMINAL.value, "operations_end"] = t0_time+datetime.timedelta(hours=1, minutes=1)
|
||||
|
||||
|
||||
(code, msg) = vr.validation_rule_fct_etd_time_not_in_operation_window(shipcall, df_times)
|
||||
assert code==StatusFlags.GREEN, f"the ignore terminal flag is active, so this validation rule is ignored. There should not be a violation"
|
||||
vr.ignore_terminal_flag = reset_to_default
|
||||
return
|
||||
|
||||
def test_validation_rule_fct_eta_time_not_in_operation_window_and_validation_rule_fct_etd_time_not_in_operation_window__always_okay(build_sql_proxy_connection):
|
||||
|
||||
Reference in New Issue
Block a user