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:
Skyramp creates a SkyrampClient with the runtime specified.
For this tutorial, we will use Docker as the runtime. You can also execute the test locally or in Kubernetes.
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
The SkyrampClient then sends the scenario over to the Skyramp worker for execution and gets an AsyncTestStatus in return.
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.
Note: For local execution, the Skyramp Library acts as the “worker” for the load test.
AsyncTestStatus is a wrapper for fetching the status of a test run (including details on load test performance + request execution results).
AsyncScenarioStatus - a wrapper for fetching the status of a specific scenario
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:
Create an AsyncScenario object.
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 ofproducts_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 ofproducts_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 queueReturns: 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 ofproducts_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 runsReturns: 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 testat_once
- Deprecated, use load_num_threadsload_count
- number of times Skyramp executes the defined requestload_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 secondsload_rampup_interval
- how often Skyramp increases the RPS until target RPS are reachedload_rampup_duration
- duration that Skyramp incrementally increases the requests per second (RPS) until the target RPS are reachedstop_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 executionReturns:
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 executionReturns:
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 checkingReturns:
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 nameReturns:
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 nameReturns:
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 formaReturns:
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:
Skyramp creates a SkyrampClient with the runtime specified.
For this tutorial, we will use Docker as the runtime. You can also execute the test locally or in Kubernetes.
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
The SkyrampClient then sends the scenario over to the Skyramp worker for execution and gets an AsyncTestStatus in return.
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.
Note: For local execution, the Skyramp Library acts as the “worker” for the load test.
AsyncTestStatus is a wrapper for fetching the status of a test run (including details on load test performance + request execution results).
AsyncScenarioStatus - a wrapper for fetching the status of a specific scenario
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:
Create an AsyncScenario object.
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 ofproducts_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 ofproducts_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 queueReturns: 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 ofproducts_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 runsReturns: 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 testat_once
- Deprecated, use load_num_threadsload_count
- number of times Skyramp executes the defined requestload_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 secondsload_rampup_interval
- how often Skyramp increases the RPS until target RPS are reachedload_rampup_duration
- duration that Skyramp incrementally increases the requests per second (RPS) until the target RPS are reachedstop_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 executionReturns:
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 executionReturns:
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 checkingReturns:
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 nameReturns:
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 nameReturns:
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 formaReturns:
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:
Skyramp creates a SkyrampClient with the runtime specified.
For this tutorial, we will use Docker as the runtime. You can also execute the test locally or in Kubernetes.
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
The SkyrampClient then sends the scenario over to the Skyramp worker for execution and gets an AsyncTestStatus in return.
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.
Note: For local execution, the Skyramp Library acts as the “worker” for the load test.
AsyncTestStatus is a wrapper for fetching the status of a test run (including details on load test performance + request execution results).
AsyncScenarioStatus - a wrapper for fetching the status of a specific scenario
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:
Create an AsyncScenario object.
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 ofproducts_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 ofproducts_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 queueReturns: 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 ofproducts_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 runsReturns: 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 testat_once
- Deprecated, use load_num_threadsload_count
- number of times Skyramp executes the defined requestload_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 secondsload_rampup_interval
- how often Skyramp increases the RPS until target RPS are reachedload_rampup_duration
- duration that Skyramp incrementally increases the requests per second (RPS) until the target RPS are reachedstop_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 executionReturns:
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 executionReturns:
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 checkingReturns:
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 nameReturns:
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 nameReturns:
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 formaReturns:
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:
