Coverage for functions \ flipdare \ service \ user_summary_service.py: 47%

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 

14from typing import TYPE_CHECKING 

15from flipdare.app_defaults import get_fallback_avatar 

16from flipdare.app_log import LOG 

17from flipdare.constants import IS_DEBUG, NO_DOC_ID 

18from flipdare.generated.shared.backend.summary_email_entry_type import SummaryEmailEntryType 

19from flipdare.result.app_result import AppResult 

20from flipdare.service._service_provider import ServiceProvider 

21from flipdare.firestore.context.dare_context import GroupDareContext, UserDareContext 

22from flipdare.firestore.context.friend_context import FriendContext 

23from flipdare.generated.model.backend.user_summary_entry_model import UserSummaryEntryModel 

24from flipdare.generated.shared.app_error_code import AppErrorCode 

25from flipdare.generated.shared.model.dare.dare_status import DareStatus 

26from flipdare.wrapper.backend.user_summary_entry_wrapper import UserSummaryEntryWrapper 

27 

28if TYPE_CHECKING: 

29 from flipdare.manager.db_manager import DbManager 

30 from flipdare.manager.backend_manager import BackendManager 

31 

32__all__ = ["UserSummaryService"] 

33 

34 

35class UserSummaryService(ServiceProvider): 

36 

37 def __init__( 

38 self, 

39 db_manager: DbManager | None = None, 

40 backend_manager: BackendManager | None = None, 

41 ) -> None: 

42 super().__init__( 

43 backend_manager=backend_manager, 

44 db_manager=db_manager, 

45 ) 

46 

47 def create_group_dare_entry( 

48 self, 

49 dare_context: GroupDareContext, 

50 ) -> AppResult[UserSummaryEntryWrapper]: 

51 doc_id = dare_context.doc_id 

52 main_result = AppResult[UserSummaryEntryWrapper](doc_id=doc_id) 

53 

54 dare = dare_context.dare 

55 from_user = dare_context.from_user 

56 to_group = dare_context.to_obj 

57 to_user = dare_context.to_user 

58 

59 owner_name = from_user.model.contact_name 

60 avatar = from_user.avatar 

61 group_name = to_group.name 

62 member_name: str | None = None 

63 accepted_name: str | None = None 

64 if to_user is not None: 

65 member_name = to_user.model.contact_name 

66 accepted_name = member_name 

67 avatar = to_user.avatar 

68 

69 to_uid = to_group.doc_id 

70 obj_id = dare.doc_id 

71 

72 entry_type = SummaryEmailEntryType.GROUP_DARE_SENT 

73 if dare.status == DareStatus.ACCEPTED: 

74 entry_type = SummaryEmailEntryType.GROUP_DARE_ACCEPTED 

75 elif dare.status.is_completing: 

76 entry_type = SummaryEmailEntryType.GROUP_DARE_COMPLETED 

77 

78 if avatar is None: 

79 # this should never happen. 

80 # even though image avatar is known, the user is assigned a fallback avatar 

81 # during processing .. 

82 avatar = get_fallback_avatar() 

83 

84 try: 

85 entry = UserSummaryEntryModel.create_group_dare( 

86 owner_name=owner_name, 

87 avatar=avatar, 

88 group_name=group_name, 

89 short_description=dare.short_description, 

90 entry_type=entry_type, 

91 obj_id=obj_id, 

92 member_name=member_name, 

93 accepted_name=accepted_name, 

94 ) 

95 

96 result = self._add_entry(to_uid, entry) 

97 except Exception as e: 

98 LOG().error(f"Failed to create group dare summary entry for {dare.doc_id}: {e}") 

99 main_result.add_error( 

100 AppErrorCode.UNKNOWN, 

101 f"Failed to create group dare summary entry: {e}", 

102 ) 

103 return main_result 

104 

105 main_result.merge(result) 

106 return main_result 

107 

108 def create_dare_entry( 

109 self, 

110 dare_context: UserDareContext, 

111 ) -> AppResult[UserSummaryEntryWrapper]: 

112 doc_id = dare_context.doc_id 

113 main_result = AppResult[UserSummaryEntryWrapper](doc_id=doc_id) 

114 

115 dare = dare_context.dare 

