locustfile.py

"""
Locust Load Testing Configuration for Online Bookstore
Simulates realistic user behavior patterns for performance testing

Installation:
    pip install locust

Usage:
    # Start with web UI
    locust -f locustfile.py --host=http://localhost:5000
    
    # Headless mode (no UI)
    locust -f locustfile.py --host=http://localhost:5000 --users 100 --spawn-rate 10 --run-time 1m --headless
    
    # With custom settings
    locust -f locustfile.py --host=http://localhost:5000 --users 50 --spawn-rate 5 --run-time 5m

Access Web UI:
    http://localhost:8089
"""

from locust import HttpUser, task, between, tag, events
import random
import json
from datetime import datetime


# Sample data for testing
BOOK_TITLES = [
    "The Great Gatsby",
    "1984",
    "Moby Dick",
    "I Ching"
]

DISCOUNT_CODES = [
    "SAVE10",
    "WELCOME20",
    "save10",  # Test case sensitivity
    "INVALID"  # Test invalid code
]

VALID_EMAILS = [
    "user1@test.com",
    "user2@test.com",
    "user3@test.com",
    "loadtest@bookstore.com",
    "performance@test.com"
]

PAYMENT_CARDS = {
    "valid": "4532123456789012",
    "failed": "4532123456781111"  # Ends in 1111 - mock failure
}


