Merge pull request #348 from HttpRunner/parser

refactor all
This commit is contained in:
debugtalk
2018-08-22 19:05:59 +08:00
committed by GitHub
26 changed files with 1950 additions and 1924 deletions

View File

@@ -18,5 +18,6 @@ coverage = "*"
coveralls = "*" coveralls = "*"
twine = "*" twine = "*"
contextlib2 = "*" contextlib2 = "*"
locustio = "*"
[scripts] [scripts]

View File

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

View File

@@ -1,3 +1,3 @@
# encoding: utf-8 # encoding: utf-8
from httprunner.task import HttpRunner from httprunner.api import HttpRunner, LocustRunner

325
httprunner/api.py Normal file
View File

@@ -0,0 +1,325 @@
# encoding: utf-8
import os
import unittest
from httprunner import (exceptions, loader, logger, parser, report, runner,
utils, validator)
class HttpRunner(object):
def __init__(self, **kwargs):
""" initialize HttpRunner.
Args:
kwargs (dict): key-value arguments used to initialize TextTestRunner.
Commonly used arguments:
resultclass (class): HtmlTestResult or TextTestResult
failfast (bool): False/True, stop the test run on the first error or failure.
dot_env_path (str): .env file path.
http_client_session (instance): requests.Session(), or locust.client.Session() instance.
Attributes:
project_mapping (dict): save project loaded api/testcases, environments and debugtalk.py module.
{
"debugtalk": {
"variables": {},
"functions": {}
},
"env": {},
"def-api": {},
"def-testcase": {}
}
"""
self.kwargs = kwargs
self.http_client_session = self.kwargs.pop("http_client_session", None)
self.__loader()
def __loader(self):
""" load project dependent files, including api/testcase definitions,
environment variables and builtin module.
"""
loader.reset_loader()
# load .env
dot_env_path = self.kwargs.pop("dot_env_path", None)
loader.load_dot_env_file(dot_env_path)
# load api/testcase definition and debugtalk.py module
project_folder_path = os.path.join(os.getcwd(), "tests") # TODO: remove tests
loader.load_project_tests(project_folder_path)
self.project_mapping = loader.project_mapping
utils.set_os_environ(self.project_mapping["env"])
def __load_testcases(self, path_or_testcases):
""" load testcases, extend and merge with api/testcase definitions.
Args:
path_or_testcases (str/dict/list): YAML/JSON testcase file path or testcase list
path (str): testcase file/folder path
testcases (dict/list): testcase dict or list of testcases
Returns:
list: valid testcases list.
[
# testcase data structure
{
"config": {
"name": "desc1",
"path": "",
"variables": [], # optional
"request": {} # optional
},
"teststeps": [
# teststep data structure
{
'name': 'test step desc2',
'variables': [], # optional
'extract': [], # optional
'validate': [],
'request': {},
'function_meta': {}
},
teststep2 # another teststep dict
]
},
{} # another testcase dict
]
"""
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)
if not testcases:
raise exceptions.TestcaseNotFound
if isinstance(testcases, dict):
testcases = [testcases]
return testcases
def __parse_testcases(self, testcases, variables_mapping=None):
""" parse testcases configs, including variables/parameters/name/request.
Args:
testcases (list): testcase list, with config unparsed.
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 = []
for testcase in testcases:
config = testcase.setdefault("config", {})
# parse config parameters
config_parameters = config.pop("parameters", [])
cartesian_product_parameters_list = parser.parse_parameters(
config_parameters,
self.project_mapping["debugtalk"]["variables"],
self.project_mapping["debugtalk"]["functions"]
) or [{}]
for parameter_mapping in cartesian_product_parameters_list:
# parse config variables
raw_config_variables = config.get("variables", [])
parsed_config_variables = parser.parse_data(
raw_config_variables,
self.project_mapping["debugtalk"]["variables"],
self.project_mapping["debugtalk"]["functions"]
)
# priority: passed in > debugtalk.py > parameters > variables
# override variables mapping with parameters mapping
config_variables = utils.override_mapping_list(
parsed_config_variables, parameter_mapping)
# merge debugtalk.py module variables
config_variables.update(self.project_mapping["debugtalk"]["variables"])
# override variables mapping with passed in variables_mapping
config_variables = utils.override_mapping_list(
config_variables, variables_mapping)
testcase["config"]["variables"] = config_variables
# parse config name
testcase["config"]["name"] = parser.parse_data(
testcase["config"].get("name", ""),
config_variables,
self.project_mapping["debugtalk"]["functions"]
)
# parse config request
testcase["config"]["request"] = parser.parse_data(
testcase["config"].get("request", {}),
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
def __initialize(self, testcases):
""" initialize test runner with parsed testcases.
Args:
testcases (list): testcases list
Returns:
tuple: (unittest.TextTestRunner(), unittest.TestSuite())
"""
self.kwargs.setdefault("resultclass", report.HtmlTestResult)
unittest_runner = unittest.TextTestRunner(**self.kwargs)
testcases_list = []
loader = unittest.TestLoader()
loaded_testcases = []
for testcase in testcases:
config = testcase.get("config", {})
test_runner = runner.Runner(config, self.http_client_session)
TestSequense = type('TestSequense', (unittest.TestCase,), {})
teststeps = testcase.get("teststeps", [])
for index, teststep_dict in enumerate(teststeps):
for times_index in range(int(teststep_dict.get("times", 1))):
# 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 = utils.add_teststep(test_runner, teststep_dict)
setattr(TestSequense, test_method_name, test_method)
loaded_testcase = loader.loadTestsFromTestCase(TestSequense)
setattr(loaded_testcase, "config", config)
setattr(loaded_testcase, "runner", test_runner)
loaded_testcases.append(loaded_testcase)
test_suite = unittest.TestSuite(loaded_testcases)
return (unittest_runner, test_suite)
def run(self, path_or_testcases, mapping=None):
""" start to run test with variables mapping.
Args:
path_or_testcases (str/list/dict): YAML/JSON testcase file path or testcase list
path: path could be in several type
- absolute/relative file path
- absolute/relative folder path
- list/set container with file(s) and/or folder(s)
testcases: testcase dict or list of testcases
- (dict) testset_dict
- (list) list of testset_dict
[
testset_dict_1,
testset_dict_2
]
mapping (dict): if mapping specified, it will override variables in config block.
Returns:
instance: HttpRunner() instance
"""
# parser
testcases_list = self.__load_testcases(path_or_testcases)
parsed_testcases_list = self.__parse_testcases(testcases_list)
# initialize
unittest_runner, test_suite = self.__initialize(parsed_testcases_list)
# aggregate
self.summary = {
"success": True,
"stat": {},
"time": {},
"platform": report.get_platform(),
"details": []
}
# execution
for testcase in test_suite:
testcase_name = testcase.config.get("name")
logger.log_info("Start to run testcase: {}".format(testcase_name))
result = unittest_runner.run(testcase)
testcase_summary = report.get_summary(result)
self.summary["success"] &= testcase_summary["success"]
testcase_summary["name"] = testcase_name
testcase_summary["base_url"] = testcase.config.get("request", {}).get("base_url", "")
in_out = utils.get_testcase_io(testcase)
utils.print_io(in_out)
testcase_summary["in_out"] = in_out
report.aggregate_stat(self.summary["stat"], testcase_summary["stat"])
report.aggregate_stat(self.summary["time"], testcase_summary["time"])
self.summary["details"].append(testcase_summary)
return self
def gen_html_report(self, html_report_name=None, html_report_template=None):
""" generate html report and return report path.
Args:
html_report_name (str): output html report file name
html_report_template (str): report template file path, template should be in Jinja2 format
Returns:
str: generated html report path
"""
return report.render_html_report(
self.summary,
html_report_name,
html_report_template
)
class LocustRunner(object):
def __init__(self, locust_client):
self.runner = HttpRunner(http_client_session=locust_client)
def run(self, path):
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"),
name=test.testcase_dict.get("request", {}).get("url"),
response_time=0,
exception=ex
)

View File

@@ -9,7 +9,7 @@ import unittest
from httprunner import logger from httprunner import logger
from httprunner.__about__ import __description__, __version__ from httprunner.__about__ import __description__, __version__
from httprunner.compat import is_py2 from httprunner.compat import is_py2
from httprunner.task import HttpRunner from httprunner.api import HttpRunner
from httprunner.utils import (create_scaffold, get_python2_retire_msg, from httprunner.utils import (create_scaffold, get_python2_retire_msg,
prettify_json_file, validate_json_file) prettify_json_file, validate_json_file)

View File

@@ -1,447 +1,145 @@
# encoding: utf-8 # encoding: utf-8
import copy import copy
import os
import random
import re
import sys
from httprunner import built_in, exceptions, loader, logger, parser, utils from httprunner import exceptions, logger, parser, utils
from httprunner.compat import OrderedDict, basestring, builtin_str, str from httprunner.compat import OrderedDict
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()}"
testset_path (str): testset file path, used for locating csv file and debugtalk.py
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)
"""
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("-")
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))
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
]
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
class Context(object): class Context(object):
""" Manages context functions and variables. """ Manages context functions and variables.
context has two levels, testset and testcase. context has two levels, testcase and teststep.
""" """
def __init__(self): def __init__(self, variables=None, functions=None):
self.testset_shared_variables_mapping = OrderedDict() """ init Context with testcase variables and functions.
self.testcase_variables_mapping = OrderedDict() """
self.testcase_parser = TestcaseParser() # 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.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, if level == "testcase":
testcase level context initializes when each testcase starts. # 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)
if level == "testset":
self.testset_functions_config = {}
self.testset_request_config = {}
self.testset_shared_variables_mapping = OrderedDict()
# testcase config shall inherit from testset configs, # teststep level context, will be altered in each teststep.
# but can not change testset configs, that's why we use copy.deepcopy here. # teststep config shall inherit from testcase configs,
self.testcase_functions_config = copy.deepcopy(self.testset_functions_config) # but can not change testcase configs, that's why we use copy.deepcopy here.
self.testcase_variables_mapping = copy.deepcopy(self.testset_shared_variables_mapping) self.teststep_variables_mapping = copy.deepcopy(self.testcase_runtime_variables_mapping)
self.testcase_parser.bind_functions(self.testcase_functions_config) def update_context_variables(self, variables, level):
self.testcase_parser.update_binded_variables(self.testcase_variables_mapping) """ update context variables, with level specified.
if level == "testset": Args:
self.import_module_items(built_in) 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): if isinstance(variables, list):
variables = utils.convert_to_order_dict(variables) variables = utils.convert_mappinglist_to_orderdict(variables)
for variable_name, value in variables.items(): for variable_name, variable_value in variables.items():
variable_eval_value = self.eval_content(value) variable_eval_value = self.eval_content(variable_value)
if level == "testset": if level == "testcase":
self.testset_shared_variables_mapping[variable_name] = variable_eval_value self.testcase_runtime_variables_mapping[variable_name] = variable_eval_value
self.bind_testcase_variable(variable_name, variable_eval_value) self.update_teststep_variables_mapping(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)
def eval_content(self, content): def eval_content(self, content):
""" evaluate content recursively, take effect on each variable and function in 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. 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": for variable_name, variable_value in variables.items():
request_dict = self.eval_content( self.testcase_runtime_variables_mapping[variable_name] = variable_value
request_dict 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( def __eval_check_item(self, validator, resp_obj):
copy.deepcopy(self.testset_request_config), """ evaluate check item in validator.
request_dict
)
parsed_request = self.eval_content(
testcase_request_config
)
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 = validator["check"]
# check_item should only be the following 5 formats: # check_item should only be the following 5 formats:
@@ -470,12 +168,22 @@ class Context(object):
validator["check_result"] = "unchecked" validator["check_result"] = "unchecked"
return validator return validator
def do_validation(self, validator_dict): def _do_validation(self, validator_dict):
""" validate with functions """ 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 # TODO: move comparator uniform to init_test_suites
comparator = utils.get_uniform_comparator(validator_dict["comparator"]) 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: if not validate_func:
raise exceptions.FunctionNotFound("comparator not found: {}".format(comparator)) raise exceptions.FunctionNotFound("comparator not found: {}".format(comparator))
@@ -516,26 +224,28 @@ class Context(object):
def validate(self, validators, resp_obj): def validate(self, validators, resp_obj):
""" make validations """ make validations
""" """
evaluated_validators = []
if not validators: if not validators:
return return evaluated_validators
logger.log_info("start to validate.") logger.log_info("start to validate.")
self.evaluated_validators = []
validate_pass = True validate_pass = True
for validator in validators: for validator in validators:
# evaluate validators with context variable mapping. # evaluate validators with context variable mapping.
evaluated_validator = self.eval_check_item( evaluated_validator = self.__eval_check_item(
parser.parse_validator(validator), parser.parse_validator(validator),
resp_obj resp_obj
) )
try: try:
self.do_validation(evaluated_validator) self._do_validation(evaluated_validator)
except exceptions.ValidationFailure: except exceptions.ValidationFailure:
validate_pass = False validate_pass = False
self.evaluated_validators.append(evaluated_validator) evaluated_validators.append(evaluated_validator)
if not validate_pass: if not validate_pass:
raise exceptions.ValidationFailure raise exceptions.ValidationFailure
return evaluated_validators

View File

@@ -6,12 +6,14 @@ import json
import os import os
import yaml import yaml
from httprunner import exceptions, logger, parser, validator from httprunner import built_in, exceptions, logger, parser, validator
from httprunner.compat import OrderedDict from httprunner.compat import OrderedDict
project_mapping = { project_mapping = {
"debugtalk": {}, "debugtalk": {
"variables": {},
"functions": {}
},
"env": {}, "env": {},
"def-api": {}, "def-api": {},
"def-testcase": {} "def-testcase": {}
@@ -296,8 +298,15 @@ def load_python_module(module):
return debugtalk_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): def load_debugtalk_module(start_path=None):
""" load debugtalk.py module. """ load project debugtalk.py module and merge with builtin module.
Args: Args:
start_path (str, optional): start locating path, maybe file path or directory path. start_path (str, optional): start locating path, maybe file path or directory path.
@@ -305,7 +314,6 @@ def load_debugtalk_module(start_path=None):
Returns: Returns:
dict: variables and functions mapping for debugtalk.py dict: variables and functions mapping for debugtalk.py
{ {
"variables": {}, "variables": {},
"functions": {} "functions": {}
@@ -318,16 +326,15 @@ def load_debugtalk_module(start_path=None):
module_path = locate_file(start_path, "debugtalk.py") module_path = locate_file(start_path, "debugtalk.py")
module_name = convert_module_name(module_path) module_name = convert_module_name(module_path)
except exceptions.FileNotFound: except exceptions.FileNotFound:
return { return
"variables": {},
"functions": {}
}
# load debugtalk.py module
imported_module = importlib.import_module(module_name) imported_module = importlib.import_module(module_name)
loaded_module = load_python_module(imported_module) debugtalk_module = load_python_module(imported_module)
project_mapping["debugtalk"] = loaded_module # override built_in module with debugtalk.py module
return loaded_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): def get_module_item(module_mapping, item_type, item_name):
@@ -501,7 +508,7 @@ def _get_block_by_name(ref_call, ref_type):
args_mapping[item] = call_args[index] args_mapping[item] = call_args[index]
if args_mapping: if args_mapping:
block = parser.parse_data(block, args_mapping) block = parser.substitute_variables(block, args_mapping)
return block return block
@@ -880,28 +887,33 @@ def load_test_folder(test_folder_path=None):
return test_definition_mapping return test_definition_mapping
def load_project_tests(folder_path=None): def reset_loader():
""" load api, testcases and debugtalk.py module. """ reset project mapping.
"""
project_mapping["debugtalk"] = {
"variables": {},
"functions": {}
}
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 builtin module.
Args: Args:
folder_path (str): folder path. folder_path (str): folder path.
If not set, defautls to current working directory.
Returns:
dict: project tests mapping.
""" """
folder_path = folder_path or os.getcwd() load_builtin_module()
load_debugtalk_module(folder_path)
load_api_folder(os.path.join(folder_path, "api")) load_api_folder(os.path.join(folder_path, "api"))
load_test_folder(os.path.join(folder_path, "suite")) load_test_folder(os.path.join(folder_path, "suite"))
return project_mapping
def load_testcases(path): def load_testcases(path):
""" load testcases from file path """ load testcases from file path, extend and merge with api/testcase definitions.
Args: Args:
path (str): testcase file/foler path. path (str): testcase file/foler path.
@@ -936,11 +948,14 @@ def load_testcases(path):
return testcases_cache_mapping[path] return testcases_cache_mapping[path]
if os.path.isdir(path): if os.path.isdir(path):
load_debugtalk_module(path)
files_list = load_folder_files(path) files_list = load_folder_files(path)
testcases_list = load_testcases(files_list) testcases_list = load_testcases(files_list)
elif os.path.isfile(path): elif os.path.isfile(path):
try: try:
dir_path = os.path.dirname(path)
load_debugtalk_module(dir_path)
testcase = _load_test_file(path) testcase = _load_test_file(path)
if testcase["teststeps"]: if testcase["teststeps"]:
testcases_list = [testcase] testcases_list = [testcase]

View File

@@ -4,12 +4,12 @@ import ast
import os import os
import re import re
from httprunner import exceptions from httprunner import exceptions, utils
from httprunner.compat import builtin_str, numeric_types, str from httprunner.compat import basestring, builtin_str, numeric_types, str
variable_regexp = r"\$([\w_]+)" variable_regexp = r"\$([\w_]+)"
function_regexp = r"\$\{([\w_]+\([\$\w\.\-_ =,]*\))\}" function_regexp = r"\$\{([\w_]+\([\$\w\.\-/_ =,]*\))\}"
function_regexp_compile = re.compile(r"^([\w_]+)\(([\$\w\.\-_ =,]*)\)$") function_regexp_compile = re.compile(r"^([\w_]+)\(([\$\w\.\-/_ =,]*)\)$")
def parse_string_value(str_value): def parse_string_value(str_value):
@@ -200,53 +200,325 @@ def parse_validator(validator):
} }
def parse_data(content, mapping): def substitute_variables(content, variables_mapping):
""" substitute variables in content with mapping """ substitute variables in content with variables_mapping
e.g.
@params
content = {
'request': {
'url': '/api/users/$uid',
'headers': {'token': '$token'}
}
}
mapping = {"$uid": 1000}
@return
{
'request': {
'url': '/api/users/1000',
'headers': {'token': '$token'}
}
}
"""
# TODO: refactor type check
# TODO: combine this with TestcaseParser
if content is None or isinstance(content, (numeric_types, bool, type)):
return content
Args:
content (str/dict/list/numeric/bool/type): content to be substituted.
variables_mapping (dict): variables mapping.
Returns:
substituted content.
Examples:
>>> content = {
'request': {
'url': '/api/users/$uid',
'headers': {'token': '$token'}
}
}
>>> variables_mapping = {"$uid": 1000}
>>> substitute_variables(content, variables_mapping)
{
'request': {
'url': '/api/users/1000',
'headers': {'token': '$token'}
}
}
"""
if isinstance(content, (list, set, tuple)): if isinstance(content, (list, set, tuple)):
return [ return [
parse_data(item, mapping) substitute_variables(item, variables_mapping)
for item in content for item in content
] ]
if isinstance(content, dict): if isinstance(content, dict):
substituted_data = {} substituted_data = {}
for key, value in content.items(): for key, value in content.items():
eval_key = parse_data(key, mapping) eval_key = substitute_variables(key, variables_mapping)
eval_value = parse_data(value, mapping) eval_value = substitute_variables(value, variables_mapping)
substituted_data[eval_key] = eval_value substituted_data[eval_key] = eval_value
return substituted_data return substituted_data
# content is in string format here if isinstance(content, basestring):
for var, value in mapping.items(): # content is in string format here
if content == var: for var, value in variables_mapping.items():
# content is a variable if content == var:
content = value # content is a variable
else: content = value
if not isinstance(value, str): else:
value = builtin_str(value) if not isinstance(value, str):
content = content.replace(var, value) value = builtin_str(value)
content = content.replace(var, value)
return content
def parse_parameters(parameters, variables_mapping, functions_mapping):
""" 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()}"
variables_mapping (dict): variables mapping loaded from debugtalk.py
functions_mapping (dict): functions mapping loaded from debugtalk.py
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)
"""
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]
# ["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 = parse_data(parameter_content, variables_mapping, functions_mapping)
# 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
]
parsed_parameters_list.append(parameter_content_list)
return utils.gen_cartesian_product(*parsed_parameters_list)
###############################################################################
## parse content with variables and functions mapping
###############################################################################
def get_mapping_variable(variable_name, variables_mapping):
""" get variable from variables_mapping.
Args:
variable_name (str): variable name
variables_mapping (dict): variables mapping
Returns:
mapping variable value.
Raises:
exceptions.VariableNotFound: variable is not found.
"""
try:
return variables_mapping[variable_name]
except KeyError:
raise exceptions.VariableNotFound("{} is not found.".format(variable_name))
def get_mapping_function(function_name, functions_mapping):
""" get function from functions_mapping,
if not found, then try to check if builtin function.
Args:
variable_name (str): variable name
variables_mapping (dict): variables mapping
Returns:
mapping function object.
Raises:
exceptions.FunctionNotFound: function is neither defined in debugtalk.py nor builtin.
"""
if function_name in functions_mapping:
return functions_mapping[function_name]
try:
# check if builtin functions
item_func = eval(function_name)
if callable(item_func):
# is builtin function
return item_func
except (NameError, TypeError):
# is not builtin function
raise exceptions.FunctionNotFound("{} is not found.".format(function_name))
def parse_string_functions(content, variables_mapping, functions_mapping):
""" parse string content with functions mapping.
Args:
content (str): string content to be parsed.
variables_mapping (dict): variables mapping.
functions_mapping (dict): functions mapping.
Returns:
str: parsed string content.
Examples:
>>> content = "abc${add_one(3)}def"
>>> functions_mapping = {"add_one": lambda x: x + 1}
>>> parse_string_functions(content, functions_mapping)
"abc4def"
"""
functions_list = extract_functions(content)
for func_content in functions_list:
function_meta = parse_function(func_content)
func_name = function_meta["func_name"]
args = function_meta.get("args", [])
kwargs = function_meta.get("kwargs", {})
args = parse_data(args, variables_mapping, functions_mapping)
kwargs = parse_data(kwargs, variables_mapping, functions_mapping)
if func_name in ["parameterize", "P"]:
from httprunner import loader
eval_value = loader.load_csv_file(*args, **kwargs)
else:
func = get_mapping_function(func_name, functions_mapping)
eval_value = func(*args, **kwargs)
func_content = "${" + func_content + "}"
if func_content == content:
# content is a function, e.g. "${add_one(3)}"
content = eval_value
else:
# content contains one or many functions, e.g. "abc${add_one(3)}def"
content = content.replace(
func_content,
str(eval_value), 1
)
return content
def parse_string_variables(content, variables_mapping):
""" parse string content with variables mapping.
Args:
content (str): string content to be parsed.
variables_mapping (dict): variables mapping.
Returns:
str: parsed string content.
Examples:
>>> content = "/api/users/$uid"
>>> variables_mapping = {"$uid": 1000}
>>> parse_string_variables(content, variables_mapping)
"/api/users/1000"
"""
variables_list = extract_variables(content)
for variable_name in variables_list:
variable_value = get_mapping_variable(variable_name, variables_mapping)
# TODO: replace variable label from $var to {{var}}
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 parse_data(content, variables_mapping=None, functions_mapping=None):
""" parse content with variables mapping
Args:
content (str/dict/list/numeric/bool/type): content to be parsed
variables_mapping (dict): variables mapping.
functions_mapping (dict): functions mapping.
Returns:
parsed content.
Examples:
>>> content = {
'request': {
'url': '/api/users/$uid',
'headers': {'token': '$token'}
}
}
>>> variables_mapping = {"uid": 1000, "token": "abcdef"}
>>> parse_data(content, variables_mapping)
{
'request': {
'url': '/api/users/1000',
'headers': {'token': 'abcdef'}
}
}
"""
# TODO: refactor type check
if content is None or isinstance(content, (numeric_types, bool, type)):
return content
if isinstance(content, (list, set, tuple)):
return [
parse_data(item, variables_mapping, functions_mapping)
for item in content
]
if isinstance(content, dict):
parsed_content = {}
for key, value in content.items():
parsed_key = parse_data(key, variables_mapping, functions_mapping)
parsed_value = parse_data(value, variables_mapping, functions_mapping)
parsed_content[parsed_key] = parsed_value
return parsed_content
if isinstance(content, basestring):
# content is in string format here
variables_mapping = variables_mapping or {}
functions_mapping = functions_mapping or {}
content = content.strip()
# replace functions with evaluated value
# Notice: _eval_content_functions must be called before _eval_content_variables
content = parse_string_functions(content, variables_mapping, functions_mapping)
# replace variables with binding value
content = parse_string_variables(content, variables_mapping)
return content return content

