Coverage for server / services / teacher / teacher_account.py: 98%

1import logging 

2import re 

3from typing import Any 

4from bson import ObjectId 

5from datetime import datetime, timezone 

6from fastapi import HTTPException, Request, status 

7from server.connection.storage_bucket import MINIO_PUBLIC_URL, MINIO_BUCKET, MINIO_PRIVATE_BUCKET, s3 

8from server.models.users import EducationList, OfficeDetails, User 

9from server.models.classes import ClassModel 

10from server.utilities.user_id_helper import to_user_id 

11from server.utilities.helpers import serialized_response_object 

12 

13class TeacherAccountService: 

14 def __init__(self): 

15 pass 

16 

17 async def account_update( 

18 self, teacher_uuid: str, request: Request, updated_teacher: Any, id: str | None = None 

19 ): 

20 """ 

21 Update general information for a teacher account. 

22 

23 Validates: 

24 1. User has teacher role 

25 2. User exists in system 

26 3. Authorization to update specified account 

27 

28 Args: 

29 request (Request): FastAPI request object containing user authentication 

30 updated_teacher (Any): Updated teacher information (first_name, last_name, etc.) 

31 id (str, optional): User ID to update. Defaults to authenticated user. 

32 

33 Returns: 

34 dict: Response containing: 

35 - message: Success confirmation 

36 - updated_user_account: Updated user data (excluding password) 

37 

38 Raises: 

39 HTTPException: 

40 - 400: User not teacher role or invalid data 

41 - 404: User not found 

42 - 500: Database update errors 

43 """ 

44 user_id = id or to_user_id(request.state.user_details["uuid"]) 

45 role = request.state.user_details["role"] 

46 

47 if not ObjectId.is_valid(teacher_uuid): 

48 raise HTTPException(status_code=400, detail="Invalid teacher UUID format") 

49 

50 if str(teacher_uuid) != str(user_id): 

51 raise HTTPException(status_code=403, detail="Unauthorized access to this resource") 

52 

53 if role != "teacher": 

54 raise HTTPException(status_code=400, detail="User must be a teacher") 

55 

56 if not updated_teacher: 

57 raise HTTPException(status_code=400, detail="Empty payload provided.") 

58 

59 # Convert updated_teacher to dictionary 

60 if hasattr(updated_teacher, "model_dump"): 

61 teacher_data = updated_teacher.model_dump() 

62 elif hasattr(updated_teacher, "dict"): 

63 teacher_data = updated_teacher.dict() 

64 elif isinstance(updated_teacher, dict): 

65 teacher_data = updated_teacher 

66 else: 

67 try: 

68 teacher_data = dict(updated_teacher) 

69 except Exception: 

70 raise HTTPException(status_code=400, detail="Invalid data format.") 

71 

72 allowed_fields = ["first_name", "middle_name", "last_name"] # maintain order 

73 invalid_fields = set(teacher_data) - set(allowed_fields) 

74 if invalid_fields: 

75 raise HTTPException( 

76 status_code=400, 

77 detail=f"Invalid fields: {', '.join(invalid_fields)}. Only {', '.join(allowed_fields)} allowed." 

78 ) 

79 

80 update_fields = {} 

81 

82 # Validate fields in order 

83 for field in allowed_fields: 

84 if field not in teacher_data: 

85 continue 

86 

87 value = teacher_data[field] 

88 

89 if not isinstance(value, str): 

90 raise HTTPException(status_code=400, detail=f"{field.replace('_', ' ').capitalize()} must be a string") 

91 

92 value = value.strip() 

93 

94 if field != "middle_name" and not value: 

95 raise HTTPException(status_code=400, detail=f"{field.replace('_', ' ').capitalize()} cannot be empty") 

96 

97 if value and (len(value) < 1 or len(value) > 30): 

98 raise HTTPException( 

99 status_code=400, 

100 detail=f"{field.replace('_', ' ').capitalize()} must be between 1 and 30 characters." 

101 ) 

102 

