# import requests # import time # from faker import Faker # class AuthClient: # """ # Python client for interacting with the Authentication API # """ # def __init__(self, base_url="http://localhost:7860/api"): # """ # Initialize the client with the API base URL # Args: # base_url (str): The base URL of the API # """ # self.base_url = base_url # self.token = None # def register(self, email, password): # """ # Register a new user # Args: # email (str): User's email # password (str): User's password (should be at least 8 characters) # Returns: # dict: The user data returned by the API # Raises: # Exception: If registration fails # """ # url = f"{self.base_url}/register" # data = { # "email": email, # "password": password # } # response = requests.post(url, json=data) # if response.status_code == 201: # return response.json() # else: # error_detail = response.json().get("detail", "Unknown error") # raise Exception(f"Registration failed: {error_detail} (Status: {response.status_code})") # def login(self, email, password): # """ # Login to obtain an authentication token # Args: # email (str): User's email # password (str): User's password # Returns: # dict: The token data returned by the API # Raises: # Exception: If login fails # """ # url = f"{self.base_url}/login" # data = { # "email": email, # "password": password # } # response = requests.post(url, json=data) # if response.status_code == 200: # token_data = response.json() # self.token = token_data["access_token"] # return token_data # else: # error_detail = response.json().get("detail", "Unknown error") # raise Exception(f"Login failed: {error_detail} (Status: {response.status_code})") # def get_current_user(self): # """ # Get information about the current logged-in user # Returns: # dict: The user data returned by the API # Raises: # Exception: If not authenticated or request fails # """ # if not self.token: # raise Exception("Not authenticated. Please login first.") # url = f"{self.base_url}/users/me" # headers = {"Authorization": f"Bearer {self.token}"} # response = requests.get(url, headers=headers) # if response.status_code == 200: # return response.json() # else: # error_detail = response.json().get("detail", "Unknown error") # raise Exception(f"Failed to get user info: {error_detail} (Status: {response.status_code})") # def logout(self): # """Clear the authentication token""" # self.token = None # # Example usage # def main(): # # Initialize the client # client = AuthClient("https://amaye15-authenticationapp.hf.space/api") # # Initialize Faker # fake = Faker() # for i in range(10): # try: # # Generate random user data # first_name = fake.first_name() # last_name = fake.last_name() # email = fake.email() # password = fake.password(length=12, special_chars=True, digits=True, upper_case=True, lower_case=True) # # Register a new user # print(f"Registering a new user: {first_name} {last_name}...") # try: # user = client.register(email, password) # print(f"Registered user: {user}") # except Exception as e: # print(f"Registration failed: {e}") # # Login # print("\nLogging in...") # token_data = client.login(email, password) # print(f"Login successful, token: {token_data['access_token'][:10]}...") # # Get current user # print("\nGetting current user info...") # user_info = client.get_current_user() # print(f"Current user: {user_info}") # # Logout # print("\nLogging out...") # client.logout() # print("Logged out successfully") # except Exception as e: # print(f"Error: {e}") # if __name__ == "__main__": # main() import asyncio import aiohttp import time from faker import Faker class AuthClient: """ Asynchronous Python client for interacting with the Authentication API """ def __init__(self, base_url="http://localhost:7860/api"): """ Initialize the client with the API base URL Args: base_url (str): The base URL of the API """ self.base_url = base_url self.token = None self.session = None async def __aenter__(self): """Create and enter an aiohttp session""" self.session = aiohttp.ClientSession() return self async def __aexit__(self, exc_type, exc_val, exc_tb): """Close the aiohttp session""" if self.session: await self.session.close() async def _get_session(self): """Get or create an aiohttp session""" if self.session is None: self.session = aiohttp.ClientSession() return self.session async def register(self, email, password): """ Register a new user Args: email (str): User's email password (str): User's password (should be at least 8 characters) Returns: dict: The user data returned by the API Raises: Exception: If registration fails """ url = f"{self.base_url}/register" data = { "email": email, "password": password } session = await self._get_session() async with session.post(url, json=data) as response: if response.status == 201: return await response.json() else: error_data = await response.json() error_detail = error_data.get("detail", "Unknown error") raise Exception(f"Registration failed: {error_detail} (Status: {response.status})") async def login(self, email, password): """ Login to obtain an authentication token Args: email (str): User's email password (str): User's password Returns: dict: The token data returned by the API Raises: Exception: If login fails """ url = f"{self.base_url}/login" data = { "email": email, "password": password } session = await self._get_session() async with session.post(url, json=data) as response: if response.status == 200: token_data = await response.json() self.token = token_data["access_token"] return token_data else: error_data = await response.json() error_detail = error_data.get("detail", "Unknown error") raise Exception(f"Login failed: {error_detail} (Status: {response.status})") async def get_current_user(self): """ Get information about the current logged-in user Returns: dict: The user data returned by the API Raises: Exception: If not authenticated or request fails """ if not self.token: raise Exception("Not authenticated. Please login first.") url = f"{self.base_url}/users/me" headers = {"Authorization": f"Bearer {self.token}"} session = await self._get_session() async with session.get(url, headers=headers) as response: if response.status == 200: return await response.json() else: error_data = await response.json() error_detail = error_data.get("detail", "Unknown error") raise Exception(f"Failed to get user info: {error_detail} (Status: {response.status})") def logout(self): """Clear the authentication token""" self.token = None # Load testing function async def load_test(num_users=10, concurrency=5, base_url="https://amaye15-authenticationapp.hf.space/api"): """ Run a load test with multiple simulated users Args: num_users (int): Total number of users to simulate concurrency (int): Number of concurrent users base_url (str): The base URL of the API """ fake = Faker() start_time = time.time() completed = 0 success_count = 0 failure_count = 0 # Semaphore to limit concurrency sem = asyncio.Semaphore(concurrency) # For progress tracking progress_lock = asyncio.Lock() async def run_single_user(): nonlocal completed, success_count, failure_count async with sem: # This limits concurrency async with AuthClient(base_url) as client: try: # Generate random user data email = fake.email() password = fake.password(length=12, special_chars=True, digits=True, upper_case=True, lower_case=True) # Complete user flow await client.register(email, password) await client.login(email, password) await client.get_current_user() client.logout() async with progress_lock: completed += 1 success_count += 1 # Print progress print(f"Progress: {completed}/{num_users} users completed", end="\r") except Exception as e: async with progress_lock: completed += 1 failure_count += 1 print(f"Error: {e}") print(f"Progress: {completed}/{num_users} users completed", end="\r") # Create all tasks tasks = [run_single_user() for _ in range(num_users)] # Display start message print(f"Starting load test with {num_users} users (max {concurrency} concurrent)...") # Run all tasks await asyncio.gather(*tasks) # Calculate stats end_time = time.time() duration = end_time - start_time # Display results print("\n\n--- Load Test Results ---") print(f"Total users: {num_users}") print(f"Concurrency level: {concurrency}") print(f"Successful flows: {success_count} ({success_count/num_users*100:.1f}%)") print(f"Failed flows: {failure_count} ({failure_count/num_users*100:.1f}%)") print(f"Total duration: {duration:.2f} seconds") if success_count > 0: print(f"Average time per successful user: {duration/success_count:.2f} seconds") print(f"Requests per second: {success_count/duration:.2f}") # Example usage async def main(): # Initialize the client base_url = "https://amaye15-authenticationapp.hf.space/api" # Run a simple example with a single user fake = Faker() async with AuthClient(base_url) as client: # Generate random user data first_name = fake.first_name() last_name = fake.last_name() email = fake.email() password = fake.password(length=12, special_chars=True, digits=True, upper_case=True, lower_case=True) try: # Register a new user print(f"Registering a new user: {first_name} {last_name}...") user = await client.register(email, password) print(f"Registered user: {user}") # Login print("\nLogging in...") token_data = await client.login(email, password) print(f"Login successful, token: {token_data['access_token'][:10]}...") # Get current user print("\nGetting current user info...") user_info = await client.get_current_user() print(f"Current user: {user_info}") # Logout print("\nLogging out...") client.logout() print("Logged out successfully") except Exception as e: print(f"Error: {e}") # Run a load test print("\nRunning load test...") await load_test(10, 5, base_url) if __name__ == "__main__": asyncio.run(main())