from locust import task, HttpUser
import json
class ContractTestUser(HttpUser):
def on_start(self):
# Expected API contracts
self.contracts = {
"/api/users": {
"status_code": 200,
"required_fields": ["id", "name", "email"],
"field_types": {
"id": int,
"name": str,
"email": str
}
},
"/api/products": {
"status_code": 200,
"required_fields": ["id", "name", "price"],
"field_types": {
"id": int,
"name": str,
"price": (int, float)
}
},
"/api/orders": {
"status_code": 200,
"required_fields": ["id", "user_id", "total"],
"field_types": {
"id": int,
"user_id": int,
"total": (int, float)
}
}
}
@task(3)
def test_users_contract(self):
"""Test users API contract"""
endpoint = "/api/users"
with self.client.get(endpoint, name="Users Contract") as response:
self.validate_contract(endpoint, response)
@task(2)
def test_products_contract(self):
"""Test products API contract"""
endpoint = "/api/products"
with self.client.get(endpoint, name="Products Contract") as response:
self.validate_contract(endpoint, response)
@task(2)
def test_orders_contract(self):
"""Test orders API contract"""
endpoint = "/api/orders"
with self.client.get(endpoint, name="Orders Contract") as response:
self.validate_contract(endpoint, response)
@task(1)
def test_single_user_contract(self):
"""Test single user API contract"""
endpoint = "/api/users/1"
with self.client.get(endpoint, name="Single User Contract") as response:
# Use same contract as users list but for single item
if response.status_code == 200:
try:
data = response.json()
self.validate_single_item_schema(data, self.contracts["/api/users"])
print("Single user contract: PASSED")
except Exception as e:
response.failure(f"Single user contract failed: {e}")
elif response.status_code == 404:
print("Single user contract: User not found (acceptable)")
else:
response.failure(f"Unexpected status code: {response.status_code}")
def validate_contract(self, endpoint, response):
"""Validate API response against contract"""
if endpoint not in self.contracts:
response.failure(f"No contract defined for {endpoint}")
return
contract = self.contracts[endpoint]
# Check status code
if response.status_code != contract["status_code"]:
response.failure(f"Status code mismatch: expected {contract['status_code']}, got {response.status_code}")
return
# Parse JSON response
try:
data = response.json()
except json.JSONDecodeError:
response.failure("Invalid JSON response")
return
# Handle both single items and arrays
if isinstance(data, list):
if len(data) > 0:
self.validate_single_item_schema(data[0], contract)
print(f"Contract validation for {endpoint}: PASSED (array with {len(data)} items)")
else:
self.validate_single_item_schema(data, contract)
print(f"Contract validation for {endpoint}: PASSED")
def validate_single_item_schema(self, item, contract):
"""Validate a single item against the contract schema"""
# Check required fields
for field in contract["required_fields"]:
if field not in item:
raise ValueError(f"Missing required field: {field}")
# Check field types
for field, expected_type in contract["field_types"].items():
if field in item:
if not isinstance(item[field], expected_type):
raise ValueError(f"Field {field} has wrong type: expected {expected_type}, got {type(item[field])}")
@task(1)
def test_error_responses(self):
"""Test error response contracts"""
# Test 404 response
with self.client.get("/api/users/99999", name="404 Error Contract") as response:
if response.status_code == 404:
try:
error_data = response.json()
# Basic error response structure
if "error" in error_data or "message" in error_data:
print("404 error contract: PASSED")
else:
response.failure("404 response missing error message")
except json.JSONDecodeError:
response.failure("404 response not valid JSON")
else:
response.failure(f"Expected 404, got {response.status_code}")
@task(1)
def test_post_contract(self):
"""Test POST request contract"""
user_data = {
"name": "Test User",
"email": "test@example.com"
}
with self.client.post(
"/api/users",
json=user_data,
name="POST Contract"
) as response:
if response.status_code in [201, 200]:
try:
created_user = response.json()
# Validate created user has required fields
required_fields = ["id", "name", "email"]
for field in required_fields:
if field not in created_user:
response.failure(f"Created user missing field: {field}")
return
# Validate the data we sent is reflected
if created_user["name"] != user_data["name"]:
response.failure("Created user name doesn't match input")
return
if created_user["email"] != user_data["email"]:
response.failure("Created user email doesn't match input")
return
print("POST contract validation: PASSED")
except json.JSONDecodeError:
response.failure("POST response not valid JSON")
elif response.status_code == 400:
# Bad request is acceptable for validation errors
print("POST contract: Validation error (acceptable)")
else:
response.failure(f"Unexpected POST status code: {response.status_code}")
