Coverage for app/tests/api/routes/test_users.py: 100%

1import uuid 

2from unittest.mock import patch 

3 

4from fastapi.testclient import TestClient 

5from sqlmodel import Session, select 

6 

7from app import crud 

8from app.core.config import settings 

9from app.core.security import verify_password 

10from app.models import User, UserCreate 

11from app.tests.utils.utils import random_email, random_lower_string 

12 

13 

14def test_get_users_superuser_me( 

15 client: TestClient, superuser_token_headers: dict[str, str] 

16) -> None: 

17 r = client.get(f"{settings.API_V1_STR}/users/me", headers=superuser_token_headers) 

18 current_user = r.json() 

19 assert current_user 

20 assert current_user["is_active"] is True 

21 assert current_user["is_superuser"] 

22 assert current_user["email"] == settings.FIRST_SUPERUSER 

23 

24 

25def test_get_users_normal_user_me( 

26 client: TestClient, normal_user_token_headers: dict[str, str] 

27) -> None: 

28 r = client.get(f"{settings.API_V1_STR}/users/me", headers=normal_user_token_headers) 

29 current_user = r.json() 

30 assert current_user 

31 assert current_user["is_active"] is True 

32 assert current_user["is_superuser"] is False 

33 assert current_user["email"] == settings.EMAIL_TEST_USER 

34 

35 

36def test_create_user_new_email( 

37 client: TestClient, superuser_token_headers: dict[str, str], db: Session 

38) -> None: 

39 with ( 

40 patch("app.utils.send_email", return_value=None), 

41 patch("app.core.config.settings.SMTP_HOST", "smtp.example.com"), 

42 patch("app.core.config.settings.SMTP_USER", "admin@example.com"), 

43 ): 

44 username = random_email() 

45 password = random_lower_string() 

46 data = {"email": username, "password": password} 

47 r = client.post( 

48 f"{settings.API_V1_STR}/users/", 

49 headers=superuser_token_headers, 

50 json=data, 

51 ) 

52 assert 200 <= r.status_code < 300 

53 created_user = r.json() 

54 user = crud.get_user_by_email(session=db, email=username) 

55 assert user 

56 assert user.email == created_user["email"] 

57 

58 

59def test_get_existing_user( 

60 client: TestClient, superuser_token_headers: dict[str, str], db: Session 

61) -> None: 

62 username = random_email() 

63 password = random_lower_string() 

64 user_in = UserCreate(email=username, password=password) 

65 user = crud.create_user(session=db, user_create=user_in) 

66 user_id = user.id 

67 r = client.get( 

68 f"{settings.API_V1_STR}/users/{user_id}", 

69 headers=superuser_token_headers, 

70 ) 

71 assert 200 <= r.status_code < 300 

72 api_user = r.json() 

73 existing_user = crud.get_user_by_email(session=db, email=username) 

74 assert existing_user 

75 assert existing_user.email == api_user["email"] 

76 

77 

78def test_get_existing_user_current_user(client: TestClient, db: Session) -> None: 

79 username = random_email() 

80 password = random_lower_string() 

81 user_in = UserCreate(email=username, password=password) 

82 user = crud.create_user(session=db, user_create=user_in) 

83 user_id = user.id 

84 

85 login_data = { 

86 "username": username, 

87 "password": password, 

88 } 

89 r = client.post(f"{settings.API_V1_STR}/login/access-token", data=login_data) 

90 tokens = r.json() 

91 a_token = tokens["access_token"] 

92 headers = {"Authorization": f"Bearer {a_token}"} 

93 

94 r = client.get( 

95 f"{settings.API_V1_STR}/users/{user_id}", 

96 headers=headers, 

97 ) 

98 assert 200 <= r.status_code < 300 

99 api_user = r.json() 

100 existing_user = crud.get_user_by_email(session=db, email=username) 

101 assert existing_user 

102 assert existing_user.email == api_user["email"] 

103 

104 

105def test_get_existing_user_permissions_error( 

106 client: TestClient, normal_user_token_headers: dict[str, str] 

107) -> None: 

108 r = client.get( 

109 f"{settings.API_V1_STR}/users/{uuid.uuid4()}", 

110 headers=normal_user_token_headers, 

111 ) 

112 assert r.status_code == 403 

113 assert r.json() == {"detail": "The user doesn't have enough privileges"} 

114 

115 

116def test_create_user_existing_username( 

117 client: TestClient, superuser_token_headers: dict[str, str], db: Session 

118) -> None: 

