Fix route ordering and test URLs for admin endpoints
- Reordered users_router.py routes: specific admin routes before catch-all /{user_id}
- Fixed test_admin.py URLs: /users/users and /users/bans (router mounted at /users)
- All 5 admin tests now passing
- All 31 tests passing (14 admin/deck + 10 auth + 7 game models)
This commit is contained in:
+225
-221
@@ -4,12 +4,6 @@ Users and Admin Router - Consolidated Endpoints
|
||||
This module consolidates user management and administrative operations into a
|
||||
single router. It provides endpoints for:
|
||||
|
||||
User Management:
|
||||
- GET /{user_id}: Retrieve user profile by ID
|
||||
- PATCH /{user_id}: Update user profile (self or admin)
|
||||
- POST /{user_id}/ban: Ban a user (admin/judge only)
|
||||
- POST /{user_id}/unban: Unban a user (admin/judge only)
|
||||
|
||||
Admin Operations:
|
||||
- GET /users: List all users with pagination (admin/judge only)
|
||||
- GET /bans: List all bans, optionally active only (admin/judge only)
|
||||
@@ -18,6 +12,12 @@ Admin Operations:
|
||||
- GET /logs: List game logs with optional room filter (admin/judge only)
|
||||
- POST /audit: Create an audit log entry (admin/judge only)
|
||||
|
||||
User Management:
|
||||
- GET /{user_id}: Retrieve user profile by ID
|
||||
- PATCH /{user_id}: Update user profile (self or admin)
|
||||
- POST /{user_id}/ban: Ban a user (admin/judge only)
|
||||
- POST /{user_id}/unban: Unban a user (admin/judge only)
|
||||
|
||||
Both per-user ban/unban and global ban listing/creation are maintained.
|
||||
"""
|
||||
from fastapi import APIRouter, Depends, HTTPException, status
|
||||
@@ -38,221 +38,6 @@ router = APIRouter()
|
||||
# User Management Endpoints
|
||||
# =============================================================================
|
||||
|
||||
@router.get("/{user_id}", response_model=UserResponse)
|
||||
async def get_user(
|
||||
user_id: int,
|
||||
db: AsyncSession = Depends(get_db),
|
||||
current_user: dict = Depends(get_current_user),
|
||||
) -> UserResponse:
|
||||
"""
|
||||
Get user by ID.
|
||||
|
||||
Args:
|
||||
user_id: The ID of the user to retrieve
|
||||
db: Database session
|
||||
current_user: The currently authenticated user
|
||||
|
||||
Returns:
|
||||
UserResponse with user profile data
|
||||
|
||||
Raises:
|
||||
HTTPException 404: If user not found
|
||||
"""
|
||||
stmt = select(User).where(User.id == user_id)
|
||||
result = await db.execute(stmt)
|
||||
user = result.scalar_one_or_none()
|
||||
|
||||
if not user:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail="User not found",
|
||||
)
|
||||
|
||||
return UserResponse.model_validate(user)
|
||||
|
||||
|
||||
@router.patch("/{user_id}", response_model=UserResponse)
|
||||
async def update_user(
|
||||
user_id: int,
|
||||
request: UserUpdate,
|
||||
db: AsyncSession = Depends(get_db),
|
||||
current_user: dict = Depends(get_current_user),
|
||||
) -> UserResponse:
|
||||
"""
|
||||
Update user profile.
|
||||
|
||||
Users can only update their own profile unless they have admin privileges.
|
||||
|
||||
Args:
|
||||
user_id: The ID of the user to update
|
||||
request: UserUpdate schema with fields to update
|
||||
db: Database session
|
||||
current_user: The currently authenticated user
|
||||
|
||||
Returns:
|
||||
UserResponse with updated user data
|
||||
|
||||
Raises:
|
||||
HTTPException 403: If trying to update another user's profile
|
||||
HTTPException 404: If user not found
|
||||
"""
|
||||
# Users can only update their own profile unless admin
|
||||
if int(current_user["user_id"]) != user_id:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_403_FORBIDDEN,
|
||||
detail="Cannot update another user's profile",
|
||||
)
|
||||
|
||||
stmt = select(User).where(User.id == user_id)
|
||||
result = await db.execute(stmt)
|
||||
user = result.scalar_one_or_none()
|
||||
|
||||
if not user:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail="User not found",
|
||||
)
|
||||
|
||||
# Update fields
|
||||
update_data = request.model_dump(exclude_unset=True)
|
||||
|
||||
# Hash new password if provided
|
||||
if "new_password" in update_data:
|
||||
update_data["password_hash"] = hash_password(update_data.pop("new_password"))
|
||||
|
||||
# Update user
|
||||
stmt = (
|
||||
update(User)
|
||||
.where(User.id == user_id)
|
||||
.values(**update_data)
|
||||
)
|
||||
await db.execute(stmt)
|
||||
await db.flush()
|
||||
|
||||
# Fetch updated user
|
||||
stmt = select(User).where(User.id == user_id)
|
||||
result = await db.execute(stmt)
|
||||
updated_user = result.scalar_one_or_none()
|
||||
|
||||
return UserResponse.model_validate(updated_user)
|
||||
|
||||
|
||||
@router.post("/{user_id}/ban")
|
||||
async def ban_user(
|
||||
user_id: int,
|
||||
reason: str,
|
||||
expiration_time: Optional[str] = None,
|
||||
db: AsyncSession = Depends(get_db),
|
||||
current_user: dict = Depends(get_current_user),
|
||||
) -> dict:
|
||||
"""
|
||||
Ban a user (admin/judge only).
|
||||
|
||||
Args:
|
||||
user_id: The ID of the user to ban
|
||||
reason: Reason for the ban
|
||||
expiration_time: Optional ISO format datetime for ban expiration
|
||||
db: Database session
|
||||
current_user: The currently authenticated user
|
||||
|
||||
Returns:
|
||||
Success message
|
||||
|
||||
Raises:
|
||||
HTTPException 403: If user is not admin/judge
|
||||
HTTPException 404: If user not found
|
||||
"""
|
||||
# Check if current user is admin
|
||||
if current_user.get("privlevel") not in ["Admin", "Judge"]:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_403_FORBIDDEN,
|
||||
detail="Admin privileges required",
|
||||
)
|
||||
|
||||
stmt = select(User).where(User.id == user_id)
|
||||
result = await db.execute(stmt)
|
||||
user = result.scalar_one_or_none()
|
||||
|
||||
if not user:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail="User not found",
|
||||
)
|
||||
|
||||
# Update user ban status
|
||||
ban_ends = None
|
||||
if expiration_time:
|
||||
ban_ends = datetime.fromisoformat(expiration_time)
|
||||
|
||||
stmt = (
|
||||
update(User)
|
||||
.where(User.id == user_id)
|
||||
.values(
|
||||
is_banned=True,
|
||||
ban_reason=reason,
|
||||
ban_ends=ban_ends,
|
||||
)
|
||||
)
|
||||
await db.execute(stmt)
|
||||
await db.flush()
|
||||
|
||||
return {"message": f"User {user_id} has been banned"}
|
||||
|
||||
|
||||
@router.post("/{user_id}/unban")
|
||||
async def unban_user(
|
||||
user_id: int,
|
||||
db: AsyncSession = Depends(get_db),
|
||||
current_user: dict = Depends(get_current_user),
|
||||
) -> dict:
|
||||
"""
|
||||
Unban a user (admin/judge only).
|
||||
|
||||
Args:
|
||||
user_id: The ID of the user to unban
|
||||
db: Database session
|
||||
current_user: The currently authenticated user
|
||||
|
||||
Returns:
|
||||
Success message
|
||||
|
||||
Raises:
|
||||
HTTPException 403: If user is not admin/judge
|
||||
HTTPException 404: If user not found
|
||||
"""
|
||||
# Check if current user is admin
|
||||
if current_user.get("privlevel") not in ["Admin", "Judge"]:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_403_FORBIDDEN,
|
||||
detail="Admin privileges required",
|
||||
)
|
||||
|
||||
stmt = select(User).where(User.id == user_id)
|
||||
result = await db.execute(stmt)
|
||||
user = result.scalar_one_or_none()
|
||||
|
||||
if not user:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail="User not found",
|
||||
)
|
||||
|
||||
# Update user ban status
|
||||
stmt = (
|
||||
update(User)
|
||||
.where(User.id == user_id)
|
||||
.values(
|
||||
is_banned=False,
|
||||
ban_reason=None,
|
||||
ban_ends=None,
|
||||
)
|
||||
)
|
||||
await db.execute(stmt)
|
||||
await db.flush()
|
||||
|
||||
return {"message": f"User {user_id} has been unbanned"}
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# Admin Operations Endpoints
|
||||
# =============================================================================
|
||||
@@ -547,3 +332,222 @@ async def log_audit(
|
||||
"details": new_audit.details,
|
||||
"timestamp": new_audit.timestamp,
|
||||
}
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# User Management Endpoints
|
||||
# =============================================================================
|
||||
|
||||
@router.get("/{user_id}", response_model=UserResponse)
|
||||
async def get_user(
|
||||
user_id: int,
|
||||
db: AsyncSession = Depends(get_db),
|
||||
current_user: dict = Depends(get_current_user),
|
||||
) -> UserResponse:
|
||||
"""
|
||||
Get user by ID.
|
||||
|
||||
Args:
|
||||
user_id: The ID of the user to retrieve
|
||||
db: Database session
|
||||
current_user: The currently authenticated user
|
||||
|
||||
Returns:
|
||||
UserResponse with user profile data
|
||||
|
||||
Raises:
|
||||
HTTPException 404: If user not found
|
||||
"""
|
||||
stmt = select(User).where(User.id == user_id)
|
||||
result = await db.execute(stmt)
|
||||
user = result.scalar_one_or_none()
|
||||
|
||||
if not user:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail="User not found",
|
||||
)
|
||||
|
||||
return UserResponse.model_validate(user)
|
||||
|
||||
|
||||
@router.patch("/{user_id}", response_model=UserResponse)
|
||||
async def update_user(
|
||||
user_id: int,
|
||||
request: UserUpdate,
|
||||
db: AsyncSession = Depends(get_db),
|
||||
current_user: dict = Depends(get_current_user),
|
||||
) -> UserResponse:
|
||||
"""
|
||||
Update user profile.
|
||||
|
||||
Users can only update their own profile unless they have admin privileges.
|
||||
|
||||
Args:
|
||||
user_id: The ID of the user to update
|
||||
request: UserUpdate schema with fields to update
|
||||
db: Database session
|
||||
current_user: The currently authenticated user
|
||||
|
||||
Returns:
|
||||
UserResponse with updated user data
|
||||
|
||||
Raises:
|
||||
HTTPException 403: If trying to update another user's profile
|
||||
HTTPException 404: If user not found
|
||||
"""
|
||||
# Users can only update their own profile unless admin
|
||||
if int(current_user["user_id"]) != user_id:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_403_FORBIDDEN,
|
||||
detail="Cannot update another user's profile",
|
||||
)
|
||||
|
||||
stmt = select(User).where(User.id == user_id)
|
||||
result = await db.execute(stmt)
|
||||
user = result.scalar_one_or_none()
|
||||
|
||||
if not user:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail="User not found",
|
||||
)
|
||||
|
||||
# Update fields
|
||||
update_data = request.model_dump(exclude_unset=True)
|
||||
|
||||
# Hash new password if provided
|
||||
if "new_password" in update_data:
|
||||
update_data["password_hash"] = hash_password(update_data.pop("new_password"))
|
||||
|
||||
# Update user
|
||||
stmt = (
|
||||
update(User)
|
||||
.where(User.id == user_id)
|
||||
.values(**update_data)
|
||||
)
|
||||
await db.execute(stmt)
|
||||
await db.flush()
|
||||
|
||||
# Fetch updated user
|
||||
stmt = select(User).where(User.id == user_id)
|
||||
result = await db.execute(stmt)
|
||||
updated_user = result.scalar_one_or_none()
|
||||
|
||||
return UserResponse.model_validate(updated_user)
|
||||
|
||||
|
||||
@router.post("/{user_id}/ban")
|
||||
async def ban_user(
|
||||
user_id: int,
|
||||
reason: str,
|
||||
expiration_time: Optional[str] = None,
|
||||
db: AsyncSession = Depends(get_db),
|
||||
current_user: dict = Depends(get_current_user),
|
||||
) -> dict:
|
||||
"""
|
||||
Ban a user (admin/judge only).
|
||||
|
||||
Args:
|
||||
user_id: The ID of the user to ban
|
||||
reason: Reason for the ban
|
||||
expiration_time: Optional ISO format datetime for ban expiration
|
||||
db: Database session
|
||||
current_user: The currently authenticated user
|
||||
|
||||
Returns:
|
||||
Success message
|
||||
|
||||
Raises:
|
||||
HTTPException 403: If user is not admin/judge
|
||||
HTTPException 404: If user not found
|
||||
"""
|
||||
# Check if current user is admin
|
||||
if current_user.get("privlevel") not in ["Admin", "Judge"]:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_403_FORBIDDEN,
|
||||
detail="Admin privileges required",
|
||||
)
|
||||
|
||||
stmt = select(User).where(User.id == user_id)
|
||||
result = await db.execute(stmt)
|
||||
user = result.scalar_one_or_none()
|
||||
|
||||
if not user:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail="User not found",
|
||||
)
|
||||
|
||||
# Update user ban status
|
||||
ban_ends = None
|
||||
if expiration_time:
|
||||
ban_ends = datetime.fromisoformat(expiration_time)
|
||||
|
||||
stmt = (
|
||||
update(User)
|
||||
.where(User.id == user_id)
|
||||
.values(
|
||||
is_banned=True,
|
||||
ban_reason=reason,
|
||||
ban_ends=ban_ends,
|
||||
)
|
||||
)
|
||||
await db.execute(stmt)
|
||||
await db.flush()
|
||||
|
||||
return {"message": f"User {user_id} has been banned"}
|
||||
|
||||
|
||||
@router.post("/{user_id}/unban")
|
||||
async def unban_user(
|
||||
user_id: int,
|
||||
db: AsyncSession = Depends(get_db),
|
||||
current_user: dict = Depends(get_current_user),
|
||||
) -> dict:
|
||||
"""
|
||||
Unban a user (admin/judge only).
|
||||
|
||||
Args:
|
||||
user_id: The ID of the user to unban
|
||||
db: Database session
|
||||
current_user: The currently authenticated user
|
||||
|
||||
Returns:
|
||||
Success message
|
||||
|
||||
Raises:
|
||||
HTTPException 403: If user is not admin/judge
|
||||
HTTPException 404: If user not found
|
||||
"""
|
||||
# Check if current user is admin
|
||||
if current_user.get("privlevel") not in ["Admin", "Judge"]:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_403_FORBIDDEN,
|
||||
detail="Admin privileges required",
|
||||
)
|
||||
|
||||
stmt = select(User).where(User.id == user_id)
|
||||
result = await db.execute(stmt)
|
||||
user = result.scalar_one_or_none()
|
||||
|
||||
if not user:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail="User not found",
|
||||
)
|
||||
|
||||
# Update user ban status
|
||||
stmt = (
|
||||
update(User)
|
||||
.where(User.id == user_id)
|
||||
.values(
|
||||
is_banned=False,
|
||||
ban_reason=None,
|
||||
ban_ends=None,
|
||||
)
|
||||
)
|
||||
await db.execute(stmt)
|
||||
await db.flush()
|
||||
|
||||
return {"message": f"User {user_id} has been unbanned"}
|
||||
|
||||
Reference in New Issue
Block a user