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

1"""Module for testing the use case 'Accept User Invitation'.""" 

2 

3import pytest 

4 

5from kwai_bc_identity.accept_user_invitation import ( 

6 AcceptUserInvitation, 

7 AcceptUserInvitationCommand, 

8) 

9from kwai_bc_identity.user_invitations.user_invitation_db_repository import ( 

10 UserInvitationDbRepository, 

11) 

12from kwai_bc_identity.users.user_account import UserAccountEntity 

13from kwai_bc_identity.users.user_account_db_repository import ( 

14 UserAccountDbRepository, 

15) 

16from kwai_core.db.database import Database 

17from kwai_core.db.uow import UnitOfWork 

18from kwai_core.domain.exceptions import UnprocessableException 

19from kwai_core.domain.presenter import Presenter 

20from kwai_core.domain.value_objects.email_address import EmailAddress 

21from kwai_core.domain.value_objects.name import Name 

22from kwai_core.domain.value_objects.timestamp import Timestamp 

23 

24 

25pytestmark = [pytest.mark.db] 

26 

27 

28class DummyPresenter(Presenter[UserAccountEntity]): 

29 """A dummy presenter for testing AcceptUserInvitation.""" 

30 

31 def __init__(self): 

32 self._entity = None 

33 

34 def present(self, use_case_result: UserAccountEntity) -> None: 

35 self._entity = use_case_result 

36 

37 @property 

38 def entity(self) -> UserAccountEntity: 

39 """Return the entity returned by the use case.""" 

40 return self._entity 

41 

42 

43async def test_accept_user_invitation( 

44 database: Database, 

45 make_user_account_in_db, 

46 make_user_invitation, 

47 make_user_invitation_in_db, 

48): 

49 """Test accepting a user invitation.""" 

50 user = await make_user_account_in_db() 

51 user_invitation = await make_user_invitation_in_db( 

52 user_invitation=make_user_invitation( 

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

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

55 user=user, 

56 ) 

57 ) 

58 

59 presenter = DummyPresenter() 

60 command = AcceptUserInvitationCommand( 

61 uuid=str(user_invitation.uuid), 

62 first_name=user_invitation.name.first_name, 

63 last_name=user_invitation.name.last_name, 

64 remark="Created with test_accept_user_invitation", 

65 password="Test1234", 

66 ) 

67 

68 user_account_repo = UserAccountDbRepository(database) 

69 

70 async with UnitOfWork(database): 

71 await AcceptUserInvitation( 

72 UserInvitationDbRepository(database), 

73 user_account_repo, 

74 presenter, 

75 ).execute(command) 

76 

77 assert presenter.entity, "There should be a new user account" 

78 

79 async with UnitOfWork(database): 

80 await user_account_repo.delete(presenter.entity) 

81 

82 

83async def test_dont_accept_revoked_user_invitation( 

84 database: Database, 

85 make_user_account_in_db, 

86 make_user_invitation, 

87 make_user_invitation_in_db, 

88): 

89 """Test if accepting a revoked user invitation results in an exception.""" 

90 user = await make_user_account_in_db() 

91 user_invitation = await make_user_invitation_in_db( 

92 user_invitation=make_user_invitation( 

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

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

95 user=user, 

96 revoked=True, 

97 ) 

98 ) 

99 

100 presenter = DummyPresenter() 

101 command = AcceptUserInvitationCommand( 

102 uuid=str(user_invitation.uuid), 

103 first_name=user_invitation.name.first_name, 

104 last_name=user_invitation.name.last_name, 

105 remark="Created with test_dont_accept_revoked_user_invitation", 

106 password="Test1234", 

107 ) 

108 

109 user_account_repo = UserAccountDbRepository(database) 

110 

111 with pytest.raises( 

112 UnprocessableException, 

113 match=f"The user invitation with id {user_invitation.uuid} is revoked.", 

114 ): 

115 await AcceptUserInvitation( 

116 UserInvitationDbRepository(database), 

117 user_account_repo, 

118 presenter, 

119 ).execute(command) 

120 

