Simple Shopping Cart API Testing

---

This guide shows how to test shopping cart APIs with common operations like adding items, updating quantities, and processing checkout.

Use Cases

• Test shopping cart item management
• Validate cart persistence and session handling
• Test checkout flow and cart validation
• Check cart performance under concurrent access

Simple Implementation

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}")

Setup Instructions

1. Replace cart endpoints with your actual shopping cart API URLs
2. Update product data with real products from your catalog
3. Adjust cart data structure based on your cart API requirements
4. Configure authentication and session handling as needed

What This Tests

• Cart Creation: Tests creating new shopping carts
• Add to Cart: Tests adding products to cart with quantities
• Cart Updates: Tests updating item quantities in cart
• Item Removal: Tests removing items from cart
• Cart Viewing: Tests retrieving cart contents and totals
• Checkout Process: Tests cart checkout and order creation
• Cart Validation: Tests error handling for invalid operations
• Cart Persistence: Tests cart recovery across sessions

Best Practices

• Test realistic shopping patterns and quantities
• Validate cart totals and item counts
• Test concurrent cart operations
• Handle cart session management properly
• Test both valid and invalid cart operations

Common Issues

• Session Management: Cart persistence across browser sessions
• Concurrency: Multiple users updating same cart simultaneously
• Inventory Sync: Cart items may become unavailable during checkout
• Price Updates: Product prices may change while items are in cart