116 from_user = dare_context.from_user 

117 to_user = dare_context.to_obj 

118 

119 from_name = from_user.model.contact_name 

120 from_avatar = from_user.avatar 

121 to_name = to_user.model.contact_name 

122 to_uid = to_user.doc_id 

123 obj_id = dare.doc_id 

124 

125 if from_avatar is None: 

126 # this should never happen. 

127 # even though image avatar is known, the user is assigned a fallback avatar 

128 # during processing .. 

129 from_avatar = get_fallback_avatar() 

130 

131 entry_type = SummaryEmailEntryType.DARE_REQUEST 

132 if dare.status == DareStatus.ACCEPTED: 

133 entry_type = SummaryEmailEntryType.DARE_ACCEPTED 

134 elif dare.status.is_completing: 

135 entry_type = SummaryEmailEntryType.DARE_COMPLETED 

136 

137 try: 

138 entry = UserSummaryEntryModel.create_dare( 

139 from_name=from_name, 

140 avatar=from_avatar, 

141 to_name=to_name, 

142 short_description=dare.short_description, 

143 entry_type=entry_type, 

144 obj_id=obj_id, 

145 ) 

146 

147 result = self._add_entry(to_uid, entry) 

148 if not result.is_error: 

149 assert result.generated is not None # narrowing 

150 main_result.generated = result.generated 

151 except Exception as e: 

152 msg = f"Failed to create dare summary entry for {dare.doc_id}: {e}" 

153 LOG().error(msg) 

154 main_result.add_error(AppErrorCode.UNKNOWN, msg) 

155 return main_result 

156 

157 if result.is_error: 

158 msg = f"Failed to create dare summary entry for {dare.doc_id}: {result.errors}" 

159 LOG().error(msg) 

160 main_result.merge(result) 

161 

162 if IS_DEBUG: 

163 msg = f"Result={main_result.is_ok} dare summary entry for dare {dare.doc_id} to user {to_uid}" 

164 LOG().debug(msg) 

165 

166 return main_result 

167 

168 def create_friend_request( 

169 self, 

170 friend_context: FriendContext, 

171 ) -> AppResult[UserSummaryEntryWrapper]: 

172 doc_id = friend_context.doc_id 

173 main_result = AppResult[UserSummaryEntryWrapper](doc_id=doc_id) 

174 

175 friend = friend_context.friend 

176 from_user = friend_context.from_user 

177 to_user = friend_context.to_user 

178 

179 assert friend is not None # narrowing 

180 assert from_user is not None # narrowing 

181 assert to_user is not None # narrowing 

182 

183 from_uid = from_user.doc_id 

184 to_uid = to_user.doc_id 

185 

186 entry = UserSummaryEntryModel.create_friend_request( 

187 from_name=from_user.model.contact_name, 

188 from_uid=from_uid, 

189 to_name=to_user.model.contact_name, 

190 is_accepted=friend.is_accepted, 

191 from_avatar=from_user.avatar, 

192 ) 

193 

194 result = self._add_entry(to_uid, entry) 

195 main_result.merge(result) 

196 return main_result 

197 

198 def _add_entry( 

199 self, 

200 to_uid: str, 

201 entry: UserSummaryEntryModel, 

202 ) -> AppResult[UserSummaryEntryWrapper]: 

203 doc_id = entry.id or NO_DOC_ID 

204 main_result = AppResult[UserSummaryEntryWrapper](doc_id=doc_id) 

205 try: 

206 summary_db = self.summary_db 

207 result = summary_db.create_report_entry(user_id=to_uid, entry=entry) 

208 if result is not None: 

209 if IS_DEBUG: 

210 LOG().debug(f"Created summary entry for user {to_uid}: {result.entry}") 

211 

212 main_result.generated = result.entry 

213 else: 

214 msg = f"Failed to save summary entry to database for {to_uid}." 

215 LOG().error(msg) 

216 main_result.add_error( 

217 AppErrorCode.CREATE_FAILED, 

218 msg, 

219 ) 

220 except Exception as e: 

221 msg = f"Failed to create summary entry for {to_uid}: {e}" 

222 LOG().error(msg) 

223 main_result.add_error(AppErrorCode.DATABASE_EX, msg) 

224 

225 return main_result