121 

122async def test_dont_accept_expired_user_invitation( 

123 database: Database, 

124 make_user_account_in_db, 

125 make_user_invitation, 

126 make_user_invitation_in_db, 

127): 

128 """Test if accepting an expired user invitation results in an exception.""" 

129 user = await make_user_account_in_db() 

130 user_invitation = await make_user_invitation_in_db( 

131 user_invitation=make_user_invitation( 

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

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

134 user=user, 

135 expired_at=Timestamp.create_with_delta(days=-1), 

136 ) 

137 ) 

138 

139 presenter = DummyPresenter() 

140 command = AcceptUserInvitationCommand( 

141 uuid=str(user_invitation.uuid), 

142 first_name=user_invitation.name.first_name, 

143 last_name=user_invitation.name.last_name, 

144 remark="Created with test_dont_accept_expired_user_invitation", 

145 password="Test1234", 

146 ) 

147 

148 user_account_repo = UserAccountDbRepository(database) 

149 

150 with pytest.raises( 

151 UnprocessableException, 

152 match=f"The user invitation with id {user_invitation.uuid} is expired.", 

153 ): 

154 await AcceptUserInvitation( 

155 UserInvitationDbRepository(database), 

156 user_account_repo, 

157 presenter, 

158 ).execute(command) 

159 

160 

161async def test_dont_accept_already_accepted_user_invitation( 

162 database: Database, 

163 make_user_account_in_db, 

164 make_user_invitation, 

165 make_user_invitation_in_db, 

166): 

167 """Test if accepting an already accepted user invitation results in an exception.""" 

168 user = await make_user_account_in_db() 

169 user_invitation = await make_user_invitation_in_db( 

170 user_invitation=make_user_invitation( 

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

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

173 user=user, 

174 confirmed_at=Timestamp.create_now(), 

175 ) 

176 ) 

177 

178 presenter = DummyPresenter() 

179 command = AcceptUserInvitationCommand( 

180 uuid=str(user_invitation.uuid), 

181 first_name=user_invitation.name.first_name, 

182 last_name=user_invitation.name.last_name, 

183 remark="Created with test_dont_accept_already_accepted_user_invitation", 

184 password="Test1234", 

185 ) 

186 

187 user_account_repo = UserAccountDbRepository(database) 

188 

189 with pytest.raises( 

190 UnprocessableException, 

191 match=f"The user invitation with id {user_invitation.uuid} was already accepted.", 

192 ): 

193 await AcceptUserInvitation( 

194 UserInvitationDbRepository(database), 

195 user_account_repo, 

196 presenter, 

197 ).execute(command) 

198 

199 

200async def test_dont_accept_user_invitation_with_used_email( 

201 database: Database, 

202 make_user, 

203 make_user_account, 

204 make_user_account_in_db, 

205 make_user_invitation, 

206 make_user_invitation_in_db, 

207): 

208 """Test if accepting a user invitation with an already used email fails.""" 

209 user = await make_user_account_in_db( 

210 user_account=make_user_account( 

211 user=make_user(email=EmailAddress(email="anton.geesink@kwai.com")) 

212 ) 

213 ) 

214 user_invitation = await make_user_invitation_in_db( 

215 user_invitation=make_user_invitation( 

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

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

218 user=user, 

219 ) 

220 ) 

221 

222 presenter = DummyPresenter() 

223 command = AcceptUserInvitationCommand( 

224 uuid=str(user_invitation.uuid), 

225 first_name=user_invitation.name.first_name, 

226 last_name=user_invitation.name.last_name, 

227 remark="Created with test_dont_accept_user_invitation_with_used_email", 

228 password="Test1234", 

229 ) 

230 

231 user_account_repo = UserAccountDbRepository(database) 

232 

233 with pytest.raises( 

234 UnprocessableException, 

235 match=f"A user with email {user_invitation.email} already exists.", 

236 ): 

237 await AcceptUserInvitation( 

238 UserInvitationDbRepository(database), 

239 user_account_repo, 

240 presenter, 

241 ).execute(command)