test cases

test_cases/test_game_api.py

+import requests
+import json 
+
+chat_url = "https://voice-agent.caprover.al-arcade.com/chat"
+chat_payload = {
+    'student_id': 'student_001',
+    'text': 'how do i open this gate ',
+    'game_context': 'The user is in the jungle level trying to open a door.'
+}
+output_filename = "agent_response_final.wav"
+
+print(f" Sending POST request to {chat_url} ")
+chat_response = requests.post(chat_url, data=chat_payload)
+response_data = chat_response.json()
+print("Server Confirmation:")
+print(json.dumps(response_data, indent=2, ensure_ascii=False))
+print("-" * 40)
+
+
+
+agent_text = response_data.get("agent_response")
+audio_url = response_data.get("audio_filepath")
+
+
+
+print(f"Downloading audio from the provided URL ")
+
+audio_response = requests.get(audio_url, stream=True)
+audio_response.raise_for_status()
+
+with open(output_filename, 'wb') as audio_file:
+    for chunk in audio_response.iter_content(chunk_size=8192):
+        audio_file.write(chunk)
+
+print("-" * 40)
+print("--- Download Complete ---")
+print(f"Agent's Text Response: '{agent_text}'")
+print(f"Audio file saved as: '{output_filename}'")
\ No newline at end of file

test_cases/test_impatient_user.py

+import requests
+import threading
+import time
+import base64
+import json
+
+# --- Configuration ---
+BASE_URL = "voice-agent.caprover.al-arcade.com"
+STUDENT_ID = "student_001"
+
+# Define three different requests for the SAME student
+REQUESTS = {
+    "req1": "What is photosynthesis?",
+    "req2": "Tell me about gravity.",
+    "req3": "How do penguins stay warm?"
+}
+
+# Thread-safe dictionary to store results
+final_results = {}
+lock = threading.Lock()
+
+def run_full_cycle(request_name, question):
+    """
+    Simulates a full user interaction: POST a question, then immediately GET the audio.
+    """
+    try:
+        # --- Step 1: POST the chat message ---
+        chat_url = f"{BASE_URL}/chat"
+        chat_payload = {'student_id': STUDENT_ID, 'text': question}
+        
+        # We don't print here to keep the final report clean
+        chat_response = requests.post(chat_url, data=chat_payload)
+        chat_response.raise_for_status()
+        
+        # --- Step 2: GET the audio response ---
+        # A short, realistic delay as if the user is waiting for the POST to finish
+        time.sleep(1) 
+        get_url = f"{BASE_URL}/get-audio-response?student_id={STUDENT_ID}"
+        
+        audio_response = requests.get(get_url)
+        audio_response.raise_for_status()
+        
+        encoded_text = audio_response.headers.get('X-Response-Text')
+        if not encoded_text:
+            raise ValueError("X-Response-Text header was missing.")
+
+        decoded_text = base64.b64decode(encoded_text).decode('utf-8')
+
+        # Safely store the successful result
+        with lock:
+            final_results[request_name] = {
+                "request": question,
+                "response": decoded_text,
+                "status": "✅ SUCCESS"
+            }
+
+    except Exception as e:
+        with lock:
+            final_results[request_name] = {
+                "request": question,
+                "error": str(e),
+                "status": "❌ FAILED"
+            }
+
+if __name__ == "__main__":
+    print(f"--- Starting 'Impatient User' test for student '{STUDENT_ID}' ---")
+    print("Sending 3 requests in rapid succession...")
+    
+    threads = []
+    
+    # Create and start a thread for each request
+    for name, q in REQUESTS.items():
+        thread = threading.Thread(target=run_full_cycle, args=(name, q))
+        threads.append(thread)
+        thread.start()
+        # A very short delay to ensure they are sent almost at the same time
+        time.sleep(0.1)
+
+    # Wait for all threads to complete
+    for thread in threads:
+        thread.join()
+        
+    print("\n" + "="*50)
+    print("---           FINAL TEST REPORT           ---")
+    print("="*50 + "\n")
+    
+    # Print the final report
+    for req_name in REQUESTS.keys():
+        result = final_results.get(req_name, {})
+        
+        print(f"Request: '{result.get('request', 'N/A')}'")
+        print(f"Status:  {result.get('status', 'UNKNOWN')}")
+        
+        if "error" in result:
+            print(f"  -> Error: {result.get('error')}")
+        else:
+            print(f"  -> Response: '{result.get('response', '')[:80]}...'") # Print a snippet
+        
+        print("-" * 50)
\ No newline at end of file

test_cases/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