Compare commits

..

2 Commits

Author SHA1 Message Date
Roberto Musso
e9fb6102ab style: align confirmation result page with adiuvAI brand
Some checks failed
Test & Deploy Waitlist / test (push) Failing after 33s
Test & Deploy Waitlist / deploy (push) Has been skipped
2026-04-11 18:58:12 +02:00
Roberto Musso
f956f0a260 feat: Brevo double opt-in + contact sync
- Add brevo.py: transactional email sending + contact list sync via Brevo API
- Add token.py: stateless HMAC-signed confirmation tokens (no DB migration needed)
- Update routes.py: POST /waitlist sends confirmation email, GET /waitlist/confirm verifies token
- Update config.py: Brevo + confirmation settings (gracefully disabled when BREVO_API_KEY is empty)
- Update .env.example with new Brevo and confirmation variables
- Add httpx dependency
- Add 8 new tests (token roundtrip/expiry/tamper, confirm endpoint, Brevo mock)
2026-04-11 18:48:58 +02:00
7 changed files with 592 additions and 4 deletions

View File

@@ -9,3 +9,14 @@ RATE_LIMIT_PER_MINUTE=5
# Set to "production" in prod to enforce strict origin checks # Set to "production" in prod to enforce strict origin checks
ENVIRONMENT=development ENVIRONMENT=development
# Brevo (email) — leave BREVO_API_KEY empty to disable email features
BREVO_API_KEY=
BREVO_SENDER_EMAIL=noreply@adiuvai.com
BREVO_SENDER_NAME=adiuvAI
BREVO_LIST_ID=0
# Confirmation link
CONFIRM_SECRET=replace-with-a-long-random-string
CONFIRM_BASE_URL=https://waitlist.adiuvai.com
CONFIRM_TOKEN_EXPIRY_HOURS=48

224
app/brevo.py Normal file
View File

