GithubBackup/httprunner
1
0
Fork 0
mirror of https://github.com/httprunner/httprunner.git synced 2026-05-12 02:21:29 +08:00
Code Issues Packages Projects Releases Wiki Activity

refactor context

Browse Source
This commit is contained in:
debugtalk
2018-08-22 16:45:47 +08:00
parent 9c45db775a
commit c8bb8b6fb3
10 changed files with 491 additions and 984 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
httprunner/api.py
View File

@@ -48,6 +48,8 @@ class HttpRunner(object):
}
"""
loader.reset_loader()
# load .env
loader.load_dot_env_file(dot_env_path)
@@ -98,6 +100,21 @@ class HttpRunner(object):
"""
if validator.is_testcases(path_or_testcases):
# TODO: refactor
if isinstance(path_or_testcases, list):
for testcase in path_or_testcases:
try:
dir_path = os.path.dirname(testcase["config"]["path"])
loader.load_debugtalk_module(dir_path)
except KeyError:
pass
else:
try:
dir_path = os.path.dirname(path_or_testcases["config"]["path"])
loader.load_debugtalk_module(dir_path)
except KeyError:
pass
testcases = path_or_testcases
else:
testcases = loader.load_testcases(path_or_testcases)
@@ -170,6 +187,9 @@ class HttpRunner(object):
config_variables,
self.project_mapping["debugtalk"]["functions"]
)
# put loaded project functions to config
testcase["config"]["functions"] = self.project_mapping["debugtalk"]["functions"]
parsed_testcases_list.append(testcase)
return parsed_testcases_list
@@ -300,6 +320,7 @@ class LocustRunner(object):
try:
self.runner.run(path)
except exceptions.MyBaseError as ex:
# TODO: refactor
from locust.events import request_failure
request_failure.fire(
request_type=test.testcase_dict.get("request", {}).get("method"),

586
httprunner/context.py
View File

@@ -1,447 +1,211 @@
# encoding: utf-8
import copy
import os
import random
import re
import sys
from httprunner import built_in, exceptions, loader, logger, parser, utils
from httprunner.compat import OrderedDict, basestring, builtin_str, str
from httprunner import exceptions, logger, parser, utils
from httprunner.compat import OrderedDict
# def parse_parameters(parameters, testset_path=None):
# """ parse parameters and generate cartesian product.
def parse_parameters(parameters, testset_path=None):
""" parse parameters and generate cartesian product.
# Args:
# parameters (list) parameters: parameter name and value in list
# parameter value may be in three types:
# (1) data list, e.g. ["iOS/10.1", "iOS/10.2", "iOS/10.3"]
# (2) call built-in parameterize function, "${parameterize(account.csv)}"
# (3) call custom function in debugtalk.py, "${gen_app_version()}"
Args:
parameters (list) parameters: parameter name and value in list
parameter value may be in three types:
(1) data list, e.g. ["iOS/10.1", "iOS/10.2", "iOS/10.3"]
(2) call built-in parameterize function, "${parameterize(account.csv)}"
(3) call custom function in debugtalk.py, "${gen_app_version()}"
# testset_path (str): testset file path, used for locating csv file and debugtalk.py
testset_path (str): testset file path, used for locating csv file and debugtalk.py
# Returns:
# list: cartesian product list
Returns:
list: cartesian product list
# Examples:
# >>> parameters = [
# {"user_agent": ["iOS/10.1", "iOS/10.2", "iOS/10.3"]},
# {"username-password": "${parameterize(account.csv)}"},
# {"app_version": "${gen_app_version()}"}
# ]
# >>> parse_parameters(parameters)
Examples:
>>> parameters = [
{"user_agent": ["iOS/10.1", "iOS/10.2", "iOS/10.3"]},
{"username-password": "${parameterize(account.csv)}"},
{"app_version": "${gen_app_version()}"}
]
>>> parse_parameters(parameters)
# """
# testcase_parser = TestcaseParser(file_path=testset_path)
"""
testcase_parser = TestcaseParser(file_path=testset_path)
# parsed_parameters_list = []
# for parameter in parameters:
# parameter_name, parameter_content = list(parameter.items())[0]
# parameter_name_list = parameter_name.split("-")
parsed_parameters_list = []
for parameter in parameters:
parameter_name, parameter_content = list(parameter.items())[0]
parameter_name_list = parameter_name.split("-")
# if isinstance(parameter_content, list):
# # (1) data list
# # e.g. {"app_version": ["2.8.5", "2.8.6"]}
# # => [{"app_version": "2.8.5", "app_version": "2.8.6"}]
# # e.g. {"username-password": [["user1", "111111"], ["test2", "222222"]}
# # => [{"username": "user1", "password": "111111"}, {"username": "user2", "password": "222222"}]
# parameter_content_list = []
# for parameter_item in parameter_content:
# if not isinstance(parameter_item, (list, tuple)):
# # "2.8.5" => ["2.8.5"]
# parameter_item = [parameter_item]
if isinstance(parameter_content, list):
# (1) data list
# e.g. {"app_version": ["2.8.5", "2.8.6"]}
# => [{"app_version": "2.8.5", "app_version": "2.8.6"}]
# e.g. {"username-password": [["user1", "111111"], ["test2", "222222"]}
# => [{"username": "user1", "password": "111111"}, {"username": "user2", "password": "222222"}]
parameter_content_list = []
for parameter_item in parameter_content:
if not isinstance(parameter_item, (list, tuple)):
# "2.8.5" => ["2.8.5"]
parameter_item = [parameter_item]
# # ["app_version"], ["2.8.5"] => {"app_version": "2.8.5"}
# # ["username", "password"], ["user1", "111111"] => {"username": "user1", "password": "111111"}
# parameter_content_dict = dict(zip(parameter_name_list, parameter_item))
# ["app_version"], ["2.8.5"] => {"app_version": "2.8.5"}
# ["username", "password"], ["user1", "111111"] => {"username": "user1", "password": "111111"}
parameter_content_dict = dict(zip(parameter_name_list, parameter_item))
# parameter_content_list.append(parameter_content_dict)
# else:
# # (2) & (3)
# parsed_parameter_content = testcase_parser.eval_content_with_bindings(parameter_content)
# # e.g. [{'app_version': '2.8.5'}, {'app_version': '2.8.6'}]
# # e.g. [{"username": "user1", "password": "111111"}, {"username": "user2", "password": "222222"}]
# if not isinstance(parsed_parameter_content, list):
# raise exceptions.ParamsError("parameters syntax error!")
parameter_content_list.append(parameter_content_dict)
else:
# (2) & (3)
parsed_parameter_content = testcase_parser.eval_content_with_bindings(parameter_content)
# e.g. [{'app_version': '2.8.5'}, {'app_version': '2.8.6'}]
# e.g. [{"username": "user1", "password": "111111"}, {"username": "user2", "password": "222222"}]
if not isinstance(parsed_parameter_content, list):
raise exceptions.ParamsError("parameters syntax error!")
# parameter_content_list = [
# # get subset by parameter name
# {key: parameter_item[key] for key in parameter_name_list}
# for parameter_item in parsed_parameter_content
# ]
parameter_content_list = [
# get subset by parameter name
{key: parameter_item[key] for key in parameter_name_list}
for parameter_item in parsed_parameter_content
]
# parsed_parameters_list.append(parameter_content_list)
parsed_parameters_list.append(parameter_content_list)
return utils.gen_cartesian_product(*parsed_parameters_list)
class TestcaseParser(object):
def __init__(self, variables={}, functions={}, file_path=None):
self.update_binded_variables(variables)
self.bind_functions(functions)
self.file_path = file_path
def update_binded_variables(self, variables):
""" bind variables to current testcase parser
@param (dict) variables, variables binds mapping
{
"authorization": "a83de0ff8d2e896dbd8efb81ba14e17d",
"random": "A2dEx",
"data": {"name": "user", "password": "123456"},
"uuid": 1000
}
"""
self.variables = variables
def bind_functions(self, functions):
""" bind functions to current testcase parser
@param (dict) functions, functions binds mapping
{
"add_two_nums": lambda a, b=1: a + b
}
"""
self.functions = functions
def _get_bind_item(self, item_type, item_name):
""" get specified function or variable.
Args:
item_type(str): functions or variables
item_name(str): function name or variable name
Returns:
object: specified function or variable object.
"""
if item_type == "functions":
if item_name in self.functions:
return self.functions[item_name]
try:
# check if builtin functions
item_func = eval(item_name)
if callable(item_func):
# is builtin function
return item_func
except (NameError, TypeError):
# is not builtin function, continue to search
pass
else:
# item_type == "variables":
if item_name in self.variables:
return self.variables[item_name]
debugtalk_module = loader.load_debugtalk_module(self.file_path)
return loader.get_module_item(debugtalk_module, item_type, item_name)
def get_bind_function(self, func_name):
return self._get_bind_item("functions", func_name)
def get_bind_variable(self, variable_name):
return self._get_bind_item("variables", variable_name)
def load_csv_list(self, csv_file_name, fetch_method="Sequential"):
""" locate csv file and load csv content.
Args:
csv_file_name (str): csv file name
fetch_method (str): fetch data method, defaults to Sequential.
If set to "random", csv data list will be reordered in random.
Returns:
list: csv data list
"""
csv_file_path = loader.locate_file(self.file_path, csv_file_name)
csv_content_list = loader.load_file(csv_file_path)
if fetch_method.lower() == "random":
random.shuffle(csv_content_list)
return csv_content_list
def _eval_content_functions(self, content):
functions_list = parser.extract_functions(content)
for func_content in functions_list:
function_meta = parser.parse_function(func_content)
func_name = function_meta['func_name']
args = function_meta.get('args', [])
kwargs = function_meta.get('kwargs', {})
args = self.eval_content_with_bindings(args)
kwargs = self.eval_content_with_bindings(kwargs)
if func_name in ["parameterize", "P"]:
eval_value = self.load_csv_list(*args, **kwargs)
else:
func = self.get_bind_function(func_name)
eval_value = func(*args, **kwargs)
func_content = "${" + func_content + "}"
if func_content == content:
# content is a variable
content = eval_value
else:
# content contains one or many variables
content = content.replace(
func_content,
str(eval_value), 1
)
return content
def _eval_content_variables(self, content):
""" replace all variables of string content with mapping value.
@param (str) content
@return (str) parsed content
e.g.
variable_mapping = {
"var_1": "abc",
"var_2": "def"
}
$var_1 => "abc"
$var_1#XYZ => "abc#XYZ"
/$var_1/$var_2/var3 => "/abc/def/var3"
${func($var_1, $var_2, xyz)} => "${func(abc, def, xyz)}"
"""
variables_list = parser.extract_variables(content)
for variable_name in variables_list:
variable_value = self.get_bind_variable(variable_name)
if "${}".format(variable_name) == content:
# content is a variable
content = variable_value
else:
# content contains one or several variables
if not isinstance(variable_value, str):
variable_value = builtin_str(variable_value)
content = content.replace(
"${}".format(variable_name),
variable_value, 1
)
return content
def eval_content_with_bindings(self, content):
""" parse content recursively, each variable and function in content will be evaluated.
@param (dict) content in any data structure
{
"url": "http://127.0.0.1:5000/api/users/$uid/${add_two_nums(1, 1)}",
"method": "POST",
"headers": {
"Content-Type": "application/json",
"authorization": "$authorization",
"random": "$random",
"sum": "${add_two_nums(1, 2)}"
},
"body": "$data"
}
@return (dict) parsed content with evaluated bind values
{
"url": "http://127.0.0.1:5000/api/users/1000/2",
"method": "POST",
"headers": {
"Content-Type": "application/json",
"authorization": "a83de0ff8d2e896dbd8efb81ba14e17d",
"random": "A2dEx",
"sum": 3
},
"body": {"name": "user", "password": "123456"}
}
"""
if content is None:
return None
if isinstance(content, (list, tuple)):
return [
self.eval_content_with_bindings(item)
for item in content
]
if isinstance(content, dict):
evaluated_data = {}
for key, value in content.items():
eval_key = self.eval_content_with_bindings(key)
eval_value = self.eval_content_with_bindings(value)
evaluated_data[eval_key] = eval_value
return evaluated_data
if isinstance(content, basestring):
# content is in string format here
content = content.strip()
# replace functions with evaluated value
# Notice: _eval_content_functions must be called before _eval_content_variables
content = self._eval_content_functions(content)
# replace variables with binding value
content = self._eval_content_variables(content)
return content
# return utils.gen_cartesian_product(*parsed_parameters_list)
class Context(object):
""" Manages context functions and variables.
context has two levels, testset and testcase.
context has two levels, testcase and teststep.
"""
def __init__(self):
self.testset_shared_variables_mapping = OrderedDict()
self.testcase_variables_mapping = OrderedDict()
self.testcase_parser = TestcaseParser()
def __init__(self, variables=None, functions=None):
""" init Context with testcase variables and functions.
"""
# testcase level context
## TESTCASE_SHARED_VARIABLES_MAPPING and TESTCASE_SHARED_FUNCTIONS_MAPPING will not change.
self.TESTCASE_SHARED_VARIABLES_MAPPING = variables or OrderedDict()
self.TESTCASE_SHARED_FUNCTIONS_MAPPING = functions or OrderedDict()
# testcase level request, will not change
self.TESTCASE_SHARED_REQUEST_MAPPING = {}
self.evaluated_validators = []
self.init_context()
self.init_context_variables(level="testcase")
def init_context_variables(self, level="testcase"):
""" initialize testcase/teststep context
Args:
level (enum): "testcase" or "teststep"
def init_context(self, level='testset'):
"""
testset level context initializes when a file is loaded,
testcase level context initializes when each testcase starts.
"""
if level == "testset":
self.testset_functions_config = {}
self.testset_request_config = {}
self.testset_shared_variables_mapping = OrderedDict()
if level == "testcase":
# testcase level runtime context, will be updated with extracted variables in each teststep.
self.testcase_runtime_variables_mapping = copy.deepcopy(self.TESTCASE_SHARED_VARIABLES_MAPPING)
# testcase config shall inherit from testset configs,
# but can not change testset configs, that's why we use copy.deepcopy here.
self.testcase_functions_config = copy.deepcopy(self.testset_functions_config)
self.testcase_variables_mapping = copy.deepcopy(self.testset_shared_variables_mapping)
# teststep level context, will be altered in each teststep.
# teststep config shall inherit from testcase configs,
# but can not change testcase configs, that's why we use copy.deepcopy here.
self.teststep_variables_mapping = copy.deepcopy(self.testcase_runtime_variables_mapping)
self.testcase_parser.bind_functions(self.testcase_functions_config)
self.testcase_parser.update_binded_variables(self.testcase_variables_mapping)
def update_context_variables(self, variables, level):
""" update context variables, with level specified.
if level == "testset":
self.import_module_items(built_in)
Args:
variables (list/OrderedDict): testcase config block or teststep block
[
{"TOKEN": "debugtalk"},
{"random": "${gen_random_string(5)}"},
{"json": {'name': 'user', 'password': '123456'}},
{"md5": "${gen_md5($TOKEN, $json, $random)}"}
]
OrderDict({
"TOKEN": "debugtalk",
"random": "${gen_random_string(5)}",
"json": {'name': 'user', 'password': '123456'},
"md5": "${gen_md5($TOKEN, $json, $random)}"
})
level (enum): "testcase" or "teststep"
def config_context(self, config_dict, level):
if level == "testset":
self.testcase_parser.file_path = config_dict.get("path", None)
variables = config_dict.get('variables') \
or config_dict.get('variable_binds', OrderedDict())
self.bind_variables(variables, level)
def bind_functions(self, function_binds, level="testcase"):
""" Bind named functions within the context
This allows for passing in self-defined functions in testing.
e.g. function_binds:
{
"add_one": lambda x: x + 1, # lambda function
"add_two_nums": "lambda x, y: x + y" # lambda function in string
}
"""
eval_function_binds = {}
for func_name, function in function_binds.items():
if isinstance(function, str):
function = eval(function)
eval_function_binds[func_name] = function
self.__update_context_functions_config(level, eval_function_binds)
def import_module_items(self, imported_module):
""" import module functions and variables and bind to testset context
"""
module_mapping = loader.load_python_module(imported_module)
self.__update_context_functions_config("testset", module_mapping["functions"])
self.bind_variables(module_mapping["variables"], "testset")
def bind_variables(self, variables, level="testcase"):
""" bind variables to testset context or current testcase context.
variables in testset context can be used in all testcases of current test suite.
@param (list or OrderDict) variables, variable can be value or custom function.
if value is function, it will be called and bind result to variable.
e.g.
OrderDict({
"TOKEN": "debugtalk",
"random": "${gen_random_string(5)}",
"json": {'name': 'user', 'password': '123456'},
"md5": "${gen_md5($TOKEN, $json, $random)}"
})
"""
if isinstance(variables, list):
variables = utils.convert_mappinglist_to_orderdict(variables)
for variable_name, value in variables.items():
variable_eval_value = self.eval_content(value)
for variable_name, variable_value in variables.items():
variable_eval_value = self.eval_content(variable_value)
if level == "testset":
self.testset_shared_variables_mapping[variable_name] = variable_eval_value
if level == "testcase":
self.testcase_runtime_variables_mapping[variable_name] = variable_eval_value
self.bind_testcase_variable(variable_name, variable_eval_value)
def bind_testcase_variable(self, variable_name, variable_value):
""" bind and update testcase variables mapping
"""
self.testcase_variables_mapping[variable_name] = variable_value
self.testcase_parser.update_binded_variables(self.testcase_variables_mapping)
def bind_extracted_variables(self, variables):
""" bind extracted variables to testset context
@param (OrderDict) variables
extracted value do not need to evaluate.
"""
for variable_name, value in variables.items():
self.testset_shared_variables_mapping[variable_name] = value
self.bind_testcase_variable(variable_name, value)
def __update_context_functions_config(self, level, config_mapping):
"""
@param level: testset or testcase
@param config_type: functions
@param config_mapping: functions config mapping
"""
if level == "testset":
self.testset_functions_config.update(config_mapping)
self.testcase_functions_config.update(config_mapping)
self.testcase_parser.bind_functions(self.testcase_functions_config)
self.update_teststep_variables_mapping(variable_name, variable_eval_value)
def eval_content(self, content):
""" evaluate content recursively, take effect on each variable and function in content.
content may be in any data structure, include dict, list, tuple, number, string, etc.
"""
return self.testcase_parser.eval_content_with_bindings(content)
return parser.parse_data(
content,
self.teststep_variables_mapping,
self.TESTCASE_SHARED_FUNCTIONS_MAPPING
)
def update_testcase_runtime_variables_mapping(self, variables):
""" update testcase_runtime_variables_mapping with extracted vairables in teststep.
Args:
variables (OrderDict): extracted variables in teststep
def get_parsed_request(self, request_dict, level="testcase"):
""" get parsed request with bind variables and functions.
@param request_dict: request config mapping
@param level: testset or testcase
"""
if level == "testset":
request_dict = self.eval_content(
request_dict
for variable_name, variable_value in variables.items():
self.testcase_runtime_variables_mapping[variable_name] = variable_value
self.update_teststep_variables_mapping(variable_name, variable_value)
def update_teststep_variables_mapping(self, variable_name, variable_value):
""" bind and update testcase variables mapping
"""
self.teststep_variables_mapping[variable_name] = variable_value
def get_parsed_request(self, request_dict, level="teststep"):
""" get parsed request with variables and functions.
Args:
request_dict (dict): request config mapping
level (enum): "testcase" or "teststep"
Returns:
dict: parsed request dict
"""
if level == "testcase":
# testcase config request dict has been parsed in __parse_testcases
self.TESTCASE_SHARED_REQUEST_MAPPING = request_dict
return request_dict
else:
# teststep
return self.eval_content(
utils.deep_update_dict(
copy.deepcopy(self.TESTCASE_SHARED_REQUEST_MAPPING),
request_dict
)
)
self.testset_request_config.update(request_dict)
testcase_request_config = utils.deep_update_dict(
copy.deepcopy(self.testset_request_config),
request_dict
)
parsed_request = self.eval_content(
testcase_request_config
)
def __eval_check_item(self, validator, resp_obj):
""" evaluate check item in validator.
return parsed_request
Args:
validator (dict): validator
{"check": "status_code", "comparator": "eq", "expect": 201}
{"check": "$resp_body_success", "comparator": "eq", "expect": True}
resp_obj (object): requests.Response() object
Returns:
dict: validator info
{
"check": "status_code",
"check_value": 200,
"expect": 201,
"comparator": "eq"
}
def eval_check_item(self, validator, resp_obj):
""" evaluate check item in validator
@param (dict) validator
{"check": "status_code", "comparator": "eq", "expect": 201}
{"check": "$resp_body_success", "comparator": "eq", "expect": True}
@param (object) resp_obj
@return (dict) validator info
{
"check": "status_code",
"check_value": 200,
"expect": 201,
"comparator": "eq"
}
"""
check_item = validator["check"]
# check_item should only be the following 5 formats:
@@ -470,12 +234,22 @@ class Context(object):
validator["check_result"] = "unchecked"
return validator
def do_validation(self, validator_dict):
def _do_validation(self, validator_dict):
""" validate with functions
Args:
validator_dict (dict): validator dict
{
"check": "status_code",
"check_value": 200,
"expect": 201,
"comparator": "eq"
}
"""
# TODO: move comparator uniform to init_test_suites
comparator = utils.get_uniform_comparator(validator_dict["comparator"])
validate_func = self.testcase_parser.get_bind_function(comparator)
validate_func = self.TESTCASE_SHARED_FUNCTIONS_MAPPING.get(comparator)
if not validate_func:
raise exceptions.FunctionNotFound("comparator not found: {}".format(comparator))
@@ -516,26 +290,28 @@ class Context(object):
def validate(self, validators, resp_obj):
""" make validations
"""
evaluated_validators = []
if not validators:
return
return evaluated_validators
logger.log_info("start to validate.")
self.evaluated_validators = []
validate_pass = True
for validator in validators:
# evaluate validators with context variable mapping.
evaluated_validator = self.eval_check_item(
evaluated_validator = self.__eval_check_item(
parser.parse_validator(validator),
resp_obj
)
try:
self.do_validation(evaluated_validator)
self._do_validation(evaluated_validator)
except exceptions.ValidationFailure:
validate_pass = False
self.evaluated_validators.append(evaluated_validator)
evaluated_validators.append(evaluated_validator)
if not validate_pass:
raise exceptions.ValidationFailure
return evaluated_validators

44
httprunner/loader.py
View File

@@ -295,8 +295,15 @@ def load_python_module(module):
return debugtalk_module
def load_builtin_module():
""" load built_in module
"""
built_in_module = load_python_module(built_in)
project_mapping["debugtalk"] = built_in_module
def load_debugtalk_module(start_path=None):
""" load built_in module and project debugtalk.py module.
""" load project debugtalk.py module and merge with builtin module.
Args:
start_path (str, optional): start locating path, maybe file path or directory path.
@@ -304,34 +311,27 @@ def load_debugtalk_module(start_path=None):
Returns:
dict: variables and functions mapping for debugtalk.py
{
"variables": {},
"functions": {}
}
"""
# load built_in module
built_in_module = load_python_module(built_in)
start_path = start_path or os.getcwd()
try:
module_path = locate_file(start_path, "debugtalk.py")
module_name = convert_module_name(module_path)
except exceptions.FileNotFound:
return built_in_module
return
# load debugtalk.py module
imported_module = importlib.import_module(module_name)
debugtalk_module = load_python_module(imported_module)
# override built_in module with debugtalk.py module
debugtalk_module["variables"].update(built_in_module["variables"])
debugtalk_module["functions"].update(built_in_module["functions"])
project_mapping["debugtalk"] = debugtalk_module
return debugtalk_module
project_mapping["debugtalk"]["variables"].update(debugtalk_module["variables"])
project_mapping["debugtalk"]["functions"].update(debugtalk_module["functions"])
def get_module_item(module_mapping, item_type, item_name):
@@ -884,22 +884,27 @@ def load_test_folder(test_folder_path=None):
return test_definition_mapping
def reset_loader():
""" reset project mapping.
"""
project_mapping["debugtalk"] = {}
project_mapping["env"] = {}
project_mapping["def-api"] = {}
project_mapping["def-testcase"] = {}
testcases_cache_mapping.clear()
def load_project_tests(folder_path):
""" load api, testcases and debugtalk.py module.
""" load api, testcases and builtin module.
Args:
folder_path (str): folder path.
Returns:
dict: project tests mapping.
"""
load_debugtalk_module(folder_path)
load_builtin_module()
load_api_folder(os.path.join(folder_path, "api"))
load_test_folder(os.path.join(folder_path, "suite"))
return project_mapping
def load_testcases(path):
""" load testcases from file path, extend and merge with api/testcase definitions.
@@ -937,11 +942,14 @@ def load_testcases(path):
return testcases_cache_mapping[path]
if os.path.isdir(path):
load_debugtalk_module(path)
files_list = load_folder_files(path)
testcases_list = load_testcases(files_list)
elif os.path.isfile(path):
try:
dir_path = os.path.dirname(path)
load_debugtalk_module(dir_path)
testcase = _load_test_file(path)
if testcase["teststeps"]:
testcases_list = [testcase]

1
httprunner/parser.py
View File

@@ -492,7 +492,6 @@ def parse_data(content, variables_mapping=None, functions_mapping=None):
"""
# TODO: refactor type check
# TODO: combine this with TestcaseParser
if content is None or isinstance(content, (numeric_types, bool, type)):
return content

195
httprunner/runner.py
View File

@@ -4,69 +4,83 @@ from unittest.case import SkipTest
from httprunner import exceptions, logger, response, utils
from httprunner.client import HttpSession
from httprunner.compat import OrderedDict
from httprunner.context import Context
class Runner(object):
def __init__(self, config_dict=None, http_client_session=None):
"""
"""
self.http_client_session = http_client_session
self.context = Context()
config_dict = config_dict or {}
self.evaluated_validators = []
# testset setup hooks
testset_setup_hooks = config_dict.pop("setup_hooks", [])
# testset teardown hooks
self.testset_teardown_hooks = config_dict.pop("teardown_hooks", [])
# testcase variables
config_variables = config_dict.get("variables", {})
# testcase functions
config_functions = config_dict.get("functions", {})
# testcase setup hooks
testcase_setup_hooks = config_dict.pop("setup_hooks", [])
# testcase teardown hooks
self.testcase_teardown_hooks = config_dict.pop("teardown_hooks", [])
self.init_config(config_dict, "testset")
self.context = Context(config_variables, config_functions)
self.init_config(config_dict, "testcase")
if testset_setup_hooks:
self.do_hook_actions(testset_setup_hooks)
if testcase_setup_hooks:
self.do_hook_actions(testcase_setup_hooks)
def __del__(self):
if self.testset_teardown_hooks:
self.do_hook_actions(self.testset_teardown_hooks)
if self.testcase_teardown_hooks:
self.do_hook_actions(self.testcase_teardown_hooks)
def init_config(self, config_dict, level):
""" create/update context variables binds
@param (dict) config_dict
@param (str) level, "testset" or "testcase"
testset:
{
"name": "smoke testset",
"path": "tests/data/demo_testset_variables.yml",
"variables": [], # optional
"request": {
"base_url": "http://127.0.0.1:5000",
"headers": {
"User-Agent": "iOS/2.8.3"
Args:
config_dict (dict):
level (enum): "testcase" or "teststep"
testcase:
{
"name": "testcase description",
"path": "tests/data/demo_testset_variables.yml",
"variables": [], # optional
"request": {
"base_url": "http://127.0.0.1:5000",
"headers": {
"User-Agent": "iOS/2.8.3"
}
}
}
}
}
testcase:
{
"name": "testcase description",
"variables": [], # optional
"request": {
"url": "/api/get-token",
"method": "POST",
"headers": {
"Content-Type": "application/json"
teststep:
{
"name": "teststep description",
"variables": [], # optional
"request": {
"url": "/api/get-token",
"method": "POST",
"headers": {
"Content-Type": "application/json"
}
},
"json": {
"sign": "f1219719911caae89ccc301679857ebfda115ca2"
}
}
},
"json": {
"sign": "f1219719911caae89ccc301679857ebfda115ca2"
}
}
@param (str) context level, testcase or testset
Returns:
dict: parsed request dict
"""
# convert keys in request headers to lowercase
config_dict = utils.lower_config_dict_key(config_dict)
self.context.init_context(level)
self.context.config_context(config_dict, level)
self.context.init_context_variables(level)
variables = config_dict.get('variables') \
or config_dict.get('variable_binds', OrderedDict())
self.context.update_context_variables(variables, level)
request_config = config_dict.get('request', {})
parsed_request = self.context.get_parsed_request(request_config, level)
@@ -76,24 +90,32 @@ class Runner(object):
return parsed_request
def _handle_skip_feature(self, testcase_dict):
""" handle skip feature for testcase
def _handle_skip_feature(self, teststep_dict):
""" handle skip feature for teststep
- skip: skip current test unconditionally
- skipIf: skip current test if condition is true
- skipUnless: skip current test unless condition is true
Args:
teststep_dict (dict): teststep info
Raises:
SkipTest: skip teststep
"""
# TODO: move skip to __initialize
skip_reason = None
if "skip" in testcase_dict:
skip_reason = testcase_dict["skip"]
if "skip" in teststep_dict:
skip_reason = teststep_dict["skip"]
elif "skipIf" in testcase_dict:
skip_if_condition = testcase_dict["skipIf"]
elif "skipIf" in teststep_dict:
skip_if_condition = teststep_dict["skipIf"]
if self.context.eval_content(skip_if_condition):
skip_reason = "{} evaluate to True".format(skip_if_condition)
elif "skipUnless" in testcase_dict:
skip_unless_condition = testcase_dict["skipUnless"]
elif "skipUnless" in teststep_dict:
skip_unless_condition = teststep_dict["skipUnless"]
if not self.context.eval_content(skip_unless_condition):
skip_reason = "{} evaluate to False".format(skip_unless_condition)
@@ -106,40 +128,49 @@ class Runner(object):
# TODO: check hook function if valid
self.context.eval_content(action)
def run_test(self, testcase_dict):
""" run single testcase.
@param (dict) testcase_dict
{
"name": "testcase description",
"skip": "skip this test unconditionally",
"times": 3,
"variables": [], # optional, override
"request": {
"url": "http://127.0.0.1:5000/api/users/1000",
"method": "POST",
"headers": {
"Content-Type": "application/json",
"authorization": "$authorization",
"random": "$random"
def run_test(self, teststep_dict):
""" run single teststep.
Args:
teststep_dict (dict): teststep info
{
"name": "teststep description",
"skip": "skip this test unconditionally",
"times": 3,
"variables": [], # optional, override
"request": {
"url": "http://127.0.0.1:5000/api/users/1000",
"method": "POST",
"headers": {
"Content-Type": "application/json",
"authorization": "$authorization",
"random": "$random"
},
"body": '{"name": "user", "password": "123456"}'
},
"body": '{"name": "user", "password": "123456"}'
},
"extract": [], # optional
"validate": [], # optional
"setup_hooks": [], # optional
"teardown_hooks": [] # optional
}
@return True or raise exception during test
"extract": [], # optional
"validate": [], # optional
"setup_hooks": [], # optional
"teardown_hooks": [] # optional
}
Raises:
exceptions.ParamsError
exceptions.ValidationFailure
exceptions.ExtractFailure
"""
# check skip
self._handle_skip_feature(testcase_dict)
self._handle_skip_feature(teststep_dict)
# prepare
parsed_request = self.init_config(testcase_dict, level="testcase")
self.context.bind_testcase_variable("request", parsed_request)
extractors = teststep_dict.pop("extract", []) or teststep_dict.pop("extractors", [])
validators = teststep_dict.pop("validate", []) or teststep_dict.pop("validators", [])
parsed_request = self.init_config(teststep_dict, level="teststep")
self.context.update_teststep_variables_mapping("request", parsed_request)
# setup hooks
setup_hooks = testcase_dict.get("setup_hooks", [])
setup_hooks = teststep_dict.get("setup_hooks", [])
setup_hooks.insert(0, "${setup_hook_prepare_kwargs($request)}")
self.do_hook_actions(setup_hooks)
@@ -171,21 +202,19 @@ class Runner(object):
resp_obj = response.ResponseObject(resp)
# teardown hooks
teardown_hooks = testcase_dict.get("teardown_hooks", [])
teardown_hooks = teststep_dict.get("teardown_hooks", [])
if teardown_hooks:
logger.log_info("start to run teardown hooks")
self.context.bind_testcase_variable("response", resp_obj)
self.context.update_teststep_variables_mapping("response", resp_obj)
self.do_hook_actions(teardown_hooks)
# extract
extractors = testcase_dict.get("extract", []) or testcase_dict.get("extractors", [])
extracted_variables_mapping = resp_obj.extract_response(extractors)
self.context.bind_extracted_variables(extracted_variables_mapping)
self.context.update_testcase_runtime_variables_mapping(extracted_variables_mapping)
# validate
validators = testcase_dict.get("validate", []) or testcase_dict.get("validators", [])
try:
self.context.validate(validators, resp_obj)
self.evaluated_validators = self.context.validate(validators, resp_obj)
except (exceptions.ParamsError, \
exceptions.ValidationFailure, exceptions.ExtractFailure):
# log request
@@ -207,7 +236,7 @@ class Runner(object):
def extract_output(self, output_variables_list):
""" extract output variables
"""
variables_mapping = self.context.testcase_variables_mapping
variables_mapping = self.context.teststep_variables_mapping
output = {}
for variable in output_variables_list:

4
httprunner/utils.py
View File

@@ -267,7 +267,7 @@ def add_teststep(test_runner, teststep_dict):
finally:
if hasattr(test_runner.http_client_session, "meta_data"):
self.meta_data = test_runner.http_client_session.meta_data
self.meta_data["validators"] = test_runner.context.evaluated_validators
self.meta_data["validators"] = test_runner.evaluated_validators
test_runner.http_client_session.init_meta_data()
test.__doc__ = teststep_dict["name"]
@@ -331,6 +331,8 @@ def print_io(in_out):
def prepare_content(var_type, in_out):
content = ""
for variable, value in in_out.items():
if isinstance(value, tuple):
continue
if is_py2:
if isinstance(variable, unicode):

15
tests/test_api.py
View File

@@ -303,26 +303,26 @@ class TestHttpRunner(ApiServerUnittest):
self.assertTrue(summary["success"])
self.assertEqual(len(summary["details"]), 1)
def test_run_testset_output(self):
def test_run_testcase_output(self):
testcase_file_path = os.path.join(
os.getcwd(), 'tests/data/demo_testset_layer.yml')
runner = HttpRunner().run(testcase_file_path)
runner = HttpRunner(failfast=True).run(testcase_file_path)
summary = runner.summary
self.assertTrue(summary["success"])
self.assertIn("token", summary["details"][0]["in_out"]["out"])
self.assertIn("user_agent", summary["details"][0]["in_out"]["in"])
def test_run_testset_with_variables_mapping(self):
def test_run_testcase_with_variables_mapping(self):
testcase_file_path = os.path.join(
os.getcwd(), 'tests/data/demo_testset_layer.yml')
variables_mapping = {
"app_version": '2.9.7'
}
runner = HttpRunner().run(testcase_file_path, mapping=variables_mapping)
runner = HttpRunner(failfast=True).run(testcase_file_path, mapping=variables_mapping)
summary = runner.summary
self.assertTrue(summary["success"])
self.assertIn("token", summary["details"][0]["in_out"]["out"])
self.assertEqual(len(summary["details"][0]["in_out"]["in"]), 7)
self.assertEqual(len(summary["details"][0]["in_out"]["in"]), 9)
def test_run_testset_with_parameters(self):
testcase_file_path = os.path.join(
@@ -340,11 +340,6 @@ class TestHttpRunner(ApiServerUnittest):
self.assertEqual(hrunner.project_mapping["env"]["PROJECT_KEY"], "ABCDEFGH")
self.assertIn("debugtalk", hrunner.project_mapping)
self.assertIn("setup_and_reset", hrunner.project_mapping["def-testcase"])
self.assertEqual(
hrunner.project_mapping["debugtalk"]["variables"]["SECRET_KEY"],
"DebugTalk"
)
self.assertIn("get_sign", hrunner.project_mapping["debugtalk"]["functions"])
self.assertIn("get_token", hrunner.project_mapping["def-api"])
self.assertIn("setup_and_reset", hrunner.project_mapping["def-testcase"])

560
tests/test_context.py
View File

@@ -1,201 +1,152 @@
import os
import time
import unittest
import requests
from httprunner import context, exceptions, loader, parser, response, runner
from httprunner import context, exceptions, loader, response
from tests.base import ApiServerUnittest
class TestContext(ApiServerUnittest):
def setUp(self):
self.context = context.Context()
project_dir = os.path.join(os.getcwd(), "tests")
loader.load_project_tests(project_dir)
loader.load_debugtalk_module(project_dir)
self.debugtalk_module = loader.project_mapping["debugtalk"]
self.context = context.Context(
self.debugtalk_module["variables"],
self.debugtalk_module["functions"]
)
testcase_file_path = os.path.join(os.getcwd(), 'tests/data/demo_binds.yml')
self.testcases = loader.load_file(testcase_file_path)
def test_context_init_functions(self):
self.assertIn("get_timestamp", self.context.testset_functions_config)
self.assertIn("gen_random_string", self.context.testset_functions_config)
def test_init_context_functions(self):
context_functions = self.context.TESTCASE_SHARED_FUNCTIONS_MAPPING
self.assertIn("gen_md5", context_functions)
self.assertIn("equals", context_functions)
def test_init_context_variables(self):
self.assertEqual(
self.context.teststep_variables_mapping["SECRET_KEY"],
"DebugTalk"
)
self.assertEqual(
self.context.testcase_runtime_variables_mapping["SECRET_KEY"],
"DebugTalk"
)
def test_update_context_testcase_level(self):
variables = [
{"random": "${gen_random_string(5)}"},
{"timestamp10": "${get_timestamp(10)}"}
{"TOKEN": "debugtalk"},
{"data": '{"name": "user", "password": "123456"}'}
]
self.context.bind_variables(variables)
context_variables = self.context.testcase_variables_mapping
self.context.update_context_variables(variables, "testcase")
self.assertEqual(
self.context.teststep_variables_mapping["TOKEN"],
"debugtalk"
)
self.assertEqual(
self.context.testcase_runtime_variables_mapping["TOKEN"],
"debugtalk"
)
self.assertEqual(len(context_variables["random"]), 5)
self.assertEqual(len(context_variables["timestamp10"]), 10)
def test_context_bind_testset_variables(self):
# testcase in JSON format
testcase1 = {
"variables": [
{"GLOBAL_TOKEN": "debugtalk"},
{"token": "$GLOBAL_TOKEN"}
]
}
# testcase in YAML format
testcase2 = self.testcases["bind_variables"]
for testcase in [testcase1, testcase2]:
variables = testcase['variables']
self.context.bind_variables(variables, level="testset")
testset_variables = self.context.testset_shared_variables_mapping
testcase_variables = self.context.testcase_variables_mapping
self.assertIn("GLOBAL_TOKEN", testset_variables)
self.assertIn("GLOBAL_TOKEN", testcase_variables)
self.assertEqual(testset_variables["GLOBAL_TOKEN"], "debugtalk")
self.assertIn("token", testset_variables)
self.assertIn("token", testcase_variables)
self.assertEqual(testset_variables["token"], "debugtalk")
def test_context_bind_testcase_variables(self):
testcase1 = {
"variables": [
{"GLOBAL_TOKEN": "debugtalk"},
{"token": "$GLOBAL_TOKEN"}
]
}
testcase2 = self.testcases["bind_variables"]
for testcase in [testcase1, testcase2]:
variables = testcase['variables']
self.context.bind_variables(variables)
testset_variables = self.context.testset_shared_variables_mapping
testcase_variables = self.context.testcase_variables_mapping
self.assertNotIn("GLOBAL_TOKEN", testset_variables)
self.assertIn("GLOBAL_TOKEN", testcase_variables)
self.assertEqual(testcase_variables["GLOBAL_TOKEN"], "debugtalk")
self.assertNotIn("token", testset_variables)
self.assertIn("token", testcase_variables)
self.assertEqual(testcase_variables["token"], "debugtalk")
def test_context_bind_lambda_functions(self):
function_binds = {
"add_one": lambda x: x + 1,
"add_two_nums": lambda x, y: x + y
}
def test_update_context_teststep_level(self):
variables = [
{"add1": "${add_one(2)}"},
{"sum2nums": "${add_two_nums(2,3)}"}
{"TOKEN": "debugtalk"},
{"data": '{"name": "user", "password": "123456"}'}
]
self.context.bind_functions(function_binds)
self.context.bind_variables(variables)
self.context.update_context_variables(variables, "teststep")
self.assertEqual(
self.context.teststep_variables_mapping["TOKEN"],
"debugtalk"
)
self.assertNotIn(
"TOKEN",
self.context.testcase_runtime_variables_mapping
)
context_variables = self.context.testcase_variables_mapping
self.assertIn("add1", context_variables)
self.assertEqual(context_variables["add1"], 3)
self.assertIn("sum2nums", context_variables)
self.assertEqual(context_variables["sum2nums"], 5)
def test_eval_content_functions(self):
content = "${sleep_N_secs(1)}"
start_time = time.time()
self.context.eval_content(content)
elapsed_time = time.time() - start_time
self.assertGreater(elapsed_time, 1)
def test_call_builtin_functions(self):
testcase1 = {
"variables": [
{"length": "${len(debugtalk)}"},
{"smallest": "${min(2, 3, 8)}"},
{"largest": "${max(2, 3, 8)}"}
]
}
testcase2 = self.testcases["builtin_functions"]
def test_eval_content_variables(self):
content = "abc$SECRET_KEY"
self.assertEqual(
self.context.eval_content(content),
"abcDebugTalk"
)
for testcase in [testcase1, testcase2]:
variables = testcase['variables']
self.context.bind_variables(variables)
# TODO: fix variable extraction
# content = "abc$SECRET_KEYdef"
# self.assertEqual(
# self.context.eval_content(content),
# "abcDebugTalkdef"
# )
context_variables = self.context.testcase_variables_mapping
self.assertEqual(context_variables["length"], 9)
self.assertEqual(context_variables["smallest"], 2)
self.assertEqual(context_variables["largest"], 8)
def test_update_testcase_runtime_variables_mapping(self):
variables = {"abc": 123}
self.context.update_testcase_runtime_variables_mapping(variables)
self.assertEqual(
self.context.testcase_runtime_variables_mapping["abc"],
123
)
self.assertEqual(
self.context.teststep_variables_mapping["abc"],
123
)
def test_import_module_items(self):
def test_update_teststep_variables_mapping(self):
self.context.update_teststep_variables_mapping("abc", 123)
self.assertEqual(
self.context.teststep_variables_mapping["abc"],
123
)
self.assertNotIn(
"abc",
self.context.testcase_runtime_variables_mapping
)
def test_get_parsed_request(self):
variables = [
{"TOKEN": "debugtalk"},
{"random": "${gen_random_string(5)}"},
{"data": '{"name": "user", "password": "123456"}'},
{"authorization": "${gen_md5($TOKEN, $data, $random)}"}
]
from tests import debugtalk
from tests.debugtalk import gen_md5
self.context.import_module_items(debugtalk)
self.context.bind_variables(variables)
context_variables = self.context.testcase_variables_mapping
self.context.update_context_variables(variables, "teststep")
self.assertIn("TOKEN", context_variables)
TOKEN = context_variables["TOKEN"]
self.assertEqual(TOKEN, "debugtalk")
self.assertIn("random", context_variables)
self.assertIsInstance(context_variables["random"], str)
self.assertEqual(len(context_variables["random"]), 5)
random = context_variables["random"]
self.assertIn("data", context_variables)
data = context_variables["data"]
self.assertIn("authorization", context_variables)
self.assertEqual(len(context_variables["authorization"]), 32)
authorization = context_variables["authorization"]
self.assertEqual(gen_md5(TOKEN, data, random), authorization)
self.assertIn("SECRET_KEY", context_variables)
SECRET_KEY = context_variables["SECRET_KEY"]
self.assertEqual(SECRET_KEY, "DebugTalk")
def test_get_parsed_request(self):
test_runner = runner.Runner()
testcase = {
"variables": [
{"TOKEN": "debugtalk"},
{"random": "${gen_random_string(5)}"},
{"data": '{"name": "user", "password": "123456"}'},
{"authorization": "${gen_md5($TOKEN, $data, $random)}"}
],
"request": {
"url": "http://127.0.0.1:5000/api/users/1000",
"method": "POST",
"headers": {
"Content-Type": "application/json",
"authorization": "$authorization",
"random": "$random",
"secret_key": "$SECRET_KEY"
},
"data": "$data"
}
request = {
"url": "http://127.0.0.1:5000/api/users/1000",
"method": "POST",
"headers": {
"Content-Type": "application/json",
"authorization": "$authorization",
"random": "$random",
"secret_key": "$SECRET_KEY"
},
"data": "$data"
}
from tests import debugtalk
self.context.import_module_items(debugtalk)
self.context.bind_variables(testcase["variables"])
parsed_request = self.context.get_parsed_request(testcase["request"])
parsed_request = self.context.get_parsed_request(request, level="teststep")
self.assertIn("authorization", parsed_request["headers"])
self.assertEqual(len(parsed_request["headers"]["authorization"]), 32)
self.assertIn("random", parsed_request["headers"])
self.assertEqual(len(parsed_request["headers"]["random"]), 5)
self.assertIn("data", parsed_request)
self.assertEqual(parsed_request["data"], testcase["variables"][2]["data"])
self.assertEqual(parsed_request["data"], variables[2]["data"])
self.assertEqual(parsed_request["headers"]["secret_key"], "DebugTalk")
def test_exec_content_functions(self):
test_runner = runner.Runner()
content = "${sleep_N_secs(1)}"
start_time = time.time()
test_runner.context.eval_content(content)
end_time = time.time()
elapsed_time = end_time - start_time
self.assertGreater(elapsed_time, 1)
def test_do_validation(self):
self.context.do_validation(
self.context._do_validation(
{"check": "check", "check_value": 1, "expect": 1, "comparator": "eq"}
)
self.context.do_validation(
self.context._do_validation(
{"check": "check", "check_value": "abc", "expect": "abc", "comparator": "=="}
)
config_dict = {
"path": 'tests/data/demo_testset_hardcode.yml'
}
self.context.config_context(config_dict, "testset")
self.context.do_validation(
self.context._do_validation(
{"check": "status_code", "check_value": "201", "expect": 3, "comparator": "sum_status_code"}
)
@@ -213,7 +164,7 @@ class TestContext(ApiServerUnittest):
{"resp_status_code": 200},
{"resp_body_success": True}
]
self.context.bind_variables(variables)
self.context.update_context_variables(variables, "teststep")
with self.assertRaises(exceptions.ValidationFailure):
self.context.validate(validators, resp_obj)
@@ -228,13 +179,7 @@ class TestContext(ApiServerUnittest):
{"resp_status_code": 201},
{"resp_body_success": True}
]
self.context.bind_variables(variables)
from tests.debugtalk import is_status_code_200
functions = {
"is_status_code_200": is_status_code_200
}
self.context.bind_functions(functions)
self.context.update_context_variables(variables, "teststep")
self.context.validate(validators, resp_obj)
def test_validate_exception(self):
@@ -248,7 +193,7 @@ class TestContext(ApiServerUnittest):
{"check": "$resp_status_code", "comparator": "eq", "expect": 201}
]
variables = []
self.context.bind_variables(variables)
self.context.update_context_variables(variables, "teststep")
with self.assertRaises(exceptions.VariableNotFound):
self.context.validate(validators, resp_obj)
@@ -257,294 +202,7 @@ class TestContext(ApiServerUnittest):
variables = [
{"resp_status_code": 200}
]
self.context.bind_variables(variables)
self.context.update_context_variables(variables, "teststep")
with self.assertRaises(exceptions.ValidationFailure):
self.context.validate(validators, resp_obj)
class TestTestcaseParser(unittest.TestCase):
def test_eval_content_variables(self):
variables = {
"var_1": "abc",
"var_2": "def",
"var_3": 123,
"var_4": {"a": 1},
"var_5": True,
"var_6": None
}
testcase_parser = context.TestcaseParser(variables=variables)
self.assertEqual(
testcase_parser._eval_content_variables("$var_1"),
"abc"
)
self.assertEqual(
testcase_parser._eval_content_variables("var_1"),
"var_1"
)
self.assertEqual(
testcase_parser._eval_content_variables("$var_1#XYZ"),
"abc#XYZ"
)
self.assertEqual(
testcase_parser._eval_content_variables("/$var_1/$var_2/var3"),
"/abc/def/var3"
)
self.assertEqual(
testcase_parser._eval_content_variables("/$var_1/$var_2/$var_1"),
"/abc/def/abc"
)
self.assertEqual(
testcase_parser._eval_content_variables("${func($var_1, $var_2, xyz)}"),
"${func(abc, def, xyz)}"
)
self.assertEqual(
testcase_parser._eval_content_variables("$var_3"),
123
)
self.assertEqual(
testcase_parser._eval_content_variables("$var_4"),
{"a": 1}
)
self.assertEqual(
testcase_parser._eval_content_variables("$var_5"),
True
)
self.assertEqual(
testcase_parser._eval_content_variables("abc$var_5"),
"abcTrue"
)
self.assertEqual(
testcase_parser._eval_content_variables("abc$var_4"),
"abc{'a': 1}"
)
self.assertEqual(
testcase_parser._eval_content_variables("$var_6"),
None
)
def test_eval_content_variables_search_upward(self):
testcase_parser = context.TestcaseParser()
with self.assertRaises(exceptions.VariableNotFound):
testcase_parser._eval_content_variables("/api/$SECRET_KEY")
testcase_parser.file_path = "tests/data/demo_testset_hardcode.yml"
content = testcase_parser._eval_content_variables("/api/$SECRET_KEY")
self.assertEqual(content, "/api/DebugTalk")
def test_parse_content_with_bindings_variables(self):
variables = {
"str_1": "str_value1",
"str_2": "str_value2"
}
testcase_parser = context.TestcaseParser(variables=variables)
self.assertEqual(
testcase_parser.eval_content_with_bindings("$str_1"),
"str_value1"
)
self.assertEqual(
testcase_parser.eval_content_with_bindings("123$str_1/456"),
"123str_value1/456"
)
with self.assertRaises(exceptions.VariableNotFound):
testcase_parser.eval_content_with_bindings("$str_3")
self.assertEqual(
testcase_parser.eval_content_with_bindings(["$str_1", "str3"]),
["str_value1", "str3"]
)
self.assertEqual(
testcase_parser.eval_content_with_bindings({"key": "$str_1"}),
{"key": "str_value1"}
)
def test_parse_content_with_bindings_multiple_identical_variables(self):
variables = {
"userid": 100,
"data": 1498
}
testcase_parser = context.TestcaseParser(variables=variables)
content = "/users/$userid/training/$data?userId=$userid&data=$data"
self.assertEqual(
testcase_parser.eval_content_with_bindings(content),
"/users/100/training/1498?userId=100&data=1498"
)
def test_parse_variables_multiple_identical_variables(self):
variables = {
"user": 100,
"userid": 1000,
"data": 1498
}
testcase_parser = context.TestcaseParser(variables=variables)
content = "/users/$user/$userid/$data?userId=$userid&data=$data"
self.assertEqual(
testcase_parser.eval_content_with_bindings(content),
"/users/100/1000/1498?userId=1000&data=1498"
)
def test_parse_content_with_bindings_functions(self):
import random, string
functions = {
"gen_random_string": lambda str_len: ''.join(random.choice(string.ascii_letters + string.digits) \
for _ in range(str_len))
}
testcase_parser = context.TestcaseParser(functions=functions)
result = testcase_parser.eval_content_with_bindings("${gen_random_string(5)}")
self.assertEqual(len(result), 5)
add_two_nums = lambda a, b=1: a + b
functions["add_two_nums"] = add_two_nums
self.assertEqual(
testcase_parser.eval_content_with_bindings("${add_two_nums(1)}"),
2
)
self.assertEqual(
testcase_parser.eval_content_with_bindings("${add_two_nums(1, 2)}"),
3
)
def test_eval_content_functions(self):
functions = {
"add_two_nums": lambda a, b=1: a + b
}
testcase_parser = context.TestcaseParser(functions=functions)
self.assertEqual(
testcase_parser._eval_content_functions("${add_two_nums(1, 2)}"),
3
)
self.assertEqual(
testcase_parser._eval_content_functions("/api/${add_two_nums(1, 2)}"),
"/api/3"
)
def test_eval_content_functions_search_upward(self):
testcase_parser = context.TestcaseParser()
with self.assertRaises(exceptions.FunctionNotFound):
testcase_parser._eval_content_functions("/api/${gen_md5(abc)}")
testcase_parser.file_path = "tests/data/demo_testset_hardcode.yml"
content = testcase_parser._eval_content_functions("/api/${gen_md5(abc)}")
self.assertEqual(content, "/api/900150983cd24fb0d6963f7d28e17f72")
def test_parse_content_with_bindings_testcase(self):
variables = {
"uid": "1000",
"random": "A2dEx",
"authorization": "a83de0ff8d2e896dbd8efb81ba14e17d",
"data": {"name": "user", "password": "123456"}
}
functions = {
"add_two_nums": lambda a, b=1: a + b,
"get_timestamp": lambda: int(time.time() * 1000)
}
testcase_template = {
"url": "http://127.0.0.1:5000/api/users/$uid/${add_two_nums(1,2)}",
"method": "POST",
"headers": {
"Content-Type": "application/json",
"authorization": "$authorization",
"random": "$random",
"sum": "${add_two_nums(1, 2)}"
},
"body": "$data"
}
parsed_testcase = context.TestcaseParser(variables, functions)\
.eval_content_with_bindings(testcase_template)
self.assertEqual(
parsed_testcase["url"],
"http://127.0.0.1:5000/api/users/1000/3"
)
self.assertEqual(
parsed_testcase["headers"]["authorization"],
variables["authorization"]
)
self.assertEqual(
parsed_testcase["headers"]["random"],
variables["random"]
)
self.assertEqual(
parsed_testcase["body"],
variables["data"]
)
self.assertEqual(
parsed_testcase["headers"]["sum"],
3
)
def test_parse_parameters_raw_list(self):
parameters = [
{"user_agent": ["iOS/10.1", "iOS/10.2", "iOS/10.3"]},
{"username-password": [("user1", "111111"), ["test2", "222222"]]}
]
cartesian_product_parameters = context.parse_parameters(parameters)
self.assertEqual(
len(cartesian_product_parameters),
3 * 2
)
self.assertEqual(
cartesian_product_parameters[0],
{'user_agent': 'iOS/10.1', 'username': 'user1', 'password': '111111'}
)
def test_parse_parameters_parameterize(self):
parameters = [
{"app_version": "${parameterize(app_version.csv)}"},
{"username-password": "${parameterize(account.csv)}"}
]
testset_path = os.path.join(
os.getcwd(),
"tests/data/demo_parameters.yml"
)
cartesian_product_parameters = context.parse_parameters(
parameters,
testset_path
)
self.assertEqual(
len(cartesian_product_parameters),
2 * 3
)
def test_parse_parameters_custom_function(self):
parameters = [
{"app_version": "${gen_app_version()}"},
{"username-password": "${get_account()}"}
]
testset_path = os.path.join(
os.getcwd(),
"tests/data/demo_parameters.yml"
)
cartesian_product_parameters = context.parse_parameters(
parameters,
testset_path
)
self.assertEqual(
len(cartesian_product_parameters),
2 * 2
)
def test_parse_parameters_mix(self):
parameters = [
{"user_agent": ["iOS/10.1", "iOS/10.2", "iOS/10.3"]},
{"app_version": "${gen_app_version()}"},
{"username-password": "${parameterize(account.csv)}"}
]
testset_path = os.path.join(
os.getcwd(),
"tests/data/demo_parameters.yml"
)
cartesian_product_parameters = context.parse_parameters(
parameters,
testset_path
)
self.assertEqual(
len(cartesian_product_parameters),
3 * 2 * 3
)

18
tests/test_loader.py
View File

@@ -180,13 +180,17 @@ class TestModuleLoader(unittest.TestCase):
self.assertNotIn("is_py3", functions_dict)
def test_load_debugtalk_module(self):
imported_module_items = loader.load_debugtalk_module()
project_dir = os.path.join(os.getcwd(), "tests")
loader.load_project_tests(project_dir)
loader.load_debugtalk_module()
imported_module_items = loader.project_mapping["debugtalk"]
self.assertIn("basestring", imported_module_items["variables"])
self.assertIn("equals", imported_module_items["functions"])
self.assertNotIn("SECRET_KEY", imported_module_items["variables"])
self.assertNotIn("alter_response", imported_module_items["functions"])
imported_module_items = loader.load_debugtalk_module("tests")
loader.load_debugtalk_module("tests")
imported_module_items = loader.project_mapping["debugtalk"]
self.assertEqual(
imported_module_items["variables"]["SECRET_KEY"],
"DebugTalk"
@@ -476,7 +480,9 @@ class TestSuiteLoader(unittest.TestCase):
def test_load_project_tests(self):
project_dir = os.path.join(os.getcwd(), "tests")
project_tests = loader.load_project_tests(project_dir)
self.assertEqual(project_tests["debugtalk"]["variables"]["SECRET_KEY"], "DebugTalk")
self.assertIn("get_token", project_tests["def-api"])
self.assertIn("setup_and_reset", project_tests["def-testcase"])
loader.load_project_tests(project_dir)
loader.load_debugtalk_module(project_dir)
project_mapping = loader.project_mapping
self.assertEqual(project_mapping["debugtalk"]["variables"]["SECRET_KEY"], "DebugTalk")
self.assertIn("get_token", project_mapping["def-api"])
self.assertIn("setup_and_reset", project_mapping["def-testcase"])

31
tests/test_runner.py
View File

@@ -10,7 +10,15 @@ from tests.base import ApiServerUnittest
class TestRunner(ApiServerUnittest):
def setUp(self):
self.test_runner = runner.Runner()
project_dir = os.path.join(os.getcwd(), "tests")
loader.load_project_tests(project_dir)
loader.load_debugtalk_module(project_dir)
self.debugtalk_module = loader.project_mapping["debugtalk"]
config_dict = {
"variables": self.debugtalk_module["variables"],
"functions": self.debugtalk_module["functions"]
}
self.test_runner = runner.Runner(config_dict)
self.reset_all()
def reset_all(self):
@@ -30,18 +38,19 @@ class TestRunner(ApiServerUnittest):
testcases = loader.load_file(testcase_file_path)
config_dict = {
"path": testcase_file_path
"variables": self.debugtalk_module["variables"],
"functions": self.debugtalk_module["functions"]
}
self.test_runner.init_config(config_dict, "testset")
test_runner = runner.Runner(config_dict)
test = testcases[0]["test"]
self.test_runner.run_test(test)
test_runner.run_test(test)
test = testcases[1]["test"]
self.test_runner.run_test(test)
test_runner.run_test(test)
test = testcases[2]["test"]
self.test_runner.run_test(test)
test_runner.run_test(test)
def test_run_single_testcase_fail(self):
test = {
@@ -75,6 +84,8 @@ class TestRunner(ApiServerUnittest):
config_dict = {
"path": os.path.join(os.getcwd(), __file__),
"name": "basic test with httpbin",
"variables": self.debugtalk_module["variables"],
"functions": self.debugtalk_module["functions"],
"request": {
"base_url": HTTPBIN_SERVER
},
@@ -123,6 +134,8 @@ class TestRunner(ApiServerUnittest):
config_dict = {
"path": os.path.join(os.getcwd(), __file__),
"name": "basic test with httpbin",
"variables": self.debugtalk_module["variables"],
"functions": self.debugtalk_module["functions"],
"request": {
"base_url": HTTPBIN_SERVER
}
@@ -177,7 +190,7 @@ class TestRunner(ApiServerUnittest):
config_dict = {
"path": os.path.join(os.getcwd(), __file__)
}
self.test_runner.init_config(config_dict, "testset")
self.test_runner.init_config(config_dict, "testcase")
start_time = time.time()
self.test_runner.run_test(test)
@@ -210,7 +223,7 @@ class TestRunner(ApiServerUnittest):
config_dict = {
"path": os.path.join(os.getcwd(), __file__)
}
self.test_runner.init_config(config_dict, "testset")
self.test_runner.init_config(config_dict, "testcase")
start_time = time.time()
self.test_runner.run_test(test)
@@ -238,7 +251,7 @@ class TestRunner(ApiServerUnittest):
config_dict = {
"path": testcase_file_path
}
self.test_runner.init_config(config_dict, "testset")
self.test_runner.init_config(config_dict, "testcase")
test = testcases[2]["test"]
self.test_runner.run_test(test)