Skip to content
2 changes: 1 addition & 1 deletion backend/app/api/deps.py
Original file line number Diff line number Diff line change
Expand Up @@ -51,6 +51,6 @@ def get_current_user(session: SessionDep, token: TokenDep) -> 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
25 changes: 18 additions & 7 deletions backend/app/api/routes/users.py
Original file line number Diff line number Diff line change
Expand Up @@ -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")
Comment thread
alejsdev marked this conversation as resolved.


@router.post("/signup", response_model=UserPublic)
def register_user(session: SessionDep, user_in: UserRegister) -> Any:
"""
Expand Down Expand Up @@ -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:
Expand All @@ -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)
Expand Down
52 changes: 34 additions & 18 deletions backend/app/tests/api/routes/test_users.py
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand Down Expand Up @@ -402,50 +402,66 @@ 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:
def test_delete_user_me(client: TestClient, 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

login_data = {
"username": username,
"password": password,
}
r = client.post(f"{settings.API_V1_STR}/login/access-token", data=login_data)
tokens = r.json()
a_token = tokens["access_token"]
headers = {"Authorization": f"Bearer {a_token}"}

r = client.delete(
f"{settings.API_V1_STR}/users/{user_id}",
headers=superuser_token_headers,
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"
Comment thread
tiangolo marked this conversation as resolved.
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_current_user(client: TestClient, db: Session) -> None:
def test_delete_user_me_as_superuser(
Comment thread
alejsdev marked this conversation as resolved.
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

login_data = {
"username": username,
"password": password,
}
r = client.post(f"{settings.API_V1_STR}/login/access-token", data=login_data)
tokens = r.json()
a_token = tokens["access_token"]
headers = {"Authorization": f"Bearer {a_token}"}

r = client.delete(
f"{settings.API_V1_STR}/users/{user_id}",
headers=headers,
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()
Expand Down