Load Test

Customization

Customization of Load Test Scenarios

This guide explains how to customize load testing scenarios using the Skyramp worker and library functions. Skyramp already generates load tests out of the box through the Skyramp CLI and agent; however, we understand that additional customization is required in many cases. We'll demonstrate using Skyramp’s Demo Shop API, a simple e-commerce API for product and order management. Learn more about the Demo Shop API.

Refer to the Installation Guide if you haven't installed Skyramp yet.

Load Test Structure

Here is an example of a simple load test in Python:

# Generated by Skyramp v1.2.1 on 2025-06-26 15:49:15.088816 -0700 PDT m=+2.872724918
# Command: skyramp generate load rest https://demoshop.skyramp.dev/api/v1/products \
# 		--api-schema https://demoshop.skyramp.dev/openapi.json \
# 		--framework pytest \
# 		--language python \
# 		--method POST \
#       --runtime docker

# Import of required libraries
import skyramp
import os
import time
# URL for test requests
URL = "https://demoshop.skyramp.dev"

load_config = skyramp.LoadTestConfig(
    load_duration=5,
    load_num_threads=1,
    load_target_rps=None,
    load_count=None,
    load_rampup_duration=None,
    load_rampup_interval=None
)
def test_products_post(load_test_config:skyramp.LoadTestConfig=load_config):
    # Invocation of Skyramp Client
    client = skyramp.Client(
        runtime="docker",
        docker_network="skyramp",
        docker_skyramp_port=35142
    )
    # Definition of authentication header
    headers = {}
    if os.getenv("SKYRAMP_TEST_TOKEN") is not None:
        headers["Authorization"] = "Bearer " + os.getenv("SKYRAMP_TEST_TOKEN")

    scenario = skyramp.AsyncScenario(name="scenario")
    # Request Body
    products_POST_request_body = r'''{
            "category": "Toys",
            "description": "Bear Soft Toy",
            "image_url": "https://images.app.goo.gl/cgcHpeehRdu5osot8",
            "in_stock": true,
            "name": "bigbear",
            "price": 9.99
        }'''
    
    # Add Request to Scenario
    products_POST_response = scenario.add_async_request(
        name="products_POST",
        url=URL,
        path="/api/v1/products",
        method="POST",
        body=products_POST_request_body,
        headers=headers,
        expected_code="201"
    )
    result = client.send_scenario(
        scenario,
        load_test_config=load_test_config
    )

    print(
        f"result: {result.get_overall_status()}"
    )


if __name__ == "__main__":
    test_products_post()


Load Test File Anatomy & Data Structure

At a high level, a Skyramp load test works as follows:

  1. Skyramp creates a SkyrampClient with the runtime specified.

    1. For this tutorial, we will use Docker as the runtime. You can also execute the test locally or in Kubernetes.

  2. A new AsyncScenario is then created. A scenario is a set of steps defined to execute for a test. A step can be:

    • an API request represented by an AsyncRequest (“create/read/update/delete this resource”)

    • a test assertion (“assert that expected value matches actual value”)

    • a sub-scenario

  3. The SkyrampClient then sends the scenario over to the Skyramp worker for execution and gets an AsyncTestStatus in return.

    1. A worker in this context is a component or process responsible for executing the test scenario. When you run a load test using the Docker runtime, Skyramp automatically spins up a specialized worker that executes the load test scenario.

      1. Note: For local execution, the Skyramp Library acts as the “worker” for the load test.

    2. AsyncTestStatus is a wrapper for fetching the status of a test run (including details on load test performance + request execution results).

    3. AsyncScenarioStatus - a wrapper for fetching the status of a specific scenario

    4. AsyncRequestStatus - a wrapper for fetching the status of a specific request

High-Level Diagram of Data Hierarchy

Why do load tests look different from other test types?

Load tests are run asynchronously. This means that all requests are packaged in a scenario, sent to the worker, and executed without any individual responses being communicated back to the program runtime.

Why would I want to execute a test async?

For load testing scenarios with multiple chained API calls and simulations of high volume traffic, async test execution enables significantly higher test performance and allows for the simulation of more realistic load on the system by enabling concurrency and targeted/maximized requests per second.

Customize Skyramp Load Test Scenarios

The Skyramp Library provides useful classes and functions that can help you customize your test scenarios.

Create Scenarios and Requests

At a very high level, scenario setup requires two key steps:

  1. Create an AsyncScenario object.

  2. Add an async request or async scenario to the AsyncScenario object using add_async_request or add_async_scenario.

Functions

