"""Email Service (SMTP) 单元测试""" from __future__ import annotations from unittest.mock import MagicMock, patch import pytest from packages.adapters.smtp.email_service import ( EmailService, NoopEmailService, get_email_service, ) from packages.domain.auth.email_service import EmailConfig @pytest.fixture def email_config(): return EmailConfig( smtp_host="smtp.example.com", smtp_port=587, from_email="noreply@example.com", from_name="小虾 SaaS", smtp_user="user", smtp_password="pass", use_tls=True, ) @pytest.fixture def email_service(email_config): return EmailService(email_config) class TestNoopEmailService: """NoopEmailService 测试""" def test_send_verification_email_returns_false(self): """验证邮件返回失败""" svc = NoopEmailService() success, msg = svc.send_verification_email( to_email="test@example.com", username="testuser", verification_url="https://example.com/verify", ) assert success is False assert "disabled" in msg.lower() def test_send_password_reset_email_returns_false(self): """密码重置邮件返回失败""" svc = NoopEmailService() success, msg = svc.send_password_reset_email( to_email="test@example.com", username="testuser", reset_url="https://example.com/reset", ) assert success is False assert "disabled" in msg.lower() class TestEmailServiceInit: """初始化测试""" def test_init_with_config(self, email_config): """使用指定配置初始化""" svc = EmailService(email_config) assert svc.config is email_config def test_init_without_config(self): """不指定配置使用默认 EmailConfig""" svc = EmailService() assert svc.config is not None assert isinstance(svc.config, EmailConfig) class TestSendEmail: """send_email 方法测试""" def test_send_success(self, email_service): """发送成功""" with patch("packages.adapters.smtp.email_service.smtplib.SMTP") as 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="测试主题", html_body="
测试内容
", ) assert success is True assert error is None mock_server.starttls.assert_called_once() mock_server.login.assert_called_once_with("user", "pass") mock_server.sendmail.assert_called_once() def test_send_without_tls(self, email_config): """不使用 TLS""" email_config.use_tls = False svc = EmailService(email_config) with patch("packages.adapters.smtp.email_service.smtplib.SMTP") as mock_smtp: mock_server = MagicMock() mock_smtp.return_value.__enter__.return_value = mock_server svc.send_email(to_email="u@e.com", subject="s", html_body="body") mock_server.starttls.assert_not_called() def test_send_without_auth(self, email_config): """不配置用户名密码时不登录""" email_config.smtp_user = "" email_config.smtp_password = "" svc = EmailService(email_config) with patch("packages.adapters.smtp.email_service.smtplib.SMTP") as mock_smtp: mock_server = MagicMock() mock_smtp.return_value.__enter__.return_value = mock_server svc.send_email(to_email="u@e.com", subject="s", html_body="body") mock_server.login.assert_not_called() def test_send_with_cc_and_bcc(self, email_service): """发送带抄送和密送""" with patch("packages.adapters.smtp.email_service.smtplib.SMTP") as mock_smtp: mock_server = MagicMock() mock_smtp.return_value.__enter__.return_value = mock_server email_service.send_email( to_email="to@example.com", subject="s", html_body="body", cc=["cc1@example.com", "cc2@example.com"], bcc=["bcc@example.com"], ) # 验证 recipients 包含所有收件人 call_args = mock_server.sendmail.call_args recipients = call_args[0][1] assert "to@example.com" in recipients assert "cc1@example.com" in recipients assert "cc2@example.com" in recipients assert "bcc@example.com" in recipients def test_send_with_text_body(self, email_service): """带纯文本正文""" with patch("packages.adapters.smtp.email_service.smtplib.SMTP") as mock_smtp: mock_server = MagicMock() mock_smtp.return_value.__enter__.return_value = mock_server email_service.send_email( to_email="u@e.com", subject="s", html_body="html
", text_body="plain text", ) mock_server.sendmail.assert_called_once() def test_send_failure_returns_false(self, email_service): """发送失败返回 False 和错误信息""" with patch("packages.adapters.smtp.email_service.smtplib.SMTP") as mock_smtp: mock_server = MagicMock() mock_server.sendmail.side_effect = Exception("Connection refused") mock_smtp.return_value.__enter__.return_value = mock_server success, error = email_service.send_email(to_email="u@e.com", subject="s", html_body="body") assert success is False assert "Connection refused" in error def test_from_header_exists(self, email_service): """From 头存在""" with patch("packages.adapters.smtp.email_service.smtplib.SMTP") as mock_smtp: mock_server = MagicMock() mock_smtp.return_value.__enter__.return_value = mock_server email_service.send_email(to_email="u@e.com", subject="s", html_body="body") call_args = mock_server.sendmail.call_args msg_str = call_args[0][2] assert "From:" in msg_str assert "To: u@e.com" in msg_str class TestSendVerificationEmail: """发送验证邮件测试""" def test_verification_email_contains_url(self, email_service): """验证邮件包含验证链接""" with patch.object(email_service, "send_email", return_value=(True, None)) as mock_send: email_service.send_verification_email( to_email="user@example.com", username="testuser", verification_url="https://app.example.com/verify?token=abc123", ) mock_send.assert_called_once() call_args = mock_send.call_args # 验证主题 assert "验证" in call_args[0][1] # HTML 正文包含用户名和链接 assert "testuser" in call_args[0][2] assert "https://app.example.com/verify?token=abc123" in call_args[0][2] def test_verification_email_has_text_body(self, email_service): """验证邮件有纯文本版""" with patch.object(email_service, "send_email", return_value=(True, None)) as mock_send: email_service.send_verification_email( to_email="u@e.com", username="u", verification_url="https://example.com/v", ) call_args = mock_send.call_args # 第四个参数是 text_body assert call_args[0][3] is not None assert len(call_args[0][3]) > 0 class TestSendPasswordResetEmail: """发送密码重置邮件测试""" def test_reset_email_contains_url(self, email_service): """重置邮件包含重置链接""" with patch.object(email_service, "send_email", return_value=(True, None)) as mock_send: email_service.send_password_reset_email( to_email="user@example.com", username="testuser", reset_url="https://app.example.com/reset?token=xyz", ) mock_send.assert_called_once() call_args = mock_send.call_args assert "重置" in call_args[0][1] assert "testuser" in call_args[0][2] assert "https://app.example.com/reset?token=xyz" in call_args[0][2] def test_reset_email_has_text_body(self, email_service): """重置邮件有纯文本版""" with patch.object(email_service, "send_email", return_value=(True, None)) as mock_send: email_service.send_password_reset_email( to_email="u@e.com", username="u", reset_url="https://example.com/r", ) call_args = mock_send.call_args assert call_args[0][3] is not None assert len(call_args[0][3]) > 0 class TestGetEmailService: """工厂函数测试""" def test_disabled_returns_noop(self): """禁用时返回 NoopEmailService""" svc = get_email_service(enabled=False) assert isinstance(svc, NoopEmailService) def test_enabled_returns_email_service(self, email_config): """启用时返回 EmailService""" svc = get_email_service(config=email_config, enabled=True) assert isinstance(svc, EmailService) def test_singleton_default(self): """默认情况下是单例""" svc1 = get_email_service() svc2 = get_email_service() # 两个都可能是 Noop 或 EmailService,取决于环境 assert type(svc1) is type(svc2)