A script to benchmark authenticated Django views

benchmark_views.py

#!/usr/bin/env python3
"""
Django Views Benchmark Script

This script benchmarks Django views by measuring their response times.
It handles authentication via Django's login system and maintains session state.

Usage:
    python benchmarks/benchmark_views.py [--host HOST] [--iterations N] [--warmup-requests N]

Example:
    python benchmarks/benchmark_views.py --host http://127.0.0.1:8000 --iterations 100 --warmup-requests 10
"""

import argparse
import getpass
import statistics
import sys
import time
from urllib.parse import urljoin, urlparse

import requests
from bs4 import BeautifulSoup

# List of paths to benchmark
PATHS = [
    "/en/posts/",
]


class ViewBenchmarker:
    """Handles benchmarking of Django views with authentication."""

    def __init__(self, host):
        self.host = host.rstrip("/")
        self.session = requests.Session()
        self.session.headers.update({"User-Agent": "Django-View-Benchmarker/1.0"})

    def get_csrf_token(self, url):
        """Extract CSRF token from a Django form page."""
        try:
            response = self.session.get(url)
            response.raise_for_status()

            soup = BeautifulSoup(response.content, "html.parser")
            csrf_input = soup.find("input", {"name": "csrfmiddlewaretoken"})

            if csrf_input:
                return csrf_input.get("value")
            else:
                print("Warning: CSRF token not found in form")
                return None

        except requests.RequestException as e:
            print(f"Error getting CSRF token: {e}")
            return None

    def login(self, username, password):
        """Login to Django application."""
        login_url = urljoin(self.host, "/en/login/")

        print(f"Attempting to login at: {login_url}")

        # Get CSRF token
        csrf_token = self.get_csrf_token(login_url)
        if not csrf_token:
            print("Failed to get CSRF token")
            return False

        # Prepare login data
        login_data = {
            "username": username,
            "password": password,
            "csrfmiddlewaretoken": csrf_token,
            "this_is_the_login_form": "1",
            "next": "/",
        }

        # Set CSRF token in headers
        self.session.headers.update(
            {
                "X-CSRFToken": csrf_token,
                "Referer": login_url,
            }
        )

        try:
            response = self.session.post(
                login_url, data=login_data, allow_redirects=False
            )

            # Check if login was successful (redirect indicates success)
            if response.status_code in (302, 303):
                print("Login successful!")
                return True
            else:
                print(f"Login failed. Status code: {response.status_code}")
                if response.status_code == 200:
                    # Parse error messages from the response
                    soup = BeautifulSoup(response.content, "html.parser")
                    error_divs = soup.find_all("div", class_="text-red-600")
                    for error_div in error_divs:
                        error_text = error_div.get_text(strip=True)
                        if error_text:
                            print(f"Error: {error_text}")
                return False

        except requests.RequestException as e:
            print(f"Login request failed: {e}")
            return False

    def benchmark_path(self, path, iterations, warmup_requests, pause):
        """Benchmark a single path."""
        url = urljoin(self.host, path)
        print(f"\nBenchmarking: {url}")

        # Warmup requests
        print(f"Performing {warmup_requests} warmup requests...")
        for i in range(warmup_requests):
            try:
                response = self.session.get(url)
                if response.status_code != 200:
                    print(
                        f"Warning: Warmup request {i + 1} returned status {response.status_code}"
                    )
            except requests.RequestException as e:
                print(f"Warning: Warmup request {i + 1} failed: {e}")

        # Actual benchmark requests
        print(f"Performing {iterations} benchmark requests...")
        response_times = []
        successful_requests = 0

        for i in range(iterations):
            try:
                start_time = time.time()
                response = self.session.get(url)
                end_time = time.time()

                response_time = (
                    end_time - start_time
                ) * 1000  # Convert to milliseconds

                if response.status_code == 200:
                    response_times.append(response_time)
                    successful_requests += 1
                else:
                    print(f"Request {i + 1} failed with status {response.status_code}")

            except requests.RequestException as e:
                print(f"Request {i + 1} failed: {e}")

            # Progress indicator
            if (i + 1) % 10 == 0:
                print(f"  Completed {i + 1}/{iterations} requests")

            # Sleep between requests to bypass rate limiters
            if i < iterations - 1:  # Don't sleep after the last request
                time.sleep(pause)

        return response_times, successful_requests

    def print_statistics(
        self, path, response_times, successful_requests, total_requests
    ):
        """Print benchmark statistics."""
        print(f"\n{'=' * 60}")
        print(f"Results for: {path}")
        print(f"{'=' * 60}")
        print(f"Total requests: {total_requests}")
        print(f"Successful requests: {successful_requests}")
        print(f"Failed requests: {total_requests - successful_requests}")

        if response_times:
            print(f"\nResponse Time Statistics (ms):")
            print(f"  Average: {statistics.mean(response_times):.2f}")
            print(f"  Median:  {statistics.median(response_times):.2f}")
            print(f"  Min:     {min(response_times):.2f}")
            print(f"  Max:     {max(response_times):.2f}")

            if len(response_times) > 1:
                print(f"  Std Dev: {statistics.stdev(response_times):.2f}")

            # Percentiles
            sorted_times = sorted(response_times)
            p95_index = int(0.95 * len(sorted_times))
            p99_index = int(0.99 * len(sorted_times))

            print(f"  95th percentile: {sorted_times[p95_index]:.2f}")
            print(f"  99th percentile: {sorted_times[p99_index]:.2f}")
        else:
            print("\nNo successful requests to analyze.")


