Explorer reports addition
We have added a new Explorer feature to reports, with a timeline scrubber and easy anomaly detection.
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
- Replace inventory endpoints with your actual inventory API URLs
- Update product data with real SKUs and product information
- Adjust location names to match your warehouse/store setup
- 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.