AsyncScenario

  • add_async_request - adds an AsyncRequest as a step in the scenario.

    • Returns: an AsyncRequest encapsulating the details of the request that was added

    • Arguments:

      • name

        • Type: str

        • Description: human-readable identifier for the request

        • Example: scenarioName

      • url

        • Type: str

        • Description: the base URL of the endpoint

        • Example: https://demoshop.skyramp.dev

      • path

        • Type: str

        • Description: the path to the endpoint of the method

        • Example: /api/v1/products

      • method

        • Type: str

        • Description: the HTTP method of the request (e.g. GET, POST)

        • Example: POST

      • body

        • Type: Optional[dict]

        • Default: None

        • Description: the body of the request

        • Example:

          {
              "name": "Macbook Pro",
              "description": "High-performance laptop",
              "price": 2499.99,
              "image_url": "https://images.app.goo.gl/jGPHo3ZEzEbHG8o2A",
              "category": "Laptop",
              "in_stock": true
          }
      • headers

        • Type: Optional[dict]

        • Default: None

        • Description: the headers of the request

        • Example:

          {
            "Authorization": "Bearer TOKEN_PLACEHOLDER" 
          }
      • path_params

        • Type: Optional[dict]

        • Default: None

        • Description: the path parameters of the request

          • Example:

            {
              "product_id": 2
            }
      • query_params

        • Type: Optional[dict]

        • Default: None

        • Description: the query parameters of the request

        • Example:

          {
            "limit": 10
          }
      • form_params

        • Type: Optional[dict]

        • Default: None

        • Description: the form parameters of the request

        • Example:

          {
              "name": "Macbook Pro",
              "description": "High-performance laptop",
              "price": 2499.99,
              "image_url": "https://images.app.goo.gl/jGPHo3ZEzEbHG8o2A",
              "category": "Laptop",
              "in_stock": true
          }
      • multipart_params

        • Type: Optional[list]

        • Default: None

        • Description: the multipart parameters of the request. Multipart parameters are sets of data that get combined into the request body. They are often used to specify the contents of binaries (e.g. files, videos, photos) and/or encoded data as part of a request.

        • Example:

          • NOTE: The Demo Shop API today does not support multi-part parameters in any of its endpoints. We will use a hypothetical example here - let’s say the POST /products endpoint specified two additional parameters called attributes and promotion_file (a text file outlining a promotional deal for the product).

            [
                MultipartParam(name="attributes", value=f'''{{
                    "product_id": 3}
                }}'''),
                MultipartParam(name="promotion_file", value=file_name, filename="a.txt"),
            ]
          • Here is the curl equivalent:

            curl -X POST https://demoshop.skyramp.dev/api/v1/products \
              -H "Content-Type: multipart/form-data" \
              -k \
              -F attributes='{"product_id": 3}' \
              -F promotion_file
            
            
      • data_override

        • Type: Optional[dict]

        • Default: None

        • Description: any attributes in the request body that need to be overridden with a new value. Refer to Generating Test Data using the Skyramp Library for more information on usage. You can also override a request body with chained response values.

        • Example:

          {
            "name": "skyramp_uuid()"
          }
      • description

        • Type: Optional[str]

        • Default: None

        • Description: a description of the request

        • Example: Create a new product called MacBook Pro

      • expected_code

        • Type: Optional[str]

        • Default: None

        • Description: the expected response status code of the request

        • Example: 200

      • if_

        • Type: Optional[str]

        • Default: ""

        • Description: condition to execute the request. The condition string is intended so that the Skyramp worker can interpret the request during runtime. The Skyramp functions referenced in the Add Assertions to Scenario section should be used to generate this parameter value.

        • Example: Assuming you have a API response called products_POST_response, you can pass in the return value of products_POST_response.request_status_check(201)

      • until

        • Type: Optional[str]

        • Default: None (interpreted as "")

        • Description: condition to stop retrying the request. Useful for polling. The condition string is intended so that the Skyramp worker can interpret the request during runtime. The Skyramp functions referenced in the Add Assertions to Scenario section should be used to generate this parameter value.

        • Example: Assuming you have a API response called products_POST_response, you can pass in the return value of products_POST_response.request_status_check(201)

      • max_retries

        • Type: Optional[int]

        • Default: 5

        • Description: maximum number of times the request can retry before failing. Used with until.

      • retry_interval

        • Type: Optional[int]

        • Default: 1

        • Description: interval in seconds between retries. Used with until.

  • add_async_scenario - adds an AsyncScenario as a step in the scenario queue

    • Returns: n/a

    • Arguments:

      • nested_scenario

        • Type: AsyncScenario

        • Description: the scenario that is being added

        • Example: AsyncScenario(name: “some nested scenario")

      • until

        • Type: Optional[str]

        • Default: ""

        • Description: condition to stop retrying the scenario. The condition string is intended so that the Skyramp worker can interpret the request during runtime. The Skyramp functions referenced in the Add Assertions to Scenario section should be used to generate this parameter value.

        • Example: Assuming you have a API response called products_POST_response, you can pass in the return value of products_POST_response.request_status_check(201)

      • max_retries

        • Type: Optional[int]

        • Default: 5

        • Description: maximum number of times the scenario can retry before failing. Used with until.

      • retry_interval

        • Type: Optional[int]

        • Default: 1

        • Description: interval in seconds between retries. Used with until.

Example Usage

Sequential Scenario Execution using add_async_scenario

Here is an example of how you can sequentially execute two scenarios subScenario1 and subScenario2 by nesting them in a parent scenario.

# Create Parent Scenario
scenario = skyramp.AsyncScenario(name="parentScenario")

# Create the first sub-scenario
subScenario1 = skyramp.AsyncScenario(name="subScenario1")

# Request Body
products_POST_request_body = r'''{
        "category": "Toys",
        "description": "Bear Soft Toy",
        "image_url": "https://images.app.goo.gl/cgcHpeehRdu5osot8",
        "in_stock": true,
        "name": "bigbear",
        "price": 9.99
    }'''
# Add Request to sub-scenario 1
products_POST_response = subScenario1.add_async_request(
    name="products_POST_1",
    url=URL,
    path="/api/v1/products",
    method="POST",
    body=products_POST_request_body,
    headers=headers,
    expected_code="201"
)

# Create the second sub-scenario
subScenario2 = skyramp.AsyncScenario(name="subScenario2")

# Add Request to sub-scenario 2
products_POST_response = subScenario2.add_async_request(
    name="products_POST_2",
    url=URL,
    path="/api/v1/products",
    method="POST",
    body=products_POST_request_body,
    headers=headers,
    expected_code="201"
)

# Add sub scenarios to the parent scenario in order of desired execution
scenario.add_async_scenario(subScenario1)
scenario.add_async_scenario(subScenario2)

# Execute parent scenario
result = client.send_scenario(
    scenario,
    load_test_config=load_test_config
)

Define Scenario Using Conditional Execution (if_)

During scenario execution, the Skyramp worker interprets the if_ argument to determine whether to execute a request or scenario. Here is an example of a scenario which first creates a product, and then creates an order if the product creation response is successful:

# Create Scenario 1
scenario = skyramp.AsyncScenario(name="scenario")

# Request Body
products_POST_request_body = r'''{
        "category": "Toys",
        "description": "Bear Soft Toy",
        "image_url": "https://images.app.goo.gl/cgcHpeehRdu5osot8",
        "in_stock": true,
        "name": "bigbear",
        "price": 9.99
    }'''
# Add Request to Scenario 1
products_POST_response = scenario.add_async_request(
    name="products_POST_1",
    url=URL,
    path="/api/v1/products",
    method="POST",
    body=products_POST_request_body,
    headers=headers,
    expected_code="201"
)

# Request Body
orders_POST_request_body = r'''{
  "customer_email": "sahil@skyramp.dev",
  "items": [
    {
      "product_id": 1,
      "unit_price": 1299.99,
      "quantity": 3
    }
  ]
}'''

# Add a POST Order request to Scenario if the Previous Request was Successful
orders_POST_response = scenario.add_async_request(
    name="orders_POST",
    url=URL,
    path="/api/v1/orders/",
    method="POST",
    body=orders_POST_request_body,
    headers=headers,
    expected_code="201",
    if_=products_POST_response.request_status_check(400)
)

result = client.send_scenario(
    scenario,
    load_test_config=load_test_config
)

Define Scenario Using Conditional Execution (until)

Similarly, the Skyramp worker interprets the until, max_retries, and retry_interval arguments to determine whether and when to stop trying a request. The most common use case for this is to allow the Skyramp worker to poll endpoints during scenario execution. Here is an example of a scenario where an order is created, and then a GET order endpoint is polled until the order is successfully created OR the scenario hits 10 retries (within 1 second intervals).

Note: The example below is to demonstrate how to use the until parameter in code only. At this time the Demo Shop API returns all new resources immediately so there is no practical use for this particular code block.

# Create Scenario 1
scenario = skyramp.AsyncScenario(name="scenario")

# Request Body
orders_POST_request_body = r'''{
  "customer_email": "sahil@skyramp.dev",
  "items": [
    {
      "product_id": 1,
      "unit_price": 1299.99,
      "quantity": 3
    }
  ]
}'''

# Add Request to Scenario 
orders_POST_response = scenario.add_async_request(
    name="orders_POST",
    url=URL,
    path="/api/v1/orders",
    method="POST",
    body=orders_POST_request_body,
    headers=headers,
    expected_code="201"
)

# Add polling of GET /orders/:orderId until either product is ready 
# or the worker hits 10 retries each within 1 second intervals  
orders_GET_response = scenario.add_async_request(
    name="products_GET",
    url=URL,
    path="/api/v1/orders/{order_id}",
    method="GET",
    path_params={"order_id": orders_POST_response.get_async_request_value("order_id")},
    headers=headers,
    expected_code="201",
    until=f"{orders_POST_response.request_status_check(201)}",
    max_retries=10,
    retry_interval=1
)

 result = client.send_scenario(
    scenario,
    load_test_config=load_test_config
)

Data Override

You can direct the Skyramp worker to override a request body with your own custom or chained parameters.

# Add Request to Scenario, override original POST request with random name
products_POST_response = scenario.add_async_request(
    name="products_POST",
    url=URL,
    path="/api/v1/products",
    method="POST",
    body=products_POST_request_body,
    headers=headers,
    expected_code="201",
    data_override={
      "name": "skyramp_uuid()"
    }
)

Manage Variables in a Scenario

Skyramp supports state management for testing scenarios via async variables.

At a high level, here’s how to manage variables in an AsyncScenario.

  • For chaining API response values to other API requests, you can use get_async_request_value.

  • For saving certain response values within the context of the scenario for use by other scenarios (particularly for async scenario execution when the variable may be needed beyond the completion of the scenario runtime), you can use export_async_var.

  • For fetching any variable for use by the Skyramp worker, use get_async_var.

Functions

AsyncRequest

  • get_async_request_value - fetches a response (or a part of a response as specified in the path). Note that this response is not human-readable as the actual response value and is only interpretable by the Skyramp worker.

    • Returns: str

    • Arguments:

      • path

        • Type: str

        • Default: None

        • Description: the path of the response variable

        • Example: name, items.0.price

AsyncScenario

  • export_async_var - saves a variable in the context of a scenario for future reference.

    • Returns: str

    • Arguments:

      • var_name

        • Type: str

        • Default: None

        • Description: name of the variable being saved

        • Example: product_id

      • value

        • Type:str

        • Default: None

        • Description: the value of the variable being saved

        • Example: The output of a function call of get_async_request_value

  • get_async_var - fetch a variable that is saved in a scenario.

    • Returns: str

    • Arguments:

      • var_name

        • Type: str

        • Default: None

        • Description: name of the variable being fetched

        • Example: product_id

  • set_async_var - sets variable at scenario level. Does not support variable override. Commonly used for initialization of variables before the load test scenario runs

    • Returns: n/a

    • Arguments:

      • var_name

        • Type: str

        • Default: None

        • Description: name of the variable being saved

        • Example: product_id

      • value

        • Type: str

        • Default: None

        • Description: the value of the variable being saved

        • Example: 57

Example Usage

Let’s say you want to run two scenarios sequentially:

  • scenario1 is to create a product

  • scenario2 is to create an order with the newly created product

To successfully chain the product ID into a separate scenario, export the product ID from the first scenario into a variable and use it in the second scenario. Here is a code example:

# Create Parent Scenario for sequential execution
scenario = skyramp.AsyncScenario(name="parentScenario")
# Create Scenario 1 - Product Creation
scenario1 = skyramp.AsyncScenario(name="scenario1")
# Request Body
products_POST_request_body = r'''{
        "category": "Toys",
        "description": "Bear Soft Toy",
        "image_url": "https://images.app.goo.gl/cgcHpeehRdu5osot8",
        "in_stock": true,
        "name": "bigbear",
        "price": 9.99
    }'''
# Add Request to scenario 1
products_POST_response = scenario1.add_async_request(
    name="products_POST",
    url=URL,
    path="/api/v1/products",
    method="POST",
    body=products_POST_request_body,
    headers=headers,
    expected_code="201"
)
# Export the response product_id to parent scenario so that you can use it in the next scenario
scenario.export_async_var("product_id", products_POST_response.get_async_request_value("product_id"))
# Create Scenario 2 - Order Creation
scenario2 = skyramp.AsyncScenario(name="scenario2")
# Request Body
orders_POST_request_body = r'''{
    "customer_email": "sahil@skyramp.dev",
    "items": [
        {
        "product_id": 1,
        "unit_price": 1299.99,
        "quantity": 3
        }
    ]
}'''
# Add a POST Order request to Scenario that chains the product ID
orders_POST_response = scenario2.add_async_request(
    name="orders_POST",
    url=URL,
    path="/api/v1/orders/",
    method="POST",
    body=orders_POST_request_body,
    headers=headers,
    expected_code="201",
    data_override={"$0.product_id": scenario.get_async_var("product_id")}
)
# Add sub scenarios to the parent scenario in order of desired execution
scenario.add_async_scenario(scenario1)
scenario.add_async_scenario(scenario2)
# Execute parent scenario
result = client.send_scenario(
    scenario,
    load_test_config=load_test_config
)

Execute Test Scenarios

To execute an AsyncTestScenario, you can send it to the Skyramp worker.

Load Test Configuration

Skyramp provides a load test configuration that allows you to customize how the load test runs. When you run skyramp generate load rest command, the load test related flags used automatically convert to this configuration object. Here is the schema:

LoadTestConfig

  • load_target_rps - the maximum RPS of the load test

  • at_once - Deprecated, use load_num_threads

  • load_count - number of times Skyramp executes the defined request

  • load_num_threads - number of concurrent threads for load test. Concurrent threads represent virtual users enabling you to test the vertical scalability of the service.

  • load_duration - duration of the load test execution in seconds

  • load_rampup_interval- how often Skyramp increases the RPS until target RPS are reached

  • load_rampup_duration - duration that Skyramp incrementally increases the requests per second (RPS) until the target RPS are reached

  • stop_on_failure - whether the load test should stop if something fails. Defaults to False.

Scenario Execution modes

There are two ways you can execute the scenario:

  • Unblocked execution: the program runtime continues without waiting for the scenario execution to complete. This is useful if you would like to proceed with executing other operations without waiting for the results of the scenario.

  • Blocked execution: the program runtime will wait for the scenario execution to complete before proceeding.

Functions

SkyrampClient

  • send_scenario - sends a scenario to the Skyramp worker for execution

    • Returns: AsyncTestResult

    • Arguments:

      • scenario

        • Type: Union[AsyncScenario, List[AsyncScenario]]

        • Default: n/a

        • Description: a set of scenarios to send to the Skyramp worker (library if local runtime) for execution. If more than one scenario is specified, scenarios will be run in parallel.

        • Example: AsyncScenario(name: “scenario”)

      • load_test_config

        • Type: Optional[_LoadTestConfig]

        • Default: None

        • Description: configuration for executing the scenario as a load test. See Load Test Documentation for more information on how to configure load tests. If no config is specified, the scenario will still execute as an integration test.

        • Example:

          skyramp.LoadTestConfig(
              load_duration=5,
              load_num_threads=1,
              load_target_rps=None,
              load_count=None,
              load_rampup_duration=None,
              load_rampup_interval=None
          )
      • dependencies_filepath

        • Type: Optional[str]

        • Default: None

        • Description: path to the dependencies file

        • Example: ./requirements.txt

      • blocked

        • Type: boolean

        • Default: True

        • Description: tells the Skyramp worker (library if local runtime) whether to continue the test execution without waiting for the scenario to finish (false) or wait to continue (true)

      • skip_cert_verification

        • Type: boolean

        • Default: False

        • Description: whether to skip SSL verification in the test scenario

  • async_poll_status - polls the Skyramp worker for status of a async test execution

    • Returns: AsyncTestStatus

    • Arguments:

      • test_id

        • Type: str

        • Default: ""

        • Description: identifier for the test run to poll

        • Example: test_id

AsyncTestResult

  • get_test_id - fetches the test run’s ID for polling / status checking

    • Returns: str

Example Usage

Unblocked Test Execution

To execute this load test without blocking the program runtime from proceeding, you can specify blocked=False in your send_scenario call (as shown below).

With blocked set to False, the function will return immediately and proceed to the next line. You can also use async_poll_status to poll the test run for results while the scenario executes. In this example, we use the asyncio library to manage the polling event loop.

# Unblocked Test Execution
# This test will proceed immediately, and the scenario execution will continue in the background
result = client.send_scenario(
  scenario,
  load_test_config=load_test_config,
  blocked=False
)

# Extract the unique identifier for this test run
# This ID can be used to track and query the test's progress
test_id = result.get_test_id()

# Poll the server until the test completes
# Using asyncio.run() to execute the async polling function in a synchronous context
test_results = asyncio.run(client.async_poll_status(test_id))

print(
    f"result: {test_results.get_overall_status()}"
)

Parallel Scenario Execution

The following example shows how to execute scenario1 and scenario2 in parallel. To shorten the example code, any additional setup of each scenario has been omitted. Please refer to above documentation and examples for more information on how to customize scenarios with requests

scenario1 = skyramp.AsyncScenario(name="scenario1")
scenario2 = skyramp.AsyncScenario(name="scenario2")

# Asynchronous Execution
# This test will proceed immediately, and the scenario execution will continue in the background
client.send_scenario(
  [scenario1, scenario2], # scenario1 and scenario2 will execute in parallel
  load_test_config=load_test_config,
  blocked=False
)

Parallelize Two Sequential Scenarios

The following example shows how to execute scenario and scenario2 in parallel, with each having two sub scenarios executed in sequential order. To shorten the example code, any additional setup of each scenario has been omitted. Please refer to above documentation and examples for more information on how to customize scenarios with requests:

# Create Scenario 1
scenario = skyramp.AsyncScenario(name="scenario1")

# Scenario 1 will run two sub-scenarios sequentially
subScenario1 = skyramp.AsyncScenario(name="subScenario1")
subScenario2 = skyramp.AsyncScenario(name="subScenario2")
scenario.add_async_scenario(nested_scenario: subScenario1)
scenario.add_async_scenario(nested_scenario: subScenario2)

# Create Scenario 2 (which will run in parallel with Scenario 1)
scenario2 = skyramp.AsyncScenario(name="scenario2")

# Scenario 2 will run two separate sub-scenarios sequentially
subScenario3 = skyramp.AsyncScenario(name="subScenario3")
subScenario4 = skyramp.AsyncScenario(name="subScenario4")
scenario2.add_async_scenario(nested_scenario: subScenario3)
scenario2.add_async_scenario(nested_scenario: subScenario4)

# Run Scenario 1 and Scenario 2 in parallel
results = client.send_scenario(
  [scenario1, scenario2], # scenario1 and scenario2 will execute in parallel
  load_test_config=load_test_config,
)


View Test Run Status

With the Skyramp Library, you can view your test run status while in progress or once completed.

Test Run Format

In general, you can fetch test run status as follows:

  • For a summary of load test execution statistics, use get_overall_status. Generally, the following statistics will be displayed per top level scenario:

    • Scenario name

    • Scenario status

      • Scenario count - number of times the scenario was executed

      • Execution count - number of successful scenario executions

      • Failure count - number of failed scenario executions

      • Latency metrics - statistical distribution of latency numbers for the scenario execution sample (including - average, min, max, 90th, 95th, and 99th percentiles)

  • For a full output of all test execution statistics and requests for a given scenario, use get_scenario or get_scenarios. Generally, the output will show the following information:

    • ALL of the information in get_overall_statusfor the scenario(s)

    • Requests - Every request that is part of the scenario has it’s own stats aggregated similar to its scenario wrapper

      • Request name

      • Request status

        • Count - number of times the request was

        • Failure count - number of failed request executions

        • Execution count - number of successful request executions

        • Latency metrics - statistical distribution of latency numbers for the request execution sample (including - average, min, max, 90th, 95th, and 99th percentiles)

        • Log table - collection of specific logs per response code collected

        • Code table - distribution of requests by response code

Functions

AsyncTestStatus

  • get_scenario - Retrieves a status object for a given scenario name

    • Returns: AsyncScenarioStatus

    • Arguments:

      • scenario_name

        • Type: str

        • Default: n/a

        • Description: name of the scenario

        • Example: "scenario"

  • get_scenarios - Retrieves a list of status objects matching the scenario name

    • Returns: List[AsyncScenarioStatus]

    • Arguments:

      • scenario_name

        • Type: str

        • Default: ''

        • Description: name of the scenario

        • Example: "scenario"

  • get_overall_status - fetch the overall status of the test run by iterating through top-level scenarios in the forma

    • Returns: str

Example Usage

get_overall_status

result = client.send_scenario(
    scenario,
    load_test_config=load_test_config
)

print(
    f"result: {result.get_overall_status()}"
)

Output:

get_scenario & get_scenarios

result = client.send_scenario(
    scenario,
    load_test_config=load_test_config
)

# get scenario status
scenario_status=result.get_scenario(scenario_name="scenario")
print(
    f"Scenario status: {scenario_status}"
)

Output:

Customization of Load Test Scenarios

This guide explains how to customize load testing scenarios using the Skyramp worker and library functions. Skyramp already generates load tests out of the box through the Skyramp CLI and agent; however, we understand that additional customization is required in many cases. We'll demonstrate using Skyramp’s Demo Shop API, a simple e-commerce API for product and order management. Learn more about the Demo Shop API.

Refer to the Installation Guide if you haven't installed Skyramp yet.

Load Test Structure

Here is an example of a simple load test in Python:

# Generated by Skyramp v1.2.1 on 2025-06-26 15:49:15.088816 -0700 PDT m=+2.872724918
# Command: skyramp generate load rest https://demoshop.skyramp.dev/api/v1/products \
# 		--api-schema https://demoshop.skyramp.dev/openapi.json \
# 		--framework pytest \
# 		--language python \
# 		--method POST \
#       --runtime docker

# Import of required libraries
import skyramp
import os
import time
# URL for test requests
URL = "https://demoshop.skyramp.dev"

load_config = skyramp.LoadTestConfig(
    load_duration=5,
    load_num_threads=1,
    load_target_rps=None,
    load_count=None,
    load_rampup_duration=None,
    load_rampup_interval=None
)
def test_products_post(load_test_config:skyramp.LoadTestConfig=load_config):
    # Invocation of Skyramp Client
    client = skyramp.Client(
        runtime="docker",
        docker_network="skyramp",
        docker_skyramp_port=35142
    )
    # Definition of authentication header
    headers = {}
    if os.getenv("SKYRAMP_TEST_TOKEN") is not None:
        headers["Authorization"] = "Bearer " + os.getenv("SKYRAMP_TEST_TOKEN")

    scenario = skyramp.AsyncScenario(name="scenario")
    # Request Body
    products_POST_request_body = r'''{
            "category": "Toys",
            "description": "Bear Soft Toy",
            "image_url": "https://images.app.goo.gl/cgcHpeehRdu5osot8",
            "in_stock": true,
            "name": "bigbear",
            "price": 9.99
        }'''
    
    # Add Request to Scenario
    products_POST_response = scenario.add_async_request(
        name="products_POST",
        url=URL,
        path="/api/v1/products",
        method="POST",
        body=products_POST_request_body,
        headers=headers,
        expected_code="201"
    )
    result = client.send_scenario(
        scenario,
        load_test_config=load_test_config
    )

    print(
        f"result: {result.get_overall_status()}"
    )


if __name__ == "__main__":
    test_products_post()


Load Test File Anatomy & Data Structure

At a high level, a Skyramp load test works as follows:

  1. Skyramp creates a SkyrampClient with the runtime specified.

    1. For this tutorial, we will use Docker as the runtime. You can also execute the test locally or in Kubernetes.

  2. A new AsyncScenario is then created. A scenario is a set of steps defined to execute for a test. A step can be:

    • an API request represented by an AsyncRequest (“create/read/update/delete this resource”)

    • a test assertion (“assert that expected value matches actual value”)

    • a sub-scenario

  3. The SkyrampClient then sends the scenario over to the Skyramp worker for execution and gets an AsyncTestStatus in return.

    1. A worker in this context is a component or process responsible for executing the test scenario. When you run a load test using the Docker runtime, Skyramp automatically spins up a specialized worker that executes the load test scenario.

      1. Note: For local execution, the Skyramp Library acts as the “worker” for the load test.

    2. AsyncTestStatus is a wrapper for fetching the status of a test run (including details on load test performance + request execution results).

    3. AsyncScenarioStatus - a wrapper for fetching the status of a specific scenario

    4. AsyncRequestStatus - a wrapper for fetching the status of a specific request

High-Level Diagram of Data Hierarchy

Why do load tests look different from other test types?

Load tests are run asynchronously. This means that all requests are packaged in a scenario, sent to the worker, and executed without any individual responses being communicated back to the program runtime.

Why would I want to execute a test async?

For load testing scenarios with multiple chained API calls and simulations of high volume traffic, async test execution enables significantly higher test performance and allows for the simulation of more realistic load on the system by enabling concurrency and targeted/maximized requests per second.

Customize Skyramp Load Test Scenarios

The Skyramp Library provides useful classes and functions that can help you customize your test scenarios.

Create Scenarios and Requests

At a very high level, scenario setup requires two key steps:

  1. Create an AsyncScenario object.

  2. Add an async request or async scenario to the AsyncScenario object using add_async_request or add_async_scenario.

Functions

AsyncScenario

  • add_async_request - adds an AsyncRequest as a step in the scenario.

    • Returns: an AsyncRequest encapsulating the details of the request that was added

    • Arguments:

      • name

        • Type: str

        • Description: human-readable identifier for the request

        • Example: scenarioName

      • url

        • Type: str

        • Description: the base URL of the endpoint

        • Example: https://demoshop.skyramp.dev

      • path

        • Type: str

        • Description: the path to the endpoint of the method

        • Example: /api/v1/products

      • method

        • Type: str

        • Description: the HTTP method of the request (e.g. GET, POST)

        • Example: POST

      • body

        • Type: Optional[dict]

        • Default: None

        • Description: the body of the request

        • Example:

          {
              "name": "Macbook Pro",
              "description": "High-performance laptop",
              "price": 2499.99,
              "image_url": "https://images.app.goo.gl/jGPHo3ZEzEbHG8o2A",
              "category": "Laptop",
              "in_stock": true
          }
      • headers

        • Type: Optional[dict]

        • Default: None

        • Description: the headers of the request

        • Example:

          {
            "Authorization": "Bearer TOKEN_PLACEHOLDER" 
          }
      • path_params

        • Type: Optional[dict]

        • Default: None

        • Description: the path parameters of the request

          • Example:

            {
              "product_id": 2
            }
      • query_params

        • Type: Optional[dict]

        • Default: None

        • Description: the query parameters of the request

        • Example:

          {
            "limit": 10
          }
      • form_params

        • Type: Optional[dict]

        • Default: None

        • Description: the form parameters of the request

        • Example:

          {
              "name": "Macbook Pro",
              "description": "High-performance laptop",
              "price": 2499.99,
              "image_url": "https://images.app.goo.gl/jGPHo3ZEzEbHG8o2A",
              "category": "Laptop",
              "in_stock": true
          }
      • multipart_params

        • Type: Optional[list]

        • Default: None

        • Description: the multipart parameters of the request. Multipart parameters are sets of data that get combined into the request body. They are often used to specify the contents of binaries (e.g. files, videos, photos) and/or encoded data as part of a request.

        • Example:

          • NOTE: The Demo Shop API today does not support multi-part parameters in any of its endpoints. We will use a hypothetical example here - let’s say the POST /products endpoint specified two additional parameters called attributes and promotion_file (a text file outlining a promotional deal for the product).

            [
                MultipartParam(name="attributes", value=f'''{{
                    "product_id": 3}
                }}'''),
                MultipartParam(name="promotion_file", value=file_name, filename="a.txt"),
            ]
          • Here is the curl equivalent:

            curl -X POST https://demoshop.skyramp.dev/api/v1/products \
              -H "Content-Type: multipart/form-data" \
              -k \
              -F attributes='{"product_id": 3}' \
              -F promotion_file
            
            
      • data_override

        • Type: Optional[dict]

        • Default: None

        • Description: any attributes in the request body that need to be overridden with a new value. Refer to Generating Test Data using the Skyramp Library for more information on usage. You can also override a request body with chained response values.

        • Example:

          {
            "name": "skyramp_uuid()"
          }
      • description

        • Type: Optional[str]

        • Default: None

        • Description: a description of the request

        • Example: Create a new product called MacBook Pro

      • expected_code

        • Type: Optional[str]

        • Default: None

        • Description: the expected response status code of the request

        • Example: 200

      • if_

        • Type: Optional[str]

        • Default: ""

        • Description: condition to execute the request. The condition string is intended so that the Skyramp worker can interpret the request during runtime. The Skyramp functions referenced in the Add Assertions to Scenario section should be used to generate this parameter value.

        • Example: Assuming you have a API response called products_POST_response, you can pass in the return value of products_POST_response.request_status_check(201)

      • until

        • Type: Optional[str]

        • Default: None (interpreted as "")

        • Description: condition to stop retrying the request. Useful for polling. The condition string is intended so that the Skyramp worker can interpret the request during runtime. The Skyramp functions referenced in the Add Assertions to Scenario section should be used to generate this parameter value.

        • Example: Assuming you have a API response called products_POST_response, you can pass in the return value of products_POST_response.request_status_check(201)

      • max_retries

        • Type: Optional[int]

        • Default: 5

        • Description: maximum number of times the request can retry before failing. Used with until.

      • retry_interval

        • Type: Optional[int]

        • Default: 1

        • Description: interval in seconds between retries. Used with until.

  • add_async_scenario - adds an AsyncScenario as a step in the scenario queue

    • Returns: n/a

    • Arguments:

      • nested_scenario

        • Type: AsyncScenario

        • Description: the scenario that is being added

        • Example: AsyncScenario(name: “some nested scenario")

      • until

        • Type: Optional[str]

        • Default: ""

        • Description: condition to stop retrying the scenario. The condition string is intended so that the Skyramp worker can interpret the request during runtime. The Skyramp functions referenced in the Add Assertions to Scenario section should be used to generate this parameter value.

        • Example: Assuming you have a API response called products_POST_response, you can pass in the return value of products_POST_response.request_status_check(201)

      • max_retries

        • Type: Optional[int]

        • Default: 5

        • Description: maximum number of times the scenario can retry before failing. Used with until.

      • retry_interval

        • Type: Optional[int]

        • Default: 1

        • Description: interval in seconds between retries. Used with until.

Example Usage

Sequential Scenario Execution using add_async_scenario

Here is an example of how you can sequentially execute two scenarios subScenario1 and subScenario2 by nesting them in a parent scenario.

# Create Parent Scenario
scenario = skyramp.AsyncScenario(name="parentScenario")

# Create the first sub-scenario
subScenario1 = skyramp.AsyncScenario(name="subScenario1")

# Request Body
products_POST_request_body = r'''{
        "category": "Toys",
        "description": "Bear Soft Toy",
        "image_url": "https://images.app.goo.gl/cgcHpeehRdu5osot8",
        "in_stock": true,
        "name": "bigbear",
        "price": 9.99
    }'''
# Add Request to sub-scenario 1
products_POST_response = subScenario1.add_async_request(
    name="products_POST_1",
    url=URL,
    path="/api/v1/products",
    method="POST",
    body=products_POST_request_body,
    headers=headers,
    expected_code="201"
)

# Create the second sub-scenario
subScenario2 = skyramp.AsyncScenario(name="subScenario2")

# Add Request to sub-scenario 2
products_POST_response = subScenario2.add_async_request(
    name="products_POST_2",
    url=URL,
    path="/api/v1/products",
    method="POST",
    body=products_POST_request_body,
    headers=headers,
    expected_code="201"
)

# Add sub scenarios to the parent scenario in order of desired execution
scenario.add_async_scenario(subScenario1)
scenario.add_async_scenario(subScenario2)

# Execute parent scenario
result = client.send_scenario(
    scenario,
    load_test_config=load_test_config
)

Define Scenario Using Conditional Execution (if_)

During scenario execution, the Skyramp worker interprets the if_ argument to determine whether to execute a request or scenario. Here is an example of a scenario which first creates a product, and then creates an order if the product creation response is successful:

# Create Scenario 1
scenario = skyramp.AsyncScenario(name="scenario")

# Request Body
products_POST_request_body = r'''{
        "category": "Toys",
        "description": "Bear Soft Toy",
        "image_url": "https://images.app.goo.gl/cgcHpeehRdu5osot8",
        "in_stock": true,
        "name": "bigbear",
        "price": 9.99
    }'''
# Add Request to Scenario 1
products_POST_response = scenario.add_async_request(
    name="products_POST_1",
    url=URL,
    path="/api/v1/products",
    method="POST",
    body=products_POST_request_body,
    headers=headers,
    expected_code="201"
)

# Request Body
orders_POST_request_body = r'''{
  "customer_email": "sahil@skyramp.dev",
  "items": [
    {
      "product_id": 1,
      "unit_price": 1299.99,
      "quantity": 3
    }
  ]
}'''

# Add a POST Order request to Scenario if the Previous Request was Successful
orders_POST_response = scenario.add_async_request(
    name="orders_POST",
    url=URL,
    path="/api/v1/orders/",
    method="POST",
    body=orders_POST_request_body,
    headers=headers,
    expected_code="201",
    if_=products_POST_response.request_status_check(400)
)

result = client.send_scenario(
    scenario,
    load_test_config=load_test_config
)

Define Scenario Using Conditional Execution (until)

Similarly, the Skyramp worker interprets the until, max_retries, and retry_interval arguments to determine whether and when to stop trying a request. The most common use case for this is to allow the Skyramp worker to poll endpoints during scenario execution. Here is an example of a scenario where an order is created, and then a GET order endpoint is polled until the order is successfully created OR the scenario hits 10 retries (within 1 second intervals).

Note: The example below is to demonstrate how to use the until parameter in code only. At this time the Demo Shop API returns all new resources immediately so there is no practical use for this particular code block.

# Create Scenario 1
scenario = skyramp.AsyncScenario(name="scenario")

# Request Body
orders_POST_request_body = r'''{
  "customer_email": "sahil@skyramp.dev",
  "items": [
    {
      "product_id": 1,
      "unit_price": 1299.99,
      "quantity": 3
    }
  ]
}'''

# Add Request to Scenario 
orders_POST_response = scenario.add_async_request(
    name="orders_POST",
    url=URL,
    path="/api/v1/orders",
    method="POST",
    body=orders_POST_request_body,
    headers=headers,
    expected_code="201"
)

# Add polling of GET /orders/:orderId until either product is ready 
# or the worker hits 10 retries each within 1 second intervals  
orders_GET_response = scenario.add_async_request(
    name="products_GET",
    url=URL,
    path="/api/v1/orders/{order_id}",
    method="GET",
    path_params={"order_id": orders_POST_response.get_async_request_value("order_id")},
    headers=headers,
    expected_code="201",
    until=f"{orders_POST_response.request_status_check(201)}",
    max_retries=10,
    retry_interval=1
)

 result = client.send_scenario(
    scenario,
    load_test_config=load_test_config
)

Data Override

You can direct the Skyramp worker to override a request body with your own custom or chained parameters.

# Add Request to Scenario, override original POST request with random name
products_POST_response = scenario.add_async_request(
    name="products_POST",
    url=URL,
    path="/api/v1/products",
    method="POST",
    body=products_POST_request_body,
    headers=headers,
    expected_code="201",
    data_override={
      "name": "skyramp_uuid()"
    }
)

Manage Variables in a Scenario

Skyramp supports state management for testing scenarios via async variables.

At a high level, here’s how to manage variables in an AsyncScenario.

  • For chaining API response values to other API requests, you can use get_async_request_value.

  • For saving certain response values within the context of the scenario for use by other scenarios (particularly for async scenario execution when the variable may be needed beyond the completion of the scenario runtime), you can use export_async_var.

  • For fetching any variable for use by the Skyramp worker, use get_async_var.

Functions

AsyncRequest

  • get_async_request_value - fetches a response (or a part of a response as specified in the path). Note that this response is not human-readable as the actual response value and is only interpretable by the Skyramp worker.

    • Returns: str

    • Arguments:

      • path

        • Type: str

        • Default: None

        • Description: the path of the response variable

        • Example: name, items.0.price

AsyncScenario

  • export_async_var - saves a variable in the context of a scenario for future reference.

    • Returns: str

    • Arguments:

      • var_name

        • Type: str

        • Default: None

        • Description: name of the variable being saved

        • Example: product_id

      • value

        • Type:str

        • Default: None

        • Description: the value of the variable being saved

        • Example: The output of a function call of get_async_request_value

  • get_async_var - fetch a variable that is saved in a scenario.

    • Returns: str

    • Arguments:

      • var_name

        • Type: str

        • Default: None

        • Description: name of the variable being fetched

        • Example: product_id

  • set_async_var - sets variable at scenario level. Does not support variable override. Commonly used for initialization of variables before the load test scenario runs

    • Returns: n/a

    • Arguments:

      • var_name

        • Type: str

        • Default: None

        • Description: name of the variable being saved

        • Example: product_id

      • value

        • Type: str

        • Default: None

        • Description: the value of the variable being saved

        • Example: 57

Example Usage

Let’s say you want to run two scenarios sequentially:

  • scenario1 is to create a product

  • scenario2 is to create an order with the newly created product

To successfully chain the product ID into a separate scenario, export the product ID from the first scenario into a variable and use it in the second scenario. Here is a code example:

# Create Parent Scenario for sequential execution
scenario = skyramp.AsyncScenario(name="parentScenario")
# Create Scenario 1 - Product Creation
scenario1 = skyramp.AsyncScenario(name="scenario1")
# Request Body
products_POST_request_body = r'''{
        "category": "Toys",
        "description": "Bear Soft Toy",
        "image_url": "https://images.app.goo.gl/cgcHpeehRdu5osot8",
        "in_stock": true,
        "name": "bigbear",
        "price": 9.99
    }'''
# Add Request to scenario 1
products_POST_response = scenario1.add_async_request(
    name="products_POST",
    url=URL,
    path="/api/v1/products",
    method="POST",
    body=products_POST_request_body,
    headers=headers,
    expected_code="201"
)
# Export the response product_id to parent scenario so that you can use it in the next scenario
scenario.export_async_var("product_id", products_POST_response.get_async_request_value("product_id"))
# Create Scenario 2 - Order Creation
scenario2 = skyramp.AsyncScenario(name="scenario2")
# Request Body
orders_POST_request_body = r'''{
    "customer_email": "sahil@skyramp.dev",
    "items": [
        {
        "product_id": 1,
        "unit_price": 1299.99,
        "quantity": 3
        }
    ]
}'''
# Add a POST Order request to Scenario that chains the product ID
orders_POST_response = scenario2.add_async_request(
    name="orders_POST",
    url=URL,
    path="/api/v1/orders/",
    method="POST",
    body=orders_POST_request_body,
    headers=headers,
    expected_code="201",
    data_override={"$0.product_id": scenario.get_async_var("product_id")}
)
# Add sub scenarios to the parent scenario in order of desired execution
scenario.add_async_scenario(scenario1)
scenario.add_async_scenario(scenario2)
# Execute parent scenario
result = client.send_scenario(
    scenario,
    load_test_config=load_test_config
)

Execute Test Scenarios

To execute an AsyncTestScenario, you can send it to the Skyramp worker.

Load Test Configuration

Skyramp provides a load test configuration that allows you to customize how the load test runs. When you run skyramp generate load rest command, the load test related flags used automatically convert to this configuration object. Here is the schema:

LoadTestConfig

  • load_target_rps - the maximum RPS of the load test

  • at_once - Deprecated, use load_num_threads

  • load_count - number of times Skyramp executes the defined request

  • load_num_threads - number of concurrent threads for load test. Concurrent threads represent virtual users enabling you to test the vertical scalability of the service.

  • load_duration - duration of the load test execution in seconds

  • load_rampup_interval- how often Skyramp increases the RPS until target RPS are reached

  • load_rampup_duration - duration that Skyramp incrementally increases the requests per second (RPS) until the target RPS are reached

  • stop_on_failure - whether the load test should stop if something fails. Defaults to False.

Scenario Execution modes

There are two ways you can execute the scenario:

  • Unblocked execution: the program runtime continues without waiting for the scenario execution to complete. This is useful if you would like to proceed with executing other operations without waiting for the results of the scenario.

  • Blocked execution: the program runtime will wait for the scenario execution to complete before proceeding.

Functions

SkyrampClient

  • send_scenario - sends a scenario to the Skyramp worker for execution

    • Returns: AsyncTestResult

    • Arguments:

      • scenario

        • Type: Union[AsyncScenario, List[AsyncScenario]]

        • Default: n/a

        • Description: a set of scenarios to send to the Skyramp worker (library if local runtime) for execution. If more than one scenario is specified, scenarios will be run in parallel.

        • Example: AsyncScenario(name: “scenario”)

      • load_test_config

        • Type: Optional[_LoadTestConfig]

        • Default: None

        • Description: configuration for executing the scenario as a load test. See Load Test Documentation for more information on how to configure load tests. If no config is specified, the scenario will still execute as an integration test.

        • Example:

          skyramp.LoadTestConfig(
              load_duration=5,
              load_num_threads=1,
              load_target_rps=None,
              load_count=None,
              load_rampup_duration=None,
              load_rampup_interval=None
          )
      • dependencies_filepath

        • Type: Optional[str]

        • Default: None

        • Description: path to the dependencies file

        • Example: ./requirements.txt

      • blocked

        • Type: boolean

        • Default: True

        • Description: tells the Skyramp worker (library if local runtime) whether to continue the test execution without waiting for the scenario to finish (false) or wait to continue (true)

      • skip_cert_verification

        • Type: boolean

        • Default: False

        • Description: whether to skip SSL verification in the test scenario

  • async_poll_status - polls the Skyramp worker for status of a async test execution

    • Returns: AsyncTestStatus

    • Arguments:

      • test_id

        • Type: str

        • Default: ""

        • Description: identifier for the test run to poll

        • Example: test_id

AsyncTestResult

  • get_test_id - fetches the test run’s ID for polling / status checking

    • Returns: str

Example Usage

Unblocked Test Execution

To execute this load test without blocking the program runtime from proceeding, you can specify blocked=False in your send_scenario call (as shown below).

With blocked set to False, the function will return immediately and proceed to the next line. You can also use async_poll_status to poll the test run for results while the scenario executes. In this example, we use the asyncio library to manage the polling event loop.

# Unblocked Test Execution
# This test will proceed immediately, and the scenario execution will continue in the background
result = client.send_scenario(
  scenario,
  load_test_config=load_test_config,
  blocked=False
)

# Extract the unique identifier for this test run
# This ID can be used to track and query the test's progress
test_id = result.get_test_id()

# Poll the server until the test completes
# Using asyncio.run() to execute the async polling function in a synchronous context
test_results = asyncio.run(client.async_poll_status(test_id))

print(
    f"result: {test_results.get_overall_status()}"
)

Parallel Scenario Execution

The following example shows how to execute scenario1 and scenario2 in parallel. To shorten the example code, any additional setup of each scenario has been omitted. Please refer to above documentation and examples for more information on how to customize scenarios with requests

scenario1 = skyramp.AsyncScenario(name="scenario1")
scenario2 = skyramp.AsyncScenario(name="scenario2")

# Asynchronous Execution
# This test will proceed immediately, and the scenario execution will continue in the background
client.send_scenario(
  [scenario1, scenario2], # scenario1 and scenario2 will execute in parallel
  load_test_config=load_test_config,
  blocked=False
)

Parallelize Two Sequential Scenarios

The following example shows how to execute scenario and scenario2 in parallel, with each having two sub scenarios executed in sequential order. To shorten the example code, any additional setup of each scenario has been omitted. Please refer to above documentation and examples for more information on how to customize scenarios with requests:

# Create Scenario 1
scenario = skyramp.AsyncScenario(name="scenario1")

# Scenario 1 will run two sub-scenarios sequentially
subScenario1 = skyramp.AsyncScenario(name="subScenario1")
subScenario2 = skyramp.AsyncScenario(name="subScenario2")
scenario.add_async_scenario(nested_scenario: subScenario1)
scenario.add_async_scenario(nested_scenario: subScenario2)

# Create Scenario 2 (which will run in parallel with Scenario 1)
scenario2 = skyramp.AsyncScenario(name="scenario2")

# Scenario 2 will run two separate sub-scenarios sequentially
subScenario3 = skyramp.AsyncScenario(name="subScenario3")
subScenario4 = skyramp.AsyncScenario(name="subScenario4")
scenario2.add_async_scenario(nested_scenario: subScenario3)
scenario2.add_async_scenario(nested_scenario: subScenario4)

# Run Scenario 1 and Scenario 2 in parallel
results = client.send_scenario(
  [scenario1, scenario2], # scenario1 and scenario2 will execute in parallel
  load_test_config=load_test_config,
)


View Test Run Status

With the Skyramp Library, you can view your test run status while in progress or once completed.

Test Run Format

In general, you can fetch test run status as follows:

  • For a summary of load test execution statistics, use get_overall_status. Generally, the following statistics will be displayed per top level scenario:

    • Scenario name

    • Scenario status

      • Scenario count - number of times the scenario was executed

      • Execution count - number of successful scenario executions

      • Failure count - number of failed scenario executions

      • Latency metrics - statistical distribution of latency numbers for the scenario execution sample (including - average, min, max, 90th, 95th, and 99th percentiles)

  • For a full output of all test execution statistics and requests for a given scenario, use get_scenario or get_scenarios. Generally, the output will show the following information:

    • ALL of the information in get_overall_statusfor the scenario(s)

    • Requests - Every request that is part of the scenario has it’s own stats aggregated similar to its scenario wrapper

      • Request name

      • Request status

        • Count - number of times the request was

        • Failure count - number of failed request executions

        • Execution count - number of successful request executions

        • Latency metrics - statistical distribution of latency numbers for the request execution sample (including - average, min, max, 90th, 95th, and 99th percentiles)

        • Log table - collection of specific logs per response code collected

        • Code table - distribution of requests by response code

Functions

AsyncTestStatus

  • get_scenario - Retrieves a status object for a given scenario name

    • Returns: AsyncScenarioStatus

    • Arguments:

      • scenario_name

        • Type: str

        • Default: n/a

        • Description: name of the scenario

        • Example: "scenario"

  • get_scenarios - Retrieves a list of status objects matching the scenario name

    • Returns: List[AsyncScenarioStatus]

    • Arguments:

      • scenario_name

        • Type: str

        • Default: ''

        • Description: name of the scenario

        • Example: "scenario"

  • get_overall_status - fetch the overall status of the test run by iterating through top-level scenarios in the forma

    • Returns: str

Example Usage

get_overall_status

result = client.send_scenario(
    scenario,
    load_test_config=load_test_config
)

print(
    f"result: {result.get_overall_status()}"
)

Output:

get_scenario & get_scenarios

result = client.send_scenario(
    scenario,
    load_test_config=load_test_config
)

# get scenario status
scenario_status=result.get_scenario(scenario_name="scenario")
print(
    f"Scenario status: {scenario_status}"
)

Output:

Customization of Load Test Scenarios

This guide explains how to customize load testing scenarios using the Skyramp worker and library functions. Skyramp already generates load tests out of the box through the Skyramp CLI and agent; however, we understand that additional customization is required in many cases. We'll demonstrate using Skyramp’s Demo Shop API, a simple e-commerce API for product and order management. Learn more about the Demo Shop API.

Refer to the Installation Guide if you haven't installed Skyramp yet.

Load Test Structure

Here is an example of a simple load test in Python:

# Generated by Skyramp v1.2.1 on 2025-06-26 15:49:15.088816 -0700 PDT m=+2.872724918
# Command: skyramp generate load rest https://demoshop.skyramp.dev/api/v1/products \
# 		--api-schema https://demoshop.skyramp.dev/openapi.json \
# 		--framework pytest \
# 		--language python \
# 		--method POST \
#       --runtime docker

# Import of required libraries
import skyramp
import os
import time
# URL for test requests
URL = "https://demoshop.skyramp.dev"

load_config = skyramp.LoadTestConfig(
    load_duration=5,
    load_num_threads=1,
    load_target_rps=None,
    load_count=None,
    load_rampup_duration=None,
    load_rampup_interval=None
)
def test_products_post(load_test_config:skyramp.LoadTestConfig=load_config):
    # Invocation of Skyramp Client
    client = skyramp.Client(
        runtime="docker",
        docker_network="skyramp",
        docker_skyramp_port=35142
    )
    # Definition of authentication header
    headers = {}
    if os.getenv("SKYRAMP_TEST_TOKEN") is not None:
        headers["Authorization"] = "Bearer " + os.getenv("SKYRAMP_TEST_TOKEN")

    scenario = skyramp.AsyncScenario(name="scenario")
    # Request Body
    products_POST_request_body = r'''{
            "category": "Toys",
            "description": "Bear Soft Toy",
            "image_url": "https://images.app.goo.gl/cgcHpeehRdu5osot8",
            "in_stock": true,
            "name": "bigbear",
            "price": 9.99
        }'''
    
    # Add Request to Scenario
    products_POST_response = scenario.add_async_request(
        name="products_POST",
        url=URL,
        path="/api/v1/products",
        method="POST",
        body=products_POST_request_body,
        headers=headers,
        expected_code="201"
    )
    result = client.send_scenario(
        scenario,
        load_test_config=load_test_config
    )

    print(
        f"result: {result.get_overall_status()}"
    )


if __name__ == "__main__":
    test_products_post()


Load Test File Anatomy & Data Structure

At a high level, a Skyramp load test works as follows:

  1. Skyramp creates a SkyrampClient with the runtime specified.

    1. For this tutorial, we will use Docker as the runtime. You can also execute the test locally or in Kubernetes.

  2. A new AsyncScenario is then created. A scenario is a set of steps defined to execute for a test. A step can be:

    • an API request represented by an AsyncRequest (“create/read/update/delete this resource”)

    • a test assertion (“assert that expected value matches actual value”)

    • a sub-scenario

  3. The SkyrampClient then sends the scenario over to the Skyramp worker for execution and gets an AsyncTestStatus in return.

    1. A worker in this context is a component or process responsible for executing the test scenario. When you run a load test using the Docker runtime, Skyramp automatically spins up a specialized worker that executes the load test scenario.

      1. Note: For local execution, the Skyramp Library acts as the “worker” for the load test.

    2. AsyncTestStatus is a wrapper for fetching the status of a test run (including details on load test performance + request execution results).

    3. AsyncScenarioStatus - a wrapper for fetching the status of a specific scenario

    4. AsyncRequestStatus - a wrapper for fetching the status of a specific request

High-Level Diagram of Data Hierarchy

Why do load tests look different from other test types?

Load tests are run asynchronously. This means that all requests are packaged in a scenario, sent to the worker, and executed without any individual responses being communicated back to the program runtime.

Why would I want to execute a test async?

For load testing scenarios with multiple chained API calls and simulations of high volume traffic, async test execution enables significantly higher test performance and allows for the simulation of more realistic load on the system by enabling concurrency and targeted/maximized requests per second.

Customize Skyramp Load Test Scenarios

The Skyramp Library provides useful classes and functions that can help you customize your test scenarios.

Create Scenarios and Requests

At a very high level, scenario setup requires two key steps:

  1. Create an AsyncScenario object.

  2. Add an async request or async scenario to the AsyncScenario object using add_async_request or add_async_scenario.

Functions

AsyncScenario

  • add_async_request - adds an AsyncRequest as a step in the scenario.

    • Returns: an AsyncRequest encapsulating the details of the request that was added

    • Arguments:

      • name

        • Type: str

        • Description: human-readable identifier for the request

        • Example: scenarioName

      • url

        • Type: str

        • Description: the base URL of the endpoint

        • Example: https://demoshop.skyramp.dev

      • path

        • Type: str

        • Description: the path to the endpoint of the method

        • Example: /api/v1/products

      • method

        • Type: str

        • Description: the HTTP method of the request (e.g. GET, POST)

        • Example: POST

      • body

        • Type: Optional[dict]

        • Default: None

        • Description: the body of the request

        • Example:

          {
              "name": "Macbook Pro",
              "description": "High-performance laptop",
              "price": 2499.99,
              "image_url": "https://images.app.goo.gl/jGPHo3ZEzEbHG8o2A",
              "category": "Laptop",
              "in_stock": true
          }
      • headers

        • Type: Optional[dict]

        • Default: None

        • Description: the headers of the request

        • Example:

          {
            "Authorization": "Bearer TOKEN_PLACEHOLDER" 
          }
      • path_params

        • Type: Optional[dict]

        • Default: None

        • Description: the path parameters of the request

          • Example:

            {
              "product_id": 2
            }
      • query_params

        • Type: Optional[dict]

        • Default: None

        • Description: the query parameters of the request

        • Example:

          {
            "limit": 10
          }
      • form_params

        • Type: Optional[dict]

        • Default: None

        • Description: the form parameters of the request

        • Example:

          {
              "name": "Macbook Pro",
              "description": "High-performance laptop",
              "price": 2499.99,
              "image_url": "https://images.app.goo.gl/jGPHo3ZEzEbHG8o2A",
              "category": "Laptop",
              "in_stock": true
          }
      • multipart_params

        • Type: Optional[list]

        • Default: None

        • Description: the multipart parameters of the request. Multipart parameters are sets of data that get combined into the request body. They are often used to specify the contents of binaries (e.g. files, videos, photos) and/or encoded data as part of a request.

        • Example:

          • NOTE: The Demo Shop API today does not support multi-part parameters in any of its endpoints. We will use a hypothetical example here - let’s say the POST /products endpoint specified two additional parameters called attributes and promotion_file (a text file outlining a promotional deal for the product).

            [
                MultipartParam(name="attributes", value=f'''{{
                    "product_id": 3}
                }}'''),
                MultipartParam(name="promotion_file", value=file_name, filename="a.txt"),
            ]
          • Here is the curl equivalent:

            curl -X POST https://demoshop.skyramp.dev/api/v1/products \
              -H "Content-Type: multipart/form-data" \
              -k \
              -F attributes='{"product_id": 3}' \
              -F promotion_file
            
            
      • data_override

        • Type: Optional[dict]

        • Default: None

        • Description: any attributes in the request body that need to be overridden with a new value. Refer to Generating Test Data using the Skyramp Library for more information on usage. You can also override a request body with chained response values.

        • Example:

          {
            "name": "skyramp_uuid()"
          }
      • description

        • Type: Optional[str]

        • Default: None

        • Description: a description of the request

        • Example: Create a new product called MacBook Pro

      • expected_code

        • Type: Optional[str]

        • Default: None

        • Description: the expected response status code of the request

        • Example: 200

      • if_

        • Type: Optional[str]

        • Default: ""

        • Description: condition to execute the request. The condition string is intended so that the Skyramp worker can interpret the request during runtime. The Skyramp functions referenced in the Add Assertions to Scenario section should be used to generate this parameter value.

        • Example: Assuming you have a API response called products_POST_response, you can pass in the return value of products_POST_response.request_status_check(201)

      • until

        • Type: Optional[str]

        • Default: None (interpreted as "")

        • Description: condition to stop retrying the request. Useful for polling. The condition string is intended so that the Skyramp worker can interpret the request during runtime. The Skyramp functions referenced in the Add Assertions to Scenario section should be used to generate this parameter value.

        • Example: Assuming you have a API response called products_POST_response, you can pass in the return value of products_POST_response.request_status_check(201)

      • max_retries

        • Type: Optional[int]

        • Default: 5

        • Description: maximum number of times the request can retry before failing. Used with until.

      • retry_interval

        • Type: Optional[int]

        • Default: 1

        • Description: interval in seconds between retries. Used with until.

  • add_async_scenario - adds an AsyncScenario as a step in the scenario queue

    • Returns: n/a

    • Arguments:

      • nested_scenario

        • Type: AsyncScenario

        • Description: the scenario that is being added

        • Example: AsyncScenario(name: “some nested scenario")

      • until

        • Type: Optional[str]

        • Default: ""

        • Description: condition to stop retrying the scenario. The condition string is intended so that the Skyramp worker can interpret the request during runtime. The Skyramp functions referenced in the Add Assertions to Scenario section should be used to generate this parameter value.

        • Example: Assuming you have a API response called products_POST_response, you can pass in the return value of products_POST_response.request_status_check(201)

      • max_retries

        • Type: Optional[int]

        • Default: 5

        • Description: maximum number of times the scenario can retry before failing. Used with until.

      • retry_interval

        • Type: Optional[int]

        • Default: 1

        • Description: interval in seconds between retries. Used with until.

Example Usage

Sequential Scenario Execution using add_async_scenario

Here is an example of how you can sequentially execute two scenarios subScenario1 and subScenario2 by nesting them in a parent scenario.

# Create Parent Scenario
scenario = skyramp.AsyncScenario(name="parentScenario")

# Create the first sub-scenario
subScenario1 = skyramp.AsyncScenario(name="subScenario1")

# Request Body
products_POST_request_body = r'''{
        "category": "Toys",
        "description": "Bear Soft Toy",
        "image_url": "https://images.app.goo.gl/cgcHpeehRdu5osot8",
        "in_stock": true,
        "name": "bigbear",
        "price": 9.99
    }'''
# Add Request to sub-scenario 1
products_POST_response = subScenario1.add_async_request(
    name="products_POST_1",
    url=URL,
    path="/api/v1/products",
    method="POST",
    body=products_POST_request_body,
    headers=headers,
    expected_code="201"
)

# Create the second sub-scenario
subScenario2 = skyramp.AsyncScenario(name="subScenario2")

# Add Request to sub-scenario 2
products_POST_response = subScenario2.add_async_request(
    name="products_POST_2",
    url=URL,
    path="/api/v1/products",
    method="POST",
    body=products_POST_request_body,
    headers=headers,
    expected_code="201"
)

# Add sub scenarios to the parent scenario in order of desired execution
scenario.add_async_scenario(subScenario1)
scenario.add_async_scenario(subScenario2)

# Execute parent scenario
result = client.send_scenario(
    scenario,
    load_test_config=load_test_config
)

Define Scenario Using Conditional Execution (if_)

During scenario execution, the Skyramp worker interprets the if_ argument to determine whether to execute a request or scenario. Here is an example of a scenario which first creates a product, and then creates an order if the product creation response is successful:

# Create Scenario 1
scenario = skyramp.AsyncScenario(name="scenario")

# Request Body
products_POST_request_body = r'''{
        "category": "Toys",
        "description": "Bear Soft Toy",
        "image_url": "https://images.app.goo.gl/cgcHpeehRdu5osot8",
        "in_stock": true,
        "name": "bigbear",
        "price": 9.99
    }'''
# Add Request to Scenario 1
products_POST_response = scenario.add_async_request(
    name="products_POST_1",
    url=URL,
    path="/api/v1/products",
    method="POST",
    body=products_POST_request_body,
    headers=headers,
    expected_code="201"
)

# Request Body
orders_POST_request_body = r'''{
  "customer_email": "sahil@skyramp.dev",
  "items": [
    {
      "product_id": 1,
      "unit_price": 1299.99,
      "quantity": 3
    }
  ]
}'''

# Add a POST Order request to Scenario if the Previous Request was Successful
orders_POST_response = scenario.add_async_request(
    name="orders_POST",
    url=URL,
    path="/api/v1/orders/",
    method="POST",
    body=orders_POST_request_body,
    headers=headers,
    expected_code="201",
    if_=products_POST_response.request_status_check(400)
)

result = client.send_scenario(
    scenario,
    load_test_config=load_test_config
)

Define Scenario Using Conditional Execution (until)

Similarly, the Skyramp worker interprets the until, max_retries, and retry_interval arguments to determine whether and when to stop trying a request. The most common use case for this is to allow the Skyramp worker to poll endpoints during scenario execution. Here is an example of a scenario where an order is created, and then a GET order endpoint is polled until the order is successfully created OR the scenario hits 10 retries (within 1 second intervals).

Note: The example below is to demonstrate how to use the until parameter in code only. At this time the Demo Shop API returns all new resources immediately so there is no practical use for this particular code block.

# Create Scenario 1
scenario = skyramp.AsyncScenario(name="scenario")

# Request Body
orders_POST_request_body = r'''{
  "customer_email": "sahil@skyramp.dev",
  "items": [
    {
      "product_id": 1,
      "unit_price": 1299.99,
      "quantity": 3
    }
  ]
}'''

# Add Request to Scenario 
orders_POST_response = scenario.add_async_request(
    name="orders_POST",
    url=URL,
    path="/api/v1/orders",
    method="POST",
    body=orders_POST_request_body,
    headers=headers,
    expected_code="201"
)

# Add polling of GET /orders/:orderId until either product is ready 
# or the worker hits 10 retries each within 1 second intervals  
orders_GET_response = scenario.add_async_request(
    name="products_GET",
    url=URL,
    path="/api/v1/orders/{order_id}",
    method="GET",
    path_params={"order_id": orders_POST_response.get_async_request_value("order_id")},
    headers=headers,
    expected_code="201",
    until=f"{orders_POST_response.request_status_check(201)}",
    max_retries=10,
    retry_interval=1
)

 result = client.send_scenario(
    scenario,
    load_test_config=load_test_config
)

Data Override

You can direct the Skyramp worker to override a request body with your own custom or chained parameters.

# Add Request to Scenario, override original POST request with random name
products_POST_response = scenario.add_async_request(
    name="products_POST",
    url=URL,
    path="/api/v1/products",
    method="POST",
    body=products_POST_request_body,
    headers=headers,
    expected_code="201",
    data_override={
      "name": "skyramp_uuid()"
    }
)

Manage Variables in a Scenario

Skyramp supports state management for testing scenarios via async variables.

At a high level, here’s how to manage variables in an AsyncScenario.

  • For chaining API response values to other API requests, you can use get_async_request_value.

  • For saving certain response values within the context of the scenario for use by other scenarios (particularly for async scenario execution when the variable may be needed beyond the completion of the scenario runtime), you can use export_async_var.

  • For fetching any variable for use by the Skyramp worker, use get_async_var.

Functions

AsyncRequest

  • get_async_request_value - fetches a response (or a part of a response as specified in the path). Note that this response is not human-readable as the actual response value and is only interpretable by the Skyramp worker.

    • Returns: str

    • Arguments:

      • path

        • Type: str

        • Default: None

        • Description: the path of the response variable

        • Example: name, items.0.price

AsyncScenario

  • export_async_var - saves a variable in the context of a scenario for future reference.

    • Returns: str

    • Arguments:

      • var_name

        • Type: str

        • Default: None

        • Description: name of the variable being saved

        • Example: product_id

      • value

        • Type:str

        • Default: None

        • Description: the value of the variable being saved

        • Example: The output of a function call of get_async_request_value

  • get_async_var - fetch a variable that is saved in a scenario.

    • Returns: str

    • Arguments:

      • var_name

        • Type: str

        • Default: None

        • Description: name of the variable being fetched

        • Example: product_id

  • set_async_var - sets variable at scenario level. Does not support variable override. Commonly used for initialization of variables before the load test scenario runs

    • Returns: n/a

    • Arguments:

      • var_name

        • Type: str

        • Default: None

        • Description: name of the variable being saved

        • Example: product_id

      • value

        • Type: str

        • Default: None

        • Description: the value of the variable being saved

        • Example: 57

Example Usage

Let’s say you want to run two scenarios sequentially:

  • scenario1 is to create a product

  • scenario2 is to create an order with the newly created product

To successfully chain the product ID into a separate scenario, export the product ID from the first scenario into a variable and use it in the second scenario. Here is a code example:

# Create Parent Scenario for sequential execution
scenario = skyramp.AsyncScenario(name="parentScenario")
# Create Scenario 1 - Product Creation
scenario1 = skyramp.AsyncScenario(name="scenario1")
# Request Body
products_POST_request_body = r'''{
        "category": "Toys",
        "description": "Bear Soft Toy",
        "image_url": "https://images.app.goo.gl/cgcHpeehRdu5osot8",
        "in_stock": true,
        "name": "bigbear",
        "price": 9.99
    }'''
# Add Request to scenario 1
products_POST_response = scenario1.add_async_request(
    name="products_POST",
    url=URL,
    path="/api/v1/products",
    method="POST",
    body=products_POST_request_body,
    headers=headers,
    expected_code="201"
)
# Export the response product_id to parent scenario so that you can use it in the next scenario
scenario.export_async_var("product_id", products_POST_response.get_async_request_value("product_id"))
# Create Scenario 2 - Order Creation
scenario2 = skyramp.AsyncScenario(name="scenario2")
# Request Body
orders_POST_request_body = r'''{
    "customer_email": "sahil@skyramp.dev",
    "items": [
        {
        "product_id": 1,
        "unit_price": 1299.99,
        "quantity": 3
        }
    ]
}'''
# Add a POST Order request to Scenario that chains the product ID
orders_POST_response = scenario2.add_async_request(
    name="orders_POST",
    url=URL,
    path="/api/v1/orders/",
    method="POST",
    body=orders_POST_request_body,
    headers=headers,
    expected_code="201",
    data_override={"$0.product_id": scenario.get_async_var("product_id")}
)
# Add sub scenarios to the parent scenario in order of desired execution
scenario.add_async_scenario(scenario1)
scenario.add_async_scenario(scenario2)
# Execute parent scenario
result = client.send_scenario(
    scenario,
    load_test_config=load_test_config
)

Execute Test Scenarios

To execute an AsyncTestScenario, you can send it to the Skyramp worker.

Load Test Configuration

Skyramp provides a load test configuration that allows you to customize how the load test runs. When you run skyramp generate load rest command, the load test related flags used automatically convert to this configuration object. Here is the schema:

LoadTestConfig

  • load_target_rps - the maximum RPS of the load test

  • at_once - Deprecated, use load_num_threads

  • load_count - number of times Skyramp executes the defined request

  • load_num_threads - number of concurrent threads for load test. Concurrent threads represent virtual users enabling you to test the vertical scalability of the service.

  • load_duration - duration of the load test execution in seconds

  • load_rampup_interval- how often Skyramp increases the RPS until target RPS are reached

  • load_rampup_duration - duration that Skyramp incrementally increases the requests per second (RPS) until the target RPS are reached

  • stop_on_failure - whether the load test should stop if something fails. Defaults to False.

Scenario Execution modes

There are two ways you can execute the scenario:

  • Unblocked execution: the program runtime continues without waiting for the scenario execution to complete. This is useful if you would like to proceed with executing other operations without waiting for the results of the scenario.

  • Blocked execution: the program runtime will wait for the scenario execution to complete before proceeding.

Functions

SkyrampClient

  • send_scenario - sends a scenario to the Skyramp worker for execution

    • Returns: AsyncTestResult

    • Arguments:

      • scenario

        • Type: Union[AsyncScenario, List[AsyncScenario]]

        • Default: n/a

        • Description: a set of scenarios to send to the Skyramp worker (library if local runtime) for execution. If more than one scenario is specified, scenarios will be run in parallel.

        • Example: AsyncScenario(name: “scenario”)

      • load_test_config

        • Type: Optional[_LoadTestConfig]

        • Default: None

        • Description: configuration for executing the scenario as a load test. See Load Test Documentation for more information on how to configure load tests. If no config is specified, the scenario will still execute as an integration test.

        • Example:

          skyramp.LoadTestConfig(
              load_duration=5,
              load_num_threads=1,
              load_target_rps=None,
              load_count=None,
              load_rampup_duration=None,
              load_rampup_interval=None
          )
      • dependencies_filepath

        • Type: Optional[str]

        • Default: None

        • Description: path to the dependencies file

        • Example: ./requirements.txt

      • blocked

        • Type: boolean

        • Default: True

        • Description: tells the Skyramp worker (library if local runtime) whether to continue the test execution without waiting for the scenario to finish (false) or wait to continue (true)

      • skip_cert_verification

        • Type: boolean

        • Default: False

        • Description: whether to skip SSL verification in the test scenario

  • async_poll_status - polls the Skyramp worker for status of a async test execution

    • Returns: AsyncTestStatus

    • Arguments:

      • test_id

        • Type: str

        • Default: ""

        • Description: identifier for the test run to poll

        • Example: test_id

AsyncTestResult

  • get_test_id - fetches the test run’s ID for polling / status checking

    • Returns: str

Example Usage

Unblocked Test Execution

To execute this load test without blocking the program runtime from proceeding, you can specify blocked=False in your send_scenario call (as shown below).

With blocked set to False, the function will return immediately and proceed to the next line. You can also use async_poll_status to poll the test run for results while the scenario executes. In this example, we use the asyncio library to manage the polling event loop.

# Unblocked Test Execution
# This test will proceed immediately, and the scenario execution will continue in the background
result = client.send_scenario(
  scenario,
  load_test_config=load_test_config,
  blocked=False
)

# Extract the unique identifier for this test run
# This ID can be used to track and query the test's progress
test_id = result.get_test_id()

# Poll the server until the test completes
# Using asyncio.run() to execute the async polling function in a synchronous context
test_results = asyncio.run(client.async_poll_status(test_id))

print(
    f"result: {test_results.get_overall_status()}"
)

Parallel Scenario Execution

The following example shows how to execute scenario1 and scenario2 in parallel. To shorten the example code, any additional setup of each scenario has been omitted. Please refer to above documentation and examples for more information on how to customize scenarios with requests

scenario1 = skyramp.AsyncScenario(name="scenario1")
scenario2 = skyramp.AsyncScenario(name="scenario2")

# Asynchronous Execution
# This test will proceed immediately, and the scenario execution will continue in the background
client.send_scenario(
  [scenario1, scenario2], # scenario1 and scenario2 will execute in parallel
  load_test_config=load_test_config,
  blocked=False
)

Parallelize Two Sequential Scenarios

The following example shows how to execute scenario and scenario2 in parallel, with each having two sub scenarios executed in sequential order. To shorten the example code, any additional setup of each scenario has been omitted. Please refer to above documentation and examples for more information on how to customize scenarios with requests:

# Create Scenario 1
scenario = skyramp.AsyncScenario(name="scenario1")

# Scenario 1 will run two sub-scenarios sequentially
subScenario1 = skyramp.AsyncScenario(name="subScenario1")
subScenario2 = skyramp.AsyncScenario(name="subScenario2")
scenario.add_async_scenario(nested_scenario: subScenario1)
scenario.add_async_scenario(nested_scenario: subScenario2)

# Create Scenario 2 (which will run in parallel with Scenario 1)
scenario2 = skyramp.AsyncScenario(name="scenario2")

# Scenario 2 will run two separate sub-scenarios sequentially
subScenario3 = skyramp.AsyncScenario(name="subScenario3")
subScenario4 = skyramp.AsyncScenario(name="subScenario4")
scenario2.add_async_scenario(nested_scenario: subScenario3)
scenario2.add_async_scenario(nested_scenario: subScenario4)

# Run Scenario 1 and Scenario 2 in parallel
results = client.send_scenario(
  [scenario1, scenario2], # scenario1 and scenario2 will execute in parallel
  load_test_config=load_test_config,
)


View Test Run Status

With the Skyramp Library, you can view your test run status while in progress or once completed.

Test Run Format

In general, you can fetch test run status as follows:

  • For a summary of load test execution statistics, use get_overall_status. Generally, the following statistics will be displayed per top level scenario:

    • Scenario name

    • Scenario status

      • Scenario count - number of times the scenario was executed

      • Execution count - number of successful scenario executions

      • Failure count - number of failed scenario executions

      • Latency metrics - statistical distribution of latency numbers for the scenario execution sample (including - average, min, max, 90th, 95th, and 99th percentiles)

  • For a full output of all test execution statistics and requests for a given scenario, use get_scenario or get_scenarios. Generally, the output will show the following information:

    • ALL of the information in get_overall_statusfor the scenario(s)

    • Requests - Every request that is part of the scenario has it’s own stats aggregated similar to its scenario wrapper

      • Request name

      • Request status

        • Count - number of times the request was

        • Failure count - number of failed request executions

        • Execution count - number of successful request executions

        • Latency metrics - statistical distribution of latency numbers for the request execution sample (including - average, min, max, 90th, 95th, and 99th percentiles)

        • Log table - collection of specific logs per response code collected

        • Code table - distribution of requests by response code

Functions

AsyncTestStatus

  • get_scenario - Retrieves a status object for a given scenario name

    • Returns: AsyncScenarioStatus

    • Arguments:

      • scenario_name

        • Type: str

        • Default: n/a

        • Description: name of the scenario

        • Example: "scenario"

  • get_scenarios - Retrieves a list of status objects matching the scenario name

    • Returns: List[AsyncScenarioStatus]

    • Arguments:

      • scenario_name

        • Type: str

        • Default: ''

        • Description: name of the scenario

        • Example: "scenario"

  • get_overall_status - fetch the overall status of the test run by iterating through top-level scenarios in the forma

    • Returns: str

Example Usage

get_overall_status

result = client.send_scenario(
    scenario,
    load_test_config=load_test_config
)

print(
    f"result: {result.get_overall_status()}"
)

Output:

get_scenario & get_scenarios

result = client.send_scenario(
    scenario,
    load_test_config=load_test_config
)

# get scenario status
scenario_status=result.get_scenario(scenario_name="scenario")
print(
    f"Scenario status: {scenario_status}"
)

Output:

© 2025 Skyramp, Inc. All rights reserved.

© 2025 Skyramp, Inc. All rights reserved.

© 2025 Skyramp, Inc. All rights reserved.