A battle-tested guide to automating email verification tests in CI/CD pipelines. Covers pytest, Playwright, GitHub Actions, webhook-driven testing, and production monitoring — with full working code examples that you can copy into your project today.
# Stop Testing Email with Gmail: A Practical Guide to Email Automation in CI/CD
I once watched a QA engineer manually check 47 registration emails in Gmail. It took 3 hours. She had a spreadsheet open in one monitor, Gmail in the other, and was copy-pasting confirmation links one by one into a browser to verify they worked. When she finished, the dev team pushed a hotfix that changed the email template, and she had to start over.
There's a better way. A *much* better way.
This post is the guide I wish I'd had five years ago, when I was the person building email verification flows and praying they worked in production because we had no automated tests for them. I'm going to show you exactly how to test email flows — registration, password reset, notifications, the whole lot — in your CI/CD pipeline, with real code you can steal.
Let's be honest about why email testing is uniquely painful compared to testing, say, a REST API endpoint.
Email is asynchronous by nature. You trigger a send, and then... you wait. How long? Could be 200 milliseconds. Could be 30 seconds. Could be never, if your SMTP server is having a bad day. There's no synchronous response that tells you "yes, the email arrived and the content is correct."
Email is an external dependency you don't control. Your application talks to an SMTP server, which talks to another SMTP server, which eventually puts something in an inbox. That's at least three systems outside your application boundary. In a unit testing world, this is a nightmare.
Email content is hard to assert against. The email your application sends is HTML. But not just any HTML — it's *email* HTML, which is a special circle of hell where <table> layouts are still best practice, inline styles are mandatory, and every email client renders things differently. Extracting a confirmation link from that mess requires parsing gnarly markup.
Email has no built-in test mode. Your database has a test instance. Your API has a staging environment. Your email? It either sends to a real inbox, or it doesn't send at all. There's no native "dry run."
And yet, email flows are *critical*. Registration, password reset, two-factor authentication, payment confirmations, account notifications — if any of these break, your users can't use your product. The irony is that the most important flows are the least tested.
Before we get to the right approach, let's roast the approaches I've seen (and used, to my shame) over the years.
Someone on the team creates [email protected] and puts the password in the team wiki. Every test sends emails there. Everyone on the team has it open in a browser tab.
Problems:
I've seen teams build elaborate subject-line conventions ([TEST-1234] Registration for user_abc) to make emails findable. At that point, you've built a bad, manual version of what should be automated.
Better than Gmail, genuinely. These services give you a fake SMTP server and a web UI to view captured emails. Your app sends emails to smtp.mailtrap.io instead of a real mail server, and nothing leaves the sandbox.
But here's where it falls apart for CI/CD:
Mailtrap is fine for development. For CI/CD, you need something with a proper API that you can poll programmatically.
The "pragmatic" developer says: "Just mock the email service in tests. Verify that sendEmail() was called with the right arguments. Done."
# The easy way out
def test_registration():
with mock.patch('app.email.send') as mock_send:
register_user('[email protected]', 'password123')
mock_send.assert_called_once()
assert 'confirm' in mock_send.call_args[1]['body'].lower()This tests that your code *tries* to send an email. It does NOT test:
Mocking is appropriate for unit tests. For integration tests and E2E tests — the ones that actually catch production bugs — you need real email delivery.
The worst option, and the most common. "We'll test it manually before release." Famous last words. I've seen production outages caused by:
localhost:3000 because someone forgot to set the APP_URL environment variableAll of these would have been caught by automated email tests.
Here's the mental model that makes email testing tractable: treat email inboxes like test fixtures.
Before each test: 1. Create a fresh, unique email address via API 2. Use that address in your test flow 3. Poll the inbox via API until the expected email arrives 4. Parse the email content and extract what you need 5. Assert and continue the test
After the test, the inbox is disposable — you don't need to clean it up, worry about conflicts with other tests, or manage any state.
This is where API-based disposable email services become invaluable. Services like EvilMail provide programmatic inbox creation and email retrieval through a clean REST API. No browser automation needed to check the inbox — it's just HTTP requests.
The key requirements for your disposable email provider:
Before we write code, let's map out the full flow:
┌─────────────────────────────────────────────────────────┐
│ CI/CD Pipeline │
│ │
│ ┌──────────┐ ┌──────────────┐ ┌───────────────┐ │
│ │ Create │───▶│ Register │───▶│ Poll Inbox │ │
│ │ Temp │ │ User with │ │ via API │ │
│ │ Email │ │ Temp Email │ │ (retry loop) │ │
│ └──────────┘ └──────────────┘ └───────┬───────┘ │
│ │ │
│ ▼ │
│ ┌──────────┐ ┌──────────────┐ ┌───────────────┐ │
│ │ Assert │◀──│ Follow │◀───│ Parse Email │ │
│ │ Success │ │ Confirm │ │ Extract │ │
│ │ State │ │ Link │ │ Link/Token │ │
│ └──────────┘ └──────────────┘ └───────────────┘ │
│ │
└─────────────────────────────────────────────────────────┘The critical insight is that the email inbox is just another API in your test. You create it, you read from it, you assert against it. No different from spinning up a test database.
Let's build a complete, working email test suite in Python. I'll use pytest because it's what most Python teams use, and requests for HTTP calls.
First, let's create a reusable helper class:
# tests/helpers/email_client.py
import time
import re
import requests
from dataclasses import dataclass
from typing import Optional
from html.parser import HTMLParser
@dataclass
class EmailMessage:
sender: str
subject: str
body_text: str
body_html: str
received_at: str
class LinkExtractor(HTMLParser):
"""Extract href values from anchor tags in HTML email content."""
def __init__(self):
super().__init__()
self.links = []
def handle_starttag(self, tag, attrs):
if tag == 'a':
for attr_name, attr_value in attrs:
if attr_name == 'href' and attr_value:
self.links.append(attr_value)
class DisposableEmailClient:
"""
Client for disposable email API.
Provides inbox creation, message polling, and content parsing.
"""
def __init__(self, api_base: str, api_key: str):
self.api_base = api_base.rstrip('/')
self.session = requests.Session()
self.session.headers.update({
'Authorization': f'Bearer {api_key}',
'Content-Type': 'application/json'
})
def create_inbox(self, prefix: str = 'test') -> str:
"""Create a new disposable email address. Returns the address."""
response = self.session.post(f'{self.api_base}/inbox/create', json={
'prefix': prefix
})
response.raise_for_status()
data = response.json()
return data['email']
def get_messages(self, email: str) -> list[EmailMessage]:
"""Fetch all messages for an email address."""
response = self.session.get(
f'{self.api_base}/inbox/messages',
params={'email': email}
)
response.raise_for_status()
return [
EmailMessage(
sender=msg['from'],
subject=msg['subject'],
body_text=msg.get('text', ''),
body_html=msg.get('html', ''),
received_at=msg['date']
)
for msg in response.json().get('messages', [])
]
def wait_for_message(
self,
email: str,
subject_contains: Optional[str] = None,
timeout: int = 30,
poll_interval: float = 1.5
) -> EmailMessage:
"""
Poll inbox until a matching message arrives.
Raises TimeoutError if no message arrives within the timeout.
"""
start = time.time()
while time.time() - start < timeout:
messages = self.get_messages(email)
for msg in messages:
if subject_contains is None:
return msg
if subject_contains.lower() in msg.subject.lower():
return msg
time.sleep(poll_interval)
elapsed = time.time() - start
raise TimeoutError(
f'No email matching subject "{subject_contains}" '
f'arrived at {email} within {elapsed:.1f}s'
)
@staticmethod
def extract_links(html_body: str) -> list[str]:
"""Extract all links from an HTML email body."""
parser = LinkExtractor()
parser.feed(html_body)
return parser.links
@staticmethod
def extract_link_by_text(html_body: str, link_text: str) -> Optional[str]:
"""
Extract a link whose visible text contains the given string.
More reliable than matching on href patterns.
"""
# Pattern: <a ...href="URL"...>...link_text...</a>
pattern = re.compile(
r'<a\s[^>]*href=["\']([^"\']+)["\'][^>]*>'
r'[^<]*' + re.escape(link_text) + r'[^<]*</a>',
re.IGNORECASE | re.DOTALL
)
match = pattern.search(html_body)
return match.group(1) if match else None
@staticmethod
def extract_otp_code(text_body: str, length: int = 6) -> Optional[str]:
"""Extract a numeric OTP code from email text."""
pattern = re.compile(r'\b(\d{' + str(length) + r'})\b')
match = pattern.search(text_body)
return match.group(1) if match else NoneNow let's wire this up as pytest fixtures:
# tests/conftest.py
import os
import pytest
import requests
from helpers.email_client import DisposableEmailClient
@pytest.fixture(scope='session')
def email_client():
"""Session-scoped email client — reuses connection pool."""
api_base = os.environ.get('EMAIL_API_BASE', 'https://evilmail.pro/api/v1')
api_key = os.environ['EMAIL_API_KEY'] # Fail fast if not set
return DisposableEmailClient(api_base, api_key)
@pytest.fixture
def temp_email(email_client):
"""Create a fresh disposable email for each test."""
return email_client.create_inbox(prefix='ci-test')
@pytest.fixture(scope='session')
def app_url():
"""Base URL for the application under test."""
return os.environ.get('APP_URL', 'http://localhost:8000')
@pytest.fixture(scope='session')
def http():
"""Reusable HTTP session for app requests."""
session = requests.Session()
yield session
session.close()# tests/test_registration_flow.py
import pytest
from urllib.parse import urlparse, parse_qs
class TestRegistrationFlow:
"""
End-to-end tests for the user registration email flow.
Creates a real email, registers a real user, verifies the real link.
"""
def test_registration_sends_confirmation_email(
self, http, app_url, email_client, temp_email
):
"""Register a new user and verify that a confirmation email arrives."""
# Step 1: Register with our disposable email
response = http.post(f'{app_url}/api/auth/register', json={
'email': temp_email,
'password': 'TestPassword123!',
'name': 'CI Test User'
})
assert response.status_code == 201, (
f'Registration failed: {response.status_code} {response.text}'
)
# Step 2: Wait for the confirmation email
msg = email_client.wait_for_message(
temp_email,
subject_contains='confirm',
timeout=30
)
# Step 3: Verify email content
assert 'confirm' in msg.subject.lower()
assert temp_email in msg.body_html or 'CI Test User' in msg.body_html
# Step 4: Extract and follow the confirmation link
confirm_link = email_client.extract_link_by_text(
msg.body_html, 'Confirm'
)
assert confirm_link is not None, (
f'No confirmation link found in email. '
f'Links found: {email_client.extract_links(msg.body_html)}'
)
# Step 5: Click the confirmation link
confirm_response = http.get(confirm_link, allow_redirects=True)
assert confirm_response.status_code == 200
# Step 6: Verify the user can now log in
login_response = http.post(f'{app_url}/api/auth/login', json={
'email': temp_email,
'password': 'TestPassword123!'
})
assert login_response.status_code == 200
assert 'token' in login_response.json()
def test_duplicate_registration_error(
self, http, app_url, email_client, temp_email
):
"""Ensure duplicate registration is handled gracefully."""
payload = {
'email': temp_email,
'password': 'TestPassword123!',
'name': 'CI Test User'
}
# Register once
http.post(f'{app_url}/api/auth/register', json=payload)
email_client.wait_for_message(temp_email, timeout=15)
# Register again — should fail with 409
response = http.post(f'{app_url}/api/auth/register', json=payload)
assert response.status_code == 409
def test_confirmation_link_single_use(
self, http, app_url, email_client, temp_email
):
"""Confirmation links should not work twice."""
http.post(f'{app_url}/api/auth/register', json={
'email': temp_email,
'password': 'TestPassword123!',
'name': 'CI Test User'
})
msg = email_client.wait_for_message(
temp_email, subject_contains='confirm', timeout=30
)
confirm_link = email_client.extract_link_by_text(
msg.body_html, 'Confirm'
)
# First click — should work
first = http.get(confirm_link, allow_redirects=True)
assert first.status_code == 200
# Second click — should fail or redirect to "already confirmed"
second = http.get(confirm_link, allow_redirects=True)
assert second.status_code in (200, 302, 410)
# If 200, check for "already confirmed" message
if second.status_code == 200:
assert 'already' in second.text.lower() or 'expired' in second.text.lower()# tests/test_password_reset.py
import pytest
class TestPasswordResetFlow:
"""End-to-end tests for password reset via email."""
@pytest.fixture
def registered_user(self, http, app_url, email_client, temp_email):
"""Create and confirm a user, return their email."""
http.post(f'{app_url}/api/auth/register', json={
'email': temp_email,
'password': 'OriginalPass123!',
'name': 'Reset Test User'
})
msg = email_client.wait_for_message(
temp_email, subject_contains='confirm', timeout=30
)
link = email_client.extract_link_by_text(msg.body_html, 'Confirm')
http.get(link, allow_redirects=True)
return temp_email
def test_password_reset_full_flow(
self, http, app_url, email_client, registered_user
):
"""Request reset, receive email, use link, set new password, log in."""
email = registered_user
# Request password reset
response = http.post(f'{app_url}/api/auth/forgot-password', json={
'email': email
})
assert response.status_code == 200
# Wait for reset email (second email in this inbox)
msg = email_client.wait_for_message(
email,
subject_contains='reset',
timeout=30
)
assert 'reset' in msg.subject.lower()
# Extract reset link
reset_link = email_client.extract_link_by_text(
msg.body_html, 'Reset'
)
assert reset_link is not None
# Extract token from the reset link
from urllib.parse import urlparse, parse_qs
parsed = urlparse(reset_link)
token = parse_qs(parsed.query).get('token', [None])[0]
assert token is not None, 'No token found in reset link'
# Use token to set new password
reset_response = http.post(
f'{app_url}/api/auth/reset-password',
json={
'token': token,
'password': 'NewPassword456!'
}
)
assert reset_response.status_code == 200
# Verify login with new password works
login = http.post(f'{app_url}/api/auth/login', json={
'email': email,
'password': 'NewPassword456!'
})
assert login.status_code == 200
# Verify login with old password fails
old_login = http.post(f'{app_url}/api/auth/login', json={
'email': email,
'password': 'OriginalPass123!'
})
assert old_login.status_code == 401
def test_reset_for_nonexistent_email(
self, http, app_url
):
"""
Requesting reset for a nonexistent email should return 200.
(Don't leak whether accounts exist.)
"""
response = http.post(f'{app_url}/api/auth/forgot-password', json={
'email': '[email protected]'
})
# Should return 200 to prevent email enumeration
assert response.status_code == 200For frontend-heavy applications, you want browser-based E2E tests that actually fill out forms and click buttons. Playwright is the gold standard for this.
// tests/helpers/email-client.ts
interface EmailMessage {
from: string;
subject: string;
text: string;
html: string;
date: string;
}
interface InboxResponse {
email: string;
}
interface MessagesResponse {
messages: Array<{
from: string;
subject: string;
text?: string;
html?: string;
date: string;
}>;
}
export class EmailTestClient {
private baseUrl: string;
private apiKey: string;
constructor(baseUrl: string, apiKey: string) {
this.baseUrl = baseUrl.replace(/\/$/, '');
this.apiKey = apiKey;
}
private async request<T>(path: string, options: RequestInit = {}): Promise<T> {
const response = await fetch(`${this.baseUrl}${path}`, {
...options,
headers: {
'Authorization': `Bearer ${this.apiKey}`,
'Content-Type': 'application/json',
...options.headers,
},
});
if (!response.ok) {
const body = await response.text();
throw new Error(`Email API error ${response.status}: ${body}`);
}
return response.json() as Promise<T>;
}
async createInbox(prefix = 'pw-test'): Promise<string> {
const data = await this.request<InboxResponse>('/inbox/create', {
method: 'POST',
body: JSON.stringify({ prefix }),
});
return data.email;
}
async getMessages(email: string): Promise<EmailMessage[]> {
const data = await this.request<MessagesResponse>(
`/inbox/messages?email=${encodeURIComponent(email)}`
);
return data.messages.map(msg => ({
from: msg.from,
subject: msg.subject,
text: msg.text ?? '',
html: msg.html ?? '',
date: msg.date,
}));
}
async waitForMessage(
email: string,
options: {
subjectContains?: string;
timeout?: number;
pollInterval?: number;
} = {}
): Promise<EmailMessage> {
const {
subjectContains,
timeout = 30_000,
pollInterval = 1_500
} = options;
const start = Date.now();
while (Date.now() - start < timeout) {
const messages = await this.getMessages(email);
for (const msg of messages) {
if (!subjectContains) return msg;
if (msg.subject.toLowerCase().includes(subjectContains.toLowerCase())) {
return msg;
}
}
await new Promise(resolve => setTimeout(resolve, pollInterval));
}
throw new Error(
`Timeout: No email matching "${subjectContains}" ` +
`arrived at ${email} within ${timeout}ms`
);
}
extractLinks(html: string): string[] {
const linkRegex = /href=["']([^"']+)["']/gi;
const links: string[] = [];
let match: RegExpExecArray | null;
while ((match = linkRegex.exec(html)) !== null) {
links.push(match[1]);
}
return links;
}
extractLinkByText(html: string, linkText: string): string | null {
const regex = new RegExp(
`<a\\s[^>]*href=["']([^"']+)["'][^>]*>[^<]*${linkText}[^<]*<\\/a>`,
'i'
);
const match = html.match(regex);
return match ? match[1] : null;
}
extractOtpCode(text: string, length = 6): string | null {
const regex = new RegExp(`\\b(\\d{${length}})\\b`);
const match = text.match(regex);
return match ? match[1] : null;
}
}// tests/e2e/registration.spec.ts
import { test, expect } from '@playwright/test';
import { EmailTestClient } from '../helpers/email-client';
const emailClient = new EmailTestClient(
process.env.EMAIL_API_BASE || 'https://evilmail.pro/api/v1',
process.env.EMAIL_API_KEY!
);
test.describe('Registration Flow', () => {
test('complete registration with email verification', async ({ page }) => {
// Create a disposable inbox
const testEmail = await emailClient.createInbox('reg');
// Fill out registration form
await page.goto('/register');
await page.getByLabel('Email').fill(testEmail);
await page.getByLabel('Password', { exact: true }).fill('SecurePass123!');
await page.getByLabel('Confirm Password').fill('SecurePass123!');
await page.getByLabel('Full Name').fill('E2E Test User');
await page.getByRole('button', { name: 'Create Account' }).click();
// Should show "check your email" message
await expect(
page.getByText(/check your email|verification link/i)
).toBeVisible({ timeout: 10_000 });
// Wait for the email to arrive
const email = await emailClient.waitForMessage(testEmail, {
subjectContains: 'confirm',
timeout: 45_000,
});
expect(email.subject.toLowerCase()).toContain('confirm');
// Extract confirmation link
const confirmLink = emailClient.extractLinkByText(email.html, 'Confirm');
expect(confirmLink).not.toBeNull();
// Navigate to confirmation link in the browser
await page.goto(confirmLink!);
// Should show success message
await expect(
page.getByText(/confirmed|verified|success/i)
).toBeVisible({ timeout: 10_000 });
// Now log in with the confirmed account
await page.goto('/login');
await page.getByLabel('Email').fill(testEmail);
await page.getByLabel('Password').fill('SecurePass123!');
await page.getByRole('button', { name: /log in|sign in/i }).click();
// Should reach dashboard
await expect(page).toHaveURL(/dashboard/, { timeout: 10_000 });
});
test('password reset via email', async ({ page }) => {
// This test assumes a confirmed user already exists.
// In practice, you'd use a setup fixture or seed data.
const testEmail = await emailClient.createInbox('reset');
// First, register and confirm
await page.goto('/register');
await page.getByLabel('Email').fill(testEmail);
await page.getByLabel('Password', { exact: true }).fill('OldPassword123!');
await page.getByLabel('Confirm Password').fill('OldPassword123!');
await page.getByLabel('Full Name').fill('Reset Test User');
await page.getByRole('button', { name: 'Create Account' }).click();
const confirmEmail = await emailClient.waitForMessage(testEmail, {
subjectContains: 'confirm',
timeout: 45_000,
});
const confirmLink = emailClient.extractLinkByText(confirmEmail.html, 'Confirm');
await page.goto(confirmLink!);
// Now go to forgot password
await page.goto('/forgot-password');
await page.getByLabel('Email').fill(testEmail);
await page.getByRole('button', { name: /reset|send/i }).click();
await expect(
page.getByText(/check your email|reset link/i)
).toBeVisible({ timeout: 10_000 });
// Wait for reset email
const resetEmail = await emailClient.waitForMessage(testEmail, {
subjectContains: 'reset',
timeout: 45_000,
});
const resetLink = emailClient.extractLinkByText(resetEmail.html, 'Reset');
expect(resetLink).not.toBeNull();
// Follow the reset link
await page.goto(resetLink!);
// Fill in new password
await page.getByLabel('New Password', { exact: true }).fill('NewPassword456!');
await page.getByLabel('Confirm New Password').fill('NewPassword456!');
await page.getByRole('button', { name: /reset|change|update/i }).click();
await expect(
page.getByText(/password.*changed|password.*reset|success/i)
).toBeVisible({ timeout: 10_000 });
// Log in with new password
await page.goto('/login');
await page.getByLabel('Email').fill(testEmail);
await page.getByLabel('Password').fill('NewPassword456!');
await page.getByRole('button', { name: /log in|sign in/i }).click();
await expect(page).toHaveURL(/dashboard/, { timeout: 10_000 });
});
test('notification email content is correct', async ({ page, request }) => {
const testEmail = await emailClient.createInbox('notif');
// Trigger some action that sends a notification
// (using API directly for speed)
const response = await request.post('/api/notifications/test', {
data: {
email: testEmail,
type: 'welcome'
}
});
expect(response.ok()).toBeTruthy();
const email = await emailClient.waitForMessage(testEmail, {
subjectContains: 'welcome',
timeout: 30_000,
});
// Assert email content structure
expect(email.html).toContain('Welcome');
expect(email.html).toContain('Get Started');
// Verify all links in the email are valid (not broken)
const links = emailClient.extractLinks(email.html);
for (const link of links) {
// Skip mailto and anchor links
if (link.startsWith('mailto:') || link.startsWith('#')) continue;
const linkResponse = await request.head(link);
expect(
linkResponse.ok(),
`Broken link in email: ${link}`
).toBeTruthy();
}
});
});// playwright.config.ts
import { defineConfig, devices } from '@playwright/test';
export default defineConfig({
testDir: './tests/e2e',
timeout: 120_000, // 2 minutes per test (email can be slow)
expect: {
timeout: 15_000,
},
retries: process.env.CI ? 2 : 0,
workers: process.env.CI ? 2 : 4, // Limit parallelism in CI
reporter: process.env.CI
? [['github'], ['html', { open: 'never' }]]
: 'html',
use: {
baseURL: process.env.APP_URL || 'http://localhost:3000',
trace: 'retain-on-failure',
screenshot: 'only-on-failure',
},
projects: [
{
name: 'chromium',
use: { ...devices['Desktop Chrome'] },
},
],
webServer: process.env.CI ? undefined : {
command: 'npm run dev',
port: 3000,
reuseExistingServer: true,
},
});Real email testing has a dozen sharp edges that toy examples never mention. Here's what I've learned the hard way.
Email HTML is not web HTML. Email clients mangle markup in creative ways, and the HTML your app *sends* may not be the HTML your test *receives*.
# Problem: Some email servers re-encode HTML entities
# Your app sends: href="https://app.com/confirm?token=abc123&type=email"
# You receive: href="https://app.com/confirm?token=abc123&type=email"
import html
def extract_link_safe(raw_html: str, pattern: str) -> str | None:
"""Extract link with HTML entity decoding."""
# First, decode HTML entities
decoded = html.unescape(raw_html)
# Then extract links from the decoded HTML
links = DisposableEmailClient.extract_links(decoded)
for link in links:
if pattern in link:
return link
return NoneEmails don't always arrive in 2 seconds. Graylisting, spam filtering, and server load can add delays.
def wait_for_message_with_backoff(
client: DisposableEmailClient,
email: str,
subject: str,
max_attempts: int = 10
) -> EmailMessage:
"""
Exponential backoff polling. Starts fast, slows down.
Total wait: ~1 + 2 + 3 + 4 + 5 + 5 + 5 + 5 + 5 + 5 = ~40s
"""
for attempt in range(max_attempts):
messages = client.get_messages(email)
for msg in messages:
if subject.lower() in msg.subject.lower():
return msg
delay = min(1 + attempt, 5) # Cap at 5 seconds
time.sleep(delay)
raise TimeoutError(f'Email "{subject}" never arrived at {email}')Internationalized content? Buckle up.
def decode_email_subject(subject: str) -> str:
"""
Handle RFC 2047 encoded subjects.
e.g., '=?UTF-8?B?Q29uZmlybSB5b3VyIGVtYWls?=' -> 'Confirm your email'
"""
import email.header
decoded_parts = email.header.decode_header(subject)
parts = []
for part, charset in decoded_parts:
if isinstance(part, bytes):
parts.append(part.decode(charset or 'utf-8', errors='replace'))
else:
parts.append(part)
return ''.join(parts)If your app sends invoices, reports, or tickets as attachments:
def test_invoice_email_has_pdf_attachment(
http, app_url, email_client, temp_email
):
"""Verify invoice emails include a valid PDF attachment."""
# Trigger invoice generation
http.post(f'{app_url}/api/invoices/generate', json={
'email': temp_email,
'order_id': 'test-order-001'
})
msg = email_client.wait_for_message(
temp_email,
subject_contains='invoice',
timeout=60 # PDF generation can be slow
)
# Check for attachment metadata (API-dependent)
assert hasattr(msg, 'attachments') and len(msg.attachments) > 0
attachment = msg.attachments[0]
assert attachment['filename'].endswith('.pdf')
assert attachment['content_type'] == 'application/pdf'
assert len(attachment['content']) > 1000 # Not an empty file
# Optionally: parse the PDF and check content
import io
from PyPDF2 import PdfReader
pdf = PdfReader(io.BytesIO(attachment['content']))
text = pdf.pages[0].extract_text()
assert 'test-order-001' in textWhen testing flows that send multiple emails (register + welcome, or reset + confirmation), you need to distinguish between them:
def wait_for_nth_message(
client: DisposableEmailClient,
email: str,
n: int,
timeout: int = 60
) -> EmailMessage:
"""Wait until at least N messages exist, return the Nth."""
start = time.time()
while time.time() - start < timeout:
messages = client.get_messages(email)
if len(messages) >= n:
# Sort by date to get consistent ordering
messages.sort(key=lambda m: m.received_at)
return messages[n - 1]
time.sleep(2)
raise TimeoutError(f'Expected {n} messages at {email}, timed out')Here's where rubber meets road. Let's set this up in real pipelines.
# .github/workflows/email-tests.yml
name: Email Integration Tests
on:
push:
branches: [main, develop]
pull_request:
branches: [main]
env:
APP_URL: http://localhost:8000
DATABASE_URL: postgresql://test:test@localhost:5432/testdb
jobs:
email-tests:
runs-on: ubuntu-latest
timeout-minutes: 15
services:
postgres:
image: postgres:16
env:
POSTGRES_USER: test
POSTGRES_PASSWORD: test
POSTGRES_DB: testdb
ports:
- 5432:5432
options: >-
--health-cmd pg_isready
--health-interval 10s
--health-timeout 5s
--health-retries 5
redis:
image: redis:7
ports:
- 6379:6379
options: >-
--health-cmd "redis-cli ping"
--health-interval 10s
--health-timeout 5s
--health-retries 5
steps:
- uses: actions/checkout@v4
- name: Setup Python
uses: actions/setup-python@v5
with:
python-version: '3.12'
cache: 'pip'
- name: Install dependencies
run: |
pip install -r requirements.txt
pip install -r requirements-test.txt
- name: Run migrations
run: python manage.py migrate
- name: Start application
run: |
python manage.py runserver &
# Wait for the server to be ready
for i in $(seq 1 30); do
curl -sf http://localhost:8000/health && break
sleep 1
done
- name: Run email integration tests
env:
EMAIL_API_KEY: ${{ secrets.EMAIL_API_KEY }}
EMAIL_API_BASE: ${{ secrets.EMAIL_API_BASE }}
run: |
pytest tests/test_registration_flow.py \
tests/test_password_reset.py \
-v \
--timeout=120 \
--tb=short \
-x # Stop on first failure
- name: Upload test artifacts
if: failure()
uses: actions/upload-artifact@v4
with:
name: email-test-failures
path: |
tests/reports/
tests/screenshots/# .gitlab-ci.yml
stages:
- test
email-integration:
stage: test
image: python:3.12
services:
- postgres:16
- redis:7
variables:
POSTGRES_DB: testdb
POSTGRES_USER: test
POSTGRES_PASSWORD: test
DATABASE_URL: postgresql://test:test@postgres:5432/testdb
REDIS_URL: redis://redis:6379
APP_URL: http://localhost:8000
before_script:
- pip install -r requirements.txt -r requirements-test.txt
- python manage.py migrate
- python manage.py runserver &
- |
for i in $(seq 1 30); do
curl -sf http://localhost:8000/health && break
sleep 1
done
script:
- pytest tests/ -k "email" -v --timeout=120 --junitxml=report.xml
artifacts:
when: always
reports:
junit: report.xml
expire_in: 7 days
rules:
- if: $CI_PIPELINE_SOURCE == "merge_request_event"
- if: $CI_COMMIT_BRANCH == "main"Never hardcode API keys. Here's the hierarchy I recommend:
# Local development: .env file (gitignored)
EMAIL_API_KEY=dev_key_12345
EMAIL_API_BASE=https://evilmail.pro/api/v1
# CI/CD: Repository secrets (GitHub) or CI/CD variables (GitLab)
# GitHub: Settings > Secrets and variables > Actions
# GitLab: Settings > CI/CD > Variables (masked + protected)
# Production monitoring: Vault, AWS Secrets Manager, etc.
# Never use the same API key for testing and production monitoringPro tip: create a *separate* API key specifically for CI. This way, if you need to rotate it (and you will), you know exactly where it's used. Name it something obvious like CI_EMAIL_TESTING_KEY.
Polling works, but it's inefficient. If your disposable email provider supports webhooks, you can flip the model: instead of asking "did the email arrive yet?" every 2 seconds, the email service *tells you* when it arrives.
App sends email ──▶ Disposable Email Service
│
│ webhook POST
▼
Your Test Webhook Server
│
│ resolves promise
▼
Test Continues// tests/helpers/webhook-email-receiver.ts
import express from 'express';
import { EventEmitter } from 'events';
interface WebhookEmail {
to: string;
from: string;
subject: string;
html: string;
text: string;
}
export class WebhookEmailReceiver {
private app: express.Application;
private server: any;
private emitter = new EventEmitter();
private receivedEmails: WebhookEmail[] = [];
constructor(private port: number = 9876) {
this.app = express();
this.app.use(express.json({ limit: '10mb' }));
this.app.post('/webhook/email', (req, res) => {
const email: WebhookEmail = {
to: req.body.to,
from: req.body.from,
subject: req.body.subject,
html: req.body.html || '',
text: req.body.text || '',
};
this.receivedEmails.push(email);
this.emitter.emit('email', email);
this.emitter.emit(`email:${email.to}`, email);
res.status(200).json({ received: true });
});
}
async start(): Promise<void> {
return new Promise((resolve) => {
this.server = this.app.listen(this.port, () => {
console.log(`Webhook receiver listening on port ${this.port}`);
resolve();
});
});
}
async stop(): Promise<void> {
return new Promise((resolve) => {
if (this.server) {
this.server.close(resolve);
} else {
resolve();
}
});
}
waitForEmail(
toAddress: string,
options: { timeout?: number; subjectContains?: string } = {}
): Promise<WebhookEmail> {
const { timeout = 30_000, subjectContains } = options;
return new Promise((resolve, reject) => {
const timer = setTimeout(() => {
this.emitter.removeAllListeners(`email:${toAddress}`);
reject(new Error(
`Webhook timeout: no email to ${toAddress} within ${timeout}ms`
));
}, timeout);
// Check already-received emails first
const existing = this.receivedEmails.find(
e => e.to === toAddress &&
(!subjectContains || e.subject.toLowerCase().includes(subjectContains.toLowerCase()))
);
if (existing) {
clearTimeout(timer);
resolve(existing);
return;
}
// Listen for new emails
const handler = (email: WebhookEmail) => {
if (subjectContains && !email.subject.toLowerCase().includes(subjectContains.toLowerCase())) {
return; // Not the email we're looking for
}
clearTimeout(timer);
this.emitter.removeListener(`email:${toAddress}`, handler);
resolve(email);
};
this.emitter.on(`email:${toAddress}`, handler);
});
}
clear(): void {
this.receivedEmails = [];
this.emitter.removeAllListeners();
}
}// tests/e2e/webhook-registration.spec.ts
import { test, expect } from '@playwright/test';
import { WebhookEmailReceiver } from '../helpers/webhook-email-receiver';
import { EmailTestClient } from '../helpers/email-client';
let webhook: WebhookEmailReceiver;
const emailClient = new EmailTestClient(
process.env.EMAIL_API_BASE!,
process.env.EMAIL_API_KEY!
);
test.beforeAll(async () => {
webhook = new WebhookEmailReceiver(9876);
await webhook.start();
});
test.afterAll(async () => {
await webhook.stop();
});
test.afterEach(() => {
webhook.clear();
});
test('registration with webhook-based email capture', async ({ page }) => {
const testEmail = await emailClient.createInbox('wh');
// Configure webhook for this inbox (API-dependent)
// Some providers let you set a webhook URL per inbox
// Start waiting BEFORE triggering the send
const emailPromise = webhook.waitForEmail(testEmail, {
subjectContains: 'confirm',
timeout: 45_000,
});
// Register the user
await page.goto('/register');
await page.getByLabel('Email').fill(testEmail);
await page.getByLabel('Password', { exact: true }).fill('SecurePass123!');
await page.getByLabel('Confirm Password').fill('SecurePass123!');
await page.getByLabel('Full Name').fill('Webhook Test');
await page.getByRole('button', { name: 'Create Account' }).click();
// Wait for webhook notification — no polling needed
const email = await emailPromise;
expect(email.subject.toLowerCase()).toContain('confirm');
// Continue with link extraction and verification...
});Webhooks are faster and more efficient than polling, but they add complexity: you need a publicly reachable URL (use ngrok in local dev, or a dedicated endpoint in CI). For most teams, polling is simpler and good enough.
Here's a technique most teams never think about: use disposable email as a synthetic monitor for your production email pipeline.
The idea is simple. Every 5 minutes, a cron job: 1. Creates a disposable inbox 2. Triggers a real email from your production system (a test endpoint, or a real flow against a test account) 3. Polls the inbox for the email 4. Measures delivery time and checks content 5. Alerts if anything is wrong
# scripts/email_monitor.py
"""
Production email delivery monitor.
Run via cron: */5 * * * * /usr/bin/python3 /opt/monitors/email_monitor.py
"""
import os
import sys
import time
import json
import requests
from datetime import datetime, timezone
API_BASE = os.environ['EMAIL_API_BASE']
API_KEY = os.environ['MONITOR_EMAIL_API_KEY']
APP_URL = os.environ['APP_URL']
ALERT_WEBHOOK = os.environ.get('SLACK_WEBHOOK_URL')
MAX_DELIVERY_SECONDS = 30
def create_inbox() -> str:
resp = requests.post(
f'{API_BASE}/inbox/create',
json={'prefix': 'monitor'},
headers={'Authorization': f'Bearer {API_KEY}'}
)
resp.raise_for_status()
return resp.json()['email']
def trigger_test_email(email: str) -> None:
"""Trigger a known email from the application."""
resp = requests.post(
f'{APP_URL}/api/internal/health/email',
json={'recipient': email},
headers={'X-Internal-Key': os.environ['INTERNAL_API_KEY']}
)
resp.raise_for_status()
def poll_for_email(email: str, timeout: int = MAX_DELIVERY_SECONDS) -> dict:
start = time.time()
while time.time() - start < timeout:
resp = requests.get(
f'{API_BASE}/inbox/messages',
params={'email': email},
headers={'Authorization': f'Bearer {API_KEY}'}
)
resp.raise_for_status()
messages = resp.json().get('messages', [])
if messages:
delivery_time = time.time() - start
return {
'delivered': True,
'delivery_seconds': round(delivery_time, 2),
'subject': messages[0]['subject'],
'has_html': bool(messages[0].get('html')),
}
time.sleep(2)
return {
'delivered': False,
'delivery_seconds': timeout,
'error': 'Timeout waiting for email delivery'
}
def alert(message: str) -> None:
"""Send alert to Slack."""
if ALERT_WEBHOOK:
requests.post(ALERT_WEBHOOK, json={
'text': f':rotating_light: Email Monitor Alert: {message}',
'username': 'Email Monitor'
})
print(f'ALERT: {message}', file=sys.stderr)
def report_metric(result: dict) -> None:
"""Push metrics to your monitoring system (e.g., Prometheus, Datadog)."""
# Example: Prometheus pushgateway
if os.environ.get('PROMETHEUS_PUSHGATEWAY'):
gateway = os.environ['PROMETHEUS_PUSHGATEWAY']
metrics = [
f'email_delivery_success {1 if result["delivered"] else 0}',
f'email_delivery_seconds {result["delivery_seconds"]}',
]
requests.post(
f'{gateway}/metrics/job/email_monitor',
data='\n'.join(metrics)
)
def main():
try:
inbox = create_inbox()
send_time = datetime.now(timezone.utc).isoformat()
trigger_test_email(inbox)
result = poll_for_email(inbox)
if not result['delivered']:
alert(f'Email not delivered within {MAX_DELIVERY_SECONDS}s')
elif result['delivery_seconds'] > 15:
alert(f'Email delivery slow: {result["delivery_seconds"]}s')
elif not result.get('has_html'):
alert('Email delivered but HTML body is empty')
else:
print(json.dumps({
'status': 'ok',
'delivery_seconds': result['delivery_seconds'],
'timestamp': send_time
}))
report_metric(result)
except Exception as e:
alert(f'Monitor script failed: {e}')
sys.exit(1)
if __name__ == '__main__':
main()I've caught so many production issues with this pattern:
All of these were caught within 5 minutes, instead of hours later when users started complaining.
Email tests are IO-bound, not CPU-bound. They spend most of their time waiting for emails to arrive. This makes them excellent candidates for parallel execution.
# pytest.ini or pyproject.toml
# [tool.pytest.ini_options]
# addopts = "-n 4" # 4 parallel workers with pytest-xdistBut there's a catch: each parallel test needs its own disposable inbox. If you're sharing inboxes across tests, parallel execution will cause flaky failures. This is why the temp_email fixture creates a *new* inbox per test.
# This works in parallel — each test has its own inbox
@pytest.fixture
def temp_email(email_client):
return email_client.create_inbox(
prefix=f'ci-{os.getpid()}' # Include PID for extra uniqueness
)Disposable email APIs have rate limits. In a large test suite, you might hit them.
Mitigation strategies:
# Strategy 1: Reuse inboxes within a test class
class TestRegistrationFlow:
@pytest.fixture(scope='class')
def shared_email(self, email_client):
"""One inbox for the whole class — messages accumulate."""
return email_client.create_inbox(prefix='shared')
# Strategy 2: Add small delays between inbox creation
@pytest.fixture
def temp_email(email_client):
email = email_client.create_inbox()
time.sleep(0.2) # 200ms breathing room
return email
# Strategy 3: Pool pre-created inboxes
@pytest.fixture(scope='session')
def email_pool(email_client):
"""Pre-create a pool of inboxes at session start."""
pool = [email_client.create_inbox(prefix=f'pool-{i}') for i in range(20)]
return iter(pool)
@pytest.fixture
def temp_email(email_pool):
return next(email_pool)Disposable emails are, by design, temporary. But if you're running hundreds of tests a day, check whether your provider auto-cleans inboxes. If not:
@pytest.fixture(autouse=True, scope='session')
def cleanup_test_inboxes(email_client):
"""Cleanup all test inboxes after the test session."""
yield
# Post-test cleanup if your API supports it
try:
email_client.session.delete(
f'{email_client.api_base}/inbox/cleanup',
json={'prefix': 'ci-test', 'older_than_hours': 1}
)
except Exception:
pass # Best-effort cleanupThe biggest source of flaky email tests is timeouts. Too short, and tests fail on slow days. Too long, and your pipeline takes forever when something is actually broken.
My recommended defaults:
TIMEOUTS = {
'email_delivery': 30, # Max seconds to wait for an email
'email_poll_interval': 1.5, # Seconds between inbox checks
'page_load': 15, # Max seconds for a page to load
'total_test': 120, # Max seconds for entire test
}And always, *always* include the timeout value in the error message:
raise TimeoutError(
f'Email not delivered within {timeout}s. '
f'Subject filter: "{subject_contains}". '
f'Inbox: {email}. '
f'Messages found: {len(messages)}'
)That extra context in the error message will save you 30 minutes of debugging when a test fails at 2 AM in CI.
Here's how I structure email tests in a real project:
tests/
├── conftest.py # Fixtures: email_client, temp_email, app_url
├── helpers/
│ ├── email_client.py # DisposableEmailClient class
│ └── assertions.py # Custom assertions for email content
├── unit/
│ ├── test_email_templates.py # Template rendering (mocked, fast)
│ └── test_email_validation.py # Address validation logic
├── integration/
│ ├── test_smtp_delivery.py # SMTP connection and delivery
│ ├── test_registration_flow.py # Full registration email flow
│ ├── test_password_reset.py # Full password reset flow
│ └── test_notification_emails.py # Notification content and delivery
└── e2e/
├── test_signup_journey.py # Browser-based signup with email
└── test_onboarding_emails.py # Multi-email onboarding sequenceUnit tests run on every commit (< 5 seconds). Integration tests run on every PR (2-3 minutes). E2E tests run on merge to main (5-10 minutes). The production monitor runs continuously.
Before we wrap up, here's a cheat sheet of patterns you'll use repeatedly:
# Pattern: Wait for email, extract link, follow it
msg = email_client.wait_for_message(email, subject_contains='confirm')
link = email_client.extract_link_by_text(msg.body_html, 'Confirm')
response = requests.get(link, allow_redirects=True)
assert response.status_code == 200
# Pattern: Wait for email, extract OTP, submit it
msg = email_client.wait_for_message(email, subject_contains='verification code')
code = email_client.extract_otp_code(msg.body_text)
response = requests.post(f'{app_url}/api/auth/verify-otp', json={'code': code})
assert response.status_code == 200
# Pattern: Verify email NOT sent (negative test)
import pytest
with pytest.raises(TimeoutError):
email_client.wait_for_message(
email,
subject_contains='reset',
timeout=10 # Short timeout for negative tests
)
# Pattern: Verify email content structure
msg = email_client.wait_for_message(email, subject_contains='welcome')
assert 'unsubscribe' in msg.body_html.lower() # CAN-SPAM compliance
assert msg.body_text.strip() # Plain text version exists
links = email_client.extract_links(msg.body_html)
assert all(link.startswith('https://') for link in links if not link.startswith('mailto:'))I started this post with a story about a QA engineer spending 3 hours manually checking emails. That same team now runs 47 email tests in under 4 minutes, in parallel, in CI, on every pull request. No human touches Gmail. No one worries about whether the registration flow works after a deploy.
The path to get there wasn't complicated. It was four things:
1. Use disposable email with an API — not Gmail, not mocks, not Mailtrap in production pipelines. A service like EvilMail that gives you programmatic inbox creation and message retrieval.
2. Treat inboxes as test fixtures — create them, use them, throw them away. One per test. No shared state.
3. Build a thin helper library — the DisposableEmailClient class we built is about 80 lines. It handles polling, parsing, link extraction, and OTP codes. You'll reuse it across every test.
4. Integrate into CI/CD properly — secrets management, reasonable timeouts, parallel execution, good error messages.
The investment is maybe a day of work. The payoff is never again hearing "the registration email is broken in production" from a customer.
Your email tests should be boring. They should be automated. They should run on every deploy. And they should catch bugs before your users do.
Now go delete that shared Gmail password from your team wiki.