Simple Snapshot Testing

---

This guide shows how to capture and compare API response snapshots to detect regressions. Perfect for ensuring API responses remain consistent over time.

Use Cases

• Detect API response changes
• Validate response structure consistency
• Catch unintended API modifications
• Monitor API stability over time

Simple Implementation

from locust import task, HttpUser
import json
import hashlib
import time

class SnapshotTestUser(HttpUser):
    def on_start(self):
        # Endpoints to snapshot
        self.endpoints = [
            "/api/users",
            "/api/products",
            "/api/orders",
            "/api/health"
        ]
        
        # Store snapshots in memory (in real use, save to file/database)
        self.snapshots = {}
        self.snapshot_mismatches = []
        
        # Initialize baseline snapshots
        self.create_baseline_snapshots()

    def create_baseline_snapshots(self):
        """Create initial snapshots to compare against"""
        print("Creating baseline snapshots...")
        
        for endpoint in self.endpoints:
            try:
                with self.client.get(endpoint, name=f"Baseline - {endpoint}") as response:
                    if response.status_code == 200:
                        snapshot_data = self.create_snapshot(response)
                        self.snapshots[endpoint] = snapshot_data
                        print(f"Baseline snapshot created for {endpoint}")
                    else:
                        print(f"Failed to create baseline for {endpoint}: {response.status_code}")
            except Exception as e:
                print(f"Error creating baseline for {endpoint}: {e}")

    @task(4)
    def test_api_snapshot(self):
        """Test API response against saved snapshot"""
        endpoint = self.random_endpoint()
        
        with self.client.get(endpoint, name=f"Snapshot Test - {endpoint}") as response:
            if response.status_code == 200:
                current_snapshot = self.create_snapshot(response)
                self.compare_snapshots(endpoint, current_snapshot)
            else:
                response.failure(f"Snapshot test failed: {response.status_code}")

    @task(2)
    def test_response_structure(self):
        """Test that response structure matches snapshot"""
        endpoint = self.random_endpoint()
        
        with self.client.get(endpoint, name=f"Structure Test - {endpoint}") as response:
            if response.status_code == 200:
                try:
                    data = response.json()
                    structure = self.extract_structure(data)
                    
                    if endpoint in self.snapshots:
                        baseline_structure = self.snapshots[endpoint].get("structure")
                        if structure == baseline_structure:
                            print(f"Structure test {endpoint}: PASSED")
                        else:
                            print(f"Structure test {endpoint}: FAILED - structure changed")
                            response.failure("Response structure changed")
                    else:
                        print(f"Structure test {endpoint}: No baseline available")
                        
                except json.JSONDecodeError:
                    response.failure("Invalid JSON response")

    @task(2)
    def test_response_fields(self):
        """Test that required fields are present"""
        endpoint = self.random_endpoint()
        
        with self.client.get(endpoint, name=f"Fields Test - {endpoint}") as response:
            if response.status_code == 200:
                try:
                    data = response.json()
                    
                    if endpoint in self.snapshots:
                        baseline_fields = self.snapshots[endpoint].get("fields", set())
                        current_fields = self.extract_fields(data)
                        
                        missing_fields = baseline_fields - current_fields
                        new_fields = current_fields - baseline_fields
                        
                        if missing_fields:
                            print(f"Fields test {endpoint}: Missing fields: {missing_fields}")
                            response.failure(f"Missing fields: {missing_fields}")
                        elif new_fields:
                            print(f"Fields test {endpoint}: New fields detected: {new_fields}")
                        else:
                            print(f"Fields test {endpoint}: PASSED")
                    else:
                        print(f"Fields test {endpoint}: No baseline available")
                        
                except json.JSONDecodeError:
                    response.failure("Invalid JSON response")

    @task(1)
    def test_response_size(self):
        """Test that response size is within expected range"""
        endpoint = self.random_endpoint()
        
        with self.client.get(endpoint, name=f"Size Test - {endpoint}") as response:
            if response.status_code == 200:
                current_size = len(response.text)
                
                if endpoint in self.snapshots:
                    baseline_size = self.snapshots[endpoint].get("size", 0)
                    size_diff = abs(current_size - baseline_size)
                    size_change_percent = (size_diff / baseline_size * 100) if baseline_size > 0 else 0
                    
                    if size_change_percent > 50:  # More than 50% change
                        print(f"Size test {endpoint}: Large size change: {size_change_percent:.1f}%")
                        response.failure(f"Response size changed significantly: {size_change_percent:.1f}%")
                    elif size_change_percent > 20:  # 20-50% change
                        print(f"Size test {endpoint}: Moderate size change: {size_change_percent:.1f}%")
                    else:
                        print(f"Size test {endpoint}: Size stable ({size_change_percent:.1f}% change)")
                else:
                    print(f"Size test {endpoint}: No baseline size available")

    def create_snapshot(self, response):
        """Create a snapshot from API response"""
        try:
            data = response.json()
            
            snapshot = {
                "timestamp": time.time(),
                "status_code": response.status_code,
                "size": len(response.text),
                "content_hash": hashlib.md5(response.text.encode()).hexdigest(),
                "structure": self.extract_structure(data),
                "fields": self.extract_fields(data),
                "headers": dict(response.headers)
            }
            
            return snapshot
            
        except json.JSONDecodeError:
            return {
                "timestamp": time.time(),
                "status_code": response.status_code,
                "size": len(response.text),
                "content_hash": hashlib.md5(response.text.encode()).hexdigest(),
                "error": "Invalid JSON"
            }

    def extract_structure(self, data):
        """Extract the structure of JSON data"""
        if isinstance(data, dict):
            return {key: self.extract_structure(value) for key, value in data.items()}
        elif isinstance(data, list):
            if data:
                return [self.extract_structure(data[0])]  # Structure of first item
            else:
                return []
        else:
            return type(data).__name__

    def extract_fields(self, data):
        """Extract all field names from JSON data"""
        fields = set()
        
        def collect_fields(obj, prefix=""):
            if isinstance(obj, dict):
                for key, value in obj.items():
                    field_name = f"{prefix}.{key}" if prefix else key
                    fields.add(field_name)
                    collect_fields(value, field_name)
            elif isinstance(obj, list) and obj:
                collect_fields(obj[0], prefix)  # Check first item structure
        
        collect_fields(data)
        return fields

    def compare_snapshots(self, endpoint, current_snapshot):
        """Compare current snapshot with baseline"""
        if endpoint not in self.snapshots:
            print(f"Snapshot comparison {endpoint}: No baseline snapshot")
            return
        
        baseline = self.snapshots[endpoint]
        
        # Compare content hash
        if current_snapshot["content_hash"] == baseline["content_hash"]:
            print(f"Snapshot comparison {endpoint}: IDENTICAL")
            return
        
        # Check what changed
        changes = []
        
        if current_snapshot["status_code"] != baseline["status_code"]:
            changes.append(f"status_code: {baseline['status_code']} -> {current_snapshot['status_code']}")
        
        if current_snapshot["structure"] != baseline["structure"]:
            changes.append("response structure changed")
        
        size_diff = current_snapshot["size"] - baseline["size"]
        if abs(size_diff) > 100:  # More than 100 bytes difference
            changes.append(f"size: {size_diff:+d} bytes")
        
        if changes:
            print(f"Snapshot comparison {endpoint}: CHANGED - {', '.join(changes)}")
            self.snapshot_mismatches.append({
                "endpoint": endpoint,
                "changes": changes,
                "timestamp": current_snapshot["timestamp"]
            })
        else:
            print(f"Snapshot comparison {endpoint}: Minor changes detected")

    def random_endpoint(self):
        """Get a random endpoint for testing"""
        import random
        return random.choice(self.endpoints)

    @task(1)
    def report_snapshot_status(self):
        """Report overall snapshot testing status"""
        total_endpoints = len(self.endpoints)
        endpoints_with_baselines = len(self.snapshots)
        total_mismatches = len(self.snapshot_mismatches)
        
        print(f"Snapshot Status: {endpoints_with_baselines}/{total_endpoints} baselines, {total_mismatches} mismatches")
        
        if total_mismatches > 0:
            print("Recent mismatches:")
            for mismatch in self.snapshot_mismatches[-3:]:  # Show last 3
                print(f"  {mismatch['endpoint']}: {', '.join(mismatch['changes'])}")

Setup Instructions

1. Run initial test to create baseline snapshots
2. Save snapshots to persistent storage (file/database) for real use
3. Configure endpoints you want to monitor
4. Set appropriate thresholds for size and structure changes

What This Tests

• Response Consistency: Detects when API responses change
• Structure Stability: Monitors JSON structure changes
• Field Presence: Ensures required fields remain present
• Response Size: Tracks significant size changes

Snapshot Components

• Content Hash: MD5 hash of entire response
• Structure: JSON structure template
• Fields: All field names in response
• Size: Response size in bytes
• Headers: HTTP response headers

Change Detection

The guide detects:

• Structure Changes: JSON schema modifications
• Missing Fields: Required fields removed
• New Fields: Additional fields added
• Size Changes: Significant response size changes
• Status Code Changes: HTTP status modifications

Best Practices

• Regular Baselines: Update baselines when changes are intentional
• Threshold Tuning: Adjust change thresholds for your API
• Persistent Storage: Save snapshots to files or database
• Change Review: Review all detected changes before deployment

Common Use Cases

• Regression Testing: Catch unintended API changes
• API Monitoring: Continuous API stability monitoring
• Version Validation: Ensure API versions remain stable
• Documentation Sync: Verify API matches documentation