"""Send Test Email: REAL send through a connected Gmail account with provider-confirmed results (message ID from Gmail), full variable substitution and optional attachments. This is the compliance-friendly way to verify end-to-end delivery before a campaign.""" import asyncio import base64 from datetime import datetime, timezone from typing import Optional from fastapi import APIRouter, Depends, HTTPException from pydantic import BaseModel, EmailStr, Field from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.api.deps import get_current_user, get_db, require_csrf from app.core.config import settings as cfg from app.core.security import log_audit from app.models import EmailTemplate, GmailAccount, OAuthToken, User from app.services import message_builder from app.services.campaign_engine import get_setting from app.services.gmail_service import build_gmail_client_from_account from app.services.variable_engine import GeneratedValues, VariableEngine from app.db.session import SyncSession router = APIRouter(prefix="/test-send", tags=["test-send"]) class TestSendIn(BaseModel): account_id: int to_email: EmailStr subject: str = Field(min_length=1, max_length=500) template_id: Optional[int] = None body_html: str = "" body_text: str = "" include_pdf: bool = False use_random_variables: bool = True @router.post("/send") async def send_test(payload: TestSendIn, db: AsyncSession = Depends(get_db), user: User = Depends(get_current_user), _: None = Depends(require_csrf)): """REAL test send through Gmail API or SMTP XOAUTH2 using an authorized account. Returns the provider's message ID — never a fake success.""" import time t0 = time.monotonic() account = await db.get(GmailAccount, payload.account_id) if not account or account.user_id != user.id: raise HTTPException(status_code=404, detail="Gmail account not found") if account.status != "authorized" or not account.is_active: raise HTTPException(status_code=409, detail=f"Account is {account.status}; re-authorize first") if account.cooldown_until and account.cooldown_until > datetime.now(timezone.utc): raise HTTPException(status_code=409, detail="Account is in cooldown") # resolve body: template (with real variable substitution) or raw body template = None if payload.template_id: template = await db.get(EmailTemplate, payload.template_id) if not template or template.user_id != user.id: raise HTTPException(status_code=404, detail="Template not found") body_html = template.body_html if template else payload.body_html body_text = template.body_text if template else payload.body_text if not body_html and not body_text: raise HTTPException(status_code=400, detail="Provide a template or message body") # plain dict recipient for variable substitution (test only). VariableEngine # accepts dicts directly; ORM-like objects would need __table__ introspection. fr = { "email": str(payload.to_email), "first_name": "Test", "last_name": "Recipient", "company_name": await get_setting(db, "company_name", "Acme Corp") or "Acme Corp", "company_phone": "555-0100", "customer_email": str(payload.to_email), "custom_fields": {}, } engine = VariableEngine({}) g = GeneratedValues() if payload.use_random_variables: g.invoice_number = engine.generate_invoice_number(set()) g.amount = engine.generate_amount() g.shipping = engine.generate_shipping() g.invoice_date = "2025-01-15" g.due_date = "2025-01-30" unsub_url = f"{cfg.BASE_URL}/unsubscribe/test-token" res_html = engine.substitute(body_html or "", fr, g, custom_fields={}, extra={"unsubscribe_link": unsub_url, "company_phone": fr["company_phone"]}).text res_text = engine.substitute(body_text or "", fr, g, custom_fields={}, extra={"unsubscribe_link": unsub_url, "company_phone": fr["company_phone"]}).text subject = engine.substitute(payload.subject, fr, g).text # Validate AFTER substitution: bodies legitimately contain %Variables% that # are substituted per recipient above. Anything still remaining is a genuine # problem (unknown variable, structural issue) and is rejected here. problems = message_builder.validate_html(res_html) if res_html else [] problems = [p for p in problems if not p.startswith("Unsubstituted variables")] if problems: raise HTTPException(status_code=400, detail=f"HTML rejected: {'; '.join(problems)}") attachments = [] if payload.include_pdf and body_html: pdf = await asyncio.get_event_loop().run_in_executor( None, message_builder.html_to_pdf, res_html) if pdf: attachments.append({"filename": "Test_Preview.pdf", "content": pdf, "mime_type": "application/pdf"}) mime = message_builder.build_mime_message( from_email=account.email, to_email=str(payload.to_email), subject=subject, html_body=res_html, text_body=res_text, attachments=attachments, unsubscribe_url=unsub_url, ) # REAL send in a worker thread with a sync DB session (token refresh writes) def _real_send(): with SyncSession() as s: acc = s.get(GmailAccount, account.id) token = s.query(OAuthToken).filter_by(account_id=account.id).first() if not token: return {"ok": False, "error": "No OAuth token stored for this account", "error_class": "auth"} client = build_gmail_client_from_account(acc, token) if client.method == "smtp": return client.send_mime_smtp(mime) return client.send_mime_api(mime) try: result = await asyncio.get_event_loop().run_in_executor(None, _real_send) except Exception as e: await log_audit(db, user.id, "test_send.error", "gmail_account", account.id, {"error": str(e)[:200]}) return {"ok": False, "stage": "temp_failed", "detail": str(e)[:400], "error_class": "unknown", "elapsed_ms": int((time.monotonic() - t0) * 1000)} elapsed = int((time.monotonic() - t0) * 1000) if result.get("ok"): await log_audit(db, user.id, "test_send.success", "gmail_account", account.id, {"to": str(payload.to_email), "message_id": result.get("message_id")}) return {"ok": True, "stage": "accepted", "detail": "Gmail accepted the message", "message_id": result.get("message_id"), "thread_id": result.get("thread_id"), "method": client_method(account), "provider_response": _clip(str(result.get("raw_response", ""))), "elapsed_ms": elapsed} # provider refused — surface the REAL error, never fake success err = str(result.get("error", "") or result.get("detail", "")) await log_audit(db, user.id, "test_send.failed", "gmail_account", account.id, {"to": str(payload.to_email), "error": err[:200]}) return {"ok": False, "stage": "perm_failed", "detail": err[:400], "error_class": result.get("error_class", "unknown"), "method": client_method(account), "provider_response": _clip(str(result.get("raw_response", ""))), "elapsed_ms": elapsed} def _clip(s: str) -> str: return s[:300] def client_method(account: GmailAccount) -> str: return account.sending_method or "api" @router.get("/accounts") async def test_accounts(db: AsyncSession = Depends(get_db), user: User = Depends(get_current_user)): """Authorized accounts available for test sends.""" rows = (await db.execute( select(GmailAccount).where(GmailAccount.user_id == user.id, GmailAccount.status == "authorized", GmailAccount.is_active == True))).scalars().all() # noqa: E712 return {"accounts": [{"id": a.id, "email": a.email, "sending_method": a.sending_method} for a in rows]} @router.get("/templates") async def test_templates(db: AsyncSession = Depends(get_db), user: User = Depends(get_current_user)): rows = (await db.execute( select(EmailTemplate).where(EmailTemplate.user_id == user.id))).scalars().all() return {"templates": [{"id": t.id, "name": t.name} for t in rows]}