steps.api

Submodules

• steps.api.cloud
• steps.api.cloud_objects
• steps.api.given
• steps.api.given_cloud
• steps.api.then
• steps.api.then_cloud
• steps.api.utils

Attributes

PAGINATION_RESULT_KEYS

Exceptions

MissingSchema	Assertion failed.
DefaultSchemaIsUsed	Assertion failed.
ForbiddenKeyword	Assertion failed.
SchemaError	Assertion failed.
MissingEndpoint	Assertion failed.

Classes

Context
Platform	Platform class representing a worker in the cloud system.
Job	Job class representing a computational job in the cloud system.
JobToken	Job token class for job-specific authentication.
Request	A user-created Request object.
Context

Functions

try_read_from_variable(method, context)
get_table(→ dict)	Convert a Gherkin table to a Python dictionary.
prepare_request(context, api_type)	Prepare a new request for a specific API type.
prepare_cloud_request(context)	Prepare a new request for the cloud API.
log_as_user(→ None)	Log in as a specific test user.
log_as_admin(→ None)	Log in as the admin user.
get_user(context, user_index)	Get information about a specific user.
create_job_token(→ None)	Create a new job token for a specific user.
delete_platform(→ None)	Delete a platform by its ID.
register_platform_simplify(→ None)
get_platform(→ steps.api.cloud.Platform)
register_platform(→ None)
fetch_platform_job(→ None)
run_platform_job(→ None)
progress_job(→ None)
finish_job(→ None)
assign_user_to_platform(context, user_index, ...)
assign_token_to_platform(→ None)
assign_user_and_token_to_platform(context, user_index, ...)
create_job_simplified(context, user_index)
create_job(→ None)
create_job_payload(→ None)
get_job_status(→ None)
get_job_result(→ None)
get_job_info(→ None)
cancel_job(→ None)
step_impl(context, state)
log_allure(→ None)
attach_swagger_link(context)
parse_nested_value(value, context)
try_read_from_variable(method, context)
try_save_value_to_variable(context, value, method)
set_header_key(→ None)	Set a specific HTTP header for the request.
get_table(→ dict)	Convert a Gherkin table to a Python dictionary.
set_body(→ None)	Set the request body from a Gherkin table.
set_param_key(context, key, value)	Set a query parameter for the request.
set_request_params(→ None)	Set request query parameters from a Gherkin table.
save_request(context)	Save the current request as a reference for later use.
prepare_request(context, api_type)	Prepare a new request for a specific API type.
send_complete_request(context)	Send the prepared request and handle response and logging.
send_request_with_path_parameter(→ None)	Make a request with a path parameter to a specific endpoint.
resend_request(→ None)
resend_request_setting_header(→ None)
update_ref_request_header(→ None)
resend_request_setting_body(→ None)
resend_request_with_path_parameter(→ None)
send_request(→ None)	Make a request to a specific endpoint.
save_random_uuid(→ None)
save_variable(→ None)
save_data_in_response(→ None)	Save a field from the response to a variable.
add_test_section_description(context, ...)
check_not_page_result(context)	Check that the response is not in pagination format.
check_expected_values_in_response(context, key, value)
get_status_exception(→ AWrongStatus)
check_forbidden_keyword(→ str)
status_code_validation(context, expected_http_status_code)	Validate that the response has the expected HTTP status code.
check_response_from_table(context)	Check that the response contains the expected key-value pairs from a table.
check_empty_response(context)	Check that the response body is empty (empty JSON object or array).
check_not_empty_response(context)	Check that the response body is not empty (has content).
response_match_defined_schema(context)	Validate that the response JSON matches the defined schema.

Package Contents

class steps.api.Context

output_dir: str = ''

logger: logging.Logger = None

apis: dict[str, steps.api.utils.api.Api]

cloud: steps.api.cloud.Cloud = None

request: steps.api.utils.api.Request = None

response = None

steps.api.try_read_from_variable(method: str, context)

class steps.api.Platform(name: str = '', type: str = 'simulator', platform_id: str = None, worker_id: str = str(uuid4()), specs: dict[str] = {'available_commands': ['probs']}, **kwargs)

