GithubBackup/httprunner
1
0
Fork 0
mirror of https://github.com/httprunner/httprunner.git synced 2026-07-26 00:07:47 +08:00
Code Issues Packages Projects Releases Wiki Activity

HttpRunner 2.0 is comming!

Browse Source 
1, new design for testcase format;
2, refactor testcase layer mechanism.
This commit is contained in:
httprunner
2018-11-22 19:20:30 +08:00
parent 87a80ecc7f
commit f05f190d24
31 changed files with 1500 additions and 1303 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
httprunner/__about__.py
View File

@@ -1,7 +1,7 @@
__title__ = 'HttpRunner'
__description__ = 'One-stop solution for HTTP(S) testing.'
__url__ = 'https://github.com/HttpRunner/HttpRunner'
__version__ = '1.6.0.alpha'
__version__ = '2.0.0.beta'
__author__ = 'debugtalk'
__author_email__ = 'mail@debugtalk.com'
__license__ = 'MIT'

74
httprunner/api.py
View File

@@ -28,17 +28,17 @@ class HttpRunner(object):
self.test_loader = unittest.TestLoader()
self.summary = None
def _add_tests(self, testcases):
def _add_tests(self, tests_mapping):
""" initialize testcase with Runner() and add to test suite.
Args:
testcases (list): parsed testcases list
tests_mapping (dict): project info and testcases list.
Returns:
unittest.TestSuite()
"""
def _add_teststep(test_runner, config, teststep_dict):
def _add_teststep(test_runner, teststep_dict):
""" add teststep to testcase.
"""
def test(self):
@@ -52,22 +52,16 @@ class HttpRunner(object):
self.meta_data["validators"] = test_runner.evaluated_validators
test_runner.http_client_session.init_meta_data()
try:
teststep_dict["name"] = parser.parse_data(
teststep_dict["name"],
config.get("variables", {}),
config.get("functions", {})
)
except exceptions.VariableNotFound:
pass
test.__doc__ = teststep_dict["name"]
# TODO: refactor
test.__doc__ = teststep_dict.get("name") or teststep_dict.get("config", {}).get("name")
return test
test_suite = unittest.TestSuite()
for testcase in testcases:
functions = tests_mapping.get("project_mapping", {}).get("functions", {})
for testcase in tests_mapping["testcases"]:
config = testcase.get("config", {})
test_runner = runner.Runner(config, self.http_client_session)
test_runner = runner.Runner(config, functions, self.http_client_session)
TestSequense = type('TestSequense', (unittest.TestCase,), {})
teststeps = testcase.get("teststeps", [])
@@ -76,12 +70,12 @@ class HttpRunner(object):
# suppose one testcase should not have more than 9999 steps,
# and one step should not run more than 999 times.
test_method_name = 'test_{:04}_{:03}'.format(index, times_index)
test_method = _add_teststep(test_runner, config, teststep_dict)
test_method = _add_teststep(test_runner, teststep_dict)
setattr(TestSequense, test_method_name, test_method)
loaded_testcase = self.test_loader.loadTestsFromTestCase(TestSequense)
setattr(loaded_testcase, "config", config)
setattr(loaded_testcase, "teststeps", testcase.get("teststeps", []))
setattr(loaded_testcase, "teststeps", teststeps)
setattr(loaded_testcase, "runner", test_runner)
test_suite.addTest(loaded_testcase)
@@ -140,49 +134,21 @@ class HttpRunner(object):
self.summary["details"].append(testcase_summary)
def _run_tests(self, testcases, mapping=None):
def _run_tests(self, tests_mapping):
""" start to run test with variables mapping.
Args:
testcases (list): list of testcase_dict, each testcase is corresponding to a YAML/JSON file
[
{ # testcase data structure
"config": {
"name": "desc1",
"path": "testcase1_path",
"variables": [], # optional
"request": {}, # optional
"functions": {},
"env": {},
"def-api": {},
"def-testcase": {}
},
"teststeps": [
# teststep data structure
{
'name': 'test step desc2',
'variables': [], # optional
'extract': [], # optional
'validate': [],
'request': {},
'function_meta': {}
},
teststep2 # another teststep dict
]
},
testcase_dict_2 # another testcase dict
]
mapping (dict): if mapping is specified, it will override variables in config block.
tests_mapping (dict): list of testcase_dict, each testcase is corresponding to a YAML/JSON file
Returns:
instance: HttpRunner() instance
"""
self.exception_stage = "parse tests"
parsed_testcases_list = parser.parse_tests(testcases, mapping)
parser.parse_tests(tests_mapping)
self.exception_stage = "add tests to test suite"
test_suite = self._add_tests(parsed_testcases_list)
test_suite = self._add_tests(tests_mapping)
self.exception_stage = "run test suite"
results = self._run_suite(test_suite)
@@ -207,16 +173,14 @@ class HttpRunner(object):
self.exception_stage = "load tests"
if validator.is_testcases(path_or_testcases):
if isinstance(path_or_testcases, dict):
testcases = [path_or_testcases]
else:
testcases = path_or_testcases
tests_mapping = path_or_testcases
elif validator.is_testcase_path(path_or_testcases):
testcases = loader.load_tests(path_or_testcases, dot_env_path)
tests_mapping = loader.load_tests(path_or_testcases, dot_env_path)
tests_mapping["project_mapping"]["variables"] = mapping or {}
else:
raise exceptions.ParamsError("invalid testcase path or testcases.")
return self._run_tests(testcases, mapping)
return self._run_tests(tests_mapping)
def gen_html_report(self, html_report_name=None, html_report_template=None):
""" generate html report and return report path.

18
httprunner/cli.py
View File

@@ -77,17 +77,13 @@ def main_hrun():
create_scaffold(project_name)
exit(0)
try:
runner = HttpRunner(
failfast=args.failfast
)
runner.run(
args.testcase_paths,
dot_env_path=args.dot_env_path
)
except Exception:
logger.log_error("!!!!!!!!!! exception stage: {} !!!!!!!!!!".format(runner.exception_stage))
raise
for path in args.testcase_paths:
try:
runner = HttpRunner(failfast=args.failfast)
runner.run(path, dot_env_path=args.dot_env_path)
except Exception:
logger.log_error("!!!!!!!!!! exception stage: {} !!!!!!!!!!".format(runner.exception_stage))
raise
if not args.no_html_report:
runner.gen_html_report(

141
httprunner/context.py
View File

@@ -1,75 +1,59 @@
# encoding: utf-8
import copy
from httprunner import exceptions, logger, parser, utils
class Context(object):
""" Manages context functions and variables.
context has two levels, testcase and teststep.
class SessionContext(object):
""" HttpRunner session, store runtime variables.
Examples:
>>> functions={...}
>>> variables = {"SECRET_KEY": "DebugTalk"}
>>> context = SessionContext(functions, variables)
Equivalent to:
>>> context = SessionContext(functions)
>>> context.update_seesion_variables(variables)
"""
def __init__(self, variables=None, functions=None):
""" init Context with testcase variables and functions.
"""
variables = variables or {}
functions = functions or {}
# testcase level context
## TESTCASE_SHARED_VARIABLES_MAPPING and TESTCASE_SHARED_FUNCTIONS_MAPPING are unchangeable.
self.TESTCASE_SHARED_VARIABLES_MAPPING = utils.ensure_mapping_format(variables)
self.TESTCASE_SHARED_FUNCTIONS_MAPPING = functions
def __init__(self, functions, variables=None):
self.session_variables_mapping = utils.ensure_mapping_format(variables or {})
self.FUNCTIONS_MAPPING = functions
self.teststep_variables_mapping = {}
self.init_teststep_variables()
# testcase level request, will not change
self.TESTCASE_SHARED_REQUEST_MAPPING = {}
self.evaluated_validators = []
self.init_context_variables(level="testcase")
def init_context_variables(self, level="testcase"):
""" initialize testcase/teststep context
def init_teststep_variables(self, variables_mapping=None):
""" init teststep variables, called when each teststep(api) starts.
variables_mapping will be evaluated first.
Args:
level (enum): "testcase" or "teststep"
"""
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)
# 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)
def update_context_variables(self, variables, level):
""" update context variables, with level specified.
Args:
variables (list/OrderedDict): testcase config block or teststep block
variables_mapping (dict/list)
[
{"TOKEN": "debugtalk"},
{"random": "${gen_random_string(5)}"},
{"json": {'name': 'user', 'password': '123456'}},
{"md5": "${gen_md5($TOKEN, $json, $random)}"}
{"data": '{"name": "user", "password": "123456"}'},
{"authorization": "${gen_md5($TOKEN, $data, $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"
"""
variables_mapping = utils.ensure_mapping_format(variables)
variables_mapping = variables_mapping or {}
variables_mapping = utils.ensure_mapping_format(variables_mapping)
for variable_name, variable_value in variables_mapping.items():
variable_eval_value = self.eval_content(variable_value)
variable_value = self.eval_content(variable_value)
self.update_teststep_variables(variable_name, variable_value)
if level == "testcase":
self.testcase_runtime_variables_mapping[variable_name] = variable_eval_value
self.teststep_variables_mapping.update(self.session_variables_mapping)
self.update_teststep_variables_mapping(variable_name, variable_eval_value)
def update_teststep_variables(self, variable_name, variable_value):
""" update teststep variables, these variables are only valid in the current teststep.
"""
self.teststep_variables_mapping[variable_name] = variable_value
def update_seesion_variables(self, variables_mapping):
""" update session with extracted variables mapping.
these variables are valid in the whole running session.
"""
variables_mapping = utils.ensure_mapping_format(variables_mapping)
self.session_variables_mapping.update(variables_mapping)
self.teststep_variables_mapping.update(self.session_variables_mapping)
def eval_content(self, content):
""" evaluate content recursively, take effect on each variable and function in content.
@@ -78,50 +62,9 @@ class Context(object):
return parser.parse_data(
content,
self.teststep_variables_mapping,
self.TESTCASE_SHARED_FUNCTIONS_MAPPING
self.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
"""
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_tests
self.TESTCASE_SHARED_REQUEST_MAPPING = copy.deepcopy(request_dict)
return self.TESTCASE_SHARED_REQUEST_MAPPING
else:
# teststep
return self.eval_content(
utils.deep_update_dict(
copy.deepcopy(self.TESTCASE_SHARED_REQUEST_MAPPING),
request_dict
)
)
def __eval_check_item(self, validator, resp_obj):
""" evaluate check item in validator.
@@ -183,7 +126,7 @@ class Context(object):
"""
# TODO: move comparator uniform to init_test_suites
comparator = utils.get_uniform_comparator(validator_dict["comparator"])
validate_func = parser.get_mapping_function(comparator, self.TESTCASE_SHARED_FUNCTIONS_MAPPING)
validate_func = parser.get_mapping_function(comparator, self.FUNCTIONS_MAPPING)
check_item = validator_dict["check"]
check_value = validator_dict["check_value"]

567
httprunner/loader.py
View File

@@ -273,15 +273,21 @@ def load_debugtalk_functions():
## testcase loader
###############################################################################
def _load_teststeps(test_block, project_mapping):
""" load teststeps with api/testcase references
project_mapping = {}
tests_def_mapping = {
"api": {},
"testcases": {}
}
def load_teststep(raw_stepinfo):
""" load teststep with api/testcase/proc references
Args:
test_block (dict): test block content, maybe in 3 formats.
raw_stepinfo (dict): teststep data, maybe in 3 formats.
# api reference
{
"name": "add product to cart",
"api": "api_add_cart()",
"api": "api_add_cart",
"variables": [],
"validate": [],
"extract": []
@@ -289,7 +295,7 @@ def _load_teststeps(test_block, project_mapping):
# testcase reference
{
"name": "add product to cart",
"suite": "create_and_check()",
"testcase": "create_and_check",
"variables": []
}
# define directly
@@ -304,33 +310,43 @@ def _load_teststeps(test_block, project_mapping):
Returns:
list: loaded teststeps list
"""
teststeps = []
Args:
raw_stepinfo (dict): teststep info
"""
# reference api
if "api" in test_block:
ref_call = test_block.pop("api")
def_block = _get_block_by_name(ref_call, "def-api", project_mapping)
extended_block = _extend_block(test_block, def_block)
teststeps.append(extended_block)
if "api" in raw_stepinfo:
api_name = raw_stepinfo["api"]
raw_stepinfo["api_def"] = _get_api_definition(api_name)
# TODO: reference proc functions
elif "func" in raw_stepinfo:
pass
# reference testcase
elif "suite" in test_block: # TODO: replace suite with testcase
ref_call = test_block.pop("suite")
def_block = _get_block_by_name(ref_call, "def-testcase", project_mapping)
# TODO: bugfix lost block config variables
for teststep in def_block["teststeps"]:
_teststeps = _load_teststeps(teststep, project_mapping)
teststeps.extend(_teststeps)
elif "testcase" in raw_stepinfo:
testcase_path = raw_stepinfo["testcase"]
if testcase_path not in tests_def_mapping["testcases"]:
testcase_path = os.path.join(
project_mapping["PWD"],
testcase_path
)
testcase_dict = load_testcase(load_file(testcase_path))
tests_def_mapping[testcase_path] = testcase_dict
else:
testcase_dict = tests_def_mapping[testcase_path]
raw_stepinfo["testcase_def"] = testcase_dict
# define directly
else:
teststeps.append(test_block)
pass
return teststeps
return raw_stepinfo
def _load_testcase(raw_testcase, project_mapping):
def load_testcase(raw_testcase):
""" load testcase/testsuite with api/testcase references
Args:
@@ -352,20 +368,18 @@ def _load_testcase(raw_testcase, project_mapping):
"test": {...}
}
]
project_mapping (dict): project_mapping
Returns:
dict: loaded testcase content
{
"name": "XYZ",
"config": {},
"teststeps": [teststep11, teststep12]
}
"""
loaded_testcase = {
"config": {},
"teststeps": []
}
config = {}
teststeps = []
for item in raw_testcase:
# TODO: add json schema validation
@@ -377,290 +391,38 @@ def _load_testcase(raw_testcase, project_mapping):
raise exceptions.FileFormatError("Testcase format error: {}".format(item))
if key == "config":
loaded_testcase["config"].update(test_block)
config.update(test_block)
elif key == "test":
loaded_testcase["teststeps"].extend(_load_teststeps(test_block, project_mapping))
teststeps.append(load_teststep(test_block))
else:
logger.log_warning(
"unexpected block key: {}. block key should only be 'config' or 'test'.".format(key)
)
return loaded_testcase
return {
"config": config,
"teststeps": teststeps
}
def _get_block_by_name(ref_call, ref_type, project_mapping):
""" get test content by reference name.
Args:
ref_call (str): call function.
e.g. api_v1_Account_Login_POST($UserName, $Password)
ref_type (enum): "def-api" or "def-testcase"
project_mapping (dict): project_mapping
def _get_api_definition(name):
""" get api definition by name.
Returns:
dict: api/testcase definition.
Raises:
exceptions.ParamsError: call args number is not equal to defined args number.
"""
function_meta = parser.parse_function(ref_call)
func_name = function_meta["func_name"]
call_args = function_meta["args"]
block = _get_test_definition(func_name, ref_type, project_mapping)
def_args = block.get("function_meta", {}).get("args", [])
if len(call_args) != len(def_args):
err_msg = "{}: call args number is not equal to defined args number!\n".format(func_name)
err_msg += "defined args: {}\n".format(def_args)
err_msg += "reference args: {}".format(call_args)
logger.log_error(err_msg)
raise exceptions.ParamsError(err_msg)
args_mapping = {}
for index, item in enumerate(def_args):
if call_args[index] == item:
continue
args_mapping[item] = call_args[index]
if args_mapping:
block = parser.substitute_variables(block, args_mapping)
return block
def _get_test_definition(name, ref_type, project_mapping):
""" get expected api or testcase.
Args:
name (str): api or testcase name
ref_type (enum): "def-api" or "def-testcase"
project_mapping (dict): project_mapping
Returns:
dict: expected api/testcase info if found.
dict: expected api definition if found.
Raises:
exceptions.ApiNotFound: api not found
exceptions.TestcaseNotFound: testcase not found
"""
block = project_mapping.get(ref_type, {}).get(name)
# NOTICE: avoid project_mapping been changed during iteration.
block = copy.deepcopy(block)
if not block:
err_msg = "{} not found!".format(name)
if ref_type == "def-api":
raise exceptions.ApiNotFound(err_msg)
else:
# ref_type == "def-testcase":
raise exceptions.TestcaseNotFound(err_msg)
return block
def _extend_block(ref_block, def_block):
""" extend ref_block with def_block, ref_block will merge and override def_block.
Args:
def_block (dict): api definition dict.
ref_block (dict): reference block
Returns:
dict: extended reference block.
Examples:
>>> def_block = {
"name": "get token 1",
"request": {...},
"validate": [{'eq': ['status_code', 200]}]
}
>>> ref_block = {
"name": "get token 2",
"extract": [{"token": "content.token"}],
"validate": [{'eq': ['status_code', 201]}, {'len_eq': ['content.token', 16]}]
}
>>> _extend_block(def_block, ref_block)
{
"name": "get token 2",
"request": {...},
"extract": [{"token": "content.token"}],
"validate": [{'eq': ['status_code', 201]}, {'len_eq': ['content.token', 16]}]
}
"""
extended_block = {}
# override name
extended_block["name"] = ref_block.pop("name", None) or def_block.pop("name", "")
# override variables
def_variables = def_block.pop("variables", [])
ref_variables = ref_block.pop("variables", [])
extended_block["variables"] = _extend_variables(
def_variables,
ref_variables
)
# merge & override validators
def_validators = def_block.pop("validate", [])
ref_validators = ref_block.pop("validate", [])
extended_block["validate"] = _extend_validators(
def_validators,
ref_validators
)
# merge & override extractors
def_extrators = def_block.pop("extract", [])
ref_extractors = ref_block.pop("extract", [])
extended_block["extract"] = _extend_variables(
def_extrators,
ref_extractors
)
# TODO: merge & override request
def_request = def_block.pop("request", {})
ref_request = ref_block.pop("request", {})
extended_block["request"] = def_request
# merge & override setup_hooks
def_setup_hooks = def_block.pop("setup_hooks", [])
ref_setup_hooks = ref_block.pop("setup_hooks", [])
extended_block["setup_hooks"] = list(set(def_setup_hooks + ref_setup_hooks))
# merge & override teardown_hooks
def_teardown_hooks = def_block.pop("teardown_hooks", [])
ref_teardown_hooks = ref_block.pop("teardown_hooks", [])
extended_block["teardown_hooks"] = list(set(def_teardown_hooks + ref_teardown_hooks))
# TODO: extend with other ref block items, e.g. times
# extended_block.update(def_block)
extended_block.update(ref_block)
return extended_block
def _convert_validators_to_mapping(validators):
""" convert validators list to mapping.
Args:
validators (list): validators in list
Returns:
dict: validators mapping, use (check, comparator) as key.
Examples:
>>> validators = [
{"check": "v1", "expect": 201, "comparator": "eq"},
{"check": {"b": 1}, "expect": 200, "comparator": "eq"}
]
>>> _convert_validators_to_mapping(validators)
{
("v1", "eq"): {"check": "v1", "expect": 201, "comparator": "eq"},
('{"b": 1}', "eq"): {"check": {"b": 1}, "expect": 200, "comparator": "eq"}
}
"""
validators_mapping = {}
for validator in validators:
validator = parser.parse_validator(validator)
if not isinstance(validator["check"], collections.Hashable):
check = json.dumps(validator["check"])
else:
check = validator["check"]
key = (check, validator["comparator"])
validators_mapping[key] = validator
return validators_mapping
def _extend_validators(def_validators, ref_validators):
""" extend ref_validators with def_validators.
ref_validators will merge and override def_validators.
Args:
def_validators (list):
ref_validators (list):
Returns:
list: extended validators
Examples:
>>> def_validators = [{'eq': ['v1', 200]}, {"check": "s2", "expect": 16, "comparator": "len_eq"}]
>>> ref_validators = [{"check": "v1", "expect": 201}, {'len_eq': ['s3', 12]}]
>>> _extend_validators(def_validators, ref_validators)
[
{"check": "v1", "expect": 201, "comparator": "eq"},
{"check": "s2", "expect": 16, "comparator": "len_eq"},
{"check": "s3", "expect": 12, "comparator": "len_eq"}
]
"""
if not def_validators:
return ref_validators
elif not ref_validators:
return def_validators
else:
def_validators_mapping = _convert_validators_to_mapping(def_validators)
ref_validators_mapping = _convert_validators_to_mapping(ref_validators)
def_validators_mapping.update(ref_validators_mapping)
return list(def_validators_mapping.values())
def _extend_variables(def_variables, ref_variables):
""" extend ref_variables with def_variables.
ref_variables will merge and override def_variables.
Args:
def_variables (list):
ref_variables (list):
Returns:
list: extended variables
Examples:
>>> def_variables = [{"var1": "val1"}, {"var2": "val2"}]
>>> ref_variables = [{"var1": "val111"}, {"var3": "val3"}]
>>> _extend_variables(def_variables, ref_variables)
[
{"var1": "val111"},
{"var2": "val2"},
{"var3": "val3"}
]
"""
if not def_variables:
return ref_variables
elif not ref_variables:
return def_variables
else:
extended_variables_dict = OrderedDict()
for def_variable in def_variables:
var_name = list(def_variable.keys())[0]
extended_variables_dict[var_name] = def_variable[var_name]
for ref_variable in ref_variables:
if not ref_variable:
continue
var_name = list(ref_variable.keys())[0]
extended_variables_dict[var_name] = ref_variable[var_name]
extended_variables = []
for key, value in extended_variables_dict.items():
extended_variables.append({key: value})
return extended_variables
try:
block = tests_def_mapping["api"][name]
# NOTICE: avoid project_mapping been changed during iteration.
return utils.deepcopy_dict(block)
except KeyError:
raise exceptions.ApiNotFound("{} not found!".format(name))
def load_folder_content(folder_path):
@@ -737,99 +499,16 @@ def load_api_folder(api_folder_path):
for api_item in api_items:
key, api_dict = api_item.popitem()
# TODO: replace def with api file path
api_def = api_dict.pop("def")
function_meta = parser.parse_function(api_def)
func_name = function_meta["func_name"]
# TODO: replace id with api file path
api_id = api_dict.get("id")
if api_id in api_definition_mapping:
logger.log_warning("API definition duplicated: {}".format(api_id))
if func_name in api_definition_mapping:
logger.log_warning("API definition duplicated: {}".format(func_name))
api_dict["function_meta"] = function_meta
api_definition_mapping[func_name] = api_dict
api_definition_mapping[api_id] = api_dict
return api_definition_mapping
def load_test_folder(test_folder_path):
""" load testcases definitions from folder.
Args:
test_folder_path (str): testcases files folder.
testcase file should be in the following format:
[
{
"config": {
"def": "create_and_check",
"request": {},
"validate": []
}
},
{
"test": {
"api": "get_user",
"validate": []
}
}
]
Returns:
dict: testcases definition mapping.
{
"create_and_check": [
{"config": {}},
{"test": {}},
{"test": {}}
],
"tests/testcases/create_and_get.yml": [
{"config": {}},
{"test": {}},
{"test": {}}
]
}
"""
test_definition_mapping = {}
test_items_mapping = load_folder_content(test_folder_path)
for test_file_path, items in test_items_mapping.items():
# TODO: add JSON schema validation
testcase = {
"config": {},
"teststeps": []
}
for item in items:
key, block = item.popitem()
if key == "config":
testcase["config"].update(block)
if "def" not in block:
test_definition_mapping[test_file_path] = testcase
continue
# TODO: replace def with testcase file path
testcase_def = block.pop("def")
function_meta = parser.parse_function(testcase_def)
func_name = function_meta["func_name"]
if func_name in test_definition_mapping:
logger.log_warning("API definition duplicated: {}".format(func_name))
testcase["function_meta"] = function_meta
test_definition_mapping[func_name] = testcase
else:
# key == "test":
### TODO: extend suite with api
testcase["teststeps"].append(block)
return test_definition_mapping
def load_debugtalk_py(start_path):
""" locate debugtalk.py file and returns PWD and debugtalk.py functions.
@@ -878,22 +557,17 @@ def load_project_tests(test_path, dot_env_path=None):
dict: project loaded api/testcases definitions, environments and debugtalk.py functions.
"""
project_mapping = {}
# locate PWD and load debugtalk.py functions
project_working_directory, debugtalk_functions = load_debugtalk_py(test_path)
project_mapping["PWD"] = project_working_directory
project_mapping["functions"] = debugtalk_functions
# load .env
dot_env_path = dot_env_path or os.path.join(project_working_directory, ".env")
project_mapping["env"] = load_dot_env_file(dot_env_path)
# load api/testcases
project_mapping["def-api"] = load_api_folder(os.path.join(project_working_directory, "api"))
# TODO: replace suite with testcases
project_mapping["def-testcase"] = load_test_folder(os.path.join(project_working_directory, "suite"))
return project_mapping
# load api
tests_def_mapping["api"] = load_api_folder(os.path.join(project_working_directory, "api"))
def load_tests(path, dot_env_path=None):
@@ -901,54 +575,44 @@ def load_tests(path, dot_env_path=None):
Args:
path (str/list): testcase file/foler path.
path could be in several types:
path could be in 2 types:
- absolute/relative file path
- absolute/relative folder path
- list/set container with file(s) and/or folder(s)
dot_env_path (str): specified .env file path
Returns:
list: testcases list, each testcase is corresponding to a file
[
{ # testcase data structure
"config": {
"name": "desc1",
"path": "testcase1_path",
"variables": [], # optional
"request": {}, # optional
dict: tests mapping, include project_mapping and testcases.
each testcase is corresponding to a file.
{
"project_mapping": {
"PWD": "XXXXX",
"functions": {},
"env": {},
"def-api": {},
"def-testcase": {}
"env": {}
},
"teststeps": [
# teststep data structure
{
'name': 'test step desc1',
'variables': [], # optional
'extract': [], # optional
'validate': [],
'request': {},
'function_meta': {}
"testcases": [
{ # testcase data structure
"config": {
"name": "desc1",
"path": "testcase1_path",
"variables": [], # optional
},
"teststeps": [
# teststep data structure
{
'name': 'test step desc1',
'variables': [], # optional
'extract': [], # optional
'validate': [],
'request': {}
},
teststep2 # another teststep dict
]
},
teststep2 # another teststep dict
testcase_dict_2 # another testcase dict
]
},
testcase_dict_2 # another testcase dict
]
}
"""
if isinstance(path, (list, set)):
testcases_list = []
for file_path in set(path):
testcases = load_tests(file_path, dot_env_path)
if not testcases:
continue
testcases_list.extend(testcases)
return testcases_list
if not os.path.exists(path):
err_msg = "path not exist: {}".format(path)
logger.log_error(err_msg)
@@ -957,22 +621,41 @@ def load_tests(path, dot_env_path=None):
if not os.path.isabs(path):
path = os.path.join(os.getcwd(), path)
load_project_tests(path, dot_env_path)
tests_mapping = {
"project_mapping": project_mapping
}
def load_test_file(path):
raw_testcase = load_file(path)
try:
testcase = load_testcase(raw_testcase)
testcase["config"]["path"] = path
except exceptions.FileFormatError:
testcase = {}
return testcase
testcases_list = []
if os.path.isdir(path):
files_list = load_folder_files(path)
testcases_list = load_tests(files_list, dot_env_path)
for path in files_list:
testcase = load_test_file(path)
if not testcase:
continue
testcases_list.append(testcase)
elif os.path.isfile(path):
try:
raw_testcase = load_file(path)
project_mapping = load_project_tests(path, dot_env_path)
testcase = _load_testcase(raw_testcase, project_mapping)
testcase["config"]["path"] = path
testcase["config"].update(project_mapping)
testcases_list = [testcase]
except exceptions.FileFormatError:
testcases_list = []
return testcases_list
testcase = load_test_file(path)
if testcase:
testcases_list.append(testcase)
tests_mapping["testcases"] = testcases_list
return tests_mapping
def load_locust_tests(path, dot_env_path=None):
@@ -999,7 +682,7 @@ def load_locust_tests(path, dot_env_path=None):
"""
raw_testcase = load_file(path)
project_mapping = load_project_tests(path, dot_env_path)
load_project_tests(path, dot_env_path)
config = {}
tests = []
@@ -1009,10 +692,10 @@ def load_locust_tests(path, dot_env_path=None):
if key == "config":
config.update(test_block)
elif key == "test":
teststeps = _load_teststeps(test_block, project_mapping)
weight = test_block.pop("weight", 1)
teststep = load_teststep(test_block)
weight = test_block.get("weight", 1)
for _ in range(weight):
tests.append(teststeps)
tests.append(teststep)
# parse config variables
raw_config_variables = config.get("variables", [])

412
httprunner/parser.py
View File

@@ -3,6 +3,7 @@
import ast
import os
import re
from collections import OrderedDict
from httprunner import exceptions, utils
from httprunner.compat import basestring, builtin_str, numeric_types, str
@@ -516,7 +517,7 @@ def parse_data(content, variables_mapping=None, functions_mapping=None):
for item in content
]
if isinstance(content, dict):
if isinstance(content, (dict, OrderedDict)):
parsed_content = {}
for key, value in content.items():
parsed_key = parse_data(key, variables_mapping, functions_mapping)
@@ -527,7 +528,7 @@ def parse_data(content, variables_mapping=None, functions_mapping=None):
if isinstance(content, basestring):
# content is in string format here
variables_mapping = variables_mapping or {}
variables_mapping = utils.ensure_mapping_format(variables_mapping or {})
functions_mapping = functions_mapping or {}
content = content.strip()
@@ -541,110 +542,327 @@ def parse_data(content, variables_mapping=None, functions_mapping=None):
return content
def parse_tests(testcases, variables_mapping=None):
""" parse testcases configs, including variables/parameters/name/request.
def _extend_with_api(teststep_dict, api_def_dict):
""" extend teststep with api definition, teststep will merge and override api definition.
Args:
testcases (list): testcase list, with config unparsed.
[
{ # testcase data structure
"config": {
"name": "desc1",
"path": "testcase1_path",
"variables": {}, # optional
"request": {} # optional
"functions": {},
"env": {},
"def-api": {},
"def-testcase": {}
},
"teststeps": [
# teststep data structure
{
'name': 'test step desc2',
'variables': [], # optional
'extract': [], # optional
'validate': [],
'request': {},
'function_meta': {}
},
teststep2 # another teststep dict
]
teststep_dict (dict): teststep block
api_def_dict (dict): api definition
Returns:
dict: extended teststep dict.
Examples:
>>> api_def_dict = {
"name": "get token 1",
"request": {...},
"validate": [{'eq': ['status_code', 200]}]
}
>>> teststep_dict = {
"name": "get token 2",
"extract": [{"token": "content.token"}],
"validate": [{'eq': ['status_code', 201]}, {'len_eq': ['content.token', 16]}]
}
>>> _extend_with_api(teststep_dict, api_def_dict)
{
"name": "get token 2",
"request": {...},
"extract": [{"token": "content.token"}],
"validate": [{'eq': ['status_code', 201]}, {'len_eq': ['content.token', 16]}]
}
"""
# override name
api_def_name = api_def_dict.pop("name", "")
teststep_dict["name"] = teststep_dict.get("name") or api_def_name
# override variables
def_variables = api_def_dict.pop("variables", [])
teststep_dict["variables"] = utils.extend_variables(
def_variables,
teststep_dict.get("variables", {})
)
# merge & override validators TODO: relocate
def_raw_validators = api_def_dict.pop("validate", [])
ref_raw_validators = teststep_dict.get("validate", [])
def_validators = [
parse_validator(validator)
for validator in def_raw_validators
]
ref_validators = [
parse_validator(validator)
for validator in ref_raw_validators
]
teststep_dict["validate"] = utils.extend_validators(
def_validators,
ref_validators
)
# merge & override extractors
def_extrators = api_def_dict.pop("extract", [])
teststep_dict["extract"] = utils.extend_variables(
def_extrators,
teststep_dict.get("extract", [])
)
# TODO: merge & override request
teststep_dict["request"] = api_def_dict.pop("request", {})
# base_url
base_url = teststep_dict.pop("base_url", None)
if base_url:
teststep_dict["request"]["url"] = "{}/{}".format(base_url.rstrip("/"), teststep_dict["request"]["url"].lstrip("/"))
# verify
if "verify" in teststep_dict:
verify = teststep_dict.pop("verify")
elif "verify" in api_def_dict:
verify = api_def_dict.pop("verify")
else:
verify = True
teststep_dict["request"]["verify"] = verify
# merge & override setup_hooks
def_setup_hooks = api_def_dict.pop("setup_hooks", [])
ref_setup_hooks = teststep_dict.get("setup_hooks", [])
extended_setup_hooks = list(set(def_setup_hooks + ref_setup_hooks))
if extended_setup_hooks:
teststep_dict["setup_hooks"] = extended_setup_hooks
# merge & override teardown_hooks
def_teardown_hooks = api_def_dict.pop("teardown_hooks", [])
ref_teardown_hooks = teststep_dict.get("teardown_hooks", [])
extended_teardown_hooks = list(set(def_teardown_hooks + ref_teardown_hooks))
if extended_teardown_hooks:
teststep_dict["teardown_hooks"] = extended_teardown_hooks
# TODO: extend with other api definition items, e.g. times
teststep_dict.update(api_def_dict)
return teststep_dict
def _extend_with_testcase(teststep_dict, testcase_def_dict):
""" extend teststep with testcase definition
teststep will merge and override testcase config definition.
Args:
teststep_dict (dict): teststep block
testcase_def_dict (dict): testcase definition
Returns:
dict: extended teststep dict.
"""
# override testcase config variables
testcase_def_dict["config"].setdefault("variables", {})
testcase_def_variables = utils.ensure_mapping_format(testcase_def_dict["config"].get("variables", {}))
testcase_def_variables.update(teststep_dict.pop("variables", {}))
testcase_def_dict["config"]["variables"] = testcase_def_variables
# override base_url, verify
# priority: testcase config > testsuite teststep
teststep_base_url = teststep_dict.pop("base_url", None)
teststep_verify = teststep_dict.pop("verify", True)
testcase_def_dict["config"].setdefault("base_url", teststep_base_url)
testcase_def_dict["config"].setdefault("verify", teststep_verify)
# override testcase config name, output, etc.
testcase_def_dict["config"].update(teststep_dict)
teststep_dict.clear()
teststep_dict.update(testcase_def_dict)
def __parse_config(config, project_mapping):
""" parse testcase config, include variables and name.
"""
# get config variables
raw_config_variables = config.pop("variables", {})
raw_config_variables_mapping = utils.ensure_mapping_format(raw_config_variables)
override_variables = utils.deepcopy_dict(project_mapping.get("variables", {}))
functions = project_mapping.get("functions", {})
# override testcase config variables with passed in variables
for key, value in raw_config_variables_mapping.items():
if key in override_variables:
# passed in
continue
else:
# config variables
try:
parsed_value = parse_data(
value,
override_variables,
functions
)
except exceptions.VariableNotFound:
pass
override_variables[key] = parsed_value
if override_variables:
config["variables"] = override_variables
# parse config name
config["name"] = parse_data(
config.get("name", ""),
override_variables,
functions
)
# parse config base_url
if "base_url" in config:
config["base_url"] = parse_data(
config["base_url"],
override_variables,
functions
)
def __parse_teststeps(teststeps, config, project_mapping):
""" override teststeps with testcase config variables, base_url and verify.
teststep maybe nested testcase.
variables priority:
testsuite config > testsuite teststep > testcase config > testcase teststep > api
base_url/verify priority:
testcase teststep > testcase config > testsuite teststep > testsuite config > api
Args:
teststeps (list):
config (dict):
Returns:
list: overrided teststeps
"""
config_variables = config.pop("variables", {})
config_base_url = config.pop("base_url", None)
config_verify = config.pop("verify", True)
functions = project_mapping.get("functions", {})
for teststep in teststeps:
# priority teststep > config
teststep.setdefault("base_url", config_base_url)
teststep.setdefault("verify", config_verify)
if "testcase_def" in teststep:
# teststep is nested testcase
# 1, testsuite config => testsuite teststeps
# override teststep variables
teststep["variables"] = utils.extend_variables(
teststep.pop("variables", {}),
config_variables
)
# parse teststep name
try:
teststep["name"] = parse_data(
teststep.pop("name", ""),
teststep["variables"],
functions
)
except exceptions.VariableNotFound:
pass
# 2, testsuite teststep => testcase config
testcase_def = teststep.pop("testcase_def")
_extend_with_testcase(teststep, testcase_def)
# 3, testcase config => testcase teststep
_parse_testcase(teststep, project_mapping)
else:
# 1, config => teststeps
# override teststep variables
teststep["variables"] = utils.extend_variables(
teststep.pop("variables", {}),
config_variables
)
# parse teststep name
try:
teststep["name"] = parse_data(
teststep.pop("name", ""),
teststep["variables"],
functions
)
except exceptions.VariableNotFound:
pass
if "api_def" in teststep:
# 2, teststep => api
api_def_dict = teststep.pop("api_def")
_extend_with_api(teststep, api_def_dict)
else:
# base_url
base_url = teststep.pop("base_url", None)
if base_url:
teststep["request"]["url"] = "{}/{}".format(base_url.rstrip("/"), teststep["request"]["url"].lstrip("/"))
def _parse_testcase(testcase, project_mapping):
__parse_config(testcase["config"], project_mapping)
__parse_teststeps(testcase["teststeps"], testcase["config"], project_mapping)
def parse_tests(tests_mapping):
""" parse testcases configs, including variables/name/request.
Args:
tests_mapping (dict): project info and testcases list.
{
"project_mapping": {
"PWD": "XXXXX",
"functions": {},
"variables": {}, # optional, priority 1
"env": {}
},
testcase_dict_2 # another testcase dict
]
"testcases": [
{ # testcase data structure
"config": {
"name": "desc1",
"path": "testcase1_path",
"variables": [], # optional, priority 2
},
"teststeps": [
# teststep data structure
{
'name': 'test step desc1',
'variables': [], # optional, priority 3
'extract': [],
'validate': [],
'api_def': {
"variables": {} # optional, priority 4
'request': {},
}
},
teststep2 # another teststep dict
]
},
testcase_dict_2 # another testcase dict
]
}
variables_mapping (dict): if variables_mapping is specified, it will override variables in config block.
Returns:
list: parsed testcases list, with config variables/parameters/name/request parsed.
"""
variables_mapping = variables_mapping or {}
parsed_testcases_list = []
project_mapping = tests_mapping.get("project_mapping", {})
for testcase in testcases:
testcase_config = testcase.setdefault("config", {})
functions = testcase_config.get("functions", {})
env_mapping = project_mapping.get("env", {})
# set OS environment variables
utils.set_os_environ(env_mapping)
env_mapping = testcase_config.get("env", {})
# set OS environment variables
utils.set_os_environ(env_mapping)
for testcase in tests_mapping["testcases"]:
testcase.setdefault("config", {})
_parse_testcase(testcase, project_mapping)
# parse config parameters
config_parameters = testcase_config.pop("parameters", [])
cartesian_product_parameters_list = parse_parameters(
config_parameters,
testcase_config.get("variables", {}),
functions
) or [{}]
for parameter_mapping in cartesian_product_parameters_list:
testcase_dict = utils.deepcopy_dict(testcase)
config = testcase_dict.get("config")
# get config variables
raw_config_variables = config.get("variables", [])
raw_config_variables_mapping = utils.ensure_mapping_format(raw_config_variables)
# priority: passed in > parameters > variables
config_variables = utils.deepcopy_dict(parameter_mapping)
config_variables.update(variables_mapping)
for key, value in raw_config_variables_mapping.items():
if key in config_variables:
# passed in & .env & parameters
continue
else:
# config variables
parsed_value = parse_data(
value,
config_variables,
functions
)
config_variables[key] = parsed_value
testcase_dict["config"]["variables"] = config_variables
# parse config name
testcase_dict["config"]["name"] = parse_data(
testcase_dict["config"].get("name", ""),
config_variables,
functions
)
# parse config request
testcase_dict["config"]["request"] = parse_data(
testcase_dict["config"].get("request", {}),
config_variables,
functions
)
# put loaded project functions to config
testcase_dict["config"]["functions"] = functions
parsed_testcases_list.append(testcase_dict)
# unset OS environment variables
utils.unset_os_environ(env_mapping)
return parsed_testcases_list
# unset OS environment variables
utils.unset_os_environ(env_mapping)

215
httprunner/runner.py
View File

@@ -4,31 +4,62 @@ 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
from httprunner.context import SessionContext
class Runner(object):
""" Running testcases.
Examples:
>>> functions={...}
>>> config = {
"name": "XXXX",
"base_url": "http://127.0.0.1",
"verify": False
}
>>> runner = Runner(config, functions)
>>> teststep = {
"name": "teststep description",
"variables": [], # optional
"request": {
"url": "http://127.0.0.1:5000/api/users/1000",
"method": "GET"
}
}
>>> runner.run_test(teststep)
"""
def __init__(self, config, functions, http_client_session=None):
""" run testcase or testsuite.
Args:
config (dict): testcase/testsuite config dict
{
"name": "ABC",
"variables": {},
"setup_hooks", [],
"teardown_hooks", []
}
http_client_session (instance): requests.Session(), or locust.client.Session() instance.
def __init__(self, config_dict=None, http_client_session=None):
""" Each testcase is corresponding to one Runner() instance
and one Context() instance.
"""
self.http_client_session = http_client_session
config_dict = config_dict or {}
base_url = config.get("base_url")
self.verify = config.get("verify", True)
self.output = config.get("output", [])
self.functions = functions
self.evaluated_validators = []
# 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_setup_hooks = config.get("setup_hooks", [])
# testcase teardown hooks
self.testcase_teardown_hooks = config_dict.pop("teardown_hooks", [])
self.testcase_teardown_hooks = config.get("teardown_hooks", [])
self.context = Context(config_variables, config_functions)
self.init_test(config_dict, "testcase")
self.http_client_session = http_client_session or HttpSession(base_url)
self.session_context = SessionContext(self.functions)
if testcase_setup_hooks:
self.do_hook_actions(testcase_setup_hooks)
@@ -37,57 +68,6 @@ class Runner(object):
if self.testcase_teardown_hooks:
self.do_hook_actions(self.testcase_teardown_hooks)
def init_test(self, test_dict, level):
""" create/update context variables binds
Args:
test_dict (dict):
level (enum): "testcase" or "teststep"
testcase:
{
"name": "testcase description",
"variables": [], # optional
"request": {
"base_url": "http://127.0.0.1:5000",
"headers": {
"User-Agent": "iOS/2.8.3"
}
}
}
teststep:
{
"name": "teststep description",
"variables": [], # optional
"request": {
"url": "/api/get-token",
"method": "POST",
"headers": {
"Content-Type": "application/json"
}
},
"json": {
"sign": "f1219719911caae89ccc301679857ebfda115ca2"
}
}
Returns:
dict: parsed request dict
"""
test_dict = utils.lower_test_dict_keys(test_dict)
self.context.init_context_variables(level)
variables = test_dict.get('variables', OrderedDict())
self.context.update_context_variables(variables, level)
request_config = test_dict.get('request', {})
parsed_request = self.context.get_parsed_request(request_config, level)
base_url = parsed_request.pop("base_url", None)
self.http_client_session = self.http_client_session or HttpSession(base_url)
return parsed_request
def _handle_skip_feature(self, teststep_dict):
""" handle skip feature for teststep
- skip: skip current test unconditionally
@@ -109,12 +89,12 @@ class Runner(object):
elif "skipIf" in teststep_dict:
skip_if_condition = teststep_dict["skipIf"]
if self.context.eval_content(skip_if_condition):
if self.session_context.eval_content(skip_if_condition):
skip_reason = "{} evaluate to True".format(skip_if_condition)
elif "skipUnless" in teststep_dict:
skip_unless_condition = teststep_dict["skipUnless"]
if not self.context.eval_content(skip_unless_condition):
if not self.session_context.eval_content(skip_unless_condition):
skip_reason = "{} evaluate to False".format(skip_unless_condition)
if skip_reason:
@@ -124,9 +104,9 @@ class Runner(object):
for action in actions:
logger.log_debug("call hook: {}".format(action))
# TODO: check hook function if valid
self.context.eval_content(action)
self.session_context.eval_content(action)
def run_test(self, teststep_dict):
def _run_teststep(self, teststep_dict):
""" run single teststep.
Args:
@@ -135,7 +115,7 @@ class Runner(object):
"name": "teststep description",
"skip": "skip this test unconditionally",
"times": 3,
"variables": [], # optional, override
"variables": [], # optional, override
"request": {
"url": "http://127.0.0.1:5000/api/users/1000",
"method": "POST",
@@ -162,10 +142,14 @@ class Runner(object):
self._handle_skip_feature(teststep_dict)
# prepare
extractors = teststep_dict.get("extract", [])
validators = teststep_dict.get("validate", [])
parsed_request = self.init_test(teststep_dict, level="teststep")
self.context.update_teststep_variables_mapping("request", parsed_request)
teststep_dict = utils.lower_test_dict_keys(teststep_dict)
teststep_variables = teststep_dict.get("variables", {})
self.session_context.init_teststep_variables(teststep_variables)
# parse teststep request
raw_request = teststep_dict.get('request', {})
parsed_teststep_request = self.session_context.eval_content(raw_request)
self.session_context.update_teststep_variables("request", parsed_teststep_request)
# setup hooks
setup_hooks = teststep_dict.get("setup_hooks", [])
@@ -173,9 +157,10 @@ class Runner(object):
self.do_hook_actions(setup_hooks)
try:
url = parsed_request.pop('url')
method = parsed_request.pop('method')
group_name = parsed_request.pop("group", None)
url = parsed_teststep_request.pop('url')
method = parsed_teststep_request.pop('method')
parsed_teststep_request.setdefault("verify", self.verify)
group_name = parsed_teststep_request.pop("group", None)
except KeyError:
raise exceptions.ParamsError("URL or METHOD missed!")
@@ -188,14 +173,14 @@ class Runner(object):
raise exceptions.ParamsError(err_msg)
logger.log_info("{method} {url}".format(method=method, url=url))
logger.log_debug("request kwargs(raw): {kwargs}".format(kwargs=parsed_request))
logger.log_debug("request kwargs(raw): {kwargs}".format(kwargs=parsed_teststep_request))
# request
resp = self.http_client_session.request(
method,
url,
name=group_name,
**parsed_request
**parsed_teststep_request
)
resp_obj = response.ResponseObject(resp)
@@ -203,21 +188,23 @@ class Runner(object):
teardown_hooks = teststep_dict.get("teardown_hooks", [])
if teardown_hooks:
logger.log_info("start to run teardown hooks")
self.context.update_teststep_variables_mapping("response", resp_obj)
self.session_context.update_teststep_variables("response", resp_obj)
self.do_hook_actions(teardown_hooks)
# extract
extractors = teststep_dict.get("extract", [])
extracted_variables_mapping = resp_obj.extract_response(extractors)
self.context.update_testcase_runtime_variables_mapping(extracted_variables_mapping)
self.session_context.update_seesion_variables(extracted_variables_mapping)
# validate
validators = teststep_dict.get("validate", [])
try:
self.evaluated_validators = self.context.validate(validators, resp_obj)
self.evaluated_validators = self.session_context.validate(validators, resp_obj)
except (exceptions.ParamsError, exceptions.ValidationFailure, exceptions.ExtractFailure):
# log request
err_req_msg = "request: \n"
err_req_msg += "headers: {}\n".format(parsed_request.pop("headers", {}))
for k, v in parsed_request.items():
err_req_msg += "headers: {}\n".format(parsed_teststep_request.pop("headers", {}))
for k, v in parsed_teststep_request.items():
err_req_msg += "{}: {}\n".format(k, repr(v))
logger.log_error(err_req_msg)
@@ -230,10 +217,66 @@ class Runner(object):
raise
def _run_testcase(self, testcase_dict):
""" run single testcase.
"""
config = testcase_dict.get("config", {})
test_runner = Runner(config, self.functions, self.http_client_session)
teststeps = testcase_dict.get("teststeps", [])
for index, teststep_dict in enumerate(teststeps):
test_runner.run_test(teststep_dict)
self.session_context.update_seesion_variables(test_runner.extract_sessions())
def run_test(self, teststep_dict):
""" run single teststep of testcase.
teststep_dict may be in 3 types.
Args:
teststep_dict (dict):
# teststep
{
"name": "teststep description",
"variables": [], # optional
"request": {
"url": "http://127.0.0.1:5000/api/users/1000",
"method": "GET"
}
}
# embeded testcase
{
"config": {...},
"teststeps": [
{...},
{...}
]
}
# TODO: function
{
"name": "exec function",
"function": "${func()}"
}
"""
if "config" in teststep_dict:
self._run_testcase(teststep_dict)
else:
# api
self._run_teststep(teststep_dict)
def extract_sessions(self):
"""
"""
return self.extract_output(self.output)
def extract_output(self, output_variables_list):
""" extract output variables
"""
variables_mapping = self.context.teststep_variables_mapping
variables_mapping = self.session_context.session_variables_mapping
output = {}
for variable in output_variables_list:

107
httprunner/utils.py
View File

@@ -268,6 +268,113 @@ def ensure_mapping_format(variables):
return variables_ordered_dict
def _convert_validators_to_mapping(validators):
""" convert validators list to mapping.
Args:
validators (list): validators in list
Returns:
dict: validators mapping, use (check, comparator) as key.
Examples:
>>> validators = [
{"check": "v1", "expect": 201, "comparator": "eq"},
{"check": {"b": 1}, "expect": 200, "comparator": "eq"}
]
>>> _convert_validators_to_mapping(validators)
{
("v1", "eq"): {"check": "v1", "expect": 201, "comparator": "eq"},
('{"b": 1}', "eq"): {"check": {"b": 1}, "expect": 200, "comparator": "eq"}
}
"""
validators_mapping = {}
for validator in validators:
if not isinstance(validator["check"], collections.Hashable):
check = json.dumps(validator["check"])
else:
check = validator["check"]
key = (check, validator["comparator"])
validators_mapping[key] = validator
return validators_mapping
def extend_validators(raw_validators, override_validators):
""" extend raw_validators with override_validators.
override_validators will merge and override raw_validators.
Args:
raw_validators (dict):
override_validators (dict):
Returns:
list: extended validators
Examples:
>>> raw_validators = [{'eq': ['v1', 200]}, {"check": "s2", "expect": 16, "comparator": "len_eq"}]
>>> override_validators = [{"check": "v1", "expect": 201}, {'len_eq': ['s3', 12]}]
>>> extend_validators(raw_validators, override_validators)
[
{"check": "v1", "expect": 201, "comparator": "eq"},
{"check": "s2", "expect": 16, "comparator": "len_eq"},
{"check": "s3", "expect": 12, "comparator": "len_eq"}
]
"""
if not raw_validators:
return override_validators
elif not override_validators:
return raw_validators
else:
def_validators_mapping = _convert_validators_to_mapping(raw_validators)
ref_validators_mapping = _convert_validators_to_mapping(override_validators)
def_validators_mapping.update(ref_validators_mapping)
return list(def_validators_mapping.values())
def extend_variables(raw_variables, override_variables):
""" extend raw_variables with override_variables.
override_variables will merge and override raw_variables.
Args:
raw_variables (list):
override_variables (list):
Returns:
OrderedDict: extended variables mapping
Examples:
>>> raw_variables = [{"var1": "val1"}, {"var2": "val2"}]
>>> override_variables = [{"var1": "val111"}, {"var3": "val3"}]
>>> extend_variables(raw_variables, override_variables)
OrderedDict([
('var1', 'val111'),
('var2', 'val2'),
('var3', 'val3')
])
"""
if not raw_variables:
return override_variables
elif not override_variables:
return raw_variables
else:
raw_variables_mapping = ensure_mapping_format(raw_variables)
override_variables_mapping = ensure_mapping_format(override_variables)
raw_variables_mapping.update(override_variables_mapping)
return raw_variables_mapping
def get_testcase_io(testcase):
""" get testcase input(variables) and output.

47
httprunner/validator.py
View File

@@ -54,21 +54,50 @@ def is_testcases(data_structure):
Args:
data_structure (dict): testcase(s) should always be in the following data structure:
{
"project_mapping": {
"PWD": "XXXXX",
"functions": {},
"env": {}
},
"testcases": [
{ # testcase data structure
"config": {
"name": "desc1",
"path": "testcase1_path",
"variables": [], # optional
},
"teststeps": [
# teststep data structure
{
'name': 'test step desc1',
'variables': [], # optional
'extract': [], # optional
'validate': [],
'request': {}
},
teststep2 # another teststep dict
]
},
testcase_dict_2 # another testcase dict
]
}
testcase_dict
or
[
testcase_dict_1,
testcase_dict_2
]
Returns:
bool: True if data_structure is valid testcase(s), otherwise False.
"""
if not isinstance(data_structure, list):
return is_testcase(data_structure)
if not isinstance(data_structure, dict):
return False
for item in data_structure:
if "testcases" not in data_structure:
return False
testcases = data_structure["testcases"]
if not isinstance(testcases, list):
return False
for item in testcases:
if not is_testcase(item):
return False