Coverage for functions \ flipdare \ mailer \ admin_mailer.py: 100%

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 

15import smtplib 

16from typing import Any, override 

17 

18import flask 

19 

20from flipdare.mailer.email_image import EmailImage 

21from flipdare.app_globals import is_valid_email 

22from flipdare.app_log import LOG 

23from flipdare.constants import IS_DEBUG, IS_TRACE 

24from flipdare.error.stack_util import StackUtil 

25from flipdare.generated.schema.email.body.admin.log_email_schema import LogEmailSchema 

26from flipdare.mailer.admin.app_contact_email import AppContactEmail 

27from flipdare.mailer.admin.app_log_email import AppLogEmail 

28from flipdare.mailer._jinja_email_template import JinjaEmailTemplate 

29from flipdare.mailer._mailer import Mailer, MailerOptions 

30from flipdare.error.error_context import ErrorContext 

31from flipdare.generated import AppErrorCode 

32from flipdare.generated.shared.app_log_category import AppLogCategory 

33from flipdare.generated.shared.backend.app_job_type import AppJobType 

34from flipdare.generated.shared.backend.system_log_type import SystemLogType 

35from flipdare.generated.shared.firestore_collections import FirestoreCollections 

36from flipdare.generated.shared.search.search_collections import SearchCollections 

37from flipdare.message.user_message import UserMessage 

38from flipdare.request.data.search_request_adapter import AppRequest 

39from flipdare.core.app_response import AppErrorResponse, AppOkResponse 

40from flipdare.util.debug_util import stringify_debug 

41from flipdare.util.time_util import TimeUtil 

42 

43__all__ = ["AdminMailer"] 

44 

45 

46class AdminMailer(Mailer): 

47 def __init__( 

48 self, 

49 options: MailerOptions | None = None, 

50 smtp_client: smtplib.SMTP_SSL | None = None, 

51 ) -> None: 

52 super().__init__(options=options, smtp_client=smtp_client) 

53 

54 @override 

55 def send( # type: ignore[override] 

56 self, 

57 email_template: JinjaEmailTemplate[Any], 

58 images: list[EmailImage] | None = None, 

59 ) -> bool: # pragma: no cover 

60 """Send admin email using SMTP.""" 

61 email = self.options.internal_service_email 

62 if IS_TRACE: 

63 msg = f"Preparing to send admin email to {email} with template {email_template.__class__.__name__}" 

64 LOG().trace(msg) 

65 

66 # dont minify admin messages, easier to troubleshoot! 

67 return self._send_template(email, email_template, should_minify=False, images=images) 

68 

69 def send_contact(self, request: AppRequest[Any]) -> flask.Response: 

70 # this is remote so we need to check the honeypot_field 

71 # If honeypot_field is NOT empty: Reject the submission immediately 

72 # (do not process the unsubscribe request). 

73 # If honeypot_field IS empty: Proceed with normal processing. 

74 from flipdare.services import get_app_logger 

75 

76 req = request.raw_request 

77 url = req.url 

78 

79 if req.method.lower() != "post": 

80 msg = f"Invalid method ({req.method}) for unsubscribe" 

81 return AppErrorResponse.from_context( 

82 ErrorContext.forbidden(url, message=msg), 

83 ).raw_response() 

84 

85 if IS_TRACE: 

86 LOG().trace( 

87 f"Received contact email with details:\n" 

88 f"URL: {req.url}\n" 

89 f"Method: {req.method}\n" 

90 f"Headers: {req.headers}\n" 

91 f"Body: {req.get_data(as_text=True)}", 

92 ) 

93 

94 email = None 

95 try: 

96 data = req.get_json() 

97 honeypot_field = data.get("pot_field", "") 

98 if honeypot_field: 

99 # if spam, return ok but with an empty message and do nothing.. 

100 if IS_DEBUG: 

101 LOG().debug(f"Spam detected in contact form for email {email}") 

102 

103 return AppOkResponse.ok().raw_response() 

104 

105 email = data.get("email", "") 

106 subject = data.get("subject", "") 

107 message = data.get("message", "") 

108 except Exception as e: 

109 msg = f"Contact error: {e}" 

110 return AppErrorResponse.message( 

111 url=url, 

112 error_code=AppErrorCode.INVALID_DATA, 

113 message=msg, 

114 ).raw_response() 

115 

116 email_result = is_valid_email(email) 

117 if email_result.is_error: 

118 msg = f"Invalid email address: {email}" 

119 return AppErrorResponse.message( 

120 url=url, 

121 error_code=AppErrorCode.INVALID_EMAIL, 

122 message=msg, 

123 ).raw_response() 

124 

125 email_content = AppContactEmail(from_email=email, message=message, from_subject=subject) 

126 

127 if IS_TRACE: 

128 LOG().trace(f"Sending contact email for {email}") 

129 try: 

130 # first we send the email, and update if the email is sent ok 

131 self.send(email_template=email_content) 

132 if IS_TRACE: 

133 LOG().trace(f"Contact email sent to admin for {email}") 

134 

135 return AppOkResponse.message(UserMessage.CONTACT_OK).raw_response() 

136 except Exception as error: 

137 cause = f"Error sending contact email to admin for {email}: {error}" 

138 get_app_logger().contact_email_error( 

139 error_code=AppErrorCode.CONTACT, 

140 email=email, 

141 error=error, 

142 message=cause, 

143 ) 

144 

145 return AppErrorResponse.message( 

146 url=url, 

147 error_code=AppErrorCode.SERVER, 

148 message=cause, 

149 ).raw_response() 

150 

151 def send_error( 

152 self, 

153 error_code: str, 

154 message: str, 

155 job_type: AppJobType | None = None, 

156 collection: FirestoreCollections | SearchCollections | None = None, 

157 category: AppLogCategory | None = None, 

158 include_stack: bool = True, 

159 ) -> None: 

160 source = job_type or collection or category or "UnknownSource" 

161 schema = LogEmailSchema( 

162 occurred_at=TimeUtil.formatted_now(), 

163 called_from=StackUtil.get_caller_str(), 

164 source=source, 

165 log_type=SystemLogType.ERROR, 

166 error_code=error_code, 

167 message=message, 

168 ) 

169 

170 if job_type is not None: 

171 schema["job_type"] = job_type 

172 if include_stack: 

173 schema["stack_trace"] = StackUtil.get_flipdare_stack() 

174 

175 self.send_log(schema) 

176 

177 def send_log( 

178 self, 

179 schema: LogEmailSchema, 

180 ) -> None: 

181 try: 

182 template = AppLogEmail(schema) 

183 self.send(template) 

184 except Exception as ex: 

185 # NOTE: the primary reason for this failure is SMTP issues, so just log 

186 msg = f"Exception thrown sending email: {ex}\n" 

187 msg += f"Schema:\n{stringify_debug(schema)}" 

188 LOG().error(msg)