View File

@@ -6,14 +6,13 @@ import platform
import time import time
import unittest import unittest
from base64 import b64encode from base64 import b64encode
from collections import Iterable, OrderedDict from collections import Iterable
from datetime import datetime from datetime import datetime
from httprunner import logger from httprunner import logger
from httprunner.__about__ import __version__ from httprunner.__about__ import __version__
from httprunner.compat import basestring, bytes, json, numeric_types from httprunner.compat import basestring, bytes, json, numeric_types
from jinja2 import Template, escape from jinja2 import Template, escape
from requests.structures import CaseInsensitiveDict
def get_platform(): def get_platform():
@@ -26,6 +25,7 @@ def get_platform():
"platform": platform.platform() "platform": platform.platform()
} }
def get_summary(result): def get_summary(result):
""" get summary from test result """ get summary from test result
""" """
@@ -58,6 +58,25 @@ def get_summary(result):
return summary return summary
def aggregate_stat(origin_stat, new_stat):
""" aggregate new_stat to origin_stat.
Args:
origin_stat (dict): origin stat dict, will be updated with new_stat dict.
new_stat (dict): new stat dict.
"""
for key in new_stat:
if key not in origin_stat:
origin_stat[key] = new_stat[key]
elif key == "start_at":
# start datetime
origin_stat[key] = min(origin_stat[key], new_stat[key])
else:
origin_stat[key] += new_stat[key]
def render_html_report(summary, html_report_name=None, html_report_template=None): def render_html_report(summary, html_report_name=None, html_report_template=None):
""" render html report with specified report name and template """ render html report with specified report name and template
if html_report_name is not specified, use current datetime if html_report_name is not specified, use current datetime
@@ -112,6 +131,7 @@ def render_html_report(summary, html_report_name=None, html_report_template=None
return report_path return report_path
def stringify_data(meta_data, request_or_response): def stringify_data(meta_data, request_or_response):
""" """
meta_data = { meta_data = {
@@ -151,6 +171,7 @@ def stringify_data(meta_data, request_or_response):
meta_data[request_or_response][key] = value meta_data[request_or_response][key] = value
class HtmlTestResult(unittest.TextTestResult): class HtmlTestResult(unittest.TextTestResult):
"""A html result class that can generate formatted html results. """A html result class that can generate formatted html results.
@@ -161,12 +182,16 @@ class HtmlTestResult(unittest.TextTestResult):
self.records = [] self.records = []
def _record_test(self, test, status, attachment=''): def _record_test(self, test, status, attachment=''):
self.records.append({ data = {
'name': test.shortDescription(), 'name': test.shortDescription(),
'status': status, 'status': status,
'attachment': attachment, 'attachment': attachment,
"meta_data": test.meta_data "meta_data": {}
}) }
if hasattr(test, "meta_data"):
data["meta_data"] = test.meta_data
self.records.append(data)
def startTestRun(self): def startTestRun(self):
self.start_at = time.time() self.start_at = time.time()

View File

@@ -221,7 +221,7 @@ class ResponseObject(object):
logger.log_info("start to extract from response object.") logger.log_info("start to extract from response object.")
extracted_variables_mapping = OrderedDict() extracted_variables_mapping = OrderedDict()
extract_binds_order_dict = utils.convert_to_order_dict(extractors) extract_binds_order_dict = utils.convert_mappinglist_to_orderdict(extractors)
for key, field in extract_binds_order_dict.items(): for key, field in extract_binds_order_dict.items():
extracted_variables_mapping[key] = self.extract_field(field) extracted_variables_mapping[key] = self.extract_field(field)

View File

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

View File

@@ -1,334 +0,0 @@
# encoding: utf-8
import copy
import sys
import unittest
from httprunner import (context, exceptions, loader, logger, runner, utils,
validator)
from httprunner.compat import is_py3
from httprunner.report import (HtmlTestResult, get_platform, get_summary,
render_html_report)
class TestCase(unittest.TestCase):
""" create a testcase.
"""
def __init__(self, test_runner, testcase_dict):
super(TestCase, self).__init__()
self.test_runner = test_runner
self.testcase_dict = copy.copy(testcase_dict)
def runTest(self):
""" run testcase and check result.
"""
try:
self.test_runner.run_test(self.testcase_dict)
except exceptions.MyBaseFailure as ex:
self.fail(repr(ex))
finally:
if hasattr(self.test_runner.http_client_session, "meta_data"):
self.meta_data = self.test_runner.http_client_session.meta_data
self.meta_data["validators"] = self.test_runner.context.evaluated_validators
self.test_runner.http_client_session.init_meta_data()
class TestSuite(unittest.TestSuite):
""" create test suite with a testcase, it may include one or several teststeps.
each suite should initialize a separate Runner() with testcase config.
Args:
testcase (dict): testcase dict
{
"config": {
"name": "testcase description",
"parameters": {},
"variables": [],
"request": {},
"output": []
},
"teststeps": [
{
"name": "teststep1 description",
"parameters": {},
"variables": [], # optional, override
"request": {},
"extract": {}, # optional
"validate": {} # optional
},
teststep2
]
}
variables_mapping (dict): passed in variables mapping, it will override variables in config block.
"""
def __init__(self, testcase, variables_mapping=None, http_client_session=None):
super(TestSuite, self).__init__()
self.test_runner_list = []
self.config = testcase.get("config", {})
self.output_variables_list = self.config.get("output", [])
self.testset_file_path = self.config.get("path")
config_dict_parameters = self.config.get("parameters", [])
config_dict_variables = self.config.get("variables", [])
variables_mapping = variables_mapping or {}
config_dict_variables = utils.override_variables_binds(config_dict_variables, variables_mapping)
config_parametered_variables_list = self._get_parametered_variables(
config_dict_variables,
config_dict_parameters
)
self.testcase_parser = context.TestcaseParser()
teststeps = testcase.get("teststeps", [])
for config_variables in config_parametered_variables_list:
# config level
self.config["variables"] = config_variables
test_runner = runner.Runner(self.config, http_client_session)
for teststep_dict in teststeps:
teststep_dict = copy.copy(teststep_dict)
# testcase level
testcase_parametered_variables_list = self._get_parametered_variables(
teststep_dict.get("variables", []),
teststep_dict.get("parameters", [])
)
for testcase_variables in testcase_parametered_variables_list:
teststep_dict["variables"] = testcase_variables
# eval testcase name with bind variables
variables = utils.override_variables_binds(
config_variables,
testcase_variables
)
self.testcase_parser.update_binded_variables(variables)
try:
testcase_name = self.testcase_parser.eval_content_with_bindings(teststep_dict["name"])
except (AssertionError, exceptions.ParamsError):
logger.log_warning("failed to eval testcase name: {}".format(teststep_dict["name"]))
testcase_name = teststep_dict["name"]
self.test_runner_list.append((test_runner, variables))
self._add_test_to_suite(testcase_name, test_runner, teststep_dict)
def _get_parametered_variables(self, variables, parameters):
""" parameterize varaibles with parameters
"""
cartesian_product_parameters = context.parse_parameters(
parameters,
self.testset_file_path
) or [{}]
parametered_variables_list = []
for parameter_mapping in cartesian_product_parameters:
parameter_mapping = parameter_mapping or {}
variables = utils.override_variables_binds(
variables,
parameter_mapping
)
parametered_variables_list.append(variables)
return parametered_variables_list
def _add_test_to_suite(self, testcase_name, test_runner, testcase_dict):
if is_py3:
TestCase.runTest.__doc__ = testcase_name
else:
TestCase.runTest.__func__.__doc__ = testcase_name
test = TestCase(test_runner, testcase_dict)
[self.addTest(test) for _ in range(int(testcase_dict.get("times", 1)))]
@property
def output(self):
outputs = []
for test_runner, variables in self.test_runner_list:
out = test_runner.extract_output(self.output_variables_list)
if not out:
continue
in_out = {
"in": dict(variables),
"out": out
}
if in_out not in outputs:
outputs.append(in_out)
return outputs
def init_test_suites(path_or_testcases, mapping=None, http_client_session=None):
""" initialize TestSuite list with testcase path or testcase(s).
Args:
path_or_testcases (str/dict/list): testcase file path or testcase dict or testcases list
testcase_dict
or
[
testcase_dict_1,
testcase_dict_2,
{
"config": {},
"teststeps": [teststep11, teststep12]
}
]
mapping (dict): passed in variables mapping, it will override variables in config block.
http_client_session (instance): requests.Session(), or locusts.client.Session() instance.
Returns:
list: TestSuite() instance list.
"""
if validator.is_testcases(path_or_testcases):
testcases = path_or_testcases
else:
testcases = loader.load_testcases(path_or_testcases)
# TODO: move comparator uniform here
mapping = mapping or {}
if not testcases:
raise exceptions.TestcaseNotFound
if isinstance(testcases, dict):
testcases = [testcases]
test_suite_list = []
for testcase in testcases:
test_suite = TestSuite(testcase, mapping, http_client_session)
test_suite_list.append(test_suite)
return test_suite_list
class HttpRunner(object):
def __init__(self, **kwargs):
""" initialize HttpRunner.
Args:
kwargs (dict): key-value arguments used to initialize TextTestRunner.
Commonly used arguments:
resultclass (class): HtmlTestResult or TextTestResult
failfast (bool): False/True, stop the test run on the first error or failure.
dot_env_path (str): .env file path.
Attributes:
project_mapping (dict): save project loaded api/testcases, environments and debugtalk.py module.
"""
dot_env_path = kwargs.pop("dot_env_path", None)
loader.load_dot_env_file(dot_env_path)
loader.load_project_tests("tests") # TODO: remove tests
self.project_mapping = loader.project_mapping
utils.set_os_environ(self.project_mapping["env"])
kwargs.setdefault("resultclass", HtmlTestResult)
self.runner = unittest.TextTestRunner(**kwargs)
def run(self, path_or_testcases, mapping=None):
""" start to run test with varaibles mapping.
Args:
path_or_testcases (str/list/dict): YAML/JSON testcase file path or testcase list
path: path could be in several type
- absolute/relative file path
- absolute/relative folder path
- list/set container with file(s) and/or folder(s)
testcases: testcase dict or list of testcases
- (dict) testset_dict
- (list) list of testset_dict
[
testset_dict_1,
testset_dict_2
]
mapping (dict): if mapping specified, it will override variables in config block.
Returns:
instance: HttpRunner() instance
"""
try:
test_suite_list = init_test_suites(path_or_testcases, mapping)
except exceptions.TestcaseNotFound:
logger.log_error("Testcases not found in {}".format(path_or_testcases))
sys.exit(1)
self.summary = {
"success": True,
"stat": {},
"time": {},
"platform": get_platform(),
"details": []
}
def accumulate_stat(origin_stat, new_stat):
"""accumulate new_stat to origin_stat."""
for key in new_stat:
if key not in origin_stat:
origin_stat[key] = new_stat[key]
elif key == "start_at":
# start datetime
origin_stat[key] = min(origin_stat[key], new_stat[key])
else:
origin_stat[key] += new_stat[key]
for test_suite in test_suite_list:
result = self.runner.run(test_suite)
test_suite_summary = get_summary(result)
self.summary["success"] &= test_suite_summary["success"]
test_suite_summary["name"] = test_suite.config.get("name")
test_suite_summary["base_url"] = test_suite.config.get("request", {}).get("base_url", "")
test_suite_summary["output"] = test_suite.output
utils.print_output(test_suite_summary["output"])
accumulate_stat(self.summary["stat"], test_suite_summary["stat"])
accumulate_stat(self.summary["time"], test_suite_summary["time"])
self.summary["details"].append(test_suite_summary)
return self
def gen_html_report(self, html_report_name=None, html_report_template=None):
""" generate html report and return report path.
Args:
html_report_name (str): output html report file name
html_report_template (str): report template file path, template should be in Jinja2 format
Returns:
str: generated html report path
"""
return render_html_report(
self.summary,
html_report_name,
html_report_template
)
class LocustTask(object):
def __init__(self, path_or_testcases, locust_client, mapping=None):
self.test_suite_list = init_test_suites(path_or_testcases, mapping, locust_client)
def run(self):
for test_suite in self.test_suite_list:
for test in test_suite:
try:
test.runTest()
except exceptions.MyBaseError as ex:
from locust.events import request_failure
request_failure.fire(
request_type=test.testcase_dict.get("request", {}).get("method"),
name=test.testcase_dict.get("request", {}).get("url"),
response_time=0,
exception=ex
)

View File

@@ -1,15 +1,18 @@
#coding: utf-8 #coding: utf-8
import zmq import zmq
from locust import HttpLocust, TaskSet, task from locust import HttpLocust, TaskSet, task
from httprunner.task import LocustTask from httprunner import LocustRunner
class WebPageTasks(TaskSet): class WebPageTasks(TaskSet):
def on_start(self): def on_start(self):
self.test_runner = LocustTask(self.locust.file_path, self.client) self.test_runner = LocustRunner(self.client)
self.file_path = self.locust.file_path
@task @task
def test_specified_scenario(self): def test_specified_scenario(self):
self.test_runner.run() self.test_runner.run(self.file_path)
class WebPageUser(HttpLocust): class WebPageUser(HttpLocust):
host = "$HOST" host = "$HOST"

View File

@@ -204,12 +204,10 @@
<th>variables</th> <th>variables</th>
<th>output</th> <th>output</th>
</tr> </tr>
{% for in_out in test_suite_summary.output %}
<tr> <tr>
<td>{{in_out.in}}</td> <td>{{test_suite_summary.in_out.in}}</td>
<td>{{in_out.out}}</td> <td>{{test_suite_summary.in_out.out}}</td>
</tr> </tr>
{% endfor %}
</table> </table>
</div> </div>
</div> </div>

View File

@@ -158,41 +158,49 @@ def lower_config_dict_key(config_dict):
return config_dict return config_dict
def convert_to_order_dict(map_list): def convert_mappinglist_to_orderdict(mapping_list):
""" convert mapping in list to ordered dict """ convert mapping list to ordered dict
@param (list) map_list
[ Args:
{"a": 1}, mapping_list (list):
{"b": 2} [
] {"a": 1},
@return (OrderDict) {"b": 2}
OrderDict({ ]
"a": 1,
"b": 2 Returns:
}) OrderedDict: converted mapping in OrderedDict
OrderDict(
{
"a": 1,
"b": 2
}
)
""" """
ordered_dict = OrderedDict() ordered_dict = OrderedDict()
for map_dict in map_list: for map_dict in mapping_list:
ordered_dict.update(map_dict) ordered_dict.update(map_dict)
return ordered_dict return ordered_dict
def update_ordered_dict(ordered_dict, override_mapping): def update_ordered_dict(ordered_dict, override_mapping):
""" override ordered_dict with new mapping """ override ordered_dict with new mapping.
@param
(OrderDict) ordered_dict Args:
OrderDict({ ordered_dict (OrderDict): original ordered dict
"a": 1, override_mapping (dict): new variables mapping
"b": 2
}) Returns:
(dict) override_mapping OrderDict: new overrided variables mapping.
{"a": 3, "c": 4}
@return (OrderDict) Examples:
OrderDict({ >>> ordered_dict = OrderDict({"a": 1, "b": 2})
"a": 3, >>> override_mapping = {"a": 3, "c": 4}
"b": 2, >>> update_ordered_dict(ordered_dict, override_mapping)
"c": 4 OrderDict({"a": 3, "b": 2, "c": 4})
})
""" """
new_ordered_dict = copy.copy(ordered_dict) new_ordered_dict = copy.copy(ordered_dict)
for var, value in override_mapping.items(): for var, value in override_mapping.items():
@@ -200,11 +208,43 @@ def update_ordered_dict(ordered_dict, override_mapping):
return new_ordered_dict return new_ordered_dict
def override_variables_binds(variables, new_mapping):
""" convert variables in testcase to ordered mapping, with new_mapping overrided def override_mapping_list(variables, new_mapping):
""" override variables with new mapping.
Args:
variables (list): variables list
[
{"var_a": 1},
{"var_b": "world"}
]
new_mapping (dict): overrided variables mapping
{
"var_a": "hello"
}
Returns:
OrderedDict: overrided variables mapping.
Examples:
>>> variables = [
{"var_a": 1},
{"var_b": "world"}
]
>>> new_mapping = {
"var_a": "hello"
}
>>> override_mapping_list(variables, new_mapping)
OrderedDict(
{
"var_a": "hello",
"var_b": "world"
}
)
""" """
if isinstance(variables, list): if isinstance(variables, list):
variables_ordered_dict = convert_to_order_dict(variables) variables_ordered_dict = convert_mappinglist_to_orderdict(variables)
elif isinstance(variables, (OrderedDict, dict)): elif isinstance(variables, (OrderedDict, dict)):
variables_ordered_dict = variables variables_ordered_dict = variables
else: else:
@@ -215,18 +255,84 @@ def override_variables_binds(variables, new_mapping):
new_mapping new_mapping
) )
def print_output(outputs):
if not outputs: def add_teststep(test_runner, teststep_dict):
return """ add teststep to testcase.
"""
def test(self):
try:
test_runner.run_test(teststep_dict)
except exceptions.MyBaseFailure as ex:
self.fail(repr(ex))
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.evaluated_validators
test_runner.http_client_session.init_meta_data()
test.__doc__ = teststep_dict["name"]
return test
def get_testcase_io(testcase):
""" get testcase input(variables) and output.
Args:
testcase (unittest.suite.TestSuite): corresponding to one YAML/JSON file, it has been set two attributes:
config: parsed config block
runner: initialized runner.Runner() with config
Returns:
dict: input(variables) and output mapping.
"""
runner = testcase.runner
variables = testcase.config.get("variables", [])
output_list = testcase.config.get("output", [])
return {
"in": dict(variables),
"out": runner.extract_output(output_list)
}
def print_io(in_out):
""" print input(variables) and output.
Args:
in_out (dict): input(variables) and output mapping.
Examples:
>>> in_out = {
"in": {
"var_a": "hello",
"var_b": "world"
},
"out": {
"status_code": 500
}
}
>>> print_io(in_out)
================== Variables & Output ==================
Type | Variable : Value
------ | ---------------- : ---------------------------
Var | var_a : hello
Var | var_b : world
Out | status_code : 500
--------------------------------------------------------
"""
content_format = "{:<6} | {:<16} : {:<}\n"
content = "\n================== Variables & Output ==================\n" content = "\n================== Variables & Output ==================\n"
content += '{:<6} | {:<16} : {:<}\n'.format("Type", "Variable", "Value") content += content_format.format("Type", "Variable", "Value")
content += '{:<6} | {:<16} : {:<}\n'.format("-" * 6, "-" * 16, "-" * 27) content += content_format.format("-" * 6, "-" * 16, "-" * 27)
def prepare_content(var_type, in_out): def prepare_content(var_type, in_out):
content = "" content = ""
for variable, value in in_out.items(): for variable, value in in_out.items():
if isinstance(value, tuple):
continue
if is_py2: if is_py2:
if isinstance(variable, unicode): if isinstance(variable, unicode):
@@ -234,21 +340,17 @@ def print_output(outputs):
if isinstance(value, unicode): if isinstance(value, unicode):
value = value.encode("utf-8") value = value.encode("utf-8")
content += '{:<6} | {:<16} : {:<}\n'.format(var_type, variable, value) content += content_format.format(var_type, variable, value)
return content return content
for output in outputs: _in = in_out["in"]
_in = output["in"] _out = in_out["out"]
_out = output["out"]
if not _out: content += prepare_content("Var", _in)
continue content += "\n"
content += prepare_content("Out", _out)
content += prepare_content("Var", _in) content += "-" * 56 + "\n"
content += "\n"
content += prepare_content("Out", _out)
content += "-" * 56 + "\n"
logger.log_debug(content) logger.log_debug(content)
@@ -336,6 +438,7 @@ def validate_json_file(file_list):
print("OK") print("OK")
def prettify_json_file(file_list): def prettify_json_file(file_list):
""" prettify JSON testset format """ prettify JSON testset format
""" """
@@ -362,6 +465,7 @@ def prettify_json_file(file_list):
print("success: {}".format(outfile)) print("success: {}".format(outfile))
def get_python2_retire_msg(): def get_python2_retire_msg():
retire_day = datetime(2020, 1, 1) retire_day = datetime(2020, 1, 1)
today = datetime.now() today = datetime.now()

View File

@@ -12,20 +12,34 @@ def is_testcase(data_structure):
data_structure (dict): testcase should always be in the following data structure: data_structure (dict): testcase should always be in the following data structure:
{ {
"name": "desc1", "config": {
"config": {}, "name": "desc1",
"api": {}, "path": "",
"testcases": [testcase11, testcase12] "variables": [], # optional
"request": {} # optional
},
"teststeps": [
teststep1,
{ # teststep2
'name': 'test step desc2',
'variables': [], # optional
'extract': [], # optional
'validate': [],
'request': {},
'function_meta': {}
}
]
} }
Returns: Returns:
bool: True if data_structure is valid testcase, otherwise False. bool: True if data_structure is valid testcase, otherwise False.
""" """
# TODO: replace with JSON schema validation
if not isinstance(data_structure, dict): if not isinstance(data_structure, dict):
return False return False
if "name" not in data_structure or "teststeps" not in data_structure: if "teststeps" not in data_structure:
return False return False
if not isinstance(data_structure["teststeps"], list): if not isinstance(data_structure["teststeps"], list):

View File

@@ -8,6 +8,7 @@
variables: variables:
- device_sn: ${gen_random_string(15)} - device_sn: ${gen_random_string(15)}
- os_platform: 'ios' - os_platform: 'ios'
- app_version: 2.8.5
request: request:
base_url: $BASE_URL base_url: $BASE_URL
headers: headers:
@@ -18,21 +19,9 @@
- test: - test:
name: get token with $user_agent and $app_version name: get token with $user_agent and $app_version
parameters:
- app_version: ${gen_app_version()}
api: get_token($user_agent, $device_sn, $os_platform, $app_version) api: get_token($user_agent, $device_sn, $os_platform, $app_version)
extract: extract:
- token: content.token - token: content.token
validate: validate:
- "eq": ["status_code", 200] - "eq": ["status_code", 200]
- "len_eq": ["content.token", 16] - "len_eq": ["content.token", 16]
# - test:
# name: create user
# parameters:
# - user_id: [1001, 1002, 1003]
# - username-password: ${P(account.csv)}
# api: create_user($user_id, $username, $password, $token)
# validate:
# - {"check": "status_code", "expect": 201}
# - {"check": "content.success", "expect": true}

View File

@@ -76,7 +76,7 @@
method: GET method: GET
validate: validate:
- eq: ["status_code", 200] - eq: ["status_code", 200]
- eq: [cookies.name, "value"] # - eq: [cookies.name, "value"]
- test: - test:
name: post data name: post data

356
tests/test_api.py Normal file
View File

@@ -0,0 +1,356 @@
import os
import shutil
import time
from httprunner import HttpRunner, LocustRunner
from locust import HttpLocust
from tests.api_server import HTTPBIN_SERVER
from tests.base import ApiServerUnittest
class TestHttpRunner(ApiServerUnittest):
def setUp(self):
self.testset_path = "tests/data/demo_testset_cli.yml"
self.testcase_file_path_list = [
os.path.join(
os.getcwd(), 'tests/data/demo_testset_hardcode.yml'),
os.path.join(
os.getcwd(), 'tests/data/demo_testset_hardcode.json')
]
self.testcase = {
'name': 'testset description',
'config': {
'path': 'docs/data/demo-quickstart-2.yml',
'name': 'testset description',
'request': {
'base_url': '',
'headers': {'User-Agent': 'python-requests/2.18.4'}
},
'variables': [],
'output': ['token']
},
'api': {},
'teststeps': [
{
'name': '/api/get-token',
'request': {
'url': 'http://127.0.0.1:5000/api/get-token',
'method': 'POST',
'headers': {'Content-Type': 'application/json', 'app_version': '2.8.6', 'device_sn': 'FwgRiO7CNA50DSU', 'os_platform': 'ios', 'user_agent': 'iOS/10.3'},
'json': {'sign': '958a05393efef0ac7c0fb80a7eac45e24fd40c27'}
},
'extract': [
{'token': 'content.token'}
],
'validate': [
{'eq': ['status_code', 200]},
{'eq': ['headers.Content-Type', 'application/json']},
{'eq': ['content.success', True]}
]
},
{
'name': '/api/users/1000',
'request': {
'url': 'http://127.0.0.1:5000/api/users/1000',
'method': 'POST',
'headers': {'Content-Type': 'application/json', 'device_sn': 'FwgRiO7CNA50DSU','token': '$token'}, 'json': {'name': 'user1', 'password': '123456'}
},
'validate': [
{'eq': ['status_code', 201]},
{'eq': ['headers.Content-Type', 'application/json']},
{'eq': ['content.success', True]},
{'eq': ['content.msg', 'user created successfully.']}
]
}
]
}
self.reset_all()
def reset_all(self):
url = "%s/api/reset-all" % self.host
headers = self.get_authenticated_headers()
return self.api_client.get(url, headers=headers)
def test_text_run_times(self):
runner = HttpRunner().run(self.testset_path)
self.assertEqual(runner.summary["stat"]["testsRun"], 10)
def test_text_skip(self):
runner = HttpRunner().run(self.testset_path)
self.assertEqual(runner.summary["stat"]["skipped"], 4)
def test_html_report(self):
kwargs = {}
output_folder_name = os.path.basename(os.path.splitext(self.testset_path)[0])
runner = HttpRunner().run(self.testset_path)
summary = runner.summary
self.assertEqual(summary["stat"]["testsRun"], 10)
self.assertEqual(summary["stat"]["skipped"], 4)
runner.gen_html_report(html_report_name=output_folder_name)
report_save_dir = os.path.join(os.getcwd(), 'reports', output_folder_name)
self.assertGreater(len(os.listdir(report_save_dir)), 0)
shutil.rmtree(report_save_dir)
def test_run_testcases(self):
testcases = [self.testcase]
runner = HttpRunner().run(testcases)
summary = runner.summary
self.assertTrue(summary["success"])
self.assertEqual(summary["stat"]["testsRun"], 2)
self.assertIn("details", summary)
self.assertIn("records", summary["details"][0])
def test_run_testcase(self):
testcases = self.testcase
runner = HttpRunner().run(testcases)
summary = runner.summary
self.assertTrue(summary["success"])
self.assertEqual(summary["stat"]["testsRun"], 2)
self.assertIn("records", summary["details"][0])
def test_run_yaml_upload(self):
testset_path = "tests/httpbin/upload.yml"
runner = HttpRunner().run(testset_path)
summary = runner.summary
self.assertTrue(summary["success"])
self.assertEqual(summary["stat"]["testsRun"], 1)
self.assertIn("details", summary)
self.assertIn("records", summary["details"][0])
def test_run_post_data(self):
testcases = [
{
"name": "post data",
"teststeps": [
{
"name": "post data",
"request": {
"url": "{}/post".format(HTTPBIN_SERVER),
"method": "POST",
"headers": {
"Content-Type": "application/json"
},
"data": "abc"
},
"validate": [
{"eq": ["status_code", 200]}
]
}
]
}
]
runner = HttpRunner().run(testcases)
summary = runner.summary
self.assertTrue(summary["success"])
self.assertEqual(summary["stat"]["testsRun"], 1)
self.assertEqual(summary["details"][0]["records"][0]["meta_data"]["response"]["json"]["data"], "abc")
def test_html_report_repsonse_image(self):
testset_path = "tests/httpbin/load_image.yml"
runner = HttpRunner().run(testset_path)
summary = runner.summary
output_folder_name = os.path.basename(os.path.splitext(testset_path)[0])
report = runner.gen_html_report(html_report_name=output_folder_name)
self.assertTrue(os.path.isfile(report))
report_save_dir = os.path.join(os.getcwd(), 'reports', output_folder_name)
shutil.rmtree(report_save_dir)
def test_testcase_layer(self):
testcase_path = "tests/testcases/smoketest.yml"
runner = HttpRunner(failfast=True).run(testcase_path)
summary = runner.summary
self.assertTrue(summary["success"])
self.assertEqual(summary["stat"]["testsRun"], 8)
def test_run_httprunner_with_hooks(self):
testcase_file_path = os.path.join(
os.getcwd(), 'tests/httpbin/hooks.yml')
start_time = time.time()
runner = HttpRunner().run(testcase_file_path)
end_time = time.time()
summary = runner.summary
self.assertTrue(summary["success"])
self.assertLess(end_time - start_time, 10)
def test_run_httprunner_with_teardown_hooks_alter_response(self):
testcases = [
{
"config": {
"name": "test teardown hooks",
'path': 'tests/httpbin/hooks.yml',
},
"teststeps": [
{
"name": "test teardown hooks",
"request": {
"url": "{}/headers".format(HTTPBIN_SERVER),
"method": "GET",
"data": "abc"
},
"teardown_hooks": [
"${alter_response($response)}"
],
"validate": [
{"eq": ["status_code", 500]},
{"eq": ["headers.content-type", "html/text"]},
{"eq": ["json.headers.Host", "127.0.0.1:8888"]},
{"eq": ["content.headers.Host", "127.0.0.1:8888"]},
{"eq": ["text.headers.Host", "127.0.0.1:8888"]},
{"eq": ["new_attribute", "new_attribute_value"]},
{"eq": ["new_attribute_dict", {"key": 123}]},
{"eq": ["new_attribute_dict.key", 123]}
]
}
]
}
]
runner = HttpRunner().run(testcases)
summary = runner.summary
self.assertTrue(summary["success"])
def test_run_httprunner_with_teardown_hooks_not_exist_attribute(self):
testcases = [
{
"name": "test teardown hooks",
"config": {
'path': 'tests/httpbin/hooks.yml',
},
"teststeps": [
{
"name": "test teardown hooks",
"request": {
"url": "{}/headers".format(HTTPBIN_SERVER),
"method": "GET",
"data": "abc"
},
"teardown_hooks": [
"${alter_response($response)}"
],
"validate": [
{"eq": ["attribute_not_exist", "new_attribute"]}
]
}
]
}
]
runner = HttpRunner().run(testcases)
summary = runner.summary
self.assertFalse(summary["success"])
self.assertEqual(summary["stat"]["errors"], 1)
def test_run_httprunner_with_teardown_hooks_error(self):
testcases = [
{
"name": "test teardown hooks",
"config": {
'path': 'tests/httpbin/hooks.yml',
},
"teststeps": [
{
"name": "test teardown hooks",
"request": {
"url": "{}/headers".format(HTTPBIN_SERVER),
"method": "GET",
"data": "abc"
},
"teardown_hooks": [
"${alter_response_error($response)}"
]
}
]
}
]
runner = HttpRunner().run(testcases)
summary = runner.summary
self.assertFalse(summary["success"])
self.assertEqual(summary["stat"]["errors"], 1)
def test_run_testset_hardcode(self):
for testcase_file_path in self.testcase_file_path_list:
runner = HttpRunner().run(testcase_file_path)
self.assertTrue(runner.summary["success"])
def test_run_testsets_hardcode(self):
runner = HttpRunner().run(self.testcase_file_path_list)
summary = runner.summary
self.assertTrue(summary["success"])
self.assertEqual(summary["stat"]["testsRun"], 6)
self.assertEqual(summary["stat"]["successes"], 6)
def test_run_testset_template_variables(self):
testcase_file_path = os.path.join(
os.getcwd(), 'tests/data/demo_testset_variables.yml')
runner = HttpRunner().run(testcase_file_path)
summary = runner.summary
self.assertTrue(summary["success"])
def test_run_testset_template_import_functions(self):
testcase_file_path = os.path.join(
os.getcwd(), 'tests/data/demo_testset_functions.yml')
runner = HttpRunner().run(testcase_file_path)
summary = runner.summary
self.assertTrue(summary["success"])
def test_run_testset_layered(self):
testcase_file_path = os.path.join(
os.getcwd(), 'tests/data/demo_testset_layer.yml')
runner = HttpRunner().run(testcase_file_path)
summary = runner.summary
self.assertTrue(summary["success"])
self.assertEqual(len(summary["details"]), 1)
def test_run_testcase_output(self):
testcase_file_path = os.path.join(
os.getcwd(), 'tests/data/demo_testset_layer.yml')
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_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(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.assertGreater(len(summary["details"][0]["in_out"]["in"]), 7)
def test_run_testset_with_parameters(self):
testcase_file_path = os.path.join(
os.getcwd(), 'tests/data/demo_parameters.yml')
runner = HttpRunner().run(testcase_file_path)
summary = runner.summary
self.assertTrue(summary["success"])
self.assertEqual(len(summary["details"]), 3 * 2)
self.assertEqual(summary["stat"]["testsRun"], 3 * 2)
self.assertIn("in", summary["details"][0]["in_out"])
self.assertIn("out", summary["details"][0]["in_out"])
def test_loader(self):
hrunner = HttpRunner(dot_env_path="tests/data/test.env")
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.assertIn("get_token", hrunner.project_mapping["def-api"])
self.assertIn("setup_and_reset", hrunner.project_mapping["def-testcase"])
class TestLocustRunner(ApiServerUnittest):
def setUp(self):
WebPageUser = type('WebPageUser', (HttpLocust,), {})
self.locust_client = WebPageUser.client
def test_LocustRunner(self):
testcase_file = os.path.join(os.getcwd(), 'tests', 'httpbin', 'basic.yml')
locust_runner = LocustRunner(self.locust_client)
locust_runner.run(testcase_file)

View File

@@ -1,201 +1,152 @@
import os import os
import time import time
import unittest
import requests import requests
from httprunner import context, exceptions, loader, parser, response, runner from httprunner import context, exceptions, loader, response
from tests.base import ApiServerUnittest from tests.base import ApiServerUnittest
class TestContext(ApiServerUnittest): class TestContext(ApiServerUnittest):
def setUp(self): 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') testcase_file_path = os.path.join(os.getcwd(), 'tests/data/demo_binds.yml')
self.testcases = loader.load_file(testcase_file_path) self.testcases = loader.load_file(testcase_file_path)
def test_context_init_functions(self): def test_init_context_functions(self):
self.assertIn("get_timestamp", self.context.testset_functions_config) context_functions = self.context.TESTCASE_SHARED_FUNCTIONS_MAPPING
self.assertIn("gen_random_string", self.context.testset_functions_config) 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 = [ variables = [
{"random": "${gen_random_string(5)}"}, {"TOKEN": "debugtalk"},
{"timestamp10": "${get_timestamp(10)}"} {"data": '{"name": "user", "password": "123456"}'}
] ]
self.context.bind_variables(variables) self.context.update_context_variables(variables, "testcase")
context_variables = self.context.testcase_variables_mapping 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) def test_update_context_teststep_level(self):
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
}
variables = [ variables = [
{"add1": "${add_one(2)}"}, {"TOKEN": "debugtalk"},
{"sum2nums": "${add_two_nums(2,3)}"} {"data": '{"name": "user", "password": "123456"}'}
] ]
self.context.bind_functions(function_binds) self.context.update_context_variables(variables, "teststep")
self.context.bind_variables(variables) 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 def test_eval_content_functions(self):
self.assertIn("add1", context_variables) content = "${sleep_N_secs(1)}"
self.assertEqual(context_variables["add1"], 3) start_time = time.time()
self.assertIn("sum2nums", context_variables) self.context.eval_content(content)
self.assertEqual(context_variables["sum2nums"], 5) elapsed_time = time.time() - start_time
self.assertGreater(elapsed_time, 1)
def test_call_builtin_functions(self): def test_eval_content_variables(self):
testcase1 = { content = "abc$SECRET_KEY"
"variables": [ self.assertEqual(
{"length": "${len(debugtalk)}"}, self.context.eval_content(content),
{"smallest": "${min(2, 3, 8)}"}, "abcDebugTalk"
{"largest": "${max(2, 3, 8)}"} )
]
}
testcase2 = self.testcases["builtin_functions"]
for testcase in [testcase1, testcase2]: # TODO: fix variable extraction
variables = testcase['variables'] # content = "abc$SECRET_KEYdef"
self.context.bind_variables(variables) # self.assertEqual(
# self.context.eval_content(content),
# "abcDebugTalkdef"
# )
context_variables = self.context.testcase_variables_mapping def test_update_testcase_runtime_variables_mapping(self):
self.assertEqual(context_variables["length"], 9) variables = {"abc": 123}
self.assertEqual(context_variables["smallest"], 2) self.context.update_testcase_runtime_variables_mapping(variables)
self.assertEqual(context_variables["largest"], 8) 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 = [ variables = [
{"TOKEN": "debugtalk"}, {"TOKEN": "debugtalk"},
{"random": "${gen_random_string(5)}"}, {"random": "${gen_random_string(5)}"},
{"data": '{"name": "user", "password": "123456"}'}, {"data": '{"name": "user", "password": "123456"}'},
{"authorization": "${gen_md5($TOKEN, $data, $random)}"} {"authorization": "${gen_md5($TOKEN, $data, $random)}"}
] ]
from tests import debugtalk self.context.update_context_variables(variables, "teststep")
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.assertIn("TOKEN", context_variables) request = {
TOKEN = context_variables["TOKEN"] "url": "http://127.0.0.1:5000/api/users/1000",
self.assertEqual(TOKEN, "debugtalk") "method": "POST",
self.assertIn("random", context_variables) "headers": {
self.assertIsInstance(context_variables["random"], str) "Content-Type": "application/json",
self.assertEqual(len(context_variables["random"]), 5) "authorization": "$authorization",
random = context_variables["random"] "random": "$random",
self.assertIn("data", context_variables) "secret_key": "$SECRET_KEY"
data = context_variables["data"] },
self.assertIn("authorization", context_variables) "data": "$data"
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"
}
} }
from tests import debugtalk parsed_request = self.context.get_parsed_request(request, level="teststep")
self.context.import_module_items(debugtalk)
self.context.bind_variables(testcase["variables"])
parsed_request = self.context.get_parsed_request(testcase["request"])
self.assertIn("authorization", parsed_request["headers"]) self.assertIn("authorization", parsed_request["headers"])
self.assertEqual(len(parsed_request["headers"]["authorization"]), 32) self.assertEqual(len(parsed_request["headers"]["authorization"]), 32)
self.assertIn("random", parsed_request["headers"]) self.assertIn("random", parsed_request["headers"])
self.assertEqual(len(parsed_request["headers"]["random"]), 5) self.assertEqual(len(parsed_request["headers"]["random"]), 5)
self.assertIn("data", parsed_request) 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") 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): def test_do_validation(self):
self.context.do_validation( self.context._do_validation(
{"check": "check", "check_value": 1, "expect": 1, "comparator": "eq"} {"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": "=="} {"check": "check", "check_value": "abc", "expect": "abc", "comparator": "=="}
) )
self.context._do_validation(
config_dict = {
"path": 'tests/data/demo_testset_hardcode.yml'
}
self.context.config_context(config_dict, "testset")
self.context.do_validation(
{"check": "status_code", "check_value": "201", "expect": 3, "comparator": "sum_status_code"} {"check": "status_code", "check_value": "201", "expect": 3, "comparator": "sum_status_code"}
) )
@@ -213,7 +164,7 @@ class TestContext(ApiServerUnittest):
{"resp_status_code": 200}, {"resp_status_code": 200},
{"resp_body_success": True} {"resp_body_success": True}
] ]
self.context.bind_variables(variables) self.context.update_context_variables(variables, "teststep")
with self.assertRaises(exceptions.ValidationFailure): with self.assertRaises(exceptions.ValidationFailure):
self.context.validate(validators, resp_obj) self.context.validate(validators, resp_obj)
@@ -228,13 +179,7 @@ class TestContext(ApiServerUnittest):
{"resp_status_code": 201}, {"resp_status_code": 201},
{"resp_body_success": True} {"resp_body_success": True}
] ]
self.context.bind_variables(variables) self.context.update_context_variables(variables, "teststep")
from tests.debugtalk import is_status_code_200
functions = {
"is_status_code_200": is_status_code_200
}
self.context.bind_functions(functions)
self.context.validate(validators, resp_obj) self.context.validate(validators, resp_obj)
def test_validate_exception(self): def test_validate_exception(self):
@@ -248,7 +193,7 @@ class TestContext(ApiServerUnittest):
{"check": "$resp_status_code", "comparator": "eq", "expect": 201} {"check": "$resp_status_code", "comparator": "eq", "expect": 201}
] ]
variables = [] variables = []
self.context.bind_variables(variables) self.context.update_context_variables(variables, "teststep")
with self.assertRaises(exceptions.VariableNotFound): with self.assertRaises(exceptions.VariableNotFound):
self.context.validate(validators, resp_obj) self.context.validate(validators, resp_obj)
@@ -257,328 +202,7 @@ class TestContext(ApiServerUnittest):
variables = [ variables = [
{"resp_status_code": 200} {"resp_status_code": 200}
] ]
self.context.bind_variables(variables) self.context.update_context_variables(variables, "teststep")
with self.assertRaises(exceptions.ValidationFailure): with self.assertRaises(exceptions.ValidationFailure):
self.context.validate(validators, resp_obj) 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_extract_functions(self):
self.assertEqual(
parser.extract_functions("${func()}"),
["func()"]
)
self.assertEqual(
parser.extract_functions("${func(5)}"),
["func(5)"]
)
self.assertEqual(
parser.extract_functions("${func(a=1, b=2)}"),
["func(a=1, b=2)"]
)
self.assertEqual(
parser.extract_functions("${func(1, $b, c=$x, d=4)}"),
["func(1, $b, c=$x, d=4)"]
)
self.assertEqual(
parser.extract_functions("/api/1000?_t=${get_timestamp()}"),
["get_timestamp()"]
)
self.assertEqual(
parser.extract_functions("/api/${add(1, 2)}"),
["add(1, 2)"]
)
self.assertEqual(
parser.extract_functions("/api/${add(1, 2)}?_t=${get_timestamp()}"),
["add(1, 2)", "get_timestamp()"]
)
self.assertEqual(
parser.extract_functions("abc${func(1, 2, a=3, b=4)}def"),
["func(1, 2, a=3, b=4)"]
)
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
)

