feat: add email system with multi-provider support

Implements a comprehensive email system with:
- Multi-provider support (SMTP, SendGrid, Mailgun, Amazon SES)
- Database-stored templates with i18n (EN, FR, DE, LB)
- Jinja2 template rendering with variable interpolation
- Email logging for debugging and compliance
- Debug mode for development (logs instead of sending)
- Welcome email integration in signup flow

New files:
- models/database/email.py: EmailTemplate and EmailLog models
- app/services/email_service.py: Provider abstraction and service
- scripts/seed_email_templates.py: Template seeding script
- tests/unit/services/test_email_service.py: 28 unit tests
- docs/features/email-system.md: Complete documentation

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
2025-12-27 21:05:50 +01:00
parent 98d082699c
commit 64fd8b5194
11 changed files with 2540 additions and 0 deletions

View File

@@ -98,6 +98,39 @@ STRIPE_PUBLISHABLE_KEY=pk_test_your_publishable_key_here
STRIPE_WEBHOOK_SECRET=whsec_your_webhook_secret_here STRIPE_WEBHOOK_SECRET=whsec_your_webhook_secret_here
STRIPE_TRIAL_DAYS=30 STRIPE_TRIAL_DAYS=30
# =============================================================================
# EMAIL CONFIGURATION
# =============================================================================
# Provider: smtp, sendgrid, mailgun, ses
EMAIL_PROVIDER=smtp
EMAIL_FROM_ADDRESS=noreply@wizamart.com
EMAIL_FROM_NAME=Wizamart
EMAIL_REPLY_TO=
# SMTP Settings (used when EMAIL_PROVIDER=smtp)
SMTP_HOST=smtp.example.com
SMTP_PORT=587
SMTP_USER=
SMTP_PASSWORD=
SMTP_USE_TLS=true
SMTP_USE_SSL=false
# SendGrid (used when EMAIL_PROVIDER=sendgrid)
# SENDGRID_API_KEY=SG.your_api_key_here
# Mailgun (used when EMAIL_PROVIDER=mailgun)
# MAILGUN_API_KEY=your_api_key_here
# MAILGUN_DOMAIN=mg.yourdomain.com
# Amazon SES (used when EMAIL_PROVIDER=ses)
# AWS_ACCESS_KEY_ID=your_access_key
# AWS_SECRET_ACCESS_KEY=your_secret_key
# AWS_REGION=eu-west-1
# Email behavior
EMAIL_ENABLED=true
EMAIL_DEBUG=false
# ============================================================================= # =============================================================================
# PLATFORM LIMITS # PLATFORM LIMITS
# ============================================================================= # =============================================================================

View File

