Web Application Stress Testing Tool - Performance Testing Utility

web_stress_test.py

#!/usr/bin/env python3
"""
Web Application Stress Testing Tool
Performance testing utility for web applications
"""

import requests
import threading
import time
import random
import os
import sys
import argparse
from datetime import datetime
from collections import deque
import json
import urllib3

# Disable SSL warnings
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

# Auto-install dependencies
def install_requirements():
    """Automatically installs required dependencies"""
    try:
        import requests
    except ImportError:
        print("Installing requests...")
        os.system(f"{sys.executable} -m pip install requests")
        import requests

# Default configuration
DEFAULT_CONFIG = {
    'target_url': "https://example.com",
    'thread_count': 50,
    'test_duration': 3600,  # 1 hour
    'log_interval': 10,
    'request_timeout': 15,
    'payload_size': 2000000,  # 2MB
    'user_agents': [
        'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36',
        'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36',
        'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36',
        'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:89.0) Gecko/20100101 Firefox/89.0',
        'Mozilla/5.0 (Macintosh; Intel Mac OS X 10.15; rv:89.0) Gecko/20100101 Firefox/89.0'
    ]
}

# Global variables
stats = {
    'total': 0,
    'success': 0,
    'failed': 0,
    'times': deque(maxlen=100),
    'server_status': 'Unknown',
    'start_time': None,
    'last_response_time': 0,
    'requests_per_second': 0,
    'avg_response_time': 0,
    'test_running': False
}
lock = threading.Lock()

def log(message, level="INFO"):
    """Simple logging"""
    timestamp = datetime.now().strftime("%H:%M:%S")
    print(f"[{timestamp}] [{level}] {message}")

def get_random_user_agent():
    """Returns random user agent"""
    return random.choice(DEFAULT_CONFIG['user_agents'])

def generate_test_data():
    """Generates test data for stress testing"""
    data = {
        'test_payload': 'X' * DEFAULT_CONFIG['payload_size'],
        'test_type': 'stress_test',
        'test_id': str(random.randint(100000, 999999)),
        'timestamp': str(time.time())
    }
    
    # Add multiple parameters for comprehensive testing
    for i in range(100):
        data[f'param_{i}'] = f'test_value_{i}_{random.randint(1000,9999)}'
    
    return data

def stress_test_request(target_url):
    """Performs stress test request"""
    try:
        test_data = generate_test_data()
        
        headers = {
            'Content-Type': 'application/x-www-form-urlencoded',
            'User-Agent': get_random_user_agent(),
            'Accept': '*/*',
            'Accept-Language': 'en-US,en;q=0.9',
            'Accept-Encoding': 'gzip, deflate',
            'Connection': 'close',
            'Cache-Control': 'no-cache',
            'X-Forwarded-For': f'{random.randint(1,255)}.{random.randint(1,255)}.{random.randint(1,255)}.{random.randint(1,255)}',
            'X-Real-IP': f'{random.randint(1,255)}.{random.randint(1,255)}.{random.randint(1,255)}.{random.randint(1,255)}',
            'X-Originating-IP': f'{random.randint(1,255)}.{random.randint(1,255)}.{random.randint(1,255)}.{random.randint(1,255)}'
        }
        
        start_time = time.time()
        response = requests.post(
            target_url,
            data=test_data,
            headers=headers,
            timeout=DEFAULT_CONFIG['request_timeout'],
            allow_redirects=False,
            verify=False
        )
        response_time = time.time() - start_time
        
        with lock:
            stats['success'] += 1
            stats['total'] += 1
            stats['times'].append(response_time)
            stats['last_response_time'] = response_time
            
            # Determine server status based on response time
            if response_time > 15:
                stats['server_status'] = 'CRITICALLY OVERLOADED'
            elif response_time > 10:
                stats['server_status'] = 'HEAVILY LOADED'
            elif response_time > 5:
                stats['server_status'] = 'OVERLOADED'
            elif response_time > 2:
                stats['server_status'] = 'UNDER LOAD'
            else:
                stats['server_status'] = 'STABLE'
            
        return True
        
    except Exception as e:
        with lock:
            stats['failed'] += 1
            stats['total'] += 1
            
        return False

def memory_stress_test(target_url):
    """Memory stress test"""
    try:
        # Large memory payload
        memory_data = {
            'memory_test': 'M' * (DEFAULT_CONFIG['payload_size'] * 2),
            'test_type': 'memory_stress',
            'test_id': str(random.randint(100000, 999999))
        }
        
        headers = {
            'Content-Type': 'application/x-www-form-urlencoded',
            'User-Agent': get_random_user_agent(),
            'Connection': 'close'
        }
        
        start_time = time.time()
        response = requests.post(
            target_url,
            data=memory_data,
            headers=headers,
            timeout=DEFAULT_CONFIG['request_timeout'],
            verify=False
        )
        response_time = time.time() - start_time
        
        with lock:
            stats['success'] += 1
            stats['total'] += 1
            stats['times'].append(response_time)
            stats['last_response_time'] = response_time
            
        return True
        
    except Exception as e:
        with lock:
            stats['failed'] += 1
            stats['total'] += 1
        return False

def cpu_stress_test(target_url):
    """CPU stress test"""
    try:
        # CPU intensive data
        cpu_data = {}
        for i in range(10000):
            key = f"cpu_test_{i:010d}"
            cpu_data[key] = f"intensive_value_{i}"
        
        headers = {
            'User-Agent': get_random_user_agent(),
            'Content-Type': 'application/x-www-form-urlencoded'
        }
        
        start_time = time.time()
        response = requests.post(
            target_url,
            data=cpu_data,
            headers=headers,
            timeout=DEFAULT_CONFIG['request_timeout'],
            verify=False
        )
        response_time = time.time() - start_time
        
        with lock:
            stats['success'] += 1
            stats['total'] += 1
            stats['times'].append(response_time)
            stats['last_response_time'] = response_time
            
        return True
        
    except Exception as e:
        with lock:
            stats['failed'] += 1
            stats['total'] += 1
        return False

