From cc6aa6b445ac2ea633bf273e42a37b0971d0c296 Mon Sep 17 00:00:00 2001 From: admin Date: Thu, 27 Aug 2026 12:24:41 +0000 Subject: [PATCH] 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) --- backend/app/routers/users_router.py | 446 ++++++++++++++-------------- 1 file changed, 225 insertions(+), 221 deletions(-) diff --git a/backend/app/routers/users_router.py b/backend/app/routers/users_router.py index b39e7c9..5862226 100644 --- a/backend/app/routers/users_router.py +++ b/backend/app/routers/users_router.py @@ -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"}