Source code for steps.api.utils.parser

import re
import json
import uuid
import operator
from functools import reduce

[docs] RANDOM_UUID_KEY = "random uuid"
[docs] ANYTHING_IN_QM_BUT_QM = r"\"([^\"]+)\""
[docs] ANY_INT = r"(\d+)"
[docs] ANY_FLOAT = r"(\d+\.\d*)"
[docs] STRING_METHOD_PATTERN = re.compile(rf"^string {ANYTHING_IN_QM_BUT_QM}$")
[docs] INT_METHOD_PATTERN = re.compile(rf"^int {ANY_INT}$")
[docs] FLOAT_METHOD_PATTERN = re.compile(rf"^float {ANY_FLOAT}$")
[docs] USER_VARIABLE_METHOD_PATTERN = re.compile(rf"^variable {ANYTHING_IN_QM_BUT_QM} of user {ANY_INT}$")
[docs] ADMIN_VARIABLE_METHOD_PATTERN = re.compile(rf"^variable {ANYTHING_IN_QM_BUT_QM} of admin$")
[docs] PLATFORM_VARIABLE_METHOD_PATTERN = re.compile(rf"^variable {ANYTHING_IN_QM_BUT_QM} of default platform$")
[docs] VARIABLE_METHOD_PATTERN = re.compile(rf"^temporary variable {ANYTHING_IN_QM_BUT_QM}$")
[docs] ENV_VARIABLE_METHOD_PATTERN = re.compile(rf"^environnement variable {ANYTHING_IN_QM_BUT_QM}$")
[docs] PARSER_DICT = { VARIABLE_METHOD_PATTERN: "temp", USER_VARIABLE_METHOD_PATTERN: "users", ADMIN_VARIABLE_METHOD_PATTERN: "admin", ENV_VARIABLE_METHOD_PATTERN: "environnement", PLATFORM_VARIABLE_METHOD_PATTERN: "default_platform" }
[docs] TYPE_DICT = { STRING_METHOD_PATTERN: str, INT_METHOD_PATTERN: int, FLOAT_METHOD_PATTERN: float }
[docs] EVAL_METHODS = ["any", "all", "none"]
[docs] def find(element, my_dict): return reduce(operator.getitem, element.split('/'), my_dict)
# Handle nested values
[docs] def to_float(value: str) -> float | None: try: return float(value) except ValueError: return None
[docs] def to_int(value: str) -> int | None: try: return int(value) except ValueError: return None
# Check value possibly from context variable
[docs] def parse_string(value: str): if value == 'None': return None if value == 'true': return True if value == 'false': return False if value.startswith("{") or value.startswith("["): return json.loads(value) temp = to_int(value) if temp is not None: return temp temp = to_float(value) if temp is not None: return temp return value
[docs] def parse_nested_value(value, context): return_value = value if isinstance(value, str): return_value = parse_string(value) if isinstance(return_value, str): return_value = try_read_from_variable(return_value, context) if isinstance(return_value, dict): return_value = return_value.copy() for key, temp_value in return_value.items(): temp_dict_value = parse_nested_value(temp_value, context) if temp_dict_value: return_value[key] = temp_dict_value elif isinstance(return_value, list): parsed_value = [] for temp_value in return_value: temp_list_value = parse_nested_value(temp_value, context) if temp_list_value: parsed_value.append(temp_list_value) else: parsed_value.append(temp_value) return_value = parsed_value return return_value
[docs] def get_true_keys(str_keys: list, my_dict: dict | list) -> list: keys = [] for key in str_keys: if isinstance(my_dict, dict): keys.append(key) elif isinstance(my_dict, list): keys.append(int(key)) else: raise NotImplementedError(f"Cannot handle type {type(dict)}") my_dict = my_dict[keys[-1]] return keys
[docs] def get_nested_value(keys: list, my_dict: dict): keys = get_true_keys(keys, my_dict) for key in keys[:-1]: if key in my_dict or isinstance(my_dict, list): my_dict = my_dict[key] elif key in my_dict.__dict__: my_dict = my_dict.__dict__[key] else: raise KeyError(f"Can't find {key} in {my_dict}") return my_dict[keys[-1]]
[docs] def delete_nested_value(keys: list, my_dict: dict): keys = get_true_keys(keys, my_dict) for key in keys[:-1]: my_dict = my_dict[key] del my_dict[keys[-1]]
[docs] def set_nested_value(keys: list, my_dict: dict, value): for key in keys[:-1]: if isinstance(key, str): my_dict = my_dict.setdefault(key, {}) elif isinstance(key, int): my_dict = my_dict[key] my_dict[keys[-1]] = value
# Read / Save in context variable
[docs] def try_read_from_variable(method: str, context): if method == RANDOM_UUID_KEY: return str(uuid.uuid4()) for parse_pattern, dict_name in PARSER_DICT.items(): keys = parse_pattern.findall(method) if keys: keys = keys[0] if not isinstance(keys, str): keys = list(keys) keys.reverse() else: keys = [keys] return get_nested_value(keys, context.cloud.__dict__[dict_name]) for type_pattern, type_to_cast in TYPE_DICT.items(): keys = type_pattern.findall(method) if keys: return type_to_cast(keys[0]) return method
[docs] def try_save_value_to_variable(context, value, method: str): for pattern, dict_name in PARSER_DICT.items(): keys = pattern.findall(method) if keys: keys = keys[0] if not isinstance(keys, str): new_keys = [] for key in keys: try: key = int(key) except: pass new_keys.insert(0, key) keys = new_keys else: keys = [keys] set_nested_value(keys, context.cloud.__dict__[dict_name], value) return raise RuntimeWarning(f"Cannot save response to variable with method {method}")
[docs] def check_expected_values_in_response(context, key: str, value: str): response = context.request.response.json() # value treatment expected_value = parse_string(value) if isinstance(expected_value, str): expected_value = try_read_from_variable(expected_value, context) # key treatment eval_method = None response_value = None for eval_method in EVAL_METHODS: if key.startswith(eval_method+"("): try: response_value = eval(key[len(eval_method)+1:-1]) except KeyError as e: raise AssertionError(f"Cannot find key {e.args[0]} for entry {key}") break if response_value is None: eval_method = None if '[' in key: dict_name = key.split('[')[0] temp_value = response[dict_name] if '{' in temp_value: temp_value = json.loads(temp_value) try: response_value = eval(key.replace(dict_name, "temp_value")) except KeyError as e: not_found_key = e.args[0] index = key.rindex(not_found_key) index = key[:index].rindex('[') assert False, f"Couldn't find key {not_found_key} in {key[:index]} from response: {response}" else: assert key in response, f"Couldn't find key {key} in {response}" response_value = response[key] # assertion treatment if eval_method: temp = [value == expected_value for value in response_value] if eval_method == 'any': result = any(temp) elif eval_method == 'all': result = all(temp) elif eval_method == "none": result = not any(temp) else: result = response_value == expected_value return result, expected_value, response_value