diff --git a/backend/app/api/deps.py b/backend/app/api/deps.py index 74103ae..b71dda5 100644 --- a/backend/app/api/deps.py +++ b/backend/app/api/deps.py @@ -51,6 +51,6 @@ CurrentUser = Annotated[User, Depends(get_current_user)] def get_current_active_superuser(current_user: CurrentUser) -> User: if not current_user.is_superuser: raise HTTPException( - status_code=400, detail="The user doesn't have enough privileges" + status_code=403, detail="The user doesn't have enough privileges" ) return current_user diff --git a/backend/app/api/routes/users.py b/backend/app/api/routes/users.py index 1fd2500..7a4066a 100644 --- a/backend/app/api/routes/users.py +++ b/backend/app/api/routes/users.py @@ -124,6 +124,22 @@ def read_user_me(current_user: CurrentUser) -> Any: return current_user +@router.delete("/me", response_model=Message) +def delete_user_me(session: SessionDep, current_user: CurrentUser) -> Any: + """ + Delete own user. + """ + if current_user.is_superuser: + raise HTTPException( + status_code=403, detail="Super users are not allowed to delete themselves" + ) + statement = delete(Item).where(col(Item.owner_id) == current_user.id) + session.exec(statement) # type: ignore + session.delete(current_user) + session.commit() + return Message(message="User deleted successfully") + + @router.post("/signup", response_model=UserPublic) def register_user(session: SessionDep, user_in: UserRegister) -> Any: """ @@ -195,7 +211,7 @@ def update_user( return db_user -@router.delete("/{user_id}") +@router.delete("/{user_id}", dependencies=[Depends(get_current_active_superuser)]) def delete_user( session: SessionDep, current_user: CurrentUser, user_id: int ) -> Message: @@ -205,15 +221,10 @@ def delete_user( user = session.get(User, user_id) if not user: raise HTTPException(status_code=404, detail="User not found") - elif user != current_user and not current_user.is_superuser: - raise HTTPException( - status_code=403, detail="The user doesn't have enough privileges" - ) - elif user == current_user and current_user.is_superuser: + if user == current_user: raise HTTPException( status_code=403, detail="Super users are not allowed to delete themselves" ) - statement = delete(Item).where(col(Item.owner_id) == user_id) session.exec(statement) # type: ignore session.delete(user) diff --git a/backend/app/tests/api/routes/test_users.py b/backend/app/tests/api/routes/test_users.py index d6923a5..465583a 100644 --- a/backend/app/tests/api/routes/test_users.py +++ b/backend/app/tests/api/routes/test_users.py @@ -142,7 +142,7 @@ def test_create_user_by_normal_user( headers=normal_user_token_headers, json=data, ) - assert r.status_code == 400 + assert r.status_code == 403 def test_retrieve_users( @@ -402,28 +402,7 @@ def test_update_user_email_exists( assert r.json()["detail"] == "User with this email already exists" -def test_delete_user_super_user( - client: TestClient, superuser_token_headers: dict[str, str], db: Session -) -> None: - username = random_email() - password = random_lower_string() - user_in = UserCreate(email=username, password=password) - user = crud.create_user(session=db, user_create=user_in) - user_id = user.id - r = client.delete( - f"{settings.API_V1_STR}/users/{user_id}", - headers=superuser_token_headers, - ) - assert r.status_code == 200 - deleted_user = r.json() - assert deleted_user["message"] == "User deleted successfully" - - user_query = select(User).where(User.id == user_id) - user_db = db.execute(user_query).first() - assert user_db is None - - -def test_delete_user_current_user(client: TestClient, db: Session) -> None: +def test_delete_user_me(client: TestClient, db: Session) -> None: username = random_email() password = random_lower_string() user_in = UserCreate(email=username, password=password) @@ -440,12 +419,49 @@ def test_delete_user_current_user(client: TestClient, db: Session) -> None: headers = {"Authorization": f"Bearer {a_token}"} r = client.delete( - f"{settings.API_V1_STR}/users/{user_id}", + f"{settings.API_V1_STR}/users/me", headers=headers, ) assert r.status_code == 200 deleted_user = r.json() assert deleted_user["message"] == "User deleted successfully" + result = db.exec(select(User).where(User.id == user_id)).first() + assert result is None + + user_query = select(User).where(User.id == user_id) + user_db = db.execute(user_query).first() + assert user_db is None + + +def test_delete_user_me_as_superuser( + client: TestClient, superuser_token_headers: dict[str, str] +) -> None: + r = client.delete( + f"{settings.API_V1_STR}/users/me", + headers=superuser_token_headers, + ) + assert r.status_code == 403 + response = r.json() + assert response["detail"] == "Super users are not allowed to delete themselves" + + +def test_delete_user_super_user( + client: TestClient, superuser_token_headers: dict[str, str], db: Session +) -> None: + username = random_email() + password = random_lower_string() + user_in = UserCreate(email=username, password=password) + user = crud.create_user(session=db, user_create=user_in) + user_id = user.id + r = client.delete( + f"{settings.API_V1_STR}/users/{user_id}", + headers=superuser_token_headers, + ) + assert r.status_code == 200 + deleted_user = r.json() + assert deleted_user["message"] == "User deleted successfully" + result = db.exec(select(User).where(User.id == user_id)).first() + assert result is None user_query = select(User).where(User.id == user_id) user_db = db.execute(user_query).first()