steps.api ========= .. py:module:: steps.api Submodules ---------- .. toctree:: :maxdepth: 1 /autoapi/steps/api/cloud/index /autoapi/steps/api/cloud_objects/index /autoapi/steps/api/given/index /autoapi/steps/api/given_cloud/index /autoapi/steps/api/then/index /autoapi/steps/api/then_cloud/index /autoapi/steps/api/utils/index Attributes ---------- .. autoapisummary:: steps.api.PAGINATION_RESULT_KEYS Exceptions ---------- .. autoapisummary:: steps.api.MissingSchema steps.api.DefaultSchemaIsUsed steps.api.ForbiddenKeyword steps.api.SchemaError steps.api.MissingEndpoint Classes ------- .. autoapisummary:: steps.api.Context steps.api.Platform steps.api.Job steps.api.JobToken steps.api.Request steps.api.Context Functions --------- .. autoapisummary:: steps.api.try_read_from_variable steps.api.get_table steps.api.prepare_request steps.api.prepare_cloud_request steps.api.log_as_user steps.api.log_as_admin steps.api.get_user steps.api.create_job_token steps.api.delete_platform steps.api.register_platform_simplify steps.api.get_platform steps.api.register_platform steps.api.fetch_platform_job steps.api.run_platform_job steps.api.progress_job steps.api.finish_job steps.api.assign_user_to_platform steps.api.assign_token_to_platform steps.api.assign_user_and_token_to_platform steps.api.create_job_simplified steps.api.create_job steps.api.create_job_payload steps.api.get_job_status steps.api.get_job_result steps.api.get_job_info steps.api.cancel_job steps.api.step_impl steps.api.log_allure steps.api.attach_swagger_link steps.api.parse_nested_value steps.api.try_read_from_variable steps.api.try_save_value_to_variable steps.api.set_header_key steps.api.get_table steps.api.set_body steps.api.set_param_key steps.api.set_request_params steps.api.save_request steps.api.prepare_request steps.api.send_complete_request steps.api.send_request_with_path_parameter steps.api.resend_request steps.api.resend_request_setting_header steps.api.update_ref_request_header steps.api.resend_request_setting_body steps.api.resend_request_with_path_parameter steps.api.send_request steps.api.save_random_uuid steps.api.save_variable steps.api.save_data_in_response steps.api.add_test_section_description steps.api.check_not_page_result steps.api.check_expected_values_in_response steps.api.get_status_exception steps.api.check_forbidden_keyword steps.api.status_code_validation steps.api.check_response_from_table steps.api.check_empty_response steps.api.check_not_empty_response steps.api.response_match_defined_schema Package Contents ---------------- .. py:class:: Context .. py:attribute:: output_dir :type: str :value: '' .. py:attribute:: logger :type: logging.Logger :value: None .. py:attribute:: apis :type: dict[str, steps.api.utils.api.Api] .. py:attribute:: cloud :type: steps.api.cloud.Cloud :value: None .. py:attribute:: request :type: steps.api.utils.api.Request :value: None .. py:attribute:: response :value: None .. py:function:: try_read_from_variable(method: str, context) .. py:class:: Platform(name: str = '', type: str = 'simulator', platform_id: str = None, worker_id: str = str(uuid4()), specs: dict[str] = {'available_commands': ['probs']}, **kwargs) Bases: :py:obj:`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. :param name: Name of the platform :param type: Type of platform (default: "simulator") :param platform_id: Platform ID (None for new platforms) :param worker_id: Worker ID (defaults to a new UUID) :param specs: Platform specifications :param kwargs: Additional attributes to set on the platform :raises AssertionError: If Platform.ADMIN is None or not an admin .. py:attribute:: ADMIN :value: None .. py:attribute:: REGISTER_ENDPOINT :value: '/admin/platforms/register' .. py:attribute:: JOB_ENDPOINT :value: '/admin/job' .. py:attribute:: ASSIGN_ENDPOINT :value: '/api/platform-permissions' .. py:method:: get_object_name() -> str :classmethod: .. py:method:: get_endpoint() -> str :classmethod: .. py:method:: get_api(context: steps.api.utils.CloudContext) -> steps.api.utils.Api :classmethod: .. py:method:: get_token_method() -> str :classmethod: .. py:attribute:: PLATFORM_NAME_KEY :value: 'platform_name' .. py:attribute:: DEFAULT_NAME :value: 'sim:qa' .. py:attribute:: name :value: '' .. py:attribute:: type :value: 'simulator' .. py:attribute:: worker_id :type: str :value: '' .. py:attribute:: platform_id :type: str .. py:attribute:: specs .. py:method:: register(context: steps.api.utils.CloudContext, register_dict: dict[str] = None) -> None Register this platform in the cloud. :param context: Cloud context for API configuration :param register_dict: Dictionary with registration parameters (optional) :raises RuntimeError: If platform registration fails .. py:method:: read(context: steps.api.utils.CloudContext) -> dict[str] Read this platform's information from the cloud. :param context: Cloud context for API configuration :return: Dictionary with platform data from response :raises RuntimeError: If reading platform data fails .. py:method:: update(context: steps.api.utils.CloudContext, update_dict: dict) -> None Update this platform's information in the cloud. :param context: Cloud context for API configuration :param update_dict: Dictionary with fields to update :raises RuntimeError: If updating platform data fails .. py:method:: delete(context: steps.api.utils.CloudContext) -> None Delete this platform from the cloud. :param context: Cloud context for API configuration :raises RuntimeError: If deleting the platform fails .. py:method:: fetch_job(context: steps.api.utils.CloudContext) -> steps.api.cloud_objects.cloud_job.Job Fetch a job for this platform to process. :param context: Cloud context for API configuration :return: Job object for the fetched job :raises RuntimeError: If fetching a job fails .. py:method:: run_job(context: steps.api.utils.CloudContext, job) -> None .. py:method:: progress_job(context: steps.api.utils.CloudContext, job, progress_dict: dict[str]) -> None .. py:method:: finish_job(context: steps.api.utils.CloudContext, job, finish_dict: dict[str]) -> None .. py:method:: assign_user(context: steps.api.utils.CloudContext, user: steps.api.cloud_objects.cloud_user.User) -> None .. py:method:: assign_token(context: steps.api.utils.CloudContext, job_token) -> None .. py:method:: assign_company(context: steps.api.utils.CloudContext, company_id: int) -> None .. py:method:: list_all(context: steps.api.utils.CloudContext) -> list[Platform] :classmethod: List all platforms in the cloud. :param context: Cloud context for API configuration :return: List of Platform objects :raises AssertionError: If ADMIN is not an admin :raises RuntimeError: If listing platforms fails .. py:class:: 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: :py:obj:`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. :param platform_name: Name of the platform to run the job on :param job_name: Name of the job :param payload: Job payload containing command, circuit, parameters, etc. :param pcvl_version: Version of the PCVL library to use :param process_id: Unique ID for the process (defaults to a new UUID) :param job_id: Job ID (empty for new jobs) :param _: Additional attributes to set on the job .. py:attribute:: JOB_NAME_KEY :value: 'job_name' .. py:attribute:: PLATFORM_NAME_KEY :value: 'platform_name' .. py:attribute:: PAYLOAD_KEY :value: 'payload' .. py:attribute:: PCVL_VERSION_KEY :value: 'pcvl_version' .. py:attribute:: PROCESS_ID_KEY :value: 'process_id' .. py:attribute:: DEFAULT_PAYLOAD .. py:method:: get_object_name() -> str :classmethod: .. py:method:: get_endpoint() -> str :classmethod: .. py:method:: get_api(context: steps.api.utils.CloudContext) -> steps.api.utils.Api :classmethod: .. py:method:: get_token_method() -> str :classmethod: .. py:attribute:: DEFAULT_PLATFORM_NAME :value: 'sim:qa' .. py:attribute:: platform_name :type: str :value: 'sim:qa' .. py:attribute:: job_name :type: str :value: 'qa_job' .. py:attribute:: payload :type: dict[str] .. py:attribute:: pcvl_version :type: str :value: '0.12.0' .. py:attribute:: process_id :type: str :value: '' .. py:attribute:: job_id :value: '' .. py:method:: create(context: steps.api.utils.CloudContext, user, create_dict: dict = {}) -> None Create this job in the cloud. :param context: Cloud context for API configuration :param user: User providing authentication :param create_dict: Additional parameters for job creation :raises RuntimeError: If job creation fails .. py:method:: read(context: steps.api.utils.CloudContext, user) -> dict[str] Read this job's information from the cloud. :param context: Cloud context for API configuration :param user: User providing authentication :return: Dictionary with job data from response :raises RuntimeError: If reading job data fails .. py:method:: cancel(context: steps.api.utils.CloudContext, user) -> dict[str] Cancel this job. :param context: Cloud context for API configuration :param user: User providing authentication :raises RuntimeError: If canceling the job fails .. py:method:: get_status(context: steps.api.utils.CloudContext, user) -> dict[str] Get the current status of this job. :param context: Cloud context for API configuration :param user: User providing authentication :return: Dictionary with job status information :raises RuntimeError: If retrieving job status fails .. py:method:: get_results(context: steps.api.utils.CloudContext, user) -> dict[str] Get the results of this job. :param context: Cloud context for API configuration :param user: User providing authentication :return: Dictionary with job results :raises RuntimeError: If retrieving job results fails .. py:method:: cancel_many(context: steps.api.utils.CloudContext, jobs: list, admin: steps.api.cloud_objects.cloud_user.User) -> None :classmethod: Cancel multiple jobs at once. :param context: Cloud context for API configuration :param jobs: List of Job objects to cancel :param admin: Admin user providing authentication :raises AssertionError: If the user is not an admin :raises RuntimeError: If canceling the jobs fails .. py:method:: list_all(context: steps.api.utils.CloudContext, admin: steps.api.cloud_objects.cloud_user.User) -> list[Job] :classmethod: List all jobs in the cloud. :param context: Cloud context for API configuration :param admin: Admin user providing authentication :return: List of Job objects :raises AssertionError: If the user is not an admin :raises RuntimeError: If listing jobs fails .. py:class:: JobToken(label: str = None, duration: int = 99999, token_id: int = -1, user_id: int = -1, token: str = '', **kwargs) Bases: :py:obj:`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. :param label: Label for the token :param duration: Token duration in seconds :param token_id: Token ID (defaults to -1 for new tokens) :param user_id: ID of the user the token belongs to :param token: Token string :param kwargs: Additional attributes to set on the token .. py:method:: get_object_name() -> str :classmethod: .. py:method:: get_endpoint() -> str :classmethod: .. py:method:: get_api(context: steps.api.utils.CloudContext) -> steps.api.utils.Api :classmethod: .. py:method:: get_token_method() -> str :classmethod: .. py:attribute:: DURATION_KEY :value: 'duration' .. py:attribute:: LABEL_KEY :value: 'label' .. py:attribute:: label :type: str :value: None .. py:attribute:: token_id :type: int .. py:attribute:: user_id :type: int :value: -1 .. py:attribute:: duration :type: int :value: 99999 .. py:attribute:: token :type: str :value: '' .. py:method:: create(context: steps.api.utils.CloudContext, user: steps.api.cloud_objects.cloud_user.User) -> None Create this job token in the cloud. :param context: Cloud context for API configuration :param user: User providing authentication and owning the token :raises AssertionError: If the user doesn't have a login token :raises RuntimeError: If job token creation fails .. py:method:: read(context: steps.api.utils.CloudContext, user: steps.api.cloud_objects.cloud_user.User) -> None Read this job token's information from the cloud. :param context: Cloud context for API configuration :param user: User providing authentication :return: Dictionary with token data from response :raises AssertionError: If the user doesn't have job tokens or the token wasn't created :raises RuntimeError: If reading token data fails .. py:method:: delete(context: steps.api.utils.CloudContext, user: steps.api.cloud_objects.cloud_user.User) -> None Delete this job token from the cloud. :param context: Cloud context for API configuration :param user: User providing authentication :raises AssertionError: If the token wasn't created :raises RuntimeError: If deleting the token fails .. py:method:: list_all(context: steps.api.utils.CloudContext, admin: steps.api.cloud_objects.cloud_user.User) -> list[JobToken] :classmethod: List all job tokens in the cloud. :param context: Cloud context for API configuration :param admin: Admin user providing authentication :return: List of JobToken objects :raises AssertionError: If the user is not an admin :raises RuntimeError: If listing tokens fails .. py:function:: 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. :param context: The Behave context containing the table :return: Dictionary with keys and values from the table .. py:function:: prepare_request(context: steps.api.utils.Context, api_type: str) Prepare a new request for a specific API type. :param context: The Behave context :param api_type: Type of API to prepare the request for :raises AssertionError: If the API type is not available in context.apis .. py:function:: prepare_cloud_request(context) Prepare a new request for the cloud API. Shorthand for creating a request to the cloud API type. :param context: The Behave context .. py:function:: log_as_user(context: steps.api.utils.Context, user_index: str) -> None Log in as a specific test user. :param context: The Behave context :param user_index: Index of the user in the users list .. py:function:: log_as_admin(context: steps.api.utils.Context) -> None Log in as the admin user. :param context: The Behave context .. py:function:: get_user(context: steps.api.utils.Context, user_index: str) Get information about a specific user. :param context: The Behave context :param user_index: Index of the user in the users list :return: User data from the response .. py:function:: create_job_token(context: steps.api.utils.Context, token_label: str, user_index: str) -> None Create a new job token for a specific user. :param context: The Behave context :param token_label: Label for the new token :param user_index: Index of the user to create the token for .. py:function:: delete_platform(context: steps.api.utils.Context, platform_id_method: str) -> None Delete a platform by its ID. :param context: The Behave context :param platform_id_method: Method to get the platform ID .. py:function:: register_platform_simplify(context: steps.api.utils.Context, platform_name_method) -> None .. py:function:: get_platform(context: steps.api.utils.Context, platform_name_or_id_method: str) -> steps.api.cloud.Platform .. py:function:: register_platform(context: steps.api.utils.Context) -> None .. py:function:: fetch_platform_job(context: steps.api.utils.Context, platform_name_method: dict = None) -> None .. py:function:: run_platform_job(context: steps.api.utils.Context, job_id_method: dict = None) -> None .. py:function:: progress_job(context: steps.api.utils.Context, job_id_method: dict = None) -> None .. py:function:: finish_job(context: steps.api.utils.Context, job_id_method: dict = None) -> None .. py:function:: assign_user_to_platform(context: steps.api.utils.Context, user_index: str, platform_id_method: str) .. py:function:: assign_token_to_platform(context: steps.api.utils.Context, job_token_id_method: str, platform_id_method: str) -> None .. py:function:: assign_user_and_token_to_platform(context: steps.api.utils.Context, user_index: str, platform_name_method: str) .. py:function:: create_job_simplified(context: steps.api.utils.Context, user_index: str) .. py:function:: create_job(context: steps.api.utils.Context, user_index: str, platform_name_method: str) -> None .. py:function:: create_job_payload(context: steps.api.utils.Context, user_index: str) -> None .. py:function:: get_job_status(context: steps.api.utils.Context, job_id_method: str, user_index: str) -> None .. py:function:: get_job_result(context: steps.api.utils.Context, job_id_method: str, user_index: str) -> None .. py:function:: get_job_info(context: steps.api.utils.Context, job_id_method: str) -> None .. py:function:: cancel_job(context: steps.api.utils.Context, job_id_method: str, user_index: str) -> None .. py:function:: step_impl(context, state) .. py:class:: Request(api=None, timeout: int = 3) Bases: :py:obj:`requests.Request` A user-created :class:`Request ` object. Used to prepare a :class:`PreparedRequest `, which is sent to the server. :param method: HTTP method to use. :param url: URL to send. :param headers: dictionary of headers to send. :param files: dictionary of {filename: fileobject} files to multipart upload. :param data: the body to attach to the request. If a dictionary or list of tuples ``[(key, value)]`` is provided, form-encoding will take place. :param json: json for the body to attach to the request (if files or data is not specified). :param params: URL parameters to append to the URL. If a dictionary or list of tuples ``[(key, value)]`` is provided, form-encoding will take place. :param auth: Auth handler or (user, pass) tuple. :param cookies: dictionary or CookieJar of cookies to attach to this request. :param hooks: dictionary of callback hooks, for internal usage. Usage:: >>> import requests >>> req = requests.Request('GET', 'https://httpbin.org/get') >>> req.prepare() .. py:attribute:: api :type: steps.api.utils.api.Api :value: None .. py:attribute:: method :value: '' .. py:attribute:: headers .. py:attribute:: params .. py:attribute:: body .. py:attribute:: original_path_url :value: '' .. py:attribute:: path_parameters .. py:attribute:: path_url :value: '' .. py:attribute:: url :value: '' .. py:attribute:: log :value: True .. py:attribute:: timeout :value: 3 .. py:attribute:: expected_status_code :value: '200' .. py:attribute:: response :value: None .. py:method:: set_header(key, value) .. py:method:: set_body(key, value) .. py:method:: set_param(key, value) .. py:method:: set_authorization_header(value) .. py:method:: set_json_content_type() .. py:method:: set_form_content_type() .. py:method:: set_url(path_url: str, path_parameters: dict = {}) .. py:method:: set_method(method: str) .. py:method:: send() .. py:method:: get_pretty_request() -> str .. py:method:: get_pretty_response() -> str .. py:method:: copy() .. py:class:: Context .. py:attribute:: output_dir :type: str :value: '' .. py:attribute:: logger :type: logging.Logger :value: None .. py:attribute:: apis :type: dict[str, steps.api.utils.api.Api] .. py:attribute:: cloud :type: steps.api.cloud.Cloud :value: None .. py:attribute:: request :type: steps.api.utils.api.Request :value: None .. py:attribute:: response :value: None .. py:function:: log_allure(title: str, text: str) -> None .. py:function:: attach_swagger_link(context: steps.api.utils.context.Context) .. py:function:: parse_nested_value(value, context) .. py:function:: try_read_from_variable(method: str, context) .. py:function:: try_save_value_to_variable(context, value, method: str) .. py:function:: 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. :param context: The Behave context :param key: Header key to set :param method: Method to get the value (variable name or literal) :raises RuntimeWarning: If the value cannot be set using the method .. py:function:: 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. :param context: The Behave context containing the table :return: Dictionary with keys and values from the table .. py:function:: 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. :param context: The Behave context containing the table .. py:function:: 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. :param context: The Behave context :param key: Parameter key to set :param value: Parameter value (raw or variable reference) .. py:function:: 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. :param context: The Behave context containing the table .. py:function:: save_request(context: steps.api.utils.Context) Save the current request as a reference for later use. :param context: The Behave context .. py:function:: prepare_request(context: steps.api.utils.Context, api_type: str) Prepare a new request for a specific API type. :param context: The Behave context :param api_type: Type of API to prepare the request for :raises AssertionError: If the API type is not available in context.apis .. py:function:: 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. :param context: The Behave context :raises AssertionError: If connection errors occur .. py:function:: 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. :param context: The Behave context :param request_verb: HTTP method (GET, POST, etc.) :param current_path: URL path template :param path_key: Key name in the path to replace with value :param method: Method to obtain the value for the path parameter :param log_response: Whether to log the response (default: True) .. py:function:: resend_request(context: steps.api.utils.Context, log_response: bool = True) -> None .. py:function:: resend_request_setting_header(context: steps.api.utils.Context, key: str, method: str, log_response: bool = True) -> None .. py:function:: update_ref_request_header(context: steps.api.utils.Context, key: str, method: str, log_response: bool = True) -> None .. py:function:: resend_request_setting_body(context: steps.api.utils.Context, log_response: bool = True) -> None .. py:function:: resend_request_with_path_parameter(context, path_key: str, method: str, log_response: bool = True) -> None .. py:function:: 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. :param context: The Behave context :param request_verb: HTTP method (GET, POST, etc.) :param current_path: URL path :param log_response: Whether to log the response (default: True) .. py:function:: save_random_uuid(context: steps.api.utils.Context, method: str) -> None .. py:function:: save_variable(context: steps.api.utils.Context, variable_key: str, method: str) -> None .. py:function:: 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. :param context: The Behave context :param variable_key: Key in the response to extract :param method: Method to save the value (variable specification) .. py:function:: add_test_section_description(context: steps.api.utils.Context, test_section_description: str) .. py:data:: PAGINATION_RESULT_KEYS :value: ['content', 'currentPage', 'totalElements', 'totalPages'] .. py:function:: 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). :param context: The Behave context :raises AssertionError: If the response has the standard pagination format .. py:function:: check_expected_values_in_response(context, key: str, value: str) .. py:function:: get_status_exception(expected_status, actual_status) -> AWrongStatus .. py:function:: check_forbidden_keyword(my_obj) -> str .. py:exception:: MissingSchema(request, schema: str = None) Bases: :py:obj:`ASchemaError` Assertion failed. .. py:method:: get_message() .. py:exception:: DefaultSchemaIsUsed(request, schema: str = None) Bases: :py:obj:`ASchemaError` Assertion failed. .. py:method:: get_message() .. py:exception:: ForbiddenKeyword(request, schema: str, errors: list[str]) Bases: :py:obj:`ASchemaError` Assertion failed. .. py:attribute:: errors .. py:method:: get_message() .. py:exception:: SchemaError(request, schema: str, errors: list[str]) Bases: :py:obj:`ASchemaError` Assertion failed. .. py:attribute:: errors .. py:method:: get_message() .. py:exception:: MissingEndpoint(request, schema: str = None) Bases: :py:obj:`ASchemaError` Assertion failed. .. py:method:: get_message() .. py:function:: status_code_validation(context, expected_http_status_code) Validate that the response has the expected HTTP status code. :param context: The Behave context :param expected_http_status_code: Expected HTTP status code as string :raises MissingEndpoint: If 404 wasn't expected but was received with the standard message .. py:function:: 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. :param context: The Behave context containing a table of keys and values .. py:function:: check_empty_response(context) Check that the response body is empty (empty JSON object or array). :param context: The Behave context .. py:function:: check_not_empty_response(context) Check that the response body is not empty (has content). :param context: The Behave context .. py:function:: 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. :param context: The Behave context