497 lines
16 KiB
Python
497 lines
16 KiB
Python
import pytest
|
||
from unittest.mock import Mock, patch, MagicMock
|
||
from app.services.email_service import (
|
||
EmailService,
|
||
EmailMessage,
|
||
EmailSendResult,
|
||
EMAIL_TEMPLATES,
|
||
)
|
||
|
||
|
||
class TestEmailMessage:
|
||
"""邮件消息数据结构测试"""
|
||
|
||
def test_email_message_creation(self):
|
||
"""测试EmailMessage数据创建"""
|
||
msg = EmailMessage(
|
||
to="test@example.com",
|
||
subject="测试邮件",
|
||
body_html="<h1>测试</h1>",
|
||
body_text="测试邮件",
|
||
)
|
||
assert msg.to == "test@example.com"
|
||
assert msg.subject == "测试邮件"
|
||
assert msg.body_html == "<h1>测试</h1>"
|
||
assert msg.body_text == "测试邮件"
|
||
assert msg.attachments == []
|
||
assert msg.metadata == {}
|
||
|
||
def test_email_message_with_attachments(self):
|
||
"""测试带附件的邮件消息"""
|
||
msg = EmailMessage(
|
||
to="test@example.com",
|
||
subject="带附件",
|
||
body_html="<p>内容</p>",
|
||
body_text="内容",
|
||
attachments=[{"filename": "report.pdf", "content": b"data"}],
|
||
)
|
||
assert len(msg.attachments) == 1
|
||
assert msg.attachments[0]["filename"] == "report.pdf"
|
||
|
||
def test_email_message_with_metadata(self):
|
||
"""测试带元数据的邮件消息"""
|
||
msg = EmailMessage(
|
||
to="test@example.com",
|
||
subject="测试",
|
||
body_html="<p>内容</p>",
|
||
body_text="内容",
|
||
metadata={"brand_name": "test_brand", "alert_type": "error"},
|
||
)
|
||
assert msg.metadata["brand_name"] == "test_brand"
|
||
|
||
|
||
class TestEmailSendResult:
|
||
"""邮件发送结果数据结构测试"""
|
||
|
||
def test_email_send_result_success(self):
|
||
"""测试成功发送结果"""
|
||
result = EmailSendResult(
|
||
success=True,
|
||
message_id="msg_123",
|
||
error=None,
|
||
retry_count=0,
|
||
)
|
||
assert result.success is True
|
||
assert result.message_id == "msg_123"
|
||
assert result.error is None
|
||
assert result.retry_count == 0
|
||
|
||
def test_email_send_result_failure(self):
|
||
"""测试失败发送结果"""
|
||
result = EmailSendResult(
|
||
success=False,
|
||
message_id=None,
|
||
error="SMTP连接失败",
|
||
retry_count=3,
|
||
)
|
||
assert result.success is False
|
||
assert result.message_id is None
|
||
assert result.error == "SMTP连接失败"
|
||
assert result.retry_count == 3
|
||
|
||
|
||
class TestEmailTemplateRendering:
|
||
"""邮件模板渲染测试"""
|
||
|
||
@pytest.fixture
|
||
def email_service(self):
|
||
"""创建邮件服务实例"""
|
||
return EmailService()
|
||
|
||
def test_render_alert_notification_template(self, email_service):
|
||
"""测试告警通知模板渲染"""
|
||
variables = {
|
||
"alert_type": "系统错误",
|
||
"brand_name": "测试品牌",
|
||
"severity": "高",
|
||
"description": "数据库连接失败",
|
||
"timestamp": "2024-01-01 12:00:00",
|
||
}
|
||
msg = email_service.render_template("alert_notification", "admin@example.com", variables)
|
||
|
||
assert msg.to == "admin@example.com"
|
||
assert "[GEO平台] 告警通知：系统错误" in msg.subject
|
||
assert "测试品牌" in msg.body_html
|
||
assert "系统错误" in msg.body_html
|
||
assert "高" in msg.body_html
|
||
|
||
def test_render_quota_warning_template(self, email_service):
|
||
"""测试额度预警模板渲染"""
|
||
variables = {
|
||
"quota_type": "API调用",
|
||
"usage_percentage": 85,
|
||
"used": 850,
|
||
"limit": 1000,
|
||
"recommended_action": "请升级套餐",
|
||
}
|
||
msg = email_service.render_template("quota_warning", "user@example.com", variables)
|
||
|
||
assert msg.to == "user@example.com"
|
||
assert "[GEO平台] 额度预警：API调用" in msg.subject
|
||
assert "85%" in msg.body_html
|
||
assert "850" in msg.body_html
|
||
assert "1000" in msg.body_html
|
||
|
||
def test_render_template_missing_variables(self, email_service):
|
||
"""测试模板渲染缺少变量"""
|
||
variables = {"alert_type": "系统错误"}
|
||
msg = email_service.render_template("alert_notification", "admin@example.com", variables)
|
||
|
||
assert msg is not None
|
||
assert msg.to == "admin@example.com"
|
||
|
||
def test_render_template_invalid_template(self, email_service):
|
||
"""测试无效模板名称"""
|
||
with pytest.raises(ValueError, match="模板不存在"):
|
||
email_service.render_template(
|
||
"invalid_template",
|
||
"admin@example.com",
|
||
{"key": "value"},
|
||
)
|
||
|
||
def test_render_template_variable_substitution(self, email_service):
|
||
"""测试模板变量替换"""
|
||
variables = {
|
||
"brand_name": "品牌A",
|
||
"alert_type": "告警B",
|
||
"severity": "严重",
|
||
"description": "描述C",
|
||
"timestamp": "时间D",
|
||
}
|
||
msg = email_service.render_template("alert_notification", "test@example.com", variables)
|
||
|
||
assert "品牌A" in msg.body_html
|
||
assert "告警B" in msg.body_html
|
||
assert "严重" in msg.body_html
|
||
assert "描述C" in msg.body_html
|
||
assert "时间D" in msg.body_html
|
||
|
||
|
||
class TestEmailGeneration:
|
||
"""邮件内容生成测试"""
|
||
|
||
@pytest.fixture
|
||
def email_service(self):
|
||
"""创建邮件服务实例"""
|
||
return EmailService()
|
||
|
||
def test_generate_alert_notification_email(self, email_service):
|
||
"""测试生成告警通知邮件"""
|
||
msg = email_service.generate_alert_email(
|
||
to="admin@example.com",
|
||
alert_type="数据库告警",
|
||
brand_name="测试品牌",
|
||
severity="高",
|
||
description="数据库CPU使用率超过90%",
|
||
timestamp="2024-01-01 12:00:00",
|
||
)
|
||
|
||
assert msg.to == "admin@example.com"
|
||
assert "数据库告警" in msg.subject
|
||
assert "测试品牌" in msg.body_html
|
||
assert "高" in msg.body_html
|
||
assert msg.body_text != ""
|
||
|
||
def test_generate_quota_warning_email(self, email_service):
|
||
"""测试生成额度预警邮件"""
|
||
msg = email_service.generate_quota_warning_email(
|
||
to="user@example.com",
|
||
quota_type="API调用",
|
||
usage_percentage=85,
|
||
used=850,
|
||
limit=1000,
|
||
recommended_action="建议升级套餐",
|
||
)
|
||
|
||
assert msg.to == "user@example.com"
|
||
assert "API调用" in msg.subject
|
||
assert "85%" in msg.body_html
|
||
assert "850" in msg.body_html
|
||
assert "1000" in msg.body_html
|
||
|
||
def test_generate_email_has_both_formats(self, email_service):
|
||
"""测试生成的邮件包含HTML和纯文本格式"""
|
||
msg = email_service.generate_alert_email(
|
||
to="test@example.com",
|
||
alert_type="测试",
|
||
brand_name="品牌",
|
||
severity="中",
|
||
description="描述",
|
||
timestamp="时间",
|
||
)
|
||
|
||
assert msg.body_html != ""
|
||
assert msg.body_text != ""
|
||
assert "<" in msg.body_html
|
||
assert "<" not in msg.body_text
|
||
|
||
|
||
class TestEmailSending:
|
||
"""邮件发送测试"""
|
||
|
||
@pytest.fixture
|
||
def email_service(self):
|
||
"""创建邮件服务实例（模拟模式）"""
|
||
return EmailService(simulate_mode=True)
|
||
|
||
def test_send_email_simulate_mode(self, email_service):
|
||
"""测试模拟模式发送邮件"""
|
||
msg = EmailMessage(
|
||
to="test@example.com",
|
||
subject="测试",
|
||
body_html="<p>测试</p>",
|
||
body_text="测试",
|
||
)
|
||
result = email_service.send_email(msg)
|
||
|
||
assert result.success is True
|
||
assert result.message_id is not None
|
||
assert result.error is None
|
||
|
||
@patch("smtplib.SMTP")
|
||
def test_send_email_real_smtp(self, mock_smtp):
|
||
"""测试真实SMTP发送（模拟SMTP）"""
|
||
mock_server = MagicMock()
|
||
mock_smtp.return_value = mock_server
|
||
|
||
service = EmailService(
|
||
simulate_mode=False,
|
||
smtp_host="smtp.example.com",
|
||
smtp_port=587,
|
||
smtp_user="user@example.com",
|
||
smtp_password="password",
|
||
)
|
||
|
||
msg = EmailMessage(
|
||
to="recipient@example.com",
|
||
subject="测试",
|
||
body_html="<p>内容</p>",
|
||
body_text="内容",
|
||
)
|
||
|
||
result = service.send_email(msg)
|
||
|
||
assert result.success is True
|
||
mock_smtp.assert_called_once_with("smtp.example.com", 587)
|
||
mock_server.starttls.assert_called_once()
|
||
mock_server.login.assert_called_once_with("user@example.com", "password")
|
||
|
||
@patch("smtplib.SMTP")
|
||
def test_send_email_smtp_failure(self, mock_smtp):
|
||
"""测试SMTP发送失败"""
|
||
mock_smtp.side_effect = Exception("连接失败")
|
||
|
||
service = EmailService(
|
||
simulate_mode=False,
|
||
smtp_host="smtp.example.com",
|
||
smtp_port=587,
|
||
smtp_user="user@example.com",
|
||
smtp_password="password",
|
||
)
|
||
|
||
msg = EmailMessage(
|
||
to="test@example.com",
|
||
subject="测试",
|
||
body_html="<p>内容</p>",
|
||
body_text="内容",
|
||
)
|
||
|
||
result = service.send_email(msg)
|
||
|
||
assert result.success is False
|
||
assert result.error is not None
|
||
assert "连接失败" in result.error
|
||
|
||
@patch("smtplib.SMTP")
|
||
def test_send_email_with_retry(self, mock_smtp):
|
||
"""测试邮件发送重试"""
|
||
mock_server = MagicMock()
|
||
mock_smtp.return_value = mock_server
|
||
|
||
service = EmailService(
|
||
simulate_mode=False,
|
||
smtp_host="smtp.example.com",
|
||
smtp_port=587,
|
||
smtp_user="user@example.com",
|
||
smtp_password="password",
|
||
max_retries=3,
|
||
)
|
||
|
||
msg = EmailMessage(
|
||
to="test@example.com",
|
||
subject="测试",
|
||
body_html="<p>内容</p>",
|
||
body_text="内容",
|
||
)
|
||
|
||
result = service.send_email(msg)
|
||
|
||
assert result.success is True
|
||
assert result.retry_count == 0
|
||
|
||
|
||
class TestEmailQueue:
|
||
"""邮件队列测试"""
|
||
|
||
@pytest.fixture
|
||
def email_service(self):
|
||
"""创建邮件服务实例"""
|
||
return EmailService(simulate_mode=True)
|
||
|
||
def test_add_to_queue(self, email_service):
|
||
"""测试添加邮件到队列"""
|
||
msg = EmailMessage(
|
||
to="test@example.com",
|
||
subject="测试",
|
||
body_html="<p>内容</p>",
|
||
body_text="内容",
|
||
)
|
||
|
||
email_service.add_to_queue(msg)
|
||
|
||
assert len(email_service.get_queue()) == 1
|
||
assert email_service.get_queue()[0].to == "test@example.com"
|
||
|
||
def test_add_multiple_to_queue(self, email_service):
|
||
"""测试批量添加邮件到队列"""
|
||
messages = [
|
||
EmailMessage(to=f"user{i}@example.com", subject=f"测试{i}", body_html="<p>内容</p>", body_text="内容")
|
||
for i in range(5)
|
||
]
|
||
|
||
for msg in messages:
|
||
email_service.add_to_queue(msg)
|
||
|
||
assert len(email_service.get_queue()) == 5
|
||
|
||
def test_send_queue(self, email_service):
|
||
"""测试发送队列中的邮件"""
|
||
messages = [
|
||
EmailMessage(to=f"user{i}@example.com", subject=f"测试{i}", body_html="<p>内容</p>", body_text="内容")
|
||
for i in range(3)
|
||
]
|
||
|
||
for msg in messages:
|
||
email_service.add_to_queue(msg)
|
||
|
||
results = email_service.send_queue()
|
||
|
||
assert len(results) == 3
|
||
assert all(r.success for r in results)
|
||
assert len(email_service.get_queue()) == 0
|
||
|
||
def test_send_queue_empty(self, email_service):
|
||
"""测试发送空队列"""
|
||
results = email_service.send_queue()
|
||
|
||
assert results == []
|
||
|
||
def test_clear_queue(self, email_service):
|
||
"""测试清空队列"""
|
||
msg = EmailMessage(
|
||
to="test@example.com",
|
||
subject="测试",
|
||
body_html="<p>内容</p>",
|
||
body_text="内容",
|
||
)
|
||
email_service.add_to_queue(msg)
|
||
assert len(email_service.get_queue()) == 1
|
||
|
||
email_service.clear_queue()
|
||
|
||
assert len(email_service.get_queue()) == 0
|
||
|
||
|
||
class TestEmailAttachments:
|
||
"""邮件附件测试"""
|
||
|
||
@pytest.fixture
|
||
def email_service(self):
|
||
"""创建邮件服务实例"""
|
||
return EmailService(simulate_mode=True)
|
||
|
||
def test_add_attachment_to_message(self, email_service):
|
||
"""测试添加附件到邮件"""
|
||
msg = EmailMessage(
|
||
to="test@example.com",
|
||
subject="带附件",
|
||
body_html="<p>内容</p>",
|
||
body_text="内容",
|
||
)
|
||
|
||
email_service.add_attachment(msg, "report.pdf", b"PDF content")
|
||
|
||
assert len(msg.attachments) == 1
|
||
assert msg.attachments[0]["filename"] == "report.pdf"
|
||
assert msg.attachments[0]["content"] == b"PDF content"
|
||
|
||
def test_add_multiple_attachments(self, email_service):
|
||
"""测试添加多个附件"""
|
||
msg = EmailMessage(
|
||
to="test@example.com",
|
||
subject="多附件",
|
||
body_html="<p>内容</p>",
|
||
body_text="内容",
|
||
)
|
||
|
||
email_service.add_attachment(msg, "file1.pdf", b"content1")
|
||
email_service.add_attachment(msg, "file2.xlsx", b"content2")
|
||
|
||
assert len(msg.attachments) == 2
|
||
assert msg.attachments[0]["filename"] == "file1.pdf"
|
||
assert msg.attachments[1]["filename"] == "file2.xlsx"
|
||
|
||
|
||
class TestEmailValidation:
|
||
"""邮箱地址验证测试"""
|
||
|
||
@pytest.fixture
|
||
def email_service(self):
|
||
"""创建邮件服务实例"""
|
||
return EmailService()
|
||
|
||
def test_validate_valid_email(self, email_service):
|
||
"""测试有效邮箱地址"""
|
||
assert email_service.validate_email("test@example.com") is True
|
||
assert email_service.validate_email("user.name@domain.org") is True
|
||
assert email_service.validate_email("user+tag@example.com") is True
|
||
|
||
def test_validate_invalid_email(self, email_service):
|
||
"""测试无效邮箱地址"""
|
||
assert email_service.validate_email("invalid") is False
|
||
assert email_service.validate_email("invalid@") is False
|
||
assert email_service.validate_email("@example.com") is False
|
||
assert email_service.validate_email("test@") is False
|
||
assert email_service.validate_email("") is False
|
||
|
||
def test_send_to_invalid_email(self, email_service):
|
||
"""测试发送到无效邮箱"""
|
||
msg = EmailMessage(
|
||
to="invalid_email",
|
||
subject="测试",
|
||
body_html="<p>内容</p>",
|
||
body_text="内容",
|
||
)
|
||
|
||
result = email_service.send_email(msg)
|
||
|
||
assert result.success is False
|
||
assert result.error is not None
|
||
|
||
|
||
class TestEmailTemplatesConstants:
|
||
"""邮件模板常量测试"""
|
||
|
||
def test_email_templates_exist(self):
|
||
"""测试邮件模板存在"""
|
||
assert "alert_notification" in EMAIL_TEMPLATES
|
||
assert "quota_warning" in EMAIL_TEMPLATES
|
||
|
||
def test_alert_notification_template_structure(self):
|
||
"""测试告警通知模板结构"""
|
||
template = EMAIL_TEMPLATES["alert_notification"]
|
||
assert "subject" in template
|
||
assert "body_html" in template
|
||
assert "body_text" in template
|
||
assert "{alert_type}" in template["subject"]
|
||
assert "{brand_name}" in template["body_html"]
|
||
|
||
def test_quota_warning_template_structure(self):
|
||
"""测试额度预警模板结构"""
|
||
template = EMAIL_TEMPLATES["quota_warning"]
|
||
assert "subject" in template
|
||
assert "body_html" in template
|
||
assert "body_text" in template
|
||
assert "{quota_type}" in template["subject"]
|
||
assert "{usage_percentage}" in template["body_html"]
|