paypal #93

Merged
russellballestrini merged 21 commits from feature/Paypal_checkout into master 2025-12-22 17:29:52 -05:00
27 changed files with 3832 additions and 10 deletions

37
CHANGELOG.rst Normal file
View file

@ -0,0 +1,37 @@
Changelog
=========
All notable changes to this project will be documented in this file.
2025-12-22 (2:30 PM)
--------------------
PayPal Integration
~~~~~~~~~~~~~~~~~~
* Added PayPal as a payment processor alongside Stripe and crypto payments
* New ``PayPalUserShop`` model for saved payment methods
* Invoice model extended with ``paypal_order_id`` and ``paypal_capture_id`` columns
* Shop settings now include PayPal client ID and secret configuration
* Checkout page supports PayPal payment option when enabled
* Added PayPal saved payment methods (vault) support
* Added ``/billing/disconnect-paypal`` route for users to manage saved PayPal
* See ``docs/PAYPAL.md`` for details
CSS Grid Lanes
~~~~~~~~~~~~~~
* Added toggleable CSS Grid Lanes (masonry layout) setting per shop
* New ``grid_lanes_enabled`` column on Shop model
Video Thumbnails
~~~~~~~~~~~~~~~~
* Added play button overlay on video thumbnails for unlocked content
* Styled video play overlay with red tint and click-to-play text
Meta Tags
~~~~~~~~~
* Added Twitter card meta tags for proper link unfurling on Matrix/Discord
* Increased meta description truncation to 500 chars

View file

@ -3,7 +3,7 @@ Make Post Sell
The `Make Post Sell <https://www.makepostsell.com>`_ monolith platform service.
You can use the SaaS or self-host! Accepts credit cards, Monero (XMR), and Dogecoin (DOGE) crypto payments.
You can use the SaaS or self-host! Accepts credit cards (Stripe), PayPal, Monero (XMR), and Dogecoin (DOGE) payments.
Our `blog acts as our user guide <https://blog.makepostsell.com/>`_ & also uses ``make_post_sell``!

View file

@ -64,8 +64,13 @@ app.bucket.secure_uploads.secret_key = ${MPS_APP_SECURE_UPLOADS_SECRET_KEY}
# stripe test mode is enabled for development & disabled by default.
app.stripe.test_mode = True
# PayPal sandbox mode is enabled for development & disabled by default.
app.paypal.sandbox_mode = ${MPS_PAYPAL_SANDBOX_MODE:-True}
app.paypal.webhook_id = ${MPS_PAYPAL_WEBHOOK_ID:-}
# Payment method toggles
app.payments.stripe.enabled = ${MPS_PAYMENTS_STRIPE_ENABLED:-True}
app.payments.paypal.enabled = ${MPS_PAYMENTS_PAYPAL_ENABLED:-True}
app.payments.monero.enabled = ${MPS_PAYMENTS_MONERO_ENABLED:-False}
app.payments.dogecoin.enabled = ${MPS_PAYMENTS_DOGECOIN_ENABLED:-False}

132
docs/ADYEN.md Normal file
View file

@ -0,0 +1,132 @@
# Adyen Payments
Adyen is a payment processor that supports cards, wallets, and local payment methods.
**Status: Implemented**
## Overview
Adyen provides similar functionality to Stripe with a server-side API for processing payments. The integration pattern would be similar to our existing Stripe implementation.
## Privacy/Verification Requirements
Like PayPal and Stripe, Adyen requires business verification. However, Adyen's process is generally less invasive than PayPal's:
**What Adyen requires:**
- Business registration documents (company registration, articles of incorporation)
- Proof of identity for account holders (government ID photo)
- Bank account verification for payouts
- Proof of address (utility bill, bank statement)
**What Adyen does NOT require (unlike PayPal):**
- No face scanning / biometric capture
- No selfies or photos of your face
- Auto-verification attempted first before manual document requests
**Onboarding process:**
1. Self-serve signup at adyen.com
2. Adyen attempts automatic verification first
3. If auto-verification fails, they request documents via dashboard
4. Once verified, you can process live payments
Similar KYC (Know Your Customer) requirements as other payment processors, but less invasive than PayPal. If privacy is a priority, use crypto payments (XMR/DOGE) instead.
## Technical Integration
### Python Library
Official library: https://github.com/Adyen/adyen-python-api-library
```bash
pip install Adyen
```
### Basic Usage
```python
import Adyen
adyen = Adyen.Adyen()
adyen.client.xapikey = "YOUR_API_KEY"
adyen.client.platform = "test" # or "live"
# Create payment
result = adyen.checkout.payments_api.payments({
"amount": {"currency": "USD", "value": 1000}, # $10.00 in cents
"reference": f"invoice_{invoice.uuid_str}",
"merchantAccount": "YOUR_MERCHANT_ACCOUNT",
"paymentMethod": {
"type": "scheme",
"number": "4111111111111111",
"expiryMonth": "03",
"expiryYear": "2030",
"cvc": "737"
},
"returnUrl": "https://your-site.com/checkout/result"
})
```
### Required Credentials
Each shop would need:
| Credential | Description |
|------------|-------------|
| `adyen_api_key` | API key from Adyen dashboard |
| `adyen_merchant_account` | Merchant account identifier |
| `adyen_client_key` | Client-side key for Drop-in/Components |
| `adyen_hmac_key` | HMAC key for webhook verification |
### Webhooks
Adyen uses HMAC-SHA256 for webhook verification:
```python
import hashlib
import hmac
import base64
def verify_hmac(hmac_key, hmac_signature, payload):
expected = hmac.new(
binascii.unhexlify(hmac_key),
payload.encode('utf-8'),
hashlib.sha256
).digest()
expected_signature = base64.b64encode(expected).decode('utf-8')
return hmac.compare_digest(hmac_signature, expected_signature)
```
### Key Events
- `AUTHORISATION` - Payment authorized
- `CAPTURE` - Payment captured
- `REFUND` - Refund processed
- `CHARGEBACK` - Dispute/chargeback created
## Implementation Plan
To add Adyen support:
1. Add shop columns: `adyen_api_key`, `adyen_merchant_account`, `adyen_client_key`, `adyen_hmac_key`, `adyen_enabled`
2. Add invoice columns: `adyen_psp_reference` (payment reference)
3. Create checkout view similar to Stripe PaymentIntent flow
4. Add webhook handler with HMAC verification
5. Add shop settings UI for Adyen credentials
## Resources
- Python Library: https://github.com/Adyen/adyen-python-api-library
- Example Integration: https://github.com/adyen-examples/adyen-python-online-payments
- API Explorer: https://docs.adyen.com/api-explorer/
- Build Your Integration: https://docs.adyen.com/online-payments/build-your-integration
- Webhooks: https://docs.adyen.com/development-resources/webhooks
## Comparison with Other Processors
| Feature | Stripe | PayPal | Adyen |
|---------|--------|--------|-------|
| Python Library | `stripe` | `requests` | `Adyen` |
| Webhook Auth | Signature secret | Signature verification API | HMAC-SHA256 |
| Test Mode | `sk_test_*` keys | Sandbox mode | Test merchant account |
| KYC Required | Yes | Yes (invasive) | Yes |
| Self-serve signup | Yes | Yes | Yes |

81
docs/PAYPAL.md Normal file
View file

