Coverage for src/tests/fixtures/identity/user_invitations.py: 100%

1"""Module for defining factory fixtures for user invitations.""" 

2 

3import pytest 

4 

5from kwai_bc_identity.user_invitations.user_invitation import UserInvitationEntity 

6from kwai_bc_identity.user_invitations.user_invitation_db_repository import ( 

7 UserInvitationDbRepository, 

8) 

9from kwai_bc_identity.users.user import UserEntity 

10from kwai_bc_identity.users.user_account import UserAccountEntity 

11from kwai_core.db.uow import UnitOfWork 

12from kwai_core.domain.value_objects.email_address import EmailAddress 

13from kwai_core.domain.value_objects.name import Name 

14from kwai_core.domain.value_objects.timestamp import Timestamp 

15 

16 

17@pytest.fixture 

18def make_user_invitation(make_user): 

19 """Factory fixture for creating a user invitation.""" 

20 

21 def _make_user_invitation( 

22 email: EmailAddress | None = None, 

23 name: Name | None = None, 

24 user: UserEntity | None = None, 

25 revoked: bool = False, 

26 expired_at: Timestamp | None = None, 

27 confirmed_at: Timestamp | None = None, 

28 ) -> UserInvitationEntity: 

29 return UserInvitationEntity( 

30 email=email or EmailAddress("ichiro.abe@kwai.com"), 

31 name=name or Name(first_name="Ichiro", last_name="Abe"), 

32 user=user or make_user(), 

33 revoked=revoked, 

34 expired_at=expired_at or Timestamp.create_with_delta(days=7), 

35 confirmed_at=confirmed_at or Timestamp(), 

36 ) 

37 

38 return _make_user_invitation 

39 

40 

41@pytest.fixture 

42async def make_user_invitation_in_db( 

43 database, make_user_invitation, make_user_account_in_db 

44): 

45 """Factory fixture for creating a user invitation in the database.""" 

46 created_entities = [] 

47 

48 async def _make_user_invitation_in_db( 

49 user_invitation: UserInvitationEntity | None = None, 

50 user_account: UserAccountEntity | None = None, 

51 ) -> UserInvitationEntity: 

52 if user_account is None: 

53 user_account = await make_user_account_in_db() 

54 user_invitation = user_invitation or make_user_invitation( 

55 user=user_account.user 

56 ) 

57 repo = UserInvitationDbRepository(database=database) 

58 

59 async with UnitOfWork(database): 

60 user_invitation = await repo.create(user_invitation) 

61 

62 created_entities.append(user_invitation) 

63 

64 return user_invitation 

65 

66 yield _make_user_invitation_in_db 

67 

68 repo = UserInvitationDbRepository(database=database) 

69 for entity in created_entities: 

70 async with UnitOfWork(database): 

71 await repo.delete(entity)