test(billing): add Stripe E2E test scripts for sandbox validation
- 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>
This commit is contained in:
124
scripts/stripe_e2e_setup.py
Normal file
124
scripts/stripe_e2e_setup.py
Normal file
@@ -0,0 +1,124 @@
|
||||
"""
|
||||
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()
|
||||
Reference in New Issue
Block a user