View File

@@ -1,157 +0,0 @@
import os
import shutil
from httprunner import HttpRunner
from tests.api_server import HTTPBIN_SERVER
from tests.base import ApiServerUnittest
class TestHttpRunner(ApiServerUnittest):
def setUp(self):
self.testset_path = "tests/data/demo_testset_cli.yml"
self.testset = {
'name': 'testset description',
'config': {
'path': 'docs/data/demo-quickstart-2.yml',
'name': 'testset description',
'request': {
'base_url': '',
'headers': {'User-Agent': 'python-requests/2.18.4'}
},
'variables': [],
'output': ['token']
},
'api': {},
'teststeps': [
{
'name': '/api/get-token',
'request': {
'url': 'http://127.0.0.1:5000/api/get-token',
'method': 'POST',
'headers': {'Content-Type': 'application/json', 'app_version': '2.8.6', 'device_sn': 'FwgRiO7CNA50DSU', 'os_platform': 'ios', 'user_agent': 'iOS/10.3'},
'json': {'sign': '958a05393efef0ac7c0fb80a7eac45e24fd40c27'}
},
'extract': [
{'token': 'content.token'}
],
'validate': [
{'eq': ['status_code', 200]},
{'eq': ['headers.Content-Type', 'application/json']},
{'eq': ['content.success', True]}
]
},
{
'name': '/api/users/1000',
'request': {
'url': 'http://127.0.0.1:5000/api/users/1000',
'method': 'POST',
'headers': {'Content-Type': 'application/json', 'device_sn': 'FwgRiO7CNA50DSU','token': '$token'}, 'json': {'name': 'user1', 'password': '123456'}
},
'validate': [
{'eq': ['status_code', 201]},
{'eq': ['headers.Content-Type', 'application/json']},
{'eq': ['content.success', True]},
{'eq': ['content.msg', 'user created successfully.']}
]
}
]
}
self.reset_all()
def reset_all(self):
url = "%s/api/reset-all" % self.host
headers = self.get_authenticated_headers()
return self.api_client.get(url, headers=headers)
def test_text_run_times(self):
runner = HttpRunner().run(self.testset_path)
self.assertEqual(runner.summary["stat"]["testsRun"], 10)
def test_text_skip(self):
runner = HttpRunner().run(self.testset_path)
self.assertEqual(runner.summary["stat"]["skipped"], 4)
def test_html_report(self):
kwargs = {}
output_folder_name = os.path.basename(os.path.splitext(self.testset_path)[0])
runner = HttpRunner().run(self.testset_path)
summary = runner.summary
self.assertEqual(summary["stat"]["testsRun"], 10)
self.assertEqual(summary["stat"]["skipped"], 4)
runner.gen_html_report(html_report_name=output_folder_name)
report_save_dir = os.path.join(os.getcwd(), 'reports', output_folder_name)
shutil.rmtree(report_save_dir)
def test_run_testsets(self):
testsets = [self.testset]
runner = HttpRunner().run(testsets)
summary = runner.summary
self.assertTrue(summary["success"])
self.assertEqual(summary["stat"]["testsRun"], 2)
self.assertIn("details", summary)
self.assertIn("records", summary["details"][0])
def test_run_testset(self):
testsets = self.testset
runner = HttpRunner().run(testsets)
summary = runner.summary
self.assertTrue(summary["success"])
self.assertEqual(summary["stat"]["testsRun"], 2)
self.assertIn("records", summary["details"][0])
def test_run_yaml_upload(self):
testset_path = "tests/httpbin/upload.yml"
runner = HttpRunner().run(testset_path)
summary = runner.summary
self.assertTrue(summary["success"])
self.assertEqual(summary["stat"]["testsRun"], 1)
self.assertIn("details", summary)
self.assertIn("records", summary["details"][0])
def test_run_post_data(self):
testsets = [
{
"name": "post data",
"teststeps": [
{
"name": "post data",
"request": {
"url": "{}/post".format(HTTPBIN_SERVER),
"method": "POST",
"headers": {
"Content-Type": "application/json"
},
"data": "abc"
},
"validate": [
{"eq": ["status_code", 200]}
]
}
]
}
]
runner = HttpRunner().run(testsets)
summary = runner.summary
self.assertTrue(summary["success"])
self.assertEqual(summary["stat"]["testsRun"], 1)
self.assertEqual(summary["details"][0]["records"][0]["meta_data"]["response"]["json"]["data"], "abc")
def test_html_report_repsonse_image(self):
testset_path = "tests/httpbin/load_image.yml"
runner = HttpRunner().run(testset_path)
summary = runner.summary
output_folder_name = os.path.basename(os.path.splitext(testset_path)[0])
report = runner.gen_html_report(html_report_name=output_folder_name)
self.assertTrue(os.path.isfile(report))
report_save_dir = os.path.join(os.getcwd(), 'reports', output_folder_name)
shutil.rmtree(report_save_dir)
def test_testcase_layer(self):
testcase_path = "tests/testcases/smoketest.yml"
runner = HttpRunner(failfast=True).run(testcase_path)
summary = runner.summary
self.assertTrue(summary["success"])
self.assertEqual(summary["stat"]["testsRun"], 8)

