If a PUT or DELETE operation is attempted on a non-existant object, 404 is returned
This commit is contained in:
parent
4531eda8f1
commit
0d7861ec36
@ -84,6 +84,9 @@ def PutShipcalls():
|
||||
# read the user data from the JWT token (set when login is performed)
|
||||
user_data = check_jwt()
|
||||
|
||||
if not InputValidationShipcall.exists_shipcall_by_id(loadedModel.get("id")):
|
||||
return create_dynamic_exception_response(ex=None, status_code=404, message="no shipcall found with the provided id")
|
||||
|
||||
# validate the PUT shipcall data and the user's authority
|
||||
InputValidationShipcall.evaluate_put_data(user_data, loadedModel, content)
|
||||
return impl.shipcalls.PutShipcalls(loadedModel)
|
||||
|
||||
@ -71,6 +71,9 @@ def PutShip():
|
||||
content = request.get_json(force=True)
|
||||
loadedModel = model.ShipSchema().load(data=content, many=False, partial=True, unknown=EXCLUDE)
|
||||
|
||||
if not InputValidationShip.exists_ship_by_dict(model=loadedModel):
|
||||
return create_dynamic_exception_response(ex=None, status_code=404, message="no ship found with the provided id")
|
||||
|
||||
# validate the request data & user permissions
|
||||
InputValidationShip.evaluate_put_data(user_data, loadedModel, content)
|
||||
return impl.ships.PutShip(loadedModel)
|
||||
@ -100,6 +103,10 @@ def DeleteShip():
|
||||
|
||||
# validate the request data & user permissions
|
||||
ship_id = request.args.get("id")
|
||||
|
||||
if not InputValidationShip.exists_ship_by_id(id=ship_id):
|
||||
return create_dynamic_exception_response(ex=None, status_code=404, message="no ship found with the provided id")
|
||||
|
||||
InputValidationShip.evaluate_delete_data(user_data, ship_id)
|
||||
return impl.ships.DeleteShip(options)
|
||||
|
||||
|
||||
@ -67,6 +67,9 @@ def PutTimes():
|
||||
# read the user data from the JWT token (set when login is performed)
|
||||
user_data = check_jwt()
|
||||
|
||||
if not InputValidationTimes.exists_times_by_id(loadedModel.get("id")):
|
||||
return create_dynamic_exception_response(ex=None, status_code=404, message="no times found with the provided id")
|
||||
|
||||
# validate the request
|
||||
InputValidationTimes.evaluate_put_data(user_data, loadedModel, content)
|
||||
return impl.times.PutTimes(loadedModel)
|
||||
@ -91,6 +94,9 @@ def DeleteTimes():
|
||||
# read the user data from the JWT token (set when login is performed)
|
||||
user_data = check_jwt()
|
||||
|
||||
if not InputValidationTimes.exists_times_by_id(options["id"]):
|
||||
return create_dynamic_exception_response(ex=None, status_code=404, message="no times found with the provided id")
|
||||
|
||||
# validate the request
|
||||
InputValidationTimes.evaluate_delete_data(user_data, times_id = request.args.get("id"))
|
||||
|
||||
|
||||
@ -4,19 +4,27 @@ import datetime
|
||||
def get_user_data_for_id(user_id:int, expiration_time:int=90):
|
||||
"""debugging function, which is useful to pull user_data from the database, which may be used to create stub data and unit tests"""
|
||||
query = "SELECT * FROM user where id = ?id?"
|
||||
pdata = execute_sql_query_standalone(query=query, param={"id":user_id})
|
||||
pdata = pdata[0] if len(pdata)>0 else None
|
||||
pdata = execute_sql_query_standalone(query=query, param={"id":user_id}, command_type="single_or_none")
|
||||
assert pdata is not None, f"could not find user with id {user_id}"
|
||||
|
||||
user_data = {k:v for k,v in pdata.items() if k in ['id','participant_id','first_name','last_name','user_name','user_phone','user_email']}
|
||||
user_data["exp"] = (datetime.datetime.now()+datetime.timedelta(minutes=expiration_time)).timestamp()
|
||||
return user_data
|
||||
|
||||
|
||||
def get_times_data_for_id(times_id:int):
|
||||
"""helper function to load previous times data from the database"""
|
||||
query = "SELECT * FROM times where id = ?id?"
|
||||
pdata = execute_sql_query_standalone(query=query, param={"id":times_id})
|
||||
pdata = pdata[0] if len(pdata)>0 else None
|
||||
assert pdata is not None, f"could not find times with id {times_id}"
|
||||
pdata = execute_sql_query_standalone(query=query, param={"id":times_id}, command_type="single_or_none")
|
||||
return pdata
|
||||
|
||||
def get_ship_data_for_id(ship_id:int):
|
||||
"""helper function to load previous ship data from the database"""
|
||||
query = "SELECT * FROM ship where id = ?id?"
|
||||
pdata = execute_sql_query_standalone(query=query, param={"id":ship_id}, command_type="single_or_none")
|
||||
return pdata
|
||||
|
||||
def get_shipcall_data_for_id(shipcall_id:int):
|
||||
"""helper function to load previous shipcall data from the database"""
|
||||
query = "SELECT * FROM shipcall where id = ?id?"
|
||||
pdata = execute_sql_query_standalone(query=query, param={"id":shipcall_id}, command_type="single_or_none")
|
||||
return pdata
|
||||
@ -8,6 +8,7 @@ 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.database.sql_utils import get_ship_data_for_id
|
||||
from BreCal.impl.participant import GetParticipant
|
||||
from BreCal.impl.ships import GetShips
|
||||
from BreCal.impl.berths import GetBerths
|
||||
@ -17,6 +18,7 @@ from BreCal.validators.input_validation_utils import check_if_user_is_bsmd_type,
|
||||
from BreCal.database.sql_handler import execute_sql_query_standalone
|
||||
from BreCal.validators.validation_base_utils import check_if_int_is_valid_flag
|
||||
from BreCal.validators.validation_base_utils import check_if_string_has_special_characters
|
||||
|
||||
import werkzeug
|
||||
|
||||
class InputValidationShip():
|
||||
@ -32,6 +34,17 @@ class InputValidationShip():
|
||||
def __init__(self) -> None:
|
||||
pass
|
||||
|
||||
@staticmethod
|
||||
def exists_ship_by_dict(model:dict):
|
||||
ship_id = model.get("id")
|
||||
if(ship_id is not None):
|
||||
return get_ship_data_for_id(ship_id) is not None
|
||||
return False
|
||||
|
||||
@staticmethod
|
||||
def exists_ship_by_id(id:int):
|
||||
return get_ship_data_for_id(id) is not None
|
||||
|
||||
@staticmethod
|
||||
def evaluate_post_data(user_data:dict, loadedModel:dict, content:dict):
|
||||
# 1.) Only users of type BSMD are allowed to POST
|
||||
|
||||
@ -11,6 +11,7 @@ from BreCal.impl.ships import GetShips
|
||||
from BreCal.impl.berths import GetBerths
|
||||
|
||||
from BreCal.database.enums import ParticipantType, ParticipantFlag
|
||||
from BreCal.database.sql_utils import get_shipcall_data_for_id
|
||||
from BreCal.validators.input_validation_utils import check_if_user_is_bsmd_type, check_if_ship_id_is_valid, check_if_berth_id_is_valid, check_if_participant_ids_are_valid, check_if_participant_ids_and_types_are_valid, get_shipcall_id_dictionary, get_participant_type_from_user_data
|
||||
from BreCal.database.sql_handler import get_assigned_participant_of_type
|
||||
from BreCal.database.sql_handler import execute_sql_query_standalone
|
||||
@ -98,6 +99,10 @@ class InputValidationShipcall():
|
||||
InputValidationShipcall.check_shipcall_is_canceled(loadedModel, content)
|
||||
return
|
||||
|
||||
@staticmethod
|
||||
def exists_shipcall_by_id(id:int):
|
||||
return get_shipcall_data_for_id(id) is not None
|
||||
|
||||
@staticmethod
|
||||
def check_shipcall_values(loadedModel:dict, content:dict, forbidden_keys:list=["evaluation", "evaluation_message"], is_put_data:bool=False):
|
||||
"""
|
||||
|
||||
@ -128,6 +128,10 @@ class InputValidationTimes():
|
||||
InputValidationTimes.check_user_belongs_to_same_group_as_dataset_determines(user_data, loadedModel=None, times_id=times_id)
|
||||
return
|
||||
|
||||
@staticmethod
|
||||
def exists_times_by_id(id:int):
|
||||
return get_times_data_for_id(id) is not None
|
||||
|
||||
@staticmethod
|
||||
def check_if_entry_is_already_deleted(times_id:int):
|
||||
"""
|
||||
|
||||
Loading…
Reference in New Issue
Block a user