def main():
    DEFAULT_HOST = "http://127.0.0.1:8000"

    parser = argparse.ArgumentParser(
        description="Benchmark Django views with authentication",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog=__doc__,
    )

    parser.add_argument(
        "--host",
        default=DEFAULT_HOST,
        help=f"Django server host (default: {DEFAULT_HOST})",
    )

    parser.add_argument(
        "--iterations",
        type=int,
        default=100,
        help="Number of benchmark iterations per path (default: 100)",
    )

    parser.add_argument(
        "--warmup-requests",
        type=int,
        default=10,
        help="Number of warmup requests to ignore (default: 10)",
    )

    args = parser.parse_args()

    # Validate host URL
    parsed_url = urlparse(args.host)
    if not parsed_url.scheme or not parsed_url.netloc:
        print(
            f"Error: Invalid host URL. Please provide a complete URL (e.g., {DEFAULT_HOST})"
        )
        sys.exit(1)

    print(f"Django Views Benchmarker")
    print(f"Host: {args.host}")
    print(f"Iterations: {args.iterations}")
    print(f"Warmup requests: {args.warmup_requests}")
    print(f"Paths to benchmark: {len(PATHS)}")

    # Get credentials
    print("\nPlease provide your login credentials:")
    username = input("Username: ").strip()
    if not username:
        print("Error: Username cannot be empty")
        sys.exit(1)

    password = getpass.getpass("Password: ")
    if not password:
        print("Error: Password cannot be empty")
        sys.exit(1)

    # Initialize benchmarker
    benchmarker = ViewBenchmarker(args.host)

    # Login
    if not benchmarker.login(username, password):
        print("Authentication failed. Exiting.")
        sys.exit(1)

    # Run benchmarks
    print(f"\nStarting benchmark of {len(PATHS)} paths...")

    # Collect results for summary
    summary_results = []

    pause = (
        0
        if args.host
        in ["http://127.0.0.1:8000", "http://localhost:8000", "http://0.0.0.0:8000"]
        else 0.5
    )

    for path in PATHS:
        response_times, successful_requests = benchmarker.benchmark_path(
            path, args.iterations, args.warmup_requests, pause
        )
        benchmarker.print_statistics(
            path, response_times, successful_requests, args.iterations
        )

        # Calculate 95th percentile for summary
        if response_times:
            sorted_times = sorted(response_times)
            p95_index = int(0.95 * len(sorted_times))
            percentile_95th = sorted_times[p95_index]
            url = urljoin(args.host, path)
            summary_results.append((url, percentile_95th))
        else:
            url = urljoin(args.host, path)
            summary_results.append((url, "N/A"))

    print(f"\n{'=' * 60}")
    print("Benchmark completed!")
    print(f"{'=' * 60}")

    # Print summary
    print(f"\nSUMMARY - All paths with 95th percentile response times:")
    print(f"{'=' * 60}")
    for url, percentile_95th in summary_results:
        if percentile_95th == "N/A":
            print(f"{url} | {percentile_95th}")
        else:
            print(f"{url} | {percentile_95th:.2f}")
    print(f"{'=' * 60}")


if __name__ == "__main__":
    main()