Bases: steps.api.cloud_objects.cloud_base.ACloudObject

Platform class representing a worker in the cloud system.

This class handles worker registration, job management, and platform permissions.

Parameters:

• name – Name of the platform

• type – Type of platform (default: “simulator”)

• platform_id – Platform ID (None for new platforms)

• worker_id – Worker ID (defaults to a new UUID)

• specs – Platform specifications

• kwargs – Additional attributes to set on the platform

Raises:

AssertionError – If Platform.ADMIN is None or not an admin

ADMIN = None

REGISTER_ENDPOINT = '/admin/platforms/register'

JOB_ENDPOINT = '/admin/job'

ASSIGN_ENDPOINT = '/api/platform-permissions'

classmethod get_object_name() → str

classmethod get_endpoint() → str

classmethod get_api(context: steps.api.utils.CloudContext) → steps.api.utils.Api

classmethod get_token_method() → str

PLATFORM_NAME_KEY = 'platform_name'

DEFAULT_NAME = 'sim:qa'

name = ''

type = 'simulator'

worker_id: str = ''

platform_id: str

specs

register(context: steps.api.utils.CloudContext, register_dict: dict[str] = None) → None

Register this platform in the cloud.

Parameters:

• context – Cloud context for API configuration

• register_dict – Dictionary with registration parameters (optional)

Raises:

RuntimeError – If platform registration fails

read(context: steps.api.utils.CloudContext) → dict[str]

Read this platform’s information from the cloud.

Parameters:

context – Cloud context for API configuration

Returns:

Dictionary with platform data from response

Raises:

RuntimeError – If reading platform data fails

update(context: steps.api.utils.CloudContext, update_dict: dict) → None

Update this platform’s information in the cloud.

Parameters:

• context – Cloud context for API configuration

• update_dict – Dictionary with fields to update

Raises:

RuntimeError – If updating platform data fails

delete(context: steps.api.utils.CloudContext) → None

Delete this platform from the cloud.

Parameters:

context – Cloud context for API configuration

Raises:

RuntimeError – If deleting the platform fails

fetch_job(context: steps.api.utils.CloudContext) → steps.api.cloud_objects.cloud_job.Job

Fetch a job for this platform to process.

Parameters:

context – Cloud context for API configuration

Returns:

Job object for the fetched job

Raises:

RuntimeError – If fetching a job fails

run_job(context: steps.api.utils.CloudContext, job) → None

progress_job(context: steps.api.utils.CloudContext, job, progress_dict: dict[str]) → None

finish_job(context: steps.api.utils.CloudContext, job, finish_dict: dict[str]) → None

assign_user(context: steps.api.utils.CloudContext, user: steps.api.cloud_objects.cloud_user.User) → None

assign_token(context: steps.api.utils.CloudContext, job_token) → None

assign_company(context: steps.api.utils.CloudContext, company_id: int) → None

classmethod list_all(context: steps.api.utils.CloudContext) → list[Platform]

List all platforms in the cloud.

Parameters:

context – Cloud context for API configuration

Returns:

List of Platform objects

Raises:

• AssertionError – If ADMIN is not an admin

• RuntimeError – If listing platforms fails

class steps.api.Job(platform_name: str = DEFAULT_PLATFORM_NAME, job_name: str = 'qa_job', payload: dict[str] = DEFAULT_PAYLOAD, pcvl_version: str = '0.12.0', process_id: str = str(uuid4()), job_id: str = '', **_)

Bases: steps.api.cloud_objects.cloud_base.ACloudObject

Job class representing a computational job in the cloud system.

This class handles job creation, monitoring, and management operations.

Parameters:

• platform_name – Name of the platform to run the job on

• job_name – Name of the job

• payload – Job payload containing command, circuit, parameters, etc.

• pcvl_version – Version of the PCVL library to use

• process_id – Unique ID for the process (defaults to a new UUID)

• job_id – Job ID (empty for new jobs)

• _ – Additional attributes to set on the job

JOB_NAME_KEY = 'job_name'

PLATFORM_NAME_KEY = 'platform_name'

PAYLOAD_KEY = 'payload'

PCVL_VERSION_KEY = 'pcvl_version'

