Michael
2bbaba954b
Wires the second source-adapter (Gumroad) plus the email delivery
that lets the server fulfill a sale end-to-end without operator
intervention.
Auth model: Gumroad doesn't HMAC the body, so we use their
recommended URL-secret pattern (?secret=...). Wrong/missing secret
returns 404 — no signal to a prober that the endpoint exists.
Webhook flow (server/app/routes/webhooks.py):
1. audit-log the raw payload (gumroad_events row) BEFORE anything
else, so a later failure leaves us replayable
2. parse via GumroadAdapter (server/app/adapters/gumroad.py)
3. mint_from_sale — UNIQUE(source, source_order_id) dedups
duplicate webhook retries
4. send the license email
5. mark gumroad_events.processed = true
Always returns 200 once auth passes. Non-2xx would trigger Gumroad's
3-day retry storm; we'd rather record the failure on the audit row
and replay manually after fixing whatever surfaced.
Product → tier mapping is per-source YAML at
server/config/products.yaml (lru_cached). Adding a SKU = edit yaml,
restart api. Unmapped product_id is an error on the audit row, not
a crash.
EmailService (server/app/email.py): provider-agnostic interface with
Postmark as the first implementation. When POSTMARK_TOKEN is unset
the factory returns LoggingEmailService instead, so the webhook
exercises end-to-end before Postmark is provisioned.
48 unit tests (was 21) including:
- Gumroad secret verify with constant-time compare
- Sale parsing: amount-in-cents, name fallback from email,
test=true tagging, missing-required fields, offer codes
- Product mapping lookups
- Email rendering text + HTML, HTML-escapes user input
- Postmark client via httpx.MockTransport (success and 4xx)
- Webhook end-to-end: secret check, audit log, idempotency on
retry, unmapped product, email failure keeps license
Smoke test (server/scripts/smoke.sh) extended to POST a synthetic
Ping payload, verify the row + audit log, prove wrong-secret is
rejected, prove duplicate sale_id stays one row.
SQLite-test compatibility:
- BigInteger primary key uses with_variant(Integer, "sqlite") since
SQLite only autoincrements INTEGER PRIMARY KEY.
- python-multipart pulled in for FastAPI Form parsing.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
144 lines
4.9 KiB
Python
144 lines
4.9 KiB
Python
"""End-to-end webhook tests — secret check, audit log, mint, email,
|
|
idempotency."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
|
|
from app.email import LicenseEmail
|
|
from app.models import GumroadEvent, License
|
|
|
|
|
|
@pytest.fixture
|
|
def with_gumroad_secret(monkeypatch):
|
|
monkeypatch.setenv("GUMROAD_WEBHOOK_SECRET", "test-gumroad-secret")
|
|
from app.config import get_settings
|
|
get_settings.cache_clear()
|
|
yield
|
|
get_settings.cache_clear()
|
|
|
|
|
|
def _form(**overrides) -> dict:
|
|
base = {
|
|
"sale_id": "GUM-2001",
|
|
"email": "jane@example.com",
|
|
"full_name": "Jane Doe",
|
|
"product_id": "datatools-core",
|
|
"price": "9900",
|
|
"currency": "usd",
|
|
"offer_code": "",
|
|
"test": "false",
|
|
}
|
|
base.update(overrides)
|
|
return base
|
|
|
|
|
|
def test_webhook_rejects_missing_secret(client, with_gumroad_secret):
|
|
r = client.post("/webhooks/gumroad", data=_form())
|
|
assert r.status_code == 404
|
|
|
|
|
|
def test_webhook_rejects_wrong_secret(client, with_gumroad_secret):
|
|
r = client.post("/webhooks/gumroad?secret=wrong", data=_form())
|
|
assert r.status_code == 404
|
|
|
|
|
|
def test_webhook_mints_and_sends_email(client, with_gumroad_secret, db_session):
|
|
captured: list[LicenseEmail] = []
|
|
|
|
class CapturingEmail:
|
|
def send_license(self, msg):
|
|
captured.append(msg)
|
|
return "captured"
|
|
|
|
with patch("app.routes.webhooks.get_email_service", lambda: CapturingEmail()):
|
|
r = client.post(
|
|
"/webhooks/gumroad?secret=test-gumroad-secret",
|
|
data=_form(),
|
|
)
|
|
assert r.status_code == 200
|
|
body = r.json()
|
|
assert body["status"] == "ok"
|
|
assert body["license_key"].startswith("DT1-CORE-")
|
|
|
|
# DB row landed.
|
|
row = db_session.query(License).filter_by(source_order_id="GUM-2001").one()
|
|
assert row.source == "gumroad"
|
|
assert row.email == "jane@example.com"
|
|
assert row.tier == "core"
|
|
assert row.blob.startswith("DTLIC2:")
|
|
|
|
# Audit row processed.
|
|
event = db_session.query(GumroadEvent).filter_by(order_id="GUM-2001").one()
|
|
assert event.processed is True
|
|
assert event.error is None
|
|
|
|
# Email captured.
|
|
assert len(captured) == 1
|
|
assert captured[0].to_email == "jane@example.com"
|
|
assert captured[0].blob == row.blob
|
|
|
|
|
|
def test_webhook_idempotent_on_duplicate_sale_id(client, with_gumroad_secret, db_session):
|
|
"""Gumroad retries on transient failures. Same sale_id must
|
|
produce one license, not two."""
|
|
with patch("app.routes.webhooks.get_email_service", lambda: _NullEmail()):
|
|
client.post("/webhooks/gumroad?secret=test-gumroad-secret", data=_form())
|
|
client.post("/webhooks/gumroad?secret=test-gumroad-secret", data=_form())
|
|
|
|
rows = db_session.query(License).filter_by(source_order_id="GUM-2001").all()
|
|
assert len(rows) == 1
|
|
# But both webhook deliveries should be in the audit log.
|
|
events = db_session.query(GumroadEvent).filter_by(order_id="GUM-2001").all()
|
|
assert len(events) == 2
|
|
|
|
|
|
def test_webhook_unmapped_product_audits_and_returns_200(client, with_gumroad_secret, db_session):
|
|
"""An unknown product_id must NOT crash and must NOT trigger a
|
|
retry storm. Audit row gets the error reason."""
|
|
with patch("app.routes.webhooks.get_email_service", lambda: _NullEmail()):
|
|
r = client.post(
|
|
"/webhooks/gumroad?secret=test-gumroad-secret",
|
|
data=_form(product_id="not-a-real-sku", sale_id="GUM-666"),
|
|
)
|
|
assert r.status_code == 200
|
|
assert r.json()["status"] == "logged-no-mint"
|
|
|
|
# No license, but audit row with error.
|
|
assert db_session.query(License).filter_by(source_order_id="GUM-666").count() == 0
|
|
event = db_session.query(GumroadEvent).filter_by(order_id="GUM-666").one()
|
|
assert event.processed is False
|
|
assert "no entry in config/products.yaml" in (event.error or "")
|
|
|
|
|
|
def test_webhook_email_failure_keeps_license(client, with_gumroad_secret, db_session):
|
|
"""If Postmark hiccups, the buyer's license is still minted and
|
|
persists in the DB. They can be served from the renewal portal
|
|
(PR 3) or a manual resend."""
|
|
from app.email import EmailDeliveryError
|
|
|
|
class FailingEmail:
|
|
def send_license(self, msg):
|
|
raise EmailDeliveryError("Postmark 503")
|
|
|
|
with patch("app.routes.webhooks.get_email_service", lambda: FailingEmail()):
|
|
r = client.post(
|
|
"/webhooks/gumroad?secret=test-gumroad-secret",
|
|
data=_form(sale_id="GUM-3001"),
|
|
)
|
|
assert r.status_code == 200
|
|
assert r.json()["status"] == "minted-email-failed"
|
|
|
|
assert db_session.query(License).filter_by(source_order_id="GUM-3001").count() == 1
|
|
event = db_session.query(GumroadEvent).filter_by(order_id="GUM-3001").one()
|
|
assert event.processed is True # we count it processed so Gumroad doesn't retry
|
|
assert "email error" in (event.error or "")
|
|
|
|
|
|
class _NullEmail:
|
|
def send_license(self, msg):
|
|
return "null"
|