File size: 3,466 Bytes
310260a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 | """
Load Tests with Locust
Locust load testing scenarios for MCP server and chat endpoint.
"""
from locust import HttpUser, task, between
import json
class ChatUser(HttpUser):
"""
Simulated user for load testing chat endpoint.
"""
wait_time = between(1, 3) # Wait 1-3 seconds between requests
def on_start(self):
"""
Setup: Login and get JWT token before starting tasks.
"""
# Signup
signup_response = self.client.post("/api/auth/signup", json={
"email": f"loadtest_{self.environment.runner.user_count}@example.com",
"password": "password123",
"name": f"Load Test User {self.environment.runner.user_count}"
})
if signup_response.status_code == 201:
# Login
login_response = self.client.post("/api/auth/signin", json={
"email": f"loadtest_{self.environment.runner.user_count}@example.com",
"password": "password123"
})
if login_response.status_code == 200:
data = login_response.json()
self.token = data["access_token"]
self.user_id = data["user"]["id"]
else:
self.token = None
self.user_id = None
else:
# User might already exist, try login
login_response = self.client.post("/api/auth/signin", json={
"email": f"loadtest_{self.environment.runner.user_count}@example.com",
"password": "password123"
})
if login_response.status_code == 200:
data = login_response.json()
self.token = data["access_token"]
self.user_id = data["user"]["id"]
else:
self.token = None
self.user_id = None
@task(3)
def chat_create_task(self):
"""
Task: Create a task via chat endpoint (weight: 3)
"""
if not self.token:
return
self.client.post(
f"/api/{self.user_id}/chat",
headers={"Authorization": f"Bearer {self.token}"},
json={"message": "Add a task to test load"}
)
@task(5)
def chat_list_tasks(self):
"""
Task: List tasks via chat endpoint (weight: 5)
"""
if not self.token:
return
self.client.post(
f"/api/{self.user_id}/chat",
headers={"Authorization": f"Bearer {self.token}"},
json={"message": "Show me my tasks"}
)
@task(2)
def chat_mark_complete(self):
"""
Task: Mark task complete via chat endpoint (weight: 2)
"""
if not self.token:
return
self.client.post(
f"/api/{self.user_id}/chat",
headers={"Authorization": f"Bearer {self.token}"},
json={"message": "Mark my first task as done"}
)
@task(1)
def chat_delete_task(self):
"""
Task: Delete task via chat endpoint (weight: 1)
"""
if not self.token:
return
self.client.post(
f"/api/{self.user_id}/chat",
headers={"Authorization": f"Bearer {self.token}"},
json={"message": "Delete my first task"}
)
# Run with: locust -f locustfile.py --host=http://localhost:8000
# Then open http://localhost:8089 to configure and start load test
|