View File

@@ -1,7 +1,8 @@
import os import os
import unittest import unittest
from httprunner import exceptions, loader, task, validator from httprunner import exceptions, loader, validator
class TestFileLoader(unittest.TestCase): class TestFileLoader(unittest.TestCase):
@@ -169,6 +170,7 @@ class TestFileLoader(unittest.TestCase):
"tests/debugtalk.py" "tests/debugtalk.py"
) )
class TestModuleLoader(unittest.TestCase): class TestModuleLoader(unittest.TestCase):
def test_filter_module_functions(self): def test_filter_module_functions(self):
@@ -178,11 +180,16 @@ class TestModuleLoader(unittest.TestCase):
self.assertNotIn("is_py3", functions_dict) self.assertNotIn("is_py3", functions_dict)
def test_load_debugtalk_module(self): def test_load_debugtalk_module(self):
imported_module_items = loader.load_debugtalk_module() project_dir = os.path.join(os.getcwd(), "tests")
self.assertEqual(imported_module_items["functions"], {}) loader.load_project_tests(project_dir)
self.assertEqual(imported_module_items["variables"], {}) loader.load_debugtalk_module()
imported_module_items = loader.project_mapping["debugtalk"]
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( self.assertEqual(
imported_module_items["variables"]["SECRET_KEY"], imported_module_items["variables"]["SECRET_KEY"],
"DebugTalk" "DebugTalk"
@@ -472,20 +479,9 @@ class TestSuiteLoader(unittest.TestCase):
def test_load_project_tests(self): def test_load_project_tests(self):
project_dir = os.path.join(os.getcwd(), "tests") project_dir = os.path.join(os.getcwd(), "tests")
project_tests = loader.load_project_tests(project_dir) loader.load_project_tests(project_dir)
self.assertEqual(project_tests["debugtalk"]["variables"]["SECRET_KEY"], "DebugTalk") loader.load_debugtalk_module(project_dir)
self.assertIn("get_token", project_tests["def-api"]) project_mapping = loader.project_mapping
self.assertIn("setup_and_reset", project_tests["def-testcase"]) self.assertEqual(project_mapping["debugtalk"]["variables"]["SECRET_KEY"], "DebugTalk")
self.assertIn("get_token", project_mapping["def-api"])
def test_loader(self): self.assertIn("setup_and_reset", project_mapping["def-testcase"])
hrunner = task.HttpRunner(dot_env_path="tests/data/test.env")
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"])

View File

@@ -2,7 +2,7 @@ import os
import time import time
import unittest import unittest
from httprunner import exceptions, parser from httprunner import exceptions, loader, parser
class TestParser(unittest.TestCase): class TestParser(unittest.TestCase):
@@ -115,6 +115,40 @@ class TestParser(unittest.TestCase):
{"check": "status_code", "comparator": "eq", "expect": 201} {"check": "status_code", "comparator": "eq", "expect": 201}
) )
def test_extract_functions(self):
self.assertEqual(
parser.extract_functions("${func()}"),
["func()"]
)
self.assertEqual(
parser.extract_functions("${func(5)}"),
["func(5)"]
)
self.assertEqual(
parser.extract_functions("${func(a=1, b=2)}"),
["func(a=1, b=2)"]
)
self.assertEqual(
parser.extract_functions("${func(1, $b, c=$x, d=4)}"),
["func(1, $b, c=$x, d=4)"]
)
self.assertEqual(
parser.extract_functions("/api/1000?_t=${get_timestamp()}"),
["get_timestamp()"]
)
self.assertEqual(
parser.extract_functions("/api/${add(1, 2)}"),
["add(1, 2)"]
)
self.assertEqual(
parser.extract_functions("/api/${add(1, 2)}?_t=${get_timestamp()}"),
["add(1, 2)", "get_timestamp()"]
)
self.assertEqual(
parser.extract_functions("abc${func(1, 2, a=3, b=4)}def"),
["func(1, 2, a=3, b=4)"]
)
def test_parse_data(self): def test_parse_data(self):
content = { content = {
'request': { 'request': {
@@ -125,19 +159,277 @@ class TestParser(unittest.TestCase):
"null": None, "null": None,
"true": True, "true": True,
"false": False, "false": False,
"empty_str": "" "empty_str": "",
"value": "abc${add_one(3)}def"
} }
} }
} }
mapping = { variables_mapping = {
"$uid": 1000, "uid": 1000,
"$method": "POST" "method": "POST",
"token": "abc123"
} }
result = parser.parse_data(content, mapping) functions_mapping = {
"add_one": lambda x: x + 1
}
result = parser.parse_data(content, variables_mapping, functions_mapping)
self.assertEqual("/api/users/1000", result["request"]["url"]) self.assertEqual("/api/users/1000", result["request"]["url"])
self.assertEqual("$token", result["request"]["headers"]["token"]) self.assertEqual("abc123", result["request"]["headers"]["token"])
self.assertEqual("POST", result["request"]["method"]) self.assertEqual("POST", result["request"]["method"])
self.assertIsNone(result["request"]["data"]["null"]) self.assertIsNone(result["request"]["data"]["null"])
self.assertTrue(result["request"]["data"]["true"]) self.assertTrue(result["request"]["data"]["true"])
self.assertFalse(result["request"]["data"]["false"]) self.assertFalse(result["request"]["data"]["false"])
self.assertEqual("", result["request"]["data"]["empty_str"]) self.assertEqual("", result["request"]["data"]["empty_str"])
self.assertEqual("abc4def", result["request"]["data"]["value"])
def test_parse_data_variables(self):
variables_mapping = {
"var_1": "abc",
"var_2": "def",
"var_3": 123,
"var_4": {"a": 1},
"var_5": True,
"var_6": None
}
self.assertEqual(
parser.parse_data("$var_1", variables_mapping),
"abc"
)
self.assertEqual(
parser.parse_data("var_1", variables_mapping),
"var_1"
)
self.assertEqual(
parser.parse_data("$var_1#XYZ", variables_mapping),
"abc#XYZ"
)
self.assertEqual(
parser.parse_data("/$var_1/$var_2/var3", variables_mapping),
"/abc/def/var3"
)
self.assertEqual(
parser.parse_data("/$var_1/$var_2/$var_1", variables_mapping),
"/abc/def/abc"
)
self.assertEqual(
parser.parse_string_variables("${func($var_1, $var_2, xyz)}", variables_mapping),
"${func(abc, def, xyz)}"
)
self.assertEqual(
parser.parse_data("$var_3", variables_mapping),
123
)
self.assertEqual(
parser.parse_data("$var_4", variables_mapping),
{"a": 1}
)
self.assertEqual(
parser.parse_data("$var_5", variables_mapping),
True
)
self.assertEqual(
parser.parse_data("abc$var_5", variables_mapping),
"abcTrue"
)
self.assertEqual(
parser.parse_data("abc$var_4", variables_mapping),
"abc{'a': 1}"
)
self.assertEqual(
parser.parse_data("$var_6", variables_mapping),
None
)
with self.assertRaises(exceptions.VariableNotFound):
parser.parse_data("/api/$SECRET_KEY", variables_mapping)
self.assertEqual(
parser.parse_data(["$var_1", "$var_2"], variables_mapping),
["abc", "def"]
)
self.assertEqual(
parser.parse_data({"$var_1": "$var_2"}, variables_mapping),
{"abc": "def"}
)
def test_parse_data_multiple_identical_variables(self):
variables_mapping = {
"userid": 100,
"data": 1498
}
content = "/users/$userid/training/$data?userId=$userid&data=$data"
self.assertEqual(
parser.parse_data(content, variables_mapping),
"/users/100/training/1498?userId=100&data=1498"
)
variables_mapping = {
"user": 100,
"userid": 1000,
"data": 1498
}
content = "/users/$user/$userid/$data?userId=$userid&data=$data"
self.assertEqual(
parser.parse_data(content, variables_mapping),
"/users/100/1000/1498?userId=1000&data=1498"
)
def test_parse_data_functions(self):
import random, string
functions_mapping = {
"gen_random_string": lambda str_len: ''.join(random.choice(string.ascii_letters + string.digits) \
for _ in range(str_len))
}
result = parser.parse_data("${gen_random_string(5)}", functions_mapping=functions_mapping)
self.assertEqual(len(result), 5)
add_two_nums = lambda a, b=1: a + b
functions_mapping["add_two_nums"] = add_two_nums
self.assertEqual(
parser.parse_data("${add_two_nums(1)}", functions_mapping=functions_mapping),
2
)
self.assertEqual(
parser.parse_data("${add_two_nums(1, 2)}", functions_mapping=functions_mapping),
3
)
self.assertEqual(
parser.parse_data("/api/${add_two_nums(1, 2)}", functions_mapping=functions_mapping),
"/api/3"
)
with self.assertRaises(exceptions.FunctionNotFound):
parser.parse_data("/api/${gen_md5(abc)}")
def test_parse_data_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 = parser.parse_data(testcase_template, variables, functions)
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_substitute_variables(self):
content = {
'request': {
'url': '/api/users/$uid',
'headers': {'token': '$token'}
}
}
variables_mapping = {"$uid": 1000}
substituted_data = parser.substitute_variables(content, variables_mapping)
self.assertEqual(substituted_data["request"]["url"], "/api/users/1000")
self.assertEqual(substituted_data["request"]["headers"], {'token': '$token'})
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"]]}
]
variables_mapping = {}
functions_mapping = {}
cartesian_product_parameters = parser.parse_parameters(
parameters, variables_mapping, functions_mapping)
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_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"
)
from tests import debugtalk
debugtalk_module = loader.load_python_module(debugtalk)
cartesian_product_parameters = parser.parse_parameters(
parameters,
debugtalk_module["variables"],
debugtalk_module["functions"]
)
self.assertEqual(
len(cartesian_product_parameters),
2 * 2
)
def test_parse_parameters_parameterize(self):
parameters = [
{"app_version": "${parameterize(tests/data/app_version.csv)}"},
{"username-password": "${parameterize(tests/data/account.csv)}"}
]
variables_mapping = {}
functions_mapping = {}
cartesian_product_parameters = parser.parse_parameters(
parameters, variables_mapping, functions_mapping)
self.assertEqual(
len(cartesian_product_parameters),
2 * 3
)
def test_parse_parameters_mix(self):
project_dir = os.path.join(os.getcwd(), "tests")
loader.load_debugtalk_module(project_dir)
project_mapping = loader.project_mapping
parameters = [
{"user_agent": ["iOS/10.1", "iOS/10.2", "iOS/10.3"]},
{"app_version": "${gen_app_version()}"},
{"username-password": "${parameterize(tests/data/account.csv)}"}
]
variables_mapping = {}
functions_mapping = project_mapping["debugtalk"]["functions"]
testset_path = os.path.join(
os.getcwd(),
"tests/data/demo_parameters.yml"
)
cartesian_product_parameters = parser.parse_parameters(
parameters, variables_mapping, functions_mapping)
self.assertEqual(
len(cartesian_product_parameters),
3 * 2 * 3
)

