157 lines
4.7 KiB
Python
157 lines
4.7 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Test script for the enhanced Meta API rate limiter.
|
|
|
|
This demonstrates the rate limiter's ability to parse and monitor
|
|
all Meta API rate limit headers.
|
|
"""
|
|
|
|
import asyncio
|
|
import json
|
|
import logging
|
|
from dataclasses import dataclass
|
|
from typing import Dict, Optional
|
|
|
|
from src.meta_api_grabber.rate_limiter import MetaRateLimiter
|
|
|
|
|
|
# Configure logging to show debug messages
|
|
logging.basicConfig(
|
|
level=logging.DEBUG,
|
|
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class MockResponse:
|
|
"""Mock API response with rate limit headers."""
|
|
headers: Dict[str, str]
|
|
|
|
|
|
async def test_rate_limiter():
|
|
"""Test the rate limiter with various header scenarios."""
|
|
|
|
# Initialize rate limiter
|
|
limiter = MetaRateLimiter(
|
|
base_delay=1.0,
|
|
throttle_threshold=75.0,
|
|
max_retry_delay=60.0,
|
|
)
|
|
|
|
print("\n" + "="*70)
|
|
print("TESTING ENHANCED META API RATE LIMITER")
|
|
print("="*70 + "\n")
|
|
|
|
# Test 1: X-App-Usage header
|
|
print("\n--- Test 1: X-App-Usage Header ---")
|
|
response1 = MockResponse(headers={
|
|
'x-app-usage': json.dumps({
|
|
'call_count': 45,
|
|
'total_time': 30,
|
|
'total_cputime': 35
|
|
})
|
|
})
|
|
limiter.update_usage(response1)
|
|
limiter.print_stats()
|
|
|
|
# Test 2: X-Ad-Account-Usage header
|
|
print("\n--- Test 2: X-Ad-Account-Usage Header ---")
|
|
response2 = MockResponse(headers={
|
|
'x-ad-account-usage': json.dumps({
|
|
'acc_id_util_pct': 78.5,
|
|
'reset_time_duration': 120,
|
|
'ads_api_access_tier': 'development_access'
|
|
})
|
|
})
|
|
limiter.update_usage(response2)
|
|
limiter.print_stats()
|
|
|
|
# Test 3: X-Business-Use-Case-Usage header
|
|
print("\n--- Test 3: X-Business-Use-Case-Usage Header ---")
|
|
response3 = MockResponse(headers={
|
|
'x-business-use-case-usage': json.dumps({
|
|
'66782684': [{
|
|
'type': 'ads_management',
|
|
'call_count': 85,
|
|
'total_cputime': 40,
|
|
'total_time': 35,
|
|
'estimated_time_to_regain_access': 5,
|
|
'ads_api_access_tier': 'development_access'
|
|
}],
|
|
'10153848260347724': [{
|
|
'type': 'ads_insights',
|
|
'call_count': 92,
|
|
'total_cputime': 50,
|
|
'total_time': 45,
|
|
'estimated_time_to_regain_access': 8,
|
|
'ads_api_access_tier': 'development_access'
|
|
}]
|
|
})
|
|
})
|
|
limiter.update_usage(response3)
|
|
limiter.print_stats()
|
|
|
|
# Test 4: Legacy x-fb-ads-insights-throttle header
|
|
print("\n--- Test 4: Legacy Header ---")
|
|
response4 = MockResponse(headers={
|
|
'x-fb-ads-insights-throttle': json.dumps({
|
|
'app_id_util_pct': 65.0,
|
|
'acc_id_util_pct': 70.5
|
|
})
|
|
})
|
|
limiter.update_usage(response4)
|
|
limiter.print_stats()
|
|
|
|
# Test 5: All headers combined (high usage scenario)
|
|
print("\n--- Test 5: High Usage Scenario (All Headers) ---")
|
|
response5 = MockResponse(headers={
|
|
'x-app-usage': json.dumps({
|
|
'call_count': 95,
|
|
'total_time': 88,
|
|
'total_cputime': 90
|
|
}),
|
|
'x-ad-account-usage': json.dumps({
|
|
'acc_id_util_pct': 97.5,
|
|
'reset_time_duration': 300,
|
|
'ads_api_access_tier': 'standard_access'
|
|
}),
|
|
'x-business-use-case-usage': json.dumps({
|
|
'12345678': [{
|
|
'type': 'ads_insights',
|
|
'call_count': 98,
|
|
'total_cputime': 95,
|
|
'total_time': 92,
|
|
'estimated_time_to_regain_access': 15,
|
|
'ads_api_access_tier': 'standard_access'
|
|
}]
|
|
}),
|
|
'x-fb-ads-insights-throttle': json.dumps({
|
|
'app_id_util_pct': 93.0,
|
|
'acc_id_util_pct': 96.0
|
|
})
|
|
})
|
|
limiter.update_usage(response5)
|
|
limiter.print_stats()
|
|
|
|
# Test throttling behavior
|
|
print("\n--- Test 6: Throttling Behavior ---")
|
|
print(f"Should throttle: {limiter.should_throttle()}")
|
|
print(f"Max usage: {limiter.get_max_usage_pct():.1f}%")
|
|
print(f"Throttle delay: {limiter.get_throttle_delay():.1f}s")
|
|
print(f"Estimated time to regain access: {limiter.estimated_time_to_regain_access} min")
|
|
print(f"Reset time duration: {limiter.reset_time_duration}s")
|
|
|
|
# Test 7: Empty/missing headers
|
|
print("\n--- Test 7: Missing Headers ---")
|
|
response6 = MockResponse(headers={})
|
|
limiter.update_usage(response6)
|
|
limiter.print_stats()
|
|
|
|
print("\n" + "="*70)
|
|
print("ALL TESTS COMPLETED")
|
|
print("="*70 + "\n")
|
|
|
|
|
|
if __name__ == '__main__':
|
|
asyncio.run(test_rate_limiter())
|