def test_worker(target_url, duration):
    """Worker thread for stress testing"""
    start_time = time.time()
    
    while stats['test_running'] and (time.time() - start_time) < duration:
        # Rotate between different test types
        test_type = random.choice([
            stress_test_request,
            memory_stress_test,
            cpu_stress_test
        ])
        
        test_type(target_url)
        
        # Small delay between requests
        time.sleep(random.uniform(0.01, 0.05))

def calculate_rps():
    """Calculates requests per second"""
    if not stats['start_time']:
        return 0
    
    elapsed = time.time() - stats['start_time']
    if elapsed > 0:
        return stats['total'] / elapsed
    return 0

def get_avg_response_time():
    """Calculates average response time"""
    if stats['times']:
        return sum(stats['times']) / len(stats['times'])
    return 0

def print_stats():
    """Prints statistics to console"""
    if stats['total'] == 0:
        return
    
    success_rate = (stats['success'] / stats['total']) * 100
    rps = calculate_rps()
    avg_time = get_avg_response_time()
    elapsed = time.time() - stats['start_time'] if stats['start_time'] else 0
    
    log("=" * 80)
    log(f"STRESS TEST STATISTICS")
    log(f"Target: {DEFAULT_CONFIG['target_url']}")
    log(f"Runtime: {elapsed:.0f}s")
    log(f"Total requests: {stats['total']:,}")
    log(f"Successful: {stats['success']:,} ({success_rate:.1f}%)")
    log(f"Failed: {stats['failed']:,}")
    log(f"Speed: {rps:.1f} req/sec")
    log(f"Avg response time: {avg_time:.3f}s")
    log(f"Server status: {stats['server_status']}")
    log("=" * 80)

def stats_monitor(log_interval):
    """Statistics monitoring thread"""
    while stats['test_running']:
        time.sleep(log_interval)
        if stats['test_running']:
            print_stats()

def parse_arguments():
    """Parses command line arguments"""
    parser = argparse.ArgumentParser(description='Web Application Stress Testing Tool')
    parser.add_argument('--target', '-t', default=DEFAULT_CONFIG['target_url'], 
                       help='Target URL for stress testing')
    parser.add_argument('--threads', '-th', type=int, default=DEFAULT_CONFIG['thread_count'],
                       help='Number of threads')
    parser.add_argument('--duration', '-d', type=int, default=DEFAULT_CONFIG['test_duration'],
                       help='Test duration in seconds')
    parser.add_argument('--log-interval', '-l', type=int, default=DEFAULT_CONFIG['log_interval'],
                       help='Statistics output interval in seconds')
    parser.add_argument('--silent', '-s', action='store_true',
                       help='Silent mode (minimal logs)')
    
    return parser.parse_args()

def main():
    """Main function"""
    global stats
    
    # Install dependencies
    install_requirements()
    
    # Parse arguments
    args = parse_arguments()
    
    # Update configuration
    DEFAULT_CONFIG['target_url'] = args.target
    DEFAULT_CONFIG['thread_count'] = args.threads
    DEFAULT_CONFIG['test_duration'] = args.duration
    DEFAULT_CONFIG['log_interval'] = args.log_interval
    
    if not args.silent:
        log("STARTING WEB APPLICATION STRESS TEST")
        log(f"Target: {DEFAULT_CONFIG['target_url']}")
        log(f"Threads: {DEFAULT_CONFIG['thread_count']}")
        log(f"Duration: {DEFAULT_CONFIG['test_duration']} seconds")
        log("Test types: Memory, CPU, and General stress tests")
    
    stats['start_time'] = time.time()
    stats['test_running'] = True
    
    # Start worker threads
    threads = []
    for i in range(DEFAULT_CONFIG['thread_count']):
        thread = threading.Thread(
            target=test_worker, 
            args=(DEFAULT_CONFIG['target_url'], DEFAULT_CONFIG['test_duration'])
        )
        thread.daemon = True
        threads.append(thread)
        thread.start()
    
    # Start statistics monitoring
    if not args.silent:
        stats_thread = threading.Thread(
            target=stats_monitor, 
            args=(DEFAULT_CONFIG['log_interval'],)
        )
        stats_thread.daemon = True
        stats_thread.start()
    
    # Wait for test completion
    try:
        start_time = time.time()
        while stats['test_running'] and (time.time() - start_time) < DEFAULT_CONFIG['test_duration']:
            time.sleep(1)
            
        stats['test_running'] = False
        
        if not args.silent:
            log("STRESS TEST COMPLETED")
            print_stats()
            
            # Final assessment
            avg_time = get_avg_response_time()
            if avg_time > 15.0:
                log("SERVER CRITICALLY OVERLOADED")
            elif avg_time > 10.0:
                log("SERVER HEAVILY LOADED")
            elif avg_time > 5.0:
                log("SERVER SIGNIFICANTLY SLOWED")
            elif avg_time > 2.0:
                log("SERVER UNDER HEAVY LOAD")
            else:
                log("SERVER HANDLED LOAD WELL")
        
        return 0
        
    except KeyboardInterrupt:
        stats['test_running'] = False
        if not args.silent:
            log("TEST STOPPED BY USER")
        return 0

if __name__ == "__main__":
    sys.exit(main())