deemanone/padelnomics
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

test: add live Resend integration tests (delivered@resend.dev)

Browse Source 
9 tests exercise the full handler→wrap→send_email→Resend API path
using Resend's @resend.dev test addresses. Skipped when RESEND_API_KEY
is not set.

With a full_access API key, tests also retrieve the sent email via
resend.Emails.get() and assert on the rendered HTML (wordmark, links,
project details, heat badges). With a sending_access key, send is
verified but HTML assertions are skipped gracefully.

Includes bounce handling test and 0.6s inter-test delay for Resend's
2 req/sec rate limit.

Run with: RESEND_API_KEY=re_xxx pytest -k resend_live

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Deeman
2026-02-23 11:19:26 +01:00
parent aafb3cfc94
commit cb1f00baf0
1 changed files with 248 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

248
web/tests/test_emails.py
View File

@@ -4,8 +4,13 @@ Tests for all transactional email worker handlers.
Each handler is tested by mocking send_email and calling the handler
directly with a payload. Assertions cover: recipient, subject keywords,
HTML content (design elements, i18n keys resolved), and from_addr.
The TestResendLive class at the bottom sends real emails via Resend's
@resend.dev test addresses. Skipped unless RESEND_API_KEY is set.
"""
import asyncio
import os
from datetime import UTC, datetime
from unittest.mock import AsyncMock, patch
@@ -591,3 +596,246 @@ async def _seed_lead_and_supplier(
await db.commit()
return lead_id, supplier_id
# ── Live Resend integration tests ────────────────────────────────
#
# These send real emails to Resend's @resend.dev test addresses,
# then retrieve the sent email via resend.Emails.get() to verify the
# rendered HTML contains expected design elements and content.
#
# Skipped unless RESEND_API_KEY is set in the environment.
# Run with: RESEND_API_KEY=re_xxx uv run pytest web/tests/test_emails.py -k resend_live -v
_has_resend_key = bool(os.environ.get("RESEND_API_KEY"))
_skip_no_key = pytest.mark.skipif(not _has_resend_key, reason="RESEND_API_KEY not set")
def _send_and_capture():
"""Patch resend.Emails.send to capture the returned email ID.
Returns (wrapper, captured) where captured is a dict that will
contain {"id": ...} after send is called. The real send still
executes — this just grabs the return value that core.send_email
discards.
"""
import resend as _resend
_original_send = _resend.Emails.send
captured = {}
def _wrapper(params):
result = _original_send(params)
captured["id"] = result.id if hasattr(result, "id") else result.get("id")
return result
return _wrapper, captured
def _retrieve_html(captured: dict) -> str | None:
"""Retrieve the sent email from Resend and return its HTML.
Returns None if:
- The send failed (rate limit, network error) — captured has no id
- The API key lacks read permission (sending_access only)
"""
import resend as _resend
email_id = captured.get("id")
if not email_id:
return None # send failed (e.g. rate limit) — can't retrieve
try:
email = _resend.Emails.get(email_id)
except Exception:
# sending_access key — can't retrieve, that's OK
return None
return email.html if hasattr(email, "html") else email.get("html", "")
@_skip_no_key
class TestResendLive:
"""Send real emails via Resend to @resend.dev test addresses.
Uses ``delivered@resend.dev`` which simulates successful delivery.
After sending, retrieves the email via resend.Emails.get() and
asserts on the rendered HTML (design elements, content, structure).
Each test calls the handler WITHOUT mocking send_email, so the full
path is exercised: handler → _email_wrap → send_email → Resend API.
"""
@pytest.fixture(autouse=True)
async def _set_resend_key(self):
"""Ensure config picks up the real API key from env.
Adds a 0.6s pause between tests to respect Resend's 2 req/sec
rate limit.
"""
import resend as _resend
original = core.config.RESEND_API_KEY
core.config.RESEND_API_KEY = os.environ["RESEND_API_KEY"]
_resend.api_key = os.environ["RESEND_API_KEY"]
yield
core.config.RESEND_API_KEY = original
await asyncio.sleep(0.6)
def _check_html(self, captured: dict, assertions: list[tuple[str, str]]):
"""Retrieve sent email and run assertions on HTML.
If the API key lacks read permission (sending_access only),
the retrieve is skipped — the test still passes because the
send itself succeeded (no exception from handler).
assertions: list of (needle, description) tuples.
"""
html = _retrieve_html(captured)
if html is None:
return # sending_access key — can't retrieve, send was enough
for needle, desc in assertions:
assert needle in html, f"Expected {desc!r} ({needle!r}) in rendered HTML"
@pytest.mark.asyncio
async def test_magic_link_delivered(self):
wrapper, captured = _send_and_capture()
with patch("resend.Emails.send", side_effect=wrapper):
await handle_send_magic_link({
"email": "delivered@resend.dev",
"token": "test_token_magic",
"lang": "en",
})
self._check_html(captured, [
("padelnomics", "wordmark"),
("Bricolage Grotesque", "brand font"),
("/auth/verify?token=test_token_magic", "verify link"),
("padelnomics.io", "footer link"),
])
@pytest.mark.asyncio
async def test_welcome_delivered(self):
wrapper, captured = _send_and_capture()
with patch("resend.Emails.send", side_effect=wrapper):
await handle_send_welcome({
"email": "delivered@resend.dev",
"name": "Test User",
"lang": "en",
})
self._check_html(captured, [
("Test", "first name"),
("/planner", "planner link"),
("/markets", "markets link"),
("/leads/quote", "quotes link"),
])
@pytest.mark.asyncio
async def test_quote_verification_delivered(self):
wrapper, captured = _send_and_capture()
with patch("resend.Emails.send", side_effect=wrapper):
await handle_send_quote_verification({
"email": "delivered@resend.dev",
"token": "test_verify_tok",
"lead_token": "test_lead_tok",
"contact_name": "Test Buyer",
"court_count": 4,
"facility_type": "Indoor",
"country": "Germany",
"lang": "en",
})
self._check_html(captured, [
("4 courts", "court count"),
("Indoor", "facility type"),
("Germany", "country"),
("token=test_verify_tok", "verify token"),
])
@pytest.mark.asyncio
async def test_waitlist_general_delivered(self):
wrapper, captured = _send_and_capture()
with patch("resend.Emails.send", side_effect=wrapper):
await handle_send_waitlist_confirmation({
"email": "delivered@resend.dev",
"intent": "signup",
"lang": "en",
})
self._check_html(captured, [
("padelnomics", "wordmark"),
("<li>", "perk bullets"),
])
@pytest.mark.asyncio
async def test_waitlist_supplier_delivered(self):
wrapper, captured = _send_and_capture()
with patch("resend.Emails.send", side_effect=wrapper):
await handle_send_waitlist_confirmation({
"email": "delivered@resend.dev",
"intent": "supplier_growth",
"lang": "en",
})
self._check_html(captured, [
("Growth", "plan name"),
])
@pytest.mark.asyncio
async def test_lead_forward_delivered(self, db):
lead_id, supplier_id = await _seed_lead_and_supplier(
db, supplier_email="delivered@resend.dev",
)
wrapper, captured = _send_and_capture()
with patch("resend.Emails.send", side_effect=wrapper):
await handle_send_lead_forward_email({
"lead_id": lead_id,
"supplier_id": supplier_id,
"lang": "en",
})
self._check_html(captured, [
("#DC2626", "HOT badge color"),
("lead@buyer.com", "contact email"),
("mailto:lead@buyer.com", "mailto link"),
("#FEF3C7", "urgency callout bg"),
])
@pytest.mark.asyncio
async def test_lead_matched_delivered(self, db):
lead_id = await _seed_lead(db, contact_email="delivered@resend.dev")
wrapper, captured = _send_and_capture()
with patch("resend.Emails.send", side_effect=wrapper):
await handle_send_lead_matched_notification({
"lead_id": lead_id,
"lang": "en",
})
self._check_html(captured, [
("#F0F9FF", "tip callout bg"),
("Indoor", "facility type"),
("Germany", "country"),
])
@pytest.mark.asyncio
async def test_supplier_enquiry_delivered(self):
wrapper, captured = _send_and_capture()
with patch("resend.Emails.send", side_effect=wrapper):
await handle_send_supplier_enquiry_email({
"supplier_email": "delivered@resend.dev",
"supplier_name": "PadelBuild GmbH",
"contact_name": "Alice Smith",
"contact_email": "alice@buyer.example",
"message": "I need 4 courts, can you quote?",
"lang": "en",
})
self._check_html(captured, [
("Alice Smith", "contact name"),
("alice@buyer.example", "contact email"),
("4 courts", "message content"),
])
@pytest.mark.asyncio
async def test_bounce_handled_gracefully(self):
"""Sending to bounced@resend.dev should not raise — send_email returns False."""
result = await core.send_email(
to="bounced@resend.dev",
subject="Bounce test",
html="<p>This should bounce.</p>",
from_addr=core.EMAIL_ADDRESSES["transactional"],
)
# Resend may return success (delivery fails async) or error;
# either way the handler must not crash.
assert isinstance(result, bool)