@@ -0,0 +1,270 @@
"""add email templates and logs tables
Revision ID: d7a4a3f06394
Revises: 404b3e2d2865
Create Date: 2025-12-27 20:48:00.661523
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = 'd7a4a3f06394'
down_revision: Union[str, None] = '404b3e2d2865'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ###
op.create_table('email_templates',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('code', sa.String(length=100), nullable=False),
sa.Column('language', sa.String(length=5), nullable=False),
sa.Column('name', sa.String(length=255), nullable=False),
sa.Column('description', sa.Text(), nullable=True),
sa.Column('category', sa.String(length=50), nullable=False),
sa.Column('subject', sa.String(length=500), nullable=False),
sa.Column('body_html', sa.Text(), nullable=False),
sa.Column('body_text', sa.Text(), nullable=True),
sa.Column('variables', sa.Text(), nullable=True),
sa.Column('is_active', sa.Boolean(), nullable=False),
sa.Column('created_at', sa.DateTime(), nullable=False),
sa.Column('updated_at', sa.DateTime(), nullable=False),
sa.PrimaryKeyConstraint('id'),
sqlite_autoincrement=True
)
op.create_index(op.f('ix_email_templates_category'), 'email_templates', ['category'], unique=False)
op.create_index(op.f('ix_email_templates_code'), 'email_templates', ['code'], unique=False)
op.create_index(op.f('ix_email_templates_id'), 'email_templates', ['id'], unique=False)
op.create_table('email_logs',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('template_code', sa.String(length=100), nullable=True),
sa.Column('template_id', sa.Integer(), nullable=True),
sa.Column('recipient_email', sa.String(length=255), nullable=False),
sa.Column('recipient_name', sa.String(length=255), nullable=True),
sa.Column('subject', sa.String(length=500), nullable=False),
sa.Column('body_html', sa.Text(), nullable=True),
sa.Column('body_text', sa.Text(), nullable=True),
sa.Column('from_email', sa.String(length=255), nullable=False),
sa.Column('from_name', sa.String(length=255), nullable=True),
sa.Column('reply_to', sa.String(length=255), nullable=True),
sa.Column('status', sa.String(length=20), nullable=False),
sa.Column('sent_at', sa.DateTime(), nullable=True),
sa.Column('delivered_at', sa.DateTime(), nullable=True),
sa.Column('opened_at', sa.DateTime(), nullable=True),
sa.Column('clicked_at', sa.DateTime(), nullable=True),
sa.Column('error_message', sa.Text(), nullable=True),
sa.Column('retry_count', sa.Integer(), nullable=False),
sa.Column('provider', sa.String(length=50), nullable=True),
sa.Column('provider_message_id', sa.String(length=255), nullable=True),
sa.Column('vendor_id', sa.Integer(), nullable=True),
sa.Column('user_id', sa.Integer(), nullable=True),
sa.Column('related_type', sa.String(length=50), nullable=True),
sa.Column('related_id', sa.Integer(), nullable=True),
sa.Column('extra_data', sa.Text(), nullable=True),
sa.Column('created_at', sa.DateTime(), nullable=False),
sa.Column('updated_at', sa.DateTime(), nullable=False),
sa.ForeignKeyConstraint(['template_id'], ['email_templates.id'], ),
sa.ForeignKeyConstraint(['user_id'], ['users.id'], ),
sa.ForeignKeyConstraint(['vendor_id'], ['vendors.id'], ),
sa.PrimaryKeyConstraint('id')
)
op.create_index(op.f('ix_email_logs_id'), 'email_logs', ['id'], unique=False)
op.create_index(op.f('ix_email_logs_provider_message_id'), 'email_logs', ['provider_message_id'], unique=False)
op.create_index(op.f('ix_email_logs_recipient_email'), 'email_logs', ['recipient_email'], unique=False)
op.create_index(op.f('ix_email_logs_status'), 'email_logs', ['status'], unique=False)
op.create_index(op.f('ix_email_logs_template_code'), 'email_logs', ['template_code'], unique=False)
op.create_index(op.f('ix_email_logs_user_id'), 'email_logs', ['user_id'], unique=False)
op.create_index(op.f('ix_email_logs_vendor_id'), 'email_logs', ['vendor_id'], unique=False)
op.alter_column('application_logs', 'created_at',
existing_type=sa.DATETIME(),
nullable=False)
op.alter_column('application_logs', 'updated_at',
existing_type=sa.DATETIME(),
nullable=False)
op.drop_index(op.f('ix_capacity_snapshots_date'), table_name='capacity_snapshots')
op.create_index('ix_capacity_snapshots_date', 'capacity_snapshots', ['snapshot_date'], unique=False)
op.create_index(op.f('ix_capacity_snapshots_snapshot_date'), 'capacity_snapshots', ['snapshot_date'], unique=True)
op.alter_column('cart_items', 'created_at',
existing_type=sa.DATETIME(),
nullable=False)
op.alter_column('cart_items', 'updated_at',
existing_type=sa.DATETIME(),
nullable=False)
op.drop_index(op.f('ix_customers_addresses_id'), table_name='customer_addresses')
op.create_index(op.f('ix_customer_addresses_id'), 'customer_addresses', ['id'], unique=False)
op.alter_column('inventory', 'warehouse',
existing_type=sa.VARCHAR(),
nullable=False)
op.alter_column('inventory', 'bin_location',
existing_type=sa.VARCHAR(),
nullable=False)
op.alter_column('inventory', 'location',
existing_type=sa.VARCHAR(),
nullable=True)
op.drop_index(op.f('idx_inventory_product_location'), table_name='inventory')
op.drop_constraint(op.f('uq_inventory_product_location'), 'inventory', type_='unique')
op.create_unique_constraint('uq_inventory_product_warehouse_bin', 'inventory', ['product_id', 'warehouse', 'bin_location'])
op.create_index(op.f('ix_marketplace_import_errors_import_job_id'), 'marketplace_import_errors', ['import_job_id'], unique=False)
op.create_index(op.f('ix_marketplace_product_translations_id'), 'marketplace_product_translations', ['id'], unique=False)
op.alter_column('marketplace_products', 'is_digital',
existing_type=sa.BOOLEAN(),
nullable=True,
existing_server_default=sa.text('0'))
op.alter_column('marketplace_products', 'is_active',
existing_type=sa.BOOLEAN(),
nullable=True,
existing_server_default=sa.text('1'))
op.drop_index(op.f('idx_mp_is_active'), table_name='marketplace_products')
op.drop_index(op.f('idx_mp_platform'), table_name='marketplace_products')
op.drop_index(op.f('idx_mp_sku'), table_name='marketplace_products')
op.create_index(op.f('ix_marketplace_products_is_active'), 'marketplace_products', ['is_active'], unique=False)
op.create_index(op.f('ix_marketplace_products_is_digital'), 'marketplace_products', ['is_digital'], unique=False)
op.create_index(op.f('ix_marketplace_products_mpn'), 'marketplace_products', ['mpn'], unique=False)
op.create_index(op.f('ix_marketplace_products_platform'), 'marketplace_products', ['platform'], unique=False)
op.create_index(op.f('ix_marketplace_products_sku'), 'marketplace_products', ['sku'], unique=False)
op.drop_index(op.f('uq_order_item_exception'), table_name='order_item_exceptions')
op.create_index(op.f('ix_order_item_exceptions_original_gtin'), 'order_item_exceptions', ['original_gtin'], unique=False)
op.create_unique_constraint(None, 'order_item_exceptions', ['order_item_id'])
op.alter_column('order_items', 'needs_product_match',
existing_type=sa.BOOLEAN(),
nullable=True,
existing_server_default=sa.text("'0'"))
op.drop_index(op.f('ix_order_items_gtin'), table_name='order_items')
op.drop_index(op.f('ix_order_items_product_id'), table_name='order_items')
op.create_index(op.f('ix_product_translations_id'), 'product_translations', ['id'], unique=False)
op.drop_index(op.f('idx_product_active'), table_name='products')
op.drop_index(op.f('idx_product_featured'), table_name='products')
op.drop_index(op.f('idx_product_gtin'), table_name='products')
op.drop_index(op.f('idx_product_vendor_gtin'), table_name='products')
op.drop_constraint(op.f('uq_product'), 'products', type_='unique')
op.create_index('idx_product_vendor_active', 'products', ['vendor_id', 'is_active'], unique=False)
op.create_index('idx_product_vendor_featured', 'products', ['vendor_id', 'is_featured'], unique=False)
op.create_index(op.f('ix_products_gtin'), 'products', ['gtin'], unique=False)
op.create_index(op.f('ix_products_vendor_sku'), 'products', ['vendor_sku'], unique=False)
op.create_unique_constraint('uq_vendor_marketplace_product', 'products', ['vendor_id', 'marketplace_product_id'])
op.drop_index(op.f('ix_vendors_domains_domain'), table_name='vendor_domains')
op.drop_index(op.f('ix_vendors_domains_id'), table_name='vendor_domains')
op.create_index(op.f('ix_vendor_domains_domain'), 'vendor_domains', ['domain'], unique=True)
op.create_index(op.f('ix_vendor_domains_id'), 'vendor_domains', ['id'], unique=False)
op.alter_column('vendor_subscriptions', 'payment_retry_count',
existing_type=sa.INTEGER(),
nullable=False,
existing_server_default=sa.text('0'))
op.create_foreign_key(None, 'vendor_subscriptions', 'subscription_tiers', ['tier_id'], ['id'])
op.drop_index(op.f('ix_vendors_themes_id'), table_name='vendor_themes')
op.create_index(op.f('ix_vendor_themes_id'), 'vendor_themes', ['id'], unique=False)
op.drop_index(op.f('ix_vendors_users_id'), table_name='vendor_users')
op.drop_index(op.f('ix_vendors_users_invitation_token'), table_name='vendor_users')
op.create_index(op.f('ix_vendor_users_id'), 'vendor_users', ['id'], unique=False)
op.create_index(op.f('ix_vendor_users_invitation_token'), 'vendor_users', ['invitation_token'], unique=False)
op.alter_column('vendors', 'company_id',
existing_type=sa.INTEGER(),
nullable=False)
# ### end Alembic commands ###
def downgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ###
op.alter_column('vendors', 'company_id',
existing_type=sa.INTEGER(),
nullable=True)
op.drop_index(op.f('ix_vendor_users_invitation_token'), table_name='vendor_users')
op.drop_index(op.f('ix_vendor_users_id'), table_name='vendor_users')
op.create_index(op.f('ix_vendors_users_invitation_token'), 'vendor_users', ['invitation_token'], unique=False)
op.create_index(op.f('ix_vendors_users_id'), 'vendor_users', ['id'], unique=False)
op.drop_index(op.f('ix_vendor_themes_id'), table_name='vendor_themes')
op.create_index(op.f('ix_vendors_themes_id'), 'vendor_themes', ['id'], unique=False)
op.drop_constraint(None, 'vendor_subscriptions', type_='foreignkey')
op.alter_column('vendor_subscriptions', 'payment_retry_count',
existing_type=sa.INTEGER(),
nullable=True,
existing_server_default=sa.text('0'))
op.drop_index(op.f('ix_vendor_domains_id'), table_name='vendor_domains')
op.drop_index(op.f('ix_vendor_domains_domain'), table_name='vendor_domains')
op.create_index(op.f('ix_vendors_domains_id'), 'vendor_domains', ['id'], unique=False)
op.create_index(op.f('ix_vendors_domains_domain'), 'vendor_domains', ['domain'], unique=1)
op.drop_constraint('uq_vendor_marketplace_product', 'products', type_='unique')
op.drop_index(op.f('ix_products_vendor_sku'), table_name='products')
op.drop_index(op.f('ix_products_gtin'), table_name='products')
op.drop_index('idx_product_vendor_featured', table_name='products')
op.drop_index('idx_product_vendor_active', table_name='products')
op.create_unique_constraint(op.f('uq_product'), 'products', ['vendor_id', 'marketplace_product_id'])
op.create_index(op.f('idx_product_vendor_gtin'), 'products', ['vendor_id', 'gtin'], unique=False)
op.create_index(op.f('idx_product_gtin'), 'products', ['gtin'], unique=False)
op.create_index(op.f('idx_product_featured'), 'products', ['vendor_id', 'is_featured'], unique=False)
op.create_index(op.f('idx_product_active'), 'products', ['vendor_id', 'is_active'], unique=False)
op.drop_index(op.f('ix_product_translations_id'), table_name='product_translations')
op.create_index(op.f('ix_order_items_product_id'), 'order_items', ['product_id'], unique=False)
op.create_index(op.f('ix_order_items_gtin'), 'order_items', ['gtin'], unique=False)
op.alter_column('order_items', 'needs_product_match',
existing_type=sa.BOOLEAN(),
nullable=False,
existing_server_default=sa.text("'0'"))
op.drop_constraint(None, 'order_item_exceptions', type_='unique')
op.drop_index(op.f('ix_order_item_exceptions_original_gtin'), table_name='order_item_exceptions')
op.create_index(op.f('uq_order_item_exception'), 'order_item_exceptions', ['order_item_id'], unique=1)
op.drop_index(op.f('ix_marketplace_products_sku'), table_name='marketplace_products')
op.drop_index(op.f('ix_marketplace_products_platform'), table_name='marketplace_products')
op.drop_index(op.f('ix_marketplace_products_mpn'), table_name='marketplace_products')
op.drop_index(op.f('ix_marketplace_products_is_digital'), table_name='marketplace_products')
op.drop_index(op.f('ix_marketplace_products_is_active'), table_name='marketplace_products')
op.create_index(op.f('idx_mp_sku'), 'marketplace_products', ['sku'], unique=False)
op.create_index(op.f('idx_mp_platform'), 'marketplace_products', ['platform'], unique=False)
op.create_index(op.f('idx_mp_is_active'), 'marketplace_products', ['is_active'], unique=False)
op.alter_column('marketplace_products', 'is_active',
existing_type=sa.BOOLEAN(),
nullable=False,
existing_server_default=sa.text('1'))
op.alter_column('marketplace_products', 'is_digital',
existing_type=sa.BOOLEAN(),
nullable=False,
existing_server_default=sa.text('0'))
op.drop_index(op.f('ix_marketplace_product_translations_id'), table_name='marketplace_product_translations')
op.drop_index(op.f('ix_marketplace_import_errors_import_job_id'), table_name='marketplace_import_errors')
op.drop_constraint('uq_inventory_product_warehouse_bin', 'inventory', type_='unique')
op.create_unique_constraint(op.f('uq_inventory_product_location'), 'inventory', ['product_id', 'location'])
op.create_index(op.f('idx_inventory_product_location'), 'inventory', ['product_id', 'location'], unique=False)
op.alter_column('inventory', 'location',
existing_type=sa.VARCHAR(),
nullable=False)
op.alter_column('inventory', 'bin_location',
existing_type=sa.VARCHAR(),
nullable=True)
op.alter_column('inventory', 'warehouse',
existing_type=sa.VARCHAR(),
nullable=True)
op.drop_index(op.f('ix_customer_addresses_id'), table_name='customer_addresses')
op.create_index(op.f('ix_customers_addresses_id'), 'customer_addresses', ['id'], unique=False)
op.alter_column('cart_items', 'updated_at',
existing_type=sa.DATETIME(),
nullable=True)
op.alter_column('cart_items', 'created_at',
existing_type=sa.DATETIME(),
nullable=True)
op.drop_index(op.f('ix_capacity_snapshots_snapshot_date'), table_name='capacity_snapshots')
op.drop_index('ix_capacity_snapshots_date', table_name='capacity_snapshots')
op.create_index(op.f('ix_capacity_snapshots_date'), 'capacity_snapshots', ['snapshot_date'], unique=1)
op.alter_column('application_logs', 'updated_at',
existing_type=sa.DATETIME(),
nullable=True)
op.alter_column('application_logs', 'created_at',
existing_type=sa.DATETIME(),
nullable=True)
op.drop_index(op.f('ix_email_logs_vendor_id'), table_name='email_logs')
op.drop_index(op.f('ix_email_logs_user_id'), table_name='email_logs')
op.drop_index(op.f('ix_email_logs_template_code'), table_name='email_logs')
op.drop_index(op.f('ix_email_logs_status'), table_name='email_logs')
op.drop_index(op.f('ix_email_logs_recipient_email'), table_name='email_logs')
op.drop_index(op.f('ix_email_logs_provider_message_id'), table_name='email_logs')
op.drop_index(op.f('ix_email_logs_id'), table_name='email_logs')
op.drop_table('email_logs')
op.drop_index(op.f('ix_email_templates_id'), table_name='email_templates')
op.drop_index(op.f('ix_email_templates_code'), table_name='email_templates')
op.drop_index(op.f('ix_email_templates_category'), table_name='email_templates')
op.drop_table('email_templates')
# ### end Alembic commands ###

View File

@@ -125,6 +125,39 @@ class Settings(BaseSettings):
stripe_webhook_secret: str = "" stripe_webhook_secret: str = ""
stripe_trial_days: int = 30 # 1-month free trial (card collected upfront but not charged) stripe_trial_days: int = 30 # 1-month free trial (card collected upfront but not charged)
# =============================================================================
# EMAIL CONFIGURATION
# =============================================================================
# Provider: smtp, sendgrid, mailgun, ses
email_provider: str = "smtp"
email_from_address: str = "noreply@wizamart.com"
email_from_name: str = "Wizamart"
email_reply_to: str = "" # Optional reply-to address
# SMTP Settings (used when email_provider=smtp)
smtp_host: str = "localhost"
smtp_port: int = 587
smtp_user: str = ""
smtp_password: str = ""
smtp_use_tls: bool = True
smtp_use_ssl: bool = False # For port 465
# SendGrid (used when email_provider=sendgrid)
sendgrid_api_key: str = ""
# Mailgun (used when email_provider=mailgun)
mailgun_api_key: str = ""
mailgun_domain: str = ""
# Amazon SES (used when email_provider=ses)
aws_access_key_id: str = ""
aws_secret_access_key: str = ""
aws_region: str = "eu-west-1"
# Email behavior
email_enabled: bool = True # Set to False to disable all emails
email_debug: bool = False # Log emails instead of sending (for development)
# ============================================================================= # =============================================================================
# DEMO/SEED DATA CONFIGURATION # DEMO/SEED DATA CONFIGURATION
# ============================================================================= # =============================================================================

View File

@@ -0,0 +1,572 @@
# app/services/email_service.py
"""
Email service with multi-provider support.
Supports:
- SMTP (default)
- SendGrid
- Mailgun
- Amazon SES
Features:
- Multi-language templates from database
- Jinja2 template rendering
- Email logging and tracking
- Queue support via background tasks
"""
import json
import logging
import smtplib
from abc import ABC, abstractmethod
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from typing import Any
from jinja2 import Environment, BaseLoader
from sqlalchemy.orm import Session
from app.core.config import settings
from models.database.email import EmailLog, EmailStatus, EmailTemplate
logger = logging.getLogger(__name__)
# =============================================================================
# EMAIL PROVIDER ABSTRACTION
# =============================================================================
class EmailProvider(ABC):
"""Abstract base class for email providers."""
@abstractmethod
def send(
self,
to_email: str,
to_name: str | None,
subject: str,
body_html: str,
body_text: str | None,
from_email: str,
from_name: str | None,
reply_to: str | None = None,
) -> tuple[bool, str | None, str | None]:
"""
Send an email.
Returns:
tuple: (success, provider_message_id, error_message)
"""
pass
class SMTPProvider(EmailProvider):
"""SMTP email provider."""
def send(
self,
to_email: str,
to_name: str | None,
subject: str,
body_html: str,
body_text: str | None,
from_email: str,
from_name: str | None,
reply_to: str | None = None,
) -> tuple[bool, str | None, str | None]:
try:
msg = MIMEMultipart("alternative")
msg["Subject"] = subject
msg["From"] = f"{from_name} <{from_email}>" if from_name else from_email
msg["To"] = f"{to_name} <{to_email}>" if to_name else to_email
if reply_to:
msg["Reply-To"] = reply_to
# Attach text and HTML parts
if body_text:
msg.attach(MIMEText(body_text, "plain", "utf-8"))
msg.attach(MIMEText(body_html, "html", "utf-8"))
# Connect and send
if settings.smtp_use_ssl:
server = smtplib.SMTP_SSL(settings.smtp_host, settings.smtp_port)
else:
server = smtplib.SMTP(settings.smtp_host, settings.smtp_port)
try:
if settings.smtp_use_tls and not settings.smtp_use_ssl:
server.starttls()
if settings.smtp_user and settings.smtp_password:
server.login(settings.smtp_user, settings.smtp_password)
server.sendmail(from_email, [to_email], msg.as_string())
return True, None, None
finally:
server.quit()
except Exception as e:
logger.error(f"SMTP send error: {e}")
return False, None, str(e)
class SendGridProvider(EmailProvider):
"""SendGrid email provider."""
def send(
self,
to_email: str,
to_name: str | None,
subject: str,
body_html: str,
body_text: str | None,
from_email: str,
from_name: str | None,
reply_to: str | None = None,
) -> tuple[bool, str | None, str | None]:
try:
from sendgrid import SendGridAPIClient
from sendgrid.helpers.mail import Mail, Email, To, Content
message = Mail(
from_email=Email(from_email, from_name),
to_emails=To(to_email, to_name),
subject=subject,
)
message.add_content(Content("text/html", body_html))
if body_text:
message.add_content(Content("text/plain", body_text))
if reply_to:
message.reply_to = Email(reply_to)
sg = SendGridAPIClient(settings.sendgrid_api_key)
response = sg.send(message)
if response.status_code in (200, 201, 202):
message_id = response.headers.get("X-Message-Id")
return True, message_id, None
else:
return False, None, f"SendGrid error: {response.status_code}"
except ImportError:
return False, None, "SendGrid library not installed. Run: pip install sendgrid"
except Exception as e:
logger.error(f"SendGrid send error: {e}")
return False, None, str(e)
class MailgunProvider(EmailProvider):
"""Mailgun email provider."""
def send(
self,
to_email: str,
to_name: str | None,
subject: str,
body_html: str,
body_text: str | None,
from_email: str,
from_name: str | None,
reply_to: str | None = None,
) -> tuple[bool, str | None, str | None]:
try:
import requests
from_str = f"{from_name} <{from_email}>" if from_name else from_email
to_str = f"{to_name} <{to_email}>" if to_name else to_email
data = {
"from": from_str,
"to": to_str,
"subject": subject,
"html": body_html,
}
if body_text:
data["text"] = body_text
if reply_to:
data["h:Reply-To"] = reply_to
response = requests.post(
f"https://api.mailgun.net/v3/{settings.mailgun_domain}/messages",
auth=("api", settings.mailgun_api_key),
data=data,
timeout=30,
)
if response.status_code == 200:
result = response.json()
return True, result.get("id"), None
else:
return False, None, f"Mailgun error: {response.status_code} - {response.text}"
except Exception as e:
logger.error(f"Mailgun send error: {e}")
return False, None, str(e)
class SESProvider(EmailProvider):
"""Amazon SES email provider."""
def send(
self,
to_email: str,
to_name: str | None,
subject: str,
body_html: str,
body_text: str | None,
from_email: str,
from_name: str | None,
reply_to: str | None = None,
) -> tuple[bool, str | None, str | None]:
try:
import boto3
ses = boto3.client(
"ses",
region_name=settings.aws_region,
aws_access_key_id=settings.aws_access_key_id,
aws_secret_access_key=settings.aws_secret_access_key,
)
from_str = f"{from_name} <{from_email}>" if from_name else from_email
body = {"Html": {"Charset": "UTF-8", "Data": body_html}}
if body_text:
body["Text"] = {"Charset": "UTF-8", "Data": body_text}
kwargs = {
"Source": from_str,
"Destination": {"ToAddresses": [to_email]},
"Message": {
"Subject": {"Charset": "UTF-8", "Data": subject},
"Body": body,
},
}
if reply_to:
kwargs["ReplyToAddresses"] = [reply_to]
response = ses.send_email(**kwargs)
return True, response.get("MessageId"), None
except ImportError:
return False, None, "boto3 library not installed. Run: pip install boto3"
except Exception as e:
logger.error(f"SES send error: {e}")
return False, None, str(e)
class DebugProvider(EmailProvider):
"""Debug provider - logs emails instead of sending."""
def send(
self,
to_email: str,
to_name: str | None,
subject: str,
body_html: str,
body_text: str | None,
from_email: str,
from_name: str | None,
reply_to: str | None = None,
) -> tuple[bool, str | None, str | None]:
logger.info(
f"\n{'='*60}\n"
f"DEBUG EMAIL\n"
f"{'='*60}\n"
f"To: {to_name} <{to_email}>\n"
f"From: {from_name} <{from_email}>\n"
f"Reply-To: {reply_to}\n"
f"Subject: {subject}\n"
f"{'='*60}\n"
f"Body (text):\n{body_text or '(none)'}\n"
f"{'='*60}\n"
f"Body (html):\n{body_html[:500]}...\n"
f"{'='*60}\n"
)
return True, f"debug-{to_email}", None
# =============================================================================
# EMAIL SERVICE
# =============================================================================
def get_provider() -> EmailProvider:
"""Get the configured email provider."""
if settings.email_debug:
return DebugProvider()
provider_map = {
"smtp": SMTPProvider,
"sendgrid": SendGridProvider,
"mailgun": MailgunProvider,
"ses": SESProvider,
}
provider_class = provider_map.get(settings.email_provider.lower())
if not provider_class:
logger.warning(f"Unknown email provider: {settings.email_provider}, using SMTP")
return SMTPProvider()
return provider_class()
class EmailService:
"""
Email service for sending templated emails.
Usage:
email_service = EmailService(db)
# Send using database template
email_service.send_template(
template_code="signup_welcome",
language="en",
to_email="user@example.com",
to_name="John Doe",
variables={"first_name": "John", "login_url": "https://..."},
vendor_id=1,
)
# Send raw email
email_service.send_raw(
to_email="user@example.com",
subject="Hello",
body_html="<h1>Hello</h1>",
)
"""
def __init__(self, db: Session):
self.db = db
self.provider = get_provider()
self.jinja_env = Environment(loader=BaseLoader())
def get_template(
self, template_code: str, language: str = "en"
) -> EmailTemplate | None:
"""Get email template from database with fallback to English."""
template = (
self.db.query(EmailTemplate)
.filter(
EmailTemplate.code == template_code,
EmailTemplate.language == language,
EmailTemplate.is_active == True,
)
.first()
)
# Fallback to English if not found
if not template and language != "en":
template = (
self.db.query(EmailTemplate)
.filter(
EmailTemplate.code == template_code,
EmailTemplate.language == "en",
EmailTemplate.is_active == True,
)
.first()
)
return template
def render_template(self, template_string: str, variables: dict[str, Any]) -> str:
"""Render a Jinja2 template string with variables."""
try:
template = self.jinja_env.from_string(template_string)
return template.render(**variables)
except Exception as e:
logger.error(f"Template rendering error: {e}")
return template_string
def send_template(
self,
template_code: str,
to_email: str,
to_name: str | None = None,
language: str = "en",
variables: dict[str, Any] | None = None,
vendor_id: int | None = None,
user_id: int | None = None,
related_type: str | None = None,
related_id: int | None = None,
) -> EmailLog:
"""
Send an email using a database template.
Args:
template_code: Template code (e.g., "signup_welcome")
to_email: Recipient email address
to_name: Recipient name (optional)
language: Language code (default: "en")
variables: Template variables dict
vendor_id: Related vendor ID for logging
user_id: Related user ID for logging
related_type: Related entity type (e.g., "order")
related_id: Related entity ID
Returns:
EmailLog record
"""
variables = variables or {}
# Get template
template = self.get_template(template_code, language)
if not template:
logger.error(f"Email template not found: {template_code} ({language})")
# Create failed log entry
log = EmailLog(
template_code=template_code,
recipient_email=to_email,
recipient_name=to_name,
subject=f"[Template not found: {template_code}]",
from_email=settings.email_from_address,
from_name=settings.email_from_name,
status=EmailStatus.FAILED.value,
error_message=f"Template not found: {template_code} ({language})",
provider=settings.email_provider,
vendor_id=vendor_id,
user_id=user_id,
related_type=related_type,
related_id=related_id,
)
self.db.add(log)
self.db.commit()
return log
# Render template
subject = self.render_template(template.subject, variables)
body_html = self.render_template(template.body_html, variables)
body_text = (
self.render_template(template.body_text, variables)
if template.body_text
else None
)
return self.send_raw(
to_email=to_email,
to_name=to_name,
subject=subject,
body_html=body_html,
body_text=body_text,
template_code=template_code,
template_id=template.id,
vendor_id=vendor_id,
user_id=user_id,
related_type=related_type,
related_id=related_id,
extra_data=json.dumps(variables) if variables else None,
)
def send_raw(
self,
to_email: str,
subject: str,
body_html: str,
to_name: str | None = None,
body_text: str | None = None,
from_email: str | None = None,
from_name: str | None = None,
reply_to: str | None = None,
template_code: str | None = None,
template_id: int | None = None,
vendor_id: int | None = None,
user_id: int | None = None,
related_type: str | None = None,
related_id: int | None = None,
extra_data: str | None = None,
) -> EmailLog:
"""
Send a raw email without using a template.
Returns:
EmailLog record
"""
from_email = from_email or settings.email_from_address
from_name = from_name or settings.email_from_name
reply_to = reply_to or settings.email_reply_to or None
# Create log entry
log = EmailLog(
template_code=template_code,
template_id=template_id,
recipient_email=to_email,
recipient_name=to_name,
subject=subject,
body_html=body_html,
body_text=body_text,
from_email=from_email,
from_name=from_name,
reply_to=reply_to,
status=EmailStatus.PENDING.value,
provider=settings.email_provider,
vendor_id=vendor_id,
user_id=user_id,
related_type=related_type,
related_id=related_id,
extra_data=extra_data,
)
self.db.add(log)
self.db.flush()
# Check if emails are disabled
if not settings.email_enabled:
log.status = EmailStatus.FAILED.value
log.error_message = "Email sending is disabled"
self.db.commit()
logger.info(f"Email sending disabled, skipping: {to_email}")
return log
# Send email
success, message_id, error = self.provider.send(
to_email=to_email,
to_name=to_name,
subject=subject,
body_html=body_html,
body_text=body_text,
from_email=from_email,
from_name=from_name,
reply_to=reply_to,
)
if success:
log.mark_sent(message_id)
logger.info(f"Email sent to {to_email}: {subject}")
else:
log.mark_failed(error or "Unknown error")
logger.error(f"Email failed to {to_email}: {error}")
self.db.commit()
return log
# =============================================================================
# CONVENIENCE FUNCTIONS
# =============================================================================
def send_email(
db: Session,
template_code: str,
to_email: str,
to_name: str | None = None,
language: str = "en",
variables: dict[str, Any] | None = None,
**kwargs,
) -> EmailLog:
"""Convenience function to send a templated email."""
service = EmailService(db)
return service.send_template(
template_code=template_code,
to_email=to_email,
to_name=to_name,
language=language,
variables=variables,
**kwargs,
)

View File

@@ -22,12 +22,14 @@ from app.exceptions import (
ResourceNotFoundException, ResourceNotFoundException,
ValidationException, ValidationException,
) )
from app.services.email_service import EmailService
from app.services.stripe_service import stripe_service from app.services.stripe_service import stripe_service
from middleware.auth import AuthManager from middleware.auth import AuthManager
from models.database.company import Company from models.database.company import Company
from models.database.subscription import ( from models.database.subscription import (
SubscriptionStatus, SubscriptionStatus,
TierCode, TierCode,
TIER_LIMITS,
VendorSubscription, VendorSubscription,
) )
from models.database.user import User from models.database.user import User
@@ -467,6 +469,62 @@ class PlatformSignupService:
return setup_intent.client_secret, stripe_customer_id return setup_intent.client_secret, stripe_customer_id
# =========================================================================
# Welcome Email
# =========================================================================
def send_welcome_email(
self,
db: Session,
user: User,
vendor: Vendor,
tier_code: str,
language: str = "fr",
) -> None:
"""
Send welcome email to new vendor.
Args:
db: Database session
user: User who signed up
vendor: Vendor that was created
tier_code: Selected tier code
language: Language for email (default: French)
"""
try:
# Get tier name
tier_enum = TierCode(tier_code)
tier_name = TIER_LIMITS.get(tier_enum, {}).get("name", tier_code.title())
# Build login URL
login_url = f"https://{settings.platform_domain}/vendor/{vendor.vendor_code}/dashboard"
email_service = EmailService(db)
email_service.send_template(
template_code="signup_welcome",
language=language,
to_email=user.email,
to_name=f"{user.first_name} {user.last_name}",
variables={
"first_name": user.first_name,
"company_name": vendor.name,
"email": user.email,
"vendor_code": vendor.vendor_code,
"login_url": login_url,
"trial_days": settings.stripe_trial_days,
"tier_name": tier_name,
},
vendor_id=vendor.id,
user_id=user.id,
related_type="signup",
)
logger.info(f"Welcome email sent to {user.email}")
except Exception as e:
# Log error but don't fail signup
logger.error(f"Failed to send welcome email to {user.email}: {e}")
# ========================================================================= # =========================================================================
# Signup Completion # Signup Completion
# ========================================================================= # =========================================================================
@@ -543,6 +601,15 @@ class PlatformSignupService:
else datetime.now(UTC) + timedelta(days=30) else datetime.now(UTC) + timedelta(days=30)
) )
# Get user for welcome email
user_id = session.get("user_id")
user = db.query(User).filter(User.id == user_id).first() if user_id else None
# Send welcome email
if user and vendor:
tier_code = session.get("tier_code", TierCode.ESSENTIAL.value)
self.send_welcome_email(db, user, vendor, tier_code)
# Clean up session # Clean up session
self.delete_session(session_id) self.delete_session(session_id)

View File

@@ -0,0 +1,331 @@
# Email System
The email system provides multi-provider support with database-stored templates and comprehensive logging for the Wizamart platform.
## Overview
The email system supports:
- **Multiple Providers**: SMTP, SendGrid, Mailgun, Amazon SES
- **Multi-language Templates**: EN, FR, DE, LB (stored in database)
- **Jinja2 Templating**: Variable interpolation in subjects and bodies
- **Email Logging**: Track all sent emails for debugging and compliance
- **Debug Mode**: Log emails instead of sending during development
## Configuration
### Environment Variables
Add these settings to your `.env` file:
```env
# Provider: smtp, sendgrid, mailgun, ses
EMAIL_PROVIDER=smtp
EMAIL_FROM_ADDRESS=noreply@wizamart.com
EMAIL_FROM_NAME=Wizamart
EMAIL_REPLY_TO=
# Behavior
EMAIL_ENABLED=true
EMAIL_DEBUG=false
# SMTP Settings (when EMAIL_PROVIDER=smtp)
SMTP_HOST=smtp.example.com
SMTP_PORT=587
SMTP_USER=
SMTP_PASSWORD=
SMTP_USE_TLS=true
SMTP_USE_SSL=false
# SendGrid (when EMAIL_PROVIDER=sendgrid)
# SENDGRID_API_KEY=SG.your_api_key_here
# Mailgun (when EMAIL_PROVIDER=mailgun)
# MAILGUN_API_KEY=your_api_key_here
# MAILGUN_DOMAIN=mg.yourdomain.com
# Amazon SES (when EMAIL_PROVIDER=ses)
# AWS_ACCESS_KEY_ID=your_access_key
# AWS_SECRET_ACCESS_KEY=your_secret_key
# AWS_REGION=eu-west-1
```
### Debug Mode
Set `EMAIL_DEBUG=true` to log emails instead of sending them. This is useful during development:
```env
EMAIL_DEBUG=true
```
Emails will be logged to the console with full details (recipient, subject, body preview).
## Database Models
### EmailTemplate
Stores multi-language email templates:
| Column | Type | Description |
|--------|------|-------------|
| id | Integer | Primary key |
| code | String(100) | Template identifier (e.g., "signup_welcome") |
| language | String(5) | Language code (en, fr, de, lb) |
| name | String(255) | Human-readable name |
| description | Text | Template purpose |
| category | String(50) | AUTH, ORDERS, BILLING, SYSTEM, MARKETING |
| subject | String(500) | Email subject (supports Jinja2) |
| body_html | Text | HTML body |
| body_text | Text | Plain text fallback |
| variables | Text | JSON list of expected variables |
| is_active | Boolean | Enable/disable template |
### EmailLog
Tracks all sent emails:
| Column | Type | Description |
|--------|------|-------------|
| id | Integer | Primary key |
| template_code | String(100) | Template used (if any) |
| recipient_email | String(255) | Recipient address |
| subject | String(500) | Email subject |
| status | String(20) | PENDING, SENT, FAILED, DELIVERED, OPENED |
| sent_at | DateTime | When email was sent |
| error_message | Text | Error details if failed |
| provider | String(50) | Provider used (smtp, sendgrid, etc.) |
| vendor_id | Integer | Related vendor (optional) |
| user_id | Integer | Related user (optional) |
## Usage
### Using EmailService
```python
from app.services.email_service import EmailService
def send_welcome_email(db, user, vendor):
email_service = EmailService(db)
email_service.send_template(
template_code="signup_welcome",
to_email=user.email,
to_name=f"{user.first_name} {user.last_name}",
language="fr", # Falls back to "en" if not found
variables={
"first_name": user.first_name,
"company_name": vendor.name,
"vendor_code": vendor.vendor_code,
"login_url": f"https://wizamart.com/vendor/{vendor.vendor_code}/dashboard",
"trial_days": 30,
"tier_name": "Essential",
},
vendor_id=vendor.id,
user_id=user.id,
related_type="signup",
)
```
### Convenience Function
```python
from app.services.email_service import send_email
send_email(
db=db,
template_code="order_confirmation",
to_email="customer@example.com",
language="en",
variables={"order_number": "ORD-001"},
)
```
### Sending Raw Emails
For one-off emails without templates:
```python
email_service = EmailService(db)
email_service.send_raw(
to_email="user@example.com",
subject="Custom Subject",
body_html="<h1>Hello</h1><p>Custom message</p>",
body_text="Hello\n\nCustom message",
)
```
## Email Templates
### Creating Templates
Templates use Jinja2 syntax for variable interpolation:
```html
<p>Hello {{ first_name }},</p>
<p>Welcome to {{ company_name }}!</p>
```
### Seeding Templates
Run the seed script to populate default templates:
```bash
python scripts/seed_email_templates.py
```
This creates templates for:
- `signup_welcome` (en, fr, de, lb)
### Available Variables
For `signup_welcome`:
| Variable | Description |
|----------|-------------|
| first_name | User's first name |
| company_name | Vendor company name |
| email | User's email address |
| vendor_code | Vendor code for dashboard URL |
| login_url | Direct link to dashboard |
| trial_days | Number of trial days |
| tier_name | Subscription tier name |
## Provider Setup
### SMTP
Standard SMTP configuration:
```env
EMAIL_PROVIDER=smtp
SMTP_HOST=smtp.gmail.com
SMTP_PORT=587
SMTP_USER=your-email@gmail.com
SMTP_PASSWORD=your-app-password
SMTP_USE_TLS=true
```
### SendGrid
1. Create account at [sendgrid.com](https://sendgrid.com)
2. Generate API key in Settings > API Keys
3. Configure:
```env
EMAIL_PROVIDER=sendgrid
SENDGRID_API_KEY=SG.xxxxxxxxxxxxxxxxxxxx
```
4. Install package: `pip install sendgrid`
### Mailgun
1. Create account at [mailgun.com](https://mailgun.com)
2. Add and verify your domain
3. Get API key from Domain Settings
4. Configure:
```env
EMAIL_PROVIDER=mailgun
MAILGUN_API_KEY=key-xxxxxxxxxxxxxxxx
MAILGUN_DOMAIN=mg.yourdomain.com
```
### Amazon SES
1. Set up SES in AWS Console
2. Verify sender domain/email
3. Create IAM user with SES permissions
4. Configure:
```env
EMAIL_PROVIDER=ses
AWS_ACCESS_KEY_ID=AKIAXXXXXXXXXXXXXXXX
AWS_SECRET_ACCESS_KEY=xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
AWS_REGION=eu-west-1
```
5. Install package: `pip install boto3`
## Email Logging
All emails are logged to the `email_logs` table. Query examples:
```python
# Get failed emails
failed = db.query(EmailLog).filter(
EmailLog.status == EmailStatus.FAILED.value
).all()
# Get emails for a vendor
vendor_emails = db.query(EmailLog).filter(
EmailLog.vendor_id == vendor_id
).order_by(EmailLog.created_at.desc()).all()
# Get recent signup emails
signups = db.query(EmailLog).filter(
EmailLog.template_code == "signup_welcome",
EmailLog.created_at >= datetime.now() - timedelta(days=7)
).all()
```
## Language Fallback
The system automatically falls back to English if a template isn't available in the requested language:
1. Request template for "de" (German)
2. If not found, try "en" (English)
3. If still not found, return None (log error)
## Testing
Run email service tests:
```bash
pytest tests/unit/services/test_email_service.py -v
```
Test coverage includes:
- Provider abstraction (Debug, SMTP, etc.)
- Template rendering with Jinja2
- Language fallback behavior
- Email sending success/failure
- EmailLog model methods
- Template variable handling
## Architecture
```
app/services/email_service.py # Email service with provider abstraction
models/database/email.py # EmailTemplate and EmailLog models
app/core/config.py # Email configuration settings
scripts/seed_email_templates.py # Template seeding script
```
### Provider Abstraction
The system uses a strategy pattern for email providers:
```
EmailProvider (ABC)
├── SMTPProvider
├── SendGridProvider
├── MailgunProvider
├── SESProvider
└── DebugProvider
```
Each provider implements the `send()` method with the same signature, making it easy to switch providers via configuration.
## Future Enhancements
Planned improvements:
1. **Email Queue**: Background task queue for high-volume sending
2. **Webhook Tracking**: Track deliveries, opens, clicks via provider webhooks
3. **Template Editor**: Admin UI for editing templates
4. **A/B Testing**: Test different email versions
5. **Scheduled Emails**: Send emails at specific times

View File

@@ -201,6 +201,7 @@ nav:
- Platform Homepage: features/platform-homepage.md - Platform Homepage: features/platform-homepage.md
- Vendor Landing Pages: features/vendor-landing-pages.md - Vendor Landing Pages: features/vendor-landing-pages.md
- Subscription & Billing: features/subscription-billing.md - Subscription & Billing: features/subscription-billing.md
- Email System: features/email-system.md
# --- User Guides --- # --- User Guides ---
- User Guides: - User Guides:

View File

@@ -18,6 +18,7 @@ from .base import Base
from .company import Company from .company import Company
from .content_page import ContentPage from .content_page import ContentPage
from .customer import Customer, CustomerAddress from .customer import Customer, CustomerAddress
from .email import EmailCategory, EmailLog, EmailStatus, EmailTemplate
from .inventory import Inventory from .inventory import Inventory
from .invoice import ( from .invoice import (
Invoice, Invoice,
@@ -101,6 +102,11 @@ __all__ = [
# Customer # Customer
"Customer", "Customer",
"CustomerAddress", "CustomerAddress",
# Email
"EmailCategory",
"EmailLog",
"EmailStatus",
"EmailTemplate",
# Product - Enums # Product - Enums
"ProductType", "ProductType",
"DigitalDeliveryMethod", "DigitalDeliveryMethod",

192
models/database/email.py Normal file
View File

@@ -0,0 +1,192 @@
# models/database/email.py
"""
Email system database models.
Provides:
- EmailTemplate: Multi-language email templates stored in database
- EmailLog: Email sending history and tracking
"""
import enum
from datetime import datetime
from sqlalchemy import (
Boolean,
Column,
DateTime,
Enum,
ForeignKey,
Integer,
String,
Text,
)
from sqlalchemy.orm import relationship
from app.core.database import Base
from .base import TimestampMixin
class EmailCategory(str, enum.Enum):
"""Email template categories."""
AUTH = "auth" # signup, password reset, verification
ORDERS = "orders" # order confirmations, shipping
BILLING = "billing" # invoices, payment failures
SYSTEM = "system" # team invites, notifications
MARKETING = "marketing" # newsletters, promotions
class EmailStatus(str, enum.Enum):
"""Email sending status."""
PENDING = "pending"
SENT = "sent"
FAILED = "failed"
BOUNCED = "bounced"
DELIVERED = "delivered"
OPENED = "opened"
CLICKED = "clicked"
class EmailTemplate(Base, TimestampMixin):
"""
Multi-language email templates.
Templates use Jinja2 syntax for variable interpolation.
Each template can have multiple language versions.
"""
__tablename__ = "email_templates"
id = Column(Integer, primary_key=True, index=True)
# Template identification
code = Column(String(100), nullable=False, index=True) # e.g., "signup_welcome"
language = Column(String(5), nullable=False, default="en") # e.g., "en", "fr", "de", "lb"
# Template metadata
name = Column(String(255), nullable=False) # Human-readable name
description = Column(Text, nullable=True) # Template purpose description
category = Column(
String(50), default=EmailCategory.SYSTEM.value, nullable=False, index=True
)
# Email content
subject = Column(String(500), nullable=False) # Subject line (supports variables)
body_html = Column(Text, nullable=False) # HTML body
body_text = Column(Text, nullable=True) # Plain text fallback
# Template variables (JSON list of expected variables)
# e.g., ["first_name", "company_name", "login_url"]
variables = Column(Text, nullable=True)
# Status
is_active = Column(Boolean, default=True, nullable=False)
# Unique constraint: one template per code+language
__table_args__ = (
{"sqlite_autoincrement": True},
)
def __repr__(self):
return f"<EmailTemplate(code='{self.code}', language='{self.language}')>"
@property
def variables_list(self) -> list[str]:
"""Parse variables JSON to list."""
import json
if not self.variables:
return []
try:
return json.loads(self.variables)
except (json.JSONDecodeError, TypeError):
return []
class EmailLog(Base, TimestampMixin):
"""
Email sending history and tracking.
Logs all sent emails for debugging, analytics, and compliance.
"""
__tablename__ = "email_logs"
id = Column(Integer, primary_key=True, index=True)
# Template reference
template_code = Column(String(100), nullable=True, index=True)
template_id = Column(Integer, ForeignKey("email_templates.id"), nullable=True)
# Recipient info
recipient_email = Column(String(255), nullable=False, index=True)
recipient_name = Column(String(255), nullable=True)
# Email content (snapshot at send time)
subject = Column(String(500), nullable=False)
body_html = Column(Text, nullable=True)
body_text = Column(Text, nullable=True)
# Sending info
from_email = Column(String(255), nullable=False)
from_name = Column(String(255), nullable=True)
reply_to = Column(String(255), nullable=True)
# Status tracking
status = Column(
String(20), default=EmailStatus.PENDING.value, nullable=False, index=True
)
sent_at = Column(DateTime, nullable=True)
delivered_at = Column(DateTime, nullable=True)
opened_at = Column(DateTime, nullable=True)
clicked_at = Column(DateTime, nullable=True)
# Error handling
error_message = Column(Text, nullable=True)
retry_count = Column(Integer, default=0, nullable=False)
# Provider info
provider = Column(String(50), nullable=True) # smtp, sendgrid, mailgun, ses
provider_message_id = Column(String(255), nullable=True, index=True)
# Context linking (optional - link to related entities)
vendor_id = Column(Integer, ForeignKey("vendors.id"), nullable=True, index=True)
user_id = Column(Integer, ForeignKey("users.id"), nullable=True, index=True)
related_type = Column(String(50), nullable=True) # e.g., "order", "subscription"
related_id = Column(Integer, nullable=True)
# Extra data (JSON for additional context)
extra_data = Column(Text, nullable=True)
# Relationships
template = relationship("EmailTemplate", foreign_keys=[template_id])
vendor = relationship("Vendor", foreign_keys=[vendor_id])
user = relationship("User", foreign_keys=[user_id])
def __repr__(self):
return f"<EmailLog(id={self.id}, recipient='{self.recipient_email}', status='{self.status}')>"
def mark_sent(self, provider_message_id: str | None = None):
"""Mark email as sent."""
self.status = EmailStatus.SENT.value
self.sent_at = datetime.utcnow()
if provider_message_id:
self.provider_message_id = provider_message_id
def mark_failed(self, error_message: str):
"""Mark email as failed."""
self.status = EmailStatus.FAILED.value
self.error_message = error_message
self.retry_count += 1
def mark_delivered(self):
"""Mark email as delivered."""
self.status = EmailStatus.DELIVERED.value
self.delivered_at = datetime.utcnow()
def mark_opened(self):
"""Mark email as opened."""
self.status = EmailStatus.OPENED.value
self.opened_at = datetime.utcnow()

View File

@@ -0,0 +1,418 @@
#!/usr/bin/env python3
"""
Seed default email templates.
Run: python scripts/seed_email_templates.py
"""
import json
import sys
from pathlib import Path
# Add project root to path
sys.path.insert(0, str(Path(__file__).parent.parent))
from app.core.database import get_db
from models.database.email import EmailCategory, EmailTemplate
# =============================================================================
# EMAIL TEMPLATES
# =============================================================================
TEMPLATES = [
# -------------------------------------------------------------------------
# SIGNUP WELCOME
# -------------------------------------------------------------------------
{
"code": "signup_welcome",
"language": "en",
"name": "Signup Welcome",
"description": "Sent to new vendors after successful signup",
"category": EmailCategory.AUTH.value,
"variables": json.dumps([
"first_name", "company_name", "email", "vendor_code",
"login_url", "trial_days", "tier_name"
]),
"subject": "Welcome to Wizamart, {{ first_name }}!",
"body_html": """<!DOCTYPE html>
<html>
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
</head>
<body style="font-family: Arial, sans-serif; line-height: 1.6; color: #333; max-width: 600px; margin: 0 auto; padding: 20px;">
<div style="background: linear-gradient(135deg, #6366f1 0%, #8b5cf6 100%); padding: 30px; border-radius: 10px 10px 0 0;">
<h1 style="color: white; margin: 0; font-size: 28px;">Welcome to Wizamart!</h1>
</div>
<div style="background: #f9fafb; padding: 30px; border-radius: 0 0 10px 10px;">
<p style="font-size: 16px;">Hi {{ first_name }},</p>
<p>Thank you for signing up for Wizamart! Your account for <strong>{{ company_name }}</strong> is now active.</p>
<div style="background: white; border-radius: 8px; padding: 20px; margin: 20px 0; border-left: 4px solid #6366f1;">
<h3 style="margin-top: 0; color: #6366f1;">Your Account Details</h3>
<p style="margin: 5px 0;"><strong>Vendor Code:</strong> {{ vendor_code }}</p>
<p style="margin: 5px 0;"><strong>Plan:</strong> {{ tier_name }}</p>
<p style="margin: 5px 0;"><strong>Trial Period:</strong> {{ trial_days }} days free</p>
</div>
<p>You can start managing your orders, inventory, and invoices right away:</p>
<div style="text-align: center; margin: 30px 0;">
<a href="{{ login_url }}" style="background: #6366f1; color: white; padding: 14px 28px; text-decoration: none; border-radius: 8px; font-weight: bold; display: inline-block;">
Go to Dashboard
</a>
</div>
<h3 style="color: #374151;">Getting Started</h3>
<ol style="color: #4b5563;">
<li>Complete your company profile</li>
<li>Connect your Letzshop API credentials</li>
<li>Import your products</li>
<li>Start syncing orders!</li>
</ol>
<p style="color: #6b7280; font-size: 14px; margin-top: 30px;">
If you have any questions, just reply to this email or visit our help center.
</p>
<p>Best regards,<br><strong>The Wizamart Team</strong></p>
</div>
<div style="text-align: center; padding: 20px; color: #9ca3af; font-size: 12px;">
<p>&copy; 2024 Wizamart. Built for Luxembourg e-commerce.</p>
</div>
</body>
</html>""",
"body_text": """Welcome to Wizamart!
Hi {{ first_name }},
Thank you for signing up for Wizamart! Your account for {{ company_name }} is now active.
Your Account Details:
- Vendor Code: {{ vendor_code }}
- Plan: {{ tier_name }}
- Trial Period: {{ trial_days }} days free
You can start managing your orders, inventory, and invoices right away.
Go to Dashboard: {{ login_url }}
Getting Started:
1. Complete your company profile
2. Connect your Letzshop API credentials
3. Import your products
4. Start syncing orders!
If you have any questions, just reply to this email.
Best regards,
The Wizamart Team
""",
},
{
"code": "signup_welcome",
"language": "fr",
"name": "Bienvenue après inscription",
"description": "Envoyé aux nouveaux vendeurs après inscription",
"category": EmailCategory.AUTH.value,
"variables": json.dumps([
"first_name", "company_name", "email", "vendor_code",
"login_url", "trial_days", "tier_name"
]),
"subject": "Bienvenue sur Wizamart, {{ first_name }} !",
"body_html": """<!DOCTYPE html>
<html>
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
</head>
<body style="font-family: Arial, sans-serif; line-height: 1.6; color: #333; max-width: 600px; margin: 0 auto; padding: 20px;">
<div style="background: linear-gradient(135deg, #6366f1 0%, #8b5cf6 100%); padding: 30px; border-radius: 10px 10px 0 0;">
<h1 style="color: white; margin: 0; font-size: 28px;">Bienvenue sur Wizamart !</h1>
</div>
<div style="background: #f9fafb; padding: 30px; border-radius: 0 0 10px 10px;">
<p style="font-size: 16px;">Bonjour {{ first_name }},</p>
<p>Merci de vous être inscrit sur Wizamart ! Votre compte pour <strong>{{ company_name }}</strong> est maintenant actif.</p>
<div style="background: white; border-radius: 8px; padding: 20px; margin: 20px 0; border-left: 4px solid #6366f1;">
<h3 style="margin-top: 0; color: #6366f1;">Détails de votre compte</h3>
<p style="margin: 5px 0;"><strong>Code vendeur :</strong> {{ vendor_code }}</p>
<p style="margin: 5px 0;"><strong>Forfait :</strong> {{ tier_name }}</p>
<p style="margin: 5px 0;"><strong>Période d'essai :</strong> {{ trial_days }} jours gratuits</p>
</div>
<p>Vous pouvez commencer à gérer vos commandes, stocks et factures dès maintenant :</p>
<div style="text-align: center; margin: 30px 0;">
<a href="{{ login_url }}" style="background: #6366f1; color: white; padding: 14px 28px; text-decoration: none; border-radius: 8px; font-weight: bold; display: inline-block;">
Accéder au tableau de bord
</a>
</div>
<h3 style="color: #374151;">Pour commencer</h3>
<ol style="color: #4b5563;">
<li>Complétez votre profil d'entreprise</li>
<li>Connectez vos identifiants API Letzshop</li>
<li>Importez vos produits</li>
<li>Commencez à synchroniser vos commandes !</li>
</ol>
<p style="color: #6b7280; font-size: 14px; margin-top: 30px;">
Si vous avez des questions, répondez simplement à cet email.
</p>
<p>Cordialement,<br><strong>L'équipe Wizamart</strong></p>
</div>
<div style="text-align: center; padding: 20px; color: #9ca3af; font-size: 12px;">
<p>&copy; 2024 Wizamart. Conçu pour le e-commerce luxembourgeois.</p>
</div>
</body>
</html>""",
"body_text": """Bienvenue sur Wizamart !
Bonjour {{ first_name }},
Merci de vous être inscrit sur Wizamart ! Votre compte pour {{ company_name }} est maintenant actif.
Détails de votre compte :
- Code vendeur : {{ vendor_code }}
- Forfait : {{ tier_name }}
- Période d'essai : {{ trial_days }} jours gratuits
Accéder au tableau de bord : {{ login_url }}
Pour commencer :
1. Complétez votre profil d'entreprise
2. Connectez vos identifiants API Letzshop
3. Importez vos produits
4. Commencez à synchroniser vos commandes !
Cordialement,
L'équipe Wizamart
""",
},
{
"code": "signup_welcome",
"language": "de",
"name": "Willkommen nach Anmeldung",
"description": "An neue Verkäufer nach erfolgreicher Anmeldung gesendet",
"category": EmailCategory.AUTH.value,
"variables": json.dumps([
"first_name", "company_name", "email", "vendor_code",
"login_url", "trial_days", "tier_name"
]),
"subject": "Willkommen bei Wizamart, {{ first_name }}!",
"body_html": """<!DOCTYPE html>
<html>
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
</head>
<body style="font-family: Arial, sans-serif; line-height: 1.6; color: #333; max-width: 600px; margin: 0 auto; padding: 20px;">
<div style="background: linear-gradient(135deg, #6366f1 0%, #8b5cf6 100%); padding: 30px; border-radius: 10px 10px 0 0;">
<h1 style="color: white; margin: 0; font-size: 28px;">Willkommen bei Wizamart!</h1>
</div>
<div style="background: #f9fafb; padding: 30px; border-radius: 0 0 10px 10px;">
<p style="font-size: 16px;">Hallo {{ first_name }},</p>
<p>Vielen Dank für Ihre Anmeldung bei Wizamart! Ihr Konto für <strong>{{ company_name }}</strong> ist jetzt aktiv.</p>
<div style="background: white; border-radius: 8px; padding: 20px; margin: 20px 0; border-left: 4px solid #6366f1;">
<h3 style="margin-top: 0; color: #6366f1;">Ihre Kontodaten</h3>
<p style="margin: 5px 0;"><strong>Verkäufercode:</strong> {{ vendor_code }}</p>
<p style="margin: 5px 0;"><strong>Tarif:</strong> {{ tier_name }}</p>
<p style="margin: 5px 0;"><strong>Testzeitraum:</strong> {{ trial_days }} Tage kostenlos</p>
</div>
<p>Sie können sofort mit der Verwaltung Ihrer Bestellungen, Bestände und Rechnungen beginnen:</p>
<div style="text-align: center; margin: 30px 0;">
<a href="{{ login_url }}" style="background: #6366f1; color: white; padding: 14px 28px; text-decoration: none; border-radius: 8px; font-weight: bold; display: inline-block;">
Zum Dashboard
</a>
</div>
<h3 style="color: #374151;">Erste Schritte</h3>
<ol style="color: #4b5563;">
<li>Vervollständigen Sie Ihr Firmenprofil</li>
<li>Verbinden Sie Ihre Letzshop API-Zugangsdaten</li>
<li>Importieren Sie Ihre Produkte</li>
<li>Starten Sie die Bestellungssynchronisierung!</li>
</ol>
<p style="color: #6b7280; font-size: 14px; margin-top: 30px;">
Bei Fragen antworten Sie einfach auf diese E-Mail.
</p>
<p>Mit freundlichen Grüßen,<br><strong>Das Wizamart-Team</strong></p>
</div>
<div style="text-align: center; padding: 20px; color: #9ca3af; font-size: 12px;">
<p>&copy; 2024 Wizamart. Entwickelt für den luxemburgischen E-Commerce.</p>
</div>
</body>
</html>""",
"body_text": """Willkommen bei Wizamart!
Hallo {{ first_name }},
Vielen Dank für Ihre Anmeldung bei Wizamart! Ihr Konto für {{ company_name }} ist jetzt aktiv.
Ihre Kontodaten:
- Verkäufercode: {{ vendor_code }}
- Tarif: {{ tier_name }}
- Testzeitraum: {{ trial_days }} Tage kostenlos
Zum Dashboard: {{ login_url }}
Erste Schritte:
1. Vervollständigen Sie Ihr Firmenprofil
2. Verbinden Sie Ihre Letzshop API-Zugangsdaten
3. Importieren Sie Ihre Produkte
4. Starten Sie die Bestellungssynchronisierung!
Mit freundlichen Grüßen,
Das Wizamart-Team
""",
},
{
"code": "signup_welcome",
"language": "lb",
"name": "Wëllkomm no der Umeldung",
"description": "Un nei Verkeefer no erfollegräicher Umeldung geschéckt",
"category": EmailCategory.AUTH.value,
"variables": json.dumps([
"first_name", "company_name", "email", "vendor_code",
"login_url", "trial_days", "tier_name"
]),
"subject": "Wëllkomm op Wizamart, {{ first_name }}!",
"body_html": """<!DOCTYPE html>
<html>
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
</head>
<body style="font-family: Arial, sans-serif; line-height: 1.6; color: #333; max-width: 600px; margin: 0 auto; padding: 20px;">
<div style="background: linear-gradient(135deg, #6366f1 0%, #8b5cf6 100%); padding: 30px; border-radius: 10px 10px 0 0;">
<h1 style="color: white; margin: 0; font-size: 28px;">Wëllkomm op Wizamart!</h1>
</div>
<div style="background: #f9fafb; padding: 30px; border-radius: 0 0 10px 10px;">
<p style="font-size: 16px;">Moien {{ first_name }},</p>
<p>Merci fir d'Umeldung op Wizamart! Äre Kont fir <strong>{{ company_name }}</strong> ass elo aktiv.</p>
<div style="background: white; border-radius: 8px; padding: 20px; margin: 20px 0; border-left: 4px solid #6366f1;">
<h3 style="margin-top: 0; color: #6366f1;">Är Kontdetailer</h3>
<p style="margin: 5px 0;"><strong>Verkeefer Code:</strong> {{ vendor_code }}</p>
<p style="margin: 5px 0;"><strong>Plang:</strong> {{ tier_name }}</p>
<p style="margin: 5px 0;"><strong>Testperiod:</strong> {{ trial_days }} Deeg gratis</p>
</div>
<p>Dir kënnt direkt ufänken Är Bestellungen, Lager a Rechnungen ze verwalten:</p>
<div style="text-align: center; margin: 30px 0;">
<a href="{{ login_url }}" style="background: #6366f1; color: white; padding: 14px 28px; text-decoration: none; border-radius: 8px; font-weight: bold; display: inline-block;">
Zum Dashboard
</a>
</div>
<h3 style="color: #374151;">Fir unzefänken</h3>
<ol style="color: #4b5563;">
<li>Fëllt Äre Firmeprofil aus</li>
<li>Verbindt Är Letzshop API Zougangsdaten</li>
<li>Importéiert Är Produkter</li>
<li>Fänkt un Bestellungen ze synchroniséieren!</li>
</ol>
<p style="color: #6b7280; font-size: 14px; margin-top: 30px;">
Wann Dir Froen hutt, äntwert einfach op dës E-Mail.
</p>
<p>Mat beschte Gréiss,<br><strong>D'Wizamart Team</strong></p>
</div>
<div style="text-align: center; padding: 20px; color: #9ca3af; font-size: 12px;">
<p>&copy; 2024 Wizamart. Gemaach fir de lëtzebuergeschen E-Commerce.</p>
</div>
</body>
</html>""",
"body_text": """Wëllkomm op Wizamart!
Moien {{ first_name }},
Merci fir d'Umeldung op Wizamart! Äre Kont fir {{ company_name }} ass elo aktiv.
Är Kontdetailer:
- Verkeefer Code: {{ vendor_code }}
- Plang: {{ tier_name }}
- Testperiod: {{ trial_days }} Deeg gratis
Zum Dashboard: {{ login_url }}
Fir unzefänken:
1. Fëllt Äre Firmeprofil aus
2. Verbindt Är Letzshop API Zougangsdaten
3. Importéiert Är Produkter
4. Fänkt un Bestellungen ze synchroniséieren!
Mat beschte Gréiss,
D'Wizamart Team
""",
},
]
def seed_templates():
"""Seed email templates into database."""
db = next(get_db())
try:
created = 0
updated = 0
for template_data in TEMPLATES:
# Check if template already exists
existing = (
db.query(EmailTemplate)
.filter(
EmailTemplate.code == template_data["code"],
EmailTemplate.language == template_data["language"],
)
.first()
)
if existing:
# Update existing template
for key, value in template_data.items():
setattr(existing, key, value)
updated += 1
print(f"Updated: {template_data['code']} ({template_data['language']})")
else:
# Create new template
template = EmailTemplate(**template_data)
db.add(template)
created += 1
print(f"Created: {template_data['code']} ({template_data['language']})")
db.commit()
print(f"\nDone! Created: {created}, Updated: {updated}")
except Exception as e:
db.rollback()
print(f"Error: {e}")
raise
finally:
db.close()
if __name__ == "__main__":
seed_templates()

View File

@@ -0,0 +1,617 @@
# tests/unit/services/test_email_service.py
"""Unit tests for EmailService - email sending and template rendering."""
import json
from unittest.mock import MagicMock, patch
import pytest
from app.services.email_service import (
DebugProvider,
EmailProvider,
EmailService,
SMTPProvider,
get_provider,
)
from models.database.email import EmailCategory, EmailLog, EmailStatus, EmailTemplate
@pytest.mark.unit
@pytest.mark.email
class TestEmailProviders:
"""Test suite for email providers."""
def test_debug_provider_send(self):
"""Test DebugProvider logs instead of sending."""
provider = DebugProvider()
success, message_id, error = provider.send(
to_email="test@example.com",
to_name="Test User",
subject="Test Subject",
body_html="<h1>Hello</h1>",
body_text="Hello",
from_email="noreply@wizamart.com",
from_name="Wizamart",
)
assert success is True
assert message_id == "debug-test@example.com"
assert error is None
def test_debug_provider_with_reply_to(self):
"""Test DebugProvider with reply-to header."""
provider = DebugProvider()
success, message_id, error = provider.send(
to_email="test@example.com",
to_name="Test User",
subject="Test Subject",
body_html="<h1>Hello</h1>",
body_text=None,
from_email="noreply@wizamart.com",
from_name="Wizamart",
reply_to="support@wizamart.com",
)
assert success is True
@patch("app.services.email_service.settings")
def test_get_provider_debug_mode(self, mock_settings):
"""Test get_provider returns DebugProvider in debug mode."""
mock_settings.email_debug = True
provider = get_provider()
assert isinstance(provider, DebugProvider)
@patch("app.services.email_service.settings")
def test_get_provider_smtp(self, mock_settings):
"""Test get_provider returns SMTPProvider for smtp config."""
mock_settings.email_debug = False
mock_settings.email_provider = "smtp"
provider = get_provider()
assert isinstance(provider, SMTPProvider)
@patch("app.services.email_service.settings")
def test_get_provider_unknown_defaults_to_smtp(self, mock_settings):
"""Test get_provider defaults to SMTP for unknown providers."""
mock_settings.email_debug = False
mock_settings.email_provider = "unknown_provider"
provider = get_provider()
assert isinstance(provider, SMTPProvider)
@pytest.mark.unit
@pytest.mark.email
class TestEmailService:
"""Test suite for EmailService."""
def test_render_template_simple(self, db):
"""Test simple template rendering."""
service = EmailService(db)
result = service.render_template(
"Hello {{ name }}!", {"name": "World"}
)
assert result == "Hello World!"
def test_render_template_multiple_vars(self, db):
"""Test template rendering with multiple variables."""
service = EmailService(db)
result = service.render_template(
"Hi {{ first_name }}, your code is {{ vendor_code }}.",
{"first_name": "John", "vendor_code": "ACME"}
)
assert result == "Hi John, your code is ACME."
def test_render_template_missing_var(self, db):
"""Test template rendering with missing variable returns empty."""
service = EmailService(db)
result = service.render_template(
"Hello {{ name }}!",
{} # No name provided
)
# Jinja2 renders missing vars as empty string by default
assert "Hello" in result
def test_render_template_error_returns_original(self, db):
"""Test template rendering error returns original string."""
service = EmailService(db)
# Invalid Jinja2 syntax
template = "Hello {{ name"
result = service.render_template(template, {"name": "World"})
assert result == template
def test_get_template_not_found(self, db):
"""Test get_template returns None for non-existent template."""
service = EmailService(db)
result = service.get_template("nonexistent_template", "en")
assert result is None
def test_get_template_with_language_fallback(self, db):
"""Test get_template falls back to English."""
# Create English template only
template = EmailTemplate(
code="test_template",
language="en",
name="Test Template",
subject="Test",
body_html="<p>Test</p>",
category=EmailCategory.SYSTEM.value,
)
db.add(template)
db.commit()
service = EmailService(db)
# Request German, should fallback to English
result = service.get_template("test_template", "de")
assert result is not None
assert result.language == "en"
# Cleanup
db.delete(template)
db.commit()
def test_get_template_specific_language(self, db):
"""Test get_template returns specific language if available."""
# Create templates in both languages
template_en = EmailTemplate(
code="test_lang_template",
language="en",
name="Test Template EN",
subject="English Subject",
body_html="<p>English</p>",
category=EmailCategory.SYSTEM.value,
)
template_fr = EmailTemplate(
code="test_lang_template",
language="fr",
name="Test Template FR",
subject="French Subject",
body_html="<p>Français</p>",
category=EmailCategory.SYSTEM.value,
)
db.add(template_en)
db.add(template_fr)
db.commit()
service = EmailService(db)
# Request French
result = service.get_template("test_lang_template", "fr")
assert result is not None
assert result.language == "fr"
assert result.subject == "French Subject"
# Cleanup
db.delete(template_en)
db.delete(template_fr)
db.commit()
@pytest.mark.unit
@pytest.mark.email
class TestEmailSending:
"""Test suite for email sending functionality."""
@patch("app.services.email_service.get_provider")
@patch("app.services.email_service.settings")
def test_send_raw_success(self, mock_settings, mock_get_provider, db):
"""Test successful raw email sending."""
# Setup mocks
mock_settings.email_enabled = True
mock_settings.email_from_address = "noreply@test.com"
mock_settings.email_from_name = "Test"
mock_settings.email_reply_to = ""
mock_settings.email_provider = "smtp"
mock_provider = MagicMock()
mock_provider.send.return_value = (True, "msg-123", None)
mock_get_provider.return_value = mock_provider
service = EmailService(db)
log = service.send_raw(
to_email="user@example.com",
to_name="User",
subject="Test Subject",
body_html="<h1>Hello</h1>",
)
assert log.status == EmailStatus.SENT.value
assert log.recipient_email == "user@example.com"
assert log.subject == "Test Subject"
assert log.provider_message_id == "msg-123"
@patch("app.services.email_service.get_provider")
@patch("app.services.email_service.settings")
def test_send_raw_failure(self, mock_settings, mock_get_provider, db):
"""Test failed raw email sending."""
# Setup mocks
mock_settings.email_enabled = True
mock_settings.email_from_address = "noreply@test.com"
mock_settings.email_from_name = "Test"
mock_settings.email_reply_to = ""
mock_settings.email_provider = "smtp"
mock_provider = MagicMock()
mock_provider.send.return_value = (False, None, "Connection refused")
mock_get_provider.return_value = mock_provider
service = EmailService(db)
log = service.send_raw(
to_email="user@example.com",
subject="Test Subject",
body_html="<h1>Hello</h1>",
)
assert log.status == EmailStatus.FAILED.value
assert log.error_message == "Connection refused"
@patch("app.services.email_service.settings")
def test_send_raw_email_disabled(self, mock_settings, db):
"""Test email sending when disabled."""
mock_settings.email_enabled = False
mock_settings.email_from_address = "noreply@test.com"
mock_settings.email_from_name = "Test"
mock_settings.email_reply_to = ""
mock_settings.email_provider = "smtp"
mock_settings.email_debug = False
service = EmailService(db)
log = service.send_raw(
to_email="user@example.com",
subject="Test Subject",
body_html="<h1>Hello</h1>",
)
assert log.status == EmailStatus.FAILED.value
assert "disabled" in log.error_message.lower()
@patch("app.services.email_service.get_provider")
@patch("app.services.email_service.settings")
def test_send_template_success(self, mock_settings, mock_get_provider, db):
"""Test successful template email sending."""
# Create test template
template = EmailTemplate(
code="test_send_template",
language="en",
name="Test Send Template",
subject="Hello {{ first_name }}",
body_html="<p>Welcome {{ first_name }} to {{ company }}</p>",
body_text="Welcome {{ first_name }} to {{ company }}",
category=EmailCategory.SYSTEM.value,
)
db.add(template)
db.commit()
# Setup mocks
mock_settings.email_enabled = True
mock_settings.email_from_address = "noreply@test.com"
mock_settings.email_from_name = "Test"
mock_settings.email_reply_to = ""
mock_settings.email_provider = "smtp"
mock_provider = MagicMock()
mock_provider.send.return_value = (True, "msg-456", None)
mock_get_provider.return_value = mock_provider
service = EmailService(db)
log = service.send_template(
template_code="test_send_template",
to_email="user@example.com",
language="en",
variables={
"first_name": "John",
"company": "ACME Corp"
},
)
assert log.status == EmailStatus.SENT.value
assert log.template_code == "test_send_template"
assert log.subject == "Hello John"
# Cleanup
db.delete(template)
db.commit()
def test_send_template_not_found(self, db):
"""Test sending with non-existent template."""
service = EmailService(db)
log = service.send_template(
template_code="nonexistent_template",
to_email="user@example.com",
)
assert log.status == EmailStatus.FAILED.value
assert "not found" in log.error_message.lower()
@pytest.mark.unit
@pytest.mark.email
class TestEmailLog:
"""Test suite for EmailLog model methods."""
def test_mark_sent(self, db):
"""Test EmailLog.mark_sent method."""
log = EmailLog(
recipient_email="test@example.com",
subject="Test",
from_email="noreply@test.com",
status=EmailStatus.PENDING.value,
)
db.add(log)
db.flush()
log.mark_sent("provider-msg-id")
assert log.status == EmailStatus.SENT.value
assert log.sent_at is not None
assert log.provider_message_id == "provider-msg-id"
db.rollback()
def test_mark_failed(self, db):
"""Test EmailLog.mark_failed method."""
log = EmailLog(
recipient_email="test@example.com",
subject="Test",
from_email="noreply@test.com",
status=EmailStatus.PENDING.value,
retry_count=0,
)
db.add(log)
db.flush()
log.mark_failed("Connection timeout")
assert log.status == EmailStatus.FAILED.value
assert log.error_message == "Connection timeout"
assert log.retry_count == 1
db.rollback()
def test_mark_delivered(self, db):
"""Test EmailLog.mark_delivered method."""
log = EmailLog(
recipient_email="test@example.com",
subject="Test",
from_email="noreply@test.com",
status=EmailStatus.SENT.value,
)
db.add(log)
db.flush()
log.mark_delivered()
assert log.status == EmailStatus.DELIVERED.value
assert log.delivered_at is not None
db.rollback()
def test_mark_opened(self, db):
"""Test EmailLog.mark_opened method."""
log = EmailLog(
recipient_email="test@example.com",
subject="Test",
from_email="noreply@test.com",
status=EmailStatus.DELIVERED.value,
)
db.add(log)
db.flush()
log.mark_opened()
assert log.status == EmailStatus.OPENED.value
assert log.opened_at is not None
db.rollback()
@pytest.mark.unit
@pytest.mark.email
class TestEmailTemplate:
"""Test suite for EmailTemplate model."""
def test_variables_list_property(self, db):
"""Test EmailTemplate.variables_list property."""
template = EmailTemplate(
code="test_vars",
language="en",
name="Test",
subject="Test",
body_html="<p>Test</p>",
category=EmailCategory.SYSTEM.value,
variables=json.dumps(["first_name", "last_name", "email"]),
)
db.add(template)
db.flush()
assert template.variables_list == ["first_name", "last_name", "email"]
db.rollback()
def test_variables_list_empty(self, db):
"""Test EmailTemplate.variables_list with no variables."""
template = EmailTemplate(
code="test_no_vars",
language="en",
name="Test",
subject="Test",
body_html="<p>Test</p>",
category=EmailCategory.SYSTEM.value,
variables=None,
)
db.add(template)
db.flush()
assert template.variables_list == []
db.rollback()
def test_variables_list_invalid_json(self, db):
"""Test EmailTemplate.variables_list with invalid JSON."""
template = EmailTemplate(
code="test_invalid_json",
language="en",
name="Test",
subject="Test",
body_html="<p>Test</p>",
category=EmailCategory.SYSTEM.value,
variables="not valid json",
)
db.add(template)
db.flush()
assert template.variables_list == []
db.rollback()
def test_template_repr(self, db):
"""Test EmailTemplate string representation."""
template = EmailTemplate(
code="signup_welcome",
language="en",
name="Welcome",
subject="Welcome",
body_html="<p>Welcome</p>",
category=EmailCategory.AUTH.value,
)
assert "signup_welcome" in repr(template)
assert "en" in repr(template)
@pytest.mark.unit
@pytest.mark.email
class TestSignupWelcomeEmail:
"""Test suite for signup welcome email integration."""
@pytest.fixture
def welcome_template(self, db):
"""Create a welcome template for testing."""
import json
template = EmailTemplate(
code="signup_welcome",
language="en",
name="Signup Welcome",
subject="Welcome {{ first_name }}!",
body_html="<p>Welcome {{ first_name }} to {{ company_name }}</p>",
body_text="Welcome {{ first_name }} to {{ company_name }}",
category=EmailCategory.AUTH.value,
variables=json.dumps([
"first_name", "company_name", "email", "vendor_code",
"login_url", "trial_days", "tier_name"
]),
)
db.add(template)
db.commit()
yield template
# Cleanup
db.delete(template)
db.commit()
def test_welcome_template_rendering(self, db, welcome_template):
"""Test that welcome template renders correctly."""
service = EmailService(db)
template = service.get_template("signup_welcome", "en")
assert template is not None
assert template.code == "signup_welcome"
# Test rendering
rendered = service.render_template(
template.subject,
{"first_name": "John"}
)
assert rendered == "Welcome John!"
def test_welcome_template_has_required_variables(self, db, welcome_template):
"""Test welcome template has all required variables."""
template = (
db.query(EmailTemplate)
.filter(
EmailTemplate.code == "signup_welcome",
EmailTemplate.language == "en",
)
.first()
)
assert template is not None
required_vars = [
"first_name",
"company_name",
"vendor_code",
"login_url",
"trial_days",
"tier_name",
]
for var in required_vars:
assert var in template.variables_list, f"Missing variable: {var}"
@patch("app.services.email_service.get_provider")
@patch("app.services.email_service.settings")
def test_welcome_email_send(self, mock_settings, mock_get_provider, db, welcome_template):
"""Test sending welcome email."""
# Setup mocks
mock_settings.email_enabled = True
mock_settings.email_from_address = "noreply@test.com"
mock_settings.email_from_name = "Test"
mock_settings.email_reply_to = ""
mock_settings.email_provider = "smtp"
mock_provider = MagicMock()
mock_provider.send.return_value = (True, "welcome-msg-123", None)
mock_get_provider.return_value = mock_provider
service = EmailService(db)
log = service.send_template(
template_code="signup_welcome",
to_email="newuser@example.com",
to_name="John Doe",
language="en",
variables={
"first_name": "John",
"company_name": "ACME Corp",
"email": "newuser@example.com",
"vendor_code": "ACME",
"login_url": "https://wizamart.com/vendor/ACME/dashboard",
"trial_days": 30,
"tier_name": "Essential",
},
vendor_id=1,
user_id=1,
related_type="signup",
)
assert log.status == EmailStatus.SENT.value
assert log.template_code == "signup_welcome"
assert log.subject == "Welcome John!"
assert log.recipient_email == "newuser@example.com"