Source code for steps.api.given

import requests

from behave import given

from .utils import Request, Context, \
    log_allure, attach_swagger_link, \
    parse_nested_value, try_read_from_variable, \
    try_save_value_to_variable


@given('I set "{key}" header to {method}')


def set_header_key(context: 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
    """
    temp_value = str(try_read_from_variable(method, context))
    if temp_value:
        if key == "Authorization":
            temp_value = "Bearer " + temp_value
        context.request.headers[key] = temp_value
        return
    raise RuntimeWarning(f"Cannot set header with method {method}")





def get_table(context: 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
    """
    table = {}
    for row in context.table:
        key = row['key'].strip()
        value = row['value']
        if isinstance(value, str):
            value = value.strip()
            value = parse_nested_value(value, context)
        table[key] = value
    return table



@given('I set the request body as follow')


def set_body(context: 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
    """
    context.request.body.update(get_table(context))





def set_param_key(context: 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)
    """
    key = key.strip()
    if isinstance(value, str):
        value = value.strip()
        value = parse_nested_value(value, context)
    value = parse_nested_value(value, context)
    context.request.params[key] = value



@given('I set the request params as follow')


def set_request_params(context: 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
    """
    for row in context.table:
        set_param_key(context, row['key'], row['value'])



@given('I am setting the previous request as reference')


def save_request(context: Context):
    """Save the current request as a reference for later use.

    :param context: The Behave context
    """
    context.ref_request = context.request.copy()



@given('I am preparing a request for {api_type} api')


def prepare_request(context: 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
    """
    assert api_type in context.apis
    context.request = Request(api=context.apis[api_type])





def send_complete_request(context: 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
    """
    status_code = None
    try:
        context.request.send()
        status_code = context.request.response.status_code
    except requests.exceptions.ConnectTimeout:
        raise AssertionError("ConnectTimeout from api call")
    except requests.exceptions.ConnectionError as e:
        raise AssertionError(f"Connection Error from api call: {e}")
    finally:
        if context.request.log and context.in_scenario:
            log_allure("Request", str(context.request))
            attach_swagger_link(context)
    if context.request.log and context.in_scenario:
        log_allure("Response", context.request.get_pretty_response())
    context.logger.info(f"{context.request.method} - {context.request.url} : {status_code}")



@given('I make a {request_verb} request to "{current_path}" with path parameter "{path_key}" as {method}')


def 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)
    """
    context.request.original_path_url = current_path
    context.request.method = request_verb
    context.request.log = log_response
    context.request.set_url(path_url=current_path, path_parameters={
                            path_key: try_read_from_variable(method, context)})
    send_complete_request(context)



@given('I make the same request')


def resend_request(context: Context, log_response: bool = True) -> None:
    context.request = context.ref_request.copy()
    context.request.log = log_response
    send_complete_request(context)



@given('I make the same request setting "{key}" header to {method}')


def resend_request_setting_header(context: Context, key: str, method: str, log_response: bool = True) -> None:
    context.request = context.ref_request.copy()
    context.request.log = log_response
    set_header_key(context, key, method)
    send_complete_request(context)



@given('I am updating the reference request setting "{key}" header to {method}')


def update_ref_request_header(context: Context, key: str, method: str, log_response: bool = True) -> None:
    context.request = context.ref_request
    context.request.log = log_response
    set_header_key(context, key, method)
    context.ref_request = context.request.copy()



@given('I make the same request setting the request body as follow')


def resend_request_setting_body(context: Context, log_response: bool = True) -> None:
    context.request = context.ref_request.copy()
    context.request.body = {}
    context.request.log = log_response
    set_body(context)
    send_complete_request(context)



@given('I make the same request with path parameter "{path_key}" as {method}')


def resend_request_with_path_parameter(
        context,
        path_key: str,
        method: str,
        log_response: bool = True) -> None:
    context.request = context.ref_request.copy()
    context.request.log = log_response
    context.request.set_url(path_url=context.request.original_path_url, path_parameters={
                            path_key: try_read_from_variable(method, context)})
    send_complete_request(context)



@given('I make a {request_verb} request to "{current_path}"')


def send_request(context: 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)
    """
    context.request.original_path_url = current_path
    context.request.method = request_verb
    context.request.log = log_response
    context.request.set_url(path_url=current_path)
    send_complete_request(context)



@given('I am saving a random uuid as {method}')


def save_random_uuid(context: Context, method: str) -> None:
    try_save_value_to_variable(context, try_read_from_variable("random uuid", context), method)



@given('I am saving "{variable_key}" as {method}')


def save_variable(context: Context, variable_key: str, method: str) -> None:
    try_save_value_to_variable(context, variable_key, method)



@given('I am saving field "{variable_key}" from response as {method}')


def save_data_in_response(context: 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)
    """
    try_save_value_to_variable(context, context.request.response.json()[variable_key], method)



@given('The next section is testing {test_section_description}')


def add_test_section_description(context: Context, test_section_description: str):
    context.logger.info(f"Testing {test_section_description}")