View File

@@ -1,7 +1,7 @@
import os import os
import time import time
from httprunner import HttpRunner, exceptions, loader, runner from httprunner import exceptions, loader, runner
from httprunner.utils import deep_update_dict from httprunner.utils import deep_update_dict
from tests.api_server import HTTPBIN_SERVER from tests.api_server import HTTPBIN_SERVER
from tests.base import ApiServerUnittest from tests.base import ApiServerUnittest
@@ -10,38 +10,47 @@ from tests.base import ApiServerUnittest
class TestRunner(ApiServerUnittest): class TestRunner(ApiServerUnittest):
def setUp(self): 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() self.reset_all()
self.testcase_file_path_list = [
os.path.join(
os.getcwd(), 'tests/data/demo_testset_hardcode.yml'),
os.path.join(
os.getcwd(), 'tests/data/demo_testset_hardcode.json')
]
def reset_all(self): def reset_all(self):
url = "%s/api/reset-all" % self.host url = "%s/api/reset-all" % self.host
headers = self.get_authenticated_headers() headers = self.get_authenticated_headers()
return self.api_client.get(url, headers=headers) return self.api_client.get(url, headers=headers)
def test_run_single_testcase(self): def test_run_single_testcase(self):
for testcase_file_path in self.testcase_file_path_list: testcase_file_path_list = [
os.path.join(
os.getcwd(), 'tests/data/demo_testset_hardcode.yml'),
os.path.join(
os.getcwd(), 'tests/data/demo_testset_hardcode.json')
]
for testcase_file_path in testcase_file_path_list:
testcases = loader.load_file(testcase_file_path) testcases = loader.load_file(testcase_file_path)
config_dict = { 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"] test = testcases[0]["test"]
self.test_runner.run_test(test) test_runner.run_test(test)
test = testcases[1]["test"] test = testcases[1]["test"]
self.test_runner.run_test(test) test_runner.run_test(test)
test = testcases[2]["test"] test = testcases[2]["test"]
self.test_runner.run_test(test) test_runner.run_test(test)
def test_run_single_testcase_fail(self): def test_run_single_testcase_fail(self):
test = { test = {
@@ -75,6 +84,8 @@ class TestRunner(ApiServerUnittest):
config_dict = { config_dict = {
"path": os.path.join(os.getcwd(), __file__), "path": os.path.join(os.getcwd(), __file__),
"name": "basic test with httpbin", "name": "basic test with httpbin",
"variables": self.debugtalk_module["variables"],
"functions": self.debugtalk_module["functions"],
"request": { "request": {
"base_url": HTTPBIN_SERVER "base_url": HTTPBIN_SERVER
}, },
@@ -123,6 +134,8 @@ class TestRunner(ApiServerUnittest):
config_dict = { config_dict = {
"path": os.path.join(os.getcwd(), __file__), "path": os.path.join(os.getcwd(), __file__),
"name": "basic test with httpbin", "name": "basic test with httpbin",
"variables": self.debugtalk_module["variables"],
"functions": self.debugtalk_module["functions"],
"request": { "request": {
"base_url": HTTPBIN_SERVER "base_url": HTTPBIN_SERVER
} }
@@ -152,110 +165,6 @@ class TestRunner(ApiServerUnittest):
test_runner = runner.Runner(config_dict) test_runner = runner.Runner(config_dict)
test_runner.run_test(test) test_runner.run_test(test)
def test_run_httprunner_with_hooks(self):
testcase_file_path = os.path.join(
os.getcwd(), 'tests/httpbin/hooks.yml')
start_time = time.time()
runner = HttpRunner().run(testcase_file_path)
end_time = time.time()
summary = runner.summary
self.assertTrue(summary["success"])
self.assertLess(end_time - start_time, 10)
def test_run_httprunner_with_teardown_hooks_alter_response(self):
testsets = [
{
"name": "test teardown hooks",
"config": {
'path': 'tests/httpbin/hooks.yml',
},
"teststeps": [
{
"name": "test teardown hooks",
"request": {
"url": "{}/headers".format(HTTPBIN_SERVER),
"method": "GET",
"data": "abc"
},
"teardown_hooks": [
"${alter_response($response)}"
],
"validate": [
{"eq": ["status_code", 500]},
{"eq": ["headers.content-type", "html/text"]},
{"eq": ["json.headers.Host", "127.0.0.1:8888"]},
{"eq": ["content.headers.Host", "127.0.0.1:8888"]},
{"eq": ["text.headers.Host", "127.0.0.1:8888"]},
{"eq": ["new_attribute", "new_attribute_value"]},
{"eq": ["new_attribute_dict", {"key": 123}]},
{"eq": ["new_attribute_dict.key", 123]}
]
}
]
}
]
runner = HttpRunner().run(testsets)
summary = runner.summary
self.assertTrue(summary["success"])
def test_run_httprunner_with_teardown_hooks_not_exist_attribute(self):
testsets = [
{
"name": "test teardown hooks",
"config": {
'path': 'tests/httpbin/hooks.yml',
},
"teststeps": [
{
"name": "test teardown hooks",
"request": {
"url": "{}/headers".format(HTTPBIN_SERVER),
"method": "GET",
"data": "abc"
},
"teardown_hooks": [
"${alter_response($response)}"
],
"validate": [
{"eq": ["attribute_not_exist", "new_attribute"]}
]
}
]
}
]
runner = HttpRunner().run(testsets)
summary = runner.summary
self.assertFalse(summary["success"])
self.assertEqual(summary["stat"]["errors"], 1)
def test_run_httprunner_with_teardown_hooks_error(self):
testsets = [
{
"name": "test teardown hooks",
"config": {
'path': 'tests/httpbin/hooks.yml',
},
"teststeps": [
{
"name": "test teardown hooks",
"request": {
"url": "{}/headers".format(HTTPBIN_SERVER),
"method": "GET",
"data": "abc"
},
"teardown_hooks": [
"${alter_response_error($response)}"
]
}
]
}
]
runner = HttpRunner().run(testsets)
summary = runner.summary
self.assertFalse(summary["success"])
self.assertEqual(summary["stat"]["errors"], 1)
def test_run_testset_with_teardown_hooks_success(self): def test_run_testset_with_teardown_hooks_success(self):
test = { test = {
"name": "get token", "name": "get token",
@@ -281,7 +190,7 @@ class TestRunner(ApiServerUnittest):
config_dict = { config_dict = {
"path": os.path.join(os.getcwd(), __file__) "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() start_time = time.time()
self.test_runner.run_test(test) self.test_runner.run_test(test)
@@ -314,7 +223,7 @@ class TestRunner(ApiServerUnittest):
config_dict = { config_dict = {
"path": os.path.join(os.getcwd(), __file__) "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() start_time = time.time()
self.test_runner.run_test(test) self.test_runner.run_test(test)
@@ -322,62 +231,6 @@ class TestRunner(ApiServerUnittest):
# check if teardown function executed # check if teardown function executed
self.assertGreater(end_time - start_time, 2) self.assertGreater(end_time - start_time, 2)
def test_run_testset_hardcode(self):
for testcase_file_path in self.testcase_file_path_list:
runner = HttpRunner().run(testcase_file_path)
self.assertTrue(runner.summary["success"])
def test_run_testsets_hardcode(self):
runner = HttpRunner().run(self.testcase_file_path_list)
summary = runner.summary
self.assertTrue(summary["success"])
self.assertEqual(summary["stat"]["testsRun"], 6)
self.assertEqual(summary["stat"]["successes"], 6)
def test_run_testset_template_variables(self):
testcase_file_path = os.path.join(
os.getcwd(), 'tests/data/demo_testset_variables.yml')
runner = HttpRunner().run(testcase_file_path)
summary = runner.summary
self.assertTrue(summary["success"])
def test_run_testset_template_import_functions(self):
testcase_file_path = os.path.join(
os.getcwd(), 'tests/data/demo_testset_functions.yml')
runner = HttpRunner().run(testcase_file_path)
summary = runner.summary
self.assertTrue(summary["success"])
def test_run_testset_layered(self):
testcase_file_path = os.path.join(
os.getcwd(), 'tests/data/demo_testset_layer.yml')
runner = HttpRunner().run(testcase_file_path)
summary = runner.summary
self.assertTrue(summary["success"])
def test_run_testset_output(self):
testcase_file_path = os.path.join(
os.getcwd(), 'tests/data/demo_testset_layer.yml')
runner = HttpRunner().run(testcase_file_path)
summary = runner.summary
self.assertTrue(summary["success"])
self.assertIn("token", summary["details"][0]["output"][0]["out"])
#TODO: fix
self.assertEqual(len(summary["details"][0]["output"]), 3)
def test_run_testset_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)
summary = runner.summary
self.assertTrue(summary["success"])
self.assertIn("token", summary["details"][0]["output"][0]["out"])
#TODO: fix
self.assertEqual(len(summary["details"][0]["output"]), 3)
def test_run_testcase_with_empty_header(self): def test_run_testcase_with_empty_header(self):
testcase_file_path = os.path.join( testcase_file_path = os.path.join(
os.getcwd(), 'tests/data/test_bugfix.yml') os.getcwd(), 'tests/data/test_bugfix.yml')
@@ -398,20 +251,11 @@ class TestRunner(ApiServerUnittest):
config_dict = { config_dict = {
"path": testcase_file_path "path": testcase_file_path
} }
self.test_runner.init_config(config_dict, "testset") self.test_runner.init_config(config_dict, "testcase")
test = testcases[2]["test"] test = testcases[2]["test"]
self.test_runner.run_test(test) self.test_runner.run_test(test)
def test_run_testset_with_parameters(self):
testcase_file_path = os.path.join(
os.getcwd(), 'tests/data/demo_parameters.yml')
runner = HttpRunner().run(testcase_file_path)
summary = runner.summary
self.assertTrue(summary["success"])
self.assertEqual(len(summary["details"][0]["output"]), 3 * 2 * 2)
self.assertEqual(summary["stat"]["testsRun"], 3 * 2 * 2)
def test_run_validate_elapsed(self): def test_run_validate_elapsed(self):
test = { test = {
"name": "get token", "name": "get token",

View File

@@ -1,80 +0,0 @@
import os
from httprunner import loader, task
from tests.base import ApiServerUnittest
class TestTask(ApiServerUnittest):
def setUp(self):
self.reset_all()
def reset_all(self):
url = "%s/api/reset-all" % self.host
headers = self.get_authenticated_headers()
return self.api_client.get(url, headers=headers)
def test_create_suite(self):
testcase_file_path = os.path.join(os.getcwd(), 'tests/data/demo_testset_variables.yml')
testset = loader._load_test_file(testcase_file_path)
suite = task.TestSuite(testset)
self.assertEqual(suite.countTestCases(), 3)
for testcase in suite:
self.assertIsInstance(testcase, task.TestCase)
def test_create_task(self):
testsets = [
{
'name': 'testset description',
'config': {
'path': 'docs/data/demo-quickstart-2.yml',
'name': 'testset description',
'request': {
'base_url': '',
'headers': {'User-Agent': 'python-requests/2.18.4'}
},
'variables': [],
'output': ['token']
},
'api': {},
'teststeps': [
{
'name': '/api/get-token',
'request': {
'url': 'http://127.0.0.1:5000/api/get-token',
'method': 'POST',
'headers': {'Content-Type': 'application/json', 'app_version': '2.8.6', 'device_sn': 'FwgRiO7CNA50DSU', 'os_platform': 'ios', 'user_agent': 'iOS/10.3'},
'json': {'sign': '958a05393efef0ac7c0fb80a7eac45e24fd40c27'}
},
'extract': [
{'token': 'content.token'}
],
'validate': [
{'eq': ['status_code', 200]},
{'eq': ['headers.Content-Type', 'application/json']},
{'eq': ['content.success', True]}
]
},
{
'name': '/api/users/1000',
'request': {
'url': 'http://127.0.0.1:5000/api/users/1000',
'method': 'POST',
'headers': {'Content-Type': 'application/json', 'device_sn': 'FwgRiO7CNA50DSU','token': '$token'}, 'json': {'name': 'user1', 'password': '123456'}
},
'validate': [
{'eq': ['status_code', 201]},
{'eq': ['headers.Content-Type', 'application/json']},
{'eq': ['content.success', True]},
{'eq': ['content.msg', 'user created successfully.']}
]
}
]
}
]
test_suite_list = task.init_test_suites(testsets)
self.assertEqual(len(test_suite_list), 1)
task_suite = test_suite_list[0]
self.assertEqual(task_suite.countTestCases(), 2)
for testcase in task_suite:
self.assertIsInstance(testcase, task.TestCase)

View File

@@ -210,7 +210,7 @@ class TestUtils(ApiServerUnittest):
{"a": 1}, {"a": 1},
{"b": 2} {"b": 2}
] ]
ordered_dict = utils.convert_to_order_dict(map_list) ordered_dict = utils.convert_mappinglist_to_orderdict(map_list)
self.assertIsInstance(ordered_dict, dict) self.assertIsInstance(ordered_dict, dict)
self.assertIn("a", ordered_dict) self.assertIn("a", ordered_dict)
@@ -219,7 +219,7 @@ class TestUtils(ApiServerUnittest):
{"a": 1}, {"a": 1},
{"b": 2} {"b": 2}
] ]
ordered_dict = utils.convert_to_order_dict(map_list) ordered_dict = utils.convert_mappinglist_to_orderdict(map_list)
override_mapping = {"a": 3, "c": 4} override_mapping = {"a": 3, "c": 4}
new_dict = utils.update_ordered_dict(ordered_dict, override_mapping) new_dict = utils.update_ordered_dict(ordered_dict, override_mapping)
self.assertEqual(3, new_dict["a"]) self.assertEqual(3, new_dict["a"])
@@ -231,7 +231,7 @@ class TestUtils(ApiServerUnittest):
{"b": 2} {"b": 2}
] ]
override_mapping = {"a": 3, "c": 4} override_mapping = {"a": 3, "c": 4}
new_dict = utils.override_variables_binds(map_list, override_mapping) new_dict = utils.override_mapping_list(map_list, override_mapping)
self.assertEqual(3, new_dict["a"]) self.assertEqual(3, new_dict["a"])
self.assertEqual(4, new_dict["c"]) self.assertEqual(4, new_dict["c"])
@@ -242,14 +242,14 @@ class TestUtils(ApiServerUnittest):
} }
) )
override_mapping = {"a": 3, "c": 4} override_mapping = {"a": 3, "c": 4}
new_dict = utils.override_variables_binds(map_list, override_mapping) new_dict = utils.override_mapping_list(map_list, override_mapping)
self.assertEqual(3, new_dict["a"]) self.assertEqual(3, new_dict["a"])
self.assertEqual(4, new_dict["c"]) self.assertEqual(4, new_dict["c"])
map_list = "invalid" map_list = "invalid"
override_mapping = {"a": 3, "c": 4} override_mapping = {"a": 3, "c": 4}
with self.assertRaises(exceptions.ParamsError): with self.assertRaises(exceptions.ParamsError):
utils.override_variables_binds(map_list, override_mapping) utils.override_mapping_list(map_list, override_mapping)
def test_create_scaffold(self): def test_create_scaffold(self):
project_path = os.path.join(os.getcwd(), "projectABC") project_path = os.path.join(os.getcwd(), "projectABC")