60fa2bc720
- test_stripe_sandbox.py: API-only validation of all 17 products (67 tests) - stripe_e2e_setup.py: webhook endpoint registration via ngrok - stripe_e2e_test.py: live webhook tests with real DB verification (67 tests) - stripe_e2e_checkout_test.py: checkout webhook tests for credit packs, sticky boosts, and business plan PDF purchases (40 tests) Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
125 lines
3.9 KiB
Python
125 lines
3.9 KiB
Python
"""
|
|
Step 1: Register a Stripe webhook endpoint via ngrok and update .env.
|
|
|
|
Run BEFORE starting the dev server:
|
|
1. Start ngrok: ngrok http 5000
|
|
2. Run this script: uv run python scripts/stripe_e2e_setup.py
|
|
3. Start dev server: make dev
|
|
4. Run E2E tests: uv run python scripts/stripe_e2e_test.py
|
|
|
|
To tear down afterward:
|
|
uv run python scripts/stripe_e2e_setup.py --teardown
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import re
|
|
import sys
|
|
import urllib.request
|
|
|
|
from dotenv import load_dotenv
|
|
|
|
load_dotenv()
|
|
|
|
import stripe
|
|
|
|
STRIPE_SECRET_KEY = os.getenv("STRIPE_SECRET_KEY", "") or os.getenv("STRIPE_API_PRIVATE_KEY", "")
|
|
if not STRIPE_SECRET_KEY:
|
|
print("ERROR: Set STRIPE_SECRET_KEY or STRIPE_API_PRIVATE_KEY in .env")
|
|
sys.exit(1)
|
|
|
|
stripe.api_key = STRIPE_SECRET_KEY
|
|
stripe.max_network_retries = 2
|
|
|
|
ENV_PATH = os.path.join(os.path.dirname(__file__), "..", ".env")
|
|
ENV_PATH = os.path.abspath(ENV_PATH)
|
|
WEBHOOK_PATH = "/billing/webhook/stripe"
|
|
NGROK_API = "http://localhost:4040/api/tunnels"
|
|
|
|
|
|
def _update_env(key, value):
|
|
"""Update a key in .env file."""
|
|
text = open(ENV_PATH).read()
|
|
pattern = rf"^{key}=.*$"
|
|
replacement = f"{key}={value}"
|
|
if re.search(pattern, text, re.MULTILINE):
|
|
text = re.sub(pattern, replacement, text, flags=re.MULTILINE)
|
|
else:
|
|
text = text.rstrip("\n") + f"\n{replacement}\n"
|
|
open(ENV_PATH, "w").write(text)
|
|
|
|
|
|
def setup():
|
|
# Get ngrok tunnel URL
|
|
try:
|
|
resp = urllib.request.urlopen(NGROK_API, timeout=5)
|
|
tunnels = json.loads(resp.read())
|
|
tunnel_url = tunnels["tunnels"][0]["public_url"]
|
|
except Exception as e:
|
|
print(f"ERROR: ngrok not running: {e}")
|
|
print("Start ngrok first: ngrok http 5000")
|
|
sys.exit(1)
|
|
|
|
webhook_url = f"{tunnel_url}{WEBHOOK_PATH}"
|
|
print(f"ngrok tunnel: {tunnel_url}")
|
|
print(f"Webhook URL: {webhook_url}")
|
|
|
|
# Check for existing E2E webhook endpoint
|
|
existing_id = os.getenv("STRIPE_WEBHOOK_ENDPOINT_ID", "")
|
|
if existing_id:
|
|
try:
|
|
ep = stripe.WebhookEndpoint.retrieve(existing_id)
|
|
if ep.url == webhook_url and ep.status == "enabled":
|
|
print(f"\nEndpoint already exists and matches: {existing_id}")
|
|
print("Ready to test. Run: uv run python scripts/stripe_e2e_test.py")
|
|
return
|
|
# URL changed (new ngrok session), delete and recreate
|
|
print(f"Existing endpoint URL mismatch, recreating...")
|
|
stripe.WebhookEndpoint.delete(existing_id)
|
|
except stripe.InvalidRequestError:
|
|
pass # Already deleted
|
|
|
|
# Create webhook endpoint
|
|
endpoint = stripe.WebhookEndpoint.create(
|
|
url=webhook_url,
|
|
enabled_events=[
|
|
"checkout.session.completed",
|
|
"customer.subscription.created",
|
|
"customer.subscription.updated",
|
|
"customer.subscription.deleted",
|
|
"invoice.payment_failed",
|
|
],
|
|
)
|
|
|
|
print(f"\nCreated endpoint: {endpoint.id}")
|
|
print(f"Webhook secret: {endpoint.secret[:25]}...")
|
|
|
|
# Update .env
|
|
_update_env("STRIPE_WEBHOOK_SECRET", endpoint.secret)
|
|
_update_env("STRIPE_WEBHOOK_ENDPOINT_ID", endpoint.id)
|
|
print("\nUpdated .env with STRIPE_WEBHOOK_SECRET and STRIPE_WEBHOOK_ENDPOINT_ID")
|
|
print("\nNext steps:")
|
|
print(" 1. Restart dev server: make dev")
|
|
print(" 2. Run E2E tests: uv run python scripts/stripe_e2e_test.py")
|
|
|
|
|
|
def teardown():
|
|
endpoint_id = os.getenv("STRIPE_WEBHOOK_ENDPOINT_ID", "")
|
|
if endpoint_id:
|
|
try:
|
|
stripe.WebhookEndpoint.delete(endpoint_id)
|
|
print(f"Deleted webhook endpoint: {endpoint_id}")
|
|
except stripe.InvalidRequestError:
|
|
print(f"Endpoint {endpoint_id} already deleted")
|
|
|
|
_update_env("STRIPE_WEBHOOK_SECRET", "")
|
|
_update_env("STRIPE_WEBHOOK_ENDPOINT_ID", "")
|
|
print("Cleared .env webhook config")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
if "--teardown" in sys.argv:
|
|
teardown()
|
|
else:
|
|
setup()
|