estel/dify
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

refactor: centralize email internationalization handling (#22752)

Browse Source 
Co-authored-by: Claude <noreply@anthropic.com>
This commit is contained in:
-LAN-
2025-07-23 00:26:00 +08:00
committed by GitHub
parent 5c7f0a533a
commit 0f4809b9b8
11 changed files with 1200 additions and 289 deletions
Download Patch File Download Diff File Expand all files Collapse all files

461
api/libs/email_i18n.py Normal file
View File

@@ -0,0 +1,461 @@
"""
Email Internationalization Module
This module provides a centralized, elegant way to handle email internationalization
in Dify. It follows Domain-Driven Design principles with proper type hints and
eliminates the need for repetitive language switching logic.
"""
from dataclasses import dataclass
from enum import Enum
from typing import Any, Optional, Protocol
from flask import render_template
from pydantic import BaseModel, Field
from extensions.ext_mail import mail
from services.feature_service import BrandingModel, FeatureService
class EmailType(Enum):
"""Enumeration of supported email types."""
RESET_PASSWORD = "reset_password"
INVITE_MEMBER = "invite_member"
EMAIL_CODE_LOGIN = "email_code_login"
CHANGE_EMAIL_OLD = "change_email_old"
CHANGE_EMAIL_NEW = "change_email_new"
OWNER_TRANSFER_CONFIRM = "owner_transfer_confirm"
OWNER_TRANSFER_OLD_NOTIFY = "owner_transfer_old_notify"
OWNER_TRANSFER_NEW_NOTIFY = "owner_transfer_new_notify"
ACCOUNT_DELETION_SUCCESS = "account_deletion_success"
ACCOUNT_DELETION_VERIFICATION = "account_deletion_verification"
ENTERPRISE_CUSTOM = "enterprise_custom"
QUEUE_MONITOR_ALERT = "queue_monitor_alert"
DOCUMENT_CLEAN_NOTIFY = "document_clean_notify"
class EmailLanguage(Enum):
"""Supported email languages with fallback handling."""
EN_US = "en-US"
ZH_HANS = "zh-Hans"
@classmethod
def from_language_code(cls, language_code: str) -> "EmailLanguage":
"""Convert a language code to EmailLanguage with fallback to English."""
if language_code == "zh-Hans":
return cls.ZH_HANS
return cls.EN_US
@dataclass(frozen=True)
class EmailTemplate:
"""Immutable value object representing an email template configuration."""
subject: str
template_path: str
branded_template_path: str
@dataclass(frozen=True)
class EmailContent:
"""Immutable value object containing rendered email content."""
subject: str
html_content: str
template_context: dict[str, Any]
class EmailI18nConfig(BaseModel):
"""Configuration for email internationalization."""
model_config = {"frozen": True, "extra": "forbid"}
templates: dict[EmailType, dict[EmailLanguage, EmailTemplate]] = Field(
default_factory=dict, description="Mapping of email types to language-specific templates"
)
def get_template(self, email_type: EmailType, language: EmailLanguage) -> EmailTemplate:
"""Get template configuration for specific email type and language."""
type_templates = self.templates.get(email_type)
if not type_templates:
raise ValueError(f"No templates configured for email type: {email_type}")
template = type_templates.get(language)
if not template:
# Fallback to English if specific language not found
template = type_templates.get(EmailLanguage.EN_US)
if not template:
raise ValueError(f"No template found for {email_type} in {language} or English")
return template
class EmailRenderer(Protocol):
"""Protocol for email template renderers."""
def render_template(self, template_path: str, **context: Any) -> str:
"""Render email template with given context."""
...
class FlaskEmailRenderer:
"""Flask-based email template renderer."""
def render_template(self, template_path: str, **context: Any) -> str:
"""Render email template using Flask's render_template."""
return render_template(template_path, **context)
class BrandingService(Protocol):
"""Protocol for branding service abstraction."""
def get_branding_config(self) -> BrandingModel:
"""Get current branding configuration."""
...
class FeatureBrandingService:
"""Feature service based branding implementation."""
def get_branding_config(self) -> BrandingModel:
"""Get branding configuration from feature service."""
return FeatureService.get_system_features().branding
class EmailSender(Protocol):
"""Protocol for email sending abstraction."""
def send_email(self, to: str, subject: str, html_content: str) -> None:
"""Send email with given parameters."""
...
class FlaskMailSender:
"""Flask-Mail based email sender."""
def send_email(self, to: str, subject: str, html_content: str) -> None:
"""Send email using Flask-Mail."""
if mail.is_inited():
mail.send(to=to, subject=subject, html=html_content)
class EmailI18nService:
"""
Main service for internationalized email handling.
This service provides a clean API for sending internationalized emails
with proper branding support and template management.
"""
def __init__(
self,
config: EmailI18nConfig,
renderer: EmailRenderer,
branding_service: BrandingService,
sender: EmailSender,
) -> None:
self._config = config
self._renderer = renderer
self._branding_service = branding_service
self._sender = sender
def send_email(
self,
email_type: EmailType,
language_code: str,
to: str,
template_context: Optional[dict[str, Any]] = None,
) -> None:
"""
Send internationalized email with branding support.
Args:
email_type: Type of email to send
language_code: Target language code
to: Recipient email address
template_context: Additional context for template rendering
"""
if template_context is None:
template_context = {}
language = EmailLanguage.from_language_code(language_code)
email_content = self._render_email_content(email_type, language, template_context)
self._sender.send_email(to=to, subject=email_content.subject, html_content=email_content.html_content)
def send_change_email(
self,
language_code: str,
to: str,
code: str,
phase: str,
) -> None:
"""
Send change email notification with phase-specific handling.
Args:
language_code: Target language code
to: Recipient email address
code: Verification code
phase: Either 'old_email' or 'new_email'
"""
if phase == "old_email":
email_type = EmailType.CHANGE_EMAIL_OLD
elif phase == "new_email":
email_type = EmailType.CHANGE_EMAIL_NEW
else:
raise ValueError(f"Invalid phase: {phase}. Must be 'old_email' or 'new_email'")
self.send_email(
email_type=email_type,
language_code=language_code,
to=to,
template_context={
"to": to,
"code": code,
},
)
def send_raw_email(
self,
to: str | list[str],
subject: str,
html_content: str,
) -> None:
"""
Send a raw email directly without template processing.
This method is provided for backward compatibility with legacy email
sending that uses pre-rendered HTML content (e.g., enterprise emails
with custom templates).
Args:
to: Recipient email address(es)
subject: Email subject
html_content: Pre-rendered HTML content
"""
if isinstance(to, list):
for recipient in to:
self._sender.send_email(to=recipient, subject=subject, html_content=html_content)
else:
self._sender.send_email(to=to, subject=subject, html_content=html_content)
def _render_email_content(
self,
email_type: EmailType,
language: EmailLanguage,
template_context: dict[str, Any],
) -> EmailContent:
"""Render email content with branding and internationalization."""
template_config = self._config.get_template(email_type, language)
branding = self._branding_service.get_branding_config()
# Determine template path based on branding
template_path = template_config.branded_template_path if branding.enabled else template_config.template_path
# Prepare template context with branding information
full_context = {
**template_context,
"branding_enabled": branding.enabled,
"application_title": branding.application_title if branding.enabled else "Dify",
}
# Render template
html_content = self._renderer.render_template(template_path, **full_context)
# Apply templating to subject with all context variables
subject = template_config.subject
try:
subject = subject.format(**full_context)
except KeyError:
# If template variables are missing, fall back to basic formatting
if branding.enabled and "{application_title}" in subject:
subject = subject.format(application_title=branding.application_title)
return EmailContent(
subject=subject,
html_content=html_content,
template_context=full_context,
)
def create_default_email_config() -> EmailI18nConfig:
"""Create default email i18n configuration with all supported templates."""
templates: dict[EmailType, dict[EmailLanguage, EmailTemplate]] = {
EmailType.RESET_PASSWORD: {
EmailLanguage.EN_US: EmailTemplate(
subject="Set Your {application_title} Password",
template_path="reset_password_mail_template_en-US.html",
branded_template_path="without-brand/reset_password_mail_template_en-US.html",
),
EmailLanguage.ZH_HANS: EmailTemplate(
subject="设置您的 {application_title} 密码",
template_path="reset_password_mail_template_zh-CN.html",
branded_template_path="without-brand/reset_password_mail_template_zh-CN.html",
),
},
EmailType.INVITE_MEMBER: {
EmailLanguage.EN_US: EmailTemplate(
subject="Join {application_title} Workspace Now",
template_path="invite_member_mail_template_en-US.html",
branded_template_path="without-brand/invite_member_mail_template_en-US.html",
),
EmailLanguage.ZH_HANS: EmailTemplate(
subject="立即加入 {application_title} 工作空间",
template_path="invite_member_mail_template_zh-CN.html",
branded_template_path="without-brand/invite_member_mail_template_zh-CN.html",
),
},
EmailType.EMAIL_CODE_LOGIN: {
EmailLanguage.EN_US: EmailTemplate(
subject="{application_title} Login Code",
template_path="email_code_login_mail_template_en-US.html",
branded_template_path="without-brand/email_code_login_mail_template_en-US.html",
),
EmailLanguage.ZH_HANS: EmailTemplate(
subject="{application_title} 登录验证码",
template_path="email_code_login_mail_template_zh-CN.html",
branded_template_path="without-brand/email_code_login_mail_template_zh-CN.html",
),
},
EmailType.CHANGE_EMAIL_OLD: {
EmailLanguage.EN_US: EmailTemplate(
subject="Check your current email",
template_path="change_mail_confirm_old_template_en-US.html",
branded_template_path="without-brand/change_mail_confirm_old_template_en-US.html",
),
EmailLanguage.ZH_HANS: EmailTemplate(
subject="检测您现在的邮箱",
template_path="change_mail_confirm_old_template_zh-CN.html",
branded_template_path="without-brand/change_mail_confirm_old_template_zh-CN.html",
),
},
EmailType.CHANGE_EMAIL_NEW: {
EmailLanguage.EN_US: EmailTemplate(
subject="Confirm your new email address",
template_path="change_mail_confirm_new_template_en-US.html",
branded_template_path="without-brand/change_mail_confirm_new_template_en-US.html",
),
EmailLanguage.ZH_HANS: EmailTemplate(
subject="确认您的邮箱地址变更",
template_path="change_mail_confirm_new_template_zh-CN.html",
branded_template_path="without-brand/change_mail_confirm_new_template_zh-CN.html",
),
},
EmailType.OWNER_TRANSFER_CONFIRM: {
EmailLanguage.EN_US: EmailTemplate(
subject="Verify Your Request to Transfer Workspace Ownership",
template_path="transfer_workspace_owner_confirm_template_en-US.html",
branded_template_path="without-brand/transfer_workspace_owner_confirm_template_en-US.html",
),
EmailLanguage.ZH_HANS: EmailTemplate(
subject="验证您转移工作空间所有权的请求",
template_path="transfer_workspace_owner_confirm_template_zh-CN.html",
branded_template_path="without-brand/transfer_workspace_owner_confirm_template_zh-CN.html",
),
},
EmailType.OWNER_TRANSFER_OLD_NOTIFY: {
EmailLanguage.EN_US: EmailTemplate(
subject="Workspace ownership has been transferred",
template_path="transfer_workspace_old_owner_notify_template_en-US.html",
branded_template_path="without-brand/transfer_workspace_old_owner_notify_template_en-US.html",
),
EmailLanguage.ZH_HANS: EmailTemplate(
subject="工作区所有权已转移",
template_path="transfer_workspace_old_owner_notify_template_zh-CN.html",
branded_template_path="without-brand/transfer_workspace_old_owner_notify_template_zh-CN.html",
),
},
EmailType.OWNER_TRANSFER_NEW_NOTIFY: {
EmailLanguage.EN_US: EmailTemplate(
subject="You are now the owner of {WorkspaceName}",
template_path="transfer_workspace_new_owner_notify_template_en-US.html",
branded_template_path="without-brand/transfer_workspace_new_owner_notify_template_en-US.html",
),
EmailLanguage.ZH_HANS: EmailTemplate(
subject="您现在是 {WorkspaceName} 的所有者",
template_path="transfer_workspace_new_owner_notify_template_zh-CN.html",
branded_template_path="without-brand/transfer_workspace_new_owner_notify_template_zh-CN.html",
),
},
EmailType.ACCOUNT_DELETION_SUCCESS: {
EmailLanguage.EN_US: EmailTemplate(
subject="Your Dify.AI Account Has Been Successfully Deleted",
template_path="delete_account_success_template_en-US.html",
branded_template_path="delete_account_success_template_en-US.html",
),
EmailLanguage.ZH_HANS: EmailTemplate(
subject="您的 Dify.AI 账户已成功删除",
template_path="delete_account_success_template_zh-CN.html",
branded_template_path="delete_account_success_template_zh-CN.html",
),
},
EmailType.ACCOUNT_DELETION_VERIFICATION: {
EmailLanguage.EN_US: EmailTemplate(
subject="Dify.AI Account Deletion and Verification",
template_path="delete_account_code_email_template_en-US.html",
branded_template_path="delete_account_code_email_template_en-US.html",
),
EmailLanguage.ZH_HANS: EmailTemplate(
subject="Dify.AI 账户删除和验证",
template_path="delete_account_code_email_template_zh-CN.html",
branded_template_path="delete_account_code_email_template_zh-CN.html",
),
},
EmailType.QUEUE_MONITOR_ALERT: {
EmailLanguage.EN_US: EmailTemplate(
subject="Alert: Dataset Queue pending tasks exceeded the limit",
template_path="queue_monitor_alert_email_template_en-US.html",
branded_template_path="queue_monitor_alert_email_template_en-US.html",
),
EmailLanguage.ZH_HANS: EmailTemplate(
subject="警报：数据集队列待处理任务超过限制",
template_path="queue_monitor_alert_email_template_zh-CN.html",
branded_template_path="queue_monitor_alert_email_template_zh-CN.html",
),
},
EmailType.DOCUMENT_CLEAN_NOTIFY: {
EmailLanguage.EN_US: EmailTemplate(
subject="Dify Knowledge base auto disable notification",
template_path="clean_document_job_mail_template-US.html",
branded_template_path="clean_document_job_mail_template-US.html",
),
EmailLanguage.ZH_HANS: EmailTemplate(
subject="Dify 知识库自动禁用通知",
template_path="clean_document_job_mail_template_zh-CN.html",
branded_template_path="clean_document_job_mail_template_zh-CN.html",
),
},
}
return EmailI18nConfig(templates=templates)
# Singleton instance for application-wide use
def get_default_email_i18n_service() -> EmailI18nService:
"""Get configured email i18n service with default dependencies."""
config = create_default_email_config()
renderer = FlaskEmailRenderer()
branding_service = FeatureBrandingService()
sender = FlaskMailSender()
return EmailI18nService(
config=config,
renderer=renderer,
branding_service=branding_service,
sender=sender,
)
# Global instance
_email_i18n_service: Optional[EmailI18nService] = None
def get_email_i18n_service() -> EmailI18nService:
"""Get global email i18n service instance."""
global _email_i18n_service
if _email_i18n_service is None:
_email_i18n_service = get_default_email_i18n_service()
return _email_i18n_service

20
api/schedule/mail_clean_document_notify_task.py
View File

@@ -3,12 +3,12 @@ import time
from collections import defaultdict
import click
from flask import render_template # type: ignore
import app
from configs import dify_config
from extensions.ext_database import db
from extensions.ext_mail import mail
from libs.email_i18n import EmailType, get_email_i18n_service
from models.account import Account, Tenant, TenantAccountJoin
from models.dataset import Dataset, DatasetAutoDisableLog
from services.feature_service import FeatureService
@@ -72,14 +72,16 @@ def mail_clean_document_notify_task():
document_count = len(document_ids)
knowledge_details.append(rf"Knowledge base {dataset.name}: {document_count} documents")
if knowledge_details:
html_content = render_template(
"clean_document_job_mail_template-US.html",
userName=account.email,
knowledge_details=knowledge_details,
url=url,
)
mail.send(
to=account.email, subject="Dify Knowledge base auto disable notification", html=html_content
email_service = get_email_i18n_service()
email_service.send_email(
email_type=EmailType.DOCUMENT_CLEAN_NOTIFY,
language_code="en-US",
to=account.email,
template_context={
"userName": account.email,
"knowledge_details": knowledge_details,
"url": url,
},
)
# update notified to True

23
api/schedule/queue_monitor_task.py
View File

@@ -3,13 +3,12 @@ from datetime import datetime
from urllib.parse import urlparse
import click
from flask import render_template
from redis import Redis
import app
from configs import dify_config
from extensions.ext_database import db
from extensions.ext_mail import mail
from libs.email_i18n import EmailType, get_email_i18n_service
# Create a dedicated Redis connection (using the same configuration as Celery)
celery_broker_url = dify_config.CELERY_BROKER_URL
@@ -39,18 +38,20 @@ def queue_monitor_task():
alter_emails = dify_config.QUEUE_MONITOR_ALERT_EMAILS
if alter_emails:
to_list = alter_emails.split(",")
email_service = get_email_i18n_service()
for to in to_list:
try:
current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
html_content = render_template(
"queue_monitor_alert_email_template_en-US.html",
queue_name=queue_name,
queue_length=queue_length,
threshold=threshold,
alert_time=current_time,
)
mail.send(
to=to, subject="Alert: Dataset Queue pending tasks exceeded the limit", html=html_content
email_service.send_email(
email_type=EmailType.QUEUE_MONITOR_ALERT,
language_code="en-US",
to=to,
template_context={
"queue_name": queue_name,
"queue_length": queue_length,
"threshold": threshold,
"alert_time": current_time,
},
)
except Exception as e:
logging.exception(click.style("Exception occurred during sending email", fg="red"))

46
api/tasks/mail_account_deletion_task.py
View File

@@ -3,14 +3,20 @@ import time
import click
from celery import shared_task # type: ignore
from flask import render_template
from extensions.ext_mail import mail
from libs.email_i18n import EmailType, get_email_i18n_service
@shared_task(queue="mail")
def send_deletion_success_task(to):
"""Send email to user regarding account deletion."""
def send_deletion_success_task(to: str, language: str = "en-US") -> None:
"""
Send account deletion success email with internationalization support.
Args:
to: Recipient email address
language: Language code for email localization
"""
if not mail.is_inited():
return
@@ -18,12 +24,16 @@ def send_deletion_success_task(to):
start_at = time.perf_counter()
try:
html_content = render_template(
"delete_account_success_template_en-US.html",
email_service = get_email_i18n_service()
email_service.send_email(
email_type=EmailType.ACCOUNT_DELETION_SUCCESS,
language_code=language,
to=to,
email=to,
template_context={
"to": to,
"email": to,
},
)
mail.send(to=to, subject="Your Dify.AI Account Has Been Successfully Deleted", html=html_content)
end_at = time.perf_counter()
logging.info(
@@ -36,12 +46,14 @@ def send_deletion_success_task(to):
@shared_task(queue="mail")
def send_account_deletion_verification_code(to, code):
"""Send email to user regarding account deletion verification code.
def send_account_deletion_verification_code(to: str, code: str, language: str = "en-US") -> None:
"""
Send account deletion verification code email with internationalization support.
Args:
to (str): Recipient email address
code (str): Verification code
to: Recipient email address
code: Verification code
language: Language code for email localization
"""
if not mail.is_inited():
return
@@ -50,8 +62,16 @@ def send_account_deletion_verification_code(to, code):
start_at = time.perf_counter()
try:
html_content = render_template("delete_account_code_email_template_en-US.html", to=to, code=code)
mail.send(to=to, subject="Dify.AI Account Deletion and Verification", html=html_content)
email_service = get_email_i18n_service()
email_service.send_email(
email_type=EmailType.ACCOUNT_DELETION_VERIFICATION,
language_code=language,
to=to,
template_context={
"to": to,
"code": code,
},
)
end_at = time.perf_counter()
logging.info(

68
api/tasks/mail_change_mail_task.py
View File

@@ -3,20 +3,21 @@ import time
import click
from celery import shared_task # type: ignore
from flask import render_template
from extensions.ext_mail import mail
from services.feature_service import FeatureService
from libs.email_i18n import get_email_i18n_service
@shared_task(queue="mail")
def send_change_mail_task(language: str, to: str, code: str, phase: str):
def send_change_mail_task(language: str, to: str, code: str, phase: str) -> None:
"""
Async Send change email mail
:param language: Language in which the email should be sent (e.g., 'en', 'zh')
:param to: Recipient email address
:param code: Change email code
:param phase: Change email phase (new_email, old_email)
Send change email notification with internationalization support.
Args:
language: Language code for email localization
to: Recipient email address
code: Email verification code
phase: Change email phase ('old_email' or 'new_email')
"""
if not mail.is_inited():
return
@@ -24,51 +25,14 @@ def send_change_mail_task(language: str, to: str, code: str, phase: str):
logging.info(click.style("Start change email mail to {}".format(to), fg="green"))
start_at = time.perf_counter()
email_config = {
"zh-Hans": {
"old_email": {
"subject": "检测您现在的邮箱",
"template_with_brand": "change_mail_confirm_old_template_zh-CN.html",
"template_without_brand": "without-brand/change_mail_confirm_old_template_zh-CN.html",
},
"new_email": {
"subject": "确认您的邮箱地址变更",
"template_with_brand": "change_mail_confirm_new_template_zh-CN.html",
"template_without_brand": "without-brand/change_mail_confirm_new_template_zh-CN.html",
},
},
"en": {
"old_email": {
"subject": "Check your current email",
"template_with_brand": "change_mail_confirm_old_template_en-US.html",
"template_without_brand": "without-brand/change_mail_confirm_old_template_en-US.html",
},
"new_email": {
"subject": "Confirm your new email address",
"template_with_brand": "change_mail_confirm_new_template_en-US.html",
"template_without_brand": "without-brand/change_mail_confirm_new_template_en-US.html",
},
},
}
# send change email mail using different languages
try:
system_features = FeatureService.get_system_features()
lang_key = "zh-Hans" if language == "zh-Hans" else "en"
if phase not in ["old_email", "new_email"]:
raise ValueError("Invalid phase")
config = email_config[lang_key][phase]
subject = config["subject"]
if system_features.branding.enabled:
template = config["template_without_brand"]
else:
template = config["template_with_brand"]
html_content = render_template(template, to=to, code=code)
mail.send(to=to, subject=subject, html=html_content)
email_service = get_email_i18n_service()
email_service.send_change_email(
language_code=language,
to=to,
code=code,
phase=phase,
)
end_at = time.perf_counter()
logging.info(

46
api/tasks/mail_email_code_login.py
View File

@@ -3,19 +3,20 @@ import time
import click
from celery import shared_task # type: ignore
from flask import render_template
from extensions.ext_mail import mail
from services.feature_service import FeatureService
from libs.email_i18n import EmailType, get_email_i18n_service
@shared_task(queue="mail")
def send_email_code_login_mail_task(language: str, to: str, code: str):
def send_email_code_login_mail_task(language: str, to: str, code: str) -> None:
"""
Async Send email code login mail
:param language: Language in which the email should be sent (e.g., 'en', 'zh')
:param to: Recipient email address
:param code: Email code to be included in the email
Send email code login email with internationalization support.
Args:
language: Language code for email localization
to: Recipient email address
code: Email verification code
"""
if not mail.is_inited():
return
@@ -23,28 +24,17 @@ def send_email_code_login_mail_task(language: str, to: str, code: str):
logging.info(click.style("Start email code login mail to {}".format(to), fg="green"))
start_at = time.perf_counter()
# send email code login mail using different languages
try:
if language == "zh-Hans":
template = "email_code_login_mail_template_zh-CN.html"
system_features = FeatureService.get_system_features()
if system_features.branding.enabled:
application_title = system_features.branding.application_title
template = "without-brand/email_code_login_mail_template_zh-CN.html"
html_content = render_template(template, to=to, code=code, application_title=application_title)
else:
html_content = render_template(template, to=to, code=code)
mail.send(to=to, subject="邮箱验证码", html=html_content)
else:
template = "email_code_login_mail_template_en-US.html"
system_features = FeatureService.get_system_features()
if system_features.branding.enabled:
application_title = system_features.branding.application_title
template = "without-brand/email_code_login_mail_template_en-US.html"
html_content = render_template(template, to=to, code=code, application_title=application_title)
else:
html_content = render_template(template, to=to, code=code)
mail.send(to=to, subject="Email Code", html=html_content)
email_service = get_email_i18n_service()
email_service.send_email(
email_type=EmailType.EMAIL_CODE_LOGIN,
language_code=language,
to=to,
template_context={
"to": to,
"code": code,
},
)
end_at = time.perf_counter()
logging.info(

11
api/tasks/mail_enterprise_task.py
View File

@@ -1,15 +1,17 @@
import logging
import time
from collections.abc import Mapping
import click
from celery import shared_task # type: ignore
from flask import render_template_string
from extensions.ext_mail import mail
from libs.email_i18n import get_email_i18n_service
@shared_task(queue="mail")
def send_enterprise_email_task(to, subject, body, substitutions):
def send_enterprise_email_task(to: list[str], subject: str, body: str, substitutions: Mapping[str, str]):
if not mail.is_inited():
return
@@ -19,11 +21,8 @@ def send_enterprise_email_task(to, subject, body, substitutions):
try:
html_content = render_template_string(body, **substitutions)
if isinstance(to, list):
for t in to:
mail.send(to=t, subject=subject, html=html_content)
else:
mail.send(to=to, subject=subject, html=html_content)
email_service = get_email_i18n_service()
email_service.send_raw_email(to=to, subject=subject, html_content=html_content)
end_at = time.perf_counter()
logging.info(

72
api/tasks/mail_invite_member_task.py
View File

@@ -3,24 +3,23 @@ import time
import click
from celery import shared_task # type: ignore
from flask import render_template
from configs import dify_config
from extensions.ext_mail import mail
from services.feature_service import FeatureService
from libs.email_i18n import EmailType, get_email_i18n_service
@shared_task(queue="mail")
def send_invite_member_mail_task(language: str, to: str, token: str, inviter_name: str, workspace_name: str):
def send_invite_member_mail_task(language: str, to: str, token: str, inviter_name: str, workspace_name: str) -> None:
"""
Async Send invite member mail
:param language
:param to
:param token
:param inviter_name
:param workspace_name
Send invite member email with internationalization support.
Usage: send_invite_member_mail_task.delay(language, to, token, inviter_name, workspace_name)
Args:
language: Language code for email localization
to: Recipient email address
token: Invitation token
inviter_name: Name of the person sending the invitation
workspace_name: Name of the workspace
"""
if not mail.is_inited():
return
@@ -30,49 +29,20 @@ def send_invite_member_mail_task(language: str, to: str, token: str, inviter_nam
)
start_at = time.perf_counter()
# send invite member mail using different languages
try:
url = f"{dify_config.CONSOLE_WEB_URL}/activate?token={token}"
if language == "zh-Hans":
template = "invite_member_mail_template_zh-CN.html"
system_features = FeatureService.get_system_features()
if system_features.branding.enabled:
application_title = system_features.branding.application_title
template = "without-brand/invite_member_mail_template_zh-CN.html"
html_content = render_template(
template,
to=to,
inviter_name=inviter_name,
workspace_name=workspace_name,
url=url,
application_title=application_title,
)
mail.send(to=to, subject=f"立即加入 {application_title} 工作空间", html=html_content)
else:
html_content = render_template(
template, to=to, inviter_name=inviter_name, workspace_name=workspace_name, url=url
)
mail.send(to=to, subject="立即加入 Dify 工作空间", html=html_content)
else:
template = "invite_member_mail_template_en-US.html"
system_features = FeatureService.get_system_features()
if system_features.branding.enabled:
application_title = system_features.branding.application_title
template = "without-brand/invite_member_mail_template_en-US.html"
html_content = render_template(
template,
to=to,
inviter_name=inviter_name,
workspace_name=workspace_name,
url=url,
application_title=application_title,
)
mail.send(to=to, subject=f"Join {application_title} Workspace Now", html=html_content)
else:
html_content = render_template(
template, to=to, inviter_name=inviter_name, workspace_name=workspace_name, url=url
)
mail.send(to=to, subject="Join Dify Workspace Now", html=html_content)
email_service = get_email_i18n_service()
email_service.send_email(
email_type=EmailType.INVITE_MEMBER,
language_code=language,
to=to,
template_context={
"to": to,
"inviter_name": inviter_name,
"workspace_name": workspace_name,
"url": url,
},
)
end_at = time.perf_counter()
logging.info(

155
api/tasks/mail_owner_transfer_task.py
View File

@@ -3,47 +3,40 @@ import time
import click
from celery import shared_task # type: ignore
from flask import render_template
from extensions.ext_mail import mail
from services.feature_service import FeatureService
from libs.email_i18n import EmailType, get_email_i18n_service
@shared_task(queue="mail")
def send_owner_transfer_confirm_task(language: str, to: str, code: str, workspace: str):
def send_owner_transfer_confirm_task(language: str, to: str, code: str, workspace: str) -> None:
"""
Async Send owner transfer confirm mail
:param language: Language in which the email should be sent (e.g., 'en', 'zh')
:param to: Recipient email address
:param workspace: Workspace name
Send owner transfer confirmation email with internationalization support.
Args:
language: Language code for email localization
to: Recipient email address
code: Verification code
workspace: Workspace name
"""
if not mail.is_inited():
return
logging.info(click.style("Start change email mail to {}".format(to), fg="green"))
logging.info(click.style("Start owner transfer confirm mail to {}".format(to), fg="green"))
start_at = time.perf_counter()
# send change email mail using different languages
try:
if language == "zh-Hans":
template = "transfer_workspace_owner_confirm_template_zh-CN.html"
system_features = FeatureService.get_system_features()
if system_features.branding.enabled:
template = "without-brand/transfer_workspace_owner_confirm_template_zh-CN.html"
html_content = render_template(template, to=to, code=code, WorkspaceName=workspace)
mail.send(to=to, subject="验证您转移工作空间所有权的请求", html=html_content)
else:
html_content = render_template(template, to=to, code=code, WorkspaceName=workspace)
mail.send(to=to, subject="验证您转移工作空间所有权的请求", html=html_content)
else:
template = "transfer_workspace_owner_confirm_template_en-US.html"
system_features = FeatureService.get_system_features()
if system_features.branding.enabled:
template = "without-brand/transfer_workspace_owner_confirm_template_en-US.html"
html_content = render_template(template, to=to, code=code, WorkspaceName=workspace)
mail.send(to=to, subject="Verify Your Request to Transfer Workspace Ownership", html=html_content)
else:
html_content = render_template(template, to=to, code=code, WorkspaceName=workspace)
mail.send(to=to, subject="Verify Your Request to Transfer Workspace Ownership", html=html_content)
email_service = get_email_i18n_service()
email_service.send_email(
email_type=EmailType.OWNER_TRANSFER_CONFIRM,
language_code=language,
to=to,
template_context={
"to": to,
"code": code,
"WorkspaceName": workspace,
},
)
end_at = time.perf_counter()
logging.info(
@@ -57,96 +50,80 @@ def send_owner_transfer_confirm_task(language: str, to: str, code: str, workspac
@shared_task(queue="mail")
def send_old_owner_transfer_notify_email_task(language: str, to: str, workspace: str, new_owner_email: str):
def send_old_owner_transfer_notify_email_task(language: str, to: str, workspace: str, new_owner_email: str) -> None:
"""
Async Send owner transfer confirm mail
:param language: Language in which the email should be sent (e.g., 'en', 'zh')
:param to: Recipient email address
:param workspace: Workspace name
:param new_owner_email: New owner email
Send old owner transfer notification email with internationalization support.
Args:
language: Language code for email localization
to: Recipient email address
workspace: Workspace name
new_owner_email: New owner email address
"""
if not mail.is_inited():
return
logging.info(click.style("Start change email mail to {}".format(to), fg="green"))
logging.info(click.style("Start old owner transfer notify mail to {}".format(to), fg="green"))
start_at = time.perf_counter()
# send change email mail using different languages
try:
if language == "zh-Hans":
template = "transfer_workspace_old_owner_notify_template_zh-CN.html"
system_features = FeatureService.get_system_features()
if system_features.branding.enabled:
template = "without-brand/transfer_workspace_old_owner_notify_template_zh-CN.html"
html_content = render_template(template, to=to, WorkspaceName=workspace, NewOwnerEmail=new_owner_email)
mail.send(to=to, subject="工作区所有权已转移", html=html_content)
else:
html_content = render_template(template, to=to, WorkspaceName=workspace, NewOwnerEmail=new_owner_email)
mail.send(to=to, subject="工作区所有权已转移", html=html_content)
else:
template = "transfer_workspace_old_owner_notify_template_en-US.html"
system_features = FeatureService.get_system_features()
if system_features.branding.enabled:
template = "without-brand/transfer_workspace_old_owner_notify_template_en-US.html"
html_content = render_template(template, to=to, WorkspaceName=workspace, NewOwnerEmail=new_owner_email)
mail.send(to=to, subject="Workspace ownership has been transferred", html=html_content)
else:
html_content = render_template(template, to=to, WorkspaceName=workspace, NewOwnerEmail=new_owner_email)
mail.send(to=to, subject="Workspace ownership has been transferred", html=html_content)
email_service = get_email_i18n_service()
email_service.send_email(
email_type=EmailType.OWNER_TRANSFER_OLD_NOTIFY,
language_code=language,
to=to,
template_context={
"to": to,
"WorkspaceName": workspace,
"NewOwnerEmail": new_owner_email,
},
)
end_at = time.perf_counter()
logging.info(
click.style(
"Send owner transfer confirm mail to {} succeeded: latency: {}".format(to, end_at - start_at),
"Send old owner transfer notify mail to {} succeeded: latency: {}".format(to, end_at - start_at),
fg="green",
)
)
except Exception:
logging.exception("owner transfer confirm email mail to {} failed".format(to))
logging.exception("old owner transfer notify email mail to {} failed".format(to))
@shared_task(queue="mail")
def send_new_owner_transfer_notify_email_task(language: str, to: str, workspace: str):
def send_new_owner_transfer_notify_email_task(language: str, to: str, workspace: str) -> None:
"""
Async Send owner transfer confirm mail
:param language: Language in which the email should be sent (e.g., 'en', 'zh')
:param to: Recipient email address
:param code: Change email code
:param workspace: Workspace name
Send new owner transfer notification email with internationalization support.
Args:
language: Language code for email localization
to: Recipient email address
workspace: Workspace name
"""
if not mail.is_inited():
return
logging.info(click.style("Start change email mail to {}".format(to), fg="green"))
logging.info(click.style("Start new owner transfer notify mail to {}".format(to), fg="green"))
start_at = time.perf_counter()
# send change email mail using different languages
try:
if language == "zh-Hans":
template = "transfer_workspace_new_owner_notify_template_zh-CN.html"
system_features = FeatureService.get_system_features()
if system_features.branding.enabled:
template = "without-brand/transfer_workspace_new_owner_notify_template_zh-CN.html"
html_content = render_template(template, to=to, WorkspaceName=workspace)
mail.send(to=to, subject=f"您现在是 {workspace} 的所有者", html=html_content)
else:
html_content = render_template(template, to=to, WorkspaceName=workspace)
mail.send(to=to, subject=f"您现在是 {workspace} 的所有者", html=html_content)
else:
template = "transfer_workspace_new_owner_notify_template_en-US.html"
system_features = FeatureService.get_system_features()
if system_features.branding.enabled:
template = "without-brand/transfer_workspace_new_owner_notify_template_en-US.html"
html_content = render_template(template, to=to, WorkspaceName=workspace)
mail.send(to=to, subject=f"You are now the owner of {workspace}", html=html_content)
else:
html_content = render_template(template, to=to, WorkspaceName=workspace)
mail.send(to=to, subject=f"You are now the owner of {workspace}", html=html_content)
email_service = get_email_i18n_service()
email_service.send_email(
email_type=EmailType.OWNER_TRANSFER_NEW_NOTIFY,
language_code=language,
to=to,
template_context={
"to": to,
"WorkspaceName": workspace,
},
)
end_at = time.perf_counter()
logging.info(
click.style(
"Send owner transfer confirm mail to {} succeeded: latency: {}".format(to, end_at - start_at),
"Send new owner transfer notify mail to {} succeeded: latency: {}".format(to, end_at - start_at),
fg="green",
)
)
except Exception:
logging.exception("owner transfer confirm email mail to {} failed".format(to))
logging.exception("new owner transfer notify email mail to {} failed".format(to))

48
api/tasks/mail_reset_password_task.py
View File

@@ -3,19 +3,20 @@ import time
import click
from celery import shared_task # type: ignore
from flask import render_template
from extensions.ext_mail import mail
from services.feature_service import FeatureService
from libs.email_i18n import EmailType, get_email_i18n_service
@shared_task(queue="mail")
def send_reset_password_mail_task(language: str, to: str, code: str):
def send_reset_password_mail_task(language: str, to: str, code: str) -> None:
"""
Async Send reset password mail
:param language: Language in which the email should be sent (e.g., 'en', 'zh')
:param to: Recipient email address
:param code: Reset password code
Send reset password email with internationalization support.
Args:
language: Language code for email localization
to: Recipient email address
code: Reset password code
"""
if not mail.is_inited():
return
@@ -23,30 +24,17 @@ def send_reset_password_mail_task(language: str, to: str, code: str):
logging.info(click.style("Start password reset mail to {}".format(to), fg="green"))
start_at = time.perf_counter()
# send reset password mail using different languages
try:
if language == "zh-Hans":
template = "reset_password_mail_template_zh-CN.html"
system_features = FeatureService.get_system_features()
if system_features.branding.enabled:
application_title = system_features.branding.application_title
template = "without-brand/reset_password_mail_template_zh-CN.html"
html_content = render_template(template, to=to, code=code, application_title=application_title)
mail.send(to=to, subject=f"设置您的 {application_title} 密码", html=html_content)
else:
html_content = render_template(template, to=to, code=code)
mail.send(to=to, subject="设置您的 Dify 密码", html=html_content)
else:
template = "reset_password_mail_template_en-US.html"
system_features = FeatureService.get_system_features()
if system_features.branding.enabled:
application_title = system_features.branding.application_title
template = "without-brand/reset_password_mail_template_en-US.html"
html_content = render_template(template, to=to, code=code, application_title=application_title)
mail.send(to=to, subject=f"Set Your {application_title} Password", html=html_content)
else:
html_content = render_template(template, to=to, code=code)
mail.send(to=to, subject="Set Your Dify Password", html=html_content)
email_service = get_email_i18n_service()
email_service.send_email(
email_type=EmailType.RESET_PASSWORD,
language_code=language,
to=to,
template_context={
"to": to,
"code": code,
},
)
end_at = time.perf_counter()
logging.info(

539
api/tests/unit_tests/libs/test_email_i18n.py Normal file
View File

@@ -0,0 +1,539 @@
"""
Unit tests for EmailI18nService
Tests the email internationalization service with mocked dependencies
following Domain-Driven Design principles.
"""
from typing import Any
from unittest.mock import MagicMock
import pytest
from libs.email_i18n import (
EmailI18nConfig,
EmailI18nService,
EmailLanguage,
EmailTemplate,
EmailType,
FlaskEmailRenderer,
FlaskMailSender,
create_default_email_config,
get_email_i18n_service,
)
from services.feature_service import BrandingModel
class MockEmailRenderer:
"""Mock implementation of EmailRenderer protocol"""
def __init__(self) -> None:
self.rendered_templates: list[tuple[str, dict[str, Any]]] = []
def render_template(self, template_path: str, **context: Any) -> str:
"""Mock render_template that returns a formatted string"""
self.rendered_templates.append((template_path, context))
return f"<html>Rendered {template_path} with {context}</html>"
class MockBrandingService:
"""Mock implementation of BrandingService protocol"""
def __init__(self, enabled: bool = False, application_title: str = "Dify") -> None:
self.enabled = enabled
self.application_title = application_title
def get_branding_config(self) -> BrandingModel:
"""Return mock branding configuration"""
branding_model = MagicMock(spec=BrandingModel)
branding_model.enabled = self.enabled
branding_model.application_title = self.application_title
return branding_model
class MockEmailSender:
"""Mock implementation of EmailSender protocol"""
def __init__(self) -> None:
self.sent_emails: list[dict[str, str]] = []
def send_email(self, to: str, subject: str, html_content: str) -> None:
"""Mock send_email that records sent emails"""
self.sent_emails.append(
{
"to": to,
"subject": subject,
"html_content": html_content,
}
)
class TestEmailI18nService:
"""Test cases for EmailI18nService"""
@pytest.fixture
def email_config(self) -> EmailI18nConfig:
"""Create test email configuration"""
return EmailI18nConfig(
templates={
EmailType.RESET_PASSWORD: {
EmailLanguage.EN_US: EmailTemplate(
subject="Reset Your {application_title} Password",
template_path="reset_password_en.html",
branded_template_path="branded/reset_password_en.html",
),
EmailLanguage.ZH_HANS: EmailTemplate(
subject="重置您的 {application_title} 密码",
template_path="reset_password_zh.html",
branded_template_path="branded/reset_password_zh.html",
),
},
EmailType.INVITE_MEMBER: {
EmailLanguage.EN_US: EmailTemplate(
subject="Join {application_title} Workspace",
template_path="invite_member_en.html",
branded_template_path="branded/invite_member_en.html",
),
},
}
)
@pytest.fixture
def mock_renderer(self) -> MockEmailRenderer:
"""Create mock email renderer"""
return MockEmailRenderer()
@pytest.fixture
def mock_branding_service(self) -> MockBrandingService:
"""Create mock branding service"""
return MockBrandingService()
@pytest.fixture
def mock_sender(self) -> MockEmailSender:
"""Create mock email sender"""
return MockEmailSender()
@pytest.fixture
def email_service(
self,
email_config: EmailI18nConfig,
mock_renderer: MockEmailRenderer,
mock_branding_service: MockBrandingService,
mock_sender: MockEmailSender,
) -> EmailI18nService:
"""Create EmailI18nService with mocked dependencies"""
return EmailI18nService(
config=email_config,
renderer=mock_renderer,
branding_service=mock_branding_service,
sender=mock_sender,
)
def test_send_email_with_english_language(
self,
email_service: EmailI18nService,
mock_renderer: MockEmailRenderer,
mock_sender: MockEmailSender,
) -> None:
"""Test sending email with English language"""
email_service.send_email(
email_type=EmailType.RESET_PASSWORD,
language_code="en-US",
to="test@example.com",
template_context={"reset_link": "https://example.com/reset"},
)
# Verify renderer was called with correct template
assert len(mock_renderer.rendered_templates) == 1
template_path, context = mock_renderer.rendered_templates[0]
assert template_path == "reset_password_en.html"
assert context["reset_link"] == "https://example.com/reset"
assert context["branding_enabled"] is False
assert context["application_title"] == "Dify"
# Verify email was sent
assert len(mock_sender.sent_emails) == 1
sent_email = mock_sender.sent_emails[0]
assert sent_email["to"] == "test@example.com"
assert sent_email["subject"] == "Reset Your Dify Password"
assert "reset_password_en.html" in sent_email["html_content"]
def test_send_email_with_chinese_language(
self,
email_service: EmailI18nService,
mock_sender: MockEmailSender,
) -> None:
"""Test sending email with Chinese language"""
email_service.send_email(
email_type=EmailType.RESET_PASSWORD,
language_code="zh-Hans",
to="test@example.com",
template_context={"reset_link": "https://example.com/reset"},
)
# Verify email was sent with Chinese subject
assert len(mock_sender.sent_emails) == 1
sent_email = mock_sender.sent_emails[0]
assert sent_email["subject"] == "重置您的 Dify 密码"
def test_send_email_with_branding_enabled(
self,
email_config: EmailI18nConfig,
mock_renderer: MockEmailRenderer,
mock_sender: MockEmailSender,
) -> None:
"""Test sending email with branding enabled"""
# Create branding service with branding enabled
branding_service = MockBrandingService(enabled=True, application_title="MyApp")
email_service = EmailI18nService(
config=email_config,
renderer=mock_renderer,
branding_service=branding_service,
sender=mock_sender,
)
email_service.send_email(
email_type=EmailType.RESET_PASSWORD,
language_code="en-US",
to="test@example.com",
)
# Verify branded template was used
assert len(mock_renderer.rendered_templates) == 1
template_path, context = mock_renderer.rendered_templates[0]
assert template_path == "branded/reset_password_en.html"
assert context["branding_enabled"] is True
assert context["application_title"] == "MyApp"
# Verify subject includes custom application title
assert len(mock_sender.sent_emails) == 1
sent_email = mock_sender.sent_emails[0]
assert sent_email["subject"] == "Reset Your MyApp Password"
def test_send_email_with_language_fallback(
self,
email_service: EmailI18nService,
mock_sender: MockEmailSender,
) -> None:
"""Test language fallback to English when requested language not available"""
# Request invite member in Chinese (not configured)
email_service.send_email(
email_type=EmailType.INVITE_MEMBER,
language_code="zh-Hans",
to="test@example.com",
)
# Should fall back to English
assert len(mock_sender.sent_emails) == 1
sent_email = mock_sender.sent_emails[0]
assert sent_email["subject"] == "Join Dify Workspace"
def test_send_email_with_unknown_language_code(
self,
email_service: EmailI18nService,
mock_sender: MockEmailSender,
) -> None:
"""Test unknown language code falls back to English"""
email_service.send_email(
email_type=EmailType.RESET_PASSWORD,
language_code="fr-FR", # French not configured
to="test@example.com",
)
# Should use English
assert len(mock_sender.sent_emails) == 1
sent_email = mock_sender.sent_emails[0]
assert sent_email["subject"] == "Reset Your Dify Password"
def test_send_change_email_old_phase(
self,
email_config: EmailI18nConfig,
mock_renderer: MockEmailRenderer,
mock_sender: MockEmailSender,
mock_branding_service: MockBrandingService,
) -> None:
"""Test sending change email for old email verification"""
# Add change email templates to config
email_config.templates[EmailType.CHANGE_EMAIL_OLD] = {
EmailLanguage.EN_US: EmailTemplate(
subject="Verify your current email",
template_path="change_email_old_en.html",
branded_template_path="branded/change_email_old_en.html",
),
}
email_service = EmailI18nService(
config=email_config,
renderer=mock_renderer,
branding_service=mock_branding_service,
sender=mock_sender,
)
email_service.send_change_email(
language_code="en-US",
to="old@example.com",
code="123456",
phase="old_email",
)
# Verify correct template and context
assert len(mock_renderer.rendered_templates) == 1
template_path, context = mock_renderer.rendered_templates[0]
assert template_path == "change_email_old_en.html"
assert context["to"] == "old@example.com"
assert context["code"] == "123456"
def test_send_change_email_new_phase(
self,
email_config: EmailI18nConfig,
mock_renderer: MockEmailRenderer,
mock_sender: MockEmailSender,
mock_branding_service: MockBrandingService,
) -> None:
"""Test sending change email for new email verification"""
# Add change email templates to config
email_config.templates[EmailType.CHANGE_EMAIL_NEW] = {
EmailLanguage.EN_US: EmailTemplate(
subject="Verify your new email",
template_path="change_email_new_en.html",
branded_template_path="branded/change_email_new_en.html",
),
}
email_service = EmailI18nService(
config=email_config,
renderer=mock_renderer,
branding_service=mock_branding_service,
sender=mock_sender,
)
email_service.send_change_email(
language_code="en-US",
to="new@example.com",
code="654321",
phase="new_email",
)
# Verify correct template and context
assert len(mock_renderer.rendered_templates) == 1
template_path, context = mock_renderer.rendered_templates[0]
assert template_path == "change_email_new_en.html"
assert context["to"] == "new@example.com"
assert context["code"] == "654321"
def test_send_change_email_invalid_phase(
self,
email_service: EmailI18nService,
) -> None:
"""Test sending change email with invalid phase raises error"""
with pytest.raises(ValueError, match="Invalid phase: invalid_phase"):
email_service.send_change_email(
language_code="en-US",
to="test@example.com",
code="123456",
phase="invalid_phase",
)
def test_send_raw_email_single_recipient(
self,
email_service: EmailI18nService,
mock_sender: MockEmailSender,
) -> None:
"""Test sending raw email to single recipient"""
email_service.send_raw_email(
to="test@example.com",
subject="Test Subject",
html_content="<html>Test Content</html>",
)
assert len(mock_sender.sent_emails) == 1
sent_email = mock_sender.sent_emails[0]
assert sent_email["to"] == "test@example.com"
assert sent_email["subject"] == "Test Subject"
assert sent_email["html_content"] == "<html>Test Content</html>"
def test_send_raw_email_multiple_recipients(
self,
email_service: EmailI18nService,
mock_sender: MockEmailSender,
) -> None:
"""Test sending raw email to multiple recipients"""
recipients = ["user1@example.com", "user2@example.com", "user3@example.com"]
email_service.send_raw_email(
to=recipients,
subject="Test Subject",
html_content="<html>Test Content</html>",
)
# Should send individual emails to each recipient
assert len(mock_sender.sent_emails) == 3
for i, recipient in enumerate(recipients):
sent_email = mock_sender.sent_emails[i]
assert sent_email["to"] == recipient
assert sent_email["subject"] == "Test Subject"
assert sent_email["html_content"] == "<html>Test Content</html>"
def test_get_template_missing_email_type(
self,
email_config: EmailI18nConfig,
) -> None:
"""Test getting template for missing email type raises error"""
with pytest.raises(ValueError, match="No templates configured for email type"):
email_config.get_template(EmailType.EMAIL_CODE_LOGIN, EmailLanguage.EN_US)
def test_get_template_missing_language_and_english(
self,
email_config: EmailI18nConfig,
) -> None:
"""Test error when neither requested language nor English fallback exists"""
# Add template without English fallback
email_config.templates[EmailType.EMAIL_CODE_LOGIN] = {
EmailLanguage.ZH_HANS: EmailTemplate(
subject="Test",
template_path="test.html",
branded_template_path="branded/test.html",
),
}
with pytest.raises(ValueError, match="No template found for"):
# Request a language that doesn't exist and no English fallback
email_config.get_template(EmailType.EMAIL_CODE_LOGIN, EmailLanguage.EN_US)
def test_subject_templating_with_variables(
self,
email_config: EmailI18nConfig,
mock_renderer: MockEmailRenderer,
mock_sender: MockEmailSender,
mock_branding_service: MockBrandingService,
) -> None:
"""Test subject templating with custom variables"""
# Add template with variable in subject
email_config.templates[EmailType.OWNER_TRANSFER_NEW_NOTIFY] = {
EmailLanguage.EN_US: EmailTemplate(
subject="You are now the owner of {WorkspaceName}",
template_path="owner_transfer_en.html",
branded_template_path="branded/owner_transfer_en.html",
),
}
email_service = EmailI18nService(
config=email_config,
renderer=mock_renderer,
branding_service=mock_branding_service,
sender=mock_sender,
)
email_service.send_email(
email_type=EmailType.OWNER_TRANSFER_NEW_NOTIFY,
language_code="en-US",
to="test@example.com",
template_context={"WorkspaceName": "My Workspace"},
)
# Verify subject was templated correctly
assert len(mock_sender.sent_emails) == 1
sent_email = mock_sender.sent_emails[0]
assert sent_email["subject"] == "You are now the owner of My Workspace"
def test_email_language_from_language_code(self) -> None:
"""Test EmailLanguage.from_language_code method"""
assert EmailLanguage.from_language_code("zh-Hans") == EmailLanguage.ZH_HANS
assert EmailLanguage.from_language_code("en-US") == EmailLanguage.EN_US
assert EmailLanguage.from_language_code("fr-FR") == EmailLanguage.EN_US # Fallback
assert EmailLanguage.from_language_code("unknown") == EmailLanguage.EN_US # Fallback
class TestEmailI18nIntegration:
"""Integration tests for email i18n components"""
def test_create_default_email_config(self) -> None:
"""Test creating default email configuration"""
config = create_default_email_config()
# Verify key email types have at least English template
expected_types = [
EmailType.RESET_PASSWORD,
EmailType.INVITE_MEMBER,
EmailType.EMAIL_CODE_LOGIN,
EmailType.CHANGE_EMAIL_OLD,
EmailType.CHANGE_EMAIL_NEW,
EmailType.OWNER_TRANSFER_CONFIRM,
EmailType.OWNER_TRANSFER_OLD_NOTIFY,
EmailType.OWNER_TRANSFER_NEW_NOTIFY,
EmailType.ACCOUNT_DELETION_SUCCESS,
EmailType.ACCOUNT_DELETION_VERIFICATION,
EmailType.QUEUE_MONITOR_ALERT,
EmailType.DOCUMENT_CLEAN_NOTIFY,
]
for email_type in expected_types:
assert email_type in config.templates
assert EmailLanguage.EN_US in config.templates[email_type]
# Verify some have Chinese translations
assert EmailLanguage.ZH_HANS in config.templates[EmailType.RESET_PASSWORD]
assert EmailLanguage.ZH_HANS in config.templates[EmailType.INVITE_MEMBER]
def test_get_email_i18n_service(self) -> None:
"""Test getting global email i18n service instance"""
service1 = get_email_i18n_service()
service2 = get_email_i18n_service()
# Should return the same instance
assert service1 is service2
def test_flask_email_renderer(self) -> None:
"""Test FlaskEmailRenderer implementation"""
renderer = FlaskEmailRenderer()
# Should raise TemplateNotFound when template doesn't exist
from jinja2.exceptions import TemplateNotFound
with pytest.raises(TemplateNotFound):
renderer.render_template("test.html", foo="bar")
def test_flask_mail_sender_not_initialized(self) -> None:
"""Test FlaskMailSender when mail is not initialized"""
sender = FlaskMailSender()
# Mock mail.is_inited() to return False
import libs.email_i18n
original_mail = libs.email_i18n.mail
mock_mail = MagicMock()
mock_mail.is_inited.return_value = False
libs.email_i18n.mail = mock_mail
try:
# Should not send email when mail is not initialized
sender.send_email("test@example.com", "Subject", "<html>Content</html>")
mock_mail.send.assert_not_called()
finally:
# Restore original mail
libs.email_i18n.mail = original_mail
def test_flask_mail_sender_initialized(self) -> None:
"""Test FlaskMailSender when mail is initialized"""
sender = FlaskMailSender()
# Mock mail.is_inited() to return True
import libs.email_i18n
original_mail = libs.email_i18n.mail
mock_mail = MagicMock()
mock_mail.is_inited.return_value = True
libs.email_i18n.mail = mock_mail
try:
# Should send email when mail is initialized
sender.send_email("test@example.com", "Subject", "<html>Content</html>")
mock_mail.send.assert_called_once_with(
to="test@example.com",
subject="Subject",
html="<html>Content</html>",
)
finally:
# Restore original mail
libs.email_i18n.mail = original_mail