f8cd37ac8c
Includes: API (Express/TypeScript), Astro site, Python workers, document generators, FCC compliance tools, Canada CRTC formation, Ansible infrastructure, and deployment scripts. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
221 lines
8.5 KiB
Python
221 lines
8.5 KiB
Python
"""
|
||
E2E checkout test — Stripe card + PayPal on dev.performancewest.net
|
||
Runs inside the prod workers container (has Playwright + network access).
|
||
|
||
Usage: docker cp to workers container, then exec.
|
||
"""
|
||
import asyncio
|
||
import json
|
||
from playwright.async_api import async_playwright, Page, ConsoleMessage
|
||
|
||
BASE = "https://dev.performancewest.net"
|
||
API = "https://api.dev.performancewest.net"
|
||
IDENTITY_SESSION = None # Will find a verified one
|
||
|
||
# ── Helpers ────<E29480><E29480><EFBFBD>───────────────────────────────────────────────────────────────
|
||
|
||
async def get_verified_identity(page):
|
||
"""Find a verified identity session from the dev DB via API."""
|
||
# Just use test_mode which bypasses identity
|
||
return "test_mode_bypass"
|
||
|
||
async def fill_order_form(page: Page, label: str):
|
||
"""Fill steps 1-5 and return on step 5 ready to submit."""
|
||
await page.goto(f"{BASE}/order/canada-crtc?test_mode=1", wait_until="domcontentloaded", timeout=60000)
|
||
await page.wait_for_timeout(2000)
|
||
|
||
# Step 1 — numbered is default, just click Next
|
||
print(f" [{label}] Step 1: company type")
|
||
await page.click("#btn-next")
|
||
await page.wait_for_timeout(500)
|
||
|
||
# Step 2 — director
|
||
print(f" [{label}] Step 2: director")
|
||
await page.fill("#director_first_name", "Test")
|
||
await page.fill("#director_last_name", "Checkout")
|
||
await page.select_option("#director_country", "US")
|
||
await page.wait_for_timeout(400)
|
||
await page.fill("#director_street", "100 Test Ave")
|
||
await page.fill("#director_city", "Austin")
|
||
await page.select_option("#director_province_select", "TX")
|
||
await page.evaluate("document.getElementById('director_province_select').dispatchEvent(new Event('change',{bubbles:true}))")
|
||
await page.wait_for_timeout(200)
|
||
await page.fill("#director_postal", "73301")
|
||
try:
|
||
await page.select_option("#director_citizenship", "United States", timeout=3000)
|
||
except Exception:
|
||
pass
|
||
await page.click("#btn-next")
|
||
await page.wait_for_timeout(500)
|
||
|
||
# Step 3 — services
|
||
print(f" [{label}] Step 3: services")
|
||
textarea = await page.query_selector("#service_description")
|
||
if textarea:
|
||
await textarea.fill("VoIP reseller services for e2e checkout test")
|
||
await page.click("#btn-next")
|
||
await page.wait_for_timeout(500)
|
||
|
||
# Step 4 — identity (test_mode bypasses)
|
||
print(f" [{label}] Step 4: identity (test_mode bypass)")
|
||
await page.click("#btn-next")
|
||
await page.wait_for_timeout(500)
|
||
|
||
# Step 5 — billing contact
|
||
print(f" [{label}] Step 5: billing contact")
|
||
await page.fill("#customer_name", "Test Checkout")
|
||
await page.fill("#customer_email", f"test+{label}@performancewest.net")
|
||
await page.fill("#customer_phone", "+15125550199")
|
||
await page.check("#consent")
|
||
await page.wait_for_timeout(300)
|
||
|
||
|
||
async def test_stripe_card():
|
||
"""Test: fill form → submit → should redirect to checkout.stripe.com"""
|
||
print("\n=== TEST: Stripe Card Checkout ===")
|
||
async with async_playwright() as pw:
|
||
browser = await pw.chromium.launch(headless=True)
|
||
page = await browser.new_page()
|
||
|
||
requests_log = []
|
||
def on_response(resp):
|
||
if "api" in resp.url and resp.request.method in ("POST", "GET"):
|
||
requests_log.append((resp.request.method, resp.url, resp.status))
|
||
page.on("response", on_response)
|
||
|
||
errors = []
|
||
def on_error(err):
|
||
errors.append(str(err))
|
||
page.on("pageerror", on_error)
|
||
|
||
await fill_order_form(page, "stripe")
|
||
|
||
# Select card payment — radio is now on step 5
|
||
await page.click('input[name="payment_method_choice"][value="card"]')
|
||
await page.wait_for_timeout(200)
|
||
|
||
# Click Submit Order
|
||
print(" [stripe] Clicking Submit Order...")
|
||
await page.click("#btn-next")
|
||
|
||
# Wait for redirect
|
||
result = "TIMEOUT"
|
||
for i in range(20):
|
||
await page.wait_for_timeout(1000)
|
||
try:
|
||
url = page.url
|
||
if "checkout.stripe.com" in url:
|
||
result = "SUCCESS"
|
||
print(f" [stripe] Redirected to Stripe Checkout at t+{i+1}s")
|
||
break
|
||
status_el = await page.query_selector("#submit-status")
|
||
if status_el:
|
||
txt = (await status_el.inner_text()).strip()
|
||
if txt and txt not in ("", "Placing your order...", "Creating your order...", "Redirecting to payment..."):
|
||
result = f"ERROR: {txt}"
|
||
break
|
||
except Exception:
|
||
# Page navigated — check new URL
|
||
await page.wait_for_timeout(2000)
|
||
url = page.url
|
||
if "checkout.stripe.com" in url:
|
||
result = "SUCCESS"
|
||
print(f" [stripe] Redirected to Stripe Checkout at t+{i+1}s")
|
||
break
|
||
|
||
if errors:
|
||
print(f" [stripe] Page errors: {errors}")
|
||
|
||
api_calls = [(m, u.split('?')[0].replace(API, ''), s) for m, u, s in requests_log if API in u]
|
||
print(f" [stripe] API calls: {json.dumps(api_calls, indent=2)}")
|
||
print(f" [stripe] Result: {result}")
|
||
await browser.close()
|
||
return result
|
||
|
||
|
||
async def test_paypal():
|
||
"""Test: fill form → submit with PayPal → should redirect to paypal.com"""
|
||
print("\n=== TEST: PayPal Direct Checkout ===")
|
||
async with async_playwright() as pw:
|
||
browser = await pw.chromium.launch(headless=True)
|
||
page = await browser.new_page()
|
||
|
||
requests_log = []
|
||
def on_response(resp):
|
||
if "api" in resp.url and resp.request.method in ("POST", "GET"):
|
||
requests_log.append((resp.request.method, resp.url, resp.status))
|
||
page.on("response", on_response)
|
||
|
||
errors = []
|
||
def on_error(err):
|
||
errors.append(str(err))
|
||
page.on("pageerror", on_error)
|
||
|
||
await fill_order_form(page, "paypal")
|
||
|
||
# Select PayPal payment — radio is now on step 5
|
||
await page.click('input[name="payment_method_choice"][value="paypal"]')
|
||
await page.wait_for_timeout(200)
|
||
|
||
# Click Submit Order
|
||
print(" [paypal] Clicking Submit Order...")
|
||
await page.click("#btn-next")
|
||
|
||
# Wait for redirect
|
||
result = "TIMEOUT"
|
||
for i in range(20):
|
||
await page.wait_for_timeout(1000)
|
||
try:
|
||
url = page.url
|
||
if "paypal.com" in url:
|
||
result = "SUCCESS"
|
||
print(f" [paypal] Redirected to PayPal at t+{i+1}s")
|
||
break
|
||
if "checkout.stripe.com" in url:
|
||
result = "WRONG_GATEWAY: redirected to Stripe instead of PayPal"
|
||
break
|
||
status_el = await page.query_selector("#submit-status")
|
||
if status_el:
|
||
txt = (await status_el.inner_text()).strip()
|
||
if txt and txt not in ("", "Placing your order...", "Creating your order...", "Redirecting to payment..."):
|
||
result = f"ERROR: {txt}"
|
||
break
|
||
except Exception:
|
||
await page.wait_for_timeout(2000)
|
||
url = page.url
|
||
if "paypal.com" in url:
|
||
result = "SUCCESS"
|
||
print(f" [paypal] Redirected to PayPal at t+{i+1}s")
|
||
elif "checkout.stripe.com" in url:
|
||
result = "WRONG_GATEWAY: redirected to Stripe instead of PayPal"
|
||
break
|
||
|
||
if errors:
|
||
print(f" [paypal] Page errors: {errors}")
|
||
|
||
api_calls = [(m, u.split('?')[0].replace(API, ''), s) for m, u, s in requests_log if API in u]
|
||
print(f" [paypal] API calls: {json.dumps(api_calls, indent=2)}")
|
||
print(f" [paypal] Result: {result}")
|
||
await browser.close()
|
||
return result
|
||
|
||
|
||
async def main():
|
||
print("E2E Checkout Tests — dev.performancewest.net")
|
||
print("=" * 50)
|
||
|
||
stripe_result = await test_stripe_card()
|
||
paypal_result = await test_paypal()
|
||
|
||
print("\n" + "=" * 50)
|
||
print("RESULTS:")
|
||
print(f" Stripe Card: {stripe_result}")
|
||
print(f" PayPal: {paypal_result}")
|
||
print("=" * 50)
|
||
|
||
if stripe_result == "SUCCESS" and paypal_result == "SUCCESS":
|
||
print("ALL TESTS PASSED")
|
||
else:
|
||
print("SOME TESTS FAILED")
|
||
|
||
asyncio.run(main())
|