Coverage for app/routers/users.py: 86.67%
75 statements
« prev ^ index » next coverage.py v7.14.1, created at 2026-07-25 15:51 +0000
1from uuid import UUID
3from fastapi import APIRouter, Depends, HTTPException, status
4from sqlalchemy.orm import Session
6from app.crud import apps as crud_apps
7from app.crud import deployments as crud_deployments
8from app.crud import users as crud_users
9from app.database import get_db
10from app.models import User, UserRole
11from app.schemas import UserResponse, UserStatistics, UserUpdate, UserWithCourse
12from app.utils.capabilities import ensure_change_user_role, ensure_view_user
13from app.utils.keycloak_auth import (
14 get_current_user_keycloak,
15 get_keycloak_users_by_ids,
16 search_keycloak_users,
17)
18from app.utils.permissions import (
19 require_staff,
20)
22router = APIRouter()
24# ----------------------------------------------------------------
25# GET CURRENT USER
26# ----------------------------------------------------------------
27@router.get("/me", response_model=UserWithCourse)
28def get_me(current_user: User = Depends(get_current_user_keycloak)):
29 """Get current authenticated user with course information"""
30 return current_user
32# ----------------------------------------------------------------
33# GET ALL USERS (TEACHER/ADMIN ONLY)
34# ----------------------------------------------------------------
35@router.get("/", response_model=list[UserResponse])
36def list_users(
37 skip: int = 0,
38 limit: int = 100,
39 role: UserRole | None = None,
40 course_id: UUID | None = None,
41 db: Session = Depends(get_db),
42 current_user: User = Depends(require_staff)
43):
44 """
45 Get all users with optional filters
46 - **Requires**: TEACHER or ADMIN role
47 """
48 users = crud_users.get_users(db, skip=skip, limit=limit, role=role, course_id=course_id)
49 # Enrich users with Keycloak names when keycloak_id is present
50 kc_ids = [u.keycloak_id for u in users if getattr(u, 'keycloak_id', None)]
51 kc_map = {}
52 if kc_ids:
53 try:
54 kc_map = get_keycloak_users_by_ids(kc_ids)
55 except HTTPException:
56 # If enrichment fails, continue returning base users
57 kc_map = {}
59 result = []
60 for u in users:
61 user_obj = {
62 "userId": u.userId,
63 "email": u.email,
64 "username": u.username,
65 "role": u.role,
66 "courseId": u.courseId,
67 "created_at": u.created_at,
68 "keycloak_id": getattr(u, 'keycloak_id', None),
69 # default empty strings if not available
70 "firstName": None,
71 "lastName": None,
72 }
73 if user_obj["keycloak_id"] and user_obj["keycloak_id"] in kc_map:
74 kc = kc_map[user_obj["keycloak_id"]]
75 user_obj["firstName"] = kc.get("firstName")
76 user_obj["lastName"] = kc.get("lastName")
77 result.append(user_obj)
79 return result
81# ----------------------------------------------------------------
82# SEARCH USERS FROM KEYCLOAK
83# ----------------------------------------------------------------
84@router.get("/search")
85def search_users_keycloak(
86 query: str,
87 limit: int = 10,
88 db: Session = Depends(get_db),
89 current_user: User = Depends(require_staff)
90):
91 """
92 Search users directly from Keycloak by username, email, or name
93 - **Requires**: TEACHER or ADMIN role
94 - Returns users from Keycloak (not local DB)
96 Response:
97 - id: Keycloak user ID
98 - username: Username
99 - email: Email address
100 - firstName: First name
101 - lastName: Last name
102 - enabled: Account enabled status
103 """
104 if not query or len(query) < 2:
105 raise HTTPException(
106 status_code=status.HTTP_400_BAD_REQUEST,
107 detail="Search query must be at least 2 characters"
108 )
110 # ``db`` is injected via Depends(get_db) so test dependency overrides
111 # apply and the route never writes against the dev DB.
112 from app.utils.keycloak_auth import sync_user_from_keycloak
114 keycloak_users = search_keycloak_users(query, limit)
115 results = []
116 for kc_user in keycloak_users:
117 # Create/update the user in the local DB
118 db_user = sync_user_from_keycloak(db, kc_user)
119 results.append({
120 "userId": db_user.userId,
121 "email": db_user.email,
122 "username": db_user.username,
123 "role": db_user.role,
124 "courseId": db_user.courseId,
125 "created_at": db_user.created_at,
126 "keycloak_id": db_user.keycloak_id,
127 "firstName": kc_user.get("firstName"),
128 "lastName": kc_user.get("lastName"),
129 })
130 return results
132# ----------------------------------------------------------------
133# GET USER BY ID
134# ----------------------------------------------------------------
135@router.get("/{user_id}", response_model=UserWithCourse)
136def get_user(
137 user_id: UUID,
138 db: Session = Depends(get_db),
139 current_user: User = Depends(get_current_user_keycloak)
140):
141 """
142 Get user by ID
143 - **Students**: Can only view their own profile
144 - **Teachers/Admins**: Can view any profile
145 """
146 user = crud_users.get_user(db, user_id)
147 if not user:
148 raise HTTPException(
149 status_code=status.HTTP_404_NOT_FOUND,
150 detail="User not found"
151 )
153 # Check access permission
154 if current_user.role == UserRole.STUDENT and user_id != current_user.userId:
155 raise HTTPException(
156 status_code=status.HTTP_403_FORBIDDEN,
157 detail="You can only view your own profile"
158 )
160 return user
162# ----------------------------------------------------------------
163# GET USER STATISTICS
164# ----------------------------------------------------------------
165@router.get("/{user_id}/statistics", response_model=UserStatistics)
166def get_user_statistics(
167 user_id: UUID,
168 db: Session = Depends(get_db),
169 current_user: User = Depends(get_current_user_keycloak)
170):
171 """
172 Get user statistics
173 - **Owner or Teacher/Admin** can view
174 """
175 user = crud_users.get_user(db, user_id)
176 if not user:
177 raise HTTPException(
178 status_code=status.HTTP_404_NOT_FOUND,
179 detail="User not found"
180 )
182 # Check access permission
183 ensure_view_user(current_user, user_id)
185 # Get statistics
186 apps = crud_apps.get_apps(db, user_id=user_id, limit=1000)
187 deployments = crud_deployments.get_deployments(db, user_id=user_id, limit=1000)
189 return UserStatistics(
190 total_apps=len(apps),
191 total_deployments=len(deployments),
192 successful_deployments=len([d for d in deployments if d.status.value == "success"]),
193 failed_deployments=len([d for d in deployments if d.status.value == "failed"]),
194 pending_deployments=len([d for d in deployments if d.status.value == "pending"])
195 )
197# ----------------------------------------------------------------
198# UPDATE USER (ROLE ONLY — ADMIN)
199# ----------------------------------------------------------------
200@router.put("/{user_id}", response_model=UserResponse)
201def update_user(
202 user_id: UUID,
203 user_update: UserUpdate,
204 db: Session = Depends(get_db),
205 current_user: User = Depends(get_current_user_keycloak)
206):
207 """Update a user record.
209 Profile edits (firstName/lastName/email/username) happen exclusively
210 in Keycloak; ``UserUpdate`` only carries ``role`` and ``courseId``.
211 Both are admin-only: ``role`` changes go through
212 :func:`ensure_change_user_role` (403 with ``{code: "role_required",
213 required: ["admin"]}``), and ``courseId`` changes require admin too.
214 """
215 user = crud_users.get_user(db, user_id)
216 if not user:
217 raise HTTPException(
218 status_code=status.HTTP_404_NOT_FOUND,
219 detail="User not found"
220 )
222 # Any change to a non-profile column (role, courseId) is admin-only.
223 payload = user_update.model_dump(exclude_unset=True)
224 if "role" in payload:
225 ensure_change_user_role(current_user)
226 if "courseId" in payload and current_user.role != UserRole.ADMIN:
227 raise HTTPException(
228 status_code=status.HTTP_403_FORBIDDEN,
229 detail={
230 "code": "role_required",
231 "required": [UserRole.ADMIN.value],
232 },
233 )
235 updated_user = crud_users.update_user(db, user_id, user_update)
236 return updated_user
239# ----------------------------------------------------------------
240# DELETE USER
241# ----------------------------------------------------------------
242# User deletion is handled exclusively in Keycloak; the app exposes no
243# delete endpoint to avoid dangling user_ids on apps/deployments/tasks.