locust from curl command

curl.txt

put here the cURL command
should work with anything you can copy as cURL in your browser

locurl.py

import re
from pathlib import Path

from locust import HttpUser, task


def parse_curl(curl_cmd: str) -> dict:
    """Parse a curl command and extract URL, method, headers, and body."""
    result = {
        "url": "",
        "method": "GET",
        "headers": {},
        "body": None,
    }

    # Normalize multiline curl command
    normalized = curl_cmd.replace("\\\n", " ").replace("\n", " ")

    # Extract URL (first quoted string after 'curl')
    url_match = re.search(r"curl\s+['\"]([^'\"]+)['\"]", normalized)
    if url_match:
        result["url"] = url_match.group(1)

    # Extract method (-X)
    method_match = re.search(r"-X\s+['\"]?(\w+)['\"]?", normalized)
    if method_match:
        result["method"] = method_match.group(1)

    # Extract headers (-H) - handle single and double quoted separately
    # Single-quoted headers (can contain double quotes inside)
    for header_match in re.finditer(r"-H\s+'([^']+)'", normalized):
        header_line = header_match.group(1)
        colon_index = header_line.find(":")
        if colon_index > 0:
            key = header_line[:colon_index].strip()
            value = header_line[colon_index + 1 :].strip()
            result["headers"][key] = value
    # Double-quoted headers (can contain single quotes inside)
    for header_match in re.finditer(r'-H\s+"([^"]+)"', normalized):
        header_line = header_match.group(1)
        colon_index = header_line.find(":")
        if colon_index > 0:
            key = header_line[:colon_index].strip()
            value = header_line[colon_index + 1 :].strip()
            result["headers"][key] = value

    # Extract body (--data-raw or --data or -d)
    data_match = re.search(r"(?:--data-raw|--data|-d)\s+'((?:[^'\\]|\\.)*)'", normalized)
    if data_match:
        result["body"] = data_match.group(1)
    else:
        # Try double quotes
        data_match = re.search(r'(?:--data-raw|--data|-d)\s+"((?:[^"\\]|\\.)*)"', normalized)
        if data_match:
            result["body"] = data_match.group(1)

    # import pdb
    #
    # pdb.set_trace()
    return result


# Parse curl command at module load time
curl_content = Path("curl.txt").read_text()
config = parse_curl(curl_content)

print("Parsed curl command:")
print(f"  URL: {config['url']}")
print(f"  Method: {config['method']}")
print(f"  Headers: {config['headers']}")
print(f"  Body length: {len(config['body']) if config['body'] else 0} chars")


class CurlUser(HttpUser):
    host = re.match(r"(https?://[^/]+)", config["url"]).group(1) if config["url"] else "http://localhost"
    path = config["url"].replace(host, "") if config["url"] else "/"

    @task
    def execute_curl(self):
        if config["method"].upper() == "GET":
            self.client.get(self.path, headers=config["headers"])
        elif config["method"].upper() == "POST":
            self.client.post(self.path, headers=config["headers"], data=config["body"])
        elif config["method"].upper() == "PUT":
            self.client.put(self.path, headers=config["headers"], data=config["body"])
        elif config["method"].upper() == "DELETE":
            self.client.delete(self.path, headers=config["headers"])
        elif config["method"].upper() == "PATCH":
            self.client.patch(self.path, headers=config["headers"], data=config["body"])