@@ -0,0 +1,224 @@
"""
Brevo (ex-Sendinblue) integration.
- Send transactional confirmation emails
- Sync confirmed contacts to a Brevo list
"""
import logging
import httpx
from app.config import settings
logger = logging.getLogger(__name__)
BREVO_API = "https://api.brevo.com/v3"
def _headers() -> dict[str, str]:
return {
"api-key": settings.BREVO_API_KEY,
"Content-Type": "application/json",
"Accept": "application/json",
}
async def send_confirmation_email(email: str, confirm_url: str) -> bool:
"""Send a double opt-in confirmation email. Returns True on success."""
if not settings.brevo_configured:
logger.warning("Brevo not configured — skipping confirmation email for %s***", email[:3])
return False
payload = {
"sender": {
"name": settings.BREVO_SENDER_NAME,
"email": settings.BREVO_SENDER_EMAIL,
},
"to": [{"email": email}],
"subject": "Confirm your spot on the adiuvAI waitlist",
"htmlContent": _confirmation_html(confirm_url),
}
try:
async with httpx.AsyncClient(timeout=10) as client:
resp = await client.post(f"{BREVO_API}/smtp/email", headers=_headers(), json=payload)
resp.raise_for_status()
logger.info("Confirmation email sent to %s***", email[:3])
return True
except httpx.HTTPError:
logger.exception("Failed to send confirmation email to %s***", email[:3])
return False
async def add_contact_to_list(email: str) -> bool:
"""Add a confirmed contact to the Brevo waitlist list. Returns True on success."""
if not settings.brevo_configured:
logger.warning("Brevo not configured — skipping contact sync for %s***", email[:3])
return False
if settings.BREVO_LIST_ID == 0:
logger.warning("BREVO_LIST_ID not set — skipping contact sync")
return False
payload = {
"email": email,
"listIds": [settings.BREVO_LIST_ID],
"updateEnabled": True,
}
try:
async with httpx.AsyncClient(timeout=10) as client:
resp = await client.post(f"{BREVO_API}/contacts", headers=_headers(), json=payload)
resp.raise_for_status()
logger.info("Contact synced to Brevo list %d: %s***", settings.BREVO_LIST_ID, email[:3])
return True
except httpx.HTTPError:
logger.exception("Failed to sync contact to Brevo: %s***", email[:3])
return False
def _confirmation_html(confirm_url: str) -> str:
"""Email template aligned with the adiuvAI landing page brand."""
return f"""\
<!DOCTYPE html>
<html lang="en" xmlns="http://www.w3.org/1999/xhtml">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width,initial-scale=1">
<meta http-equiv="X-UA-Compatible" content="IE=edge">
<meta name="color-scheme" content="light">
<meta name="supported-color-schemes" content="light">
<title>Confirm your spot — adiuvAI</title>
<!--[if mso]><style>table,td{{font-family:Arial,sans-serif!important;}}</style><![endif]-->
</head>
<body style="margin:0;padding:0;width:100%;background-color:#f4edf3;
font-family:-apple-system,BlinkMacSystemFont,'Segoe UI',Roboto,'Helvetica Neue',Arial,sans-serif;
-webkit-font-smoothing:antialiased;-moz-osx-font-smoothing:grayscale;">
<!-- Preheader (hidden text for inbox preview) -->
<div style="display:none;font-size:1px;color:#f4edf3;line-height:1px;max-height:0;max-width:0;opacity:0;overflow:hidden;">
Confirm your email to secure your early access spot on the adiuvAI waitlist.
</div>
<!-- Outer wrapper -->
<table role="presentation" width="100%" cellpadding="0" cellspacing="0" border="0"
style="background-color:#f4edf3;">
<tr><td align="center" style="padding:48px 16px;">
<!-- Card -->
<table role="presentation" width="520" cellpadding="0" cellspacing="0" border="0"
style="max-width:520px;width:100%;background-color:#ffffff;
border-radius:20px;overflow:hidden;
box-shadow:0 4px 16px rgba(0,0,0,0.03),0 1px 2px rgba(0,0,0,0.02);">
<!-- Gold accent bar -->
<tr><td style="height:4px;background:linear-gradient(135deg,#fbc881 0%,#e5a94e 100%);
font-size:0;line-height:0;">&nbsp;</td></tr>
<!-- Logo row -->
<tr><td align="center" style="padding:36px 40px 0;">
<table role="presentation" cellpadding="0" cellspacing="0" border="0">
<tr>
<!-- Compass SVG (inline, matching the website nav-logo) -->
<td style="padding-right:8px;vertical-align:middle;">
<!--[if mso]>
<v:group style="width:28px;height:28px;" coordsize="64,64">
<v:shape style="width:64;height:64;" path="M32,4 L48,32 L16,32 Z" fillcolor="#fbc881" stroked="f"/>
<v:shape style="width:64;height:64;" path="M16,32 L48,32 L32,60 Z" fillcolor="#040404" stroked="f"/>
</v:group>
<![endif]-->
<!--[if !mso]><!-->
<svg viewBox="0 0 64 64" width="28" height="28" xmlns="http://www.w3.org/2000/svg"
style="display:block;width:28px;height:28px;">
<path d="M32,4 L48,32 L16,32 Z" fill="#fbc881"/>
<path d="M16,32 L48,32 L32,60 Z" fill="#040404"/>
<circle cx="32" cy="32" r="2.5" fill="#040404" opacity="0.18"/>
</svg>
<!--<![endif]-->
</td>
<td style="vertical-align:middle;font-size:18px;font-weight:400;
color:#040404;letter-spacing:-0.02em;">
adiuv<span style="font-weight:700;color:#e5a94e;">AI</span>
</td>
</tr>
</table>
</td></tr>
<!-- Body -->
<tr><td style="padding:32px 40px 0;">
<!-- Badge -->
<table role="presentation" cellpadding="0" cellspacing="0" border="0"
style="margin:0 auto 24px;">
<tr><td style="padding:5px 14px;border-radius:50px;
background-color:rgba(251,200,129,0.12);
border:1px solid rgba(251,200,129,0.2);
font-size:11px;font-weight:600;letter-spacing:0.06em;
text-transform:uppercase;color:#e5a94e;">
&#9679;&ensp;One more step
</td></tr>
</table>
<!-- Heading -->
<h1 style="margin:0 0 16px;font-size:24px;font-weight:600;
letter-spacing:-0.03em;line-height:1.2;color:#040404;text-align:center;">
Confirm your spot on<br>the waitlist
</h1>
<!-- Paragraph -->
<p style="margin:0 0 28px;font-size:15px;line-height:1.7;color:#8a8ea9;
text-align:center;">
Thanks for signing up! Please confirm your email address
so we can keep you in the loop when adiuvAI launches.
</p>
<!-- CTA Button -->
<table role="presentation" cellpadding="0" cellspacing="0" border="0"
style="margin:0 auto;">
<tr><td align="center" style="border-radius:50px;background-color:#040404;">
<a href="{confirm_url}"
target="_blank"
style="display:inline-block;padding:14px 36px;
font-size:15px;font-weight:600;
color:#fbfbfb;text-decoration:none;
border-radius:50px;
font-family:-apple-system,BlinkMacSystemFont,'Segoe UI',Roboto,sans-serif;">
Confirm my email
</a>
</td></tr>
</table>
<!-- Expiry note -->
<p style="margin:20px 0 0;font-size:13px;color:#c8c3cd;text-align:center;">
This link expires in {settings.CONFIRM_TOKEN_EXPIRY_HOURS} hours.
</p>
</td></tr>
<!-- Divider -->
<tr><td style="padding:32px 40px 0;">
<table role="presentation" width="100%" cellpadding="0" cellspacing="0" border="0">
<tr><td style="height:1px;background-color:#f4edf3;font-size:0;line-height:0;">&nbsp;</td></tr>
</table>
</td></tr>
<!-- Footer -->
<tr><td style="padding:20px 40px 32px;text-align:center;">
<p style="margin:0;font-size:12px;color:#c8c3cd;line-height:1.5;">
If you didn't sign up, simply ignore this email.
</p>
<p style="margin:8px 0 0;font-size:12px;color:#c8c3cd;">
<a href="https://adiuvai.com" style="color:#8a8ea9;text-decoration:underline;
text-underline-offset:2px;">adiuvai.com</a>
</p>
</td></tr>
</table>
<!-- /Card -->
</td></tr>
</table>
</body>
</html>"""

