maintenance of API Input Validation (ship & times)

src/server/BreCal/database/sql_queries.py

@@ -266,6 +266,11 @@ class SQLQuery():
         query = "SELECT id, name, imo, callsign, participant_id, length, width, is_tug, bollard_pull, eni, created, modified, deleted FROM ship ORDER BY name"
         return query
     
+    @staticmethod
+    def get_ship_by_id()->str:
+        query = "SELECT * FROM ship where id = ?id?"
+        return query
+    
     @staticmethod
     def get_times()->str:
         query = "SELECT id, eta_berth, eta_berth_fixed, etd_berth, etd_berth_fixed, lock_time, lock_time_fixed, " + \

src/server/BreCal/stubs/times.py

@@ -0,0 +1,12 @@
+import datetime
+from BreCal.schemas import model
+from BreCal.schemas.model import ParticipantType
+
+def get_schema_model_stub_departure():
+    schemaModel = {'id': 0, 'eta_berth': None, 'eta_berth_fixed': None, 'etd_berth': datetime.datetime(2024, 9, 7, 15, 12, 58), 'etd_berth_fixed': None, 'lock_time': None, 'lock_time_fixed': None, 'zone_entry': None, 'zone_entry_fixed': None, 'operations_start': None, 'operations_end': None, 'remarks': 'test', 'participant_id': 10, 'berth_id': 146, 'berth_info': '', 'pier_side': None, 'shipcall_id': 115, 'participant_type': ParticipantType.AGENCY, 'ata': None, 'atd': None, 'eta_interval_end': None, 'etd_interval_end': None, 'created': None, 'modified': None}
+    return schemaModel
+
+def get_schema_model_stub_arrival():
+    schemaModel = {'id': 0, 'eta_berth': datetime.datetime(2024, 9, 7, 15, 12, 58), 'eta_berth_fixed': None, 'etd_berth': None, 'etd_berth_fixed': None, 'lock_time': None, 'lock_time_fixed': None, 'zone_entry': None, 'zone_entry_fixed': None, 'operations_start': None, 'operations_end': None, 'remarks': 'test', 'participant_id': 10, 'berth_id': 146, 'berth_info': '', 'pier_side': None, 'shipcall_id': 115, 'participant_type': ParticipantType.AGENCY, 'ata': None, 'atd': None, 'eta_interval_end': None, 'etd_interval_end': None, 'created': None, 'modified': None}
+    return schemaModel
+

src/server/BreCal/validators/input_validation_ship.py

@@ -6,6 +6,8 @@ from marshmallow import ValidationError
 from string import ascii_letters, digits
 
 from BreCal.schemas.model import Ship, Shipcall, Berth, User, Participant, ShipcallType
+from BreCal.database.sql_handler import execute_sql_query_standalone    
+from BreCal.database.sql_queries import SQLQuery
 from BreCal.impl.participant import GetParticipant
 from BreCal.impl.ships import GetShips
 from BreCal.impl.berths import GetBerths
@@ -47,14 +49,14 @@ class InputValidationShip():
         # 1.) Only users of type BSMD are allowed to PUT
         InputValidationShip.check_user_is_bsmd_type(user_data)
 
-        # 2.) The IMO number field may not be changed
+        # 2.) ID field is mandatory
+        InputValidationShip.content_contains_ship_id(content)
+
+        # 3.) The IMO number field may not be changed
         InputValidationShip.put_content_may_not_contain_imo_number(content)
 
-        # 3.) Check for reasonable Values (see BreCal.schemas.model.ShipSchema)
+        # 4.) Check for reasonable Values (see BreCal.schemas.model.ShipSchema)
         InputValidationShip.optionally_evaluate_bollard_pull_value(content)
-
-        # 4.) ID field is mandatory
-        InputValidationShip.content_contains_ship_id(content)
         return
     
     @staticmethod
@@ -101,11 +103,15 @@ class InputValidationShip():
     
     @staticmethod
     def put_content_may_not_contain_imo_number(content:dict):
+        # IMO is a required field, so it will never be None outside of tests. If so, there is no violation
         put_data_ship_imo = content.get("imo",None)
+        if put_data_ship_imo is None:
+            return
+        
+        # grab the ship by its ID and compare, whether the IMO is unchanged
+        ship = execute_sql_query_standalone(SQLQuery.get_ship_by_id(), param={"id":content.get("id")}, command_type="single", model=Ship)
 
-        # #TODO: Aktuelle IMO abfragen und nach Änderung suchen, bevor eine Fehlermeldung erstellt wird
-
-        if put_data_ship_imo is not None:
+        if put_data_ship_imo != ship.imo:
             raise ValidationError(f"The IMO number field may not be changed since it serves the purpose of a primary (matching) key.")
         return
     

src/server/BreCal/validators/input_validation_times.py

