youngkingdom/dify
1
0
Fork 0
mirror of https://github.com/langgenius/dify.git synced 2026-05-03 00:48:04 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/main' into feat/trigger

Browse Source
This commit is contained in:
lyzno1
2025-10-07 18:25:43 +08:00
parent 04d94e3337 11f7a89e25
commit 02222752f0
21 changed files with 786 additions and 209 deletions
Download Patch File Download Diff File Expand all files Collapse all files

282
api/tests/test_containers_integration_tests/tasks/test_mail_change_mail_task.py Normal file
View File

@ -0,0 +1,282 @@
from unittest.mock import MagicMock, patch
import pytest
from faker import Faker
from libs.email_i18n import EmailType
from models.account import Account, Tenant, TenantAccountJoin, TenantAccountRole
from tasks.mail_change_mail_task import send_change_mail_completed_notification_task, send_change_mail_task
class TestMailChangeMailTask:
"""Integration tests for mail_change_mail_task using testcontainers."""
@pytest.fixture
def mock_external_service_dependencies(self):
"""Mock setup for external service dependencies."""
with (
patch("tasks.mail_change_mail_task.mail") as mock_mail,
patch("tasks.mail_change_mail_task.get_email_i18n_service") as mock_get_email_i18n_service,
):
# Setup mock mail service
mock_mail.is_inited.return_value = True
# Setup mock email i18n service
mock_email_service = MagicMock()
mock_get_email_i18n_service.return_value = mock_email_service
yield {
"mail": mock_mail,
"email_i18n_service": mock_email_service,
"get_email_i18n_service": mock_get_email_i18n_service,
}
def _create_test_account(self, db_session_with_containers):
"""
Helper method to create a test account for testing.
Args:
db_session_with_containers: Database session from testcontainers infrastructure
Returns:
Account: Created account instance
"""
fake = Faker()
# Create account
account = Account(
email=fake.email(),
name=fake.name(),
interface_language="en-US",
status="active",
)
db_session_with_containers.add(account)
db_session_with_containers.commit()
# Create tenant
tenant = Tenant(
name=fake.company(),
status="normal",
)
db_session_with_containers.add(tenant)
db_session_with_containers.commit()
# Create tenant-account join
join = TenantAccountJoin(
tenant_id=tenant.id,
account_id=account.id,
role=TenantAccountRole.OWNER.value,
current=True,
)
db_session_with_containers.add(join)
db_session_with_containers.commit()
return account
def test_send_change_mail_task_success_old_email_phase(
self, db_session_with_containers, mock_external_service_dependencies
):
"""
Test successful change email task execution for old_email phase.
This test verifies:
- Proper mail service initialization check
- Correct email service method call with old_email phase
- Successful task completion
"""
# Arrange: Create test data
account = self._create_test_account(db_session_with_containers)
test_language = "en-US"
test_email = account.email
test_code = "123456"
test_phase = "old_email"
# Act: Execute the task
send_change_mail_task(test_language, test_email, test_code, test_phase)
# Assert: Verify the expected outcomes
mock_external_service_dependencies["mail"].is_inited.assert_called_once()
mock_external_service_dependencies["get_email_i18n_service"].assert_called_once()
mock_external_service_dependencies["email_i18n_service"].send_change_email.assert_called_once_with(
language_code=test_language,
to=test_email,
code=test_code,
phase=test_phase,
)
def test_send_change_mail_task_success_new_email_phase(
self, db_session_with_containers, mock_external_service_dependencies
):
"""
Test successful change email task execution for new_email phase.
This test verifies:
- Proper mail service initialization check
- Correct email service method call with new_email phase
- Successful task completion
"""
# Arrange: Create test data
account = self._create_test_account(db_session_with_containers)
test_language = "zh-Hans"
test_email = "new@example.com"
test_code = "789012"
test_phase = "new_email"
# Act: Execute the task
send_change_mail_task(test_language, test_email, test_code, test_phase)
# Assert: Verify the expected outcomes
mock_external_service_dependencies["mail"].is_inited.assert_called_once()
mock_external_service_dependencies["get_email_i18n_service"].assert_called_once()
mock_external_service_dependencies["email_i18n_service"].send_change_email.assert_called_once_with(
language_code=test_language,
to=test_email,
code=test_code,
phase=test_phase,
)
def test_send_change_mail_task_mail_not_initialized(
self, db_session_with_containers, mock_external_service_dependencies
):
"""
Test change email task when mail service is not initialized.
This test verifies:
- Early return when mail service is not initialized
- No email service calls when mail is not available
"""
# Arrange: Setup mail service as not initialized
mock_external_service_dependencies["mail"].is_inited.return_value = False
test_language = "en-US"
test_email = "test@example.com"
test_code = "123456"
test_phase = "old_email"
# Act: Execute the task
send_change_mail_task(test_language, test_email, test_code, test_phase)
# Assert: Verify no email service calls
mock_external_service_dependencies["mail"].is_inited.assert_called_once()
mock_external_service_dependencies["get_email_i18n_service"].assert_not_called()
mock_external_service_dependencies["email_i18n_service"].send_change_email.assert_not_called()
def test_send_change_mail_task_email_service_exception(
self, db_session_with_containers, mock_external_service_dependencies
):
"""
Test change email task when email service raises an exception.
This test verifies:
- Exception is properly caught and logged
- Task completes without raising exception
"""
# Arrange: Setup email service to raise exception
mock_external_service_dependencies["email_i18n_service"].send_change_email.side_effect = Exception(
"Email service failed"
)
test_language = "en-US"
test_email = "test@example.com"
test_code = "123456"
test_phase = "old_email"
# Act: Execute the task (should not raise exception)
send_change_mail_task(test_language, test_email, test_code, test_phase)
# Assert: Verify email service was called despite exception
mock_external_service_dependencies["mail"].is_inited.assert_called_once()
mock_external_service_dependencies["get_email_i18n_service"].assert_called_once()
mock_external_service_dependencies["email_i18n_service"].send_change_email.assert_called_once_with(
language_code=test_language,
to=test_email,
code=test_code,
phase=test_phase,
)
def test_send_change_mail_completed_notification_task_success(
self, db_session_with_containers, mock_external_service_dependencies
):
"""
Test successful change email completed notification task execution.
This test verifies:
- Proper mail service initialization check
- Correct email service method call with CHANGE_EMAIL_COMPLETED type
- Template context is properly constructed
- Successful task completion
"""
# Arrange: Create test data
account = self._create_test_account(db_session_with_containers)
test_language = "en-US"
test_email = account.email
# Act: Execute the task
send_change_mail_completed_notification_task(test_language, test_email)
# Assert: Verify the expected outcomes
mock_external_service_dependencies["mail"].is_inited.assert_called_once()
mock_external_service_dependencies["get_email_i18n_service"].assert_called_once()
mock_external_service_dependencies["email_i18n_service"].send_email.assert_called_once_with(
email_type=EmailType.CHANGE_EMAIL_COMPLETED,
language_code=test_language,
to=test_email,
template_context={
"to": test_email,
"email": test_email,
},
)
def test_send_change_mail_completed_notification_task_mail_not_initialized(
self, db_session_with_containers, mock_external_service_dependencies
):
"""
Test change email completed notification task when mail service is not initialized.
This test verifies:
- Early return when mail service is not initialized
- No email service calls when mail is not available
"""
# Arrange: Setup mail service as not initialized
mock_external_service_dependencies["mail"].is_inited.return_value = False
test_language = "en-US"
test_email = "test@example.com"
# Act: Execute the task
send_change_mail_completed_notification_task(test_language, test_email)
# Assert: Verify no email service calls
mock_external_service_dependencies["mail"].is_inited.assert_called_once()
mock_external_service_dependencies["get_email_i18n_service"].assert_not_called()
mock_external_service_dependencies["email_i18n_service"].send_email.assert_not_called()
def test_send_change_mail_completed_notification_task_email_service_exception(
self, db_session_with_containers, mock_external_service_dependencies
):
"""
Test change email completed notification task when email service raises an exception.
This test verifies:
- Exception is properly caught and logged
- Task completes without raising exception
"""
# Arrange: Setup email service to raise exception
mock_external_service_dependencies["email_i18n_service"].send_email.side_effect = Exception(
"Email service failed"
)
test_language = "en-US"
test_email = "test@example.com"
# Act: Execute the task (should not raise exception)
send_change_mail_completed_notification_task(test_language, test_email)
# Assert: Verify email service was called despite exception
mock_external_service_dependencies["mail"].is_inited.assert_called_once()
mock_external_service_dependencies["get_email_i18n_service"].assert_called_once()
mock_external_service_dependencies["email_i18n_service"].send_email.assert_called_once_with(
email_type=EmailType.CHANGE_EMAIL_COMPLETED,
language_code=test_language,
to=test_email,
template_context={
"to": test_email,
"email": test_email,
},
)

