168 lines
6.6 KiB
Python
168 lines
6.6 KiB
Python
import pytest
|
|
from BreCal.validators.validation_rule_functions import ValidationRuleFunctions
|
|
from BreCal.validators.validation_rules import ValidationRules
|
|
from BreCal.database.sql_handler import SQLHandler
|
|
|
|
@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_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
|
|
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_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
|
|
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_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
|