Skip to main content
This guide shows how to test inventory management APIs with common operations like checking stock, updating inventory, and managing product availability.

​
Use Cases

  • Test inventory stock level checking
  • Validate inventory updates and adjustments
  • Test low stock alerts and notifications
  • Check inventory synchronization across systems

​
Simple Implementation

from locust import task, HttpUser
import random
import json

class InventoryTestUser(HttpUser):
    def on_start(self):
        # Sample product data for testing
        self.products = [
            {"id": "PROD001", "sku": "LAPTOP-HP-001", "name": "HP Laptop"},
            {"id": "PROD002", "sku": "PHONE-IPH-001", "name": "iPhone 13"},
            {"id": "PROD003", "sku": "HEADP-SON-001", "name": "Sony Headphones"},
            {"id": "PROD004", "sku": "TABLET-IPD-001", "name": "iPad Pro"},
            {"id": "PROD005", "sku": "WATCH-APL-001", "name": "Apple Watch"}
        ]
        
        # Inventory operations
        self.operations = ["restock", "adjustment", "sale", "return"]
        
        # API endpoints
        self.inventory_endpoints = {
            "check": "/api/inventory/check",
            "update": "/api/inventory/update",
            "bulk": "/api/inventory/bulk-update",
            "alerts": "/api/inventory/alerts"
        }

    @task(4)
    def test_check_inventory(self):
        """Test checking inventory levels"""
        product = random.choice(self.products)
        
        params = {
            "sku": product["sku"],
            "location": random.choice(["warehouse_1", "store_001", "online"])
        }
        
        with self.client.get(
            self.inventory_endpoints["check"],
            params=params,
            name="Check Inventory"
        ) as response:
            if response.status_code == 200:
                try:
                    inventory = response.json()
                    
                    available = inventory.get("available", 0)
                    reserved = inventory.get("reserved", 0)
                    total = inventory.get("total", 0)
                    
                    print(f"Inventory check {product['sku']}: {available}/{total} available")
                    
                    # Validate inventory data
                    if available < 0:
                        response.failure("Negative available inventory")
                    
                except json.JSONDecodeError:
                    response.failure("Invalid JSON response")
            elif response.status_code == 404:
                print(f"Product not found: {product['sku']}")
            else:
                response.failure(f"Inventory check failed: {response.status_code}")

    @task(3)
    def test_update_inventory(self):
        """Test updating inventory levels"""
        product = random.choice(self.products)
        operation = random.choice(self.operations)
        
        # Generate appropriate quantity based on operation
        if operation == "restock":
            quantity = random.randint(10, 100)
        elif operation == "sale":
            quantity = -random.randint(1, 5)
        elif operation == "return":
            quantity = random.randint(1, 3)
        else:  # adjustment
            quantity = random.randint(-10, 10)
        
        update_data = {
            "sku": product["sku"],
            "quantity": quantity,
            "operation": operation,
            "location": "warehouse_1",
            "reason": f"Test {operation} operation"
        }
        
        with self.client.post(
            self.inventory_endpoints["update"],
            json=update_data,
            name="Update Inventory"
        ) as response:
            if response.status_code == 200:
                try:
                    result = response.json()
                    new_quantity = result.get("new_quantity", 0)
                    
                    print(f"Inventory updated {product['sku']}: {operation} {quantity}, new total: {new_quantity}")
                    
                    # Validate update
                    if new_quantity < 0 and operation != "adjustment":
                        print(f"Warning: Negative inventory after {operation}")
                    
                except json.JSONDecodeError:
                    response.failure("Invalid JSON response")
            elif response.status_code == 400:
                print(f"Invalid inventory update: {operation} {quantity}")
            elif response.status_code == 409:
                print(f"Inventory conflict: insufficient stock for {operation}")
            else:
                response.failure(f"Inventory update failed: {response.status_code}")

    @task(2)
    def test_bulk_inventory_update(self):
        """Test bulk inventory updates"""
        # Select 2-3 products for bulk update
        selected_products = random.sample(self.products, random.randint(2, 3))
        
        bulk_updates = []
        for product in selected_products:
            bulk_updates.append({
                "sku": product["sku"],
                "quantity": random.randint(5, 50),
                "operation": "restock",
                "location": "warehouse_1"
            })
        
        bulk_data = {
            "updates": bulk_updates,
            "batch_id": f"BATCH_{random.randint(1000, 9999)}"
        }
        
        with self.client.post(
            self.inventory_endpoints["bulk"],
            json=bulk_data,
            name="Bulk Inventory Update"
        ) as response:
            if response.status_code == 200:
                try:
                    result = response.json()
                    
                    processed = result.get("processed", 0)
                    failed = result.get("failed", 0)
                    batch_id = result.get("batch_id")
                    
                    print(f"Bulk update: {processed} processed, {failed} failed (batch: {batch_id})")
                    
                    if failed > 0:
                        print(f"Some bulk updates failed: {failed}/{len(bulk_updates)}")
                    
                except json.JSONDecodeError:
                    response.failure("Invalid JSON response")
            else:
                response.failure(f"Bulk inventory update failed: {response.status_code}")

    @task(2)
    def test_inventory_availability(self):
        """Test inventory availability checking"""
        product = random.choice(self.products)
        requested_quantity = random.randint(1, 10)
        
        availability_params = {
            "sku": product["sku"],
            "quantity": requested_quantity,
            "location": random.choice(["warehouse_1", "store_001", "all"])
        }
        
        with self.client.get(
            f"{self.inventory_endpoints['check']}/availability",
            params=availability_params,
            name="Check Availability"
        ) as response:
            if response.status_code == 200:
                try:
                    availability = response.json()
                    
                    available = availability.get("available", False)
                    can_fulfill = availability.get("can_fulfill", 0)
                    locations = availability.get("locations", [])
                    
                    print(f"Availability {product['sku']}: {can_fulfill}/{requested_quantity} available")
                    
                    if not available and can_fulfill > 0:
                        print(f"Inconsistent availability data")
                    
                except json.JSONDecodeError:
                    response.failure("Invalid JSON response")
            else:
                response.failure(f"Availability check failed: {response.status_code}")

    @task(1)
    def test_low_stock_alerts(self):
        """Test low stock alert system"""
        with self.client.get(
            self.inventory_endpoints["alerts"],
            params={"threshold": random.choice([5, 10, 20])},
            name="Low Stock Alerts"
        ) as response:
            if response.status_code == 200:
                try:
                    alerts = response.json()
                    
                    if isinstance(alerts, dict):
                        items = alerts.get("alerts", alerts.get("items", []))
                    else:
                        items = alerts if isinstance(alerts, list) else []
                    
                    print(f"Low stock alerts: {len(items)} products below threshold")
                    
                    # Validate alert data
                    for alert in items[:3]:  # Check first 3 alerts
                        if isinstance(alert, dict):
                            sku = alert.get("sku")
                            current_stock = alert.get("current_stock", 0)
                            threshold = alert.get("threshold", 0)
                            
                            if current_stock > threshold:
                                print(f"Invalid alert: {sku} stock {current_stock} > threshold {threshold}")
                    
                except json.JSONDecodeError:
                    response.failure("Invalid JSON response")
            else:
                response.failure(f"Low stock alerts failed: {response.status_code}")

    @task(1)
    def test_inventory_reservation(self):
        """Test inventory reservation for orders"""
        product = random.choice(self.products)
        reserve_quantity = random.randint(1, 5)
        
        reservation_data = {
            "sku": product["sku"],
            "quantity": reserve_quantity,
            "order_id": f"ORDER_{random.randint(10000, 99999)}",
            "expires_in": 900  # 15 minutes
        }
        
        with self.client.post(
            f"{self.inventory_endpoints['update']}/reserve",
            json=reservation_data,
            name="Reserve Inventory"
        ) as response:
            if response.status_code == 200:
                try:
                    result = response.json()
                    
                    reservation_id = result.get("reservation_id")
                    reserved_quantity = result.get("reserved_quantity", 0)
                    expires_at = result.get("expires_at")
                    
                    print(f"Reserved {reserved_quantity} units of {product['sku']} (ID: {reservation_id})")
                    
                    # Test releasing the reservation
                    if reservation_id:
                        self._release_reservation(reservation_id)
                    
                except json.JSONDecodeError:
                    response.failure("Invalid JSON response")
            elif response.status_code == 409:
                print(f"Insufficient inventory to reserve {reserve_quantity} units")
            else:
                response.failure(f"Inventory reservation failed: {response.status_code}")

    def _release_reservation(self, reservation_id):
        """Helper method to release inventory reservation"""
        with self.client.delete(
            f"{self.inventory_endpoints['update']}/reserve/{reservation_id}",
            name="Release Reservation"
        ) as response:
            if response.status_code == 200:
                print(f"Released reservation: {reservation_id}")
            else:
                print(f"Failed to release reservation: {reservation_id}")

​
Setup Instructions

  1. Replace inventory endpoints with your actual inventory API URLs
  2. Update product data with real SKUs and product information
  3. Adjust location names to match your warehouse/store setup
  4. Configure authentication headers if required by your inventory API

​
What This Tests

  • Inventory Checking: Tests stock level retrieval and availability
  • Inventory Updates: Tests stock adjustments and operations
  • Bulk Operations: Tests batch inventory updates
  • Availability Checks: Tests product availability validation
  • Low Stock Alerts: Tests inventory monitoring and alerts
  • Reservations: Tests inventory reservation and release

​
Best Practices

  • Use realistic product SKUs and inventory operations
  • Test both positive and negative inventory adjustments
  • Validate inventory consistency across operations
  • Monitor performance with bulk operations
  • Test edge cases like negative inventory and conflicts

​
Common Issues

  • Concurrency: Multiple updates to same SKU may cause conflicts
  • Negative Inventory: Some systems allow, others don’t
  • Location Sync: Multi-location inventory may have sync delays
  • Reservation Expiry: Test reservation timeout handling

This guide provides comprehensive inventory management testing patterns for e-commerce and retail applications with proper stock tracking and warehouse operations.