View File

@@ -1,3 +1,5 @@
import secrets
from pydantic_settings import BaseSettings from pydantic_settings import BaseSettings
@@ -7,6 +9,17 @@ class Settings(BaseSettings):
RATE_LIMIT_PER_MINUTE: int = 5 RATE_LIMIT_PER_MINUTE: int = 5
ENVIRONMENT: str = "development" ENVIRONMENT: str = "development"
# Brevo (email)
BREVO_API_KEY: str = ""
BREVO_SENDER_EMAIL: str = "noreply@adiuvai.com"
BREVO_SENDER_NAME: str = "adiuvAI"
BREVO_LIST_ID: int = 0 # Brevo contact list ID for waitlist subscribers
# Confirmation link
CONFIRM_SECRET: str = secrets.token_hex(32) # override in production .env
CONFIRM_BASE_URL: str = "https://waitlist.adiuvai.com"
CONFIRM_TOKEN_EXPIRY_HOURS: int = 48
model_config = {"env_file": ".env", "env_file_encoding": "utf-8"} model_config = {"env_file": ".env", "env_file_encoding": "utf-8"}
@property @property
@@ -17,5 +30,9 @@ class Settings(BaseSettings):
def sync_database_url(self) -> str: def sync_database_url(self) -> str:
return self.DATABASE_URL.replace("+asyncpg", "+psycopg2") return self.DATABASE_URL.replace("+asyncpg", "+psycopg2")
@property
def brevo_configured(self) -> bool:
return bool(self.BREVO_API_KEY)
settings = Settings() settings = Settings()

View File

