Coverage for functions \ flipdare \ service \ _email_mixin.py: 39%

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 

13from __future__ import annotations 

14 

15from typing import Any, Concatenate, ParamSpec, Protocol, TypeVar, runtime_checkable 

16from collections.abc import Callable 

17from flipdare.backend.app_logger import AppLogger 

18from flipdare.app_log import LOG 

19from flipdare.result.outcome import Outcome 

20from flipdare.mailer.admin_mailer import AdminMailer 

21from flipdare.mailer.user_mailer import UserMailer 

22from flipdare.firestore.context.group_context import GroupContextFactory 

23from flipdare.generated.shared.app_error_code import AppErrorCode 

24from flipdare.mailer._jinja_email_template import JinjaEmailTemplate 

25from flipdare.generated.shared.firestore_collections import FirestoreCollections 

26from flipdare.wrapper.group_wrapper import GroupWrapper 

27from flipdare.wrapper.user_wrapper import UserWrapper 

28 

29P = ParamSpec("P") 

30T = TypeVar("T", bound=JinjaEmailTemplate[Any]) 

31 

32 

33@runtime_checkable 

34class EmailMixinRequirements(Protocol): 

35 """ 

36 We need this so can call email mixin methods 

37 from within the email mixin .. 

38 """ 

39 

40 @property 

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

42 @property 

43 def user_mailer(self) -> UserMailer: ... 

44 @property 

45 def admin_mailer(self) -> AdminMailer: ... 

46 

47 def send_user_email( 

48 self: EmailMixinRequirements, 

49 user: UserWrapper, 

50 template: JinjaEmailTemplate[Any], 

51 doc_id: str, 

52 collection: FirestoreCollections | None = None, 

53 ) -> Outcome: ... 

54 def send_group_email( 

55 self: EmailMixinRequirements, 

56 group: GroupWrapper, 

57 email_callback: Callable[Concatenate[UserWrapper, P], T], 

58 doc_id: str, 

59 collection: FirestoreCollections | None = None, 

60 *args: P.args, 

61 **kwargs: P.kwargs, 

62 ) -> Outcome: ... 

63 

64 

65class EmailMixin: 

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

67 __slots__ = () 

68 

69 def send_user_email( 

70 self: EmailMixinRequirements, 

71 user: UserWrapper, 

72 template: JinjaEmailTemplate[Any], 

73 doc_id: str, 

74 collection: FirestoreCollections | None = None, 

75 ) -> Outcome: 

76 try: 

77 self.user_mailer.send(user=user, email_template=template, notif_check=False) 

78 return Outcome.OK 

79 

80 except Exception as error: 

81 msg = f"Failed to send email to {user.email}: {error}" 

82 LOG().error(msg) 

83 self.app_logger.system_error( 

84 message=msg, 

85 error_code=AppErrorCode.INVALID_EMAIL, 

86 collection=collection, 

87 doc_id=doc_id, 

88 ) 

89 return Outcome.ERROR 

90 

91 def send_group_email( 

92 self: EmailMixinRequirements, 

93 group: GroupWrapper, 

94 email_callback: Callable[Concatenate[UserWrapper, P], T], 

95 doc_id: str, 

96 collection: FirestoreCollections | None = None, 

97 *args: P.args, 

98 **kwargs: P.kwargs, 

99 ) -> Outcome: 

100 try: 

101 ctx = GroupContextFactory().create(group) 

102 if ctx is None: 

103 self.app_logger.system_error( 

104 message=f"Could not create group context for group {group.doc_id}", 

105 error_code=AppErrorCode.INVALID_EMAIL, 

106 collection=collection, 

107 doc_id=doc_id, 

108 ) 

109 return Outcome.ERROR 

110 

111 if not ctx.validate(): 

112 self.app_logger.system_error( 

113 message=f"Invalid group context for group {group.doc_id}", 

114 error_code=AppErrorCode.INVALID_EMAIL, 

115 collection=collection, 

116 doc_id=doc_id, 

117 ) 

118 return Outcome.ERROR 

119 

120 result = Outcome.OK 

121 users = ctx.users() 

122 

123 if users is None: 

124 self.app_logger.system_error( 

125 message=f"Group {group.doc_id} has no members", 

126 error_code=AppErrorCode.INVALID_EMAIL, 

127 collection=collection, 

128 doc_id=doc_id, 

129 ) 

130 return Outcome.ERROR 

131 

132 error_ct = 0 

133 for user in users: 

134 template = email_callback(user, *args, **kwargs) 

135 send_result = self.send_user_email( 

136 user=user, 

137 template=template, 

138 doc_id=doc_id, 

139 collection=collection, 

140 ) 

141 if send_result.is_error: 

142 error_ct += 1 

143 

144 if error_ct > 0: 

145 result = Outcome.ERROR if error_ct == len(users) else Outcome.WARNING 

146 

147 return result 

148 

149 except Exception as error: 

150 msg = f"Failed to send email to group {group.doc_id}: {error}" 

151 self.app_logger.system_error( 

152 message=msg, 

153 error_code=AppErrorCode.INVALID_EMAIL, 

154 collection=collection, 

155 doc_id=doc_id, 

156 ) 

157 return Outcome.ERROR