diff --git a/.gitignore b/.gitignore index 67e970b..790e347 100644 --- a/.gitignore +++ b/.gitignore @@ -442,5 +442,8 @@ src/notebooks_metz docs/traffic_light_examples **/.~lock* misc/berths_and_terminals.csv +misc/mysql-workbench-community_8.0.34-1ubuntu22.04_amd64.deb times.md +Ampelfunktion.md + diff --git a/src/lib_brecal_utils/brecal_utils/request_status_code.py b/src/lib_brecal_utils/brecal_utils/request_status_code.py new file mode 100644 index 0000000..c3d72ea --- /dev/null +++ b/src/lib_brecal_utils/brecal_utils/request_status_code.py @@ -0,0 +1,131 @@ +import json +from abc import ABC, abstractmethod +from BreCal.schemas.model import obj_dict + + +"""implementation of default objects for http request codes. this enforces standardized outputs in the (response, code, headers)-style""" + +def get_request_code(code_id): + """convenience function, which returns the desired request code object""" + request_code_dict = { + 200:RequestCode_HTTP_200_OK, + 201:RequestCode_HTTP_201_CREATED, + 400:RequestCode_HTTP_400_BAD_REQUEST, + 403:RequestCode_HTTP_403_FORBIDDEN, + 404:RequestCode_HTTP_404_NOT_FOUND, + 500:RequestCode_HTTP_500_INTERNAL_SERVER_ERROR + } + + assert code_id in list(request_code_dict.keys()), f"unsupported request code: {code_id}. \nAvailable codes: {request_code_dict}" + return request_code_dict.get(code_id)() + +class RequestStatusCode(ABC): + def __init__(self): + return + + @abstractmethod + def __call__(self, data): + raise NotImplementedError("any default status code object must be callable") + + @abstractmethod + def status_code(self): + raise NotImplementedError("any default status code object should return an integer") + + @abstractmethod + def response(self, data): + raise NotImplementedError("the response method should return a binary json object. typically, json.dumps is used") + + + def headers(self): + return {'Content-Type': 'application/json; charset=utf-8'} + + +class RequestCode_HTTP_200_OK(RequestStatusCode): + def __init__(self) -> None: + super().__init__() + + def __call__(self, data): + return (self.response(data), self.status_code(), self.headers()) + + def status_code(self): + return 200 + + def response(self, data): + return json.dumps(data, default=obj_dict) + + +class RequestCode_HTTP_201_CREATED(RequestStatusCode): + def __init__(self) -> None: + super().__init__() + + def __call__(self, data): + return (self.response(data), self.status_code(), self.headers()) + + def status_code(self): + return 201 + + def response(self, new_id): + return json.dumps({"id":new_id}) + + +class RequestCode_HTTP_400_BAD_REQUEST(RequestStatusCode): + def __init__(self) -> None: + super().__init__() + + def __call__(self, data): + return (self.response(data), self.status_code(), self.headers()) + + def status_code(self): + return 400 + + def response(self, data): + return json.dumps(data) + + +class RequestCode_HTTP_403_FORBIDDEN(RequestStatusCode): + def __init__(self) -> None: + super().__init__() + + def __call__(self, data): + return (self.response(data), self.status_code(), self.headers()) + + def status_code(self): + return 403 + + def response(self, message="invalid credentials"): + result = {} + result["message"] = message + return json.dumps(result) + + +class RequestCode_HTTP_404_NOT_FOUND(RequestStatusCode): + def __init__(self) -> None: + super().__init__() + + def __call__(self, data): + return (self.response(data), self.status_code(), self.headers()) + + def status_code(self): + return 404 + + def response(self, message="no such record"): + result = {} + result["message"] = message + return json.dumps(result) + + +class RequestCode_HTTP_500_INTERNAL_SERVER_ERROR(RequestStatusCode): + def __init__(self) -> None: + super().__init__() + + def __call__(self, data): + return (self.response(data), self.status_code(), self.headers()) + + def status_code(self): + return 500 + + def response(self, message="credential lookup mismatch"): + result = {} + result["message"] = message + return json.dumps(result) + diff --git a/src/lib_brecal_utils/brecal_utils/stubs/times_tug.py b/src/lib_brecal_utils/brecal_utils/stubs/times_tug.py deleted file mode 100644 index e69de29..0000000 diff --git a/src/lib_brecal_utils/brecal_utils/stubs/tug.py b/src/lib_brecal_utils/brecal_utils/stubs/tug.py deleted file mode 100644 index bcaa425..0000000 --- a/src/lib_brecal_utils/brecal_utils/stubs/tug.py +++ /dev/null @@ -1,6 +0,0 @@ - - -def get_tug_simple(): - raise NotImplementedError("unclarified") - return - diff --git a/src/lib_brecal_utils/brecal_utils/validators/input_validation.py b/src/lib_brecal_utils/brecal_utils/validators/input_validation.py new file mode 100644 index 0000000..57200e8 --- /dev/null +++ b/src/lib_brecal_utils/brecal_utils/validators/input_validation.py @@ -0,0 +1,165 @@ + +####################################### InputValidation ####################################### + +from abc import ABC, abstractmethod +from BreCal.schemas.model import Ship, Shipcall, Berth, User, Participant + +class InputValidation(): + def __init__(self): + self.build_supported_models_dictionary() + return + + def build_supported_models_dictionary(self): + self.supported_models = { + Ship:ShipValidation(), + Shipcall:ShipcallValidation(), + Berth:BerthValidation(), + User:UserValidation(), + Participant:ParticipantValidation(), + } + return + + def assert_if_not_supported(self, dataclass_object): + assert type(dataclass_object) in self.supported_models, f"unsupported model. Found: {type(dataclass_object)}" + return + + def verify(self, dataclass_object): + self.assert_if_not_supported(dataclass_object) + + # determine the type of the dataclass object. The internal dictionary 'supported_models' matches the dataclass object + # to the respective validation protocol + validator = self.supported_models.get(type(dataclass_object)) + + # check the object based on the rules within the matched validator + input_validation_state = validator.check(dataclass_object) + return input_validation_state + + +class DataclassValidation(ABC): + """parent class of dataclas validators, which determines the outline of every object""" + def __init__(self): + return + + def check(self, dataclass_object) -> (list, bool): + """ + the 'check' method provides a default style, how each dataclass object is validated. It returns a list of violations + and a boolean, which determines, whether the check is passed successfully + """ + all_rules = self.apply_all_rules(dataclass_object) + violations = self.filter_violations(all_rules) + input_validation_state = self.evaluate(violations) + return (violations, input_validation_state) + + @abstractmethod + def apply_all_rules(self, dataclass_object) -> list: + """ + the 'apply_all_rules' method is mandatory for any dataclass validation object. It should execute all validation rules and + return a list of tuples, where each element is (output_boolean, validation_name) + """ + all_rules = [(True, 'blank_validation_rule')] + return all_rules + + def filter_violations(self, all_rules): + """input: all_rules, a list of tuples, where each element is (output, validation_name), which are (bool, str). """ + # if output is False, a violation is observed + violations = [result[1] for result in all_rules if not result[0]] + return violations + + def evaluate(self, violations) -> bool: + input_validation_state = len(violations)==0 + return input_validation_state + + + +class ShipcallValidation(DataclassValidation): + """an object that validates a Shipcall dataclass object""" + def __init__(self): + super().__init__() + return + + def apply_all_rules(self, dataclass_object) -> list: + """apply all input validation rules to determine, whether there are violations. returns a list of tuples (output, validation_name)""" + raise NotImplementedError() + return all_rules + + +from brecal_utils.validators.schema_validation import ship_bollard_pull_is_defined_or_is_not_tug, ship_bollard_pull_is_none_or_in_range, ship_callsign_len_is_seven_at_maximum, ship_eni_len_is_eight, ship_imo_len_is_seven, ship_length_in_range, ship_participant_id_is_defined_or_is_not_tug, ship_participant_id_is_none_or_int, ship_width_in_range +# skip: ship_max_draft_is_defined_or_is_not_tug, ship_max_draft_is_none_or_in_range, +class ShipValidation(DataclassValidation): + """an object that validates a Ship dataclass object""" + def __init__(self): + super().__init__() + return + + def apply_all_rules(self, dataclass_object) -> list: + """apply all input validation rules to determine, whether there are violations. returns a list of tuples (output, validation_name)""" + # skip: ship_max_draft_is_defined_or_is_not_tug, ship_max_draft_is_none_or_in_range, + """ + #TODO_ship_max_draft + with pytest.raises(AttributeError, match="'Ship' object has no attribute 'max_draft'"): + assert ship_max_draft_in_range(ship)[0], f"max draft of a ship must be between 0 and 20 meters" + assert ship_max_draft_is_none_or_in_range(ship)[0], f"the max_draft should either be undefined or between 0 and 20 meters" + """ + + # list comprehension: every function becomes part of the loop and will be executed. Each function is wrapped and provides (output, validation_name) + all_rules = [ + # tuple: (output, validation_name) + check_rule(dataclass_object) + + for check_rule in [ + ship_bollard_pull_is_defined_or_is_not_tug, + ship_bollard_pull_is_none_or_in_range, + ship_callsign_len_is_seven_at_maximum, + ship_eni_len_is_eight, + ship_imo_len_is_seven, + ship_length_in_range, + ship_participant_id_is_defined_or_is_not_tug, + ship_participant_id_is_none_or_int, + ship_width_in_range + ] + ] + return all_rules + +class BerthValidation(DataclassValidation): + """an object that validates a Berth dataclass object""" + def __init__(self): + super().__init__() + return + + def apply_all_rules(self, dataclass_object) -> list: + """apply all input validation rules to determine, whether there are violations. returns a list of tuples (output, validation_name)""" + raise NotImplementedError() + return all_rules + +class UserValidation(DataclassValidation): + """an object that validates a User dataclass object""" + def __init__(self): + super().__init__() + return + + def apply_all_rules(self, dataclass_object) -> list: + """apply all input validation rules to determine, whether there are violations. returns a list of tuples (output, validation_name)""" + raise NotImplementedError() + return all_rules + +from brecal_utils.validators.schema_validation import participant_postal_code_len_is_five +class ParticipantValidation(DataclassValidation): + """an object that validates a Participant dataclass object""" + def __init__(self): + super().__init__() + return + + def apply_all_rules(self, dataclass_object) -> list: + """apply all input validation rules to determine, whether there are violations. returns a list of tuples (output, validation_name)""" + + # list comprehension: every function becomes part of the loop and will be executed. Each function is wrapped and provides (output, validation_name) + all_rules = [ + # tuple: (output, validation_name) + check_rule(dataclass_object) + + for check_rule in [ + participant_postal_code_len_is_five, + ] + ] + return all_rules + diff --git a/src/lib_brecal_utils/brecal_utils/validators/schema_validation.py b/src/lib_brecal_utils/brecal_utils/validators/schema_validation.py index e47d59a..384b2f6 100644 --- a/src/lib_brecal_utils/brecal_utils/validators/schema_validation.py +++ b/src/lib_brecal_utils/brecal_utils/validators/schema_validation.py @@ -6,7 +6,7 @@ def validation_state_and_validation_name(validation_name): """ - can wrap arbitrary functions, so they returned (output, validation_name)-tuples + can wrap arbitrary functions, so they return (output, validation_name)-tuples usage example: @validation_state_and_validation_name("ship_eni_length") def validate_ship_eni_length(ship): @@ -31,6 +31,8 @@ def length_matches_exactly(query_value, length_value): return len(str(query_value)) == length_value + + ####################################### dataclass specifics ####################################### ### Ship dataclass (BreCal.schema.model.Ship) ### diff --git a/src/lib_brecal_utils/brecal_utils/validators/validation_rules.py b/src/lib_brecal_utils/brecal_utils/validators/validation_rules.py index a298552..d1e3a50 100644 --- a/src/lib_brecal_utils/brecal_utils/validators/validation_rules.py +++ b/src/lib_brecal_utils/brecal_utils/validators/validation_rules.py @@ -14,7 +14,7 @@ class ValidationRules(): self.times = times self.validation_state = self.determine_validation_state() - self.notification_state = self.determine_notification_state() + self.notification_state = self.determine_notification_state() # (state:str, should_notify:bool) return def determine_validation_state(self) -> str: @@ -37,7 +37,7 @@ class ValidationRules(): returns: notification_state_new (str), should_notify (bool) """ - state_new = self.undefined_method() # determien the successor + state_new = self.undefined_method() # determine the successor should_notify = self.identify_notification_state_change(state_new) self.notification_state = state_new # overwrite the predecessor return state_new, should_notify @@ -62,5 +62,7 @@ class ValidationRules(): return state_mapping[state_new] > state_mapping[state_old] def undefined_method(self) -> str: - return 'green' + """this function should apply the ValidationRules to the respective .shipcall, in regards to .times""" + # #TODO_traffic_state + return ('green', False) # (state:str, should_notify:bool) diff --git a/src/lib_brecal_utils/tests/stubs/test_stub_objects.py b/src/lib_brecal_utils/tests/stubs/test_stub_objects.py index 9054cce..ccd1655 100644 --- a/src/lib_brecal_utils/tests/stubs/test_stub_objects.py +++ b/src/lib_brecal_utils/tests/stubs/test_stub_objects.py @@ -28,15 +28,6 @@ def test_build_stub_ship(): assert isinstance(ship, Ship) return -def test_build_stub_tug(): - from brecal_utils.stubs.tug import get_tug_simple - with pytest.raises(ImportError): - from BreCal.schemas.model import Tug - with pytest.raises(NotImplementedError, match="unclarified"): - tug = get_tug_simple() - assert isinstance(tug, Tug) - return - def test_build_stub_shipcall(): from BreCal.schemas.model import Shipcall from brecal_utils.stubs.shipcall import get_shipcall_simple diff --git a/src/lib_brecal_utils/tests/validators/test_input_validation.py b/src/lib_brecal_utils/tests/validators/test_input_validation.py new file mode 100644 index 0000000..a962470 --- /dev/null +++ b/src/lib_brecal_utils/tests/validators/test_input_validation.py @@ -0,0 +1,63 @@ +import pytest + +@pytest.fixture() +def build_input_validation(): + from brecal_utils.validators.input_validation import InputValidation + iv = InputValidation() + return locals() + + +def test_build_input_validation(): + from brecal_utils.validators.input_validation import InputValidation + iv = InputValidation() + return + +def test_all_models_are_supported(build_input_validation): + iv = build_input_validation["iv"] + + from brecal_utils.stubs.ship import get_ship_simple + ship = get_ship_simple() + iv.assert_if_not_supported(ship) + + from brecal_utils.stubs.shipcall import get_shipcall_simple + shipcall = get_shipcall_simple() + iv.assert_if_not_supported(shipcall) + + from brecal_utils.stubs.berth import get_berth_simple + berth = get_berth_simple() + iv.assert_if_not_supported(berth) + + from brecal_utils.stubs.participant import get_participant_simple + participant = get_participant_simple() + iv.assert_if_not_supported(participant) + + from brecal_utils.stubs.user import get_user_simple + user = get_user_simple() + iv.assert_if_not_supported(user) + + # placeholder: how to handle times? + return + +def test_ship_input_validation(build_input_validation): + iv = build_input_validation["iv"] + + from brecal_utils.stubs.ship import get_ship_simple + ship = get_ship_simple() + violations, state = iv.verify(ship) + assert state, f"found violations: {violations}" + return + +def test_participant_input_validation(build_input_validation): + iv = build_input_validation["iv"] + + from brecal_utils.stubs.participant import get_participant_simple + participant = get_participant_simple() + violations, state = iv.verify(participant) + assert state, f"found violations: {violations}" + return + + + +if __name__=="__main__": + pass + diff --git a/src/lib_brecal_utils/tests/validators/test_schema_validation_participant.py b/src/lib_brecal_utils/tests/validators/test_schema_validation_participant.py index 2c75223..9c51f60 100644 --- a/src/lib_brecal_utils/tests/validators/test_schema_validation_participant.py +++ b/src/lib_brecal_utils/tests/validators/test_schema_validation_participant.py @@ -25,4 +25,4 @@ def test_participant_postal_code_len_is_six_should_assert(): if __name__=="__main__": test_participant_postal_code_len_is_five() test_participant_postal_code_len_is_six_should_assert() - + diff --git a/src/server/BreCal/schemas/model.py b/src/server/BreCal/schemas/model.py index 2b096c1..23d1c50 100644 --- a/src/server/BreCal/schemas/model.py +++ b/src/server/BreCal/schemas/model.py @@ -88,6 +88,8 @@ class ShipcallSchema(Schema): participants = fields.List(fields.Int) created = fields.DateTime(Required = False, allow_none=True) modified = fields.DateTime(Required = False, allow_none=True) + validation_state = fields.Str(Required = False, allow_none=True) + validation_state_changed = fields.DateTime(Required = False, allow_none=True) @dataclass class Shipcall: @@ -117,6 +119,8 @@ class Shipcall: canceled: bool created: datetime modified: datetime + validation_state: str + validation_state_changed: datetime participants: List[int] = field(default_factory=list) class ShipcallId(Schema):