@@ -49,12 +49,12 @@ def build_post_data_type_dependent_required_fields_dict()->dict[ShipcallType,dic
         ShipcallType.shifting:{
             ParticipantType.undefined:[], # should not be set in POST requests
             ParticipantType.BSMD:[], # should not be set in POST requests
-            ParticipantType.TERMINAL:["operations_start", "operations_end"],
-            ParticipantType.AGENCY:["eta_berth", "etd_berth"],
-            ParticipantType.MOORING:["eta_berth", "etd_berth"],
-            ParticipantType.PILOT:["eta_berth", "etd_berth"],
-            ParticipantType.PORT_ADMINISTRATION:["eta_berth", "etd_berth"],
-            ParticipantType.TUG:["eta_berth", "etd_berth"],
+            ParticipantType.TERMINAL:["operations_start"],
+            ParticipantType.AGENCY:["etd_berth"],
+            ParticipantType.MOORING:["etd_berth"],
+            ParticipantType.PILOT:["etd_berth"],
+            ParticipantType.PORT_ADMINISTRATION:["etd_berth"],
+            ParticipantType.TUG:["etd_berth"],
         },
     }
     return post_data_type_dependent_required_fields_dict

src/server/tests/database/test_sql_queries.py

@@ -87,6 +87,13 @@ def test_sql_get_ships():
     assert all([isinstance(ship, model.Ship) for ship in ships])
     return
 
+def test_sql_get_ship_by_id():
+    query =  #"SELECT * FROM ship where id = ?id?" # #TODO_refactor: put into the SQLQuery object
+    ship = execute_sql_query_standalone(SQLQuery.get_ship_by_id(), param={"id":1}, command_type="single", model=model.Ship)
+    assert isinstance(ship, model.Ship)
+    return
+
+
 def test_sql_get_times():
     options = {'shipcall_id':153}
     times = execute_sql_query_standalone(query=SQLQuery.get_times(), model=model.Times, param={"scid" : options["shipcall_id"]})

src/server/tests/validators/test_input_validation_times.py

@@ -409,3 +409,60 @@ def test_input_validation_times_delete_request_fails_when_user_belongs_to_wrong_
     InputValidationTimes.check_user_belongs_to_same_group_as_dataset_determines(user_data, loadedModel=None, times_id=times_id, pdata=pdata)
     return
 
+def test_input_validation_times_check_dataset_values_for_time_intervals():
+    import datetime
+    from BreCal.schemas import model
+    from BreCal.validators.input_validation_times import InputValidationTimes
+    from BreCal.stubs.times import get_schema_model_stub_arrival, get_schema_model_stub_departure
+    from marshmallow import ValidationError
+
+    ### ETD (departure)
+    # expected to pass
+    schemaModel = get_schema_model_stub_departure()
+    schemaModel["etd_interval_end"] = schemaModel["etd_berth"] + datetime.timedelta(minutes=2)
+
+    schemaModel["etd_berth"] = schemaModel["etd_berth"].isoformat()
+    schemaModel["etd_interval_end"] = schemaModel["etd_interval_end"].isoformat()
+    content = schemaModel
+    loadedModel = model.TimesSchema().load(data=schemaModel, many=False, partial=True)
+
+    InputValidationTimes.check_dataset_values(user_data={}, loadedModel=loadedModel, content=content)
+
+    # expected to fail: the from-to-interval is incorrectly set.
+    schemaModel = get_schema_model_stub_departure()
+    schemaModel["etd_interval_end"] = schemaModel["etd_berth"] - datetime.timedelta(minutes=2)
+
+    schemaModel["etd_berth"] = schemaModel["etd_berth"].isoformat()
+    schemaModel["etd_interval_end"] = schemaModel["etd_interval_end"].isoformat()
+    content = schemaModel
+    loadedModel = model.TimesSchema().load(data=schemaModel, many=False, partial=True)
+
+    with pytest.raises(ValidationError, match="The provided time interval for the estimated departure time is invalid"):
+        InputValidationTimes.check_dataset_values(user_data={}, loadedModel=loadedModel, content=content)
+
+
+    ### ETA (arrival)
+    # expected to pass
+    schemaModel = get_schema_model_stub_arrival()
+    schemaModel["eta_interval_end"] = schemaModel["eta_berth"] + datetime.timedelta(minutes=2)
+
+    schemaModel["eta_berth"] = schemaModel["eta_berth"].isoformat()
+    schemaModel["eta_interval_end"] = schemaModel["eta_interval_end"].isoformat()
+    content = schemaModel
+    loadedModel = model.TimesSchema().load(data=schemaModel, many=False, partial=True)
+
+    InputValidationTimes.check_dataset_values(user_data={}, loadedModel=loadedModel, content=content)
+
+    # expected to fail: the from-to-interval is incorrectly set.
+    schemaModel = get_schema_model_stub_arrival()
+    schemaModel["eta_interval_end"] = schemaModel["eta_berth"] - datetime.timedelta(minutes=2)
+
+    schemaModel["eta_berth"] = schemaModel["eta_berth"].isoformat()
+    schemaModel["eta_interval_end"] = schemaModel["eta_interval_end"].isoformat()
+    content = schemaModel
+    loadedModel = model.TimesSchema().load(data=schemaModel, many=False, partial=True)
+
+    with pytest.raises(ValidationError, match="The provided time interval for the estimated arrival time is invalid"):
+        InputValidationTimes.check_dataset_values(user_data={}, loadedModel=loadedModel, content=content)
+    return
+