Coverage for microservice_websocket/app/services/database/user_manager.py: 84%

1from typing import Optional 

2 

3from beanie import PydanticObjectId 

4from passlib.context import CryptContext 

5 

6from ...blueprints.api.models import UpdateUserPayload 

7from ...utils.exceptions import NotFoundException 

8from ..database import User 

9 

10pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") 

11 

12 

13def verify_password(plain_password: str, hashed_password: str) -> bool: 

14 return pwd_context.verify(plain_password, hashed_password) 

15 

16 

17async def update_password(user: User, old_password: str, new_password: str) -> bool: 

18 if not await auth_user(user.email, old_password): 18 ↛ 19line 18 didn't jump to line 19, because the condition on line 18 was never true

19 return False 

20 

21 user.hashed_password = hash_password(new_password) 

22 await user.save() 

23 

24 return True 

25 

26 

27def hash_password(plain_password: str) -> str: 

28 return pwd_context.hash(plain_password) 

29 

30 

31async def auth_user(email: str, password: str) -> Optional[User]: 

32 user = await User.find(User.email == email).first_or_none() 

33 

34 if not user: 

35 return None 

36 if not verify_password(password, user.hashed_password): 36 ↛ 37line 36 didn't jump to line 37, because the condition on line 36 was never true

37 return None 

38 return user 

39 

40 

41async def get_user_from_mail(email: str) -> Optional[User]: 

42 return await User.find(User.email == email).first_or_none() 

43 

44 

45async def create_user( 

46 email: str, password: str, first_name: str, last_name: str, role: str = "standard" 

47) -> Optional[User]: 

48 if await get_user_from_mail(email): 

49 return None 

50 

51 user = User( 

52 email=email, 

53 hashed_password=hash_password(password), 

54 first_name=first_name, 

55 last_name=last_name, 

56 role=role, 

57 ) 

58 await user.save() 

59 return user 

60 

61 

62async def update_user(user_id: str, payload: UpdateUserPayload) -> bool: 

63 user = await User.get(PydanticObjectId(user_id)) 

64 if user is None: 

65 raise NotFoundException("User") 

66 

67 if payload.email is not None: 67 ↛ 70line 67 didn't jump to line 70, because the condition on line 67 was never false

68 user.email = payload.email 

69 

70 if payload.first_name is not None: 70 ↛ 71line 70 didn't jump to line 71, because the condition on line 70 was never true

71 user.first_name = payload.first_name 

72 

73 if payload.last_name is not None: 73 ↛ 74line 73 didn't jump to line 74, because the condition on line 73 was never true

74 user.last_name = payload.last_name 

75 

76 if payload.role is not None: 76 ↛ 79line 76 didn't jump to line 79, because the condition on line 76 was never false

77 user.role = payload.role 

78 

79 if ( 79 ↛ 85line 79 didn't jump to line 85

80 payload.old_password 

81 and payload.new_password 

82 and payload.old_password != payload.new_password 

83 and not await update_password(user, payload.old_password, payload.new_password) 

84 ): 

85 return False 

86 

87 await user.save() 

88 

89 return True 

90 

91 

92async def delete_user(user_id: str): 

93 user = await User.get(PydanticObjectId(user_id)) 

94 if user is None: 

95 raise NotFoundException("User") 

96 

97 await user.delete()