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

1"""Module for testing the use case: invite a user.""" 

2 

3import pytest 

4 

5from kwai_bc_identity.invite_user import InviteUser, InviteUserCommand 

6from kwai_bc_identity.user_invitations.user_invitation import UserInvitationEntity 

7from kwai_bc_identity.user_invitations.user_invitation_db_repository import ( 

8 UserInvitationDbRepository, 

9) 

10from kwai_bc_identity.user_invitations.user_invitation_repository import ( 

11 UserInvitationRepository, 

12) 

13from kwai_bc_identity.users.user import UserEntity 

14from kwai_bc_identity.users.user_db_repository import UserDbRepository 

15from kwai_core.db.database import Database 

16from kwai_core.domain.exceptions import UnprocessableException 

17from kwai_core.domain.presenter import EntityPresenter 

18from kwai_core.events.publisher import Publisher 

19 

20 

21pytestmark = [pytest.mark.db, pytest.mark.bus] 

22 

23 

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

25def repo(database: Database) -> UserInvitationRepository: 

26 """Create a user invitation repository.""" 

27 return UserInvitationDbRepository(database) 

28 

29 

30async def test_invite_user( 

31 database: Database, 

32 repo: UserInvitationRepository, 

33 user: UserEntity, 

34 publisher: Publisher, 

35): 

36 """Test use case: invite a user.""" 

37 user_repo = UserDbRepository(database) 

38 command = InviteUserCommand( 

39 email="ichiro.abe@kwai.com", 

40 first_name="Ichiro", 

41 last_name="Abe", 

42 remark="Created with pytest test_invite_user", 

43 ) 

44 presenter = EntityPresenter[UserInvitationEntity]() 

45 await InviteUser( 

46 user=user, 

47 user_repo=user_repo, 

48 user_invitation_repo=repo, 

49 presenter=presenter, 

50 publisher=publisher, 

51 ).execute(command) 

52 

53 assert presenter.entity is not None, "There should be a user invitation" 

54 

55 

56async def test_user_already_exists( 

57 database: Database, 

58 repo: UserInvitationRepository, 

59 make_user_account_in_db, 

60 publisher: Publisher, 

61): 

62 """Test if an exception is raised when a user with the email already exists.""" 

63 user_account = await make_user_account_in_db() 

64 user_repo = UserDbRepository(database) 

65 presenter = EntityPresenter[UserInvitationEntity]() 

66 command = InviteUserCommand( 

67 email=str(user_account.user.email), 

68 first_name="Jigoro", 

69 last_name="Kano", 

70 remark="Created with pytest test_user_already_exists", 

71 ) 

72 

73 with pytest.raises(UnprocessableException): 

74 await InviteUser( 

75 user=user_account.user, 

76 user_repo=user_repo, 

77 user_invitation_repo=repo, 

78 presenter=presenter, 

79 publisher=publisher, 

80 ).execute(command) 

81 

82 

83async def test_already_invited_user( 

84 database: Database, 

85 repo: UserInvitationRepository, 

86 make_user_account_in_db, 

87 make_user_invitation_in_db, 

88 publisher: Publisher, 

89): 

90 """Test if an exception is raised when there is an invitation pending.""" 

91 user_account = await make_user_account_in_db() 

92 # Make already a user invitation 

93 user_invitation = await make_user_invitation_in_db() 

94 

95 # Try to make the same user invitation 

96 user_repo = UserDbRepository(database) 

97 presenter = EntityPresenter[UserInvitationEntity]() 

98 

99 command = InviteUserCommand( 

100 email=str(user_invitation.email), 

101 first_name=user_account.user.name.first_name, 

102 last_name=user_account.user.name.last_name, 

103 remark="Created with pytest test_already_invited_user", 

104 ) 

105 

106 with pytest.raises(UnprocessableException): 

107 await InviteUser( 

108 user=user_account.user, 

109 user_repo=user_repo, 

110 user_invitation_repo=repo, 

111 presenter=presenter, 

112 publisher=publisher, 

113 ).execute(command)