class BookstoreUser(HttpUser):
    """
    Simulates a typical user browsing and purchasing from the bookstore
    
    Wait time: Between 1-3 seconds between tasks (simulates reading/thinking time)
    """
    wait_time = between(1, 3)
    
    def on_start(self):
        """
        Called when a user starts (simulates user session initialization)
        Initialize user-specific data
        """
        self.email = random.choice(VALID_EMAILS)
        self.password = "password123"
        self.name = f"LoadTest User {random.randint(1, 1000)}"
        self.cart_items = []
        
        # Optional: Register and login
        # Uncomment if you want users to register/login
        # self.register_user()
        # self.login_user()
    
    def on_stop(self):
        """Called when a user stops (cleanup)"""
        # Optional: Logout
        pass
    
    # ==================== BROWSING TASKS ====================
    
    @task(10)
    @tag('browsing', 'critical')
    def view_homepage(self):
        """
        View homepage with book catalog
        Weight: 10 (most frequent action)
        """
        with self.client.get("/", catch_response=True, name="GET /homepage") as response:
            if response.status_code == 200 and b"Bookstore" in response.content:
                response.success()
            else:
                response.failure("Homepage did not load correctly")
    
    @task(3)
    @tag('browsing')
    def view_cart(self):
        """
        View shopping cart
        Weight: 3 (moderate frequency)
        """
        with self.client.get("/cart", catch_response=True, name="GET /cart") as response:
            if response.status_code == 200:
                response.success()
            else:
                response.failure("Cart page did not load")
    
    # ==================== CART MANAGEMENT TASKS ====================
    
    @task(5)
    @tag('cart', 'critical')
    def add_to_cart(self):
        """
        Add a random book to cart
        Weight: 5 (frequent action)
        """
        book_title = random.choice(BOOK_TITLES)
        quantity = random.randint(1, 5)
        
        data = {
            'title': book_title,
            'quantity': str(quantity)
        }
        
        with self.client.post("/add-to-cart", data=data, catch_response=True, name="POST /add-to-cart") as response:
            if response.status_code in [200, 302]:
                self.cart_items.append(book_title)
                response.success()
            else:
                response.failure(f"Failed to add {book_title} to cart")
    
    @task(2)
    @tag('cart')
    def update_cart_quantity(self):
        """
        Update quantity of an item in cart
        Weight: 2 (less frequent)
        """
        if self.cart_items:
            book_title = random.choice(self.cart_items)
            new_quantity = random.randint(1, 10)
            
            data = {
                'title': book_title,
                'quantity': str(new_quantity)
            }
            
            with self.client.post("/update-cart", data=data, catch_response=True, name="POST /update-cart") as response:
                if response.status_code in [200, 302]:
                    response.success()
                else:
                    response.failure("Failed to update cart")
    
    @task(1)
    @tag('cart')
    def remove_from_cart(self):
        """
        Remove an item from cart
        Weight: 1 (occasional action)
        """
        if self.cart_items:
            book_title = random.choice(self.cart_items)
            
            data = {'title': book_title}
            
            with self.client.post("/remove-from-cart", data=data, catch_response=True, name="POST /remove-from-cart") as response:
                if response.status_code in [200, 302]:
                    self.cart_items.remove(book_title)
                    response.success()
                else:
                    response.failure("Failed to remove from cart")
    
    @task(1)
    @tag('cart')
    def clear_cart(self):
        """
        Clear entire cart
        Weight: 1 (rare action)
        """
        with self.client.post("/clear-cart", catch_response=True, name="POST /clear-cart") as response:
            if response.status_code in [200, 302]:
                self.cart_items.clear()
                response.success()
            else:
                response.failure("Failed to clear cart")
    
    # ==================== CHECKOUT TASKS ====================
    
    @task(2)
    @tag('checkout', 'critical')
    def view_checkout(self):
        """
        View checkout page
        Weight: 2 (moderate frequency)
        """
        with self.client.get("/checkout", catch_response=True, name="GET /checkout") as response:
            if response.status_code in [200, 302]:
                response.success()
            else:
                response.failure("Checkout page failed")
    
    @task(1)
    @tag('checkout', 'critical', 'transaction')
    def complete_checkout_success(self):
        """
        Complete a successful checkout process
        Weight: 1 (critical but less frequent)
        """
        # First add items to cart
        for _ in range(random.randint(1, 3)):
            book_title = random.choice(BOOK_TITLES)
            self.client.post("/add-to-cart", data={
                'title': book_title,
                'quantity': str(random.randint(1, 3))
            })
        
        # Process checkout
        discount_code = random.choice(DISCOUNT_CODES) if random.random() > 0.5 else ""
        
        checkout_data = {
            'name': self.name,
            'email': self.email,
            'address': f'{random.randint(1, 999)} Test St',
            'city': 'Test City',
            'zip_code': str(random.randint(10000, 99999)),
            'payment_method': 'credit_card',
            'card_number': PAYMENT_CARDS['valid'],
            'expiry_date': '12/25',
            'cvv': '123',
            'discount_code': discount_code
        }
        
        with self.client.post("/process-checkout", data=checkout_data, catch_response=True, name="POST /process-checkout [Success]") as response:
            if response.status_code in [200, 302]:
                self.cart_items.clear()
                response.success()
            else:
                response.failure("Checkout failed")
    
    @task(1)
    @tag('checkout', 'error-handling')
    def complete_checkout_failed_payment(self):
        """
        Test checkout with failed payment
        Weight: 1 (test error handling)
        """
        # Add item to cart
        book_title = random.choice(BOOK_TITLES)
        self.client.post("/add-to-cart", data={
            'title': book_title,
            'quantity': '1'
        })
        
        # Attempt checkout with failing card
        checkout_data = {
            'name': self.name,
            'email': self.email,
            'address': '123 Test St',
            'city': 'Test City',
            'zip_code': '12345',
            'payment_method': 'credit_card',
            'card_number': PAYMENT_CARDS['failed'],  # This will fail
            'expiry_date': '12/25',
            'cvv': '123'
        }
        
        with self.client.post("/process-checkout", data=checkout_data, catch_response=True, name="POST /process-checkout [Failed Payment]") as response:
            if response.status_code in [200, 302] and b'Payment failed' in response.content:
                response.success()
            else:
                response.failure("Failed payment not handled correctly")
    
    # ==================== AUTHENTICATION TASKS ====================
    
    @task(1)
    @tag('auth')
    def register_user(self):
        """
        Register a new user account
        Weight: 1 (occasional action)
        """
        unique_email = f"loadtest_{random.randint(1, 999999)}@test.com"
        
        register_data = {
            'email': unique_email,
            'password': 'password123',
            'name': f'LoadTest User {random.randint(1, 9999)}',
            'address': f'{random.randint(1, 999)} Test Street'
        }
        
        with self.client.post("/register", data=register_data, catch_response=True, name="POST /register") as response:
            if response.status_code in [200, 302]:
                response.success()
            else:
                response.failure("Registration failed")
    
    @task(1)
    @tag('auth')
    def login_user(self):
        """
        Login with demo account
        Weight: 1 (occasional action)
        """
        login_data = {
            'email': 'demo@bookstore.com',
            'password': 'demo123'
        }
        
        with self.client.post("/login", data=login_data, catch_response=True, name="POST /login") as response:
            if response.status_code in [200, 302]:
                response.success()
            else:
                response.failure("Login failed")
    
    @task(1)
    @tag('auth')
    def view_account(self):
        """
        View account page
        Weight: 1 (occasional action)
        """
        with self.client.get("/account", catch_response=True, name="GET /account") as response:
            if response.status_code in [200, 302]:
                response.success()
            else:
                response.failure("Account page failed")
    
    # ==================== EDGE CASE TESTS ====================
    
    @task(1)
    @tag('edge-case', 'error-handling')
    def test_invalid_quantity(self):
        """
        Test adding item with invalid quantity (error handling)
        Weight: 1 (test robustness)
        """
        invalid_quantities = ['abc', '-1', '0', '!@#$', '']
        
        data = {
            'title': random.choice(BOOK_TITLES),
            'quantity': random.choice(invalid_quantities)
        }
        
        with self.client.post("/add-to-cart", data=data, catch_response=True, name="POST /add-to-cart [Invalid Qty]") as response:
            # Should handle gracefully, not crash
            if response.status_code in [200, 302, 400]:
                response.success()
            else:
                response.failure("Invalid quantity not handled")
    
    @task(1)
    @tag('edge-case')
    def test_nonexistent_book(self):
        """
        Test adding non-existent book
        Weight: 1 (test validation)
        """
        data = {
            'title': 'Nonexistent Book Title 12345',
            'quantity': '1'
        }
        
        with self.client.post("/add-to-cart", data=data, catch_response=True, name="POST /add-to-cart [Invalid Book]") as response:
            if response.status_code in [200, 302, 404]:
                response.success()
            else:
                response.failure("Nonexistent book not handled")