@@ -1,13 +1,19 @@
import asyncio
import logging import logging
from urllib.parse import urlencode
from fastapi import APIRouter, Depends, Request from fastapi import APIRouter, Depends, Request
from fastapi.responses import HTMLResponse
from sqlalchemy import select from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.ext.asyncio import AsyncSession
from app.brevo import add_contact_to_list, send_confirmation_email
from app.config import settings
from app.db import get_db from app.db import get_db
from app.rate_limit import _get_client_ip from app.rate_limit import _get_client_ip
from app.schemas import WaitlistRequest, WaitlistResponse from app.schemas import WaitlistRequest, WaitlistResponse
from app.models import WaitlistEntry from app.models import WaitlistEntry
from app.token import generate_token, verify_token
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
router = APIRouter() router = APIRouter()
@@ -25,6 +31,7 @@ async def join_waitlist(
- Honeypot: if `website` field is non-empty, silently succeed (bot trap). - Honeypot: if `website` field is non-empty, silently succeed (bot trap).
- Duplicate emails: idempotent — returns success without error. - Duplicate emails: idempotent — returns success without error.
- Stores the Cloudflare-resolved client IP for analytics (not exposed). - Stores the Cloudflare-resolved client IP for analytics (not exposed).
- Sends a confirmation email via Brevo (fire-and-forget).
""" """
# Honeypot — bots fill hidden fields; silently "succeed" # Honeypot — bots fill hidden fields; silently "succeed"
if body.website: if body.website:
@@ -35,14 +42,165 @@ async def join_waitlist(
# Check for existing entry — idempotent # Check for existing entry — idempotent
existing = await db.execute( existing = await db.execute(
select(WaitlistEntry.id).where(WaitlistEntry.email == email) select(WaitlistEntry).where(WaitlistEntry.email == email)
) )
if existing.scalar_one_or_none() is not None: entry = existing.scalar_one_or_none()
if entry is not None:
# Re-send confirmation if not yet confirmed
if not entry.confirmed and settings.brevo_configured:
token = generate_token(email)
confirm_url = f"{settings.CONFIRM_BASE_URL}/api/v1/waitlist/confirm?token={token}"
asyncio.create_task(send_confirmation_email(email, confirm_url))
return WaitlistResponse() return WaitlistResponse()
entry = WaitlistEntry(email=email, ip_address=ip, source="website") entry = WaitlistEntry(email=email, ip_address=ip, source="website")
db.add(entry) db.add(entry)
await db.commit() await db.commit()
logger.info("New waitlist signup: %s", email[:3] + "***") logger.info("New waitlist signup: %s***", email[:3])
# Fire-and-forget: send confirmation email via Brevo
if settings.brevo_configured:
token = generate_token(email)
confirm_url = f"{settings.CONFIRM_BASE_URL}/api/v1/waitlist/confirm?token={token}"
asyncio.create_task(send_confirmation_email(email, confirm_url))
return WaitlistResponse() return WaitlistResponse()
@router.get("/waitlist/confirm", response_class=HTMLResponse)
async def confirm_email(
token: str,
db: AsyncSession = Depends(get_db),
) -> HTMLResponse:
"""
Double opt-in confirmation endpoint.
Verifies the HMAC token, marks the entry as confirmed,
and syncs the contact to Brevo.
"""
email = verify_token(token)
if email is None:
return HTMLResponse(content=_result_page(success=False), status_code=400)
result = await db.execute(
select(WaitlistEntry).where(WaitlistEntry.email == email)
)
entry = result.scalar_one_or_none()
if entry is None:
return HTMLResponse(content=_result_page(success=False), status_code=400)
if not entry.confirmed:
entry.confirmed = True
await db.commit()
logger.info("Email confirmed: %s***", email[:3])
# Sync confirmed contact to Brevo list (fire-and-forget)
if settings.brevo_configured:
asyncio.create_task(add_contact_to_list(email))
return HTMLResponse(content=_result_page(success=True))
def _result_page(*, success: bool) -> str:
"""Branded HTML response for confirmation result, matching the adiuvAI landing page."""
if success:
title = "You&#8217;re confirmed!"
message = "Your email has been verified. We&#8217;ll notify you when adiuvAI is ready."
icon_bg = "rgba(251,200,129,0.12)"
icon_border = "rgba(251,200,129,0.25)"
icon_svg = '<svg viewBox="0 0 24 24" width="24" height="24" fill="none" stroke="#e5a94e" stroke-width="2.5" stroke-linecap="round" stroke-linejoin="round"><polyline points="20 6 9 17 4 12"/></svg>'
else:
title = "Invalid or expired link"
message = "This confirmation link is no longer valid. Please sign up again."
icon_bg = "rgba(220,38,38,0.08)"
icon_border = "rgba(220,38,38,0.18)"
icon_svg = '<svg viewBox="0 0 24 24" width="24" height="24" fill="none" stroke="#dc2626" stroke-width="2.5" stroke-linecap="round" stroke-linejoin="round"><line x1="18" y1="6" x2="6" y2="18"/><line x1="6" y1="6" x2="18" y2="18"/></svg>'
return f"""\
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width,initial-scale=1">
<title>{title} — adiuvAI</title>
<style>
body {{
margin: 0; padding: 0; min-height: 100vh;
display: flex; align-items: center; justify-content: center;
background: #f4edf3;
font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, sans-serif;
-webkit-font-smoothing: antialiased;
}}
body::before {{
content: '';
position: fixed; inset: 0;
background:
radial-gradient(ellipse 80% 60% at 25% 20%, rgba(251,200,129,0.10) 0%, transparent 60%),
radial-gradient(ellipse 70% 50% at 75% 70%, rgba(138,142,169,0.08) 0%, transparent 50%);
pointer-events: none; z-index: 0;
}}
.card {{
position: relative; z-index: 1;
text-align: center; max-width: 420px; width: 100%;
margin: 24px;
padding: 48px 40px 40px;
background: rgba(255,255,255,0.55);
backdrop-filter: blur(20px); -webkit-backdrop-filter: blur(20px);
border: 1px solid rgba(255,255,255,0.65);
border-radius: 20px;
box-shadow: 0 4px 16px rgba(0,0,0,0.03), 0 1px 2px rgba(0,0,0,0.02);
}}
.logo {{
display: inline-flex; align-items: center; gap: 8px;
font-size: 18px; font-weight: 400; color: #040404;
letter-spacing: -0.02em; margin-bottom: 28px;
}}
.logo .ai {{ font-weight: 700; color: #e5a94e; }}
.icon {{
width: 56px; height: 56px; border-radius: 50%;
background: {icon_bg}; border: 1.5px solid {icon_border};
display: flex; align-items: center; justify-content: center;
margin: 0 auto 20px;
}}
h1 {{
font-size: 22px; font-weight: 600; color: #040404;
letter-spacing: -0.03em; margin: 0 0 10px;
}}
p {{
font-size: 15px; color: #8a8ea9; line-height: 1.7;
margin: 0 0 28px;
}}
.btn {{
display: inline-block;
padding: 12px 28px; border-radius: 50px;
background: #040404; color: #fbfbfb;
text-decoration: none;
font-size: 14px; font-weight: 600;
transition: all 0.4s cubic-bezier(0.16, 1, 0.3, 1);
}}
.btn:hover {{
background: #222;
transform: translateY(-1px);
box-shadow: 0 8px 24px rgba(0,0,0,0.12);
}}
</style>
</head>
<body>
<div class="card">
<div class="logo">
<svg viewBox="0 0 64 64" width="28" height="28">
<path d="M32,4 L48,32 L16,32 Z" fill="#fbc881"/>
<path d="M16,32 L48,32 L32,60 Z" fill="#040404"/>
<circle cx="32" cy="32" r="2.5" fill="#040404" opacity="0.18"/>
</svg>
<span>adiuv<span class="ai">AI</span></span>
</div>
<div class="icon">{icon_svg}</div>
<h1>{title}</h1>
<p>{message}</p>
<a href="https://adiuvai.com" class="btn">Go to adiuvai.com</a>
</div>
</body>
</html>"""

60
app/token.py Normal file
View File

@@ -0,0 +1,60 @@
"""
Stateless HMAC-signed confirmation tokens.
Token format: base64url( email : unix_timestamp : hmac_sha256 )
No database storage needed — the signature proves authenticity.
"""
import base64
import hashlib
import hmac
import time
from app.config import settings
def generate_token(email: str) -> str:
"""Create a URL-safe confirmation token for the given email."""
timestamp = str(int(time.time()))
payload = f"{email}:{timestamp}"
sig = hmac.new(
settings.CONFIRM_SECRET.encode(),
payload.encode(),
hashlib.sha256,
).hexdigest()
return base64.urlsafe_b64encode(f"{payload}:{sig}".encode()).decode()
def verify_token(token: str) -> str | None:
"""
Verify a confirmation token. Returns the email if valid, None otherwise.
Checks both HMAC signature and expiry.
"""
try:
decoded = base64.urlsafe_b64decode(token.encode()).decode()
# rsplit from the right: email may contain colons (unlikely but safe)
parts = decoded.rsplit(":", 2)
if len(parts) != 3:
return None
email, timestamp, sig = parts
# Verify HMAC
expected = hmac.new(
settings.CONFIRM_SECRET.encode(),
f"{email}:{timestamp}".encode(),
hashlib.sha256,
).hexdigest()
if not hmac.compare_digest(sig, expected):
return None
# Check expiry
age_seconds = time.time() - int(timestamp)
if age_seconds > settings.CONFIRM_TOKEN_EXPIRY_HOURS * 3600:
return None
if age_seconds < 0:
return None # future timestamp — tampered
return email
except Exception:
return None

View File

@@ -8,3 +8,4 @@ pydantic>=2.0,<3.0
pydantic-settings>=2.0,<3.0 pydantic-settings>=2.0,<3.0
email-validator>=2.0,<3.0 email-validator>=2.0,<3.0
python-dotenv>=1.0,<2.0 python-dotenv>=1.0,<2.0
httpx>=0.27,<1.0

View File

@@ -1,12 +1,17 @@
import time
from unittest.mock import AsyncMock, patch
import pytest import pytest
import pytest_asyncio import pytest_asyncio
from httpx import ASGITransport, AsyncClient from httpx import ASGITransport, AsyncClient
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from app.main import app from app.main import app
from app.models import Base from app.models import Base, WaitlistEntry
from app.db import get_db from app.db import get_db
from app.rate_limit import _hits_store from app.rate_limit import _hits_store
from app.token import generate_token, verify_token
# Use SQLite for tests (no Postgres dependency) # Use SQLite for tests (no Postgres dependency)
TEST_DB_URL = "sqlite+aiosqlite:///./test_waitlist.db" TEST_DB_URL = "sqlite+aiosqlite:///./test_waitlist.db"
@@ -107,3 +112,115 @@ async def test_rate_limit(client):
# The 6th request should be rate-limited (limit is 5) # The 6th request should be rate-limited (limit is 5)
assert resp.status_code == 429 assert resp.status_code == 429
assert "Retry-After" in resp.headers assert "Retry-After" in resp.headers
# ── Confirmation token tests ─────────────────────────────────────────
@pytest.mark.asyncio
async def test_token_roundtrip():
"""A generated token should verify back to the same email."""
email = "token@example.com"
token = generate_token(email)
assert verify_token(token) == email
@pytest.mark.asyncio
async def test_token_expired():
"""An expired token should return None."""
email = "expired@example.com"
with patch("app.token.time") as mock_time:
# Generate token "49 hours ago"
past = time.time() - 49 * 3600
mock_time.time.return_value = past
token = generate_token(email)
# Now verify with real time — should be expired (>48h)
assert verify_token(token) is None
@pytest.mark.asyncio
async def test_token_tampered():
"""A tampered token should return None."""
token = generate_token("legit@example.com")
# Flip a character in the token
tampered = token[:-1] + ("A" if token[-1] != "A" else "B")
assert verify_token(tampered) is None
# ── Confirm endpoint tests ───────────────────────────────────────────
@pytest.mark.asyncio
async def test_confirm_valid_token(client, db_session):
"""GET /confirm with valid token marks email as confirmed."""
# Seed an unconfirmed entry
entry = WaitlistEntry(email="confirm@example.com", source="website")
db_session.add(entry)
await db_session.commit()
token = generate_token("confirm@example.com")
resp = await client.get(f"/api/v1/waitlist/confirm?token={token}")
assert resp.status_code == 200
assert "confirmed" in resp.text.lower() or "verified" in resp.text.lower()
# Verify DB state
result = await db_session.execute(
select(WaitlistEntry).where(WaitlistEntry.email == "confirm@example.com")
)
assert result.scalar_one().confirmed is True
@pytest.mark.asyncio
async def test_confirm_invalid_token(client):
"""GET /confirm with invalid token returns 400."""
resp = await client.get("/api/v1/waitlist/confirm?token=garbage")
assert resp.status_code == 400
assert "invalid" in resp.text.lower() or "expired" in resp.text.lower()
@pytest.mark.asyncio
async def test_confirm_idempotent(client, db_session):
"""Confirming an already confirmed email returns 200 (idempotent)."""
entry = WaitlistEntry(email="idem@example.com", source="website", confirmed=True)
db_session.add(entry)
await db_session.commit()
token = generate_token("idem@example.com")
resp = await client.get(f"/api/v1/waitlist/confirm?token={token}")
assert resp.status_code == 200
@pytest.mark.asyncio
async def test_confirm_unknown_email(client):
"""Token for a non-existent email returns 400."""
token = generate_token("unknown@example.com")
resp = await client.get(f"/api/v1/waitlist/confirm?token={token}")
assert resp.status_code == 400
# ── Brevo integration tests (mocked) ────────────────────────────────
@pytest.mark.asyncio
async def test_signup_triggers_confirmation_email(client, db_session):
"""When Brevo is configured, signup sends a confirmation email."""
with patch("app.routes.settings") as mock_settings, \
patch("app.routes.send_confirmation_email", new_callable=AsyncMock) as mock_send:
mock_settings.brevo_configured = True
mock_settings.CONFIRM_BASE_URL = "http://test"
resp = await client.post(
"/api/v1/waitlist",
json={"email": "brevo@example.com"},
)
assert resp.status_code == 200
# Wait for fire-and-forget task
import asyncio
await asyncio.sleep(0.1)
mock_send.assert_called_once()
call_args = mock_send.call_args
assert call_args[0][0] == "brevo@example.com"
assert "confirm" in call_args[0][1]