Add subcription feature 3
This commit is contained in:
130
gateway/test_webhook_fix.py
Normal file
130
gateway/test_webhook_fix.py
Normal file
@@ -0,0 +1,130 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Test script to verify that webhook routes work correctly after the fix.
|
||||
"""
|
||||
|
||||
import sys
|
||||
import os
|
||||
import re
|
||||
|
||||
# Add the gateway directory to the path
|
||||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||||
|
||||
def test_subscription_middleware_skip_patterns():
|
||||
"""Test that webhook routes are in the skip patterns."""
|
||||
|
||||
# Read the subscription middleware file
|
||||
with open('app/middleware/subscription.py', 'r') as f:
|
||||
content = f.read()
|
||||
|
||||
# Check if webhook routes are in the skip patterns
|
||||
webhook_pattern = r'/api/v1/webhooks/.*'
|
||||
if webhook_pattern in content:
|
||||
print("✅ Webhook routes found in SubscriptionMiddleware skip patterns")
|
||||
return True
|
||||
else:
|
||||
print("❌ Webhook routes NOT found in SubscriptionMiddleware skip patterns")
|
||||
return False
|
||||
|
||||
def test_read_only_middleware_whitelist():
|
||||
"""Test that webhook routes are in the whitelist."""
|
||||
|
||||
# Read the read-only middleware file
|
||||
with open('app/middleware/read_only_mode.py', 'r') as f:
|
||||
content = f.read()
|
||||
|
||||
# Check if webhook routes are in the whitelist
|
||||
webhook_pattern = r'^/api/v1/webhooks/.*'
|
||||
if webhook_pattern in content:
|
||||
print("✅ Webhook routes found in ReadOnlyModeMiddleware whitelist")
|
||||
return True
|
||||
else:
|
||||
print("❌ Webhook routes NOT found in ReadOnlyModeMiddleware whitelist")
|
||||
return False
|
||||
|
||||
def test_webhook_route_definitions():
|
||||
"""Test that webhook routes are correctly defined."""
|
||||
|
||||
# Read the webhooks file
|
||||
with open('app/routes/webhooks.py', 'r') as f:
|
||||
content = f.read()
|
||||
|
||||
# Check if webhook routes are defined with full paths
|
||||
stripe_route = '@router.post("/api/v1/webhooks/stripe")'
|
||||
generic_route = '@router.post("/api/v1/webhooks/generic")'
|
||||
|
||||
if stripe_route in content and generic_route in content:
|
||||
print("✅ Webhook routes correctly defined with full paths")
|
||||
return True
|
||||
else:
|
||||
print("❌ Webhook routes NOT correctly defined")
|
||||
return False
|
||||
|
||||
def test_webhook_router_inclusion():
|
||||
"""Test that webhook router is included with correct prefix."""
|
||||
|
||||
# Read the main file
|
||||
with open('app/main.py', 'r') as f:
|
||||
content = f.read()
|
||||
|
||||
# Check if webhook router is included with prefix=""
|
||||
webhook_inclusion = "app.include_router(webhooks.router, prefix=\"\", tags=[\"webhooks\"])"
|
||||
|
||||
if webhook_inclusion in content:
|
||||
print("✅ Webhook router correctly included with prefix=''")
|
||||
return True
|
||||
else:
|
||||
print("❌ Webhook router NOT correctly included")
|
||||
return False
|
||||
|
||||
def test_auth_middleware_public_routes():
|
||||
"""Test that webhook routes are in the public routes."""
|
||||
|
||||
# Read the auth middleware file
|
||||
with open('app/middleware/auth.py', 'r') as f:
|
||||
content = f.read()
|
||||
|
||||
# Check if webhook routes are in the public routes
|
||||
stripe_public = '/api/v1/webhooks/stripe'
|
||||
generic_public = '/api/v1/webhooks/generic'
|
||||
|
||||
if stripe_public in content and generic_public in content:
|
||||
print("✅ Webhook routes found in AuthMiddleware public routes")
|
||||
return True
|
||||
else:
|
||||
print("❌ Webhook routes NOT found in AuthMiddleware public routes")
|
||||
return False
|
||||
|
||||
def main():
|
||||
"""Run all tests."""
|
||||
print("Testing webhook route configuration...")
|
||||
print()
|
||||
|
||||
tests = [
|
||||
test_webhook_route_definitions,
|
||||
test_webhook_router_inclusion,
|
||||
test_auth_middleware_public_routes,
|
||||
test_subscription_middleware_skip_patterns,
|
||||
test_read_only_middleware_whitelist,
|
||||
]
|
||||
|
||||
results = []
|
||||
for test in tests:
|
||||
try:
|
||||
result = test()
|
||||
results.append(result)
|
||||
except Exception as e:
|
||||
print(f"❌ Error running {test.__name__}: {e}")
|
||||
results.append(False)
|
||||
print()
|
||||
|
||||
if all(results):
|
||||
print("🎉 All tests passed! Webhook routes should work correctly.")
|
||||
return True
|
||||
else:
|
||||
print("❌ Some tests failed. Webhook routes may not work correctly.")
|
||||
return False
|
||||
|
||||
if __name__ == "__main__":
|
||||
success = main()
|
||||
sys.exit(0 if success else 1)
|
||||
Reference in New Issue
Block a user