This commit is contained in:
Tudor Sitaru
275
tests/test_api.py
Normal file
275
tests/test_api.py
Normal file
@@ -0,0 +1,275 @@
|
||||
#!/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())
|
||||
Reference in New Issue
Block a user