275 lines
10 KiB
Python
275 lines
10 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
API Test Script
|
|
|
|
This script helps test your API endpoints before running the full image downloader.
|
|
It will check if the list endpoint returns valid data and if the download endpoint
|
|
is accessible.
|
|
|
|
Usage:
|
|
python test_api.py --api-url <base_url> --list-endpoint <endpoint> --download-endpoint <endpoint>
|
|
"""
|
|
|
|
import argparse
|
|
import asyncio
|
|
import aiohttp
|
|
import json
|
|
from urllib.parse import urljoin
|
|
from typing import Dict, Any
|
|
|
|
|
|
class APITester:
|
|
def __init__(self, api_url: str, list_endpoint: str, download_endpoint: str, timeout: int = 30, api_key: str = None):
|
|
self.api_url = api_url.rstrip('/')
|
|
self.list_endpoint = list_endpoint.lstrip('/')
|
|
self.download_endpoint = download_endpoint.lstrip('/')
|
|
self.timeout = timeout
|
|
self.api_key = api_key
|
|
|
|
async def test_list_endpoint(self, session: aiohttp.ClientSession) -> Dict[str, Any]:
|
|
"""Test the list endpoint and return information about the response."""
|
|
url = urljoin(self.api_url, self.list_endpoint)
|
|
print(f"Testing list endpoint: {url}")
|
|
|
|
try:
|
|
headers = {}
|
|
if self.api_key:
|
|
headers['x-api-key'] = self.api_key
|
|
|
|
async with session.get(url, headers=headers, timeout=self.timeout) as response:
|
|
print(f"Status Code: {response.status}")
|
|
print(f"Content-Type: {response.headers.get('content-type', 'Not specified')}")
|
|
|
|
if response.status == 200:
|
|
data = await response.json()
|
|
print(f"Response type: {type(data)}")
|
|
|
|
# Analyze the response structure
|
|
if isinstance(data, list):
|
|
print(f"Found {len(data)} assets in array")
|
|
if data:
|
|
print(f"First asset keys: {list(data[0].keys())}")
|
|
elif isinstance(data, dict):
|
|
print(f"Response keys: {list(data.keys())}")
|
|
|
|
# Check common patterns
|
|
for key in ['data', 'results', 'items', 'assets', 'images']:
|
|
if key in data and isinstance(data[key], list):
|
|
print(f"Found {len(data[key])} assets in '{key}' field")
|
|
if data[key]:
|
|
print(f"First asset keys: {list(data[key][0].keys())}")
|
|
break
|
|
else:
|
|
print("No recognized array field found in response")
|
|
else:
|
|
print(f"Unexpected response format: {type(data)}")
|
|
|
|
return {
|
|
'success': True,
|
|
'data': data,
|
|
'url': url
|
|
}
|
|
else:
|
|
print(f"Error: HTTP {response.status_code}")
|
|
return {
|
|
'success': False,
|
|
'error': f"HTTP {response.status_code}",
|
|
'url': url
|
|
}
|
|
|
|
except Exception as e:
|
|
print(f"Error testing list endpoint: {e}")
|
|
return {
|
|
'success': False,
|
|
'error': str(e),
|
|
'url': url
|
|
}
|
|
|
|
async def test_download_endpoint(self, session: aiohttp.ClientSession, asset_id: str) -> Dict[str, Any]:
|
|
"""Test the download endpoint with a sample asset ID."""
|
|
url = urljoin(self.api_url, f"{self.download_endpoint}/{asset_id}")
|
|
print(f"\nTesting download endpoint: {url}")
|
|
|
|
try:
|
|
headers = {}
|
|
if self.api_key:
|
|
headers['x-api-key'] = self.api_key
|
|
|
|
async with session.get(url, headers=headers, timeout=self.timeout) as response:
|
|
print(f"Status Code: {response.status}")
|
|
print(f"Content-Type: {response.headers.get('content-type', 'Not specified')}")
|
|
print(f"Content-Length: {response.headers.get('content-length', 'Not specified')}")
|
|
|
|
if response.status == 200:
|
|
content_type = response.headers.get('content-type', '')
|
|
if content_type.startswith('image/'):
|
|
print("✓ Download endpoint returns image content")
|
|
return {
|
|
'success': True,
|
|
'url': url,
|
|
'content_type': content_type
|
|
}
|
|
else:
|
|
print(f"⚠ Warning: Content type is not an image: {content_type}")
|
|
return {
|
|
'success': True,
|
|
'url': url,
|
|
'content_type': content_type,
|
|
'warning': 'Not an image'
|
|
}
|
|
else:
|
|
print(f"Error: HTTP {response.status}")
|
|
return {
|
|
'success': False,
|
|
'error': f"HTTP {response.status}",
|
|
'url': url
|
|
}
|
|
|
|
except Exception as e:
|
|
print(f"Error testing download endpoint: {e}")
|
|
return {
|
|
'success': False,
|
|
'error': str(e),
|
|
'url': url
|
|
}
|
|
|
|
async def run_tests(self):
|
|
"""Run all API tests."""
|
|
print("=" * 60)
|
|
print("API Endpoint Test")
|
|
print("=" * 60)
|
|
|
|
timeout = aiohttp.ClientTimeout(total=self.timeout)
|
|
async with aiohttp.ClientSession(timeout=timeout) as session:
|
|
# Test list endpoint
|
|
list_result = await self.test_list_endpoint(session)
|
|
|
|
if list_result['success']:
|
|
# Try to test download endpoint with first asset
|
|
data = list_result['data']
|
|
asset_id = None
|
|
|
|
# Find an asset ID to test with
|
|
if isinstance(data, list) and data:
|
|
asset = data[0]
|
|
for key in ['id', 'asset_id', 'image_id', 'file_id', 'uuid', 'key']:
|
|
if key in asset:
|
|
asset_id = asset[key]
|
|
break
|
|
elif isinstance(data, dict):
|
|
for key in ['data', 'results', 'items', 'assets', 'images']:
|
|
if key in data and isinstance(data[key], list) and data[key]:
|
|
asset = data[key][0]
|
|
for id_key in ['id', 'asset_id', 'image_id', 'file_id', 'uuid', 'key']:
|
|
if id_key in asset:
|
|
asset_id = asset[id_key]
|
|
break
|
|
if asset_id:
|
|
break
|
|
|
|
if asset_id:
|
|
print(f"\nUsing asset ID '{asset_id}' for download test")
|
|
download_result = await self.test_download_endpoint(session, asset_id)
|
|
else:
|
|
print("\n⚠ Could not find an asset ID to test download endpoint")
|
|
print("You may need to manually test the download endpoint")
|
|
|
|
# Print summary
|
|
print("\n" + "=" * 60)
|
|
print("TEST SUMMARY")
|
|
print("=" * 60)
|
|
|
|
if list_result['success']:
|
|
print("✓ List endpoint: Working")
|
|
else:
|
|
print("✗ List endpoint: Failed")
|
|
print(f" Error: {list_result['error']}")
|
|
|
|
if 'download_result' in locals():
|
|
if download_result['success']:
|
|
print("✓ Download endpoint: Working")
|
|
if 'warning' in download_result:
|
|
print(f" Warning: {download_result['warning']}")
|
|
else:
|
|
print("✗ Download endpoint: Failed")
|
|
print(f" Error: {download_result['error']}")
|
|
|
|
print("\nRecommendations:")
|
|
if list_result['success']:
|
|
print("- List endpoint is working correctly")
|
|
print("- You can proceed with the image downloader")
|
|
else:
|
|
print("- Check your API URL and list endpoint")
|
|
print("- Verify the API is accessible")
|
|
print("- Check if authentication is required")
|
|
|
|
if 'download_result' in locals() and not download_result['success']:
|
|
print("- Check your download endpoint format")
|
|
print("- Verify asset IDs are being passed correctly")
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(
|
|
description="Test API endpoints for image downloader",
|
|
formatter_class=argparse.RawDescriptionHelpFormatter,
|
|
epilog="""
|
|
Examples:
|
|
python test_api.py --api-url "https://api.example.com" \\
|
|
--list-endpoint "/assets" \\
|
|
--download-endpoint "/download"
|
|
"""
|
|
)
|
|
|
|
parser.add_argument(
|
|
'--api-url',
|
|
required=True,
|
|
help='Base URL of the API (e.g., https://api.example.com)'
|
|
)
|
|
|
|
parser.add_argument(
|
|
'--list-endpoint',
|
|
required=True,
|
|
help='Endpoint to get the list of assets (e.g., /assets or /images)'
|
|
)
|
|
|
|
parser.add_argument(
|
|
'--download-endpoint',
|
|
required=True,
|
|
help='Endpoint to download individual assets (e.g., /download or /assets)'
|
|
)
|
|
|
|
parser.add_argument(
|
|
'--timeout',
|
|
type=int,
|
|
default=30,
|
|
help='Request timeout in seconds (default: 30)'
|
|
)
|
|
|
|
parser.add_argument(
|
|
'--api-key',
|
|
help='API key for authentication (x-api-key header)'
|
|
)
|
|
|
|
args = parser.parse_args()
|
|
|
|
tester = APITester(
|
|
api_url=args.api_url,
|
|
list_endpoint=args.list_endpoint,
|
|
download_endpoint=args.download_endpoint,
|
|
timeout=args.timeout,
|
|
api_key=args.api_key
|
|
)
|
|
|
|
try:
|
|
asyncio.run(tester.run_tests())
|
|
except KeyboardInterrupt:
|
|
print("\nTest interrupted by user")
|
|
except Exception as e:
|
|
print(f"Error: {e}")
|
|
return 1
|
|
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
exit(main()) |