Coverage for src/lexigram/notification/mailer/smtp_mailer.py: 30%

1"""SMTP mailer backend.""" 

2 

3from __future__ import annotations 

4 

5import asyncio 

6from email.errors import MessageError 

7from email.mime.multipart import MIMEMultipart 

8from email.mime.text import MIMEText 

9import smtplib 

10import ssl 

11from typing import TYPE_CHECKING 

12import uuid 

13 

14from lexigram.contracts.core import HealthCheckResult, HealthStatus 

15from lexigram.contracts.mailer import ( 

16 EmailMessage, 

17 MessageDeliveryReceipt, 

18) 

19from lexigram.logging import get_logger 

20from lexigram.notification.constants import DEFAULT_FROM_EMAIL 

21from lexigram.notification.exceptions import SMTPMailerError 

22from lexigram.result import Err, Ok, Result 

23 

24if TYPE_CHECKING: 

25 from lexigram.contracts.mailer.errors import MailerError 

26 

27logger = get_logger(__name__) 

28 

29 

30class SMTPMailer: 

31 """Email backend that sends via SMTP. 

32 

33 Implements :class:`~lexigram.contracts.mailer.protocols.MailerProtocol`. 

34 SMTP operations are blocking; they are run in a thread pool via 

35 ``run_in_executor`` to avoid blocking the event loop. 

36 

37 Args: 

38 host: SMTP server hostname. 

39 port: SMTP port (587 for TLS, 465 for SSL, 25 for plain). 

40 username: SMTP authentication username. 

41 password: SMTP authentication password. 

42 use_tls: Use STARTTLS after connection (port 587). 

43 use_ssl: Use SSL from the start (port 465). 

44 timeout: Connection timeout in seconds. 

45 from_email: Default sender address. 

46 from_name: Default sender display name. 

47 """ 

48 

49 def __init__( 

50 self, 

51 host: str = "localhost", 

52 port: int = 587, 

53 username: str | None = None, 

54 password: str | None = None, 

55 use_tls: bool = True, 

56 use_ssl: bool = False, 

57 timeout: int = 30, 

58 from_email: str | None = None, 

59 from_name: str | None = None, 

60 ) -> None: 

61 self.host = host 

62 self.port = port 

63 self.username = username 

64 self.password = password 

65 self.use_tls = use_tls 

66 self.use_ssl = use_ssl 

67 self.timeout = timeout 

68 self.from_email = from_email or DEFAULT_FROM_EMAIL 

69 self.from_name = from_name 

70 

71 def _build_mime(self, message: EmailMessage) -> MIMEMultipart: 

72 """Build MIME message from EmailMessage.""" 

73 mime = MIMEMultipart("alternative") 

74 sender = ( 

75 message.from_email if message.from_email is not None else self.from_email 

76 ) 

77 display_name = ( 

78 message.from_name if message.from_name is not None else self.from_name 

79 ) 

80 mime["From"] = f"{display_name} <{sender}>" if display_name else sender 

81 mime["To"] = ", ".join(message.to) 

82 mime["Subject"] = message.subject 

83 if message.cc: 

84 mime["Cc"] = ", ".join(message.cc) 

85 for key, value in message.headers.items(): 

86 mime[key] = value 

87 if message.body: 

88 mime.attach(MIMEText(message.body, "plain", "utf-8")) 

89 if message.html_body: 

90 mime.attach(MIMEText(message.html_body, "html", "utf-8")) 

91 return mime 

92 

93 def _send_sync(self, mime: MIMEMultipart, recipients: list[str]) -> None: 

94 """Synchronous SMTP send (runs in executor to not block event loop).""" 

95 ctx = ssl.create_default_context() if (self.use_tls or self.use_ssl) else None 

96 if self.use_ssl and ctx: 

97 server: smtplib.SMTP = smtplib.SMTP_SSL( 

98 self.host, self.port, context=ctx, timeout=self.timeout 

99 ) 

100 else: 

101 server = smtplib.SMTP(self.host, self.port, timeout=self.timeout) 

102 with server: 

103 if self.use_tls and not self.use_ssl and ctx: 

104 server.starttls(context=ctx) 

105 if self.username and self.password: 

106 server.login(self.username, self.password) 

107 server.sendmail(mime["From"], recipients, mime.as_string()) 

108 

109 async def send( 

110 self, message: EmailMessage 

111 ) -> Result[MessageDeliveryReceipt, MailerError]: 

112 """Send an email via SMTP. 

113 

114 Args: 

115 message: The email message to send. 

116 

117 Returns: 

118 ``Ok(MessageDeliveryReceipt)`` on success. 

119 ``Err(SMTPMailerError)`` for SMTP delivery failures. 

120 """ 

121 try: 

122 mime = self._build_mime(message) 

123 recipients = message.to + message.cc + message.bcc 

124 loop = asyncio.get_running_loop() 

125 await loop.run_in_executor(None, self._send_sync, mime, recipients) 

126 except (smtplib.SMTPException, MessageError) as exc: 

127 # MessageError covers HeaderParseError/HeaderWriteError raised 

128 # by as_string() on CRLF-infiltrated headers (defense-in-depth: 

129 # EmailMessage validation rejects these at construction). 

130 logger.warning("smtp_send_failed", error=str(exc), host=self.host) 

131 return Err(SMTPMailerError(str(exc))) 

132 

133 receipt = MessageDeliveryReceipt( 

134 message_id=str(uuid.uuid4()), 

135 backend="smtp", 

136 channel="email", 

137 ) 

138 logger.info("smtp_sent", to=message.to, subject=message.subject) 

139 return Ok(receipt) 

140 

141 async def health_check(self, timeout: float = 5.0) -> HealthCheckResult: 

142 """Check SMTP connectivity by attempting a connection. 

143 

144 Args: 

145 timeout: Max seconds to wait for the connection. 

146 

147 Returns: 

148 :class:`~lexigram.contracts.core.HealthCheckResult`. 

149 """ 

150 

151 def _check() -> None: 

152 with smtplib.SMTP(self.host, self.port, timeout=timeout): 

153 pass 

154 

155 try: 

156 loop = asyncio.get_running_loop() 

157 await loop.run_in_executor(None, _check) 

158 return HealthCheckResult( 

159 component="smtp", 

160 status=HealthStatus.HEALTHY, 

161 details={"host": self.host, "port": self.port}, 

162 ) 

163 except OSError as exc: 

164 return HealthCheckResult( 

165 component="smtp", 

166 status=HealthStatus.UNHEALTHY, 

167 details={"host": self.host, "error": str(exc)}, 

168 ) 

169 

170 

171__all__ = ["SMTPMailer"]