multi user stress test script

self_hosted_env/voice_agent/test_multi_users.py

+import requests
+import threading
+import time
+import base64
+import statistics
+
+# --- Configuration ---
+BASE_URL = "https://voice-agent.caprover.al-arcade.com"
+# A challenging but reasonable concurrency level for a single container.
+CONCURRENCY_LEVEL = 8 
+
+# Use 8 unique students to test different contexts.
+STUDENTS_AND_QUESTIONS = {
+    "student_001": "What is photosynthesis and why is it important for life on Earth?",
+    "student_002": "Explain Newton's first law of motion with a simple example.",
+    "student_003": "How do polar bears adapt to survive in the freezing Arctic environment?",
+    "student_004": "What is the difference between a star and a planet?",
+    "student_005": "Can you explain the main stages of the water cycle?",
+    "student_006": "What is static electricity and how can I see it at home?",
+    "student_007": "Why do some animals, like birds, migrate every year?",
+    "student_008": "What is DNA and what does it do in our bodies?"
+}
+
+# Thread-safe dictionary to store results
+final_results = {}
+lock = threading.Lock()
+
+
+def run_test_for_student(student_id, question):
+    """
+    Performs the full POST -> GET cycle and records detailed timing information.
+    Includes a 60-second timeout to prevent the script from hanging.
+    """
+    start_time = time.time()
+    
+    try:
+        # --- Step 1: POST the chat message ---
+        chat_url = f"{BASE_URL}/chat"
+        chat_payload = {'student_id': student_id, 'text': question}
+        
+        post_start = time.time()
+        # CRITICAL: Add a timeout to prevent hanging
+        chat_response = requests.post(chat_url, data=chat_payload, timeout=60)
+        post_end = time.time()
+        chat_response.raise_for_status()  # Will raise an exception for 4xx or 5xx errors
+        
+        # --- Step 2: GET the audio response ---
+        get_url = f"{BASE_URL}/get-audio-response?student_id={student_id}"
+        
+        get_start = time.time()
+        # CRITICAL: Add a timeout here as well
+        audio_response = requests.get(get_url, timeout=60)
+        get_end = time.time()
+        audio_response.raise_for_status()
+
+        end_time = time.time()
+        
+        encoded_text = audio_response.headers.get('X-Response-Text', '')
+        decoded_text = base64.b64decode(encoded_text).decode('utf-8') if encoded_text else 'NO RESPONSE TEXT'
+
+        with lock:
+            final_results[student_id] = {
+                "request": question,
+                "response": decoded_text,
+                "duration": end_time - start_time,
+                "post_duration": post_end - post_start,
+                "get_duration": get_end - get_start,
+                "status": "SUCCESS"
+            }
+
+    except Exception as e:
+        end_time = time.time()
+        with lock:
+            final_results[student_id] = {
+                "request": question,
+                "error": str(e),
+                "duration": end_time - start_time,
+                "status": "FAILED"
+            }
+
+
+if __name__ == "__main__":
+    script_start_time = time.time()
+    
+    print("="*70)
+    print("🔥 REASONABLE CONCURRENCY TEST 🔥")
+    print("="*70)
+    print(f"Concurrency Level: {CONCURRENCY_LEVEL} simultaneous requests")
+    print(f"Target URL: {BASE_URL}")
+    print(f"Started at: {time.strftime('%H:%M:%S')}")
+    print("="*70 + "\n")
+    
+    threads = []
+    
+    # Create and start a thread for each student
+    for sid, q in STUDENTS_AND_QUESTIONS.items():
+        thread = threading.Thread(target=run_test_for_student, args=(sid, q))
+        threads.append(thread)
+        thread.start()
+        # Stagger the start of each request slightly to simulate a more realistic load
+        time.sleep(0.2)
+
+    # Wait for all threads to complete
+    for thread in threads:
+        thread.join()
+        
+    total_time = time.time() - script_start_time
+    
+    # --- Analysis and Reporting ---
+    print("\n" + "="*70)
+    print("📊              DETAILED RESULTS & ANALYSIS              📊")
+    print("="*70 + "\n")
+    
+    sorted_results = sorted(final_results.items())
+    
+    for student_id, result in sorted_results:
+        status_icon = "✅" if result['status'] == "SUCCESS" else "❌"
+        
+        print(f"{status_icon} Student ID: {student_id}")
+        print(f"    ▶️  Request:  '{result.get('request', 'N/A')}'")
+        
+        if result['status'] == "SUCCESS":
+            print(f"    ◀️  Response: '{result.get('response', 'N/A')[:80]}...'")
+            print(f"    ⏱️  Timing:   Total={result['duration']:.2f}s (POST={result['post_duration']:.2f}s, GET={result['get_duration']:.2f}s)")
+        else:
+            print(f"    ❌ ERROR:    {result.get('error')}")
+            print(f"    ⏱️  Failed after {result['duration']:.2f}s")
+        
+        print("-" * 70)
+
+    # --- Summary ---
+    success_count = len([res for res in final_results.values() if res['status'] == 'SUCCESS'])
+    failed_count = len(final_results) - success_count
+    
+    print("\n" + "="*70)
+    print("📈                  SUMMARY STATISTICS                   📈")
+    print("="*70)
+    print(f"✅ Successful Requests: {success_count} / {CONCURRENCY_LEVEL}")
+    print(f"❌ Failed Requests:     {failed_count} / {CONCURRENCY_LEVEL}")
+    print(f"⏱️  Total Test Runtime:  {total_time:.2f}s")
+    
+    if success_count > 0:
+        durations = [res['duration'] for res in final_results.values() if res['status'] == 'SUCCESS']
+        print(f"📊 Average Success Time: {statistics.mean(durations):.2f}s")
+        print(f"🚀 Fastest Success Time: {min(durations):.2f}s")
+        print(f"🐢 Slowest Success Time: {max(durations):.2f}s")
+    print("="*70)
\ No newline at end of file