328 lines
11 KiB
Python
328 lines
11 KiB
Python
from __future__ import annotations
|
|
|
|
import uuid
|
|
from datetime import date, timedelta
|
|
from pathlib import Path
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
import pytest_asyncio
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
|
|
from sqlalchemy.pool import StaticPool
|
|
|
|
from app.database import Base
|
|
from app.models.subscription import Subscription
|
|
from app.models.user import User
|
|
from app.services.email.email_scheduler import EmailScheduler
|
|
from app.services.email_service import EmailService, EmailMessage
|
|
from tests.fixtures.auth import _to_uuid
|
|
|
|
|
|
TEMPLATES_DIR = Path(__file__).resolve().parent.parent.parent / "app" / "templates"
|
|
|
|
|
|
def _make_user(
|
|
user_id: str | None = None,
|
|
email: str = "test@example.com",
|
|
plan: str = "free",
|
|
) -> User:
|
|
uid = user_id or str(uuid.uuid4())
|
|
user = User(
|
|
id=uid,
|
|
email=email,
|
|
password="hashed_password",
|
|
firstName="Test",
|
|
lastName="User",
|
|
isActive=True,
|
|
emailVerified=True,
|
|
)
|
|
user.plan = plan
|
|
user.max_queries = 50 if plan != "free" else 5
|
|
return user
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def async_engine():
|
|
engine = create_async_engine(
|
|
"sqlite+aiosqlite:///:memory:",
|
|
connect_args={"check_same_thread": False},
|
|
poolclass=StaticPool,
|
|
)
|
|
async with engine.begin() as conn:
|
|
await conn.run_sync(Base.metadata.create_all)
|
|
yield engine
|
|
await engine.dispose()
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def async_session(async_engine):
|
|
maker = async_sessionmaker(
|
|
async_engine,
|
|
class_=AsyncSession,
|
|
expire_on_commit=False,
|
|
autoflush=False,
|
|
autocommit=False,
|
|
)
|
|
async with maker() as session:
|
|
yield session
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_email_service():
|
|
return EmailService(simulate_mode=True)
|
|
|
|
|
|
@pytest.fixture
|
|
def email_scheduler(mock_email_service):
|
|
return EmailScheduler(email_service=mock_email_service)
|
|
|
|
|
|
class TestEmailSchedulerInstantiation:
|
|
def test_email_scheduler_can_be_instantiated(self, mock_email_service):
|
|
scheduler = EmailScheduler(email_service=mock_email_service)
|
|
assert scheduler is not None
|
|
assert scheduler.email_service is mock_email_service
|
|
|
|
def test_email_scheduler_default_creates_mock_service(self):
|
|
scheduler = EmailScheduler()
|
|
assert scheduler is not None
|
|
assert scheduler.email_service is not None
|
|
|
|
|
|
class TestWeeklyReportEmail:
|
|
@pytest.mark.asyncio
|
|
async def test_weekly_report_sends_to_active_users(self, async_session, email_scheduler):
|
|
user1 = _make_user(email="user1@example.com")
|
|
user2 = _make_user(email="user2@example.com")
|
|
async_session.add(user1)
|
|
async_session.add(user2)
|
|
await async_session.commit()
|
|
|
|
sent_count = await email_scheduler.send_geo_weekly_report(async_session)
|
|
assert sent_count == 2
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_weekly_report_skips_inactive_users(self, async_session, email_scheduler):
|
|
active_user = _make_user(email="active@example.com")
|
|
inactive_user = _make_user(email="inactive@example.com")
|
|
inactive_user.isActive = False
|
|
async_session.add(active_user)
|
|
async_session.add(inactive_user)
|
|
await async_session.commit()
|
|
|
|
sent_count = await email_scheduler.send_geo_weekly_report(async_session)
|
|
assert sent_count == 1
|
|
|
|
|
|
class TestRenewalReminder:
|
|
@pytest.mark.asyncio
|
|
async def test_renewal_reminder_sends_for_expiring_subscriptions(self, async_session, email_scheduler):
|
|
user = _make_user(email="renew@example.com", plan="pro")
|
|
async_session.add(user)
|
|
await async_session.commit()
|
|
|
|
sub = Subscription(
|
|
user_id=user.id,
|
|
plan="pro",
|
|
status="active",
|
|
start_date=date.today() - timedelta(days=23),
|
|
end_date=date.today() + timedelta(days=7),
|
|
amount=599.0,
|
|
payment_method="mock",
|
|
)
|
|
async_session.add(sub)
|
|
await async_session.commit()
|
|
|
|
sent_count = await email_scheduler.send_renewal_reminder(async_session)
|
|
assert sent_count >= 1
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_renewal_reminder_no_reminder_for_non_expiring(self, async_session, email_scheduler):
|
|
user = _make_user(email="notexpiring@example.com", plan="pro")
|
|
async_session.add(user)
|
|
await async_session.commit()
|
|
|
|
sub = Subscription(
|
|
user_id=user.id,
|
|
plan="pro",
|
|
status="active",
|
|
start_date=date.today(),
|
|
end_date=date.today() + timedelta(days=30),
|
|
amount=599.0,
|
|
payment_method="mock",
|
|
)
|
|
async_session.add(sub)
|
|
await async_session.commit()
|
|
|
|
sent_count = await email_scheduler.send_renewal_reminder(async_session)
|
|
assert sent_count == 0
|
|
|
|
|
|
class TestWelcomeEmail:
|
|
@pytest.mark.asyncio
|
|
async def test_welcome_email_sends_successfully(self, email_scheduler):
|
|
result = await email_scheduler.send_welcome_email(
|
|
"newuser@example.com", "新用户"
|
|
)
|
|
assert result is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_welcome_email_with_empty_name(self, email_scheduler):
|
|
result = await email_scheduler.send_welcome_email(
|
|
"newuser@example.com", ""
|
|
)
|
|
assert result is True
|
|
|
|
|
|
class TestMockMode:
|
|
def test_mock_mode_logs_instead_of_sending(self, mock_email_service):
|
|
msg = EmailMessage(
|
|
to="test@example.com",
|
|
subject="Test Subject",
|
|
body_html="<p>Test</p>",
|
|
body_text="Test",
|
|
)
|
|
result = mock_email_service.send_email(msg)
|
|
assert result.success is True
|
|
assert result.message_id is not None
|
|
assert result.message_id.startswith("sim_")
|
|
assert result.error is None
|
|
|
|
def test_mock_mode_simulate_flag(self, mock_email_service):
|
|
assert mock_email_service.simulate_mode is True
|
|
|
|
def test_production_mode_would_use_smtp(self):
|
|
service = EmailService(
|
|
simulate_mode=False,
|
|
smtp_host="localhost",
|
|
smtp_port=587,
|
|
smtp_user="test",
|
|
smtp_password="test",
|
|
)
|
|
assert service.simulate_mode is False
|
|
|
|
|
|
class TestEmailTemplates:
|
|
def test_geo_weekly_report_template_exists(self):
|
|
template_path = TEMPLATES_DIR / "geo_weekly_report.html"
|
|
assert template_path.exists()
|
|
|
|
def test_renewal_reminder_template_exists(self):
|
|
template_path = TEMPLATES_DIR / "renewal_reminder.html"
|
|
assert template_path.exists()
|
|
|
|
def test_trial_expiring_template_exists(self):
|
|
template_path = TEMPLATES_DIR / "trial_expiring.html"
|
|
assert template_path.exists()
|
|
|
|
def test_welcome_template_exists(self):
|
|
template_path = TEMPLATES_DIR / "welcome.html"
|
|
assert template_path.exists()
|
|
|
|
def test_geo_weekly_report_renders_with_context(self, email_scheduler):
|
|
template_html = email_scheduler._load_template("geo_weekly_report.html")
|
|
context = {
|
|
"user_name": "测试用户",
|
|
"score_change": "+5",
|
|
"current_score": "78",
|
|
"previous_score": "73",
|
|
"top_improved": "内容质量 (+12%)",
|
|
"top_declined": "品牌权威 (-3%)",
|
|
"suggestions": "建议增加技术白皮书",
|
|
"report_link": "https://example.com",
|
|
"year": "2026",
|
|
}
|
|
rendered = email_scheduler._render_template(template_html, context)
|
|
assert "测试用户" in rendered
|
|
assert "+5" in rendered
|
|
assert "78" in rendered
|
|
assert "内容质量 (+12%)" in rendered
|
|
|
|
def test_renewal_reminder_renders_with_context(self, email_scheduler):
|
|
template_html = email_scheduler._load_template("renewal_reminder.html")
|
|
context = {
|
|
"user_name": "续费用户",
|
|
"plan_name": "专业版",
|
|
"end_date": "2026-06-30",
|
|
"days_remaining": "7",
|
|
"plan_price": "599",
|
|
"renew_link": "https://example.com",
|
|
"year": "2026",
|
|
}
|
|
rendered = email_scheduler._render_template(template_html, context)
|
|
assert "续费用户" in rendered
|
|
assert "专业版" in rendered
|
|
assert "7" in rendered
|
|
assert "599" in rendered
|
|
|
|
def test_trial_expiring_renders_with_context(self, email_scheduler):
|
|
template_html = email_scheduler._load_template("trial_expiring.html")
|
|
context = {
|
|
"user_name": "试用用户",
|
|
"days_remaining": "3",
|
|
"upgrade_link": "https://example.com",
|
|
"year": "2026",
|
|
}
|
|
rendered = email_scheduler._render_template(template_html, context)
|
|
assert "试用用户" in rendered
|
|
assert "3" in rendered
|
|
|
|
def test_welcome_renders_with_context(self, email_scheduler):
|
|
template_html = email_scheduler._load_template("welcome.html")
|
|
context = {
|
|
"user_name": "新用户",
|
|
"dashboard_link": "https://example.com",
|
|
"diagnosis_link": "https://example.com/diagnosis",
|
|
"help_link": "https://example.com/help",
|
|
"year": "2026",
|
|
}
|
|
rendered = email_scheduler._render_template(template_html, context)
|
|
assert "新用户" in rendered
|
|
assert "3" in rendered
|
|
|
|
|
|
class TestSendTemplateEmail:
|
|
def test_send_template_email_with_welcome(self, mock_email_service):
|
|
result = mock_email_service.send_template_email(
|
|
to="test@example.com",
|
|
subject="欢迎",
|
|
template_name="welcome.html",
|
|
context={
|
|
"user_name": "测试",
|
|
"dashboard_link": "https://example.com",
|
|
"diagnosis_link": "https://example.com/d",
|
|
"help_link": "https://example.com/h",
|
|
"year": "2026",
|
|
},
|
|
)
|
|
assert result.success is True
|
|
|
|
def test_send_template_email_invalid_template(self, mock_email_service):
|
|
with pytest.raises(ValueError, match="模板文件不存在"):
|
|
mock_email_service.send_template_email(
|
|
to="test@example.com",
|
|
subject="Test",
|
|
template_name="nonexistent.html",
|
|
context={},
|
|
)
|
|
|
|
def test_send_template_email_renders_context(self, mock_email_service):
|
|
result = mock_email_service.send_template_email(
|
|
to="test@example.com",
|
|
subject="周报",
|
|
template_name="geo_weekly_report.html",
|
|
context={
|
|
"user_name": "渲染测试",
|
|
"score_change": "+10",
|
|
"current_score": "85",
|
|
"previous_score": "75",
|
|
"top_improved": "测试提升",
|
|
"top_declined": "测试下降",
|
|
"suggestions": "测试建议",
|
|
"report_link": "https://example.com",
|
|
"year": "2026",
|
|
},
|
|
)
|
|
assert result.success is True
|