Coverage for bc/kwai-bc-identity/src/kwai_bc_identity/tokens/log_user_login_db_service.py: 100%

1"""Module that defines a LogUserLoginService with a database.""" 

2 

3from kwai_core.db.database import Database 

4 

5from kwai_bc_identity.tokens.log_user_login_service import LogUserLoginService 

6from kwai_bc_identity.tokens.refresh_token import RefreshTokenEntity 

7from kwai_bc_identity.tokens.user_log import UserLogEntity 

8from kwai_bc_identity.tokens.user_log_db_repository import UserLogDbRepository 

9from kwai_bc_identity.tokens.value_objects import IpAddress, OpenId 

10from kwai_bc_identity.users.user_account import UserAccountEntity 

11 

12 

13class LogUserLoginDbService(LogUserLoginService): 

14 """Logs a login user request to the database.""" 

15 

16 def __init__( 

17 self, 

18 database: Database, 

19 *, 

20 email: str, 

21 client_ip: str, 

22 user_agent: str, 

23 open_id_sub: str = "", 

24 open_id_provider: str = "", 

25 ): 

26 self._db = database 

27 self._email = email 

28 if client_ip == "testclient": 

29 self._client_ip = IpAddress.create("127.0.0.1") 

30 else: 

31 self._client_ip = IpAddress.create(client_ip) 

32 self._user_agent = user_agent 

33 self._openId = OpenId(sub=open_id_sub, provider=open_id_provider) 

34 

35 async def notify_failure( 

36 self, 

37 message: str = "", 

38 *, 

39 user_account: UserAccountEntity | None = None, 

40 refresh_token: RefreshTokenEntity | None = None, 

41 ) -> None: 

42 repo = UserLogDbRepository(self._db) 

43 await repo.create( 

44 UserLogEntity( 

45 success=False, 

46 email=self._email, 

47 user_account=user_account, 

48 client_ip=self._client_ip, 

49 user_agent=self._user_agent, 

50 remark=message, 

51 openid=self._openId, 

52 ) 

53 ) 

54 

55 async def notify_success( 

56 self, 

57 *, 

58 user_account: UserAccountEntity | None = None, 

59 refresh_token: RefreshTokenEntity | None = None, 

60 ) -> None: 

61 repo = UserLogDbRepository(self._db) 

62 await repo.create( 

63 UserLogEntity( 

64 success=True, 

65 email=self._email, 

66 user_account=user_account, 

67 refresh_token=refresh_token, 

68 client_ip=self._client_ip, 

69 user_agent=self._user_agent, 

70 openid=self._openId, 

71 ) 

72 )