API Versioning Testing

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.