103 lines
3.6 KiB
Python
103 lines
3.6 KiB
Python
"""
|
|
Test script for webhook endpoint
|
|
Sends the Square inventory.count.updated webhook payload
|
|
"""
|
|
|
|
import asyncio
|
|
import aiohttp
|
|
import json
|
|
import ssl
|
|
import sys
|
|
|
|
|
|
async def test_webhook(verify_ssl=True):
|
|
url = "https://studspider.sh:8443/hook"
|
|
|
|
# Your Square webhook payload
|
|
payload = {
|
|
"merchant_id": "6SSW7HV8K2ST5",
|
|
"type": "inventory.count.updated",
|
|
"event_id": "df5f3813-a913-45a1-94e9-fdc3f7d5e3b6",
|
|
"created_at": "2019-10-29T18:38:45.455006797Z",
|
|
"data": {
|
|
"type": "inventory_counts",
|
|
"id": "84e4ac73-d605-4dbd-a9e5-ffff794ddb9d",
|
|
"object": {
|
|
"inventory_counts": [
|
|
{
|
|
"calculated_at": "2019-10-29T18:38:45.10296Z",
|
|
"catalog_object_id": "FGQ5JJWT2PYTHF35CKZ2DSKP",
|
|
"catalog_object_type": "ITEM_VARIATION",
|
|
"location_id": "YYQR03DGCTXA4",
|
|
"quantity": "10",
|
|
"state": "IN_STOCK",
|
|
}
|
|
]
|
|
},
|
|
},
|
|
}
|
|
|
|
print("Sending webhook test payload...")
|
|
print(f"URL: {url}")
|
|
print(
|
|
f"SSL Verification: {'Enabled' if verify_ssl else 'Disabled (accepting self-signed certs)'}"
|
|
)
|
|
print(json.dumps(payload, indent=2))
|
|
print("\n" + "=" * 60 + "\n")
|
|
|
|
# Create SSL context that allows self-signed certificates if verify_ssl is False
|
|
ssl_context = None
|
|
if not verify_ssl:
|
|
ssl_context = ssl.create_default_context()
|
|
ssl_context.check_hostname = False
|
|
ssl_context.verify_mode = ssl.CERT_NONE
|
|
|
|
try:
|
|
connector = aiohttp.TCPConnector(ssl=ssl_context)
|
|
async with aiohttp.ClientSession(connector=connector) as session:
|
|
async with session.post(url, json=payload) as resp:
|
|
print(f"Status: {resp.status}")
|
|
print(f"Response headers: {dict(resp.headers)}")
|
|
|
|
response_text = await resp.text()
|
|
print(f"\nResponse body:")
|
|
|
|
try:
|
|
response_json = json.loads(response_text)
|
|
print(json.dumps(response_json, indent=2))
|
|
except json.JSONDecodeError:
|
|
print(response_text)
|
|
|
|
except aiohttp.ClientConnectorError as e:
|
|
print(f"Connection error: {e}")
|
|
print("\nTroubleshooting:")
|
|
print("1. Make sure the server is running")
|
|
print("2. Check if the hostname and port are correct")
|
|
print("3. For self-signed certificates, the script now accepts them by default")
|
|
print("\nServer start command:")
|
|
print(
|
|
" python src/test_server.py --ssl-cert cert.pem --ssl-key key.pem --hook-port 8443 --hook-ssl-only"
|
|
)
|
|
except aiohttp.ServerTimeoutError as e:
|
|
print(f"Timeout error: {e}")
|
|
print("The server took too long to respond")
|
|
except Exception as e:
|
|
print(f"Error: {type(e).__name__}: {e}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
# Check command line arguments
|
|
verify_ssl = True
|
|
if len(sys.argv) > 1 and sys.argv[1] in ["--no-verify", "--insecure", "-k"]:
|
|
verify_ssl = False
|
|
print("⚠️ SSL verification disabled - accepting self-signed certificates\n")
|
|
elif len(sys.argv) > 1:
|
|
print("Usage: python src/test_webhook.py [--no-verify]")
|
|
print(
|
|
" --no-verify: Disable SSL certificate verification (for self-signed certs)"
|
|
)
|
|
sys.exit(1)
|
|
|
|
# Default to no verification since we're likely using self-signed certs
|
|
asyncio.run(test_webhook(verify_ssl=False))
|