From 0f4809b9b848f417840d48a29b4d1350cee5a98d Mon Sep 17 00:00:00 2001 From: -LAN- Date: Wed, 23 Jul 2025 00:26:00 +0800 Subject: [PATCH] refactor: centralize email internationalization handling (#22752) Co-authored-by: Claude --- api/libs/email_i18n.py | 461 +++++++++++++++ .../mail_clean_document_notify_task.py | 20 +- api/schedule/queue_monitor_task.py | 23 +- api/tasks/mail_account_deletion_task.py | 46 +- api/tasks/mail_change_mail_task.py | 68 +-- api/tasks/mail_email_code_login.py | 46 +- api/tasks/mail_enterprise_task.py | 11 +- api/tasks/mail_invite_member_task.py | 72 +-- api/tasks/mail_owner_transfer_task.py | 155 +++-- api/tasks/mail_reset_password_task.py | 48 +- api/tests/unit_tests/libs/test_email_i18n.py | 539 ++++++++++++++++++ 11 files changed, 1200 insertions(+), 289 deletions(-) create mode 100644 api/libs/email_i18n.py create mode 100644 api/tests/unit_tests/libs/test_email_i18n.py diff --git a/api/libs/email_i18n.py b/api/libs/email_i18n.py new file mode 100644 index 000000000..bfbf41a07 --- /dev/null +++ b/api/libs/email_i18n.py @@ -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 diff --git a/api/schedule/mail_clean_document_notify_task.py b/api/schedule/mail_clean_document_notify_task.py index 5ee813e1d..12e4f6ebf 100644 --- a/api/schedule/mail_clean_document_notify_task.py +++ b/api/schedule/mail_clean_document_notify_task.py @@ -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 diff --git a/api/schedule/queue_monitor_task.py b/api/schedule/queue_monitor_task.py index e3a7021b9..a05e1358e 100644 --- a/api/schedule/queue_monitor_task.py +++ b/api/schedule/queue_monitor_task.py @@ -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")) diff --git a/api/tasks/mail_account_deletion_task.py b/api/tasks/mail_account_deletion_task.py index 0c60ae53d..a6f8ce2f0 100644 --- a/api/tasks/mail_account_deletion_task.py +++ b/api/tasks/mail_account_deletion_task.py @@ -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( diff --git a/api/tasks/mail_change_mail_task.py b/api/tasks/mail_change_mail_task.py index da44040b7..ea1875901 100644 --- a/api/tasks/mail_change_mail_task.py +++ b/api/tasks/mail_change_mail_task.py @@ -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( diff --git a/api/tasks/mail_email_code_login.py b/api/tasks/mail_email_code_login.py index ddad33172..34220784e 100644 --- a/api/tasks/mail_email_code_login.py +++ b/api/tasks/mail_email_code_login.py @@ -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( diff --git a/api/tasks/mail_enterprise_task.py b/api/tasks/mail_enterprise_task.py index b9d8fd55d..a1c290862 100644 --- a/api/tasks/mail_enterprise_task.py +++ b/api/tasks/mail_enterprise_task.py @@ -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( diff --git a/api/tasks/mail_invite_member_task.py b/api/tasks/mail_invite_member_task.py index 7ca85c7f2..8c73de011 100644 --- a/api/tasks/mail_invite_member_task.py +++ b/api/tasks/mail_invite_member_task.py @@ -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( diff --git a/api/tasks/mail_owner_transfer_task.py b/api/tasks/mail_owner_transfer_task.py index 8d05c6dc0..e566a6bc5 100644 --- a/api/tasks/mail_owner_transfer_task.py +++ b/api/tasks/mail_owner_transfer_task.py @@ -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)) diff --git a/api/tasks/mail_reset_password_task.py b/api/tasks/mail_reset_password_task.py index d4f4482a4..e2482f210 100644 --- a/api/tasks/mail_reset_password_task.py +++ b/api/tasks/mail_reset_password_task.py @@ -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( diff --git a/api/tests/unit_tests/libs/test_email_i18n.py b/api/tests/unit_tests/libs/test_email_i18n.py new file mode 100644 index 000000000..aeb30438e --- /dev/null +++ b/api/tests/unit_tests/libs/test_email_i18n.py @@ -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"Rendered {template_path} with {context}" + + +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="Test Content", + ) + + 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"] == "Test Content" + + 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="Test Content", + ) + + # 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"] == "Test Content" + + 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", "Content") + 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", "Content") + mock_mail.send.assert_called_once_with( + to="test@example.com", + subject="Subject", + html="Content", + ) + finally: + # Restore original mail + libs.email_i18n.mail = original_mail