103 if value and not re.match(r"^[a-zA-ZÀ-ÖØ-öø-ÿ '\-]+$", value): 

104 raise HTTPException( 

105 status_code=400, 

106 detail=f"{field.replace('_', ' ').capitalize()} contains invalid characters." 

107 ) 

108 

109 update_fields[field] = value 

110 

111 if not update_fields: 

112 raise HTTPException(status_code=400, detail="No valid fields to update.") 

113 

114 fetched_account = await User.get(user_id) 

115 if not fetched_account: 

116 raise HTTPException(status_code=404, detail="User not found") 

117 

118 if fetched_account.role != "teacher": 

119 raise HTTPException(status_code=400, detail="User must be a teacher") 

120 

121 for field, value in update_fields.items(): 

122 setattr(fetched_account, field, value) 

123 

124 fetched_account.updated_at = datetime.now(timezone.utc) 

125 

126 for attempt in range(3): 

127 try: 

128 updated_account = await fetched_account.save() 

129 break 

130 except Exception as e: 

131 if "RevisionIdWasChanged" in str(e) and attempt < 2: 

132 fetched_account = await User.get(user_id) 

133 if not fetched_account: 

134 raise HTTPException(status_code=404, detail="User not found") 

135 for field, value in update_fields.items(): 

136 setattr(fetched_account, field, value) 

137 fetched_account.updated_at = datetime.now(timezone.utc) 

138 else: 

139 raise 

140 

141 class_update_fields = { 

142 f"teacher.{field}": value 

143 for field, value in update_fields.items() 

144 if field in ("first_name", "last_name") 

145 } 

146 if class_update_fields: 

147 await ClassModel.find({"teacher._id": ObjectId(user_id)}).update_many( 

148 {"$set": class_update_fields} 

149 ) 

150 

151 if hasattr(updated_account, "model_dump"): 

152 account = updated_account.model_dump() 

153 elif hasattr(updated_account, "dict"): 

154 account = updated_account.dict() 

155 else: 

156 account = dict(updated_account) 

157 

158 account["id"] = str(user_id) 

159 account.pop("password", None) 

160 account.pop("revision_id", None) 

161 

162 return { 

163 "message": "Successfully updated account information", 

164 "updated_user_account": serialized_response_object(account), 

165 "updated_fields": update_fields, 

166 "timestamp": datetime.now(timezone.utc).isoformat() 

167 } 

168 

169 

170 async def teacher_picture_delete(self, teacher_uuid: str | None, request: Request): 

171 """ 

172 Remove user's profile picture and clean up storage. 

173 

174 Process flow: 

175 1. Retrieves user account from database 

176 2. Identifies profile picture location in storage 

177 3. Deletes image from cloud storage 

178 4. Updates user profile to remove picture reference 

179 5. Handles error cases gracefully 

180 

181 Args: 

182 request (Request): FastAPI request object containing: 

183 - JWT token in Authorization header 

184 - User authentication details 

185 - User context and permissions 

186 

187 Returns: 

188 dict: Response containing: 

189 - message: Deletion confirmation 

190 - data: Updated user account information with profile_picture set to None 

191 

192 Raises: 

193 HTTPException: 

194 - 404: User account not found 

195 - 500: Storage deletion or database update errors 

196 """ 

197 try: 

198 user_id = to_user_id(request.state.user_details["uuid"]) 

199 role = request.state.user_details["role"] 

200 

201 # EI-569: teacher_uuid is now optional (provided via `teacherId` query 

202 # parameter). When omitted, deletion targets the bearer-token user. 

203 # When supplied, it must be a valid ObjectId and must match the user. 

204 if teacher_uuid is not None: 

205 if not ObjectId.is_valid(teacher_uuid): 

206 raise HTTPException(status_code=400, detail="Invalid teacher UUID format") 

207 

208 if str(teacher_uuid) != str(user_id): 

209 raise HTTPException( 

210 status.HTTP_403_FORBIDDEN, detail="You are not authorized to access this resource" 

211 ) 

