chore(test): remove unusful tests
This commit is contained in:
YuehuCao
parent
350270fe6e
commit
c968f3a594
@ -1,708 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Integration test: Data Layer ↔ Service Layer
|
||||
"""
|
||||
import asyncio
|
||||
import sys
|
||||
import os
|
||||
from motor.motor_asyncio import AsyncIOMotorClient
|
||||
from beanie import init_beanie
|
||||
from backend.models.models import MessageTemplateDoc, EmailSenderDoc
|
||||
from backend.services.template_message_service import TemplateMessageService
|
||||
from backend.services.email_sender_service import EmailSenderService
|
||||
from common.constants.region import UserRegion
|
||||
|
||||
# Set the test database
|
||||
TEST_MONGODB_URI = "mongodb://localhost:27017"
|
||||
TEST_DB_NAME = "test_freeleaps2_enhanced"
|
||||
|
||||
async def setup_database():
|
||||
"""Setup the test database """
|
||||
print("🔧 Setup the test database...")
|
||||
client = AsyncIOMotorClient(TEST_MONGODB_URI)
|
||||
db = client[TEST_DB_NAME]
|
||||
|
||||
# Initialize Beanie
|
||||
await init_beanie(database=db, document_models=[MessageTemplateDoc, EmailSenderDoc])
|
||||
|
||||
return client, db
|
||||
|
||||
async def cleanup_database(client, db):
|
||||
"""Clean up the test database"""
|
||||
print("🧹 Clean up the test database...")
|
||||
await client.drop_database(TEST_DB_NAME)
|
||||
client.close()
|
||||
|
||||
async def test_complete_template_workflow():
|
||||
"""Test the complete template workflow"""
|
||||
print("\n📝 Test the complete template workflow...")
|
||||
|
||||
client, db = await setup_database()
|
||||
|
||||
try:
|
||||
template_service = TemplateMessageService()
|
||||
email_sender_service = EmailSenderService()
|
||||
tenant_id = "test_workflow_tenant"
|
||||
|
||||
# 1. Create global template
|
||||
print(" 1. Create global template...")
|
||||
global_template = MessageTemplateDoc(
|
||||
template_id="workflow_template",
|
||||
tenant_id=None,
|
||||
region=UserRegion.OTHER,
|
||||
subject="Welcome {name}",
|
||||
body="Hello {name}, your verification code is {code}",
|
||||
is_active=True
|
||||
)
|
||||
|
||||
saved_template = await template_service.create_global_template(global_template)
|
||||
assert saved_template.template_id == "workflow_template"
|
||||
print(" ✅ Global template created successfully")
|
||||
|
||||
# 2. Assign to tenant
|
||||
print(" 2. Assign to tenant...")
|
||||
results = await template_service.assign_template_to_tenant(
|
||||
template_ids=["workflow_template"],
|
||||
region=UserRegion.OTHER,
|
||||
tenant_id=tenant_id
|
||||
)
|
||||
assert results[0]["success"] is True
|
||||
print(" ✅ Template assigned successfully")
|
||||
|
||||
# 3. Get tenant template
|
||||
print(" 3. Get tenant template...")
|
||||
template = await template_service.get_template(
|
||||
template_id="workflow_template",
|
||||
tenant_id=tenant_id,
|
||||
region=UserRegion.OTHER
|
||||
)
|
||||
assert template is not None
|
||||
assert template.tenant_id == tenant_id
|
||||
print(" ✅ Tenant template retrieved successfully")
|
||||
|
||||
# 4. Render template
|
||||
print(" 4. Render template...")
|
||||
properties = {
|
||||
"name": "John Doe",
|
||||
"code": "123456"
|
||||
}
|
||||
subject, body = await template_service.render_template(template, properties)
|
||||
assert subject == "Welcome John Doe"
|
||||
assert body == "Hello John Doe, your verification code is 123456"
|
||||
print(" ✅ Template rendered successfully")
|
||||
|
||||
print(" 🎉 Template workflow test passed!")
|
||||
|
||||
finally:
|
||||
await cleanup_database(client, db)
|
||||
|
||||
async def test_email_sender_workflow():
|
||||
"""Test the email sender workflow"""
|
||||
print("\n📧 Test the email sender workflow...")
|
||||
|
||||
client, db = await setup_database()
|
||||
|
||||
try:
|
||||
email_sender_service = EmailSenderService()
|
||||
tenant_id = "test_email_workflow_tenant"
|
||||
|
||||
# 1. Set initial email sender
|
||||
print(" 1. Set initial email sender...")
|
||||
result = await email_sender_service.set_email_sender(
|
||||
tenant_id=tenant_id,
|
||||
email_senders=["initial@example.com"]
|
||||
)
|
||||
assert result["success"] is True
|
||||
assert len(result["email_senders"]) == 1
|
||||
print(" ✅ Initial sender set successfully")
|
||||
|
||||
# 2. Add more senders
|
||||
print(" 2. Add more senders...")
|
||||
result = await email_sender_service.add_email_senders(
|
||||
tenant_id=tenant_id,
|
||||
new_senders=["added@example.com", "support@example.com"]
|
||||
)
|
||||
assert result["success"] is True
|
||||
assert len(result["email_senders"]) == 3
|
||||
print(" ✅ Senders added successfully")
|
||||
|
||||
# 3. Get final senders
|
||||
print(" 3. Get final senders...")
|
||||
senders = await email_sender_service.get_email_sender(tenant_id)
|
||||
assert len(senders) == 3
|
||||
assert "initial@example.com" in senders
|
||||
assert "added@example.com" in senders
|
||||
assert "support@example.com" in senders
|
||||
print(" ✅ Senders retrieved successfully")
|
||||
|
||||
print(" 🎉 Email sender workflow test passed!")
|
||||
|
||||
finally:
|
||||
await cleanup_database(client, db)
|
||||
|
||||
async def test_email_sender_edge_cases():
|
||||
"""Test the email sender edge cases"""
|
||||
print("\n🔍 Test the email sender edge cases...")
|
||||
|
||||
client, db = await setup_database()
|
||||
|
||||
try:
|
||||
email_sender_service = EmailSenderService()
|
||||
tenant_id = "test_edge_cases_tenant"
|
||||
|
||||
# 1. Test adding empty list
|
||||
print(" 1. Test adding empty list...")
|
||||
result = await email_sender_service.add_email_senders(tenant_id, [])
|
||||
assert result["success"] is False
|
||||
assert result["msg"] == "No sender provided"
|
||||
print(" ✅ Empty list handled correctly")
|
||||
|
||||
# 2. Test adding existing senders
|
||||
print(" 2. Test adding existing senders...")
|
||||
|
||||
await email_sender_service.set_email_sender(tenant_id, ["existing@example.com"])
|
||||
|
||||
|
||||
result = await email_sender_service.add_email_senders(tenant_id, ["existing@example.com"])
|
||||
assert result["success"] is False
|
||||
assert result["msg"] == "All senders already exist"
|
||||
print(" ✅ Duplicate senders handled correctly")
|
||||
|
||||
# 3. Test removing senders
|
||||
print(" 3. Test removing senders...")
|
||||
result = await email_sender_service.remove_email_senders(tenant_id, ["existing@example.com"])
|
||||
assert result["success"] is True
|
||||
assert len(result["remaining"]) == 0
|
||||
print(" ✅ Senders removed successfully")
|
||||
|
||||
# 4. Test removing nonexistent senders
|
||||
print(" 4. Test removing nonexistent senders...")
|
||||
#
|
||||
await email_sender_service.set_email_sender(tenant_id, ["test1@example.com", "test2@example.com"])
|
||||
result = await email_sender_service.remove_email_senders(tenant_id, ["nonexistent@example.com"])
|
||||
assert result["success"] is False
|
||||
assert result["msg"] == "No sender matched for removal"
|
||||
print(" ✅ Removing nonexistent senders handled correctly")
|
||||
|
||||
# 5. Test clearing senders
|
||||
print(" 5. Test clearing senders...")
|
||||
await email_sender_service.set_email_sender(tenant_id, ["test@example.com"])
|
||||
result = await email_sender_service.clear_email_senders(tenant_id)
|
||||
assert result["success"] is True
|
||||
print(" ✅ Clearing senders successfully")
|
||||
|
||||
# 6. Test deleting sender configuration
|
||||
print(" 6. Test deleting sender configuration...")
|
||||
result = await email_sender_service.delete_email_sender(tenant_id)
|
||||
assert result["success"] is True
|
||||
print(" ✅ Deleting sender configuration successfully")
|
||||
|
||||
# 7. Test getting nonexistent senders
|
||||
print(" 7. Test getting nonexistent senders...")
|
||||
senders = await email_sender_service.get_email_sender("nonexistent_tenant")
|
||||
assert senders == []
|
||||
print(" ✅ Getting nonexistent senders handled correctly")
|
||||
|
||||
print(" 🎉 Email sender edge cases test passed!")
|
||||
|
||||
finally:
|
||||
await cleanup_database(client, db)
|
||||
|
||||
async def test_email_sender_remaining_edge_cases():
|
||||
"""Test the remaining email sender edge cases"""
|
||||
print("\n🔍 Test the remaining email sender edge cases...")
|
||||
|
||||
client, db = await setup_database()
|
||||
|
||||
try:
|
||||
email_sender_service = EmailSenderService()
|
||||
tenant_id = "test_remaining_edge_tenant"
|
||||
|
||||
# 1. Test adding non-list type
|
||||
print(" 1. Test adding non-list type...")
|
||||
result = await email_sender_service.add_email_senders(tenant_id, "not_a_list")
|
||||
assert result["success"] is False
|
||||
assert result["msg"] == "No sender provided"
|
||||
print(" ✅ Non-list type handled correctly")
|
||||
|
||||
# 2. Test adding None
|
||||
print(" 2. Test adding None...")
|
||||
result = await email_sender_service.add_email_senders(tenant_id, None)
|
||||
assert result["success"] is False
|
||||
assert result["msg"] == "No sender provided"
|
||||
print(" ✅ None handled correctly")
|
||||
|
||||
# 2.5. Test adding non-list type (ensure isinstance check is covered)
|
||||
print(" 2.5. Test adding non-list type (ensure isinstance check is covered)...")
|
||||
result = await email_sender_service.add_email_senders(tenant_id, "not_a_list")
|
||||
assert result["success"] is False
|
||||
assert result["msg"] == "No sender provided"
|
||||
print(" ✅ Non-list type handled correctly")
|
||||
|
||||
# 2.6. Test adding number type
|
||||
print(" 2.6. Test adding number type...")
|
||||
result = await email_sender_service.add_email_senders(tenant_id, 123)
|
||||
assert result["success"] is False
|
||||
assert result["msg"] == "No sender provided"
|
||||
print(" ✅ Number type handled correctly")
|
||||
|
||||
# 2.7. Test adding dictionary type
|
||||
print(" 2.7. Test adding dictionary type...")
|
||||
result = await email_sender_service.add_email_senders(tenant_id, {"key": "value"})
|
||||
assert result["success"] is False
|
||||
assert result["msg"] == "No sender provided"
|
||||
print(" ✅ Dictionary type handled correctly")
|
||||
|
||||
# 3. Test removing senders when document exists but email_senders is empty
|
||||
print(" 3. Test removing senders when document exists but email_senders is empty...")
|
||||
# Create an empty sender document
|
||||
doc = EmailSenderDoc(tenant_id=tenant_id, email_senders=[])
|
||||
await doc.create()
|
||||
|
||||
result = await email_sender_service.remove_email_senders(tenant_id, ["test@example.com"])
|
||||
assert result["success"] is False
|
||||
assert result["msg"] == "No sender found"
|
||||
print(" ✅ Empty sender list handled correctly")
|
||||
|
||||
# 4. Test clearing senders when document does not exist
|
||||
print(" 4. Test clearing senders when document does not exist...")
|
||||
result = await email_sender_service.clear_email_senders("nonexistent_tenant")
|
||||
assert result["success"] is False
|
||||
assert result["msg"] == "No sender config found"
|
||||
print(" ✅ No document found for clearing senders")
|
||||
|
||||
# 5. Test deleting sender configuration when document does not exist
|
||||
print(" 5. Test deleting sender configuration when document does not exist...")
|
||||
result = await email_sender_service.delete_email_sender("another_nonexistent_tenant")
|
||||
assert result["success"] is False
|
||||
assert result["msg"] == "No sender config found"
|
||||
print(" ✅ No document found for deleting sender configuration")
|
||||
|
||||
print(" 🎉 Email sender remaining edge cases test passed!")
|
||||
|
||||
finally:
|
||||
await cleanup_database(client, db)
|
||||
|
||||
async def test_template_edge_cases():
|
||||
"""Test the template edge cases"""
|
||||
print("\n🔍 Test the template edge cases...")
|
||||
|
||||
client, db = await setup_database()
|
||||
|
||||
try:
|
||||
template_service = TemplateMessageService()
|
||||
tenant_id = "test_template_edge_tenant"
|
||||
|
||||
# 1. Test assigning nonexistent template
|
||||
print(" 1. Test assigning nonexistent template...")
|
||||
results = await template_service.assign_template_to_tenant(
|
||||
template_ids=["nonexistent_template"],
|
||||
region=UserRegion.OTHER,
|
||||
tenant_id=tenant_id
|
||||
)
|
||||
assert results[0]["success"] is False
|
||||
assert results[0]["msg"] == "Template not found"
|
||||
print(" ✅ Assigning nonexistent template handled correctly")
|
||||
|
||||
# 2. Test assigning already assigned template
|
||||
print(" 2. Test assigning already assigned template...")
|
||||
# Create global template
|
||||
global_template = MessageTemplateDoc(
|
||||
template_id="duplicate_template",
|
||||
tenant_id=None,
|
||||
region=UserRegion.OTHER,
|
||||
subject="Test",
|
||||
body="Test",
|
||||
is_active=True
|
||||
)
|
||||
await template_service.create_global_template(global_template)
|
||||
|
||||
# First assignment
|
||||
results = await template_service.assign_template_to_tenant(
|
||||
template_ids=["duplicate_template"],
|
||||
region=UserRegion.OTHER,
|
||||
tenant_id=tenant_id
|
||||
)
|
||||
assert results[0]["success"] is True
|
||||
|
||||
# Second assignment
|
||||
results = await template_service.assign_template_to_tenant(
|
||||
template_ids=["duplicate_template"],
|
||||
region=UserRegion.OTHER,
|
||||
tenant_id=tenant_id
|
||||
)
|
||||
assert results[0]["success"] is False
|
||||
assert results[0]["msg"] == "Template already assigned"
|
||||
print(" ✅ Duplicate assignment of template handled correctly")
|
||||
|
||||
# 3. Test template rendering error
|
||||
print(" 3. Test template rendering error...")
|
||||
template = MessageTemplateDoc(
|
||||
template_id="error_template",
|
||||
tenant_id=tenant_id,
|
||||
region=UserRegion.OTHER,
|
||||
subject="Hello {name}",
|
||||
body="Welcome {name}, code: {code}",
|
||||
is_active=True
|
||||
)
|
||||
await template_service.create_template(template, tenant_id)
|
||||
|
||||
# Rendering with missing parameters
|
||||
try:
|
||||
await template_service.render_template(template, {"name": "John"})
|
||||
assert False, "Should raise an exception"
|
||||
except ValueError as e:
|
||||
assert "Missing template parameter" in str(e)
|
||||
print(" ✅ Template rendering error handled correctly")
|
||||
|
||||
# 4. Test updating template
|
||||
print(" 4. Test updating template...")
|
||||
template_id = str(template.id)
|
||||
result = await template_service.update_template(
|
||||
template_id, tenant_id, {"subject": "Updated Hello {name}"}
|
||||
)
|
||||
assert result["success"] is True
|
||||
print(" ✅ Template updated successfully")
|
||||
|
||||
# 5. Test deleting template
|
||||
print(" 5. Test deleting template...")
|
||||
result = await template_service.delete_template(template_id, tenant_id)
|
||||
assert result["success"] is True
|
||||
print(" ✅ Template deleted successfully")
|
||||
|
||||
print(" 🎉 Template edge cases test passed!")
|
||||
|
||||
finally:
|
||||
await cleanup_database(client, db)
|
||||
|
||||
async def test_template_remaining_edge_cases():
|
||||
"""Test the remaining template edge cases"""
|
||||
print("\n🔍 Test the remaining template edge cases...")
|
||||
|
||||
client, db = await setup_database()
|
||||
|
||||
try:
|
||||
template_service = TemplateMessageService()
|
||||
tenant_id = "test_template_remaining_tenant"
|
||||
|
||||
# 1. Test updating nonexistent template
|
||||
print(" 1. Test updating nonexistent template...")
|
||||
try:
|
||||
await template_service.update_template("nonexistent_id", tenant_id, {"subject": "Test"})
|
||||
assert False, "Should raise an exception"
|
||||
except Exception as e:
|
||||
# Here might raise different exceptions, depending on Beanie's implementation
|
||||
print(" ✅ Updating nonexistent template handled correctly")
|
||||
|
||||
# 2. Test deleting nonexistent template
|
||||
print(" 2. Test deleting nonexistent template...")
|
||||
try:
|
||||
await template_service.delete_template("nonexistent_id", tenant_id)
|
||||
assert False, "Should raise an exception"
|
||||
except Exception as e:
|
||||
# TODO: Here might raise different exceptions, depending on Beanie's implementation
|
||||
print(" ✅ Deleting nonexistent template handled correctly")
|
||||
|
||||
# 3. Test updating global template when template does not exist
|
||||
print(" 3. Test updating global template when template does not exist...")
|
||||
try:
|
||||
await template_service.update_global_template("nonexistent_id", {"subject": "Test"})
|
||||
assert False, "Should raise an exception"
|
||||
except Exception as e:
|
||||
print(" ✅ Updating nonexistent global template handled correctly")
|
||||
|
||||
# 4. Test deleting global template when template does not exist
|
||||
print(" 4. Test deleting global template when template does not exist...")
|
||||
try:
|
||||
await template_service.delete_global_template("nonexistent_id")
|
||||
assert False, "Should raise an exception"
|
||||
except Exception as e:
|
||||
print(" ✅ Deleting nonexistent global template handled correctly")
|
||||
|
||||
# 5. Test updating global template when template belongs to tenant
|
||||
print(" 5. Test updating global template when template belongs to tenant...")
|
||||
# Create tenant template
|
||||
tenant_template = MessageTemplateDoc(
|
||||
template_id="tenant_template_for_admin_test",
|
||||
tenant_id=tenant_id,
|
||||
region=UserRegion.OTHER,
|
||||
subject="Test",
|
||||
body="Test",
|
||||
is_active=True
|
||||
)
|
||||
saved_tenant_template = await template_service.create_template(tenant_template, tenant_id)
|
||||
|
||||
try:
|
||||
await template_service.update_global_template(str(saved_tenant_template.id), {"subject": "Test"})
|
||||
assert False, "Should raise an exception"
|
||||
except PermissionError as e:
|
||||
assert "Not a global template" in str(e)
|
||||
print(" ✅ Tenant template as global template update handled correctly")
|
||||
|
||||
# 6. Test deleting global template when template belongs to tenant
|
||||
print(" 6. Test deleting global template when template belongs to tenant...")
|
||||
try:
|
||||
await template_service.delete_global_template(str(saved_tenant_template.id))
|
||||
assert False, "Should raise an exception"
|
||||
except PermissionError as e:
|
||||
assert "Not a global template" in str(e)
|
||||
print(" ✅ Tenant template as global template delete handled correctly")
|
||||
|
||||
print(" 🎉 Template remaining edge cases test passed!")
|
||||
|
||||
finally:
|
||||
await cleanup_database(client, db)
|
||||
|
||||
async def test_admin_template_operations():
|
||||
"""Test admin template operations"""
|
||||
print("\n👑 Test admin template operations...")
|
||||
|
||||
client, db = await setup_database()
|
||||
|
||||
try:
|
||||
template_service = TemplateMessageService()
|
||||
|
||||
# 1. Create global template
|
||||
print(" 1. Create global template...")
|
||||
global_template = MessageTemplateDoc(
|
||||
template_id="admin_template",
|
||||
tenant_id=None,
|
||||
region=UserRegion.OTHER,
|
||||
subject="Admin Template {name}",
|
||||
body="Admin message for {name}",
|
||||
is_active=True
|
||||
)
|
||||
saved_template = await template_service.create_global_template(global_template)
|
||||
template_id = str(saved_template.id)
|
||||
print(" ✅ Global template created successfully")
|
||||
|
||||
# 2. Update global template
|
||||
print(" 2. Update global template...")
|
||||
result = await template_service.update_global_template(
|
||||
template_id, {"subject": "Updated Admin Template {name}"}
|
||||
)
|
||||
assert result["success"] is True
|
||||
print(" ✅ Global template updated successfully")
|
||||
|
||||
# 3. Delete global template
|
||||
print(" 3. Delete global template...")
|
||||
result = await template_service.delete_global_template(template_id)
|
||||
assert result["success"] is True
|
||||
print(" ✅ Global template deleted successfully")
|
||||
|
||||
print(" 🎉 Admin template operations test passed!")
|
||||
|
||||
finally:
|
||||
await cleanup_database(client, db)
|
||||
|
||||
async def test_permission_errors():
|
||||
"""Test permission errors"""
|
||||
print("\n🚫 Test permission errors...")
|
||||
|
||||
client, db = await setup_database()
|
||||
|
||||
try:
|
||||
template_service = TemplateMessageService()
|
||||
tenant_id = "test_permission_tenant"
|
||||
|
||||
# 1. Create tenant template
|
||||
print(" 1. Create tenant template...")
|
||||
template = MessageTemplateDoc(
|
||||
template_id="permission_template",
|
||||
tenant_id=tenant_id,
|
||||
region=UserRegion.OTHER,
|
||||
subject="Test",
|
||||
body="Test",
|
||||
is_active=True
|
||||
)
|
||||
saved_template = await template_service.create_template(template, tenant_id)
|
||||
template_id = str(saved_template.id)
|
||||
print(" ✅ Tenant template created successfully")
|
||||
|
||||
# 2. Test error tenant update template
|
||||
print(" 2. Test error tenant update template...")
|
||||
try:
|
||||
await template_service.update_template(template_id, "wrong_tenant", {"subject": "Wrong"})
|
||||
assert False, "Should raise an exception"
|
||||
except PermissionError as e:
|
||||
assert "Forbidden" in str(e)
|
||||
print(" ✅ Permission error handled correctly")
|
||||
|
||||
# 3. Test error tenant delete template
|
||||
print(" 3. Test error tenant delete template...")
|
||||
try:
|
||||
await template_service.delete_template(template_id, "wrong_tenant")
|
||||
assert False, "应该抛出权限错误"
|
||||
except PermissionError as e:
|
||||
assert "Forbidden" in str(e)
|
||||
print(" ✅ Permission error handled correctly")
|
||||
|
||||
print(" 🎉 Permission errors test passed!")
|
||||
|
||||
finally:
|
||||
await cleanup_database(client, db)
|
||||
|
||||
async def test_tenant_isolation():
|
||||
"""Test tenant isolation"""
|
||||
print("\n🔒 Test tenant isolation...")
|
||||
|
||||
client, db = await setup_database()
|
||||
|
||||
try:
|
||||
template_service = TemplateMessageService()
|
||||
email_sender_service = EmailSenderService()
|
||||
tenant_a = "tenant_a"
|
||||
tenant_b = "tenant_b"
|
||||
|
||||
# Create two tenant templates
|
||||
print(" 1. Create two tenant templates...")
|
||||
template_a = MessageTemplateDoc(
|
||||
template_id="same_template_id",
|
||||
tenant_id=tenant_a,
|
||||
region=UserRegion.OTHER,
|
||||
subject="Tenant A: Welcome {name}",
|
||||
body="Hello {name} from Tenant A",
|
||||
is_active=True
|
||||
)
|
||||
await template_service.create_template(template_a, tenant_a)
|
||||
|
||||
template_b = MessageTemplateDoc(
|
||||
template_id="same_template_id",
|
||||
tenant_id=tenant_b,
|
||||
region=UserRegion.OTHER,
|
||||
subject="Tenant B: Welcome {name}",
|
||||
body="Hello {name} from Tenant B",
|
||||
is_active=True
|
||||
)
|
||||
await template_service.create_template(template_b, tenant_b)
|
||||
|
||||
# Set different email senders
|
||||
await email_sender_service.set_email_sender(
|
||||
tenant_id=tenant_a,
|
||||
email_senders=["a@example.com"]
|
||||
)
|
||||
|
||||
await email_sender_service.set_email_sender(
|
||||
tenant_id=tenant_b,
|
||||
email_senders=["b@example.com"]
|
||||
)
|
||||
|
||||
print(" ✅ Tenant data prepared")
|
||||
|
||||
# Verify tenant isolation
|
||||
print(" 2. Verify tenant isolation...")
|
||||
templates_a = await template_service.list_tenant_templates(tenant_a, UserRegion.OTHER)
|
||||
templates_b = await template_service.list_tenant_templates(tenant_b, UserRegion.OTHER)
|
||||
|
||||
assert len(templates_a) == 1
|
||||
assert len(templates_b) == 1
|
||||
assert templates_a[0].tenant_id == tenant_a
|
||||
assert templates_b[0].tenant_id == tenant_b
|
||||
|
||||
# Verify email sender isolation
|
||||
senders_a = await email_sender_service.get_email_sender(tenant_a)
|
||||
senders_b = await email_sender_service.get_email_sender(tenant_b)
|
||||
|
||||
assert senders_a == ["a@example.com"]
|
||||
assert senders_b == ["b@example.com"]
|
||||
|
||||
print(" ✅ Tenant isolation verified")
|
||||
|
||||
print(" 🎉 Tenant isolation test passed!")
|
||||
|
||||
finally:
|
||||
await cleanup_database(client, db)
|
||||
|
||||
async def test_list_templates():
|
||||
"""Test template list functionality"""
|
||||
print("\n📋 Test template list functionality...")
|
||||
|
||||
client, db = await setup_database()
|
||||
|
||||
try:
|
||||
template_service = TemplateMessageService()
|
||||
tenant_id = "test_list"
|
||||
|
||||
# 1. Create multiple global templates
|
||||
print(" 1. Create multiple global templates...")
|
||||
global_templates = []
|
||||
for i in range(3):
|
||||
template = MessageTemplateDoc(
|
||||
template_id=f"global_template_{i}",
|
||||
tenant_id=None,
|
||||
region=UserRegion.OTHER,
|
||||
subject=f"Global Template {i}",
|
||||
body=f"Global body {i}",
|
||||
is_active=True
|
||||
)
|
||||
saved_template = await template_service.create_global_template(template)
|
||||
global_templates.append(saved_template)
|
||||
print(" ✅ Global templates created successfully")
|
||||
|
||||
# 2. Test listing global templates
|
||||
print(" 2. Test listing global templates...")
|
||||
global_list = await template_service.list_global_templates(UserRegion.OTHER)
|
||||
assert len(global_list) == 3
|
||||
print(" ✅ Global templates list retrieved successfully")
|
||||
|
||||
# 3. Create multiple tenant templates
|
||||
print(" 3. Create multiple tenant templates...")
|
||||
tenant_templates = []
|
||||
for i in range(2):
|
||||
template = MessageTemplateDoc(
|
||||
template_id=f"tenant_template_{i}",
|
||||
tenant_id=tenant_id,
|
||||
region=UserRegion.OTHER,
|
||||
subject=f"Tenant Template {i}",
|
||||
body=f"Tenant body {i}",
|
||||
is_active=True
|
||||
)
|
||||
saved_template = await template_service.create_template(template, tenant_id)
|
||||
tenant_templates.append(saved_template)
|
||||
print(" ✅ Tenant templates created successfully")
|
||||
|
||||
# 4. Test listing tenant templates
|
||||
print(" 4. Test listing tenant templates...")
|
||||
tenant_list = await template_service.list_tenant_templates(tenant_id, UserRegion.OTHER)
|
||||
assert len(tenant_list) == 2
|
||||
print(" ✅ Tenant templates list retrieved successfully")
|
||||
|
||||
print(" 🎉 Template list functionality test passed!")
|
||||
|
||||
finally:
|
||||
await cleanup_database(client, db)
|
||||
|
||||
async def main():
|
||||
"""Main test function"""
|
||||
print("🚀 Start integration test...")
|
||||
|
||||
try:
|
||||
await test_complete_template_workflow()
|
||||
await test_email_sender_workflow()
|
||||
await test_email_sender_edge_cases()
|
||||
await test_email_sender_remaining_edge_cases()
|
||||
await test_template_edge_cases()
|
||||
await test_template_remaining_edge_cases()
|
||||
await test_admin_template_operations()
|
||||
await test_permission_errors()
|
||||
await test_tenant_isolation()
|
||||
await test_list_templates()
|
||||
|
||||
print("\n🎉 All tests passed!")
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
print(f"\n❌ Test failed: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
return False
|
||||
|
||||
if __name__ == "__main__":
|
||||
# Set PYTHONPATH
|
||||
current_dir = os.path.dirname(os.path.abspath(__file__))
|
||||
project_root = os.path.dirname(os.path.dirname(os.path.dirname(current_dir)))
|
||||
sys.path.insert(0, project_root)
|
||||
|
||||
# Run tests
|
||||
success = asyncio.run(main())
|
||||
sys.exit(0 if success else 1)
|
||||
Loading…
Reference in New Issue
Block a user