adding stubs, validators and tests for these. Adaptation of data model 'shipcall' (validation_state and validation_state_changed). Adding request status codes (HTTP default format) and input validation. A tester for input validation is also prepared.

.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
+
 

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)
+    

src/lib_brecal_utils/brecal_utils/stubs/tug.py

@@ -1,6 +0,0 @@
-
-
-def get_tug_simple():
-    raise NotImplementedError("unclarified")
-    return
-

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
+    

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) ###

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)
 

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

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
+

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()
-    
+

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):