119 username = random_email() 

120 # username = email 

121 password = random_lower_string() 

122 user_in = UserCreate(email=username, password=password) 

123 crud.create_user(session=db, user_create=user_in) 

124 data = {"email": username, "password": password} 

125 r = client.post( 

126 f"{settings.API_V1_STR}/users/", 

127 headers=superuser_token_headers, 

128 json=data, 

129 ) 

130 created_user = r.json() 

131 assert r.status_code == 400 

132 assert "_id" not in created_user 

133 

134 

135def test_create_user_by_normal_user( 

136 client: TestClient, normal_user_token_headers: dict[str, str] 

137) -> None: 

138 username = random_email() 

139 password = random_lower_string() 

140 data = {"email": username, "password": password} 

141 r = client.post( 

142 f"{settings.API_V1_STR}/users/", 

143 headers=normal_user_token_headers, 

144 json=data, 

145 ) 

146 assert r.status_code == 403 

147 

148 

149def test_retrieve_users( 

150 client: TestClient, superuser_token_headers: dict[str, str], db: Session 

151) -> None: 

152 username = random_email() 

153 password = random_lower_string() 

154 user_in = UserCreate(email=username, password=password) 

155 crud.create_user(session=db, user_create=user_in) 

156 

157 username2 = random_email() 

158 password2 = random_lower_string() 

159 user_in2 = UserCreate(email=username2, password=password2) 

160 crud.create_user(session=db, user_create=user_in2) 

161 

162 r = client.get(f"{settings.API_V1_STR}/users/", headers=superuser_token_headers) 

163 all_users = r.json() 

164 

165 assert len(all_users["data"]) > 1 

166 assert "count" in all_users 

167 for item in all_users["data"]: 

168 assert "email" in item 

169 

170 

171def test_update_user_me( 

172 client: TestClient, normal_user_token_headers: dict[str, str], db: Session 

173) -> None: 

174 full_name = "Updated Name" 

175 email = random_email() 

176 data = {"full_name": full_name, "email": email} 

177 r = client.patch( 

178 f"{settings.API_V1_STR}/users/me", 

179 headers=normal_user_token_headers, 

180 json=data, 

181 ) 

182 assert r.status_code == 200 

183 updated_user = r.json() 

184 assert updated_user["email"] == email 

185 assert updated_user["full_name"] == full_name 

186 

187 user_query = select(User).where(User.email == email) 

188 user_db = db.exec(user_query).first() 

189 assert user_db 

190 assert user_db.email == email 

191 assert user_db.full_name == full_name 

192 

193 

194def test_update_password_me( 

195 client: TestClient, superuser_token_headers: dict[str, str], db: Session 

196) -> None: 

197 new_password = random_lower_string() 

198 data = { 

199 "current_password": settings.FIRST_SUPERUSER_PASSWORD, 

200 "new_password": new_password, 

201 } 

202 r = client.patch( 

203 f"{settings.API_V1_STR}/users/me/password", 

204 headers=superuser_token_headers, 

205 json=data, 

206 ) 

207 assert r.status_code == 200 

208 updated_user = r.json() 

209 assert updated_user["message"] == "Password updated successfully" 

210 

211 user_query = select(User).where(User.email == settings.FIRST_SUPERUSER) 

212 user_db = db.exec(user_query).first() 

213 assert user_db 

214 assert user_db.email == settings.FIRST_SUPERUSER 

215 assert verify_password(new_password, user_db.hashed_password) 

216 

217 # Revert to the old password to keep consistency in test 

218 old_data = { 

219 "current_password": new_password, 

220 "new_password": settings.FIRST_SUPERUSER_PASSWORD, 

221 } 

222 r = client.patch( 

223 f"{settings.API_V1_STR}/users/me/password", 

224 headers=superuser_token_headers, 

225 json=old_data, 

226 ) 

227 db.refresh(user_db) 

228 

229 assert r.status_code == 200 

230 assert verify_password(settings.FIRST_SUPERUSER_PASSWORD, user_db.hashed_password) 

231 

232 

233def test_update_password_me_incorrect_password( 

234 client: TestClient, superuser_token_headers: dict[str, str] 

235) -> None: 

236 new_password = random_lower_string() 

237 data = {"current_password": new_password, "new_password": new_password} 

238 r = client.patch( 

239 f"{settings.API_V1_STR}/users/me/password", 

240 headers=superuser_token_headers, 

241 json=data, 

242 ) 

