from locust import HttpUser, task, between
import random
import time
import json
from datetime import datetime, timezone
class MultiRegionUser(HttpUser):
"""Simulate users from different geographic regions"""
wait_time = between(1, 3)
def on_start(self):
"""Initialize region-specific configuration"""
# Simulate different regional characteristics
self.region_profiles = {
'us_east': {
'timezone': 'America/New_York',
'language': 'en-US',
'currency': 'USD',
'expected_latency': 50, # ms baseline
'peak_hours': [9, 10, 11, 14, 15, 16]
},
'us_west': {
'timezone': 'America/Los_Angeles',
'language': 'en-US',
'currency': 'USD',
'expected_latency': 80,
'peak_hours': [9, 10, 11, 14, 15, 16]
},
'europe': {
'timezone': 'Europe/London',
'language': 'en-GB',
'currency': 'EUR',
'expected_latency': 120,
'peak_hours': [8, 9, 10, 13, 14, 15]
},
'asia_pacific': {
'timezone': 'Asia/Singapore',
'language': 'en-SG',
'currency': 'SGD',
'expected_latency': 200,
'peak_hours': [9, 10, 11, 14, 15, 16]
}
}
# Select random region profile
self.current_region = random.choice(list(self.region_profiles.keys()))
self.profile = self.region_profiles[self.current_region]
# Set region-specific headers
self.client.headers.update({
'Accept-Language': self.profile['language'],
'X-User-Region': self.current_region,
'X-Currency': self.profile['currency'],
'X-Timezone': self.profile['timezone']
})
# Track latency measurements
self.latency_measurements = []
print(f"User initialized for region: {self.current_region}")
@task(3)
def test_api_latency(self):
"""Test API response times and validate against regional expectations"""
start_time = time.time()
with self.client.get(
"/api/v1/data",
name=f"API Latency - {self.current_region}",
catch_response=True
) as response:
latency = (time.time() - start_time) * 1000 # Convert to ms
self.latency_measurements.append(latency)
if response.status_code == 200:
# Validate latency against regional expectations
expected_max = self.profile['expected_latency'] * 2 # Allow 2x baseline
if latency <= expected_max:
response.success()
print(f"Good latency from {self.current_region}: {latency:.1f}ms")
else:
response.failure(f"High latency from {self.current_region}: {latency:.1f}ms (expected < {expected_max}ms)")
else:
response.failure(f"API request failed: {response.status_code}")
@task(2)
def test_cdn_performance(self):
"""Test CDN content delivery performance"""
# Test different types of content
content_types = [
{'path': '/static/images/hero.jpg', 'type': 'image'},
{'path': '/static/css/main.css', 'type': 'css'},
{'path': '/static/js/app.js', 'type': 'javascript'},
{'path': '/api/v1/config.json', 'type': 'json'}
]
content = random.choice(content_types)
start_time = time.time()
with self.client.get(
content['path'],
name=f"CDN {content['type']} - {self.current_region}",
catch_response=True
) as response:
download_time = (time.time() - start_time) * 1000
if response.status_code == 200:
# Check for CDN headers
cdn_headers = ['X-Cache', 'CF-Cache-Status', 'X-Served-By', 'X-CDN']
cdn_hit = any(header in response.headers for header in cdn_headers)
if cdn_hit:
print(f"CDN hit for {content['type']} from {self.current_region}: {download_time:.1f}ms")
response.success()
else:
print(f"CDN miss for {content['type']} from {self.current_region}: {download_time:.1f}ms")
response.success() # Still successful, just noting CDN status
else:
response.failure(f"CDN content failed: {response.status_code}")
@task(2)
def test_regional_content(self):
"""Test region-specific content delivery"""
regional_endpoints = [
f"/api/v1/content/regional?region={self.current_region}",
f"/api/v1/pricing?currency={self.profile['currency']}",
f"/api/v1/localization?lang={self.profile['language']}"
]
endpoint = random.choice(regional_endpoints)
with self.client.get(
endpoint,
name=f"Regional Content - {self.current_region}",
catch_response=True
) as response:
if response.status_code == 200:
try:
data = response.json()
# Validate region-specific content
if 'currency' in data and data['currency'] == self.profile['currency']:
print(f"Correct currency for {self.current_region}: {data['currency']}")
if 'language' in data and data['language'] == self.profile['language']:
print(f"Correct language for {self.current_region}: {data['language']}")
response.success()
except json.JSONDecodeError:
response.success() # Content might not be JSON
else:
response.failure(f"Regional content failed: {response.status_code}")
@task(1)
def test_regional_failover(self):
"""Test regional failover mechanisms"""
# Simulate accessing primary and backup regional endpoints
endpoints = [
f"/api/v1/health/region/{self.current_region}",
f"/api/v1/status/regional"
]
for endpoint in endpoints:
with self.client.get(
endpoint,
name=f"Regional Failover - {self.current_region}",
catch_response=True
) as response:
if response.status_code == 200:
try:
health_data = response.json()
if health_data.get('status') == 'healthy':
print(f"Regional health check passed for {self.current_region}")
response.success()
break # Primary region is healthy
else:
print(f"Regional health degraded for {self.current_region}")
continue # Try backup
except json.JSONDecodeError:
response.success() # Health check might return simple text
break
else:
print(f"Regional endpoint failed: {response.status_code}")
continue # Try next endpoint
def on_stop(self):
"""Report latency statistics for this region"""
if self.latency_measurements:
avg_latency = sum(self.latency_measurements) / len(self.latency_measurements)
min_latency = min(self.latency_measurements)
max_latency = max(self.latency_measurements)
print(f"Region {self.current_region} latency stats:")
print(f" Average: {avg_latency:.1f}ms")
print(f" Min: {min_latency:.1f}ms")
print(f" Max: {max_latency:.1f}ms")
print(f" Expected baseline: {self.profile['expected_latency']}ms")
class RegionalLoadPatternUser(MultiRegionUser):
"""Simulate realistic regional load patterns based on business hours"""
def on_start(self):
super().on_start()
# Adjust request frequency based on regional business hours
current_hour = datetime.now().hour
if current_hour in self.profile['peak_hours']:
self.wait_time = between(0.5, 1.5) # Higher frequency during peak
print(f"Peak hours for {self.current_region} - increased load")
else:
self.wait_time = between(2, 5) # Lower frequency during off-peak
print(f"Off-peak hours for {self.current_region} - reduced load")
@task(4)
def simulate_business_hours_activity(self):
"""Simulate typical business hours activity patterns"""
current_hour = datetime.now().hour
if current_hour in self.profile['peak_hours']:
# Peak hours - more intensive operations
operations = [
'/api/v1/dashboard',
'/api/v1/reports/generate',
'/api/v1/transactions',
'/api/v1/analytics'
]
else:
# Off-peak - lighter operations
operations = [
'/api/v1/status',
'/api/v1/notifications',
'/api/v1/settings'
]
operation = random.choice(operations)
with self.client.get(
operation,
name=f"Business Hours Activity - {self.current_region}",
catch_response=True
) as response:
if response.status_code == 200:
response.success()
else:
response.failure(f"Business hours operation failed: {response.status_code}")
