Add route and test for deleting users

src/app.py

@@ -9,7 +9,7 @@
 
 from db.database import db
 from routes.model_routes import model_routes
-from routes.authentication_routes import authentication_routes
+from routes.user_routes import user_routes
 from routes.util_routes import util_routes
 from routes.inference_routes import inference_routes
 
@@ -39,7 +39,7 @@ def index():
     return json.dumps({"status": "OK"})
 
 app.logger.setLevel(logging.INFO)
-app.register_blueprint(authentication_routes)
+app.register_blueprint(user_routes)
 app.register_blueprint(model_routes)
 app.register_blueprint(inference_routes)
 app.register_blueprint(util_routes)

src/routes/authentication_routes.py → src/routes/user_routes.py

@@ -1,17 +1,18 @@
 from flask import Blueprint, request, current_app
+from flask_jwt_extended import jwt_required, get_jwt_identity
 from pydantic import ValidationError
 from sqlalchemy.exc import SQLAlchemyError 
 from flask_jwt_extended import create_access_token
 
-from schemas.user_schema import LoginSchema, SignupSchema
-from services.user_services import add_user, check_user_login
-from utils import success_response, error_response
+from schemas.user_schema import LoginSchema, SignupSchema, DeleteUser
+from services.user_services import add_user, check_user_login, delete_user
+from utils import check_user_existence, success_response, error_response
 from errors import CustomError
 
-authentication_routes = Blueprint("authentication", __name__)
+user_routes = Blueprint("user", __name__)
 
 
-@authentication_routes.route("/signup", methods=["POST"])
+@user_routes.route("/signup", methods=["POST"])
 def register_user_route():
     """ Route for signing up the user """
 
@@ -50,7 +51,7 @@ def register_user_route():
         )
 
 
-@authentication_routes.route("/login", methods=["POST"])
+@user_routes.route("/login", methods=["POST"])
 def login_user_route():
     """ Route for logging in the user """
 
@@ -91,3 +92,46 @@ def login_user_route():
         return error_response(
             error_code="UNEXPECTED_ERROR", message="Unexpected error.", status_code=500
         )
+
+
+@user_routes.route("/delete-user", methods=["DELETE"])
+@jwt_required()
+def delete_user_route():
+    """ Route for deleting a user """
+
+    current_app.logger.info(f"Received user deletion request.")
+
+    try:
+        # Retrieve user id by decoding JWT token and check if the user still exists
+        user_id = get_jwt_identity()
+        check_user_existence(user_id)
+
+        request_data = DeleteUser(**request.json)
+
+        delete_user(user_id, request_data)
+
+        return success_response("User deletion successful.")
+
+    except ValidationError as e:
+        current_app.logger.error(f"User deletion data validation failed. Error:\n{e}")
+        return error_response(
+            error_code="INVALID_USER_DELETION_DATA", message="Invalid user deletion data.", status_code=400
+        )
+
+    except CustomError as e:
+        current_app.logger.error(f"Failed to delete user. Error:\n{e.message}")
+        return error_response(
+            error_code=e.error_code, message=e.message, status_code=e.status_code
+        )
+
+    except SQLAlchemyError as e:
+        current_app.logger.error(f"Failed to query user data. Error:\n{e}")
+        return error_response(
+            error_code="USER_DELETION_FAILED", message="Email or password not found.", status_code=500
+        )
+
+    except Exception as e:
+        current_app.logger.error(f"Unexpected error while trying to delete user. Error:\n{e}")
+        return error_response(
+            error_code="UNEXPECTED_ERROR", message="Unexpected error.", status_code=500
+        )

src/schemas/model_schema.py

@@ -2,17 +2,17 @@
 
 
 class AddModelSchema(BaseModel):
-    """ Model schema to validate the /models POST request data """ 
+    """ Schema to validate the /models POST request data """ 
     name: str
     nationalities: list[str]
 
 
 class DeleteModelSchema(BaseModel):
-    """ Model schema to validate the /models POST request data """ 
+    """ Schema to validate the /models POST request data """ 
     names: list[str]
 
 
 class ClassificationSchema(BaseModel):
-    """ Model schema to validate the /classify POST request data """ 
+    """ Schema to validate the /classify POST request data """ 
     modelName: str
     names: list[str]

src/schemas/user_schema.py

@@ -2,15 +2,20 @@
 
 
 class LoginSchema(BaseModel):
-    """ Model schema to validate the /login POST request data """ 
+    """ Schema to validate the /login POST request data """ 
     email: str
     password: str
 
 
 class SignupSchema(BaseModel):
-    """ Model schema to validate the /signup POST request data """
+    """ Schema to validate the /signup POST request data """
     email: str
     password: str
     name: str
     role: str
-    consented: bool
+    consented: bool
+
+
+class DeleteUser(BaseModel):
+    """ Schema to validate the /delete-user POST request data """
+    password: str

src/services/user_services.py

@@ -1,6 +1,5 @@
 import bcrypt
-from functools import wraps
-from schemas.user_schema import LoginSchema, SignupSchema
+from schemas.user_schema import LoginSchema, SignupSchema, DeleteUser
 from db.tables import User
 from db.database import db
 from utils import is_strong_password, is_valid_email
@@ -100,3 +99,26 @@ def add_user(data: SignupSchema) -> None:
     db.session.add(new_user)
     db.session.commit()
 
+
+def delete_user(user_id: str, data: DeleteUser) -> None:
+    """
+    Deletes a user from the database
+    :param user_id: ID of the user to delete
+    :param data: User deletion data (contains just the confirmation password)
+    """
+
+    user = User.query.filter_by(id=user_id).first()
+
+    true_password_bytes = user.password.encode("utf-8")
+    user_password_bytes = data.password.encode("utf-8")
+
+    successful_authentication = bcrypt.checkpw(user_password_bytes, true_password_bytes)
+    if not successful_authentication:
+        raise CustomError(
+            error_code="USER_DELETION_FAILED",
+            message="Password incorrect.",
+            status_code=401
+        )
+
+    db.session.delete(user)
+    db.session.commit()

tests/test_routes/test_user_routes.py

@@ -82,3 +82,35 @@ def test_login_user_fail(test_client):
 
     assert response.status_code == 400
     assert json.loads(response.data)["errorCode"] == "INVALID_LOGIN_DATA"
+
+
+def test_delete_user(test_client):
+    login_data = {
+        "email": TEST_USER_DATA["email"],
+        "password": TEST_USER_DATA["password"]
+    }
+
+    response = test_client.post("/login", json=login_data)
+
+    # Check that the response status code is 200 (successful login)
+    assert response.status_code == 200
+
+    token = json.loads(response.data)["data"]["accessToken"]
+
+    test_header = {
+        "Authorization": f"Bearer {token}"
+    }
+    test_body = {
+        "password": TEST_USER_DATA["password"]
+    }
+    response = test_client.delete("/delete-user", json=test_body, headers=test_header)
+
+    expected_response_data = {
+        "message": "User deletion successful."
+    }
+
+    assert response.status_code == 200
+    assert json.loads(response.data) == expected_response_data
+
+    deleted_user = User.query.filter_by(email=TEST_USER_DATA["email"]).first()
+    assert deleted_user is None