From c72be747276e150e95db0a5e84f84de46b94ee6c Mon Sep 17 00:00:00 2001 From: Xiaoxia AI Date: Wed, 17 Jun 2026 01:17:19 +0800 Subject: [PATCH] feat(auth): add email service with SMTP support - Implement EmailService class with configurable SMTP - Support verification/password-reset/invitation email templates - Support HTML and plain text fallback - Support CC/BCC recipients - Add 8 comprehensive unit tests with Mock SMTP (all passed) Phase 4 Task 4/68 completed --- packages/domain/auth/__init__.py | 4 + packages/domain/auth/email_service.py | 329 ++++++++++++++++++++++++++ tests/unit/test_email_service.py | 197 +++++++++++++++ 3 files changed, 530 insertions(+) create mode 100644 packages/domain/auth/email_service.py create mode 100644 tests/unit/test_email_service.py diff --git a/packages/domain/auth/__init__.py b/packages/domain/auth/__init__.py index 358139daf..0e594885e 100644 --- a/packages/domain/auth/__init__.py +++ b/packages/domain/auth/__init__.py @@ -7,6 +7,7 @@ from packages.domain.auth.password_hasher import ( password_validator, ) from packages.domain.auth.session_store import SessionStore, RedisConfig, session_store +from packages.domain.auth.email_service import EmailService, EmailConfig, email_service __all__ = [ "JWTService", @@ -20,4 +21,7 @@ __all__ = [ "SessionStore", "RedisConfig", "session_store", + "EmailService", + "EmailConfig", + "email_service", ] diff --git a/packages/domain/auth/email_service.py b/packages/domain/auth/email_service.py new file mode 100644 index 000000000..ae1f5824a --- /dev/null +++ b/packages/domain/auth/email_service.py @@ -0,0 +1,329 @@ +""" +邮件服务 +支持 SMTP 发送邮件（验证/重置密码/邀请等） +""" +import smtplib +from email.mime.text import MIMEText +from email.mime.multipart import MIMEMultipart +from typing import Optional, List +from dataclasses import dataclass + + +@dataclass +class EmailConfig: + """邮件配置""" + smtp_host: str = "smtp.gmail.com" + smtp_port: int = 587 + smtp_user: str = "" + smtp_password: str = "" + from_email: str = "" + from_name: str = "小虾 SaaS" + use_tls: bool = True + + +class EmailService: + """邮件服务类""" + + def __init__(self, config: Optional[EmailConfig] = None): + """ + 初始化邮件服务 + + Args: + config: 邮件配置 + """ + self.config = config or EmailConfig() + + def send_email( + self, + to_email: str, + subject: str, + html_body: str, + text_body: Optional[str] = None, + cc: Optional[List[str]] = None, + bcc: Optional[List[str]] = None, + ) -> tuple[bool, Optional[str]]: + """ + 发送邮件 + + Args: + to_email: 收件人邮箱 + subject: 邮件主题 + html_body: HTML 正文 + text_body: 纯文本正文（可选，作为 HTML 的备用） + cc: 抄送列表 + bcc: 密送列表 + + Returns: + (是否成功, 错误信息) + """ + try: + # 创建邮件 + msg = MIMEMultipart("alternative") + msg["Subject"] = subject + msg["From"] = f"{self.config.from_name} <{self.config.from_email}>" + msg["To"] = to_email + + if cc: + msg["Cc"] = ", ".join(cc) + + # 添加纯文本正文 + if text_body: + part1 = MIMEText(text_body, "plain", "utf-8") + msg.attach(part1) + + # 添加 HTML 正文 + part2 = MIMEText(html_body, "html", "utf-8") + msg.attach(part2) + + # 连接 SMTP 服务器 + with smtplib.SMTP(self.config.smtp_host, self.config.smtp_port) as server: + if self.config.use_tls: + server.starttls() + + # 登录 + if self.config.smtp_user and self.config.smtp_password: + server.login(self.config.smtp_user, self.config.smtp_password) + + # 发送 + recipients = [to_email] + if cc: + recipients.extend(cc) + if bcc: + recipients.extend(bcc) + + server.sendmail( + self.config.from_email, + recipients, + msg.as_string() + ) + + return True, None + + except Exception as e: + return False, str(e) + + def send_verification_email( + self, + to_email: str, + username: str, + verification_url: str, + ) -> tuple[bool, Optional[str]]: + """ + 发送邮箱验证邮件 + + Args: + to_email: 收件人邮箱 + username: 用户名 + verification_url: 验证链接 + + Returns: + (是否成功, 错误信息) + """ + subject = "验证您的邮箱 - 小虾 SaaS" + + html_body = f""" + + + + + + +
+

欢迎加入小虾 SaaS！

+

你好 {username}，

+

感谢您注册小虾 SaaS！请点击下面的按钮验证您的邮箱地址：

+
+ + 验证邮箱 + +
+

+ 如果按钮无法点击，请复制以下链接到浏览器：
+ {verification_url} +

+

+ 此链接将在 24 小时后过期。 +

+
+

+ 如果您没有注册小虾 SaaS，请忽略此邮件。 +

+
+ + + """ + + text_body = f""" + 欢迎加入小虾 SaaS！ + + 你好 {username}， + + 感谢您注册小虾 SaaS！请访问以下链接验证您的邮箱地址： + + {verification_url} + + 此链接将在 24 小时后过期。 + + 如果您没有注册小虾 SaaS，请忽略此邮件。 + """ + + return self.send_email(to_email, subject, html_body, text_body) + + def send_password_reset_email( + self, + to_email: str, + username: str, + reset_url: str, + ) -> tuple[bool, Optional[str]]: + """ + 发送密码重置邮件 + + Args: + to_email: 收件人邮箱 + username: 用户名 + reset_url: 重置链接 + + Returns: + (是否成功, 错误信息) + """ + subject = "重置您的密码 - 小虾 SaaS" + + html_body = f""" + + + + + + +
+

重置密码请求

+

你好 {username}，

+

我们收到了重置您账号密码的请求。请点击下面的按钮重置密码：

+
+ + 重置密码 + +
+

+ 如果按钮无法点击，请复制以下链接到浏览器：
+ {reset_url} +

+

+ 此链接将在 1 小时后过期。 +

+
+

+ 如果您没有请求重置密码，请忽略此邮件，您的密码不会被更改。 +

+
+ + + """ + + text_body = f""" + 重置密码请求 + + 你好 {username}， + + 我们收到了重置您账号密码的请求。请访问以下链接重置密码： + + {reset_url} + + 此链接将在 1 小时后过期。 + + 如果您没有请求重置密码，请忽略此邮件，您的密码不会被更改。 + """ + + return self.send_email(to_email, subject, html_body, text_body) + + def send_workspace_invitation_email( + self, + to_email: str, + inviter_name: str, + workspace_name: str, + role: str, + invitation_url: str, + ) -> tuple[bool, Optional[str]]: + """ + 发送 Workspace 邀请邮件 + + Args: + to_email: 收件人邮箱 + inviter_name: 邀请人姓名 + workspace_name: 工作空间名称 + role: 角色（Admin/Member/Viewer） + invitation_url: 邀请链接 + + Returns: + (是否成功, 错误信息) + """ + subject = f"{inviter_name} 邀请您加入 {workspace_name} - 小虾 SaaS" + + role_names = { + "owner": "所有者", + "admin": "管理员", + "member": "成员", + "viewer": "查看者", + } + role_display = role_names.get(role.lower(), role) + + html_body = f""" + + + + + + +
+

工作空间邀请

+

{inviter_name} 邀请您以 {role_display} 身份加入工作空间：

+
+

{workspace_name}

+

角色：{role_display}

+
+
+ + 接受邀请 + +
+

+ 如果按钮无法点击，请复制以下链接到浏览器：
+ {invitation_url} +

+

+ 此邀请将在 7 天后过期。 +

+
+

+ 如果您不认识邀请人或不想加入此工作空间，请忽略此邮件。 +

+
+ + + """ + + text_body = f""" + 工作空间邀请 + + {inviter_name} 邀请您以 {role_display} 身份加入工作空间：{workspace_name} + + 请访问以下链接接受邀请： + + {invitation_url} + + 此邀请将在 7 天后过期。 + + 如果您不认识邀请人或不想加入此工作空间，请忽略此邮件。 + """ + + return self.send_email(to_email, subject, html_body, text_body) + + +# 全局实例（生产环境应该从配置读取） +email_service = EmailService() diff --git a/tests/unit/test_email_service.py b/tests/unit/test_email_service.py new file mode 100644 index 000000000..116687fef --- /dev/null +++ b/tests/unit/test_email_service.py @@ -0,0 +1,197 @@ +""" +邮件服务测试 +""" +import pytest +from unittest.mock import Mock, patch, MagicMock +from packages.domain.auth.email_service import EmailService, EmailConfig + + +class TestEmailService: + """邮件服务测试""" + + @pytest.fixture + def email_config(self): + """创建测试邮件配置""" + return EmailConfig( + smtp_host="smtp.test.com", + smtp_port=587, + smtp_user="test@test.com", + smtp_password="test-password", + from_email="noreply@test.com", + from_name="Test Service", + use_tls=True, + ) + + @pytest.fixture + def email_service(self, email_config): + """创建邮件服务实例""" + return EmailService(config=email_config) + + @patch('smtplib.SMTP') + def test_send_email_success(self, mock_smtp, email_service): + """测试发送邮件成功""" + # Mock SMTP 服务器 + mock_server = MagicMock() + mock_smtp.return_value.__enter__.return_value = mock_server + + success, error = email_service.send_email( + to_email="user@example.com", + subject="Test Email", + html_body="

Test

", + text_body="Test", + ) + + assert success is True + assert error is None + + # 验证 SMTP 调用 + mock_smtp.assert_called_once_with("smtp.test.com", 587) + mock_server.starttls.assert_called_once() + mock_server.login.assert_called_once_with("test@test.com", "test-password") + mock_server.sendmail.assert_called_once() + + @patch('smtplib.SMTP') + def test_send_email_with_cc_bcc(self, mock_smtp, email_service): + """测试发送邮件带抄送和密送""" + mock_server = MagicMock() + mock_smtp.return_value.__enter__.return_value = mock_server + + success, error = email_service.send_email( + to_email="user@example.com", + subject="Test Email", + html_body="

Test

", + cc=["cc1@example.com", "cc2@example.com"], + bcc=["bcc@example.com"], + ) + + assert success is True + + # 验证收件人列表包含所有人 + call_args = mock_server.sendmail.call_args + recipients = call_args[0][1] + assert "user@example.com" in recipients + assert "cc1@example.com" in recipients + assert "cc2@example.com" in recipients + assert "bcc@example.com" in recipients + + @patch('smtplib.SMTP') + def test_send_email_smtp_error(self, mock_smtp, email_service): + """测试 SMTP 错误处理""" + # Mock SMTP 抛出异常 + mock_smtp.side_effect = Exception("SMTP connection failed") + + success, error = email_service.send_email( + to_email="user@example.com", + subject="Test", + html_body="

Test

", + ) + + assert success is False + assert error is not None + assert "SMTP connection failed" in error + + @patch('smtplib.SMTP') + def test_send_verification_email(self, mock_smtp, email_service): + """测试发送验证邮件""" + mock_server = MagicMock() + mock_smtp.return_value.__enter__.return_value = mock_server + + success, error = email_service.send_verification_email( + to_email="user@example.com", + username="TestUser", + verification_url="https://example.com/verify?token=abc123", + ) + + assert success is True + assert error is None + + # 验证发送了邮件 + mock_server.sendmail.assert_called_once() + call_args = mock_server.sendmail.call_args + + # 验证收件人 + assert call_args[0][1] == ["user@example.com"] + + @patch('smtplib.SMTP') + def test_send_password_reset_email(self, mock_smtp, email_service): + """测试发送密码重置邮件""" + mock_server = MagicMock() + mock_smtp.return_value.__enter__.return_value = mock_server + + success, error = email_service.send_password_reset_email( + to_email="user@example.com", + username="TestUser", + reset_url="https://example.com/reset?token=xyz789", + ) + + assert success is True + assert error is None + + # 验证发送了邮件 + mock_server.sendmail.assert_called_once() + call_args = mock_server.sendmail.call_args + + # 验证收件人 + assert call_args[0][1] == ["user@example.com"] + + @patch('smtplib.SMTP') + def test_send_workspace_invitation_email(self, mock_smtp, email_service): + """测试发送工作空间邀请邮件""" + mock_server = MagicMock() + mock_smtp.return_value.__enter__.return_value = mock_server + + success, error = email_service.send_workspace_invitation_email( + to_email="user@example.com", + inviter_name="Alice", + workspace_name="My Workspace", + role="admin", + invitation_url="https://example.com/invite?token=inv123", + ) + + assert success is True + assert error is None + + # 验证发送了邮件 + mock_server.sendmail.assert_called_once() + call_args = mock_server.sendmail.call_args + + # 验证收件人 + assert call_args[0][1] == ["user@example.com"] + + @patch('smtplib.SMTP') + def test_email_without_tls(self, mock_smtp): + """测试不使用 TLS 发送邮件""" + config = EmailConfig( + smtp_host="smtp.test.com", + smtp_port=25, + smtp_user="", + smtp_password="", + from_email="test@test.com", + use_tls=False, + ) + service = EmailService(config=config) + + mock_server = MagicMock() + mock_smtp.return_value.__enter__.return_value = mock_server + + success, error = service.send_email( + to_email="user@example.com", + subject="Test", + html_body="Test", + ) + + assert success is True + + # 验证不调用 starttls + mock_server.starttls.assert_not_called() + # 验证不调用 login（没有用户名密码） + mock_server.login.assert_not_called() + + def test_default_config(self): + """测试默认配置""" + service = EmailService() + + assert service.config.smtp_host == "smtp.gmail.com" + assert service.config.smtp_port == 587 + assert service.config.use_tls is True + assert service.config.from_name == "小虾 SaaS"