Coverage for microservice_websocket/app/blueprints/api/user.py: 89%

1from fastapi import APIRouter, Depends, HTTPException, status 

2from pydantic import BaseModel 

3 

4from ...services.database import User 

5from ...services.jwt import get_user_from_jwt 

6from ...utils.admin import verify_admin 

7from ...utils.user import ( 

8 create_user, 

9 delete_user, 

10 get_user_info, 

11 get_user_list, 

12 update_user, 

13) 

14from .models import CreateUserPayload, UpdateUserPayload 

15 

16user_router = APIRouter() 

17 

18 

19class UserListResponse(BaseModel): 

20 users: list[User.Serialized] 

21 

22 

23@user_router.get("/users", response_model=UserListResponse) 

24async def get_user_list_route(user: User = Depends(get_user_from_jwt)): 

25 verify_admin(user) 

26 

27 users: list[User] = await get_user_list() 

28 

29 return {"users": [x.serialize() for x in users]} 

30 

31 

32@user_router.get("/user/{user_id}", response_model=User.Serialized) 

33async def get_user_info_route(user_id: str, user: User = Depends(get_user_from_jwt)): 

34 verify_admin(user) 

35 

36 user_info = await get_user_info(user_id) 

37 

38 return user_info.serialize() 

39 

40 

41@user_router.get("/user", response_model=User.Serialized) 

42async def get_own_user_info_route(user: User = Depends(get_user_from_jwt)): 

43 return user.serialize() 

44 

45 

46@user_router.post("/user") 

47async def create_user_route( 

48 payload: CreateUserPayload, user: User = Depends(get_user_from_jwt) 

49): 

50 verify_admin(user) 

51 

52 if not await create_user(payload): 

53 raise HTTPException( 

54 status_code=status.HTTP_400_BAD_REQUEST, detail="User Already Existing" 

55 ) 

56 

57 return {"message": "Created"} 

58 

59 

60@user_router.put("/user/{user_id}") 

61async def update_user_route( 

62 payload: UpdateUserPayload, user_id: str, user: User = Depends(get_user_from_jwt) 

63): 

64 if str(user.id) != user_id: 64 ↛ 67line 64 didn't jump to line 67, because the condition on line 64 was never false

65 verify_admin(user) 

66 

67 if not await update_user(user_id, payload): 67 ↛ 68line 67 didn't jump to line 68, because the condition on line 67 was never true

68 raise HTTPException( 

69 status_code=status.HTTP_401_UNAUTHORIZED, 

70 detail="Old Password Doesn't Match", 

71 ) 

72 

73 return {"message": "Updated"} 

74 

75 

76@user_router.delete("/user/{user_id}") 

77async def delete_user_route(user_id: str, user: User = Depends(get_user_from_jwt)): 

78 if str(user.id) == user_id: 78 ↛ 79line 78 didn't jump to line 79, because the condition on line 78 was never true

79 return {"message": "Cannot Delete Current Active User"}, 400 

80 

81 verify_admin(user) 

82 

83 await delete_user(user_id) 

84 

85 return {"message": "Deleted"}