@ -0,0 +1,81 @@
# PayPal Payments
PayPal is available as a payment method alongside Stripe and crypto (XMR/DOGE).
## Privacy Warning
PayPal requires invasive identity verification to receive payments:
- ML-based face scanning (biometric capture)
- Two images of your face from different angles
- Photos of both sides of government-issued ID (driver's license, passport)
- Business verification for merchant accounts
This verification is required to move from sandbox to live production payments. There is no way to accept PayPal payments anonymously or privately. If privacy is important to you, consider crypto payments (XMR/DOGE) instead.
## Shop Setup
1. Go to https://developer.paypal.com/dashboard/
2. Create an app (sandbox for testing, live for production)
3. Complete identity verification (face scan + government ID)
4. Copy Client ID and Secret
5. In Shop Settings → PayPal Settings, enter credentials and save
## Configuration
Global settings in `development.ini`:
```ini
app.payments.paypal.enabled = True
app.paypal.sandbox_mode = True
```
Or environment variables:
```bash
export MPS_PAYMENTS_PAYPAL_ENABLED=True
export MPS_PAYPAL_SANDBOX_MODE=True
```
## How It Works
- Each shop configures their own PayPal credentials
- PayPal button appears at checkout when enabled
- Payment info stored on Invoice (`paypal_order_id`, `paypal_capture_id`)
- `Invoice.payment_method` returns "paypal" for PayPal payments
- Saved payment methods stored in `PayPalUserShop`
## Webhooks
Webhooks provide resilience when JavaScript callbacks fail. Configure in PayPal Developer Dashboard:
1. Go to https://developer.paypal.com/dashboard/applications
2. Select your app → Webhooks → Add Webhook
3. Enter URL: `https://yourdomain.com/webhooks/paypal`
4. Subscribe to events:
- `PAYMENT.CAPTURE.COMPLETED`
- `CHECKOUT.ORDER.APPROVED`
- `PAYMENT.CAPTURE.DENIED`
- `CUSTOMER.DISPUTE.CREATED`
5. Copy the Webhook ID
6. Set in your ini: `paypal.webhook_id = YOUR_WEBHOOK_ID`
## Code
- `make_post_sell/views/cart.py` - PayPal checkout functions (`paypal_create_order`, `paypal_complete_checkout`)
- `make_post_sell/views/billing.py` - Disconnect saved PayPal
- `make_post_sell/views/webhooks.py` - Webhook handlers
## Database
**mps_shop columns:**
- `paypal_client_id` - Shop's PayPal Client ID
- `paypal_secret` - Shop's PayPal Secret
- `paypal_enabled` - Toggle PayPal on/off
**mps_invoice columns:**
- `paypal_order_id` - PayPal order reference
- `paypal_capture_id` - PayPal capture reference
**mps_paypal_user_shop table:**
- Tracks saved PayPal payment methods per user/shop

View file

@ -25,6 +25,7 @@ from .crypto_processor import *
from .user_crypto_refund_address import *
from .stripe_user_shop import *
from .paypal_user_shop import *
from .shop_search_request import *
from .comment import *

View file

@ -103,6 +103,17 @@ class Invoice(RBase, Base):
# denormalized address of user for shipping products.
delivery_address = Column(UnicodeText, nullable=True)
# PayPal payment tracking (nullable - only set for PayPal payments)
paypal_order_id = Column(Unicode(64), nullable=True)
paypal_capture_id = Column(Unicode(64), nullable=True)
# Stripe payment tracking (nullable - only set for Stripe payments)
stripe_payment_intent_id = Column(Unicode(64), nullable=True)
stripe_charge_id = Column(Unicode(64), nullable=True)
# Adyen payment tracking (nullable - only set for Adyen payments)
adyen_psp_reference = Column(Unicode(64), nullable=True)
# one to one.
user = relationship(argument="User", uselist=False, lazy="joined")
@ -260,16 +271,16 @@ class Invoice(RBase, Base):
@property
def payment_status(self):
"""Get payment status from crypto_payment or assume paid for Stripe."""
"""Get payment status from crypto_payment or assume paid for Stripe/PayPal."""
if hasattr(self, "crypto_payment") and self.crypto_payment:
return self.crypto_payment.status
else:
# If invoice exists without crypto_payment, it's a successful Stripe payment
# If invoice exists without crypto_payment, it's a successful Stripe/PayPal payment
return "paid"
@property
def payment_method(self):
"""Get payment method from crypto_payment or return 'stripe' for card payments."""
"""Get payment method: crypto, paypal, or stripe."""
try:
if hasattr(self, "crypto_payment") and self.crypto_payment:
# crypto_payment is a collection, get the first one
@ -281,8 +292,17 @@ class Invoice(RBase, Base):
elif hasattr(self.crypto_payment, "coin_type"):
return self.crypto_payment.coin_type.lower()
except Exception:
# Fall back to stripe if there's any issue accessing crypto_payment
pass
# Check for PayPal payment
if self.paypal_order_id:
return "paypal"
# Check for Adyen payment
if self.adyen_psp_reference:
return "adyen"
# Check for Stripe payment (or assume Stripe for legacy invoices)
if self.stripe_payment_intent_id:
return "stripe"
# Default to stripe for legacy invoices without explicit payment tracking
return "stripe"
@property
@ -301,6 +321,27 @@ def get_invoice_by_id(dbsession, invoice_id):
return get_object_by_id(dbsession, invoice_id, Invoice)
def get_invoice_by_paypal_order_id(dbsession, paypal_order_id):
"""Try to get Invoice object by PayPal order ID or return None."""
return dbsession.query(Invoice).filter(
Invoice.paypal_order_id == paypal_order_id
).first()
def get_invoice_by_stripe_payment_intent_id(dbsession, payment_intent_id):
"""Try to get Invoice object by Stripe payment intent ID or return None."""
return dbsession.query(Invoice).filter(
Invoice.stripe_payment_intent_id == payment_intent_id
).first()
def get_invoice_by_adyen_psp_reference(dbsession, psp_reference):
"""Try to get Invoice object by Adyen PSP reference or return None."""
return dbsession.query(Invoice).filter(
Invoice.adyen_psp_reference == psp_reference
).first()
def delete_invoice_by_id(dbsession, invoice_id):
"""
Safely delete an invoice and its line items, but only if it's from a terminated/unsuccessful crypto payment.

View file

@ -35,6 +35,7 @@ CLASS_TO_TABLE = {
"InvoiceLineItem": "mps_invoice_line_item",
"ShopSearchRequest": "mps_shop_search_request",
"StripeUserShop": "mps_stripe_user_shop",
"PayPalUserShop": "mps_paypal_user_shop",
"Market": "mps_market",
"Comment": "mps_comment",
"CryptoPayment": "mps_crypto_payment",

View file

@ -0,0 +1,55 @@
import uuid
from sqlalchemy import Column, Unicode
from .meta import Base, RBase, UUIDType, foreign_key
from sqlalchemy.orm import relationship, backref
class PayPalUserShop(RBase, Base):
"""
A user may have zero or many unique PayPal payer IDs for each shop it makes purchases on.
This tracks saved PayPal payment methods and customer relationships per shop.
"""
id = Column(UUIDType, primary_key=True, index=True)
user_id = Column(UUIDType, foreign_key("User", "id"), nullable=False)
shop_id = Column(UUIDType, foreign_key("Shop", "id"), nullable=False)
# PayPal payer ID (e.g., "PAYERID123ABC")
payer_id = Column(Unicode(64), nullable=True)
# Billing agreement ID for reference transactions (optional)
billing_agreement_id = Column(Unicode(64), nullable=True)
# Active payment token for saved payment methods (optional)
active_payment_token = Column(Unicode(128), nullable=True)
user = relationship(
argument="User", backref=backref("paypal_user", cascade="all, delete-orphan")
)
shop = relationship(
argument="Shop", backref=backref("paypal_shop", cascade="all, delete-orphan")
)
def __init__(self, user=None, shop=None):
self.id = uuid.uuid1()
self.user = user
self.shop = shop
@property
def has_billing_agreement(self):
"""Check if this user has an active billing agreement with PayPal."""
return self.billing_agreement_id is not None
@property
def has_saved_payment_method(self):
"""Check if this user has a saved payment token."""
return self.active_payment_token is not None
def get_all_paypal_user_shop_objects(dbsession):
"""Return all PayPalUserShop objects."""
return dbsession.query(PayPalUserShop).all()

View file

@ -79,6 +79,18 @@ class Shop(RBase, Base):
stripe_public_api_key = Column(Unicode(128), nullable=True)
stripe_enabled = Column(Boolean, default=True)
# PayPal API credentials for accepting payments
paypal_client_id = Column(Unicode(128), nullable=True)
paypal_secret = Column(Unicode(128), nullable=True)
paypal_enabled = Column(Boolean, default=True)
# Adyen API credentials for accepting payments
adyen_api_key = Column(Unicode(128), nullable=True)
adyen_merchant_account = Column(Unicode(128), nullable=True)
adyen_client_key = Column(Unicode(128), nullable=True)
adyen_hmac_key = Column(Unicode(128), nullable=True)
adyen_enabled = Column(Boolean, default=True)
created_timestamp = Column(BigInteger, nullable=False)
updated_timestamp = Column(BigInteger, nullable=False)
@ -230,6 +242,28 @@ class Shop(RBase, Base):
def is_stripe_not_ready(self):
return not self.is_stripe_ready
@property
def is_paypal_ready(self):
"""Check if shop has PayPal API credentials configured."""
if self.paypal_client_id and self.paypal_secret:
return True
return False
@property
def is_paypal_not_ready(self):
return not self.is_paypal_ready
@property
def is_adyen_ready(self):
"""Check if shop has Adyen API credentials configured."""
if self.adyen_api_key and self.adyen_merchant_account:
return True
return False
@property
def is_adyen_not_ready(self):
return not self.is_adyen_ready
def is_ready_for_payment(self, request):
"""Check if shop is ready based on enabled payment methods."""
# If Stripe is enabled, shop needs Stripe API keys
@ -237,6 +271,16 @@ class Shop(RBase, Base):
if self.is_stripe_ready:
return True
# If PayPal is enabled, shop needs PayPal API credentials
if request.paypal_enabled:
if self.is_paypal_ready:
return True
# If Adyen is enabled, shop needs Adyen API credentials
if getattr(request, "adyen_enabled", False):
if self.is_adyen_ready:
return True
# If Monero is enabled, check if shop has configured processor and RPC is available
if request.monero_enabled and request.monero_rpc_available:
from .crypto_processor import CryptoProcessor
@ -339,6 +383,84 @@ class Shop(RBase, Base):
return self.stripe.Charge.list(customer=stripe_customer)
return self.stripe.Charge.list()
@property
def paypal(self):
"""Return a PayPal SDK API instance using this shop's credentials."""
if hasattr(self, "_paypal") == False:
if self.paypal_client_id and self.paypal_secret:
import paypalrestsdk
# Get sandbox mode from request/config if available
# Default to sandbox for safety
mode = "sandbox"
if hasattr(self, "dbsession") and self.dbsession:
try:
from pyramid.threadlocal import get_current_request
request = get_current_request()
if request and hasattr(request, "app"):
sandbox_mode = request.app.get("paypal.sandbox_mode", True)
if isinstance(sandbox_mode, str):
sandbox_mode = sandbox_mode.strip().lower() in ("1", "true", "yes", "on")
if not sandbox_mode:
mode = "live"
except:
pass
api = paypalrestsdk.Api({
'mode': mode,
'client_id': self.paypal_client_id,
'client_secret': self.paypal_secret
})
self._paypal = api
else:
self._paypal = None
return self._paypal
def paypal_user_shop(self, user):
"""Return the paypal_user_shop object from our database for this user, or None."""
from .paypal_user_shop import PayPalUserShop
return (
self.dbsession.query(PayPalUserShop)
.filter(
PayPalUserShop.user == user,
PayPalUserShop.shop == self,
)
.one_or_none()
)
@property
def adyen(self):
"""Return an Adyen SDK instance using this shop's credentials."""
if hasattr(self, "_adyen") == False:
if self.adyen_api_key and self.adyen_merchant_account:
import Adyen
adyen = Adyen.Adyen()
adyen.client.xapikey = self.adyen_api_key
# Get test/live mode from request/config if available
# Default to test for safety
platform = "test"
if hasattr(self, "dbsession") and self.dbsession:
try:
from pyramid.threadlocal import get_current_request
request = get_current_request()
if request and hasattr(request, "app"):
test_mode = request.app.get("adyen.test_mode", True)
if isinstance(test_mode, str):
test_mode = test_mode.strip().lower() in ("1", "true", "yes", "on")
if not test_mode:
platform = "live"
except:
pass
adyen.client.platform = platform
self._adyen = adyen
else:
self._adyen = None
return self._adyen
@property
def theme_base_color(self):
# return the user defined base for shop or default.

View file

@ -201,6 +201,48 @@ def includeme(config):
return val
return False
def add_paypal_enabled(request):
"""Check if PayPal payments are enabled globally and for the current shop."""
# If globally disabled, return False
if not request.paypal_globally_enabled:
return False
# Check per-shop setting if shop is available
if hasattr(request, "shop") and request.shop:
return getattr(request.shop, "paypal_enabled", True)
return request.paypal_globally_enabled
def add_paypal_globally_enabled(request):
"""Check if PayPal payments are enabled globally (ignoring per-shop setting)."""
val = request.app.get("payments.paypal.enabled")
if isinstance(val, str):
return val.strip().lower() in ("1", "true", "yes", "on")
elif isinstance(val, bool):
return val
return False
def add_adyen_enabled(request):
"""Check if Adyen payments are enabled globally and for the current shop."""
# If globally disabled, return False
if not request.adyen_globally_enabled:
return False
# Check per-shop setting if shop is available
if hasattr(request, "shop") and request.shop:
return getattr(request.shop, "adyen_enabled", True)
return request.adyen_globally_enabled
def add_adyen_globally_enabled(request):
"""Check if Adyen payments are enabled globally (ignoring per-shop setting)."""
val = request.app.get("payments.adyen.enabled")
if isinstance(val, str):
return val.strip().lower() in ("1", "true", "yes", "on")
elif isinstance(val, bool):
return val
return False
def add_monero_enabled(request):
"""Check if Monero payments are enabled globally."""
try:
@ -325,6 +367,14 @@ def includeme(config):
config.add_request_method(
add_stripe_globally_enabled, "stripe_globally_enabled", reify=True
)
config.add_request_method(add_paypal_enabled, "paypal_enabled", reify=True)
config.add_request_method(
add_paypal_globally_enabled, "paypal_globally_enabled", reify=True
)
config.add_request_method(add_adyen_enabled, "adyen_enabled", reify=True)
config.add_request_method(
add_adyen_globally_enabled, "adyen_globally_enabled", reify=True
)
config.add_request_method(add_monero_enabled, "monero_enabled", reify=True)
config.add_request_method(
add_monero_rpc_available, "monero_rpc_available", reify=True

View file

@ -24,6 +24,20 @@ def includeme(config):
"confirm-update-card", "/billing/confirm-update-card/{action}/{card_id}"
)
config.add_route("update-card", "/billing/update-card")
config.add_route("disconnect-paypal", "/billing/disconnect-paypal")
# PayPal routes
config.add_route("paypal_create_order", "/paypal/create-order/{cart_id}")
config.add_route("paypal_complete_checkout", "/paypal/complete-checkout/{cart_id}")
# Adyen routes
config.add_route("adyen_create_session", "/adyen/create-session/{cart_id}")
config.add_route("adyen_complete_checkout", "/adyen/complete-checkout/{cart_id}")
# Webhook routes
config.add_route("paypal_webhook", "/webhooks/paypal")
config.add_route("stripe_webhook", "/webhooks/stripe")
config.add_route("adyen_webhook", "/webhooks/adyen")
# user routes.
config.add_route("user_settings", "/u/settings")

View file

@ -0,0 +1,58 @@
"""add adyen payment columns to shop and invoice
Revision ID: 3734955e7379
Revises: 63d935094f97
Create Date: 2025-12-22 16:19:40.557406
"""
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision = '3734955e7379'
down_revision = '63d935094f97'
branch_labels = None
depends_on = None
def upgrade():
# Add Adyen columns to mps_shop
op.add_column(
'mps_shop',
sa.Column('adyen_api_key', sa.Unicode(128), nullable=True)
)
op.add_column(
'mps_shop',
sa.Column('adyen_merchant_account', sa.Unicode(128), nullable=True)
)
op.add_column(
'mps_shop',
sa.Column('adyen_client_key', sa.Unicode(128), nullable=True)
)
op.add_column(
'mps_shop',
sa.Column('adyen_hmac_key', sa.Unicode(128), nullable=True)
)
op.add_column(
'mps_shop',
sa.Column('adyen_enabled', sa.Boolean(), nullable=False, server_default='1')
)
# Add Adyen PSP reference column to mps_invoice
op.add_column(
'mps_invoice',
sa.Column('adyen_psp_reference', sa.Unicode(64), nullable=True)
)
def downgrade():
# Remove Invoice column
op.drop_column('mps_invoice', 'adyen_psp_reference')
# Remove Shop columns
op.drop_column('mps_shop', 'adyen_enabled')
op.drop_column('mps_shop', 'adyen_hmac_key')
op.drop_column('mps_shop', 'adyen_client_key')
op.drop_column('mps_shop', 'adyen_merchant_account')
op.drop_column('mps_shop', 'adyen_api_key')

View file

@ -0,0 +1,51 @@
"""add paypal support
Revision ID: 418933067d81
Revises: a7c3e8f1d2b4
Create Date: 2025-12-22 12:06:14.945882
"""
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision = '418933067d81'
down_revision = 'a7c3e8f1d2b4'
branch_labels = None
depends_on = None
def upgrade():
# Add PayPal credentials columns to mps_shop table
op.add_column(
"mps_shop",
sa.Column("paypal_client_id", sa.Unicode(128), nullable=True),
)
op.add_column(
"mps_shop",
sa.Column("paypal_secret", sa.Unicode(128), nullable=True),
)
op.add_column(
"mps_shop",
sa.Column("paypal_enabled", sa.Boolean(), nullable=False, server_default="1"),
)
# Add PayPal payment tracking columns to mps_invoice table
op.add_column(
"mps_invoice",
sa.Column("paypal_order_id", sa.Unicode(64), nullable=True),
)
op.add_column(
"mps_invoice",
sa.Column("paypal_capture_id", sa.Unicode(64), nullable=True),
)
def downgrade():
# Remove PayPal columns from mps_invoice table
op.drop_column("mps_invoice", "paypal_capture_id")
op.drop_column("mps_invoice", "paypal_order_id")
# Remove PayPal columns from mps_shop table
op.drop_column("mps_shop", "paypal_enabled")
op.drop_column("mps_shop", "paypal_secret")
op.drop_column("mps_shop", "paypal_client_id")

View file

@ -0,0 +1,33 @@
"""add stripe payment tracking columns to invoice
Revision ID: 63d935094f97
Revises: 418933067d81
Create Date: 2025-12-22 15:54:08.424454
"""
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision = '63d935094f97'
down_revision = '418933067d81'
branch_labels = None
depends_on = None
def upgrade():
# Add Stripe payment tracking columns to mps_invoice
op.add_column(
'mps_invoice',
sa.Column('stripe_payment_intent_id', sa.Unicode(64), nullable=True)
)
op.add_column(
'mps_invoice',
sa.Column('stripe_charge_id', sa.Unicode(64), nullable=True)
)
def downgrade():
op.drop_column('mps_invoice', 'stripe_charge_id')
op.drop_column('mps_invoice', 'stripe_payment_intent_id')

View file

@ -38,6 +38,38 @@
</section>
{# PayPal Saved Payment Method Section #}
{% if paypal_enabled and paypal_user_shop %}
<section class="well" style="margin-top: 20px;">
<h3>PayPal Payment Method</h3>
{% if paypal_user_shop.has_saved_payment_method %}
<div style="padding: 15px; background-color: #f0f7ff; border-left: 4px solid #0070ba; border-radius: 4px; margin-top: 10px;">
<div style="display: flex; align-items: center; gap: 10px; margin-bottom: 8px;">
<svg width="24" height="24" viewBox="0 0 24 24" fill="#0070ba">
<path d="M20.067 8.478c.492.88.556 2.014.3 3.327-.74 3.806-3.276 5.12-6.514 5.12h-.5a.805.805 0 00-.794.68l-.04.22-.63 3.993-.032.17a.804.804 0 01-.794.679H7.72a.483.483 0 01-.477-.558L9.01 8.975a.964.964 0 01.951-.814h2.053c3.893 0 6.535 1.62 7.055 6.317z"/>
<path d="M10.964 8.975a.964.964 0 00-.95.814L8.247 21.891a.483.483 0 00.476.558h3.344c.336 0 .622-.243.678-.574l.028-.15.63-3.994.04-.218a.805.805 0 01.795-.68h.5c3.238 0 5.774-1.314 6.514-5.12.256-1.313.192-2.447-.3-3.327a3.918 3.918 0 00-1.172-1.241c-.67-.437-1.53-.695-2.527-.827-.27-.036-.549-.063-.837-.083h-.18z" opacity=".7"/>
</svg>
<span style="font-weight: 500; font-size: 16px;">PayPal Account Connected</span>
</div>
<p style="margin: 0 0 12px 0; color: #666; font-size: 14px;">
Your PayPal account is saved for quick checkout. Click the PayPal button at checkout to complete your purchase.
</p>
<form method="POST" action="/billing/disconnect-paypal" style="margin: 0;">
<input type="hidden" name="csrf_token" value="{{ request.session.get_csrf_token() }}">
<button type="submit" style="padding: 8px 16px; background-color: #fff; border: 1px solid #ccc; border-radius: 4px; cursor: pointer; font-size: 14px; color: #333;">
Disconnect PayPal
</button>
</form>
</div>
{% else %}
<p style="color: #666; margin-top: 10px;">
No PayPal account connected. Check the "Save PayPal" box during checkout to save your PayPal account for faster future purchases.
</p>
{% endif %}
</section>
{% endif %}
<center>
<a href="/cart" class="product-new-button mps-button">Review Order</a>
&nbsp;

View file

@ -56,6 +56,150 @@
</form>
{% endif %}
{# PayPal checkout button #}
{% if paypal_enabled and request.shop and request.shop.is_paypal_ready and cart.requires_payment %}
<br/>
{# Saved PayPal status or save checkbox #}
{% if paypal_user_shop and paypal_user_shop.has_saved_payment_method %}
{# User has PayPal saved #}
<div style="margin-bottom: 15px; padding: 12px; background-color: #f0f7ff; border-left: 4px solid: #0070ba; border-radius: 4px;">
<div style="display: flex; align-items: center; gap: 8px;">
<span style="font-size: 20px;">✓</span>
<span style="font-weight: 500; color: #0070ba;">PayPal saved for quick checkout</span>
</div>
<small style="display: block; margin-top: 4px; color: #666;">
Click the PayPal button below to complete your purchase
</small>
</div>
{% else %}
{# User does not have PayPal saved - show checkbox #}
<div style="margin-bottom: 15px;">
<label style="display: flex; align-items: center; gap: 8px; cursor: pointer;">
<input type="checkbox" id="save-paypal-checkbox" name="save_paypal" value="true" style="cursor: pointer;">
<span>💾 Save PayPal for faster checkout next time</span>
</label>
<small style="display: block; margin-left: 24px; color: #666; margin-top: 4px;">
You can manage saved payment methods in your account settings
</small>
</div>
{% endif %}
<div id="paypal-button-container"></div>
<script src="https://www.paypal.com/sdk/js?client-id={{ request.shop.paypal_client_id }}&currency=USD"></script>
<script>
// Store all order IDs for multi-shop support
var allPayPalOrderIds = [];
// Double-click protection flags
var isCreatingOrder = false;
var isProcessingApproval = false;
paypal.Buttons({
style: {
layout: 'vertical',
color: 'gold',
shape: 'rect',
label: 'paypal'
},
createOrder: function(data, actions) {
// Prevent duplicate order creation
if (isCreatingOrder) {
console.warn('PayPal order creation already in progress');
return Promise.reject(new Error('Order creation already in progress'));
}
isCreatingOrder = true;
// Check if user wants to save PayPal for future use
var savePayPalCheckbox = document.getElementById('save-paypal-checkbox');
var savePayPal = savePayPalCheckbox ? savePayPalCheckbox.checked : false;
return fetch("{{ request.route_url('paypal_create_order', cart_id=cart.id) }}?save_paypal=" + (savePayPal ? 'true' : 'false'), {
method: 'POST',
headers: {
'Content-Type': 'application/json',
'X-CSRF-Token': '{{ request.session.get_csrf_token() }}'
}
})
.then(response => response.json())
.then(data => {
if (data.error) {
alert('Error creating PayPal order: ' + data.error);
isCreatingOrder = false;
throw new Error(data.error);
}
// Handle both single order_id (old format) and order_ids array (new format)
if (data.order_ids) {
allPayPalOrderIds = data.order_ids;
isCreatingOrder = false;
// Return first order ID for PayPal SDK to process
return data.order_ids[0];
} else if (data.order_id) {
allPayPalOrderIds = [data.order_id];
isCreatingOrder = false;
return data.order_id;
}
})
.catch(error => {
isCreatingOrder = false;
throw error;
});
},
onApprove: function(data, actions) {
// Prevent duplicate submission after approval
if (isProcessingApproval) {
console.warn('PayPal approval already being processed');
return Promise.reject(new Error('Approval already being processed'));
}
isProcessingApproval = true;
// For multi-shop carts, we have multiple order IDs
// PayPal SDK only handles the first one, so we pass all IDs to backend
var orderIdsToSubmit = allPayPalOrderIds.join(',');
// Create form to submit PayPal order ID(s)
var form = document.createElement('form');
form.method = 'POST';
form.action = '{{ request.route_url('paypal_complete_checkout', cart_id=cart.id) }}';
var csrfInput = document.createElement('input');
csrfInput.type = 'hidden';
csrfInput.name = 'csrf_token';
csrfInput.value = '{{ request.session.get_csrf_token() }}';
form.appendChild(csrfInput);
var orderIdInput = document.createElement('input');
orderIdInput.type = 'hidden';
orderIdInput.name = 'paypal_order_id';
orderIdInput.value = orderIdsToSubmit;
form.appendChild(orderIdInput);
document.body.appendChild(form);
form.submit();
// Note: We don't reset isProcessingApproval because the page will redirect
// If form.submit() fails, the page stays and user can't retry anyway
},
onError: function(err) {
console.error('PayPal error:', err);
alert('PayPal payment failed. Please try again or use another payment method.');
// Reset flags on error so user can retry
isCreatingOrder = false;
isProcessingApproval = false;
},
onCancel: function(data) {
// Reset flags if user cancels PayPal flow
console.log('PayPal payment cancelled by user');
isCreatingOrder = false;
isProcessingApproval = false;
}
}).render('#paypal-button-container');
</script>
{% endif %}{# End PayPal enabled check #}
{# For free checkouts (e.g., with coupons), provide a simple confirmation button #}
{% if not cart.requires_payment %}
<form method="POST" action="{{ request.route_url('user_cart_complete_checkout', cart_id=cart.id) }}">

View file

@ -200,10 +200,200 @@
<br />
{% endif %}
{% if request.paypal_globally_enabled %}
<section class="one-column">
<section class="shop-settings well">
<h3>PayPal Settings 💰</h3>
<form method="post" action="/s/{{ request.shop.id }}/settings" onsubmit="submit.disabled = true; return true;">
<input type="hidden" name="form_section" value="paypal-settings" />
<label for="paypal_client_id_input">PayPal API Keys</label>
<br />
<input type="checkbox" id="toggle-paypal">
<label for="toggle-paypal" class="inline-label">Show PayPal Client ID & Secret API Keys</label>
<div class="hidden-control">
<br />
<label for="paypal_client_id_input">PayPal Client ID</label>
<input
name = "paypal_client_id"
type = "text"
id = "paypal_client_id"
class = "mps-paypal-client-id{% if not request.shop.paypal_enabled %} disabled-input{% endif %}"
value = "{% if paypal_client_id %}{{ paypal_client_id }}{% endif %}"
placeholder = "PayPal Client ID (from developer.paypal.com)"
{% if not request.shop.paypal_enabled %}readonly{% endif %}
/>
<br />
<br />
<label for="paypal_secret_input">PayPal Secret Key</label>
<input
name = "paypal_secret"
type = "text"
id = "paypal_secret"
class = "mps-paypal-secret{% if not request.shop.paypal_enabled %} disabled-input{% endif %}"
value = "{% if paypal_secret %}{{ paypal_secret }}{% endif %}"
placeholder = "PayPal Secret Key (from developer.paypal.com)"
{% if not request.shop.paypal_enabled %}readonly{% endif %}
/>
<br />
<br />
{% if request.shop.paypal_enabled %}
<small class="success-indicator">✓ PayPal configured and ready to accept PayPal payments</small>
<br />
<br />
<input type="submit" name="submit" class="mps-submit" value="Save PayPal Settings" />
<input type="submit" name="disable_paypal" class="payment-toggle-button disable" value="Disable PayPal" />
{% else %}
<small class="error-indicator">✗ PayPal payments are currently disabled</small>
<br />
<br />
<small class="status-message">Your API keys are preserved but customers cannot select PayPal as a payment method.</small>
<br />
<br />
<input type="submit" name="submit" class="payment-toggle-button enable" value="Re-enable PayPal" />
<small class="status-message">Re-enable PayPal payments to update your API keys</small>
{% endif %}
</div>
<br />
<br />
<br />
</form>
</section>
</section>
<br />
<br />
{% endif %}
{% if request.adyen_globally_enabled %}
<section class="one-column">
<section class="shop-settings well">
<h3>Adyen Settings 💳</h3>
<form method="post" action="/s/{{ request.shop.id }}/settings" onsubmit="submit.disabled = true; return true;">
<input type="hidden" name="form_section" value="adyen-settings" />
<label for="adyen_api_key_input">Adyen API Keys</label>
<br />
<input type="checkbox" id="toggle-adyen">
<label for="toggle-adyen" class="inline-label">Show Adyen API Keys</label>
<div class="hidden-control" id="adyen-controls">
<br />
<label for="adyen_api_key_input">Adyen API Key</label>
<input
name = "adyen_api_key"
type = "text"
id = "adyen_api_key"
class = "mps-adyen-api-key{% if not adyen_enabled %} disabled-input{% endif %}"
value = "{% if adyen_api_key %}{{ adyen_api_key }}{% endif %}"
placeholder = "Adyen API Key (from Adyen dashboard)"
{% if not adyen_enabled %}readonly{% endif %}
/>
<br />
<br />
<label for="adyen_merchant_account_input">Adyen Merchant Account</label>
<input
name = "adyen_merchant_account"
type = "text"
id = "adyen_merchant_account"
class = "mps-adyen-merchant-account{% if not adyen_enabled %} disabled-input{% endif %}"
value = "{% if adyen_merchant_account %}{{ adyen_merchant_account }}{% endif %}"
placeholder = "Adyen Merchant Account"
{% if not adyen_enabled %}readonly{% endif %}
/>
<br />
<br />
<label for="adyen_client_key_input">Adyen Client Key</label>
<input
name = "adyen_client_key"
type = "text"
id = "adyen_client_key"
class = "mps-adyen-client-key{% if not adyen_enabled %} disabled-input{% endif %}"
value = "{% if adyen_client_key %}{{ adyen_client_key }}{% endif %}"
placeholder = "Adyen Client Key (for Drop-in/Components)"
{% if not adyen_enabled %}readonly{% endif %}
/>
<br />
<br />
<label for="adyen_hmac_key_input">Adyen HMAC Key (for webhooks)</label>
<input
name = "adyen_hmac_key"
type = "text"
id = "adyen_hmac_key"
class = "mps-adyen-hmac-key{% if not adyen_enabled %} disabled-input{% endif %}"
value = "{% if adyen_hmac_key %}{{ adyen_hmac_key }}{% endif %}"
placeholder = "Adyen HMAC Key (for webhook verification)"
{% if not adyen_enabled %}readonly{% endif %}
/>
<br />
<br />
{% if adyen_enabled %}
<small class="success-indicator">✓ Adyen configured and ready to accept payments</small>
<br />
<br />
<input type="submit" name="submit" class="mps-submit" value="Save Adyen Settings" />
<input type="submit" name="disable_adyen" class="payment-toggle-button disable" value="Disable Adyen" />
{% else %}
<small class="error-indicator">✗ Adyen payments are currently disabled</small>
<br />
<br />
<small class="status-message">Your API keys are preserved but customers cannot select Adyen as a payment method.</small>
<br />
<br />
<input type="submit" name="submit" class="payment-toggle-button enable" value="Re-enable Adyen" />
<small class="status-message">Re-enable Adyen payments to update your API keys</small>
{% endif %}
</div>
<br />
<br />
<br />
</form>
</section>
</section>
<br />
<br />
{% endif %}
{% if request.monero_enabled %}
<section class="one-column">
<section class="shop-settings well">
<h3>Crypto Settings 🪙</h3>
<h4>Payment Risk Thresholds</h4>
@ -684,15 +874,32 @@ document.addEventListener('DOMContentLoaded', function() {
if (stripeToggle) {
stripeToggle.addEventListener('change', function() {
const stripeControls = document.querySelector('#toggle-stripe ~ .hidden-control');
// Save state to localStorage
localStorage.setItem('show-stripe-keys', this.checked);
if (stripeControls) {
stripeControls.style.display = this.checked ? 'block' : 'none';
}
});
}
// Adyen toggle
const adyenToggle = document.getElementById('toggle-adyen');
const adyenControls = document.getElementById('adyen-controls');
if (adyenToggle && adyenControls) {
// Restore state from localStorage
const adyenState = localStorage.getItem('show-adyen-keys') === 'true';
adyenToggle.checked = adyenState;
adyenControls.style.display = adyenState ? 'block' : 'none';
adyenToggle.addEventListener('change', function() {
// Save state to localStorage
localStorage.setItem('show-adyen-keys', this.checked);
adyenControls.style.display = this.checked ? 'block' : 'none';
});
}
});
</script>

View file

@ -560,9 +560,11 @@ class AuthenticatedFunctionalTests(FunctionalTests):
"csrf_token": csrf_token, # Include CSRF token as a form value.
},
)
res_csrf_checkout = res_csrf_checkout.follow().follow()
# With multiple payment methods enabled (Stripe + PayPal), checkout
# renders directly instead of redirecting to /billing for Stripe setup
self.assertEqual(200, res_csrf_checkout.status_int)
self.assertIn(
"Please enter your payment information.", res_csrf_checkout.body.decode()
"Please confirm your order.", res_csrf_checkout.body.decode()
)
def test_cart_checkout_logic_with_none_stripe_user_shop(self):
@ -1626,3 +1628,604 @@ class AuthenticatedFunctionalTests(FunctionalTests):
self.assertEqual(
shop.comments_require_approval, initial_values["comments_require_approval"]
)
@mock.patch("smtplib.SMTP")
def test_shop_paypal_credentials_can_be_set(self, mock_smtp):
"""Test that PayPal credentials can be set on a shop via the model."""
# Create shop using helper
shop = self._create_shop_helper(
user_creds=self.user1_creds,
shop_params=self.shop1_params,
)
# Set PayPal credentials directly on the model
shop.paypal_client_id = "test_client_id_abc123"
shop.paypal_secret = "test_secret_xyz789"
shop.paypal_enabled = True
self.dbsession.add(shop)
self.dbsession.flush()
transaction.manager.commit()
# Re-query shop from database to verify persistence
shop = get_shop_by_name(self.dbsession, self.shop1_params["name"])
# Verify PayPal is configured
self.assertEqual(shop.paypal_client_id, "test_client_id_abc123")
self.assertEqual(shop.paypal_secret, "test_secret_xyz789")
self.assertTrue(shop.paypal_enabled)
@mock.patch("smtplib.SMTP")
def test_paypal_create_order_requires_cart(self, mock_smtp):
"""Test that PayPal create order endpoint requires a cart."""
self.log_in_user(self.user1_creds)
# Try to create PayPal order without a cart
res = self.testapp.post_json(
"/paypal/create-order",
{"shop_id": "nonexistent-shop-id"},
expect_errors=True,
)
# Should fail with error (no active cart)
self.assertIn(res.status_int, [400, 404, 500])
@mock.patch("smtplib.SMTP")
def test_paypal_complete_checkout_requires_order_id(self, mock_smtp):
"""Test that PayPal complete checkout requires order ID."""
self.log_in_user(self.user1_creds)
# Try to complete checkout without order ID
res = self.testapp.post_json(
"/paypal/complete-checkout",
{"shop_id": "nonexistent-shop-id"},
expect_errors=True,
)
# Should fail with error
self.assertIn(res.status_int, [400, 404, 500])
@mock.patch("smtplib.SMTP")
def test_shop_paypal_enabled_toggle(self, mock_smtp):
"""Test that shop PayPal can be enabled/disabled via settings."""
# Create shop using helper
shop = self._create_shop_helper(
user_creds=self.user1_creds,
shop_params=self.shop1_params,
)
# Initially PayPal should be disabled (no credentials)
self.dbsession.refresh(shop)
# paypal_enabled defaults to True but without credentials it's not really enabled
# Set PayPal credentials
shop.paypal_client_id = "test_client_id"
shop.paypal_secret = "test_secret"
shop.paypal_enabled = True
self.dbsession.add(shop)
self.dbsession.flush()
# Verify it's enabled
self.dbsession.refresh(shop)
self.assertTrue(shop.paypal_enabled)
self.assertEqual(shop.paypal_client_id, "test_client_id")
self.assertEqual(shop.paypal_secret, "test_secret")
# ========================================================================
# PayPal Sandbox Integration Tests
# These tests require real PayPal sandbox credentials in environment vars:
# MPS_TEST_PAYPAL_CLIENT_ID
# MPS_TEST_PAYPAL_SECRET
# ========================================================================
@mock.patch("smtplib.SMTP")
@mock.patch("make_post_sell.models.Product.is_ready", mock_always_true)
def test_paypal_sandbox_authentication(self, mock_smtp):
"""Test that PayPal sandbox authentication works with real credentials.
Requires MPS_TEST_PAYPAL_CLIENT_ID and MPS_TEST_PAYPAL_SECRET env vars.
"""
paypal_client_id = environ.get("MPS_TEST_PAYPAL_CLIENT_ID", "")
paypal_secret = environ.get("MPS_TEST_PAYPAL_SECRET", "")
if not paypal_client_id or not paypal_secret:
self.skipTest("PayPal sandbox credentials not configured in environment")
import requests
# Test sandbox authentication
base_url = "https://api-m.sandbox.paypal.com"
auth_response = requests.post(
f"{base_url}/v1/oauth2/token",
headers={"Accept": "application/json", "Accept-Language": "en_US"},
data={"grant_type": "client_credentials"},
auth=(paypal_client_id, paypal_secret)
)
self.assertEqual(auth_response.status_code, 200)
response_json = auth_response.json()
self.assertIn("access_token", response_json)
self.assertIn("token_type", response_json)
self.assertEqual(response_json["token_type"], "Bearer")
@mock.patch("smtplib.SMTP")
@mock.patch("make_post_sell.models.Product.is_ready", mock_always_true)
def test_paypal_sandbox_create_order_api(self, mock_smtp):
"""Test creating a PayPal order via sandbox API directly.
Requires MPS_TEST_PAYPAL_CLIENT_ID and MPS_TEST_PAYPAL_SECRET env vars.
"""
paypal_client_id = environ.get("MPS_TEST_PAYPAL_CLIENT_ID", "")
paypal_secret = environ.get("MPS_TEST_PAYPAL_SECRET", "")
if not paypal_client_id or not paypal_secret:
self.skipTest("PayPal sandbox credentials not configured in environment")
import requests
# Get access token
base_url = "https://api-m.sandbox.paypal.com"
auth_response = requests.post(
f"{base_url}/v1/oauth2/token",
headers={"Accept": "application/json", "Accept-Language": "en_US"},
data={"grant_type": "client_credentials"},
auth=(paypal_client_id, paypal_secret)
)
self.assertEqual(auth_response.status_code, 200)
access_token = auth_response.json()["access_token"]
# Create a test order
order_json = {
"intent": "CAPTURE",
"purchase_units": [{
"amount": {
"currency_code": "USD",
"value": "10.00"
},
"description": "Test purchase from functional test"
}]
}
order_response = requests.post(
f"{base_url}/v2/checkout/orders",
headers={
"Content-Type": "application/json",
"Authorization": f"Bearer {access_token}"
},
json=order_json
)
self.assertEqual(order_response.status_code, 201)
order_data = order_response.json()
self.assertIn("id", order_data)
self.assertEqual(order_data["status"], "CREATED")
# Verify we got a valid order ID (PayPal order IDs are alphanumeric)
self.assertTrue(len(order_data["id"]) > 10)
@mock.patch("smtplib.SMTP")
@mock.patch("make_post_sell.models.Product.is_ready", mock_always_true)
def test_paypal_sandbox_create_order_via_endpoint(self, mock_smtp):
"""Test PayPal order creation through our endpoint with real sandbox credentials.
Requires MPS_TEST_PAYPAL_CLIENT_ID and MPS_TEST_PAYPAL_SECRET env vars.
"""
paypal_client_id = environ.get("MPS_TEST_PAYPAL_CLIENT_ID", "")
paypal_secret = environ.get("MPS_TEST_PAYPAL_SECRET", "")
if not paypal_client_id or not paypal_secret:
self.skipTest("PayPal sandbox credentials not configured in environment")
# Create shop and product
self.test_new_product(
user_creds=self.user1_creds,
shop_params=self.shop1_params,
product_params=self.product1_params,
)
# Get shop and configure PayPal with real sandbox credentials
shop = get_shop_by_name(self.dbsession, self.shop1_params["name"])
shop.paypal_client_id = paypal_client_id
shop.paypal_secret = paypal_secret
shop.paypal_enabled = True
self.dbsession.add(shop)
self.dbsession.flush()
transaction.manager.commit()
# Get the product
all_products = get_all_products(self.dbsession)
product = all_products.first()
# Re-query shop
shop = get_shop_by_name(self.dbsession, self.shop1_params["name"])
# Log out shop owner, log in as customer
self.testapp.get("/log-out")
self.log_in_user(self.user2_creds)
# Add product to cart
add_to_cart_res = self.testapp.post(
"/cart/add",
{
"product_id": product.id,
"shop_id": shop.id,
"csrf_token": self.get_csrf_token(shop.uuid_str),
},
)
# Re-query user2 after transaction commit
user2 = get_or_create_user_by_email(self.dbsession, self.user2_creds[0])
# Get cart
carts = get_all_carts(self.dbsession)
cart = carts.filter_by(user_id=user2.id, shop_id=shop.id).first()
self.assertIsNotNone(cart)
# Try to create PayPal order via our endpoint
res = self.testapp.post_json(
f"/paypal/create-order?shop_id={shop.uuid_str}&cart_id={cart.uuid_str}",
{},
expect_errors=True,
)
# The response depends on global PayPal configuration
# 200 with order_ids = success
# 200 with error = PayPal disabled globally
# 404 = route not configured
# 400/403/500 = various error conditions
if res.status_int == 200:
response_json = res.json
if "order_ids" in response_json:
# Success - PayPal is enabled and order was created
self.assertTrue(len(response_json["order_ids"]) > 0)
# Verify order ID format (PayPal order IDs are alphanumeric)
for order_id in response_json["order_ids"]:
self.assertTrue(len(order_id) > 10)
elif "error" in response_json:
# PayPal might be disabled globally - this is acceptable
pass
else:
# Non-200 responses are acceptable depending on configuration
self.assertIn(res.status_int, [400, 403, 404, 500])
@mock.patch("smtplib.SMTP")
@mock.patch("make_post_sell.models.Product.is_ready", mock_always_true)
def test_paypal_sandbox_order_retrieval(self, mock_smtp):
"""Test that we can retrieve a PayPal order after creation.
Requires MPS_TEST_PAYPAL_CLIENT_ID and MPS_TEST_PAYPAL_SECRET env vars.
"""
paypal_client_id = environ.get("MPS_TEST_PAYPAL_CLIENT_ID", "")
paypal_secret = environ.get("MPS_TEST_PAYPAL_SECRET", "")
if not paypal_client_id or not paypal_secret:
self.skipTest("PayPal sandbox credentials not configured in environment")
import requests
# Get access token
base_url = "https://api-m.sandbox.paypal.com"
auth_response = requests.post(
f"{base_url}/v1/oauth2/token",
headers={"Accept": "application/json", "Accept-Language": "en_US"},
data={"grant_type": "client_credentials"},
auth=(paypal_client_id, paypal_secret)
)
access_token = auth_response.json()["access_token"]
# Create an order
order_response = requests.post(
f"{base_url}/v2/checkout/orders",
headers={
"Content-Type": "application/json",
"Authorization": f"Bearer {access_token}"
},
json={
"intent": "CAPTURE",
"purchase_units": [{
"amount": {"currency_code": "USD", "value": "5.00"},
"description": "Test retrieval order"
}]
}
)
order_id = order_response.json()["id"]
# Retrieve the order
get_response = requests.get(
f"{base_url}/v2/checkout/orders/{order_id}",
headers={"Authorization": f"Bearer {access_token}"}
)
self.assertEqual(get_response.status_code, 200)
order_data = get_response.json()
self.assertEqual(order_data["id"], order_id)
self.assertEqual(order_data["status"], "CREATED")
# =========================================================================
# Stripe Webhook Tests
# =========================================================================
@patch("smtplib.SMTP")
def test_stripe_webhook_endpoint_accepts_post(self, mock_smtp):
"""Test that the Stripe webhook endpoint accepts POST requests."""
# Send a minimal valid webhook event
webhook_payload = {
"type": "payment_intent.succeeded",
"data": {
"object": {
"id": "pi_test_12345",
"latest_charge": "ch_test_12345"
}
}
}
response = self.testapp.post_json(
"/webhooks/stripe",
webhook_payload,
status=200
)
self.assertEqual(response.status_int, 200)
self.assertIn("success", response.json.get("status", ""))
@patch("smtplib.SMTP")
def test_stripe_webhook_payment_intent_succeeded_no_invoice(self, mock_smtp):
"""Test webhook with payment_intent.succeeded but no matching invoice."""
webhook_payload = {
"type": "payment_intent.succeeded",
"data": {
"object": {
"id": "pi_nonexistent_12345",
"latest_charge": "ch_nonexistent_12345"
}
}
}
response = self.testapp.post_json(
"/webhooks/stripe",
webhook_payload,
status=200
)
# Should succeed even without matching invoice (idempotent)
self.assertEqual(response.status_int, 200)
@patch("smtplib.SMTP")
def test_stripe_webhook_payment_failed_event(self, mock_smtp):
"""Test handling payment_intent.payment_failed event."""
webhook_payload = {
"type": "payment_intent.payment_failed",
"data": {
"object": {
"id": "pi_failed_12345",
"last_payment_error": {
"message": "Your card was declined."
}
}
}
}
response = self.testapp.post_json(
"/webhooks/stripe",
webhook_payload,
status=200
)
self.assertEqual(response.status_int, 200)
@patch("smtplib.SMTP")
def test_stripe_webhook_charge_refunded_event(self, mock_smtp):
"""Test handling charge.refunded event."""
webhook_payload = {
"type": "charge.refunded",
"data": {
"object": {
"id": "ch_refund_12345",
"amount_refunded": 1000 # $10.00 in cents
}
}
}
response = self.testapp.post_json(
"/webhooks/stripe",
webhook_payload,
status=200
)
self.assertEqual(response.status_int, 200)
@patch("smtplib.SMTP")
def test_stripe_webhook_dispute_created_event(self, mock_smtp):
"""Test handling charge.dispute.created event."""
webhook_payload = {
"type": "charge.dispute.created",
"data": {
"object": {
"id": "dp_dispute_12345",
"charge": "ch_disputed_12345",
"amount": 5000, # $50.00 in cents
"reason": "fraudulent"
}
}
}
response = self.testapp.post_json(
"/webhooks/stripe",
webhook_payload,
status=200
)
self.assertEqual(response.status_int, 200)
@patch("smtplib.SMTP")
def test_stripe_webhook_unknown_event_type(self, mock_smtp):
"""Test that unknown event types are handled gracefully."""
webhook_payload = {
"type": "unknown.event.type",
"data": {
"object": {}
}
}
response = self.testapp.post_json(
"/webhooks/stripe",
webhook_payload,
status=200
)
# Should return success for unknown events (don't block)
self.assertEqual(response.status_int, 200)
@patch("smtplib.SMTP")
def test_stripe_webhook_malformed_json(self, mock_smtp):
"""Test that malformed JSON returns an error response."""
response = self.testapp.post(
"/webhooks/stripe",
"not valid json",
content_type="application/json",
status=200 # Returns 200 to prevent retries
)
# Should handle gracefully
self.assertEqual(response.status_int, 200)
def test_adyen_settings_form_save_credentials(self):
"""Test that Adyen credentials can be saved through settings form."""
# Create shop using helper
shop = self._create_shop_helper(
user_creds=self.user1_creds,
shop_params=self.shop1_params,
)
# Save Adyen settings
adyen_settings_data = {
"form_section": "adyen-settings",
"adyen_api_key": "test_api_key_12345",
"adyen_merchant_account": "TestMerchantAccount",
"adyen_client_key": "test_client_key",
"adyen_hmac_key": "test_hmac_key",
"csrf_token": self.get_csrf_token(shop.uuid_str),
}
settings_res = self.testapp.post(
f"/s/{shop.id}/settings", adyen_settings_data, status=302
)
# Refresh shop from DB
self.dbsession.expire(shop)
self.assertEqual(shop.adyen_api_key, "test_api_key_12345")
self.assertEqual(shop.adyen_merchant_account, "TestMerchantAccount")
self.assertEqual(shop.adyen_client_key, "test_client_key")
self.assertEqual(shop.adyen_hmac_key, "test_hmac_key")
def test_adyen_settings_disable_and_reenable(self):
"""Test that Adyen can be disabled and re-enabled."""
# Create shop using helper
shop = self._create_shop_helper(
user_creds=self.user1_creds,
shop_params=self.shop1_params,
)
# Get csrf token once before modifying anything
csrf_token = self.get_csrf_token(shop.uuid_str)
# First set Adyen credentials via form
setup_data = {
"form_section": "adyen-settings",
"adyen_api_key": "test_key",
"adyen_merchant_account": "TestMerchant",
"csrf_token": csrf_token,
}
self.testapp.post(f"/s/{shop.id}/settings", setup_data, status=302)
# Refresh shop from DB - should have Adyen enabled by default
self.dbsession.expire(shop)
self.assertTrue(shop.adyen_enabled)
# Disable Adyen
disable_data = {
"form_section": "adyen-settings",
"disable_adyen": "1",
"csrf_token": csrf_token,
}
self.testapp.post(f"/s/{shop.id}/settings", disable_data, status=302)
# Refresh shop from DB
self.dbsession.expire(shop)
self.assertFalse(shop.adyen_enabled)
# Re-enable Adyen
enable_data = {
"form_section": "adyen-settings",
"csrf_token": csrf_token,
}
self.testapp.post(f"/s/{shop.id}/settings", enable_data, status=302)
# Refresh shop from DB
self.dbsession.expire(shop)
self.assertTrue(shop.adyen_enabled)
@patch("smtplib.SMTP")
def test_adyen_webhook_authorisation_event(self, mock_smtp):
"""Test that Adyen AUTHORISATION webhook is handled correctly."""
# Post webhook notification (doesn't require shop setup, just tests handler)
webhook_payload = {
"notificationItems": [{
"NotificationRequestItem": {
"eventCode": "AUTHORISATION",
"success": "true",
"pspReference": "PSP_TEST_12345",
"merchantReference": "test_reference",
"amount": {"value": 1000, "currency": "USD"},
}
}]
}
response = self.testapp.post_json(
"/webhooks/adyen",
webhook_payload,
status=200
)
# Should return [accepted]
self.assertIn("[accepted]", response.body.decode())
@patch("smtplib.SMTP")
def test_adyen_webhook_chargeback_event(self, mock_smtp):
"""Test that Adyen CHARGEBACK webhook is handled correctly."""
# Post chargeback webhook notification
webhook_payload = {
"notificationItems": [{
"NotificationRequestItem": {
"eventCode": "CHARGEBACK",
"pspReference": "PSP_CHARGEBACK_123",
"merchantReference": "test_reference",
"amount": {"value": 5000, "currency": "USD"},
"reason": "Goods not received",
}
}]
}
response = self.testapp.post_json(
"/webhooks/adyen",
webhook_payload,
status=200
)
# Should return [accepted]
self.assertIn("[accepted]", response.body.decode())
@patch("smtplib.SMTP")
def test_adyen_webhook_malformed_json(self, mock_smtp):
"""Test that malformed JSON in Adyen webhook is handled gracefully."""
response = self.testapp.post(
"/webhooks/adyen",
"not valid json",
content_type="application/json",
status=200 # Returns 200 to prevent retries
)
# Should handle gracefully and return [accepted]
self.assertIn("[accepted]", response.body.decode())

View file

@ -19,6 +19,8 @@ from ..models.cart import Cart
from ..models.coupon import Coupon
from ..models.cart_coupon import CartCoupon
from ..models.stripe_user_shop import StripeUserShop
from ..models.paypal_user_shop import PayPalUserShop
from ..models.invoice import get_invoice_by_paypal_order_id
from ..models.invoice import Invoice, InvoiceLineItem
from ..models.coupon_redemption import CouponRedemption
from ..models.price import Price
@ -2542,3 +2544,503 @@ class DogecoinPaymentIntegration(DatabaseIntegrationTests):
self.assertEqual(extracted_monero_invoice, extracted_dogecoin_invoice)
transaction.commit()
class TestPayPalUserShopIntegration(DatabaseIntegrationTests):
"""Integration tests for PayPalUserShop model with real database."""
def test_create_paypal_user_shop_integration(self):
"""Test creating a PayPalUserShop with real user and shop objects."""
# Create real user
user = get_or_create_user_by_email(self.dbsession, "paypal_test@example.com")
self.dbsession.add(user)
# Create real shop
shop = Shop(
name="PayPal Test Shop",
phone_number="555-555-5555",
billing_address="123 Test St",
description="A shop that accepts PayPal",
)
shop.stripe_public_api_key = "pk_test_123"
shop.stripe_secret_api_key = "sk_test_123"
shop.domain_name = "paypaltest.com"
shop.paypal_client_id = "test_client_id_123"
shop.paypal_secret = "test_secret_456"
shop.paypal_enabled = True
self.dbsession.add(shop)
self.dbsession.flush()
# Create PayPalUserShop
paypal_user_shop = PayPalUserShop(user=user, shop=shop)
paypal_user_shop.payer_id = "PAYER123456789"
paypal_user_shop.billing_agreement_id = "BA-ABC123DEF456"
paypal_user_shop.active_payment_token = "TOKEN123"
self.dbsession.add(paypal_user_shop)
self.dbsession.flush()
# Verify relationships
self.assertEqual(paypal_user_shop.user_id, user.id)
self.assertEqual(paypal_user_shop.shop_id, shop.id)
# Query back from database
queried = self.dbsession.query(PayPalUserShop).filter(
PayPalUserShop.user_id == user.id,
PayPalUserShop.shop_id == shop.id
).first()
self.assertIsNotNone(queried)
self.assertEqual(queried.payer_id, "PAYER123456789")
self.assertEqual(queried.billing_agreement_id, "BA-ABC123DEF456")
self.assertEqual(queried.active_payment_token, "TOKEN123")
transaction.commit()
def test_paypal_user_shop_unique_constraint(self):
"""Test that user can only have one PayPalUserShop per shop."""
# Create user and shop
user = get_or_create_user_by_email(self.dbsession, "unique_test@example.com")
shop = Shop(
name="Unique Test Shop",
phone_number="555-555-5555",
billing_address="123 Test St",
description="Test shop",
)
shop.stripe_public_api_key = "pk_test_123"
shop.stripe_secret_api_key = "sk_test_123"
shop.domain_name = "uniquetest.com"
self.dbsession.add(user)
self.dbsession.add(shop)
self.dbsession.flush()
# Create first PayPalUserShop
paypal_user_shop1 = PayPalUserShop(user=user, shop=shop)
paypal_user_shop1.payer_id = "PAYER111"
self.dbsession.add(paypal_user_shop1)
self.dbsession.flush()
# Verify only one exists
count = self.dbsession.query(PayPalUserShop).filter(
PayPalUserShop.user_id == user.id,
PayPalUserShop.shop_id == shop.id
).count()
self.assertEqual(count, 1)
transaction.commit()
class TestInvoicePayPalIntegration(DatabaseIntegrationTests):
"""Integration tests for Invoice PayPal functionality with real database."""
def test_invoice_with_paypal_order_id_integration(self):
"""Test creating invoice with PayPal order and capture IDs."""
# Create real user
user = get_or_create_user_by_email(self.dbsession, "invoice_paypal@example.com")
self.dbsession.add(user)
# Create real shop
shop = Shop(
name="PayPal Invoice Test Shop",
phone_number="555-555-5555",
billing_address="123 Test St",
description="A shop for testing PayPal invoices",
)
shop.stripe_public_api_key = "pk_test_123"
shop.stripe_secret_api_key = "sk_test_123"
shop.domain_name = "invoicetest.com"
shop.paypal_enabled = True
self.dbsession.add(shop)
self.dbsession.flush()
# Create invoice with PayPal payment info
invoice = Invoice(user=user)
invoice.shop = shop
invoice.paypal_order_id = "ORDER123456789"
invoice.paypal_capture_id = "CAPTURE987654321"
self.dbsession.add(invoice)
self.dbsession.flush()
# Verify payment method detection
self.assertEqual(invoice.payment_method, "paypal")
# Verify query by PayPal order ID
queried = get_invoice_by_paypal_order_id(self.dbsession, "ORDER123456789")
self.assertIsNotNone(queried)
self.assertEqual(queried.id, invoice.id)
self.assertEqual(queried.paypal_capture_id, "CAPTURE987654321")
transaction.commit()
def test_invoice_payment_method_priority_integration(self):
"""Test payment method detection priority: crypto > paypal > stripe."""
user = get_or_create_user_by_email(self.dbsession, "priority_test@example.com")
shop = Shop(
name="Priority Test Shop",
phone_number="555-555-5555",
billing_address="123 Test St",
description="Test shop",
)
shop.stripe_public_api_key = "pk_test_123"
shop.stripe_secret_api_key = "sk_test_123"
shop.domain_name = "prioritytest.com"
self.dbsession.add(user)
self.dbsession.add(shop)
self.dbsession.flush()
# Invoice with no payment info defaults to stripe
invoice_stripe = Invoice(user=user)
invoice_stripe.shop = shop
self.dbsession.add(invoice_stripe)
self.dbsession.flush()
self.assertEqual(invoice_stripe.payment_method, "stripe")
# Invoice with PayPal order ID returns paypal
invoice_paypal = Invoice(user=user)
invoice_paypal.shop = shop
invoice_paypal.paypal_order_id = "ORDER_PAYPAL_123"
self.dbsession.add(invoice_paypal)
self.dbsession.flush()
self.assertEqual(invoice_paypal.payment_method, "paypal")
transaction.commit()
def test_get_invoice_by_paypal_order_id_not_found(self):
"""Test querying for non-existent PayPal order ID returns None."""
result = get_invoice_by_paypal_order_id(self.dbsession, "NONEXISTENT_ORDER_ID")
self.assertIsNone(result)
def test_invoice_with_paypal_is_paid_status(self):
"""Test is_paid property works correctly for PayPal invoices."""
user = get_or_create_user_by_email(self.dbsession, "paid_test@example.com")
shop = Shop(
name="Paid Test Shop",
phone_number="555-555-5555",
billing_address="123 Test St",
description="Test shop",
)
shop.stripe_public_api_key = "pk_test_123"
shop.stripe_secret_api_key = "sk_test_123"
shop.domain_name = "paidtest.com"
self.dbsession.add(user)
self.dbsession.add(shop)
self.dbsession.flush()
# PayPal invoice without crypto_payment is considered "paid"
invoice = Invoice(user=user)
invoice.shop = shop
invoice.paypal_order_id = "ORDER_PAID_TEST"
invoice.paypal_capture_id = "CAPTURE_PAID_TEST"
self.dbsession.add(invoice)
self.dbsession.flush()
# payment_status returns "paid" for non-crypto invoices
self.assertEqual(invoice.payment_status, "paid")
self.assertTrue(invoice.is_paid)
transaction.commit()
class TestInvoiceStripeIntegration(DatabaseIntegrationTests):
"""Integration tests for Invoice Stripe functionality with real database."""
def test_invoice_with_stripe_payment_intent_id_integration(self):
"""Test creating invoice with Stripe payment intent and charge IDs."""
# Create real user
user = get_or_create_user_by_email(self.dbsession, "invoice_stripe@example.com")
self.dbsession.add(user)
# Create real shop
shop = Shop(
name="Stripe Invoice Test Shop",
phone_number="555-555-5555",
billing_address="123 Test St",
description="A shop for testing Stripe invoices",
)
shop.stripe_public_api_key = "pk_test_123"
shop.stripe_secret_api_key = "sk_test_123"
shop.domain_name = "stripeinvoicetest.com"
self.dbsession.add(shop)
self.dbsession.flush()
# Create invoice with Stripe payment info
invoice = Invoice(user=user)
invoice.shop = shop
invoice.stripe_payment_intent_id = "pi_1234567890abcdef"
invoice.stripe_charge_id = "ch_1234567890abcdef"
self.dbsession.add(invoice)
self.dbsession.flush()
# Verify payment method detection
self.assertEqual(invoice.payment_method, "stripe")
# Verify query by Stripe payment intent ID
from make_post_sell.models.invoice import get_invoice_by_stripe_payment_intent_id
queried = get_invoice_by_stripe_payment_intent_id(self.dbsession, "pi_1234567890abcdef")
self.assertIsNotNone(queried)
self.assertEqual(queried.id, invoice.id)
self.assertEqual(queried.stripe_charge_id, "ch_1234567890abcdef")
transaction.commit()
def test_get_invoice_by_stripe_payment_intent_id_not_found(self):
"""Test querying for non-existent Stripe payment intent ID returns None."""
from make_post_sell.models.invoice import get_invoice_by_stripe_payment_intent_id
result = get_invoice_by_stripe_payment_intent_id(self.dbsession, "pi_nonexistent")
self.assertIsNone(result)
def test_invoice_stripe_payment_persists_across_sessions(self):
"""Test that Stripe payment IDs persist correctly in the database."""
user = get_or_create_user_by_email(self.dbsession, "persist_test@example.com")
shop = Shop(
name="Persist Test Shop",
phone_number="555-555-5555",
billing_address="123 Test St",
description="Test shop",
)
shop.stripe_public_api_key = "pk_test_123"
shop.stripe_secret_api_key = "sk_test_123"
shop.domain_name = "persisttest.com"
self.dbsession.add(user)
self.dbsession.add(shop)
self.dbsession.flush()
# Create invoice with Stripe payment info
invoice = Invoice(user=user)
invoice.shop = shop
invoice.stripe_payment_intent_id = "pi_persist_test_12345"
invoice.stripe_charge_id = "ch_persist_test_12345"
self.dbsession.add(invoice)
self.dbsession.flush()
invoice_id = invoice.id
# Commit and clear session to simulate new request
transaction.commit()
self.dbsession.expire_all()
# Re-query and verify
from make_post_sell.models.invoice import get_invoice_by_id
reloaded_invoice = get_invoice_by_id(self.dbsession, invoice_id)
self.assertIsNotNone(reloaded_invoice)
self.assertEqual(reloaded_invoice.stripe_payment_intent_id, "pi_persist_test_12345")
self.assertEqual(reloaded_invoice.stripe_charge_id, "ch_persist_test_12345")
self.assertEqual(reloaded_invoice.payment_method, "stripe")
def test_invoice_with_stripe_is_paid_status(self):
"""Test is_paid property works correctly for Stripe invoices."""
user = get_or_create_user_by_email(self.dbsession, "stripe_paid_test@example.com")
shop = Shop(
name="Stripe Paid Test Shop",
phone_number="555-555-5555",
billing_address="123 Test St",
description="Test shop",
)
shop.stripe_public_api_key = "pk_test_123"
shop.stripe_secret_api_key = "sk_test_123"
shop.domain_name = "stripepaidtest.com"
self.dbsession.add(user)
self.dbsession.add(shop)
self.dbsession.flush()
# Stripe invoice is considered "paid"
invoice = Invoice(user=user)
invoice.shop = shop
invoice.stripe_payment_intent_id = "pi_paid_test"
invoice.stripe_charge_id = "ch_paid_test"
self.dbsession.add(invoice)
self.dbsession.flush()
# payment_status returns "paid" for non-crypto invoices
self.assertEqual(invoice.payment_status, "paid")
self.assertTrue(invoice.is_paid)
transaction.commit()
def test_invoice_payment_method_priority_with_stripe(self):
"""Test payment method detection with Stripe: crypto > paypal > stripe."""
user = get_or_create_user_by_email(self.dbsession, "stripe_priority@example.com")
shop = Shop(
name="Stripe Priority Shop",
phone_number="555-555-5555",
billing_address="123 Test St",
description="Test shop",
)
shop.stripe_public_api_key = "pk_test_123"
shop.stripe_secret_api_key = "sk_test_123"
shop.domain_name = "stripeprioritytest.com"
self.dbsession.add(user)
self.dbsession.add(shop)
self.dbsession.flush()
# Invoice with only Stripe payment intent returns stripe
invoice_stripe = Invoice(user=user)
invoice_stripe.shop = shop
invoice_stripe.stripe_payment_intent_id = "pi_priority_test"
self.dbsession.add(invoice_stripe)
self.dbsession.flush()
self.assertEqual(invoice_stripe.payment_method, "stripe")
# Invoice with both PayPal and Stripe - PayPal takes priority
invoice_both = Invoice(user=user)
invoice_both.shop = shop
invoice_both.paypal_order_id = "ORDER_PRIORITY"
invoice_both.stripe_payment_intent_id = "pi_priority_both"
self.dbsession.add(invoice_both)
self.dbsession.flush()
self.assertEqual(invoice_both.payment_method, "paypal")
transaction.commit()
class TestAdyenIntegration(DatabaseIntegrationTests):
"""Integration tests for Adyen payment integration."""
def test_shop_adyen_credentials_persistence(self):
"""Test that Adyen credentials are properly persisted to the database."""
# Create shop with Adyen credentials
shop = Shop(
name="Adyen Test Shop",
phone_number="555-555-5555",
billing_address="123 Test St",
description="Test shop for Adyen",
)
shop.adyen_api_key = "test_api_key_12345"
shop.adyen_merchant_account = "TestMerchantAccount"
shop.adyen_client_key = "test_client_key"
shop.adyen_hmac_key = "test_hmac_key"
self.dbsession.add(shop)
self.dbsession.flush()
# Re-query the shop to verify persistence
shop_id = shop.id
self.dbsession.expunge(shop)
from ..models.shop import get_shop_by_id
reloaded_shop = get_shop_by_id(self.dbsession, shop_id)
self.assertEqual(reloaded_shop.adyen_api_key, "test_api_key_12345")
self.assertEqual(reloaded_shop.adyen_merchant_account, "TestMerchantAccount")
self.assertEqual(reloaded_shop.adyen_client_key, "test_client_key")
self.assertEqual(reloaded_shop.adyen_hmac_key, "test_hmac_key")
transaction.commit()
def test_shop_is_adyen_ready_integration(self):
"""Test is_adyen_ready with persisted shop."""
# Shop without credentials
shop1 = Shop(
name="Shop Without Adyen",
phone_number="555-555-0001",
billing_address="123 Test St",
description="No Adyen",
)
self.dbsession.add(shop1)
self.dbsession.flush()
self.assertFalse(shop1.is_adyen_ready)
# Shop with credentials
shop2 = Shop(
name="Shop With Adyen",
phone_number="555-555-0002",
billing_address="456 Test St",
description="Has Adyen",
)
shop2.adyen_api_key = "test_api_key"
shop2.adyen_merchant_account = "TestMerchant"
self.dbsession.add(shop2)
self.dbsession.flush()
self.assertTrue(shop2.is_adyen_ready)
transaction.commit()
def test_invoice_adyen_psp_reference_integration(self):
"""Test that Adyen PSP reference is properly stored on invoice."""
# Create user and shop
user = get_or_create_user_by_email(self.dbsession, "adyen@test.com")
shop = Shop(
name="Adyen Invoice Test",
phone_number="555-555-5555",
billing_address="123 Test St",
description="Testing Adyen invoices",
)
shop.adyen_api_key = "test_api_key"
shop.adyen_merchant_account = "TestMerchant"
self.dbsession.add(user)
self.dbsession.add(shop)
self.dbsession.flush()
# Create invoice with Adyen payment
invoice = Invoice(user=user)
invoice.shop = shop
invoice.adyen_psp_reference = "882619391893263J"
self.dbsession.add(invoice)
self.dbsession.flush()
# Verify payment method detection
self.assertEqual(invoice.payment_method, "adyen")
# Verify persistence
invoice_id = invoice.id
self.dbsession.expunge(invoice)
from ..models.invoice import get_invoice_by_adyen_psp_reference
reloaded_invoice = get_invoice_by_adyen_psp_reference(
self.dbsession, "882619391893263J"
)
self.assertIsNotNone(reloaded_invoice)
self.assertEqual(str(reloaded_invoice.id), str(invoice_id))
self.assertEqual(reloaded_invoice.adyen_psp_reference, "882619391893263J")
transaction.commit()
def test_get_invoice_by_adyen_psp_reference_not_found(self):
"""Test that get_invoice_by_adyen_psp_reference returns None for non-existent reference."""
from ..models.invoice import get_invoice_by_adyen_psp_reference
result = get_invoice_by_adyen_psp_reference(
self.dbsession, "nonexistent_psp_reference"
)
self.assertIsNone(result)
def test_invoice_payment_method_priority_with_adyen_integration(self):
"""Test payment method priority when multiple payment references exist."""
user = get_or_create_user_by_email(self.dbsession, "priority@test.com")
shop = Shop(
name="Priority Test Shop",
phone_number="555-555-5555",
billing_address="123 Test St",
description="Testing priority",
)
self.dbsession.add(user)
self.dbsession.add(shop)
self.dbsession.flush()
# Invoice with only Adyen payment
invoice_adyen = Invoice(user=user)
invoice_adyen.shop = shop
invoice_adyen.adyen_psp_reference = "PSP_ONLY"
self.dbsession.add(invoice_adyen)
self.dbsession.flush()
self.assertEqual(invoice_adyen.payment_method, "adyen")
# Invoice with Adyen and Stripe - Adyen takes priority
invoice_adyen_stripe = Invoice(user=user)
invoice_adyen_stripe.shop = shop
invoice_adyen_stripe.adyen_psp_reference = "PSP_PRIORITY"
invoice_adyen_stripe.stripe_payment_intent_id = "pi_lower_priority"
self.dbsession.add(invoice_adyen_stripe)
self.dbsession.flush()
self.assertEqual(invoice_adyen_stripe.payment_method, "adyen")
# Invoice with PayPal and Adyen - PayPal takes priority
invoice_paypal_adyen = Invoice(user=user)
invoice_paypal_adyen.shop = shop
invoice_paypal_adyen.paypal_order_id = "PP_HIGHEST"
invoice_paypal_adyen.adyen_psp_reference = "PSP_SECOND"
self.dbsession.add(invoice_paypal_adyen)
self.dbsession.flush()
self.assertEqual(invoice_paypal_adyen.payment_method, "paypal")
transaction.commit()

View file

@ -1547,3 +1547,308 @@ class TestCryptoPayment(unittest.TestCase):
result = self.payment.format_confirmation_status()
self.assertEqual(result, "15/10")
class TestPayPalUserShop(unittest.TestCase):
"""Unit tests for PayPalUserShop model."""
@mock.patch("make_post_sell.models.user.is_user_name_available", mock_always_true)
def setUp(self):
self.user = User("paypal@example.com")
self.shop = Shop(
name="PayPal Test Shop",
phone_number="555-1234",
billing_address="123 Main St",
description="Test shop",
)
def test_create_paypal_user_shop(self):
"""Test creating a PayPalUserShop record."""
from ..models.paypal_user_shop import PayPalUserShop
pus = PayPalUserShop(user=self.user, shop=self.shop)
self.assertEqual(pus.user, self.user)
self.assertEqual(pus.shop, self.shop)
self.assertIsNotNone(pus.id)
self.assertIsNone(pus.payer_id)
self.assertIsNone(pus.billing_agreement_id)
self.assertIsNone(pus.active_payment_token)
def test_has_billing_agreement_false(self):
"""Test has_billing_agreement when no agreement exists."""
from ..models.paypal_user_shop import PayPalUserShop
pus = PayPalUserShop(user=self.user, shop=self.shop)
self.assertFalse(pus.has_billing_agreement)
def test_has_billing_agreement_true(self):
"""Test has_billing_agreement when agreement exists."""
from ..models.paypal_user_shop import PayPalUserShop
pus = PayPalUserShop(user=self.user, shop=self.shop)
pus.billing_agreement_id = "BA-1234567890"
self.assertTrue(pus.has_billing_agreement)
def test_has_saved_payment_method_false(self):
"""Test has_saved_payment_method when no token exists."""
from ..models.paypal_user_shop import PayPalUserShop
pus = PayPalUserShop(user=self.user, shop=self.shop)
self.assertFalse(pus.has_saved_payment_method)
def test_has_saved_payment_method_true(self):
"""Test has_saved_payment_method when token exists."""
from ..models.paypal_user_shop import PayPalUserShop
pus = PayPalUserShop(user=self.user, shop=self.shop)
pus.active_payment_token = "VAULT-TOKEN-123"
self.assertTrue(pus.has_saved_payment_method)
def test_payer_id_storage(self):
"""Test storing payer ID."""
from ..models.paypal_user_shop import PayPalUserShop
pus = PayPalUserShop(user=self.user, shop=self.shop)
pus.payer_id = "PAYERID123ABC"
self.assertEqual(pus.payer_id, "PAYERID123ABC")
class TestInvoicePayPal(unittest.TestCase):
"""Unit tests for Invoice PayPal functionality."""
@mock.patch("make_post_sell.models.user.is_user_name_available", mock_always_true)
def setUp(self):
self.user = User("paypal_invoice@example.com")
self.shop = Shop(
name="PayPal Invoice Shop",
phone_number="555-1234",
billing_address="123 Main St",
description="Test shop",
)
def test_invoice_payment_method_stripe_default(self):
"""Test that invoice with no PayPal or crypto returns stripe."""
invoice = Invoice(self.user)
invoice.shop = self.shop
self.assertEqual(invoice.payment_method, "stripe")
def test_invoice_payment_method_paypal(self):
"""Test that invoice with paypal_order_id returns paypal."""
invoice = Invoice(self.user)
invoice.shop = self.shop
invoice.paypal_order_id = "PAYPAL-ORDER-123"
self.assertEqual(invoice.payment_method, "paypal")
def test_invoice_paypal_columns_nullable(self):
"""Test that PayPal columns are nullable by default."""
invoice = Invoice(self.user)
invoice.shop = self.shop
self.assertIsNone(invoice.paypal_order_id)
self.assertIsNone(invoice.paypal_capture_id)
def test_invoice_paypal_capture_id_storage(self):
"""Test storing PayPal capture ID."""
invoice = Invoice(self.user)
invoice.shop = self.shop
invoice.paypal_order_id = "PAYPAL-ORDER-123"
invoice.paypal_capture_id = "CAPTURE-456"
self.assertEqual(invoice.paypal_order_id, "PAYPAL-ORDER-123")
self.assertEqual(invoice.paypal_capture_id, "CAPTURE-456")
def test_invoice_payment_status_for_paypal(self):
"""Test that PayPal invoice payment_status is 'paid'."""
invoice = Invoice(self.user)
invoice.shop = self.shop
invoice.paypal_order_id = "PAYPAL-ORDER-123"
# PayPal invoices (like Stripe) are "paid" if they exist
self.assertEqual(invoice.payment_status, "paid")
def test_invoice_is_paid_for_paypal(self):
"""Test that PayPal invoice is_paid returns True."""
invoice = Invoice(self.user)
invoice.shop = self.shop
invoice.paypal_order_id = "PAYPAL-ORDER-123"
self.assertTrue(invoice.is_paid)
class TestInvoiceStripe(unittest.TestCase):
"""Unit tests for Invoice Stripe functionality."""
@mock.patch("make_post_sell.models.user.is_user_name_available", mock_always_true)
def setUp(self):
self.user = User("stripe_invoice@example.com")
self.shop = Shop(
name="Stripe Invoice Shop",
phone_number="555-1234",
billing_address="123 Main St",
description="Test shop",
)
def test_invoice_stripe_columns_nullable(self):
"""Test that Stripe columns are nullable by default."""
invoice = Invoice(self.user)
invoice.shop = self.shop
self.assertIsNone(invoice.stripe_payment_intent_id)
self.assertIsNone(invoice.stripe_charge_id)
def test_invoice_stripe_payment_intent_id_storage(self):
"""Test storing Stripe payment intent ID."""
invoice = Invoice(self.user)
invoice.shop = self.shop
invoice.stripe_payment_intent_id = "pi_1234567890abcdef"
self.assertEqual(invoice.stripe_payment_intent_id, "pi_1234567890abcdef")
def test_invoice_stripe_charge_id_storage(self):
"""Test storing Stripe charge ID."""
invoice = Invoice(self.user)
invoice.shop = self.shop
invoice.stripe_payment_intent_id = "pi_1234567890abcdef"
invoice.stripe_charge_id = "ch_1234567890abcdef"
self.assertEqual(invoice.stripe_charge_id, "ch_1234567890abcdef")
def test_invoice_payment_method_stripe_explicit(self):
"""Test that invoice with stripe_payment_intent_id returns stripe."""
invoice = Invoice(self.user)
invoice.shop = self.shop
invoice.stripe_payment_intent_id = "pi_1234567890abcdef"
self.assertEqual(invoice.payment_method, "stripe")
def test_invoice_payment_method_stripe_default(self):
"""Test that invoice with no payment refs returns stripe (legacy)."""
invoice = Invoice(self.user)
invoice.shop = self.shop
# Legacy invoices without payment tracking default to stripe
self.assertEqual(invoice.payment_method, "stripe")
def test_invoice_payment_status_for_stripe(self):
"""Test that Stripe invoice payment_status is 'paid'."""
invoice = Invoice(self.user)
invoice.shop = self.shop
invoice.stripe_payment_intent_id = "pi_1234567890abcdef"
self.assertEqual(invoice.payment_status, "paid")
def test_invoice_is_paid_for_stripe(self):
"""Test that Stripe invoice is_paid returns True."""
invoice = Invoice(self.user)
invoice.shop = self.shop
invoice.stripe_payment_intent_id = "pi_1234567890abcdef"
self.assertTrue(invoice.is_paid)
def test_invoice_payment_method_priority_paypal_over_stripe(self):
"""Test that PayPal takes priority if both are set (edge case)."""
invoice = Invoice(self.user)
invoice.shop = self.shop
invoice.paypal_order_id = "PAYPAL-ORDER-123"
invoice.stripe_payment_intent_id = "pi_1234567890abcdef"
# PayPal is checked first in payment_method
self.assertEqual(invoice.payment_method, "paypal")
def test_invoice_payment_method_adyen(self):
"""Test that Adyen payment method is detected correctly."""
invoice = Invoice(self.user)
invoice.shop = self.shop
invoice.adyen_psp_reference = "PSP123456789"
self.assertEqual(invoice.payment_method, "adyen")
def test_invoice_payment_method_priority_paypal_over_adyen(self):
"""Test that PayPal takes priority over Adyen (edge case)."""
invoice = Invoice(self.user)
invoice.shop = self.shop
invoice.paypal_order_id = "PAYPAL-ORDER-123"
invoice.adyen_psp_reference = "PSP123456789"
# PayPal is checked first in payment_method
self.assertEqual(invoice.payment_method, "paypal")
def test_invoice_payment_method_priority_adyen_over_stripe(self):
"""Test that Adyen takes priority over Stripe (edge case)."""
invoice = Invoice(self.user)
invoice.shop = self.shop
invoice.adyen_psp_reference = "PSP123456789"
invoice.stripe_payment_intent_id = "pi_1234567890abcdef"
# Adyen is checked before Stripe in payment_method
self.assertEqual(invoice.payment_method, "adyen")
class TestAdyen(unittest.TestCase):
"""Test Adyen-related model functionality."""
@mock.patch("make_post_sell.models.user.is_user_name_available", mock_always_true)
def setUp(self):
# Create test shop
self.shop = Shop(
"test-shop",
"555-555-5555",
"123 Test St",
"Test shop description",
)
# Create test user
self.user = User("test@example.com")
def test_shop_is_adyen_ready_without_credentials(self):
"""Test that shop is not Adyen-ready without credentials."""
self.assertFalse(self.shop.is_adyen_ready)
def test_shop_is_adyen_ready_with_api_key_only(self):
"""Test that shop is not Adyen-ready with just API key."""
self.shop.adyen_api_key = "test_api_key"
self.assertFalse(self.shop.is_adyen_ready)
def test_shop_is_adyen_ready_with_merchant_account_only(self):
"""Test that shop is not Adyen-ready with just merchant account."""
self.shop.adyen_merchant_account = "TestMerchant"
self.assertFalse(self.shop.is_adyen_ready)
def test_shop_is_adyen_ready_with_both_credentials(self):
"""Test that shop is Adyen-ready with both API key and merchant account."""
self.shop.adyen_api_key = "test_api_key"
self.shop.adyen_merchant_account = "TestMerchant"
self.assertTrue(self.shop.is_adyen_ready)
def test_shop_is_adyen_not_ready(self):
"""Test the inverse property for convenience."""
self.assertTrue(self.shop.is_adyen_not_ready)
self.shop.adyen_api_key = "test_api_key"
self.shop.adyen_merchant_account = "TestMerchant"
self.assertFalse(self.shop.is_adyen_not_ready)
def test_shop_adyen_enabled_default(self):
"""Test that Adyen enabled is None before DB insert (server_default handles it)."""
# Before DB insert, the value is None (server_default of '1' applies on insert)
# When retrieved from DB after insert, it would be True
self.assertIsNone(self.shop.adyen_enabled)
def test_invoice_adyen_psp_reference_default(self):
"""Test that Adyen PSP reference is None by default."""
invoice = Invoice(self.user)
invoice.shop = self.shop
self.assertIsNone(invoice.adyen_psp_reference)
def test_invoice_adyen_psp_reference_set(self):
"""Test setting Adyen PSP reference."""
invoice = Invoice(self.user)
invoice.shop = self.shop
invoice.adyen_psp_reference = "882619391893263J"
self.assertEqual(invoice.adyen_psp_reference, "882619391893263J")

View file

@ -40,10 +40,18 @@ def billing(request):
setup_intent = request.shop.stripe.SetupIntent.create(
payment_method_types=["card"], customer=stripe_user_shop.cus_id
)
# Get PayPal saved payment method if exists
paypal_user_shop = None
if request.shop.is_paypal_ready:
paypal_user_shop = request.shop.paypal_user_shop(request.user)
return {
"client_secret": setup_intent.client_secret,
"cards": cards,
"active_card": active_card,
"paypal_user_shop": paypal_user_shop,
"paypal_enabled": request.paypal_enabled if hasattr(request, 'paypal_enabled') else False,
}
@ -129,3 +137,31 @@ def update_card(request):
request.dbsession.flush()
request.session.flash(("You set the active card.", "success"))
return HTTPFound("/billing")
@view_config(route_name="disconnect-paypal")
@user_required()
@shop_is_ready_required()
def disconnect_paypal(request):
"""Disconnect (remove) saved PayPal payment method."""
# Check if shop has PayPal configured
if not request.shop.is_paypal_ready:
request.session.flash(("PayPal is not configured for this shop.", "error"))
return HTTPFound("/billing")
paypal_user_shop = request.shop.paypal_user_shop(request.user)
if paypal_user_shop is None:
request.session.flash(("No PayPal account connected.", "info"))
return HTTPFound("/billing")
# Clear saved payment token and payer ID
paypal_user_shop.active_payment_token = None
paypal_user_shop.payer_id = None
paypal_user_shop.billing_agreement_id = None
request.dbsession.add(paypal_user_shop)
request.dbsession.flush()
request.session.flash(("PayPal account disconnected successfully.", "success"))
return HTTPFound("/billing")

View file

@ -18,6 +18,8 @@ from ..lib.mail import (
)
import stripe
import traceback
from datetime import datetime
def get_cart_from_matchdict(request):
@ -452,6 +454,7 @@ def cart_handling_option(request):
@shop_is_ready_required()
def cart_checkout(request):
stripe_user_shop = request.shop.stripe_user_shop(request.user)
paypal_user_shop = request.shop.paypal_user_shop(request.user)
if "cart_id" not in request.matchdict:
return HTTPFound(f"/u/cart/{request.active_cart.id}/checkout")
@ -503,6 +506,7 @@ def cart_checkout(request):
# Only force Stripe flow if Stripe is the ONLY enabled payment method
only_stripe_enabled = (
request.stripe_enabled
and not request.paypal_enabled
and not request.monero_enabled
and not request.dogecoin_enabled
)
@ -577,6 +581,8 @@ def cart_checkout(request):
"products": cart.products,
"active_card": stripe_user_shop.active_card if stripe_user_shop else None,
"stripe_enabled": request.stripe_enabled,
"paypal_enabled": request.paypal_enabled,
"paypal_user_shop": paypal_user_shop,
"monero_enabled": request.monero_enabled,
"monero_synced": request.monero_synced,
"xmr_processor_enabled": xmr_processor_enabled,
@ -675,6 +681,9 @@ def cart_complete_checkout(request):
request.session.flash(msg)
return HTTPFound("/billing")
# Use idempotency key to prevent duplicate charges on retry
idempotency_key = f"checkout_{cart.uuid_str}_{shop.uuid_str}_{invoice.total_in_cents}"
payment_intent = shop.stripe.PaymentIntent.create(
amount=invoice.total_in_cents,
currency="usd",
@ -682,8 +691,14 @@ def cart_complete_checkout(request):
payment_method=stripe_user_shop.active_card_id,
off_session=True,
confirm=True,
idempotency_key=idempotency_key,
)
# Store Stripe payment references on invoice
invoice.stripe_payment_intent_id = payment_intent.id
if payment_intent.latest_charge:
invoice.stripe_charge_id = payment_intent.latest_charge
# Only persist data after successful payment
for invoice in invoices:
for line_item in invoice.line_items:
@ -725,3 +740,546 @@ def cart_complete_checkout(request):
msg = (f"Payment failed: {str(e)}", "error")
request.session.flash(msg)
return HTTPFound("/billing")
@view_config(
route_name="paypal_complete_checkout", request_method="POST", require_csrf=True
)
@user_required()
@shop_is_ready_required()
def paypal_complete_checkout(request):
"""Complete checkout using PayPal payment."""
if not request.paypal_enabled:
request.session.flash(("PayPal payments are disabled.", "error"))
return HTTPFound("/cart")
cart = get_cart_from_matchdict(request)
if cart is None:
request.session.flash(("That cart_id does not exist.", "error"))
return HTTPFound(get_referer_or_home(request))
elif cart.is_not_public and request.user.does_not_own_cart(cart):
request.session.flash(("That cart is not public and you do not own that cart.", "error"))
return HTTPFound(get_referer_or_home(request))
elif cart.is_empty:
request.session.flash(("That cart is empty, you cannot checkout.", "error"))
return HTTPFound(get_referer_or_home(request))
error_messages = cart.validate_attached_coupons()
if error_messages:
for error_message in error_messages:
request.session.flash((error_message, "error"))
return HTTPFound(get_referer_or_home(request))
paypal_order_ids_param = request.params.get("paypal_order_id")
if not paypal_order_ids_param:
request.session.flash(("PayPal order ID is missing.", "error"))
return HTTPFound("/cart")
paypal_order_ids = [oid.strip() for oid in paypal_order_ids_param.split(",")]
successful_invoices = []
failed_shops = []
try:
import requests
invoice_map = {}
for shop_id, product_quantity_tuple in cart.shop_product_dict.items():
shop = cart.shops[shop_id]
invoice = Invoice(request.user)
invoice.shop = shop
invoice.shop_id = shop.id
invoice.handling_option = cart.handling_option
invoice.handling_cost_in_cents = cart.handling_cost_in_cents
if cart.physical_products:
invoice.delivery_address = request.user.active_address.data
for product, quantity in product_quantity_tuple:
invoice.new_line_item(product=product, quantity=quantity)
for coupon in cart.coupons:
if hasattr(coupon, 'is_active') and not coupon.is_active:
continue
invoice.new_coupon_redemption(coupon)
invoice_map[shop_id] = invoice
invoices_requiring_payment = [inv for inv in invoice_map.values() if inv.requires_payment]
if len(paypal_order_ids) != len(invoices_requiring_payment):
request.session.flash((f"PayPal order count mismatch.", "error"))
return HTTPFound("/cart")
for idx, invoice in enumerate(invoices_requiring_payment):
shop = invoice.shop
paypal_order_id = paypal_order_ids[idx]
try:
api = shop.paypal
mode = api.mode if hasattr(api, 'mode') else 'sandbox'
base_url = "https://api-m.sandbox.paypal.com" if mode == "sandbox" else "https://api-m.paypal.com"
auth_response = requests.post(
f"{base_url}/v1/oauth2/token",
headers={"Accept": "application/json", "Accept-Language": "en_US"},
data={"grant_type": "client_credentials"},
auth=(shop.paypal_client_id, shop.paypal_secret)
)
if auth_response.status_code != 200:
failed_shops.append((shop, "Payment processor configuration error."))
continue
access_token = auth_response.json()["access_token"]
capture_response = requests.post(
f"{base_url}/v2/checkout/orders/{paypal_order_id}/capture",
headers={
"Content-Type": "application/json",
"Authorization": f"Bearer {access_token}"
}
)
if capture_response.status_code not in [200, 201]:
failed_shops.append((shop, "Payment could not be processed."))
continue
order = capture_response.json()
try:
captured_amount = float(order["purchase_units"][0]["payments"]["captures"][0]["amount"]["value"])
if abs(captured_amount - invoice.total) > 0.01:
failed_shops.append((shop, "Payment amount verification failed."))
continue
except (KeyError, ValueError, IndexError):
failed_shops.append((shop, "Payment processing error."))
continue
invoice.paypal_order_id = paypal_order_id
if (
"purchase_units" in order
and len(order["purchase_units"]) > 0
and "payments" in order["purchase_units"][0]
and "captures" in order["purchase_units"][0]["payments"]
and len(order["purchase_units"][0]["payments"]["captures"]) > 0
):
invoice.paypal_capture_id = order["purchase_units"][0]["payments"]["captures"][0]["id"]
# Handle vaulting
try:
if "payment_source" in order and "paypal" in order["payment_source"]:
paypal_source = order["payment_source"]["paypal"]
if "attributes" in paypal_source and "vault" in paypal_source["attributes"]:
vault_info = paypal_source["attributes"]["vault"]
if vault_info.get("status") == "VAULTED" and vault_info.get("id"):
from ..models.paypal_user_shop import PayPalUserShop
paypal_user_shop = shop.paypal_user_shop(request.user)
if paypal_user_shop is None:
paypal_user_shop = PayPalUserShop(user=request.user, shop=shop)
request.dbsession.add(paypal_user_shop)
paypal_user_shop.active_payment_token = vault_info["id"]
if "payer" in order and "payer_id" in order["payer"]:
paypal_user_shop.payer_id = order["payer"]["payer_id"]
request.dbsession.add(paypal_user_shop)
except Exception:
pass
for line_item in invoice.line_items:
line_item.product.unlock_for_user(request.user)
request.dbsession.add(line_item.product)
request.dbsession.add(invoice)
successful_invoices.append(invoice)
except Exception as e:
failed_shops.append((shop, f"Unexpected error: {str(e)}"))
continue
if successful_invoices:
for invoice in successful_invoices:
for line_item in invoice.line_items:
cart.remove_product(line_item.product, line_item.quantity)
cart.update_inventory(request.shop_location)
for invoice in successful_invoices:
send_purchase_email(
request,
request.user.email,
[item.product for item in invoice.line_items],
invoice.total,
)
send_sale_email(
request,
invoice.shop,
[item.product for item in invoice.line_items],
invoice.total,
)
if successful_invoices and not failed_shops:
request.session.flash(("Success! You have completed the purchase.", "success"))
elif successful_invoices and failed_shops:
request.session.flash(("Partial success. Some payments failed.", "warning"))
for shop, error_msg in failed_shops:
request.session.flash((f"{shop.name}: {error_msg}", "error"))
else:
request.session.flash(("All payments failed.", "error"))
for shop, error_msg in failed_shops:
request.session.flash((f"{shop.name}: {error_msg}", "error"))
save_cart(request)
if successful_invoices:
return HTTPFound(get_smart_purchase_redirect_url(successful_invoices))
return HTTPFound("/cart")
except Exception as e:
request.tm.abort()
request.session.flash((f"Payment processing failed: {str(e)}", "error"))
return HTTPFound("/cart")
@view_config(route_name="adyen_create_session", request_method="POST", renderer="json")
@user_required()
@shop_is_ready_required()
def adyen_create_session(request):
"""Create Adyen checkout session(s) for the cart."""
if not getattr(request, "adyen_enabled", False):
return {"error": "Adyen payments are disabled"}
cart = get_cart_from_matchdict(request)
if cart is None:
return {"error": "Cart not found"}
if cart.is_empty:
return {"error": "Cart is empty"}
try:
sessions = []
invoice_map = {}
for shop_id, product_quantity_tuple in cart.shop_product_dict.items():
shop = cart.shops[shop_id]
invoice = Invoice(request.user)
invoice.shop = shop
invoice.shop_id = shop.id
invoice.handling_option = cart.handling_option
invoice.handling_cost_in_cents = cart.handling_cost_in_cents
if cart.physical_products:
invoice.delivery_address = request.user.active_address.data
for product, quantity in product_quantity_tuple:
invoice.new_line_item(product=product, quantity=quantity)
for coupon in cart.coupons:
invoice.new_coupon_redemption(coupon)
invoice_map[shop_id] = invoice
for shop_id, invoice in invoice_map.items():
shop = invoice.shop
if not shop.adyen or not shop.is_adyen_ready:
return {"error": f"Adyen is not configured for shop: {shop.name}"}
# Create Adyen checkout session
adyen = shop.adyen
result = adyen.checkout.payments_api.sessions({
"amount": {
"currency": "USD",
"value": invoice.total_in_cents
},
"reference": f"cart_{cart.uuid_str}_{shop.uuid_str}",
"merchantAccount": shop.adyen_merchant_account,
"returnUrl": f"{request.host_url}/adyen/complete-checkout/{cart.id}",
"shopperReference": request.user.uuid_str,
"shopperEmail": request.user.email,
})
if result.status_code == 201:
session_data = result.message
sessions.append({
"shop_id": shop.uuid_str,
"session_id": session_data.get("id"),
"session_data": session_data.get("sessionData"),
"client_key": shop.adyen_client_key,
})
else:
return {"error": f"Failed to create Adyen session for {shop.name}"}
return {"sessions": sessions}
except Exception as e:
return {"error": str(e)}
@view_config(
route_name="adyen_complete_checkout", request_method="POST", require_csrf=True
)
@user_required()
@shop_is_ready_required()
def adyen_complete_checkout(request):
"""Complete checkout using Adyen payment."""
if not getattr(request, "adyen_enabled", False):
request.session.flash(("Adyen payments are disabled.", "error"))
return HTTPFound("/cart")
cart = get_cart_from_matchdict(request)
if cart is None:
request.session.flash(("That cart_id does not exist.", "error"))
return HTTPFound(get_referer_or_home(request))
elif cart.is_not_public and request.user.does_not_own_cart(cart):
request.session.flash(("That cart is not public and you do not own that cart.", "error"))
return HTTPFound(get_referer_or_home(request))
elif cart.is_empty:
request.session.flash(("That cart is empty, you cannot checkout.", "error"))
return HTTPFound(get_referer_or_home(request))
error_messages = cart.validate_attached_coupons()
if error_messages:
for error_message in error_messages:
request.session.flash((error_message, "error"))
return HTTPFound(get_referer_or_home(request))
# Get PSP references from Adyen redirectResultCode or params
psp_references_param = request.params.get("psp_references")
if not psp_references_param:
request.session.flash(("Adyen payment reference is missing.", "error"))
return HTTPFound("/cart")
psp_references = [ref.strip() for ref in psp_references_param.split(",")]
successful_invoices = []
failed_shops = []
try:
invoice_map = {}
for shop_id, product_quantity_tuple in cart.shop_product_dict.items():
shop = cart.shops[shop_id]
invoice = Invoice(request.user)
invoice.shop = shop
invoice.shop_id = shop.id
invoice.handling_option = cart.handling_option
invoice.handling_cost_in_cents = cart.handling_cost_in_cents
if cart.physical_products:
invoice.delivery_address = request.user.active_address.data
for product, quantity in product_quantity_tuple:
invoice.new_line_item(product=product, quantity=quantity)
for coupon in cart.coupons:
if hasattr(coupon, 'is_active') and not coupon.is_active:
continue
invoice.new_coupon_redemption(coupon)
invoice_map[shop_id] = invoice
invoices_requiring_payment = [inv for inv in invoice_map.values() if inv.requires_payment]
if len(psp_references) != len(invoices_requiring_payment):
request.session.flash((f"Adyen payment reference count mismatch.", "error"))
return HTTPFound("/cart")
for idx, invoice in enumerate(invoices_requiring_payment):
shop = invoice.shop
psp_reference = psp_references[idx]
try:
# Verify the payment with Adyen
adyen = shop.adyen
payment_details = adyen.checkout.payments_api.get_result_of_payment_session({
"sessionId": psp_reference
})
if payment_details.status_code != 200:
failed_shops.append((shop, "Payment verification failed."))
continue
result = payment_details.message
result_code = result.get("resultCode", "")
if result_code not in ["Authorised", "Received"]:
failed_shops.append((shop, f"Payment not authorized: {result_code}"))
continue
# Store the PSP reference
invoice.adyen_psp_reference = result.get("pspReference", psp_reference)
for line_item in invoice.line_items:
line_item.product.unlock_for_user(request.user)
request.dbsession.add(line_item.product)
request.dbsession.add(invoice)
successful_invoices.append(invoice)
except Exception as e:
failed_shops.append((shop, f"Unexpected error: {str(e)}"))
continue
if successful_invoices:
for invoice in successful_invoices:
for line_item in invoice.line_items:
cart.remove_product(line_item.product, line_item.quantity)
cart.update_inventory(request.shop_location)
for invoice in successful_invoices:
send_purchase_email(
request,
request.user.email,
[item.product for item in invoice.line_items],
invoice.total,
)
send_sale_email(
request,
invoice.shop,
[item.product for item in invoice.line_items],
invoice.total,
)
if successful_invoices and not failed_shops:
request.session.flash(("Success! You have completed the purchase.", "success"))
elif successful_invoices and failed_shops:
request.session.flash(("Partial success. Some payments failed.", "warning"))
for shop, error_msg in failed_shops:
request.session.flash((f"{shop.name}: {error_msg}", "error"))
else:
request.session.flash(("All payments failed.", "error"))
for shop, error_msg in failed_shops:
request.session.flash((f"{shop.name}: {error_msg}", "error"))
save_cart(request)
if successful_invoices:
return HTTPFound(get_smart_purchase_redirect_url(successful_invoices))
return HTTPFound("/cart")
except Exception as e:
request.tm.abort()
request.session.flash((f"Payment processing failed: {str(e)}", "error"))
return HTTPFound("/cart")
@view_config(route_name="paypal_create_order", request_method="POST", renderer="json")
@user_required()
@shop_is_ready_required()
def paypal_create_order(request):
"""Create PayPal order(s) for the cart."""
if not request.paypal_enabled:
return {"error": "PayPal payments are disabled"}
cart = get_cart_from_matchdict(request)
if cart is None:
return {"error": "Cart not found"}
if cart.is_empty:
return {"error": "Cart is empty"}
try:
import requests
order_ids = []
invoice_map = {}
for shop_id, product_quantity_tuple in cart.shop_product_dict.items():
shop = cart.shops[shop_id]
invoice = Invoice(request.user)
invoice.shop = shop
invoice.shop_id = shop.id
invoice.handling_option = cart.handling_option
invoice.handling_cost_in_cents = cart.handling_cost_in_cents
if cart.physical_products:
invoice.delivery_address = request.user.active_address.data
for product, quantity in product_quantity_tuple:
invoice.new_line_item(product=product, quantity=quantity)
for coupon in cart.coupons:
invoice.new_coupon_redemption(coupon)
invoice_map[shop_id] = invoice
for shop_id, invoice in invoice_map.items():
shop = invoice.shop
if not shop.paypal or not shop.is_paypal_ready:
return {"error": f"PayPal is not configured for shop: {shop.name}"}
shop_total_dollars = invoice.total
api = shop.paypal
mode = api.mode if hasattr(api, 'mode') else 'sandbox'
base_url = "https://api-m.sandbox.paypal.com" if mode == "sandbox" else "https://api-m.paypal.com"
auth_response = requests.post(
f"{base_url}/v1/oauth2/token",
headers={"Accept": "application/json", "Accept-Language": "en_US"},
data={"grant_type": "client_credentials"},
auth=(shop.paypal_client_id, shop.paypal_secret)
)
if auth_response.status_code != 200:
return {"error": f"PayPal unavailable for {shop.name}."}
access_token = auth_response.json()["access_token"]
currency = getattr(shop, 'currency', None) or 'USD'
save_paypal = request.params.get("save_paypal", "false") == "true"
order_json = {
"intent": "CAPTURE",
"purchase_units": [{
"amount": {
"currency_code": currency,
"value": f"{shop_total_dollars:.2f}"
},
"description": f"Purchase from {shop.name}"
}]
}
if save_paypal:
order_json["payment_source"] = {
"paypal": {
"attributes": {
"vault": {
"store_in_vault": "ON_SUCCESS",
"usage_type": "MERCHANT"
}
},
"experience_context": {
"payment_method_preference": "IMMEDIATE_PAYMENT_REQUIRED",
"user_action": "PAY_NOW"
}
}
}
order_response = requests.post(
f"{base_url}/v2/checkout/orders",
headers={
"Content-Type": "application/json",
"Authorization": f"Bearer {access_token}"
},
json=order_json
)
if order_response.status_code not in [200, 201]:
return {"error": f"Unable to create PayPal order for {shop.name}."}
order_data = order_response.json()
order_ids.append(order_data["id"])
return {"order_ids": order_ids}
except Exception as e:
return {"error": str(e)}

View file

@ -648,6 +648,38 @@ def shop_settings(request):
)
request.session.flash(msg)
# Handle Adyen settings form
if form_section == "adyen-settings":
adyen_api_key = request.params.get("adyen_api_key", "").strip()
adyen_merchant_account = request.params.get("adyen_merchant_account", "").strip()
adyen_client_key = request.params.get("adyen_client_key", "").strip()
adyen_hmac_key = request.params.get("adyen_hmac_key", "").strip()
# Handle disable action
if request.params.get("disable_adyen"):
shop.adyen_enabled = False
request.session.flash(("Adyen payments disabled", "success"))
# Handle re-enable action
elif not shop.adyen_enabled:
shop.adyen_enabled = True
request.session.flash(("Adyen payments re-enabled", "success"))
else:
if adyen_api_key and adyen_api_key != shop.adyen_api_key:
shop.adyen_api_key = adyen_api_key
request.session.flash(("You set the shop's Adyen API Key.", "success"))
if adyen_merchant_account and adyen_merchant_account != shop.adyen_merchant_account:
shop.adyen_merchant_account = adyen_merchant_account
request.session.flash(("You set the shop's Adyen Merchant Account.", "success"))
if adyen_client_key and adyen_client_key != shop.adyen_client_key:
shop.adyen_client_key = adyen_client_key
request.session.flash(("You set the shop's Adyen Client Key.", "success"))
if adyen_hmac_key and adyen_hmac_key != shop.adyen_hmac_key:
shop.adyen_hmac_key = adyen_hmac_key
request.session.flash(("You set the shop's Adyen HMAC Key.", "success"))
# Handle maintenance settings form
if form_section == "maintenance-settings":
if shop.maint_mode != maint_mode:
@ -911,6 +943,11 @@ def shop_settings(request):
"plausible_domain_name": shop.plausible_domain_name or "",
"stripe_public_api_key": shop.stripe_public_api_key or "",
"stripe_secret_api_key": shop.stripe_secret_api_key or "",
"adyen_api_key": shop.adyen_api_key or "",
"adyen_merchant_account": shop.adyen_merchant_account or "",
"adyen_client_key": shop.adyen_client_key or "",
"adyen_hmac_key": shop.adyen_hmac_key or "",
"adyen_enabled": shop.adyen_enabled,
"crypto_quote_expiry_seconds": shop.crypto_quote_expiry_seconds,
"payment_risk_threshold_mid_dollars": cents_to_dollars(
shop.payment_risk_threshold_mid_cents

View file

@ -0,0 +1,710 @@
"""
Webhook handlers for payment providers.
This module handles incoming webhooks from:
- PayPal: Payment capture, order approval, disputes
- Stripe: Payment success, failure, refunds, disputes
Webhooks provide resilience when checkout flows fail after
the payment provider has successfully processed the payment.
"""
import json
import logging
from pyramid.view import view_config
from pyramid.response import Response
import requests
from ..models.invoice import get_invoice_by_paypal_order_id
from ..lib.mail import send_purchase_email, send_sale_email
log = logging.getLogger(__name__)
# =============================================================================
# PayPal Webhooks
# =============================================================================
@view_config(route_name="paypal_webhook", request_method="POST")
def paypal_webhook(request):
"""
Handle PayPal webhook notifications.
Key events:
- PAYMENT.CAPTURE.COMPLETED - Payment was captured successfully
- CHECKOUT.ORDER.APPROVED - Order approved, needs capture (backup)
- PAYMENT.CAPTURE.DENIED - Payment was denied
- CUSTOMER.DISPUTE.CREATED - Dispute opened
Note: PAYMENT.CAPTURE.REFUNDED is NOT handled. Refunds are managed
externally by PayPal and the shop owner without platform involvement.
"""
try:
webhook_event = json.loads(request.body.decode("utf-8"))
event_type = webhook_event.get("event_type")
resource = webhook_event.get("resource", {})
log.info(f"PayPal webhook received: {event_type}")
# Extract order ID based on event type
order_id = _extract_order_id(event_type, resource)
if not order_id:
log.warning(f"PayPal webhook: Could not extract order_id from {event_type}")
return _json_response({"status": "ok", "message": "No order_id found"}, 200)
# Look up the invoice
invoice = get_invoice_by_paypal_order_id(request.dbsession, order_id)
if not invoice:
log.warning(f"PayPal webhook: No invoice found for order {order_id}")
return _json_response({"status": "ok", "message": "Invoice not found"}, 200)
shop = invoice.shop
if not shop:
log.error(f"PayPal webhook: Invoice {invoice.id} has no shop")
return _json_response({"status": "error", "message": "No shop"}, 400)
# Verify webhook signature
if not _verify_webhook_signature(request, webhook_event, shop):
log.warning(f"PayPal webhook: Signature verification failed for order {order_id}")
# Still return 200 to prevent PayPal from retrying endlessly
# Log it for investigation
return _json_response({"status": "ok", "message": "Signature verification skipped"}, 200)
# Process the event
if event_type == "PAYMENT.CAPTURE.COMPLETED":
_handle_capture_completed(request, invoice, resource)
elif event_type == "CHECKOUT.ORDER.APPROVED":
_handle_order_approved(request, invoice, shop, order_id)
elif event_type == "PAYMENT.CAPTURE.DENIED":
log.warning(f"PayPal payment DENIED for order {order_id}, invoice {invoice.id}")
elif event_type == "CUSTOMER.DISPUTE.CREATED":
_handle_dispute_created(resource, order_id)
return _json_response({"status": "success"}, 200)
except Exception as e:
log.exception(f"PayPal webhook error: {str(e)}")
# Return 200 anyway to prevent infinite retries
return _json_response({"status": "error", "message": str(e)}, 200)
def _extract_order_id(event_type, resource):
"""Extract order ID from webhook resource based on event type."""
# For capture events, order_id is in supplementary_data
if "supplementary_data" in resource:
related_ids = resource.get("supplementary_data", {}).get("related_ids", {})
order_id = related_ids.get("order_id")
if order_id:
return order_id
# For order events, the resource itself is the order
if event_type and "ORDER" in event_type:
return resource.get("id")
# Fallback: try common locations
return resource.get("order_id") or resource.get("id")
def _verify_webhook_signature(request, webhook_event, shop):
"""
Verify PayPal webhook signature.
Returns True if verified, False if verification failed or skipped.
"""
# Get webhook ID from app settings (configured per-deployment)
webhook_id = request.registry.settings.get("paypal.webhook_id")
if not webhook_id:
log.debug("PayPal webhook_id not configured, skipping verification")
return True # Skip verification if not configured
# Get required headers
headers = {
"transmission_id": request.headers.get("PAYPAL-TRANSMISSION-ID"),
"transmission_time": request.headers.get("PAYPAL-TRANSMISSION-TIME"),
"cert_url": request.headers.get("PAYPAL-CERT-URL"),
"auth_algo": request.headers.get("PAYPAL-AUTH-ALGO"),
"transmission_sig": request.headers.get("PAYPAL-TRANSMISSION-SIG"),
}
if not all(headers.values()):
log.warning("PayPal webhook: Missing verification headers")
return False
try:
# Determine API base URL
sandbox_mode = request.registry.settings.get("app.paypal.sandbox_mode", "True")
is_sandbox = str(sandbox_mode).lower() in ("true", "1", "yes")
base_url = "https://api-m.sandbox.paypal.com" if is_sandbox else "https://api-m.paypal.com"
# Get OAuth token
auth_response = requests.post(
f"{base_url}/v1/oauth2/token",
headers={"Accept": "application/json"},
data={"grant_type": "client_credentials"},
auth=(shop.paypal_client_id, shop.paypal_secret),
timeout=10
)
if auth_response.status_code != 200:
log.error(f"PayPal webhook auth failed: {auth_response.text}")
return False
access_token = auth_response.json()["access_token"]
# Verify signature
verify_response = requests.post(
f"{base_url}/v1/notifications/verify-webhook-signature",
headers={
"Content-Type": "application/json",
"Authorization": f"Bearer {access_token}"
},
json={
"transmission_id": headers["transmission_id"],
"transmission_time": headers["transmission_time"],
"cert_url": headers["cert_url"],
"auth_algo": headers["auth_algo"],
"transmission_sig": headers["transmission_sig"],
"webhook_id": webhook_id,
"webhook_event": webhook_event
},
timeout=10
)
if verify_response.status_code != 200:
log.error(f"PayPal webhook verification API error: {verify_response.text}")
return False
result = verify_response.json()
if result.get("verification_status") == "SUCCESS":
log.debug("PayPal webhook signature verified")
return True
else:
log.warning(f"PayPal webhook signature invalid: {result}")
return False
except Exception as e:
log.exception(f"PayPal webhook verification error: {e}")
return False
def _handle_capture_completed(request, invoice, resource):
"""
Handle PAYMENT.CAPTURE.COMPLETED event.
This is the backup for when the JS callback fails but PayPal
successfully captured the payment.
"""
capture_id = resource.get("id")
# Idempotency: check if already processed
if invoice.paypal_capture_id:
log.info(f"PayPal webhook: Invoice {invoice.id} already has capture_id, skipping")
return
log.info(f"PayPal webhook: Processing capture {capture_id} for invoice {invoice.id}")
# Update invoice with capture ID
invoice.paypal_capture_id = capture_id
request.dbsession.add(invoice)
# Unlock products for the user
user = invoice.user
if user:
for line_item in invoice.line_items:
if not line_item.product.is_unlocked_for_user(user):
line_item.product.unlock_for_user(user)
request.dbsession.add(line_item.product)
log.info(f"PayPal webhook: Unlocked product {line_item.product.id} for user {user.id}")
# Send confirmation emails
try:
products = [item.product for item in invoice.line_items]
send_purchase_email(request, user.email, products, invoice.total)
send_sale_email(request, invoice.shop, products, invoice.total)
log.info(f"PayPal webhook: Sent confirmation emails for invoice {invoice.id}")
except Exception as e:
log.exception(f"PayPal webhook: Failed to send emails for invoice {invoice.id}: {e}")
request.dbsession.flush()
def _handle_order_approved(request, invoice, shop, order_id):
"""
Handle CHECKOUT.ORDER.APPROVED event.
This means the customer approved the payment in PayPal, but we haven't
captured it yet. This is a backup in case our JS onApprove failed.
"""
# Check if already captured
if invoice.paypal_capture_id:
log.info(f"PayPal webhook: Order {order_id} already captured, skipping")
return
log.info(f"PayPal webhook: Order {order_id} approved but not captured, attempting capture")
try:
# Determine API base URL
sandbox_mode = request.registry.settings.get("app.paypal.sandbox_mode", "True")
is_sandbox = str(sandbox_mode).lower() in ("true", "1", "yes")
base_url = "https://api-m.sandbox.paypal.com" if is_sandbox else "https://api-m.paypal.com"
# Get OAuth token
auth_response = requests.post(
f"{base_url}/v1/oauth2/token",
headers={"Accept": "application/json"},
data={"grant_type": "client_credentials"},
auth=(shop.paypal_client_id, shop.paypal_secret),
timeout=10
)
if auth_response.status_code != 200:
log.error(f"PayPal webhook: Auth failed for capture attempt: {auth_response.text}")
return
access_token = auth_response.json()["access_token"]
# Capture the payment
capture_response = requests.post(
f"{base_url}/v2/checkout/orders/{order_id}/capture",
headers={
"Content-Type": "application/json",
"Authorization": f"Bearer {access_token}"
},
timeout=15
)
if capture_response.status_code not in [200, 201]:
log.error(f"PayPal webhook: Capture failed for order {order_id}: {capture_response.text}")
return
order = capture_response.json()
log.info(f"PayPal webhook: Successfully captured order {order_id}")
# Extract capture ID
try:
capture_id = order["purchase_units"][0]["payments"]["captures"][0]["id"]
invoice.paypal_capture_id = capture_id
request.dbsession.add(invoice)
except (KeyError, IndexError):
log.warning(f"PayPal webhook: Could not extract capture_id from response")
# Unlock products and send emails
user = invoice.user
if user:
for line_item in invoice.line_items:
if not line_item.product.is_unlocked_for_user(user):
line_item.product.unlock_for_user(user)
request.dbsession.add(line_item.product)
try:
products = [item.product for item in invoice.line_items]
send_purchase_email(request, user.email, products, invoice.total)
send_sale_email(request, invoice.shop, products, invoice.total)
log.info(f"PayPal webhook: Sent emails after webhook-initiated capture")
except Exception as e:
log.exception(f"PayPal webhook: Failed to send emails: {e}")
request.dbsession.flush()
except Exception as e:
log.exception(f"PayPal webhook: Error capturing order {order_id}: {e}")
def _handle_dispute_created(resource, order_id):
"""Handle CUSTOMER.DISPUTE.CREATED event."""
dispute_id = resource.get("dispute_id", "unknown")
dispute_amount = resource.get("dispute_amount", {}).get("value", "unknown")
dispute_reason = resource.get("reason", "unknown")
log.critical(
f"PayPal DISPUTE created - Order: {order_id}, "
f"Dispute ID: {dispute_id}, Amount: ${dispute_amount}, Reason: {dispute_reason}"
)
def _json_response(data, status):
"""Create a JSON response."""
return Response(
json.dumps(data),
content_type="application/json; charset=utf-8",
status=status
)
# =============================================================================
# Stripe Webhooks
# =============================================================================
@view_config(route_name="stripe_webhook", request_method="POST")
def stripe_webhook(request):
"""
Handle Stripe webhook notifications.
Key events:
- payment_intent.succeeded - Payment completed successfully
- payment_intent.payment_failed - Payment failed
- charge.refunded - Refund processed
- charge.dispute.created - Dispute opened
This provides resilience when the checkout flow fails after Stripe
has successfully processed the payment.
"""
import stripe
try:
payload = request.body
sig_header = request.headers.get("Stripe-Signature")
# Get webhook secret from settings
webhook_secret = request.registry.settings.get("stripe.webhook_secret")
if webhook_secret and sig_header:
try:
event = stripe.Webhook.construct_event(
payload, sig_header, webhook_secret
)
except ValueError:
log.warning("Stripe webhook: Invalid payload")
return _json_response({"error": "Invalid payload"}, 400)
except stripe.error.SignatureVerificationError:
log.warning("Stripe webhook: Invalid signature")
return _json_response({"error": "Invalid signature"}, 400)
else:
# No webhook secret configured, parse without verification
event = json.loads(payload.decode("utf-8"))
log.debug("Stripe webhook: No webhook_secret configured, skipping verification")
event_type = event.get("type") if isinstance(event, dict) else event.type
data_object = event.get("data", {}).get("object", {}) if isinstance(event, dict) else event.data.object
log.info(f"Stripe webhook received: {event_type}")
if event_type == "payment_intent.succeeded":
_handle_stripe_payment_succeeded(request, data_object)
elif event_type == "payment_intent.payment_failed":
_handle_stripe_payment_failed(request, data_object)
elif event_type == "charge.refunded":
_handle_stripe_refund(request, data_object)
elif event_type == "charge.dispute.created":
_handle_stripe_dispute(request, data_object)
return _json_response({"status": "success"}, 200)
except Exception as e:
log.exception(f"Stripe webhook error: {str(e)}")
return _json_response({"status": "error", "message": str(e)}, 200)
def _handle_stripe_payment_succeeded(request, payment_intent):
"""
Handle payment_intent.succeeded event.
This is the backup for when our checkout flow fails after Stripe
has successfully charged the card.
"""
from ..models.invoice import get_invoice_by_stripe_payment_intent_id
payment_intent_id = payment_intent.get("id") if isinstance(payment_intent, dict) else payment_intent.id
invoice = get_invoice_by_stripe_payment_intent_id(request.dbsession, payment_intent_id)
if not invoice:
log.warning(f"Stripe webhook: No invoice found for payment_intent {payment_intent_id}")
return
# Check if already processed (has charge_id)
if invoice.stripe_charge_id:
log.info(f"Stripe webhook: Invoice {invoice.id} already processed, skipping")
return
log.info(f"Stripe webhook: Processing payment_intent {payment_intent_id} for invoice {invoice.id}")
# Update invoice with charge ID
latest_charge = payment_intent.get("latest_charge") if isinstance(payment_intent, dict) else payment_intent.latest_charge
if latest_charge:
invoice.stripe_charge_id = latest_charge
request.dbsession.add(invoice)
# Unlock products for the user
user = invoice.user
if user:
for line_item in invoice.line_items:
if not line_item.product.is_unlocked_for_user(user):
line_item.product.unlock_for_user(user)
request.dbsession.add(line_item.product)
log.info(f"Stripe webhook: Unlocked product {line_item.product.id} for user {user.id}")
# Send confirmation emails
try:
products = [item.product for item in invoice.line_items]
send_purchase_email(request, user.email, products, invoice.total)
send_sale_email(request, invoice.shop, products, invoice.total)
log.info(f"Stripe webhook: Sent confirmation emails for invoice {invoice.id}")
except Exception as e:
log.exception(f"Stripe webhook: Failed to send emails for invoice {invoice.id}: {e}")
request.dbsession.flush()
def _handle_stripe_payment_failed(request, payment_intent):
"""Handle payment_intent.payment_failed event."""
payment_intent_id = payment_intent.get("id") if isinstance(payment_intent, dict) else payment_intent.id
error_message = ""
if isinstance(payment_intent, dict):
last_error = payment_intent.get("last_payment_error", {})
error_message = last_error.get("message", "Unknown error")
else:
if payment_intent.last_payment_error:
error_message = payment_intent.last_payment_error.message or "Unknown error"
log.warning(f"Stripe payment FAILED for payment_intent {payment_intent_id}: {error_message}")
def _handle_stripe_refund(request, charge):
"""Handle charge.refunded event."""
charge_id = charge.get("id") if isinstance(charge, dict) else charge.id
amount_refunded = charge.get("amount_refunded", 0) if isinstance(charge, dict) else charge.amount_refunded
# Convert from cents to dollars
amount_dollars = amount_refunded / 100
log.info(f"Stripe REFUND processed - Charge: {charge_id}, Amount: ${amount_dollars:.2f}")
def _handle_stripe_dispute(request, dispute):
"""Handle charge.dispute.created event."""
dispute_id = dispute.get("id") if isinstance(dispute, dict) else dispute.id
charge_id = dispute.get("charge") if isinstance(dispute, dict) else dispute.charge
amount = dispute.get("amount", 0) if isinstance(dispute, dict) else dispute.amount
reason = dispute.get("reason", "unknown") if isinstance(dispute, dict) else dispute.reason
# Convert from cents to dollars
amount_dollars = amount / 100
log.critical(
f"Stripe DISPUTE created - Dispute ID: {dispute_id}, "
f"Charge: {charge_id}, Amount: ${amount_dollars:.2f}, Reason: {reason}"
)
# =============================================================================
# Adyen Webhooks
# =============================================================================
@view_config(route_name="adyen_webhook", request_method="POST")
def adyen_webhook(request):
"""
Handle Adyen webhook notifications.
Key events:
- AUTHORISATION - Payment authorized
- CAPTURE - Payment captured
- REFUND - Refund processed
- CHARGEBACK - Dispute/chargeback created
Adyen uses HMAC-SHA256 for webhook verification.
"""
import hashlib
import hmac
import base64
import binascii
try:
payload = request.body.decode("utf-8")
notification = json.loads(payload)
# Adyen sends notifications in a wrapper
notification_items = notification.get("notificationItems", [])
for item in notification_items:
notification_request = item.get("NotificationRequestItem", {})
event_code = notification_request.get("eventCode")
psp_reference = notification_request.get("pspReference")
merchant_reference = notification_request.get("merchantReference")
log.info(f"Adyen webhook received: {event_code} for PSP ref {psp_reference}")
# Find the invoice by PSP reference
from ..models.invoice import get_invoice_by_adyen_psp_reference
invoice = get_invoice_by_adyen_psp_reference(request.dbsession, psp_reference)
if not invoice:
# Try to find by merchant reference (which includes cart/shop info)
log.info(f"Adyen webhook: No invoice found for PSP ref {psp_reference}, trying merchant ref")
# We can't verify the signature without the shop, so just acknowledge
continue
shop = invoice.shop
if not shop:
log.error(f"Adyen webhook: Invoice {invoice.id} has no shop")
continue
# Verify HMAC signature
if shop.adyen_hmac_key:
hmac_signature = request.headers.get("X-Adyen-Hmac-Signature")
if hmac_signature:
if not _verify_adyen_hmac(shop.adyen_hmac_key, hmac_signature, notification_request):
log.warning(f"Adyen webhook: HMAC verification failed for PSP ref {psp_reference}")
continue
# Process the event
if event_code == "AUTHORISATION":
success = notification_request.get("success") == "true"
if success:
_handle_adyen_authorisation(request, invoice, notification_request)
else:
reason = notification_request.get("reason", "Unknown")
log.warning(f"Adyen AUTHORISATION failed for PSP ref {psp_reference}: {reason}")
elif event_code == "CAPTURE":
_handle_adyen_capture(request, invoice, notification_request)
elif event_code == "REFUND":
_handle_adyen_refund(notification_request, psp_reference)
elif event_code == "CHARGEBACK":
_handle_adyen_chargeback(notification_request, psp_reference)
# Adyen expects [accepted] as response
return Response("[accepted]", content_type="text/plain; charset=utf-8", status=200)
except Exception as e:
log.exception(f"Adyen webhook error: {str(e)}")
# Return accepted to prevent infinite retries
return Response("[accepted]", content_type="text/plain; charset=utf-8", status=200)
def _verify_adyen_hmac(hmac_key, hmac_signature, notification_request):
"""
Verify Adyen webhook HMAC signature.
Adyen's HMAC is computed from a specific concatenation of fields.
"""
import hashlib
import hmac
import base64
import binascii
try:
# Build the signing string according to Adyen's specification
# Fields: pspReference, originalReference, merchantAccountCode, merchantReference,
# amount.value, amount.currency, eventCode, success
psp_reference = notification_request.get("pspReference", "")
original_reference = notification_request.get("originalReference", "")
merchant_account = notification_request.get("merchantAccountCode", "")
merchant_reference = notification_request.get("merchantReference", "")
amount = notification_request.get("amount", {})
amount_value = str(amount.get("value", ""))
amount_currency = amount.get("currency", "")
event_code = notification_request.get("eventCode", "")
success = notification_request.get("success", "")
# Concatenate with colons
signing_string = ":".join([
psp_reference,
original_reference,
merchant_account,
merchant_reference,
amount_value,
amount_currency,
event_code,
success
])
# Compute HMAC-SHA256
expected = hmac.new(
binascii.unhexlify(hmac_key),
signing_string.encode('utf-8'),
hashlib.sha256
).digest()
expected_signature = base64.b64encode(expected).decode('utf-8')
return hmac.compare_digest(hmac_signature, expected_signature)
except Exception as e:
log.exception(f"Adyen HMAC verification error: {e}")
return False
def _handle_adyen_authorisation(request, invoice, notification):
"""
Handle AUTHORISATION event.
This confirms the payment was authorized successfully.
"""
psp_reference = notification.get("pspReference")
# Idempotency check
if invoice.adyen_psp_reference == psp_reference:
log.info(f"Adyen webhook: Invoice {invoice.id} already has PSP ref, checking products")
# Update invoice if needed
if not invoice.adyen_psp_reference:
invoice.adyen_psp_reference = psp_reference
request.dbsession.add(invoice)
# Unlock products for the user
user = invoice.user
if user:
for line_item in invoice.line_items:
if not line_item.product.is_unlocked_for_user(user):
line_item.product.unlock_for_user(user)
request.dbsession.add(line_item.product)
log.info(f"Adyen webhook: Unlocked product {line_item.product.id} for user {user.id}")
# Send confirmation emails
try:
products = [item.product for item in invoice.line_items]
send_purchase_email(request, user.email, products, invoice.total)
send_sale_email(request, invoice.shop, products, invoice.total)
log.info(f"Adyen webhook: Sent confirmation emails for invoice {invoice.id}")
except Exception as e:
log.exception(f"Adyen webhook: Failed to send emails for invoice {invoice.id}: {e}")
request.dbsession.flush()
def _handle_adyen_capture(request, invoice, notification):
"""Handle CAPTURE event - payment was captured."""
psp_reference = notification.get("pspReference")
log.info(f"Adyen CAPTURE processed for invoice {invoice.id}, PSP ref {psp_reference}")
def _handle_adyen_refund(notification, psp_reference):
"""Handle REFUND event."""
amount = notification.get("amount", {})
amount_value = amount.get("value", 0)
currency = amount.get("currency", "USD")
# Convert from minor units
amount_dollars = amount_value / 100
log.info(f"Adyen REFUND processed - PSP ref: {psp_reference}, Amount: {amount_dollars:.2f} {currency}")
def _handle_adyen_chargeback(notification, psp_reference):
"""Handle CHARGEBACK event."""
amount = notification.get("amount", {})
amount_value = amount.get("value", 0)
currency = amount.get("currency", "USD")
reason = notification.get("reason", "unknown")
# Convert from minor units
amount_dollars = amount_value / 100
log.critical(
f"Adyen CHARGEBACK created - PSP ref: {psp_reference}, "
f"Amount: {amount_dollars:.2f} {currency}, Reason: {reason}"
)

View file

@ -29,6 +29,9 @@ bcrypt
# credit card storage and processing.
stripe
# PayPal REST API SDK for payments.
paypalrestsdk
# DKIM Signed Email from Python, lot's of extras in here like async.
# https://git.launchpad.net/dkimpy/tree/setup.py#n84
dkimpy

View file

@ -33,6 +33,10 @@ app.stripe.test_mode = True
# Payment method toggles
app.payments.stripe.enabled = True
app.payments.monero.enabled = False
app.payments.paypal.enabled = True
# PayPal sandbox mode for testing
app.paypal.sandbox_mode = True
# Monero RPC Configuration for tests
monero.rpc_url = http://127.0.0.1:18083/json_rpc