File size: 3,551 Bytes
b66240d |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 |
#!/usr/bin/env python3
"""Test script for HuggingFace Crypto Data Engine API"""
import asyncio
import httpx
import json
from typing import Optional
BASE_URL = "http://localhost:8000"
async def test_endpoint(client: httpx.AsyncClient, name: str, url: str, params: Optional[dict] = None) -> bool:
"""Test a single endpoint"""
print(f"\n{'='*60}")
print(f"Testing: {name}")
print(f"URL: {url}")
if params:
print(f"Params: {json.dumps(params, indent=2)}")
try:
response = await client.get(url, params=params, timeout=30.0)
response.raise_for_status()
data = response.json()
print(f"✅ SUCCESS - Status: {response.status_code}")
print(f"Response preview:")
print(json.dumps(data, indent=2)[:500] + "...")
return True
except Exception as e:
print(f"❌ FAILED - Error: {e}")
return False
async def main():
"""Run all API tests"""
print("🚀 HuggingFace Crypto Data Engine - API Test Suite")
print(f"Base URL: {BASE_URL}")
results = []
async with httpx.AsyncClient(base_url=BASE_URL) as client:
# Test 1: Root endpoint
results.append(await test_endpoint(
client,
"Root Endpoint",
"/"
))
# Test 2: Health check
results.append(await test_endpoint(
client,
"Health Check",
"/api/health"
))
# Test 3: OHLCV - BTC 1h
results.append(await test_endpoint(
client,
"OHLCV Data (BTC 1h)",
"/api/ohlcv",
{"symbol": "BTC", "interval": "1h", "limit": 10}
))
# Test 4: OHLCV - ETH 5m
results.append(await test_endpoint(
client,
"OHLCV Data (ETH 5m)",
"/api/ohlcv",
{"symbol": "ETH", "interval": "5m", "limit": 20}
))
# Test 5: Prices - Single symbol
results.append(await test_endpoint(
client,
"Prices (BTC)",
"/api/prices",
{"symbols": "BTC"}
))
# Test 6: Prices - Multiple symbols
results.append(await test_endpoint(
client,
"Prices (BTC, ETH, SOL)",
"/api/prices",
{"symbols": "BTC,ETH,SOL"}
))
# Test 7: Prices - All symbols
results.append(await test_endpoint(
client,
"Prices (All Symbols)",
"/api/prices"
))
# Test 8: Sentiment
results.append(await test_endpoint(
client,
"Market Sentiment",
"/api/sentiment"
))
# Test 9: Market Overview
results.append(await test_endpoint(
client,
"Market Overview",
"/api/market/overview"
))
# Test 10: Cache Stats
results.append(await test_endpoint(
client,
"Cache Statistics",
"/api/cache/stats"
))
# Summary
print(f"\n{'='*60}")
print("📊 TEST SUMMARY")
print(f"{'='*60}")
print(f"Total Tests: {len(results)}")
print(f"Passed: {sum(results)}")
print(f"Failed: {len(results) - sum(results)}")
print(f"Success Rate: {(sum(results) / len(results) * 100):.1f}%")
if all(results):
print("\n✅ All tests passed!")
return 0
else:
print("\n❌ Some tests failed!")
return 1
if __name__ == "__main__":
exit_code = asyncio.run(main())
exit(exit_code)
|