Coverage for apps/inners/use_cases/authentications/login_authentication.py: 58%

1import uuid 

2from datetime import datetime, timezone, timedelta 

3 

4import bcrypt 

5from starlette.datastructures import State 

6 

7from apps.inners.exceptions import repository_exception 

8from apps.inners.exceptions import use_case_exception 

9from apps.inners.models.daos.account import Account 

10from apps.inners.models.daos.session import Session 

11from apps.inners.models.dtos.contracts.requests.authentications.logins.login_by_email_and_password_body import \ 

12 LoginByEmailAndPasswordBody 

13from apps.inners.models.dtos.contracts.responses.authentications.logins.login_response import LoginResponse 

14from apps.inners.use_cases.managements.account_management import AccountManagement 

15from apps.inners.use_cases.managements.session_management import SessionManagement 

16 

17 

18class LoginAuthentication: 

19 def __init__( 

20 self, 

21 account_management: AccountManagement, 

22 session_management: SessionManagement 

23 ): 

24 self.account_management = account_management 

25 self.session_management = session_management 

26 

27 async def login_by_email_and_password(self, state: State, body: LoginByEmailAndPasswordBody) \ 

28 -> LoginResponse: 

29 found_account_by_email: Account = await self.account_management.find_one_by_email( 

30 state=state, 

31 email=body.email 

32 ) 

33 is_password_matched: bool = bcrypt.checkpw( 

34 body.password.encode(), 

35 found_account_by_email.password.encode() 

36 ) 

37 if not is_password_matched: 

38 raise use_case_exception.PasswordNotMatched() 

39 

40 try: 

41 found_session: Session = await self.session_management.find_one_by_account_id( 

42 state=state, 

43 account_id=found_account_by_email.id 

44 ) 

45 current_session: Session = found_session 

46 except repository_exception.NotFound: 

47 current_time = datetime.now(tz=timezone.utc) 

48 session_creator: Session = Session( 

49 id=uuid.uuid4(), 

50 account_id=found_account_by_email.id, 

51 access_token=str(uuid.uuid4()), 

52 refresh_token=str(uuid.uuid4()), 

53 access_token_expired_at=current_time + timedelta(minutes=10), 

54 refresh_token_expired_at=current_time + timedelta(days=7), 

55 ) 

56 created_session: Session = self.session_management.create_one_raw( 

57 state=state, 

58 session_creator=session_creator 

59 ) 

60 current_session: Session = created_session 

61 

62 login_response: LoginResponse = LoginResponse( 

63 account=found_account_by_email, 

64 session=current_session 

65 ) 

66 return login_response