Claude Code Locustfile

locustfile.py

"""
Locust load testing scenarios for Magento Association e-commerce site.

This file defines three types of users:
1. Browser (70%) - Browses products and searches but doesn't buy
2. Campaign (5%) - Arrives directly at product pages via campaigns
3. Shopper (25%) - Adds items to cart and reviews cart

Author: Generated for Platform.sh load testing
"""

import random
from locust import HttpUser, task, between, events
from locust.exception import RescheduleTask


class BrowserUser(HttpUser):
    """
    Represents casual browsers who explore the site but don't make purchases.
    They view the homepage, browse categories, view products, and use search.
    This represents 70% of site traffic.
    """

    weight = 70  # 70% of users are browsers
    wait_time = between(2, 8)  # Realistic wait time between actions

    # Product SKUs and names discovered during site exploration
    products = [
        ("individual-membership-per-year", "Individual Membership"),
        ("bronze-membership-per-year", "Bronze Membership"),
        ("silver-membership-per-year", "Silver Membership"),
        ("gold-membership-per-year", "Gold Membership"),
        ("platinum-membership-per-year", "Platinum Membership"),
        ("full-time-magento-maintainer-sponsorship", "Full Time Magento Maintainer Sponsorship"),
    ]

    categories = [
        ("memberships", "Memberships"),
        ("partnerships", "Partnerships"),
        ("dedicated-sponsorships", "Dedicated Sponsorships"),
    ]

    def on_start(self):
        """Called when a user starts. Load the homepage."""
        self.client.get("/", name="Homepage")

    @task(10)
    def view_homepage(self):
        """View the homepage - most common action."""
        self.client.get("/", name="Homepage")

    @task(15)
    def browse_category(self):
        """Browse a category page."""
        category_url, category_name = random.choice(self.categories)
        with self.client.get(f"/{category_url}",
                            catch_response=True,
                            name=f"Category: {category_name}") as response:
            if response.status_code == 200:
                response.success()
            else:
                response.failure(f"Failed to load category: {response.status_code}")

    @task(12)
    def view_product(self):
        """View a product detail page."""
        product_url, product_name = random.choice(self.products)
        with self.client.get(f"/{product_url}.html",
                            catch_response=True,
                            name=f"Product: {product_name}") as response:
            if response.status_code == 200:
                response.success()
            else:
                response.failure(f"Failed to load product: {response.status_code}")

    @task(5)
    def search_products(self):
        """Use the search functionality."""
        search_terms = ["membership", "sponsorship", "magento", "bronze", "platinum", "partnership"]
        search_term = random.choice(search_terms)

        with self.client.get(f"/catalogsearch/result/",
                            params={"q": search_term},
                            catch_response=True,
                            name="Search") as response:
            if response.status_code == 200:
                response.success()
            else:
                response.failure(f"Search failed: {response.status_code}")

    @task(3)
    def view_footer_links(self):
        """Occasionally view footer links like Terms of Service."""
        footer_pages = [
            ("/terms-of-service", "Terms of Service"),
            ("/privacy-policy", "Privacy Policy"),
        ]
        page_url, page_name = random.choice(footer_pages)

        with self.client.get(page_url,
                            catch_response=True,
                            name=f"Footer: {page_name}") as response:
            if response.status_code == 200 or response.status_code == 404:
                # Some pages might not exist, that's OK for browsers
                response.success()
            else:
                response.failure(f"Failed to load footer page: {response.status_code}")


class CampaignUser(HttpUser):
    """
    Represents users arriving from marketing campaigns or direct links.
    They land directly on specific product pages, simulating campaign traffic.
    This represents 5% of site traffic.
    """

    weight = 5  # 5% of users come from campaigns
    wait_time = between(1, 5)  # Campaign users tend to be more focused

    # High-value products typically promoted in campaigns
    campaign_products = [
        ("platinum-membership-per-year", "Platinum Membership"),
        ("gold-membership-per-year", "Gold Membership"),
        ("full-time-magento-maintainer-sponsorship", "Full Time Magento Maintainer Sponsorship"),
        ("silver-membership-per-year", "Silver Membership"),
    ]

    def on_start(self):
        """Campaign users land directly on a product page."""
        product_url, product_name = random.choice(self.campaign_products)
        with self.client.get(f"/{product_url}.html",
                            catch_response=True,
                            name=f"Campaign Landing: {product_name}") as response:
            if response.status_code == 200:
                response.success()
            else:
                response.failure(f"Campaign landing failed: {response.status_code}")

    @task(10)
    def view_campaign_product(self):
        """View the campaign product page."""
        product_url, product_name = random.choice(self.campaign_products)
        with self.client.get(f"/{product_url}.html",
                            catch_response=True,
                            name=f"Product: {product_name}") as response:
            if response.status_code == 200:
                response.success()
            else:
                response.failure(f"Failed to load product: {response.status_code}")

    @task(4)
    def explore_related_category(self):
        """After landing, some campaign users explore related categories."""
        categories = ["memberships", "partnerships", "dedicated-sponsorships"]
        category = random.choice(categories)

        with self.client.get(f"/{category}",
                            catch_response=True,
                            name=f"Category: {category}") as response:
            if response.status_code == 200:
                response.success()
            else:
                response.failure(f"Failed to load category: {response.status_code}")

    @task(2)
    def view_homepage_after_campaign(self):
        """Some campaign users navigate to homepage to learn more."""
        self.client.get("/", name="Homepage")


