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>
98 lines
2.7 KiB
Python
98 lines
2.7 KiB
Python
"""Gumroad adapter: secret check, Ping parsing, refunds stub."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from decimal import Decimal
|
|
|
|
import pytest
|
|
|
|
from app.adapters.gumroad import GumroadAdapter, UnmappedProductError
|
|
|
|
|
|
def _sale_payload(**overrides) -> dict:
|
|
base = {
|
|
"sale_id": "GUM-1001",
|
|
"email": "jane@example.com",
|
|
"full_name": "Jane Doe",
|
|
"product_id": "datatools-core",
|
|
"price": "9900", # cents
|
|
"currency": "usd",
|
|
"offer_code": "",
|
|
"test": "false",
|
|
}
|
|
base.update(overrides)
|
|
return base
|
|
|
|
|
|
def test_verify_secret_correct():
|
|
a = GumroadAdapter(secret="abc123")
|
|
assert a.verify_secret("abc123") is True
|
|
|
|
|
|
def test_verify_secret_wrong():
|
|
a = GumroadAdapter(secret="abc123")
|
|
assert a.verify_secret("nope") is False
|
|
|
|
|
|
def test_verify_secret_unset_rejects_all():
|
|
a = GumroadAdapter(secret=None)
|
|
assert a.verify_secret("anything") is False
|
|
assert a.verify_secret(None) is False
|
|
|
|
|
|
def test_verify_secret_missing_presented_value():
|
|
a = GumroadAdapter(secret="abc123")
|
|
assert a.verify_secret(None) is False
|
|
assert a.verify_secret("") is False
|
|
|
|
|
|
def test_parse_sale_happy_path():
|
|
a = GumroadAdapter(secret="x")
|
|
sale = a.parse_sale(_sale_payload())
|
|
assert sale is not None
|
|
assert sale.source == "gumroad"
|
|
assert sale.source_order_id == "GUM-1001"
|
|
assert sale.buyer_email == "jane@example.com"
|
|
assert sale.buyer_name == "Jane Doe"
|
|
assert sale.tier == "core"
|
|
assert sale.years == 1
|
|
assert sale.amount_paid == Decimal("99.00")
|
|
assert sale.currency == "USD"
|
|
assert sale.promotion is None
|
|
|
|
|
|
def test_parse_sale_with_offer_code():
|
|
a = GumroadAdapter(secret="x")
|
|
sale = a.parse_sale(_sale_payload(offer_code="LAUNCH50"))
|
|
assert sale.promotion == "LAUNCH50"
|
|
|
|
|
|
def test_parse_sale_test_ping_tagged():
|
|
a = GumroadAdapter(secret="x")
|
|
sale = a.parse_sale(_sale_payload(test="true"))
|
|
assert sale.notes == "gumroad test ping"
|
|
|
|
|
|
def test_parse_sale_name_fallback_from_email():
|
|
a = GumroadAdapter(secret="x")
|
|
sale = a.parse_sale(_sale_payload(full_name="", email="john.doe@example.com"))
|
|
assert sale.buyer_name == "John Doe"
|
|
|
|
|
|
def test_parse_sale_missing_required_returns_none():
|
|
a = GumroadAdapter(secret="x")
|
|
assert a.parse_sale(_sale_payload(sale_id="")) is None
|
|
assert a.parse_sale(_sale_payload(email="")) is None
|
|
assert a.parse_sale(_sale_payload(product_id="")) is None
|
|
|
|
|
|
def test_parse_sale_unmapped_product_raises():
|
|
a = GumroadAdapter(secret="x")
|
|
with pytest.raises(UnmappedProductError):
|
|
a.parse_sale(_sale_payload(product_id="no-such-sku"))
|
|
|
|
|
|
def test_parse_refund_stub_returns_none():
|
|
a = GumroadAdapter(secret="x")
|
|
assert a.parse_refund({"any": "payload"}) is None
|