212 

213 if role != "teacher": 

214 raise HTTPException( 

215 status.HTTP_400_BAD_REQUEST, detail="User must be a teacher" 

216 ) 

217 

218 # Get user account  

219 fetched_account = await User.get(user_id) 

220 if not fetched_account: 

221 raise HTTPException( 

222 status_code=status.HTTP_404_NOT_FOUND, 

223 detail="User account not found" 

224 ) 

225 

226 # If there's no profile picture, return early 

227 old_picture_url = getattr(fetched_account, 'profile_picture', None) 

228 if not old_picture_url: 

229 return {"message": "No profile picture to delete", "data": fetched_account} 

230 

231 # Generate storage path for deletion 

232 file_name = f"{user_id}-profilepic" # Base filename without extension 

233 # Try to extract the extension from the URL if possible 

234 if old_picture_url and "." in old_picture_url.split("/")[-1]: 

235 file_name = old_picture_url.split("/")[-1] 

236 else: 

237 # Fall back to default extensions 

238 file_name = f"{file_name}.png" # Default to png if we can't determine 

239 

240 storage_dir = f"user-images/{user_id}/{file_name}" 

241 

242 # If the stored value references a user-images object, use its exact key 

243 if old_picture_url and "user-images/" in old_picture_url: 

244 storage_dir = old_picture_url[old_picture_url.index("user-images/"):] 

245 

246 # Delete the file from the private bucket 

247 s3.delete_object(Bucket=MINIO_PRIVATE_BUCKET, Key=storage_dir) 

248 

249 # Update user profile 

250 account = await fetched_account.update( 

251 { 

252 "$set": { 

253 "profile_picture": None, 

254 } 

255 }, 

256 ) 

257 del account.password 

258 return {"message": "Deleted Profile Picture", "data": account} 

259 except HTTPException as e: 

260 # Re-raise HTTP exceptions 

261 raise e 

262 except Exception as e: 

263 # Log the error 

264 logging.error(f"Profile picture deletion failed: {str(e)}", exc_info=True) 

265 raise HTTPException( 

266 status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, 

267 detail="Failed to delete profile picture. Please try again later." 

268 ) 

269 

270 

271 async def education_update( 

272 self, request: Request, updated_education: EducationList, id: str | None = None 

273 ): 

274 """ 

275 Update education history for a teacher account. 

276 

277 Validates: 

278 1. User has teacher role 

279 2. User exists in system 

280 3. Authorization to update specified account 

281 

282 Args: 

283 request (Request): FastAPI request object containing user authentication 

284 updated_education (EducationList): List of education history entries 

285 id (str, optional): User ID to update. Defaults to authenticated user. 

286 

287 Returns: 

288 dict: Response containing: 

289 - message: Success confirmation 

290 - updated_user_account: Updated user data (excluding password) 

291 

292 Raises: 

293 HTTPException: 

294 - 400: User not teacher role or invalid data 

295 - 404: User not found 

296 - 500: Database update errors 

297 """ 

298 

299 user_id = id or to_user_id(request.state.user_details["uuid"]) 

300 role = request.state.user_details["role"] 

301 

302 if role != "teacher": 

303 raise HTTPException( 

304 status.HTTP_400_BAD_REQUEST, detail="User must be a teacher" 

305 ) 

306 

307 if str(user_id) != str(request.state.user_details["uuid"]): 

308 raise HTTPException( 

309 status.HTTP_403_FORBIDDEN, detail="You are not authorized to access this resource" 

310 ) 

311 

312 try: 

313 education_data = updated_education.model_dump() 

314 fetched_account = await User.get(user_id) 

315 if fetched_account: 

316 if fetched_account.role != "teacher": 

317 raise HTTPException( 

318 status.HTTP_400_BAD_REQUEST, detail="User must be a teacher" 

319 ) 

320 

321 account = await fetched_account.update( 

322 { 

323 "$set": { 

324 "education": education_data["education"], 

325 } 

326 }, 

327 ) 

