tests/performance/load_test_otp.py

import asyncio
import httpx
import time

BASE_URL = "http://127.0.0.1:8000/api/v1/otp"

async def verify_otp(session, email_or_mobile, otp):
    response = await session.post(
        f"{BASE_URL}/verify",
        json={"email_or_mobile": email_or_mobile, "otp": otp},
    )
    return response.status_code, response.text

async def load_test_otp(num_requests):
    async with httpx.AsyncClient() as session:
        tasks = []

        for i in range(num_requests):
            email_or_mobile = f"user{i}@example.com"
            tasks.append(verify_otp(session, email_or_mobile, "123456"))

        start_time = time.time()
        results = await asyncio.gather(*tasks)
        end_time = time.time()

    total_time = end_time - start_time
    successful_requests = sum(1 for status, _ in results if status == 200)
    failed_requests = num_requests - successful_requests

    print(f"Total Requests: {num_requests}")
    print(f"Successful Requests: {successful_requests}")
    print(f"Failed Requests: {failed_requests}")
    print(f"Total Time: {total_time:.2f} seconds")
    print(f"Requests per Second: {num_requests / total_time:.2f}")

if __name__ == "__main__":
    num_requests = 500  # Number of requests
    asyncio.run(load_test_otp(num_requests))