243 assert r.status_code == 400 

244 updated_user = r.json() 

245 assert updated_user["detail"] == "Incorrect password" 

246 

247 

248def test_update_user_me_email_exists( 

249 client: TestClient, normal_user_token_headers: dict[str, str], db: Session 

250) -> None: 

251 username = random_email() 

252 password = random_lower_string() 

253 user_in = UserCreate(email=username, password=password) 

254 user = crud.create_user(session=db, user_create=user_in) 

255 

256 data = {"email": user.email} 

257 r = client.patch( 

258 f"{settings.API_V1_STR}/users/me", 

259 headers=normal_user_token_headers, 

260 json=data, 

261 ) 

262 assert r.status_code == 409 

263 assert r.json()["detail"] == "User with this email already exists" 

264 

265 

266def test_update_password_me_same_password_error( 

267 client: TestClient, superuser_token_headers: dict[str, str] 

268) -> None: 

269 data = { 

270 "current_password": settings.FIRST_SUPERUSER_PASSWORD, 

271 "new_password": settings.FIRST_SUPERUSER_PASSWORD, 

272 } 

273 r = client.patch( 

274 f"{settings.API_V1_STR}/users/me/password", 

275 headers=superuser_token_headers, 

276 json=data, 

277 ) 

278 assert r.status_code == 400 

279 updated_user = r.json() 

280 assert ( 

281 updated_user["detail"] == "New password cannot be the same as the current one" 

282 ) 

283 

284 

285def test_register_user(client: TestClient, db: Session) -> None: 

286 username = random_email() 

287 password = random_lower_string() 

288 full_name = random_lower_string() 

289 data = {"email": username, "password": password, "full_name": full_name} 

290 r = client.post( 

291 f"{settings.API_V1_STR}/users/signup", 

292 json=data, 

293 ) 

294 assert r.status_code == 200 

295 created_user = r.json() 

296 assert created_user["email"] == username 

297 assert created_user["full_name"] == full_name 

298 

299 user_query = select(User).where(User.email == username) 

300 user_db = db.exec(user_query).first() 

301 assert user_db 

302 assert user_db.email == username 

303 assert user_db.full_name == full_name 

304 assert verify_password(password, user_db.hashed_password) 

305 

306 

307def test_register_user_already_exists_error(client: TestClient) -> None: 

308 password = random_lower_string() 

309 full_name = random_lower_string() 

310 data = { 

311 "email": settings.FIRST_SUPERUSER, 

312 "password": password, 

313 "full_name": full_name, 

314 } 

315 r = client.post( 

316 f"{settings.API_V1_STR}/users/signup", 

317 json=data, 

318 ) 

319 assert r.status_code == 400 

320 assert r.json()["detail"] == "The user with this email already exists in the system" 

321 

322 

323def test_update_user( 

324 client: TestClient, superuser_token_headers: dict[str, str], db: Session 

325) -> None: 

326 username = random_email() 

327 password = random_lower_string() 

328 user_in = UserCreate(email=username, password=password) 

329 user = crud.create_user(session=db, user_create=user_in) 

330 

331 data = {"full_name": "Updated_full_name"} 

332 r = client.patch( 

333 f"{settings.API_V1_STR}/users/{user.id}", 

334 headers=superuser_token_headers, 

335 json=data, 

336 ) 

337 assert r.status_code == 200 

338 updated_user = r.json() 

339 

340 assert updated_user["full_name"] == "Updated_full_name" 

341 

342 user_query = select(User).where(User.email == username) 

343 user_db = db.exec(user_query).first() 

344 db.refresh(user_db) 

345 assert user_db 

346 assert user_db.full_name == "Updated_full_name" 

347 

348 

349def test_update_user_not_exists( 

350 client: TestClient, superuser_token_headers: dict[str, str] 

351) -> None: 

352 data = {"full_name": "Updated_full_name"} 

353 r = client.patch( 

354 f"{settings.API_V1_STR}/users/{uuid.uuid4()}", 

355 headers=superuser_token_headers, 

356 json=data, 

357 ) 

358 assert r.status_code == 404 

359 assert r.json()["detail"] == "The user with this id does not exist in the system" 

360 

361 

362def test_update_user_email_exists( 

363 client: TestClient, superuser_token_headers: dict[str, str], db: Session 

364) -> None: 

365 username = random_email() 

366 password = random_lower_string() 

367 user_in = UserCreate(email=username, password=password) 