PROCESS_ID_KEY = 'process_id'

DEFAULT_PAYLOAD

classmethod get_object_name() → str

classmethod get_endpoint() → str

classmethod get_api(context: steps.api.utils.CloudContext) → steps.api.utils.Api

classmethod get_token_method() → str

DEFAULT_PLATFORM_NAME = 'sim:qa'

platform_name: str = 'sim:qa'

job_name: str = 'qa_job'

payload: dict[str]

pcvl_version: str = '0.12.0'

process_id: str = ''

job_id = ''

create(context: steps.api.utils.CloudContext, user, create_dict: dict = {}) → None

Create this job in the cloud.

Parameters:

• context – Cloud context for API configuration

• user – User providing authentication

• create_dict – Additional parameters for job creation

Raises:

RuntimeError – If job creation fails

read(context: steps.api.utils.CloudContext, user) → dict[str]

Read this job’s information from the cloud.

Parameters:

• context – Cloud context for API configuration

• user – User providing authentication

Returns:

Dictionary with job data from response

Raises:

RuntimeError – If reading job data fails

cancel(context: steps.api.utils.CloudContext, user) → dict[str]

Cancel this job.

Parameters:

• context – Cloud context for API configuration

• user – User providing authentication

Raises:

RuntimeError – If canceling the job fails

get_status(context: steps.api.utils.CloudContext, user) → dict[str]

Get the current status of this job.

Parameters:

• context – Cloud context for API configuration

• user – User providing authentication

Returns:

Dictionary with job status information

Raises:

RuntimeError – If retrieving job status fails

get_results(context: steps.api.utils.CloudContext, user) → dict[str]

Get the results of this job.

Parameters:

• context – Cloud context for API configuration

• user – User providing authentication

Returns:

Dictionary with job results

Raises:

RuntimeError – If retrieving job results fails

classmethod cancel_many(context: steps.api.utils.CloudContext, jobs: list, admin: steps.api.cloud_objects.cloud_user.User) → None

Cancel multiple jobs at once.

Parameters:

• context – Cloud context for API configuration

• jobs – List of Job objects to cancel

• admin – Admin user providing authentication

Raises:

• AssertionError – If the user is not an admin

• RuntimeError – If canceling the jobs fails

classmethod list_all(context: steps.api.utils.CloudContext, admin: steps.api.cloud_objects.cloud_user.User) → list[Job]

List all jobs in the cloud.

Parameters:

• context – Cloud context for API configuration

• admin – Admin user providing authentication

Returns:

List of Job objects

Raises:

• AssertionError – If the user is not an admin

• RuntimeError – If listing jobs fails

class steps.api.JobToken(label: str = None, duration: int = 99999, token_id: int = -1, user_id: int = -1, token: str = '', **kwargs)

Bases: steps.api.cloud_objects.cloud_base.ACloudObject

Job token class for job-specific authentication.

This class handles the creation and management of tokens used for job operations.

Parameters:

• label – Label for the token

• duration – Token duration in seconds

• token_id – Token ID (defaults to -1 for new tokens)

• user_id – ID of the user the token belongs to

• token – Token string

• kwargs – Additional attributes to set on the token

classmethod get_object_name() → str

classmethod get_endpoint() → str

classmethod get_api(context: steps.api.utils.CloudContext) → steps.api.utils.Api

classmethod get_token_method() → str

DURATION_KEY = 'duration'

LABEL_KEY = 'label'

label: str = None

token_id: int

user_id: int = -1

duration: int = 99999

token: str = ''

create(context: steps.api.utils.CloudContext, user: steps.api.cloud_objects.cloud_user.User) → None

Create this job token in the cloud.

Parameters:

• context – Cloud context for API configuration

• user – User providing authentication and owning the token

Raises:

• AssertionError – If the user doesn’t have a login token

• RuntimeError – If job token creation fails

read(context: steps.api.utils.CloudContext, user: steps.api.cloud_objects.cloud_user.User) → None

Read this job token’s information from the cloud.

Parameters:

• context – Cloud context for API configuration

• user – User providing authentication

Returns:

Dictionary with token data from response

Raises:

• AssertionError – If the user doesn’t have job tokens or the token wasn’t created

