Simple Multi-Tenant API Testing

---

This guide shows how to test multi-tenant APIs with tenant-specific operations, data isolation, and cross-tenant access validation.

Use Cases

• Test tenant data isolation and segregation
• Validate tenant-specific API access and permissions
• Test tenant provisioning and configuration
• Check cross-tenant security boundaries

Simple Implementation

from locust import task, HttpUser
import random
import json

class MultiTenantTestUser(HttpUser):
    def on_start(self):
        # Sample tenant configurations
        self.tenants = [
            {"id": "tenant001", "name": "Acme Corp", "plan": "enterprise", "subdomain": "acme"},
            {"id": "tenant002", "name": "Beta LLC", "plan": "professional", "subdomain": "beta"},
            {"id": "tenant003", "name": "Gamma Inc", "plan": "starter", "subdomain": "gamma"}
        ]
        
        # Select a tenant for this test session
        self.current_tenant = random.choice(self.tenants)
        self.tenant_id = self.current_tenant["id"]
        
        # API endpoints
        self.tenant_endpoints = {
            "data": "/api/tenants/{tenant_id}/data",
            "users": "/api/tenants/{tenant_id}/users",
            "config": "/api/tenants/{tenant_id}/config",
            "resources": "/api/tenants/{tenant_id}/resources"
        }
        
        # Authentication tokens (would be obtained via login in real scenario)
        self.tenant_tokens = {
            "tenant001": "token_acme_123",
            "tenant002": "token_beta_456", 
            "tenant003": "token_gamma_789"
        }

    def get_tenant_headers(self, tenant_id=None):
        """Get headers with tenant-specific authentication"""
        if not tenant_id:
            tenant_id = self.tenant_id
            
        return {
            "Authorization": f"Bearer {self.tenant_tokens.get(tenant_id, 'invalid_token')}",
            "X-Tenant-ID": tenant_id,
            "Content-Type": "application/json"
        }

    @task(4)
    def test_tenant_data_access(self):
        """Test accessing tenant-specific data"""
        endpoint = self.tenant_endpoints["data"].format(tenant_id=self.tenant_id)
        headers = self.get_tenant_headers()
        
        with self.client.get(
            endpoint,
            headers=headers,
            name="Tenant Data Access"
        ) as response:
            if response.status_code == 200:
                try:
                    data = response.json()
                    
                    items = data.get("data", data.get("items", []))
                    tenant_context = data.get("tenant_id")
                    
                    print(f"Tenant {self.tenant_id} data: {len(items)} items")
                    
                    # Validate tenant context
                    if tenant_context and tenant_context != self.tenant_id:
                        response.failure(f"Data leakage: Expected {self.tenant_id}, got {tenant_context}")
                    
                except json.JSONDecodeError:
                    response.failure("Invalid JSON response")
            elif response.status_code == 403:
                response.failure("Access denied to tenant data")
            else:
                response.failure(f"Tenant data access failed: {response.status_code}")

    @task(3)
    def test_tenant_user_management(self):
        """Test tenant-specific user operations"""
        endpoint = self.tenant_endpoints["users"].format(tenant_id=self.tenant_id)
        headers = self.get_tenant_headers()
        
        # Create a test user for this tenant
        user_data = {
            "username": f"testuser_{random.randint(1000, 9999)}",
            "email": f"test{random.randint(100, 999)}@{self.current_tenant['subdomain']}.example.com",
            "role": random.choice(["admin", "user", "viewer"]),
            "tenant_id": self.tenant_id
        }
        
        with self.client.post(
            endpoint,
            json=user_data,
            headers=headers,
            name="Create Tenant User"
        ) as response:
            if response.status_code in [200, 201]:
                try:
                    result = response.json()
                    user_id = result.get("user_id") or result.get("id")
                    created_tenant = result.get("tenant_id")
                    
                    print(f"Created user {user_id} for tenant {self.tenant_id}")
                    
                    # Validate tenant assignment
                    if created_tenant != self.tenant_id:
                        response.failure(f"User created in wrong tenant: {created_tenant}")
                    
                    # Test retrieving the user
                    if user_id:
                        self._test_get_tenant_user(user_id)
                    
                except json.JSONDecodeError:
                    response.failure("Invalid JSON response")
            else:
                response.failure(f"Create tenant user failed: {response.status_code}")

    def _test_get_tenant_user(self, user_id):
        """Helper method to test retrieving tenant user"""
        endpoint = f"{self.tenant_endpoints['users'].format(tenant_id=self.tenant_id)}/{user_id}"
        headers = self.get_tenant_headers()
        
        with self.client.get(
            endpoint,
            headers=headers,
            name="Get Tenant User"
        ) as response:
            if response.status_code == 200:
                try:
                    user = response.json()
                    user_tenant = user.get("tenant_id")
                    
                    if user_tenant != self.tenant_id:
                        response.failure(f"User tenant mismatch: {user_tenant}")
                    
                except json.JSONDecodeError:
                    response.failure("Invalid JSON response")

    @task(2)
    def test_tenant_configuration(self):
        """Test tenant-specific configuration management"""
        endpoint = self.tenant_endpoints["config"].format(tenant_id=self.tenant_id)
        headers = self.get_tenant_headers()
        
        # Update tenant configuration
        config_data = {
            "features": {
                "advanced_analytics": self.current_tenant["plan"] == "enterprise",
                "api_rate_limit": 1000 if self.current_tenant["plan"] == "enterprise" else 100,
                "storage_limit_gb": 100 if self.current_tenant["plan"] == "enterprise" else 10
            },
            "branding": {
                "theme_color": f"#{random.randint(100000, 999999):06x}",
                "logo_url": f"https://{self.current_tenant['subdomain']}.example.com/logo.png"
            }
        }
        
        with self.client.put(
            endpoint,
            json=config_data,
            headers=headers,
            name="Update Tenant Config"
        ) as response:
            if response.status_code == 200:
                try:
                    result = response.json()
                    updated_tenant = result.get("tenant_id")
                    
                    print(f"Updated config for tenant {self.tenant_id}")
                    
                    if updated_tenant != self.tenant_id:
                        response.failure(f"Config updated for wrong tenant: {updated_tenant}")
                    
                except json.JSONDecodeError:
                    response.failure("Invalid JSON response")
            else:
                response.failure(f"Update tenant config failed: {response.status_code}")

    @task(2)
    def test_cross_tenant_access_denied(self):
        """Test that cross-tenant access is properly denied"""
        # Try to access a different tenant's data
        other_tenants = [t for t in self.tenants if t["id"] != self.tenant_id]
        if not other_tenants:
            return
            
        other_tenant = random.choice(other_tenants)
        other_tenant_id = other_tenant["id"]
        
        # Use current tenant's token to try accessing other tenant's data
        endpoint = self.tenant_endpoints["data"].format(tenant_id=other_tenant_id)
        headers = self.get_tenant_headers()  # Uses current tenant's token
        
        with self.client.get(
            endpoint,
            headers=headers,
            name="Cross-Tenant Access Test"
        ) as response:
            if response.status_code == 403:
                print(f"Cross-tenant access correctly denied: {self.tenant_id} -> {other_tenant_id}")
            elif response.status_code == 404:
                print(f"Cross-tenant resource not found (acceptable): {other_tenant_id}")
            elif response.status_code == 200:
                response.failure(f"Security breach: Cross-tenant access allowed {self.tenant_id} -> {other_tenant_id}")
            else:
                print(f"Cross-tenant access returned: {response.status_code}")

    @task(1)
    def test_tenant_resource_limits(self):
        """Test tenant-specific resource limits"""
        endpoint = self.tenant_endpoints["resources"].format(tenant_id=self.tenant_id)
        headers = self.get_tenant_headers()
        
        with self.client.get(
            endpoint,
            headers=headers,
            name="Tenant Resource Limits"
        ) as response:
            if response.status_code == 200:
                try:
                    resources = response.json()
                    
                    used_storage = resources.get("storage_used_gb", 0)
                    storage_limit = resources.get("storage_limit_gb", 0)
                    api_calls_used = resources.get("api_calls_used", 0)
                    api_calls_limit = resources.get("api_calls_limit", 0)
                    
                    print(f"Tenant {self.tenant_id} resources: Storage {used_storage}/{storage_limit}GB, API {api_calls_used}/{api_calls_limit}")
                    
                    # Validate plan-based limits
                    plan = self.current_tenant["plan"]
                    expected_storage = 100 if plan == "enterprise" else 10
                    
                    if storage_limit != expected_storage:
                        print(f"Warning: Storage limit {storage_limit} doesn't match plan {plan}")
                    
                except json.JSONDecodeError:
                    response.failure("Invalid JSON response")
            else:
                response.failure(f"Get tenant resources failed: {response.status_code}")

    @task(1)
    def test_tenant_subdomain_access(self):
        """Test tenant subdomain-based access"""
        subdomain = self.current_tenant["subdomain"]
        
        # Test subdomain-specific endpoint
        subdomain_endpoint = f"/api/subdomain/{subdomain}/info"
        headers = self.get_tenant_headers()
        
        with self.client.get(
            subdomain_endpoint,
            headers=headers,
            name="Subdomain Access"
        ) as response:
            if response.status_code == 200:
                try:
                    info = response.json()
                    
                    returned_tenant = info.get("tenant_id")
                    returned_subdomain = info.get("subdomain")
                    
                    print(f"Subdomain access: {subdomain} -> {returned_tenant}")
                    
                    if returned_tenant != self.tenant_id:
                        response.failure(f"Subdomain mapped to wrong tenant: {returned_tenant}")
                    
                    if returned_subdomain != subdomain:
                        response.failure(f"Subdomain mismatch: {returned_subdomain}")
                    
                except json.JSONDecodeError:
                    response.failure("Invalid JSON response")
            else:
                response.failure(f"Subdomain access failed: {response.status_code}")

Setup Instructions

1. Replace tenant endpoints with your actual multi-tenant API URLs
2. Update tenant configurations with real tenant data
3. Configure authentication tokens for each tenant
4. Adjust plan-based feature validation for your SaaS model

What This Tests

• Tenant Data Access: Tests tenant-specific data retrieval and isolation
• User Management: Tests tenant-scoped user operations
• Configuration: Tests tenant-specific configuration management
• Cross-Tenant Security: Tests that tenants cannot access each other's data
• Resource Limits: Tests plan-based resource limits and quotas
• Subdomain Access: Tests subdomain-based tenant routing

Best Practices

• Always validate tenant context in API responses
• Test both positive and negative access scenarios
• Verify plan-based feature restrictions
• Test tenant isolation under concurrent load
• Validate subdomain and routing mechanisms

Common Issues

• Data Leakage: Ensure no cross-tenant data is returned
• Authentication: Verify tenant-specific token validation
• Resource Limits: Plan upgrades/downgrades may not be immediate
• Subdomain Conflicts: Handle subdomain uniqueness and conflicts