Test API versioning strategies including v1 vs v2 endpoints and migration
LoadForge can record your browser, graphically build tests, scan your site with a wizard and more. Sign up now to run your first test.
This guide demonstrates how to test API versioning with LoadForge, covering version compatibility, migration scenarios, and deprecation handling.
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
This guide provides comprehensive API versioning testing patterns for maintaining backward compatibility and smooth version transitions.