Coverage for src/kwai/modules/teams/create_team_member.py: 100%

1"""Module that defines the use case for creating a team member.""" 

2 

3from dataclasses import dataclass 

4 

5from kwai.core.domain.presenter import Presenter 

6from kwai.core.domain.value_objects.unique_id import UniqueId 

7from kwai.modules.teams.domain.team import TeamEntity, TeamIdentifier 

8from kwai.modules.teams.domain.team_member import TeamMember 

9from kwai.modules.teams.repositories.member_repository import MemberRepository 

10from kwai.modules.teams.repositories.team_repository import TeamRepository 

11 

12 

13@dataclass(frozen=True, kw_only=True, slots=True) 

14class CreateTeamMemberCommand: 

15 """Input for the 'Create Team Member' use case.""" 

16 

17 team_id: int 

18 member_id: str 

19 active: bool = True 

20 

21 

22class CreateTeamMember: 

23 """Implements the 'Create Team Member' use case.""" 

24 

25 def __init__( 

26 self, 

27 team_repository: TeamRepository, 

28 member_repository: MemberRepository, 

29 presenter: Presenter[tuple[TeamMember, TeamEntity]], 

30 ): 

31 """Initialize the use case. 

32 

33 Args: 

34 team_repository: A repository for retrieving the team. 

35 member_repository: A repository for retrieving the member. 

36 presenter: A Presenter for processing the result. 

37 """ 

38 self._team_repository = team_repository 

39 self._member_repository = member_repository 

40 self._presenter = presenter 

41 

42 async def execute(self, command: CreateTeamMemberCommand) -> None: 

43 """Execute the use case. 

44 

45 Raises: 

46 TeamNotFoundException: If the team does not exist. 

47 MemberNotFoundException: If the member does not exist. 

48 TeamMemberAlreadyExistException: If the member is already part of the team. 

49 """ 

50 team_query = self._team_repository.create_query().filter_by_id( 

51 TeamIdentifier(command.team_id) 

52 ) 

53 team = await self._team_repository.get(team_query) 

54 

55 member_query = self._member_repository.create_query().filter_by_uuid( 

56 UniqueId.create_from_string(command.member_id) 

57 ) 

58 member = await self._member_repository.get(member_query) 

59 

60 team_member = TeamMember(member=member, active=command.active) 

61 team.add_member(team_member) 

62 

63 await self._team_repository.add_team_member(team, team_member) 

64 

65 self._presenter.present((team_member, team))