• RuntimeError – If reading token data fails

delete(context: steps.api.utils.CloudContext, user: steps.api.cloud_objects.cloud_user.User) → None

Delete this job token from the cloud.

Parameters:

• context – Cloud context for API configuration

• user – User providing authentication

Raises:

• AssertionError – If the token wasn’t created

• RuntimeError – If deleting the token fails

classmethod list_all(context: steps.api.utils.CloudContext, admin: steps.api.cloud_objects.cloud_user.User) → list[JobToken]

List all job tokens in the cloud.

Parameters:

• context – Cloud context for API configuration

• admin – Admin user providing authentication

Returns:

List of JobToken objects

Raises:

• AssertionError – If the user is not an admin

• RuntimeError – If listing tokens fails

steps.api.get_table(context: steps.api.utils.Context) → dict

Convert a Gherkin table to a Python dictionary.

Processes a Behave table with ‘key’ and ‘value’ columns, parsing nested values and variable references in the process.

Parameters:

context – The Behave context containing the table

Returns:

Dictionary with keys and values from the table

steps.api.prepare_request(context: steps.api.utils.Context, api_type: str)

Prepare a new request for a specific API type.

Parameters:

• context – The Behave context

• api_type – Type of API to prepare the request for

Raises:

AssertionError – If the API type is not available in context.apis

steps.api.prepare_cloud_request(context)

Prepare a new request for the cloud API.

Shorthand for creating a request to the cloud API type.

Parameters:

context – The Behave context

steps.api.log_as_user(context: steps.api.utils.Context, user_index: str) → None

Log in as a specific test user.

Parameters:

• context – The Behave context

• user_index – Index of the user in the users list

steps.api.log_as_admin(context: steps.api.utils.Context) → None

Log in as the admin user.

Parameters:

context – The Behave context

steps.api.get_user(context: steps.api.utils.Context, user_index: str)

Get information about a specific user.

Parameters:

• context – The Behave context

• user_index – Index of the user in the users list

Returns:

User data from the response

steps.api.create_job_token(context: steps.api.utils.Context, token_label: str, user_index: str) → None

Create a new job token for a specific user.

Parameters:

• context – The Behave context

• token_label – Label for the new token

• user_index – Index of the user to create the token for

steps.api.delete_platform(context: steps.api.utils.Context, platform_id_method: str) → None

Delete a platform by its ID.

Parameters:

• context – The Behave context

• platform_id_method – Method to get the platform ID

steps.api.register_platform_simplify(context: steps.api.utils.Context, platform_name_method) → None

steps.api.get_platform(context: steps.api.utils.Context, platform_name_or_id_method: str) → steps.api.cloud.Platform

steps.api.register_platform(context: steps.api.utils.Context) → None

steps.api.fetch_platform_job(context: steps.api.utils.Context, platform_name_method: dict = None) → None

steps.api.run_platform_job(context: steps.api.utils.Context, job_id_method: dict = None) → None

steps.api.progress_job(context: steps.api.utils.Context, job_id_method: dict = None) → None

steps.api.finish_job(context: steps.api.utils.Context, job_id_method: dict = None) → None

steps.api.assign_user_to_platform(context: steps.api.utils.Context, user_index: str, platform_id_method: str)

steps.api.assign_token_to_platform(context: steps.api.utils.Context, job_token_id_method: str, platform_id_method: str) → None

steps.api.assign_user_and_token_to_platform(context: steps.api.utils.Context, user_index: str, platform_name_method: str)

steps.api.create_job_simplified(context: steps.api.utils.Context, user_index: str)

steps.api.create_job(context: steps.api.utils.Context, user_index: str, platform_name_method: str) → None

steps.api.create_job_payload(context: steps.api.utils.Context, user_index: str) → None

steps.api.get_job_status(context: steps.api.utils.Context, job_id_method: str, user_index: str) → None

steps.api.get_job_result(context: steps.api.utils.Context, job_id_method: str, user_index: str) → None

steps.api.get_job_info(context: steps.api.utils.Context, job_id_method: str) → None

steps.api.cancel_job(context: steps.api.utils.Context, job_id_method: str, user_index: str) → None

