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

1"""Module for testing the recreate user invitation use case.""" 

2 

3import pytest 

4 

5from kwai_bc_identity.recreate_user_invitation import ( 

6 RecreateUserInvitation, 

7 RecreateUserInvitationCommand, 

8) 

9from kwai_bc_identity.user_invitations.user_invitation_db_repository import ( 

10 UserInvitationDbRepository, 

11) 

12from kwai_bc_identity.users.user import UserEntity 

13from kwai_bc_identity.users.user_db_repository import UserDbRepository 

14from kwai_core.db.database import Database 

15from kwai_core.db.uow import UnitOfWork 

16from kwai_core.domain.value_objects.email_address import EmailAddress 

17from kwai_core.domain.value_objects.name import Name 

18 

19 

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

21 

22 

23async def test_recreate_user_invitation( 

24 database: Database, 

25 user: UserEntity, 

26 publisher, 

27 make_user_invitation, 

28 make_user_invitation_in_db, 

29): 

30 """Test recreate user invitation.""" 

31 user_invitation = await make_user_invitation_in_db( 

32 user_invitation=make_user_invitation( 

33 email=EmailAddress(email="anton.geesink@kwai.com"), 

34 name=Name(first_name="Anton", last_name="Geesink"), 

35 user=user, 

36 ) 

37 ) 

38 user_invitation_repo = UserInvitationDbRepository(database) 

39 async with UnitOfWork(database): 

40 new_user_invitation = await RecreateUserInvitation( 

41 user, 

42 UserDbRepository(database), 

43 user_invitation_repo, 

44 publisher, 

45 ).execute(RecreateUserInvitationCommand(uuid=str(user_invitation.uuid))) 

46 assert new_user_invitation is not None, "There should be a user invitation" 

47 

48 async with UnitOfWork(database): 

49 await user_invitation_repo.delete(new_user_invitation) 

50 

51 old_user_invitation = await user_invitation_repo.get_invitation_by_id( 

52 user_invitation.id 

53 ) 

54 assert old_user_invitation.revoked, "Old user invitation should be revoked"