class BrowsingUser(HttpUser):
    """
    User that only browses without purchasing
    Simulates window shoppers
    """
    wait_time = between(2, 5)
    
    @task(10)
    @tag('browsing')
    def browse_homepage(self):
        self.client.get("/")
    
    @task(3)
    @tag('browsing')
    def view_cart_empty(self):
        self.client.get("/cart")


class PowerUser(HttpUser):
    """
    Aggressive user that makes many rapid requests
    Tests system under stress
    """
    wait_time = between(0.5, 1)
    
    @task(5)
    def rapid_browsing(self):
        self.client.get("/")
    
    @task(3)
    def rapid_cart_actions(self):
        book = random.choice(BOOK_TITLES)
        self.client.post("/add-to-cart", data={
            'title': book,
            'quantity': str(random.randint(1, 5))
        })


# ==================== CUSTOM EVENTS & REPORTING ====================

@events.test_start.add_listener
def on_test_start(environment, **kwargs):
    """
    Called when load test starts
    """
    print("\n" + "="*60)
    print("🚀 LOAD TEST STARTED")
    print("="*60)
    print(f"Target Host: {environment.host}")
    print(f"Start Time: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    print("="*60 + "\n")


@events.test_stop.add_listener
def on_test_stop(environment, **kwargs):
    """
    Called when load test stops
    """
    print("\n" + "="*60)
    print("🏁 LOAD TEST COMPLETED")
    print("="*60)
    print(f"End Time: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    print("="*60 + "\n")


# ==================== CUSTOM TASKS FOR SPECIFIC SCENARIOS ====================

class CheckoutOnlyUser(HttpUser):
    """
    User focused on checkout process
    Good for testing payment gateway performance
    """
    wait_time = between(2, 4)
    
    @task(1)
    @tag('checkout', 'stress')
    def quick_purchase(self):
        """Complete purchase quickly"""
        # Add to cart
        self.client.post("/add-to-cart", data={
            'title': random.choice(BOOK_TITLES),
            'quantity': '1'
        })
        
        # Checkout
        self.client.post("/process-checkout", data={
            'name': 'Speed Tester',
            'email': f'speed{random.randint(1,9999)}@test.com',
            'address': '123 Speed St',
            'city': 'Fast City',
            'zip_code': '12345',
            'payment_method': 'credit_card',
            'card_number': PAYMENT_CARDS['valid'],
            'expiry_date': '12/25',
            'cvv': '123'
        })


# ==================== USAGE EXAMPLES ====================

"""
EXAMPLE COMMANDS:

1. Basic load test with web UI:
   locust -f locustfile.py --host=http://localhost:5000

2. Headless mode (100 users, spawn 10/second, run 5 minutes):
   locust -f locustfile.py --host=http://localhost:5000 --users 100 --spawn-rate 10 --run-time 5m --headless

3. Stress test (500 users, aggressive spawn):
   locust -f locustfile.py --host=http://localhost:5000 --users 500 --spawn-rate 50 --run-time 10m --headless

4. Test specific tags only (critical paths):
   locust -f locustfile.py --host=http://localhost:5000 --tags critical

5. Generate HTML report:
   locust -f locustfile.py --host=http://localhost:5000 --users 100 --spawn-rate 10 --run-time 2m --headless --html report.html

6. Run with specific user class:
   locust -f locustfile.py --host=http://localhost:5000 BookstoreUser

7. Distributed load testing (master):
   locust -f locustfile.py --host=http://localhost:5000 --master

8. Distributed load testing (worker):
   locust -f locustfile.py --host=http://localhost:5000 --worker --master-host=<master-ip>

PERFORMANCE TARGETS:

- Homepage load time: < 200ms (50th percentile)
- Cart operations: < 300ms (50th percentile)
- Checkout process: < 1000ms (50th percentile)
- Error rate: < 1%
- Success rate: > 99%

METRICS TO MONITOR:

- Response times (min, max, median, 95th percentile)
- Requests per second (RPS)
- Failure rate
- Current users
- Response time distribution
"""