from locust import task, HttpUser
import json
import random
class BatchOperationsUser(HttpUser):
def on_start(self):
self.created_items = [] # Track items we create
@task(4)
def test_batch_create_users(self):
"""Test creating multiple users in a single batch request"""
# Create batch of 3-5 users
batch_size = random.randint(3, 5)
users_batch = []
for i in range(batch_size):
user = {
"name": f"User {random.randint(1000, 9999)}",
"email": f"user{random.randint(1000, 9999)}@example.com",
"active": True
}
users_batch.append(user)
batch_data = {"users": users_batch}
with self.client.post(
"/api/users/batch",
json=batch_data,
name="Batch Create Users"
) as response:
if response.status_code in [200, 201]:
try:
result = response.json()
created_count = len(result.get("created", []))
print(f"Batch create: {created_count}/{batch_size} users created")
# Store created IDs for later use
if "created" in result:
for item in result["created"]:
if "id" in item:
self.created_items.append(item["id"])
except json.JSONDecodeError:
print("Batch create successful but no JSON response")
else:
response.failure(f"Batch create failed: {response.status_code}")
@task(3)
def test_batch_update_users(self):
"""Test updating multiple users in a single batch request"""
if len(self.created_items) < 2:
return # Need at least 2 items to update
# Select 2-3 items to update
update_count = min(3, len(self.created_items))
items_to_update = random.sample(self.created_items, update_count)
updates_batch = []
for item_id in items_to_update:
update = {
"id": item_id,
"name": f"Updated User {random.randint(1000, 9999)}",
"active": random.choice([True, False])
}
updates_batch.append(update)
batch_data = {"updates": updates_batch}
with self.client.put(
"/api/users/batch",
json=batch_data,
name="Batch Update Users"
) as response:
if response.status_code == 200:
try:
result = response.json()
updated_count = len(result.get("updated", []))
print(f"Batch update: {updated_count}/{update_count} users updated")
except json.JSONDecodeError:
print("Batch update successful")
else:
response.failure(f"Batch update failed: {response.status_code}")
@task(2)
def test_batch_create_products(self):
"""Test creating multiple products in batch"""
batch_size = random.randint(2, 4)
products_batch = []
for i in range(batch_size):
product = {
"name": f"Product {random.randint(1000, 9999)}",
"price": round(random.uniform(10.0, 100.0), 2),
"category": random.choice(["electronics", "books", "clothing"])
}
products_batch.append(product)
batch_data = {"products": products_batch}
with self.client.post(
"/api/products/batch",
json=batch_data,
name="Batch Create Products"
) as response:
if response.status_code in [200, 201]:
print(f"Batch created {batch_size} products")
else:
response.failure(f"Product batch create failed: {response.status_code}")
@task(2)
def test_large_batch_operation(self):
"""Test larger batch to check size limits"""
# Try a larger batch (10 items)
batch_size = 10
large_batch = []
for i in range(batch_size):
item = {
"name": f"Bulk Item {i}",
"value": random.randint(1, 100),
"type": "test"
}
large_batch.append(item)
batch_data = {"items": large_batch}
with self.client.post(
"/api/items/batch",
json=batch_data,
name="Large Batch Operation"
) as response:
if response.status_code in [200, 201]:
print(f"Large batch ({batch_size} items) successful")
elif response.status_code == 413:
print(f"Large batch rejected - payload too large")
response.failure("Batch size limit exceeded")
elif response.status_code == 400:
print(f"Large batch rejected - bad request")
response.failure("Large batch validation failed")
else:
response.failure(f"Large batch failed: {response.status_code}")
@task(1)
def test_batch_delete(self):
"""Test deleting multiple items in batch"""
if len(self.created_items) < 2:
return # Need items to delete
# Delete 1-2 items
delete_count = min(2, len(self.created_items))
items_to_delete = []
for _ in range(delete_count):
if self.created_items:
item_id = self.created_items.pop()
items_to_delete.append(item_id)
if not items_to_delete:
return
batch_data = {"ids": items_to_delete}
with self.client.delete(
"/api/users/batch",
json=batch_data,
name="Batch Delete Users"
) as response:
if response.status_code in [200, 204]:
print(f"Batch deleted {len(items_to_delete)} users")
else:
response.failure(f"Batch delete failed: {response.status_code}")
@task(1)
def test_empty_batch(self):
"""Test batch operation with empty data"""
empty_batch = {"users": []}
with self.client.post(
"/api/users/batch",
json=empty_batch,
name="Empty Batch Test"
) as response:
if response.status_code == 400:
print("Empty batch correctly rejected")
elif response.status_code in [200, 201]:
print("Empty batch accepted (no-op)")
else:
response.failure(f"Empty batch unexpected response: {response.status_code}")
@task(1)
def test_mixed_batch_operation(self):
"""Test batch with mix of valid and invalid items"""
mixed_batch = [
{"name": "Valid User", "email": "valid@example.com"},
{"name": "", "email": "invalid-email"}, # Invalid
{"name": "Another Valid User", "email": "valid2@example.com"}
]
batch_data = {"users": mixed_batch}
with self.client.post(
"/api/users/batch",
json=batch_data,
name="Mixed Batch Test"
) as response:
if response.status_code in [200, 201]:
try:
result = response.json()
created = len(result.get("created", []))
errors = len(result.get("errors", []))
print(f"Mixed batch: {created} created, {errors} errors")
except json.JSONDecodeError:
print("Mixed batch processed")
elif response.status_code == 400:
print("Mixed batch rejected due to invalid items")
else:
response.failure(f"Mixed batch failed: {response.status_code}")
