Build an AI Load Test Scenario Generator with Claude API

pip install anthropic pyyaml
# k6 CLI installed separately: https://k6.io/docs/get-started/installation/

import anthropic
import json
from collections import Counter
 
client = anthropic.Anthropic()
 
 
def analyze_access_logs(log_sample: list[dict]) -> dict:
    """Extract realistic traffic patterns from sampled production access logs."""
    endpoints = Counter(f"{log['method']} {log['path']}" for log in log_sample)
    total = sum(endpoints.values())
 
    return {
        "endpoint_distribution": {ep: round(count / total * 100, 1) for ep, count in endpoints.most_common(20)},
        "avg_requests_per_session": estimate_session_length(log_sample),
        "peak_to_average_ratio": calculate_peak_ratio(log_sample),
    }
 
 
def estimate_session_length(log_sample: list[dict]) -> float:
    sessions = Counter(log.get("session_id") for log in log_sample if log.get("session_id"))
    return sum(sessions.values()) / len(sessions) if sessions else 1.0

GENERATE_PROMPT = """Generate a realistic k6 load test script based on this
production traffic pattern data.
 
OpenAPI spec (relevant endpoints):
{openapi_spec}
 
Real traffic distribution (% of requests per endpoint):
{endpoint_distribution}
 
Average requests per user session: {avg_session_length}
Peak-to-average traffic ratio: {peak_ratio}
 
Requirements:
- Model realistic user sessions (login -> browse -> action -> logout pattern),
  not independent random requests to each endpoint
- Weight virtual user behavior to match the real endpoint distribution
- Include a ramp-up stage, sustained peak stage (using the peak ratio), and ramp-down
- Add realistic think-time between requests (1-3s), not zero-delay hammering
- Include response time and error rate thresholds as pass/fail criteria
 
Generate a complete, runnable k6 script."""
 
 
def generate_load_test(openapi_spec: dict, traffic_patterns: dict) -> str:
    response = client.messages.create(
        model="claude-sonnet-5",
        max_tokens=3000,
        messages=[{
            "role": "user",
            "content": GENERATE_PROMPT.format(
                openapi_spec=json.dumps(openapi_spec, indent=2)[:3000],
                endpoint_distribution=traffic_patterns["endpoint_distribution"],
                avg_session_length=traffic_patterns["avg_requests_per_session"],
                peak_ratio=traffic_patterns["peak_to_average_ratio"],
            )
        }]
    )
    text = response.content[0].text
    if "```javascript" in text or "```js" in text:
        text = text.split("```")[1]
        text = text.replace("javascript", "", 1).replace("js", "", 1).strip()
    return text

import http from 'k6/http';
import { check, sleep } from 'k6';
 
export const options = {
  stages: [
    { duration: '2m', target: 50 },   // ramp-up
    { duration: '10m', target: 200 }, // sustained peak (4x average, from real ratio)
    { duration: '2m', target: 0 },    // ramp-down
  ],
  thresholds: {
    http_req_duration: ['p(95)<500'],
    http_req_failed: ['rate<0.01'],
  },
};
 
export default function () {
  // Weighted to match real traffic: 45% browse, 30% search, 15% checkout, 10% account
  const rand = Math.random();
 
  http.get('https://api.myapp.com/health');    // session start
  sleep(1);
 
  if (rand < 0.45) {
    http.get('https://api.myapp.com/products?page=1');
    sleep(2);
    http.get(`https://api.myapp.com/products/${randomProductId()}`);
  } else if (rand < 0.75) {
    http.get(`https://api.myapp.com/search?q=${randomSearchTerm()}`);
  } else if (rand < 0.90) {
    const cartRes = http.post('https://api.myapp.com/cart', JSON.stringify({ item_id: randomProductId() }));
    check(cartRes, { 'add to cart succeeds': (r) => r.status === 200 });
    sleep(1);
    http.post('https://api.myapp.com/checkout');
  } else {
    http.get('https://api.myapp.com/account');
  }
 
  sleep(Math.random() * 2 + 1);    // realistic think-time
}

# .github/workflows/load-test.yml
name: Weekly Load Test
on:
  schedule:
    - cron: "0 3 * * 0"    # Weekly, low-traffic window
  workflow_dispatch:
 
jobs:
  load-test:
    runs-on: ubuntu-latest
    steps:
      - uses: grafana/k6-action@v0.3.1
        with:
          filename: generated-load-test.js
      - name: Fail if thresholds not met
        run: echo "k6 exits non-zero automatically on threshold failure"