222 lines
8.2 KiB
Python
222 lines
8.2 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
Test script for the Secure Wix Form Handler API
|
||
"""
|
||
|
||
import asyncio
|
||
import aiohttp
|
||
import json
|
||
import os
|
||
import sys
|
||
from datetime import datetime
|
||
|
||
# Add parent directory to path to import from src
|
||
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||
|
||
# API Configuration
|
||
BASE_URL = "http://localhost:8000"
|
||
|
||
# API Keys for testing - replace with your actual keys
|
||
TEST_API_KEY = os.getenv("WIX_API_KEY", "sk_live_your_secure_api_key_here")
|
||
ADMIN_API_KEY = os.getenv("ADMIN_API_KEY", "sk_admin_your_admin_key_here")
|
||
|
||
# Sample Wix form data based on your example
|
||
SAMPLE_WIX_DATA = {
|
||
"formName": "Contact Form",
|
||
"submissions": [],
|
||
"submissionTime": "2024-03-20T10:30:00+00:00",
|
||
"formFieldMask": ["email", "name", "phone"],
|
||
"submissionId": "test-submission-123",
|
||
"contactId": "test-contact-456",
|
||
"submissionsLink": "https://www.wix.app/forms/test-form/submissions",
|
||
"submissionPdf": {
|
||
"url": "https://example.com/submission.pdf",
|
||
"filename": "submission.pdf",
|
||
},
|
||
"formId": "test-form-789",
|
||
"field:email_5139": "test@example.com",
|
||
"field:first_name_abae": "John",
|
||
"field:last_name_d97c": "Doe",
|
||
"field:phone_4c77": "+1234567890",
|
||
"field:anrede": "Herr",
|
||
"field:anzahl_kinder": "2",
|
||
"field:alter_kind_3": "8",
|
||
"field:alter_kind_4": "12",
|
||
"field:long_answer_3524": "This is a long answer field with more details about the inquiry.",
|
||
"contact": {
|
||
"name": {"first": "John", "last": "Doe"},
|
||
"email": "test@example.com",
|
||
"locale": "de",
|
||
"company": "Test Company",
|
||
"birthdate": "1985-05-15",
|
||
"labelKeys": {},
|
||
"contactId": "test-contact-456",
|
||
"address": {
|
||
"street": "Test Street 123",
|
||
"city": "Test City",
|
||
"country": "Germany",
|
||
"postalCode": "12345",
|
||
},
|
||
"jobTitle": "Manager",
|
||
"phone": "+1234567890",
|
||
"createdDate": "2024-03-20T10:00:00.000Z",
|
||
"updatedDate": "2024-03-20T10:30:00.000Z",
|
||
},
|
||
}
|
||
|
||
|
||
async def test_api():
|
||
"""Test the API endpoints with authentication"""
|
||
|
||
headers_with_auth = {
|
||
"Content-Type": "application/json",
|
||
"Authorization": f"Bearer {TEST_API_KEY}",
|
||
}
|
||
|
||
admin_headers = {
|
||
"Content-Type": "application/json",
|
||
"Authorization": f"Bearer {ADMIN_API_KEY}",
|
||
}
|
||
|
||
async with aiohttp.ClientSession() as session:
|
||
# Test health endpoint (no auth required)
|
||
print("1. Testing health endpoint (no auth)...")
|
||
try:
|
||
async with session.get(f"{BASE_URL}/api/health") as response:
|
||
result = await response.json()
|
||
print(f" ✅ Health check: {response.status} - {result.get('status')}")
|
||
except Exception as e:
|
||
print(f" ❌ Health check failed: {e}")
|
||
|
||
# Test root endpoint (no auth required)
|
||
print("\n2. Testing root endpoint (no auth)...")
|
||
try:
|
||
async with session.get(f"{BASE_URL}/api/") as response:
|
||
result = await response.json()
|
||
print(f" ✅ Root: {response.status} - {result.get('message')}")
|
||
except Exception as e:
|
||
print(f" ❌ Root endpoint failed: {e}")
|
||
|
||
# Test webhook endpoint without auth (should fail)
|
||
print("\n3. Testing webhook endpoint WITHOUT auth (should fail)...")
|
||
try:
|
||
async with session.post(
|
||
f"{BASE_URL}/api/webhook/wix-form",
|
||
json=SAMPLE_WIX_DATA,
|
||
headers={"Content-Type": "application/json"},
|
||
) as response:
|
||
result = await response.json()
|
||
if response.status == 401:
|
||
print(
|
||
f" ✅ Correctly rejected: {response.status} - {result.get('detail')}"
|
||
)
|
||
else:
|
||
print(f" ❌ Unexpected response: {response.status} - {result}")
|
||
except Exception as e:
|
||
print(f" ❌ Test failed: {e}")
|
||
|
||
# Test webhook endpoint with valid auth
|
||
print("\n4. Testing webhook endpoint WITH valid auth...")
|
||
try:
|
||
async with session.post(
|
||
f"{BASE_URL}/api/webhook/wix-form",
|
||
json=SAMPLE_WIX_DATA,
|
||
headers=headers_with_auth,
|
||
) as response:
|
||
result = await response.json()
|
||
if response.status == 200:
|
||
print(
|
||
f" ✅ Webhook success: {response.status} - {result.get('status')}"
|
||
)
|
||
else:
|
||
print(f" ❌ Webhook failed: {response.status} - {result}")
|
||
except Exception as e:
|
||
print(f" ❌ Webhook test failed: {e}")
|
||
|
||
# Test test endpoint with auth
|
||
print("\n5. Testing simple test endpoint WITH auth...")
|
||
try:
|
||
async with session.post(
|
||
f"{BASE_URL}/api/webhook/wix-form/test",
|
||
json={"test": "data", "timestamp": datetime.now().isoformat()},
|
||
headers=headers_with_auth,
|
||
) as response:
|
||
result = await response.json()
|
||
if response.status == 200:
|
||
print(
|
||
f" ✅ Test endpoint: {response.status} - {result.get('status')}"
|
||
)
|
||
else:
|
||
print(f" ❌ Test endpoint failed: {response.status} - {result}")
|
||
except Exception as e:
|
||
print(f" ❌ Test endpoint failed: {e}")
|
||
|
||
# Test rate limiting by making multiple rapid requests
|
||
print("\n6. Testing rate limiting (making 5 rapid requests)...")
|
||
rate_limit_test_count = 0
|
||
for i in range(5):
|
||
try:
|
||
async with session.get(f"{BASE_URL}/api/health") as response:
|
||
if response.status == 200:
|
||
rate_limit_test_count += 1
|
||
elif response.status == 429:
|
||
print(f" ✅ Rate limit triggered on request {i + 1}")
|
||
break
|
||
except Exception as e:
|
||
print(f" ❌ Rate limit test failed: {e}")
|
||
break
|
||
|
||
if rate_limit_test_count == 5:
|
||
print(" ℹ️ No rate limit reached (normal for low request volume)")
|
||
|
||
# Test admin endpoint (if admin key is configured)
|
||
print("\n7. Testing admin stats endpoint...")
|
||
try:
|
||
async with session.get(
|
||
f"{BASE_URL}/api/admin/stats", headers=admin_headers
|
||
) as response:
|
||
result = await response.json()
|
||
if response.status == 200:
|
||
print(
|
||
f" ✅ Admin stats: {response.status} - {result.get('status')}"
|
||
)
|
||
elif response.status == 401:
|
||
print(
|
||
f" ⚠️ Admin access denied (API key not configured): {result.get('detail')}"
|
||
)
|
||
else:
|
||
print(f" ❌ Admin endpoint failed: {response.status} - {result}")
|
||
except Exception as e:
|
||
print(f" ❌ Admin test failed: {e}")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
print("🔒 Testing Secure Wix Form Handler API...")
|
||
print("=" * 60)
|
||
print("📍 API URL:", BASE_URL)
|
||
print(
|
||
"🔑 Using API Key:",
|
||
TEST_API_KEY[:20] + "..." if len(TEST_API_KEY) > 20 else TEST_API_KEY,
|
||
)
|
||
print(
|
||
"🔐 Using Admin Key:",
|
||
ADMIN_API_KEY[:20] + "..." if len(ADMIN_API_KEY) > 20 else ADMIN_API_KEY,
|
||
)
|
||
print("=" * 60)
|
||
print("Make sure the API is running with: python3 run_api.py")
|
||
print("-" * 60)
|
||
|
||
try:
|
||
asyncio.run(test_api())
|
||
print("\n" + "=" * 60)
|
||
print("✅ Testing completed!")
|
||
print("\n📋 Quick Setup Reminder:")
|
||
print("1. Set environment variables:")
|
||
print(" export WIX_API_KEY='your_secure_api_key'")
|
||
print(" export ADMIN_API_KEY='your_admin_key'")
|
||
print("2. Configure Wix webhook URL: https://yourdomain.com/webhook/wix-form")
|
||
print("3. Add Authorization header: Bearer your_api_key")
|
||
except Exception as e:
|
||
print(f"\n❌ Error testing API: {e}")
|
||
print("Make sure the API server is running!")
|