Coverage for src/tests/modules/identity/test_mail_user_recovery.py: 100%

1"""Module for testing mail_user_recovery.""" 

2 

3from typing import AsyncGenerator 

4 

5import pytest 

6 

7from kwai_bc_identity.mail_user_recovery import ( 

8 MailUserRecovery, 

9 MailUserRecoveryCommand, 

10) 

11from kwai_bc_identity.user_recoveries.user_recovery import UserRecoveryEntity 

12from kwai_bc_identity.user_recoveries.user_recovery_db_repository import ( 

13 UserRecoveryDbRepository, 

14) 

15from kwai_bc_identity.user_recoveries.user_recovery_repository import ( 

16 UserRecoveryRepository, 

17) 

18from kwai_bc_identity.users.user import UserEntity 

19from kwai_core.db.database import Database 

20from kwai_core.domain.exceptions import UnprocessableException 

21from kwai_core.domain.value_objects.timestamp import Timestamp 

22from kwai_core.mail.mailer import Mailer 

23from kwai_core.mail.recipient import Recipients 

24from kwai_core.template.mail_template import MailTemplate 

25 

26 

27pytestmark = [pytest.mark.db, pytest.mark.mail] 

28 

29 

30@pytest.fixture(scope="module") 

31def repo(database: Database) -> UserRecoveryRepository: 

32 """Create a user recovery repository.""" 

33 return UserRecoveryDbRepository(database) 

34 

35 

36@pytest.fixture(scope="module") 

37async def user_recovery( 

38 repo: UserRecoveryRepository, user: UserEntity 

39) -> AsyncGenerator[UserRecoveryEntity, None]: 

40 """Create a user recovery.""" 

41 user_recovery = UserRecoveryEntity( 

42 expiration=Timestamp.create_with_delta(hours=2), 

43 user=user, 

44 ) 

45 entity = await repo.create(user_recovery) 

46 yield entity 

47 await repo.delete(entity) 

48 

49 

50async def test_mail_user_recovery( 

51 repo: UserRecoveryRepository, 

52 user_recovery: UserRecoveryEntity, 

53 mailer: Mailer, 

54 recipients: Recipients, 

55 recovery_mail_template: MailTemplate, 

56): 

57 """Test use case mail user recovery.""" 

58 command = MailUserRecoveryCommand(uuid=str(user_recovery.uuid)) 

59 updated_user_recovery = await MailUserRecovery( 

60 repo, 

61 mailer, 

62 recipients, 

63 recovery_mail_template, 

64 ).execute(command) 

65 

66 assert updated_user_recovery.mailed is not None, "mailed should be set." 

67 

68 

69async def test_mail_user_recovery_already_mailed( 

70 repo: UserRecoveryRepository, 

71 user_recovery: UserRecoveryEntity, 

72 mailer: Mailer, 

73 recipients: Recipients, 

74 recovery_mail_template: MailTemplate, 

75): 

76 """Test when a user recovery is already mailed.""" 

77 command = MailUserRecoveryCommand(uuid=str(user_recovery.uuid)) 

78 with pytest.raises(UnprocessableException): 

79 await MailUserRecovery( 

80 repo, 

81 mailer, 

82 recipients, 

83 recovery_mail_template, 

84 ).execute(command)