steps.api.step_impl(context, state)

class steps.api.Request(api=None, timeout: int = 3)

Bases: requests.Request

A user-created Request object.

Used to prepare a PreparedRequest, which is sent to the server.

Parameters:

• method – HTTP method to use.

• url – URL to send.

• headers – dictionary of headers to send.

• files – dictionary of {filename: fileobject} files to multipart upload.

• data – the body to attach to the request. If a dictionary or list of tuples [(key, value)] is provided, form-encoding will take place.

• json – json for the body to attach to the request (if files or data is not specified).

• params – URL parameters to append to the URL. If a dictionary or list of tuples [(key, value)] is provided, form-encoding will take place.

• auth – Auth handler or (user, pass) tuple.

• cookies – dictionary or CookieJar of cookies to attach to this request.

• hooks – dictionary of callback hooks, for internal usage.

Usage:

>>> import requests
>>> req = requests.Request('GET', 'https://httpbin.org/get')
>>> req.prepare()
<PreparedRequest [GET]>

api: steps.api.utils.api.Api = None

method = ''

headers

params

body

original_path_url = ''

path_parameters

path_url = ''

url = ''

log = True

timeout = 3

expected_status_code = '200'

response = None

set_header(key, value)

set_body(key, value)

set_param(key, value)

set_authorization_header(value)

set_json_content_type()

set_form_content_type()

set_url(path_url: str, path_parameters: dict = {})

set_method(method: str)

send()

get_pretty_request() → str

get_pretty_response() → str

copy()

class steps.api.Context

output_dir: str = ''

logger: logging.Logger = None

apis: dict[str, steps.api.utils.api.Api]

cloud: steps.api.cloud.Cloud = None

request: steps.api.utils.api.Request = None

response = None

steps.api.log_allure(title: str, text: str) → None

steps.api.attach_swagger_link(context: steps.api.utils.context.Context)

steps.api.parse_nested_value(value, context)

steps.api.try_read_from_variable(method: str, context)

steps.api.try_save_value_to_variable(context, value, method: str)

steps.api.set_header_key(context: steps.api.utils.Context, key: str, method: str) → None

Set a specific HTTP header for the request.

Attempts to read the value from a variable using the provided method, and adds special handling for Authorization headers.

Parameters:

• context – The Behave context

• key – Header key to set

• method – Method to get the value (variable name or literal)

Raises:

RuntimeWarning – If the value cannot be set using the method

steps.api.get_table(context: steps.api.utils.Context) → dict

Convert a Gherkin table to a Python dictionary.

Processes a Behave table with ‘key’ and ‘value’ columns, parsing nested values and variable references in the process.

Parameters:

context – The Behave context containing the table

Returns:

Dictionary with keys and values from the table

steps.api.set_body(context: steps.api.utils.Context) → None

Set the request body from a Gherkin table.

Updates the request body with key-value pairs from the table.

Parameters:

context – The Behave context containing the table

steps.api.set_param_key(context: steps.api.utils.Context, key: str, value)

Set a query parameter for the request.

Processes the value to handle variable references and nested values.

Parameters:

• context – The Behave context

• key – Parameter key to set

• value – Parameter value (raw or variable reference)

steps.api.set_request_params(context: steps.api.utils.Context) → None

Set request query parameters from a Gherkin table.

Processes each row in the table to set request parameters.

Parameters:

context – The Behave context containing the table

steps.api.save_request(context: steps.api.utils.Context)

Save the current request as a reference for later use.

Parameters:

context – The Behave context

steps.api.prepare_request(context: steps.api.utils.Context, api_type: str)

Prepare a new request for a specific API type.

Parameters:

• context – The Behave context

• api_type – Type of API to prepare the request for

Raises:

AssertionError – If the API type is not available in context.apis

steps.api.send_complete_request(context: steps.api.utils.Context)

Send the prepared request and handle response and logging.

Sends the request, handles potential exceptions, and logs request and response details based on the context configuration.

Parameters:

context – The Behave context

Raises:

AssertionError – If connection errors occur

steps.api.send_request_with_path_parameter(context, request_verb: str, current_path: str, path_key: str, method: str, log_response: bool = True) → None

