Coverage for functions \ flipdare \ service \ _user_mixin.py: 71%

1#!/usr/bin/env python 

2# Copyright (c) 2026 Flipdare Pty Ltd. All rights reserved. 

3# 

4# This file is part of Flipdare's proprietary software and contains 

5# confidential and copyrighted material. Unauthorised copying, 

6# modification, distribution, or use of this file is strictly 

7# prohibited without prior written permission from Flipdare Pty Ltd. 

8# 

9# This software includes third-party components licensed under MIT, 

10# BSD, and Apache 2.0 licences. See THIRD_PARTY_NOTICES for details. 

11# 

12 

13 

14from typing import Any, NoReturn, Protocol 

15from typing import runtime_checkable 

16 

17from flipdare.backend.app_logger import AppLogger 

18from flipdare.app_log import LOG 

19from flipdare.app_types import DatabaseDict 

20from flipdare.error.app_error import AppError 

21from flipdare.error.data_load_error import DataLoadError 

22from flipdare.firestore.user_db import UserDb 

23from flipdare.generated.shared.app_error_code import AppErrorCode 

24from flipdare.message.error_message import ErrorMessage 

25from flipdare.request.app_request import AppRequest 

26from flipdare.util.error_util import ErrorUtil 

27from flipdare.wrapper.user_wrapper import UserWrapper 

28 

29 

30@runtime_checkable 

31class UserMixinRequirements(Protocol): 

32 """ 

33 We need this so can call user mixin methods 

34 from within the user mixin .. 

35 """ 

36 

37 @property 

38 def app_logger(self) -> AppLogger: ... 

39 @property 

40 def user_db(self) -> UserDb: ... 

41 

42 def log_and_throw( 

43 self, 

44 endpoint: str, 

45 error_code: AppErrorCode = AppErrorCode.SERVER, 

46 message: str | None = None, 

47 cause: Any | None = None, 

48 ) -> NoReturn: ... 

49 

50 

51class UserMixin: 

52 # NOTE: must have no slots otherwise conflicts can occur. 

53 __slots__ = () 

54 

55 def authenticate_request(self: UserMixinRequirements, req: AppRequest[Any]) -> None: 

56 try: 

57 req.is_authenticated() 

58 except DataLoadError as e: 

59 msg = f"Pin request not authenticated: {e!s}" 

60 LOG().error(msg) 

61 raise 

62 except AppError as error: 

63 msg = f"Invalid HTTP or not authenticated in confirm_pin: {req.method}:\n\t{error!s}" 

64 LOG().error(msg) 

65 raise 

66 

67 def get_user_by_email(self: UserMixinRequirements, endpoint: str, email: str) -> UserWrapper: 

68 # dont use user is not None, lint checkers dont recognize throw_ wont return 

69 user = self.user_db.get_user_by_email(email) 

70 if user is None: 

71 LOG().error(f"No user found for email {email}") 

72 code, msg = ErrorUtil.missing_email_msg(email) 

73 self.log_and_throw(endpoint=endpoint, message=msg, error_code=code) 

74 

75 return user 

76 

77 def get_user_by_id(self: UserMixinRequirements, endpoint: str, uid: str) -> UserWrapper: 

78 # dont use user is not None, lint checkers dont recognize throw_ wont return 

79 user = self.user_db.get(uid) 

80 if user is None: 

81 LOG().error(f"No user found for id {uid}") 

82 code, msg = ErrorUtil.missing_uid_msg(uid) 

83 

84 self.log_and_throw(endpoint=endpoint, message=msg, error_code=code) 

85 

86 return user 

87 

88 def update_user( 

89 self: UserMixinRequirements, 

90 endpoint: str, 

91 on_error_msg: ErrorMessage, 

92 user: UserWrapper, 

93 manual_updates: DatabaseDict | None = None, 

94 ) -> UserWrapper: 

95 updates: DatabaseDict | None = None 

96 

97 if manual_updates is not None: 

98 updates = manual_updates 

99 else: 

100 updates = user.get_updates() 

101 if not updates: 

102 LOG().warning(f"No updates for user {user.email}") 

103 self.log_and_throw( 

104 endpoint, 

105 message=on_error_msg, 

106 error_code=AppErrorCode.SERVER, 

107 ) 

108 

109 saved_model = self.user_db.update(user.doc_id, updates) 

110 if saved_model is None: 

111 self.log_and_throw(endpoint, message=on_error_msg, error_code=AppErrorCode.SERVER) 

112 try: 

113 return saved_model 

114 except Exception as e: 

115 msg = f"Error creating PersistedModel for user {user.email} after update: {e!s}" 

116 LOG().error(msg) 

117 self.log_and_throw(endpoint, message=on_error_msg, error_code=AppErrorCode.SERVER)