meta_api_grabber/test_rate_limiter.py

#!/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 (first account)
    print("\n--- Test 2: X-Ad-Account-Usage Header (Account 1) ---")
    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, account_id='act_123456789')
    limiter.print_stats()

    # Test 2b: X-Ad-Account-Usage header (second account)
    print("\n--- Test 2b: X-Ad-Account-Usage Header (Account 2) ---")
    response2b = MockResponse(headers={
        'x-ad-account-usage': json.dumps({
            'acc_id_util_pct': 45.2,
            'reset_time_duration': 80,
            'ads_api_access_tier': 'standard_access'
        })
    })
    limiter.update_usage(response2b, account_id='act_987654321')
    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())