Make a request with a path parameter to a specific endpoint.

Configures and sends a request with a path parameter replaced by the value obtained from the specified method.

Parameters:

• context – The Behave context

• request_verb – HTTP method (GET, POST, etc.)

• current_path – URL path template

• path_key – Key name in the path to replace with value

• method – Method to obtain the value for the path parameter

• log_response – Whether to log the response (default: True)

steps.api.resend_request(context: steps.api.utils.Context, log_response: bool = True) → None

steps.api.resend_request_setting_header(context: steps.api.utils.Context, key: str, method: str, log_response: bool = True) → None

steps.api.update_ref_request_header(context: steps.api.utils.Context, key: str, method: str, log_response: bool = True) → None

steps.api.resend_request_setting_body(context: steps.api.utils.Context, log_response: bool = True) → None

steps.api.resend_request_with_path_parameter(context, path_key: str, method: str, log_response: bool = True) → None

steps.api.send_request(context: steps.api.utils.Context, request_verb: str, current_path: str, log_response: bool = True) → None

Make a request to a specific endpoint.

Configures and sends a request to the specified path.

Parameters:

• context – The Behave context

• request_verb – HTTP method (GET, POST, etc.)

• current_path – URL path

• log_response – Whether to log the response (default: True)

steps.api.save_random_uuid(context: steps.api.utils.Context, method: str) → None

steps.api.save_variable(context: steps.api.utils.Context, variable_key: str, method: str) → None

steps.api.save_data_in_response(context: steps.api.utils.Context, variable_key: str, method: str) → None

Save a field from the response to a variable.

Extracts a field from the response JSON and saves it using the specified method.

Parameters:

• context – The Behave context

• variable_key – Key in the response to extract

• method – Method to save the value (variable specification)

steps.api.add_test_section_description(context: steps.api.utils.Context, test_section_description: str)

steps.api.PAGINATION_RESULT_KEYS = ['content', 'currentPage', 'totalElements', 'totalPages']

steps.api.check_not_page_result(context)

Check that the response is not in pagination format.

Verifies that the response doesn’t contain the standard pagination fields (content, currentPage, totalElements, totalPages).

Parameters:

context – The Behave context

Raises:

AssertionError – If the response has the standard pagination format

steps.api.check_expected_values_in_response(context, key: str, value: str)

steps.api.get_status_exception(expected_status, actual_status) → AWrongStatus

steps.api.check_forbidden_keyword(my_obj) → str

exception steps.api.MissingSchema(request, schema: str = None)

Bases: ASchemaError

Assertion failed.

get_message()

exception steps.api.DefaultSchemaIsUsed(request, schema: str = None)

Bases: ASchemaError

Assertion failed.

get_message()

exception steps.api.ForbiddenKeyword(request, schema: str, errors: list[str])

Bases: ASchemaError

Assertion failed.

errors

get_message()

exception steps.api.SchemaError(request, schema: str, errors: list[str])

Bases: ASchemaError

Assertion failed.

errors

get_message()

exception steps.api.MissingEndpoint(request, schema: str = None)

Bases: ASchemaError

Assertion failed.

get_message()

steps.api.status_code_validation(context, expected_http_status_code)

Validate that the response has the expected HTTP status code.

Parameters:

• context – The Behave context

• expected_http_status_code – Expected HTTP status code as string

Raises:

MissingEndpoint – If 404 wasn’t expected but was received with the standard message

steps.api.check_response_from_table(context)

Check that the response contains the expected key-value pairs from a table.

Expects a Gherkin table with ‘key’ and ‘value’ columns, and validates each key-value pair against the response JSON.

Parameters:

context – The Behave context containing a table of keys and values

steps.api.check_empty_response(context)

Check that the response body is empty (empty JSON object or array).

Parameters:

context – The Behave context

steps.api.check_not_empty_response(context)

Check that the response body is not empty (has content).

Parameters:

context – The Behave context

steps.api.response_match_defined_schema(context)

Validate that the response JSON matches the defined schema.

Performs schema validation against the defined schema for the current endpoint, checking for various schema-related issues including default schemas, forbidden keywords, and data validation errors.

Parameters:

context – The Behave context