261
api/tests/test_containers_integration_tests/tasks/test_mail_inner_task.py Normal file
View File

@ -0,0 +1,261 @@
from unittest.mock import MagicMock, patch
import pytest
from faker import Faker
from tasks.mail_inner_task import send_inner_email_task
class TestMailInnerTask:
"""Integration tests for send_inner_email_task using testcontainers."""
@pytest.fixture
def mock_external_service_dependencies(self):
"""Mock setup for external service dependencies."""
with (
patch("tasks.mail_inner_task.mail") as mock_mail,
patch("tasks.mail_inner_task.get_email_i18n_service") as mock_get_email_i18n_service,
patch("tasks.mail_inner_task._render_template_with_strategy") as mock_render_template,
):
# Setup mock mail service
mock_mail.is_inited.return_value = True
# Setup mock email i18n service
mock_email_service = MagicMock()
mock_get_email_i18n_service.return_value = mock_email_service
# Setup mock template rendering
mock_render_template.return_value = "<html>Test email content</html>"
yield {
"mail": mock_mail,
"email_service": mock_email_service,
"render_template": mock_render_template,
}
def _create_test_email_data(self, fake: Faker) -> dict:
"""
Helper method to create test email data for testing.
Args:
fake: Faker instance for generating test data
Returns:
dict: Test email data including recipients, subject, body, and substitutions
"""
return {
"to": [fake.email() for _ in range(3)],
"subject": fake.sentence(nb_words=4),
"body": "Hello {{name}}, this is a test email from {{company}}.",
"substitutions": {
"name": fake.name(),
"company": fake.company(),
"date": fake.date(),
},
}
def test_send_inner_email_success(self, db_session_with_containers, mock_external_service_dependencies):
"""
Test successful email sending with valid data.
This test verifies:
- Proper email service initialization check
- Template rendering with substitutions
- Email service integration
- Multiple recipient handling
"""
# Arrange: Create test data
fake = Faker()
email_data = self._create_test_email_data(fake)
# Act: Execute the task
send_inner_email_task(
to=email_data["to"],
subject=email_data["subject"],
body=email_data["body"],
substitutions=email_data["substitutions"],
)
# Assert: Verify the expected outcomes
# Verify mail service was checked for initialization
mock_external_service_dependencies["mail"].is_inited.assert_called_once()
# Verify template rendering was called with correct parameters
mock_external_service_dependencies["render_template"].assert_called_once_with(
email_data["body"], email_data["substitutions"]
)
# Verify email service was called once with the full recipient list
mock_email_service = mock_external_service_dependencies["email_service"]
mock_email_service.send_raw_email.assert_called_once_with(
to=email_data["to"],
subject=email_data["subject"],
html_content="<html>Test email content</html>",
)
def test_send_inner_email_single_recipient(self, db_session_with_containers, mock_external_service_dependencies):
"""
Test email sending with single recipient.
This test verifies:
- Single recipient handling
- Template rendering
- Email service integration
"""
# Arrange: Create test data with single recipient
fake = Faker()
email_data = {
"to": [fake.email()],
"subject": fake.sentence(nb_words=3),
"body": "Welcome {{user_name}}!",
"substitutions": {
"user_name": fake.name(),
},
}
# Act: Execute the task
send_inner_email_task(
to=email_data["to"],
subject=email_data["subject"],
body=email_data["body"],
substitutions=email_data["substitutions"],
)
# Assert: Verify the expected outcomes
mock_email_service = mock_external_service_dependencies["email_service"]
mock_email_service.send_raw_email.assert_called_once_with(
to=email_data["to"],
subject=email_data["subject"],
html_content="<html>Test email content</html>",
)
def test_send_inner_email_empty_substitutions(self, db_session_with_containers, mock_external_service_dependencies):
"""
Test email sending with empty substitutions.
This test verifies:
- Template rendering with empty substitutions
- Email service integration
- Handling of minimal template context
"""
# Arrange: Create test data with empty substitutions
fake = Faker()
email_data = {
"to": [fake.email()],
"subject": fake.sentence(nb_words=3),
"body": "This is a simple email without variables.",
"substitutions": {},
}
# Act: Execute the task
send_inner_email_task(
to=email_data["to"],
subject=email_data["subject"],
body=email_data["body"],
substitutions=email_data["substitutions"],
)
# Assert: Verify the expected outcomes
mock_external_service_dependencies["render_template"].assert_called_once_with(email_data["body"], {})
mock_email_service = mock_external_service_dependencies["email_service"]
mock_email_service.send_raw_email.assert_called_once_with(
to=email_data["to"],
subject=email_data["subject"],
html_content="<html>Test email content</html>",
)
def test_send_inner_email_mail_not_initialized(
self, db_session_with_containers, mock_external_service_dependencies
):
"""
Test email sending when mail service is not initialized.
This test verifies:
- Early return when mail service is not initialized
- No template rendering occurs
- No email service calls
- No exceptions raised
"""
# Arrange: Setup mail service as not initialized
mock_external_service_dependencies["mail"].is_inited.return_value = False
fake = Faker()
email_data = self._create_test_email_data(fake)
# Act: Execute the task
send_inner_email_task(
to=email_data["to"],
subject=email_data["subject"],
body=email_data["body"],
substitutions=email_data["substitutions"],
)
# Assert: Verify no processing occurred
mock_external_service_dependencies["render_template"].assert_not_called()
mock_external_service_dependencies["email_service"].send_raw_email.assert_not_called()
def test_send_inner_email_template_rendering_error(
self, db_session_with_containers, mock_external_service_dependencies
):
"""
Test email sending when template rendering fails.
This test verifies:
- Exception handling during template rendering
- No email service calls when template fails
"""
# Arrange: Setup template rendering to raise an exception
mock_external_service_dependencies["render_template"].side_effect = Exception("Template rendering failed")
fake = Faker()
email_data = self._create_test_email_data(fake)
# Act: Execute the task
send_inner_email_task(
to=email_data["to"],
subject=email_data["subject"],
body=email_data["body"],
substitutions=email_data["substitutions"],
)
# Assert: Verify template rendering was attempted
mock_external_service_dependencies["render_template"].assert_called_once()
# Verify no email service calls due to exception
mock_external_service_dependencies["email_service"].send_raw_email.assert_not_called()
def test_send_inner_email_service_error(self, db_session_with_containers, mock_external_service_dependencies):
"""
Test email sending when email service fails.
This test verifies:
- Exception handling during email sending
- Graceful error handling
"""
# Arrange: Setup email service to raise an exception
mock_external_service_dependencies["email_service"].send_raw_email.side_effect = Exception(
"Email service failed"
)
fake = Faker()
email_data = self._create_test_email_data(fake)
# Act: Execute the task
send_inner_email_task(
to=email_data["to"],
subject=email_data["subject"],
body=email_data["body"],
substitutions=email_data["substitutions"],
)
# Assert: Verify template rendering occurred
mock_external_service_dependencies["render_template"].assert_called_once()
# Verify email service was called (and failed)
mock_email_service = mock_external_service_dependencies["email_service"]
mock_email_service.send_raw_email.assert_called_once_with(
to=email_data["to"],
subject=email_data["subject"],
html_content="<html>Test email content</html>",
)