test_api.py

import httpx
import asyncio
import json
import time

# Define the API endpoint
url = "https://unicone-studio-search.hf.space/api/search"

# Define the search queries you want to test
test_queries = [
    {"query": "Here She Is"},
    {"query": "my py"},
    {"query": "ahoh girl"},
    {"query": "yarian"},
    {"query": "grand blue"},
    {"query": "Summertime"},
    {"query": "yariyan 203"},
    {"query": "Wahlberg Baigan and the Magic of Darkness"},  # Intentional typo to test fuzzy matching

    # Additional misspelled and errored queries
    {"query": "my pi"},
    {"query": "aho gir"},
    {"query": "yaariyan"},
    {"query": "grand ble"},
    {"query": "non exsitent title"},
    {"query": "yariyan 20"},
    {"query": "My spay"},

    # Misspellings with slight variations
    {"query": "yariiyn"},
    {"query": "goand blue"},
    {"query": "grnd blue"},
    {"query": "aho grl"},
    {"query": "yaariyan 2023"},
    {"query": "my pye"},

    # Variations in spacing and formatting
    {"query": "my   py"},
    {"query": "aho  girl"},
    {"query": " yarian"},
    {"query": "grandblue"},
    {"query": "   non-existent title"},
    {"query": "yariyan203"},
    {"query": "My   spey"},

    # More complex misspellings and typos
    {"query": "ma py"},
    {"query": "aho girrl"},
    {"query": "yarran"},
    {"query": "grnd blu"},
    {"query": "nonxistent title"},
    {"query": "yariyan 2-3"},
    {"query": "Mey spey"},

    # Phonetic errors and similar sounding words
    {"query": "my pie"},
    {"query": "aho girl"},
    {"query": "yariyan"},
    {"query": "grand bluee"},
    {"query": "noe existent title"},
    {"query": "yarryan"},
    {"query": "Mai spay"},

    # Common typographical errors
    {"query": "myy py"},
    {"query": "aho giirl"},
    {"query": "yaaryan"},
    {"query": "grand ble"},
    {"query": "non existant title"},
    {"query": "yarian 203"},
    {"query": "My speay"},

    # Random and varied input
    {"query": "py my"},
    {"query": "giri aho"},
    {"query": "yariiian"},
    {"query": "blue grand"},
    {"query": "existant non title"},
    {"query": "203 yariyan"},
    {"query": "spey my"},

    # Edge cases
    {"query": " "},
    {"query": "!@#$%^&*()"},
    {"query": "123456"},
    {"query": "aaaaaaa"},
    {"query": "zzzzzzz"},
    {"query": "   "},
    {"query": " "},
]

async def fetch_results(client, query):
    start_time = time.time()  # Start timing
    try:
        response = await client.post(url, json=query)
        response.raise_for_status()  # Raise an error for HTTP error responses
        results = response.json()
        duration = time.time() - start_time  # End timing
        return query['query'], results, duration
    except httpx.HTTPStatusError as e:
        duration = time.time() - start_time
        return query['query'], f"Failed with status code {e.response.status_code}: {e.response.text}", duration
    except Exception as e:
        duration = time.time() - start_time
        return query['query'], str(e), duration

async def stress_test_api(num_times):
    async with httpx.AsyncClient() as client:
        all_results = []
        total_start_time = time.time()  # Start timing for stress test
        
        for _ in range(num_times):
            queries = test_queries[:]  # Create a copy of the test queries
            tasks = [fetch_results(client, query) for query in queries]
            results = await asyncio.gather(*tasks)
            all_results.extend(results)  # Collect all results
            
        total_duration = time.time() - total_start_time

        # Write results to a file
        with open('test_results.json', 'w') as f:
            json.dump({
                'total_requests': num_times * len(test_queries),
                'total_time_seconds': total_duration,
                'average_request_time_seconds': total_duration / (num_times * len(test_queries)),
                'results': [
                    {
                        'query': query,
                        'result': result,
                        'duration_seconds': duration
                    }
                    for query, result, duration in all_results
                ]
            }, f, indent=4)

        print(f"Stress Test Completed: {num_times * len(test_queries)} requests in {total_duration:.2f} seconds")
        print(f"Average Request Time: {total_duration / (num_times * len(test_queries)):.2f} seconds")

if __name__ == "__main__":
    num_times = 1
    asyncio.run(stress_test_api(num_times))