from locust import HttpUser, task, between
import json
import time
import random
import uuid
class APIVersioningUser(HttpUser):
wait_time = between(1, 3)
def on_start(self):
"""Initialize API versioning testing"""
self.api_versions = ["v1", "v2", "v3"]
self.auth_token = None
self.test_data = {}
# Authenticate (version-agnostic)
self._authenticate()
def _authenticate(self):
"""Authenticate with API (usually version-agnostic)"""
auth_data = {
"username": f"testuser_{random.randint(1000, 9999)}",
"password": "test_password_123"
}
response = self.client.post('/auth/login',
json=auth_data,
headers={'Content-Type': 'application/json'},
name="auth_login")
if response.status_code == 200:
try:
auth_result = response.json()
self.auth_token = auth_result.get('access_token')
except json.JSONDecodeError:
pass
def _get_headers(self, version=None):
"""Get headers with optional API version"""
headers = {'Content-Type': 'application/json'}
if self.auth_token:
headers['Authorization'] = f'Bearer {self.auth_token}'
if version:
# Different versioning strategies
headers['API-Version'] = version # Header-based versioning
headers['Accept'] = f'application/vnd.api+json;version={version}' # Accept header
return headers
@task(4)
def test_version_header_strategy(self):
"""Test header-based API versioning"""
version = random.choice(self.api_versions)
# Create user with specific API version
user_data = {
"name": f"Test User {random.randint(100, 999)}",
"email": f"test{random.randint(1000, 9999)}@example.com",
"age": random.randint(18, 80)
}
# Add version-specific fields
if version == "v1":
user_data["phone"] = f"+1555{random.randint(1000000, 9999999)}"
elif version == "v2":
user_data.update({
"phone_number": f"+1555{random.randint(1000000, 9999999)}",
"preferences": {"newsletter": True, "sms": False}
})
elif version == "v3":
user_data.update({
"contact": {
"phone": f"+1555{random.randint(1000000, 9999999)}",
"preferred_method": "email"
},
"settings": {
"notifications": {"email": True, "push": True},
"privacy": {"profile_public": False}
}
})
response = self.client.post('/api/users',
json=user_data,
headers=self._get_headers(version),
name=f"create_user_{version}_header")
if response.status_code in [200, 201]:
try:
created_user = response.json()
user_id = created_user.get('id')
if user_id:
self.test_data[f"user_{version}"] = user_id
self._test_version_response_format(version, user_id)
except json.JSONDecodeError:
pass
def _test_version_response_format(self, version, user_id):
"""Test version-specific response formats"""
response = self.client.get(f'/api/users/{user_id}',
headers=self._get_headers(version),
name=f"get_user_{version}_format")
if response.status_code == 200:
try:
user_data = response.json()
# Validate version-specific response structure
if version == "v1":
# v1 might have flat structure
assert 'phone' in user_data, f"v1 should have 'phone' field"
elif version == "v2":
# v2 might have phone_number instead of phone
assert 'phone_number' in user_data, f"v2 should have 'phone_number' field"
assert 'preferences' in user_data, f"v2 should have 'preferences' field"
elif version == "v3":
# v3 might have nested contact structure
assert 'contact' in user_data, f"v3 should have 'contact' field"
assert 'settings' in user_data, f"v3 should have 'settings' field"
except (json.JSONDecodeError, AssertionError) as e:
print(f"Version {version} response validation failed: {e}")
@task(3)
def test_url_path_versioning(self):
"""Test URL path-based API versioning"""
version = random.choice(self.api_versions)
# Test different endpoint structures per version
product_data = {
"name": f"Test Product {random.randint(100, 999)}",
"price": round(random.uniform(10.00, 999.99), 2),
"category": random.choice(["electronics", "clothing", "books"])
}
# Version-specific endpoint paths
if version == "v1":
endpoint = f'/api/{version}/products'
product_data["description"] = "Simple description"
elif version == "v2":
endpoint = f'/api/{version}/products'
product_data.update({
"description": "Enhanced description",
"tags": ["test", "product"],
"metadata": {"source": "api_test"}
})
elif version == "v3":
endpoint = f'/api/{version}/catalog/products' # Different path structure
product_data.update({
"description": {
"short": "Brief description",
"long": "Detailed product description"
},
"taxonomy": {
"category": product_data["category"],
"subcategory": "test_subcategory"
},
"attributes": {"color": "blue", "size": "medium"}
})
response = self.client.post(endpoint,
json=product_data,
headers=self._get_headers(),
name=f"create_product_{version}_path")
@task(2)
def test_version_compatibility(self):
"""Test backward compatibility between versions"""
# Create data with v1 API
v1_data = {
"title": f"V1 Article {random.randint(100, 999)}",
"content": "Article created with v1 API",
"author": "Test Author",
"published": True
}
v1_response = self.client.post('/api/v1/articles',
json=v1_data,
headers=self._get_headers(),
name="create_article_v1")
if v1_response.status_code in [200, 201]:
try:
v1_article = v1_response.json()
article_id = v1_article.get('id')
if article_id:
# Try to access same data with v2 API
v2_response = self.client.get(f'/api/v2/articles/{article_id}',
headers=self._get_headers("v2"),
name="get_article_v1_data_via_v2")
# Try to update v1 data with v2 API
v2_update = {
"title": f"Updated via V2 - {random.randint(100, 999)}",
"content": "Updated content via v2 API",
"metadata": {"updated_via": "v2", "timestamp": int(time.time())}
}
self.client.put(f'/api/v2/articles/{article_id}',
json=v2_update,
headers=self._get_headers("v2"),
name="update_v1_article_via_v2")
except json.JSONDecodeError:
pass
@task(1)
def test_version_deprecation(self):
"""Test deprecated version handling"""
# Test deprecated v1 endpoint
deprecated_data = {
"old_field": "value",
"legacy_format": True
}
response = self.client.post('/api/v1/legacy-endpoint',
json=deprecated_data,
headers=self._get_headers("v1"),
name="test_deprecated_v1_endpoint")
# Check for deprecation warnings in headers
if response.status_code == 200:
deprecation_warning = response.headers.get('Deprecation')
sunset_header = response.headers.get('Sunset')
if deprecation_warning:
print(f"Deprecation warning received: {deprecation_warning}")
if sunset_header:
print(f"Sunset date: {sunset_header}")
def on_stop(self):
"""Cleanup test data across all versions"""
for version in self.api_versions:
user_id = self.test_data.get(f"user_{version}")
if user_id:
try:
self.client.delete(f'/api/users/{user_id}',
headers=self._get_headers(version),
name=f"cleanup_user_{version}")
except:
pass