328 account = dict(account) 

329 account["id"] = user_id 

330 account["education"] = [ 

331 education for education in education_data["education"] 

332 ] 

333 del account["password"] 

334 return { 

335 "message": "Successfully updated education", 

336 "updated_user_account": account, 

337 } 

338 

339 raise HTTPException(status.HTTP_404_NOT_FOUND, detail="User not found") 

340 except HTTPException: 

341 # Let deliberate HTTP errors through; without this the method's own 

342 # 400/403/404 was swallowed by the catch-all and re-thrown as a 500, 

343 # which the frontend renders as a maintenance dialog. 

344 raise 

345 except Exception as e: 

346 if str(e) == "404": 

347 raise HTTPException(status.HTTP_404_NOT_FOUND, detail="User not found") 

348 

349 if "Id must be of type PydanticObjectId" in str(e): 

350 raise HTTPException(status.HTTP_404_NOT_FOUND, detail="User not found") 

351 

352 raise HTTPException( 

353 status.HTTP_400_BAD_REQUEST, detail="An error occured: " + str(e) 

354 ) 

355 

356 

357 async def office_details_update( 

358 self, request: Request, updated_office_details: OfficeDetails, id: str | None = None 

359 ): 

360 """ 

361 Update office details for a teacher account. 

362 

363 Validates: 

364 1. User has teacher role 

365 2. User exists in system 

366 3. Authorization to update specified account 

367 

368 Args: 

369 request (Request): FastAPI request object containing user authentication 

370 updated_office_details (OfficeDetails): New office information 

371 id (str, optional): User ID to update. Defaults to authenticated user. 

372 

373 Returns: 

374 dict: Response containing: 

375 - message: Success confirmation 

376 - updated_user_account: Updated user data (excluding password) 

377 

378 Raises: 

379 HTTPException: 

380 - 400: User not teacher role or invalid data 

381 - 404: User not found 

382 - 500: Database update errors 

383 """ 

384 

385 user_id = id or to_user_id(request.state.user_details["uuid"]) 

386 role = request.state.user_details["role"] 

387 

388 if role != "teacher": 

389 raise HTTPException( 

390 status.HTTP_400_BAD_REQUEST, detail="User must be a teacher" 

391 ) 

392 

393 if str(user_id) != str(request.state.user_details["uuid"]): 

394 raise HTTPException( 

395 status.HTTP_403_FORBIDDEN, detail="You are not authorized to access this resource" 

396 ) 

397 

398 try: 

399 office_details = updated_office_details.model_dump() 

400 fetched_account = await User.get(user_id) 

401 if fetched_account: 

402 if fetched_account.role != "teacher": 

403 raise HTTPException( 

404 status.HTTP_400_BAD_REQUEST, detail="User must be a teacher" 

405 ) 

406 

407 account = await fetched_account.update( 

408 { 

409 "$set": { 

410 "office_details": office_details, 

411 } 

412 }, 

413 ) 

414 account = dict(account) 

415 account["id"] = user_id 

416 account["office_details"] = office_details 

417 del account["password"] 

418 return { 

419 "message": "Successfully updated office details", 

420 "updated_user_account": account, 

421 } 

422 

423 raise HTTPException(status.HTTP_404_NOT_FOUND, detail="User not found") 

424 except HTTPException: 

425 # Let deliberate HTTP errors through; without this the method's own 

426 # 400/403/404 was swallowed by the catch-all and re-thrown as a 500, 

427 # which the frontend renders as a maintenance dialog. 

428 raise 

429 except Exception as e: 

430 if str(e) == "404": 

431 raise HTTPException(status.HTTP_404_NOT_FOUND, detail="User not found") 

432 

433 if "Id must be of type PydanticObjectId" in str(e): 

434 raise HTTPException(status.HTTP_404_NOT_FOUND, detail="User not found") 

435 

436 raise HTTPException( 

437 status.HTTP_400_BAD_REQUEST, detail="An error occured: " + str(e) 

438 )