368 user = crud.create_user(session=db, user_create=user_in) 

369 

370 username2 = random_email() 

371 password2 = random_lower_string() 

372 user_in2 = UserCreate(email=username2, password=password2) 

373 user2 = crud.create_user(session=db, user_create=user_in2) 

374 

375 data = {"email": user2.email} 

376 r = client.patch( 

377 f"{settings.API_V1_STR}/users/{user.id}", 

378 headers=superuser_token_headers, 

379 json=data, 

380 ) 

381 assert r.status_code == 409 

382 assert r.json()["detail"] == "User with this email already exists" 

383 

384 

385def test_delete_user_me(client: TestClient, db: Session) -> None: 

386 username = random_email() 

387 password = random_lower_string() 

388 user_in = UserCreate(email=username, password=password) 

389 user = crud.create_user(session=db, user_create=user_in) 

390 user_id = user.id 

391 

392 login_data = { 

393 "username": username, 

394 "password": password, 

395 } 

396 r = client.post(f"{settings.API_V1_STR}/login/access-token", data=login_data) 

397 tokens = r.json() 

398 a_token = tokens["access_token"] 

399 headers = {"Authorization": f"Bearer {a_token}"} 

400 

401 r = client.delete( 

402 f"{settings.API_V1_STR}/users/me", 

403 headers=headers, 

404 ) 

405 assert r.status_code == 200 

406 deleted_user = r.json() 

407 assert deleted_user["message"] == "User deleted successfully" 

408 result = db.exec(select(User).where(User.id == user_id)).first() 

409 assert result is None 

410 

411 user_query = select(User).where(User.id == user_id) 

412 user_db = db.execute(user_query).first() 

413 assert user_db is None 

414 

415 

416def test_delete_user_me_as_superuser( 

417 client: TestClient, superuser_token_headers: dict[str, str] 

418) -> None: 

419 r = client.delete( 

420 f"{settings.API_V1_STR}/users/me", 

421 headers=superuser_token_headers, 

422 ) 

423 assert r.status_code == 403 

424 response = r.json() 

425 assert response["detail"] == "Super users are not allowed to delete themselves" 

426 

427 

428def test_delete_user_super_user( 

429 client: TestClient, superuser_token_headers: dict[str, str], db: Session 

430) -> None: 

431 username = random_email() 

432 password = random_lower_string() 

433 user_in = UserCreate(email=username, password=password) 

434 user = crud.create_user(session=db, user_create=user_in) 

435 user_id = user.id 

436 r = client.delete( 

437 f"{settings.API_V1_STR}/users/{user_id}", 

438 headers=superuser_token_headers, 

439 ) 

440 assert r.status_code == 200 

441 deleted_user = r.json() 

442 assert deleted_user["message"] == "User deleted successfully" 

443 result = db.exec(select(User).where(User.id == user_id)).first() 

444 assert result is None 

445 

446 

447def test_delete_user_not_found( 

448 client: TestClient, superuser_token_headers: dict[str, str] 

449) -> None: 

450 r = client.delete( 

451 f"{settings.API_V1_STR}/users/{uuid.uuid4()}", 

452 headers=superuser_token_headers, 

453 ) 

454 assert r.status_code == 404 

455 assert r.json()["detail"] == "User not found" 

456 

457 

458def test_delete_user_current_super_user_error( 

459 client: TestClient, superuser_token_headers: dict[str, str], db: Session 

460) -> None: 

461 super_user = crud.get_user_by_email(session=db, email=settings.FIRST_SUPERUSER) 

462 assert super_user 

463 user_id = super_user.id 

464 

465 r = client.delete( 

466 f"{settings.API_V1_STR}/users/{user_id}", 

467 headers=superuser_token_headers, 

468 ) 

469 assert r.status_code == 403 

470 assert r.json()["detail"] == "Super users are not allowed to delete themselves" 

471 

472 

473def test_delete_user_without_privileges( 

474 client: TestClient, normal_user_token_headers: dict[str, str], db: Session 

475) -> None: 

476 username = random_email() 

477 password = random_lower_string() 

478 user_in = UserCreate(email=username, password=password) 

479 user = crud.create_user(session=db, user_create=user_in) 

480 

481 r = client.delete( 

482 f"{settings.API_V1_STR}/users/{user.id}", 

483 headers=normal_user_token_headers, 

484 ) 

485 assert r.status_code == 403 

486 assert r.json()["detail"] == "The user doesn't have enough privileges"