251 lines
8.9 KiB
Python
251 lines
8.9 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
Test script for Socket Broker HTTP API
|
||
"""
|
||
|
||
import requests
|
||
import json
|
||
import time
|
||
import os
|
||
from datetime import datetime
|
||
|
||
# Configuration
|
||
API_BASE_URL = "http://localhost:8080"
|
||
TOPIC_NAME = "test-topic"
|
||
|
||
# Get API key from environment variable or use default
|
||
API_KEY = os.getenv('API_KEY', 'default-api-key-change-me')
|
||
|
||
def get_auth_headers():
|
||
"""Get authentication headers for API requests"""
|
||
return {
|
||
'Authorization': f'Bearer {API_KEY}',
|
||
'Content-Type': 'application/json'
|
||
}
|
||
|
||
def test_health():
|
||
"""Test the health endpoint"""
|
||
print("Testing health endpoint...")
|
||
try:
|
||
response = requests.get(f"{API_BASE_URL}/health")
|
||
if response.status_code == 200:
|
||
data = response.json()
|
||
print(f"✅ Health check passed")
|
||
print(f" Status: {data['status']}")
|
||
print(f" Connections: {data['connections']}")
|
||
print(f" Topics: {data['topics']}")
|
||
print(f" Timestamp: {data['timestamp']}")
|
||
print(f" API Key Configured: {data.get('apiKeyConfigured', 'Unknown')}")
|
||
return True
|
||
else:
|
||
print(f"❌ Health check failed: {response.status_code}")
|
||
return False
|
||
except requests.exceptions.RequestException as e:
|
||
print(f"❌ Health check error: {e}")
|
||
return False
|
||
|
||
def test_topics():
|
||
"""Test the topics endpoint"""
|
||
print("\nTesting topics endpoint...")
|
||
try:
|
||
response = requests.get(f"{API_BASE_URL}/topics")
|
||
if response.status_code == 200:
|
||
data = response.json()
|
||
print(f"✅ Topics retrieved successfully")
|
||
if data['topics']:
|
||
for topic in data['topics']:
|
||
print(f" Topic: {topic['topic']}, Subscribers: {topic['subscribers']}")
|
||
else:
|
||
print(" No topics found")
|
||
return True
|
||
else:
|
||
print(f"❌ Topics request failed: {response.status_code}")
|
||
return False
|
||
except requests.exceptions.RequestException as e:
|
||
print(f"❌ Topics request error: {e}")
|
||
return False
|
||
|
||
def test_topic_info(topic):
|
||
"""Test getting info for a specific topic"""
|
||
print(f"\nTesting topic info for '{topic}'...")
|
||
try:
|
||
response = requests.get(f"{API_BASE_URL}/topics/{topic}")
|
||
if response.status_code == 200:
|
||
data = response.json()
|
||
print(f"✅ Topic info retrieved successfully")
|
||
print(f" Topic: {data['topic']}")
|
||
print(f" Subscribers: {data['subscribers']}")
|
||
return True
|
||
elif response.status_code == 404:
|
||
print(f"ℹ️ Topic '{topic}' not found (no subscribers)")
|
||
return True
|
||
else:
|
||
print(f"❌ Topic info request failed: {response.status_code}")
|
||
return False
|
||
except requests.exceptions.RequestException as e:
|
||
print(f"❌ Topic info request error: {e}")
|
||
return False
|
||
|
||
def test_publish_message(topic, message, data=None):
|
||
"""Test publishing a message to a topic"""
|
||
print(f"\nTesting message publish to topic '{topic}'...")
|
||
|
||
payload = {
|
||
"topic": topic,
|
||
"message": message
|
||
}
|
||
|
||
if data:
|
||
payload["data"] = data
|
||
|
||
try:
|
||
response = requests.post(f"{API_BASE_URL}/publish", json=payload, headers=get_auth_headers())
|
||
if response.status_code == 200:
|
||
response_data = response.json()
|
||
print(f"✅ Message published successfully")
|
||
print(f" Subscribers: {response_data['subscribers']}")
|
||
print(f" Message: {response_data['payload']['message']}")
|
||
if response_data['payload']['data']:
|
||
print(f" Data: {json.dumps(response_data['payload']['data'], indent=2)}")
|
||
return True
|
||
elif response.status_code == 204:
|
||
print(f"ℹ️ Message published successfully but no subscribers for topic '{topic}'")
|
||
return True
|
||
elif response.status_code == 401:
|
||
print(f"❌ Authentication failed: {response.json().get('message', 'Unknown error')}")
|
||
return False
|
||
else:
|
||
print(f"❌ Message publish failed: {response.status_code}")
|
||
print(f" Response: {response.text}")
|
||
return False
|
||
except requests.exceptions.RequestException as e:
|
||
print(f"❌ Message publish error: {e}")
|
||
return False
|
||
|
||
def test_publish_with_url_topic(topic, message, data=None):
|
||
"""Test publishing using the URL-based endpoint"""
|
||
print(f"\nTesting message publish to topic '{topic}' (URL endpoint)...")
|
||
|
||
payload = {}
|
||
if message:
|
||
payload["message"] = message
|
||
if data:
|
||
payload["data"] = data
|
||
|
||
try:
|
||
response = requests.post(f"{API_BASE_URL}/publish/{topic}", json=payload, headers=get_auth_headers())
|
||
if response.status_code == 200:
|
||
response_data = response.json()
|
||
print(f"✅ Message published successfully (URL endpoint)")
|
||
print(f" Subscribers: {response_data['subscribers']}")
|
||
print(f" Message: {response_data['payload']['message']}")
|
||
if response_data['payload']['data']:
|
||
print(f" Data: {json.dumps(response_data['payload']['data'], indent=2)}")
|
||
return True
|
||
elif response.status_code == 204:
|
||
print(f"ℹ️ Message published successfully but no subscribers for topic '{topic}'")
|
||
return True
|
||
elif response.status_code == 401:
|
||
print(f"❌ Authentication failed: {response.json().get('message', 'Unknown error')}")
|
||
return False
|
||
else:
|
||
print(f"❌ Message publish failed (URL endpoint): {response.status_code}")
|
||
print(f" Response: {response.text}")
|
||
return False
|
||
except requests.exceptions.RequestException as e:
|
||
print(f"❌ Message publish error (URL endpoint): {e}")
|
||
return False
|
||
|
||
def test_authentication_errors():
|
||
"""Test authentication error cases"""
|
||
print(f"\n🧪 Testing authentication error cases...")
|
||
|
||
# Test without API key
|
||
print("Testing request without API key...")
|
||
try:
|
||
response = requests.post(f"{API_BASE_URL}/publish", json={
|
||
"topic": "test-topic",
|
||
"message": "This should fail"
|
||
})
|
||
if response.status_code == 401:
|
||
print("✅ Correctly rejected request without API key")
|
||
else:
|
||
print(f"❌ Expected 401, got {response.status_code}")
|
||
except requests.exceptions.RequestException as e:
|
||
print(f"❌ Request error: {e}")
|
||
|
||
# Test with wrong API key
|
||
print("Testing request with wrong API key...")
|
||
try:
|
||
response = requests.post(f"{API_BASE_URL}/publish", json={
|
||
"topic": "test-topic",
|
||
"message": "This should fail"
|
||
}, headers={'Authorization': 'Bearer wrong-key'})
|
||
if response.status_code == 401:
|
||
print("✅ Correctly rejected request with wrong API key")
|
||
else:
|
||
print(f"❌ Expected 401, got {response.status_code}")
|
||
except requests.exceptions.RequestException as e:
|
||
print(f"❌ Request error: {e}")
|
||
|
||
def run_demo():
|
||
"""Run a complete demo"""
|
||
print("🚀 Socket Broker HTTP API Test")
|
||
print("=" * 50)
|
||
print(f"🔑 Using API Key: {API_KEY[:10]}{'...' if len(API_KEY) > 10 else ''}")
|
||
print(f"🌐 API Base URL: {API_BASE_URL}")
|
||
print("=" * 50)
|
||
|
||
# Test basic endpoints
|
||
if not test_health():
|
||
print("❌ Cannot proceed - health check failed")
|
||
return
|
||
|
||
test_topics()
|
||
|
||
# Test topic info
|
||
test_topic_info(TOPIC_NAME)
|
||
|
||
# Test publishing messages
|
||
print(f"\n📤 Publishing test messages...")
|
||
|
||
# Simple message
|
||
test_publish_message(TOPIC_NAME, "Hello from Python test script!")
|
||
|
||
# Message with data
|
||
test_data = {
|
||
"source": "python-test",
|
||
"timestamp": datetime.now().isoformat(),
|
||
"user": "test-user",
|
||
"action": "demo"
|
||
}
|
||
test_publish_message(TOPIC_NAME, "Test message with data", test_data)
|
||
|
||
# Test URL-based endpoint
|
||
test_publish_with_url_topic(TOPIC_NAME, "Message via URL endpoint", {"method": "url"})
|
||
|
||
# Test error cases
|
||
print(f"\n🧪 Testing error cases...")
|
||
|
||
# Empty topic
|
||
test_publish_message("", "This should fail")
|
||
|
||
# No message or data
|
||
test_publish_message(TOPIC_NAME, None)
|
||
|
||
# Test authentication errors
|
||
test_authentication_errors()
|
||
|
||
print(f"\n✅ Demo completed!")
|
||
print(f"\n💡 To see real-time messages, open test_client.html in a browser")
|
||
print(f" and subscribe to topic '{TOPIC_NAME}'")
|
||
print(f"\n🔐 API Key is required for publishing messages")
|
||
print(f" Set API_KEY environment variable or use the default key")
|
||
|
||
if __name__ == "__main__":
|
||
try:
|
||
run_demo()
|
||
except KeyboardInterrupt:
|
||
print("\n\n⏹️ Demo interrupted by user")
|
||
except Exception as e:
|
||
print(f"\n❌ Demo failed with error: {e}") |