import pytest import datetime import pandas as pd from BreCal.validators.validation_rule_functions import ValidationRuleFunctions from BreCal.validators.validation_rules import ValidationRules from BreCal.database.sql_handler import SQLHandler from BreCal.database.enums import ParticipantwiseTimeDelta, ParticipantType, StatusFlags, ShipcallType from BreCal.stubs.shipcall import get_shipcall_simple from BreCal.stubs.df_times import get_df_times, random_time_perturbation, get_df_times_participants_disagree, build_stub_df_times @pytest.fixture(scope="session") def build_sql_proxy_connection(): import mysql.connector conn_from_pool = mysql.connector.connect(**{'host':'localhost', 'port':3306, 'user':'root', 'password':'HalloWach_2323XXL!!', 'pool_name':'brecal_pool', 'pool_size':20, 'database':'bremen_calling', 'autocommit': True}) sql_handler = SQLHandler(sql_connection=conn_from_pool, read_all=True) vr = ValidationRules(sql_handler) return locals() def test_build_validation_rule_functions(build_sql_proxy_connection): import types sql_handler = build_sql_proxy_connection["sql_handler"] vr = build_sql_proxy_connection["vr"] validation_rule_functions = vr.get_validation_rule_functions() assert isinstance(validation_rule_functions, list), f"must return a list of methods" for vrule in validation_rule_functions: assert isinstance(vrule,types.MethodType), f"every element returned from get_validation_rule_functions must be a method. found: {type(vrule)}" assert len(validation_rule_functions)>0, f"must return at least one method!" return def test_check_time_delta_violation_query_time_to_now_key_time_is_defined(build_sql_proxy_connection): vr = build_sql_proxy_connection["vr"] # ship arrives in three hours, while the threshold for an alert is (e.g.) 5 hours # key time is given, so the function should always return False (no violation) query_time = datetime.datetime.now() + datetime.timedelta(hours=3) key_time = datetime.datetime.now() + datetime.timedelta(hours=7, minutes=30) threshold = 60*5 violation_state = vr.check_time_delta_violation_query_time_to_now(query_time=query_time, key_time=key_time, threshold=threshold) assert not violation_state, f"the key time is filled in, so there should not be a violation" return def test_check_time_delta_violation_query_time_to_now_no_key_time_but_event_in_distant_future(build_sql_proxy_connection): vr = build_sql_proxy_connection["vr"] # ship arrives in three hours, while the threshold for an alert is (e.g.) 5 hours # key time is given, so the function should always return False (no violation) # query time (-> delta) & threshold have the same time -> no violation query_time = datetime.datetime.now() + datetime.timedelta(hours=5, seconds=10) # when the delta & threshold are identical, microseconds between checking the time and defining it here, raise the violation key_time = None threshold = 60*5 violation_state = vr.check_time_delta_violation_query_time_to_now(query_time=query_time, key_time=key_time, threshold=threshold) assert not violation_state, f"the event is still far enough away, so there should not be a violation." return def test_check_time_delta_violation_query_time_to_now_key_time_is_none(build_sql_proxy_connection): vr = build_sql_proxy_connection["vr"] # ship arrives in three hours, while the threshold for an alert is (e.g.) 5 hours # key time is given, so the function should always return False (no violation) query_time = datetime.datetime.now() + datetime.timedelta(hours=3) key_time = None threshold = 60*5 # minutes violation_state = vr.check_time_delta_violation_query_time_to_now(query_time=query_time, key_time=key_time, threshold=threshold) assert violation_state, f"when the key time is not filled in and the query time is 'dangerously close', there should be a violation to indicate the traffic state" return def test_check_time_delta_violation_query_time_to_now_key_time_is_pd_nat(build_sql_proxy_connection): vr = build_sql_proxy_connection["vr"] # ship arrives in three hours, while the threshold for an alert is (e.g.) 5 hours # key time is given, so the function should always return False (no violation) query_time = datetime.datetime.now() + datetime.timedelta(hours=3) key_time = pd.NaT threshold = 60*5 # minutes violation_state = vr.check_time_delta_violation_query_time_to_now(query_time=query_time, key_time=key_time, threshold=threshold) assert violation_state, f"when the key time is not filled in and the query time is 'dangerously close', there should be a violation to indicate the traffic state" return def test_validation_rule_fct_missing_time_agency_berth_eta__missing_time_agency_no_violation(build_sql_proxy_connection): """0001-A validation_rule_fct_missing_time_agency_berth_eta""" vr = build_sql_proxy_connection['vr'] shipcall = get_shipcall_simple() df_times = get_df_times(shipcall) # artificially remove the agency, so the function is properly checked df_times = df_times.loc[df_times["participant_type"]!=ParticipantType.AGENCY.value] # apply the validation rule (state, msg) = vr.validation_rule_fct_missing_time_agency_berth_eta(shipcall=shipcall, df_times=df_times) # expectation: green state, no msg assert state==StatusFlags.GREEN, f"function should return 'green', because the agency's entry is not present" assert msg is None, f"with a 'green' state, there should be no message returned" return def test_validation_rule_fct_missing_time_agency_berth_eta__shipcall_eta_dangerously_close_no_times_agency(build_sql_proxy_connection): """0001-A validation_rule_fct_missing_time_agency_berth_eta""" vr = build_sql_proxy_connection['vr'] shipcall = get_shipcall_simple() df_times = get_df_times(shipcall) # the shipcall happens 'soon' shipcall.eta = datetime.datetime.now() + datetime.timedelta(minutes=ParticipantwiseTimeDelta.AGENCY-10) # set times agency to be undetermined df_times.loc[df_times["participant_type"]==ParticipantType.AGENCY.value, "eta_berth"] = None # 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] 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}, ] ) 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_agency_berth_eta(shipcall=shipcall, df_times=df_times) # expectation: green state, no msg assert state==StatusFlags.YELLOW, f"function should return 'yellow', because the agency's entry is not present and the shipcall takes place soon" return def test_validation_rule_fct_missing_time_agency_berth_eta__shipcall_eta_distant_enough_no_times_agency(build_sql_proxy_connection): """0001-A validation_rule_fct_missing_time_agency_berth_eta""" vr = build_sql_proxy_connection['vr'] shipcall = get_shipcall_simple() df_times = get_df_times(shipcall) # the shipcall happens 'soon' shipcall.eta = datetime.datetime.now() + datetime.timedelta(minutes=ParticipantwiseTimeDelta.AGENCY+10) # set times agency to be undetermined df_times.loc[df_times["participant_type"]==ParticipantType.AGENCY.value, "eta_berth"] = None # 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] 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}, ] ) 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_agency_berth_eta(shipcall=shipcall, df_times=df_times) # expectation: green state, no msg assert state==StatusFlags.GREEN, f"function should return 'yellow', because the agency's entry is not present and the shipcall takes place soon" return def test_validation_rule_fct_missing_time_agency_berth_eta__shipcall_eta_is_undefined_agency_eta_is_defined(build_sql_proxy_connection): """0001-A validation_rule_fct_missing_time_agency_berth_eta""" vr = build_sql_proxy_connection['vr'] shipcall = get_shipcall_simple() df_times = get_df_times(shipcall) # the shipcall is undefined shipcall.eta = None # set times agency to be undetermined df_times.loc[df_times["participant_type"]==ParticipantType.AGENCY.value, "eta_berth"] = None # 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] 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}, ] ) 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_agency_berth_eta(shipcall=shipcall, df_times=df_times) # expectation: green state, no msg assert state==StatusFlags.GREEN, f"function should return 'yellow', because the agency's entry is not present and the shipcall takes place soon" return def test_validation_rule_fct_missing_time_agency_berth_etd__shipcall_etd_is_undefined_agency_etd_is_defined(build_sql_proxy_connection): """0001-B validation_rule_fct_missing_time_agency_berth_etd""" vr = build_sql_proxy_connection['vr'] shipcall = get_shipcall_simple() df_times = get_df_times(shipcall) # the shipcall etd is 'soon' shipcall.type = ShipcallType.OUTGOING.value shipcall.etd = datetime.datetime.now() + datetime.timedelta(minutes=ParticipantwiseTimeDelta.AGENCY-10) # set times agency to be undetermined df_times.loc[df_times["participant_type"]==ParticipantType.AGENCY.value, "etd_berth"] = None # 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] 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}, ] ) 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_agency_berth_etd(shipcall=shipcall, df_times=df_times) # expectation: green state, no msg assert state==StatusFlags.YELLOW, f"function should return 'yellow', because the agency's entry is not present and the shipcall takes place soon" return def test_validation_rule_fct_missing_time_mooring_berth_eta__shipcall_soon_but_participant_estimated_time_undefined(build_sql_proxy_connection): """0001-C validation_rule_fct_missing_time_mooring_berth_eta""" vr = build_sql_proxy_connection['vr'] 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.MOORING-10) # set times agency to be undetermined df_times.loc[df_times["participant_type"]==ParticipantType.MOORING.value, "eta_berth"] = None # 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] mooring_participant_id = df_times.loc[df_times["participant_type"]==ParticipantType.MOORING.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":mooring_participant_id, "type":ParticipantType.MOORING.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_mooring_berth_eta(shipcall=shipcall, df_times=df_times) # 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)" return def test_validation_rule_fct_missing_time_mooring_berth_etd__shipcall_soon_but_participant_estimated_time_undefined(build_sql_proxy_connection): """0001-D validation_rule_fct_missing_time_mooring_berth_etd""" vr = build_sql_proxy_connection['vr'] 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.MOORING-10) # set times agency to be undetermined df_times.loc[df_times["participant_type"]==ParticipantType.MOORING.value, "etd_berth"] = None # 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] mooring_participant_id = df_times.loc[df_times["participant_type"]==ParticipantType.MOORING.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":mooring_participant_id, "type":ParticipantType.MOORING.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_mooring_berth_etd(shipcall=shipcall, df_times=df_times) # 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)" return def test_validation_rule_fct_missing_time_portadministration_berth_eta__shipcall_soon_but_participant_estimated_time_undefined(build_sql_proxy_connection): """0001-F validation_rule_fct_missing_time_portadministration_berth_eta""" vr = build_sql_proxy_connection['vr'] 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.PORT_ADMINISTRATION-10) # set times agency to be undetermined df_times.loc[df_times["participant_type"]==ParticipantType.PORT_ADMINISTRATION.value, "eta_berth"] = None # 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] pa_participant_id = df_times.loc[df_times["participant_type"]==ParticipantType.PORT_ADMINISTRATION.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":pa_participant_id, "type":ParticipantType.PORT_ADMINISTRATION.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_portadministration_berth_eta(shipcall=shipcall, df_times=df_times) if not vr.ignore_port_administration_flag: 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)" else: assert state==StatusFlags.GREEN, f"function should return 'green', because the feature flag is set, which disables this validation rule" return def test_validation_rule_fct_missing_time_portadministration_berth_etd__shipcall_soon_but_participant_estimated_time_undefined(build_sql_proxy_connection): """0001-G validation_rule_fct_missing_time_portadministration_berth_etd""" vr = build_sql_proxy_connection['vr'] shipcall = get_shipcall_simple() shipcall.type = ShipcallType.SHIFTING.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.PORT_ADMINISTRATION-10) # set times agency to be undetermined df_times.loc[df_times["participant_type"]==ParticipantType.PORT_ADMINISTRATION.value, "etd_berth"] = None # 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] pa_participant_id = df_times.loc[df_times["participant_type"]==ParticipantType.PORT_ADMINISTRATION.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":pa_participant_id, "type":ParticipantType.PORT_ADMINISTRATION.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_portadministration_berth_etd(shipcall=shipcall, df_times=df_times) if not vr.ignore_port_administration_flag: 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)" else: assert state==StatusFlags.GREEN, f"function should return 'green', because the ignore flag is set" return def test_validation_rule_fct_missing_time_pilot_berth_eta__shipcall_soon_but_participant_estimated_time_undefined(build_sql_proxy_connection): """0001-H validation_rule_fct_missing_time_pilot_berth_eta""" vr = build_sql_proxy_connection['vr'] 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.PILOT-10) # set times agency to be undetermined df_times.loc[df_times["participant_type"]==ParticipantType.PILOT.value, "eta_berth"] = None # 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] pilot_participant_id = df_times.loc[df_times["participant_type"]==ParticipantType.PILOT.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":pilot_participant_id, "type":ParticipantType.PILOT.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_pilot_berth_eta(shipcall=shipcall, df_times=df_times) # expectation: yellow state 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)" return def test_validation_rule_fct_missing_time_pilot_berth_etd__shipcall_soon_but_participant_estimated_time_undefined(build_sql_proxy_connection): """0001-I validation_rule_fct_missing_time_pilot_berth_etd""" vr = build_sql_proxy_connection['vr'] 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.PILOT-10) # set times agency to be undetermined df_times.loc[df_times["participant_type"]==ParticipantType.PILOT.value, "etd_berth"] = None # 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] pilot_participant_id = df_times.loc[df_times["participant_type"]==ParticipantType.PILOT.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":pilot_participant_id, "type":ParticipantType.PILOT.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_pilot_berth_etd(shipcall=shipcall, df_times=df_times) # 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)" return def test_validation_rule_fct_missing_time_pilot_berth_etd__shipcall_soon_but_participant_unassigned__return_green(build_sql_proxy_connection): """0001-I validation_rule_fct_missing_time_pilot_berth_etd""" vr = build_sql_proxy_connection['vr'] 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.PILOT-10) # set times agency to be undetermined df_times.loc[df_times["participant_type"]==ParticipantType.PILOT.value, "etd_berth"] = None # 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] vr.sql_handler.read_all(vr.sql_handler.all_schemas) 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} ] ) 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_pilot_berth_etd(shipcall=shipcall, df_times=df_times) # expectation: green state, no msg assert state==StatusFlags.GREEN, f"function should return 'green', because the pilot is not assigned yet" return def test_validation_rule_fct_missing_time_pilot_berth_etd__shipcall_soon_but_participant_estimated_time_undefined_multiple_pilot_assignments_due_to_bug(build_sql_proxy_connection): """ 0001-I validation_rule_fct_missing_time_pilot_berth_etd. Checks, whether the function still works in case of a buggy input. When there is more than one pilot assignment, the validation rule should still work and return 'yellow' properly. """ vr = build_sql_proxy_connection['vr'] 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.PILOT-10) # set times agency to be undetermined df_times.loc[df_times["participant_type"]==ParticipantType.PILOT.value, "etd_berth"] = None # 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] pilot_participant_id = df_times.loc[df_times["participant_type"]==ParticipantType.PILOT.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":pilot_participant_id, "type":ParticipantType.PILOT.value, "created":pd.Timestamp(datetime.datetime.now().isoformat()), "modified":None}, {"id":10003, "shipcall_id":shipcall.id, "participant_id":pilot_participant_id, "type":ParticipantType.PILOT.value, "created":pd.Timestamp(datetime.datetime.now().isoformat()), "modified":None}, {"id":10004, "shipcall_id":shipcall.id, "participant_id":pilot_participant_id, "type":ParticipantType.PILOT.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_pilot_berth_etd(shipcall=shipcall, df_times=df_times) # 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)" return def test_validation_rule_fct_missing_time_pilot_berth_etd__agency_and_pilot_assigned_pilot_no_times_returns_yellow(build_sql_proxy_connection): """ 0001-I validation_rule_fct_missing_time_pilot_berth_etd. Checks the default behaviour, where an agency's time might exist, while a time by pilot may not exist. In these cases, a yellow state is expected. """ vr = build_sql_proxy_connection['vr'] 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.PILOT-10) # 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] pilot_participant_id = df_times.loc[df_times["participant_type"]==ParticipantType.PILOT.value, "participant_id"].iloc[0] # set times of PILOT: should not exist df_times.loc[df_times["participant_type"]==ParticipantType.PILOT.value,"participant_type"] = ParticipantType.BSMD.value 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":pilot_participant_id, "type":ParticipantType.PILOT.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_pilot_berth_etd(shipcall=shipcall, df_times=df_times) # 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)" return def test_validation_rule_fct_missing_time_tug_berth_eta__shipcall_soon_but_participant_estimated_time_undefined(build_sql_proxy_connection): """0001-J validation_rule_fct_missing_time_tug_berth_eta""" vr = build_sql_proxy_connection['vr'] shipcall = get_shipcall_simple() shipcall.type = ShipcallType.INCOMING.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, "eta_berth"] = datetime.datetime.now() + datetime.timedelta(minutes=ParticipantwiseTimeDelta.TUG-10) # set times agency to be undetermined df_times.loc[df_times["participant_type"]==ParticipantType.TUG.value, "eta_berth"] = None # 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] tug_participant_id = df_times.loc[df_times["participant_type"]==ParticipantType.TUG.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":tug_participant_id, "type":ParticipantType.TUG.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_tug_berth_eta(shipcall=shipcall, df_times=df_times) # 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)" return def test_validation_rule_fct_missing_time_tug_berth_etd__shipcall_soon_but_participant_estimated_time_undefined(build_sql_proxy_connection): """0001-K validation_rule_fct_missing_time_tug_berth_etd""" vr = build_sql_proxy_connection['vr'] 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.TUG-10) # set times agency to be undetermined df_times.loc[df_times["participant_type"]==ParticipantType.TUG.value, "etd_berth"] = None # 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] tug_participant_id = df_times.loc[df_times["participant_type"]==ParticipantType.TUG.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":tug_participant_id, "type":ParticipantType.TUG.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_tug_berth_etd(shipcall=shipcall, df_times=df_times) # 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)" return 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'] 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.YELLOW, f"function should return 'yellow', because the participant did not provide a time and the shipcall takes place soon (according to the agency)" return 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'] 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.YELLOW, f"function should return 'yellow', because the participant did not provide a time and the shipcall takes place soon (according to the agency)" return def test_validation_rule_fct_shipcall_incoming_participants_disagree_on_eta__participants_disagree_on_time_but_different_shipcall_type(build_sql_proxy_connection): """0002-A validation_rule_fct_shipcall_incoming_participants_disagree_on_eta""" vr = build_sql_proxy_connection['vr'] shipcall = get_shipcall_simple() # set shipcall type to NOT match the function -> returns 'green' query = "eta_berth" shipcall.type = ShipcallType.SHIFTING.value df_times = get_df_times_participants_disagree(query=query, shipcall=shipcall) # apply the validation rule (state, msg) = vr.validation_rule_fct_shipcall_incoming_participants_disagree_on_eta(shipcall=shipcall, df_times=df_times) # expectation: green state, no msg assert state==StatusFlags.GREEN, f"function should return 'green', because the shipcall type does not match the function" return def test_validation_rule_fct_shipcall_incoming_participants_disagree_on_eta__participants_disagree_on_time(build_sql_proxy_connection): """0002-A validation_rule_fct_shipcall_incoming_participants_disagree_on_eta""" vr = build_sql_proxy_connection['vr'] shipcall = get_shipcall_simple() # set shipcall type to match the function query = "eta_berth" shipcall.type = ShipcallType.INCOMING.value df_times = get_df_times_participants_disagree(query=query, shipcall=shipcall) # makes sure that there is disagreement among participants # apply the validation rule (state, msg) = vr.validation_rule_fct_shipcall_incoming_participants_disagree_on_eta(shipcall=shipcall, df_times=df_times) # expectation: green state, no msg assert state==StatusFlags.RED, f"function should return 'red', because agency and pilot disagree on the query" return def test_validation_rule_fct_shipcall_incoming_participants_disagree_on_eta__participants_agree_on_time(build_sql_proxy_connection): """0002-A validation_rule_fct_shipcall_incoming_participants_disagree_on_eta""" vr = build_sql_proxy_connection['vr'] shipcall = get_shipcall_simple() # set shipcall type to NOT match the function -> returns 'green' query = "eta_berth" shipcall.type = ShipcallType.INCOMING.value df_times = get_df_times(shipcall) df_times.loc[:,query] = datetime.datetime.now() # apply the validation rule (state, msg) = vr.validation_rule_fct_shipcall_incoming_participants_disagree_on_eta(shipcall=shipcall, df_times=df_times) # expectation: green state, no msg assert state==StatusFlags.GREEN, f"function should return 'green', because the participants fully agree on the time" return def test_validation_rule_fct_shipcall_outgoing_participants_disagree_on_etd__participants_disagree_on_time_but_different_shipcall_type(build_sql_proxy_connection): """0002-B validation_rule_fct_shipcall_outgoing_participants_disagree_on_etd""" vr = build_sql_proxy_connection['vr'] shipcall = get_shipcall_simple() # set shipcall type to NOT match the function -> returns 'green' query = "etd_berth" shipcall.type = ShipcallType.SHIFTING.value df_times = get_df_times_participants_disagree(query=query, shipcall=shipcall) # apply the validation rule (state, msg) = vr.validation_rule_fct_shipcall_outgoing_participants_disagree_on_etd(shipcall=shipcall, df_times=df_times) # expectation: green state, no msg assert state==StatusFlags.GREEN, f"function should return 'green', because the shipcall type does not match the function" return def test_validation_rule_fct_shipcall_outgoing_participants_disagree_on_etd__participants_disagree_on_time(build_sql_proxy_connection): """0002-B validation_rule_fct_shipcall_outgoing_participants_disagree_on_etd""" vr = build_sql_proxy_connection['vr'] shipcall = get_shipcall_simple() # set shipcall type to match the function query = "etd_berth" shipcall.type = ShipcallType.OUTGOING.value df_times = get_df_times_participants_disagree(query=query, shipcall=shipcall) # makes sure that there is disagreement among participants # apply the validation rule (state, msg) = vr.validation_rule_fct_shipcall_outgoing_participants_disagree_on_etd(shipcall=shipcall, df_times=df_times) # expectation: green state, no msg assert state==StatusFlags.RED, f"function should return 'red', because agency and pilot disagree on the query" return def test_validation_rule_fct_shipcall_outgoing_participants_disagree_on_etd__participants_agree_on_time(build_sql_proxy_connection): """0002-B validation_rule_fct_shipcall_outgoing_participants_disagree_on_etd""" vr = build_sql_proxy_connection['vr'] shipcall = get_shipcall_simple() # set shipcall type to NOT match the function -> returns 'green' query = "etd_berth" shipcall.type = ShipcallType.OUTGOING.value df_times = get_df_times(shipcall) df_times.loc[:,query] = datetime.datetime.now() # apply the validation rule (state, msg) = vr.validation_rule_fct_shipcall_outgoing_participants_disagree_on_etd(shipcall=shipcall, df_times=df_times) # expectation: green state, no msg assert state==StatusFlags.GREEN, f"function should return 'green', because the participants fully agree on the time" return def test_validation_rule_fct_shipcall_shifting_participants_disagree_on_eta_or_etd__participants_disagree_on_time_but_different_shipcall_type(build_sql_proxy_connection): """0002-C validation_rule_fct_shipcall_shifting_participants_disagree_on_eta_or_etd""" vr = build_sql_proxy_connection['vr'] shipcall = get_shipcall_simple() shipcall.type = ShipcallType.INCOMING.value df_times = get_df_times(shipcall) # set shipcall type to match the function query = "eta_berth" df_times = get_df_times_participants_disagree(query=query, shipcall=shipcall, df_times=df_times) # makes sure that there is disagreement among participants # set shipcall type to match the function query = "etd_berth" df_times = get_df_times_participants_disagree(query=query, shipcall=shipcall, df_times=df_times) # makes sure that there is disagreement among participants # apply the validation rule (state, msg) = vr.validation_rule_fct_shipcall_shifting_participants_disagree_on_eta_or_etd(shipcall=shipcall, df_times=df_times) # expectation: green state, no msg assert state==StatusFlags.GREEN, f"function should return 'green', because the shipcall type does not match the function" return def test_validation_rule_fct_shipcall_shifting_participants_disagree_on_eta_or_etd__participants_disagree_on_time(build_sql_proxy_connection): """0002-C validation_rule_fct_shipcall_shifting_participants_disagree_on_eta_or_etd""" vr = build_sql_proxy_connection['vr'] shipcall = get_shipcall_simple() shipcall.type = ShipcallType.SHIFTING.value df_times = get_df_times(shipcall) # set shipcall type to match the function query = "eta_berth" df_times = get_df_times_participants_disagree(query=query, shipcall=shipcall, df_times=df_times) # makes sure that there is disagreement among participants # set shipcall type to match the function query = "etd_berth" df_times = get_df_times_participants_disagree(query=query, shipcall=shipcall, df_times=df_times) # makes sure that there is disagreement among participants # apply the validation rule (state, msg) = vr.validation_rule_fct_shipcall_shifting_participants_disagree_on_eta_or_etd(shipcall=shipcall, df_times=df_times) # expectation: green state, no msg assert state==StatusFlags.RED, f"function should return 'red', because agency and pilot disagree on the query" return def test_validation_rule_fct_shipcall_shifting_participants_disagree_on_eta_or_etd__participants_agree_on_time(build_sql_proxy_connection): """0002-C validation_rule_fct_shipcall_shifting_participants_disagree_on_eta_or_etd""" vr = build_sql_proxy_connection['vr'] shipcall = get_shipcall_simple() shipcall.type = ShipcallType.SHIFTING.value df_times = get_df_times(shipcall) # set shipcall type to match the function query = "eta_berth" df_times.loc[:,query] = datetime.datetime.now() # set shipcall type to match the function query = "etd_berth" df_times.loc[:,query] = datetime.datetime.now() # apply the validation rule (state, msg) = vr.validation_rule_fct_shipcall_shifting_participants_disagree_on_eta_or_etd(shipcall=shipcall, df_times=df_times) # expectation: green state, no msg assert state==StatusFlags.GREEN, f"function should return 'green', because the participants fully agree on the time" return 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'] 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.RED, f"status flag should be 'red', because the planned operations start is BEFORE the estimated time of arrival for the shipcall" 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'] 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.RED, f"status flag should be 'red', because the planned operations end is AFTER the estimated time of departure for the shipcall" 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): """ 0003-A validation_rule_fct_eta_time_not_in_operation_window 0003-B validation_rule_fct_etd_time_not_in_operation_window """ vr = build_sql_proxy_connection['vr'] import random shipcall = get_shipcall_simple() shipcall.type = ShipcallType.SHIFTING.value df_times = get_df_times(shipcall) t0_time = datetime.datetime.now() # 10 random permutations of None/pd.NaT/suitable values # each of these combinations is okay and should return a 'green' state for _i in range(10): # eta_berth & operations start df_times.loc[df_times["participant_type"]==ParticipantType.AGENCY.value, "eta_berth"] = random.sample([None, pd.NaT, t0_time + datetime.timedelta(minutes=0)],k=1)[0] df_times.loc[df_times["participant_type"]==ParticipantType.TERMINAL.value, "operations_start"] = random.sample([None, pd.NaT, t0_time + datetime.timedelta(minutes=0)], k=1)[0] # etd_berth & operations start df_times.loc[df_times["participant_type"]==ParticipantType.AGENCY.value, "etd_berth"] = random.sample([None, pd.NaT, t0_time + datetime.timedelta(hours=1)],k=1)[0] df_times.loc[df_times["participant_type"]==ParticipantType.TERMINAL.value, "operations_end"] = random.sample([None, pd.NaT, t0_time+datetime.timedelta(hours=1)],k=1)[0] (code, msg) = vr.validation_rule_fct_eta_time_not_in_operation_window(shipcall=shipcall, df_times=df_times) assert code==StatusFlags.GREEN, f"status flag should be 'green', as any of these perturbations sets operation & estimated time to be on par ot one the values missed" (code, msg) = vr.validation_rule_fct_etd_time_not_in_operation_window(shipcall=shipcall, df_times=df_times) assert code==StatusFlags.GREEN, f"status flag should be 'green', as any of these perturbations sets operation & estimated time to be on par ot one the values missed" return def test_validation_rule_fct_eta_time_not_in_tidal_window__is_okay(build_sql_proxy_connection): """0004-A validation_rule_fct_eta_time_not_in_tidal_window""" vr = build_sql_proxy_connection['vr'] shipcall = get_shipcall_simple() df_times = get_df_times(shipcall) t0_time = datetime.datetime.now() # tidal window: [t0 +1min, t0 +1hr) # eta berth: shipcall.tidal_window_from = t0_time + datetime.timedelta(minutes=1) shipcall.tidal_window_to = t0_time + datetime.timedelta(hours=1) df_times.loc[df_times["participant_type"]==ParticipantType.AGENCY.value, "eta_berth"] = t0_time + datetime.timedelta(minutes=1) (code, msg) = vr.validation_rule_fct_eta_time_not_in_tidal_window(shipcall, df_times) assert code==StatusFlags.GREEN, f"state should be 'green', because eta_berth matches precisely the tidal_window_from" df_times.loc[df_times["participant_type"]==ParticipantType.AGENCY.value, "eta_berth"] = t0_time + datetime.timedelta(hours=1) (code, msg) = vr.validation_rule_fct_eta_time_not_in_tidal_window(shipcall, df_times) assert code==StatusFlags.GREEN, f"state should be 'green', because eta_berth matches precisely the tidal_window_to (in this case, the etd_berth would likely cause an issue)" return def test_validation_rule_fct_eta_time_not_in_tidal_window__eta_outside_tidal_window(build_sql_proxy_connection): """0004-A validation_rule_fct_eta_time_not_in_tidal_window""" vr = build_sql_proxy_connection['vr'] shipcall = get_shipcall_simple() df_times = get_df_times(shipcall) t0_time = datetime.datetime.now() # tidal window: [t0 +1min, t0 +1hr) # eta berth: t0+0min shipcall.tidal_window_from = t0_time + datetime.timedelta(minutes=1) shipcall.tidal_window_to = t0_time + datetime.timedelta(hours=1) df_times.loc[df_times["participant_type"]==ParticipantType.AGENCY.value, "eta_berth"] = t0_time + datetime.timedelta(minutes=0) (code, msg) = vr.validation_rule_fct_eta_time_not_in_tidal_window(shipcall, df_times) assert code==StatusFlags.RED, f"state should be 'red', eta_berth takes place before the tidal window" df_times.loc[df_times["participant_type"]==ParticipantType.AGENCY.value, "eta_berth"] = t0_time + datetime.timedelta(hours=1, minutes=1) (code, msg) = vr.validation_rule_fct_eta_time_not_in_tidal_window(shipcall, df_times) assert code==StatusFlags.RED, f"state should be 'red', eta_berth takes place after the tidal window" return def test_validation_rule_fct_etd_time_not_in_tidal_window__is_okay(build_sql_proxy_connection): """0004-B validation_rule_fct_etd_time_not_in_tidal_window""" vr = build_sql_proxy_connection['vr'] shipcall = get_shipcall_simple() df_times = get_df_times(shipcall) t0_time = datetime.datetime.now() # tidal window: [t0 +1min, t0 +1hr) # etd berth: shipcall.tidal_window_from = t0_time + datetime.timedelta(minutes=1) shipcall.tidal_window_to = t0_time + datetime.timedelta(hours=1) df_times.loc[df_times["participant_type"]==ParticipantType.AGENCY.value, "etd_berth"] = t0_time + datetime.timedelta(minutes=1) (code, msg) = vr.validation_rule_fct_etd_time_not_in_tidal_window(shipcall, df_times) assert code==StatusFlags.GREEN, f"state should be 'green', because etd_berth matches precisely the tidal_window_from" df_times.loc[df_times["participant_type"]==ParticipantType.AGENCY.value, "etd_berth"] = t0_time + datetime.timedelta(hours=1) (code, msg) = vr.validation_rule_fct_etd_time_not_in_tidal_window(shipcall, df_times) assert code==StatusFlags.GREEN, f"state should be 'green', because etd_berth matches precisely the tidal_window_to (in this case, the etd_berth would likely cause an issue)" return def test_validation_rule_fct_etd_time_not_in_tidal_window__etd_outside_tidal_window(build_sql_proxy_connection): """0004-B validation_rule_fct_etd_time_not_in_tidal_window""" vr = build_sql_proxy_connection['vr'] shipcall = get_shipcall_simple() shipcall.type = ShipcallType.OUTGOING.value df_times = get_df_times(shipcall) t0_time = datetime.datetime.now() # tidal window: [t0 +1min, t0 +1hr) # etd berth: shipcall.tidal_window_from = t0_time + datetime.timedelta(minutes=1) shipcall.tidal_window_to = t0_time + datetime.timedelta(hours=1) df_times.loc[df_times["participant_type"]==ParticipantType.AGENCY.value, "etd_berth"] = t0_time + datetime.timedelta(minutes=0) (code, msg) = vr.validation_rule_fct_etd_time_not_in_tidal_window(shipcall, df_times) assert code==StatusFlags.RED, f"state should be 'red', etd_berth takes place before the tidal window" df_times.loc[df_times["participant_type"]==ParticipantType.AGENCY.value, "etd_berth"] = t0_time + datetime.timedelta(hours=1, minutes=1) (code, msg) = vr.validation_rule_fct_etd_time_not_in_tidal_window(shipcall, df_times) assert code==StatusFlags.RED, f"state should be 'red', etd_berth takes place after the tidal window" return def test_validation_rule_fct_tidal_window_is_precisely_eta_or_etd__is_okay(build_sql_proxy_connection): """0004-A & 0004-B eta/etd is exactly tidal window: no violation""" vr = build_sql_proxy_connection['vr'] shipcall = get_shipcall_simple() df_times = get_df_times(shipcall) times_agency = vr.sql_handler.get_times_for_participant_type(df_times, participant_type=ParticipantType.AGENCY.value) # tidal window: [t0 +1min, t0 +1hr) # etd berth: t0_time = datetime.datetime.now() times_agency.eta_berth = t0_time shipcall.tidal_window_from = t0_time shipcall.tidal_window_to = t0_time+datetime.timedelta(minutes=3) violation_state = not vr.time_logic.time_inbetween( query_time=times_agency.eta_berth, start_time=shipcall.tidal_window_from, end_time=shipcall.tidal_window_to ) assert violation_state==False t0_time = datetime.datetime.now() times_agency.etd_berth = t0_time shipcall.tidal_window_from = t0_time+datetime.timedelta(minutes=-3) shipcall.tidal_window_to = t0_time violation_state = not vr.time_logic.time_inbetween( query_time=times_agency.etd_berth, start_time=shipcall.tidal_window_from, end_time=shipcall.tidal_window_to ) assert violation_state==False return def test_validation_rule_fct_too_many_identical_eta_times__is_violated_by_too_many_identical_times(build_sql_proxy_connection): """0005-A validation_rule_fct_too_many_identical_eta_times""" vr = build_sql_proxy_connection['vr'] query = "eta_berth" reference_time = pd.Timestamp(datetime.datetime.now()) reference_time = reference_time.round("min") shipcall = get_shipcall_simple() df_times = get_df_times(shipcall) df_times = df_times.loc[df_times["participant_type"]==ParticipantType.AGENCY.value] df_times.loc[:,query] = reference_time + datetime.timedelta(seconds=12) all_times_df = build_stub_df_times(shipcall, query, reference_time) (code, msg) = vr.validation_rule_fct_too_many_identical_eta_times(shipcall=shipcall, df_times=df_times, all_times_agency=all_times_df) assert code == StatusFlags.YELLOW, f"status should be 'yellow', because the artificial 'all_times_df' contains five shipcalls with identical times, which exceeds the threshold" return def test_validation_rule_fct_too_many_identical_etd_times__is_violated_by_too_many_identical_times(build_sql_proxy_connection): """0005-B validation_rule_fct_too_many_identical_etd_times""" vr = build_sql_proxy_connection['vr'] query = "etd_berth" reference_time = pd.Timestamp(datetime.datetime.now()) reference_time = reference_time.round("min") shipcall = get_shipcall_simple() df_times = get_df_times(shipcall) df_times = df_times.loc[df_times["participant_type"]==ParticipantType.AGENCY.value] df_times.loc[:,query] = reference_time + datetime.timedelta(seconds=12) all_times_df = build_stub_df_times(shipcall, query, reference_time) (code, msg) = vr.validation_rule_fct_too_many_identical_etd_times(shipcall=shipcall, df_times=df_times, all_times_agency=all_times_df) assert code == StatusFlags.YELLOW, f"status should be 'yellow', because the artificial 'all_times_df' contains five shipcalls with identical times, which exceeds the threshold" return def test_validation_rule_fct_agency_and_terminal_berth_id_disagreement__agency_and_terminal_agree(build_sql_proxy_connection): """0006-A validation_rule_fct_agency_and_terminal_berth_id_disagreement""" vr = build_sql_proxy_connection['vr'] shipcall = get_shipcall_simple() df_times = get_df_times(shipcall) df_times.loc[df_times["participant_type"]==ParticipantType.AGENCY.value, "berth_id"] = 143 df_times.loc[df_times["participant_type"]==ParticipantType.TERMINAL.value, "berth_id"] = 143 (code, msg) = vr.validation_rule_fct_agency_and_terminal_berth_id_disagreement(shipcall=shipcall, df_times=df_times) assert code==StatusFlags.GREEN, f"status should be 'green', because agency and terminal agree on the selected berth id" return def test_validation_rule_fct_agency_and_terminal_berth_id_disagreement__agency_and_terminal_disagree(build_sql_proxy_connection): """0006-A validation_rule_fct_agency_and_terminal_berth_id_disagreement""" vr = build_sql_proxy_connection['vr'] shipcall = get_shipcall_simple() df_times = get_df_times(shipcall) df_times.loc[df_times["participant_type"]==ParticipantType.AGENCY.value, "berth_id"] = 143 df_times.loc[df_times["participant_type"]==ParticipantType.TERMINAL.value, "berth_id"] = 145 (code, msg) = vr.validation_rule_fct_agency_and_terminal_berth_id_disagreement(shipcall=shipcall, df_times=df_times) assert code==StatusFlags.YELLOW, f"status should be 'yellow', because agency and terminal do not agree on the selected berth id" return def test_validation_rule_fct_agency_and_terminal_pier_side_disagreement__agency_and_terminal_agree(build_sql_proxy_connection): """0006-B validation_rule_fct_agency_and_terminal_pier_side_disagreement""" vr = build_sql_proxy_connection['vr'] shipcall = get_shipcall_simple() df_times = get_df_times(shipcall) shipcall.pier_side = True # df_times.loc[df_times["participant_type"]==ParticipantType.AGENCY.value, "pier_side"] = True df_times.loc[df_times["participant_type"]==ParticipantType.TERMINAL.value, "pier_side"] = True (code, msg) = vr.validation_rule_fct_agency_and_terminal_pier_side_disagreement(shipcall=shipcall, df_times=df_times) assert code==StatusFlags.GREEN, f"status should be 'green', because agency and terminal agree on the selected pier side" return def test_validation_rule_fct_agency_and_terminal_pier_side_disagreement__agency_and_terminal_disagree(build_sql_proxy_connection): """0006-B validation_rule_fct_agency_and_terminal_pier_side_disagreement""" vr = build_sql_proxy_connection['vr'] shipcall = get_shipcall_simple() df_times = get_df_times(shipcall) shipcall.pier_side = True #df_times.loc[df_times["participant_type"]==ParticipantType.AGENCY.value, "pier_side"] = True df_times.loc[df_times["participant_type"]==ParticipantType.TERMINAL.value, "pier_side"] = False (code, msg) = vr.validation_rule_fct_agency_and_terminal_pier_side_disagreement(shipcall=shipcall, df_times=df_times) assert code==StatusFlags.YELLOW, f"status should be 'yellow', because agency and terminal do not agree on the selected pier side" return def test_validation_rule_fct_agency_and_terminal_pier_side_disagreement__agency_and_terminal_disagree_terminal_is_none(build_sql_proxy_connection): """0006-B validation_rule_fct_agency_and_terminal_pier_side_disagreement""" vr = build_sql_proxy_connection['vr'] shipcall = get_shipcall_simple() df_times = get_df_times(shipcall) shipcall.pier_side = True # df_times.loc[df_times["participant_type"]==ParticipantType.AGENCY.value, "pier_side"] = True df_times.loc[df_times["participant_type"]==ParticipantType.TERMINAL.value, "pier_side"] = None (code, msg) = vr.validation_rule_fct_agency_and_terminal_pier_side_disagreement(shipcall=shipcall, df_times=df_times) assert code==StatusFlags.GREEN, f"status should be 'green', because the terminal's pier_side is not provided" return def test_validation_rule_fct_agency_and_terminal_pier_side_disagreement__agency_and_terminal_disagree_terminal_is_nan(build_sql_proxy_connection): """0006-B validation_rule_fct_agency_and_terminal_pier_side_disagreement""" vr = build_sql_proxy_connection['vr'] shipcall = get_shipcall_simple() df_times = get_df_times(shipcall) df_times.loc[df_times["participant_type"]==ParticipantType.AGENCY.value, "pier_side"] = True df_times.loc[df_times["participant_type"]==ParticipantType.TERMINAL.value, "pier_side"] = float("nan") (code, msg) = vr.validation_rule_fct_agency_and_terminal_pier_side_disagreement(shipcall=shipcall, df_times=df_times) assert code==StatusFlags.GREEN, f"status should be 'green', because the terminal's pier_side is not provided" return def test_validation_rule_fct_agency_and_terminal_pier_side_agreement(build_sql_proxy_connection): """#0006-A validation_rule_fct_agency_and_terminal_pier_side_disagreement""" import pandas as pd from BreCal.stubs.times_full import get_times_full_simple from BreCal.stubs.shipcall import get_shipcall_simple from BreCal.database.enums import ParticipantType from BreCal.database.enums import StatusFlags vr = build_sql_proxy_connection["vr"] shipcall = get_shipcall_simple() t1 = get_times_full_simple() t2 = get_times_full_simple() # roles: agency & terminal t1.participant_type = ParticipantType.AGENCY.value t2.participant_type = ParticipantType.TERMINAL.value # agreement shipcall.pier_side = True # t1.pier_side = True t2.pier_side = True time_objects = [t1, t2] df_times = pd.DataFrame.from_records([to_.__dict__ for to_ in time_objects]) df_times.set_index('id',inplace=True) (state, description) = vr.validation_rule_fct_agency_and_terminal_pier_side_disagreement(shipcall, df_times) assert state.value == StatusFlags.GREEN.value, f"no violation should be observed" assert description is None, f"no violation should be observed" return def test_validation_rule_fct_agency_and_terminal_pier_side_disagreement(build_sql_proxy_connection): """#0006-A validation_rule_fct_agency_and_terminal_pier_side_disagreement""" import pandas as pd from BreCal.stubs.times_full import get_times_full_simple from BreCal.stubs.shipcall import get_shipcall_simple from BreCal.database.enums import ParticipantType from BreCal.database.enums import StatusFlags vr = build_sql_proxy_connection["vr"] shipcall = get_shipcall_simple() t1 = get_times_full_simple() t2 = get_times_full_simple() # roles: agency & terminal t1.participant_type = ParticipantType.AGENCY.value t2.participant_type = ParticipantType.TERMINAL.value # disagreement shipcall.pier_side = True # t1.pier_side = True t2.pier_side = False time_objects = [t1, t2] df_times = pd.DataFrame.from_records([to_.__dict__ for to_ in time_objects]) df_times.set_index('id',inplace=True) (state, description) = vr.validation_rule_fct_agency_and_terminal_pier_side_disagreement(shipcall, df_times) assert state.value > StatusFlags.GREEN.value, f"a violation must be identified" assert description is not None, f"a violation description must be identified" return def test_validation_rule_fct_agency_and_terminal_berth_id_disagreement(build_sql_proxy_connection): """#0006-B validation_rule_fct_agency_and_terminal_pier_side_disagreement""" import pandas as pd from BreCal.stubs.times_full import get_times_full_simple from BreCal.stubs.shipcall import get_shipcall_simple from BreCal.database.enums import ParticipantType from BreCal.database.enums import StatusFlags vr = build_sql_proxy_connection["vr"] shipcall = get_shipcall_simple() t1 = get_times_full_simple() t2 = get_times_full_simple() # roles: agency & terminal t1.participant_type = ParticipantType.AGENCY.value t2.participant_type = ParticipantType.TERMINAL.value # disagreement t1.berth_id = 1 t2.berth_id = 2 time_objects = [t1, t2] df_times = pd.DataFrame.from_records([to_.__dict__ for to_ in time_objects]) df_times.set_index('id',inplace=True) (state, description) = vr.validation_rule_fct_agency_and_terminal_berth_id_disagreement(shipcall, df_times) assert state.value > StatusFlags.GREEN.value, f"a violation must be identified" assert description is not None, f"a violation description must be identified" return def test_validation_rule_fct_agency_and_terminal_berth_id_agreement(build_sql_proxy_connection): """#0006-B validation_rule_fct_agency_and_terminal_pier_side_disagreement""" import pandas as pd from BreCal.stubs.times_full import get_times_full_simple from BreCal.stubs.shipcall import get_shipcall_simple from BreCal.database.enums import ParticipantType from BreCal.database.enums import StatusFlags vr = build_sql_proxy_connection["vr"] shipcall = get_shipcall_simple() t1 = get_times_full_simple() t2 = get_times_full_simple() # roles: agency & terminal t1.participant_type = ParticipantType.AGENCY.value t2.participant_type = ParticipantType.TERMINAL.value # agreement t1.berth_id = 21 t2.berth_id = 21 time_objects = [t1, t2] df_times = pd.DataFrame.from_records([to_.__dict__ for to_ in time_objects]) df_times.set_index('id',inplace=True) (state, description) = vr.validation_rule_fct_agency_and_terminal_berth_id_disagreement(shipcall, df_times) assert state.value == StatusFlags.GREEN.value, f"no violation should be observed" assert description is None, f"no violation should be observed" return def test_all_validation_rule_fcts_have_a_description(): from BreCal.validators.validation_rule_functions import error_message_dict, ValidationRuleFunctions import types vr = ValidationRuleFunctions(sql_handler=None) assert all( [mthd_ in list(error_message_dict.keys()) for mthd_ in dir(vr) if ('validation_rule_fct' in mthd_)] ), f"one of the validation_rule_fcts is currently not defined in the error_message_dict and will create cryptic descriptions! Please add it to the error_message_dict BreCal.validators.validation_rule_functions" return