d92909321c
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (push) Successful in 1m14s
CI/CD Pipeline / Check if frontend-only change (push) Has been skipped
CI/CD Pipeline / Validate - Code Quality (push) Failing after 2m12s
CI/CD Pipeline / Validate - Type Check (mypy) (push) Successful in 1m28s
CI/CD Pipeline / Validate - Migration (alembic) (push) Successful in 1m24s
CI/CD Pipeline / Unit Tests (push) Successful in 6m25s
CI/CD Pipeline / Integration Tests (push) Successful in 5m4s
CI/CD Pipeline / Frontend Lint (push) Successful in 45s
CI/CD Pipeline / Frontend Unit Tests (push) Successful in 1m23s
CI/CD Pipeline / PR Build API Image (push) Has been skipped
CI/CD Pipeline / PR Build Web Image (push) Has been skipped
CI/CD Pipeline / PR Build Worker Image (push) Has been skipped
CI/CD Pipeline / Build Staging API Image (push) Successful in 11m44s
CI/CD Pipeline / Build Staging Web Image (push) Successful in 1m17s
CI/CD Pipeline / Build Staging Worker Image (push) Successful in 2m36s
CI/CD Pipeline / Build Production API Image (push) Has been skipped
CI/CD Pipeline / Build Production Web Image (push) Has been skipped
CI/CD Pipeline / Build Production Worker Image (push) Has been skipped
CI/CD Pipeline / Deploy Production (push) Has been skipped
CI/CD Pipeline / Production Browser E2E (push) Has been skipped
CI/CD Pipeline / Staging E2E Tests (push) Has been cancelled
CI/CD Pipeline / Staging API Integration Tests (push) Has been cancelled
CI/CD Pipeline / ACR Image Cleanup (push) Has been cancelled
177 lines
5.5 KiB
Python
Executable File
177 lines
5.5 KiB
Python
Executable File
"""
|
|
邮件服务测试
|
|
"""
|
|
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from packages.adapters.smtp.email_service import EmailConfig, EmailService
|
|
|
|
|
|
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="<h1>Test</h1>",
|
|
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="<h1>Test</h1>",
|
|
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="<h1>Test</h1>",
|
|
)
|
|
|
|
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_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"
|