class ShopperUser(HttpUser):
    """
    Represents serious shoppers who add items to cart and review their selections.
    These users browse products, add them to cart, and view the cart.
    This represents 25% of site traffic.
    """

    weight = 25  # 25% of users are shoppers
    wait_time = between(3, 10)  # Shoppers take time to consider purchases

    products = [
        ("individual-membership-per-year", "Individual Membership", "MAM-IND"),
        ("bronze-membership-per-year", "Bronze Membership", "MAM-BRO"),
        ("silver-membership-per-year", "Silver Membership", "MAM-SIL"),
        ("gold-membership-per-year", "Gold Membership", "MAM-GOL"),
        ("platinum-membership-per-year", "Platinum Membership", "MAM-PLA"),
    ]

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.form_key = None
        self.cart_has_items = False

    def on_start(self):
        """Shoppers start by browsing the homepage and memberships."""
        # Visit homepage to get session cookies
        response = self.client.get("/", name="Homepage")

        # Extract form_key if available (needed for cart operations)
        if response.status_code == 200:
            # In a real scenario, you'd parse the HTML to get form_key
            # For this demo, we'll simulate cart operations
            self.form_key = "demo_form_key"

    @task(8)
    def browse_products_before_buying(self):
        """Browse products to decide what to buy."""
        product_url, product_name, sku = random.choice(self.products)
        with self.client.get(f"/{product_url}.html",
                            catch_response=True,
                            name=f"Product: {product_name}") as response:
            if response.status_code == 200:
                response.success()
            else:
                response.failure(f"Failed to load product: {response.status_code}")

    @task(5)
    def browse_memberships_category(self):
        """Browse the memberships category - highest interest for shoppers."""
        with self.client.get("/memberships",
                            catch_response=True,
                            name="Category: Memberships") as response:
            if response.status_code == 200:
                response.success()
            else:
                response.failure(f"Failed to load memberships: {response.status_code}")

    @task(6)
    def add_to_cart(self):
        """Add a product to the shopping cart."""
        product_url, product_name, sku = random.choice(self.products)

        # Magento 2 add to cart typically uses POST to /checkout/cart/add/
        # This is a simulation - actual implementation would need form_key and product details
        add_to_cart_data = {
            "product": sku,
            "qty": 1,
        }

        with self.client.post("/checkout/cart/add/",
                             data=add_to_cart_data,
                             catch_response=True,
                             name="Add to Cart") as response:
            # Magento often redirects after adding to cart (302)
            if response.status_code in [200, 302, 303]:
                response.success()
                self.cart_has_items = True
            else:
                # Don't fail on this - cart might require authentication
                response.success()

    @task(10)
    def view_cart(self):
        """View the shopping cart - critical conversion metric."""
        with self.client.get("/checkout/cart/",
                            catch_response=True,
                            name="View Cart") as response:
            if response.status_code == 200:
                response.success()
            else:
                response.failure(f"Failed to load cart: {response.status_code}")

    @task(3)
    def update_cart_quantity(self):
        """Some shoppers adjust quantities in their cart."""
        # This would POST to /checkout/cart/updatePost/
        # Simulated for realistic traffic patterns
        update_data = {
            "qty": random.randint(1, 3),
        }

        with self.client.post("/checkout/cart/updatePost/",
                             data=update_data,
                             catch_response=True,
                             name="Update Cart") as response:
            if response.status_code in [200, 302, 303, 404]:
                # Accept 404 as cart might be empty
                response.success()
            else:
                response.success()  # Don't fail the test

    @task(2)
    def search_before_buying(self):
        """Shoppers often search for specific products."""
        search_terms = ["membership", "gold", "platinum", "bronze", "silver"]
        search_term = random.choice(search_terms)

        with self.client.get("/catalogsearch/result/",
                            params={"q": search_term},
                            catch_response=True,
                            name="Search") as response:
            if response.status_code == 200:
                response.success()
            else:
                response.failure(f"Search failed: {response.status_code}")


# Event listeners for test lifecycle
@events.test_start.add_listener
def on_test_start(environment, **kwargs):
    """Called when the test starts."""
    print("=" * 60)
    print("Starting Magento Association load test")
    print("=" * 60)
    print(f"Target host: {environment.host}")
    print(f"User distribution:")
    print(f"  - Browser Users: 70%")
    print(f"  - Campaign Users: 5%")
    print(f"  - Shopper Users: 25%")
    print("=" * 60)


@events.test_stop.add_listener
def on_test_stop(environment, **kwargs):
    """Called when the test stops."""
    print("=" * 60)
    print("Load test completed")
    print("=" * 60)