Class IntegratedBenchmark

benchmarks/dirty_benchmark.py:370–522

Run benchmarks against gunicorn with dirty pool via HTTP. Uses wrk or ab for load testing, or falls back to Python requests.

class IntegratedBenchmark:
    """
    Run benchmarks against gunicorn with dirty pool via HTTP.

    Uses wrk or ab for load testing, or falls back to Python requests.
    """

    def __init__(
        self,
        url: str = "http://127.0.0.1:8000",
        verbose: bool = False,
    ):
        self.url = url.rstrip('/')
        self.verbose = verbose
        self._tool = None

    def check_dependencies(self) -> str | None:
        """Check for available load testing tools."""
        for tool in ['wrk', 'ab']:
            try:
                subprocess.run([tool, '--version'], capture_output=True,
                               check=False)
                return tool
            except FileNotFoundError:
                continue
        return None

    def warmup(self, requests: int = 10):
        """Warm up the server."""
        import urllib.request
        for _ in range(requests):
            try:
                urllib.request.urlopen(f"{self.url}/health", timeout=5)
            except Exception:
                pass

    def run_wrk(
        self,
        path: str,
        duration: int = 10,
        threads: int = 4,
        connections: int = 100,
    ) -> dict:
        """Run wrk benchmark and parse results."""
        url = f"{self.url}{path}"
        cmd = [
            'wrk',
            '-t', str(threads),
            '-c', str(connections),
            '-d', f'{duration}s',
            '--latency',
            url,
        ]

        result = subprocess.run(cmd, capture_output=True, text=True,
                                check=False)
        return self._parse_wrk_output(result.stdout)