♻️ Add delete_user_me endpoint and corresponding test cases (#1179)

Co-authored-by: Sebastián Ramírez <tiangolo@gmail.com>
This commit is contained in:
Alejandra
2024-04-26 14:35:12 -05:00
committed by GitHub
parent d1cc75903e
commit b81cec8ea5
3 changed files with 59 additions and 32 deletions

View File

@@ -51,6 +51,6 @@ CurrentUser = Annotated[User, Depends(get_current_user)]
def get_current_active_superuser(current_user: CurrentUser) -> User: def get_current_active_superuser(current_user: CurrentUser) -> User:
if not current_user.is_superuser: if not current_user.is_superuser:
raise HTTPException( 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 return current_user

View File

@@ -124,6 +124,22 @@ def read_user_me(current_user: CurrentUser) -> Any:
return current_user 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) @router.post("/signup", response_model=UserPublic)
def register_user(session: SessionDep, user_in: UserRegister) -> Any: def register_user(session: SessionDep, user_in: UserRegister) -> Any:
""" """
@@ -195,7 +211,7 @@ def update_user(
return db_user return db_user
@router.delete("/{user_id}") @router.delete("/{user_id}", dependencies=[Depends(get_current_active_superuser)])
def delete_user( def delete_user(
session: SessionDep, current_user: CurrentUser, user_id: int session: SessionDep, current_user: CurrentUser, user_id: int
) -> Message: ) -> Message:
@@ -205,15 +221,10 @@ def delete_user(
user = session.get(User, user_id) user = session.get(User, user_id)
if not user: if not user:
raise HTTPException(status_code=404, detail="User not found") raise HTTPException(status_code=404, detail="User not found")
elif user != current_user and not current_user.is_superuser: if user == current_user:
raise HTTPException(
status_code=403, detail="The user doesn't have enough privileges"
)
elif user == current_user and current_user.is_superuser:
raise HTTPException( raise HTTPException(
status_code=403, detail="Super users are not allowed to delete themselves" status_code=403, detail="Super users are not allowed to delete themselves"
) )
statement = delete(Item).where(col(Item.owner_id) == user_id) statement = delete(Item).where(col(Item.owner_id) == user_id)
session.exec(statement) # type: ignore session.exec(statement) # type: ignore
session.delete(user) session.delete(user)

View File

@@ -142,7 +142,7 @@ def test_create_user_by_normal_user(
headers=normal_user_token_headers, headers=normal_user_token_headers,
json=data, json=data,
) )
assert r.status_code == 400 assert r.status_code == 403
def test_retrieve_users( def test_retrieve_users(
@@ -402,28 +402,7 @@ def test_update_user_email_exists(
assert r.json()["detail"] == "User with this email already exists" assert r.json()["detail"] == "User with this email already exists"
def test_delete_user_super_user( def test_delete_user_me(client: TestClient, db: Session) -> None:
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:
username = random_email() username = random_email()
password = random_lower_string() password = random_lower_string()
user_in = UserCreate(email=username, password=password) 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}"} headers = {"Authorization": f"Bearer {a_token}"}
r = client.delete( r = client.delete(
f"{settings.API_V1_STR}/users/{user_id}", f"{settings.API_V1_STR}/users/me",
headers=headers, headers=headers,
) )
assert r.status_code == 200 assert r.status_code == 200
deleted_user = r.json() deleted_user = r.json()
assert deleted_user["message"] == "User deleted successfully" 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_query = select(User).where(User.id == user_id)
user_db = db.execute(user_query).first() user_db = db.execute(user_query).first()