"""
Test script for the new conversation chat endpoint
"""
import asyncio
import json
import requests
import uuid
BASE_URL = "http://localhost:8000"
def test_health():
"""Test health endpoint"""
print("Testing health endpoint...")
response = requests.get(f"{BASE_URL}/health")
print(f"Status: {response.status_code}")
print(f"Response: {response.json()}")
return response.status_code == 200
def test_conversation_chat():
"""Test the new conversation chat endpoint"""
print("\nTesting conversation chat endpoint...")
# Generate a unique conversation ID
conversation_id = f"test_conv_{uuid.uuid4().hex[:8]}"
print(f"Using conversation ID: {conversation_id}")
# Test messages
test_messages = [
"Hello, my name is Alice. What's the weather like?",
"Can you remember my name?",
"What did I ask you about earlier?"
]
for i, message in enumerate(test_messages, 1):
print(f"\n--- Message {i} ---")
print(f"User: {message}")
payload = {
"message": message,
"conversation_id": conversation_id,
"model": "mistral:latest",
"include_context": True,
"context_limit": 5
}
try:
response = requests.post(
f"{BASE_URL}/api/conversation/chat",
json=payload,
headers={"Content-Type": "application/json"}
)
if response.status_code == 200:
data = response.json()
print(f"Assistant: {data['response']}")
print(f"Metadata: {data.get('metadata', {})}")
else:
print(f"Error {response.status_code}: {response.text}")
except Exception as e:
print(f"Request failed: {e}")
def test_memory_retrieval():
"""Test memory retrieval for the conversation"""
print("\n--- Testing Memory Retrieval ---")
# This would require the conversation ID from the previous test
# For now, just test the endpoint structure
payload = {
"conversation_id": "test_conv_12345678",
"limit": 10
}
try:
response = requests.post(
f"{BASE_URL}/api/memory/get",
json=payload,
headers={"Content-Type": "application/json"}
)
print(f"Memory retrieval status: {response.status_code}")
if response.status_code == 200:
data = response.json()
print(f"Retrieved {data.get('count', 0)} memories")
else:
print(f"Error: {response.text}")
except Exception as e:
print(f"Memory retrieval failed: {e}")
if __name__ == "__main__":
print("=== Conversation Chat API Test ===")
# Check if server is running
if not test_health():
print("Server is not healthy. Make sure it's running on localhost:8000")
exit(1)
# Test the conversation chat
test_conversation_chat()
# Test memory retrieval
test_memory_retrieval()
print("\n=== Test Complete ===")
