from locust import task, HttpUser
import random
import json
class ShoppingCartTestUser(HttpUser):
def on_start(self):
# Sample products for cart testing
self.products = [
{"id": "PROD001", "name": "Laptop", "price": 999.99, "sku": "LAP001"},
{"id": "PROD002", "name": "Mouse", "price": 29.99, "sku": "MOU001"},
{"id": "PROD003", "name": "Keyboard", "price": 79.99, "sku": "KEY001"},
{"id": "PROD004", "name": "Monitor", "price": 299.99, "sku": "MON001"},
{"id": "PROD005", "name": "Headphones", "price": 149.99, "sku": "HEAD001"}
]
# Cart and session management
self.cart_id = None
self.session_id = f"sess_{random.randint(100000, 999999)}"
self.cart_items = []
# API endpoints
self.cart_endpoints = {
"create": "/api/cart",
"add": "/api/cart/items",
"update": "/api/cart/items/{item_id}",
"remove": "/api/cart/items/{item_id}",
"checkout": "/api/cart/checkout"
}
# Create a cart for this session
self.create_cart()
def create_cart(self):
"""Create a new shopping cart"""
cart_data = {
"session_id": self.session_id,
"user_id": f"user_{random.randint(1000, 9999)}"
}
with self.client.post(
self.cart_endpoints["create"],
json=cart_data,
name="Create Cart"
) as response:
if response.status_code in [200, 201]:
try:
result = response.json()
self.cart_id = result.get("cart_id") or result.get("id")
print(f"Created cart: {self.cart_id}")
except json.JSONDecodeError:
print("Cart created but invalid JSON response")
@task(4)
def test_add_to_cart(self):
"""Test adding items to cart"""
if not self.cart_id:
self.create_cart()
product = random.choice(self.products)
quantity = random.randint(1, 3)
item_data = {
"cart_id": self.cart_id,
"product_id": product["id"],
"sku": product["sku"],
"quantity": quantity,
"price": product["price"]
}
with self.client.post(
self.cart_endpoints["add"],
json=item_data,
name="Add to Cart"
) as response:
if response.status_code in [200, 201]:
try:
result = response.json()
item_id = result.get("item_id") or result.get("id")
if item_id:
self.cart_items.append({
"item_id": item_id,
"product_id": product["id"],
"quantity": quantity
})
print(f"Added to cart: {quantity}x {product['name']}")
except json.JSONDecodeError:
response.failure("Invalid JSON response")
elif response.status_code == 400:
print(f"Invalid add to cart request")
else:
response.failure(f"Add to cart failed: {response.status_code}")
@task(3)
def test_update_cart_item(self):
"""Test updating cart item quantities"""
if not self.cart_items:
return
cart_item = random.choice(self.cart_items)
new_quantity = random.randint(1, 5)
update_data = {
"quantity": new_quantity
}
update_url = self.cart_endpoints["update"].format(item_id=cart_item["item_id"])
with self.client.put(
update_url,
json=update_data,
name="Update Cart Item"
) as response:
if response.status_code == 200:
try:
result = response.json()
updated_quantity = result.get("quantity", new_quantity)
# Update our local tracking
cart_item["quantity"] = updated_quantity
print(f"Updated cart item to quantity: {updated_quantity}")
except json.JSONDecodeError:
response.failure("Invalid JSON response")
elif response.status_code == 404:
print(f"Cart item not found: {cart_item['item_id']}")
# Remove from our tracking
self.cart_items.remove(cart_item)
else:
response.failure(f"Update cart item failed: {response.status_code}")
@task(2)
def test_remove_from_cart(self):
"""Test removing items from cart"""
if not self.cart_items:
return
cart_item = random.choice(self.cart_items)
remove_url = self.cart_endpoints["remove"].format(item_id=cart_item["item_id"])
with self.client.delete(
remove_url,
name="Remove from Cart"
) as response:
if response.status_code in [200, 204]:
print(f"Removed item from cart: {cart_item['item_id']}")
self.cart_items.remove(cart_item)
elif response.status_code == 404:
print(f"Cart item already removed: {cart_item['item_id']}")
self.cart_items.remove(cart_item)
else:
response.failure(f"Remove from cart failed: {response.status_code}")
@task(2)
def test_view_cart(self):
"""Test viewing cart contents"""
if not self.cart_id:
return
params = {"cart_id": self.cart_id}
with self.client.get(
self.cart_endpoints["create"],
params=params,
name="View Cart"
) as response:
if response.status_code == 200:
try:
cart = response.json()
items = cart.get("items", [])
total = cart.get("total", 0)
item_count = cart.get("item_count", len(items))
print(f"Cart view: {item_count} items, total: ${total}")
# Validate cart data consistency
if len(items) != len(self.cart_items):
print(f"Cart item count mismatch: API={len(items)}, local={len(self.cart_items)}")
except json.JSONDecodeError:
response.failure("Invalid JSON response")
elif response.status_code == 404:
print(f"Cart not found: {self.cart_id}")
self.cart_id = None
self.cart_items = []
else:
response.failure(f"View cart failed: {response.status_code}")
@task(1)
def test_cart_checkout(self):
"""Test cart checkout process"""
if not self.cart_items or not self.cart_id:
return
checkout_data = {
"cart_id": self.cart_id,
"shipping_address": {
"street": "123 Test St",
"city": "Test City",
"state": "TS",
"zip": "12345"
},
"payment_method": {
"type": "credit_card",
"card_number": "4111111111111111",
"exp_month": "12",
"exp_year": "2025",
"cvv": "123"
}
}
with self.client.post(
self.cart_endpoints["checkout"],
json=checkout_data,
name="Cart Checkout"
) as response:
if response.status_code == 200:
try:
result = response.json()
order_id = result.get("order_id")
status = result.get("status", "unknown")
total = result.get("total", 0)
print(f"Checkout successful: Order {order_id}, status: {status}, total: ${total}")
# Reset cart after successful checkout
self.cart_items = []
self.cart_id = None
except json.JSONDecodeError:
response.failure("Invalid JSON response")
elif response.status_code == 400:
print("Checkout failed: Invalid cart or payment data")
elif response.status_code == 409:
print("Checkout failed: Cart conflict or inventory issue")
else:
response.failure(f"Cart checkout failed: {response.status_code}")
@task(1)
def test_cart_validation(self):
"""Test cart validation and error handling"""
if not self.cart_id:
return
# Test adding invalid item
invalid_item = {
"cart_id": self.cart_id,
"product_id": "INVALID_PRODUCT",
"quantity": -1, # Invalid quantity
"price": -10.00 # Invalid price
}
with self.client.post(
self.cart_endpoints["add"],
json=invalid_item,
name="Invalid Cart Operation"
) as response:
if response.status_code == 400:
print("Invalid cart operation correctly rejected")
elif response.status_code == 404:
print("Invalid product correctly rejected")
elif response.status_code == 200:
response.failure("Invalid cart operation was accepted")
else:
print(f"Invalid cart operation returned: {response.status_code}")
@task(1)
def test_cart_persistence(self):
"""Test cart persistence across sessions"""
if not self.cart_id:
return
# Simulate session interruption and recovery
original_session = self.session_id
# Try to recover cart with session ID
recovery_params = {
"session_id": original_session
}
with self.client.get(
f"{self.cart_endpoints['create']}/recover",
params=recovery_params,
name="Recover Cart"
) as response:
if response.status_code == 200:
try:
recovered_cart = response.json()
recovered_id = recovered_cart.get("cart_id")
if recovered_id == self.cart_id:
print(f"Cart persistence verified: {recovered_id}")
else:
print(f"Cart recovery returned different ID: {recovered_id}")
except json.JSONDecodeError:
response.failure("Invalid JSON response")
elif response.status_code == 404:
print("No cart found for session recovery")
else:
print(f"Cart recovery returned: {response.status_code}")
