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

Merge pull request #348 from HttpRunner/parser

Browse Source 
refactor all
This commit is contained in:
debugtalk
2018-08-22 19:05:59 +08:00
committed by GitHub
parent 27f8ee7db1 b68bb5fdab
commit 36df7931e0
26 changed files with 1950 additions and 1924 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
Pipfile
View File

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

2
httprunner/__about__.py
View File

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

2
httprunner/__init__.py
View File

@@ -1,3 +1,3 @@
# 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
)

2
httprunner/cli.py
View File

@@ -9,7 +9,7 @@ import unittest
from httprunner import logger
from httprunner.__about__ import __description__, __version__
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,
prettify_json_file, validate_json_file)

548
httprunner/context.py
View File

@@ -1,447 +1,145 @@
# encoding: utf-8
import copy
import os
import random
import re
import sys
from httprunner import built_in, exceptions, loader, logger, parser, utils
from httprunner.compat import OrderedDict, basestring, builtin_str, str
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
from httprunner import exceptions, logger, parser, utils
from httprunner.compat import OrderedDict
class Context(object):
""" Manages context functions and variables.
context has two levels, testset and testcase.
context has two levels, testcase and teststep.
"""
def __init__(self):
self.testset_shared_variables_mapping = OrderedDict()
self.testcase_variables_mapping = OrderedDict()
self.testcase_parser = TestcaseParser()
def __init__(self, variables=None, functions=None):
""" init Context with testcase variables and functions.
"""
# testcase level context
## TESTCASE_SHARED_VARIABLES_MAPPING and TESTCASE_SHARED_FUNCTIONS_MAPPING will not change.
self.TESTCASE_SHARED_VARIABLES_MAPPING = variables or OrderedDict()
self.TESTCASE_SHARED_FUNCTIONS_MAPPING = functions or OrderedDict()
# testcase level request, will not change
self.TESTCASE_SHARED_REQUEST_MAPPING = {}
self.evaluated_validators = []
self.init_context()
self.init_context_variables(level="testcase")
def init_context_variables(self, level="testcase"):
""" initialize testcase/teststep context
Args:
level (enum): "testcase" or "teststep"
def init_context(self, level='testset'):
"""
testset level context initializes when a file is loaded,
testcase level context initializes when each testcase starts.
"""
if level == "testset":
self.testset_functions_config = {}
self.testset_request_config = {}
self.testset_shared_variables_mapping = OrderedDict()
if level == "testcase":
# testcase level runtime context, will be updated with extracted variables in each teststep.
self.testcase_runtime_variables_mapping = copy.deepcopy(self.TESTCASE_SHARED_VARIABLES_MAPPING)
# testcase config shall inherit from testset configs,
# but can not change testset configs, that's why we use copy.deepcopy here.
self.testcase_functions_config = copy.deepcopy(self.testset_functions_config)
self.testcase_variables_mapping = copy.deepcopy(self.testset_shared_variables_mapping)
# teststep level context, will be altered in each teststep.
# teststep config shall inherit from testcase configs,
# but can not change testcase configs, that's why we use copy.deepcopy here.
self.teststep_variables_mapping = copy.deepcopy(self.testcase_runtime_variables_mapping)
self.testcase_parser.bind_functions(self.testcase_functions_config)
self.testcase_parser.update_binded_variables(self.testcase_variables_mapping)
def update_context_variables(self, variables, level):
""" update context variables, with level specified.
if level == "testset":
self.import_module_items(built_in)
Args:
variables (list/OrderedDict): testcase config block or teststep block
[
{"TOKEN": "debugtalk"},
{"random": "${gen_random_string(5)}"},
{"json": {'name': 'user', 'password': '123456'}},
{"md5": "${gen_md5($TOKEN, $json, $random)}"}
]
OrderDict({
"TOKEN": "debugtalk",
"random": "${gen_random_string(5)}",
"json": {'name': 'user', 'password': '123456'},
"md5": "${gen_md5($TOKEN, $json, $random)}"
})
level (enum): "testcase" or "teststep"
def config_context(self, config_dict, level):
if level == "testset":
self.testcase_parser.file_path = config_dict.get("path", None)
variables = config_dict.get('variables') \
or config_dict.get('variable_binds', OrderedDict())
self.bind_variables(variables, level)
def bind_functions(self, function_binds, level="testcase"):
""" Bind named functions within the context
This allows for passing in self-defined functions in testing.
e.g. function_binds:
{
"add_one": lambda x: x + 1, # lambda function
"add_two_nums": "lambda x, y: x + y" # lambda function in string
}
"""
eval_function_binds = {}
for func_name, function in function_binds.items():
if isinstance(function, str):
function = eval(function)
eval_function_binds[func_name] = function
self.__update_context_functions_config(level, eval_function_binds)
def import_module_items(self, imported_module):
""" import module functions and variables and bind to testset context
"""
module_mapping = loader.load_python_module(imported_module)
self.__update_context_functions_config("testset", module_mapping["functions"])
self.bind_variables(module_mapping["variables"], "testset")
def bind_variables(self, variables, level="testcase"):
""" bind variables to testset context or current testcase context.
variables in testset context can be used in all testcases of current test suite.
@param (list or OrderDict) variables, variable can be value or custom function.
if value is function, it will be called and bind result to variable.
e.g.
OrderDict({
"TOKEN": "debugtalk",
"random": "${gen_random_string(5)}",
"json": {'name': 'user', 'password': '123456'},
"md5": "${gen_md5($TOKEN, $json, $random)}"
})
"""
if isinstance(variables, list):
variables = utils.convert_to_order_dict(variables)
variables = utils.convert_mappinglist_to_orderdict(variables)
for variable_name, value in variables.items():
variable_eval_value = self.eval_content(value)
for variable_name, variable_value in variables.items():
variable_eval_value = self.eval_content(variable_value)
if level == "testset":
self.testset_shared_variables_mapping[variable_name] = variable_eval_value
if level == "testcase":
self.testcase_runtime_variables_mapping[variable_name] = variable_eval_value
self.bind_testcase_variable(variable_name, variable_eval_value)
def bind_testcase_variable(self, variable_name, variable_value):
""" bind and update testcase variables mapping
"""
self.testcase_variables_mapping[variable_name] = variable_value
self.testcase_parser.update_binded_variables(self.testcase_variables_mapping)
def bind_extracted_variables(self, variables):
""" bind extracted variables to testset context
@param (OrderDict) variables
extracted value do not need to evaluate.
"""
for variable_name, value in variables.items():
self.testset_shared_variables_mapping[variable_name] = value
self.bind_testcase_variable(variable_name, value)
def __update_context_functions_config(self, level, config_mapping):
"""
@param level: testset or testcase
@param config_type: functions
@param config_mapping: functions config mapping
"""
if level == "testset":
self.testset_functions_config.update(config_mapping)
self.testcase_functions_config.update(config_mapping)
self.testcase_parser.bind_functions(self.testcase_functions_config)
self.update_teststep_variables_mapping(variable_name, variable_eval_value)
def eval_content(self, content):
""" evaluate content recursively, take effect on each variable and function in content.
content may be in any data structure, include dict, list, tuple, number, string, etc.
"""
return self.testcase_parser.eval_content_with_bindings(content)
return parser.parse_data(
content,
self.teststep_variables_mapping,
self.TESTCASE_SHARED_FUNCTIONS_MAPPING
)
def update_testcase_runtime_variables_mapping(self, variables):
""" update testcase_runtime_variables_mapping with extracted vairables in teststep.
Args:
variables (OrderDict): extracted variables in teststep
def get_parsed_request(self, request_dict, level="testcase"):
""" get parsed request with bind variables and functions.
@param request_dict: request config mapping
@param level: testset or testcase
"""
if level == "testset":
request_dict = self.eval_content(
request_dict
for variable_name, variable_value in variables.items():
self.testcase_runtime_variables_mapping[variable_name] = variable_value
self.update_teststep_variables_mapping(variable_name, variable_value)
def update_teststep_variables_mapping(self, variable_name, variable_value):
""" bind and update testcase variables mapping
"""
self.teststep_variables_mapping[variable_name] = variable_value
def get_parsed_request(self, request_dict, level="teststep"):
""" get parsed request with variables and functions.
Args:
request_dict (dict): request config mapping
level (enum): "testcase" or "teststep"
Returns:
dict: parsed request dict
"""
if level == "testcase":
# testcase config request dict has been parsed in __parse_testcases
self.TESTCASE_SHARED_REQUEST_MAPPING = request_dict
return request_dict
else:
# teststep
return self.eval_content(
utils.deep_update_dict(
copy.deepcopy(self.TESTCASE_SHARED_REQUEST_MAPPING),
request_dict
)
)
self.testset_request_config.update(request_dict)
testcase_request_config = utils.deep_update_dict(
copy.deepcopy(self.testset_request_config),
request_dict
)
parsed_request = self.eval_content(
testcase_request_config
)
def __eval_check_item(self, validator, resp_obj):
""" evaluate check item in validator.
return parsed_request
Args:
validator (dict): validator
{"check": "status_code", "comparator": "eq", "expect": 201}
{"check": "$resp_body_success", "comparator": "eq", "expect": True}
resp_obj (object): requests.Response() object
Returns:
dict: validator info
{
"check": "status_code",
"check_value": 200,
"expect": 201,
"comparator": "eq"
}
def eval_check_item(self, validator, resp_obj):
""" evaluate check item in validator
@param (dict) validator
{"check": "status_code", "comparator": "eq", "expect": 201}
{"check": "$resp_body_success", "comparator": "eq", "expect": True}
@param (object) resp_obj
@return (dict) validator info
{
"check": "status_code",
"check_value": 200,
"expect": 201,
"comparator": "eq"
}
"""
check_item = validator["check"]
# check_item should only be the following 5 formats:
@@ -470,12 +168,22 @@ class Context(object):
validator["check_result"] = "unchecked"
return validator
def do_validation(self, validator_dict):
def _do_validation(self, validator_dict):
""" validate with functions
Args:
validator_dict (dict): validator dict
{
"check": "status_code",
"check_value": 200,
"expect": 201,
"comparator": "eq"
}
"""
# TODO: move comparator uniform to init_test_suites
comparator = utils.get_uniform_comparator(validator_dict["comparator"])
validate_func = self.testcase_parser.get_bind_function(comparator)
validate_func = self.TESTCASE_SHARED_FUNCTIONS_MAPPING.get(comparator)
if not validate_func:
raise exceptions.FunctionNotFound("comparator not found: {}".format(comparator))
@@ -516,26 +224,28 @@ class Context(object):
def validate(self, validators, resp_obj):
""" make validations
"""
evaluated_validators = []
if not validators:
return
return evaluated_validators
logger.log_info("start to validate.")
self.evaluated_validators = []
validate_pass = True
for validator in validators:
# evaluate validators with context variable mapping.
evaluated_validator = self.eval_check_item(
evaluated_validator = self.__eval_check_item(
parser.parse_validator(validator),
resp_obj
)
try:
self.do_validation(evaluated_validator)
self._do_validation(evaluated_validator)
except exceptions.ValidationFailure:
validate_pass = False
self.evaluated_validators.append(evaluated_validator)
evaluated_validators.append(evaluated_validator)
if not validate_pass:
raise exceptions.ValidationFailure
return evaluated_validators

65
httprunner/loader.py
View File

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

350
httprunner/parser.py
View File

@@ -4,12 +4,12 @@ import ast
import os
import re
from httprunner import exceptions
from httprunner.compat import builtin_str, numeric_types, str
from httprunner import exceptions, utils
from httprunner.compat import basestring, builtin_str, numeric_types, str
variable_regexp = r"\$([\w_]+)"
function_regexp = r"\$\{([\w_]+\([\$\w\.\-_ =,]*\))\}"
function_regexp_compile = re.compile(r"^([\w_]+)\(([\$\w\.\-_ =,]*)\)$")
function_regexp = r"\$\{([\w_]+\([\$\w\.\-/_ =,]*\))\}"
function_regexp_compile = re.compile(r"^([\w_]+)\(([\$\w\.\-/_ =,]*)\)$")
def parse_string_value(str_value):
@@ -200,53 +200,325 @@ def parse_validator(validator):
}
def parse_data(content, mapping):
""" substitute variables in content with 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
def substitute_variables(content, variables_mapping):
""" substitute variables in content with variables_mapping
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)):
return [
parse_data(item, mapping)
substitute_variables(item, variables_mapping)
for item in content
]
if isinstance(content, dict):
substituted_data = {}
for key, value in content.items():
eval_key = parse_data(key, mapping)
eval_value = parse_data(value, mapping)
eval_key = substitute_variables(key, variables_mapping)
eval_value = substitute_variables(value, variables_mapping)
substituted_data[eval_key] = eval_value
return substituted_data
# content is in string format here
for var, value in mapping.items():
if content == var:
# content is a variable
content = value
else:
if not isinstance(value, str):
value = builtin_str(value)
content = content.replace(var, value)
if isinstance(content, basestring):
# content is in string format here
for var, value in variables_mapping.items():
if content == var:
# content is a variable
content = value
else:
if not isinstance(value, str):
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

35
httprunner/report.py
View File

@@ -6,14 +6,13 @@ import platform
import time
import unittest
from base64 import b64encode
from collections import Iterable, OrderedDict
from collections import Iterable
from datetime import datetime
from httprunner import logger
from httprunner.__about__ import __version__
from httprunner.compat import basestring, bytes, json, numeric_types
from jinja2 import Template, escape
from requests.structures import CaseInsensitiveDict
def get_platform():
@@ -26,6 +25,7 @@ def get_platform():
"platform": platform.platform()
}
def get_summary(result):
""" get summary from test result
"""
@@ -58,6 +58,25 @@ def get_summary(result):
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):
""" render html report with specified report name and template
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
def stringify_data(meta_data, request_or_response):
"""
meta_data = {
@@ -151,6 +171,7 @@ def stringify_data(meta_data, request_or_response):
meta_data[request_or_response][key] = value
class HtmlTestResult(unittest.TextTestResult):
"""A html result class that can generate formatted html results.
@@ -161,12 +182,16 @@ class HtmlTestResult(unittest.TextTestResult):
self.records = []
def _record_test(self, test, status, attachment=''):
self.records.append({
data = {
'name': test.shortDescription(),
'status': status,
'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):
self.start_at = time.time()

2
httprunner/response.py
View File

@@ -221,7 +221,7 @@ class ResponseObject(object):
logger.log_info("start to extract from response object.")
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():
extracted_variables_mapping[key] = self.extract_field(field)

195
httprunner/runner.py
View File

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

334
httprunner/task.py
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
)

9
httprunner/templates/locustfile_template
View File

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

6
httprunner/templates/report_template.html
View File

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

198
httprunner/utils.py
View File

@@ -158,41 +158,49 @@ def lower_config_dict_key(config_dict):
return config_dict
def convert_to_order_dict(map_list):
""" convert mapping in list to ordered dict
@param (list) map_list
[
{"a": 1},
{"b": 2}
]
@return (OrderDict)
OrderDict({
"a": 1,
"b": 2
})
def convert_mappinglist_to_orderdict(mapping_list):
""" convert mapping list to ordered dict
Args:
mapping_list (list):
[
{"a": 1},
{"b": 2}
]
Returns:
OrderedDict: converted mapping in OrderedDict
OrderDict(
{
"a": 1,
"b": 2
}
)
"""
ordered_dict = OrderedDict()
for map_dict in map_list:
for map_dict in mapping_list:
ordered_dict.update(map_dict)
return ordered_dict
def update_ordered_dict(ordered_dict, override_mapping):
""" override ordered_dict with new mapping
@param
(OrderDict) ordered_dict
OrderDict({
"a": 1,
"b": 2
})
(dict) override_mapping
{"a": 3, "c": 4}
@return (OrderDict)
OrderDict({
"a": 3,
"b": 2,
"c": 4
})
""" override ordered_dict with new mapping.
Args:
ordered_dict (OrderDict): original ordered dict
override_mapping (dict): new variables mapping
Returns:
OrderDict: new overrided variables mapping.
Examples:
>>> ordered_dict = OrderDict({"a": 1, "b": 2})
>>> override_mapping = {"a": 3, "c": 4}
>>> update_ordered_dict(ordered_dict, override_mapping)
OrderDict({"a": 3, "b": 2, "c": 4})
"""
new_ordered_dict = copy.copy(ordered_dict)
for var, value in override_mapping.items():
@@ -200,11 +208,43 @@ def update_ordered_dict(ordered_dict, override_mapping):
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):
variables_ordered_dict = convert_to_order_dict(variables)
variables_ordered_dict = convert_mappinglist_to_orderdict(variables)
elif isinstance(variables, (OrderedDict, dict)):
variables_ordered_dict = variables
else:
@@ -215,18 +255,84 @@ def override_variables_binds(variables, new_mapping):
new_mapping
)
def print_output(outputs):
if not outputs:
return
def add_teststep(test_runner, teststep_dict):
""" 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 += '{:<6} | {:<16} : {:<}\n'.format("Type", "Variable", "Value")
content += '{:<6} | {:<16} : {:<}\n'.format("-" * 6, "-" * 16, "-" * 27)
content += content_format.format("Type", "Variable", "Value")
content += content_format.format("-" * 6, "-" * 16, "-" * 27)
def prepare_content(var_type, in_out):
content = ""
for variable, value in in_out.items():
if isinstance(value, tuple):
continue
if is_py2:
if isinstance(variable, unicode):
@@ -234,21 +340,17 @@ def print_output(outputs):
if isinstance(value, unicode):
value = value.encode("utf-8")
content += '{:<6} | {:<16} : {:<}\n'.format(var_type, variable, value)
content += content_format.format(var_type, variable, value)
return content
for output in outputs:
_in = output["in"]
_out = output["out"]
_in = in_out["in"]
_out = in_out["out"]
if not _out:
continue
content += prepare_content("Var", _in)
content += "\n"
content += prepare_content("Out", _out)
content += "-" * 56 + "\n"
content += prepare_content("Var", _in)
content += "\n"
content += prepare_content("Out", _out)
content += "-" * 56 + "\n"
logger.log_debug(content)
@@ -336,6 +438,7 @@ def validate_json_file(file_list):
print("OK")
def prettify_json_file(file_list):
""" prettify JSON testset format
"""
@@ -362,6 +465,7 @@ def prettify_json_file(file_list):
print("success: {}".format(outfile))
def get_python2_retire_msg():
retire_day = datetime(2020, 1, 1)
today = datetime.now()

24
httprunner/validator.py
View File

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

13
tests/data/demo_parameters.yml
View File

@@ -8,6 +8,7 @@
variables:
- device_sn: ${gen_random_string(15)}
- os_platform: 'ios'
- app_version: 2.8.5
request:
base_url: $BASE_URL
headers:
@@ -18,21 +19,9 @@
- test:
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)
extract:
- token: content.token
validate:
- "eq": ["status_code", 200]
- "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}

2
tests/httpbin/basic.yml
View File

@@ -76,7 +76,7 @@
method: GET
validate:
- eq: ["status_code", 200]
- eq: [cookies.name, "value"]
# - eq: [cookies.name, "value"]
- test:
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)

594
tests/test_context.py
View File

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

157
tests/test_httprunner.py
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)

40
tests/test_loader.py
View File

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

306
tests/test_parser.py
View File

@@ -2,7 +2,7 @@ import os
import time
import unittest
from httprunner import exceptions, parser
from httprunner import exceptions, loader, parser
class TestParser(unittest.TestCase):
@@ -115,6 +115,40 @@ class TestParser(unittest.TestCase):
{"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):
content = {
'request': {
@@ -125,19 +159,277 @@ class TestParser(unittest.TestCase):
"null": None,
"true": True,
"false": False,
"empty_str": ""
"empty_str": "",
"value": "abc${add_one(3)}def"
}
}
}
mapping = {
"$uid": 1000,
"$method": "POST"
variables_mapping = {
"uid": 1000,
"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("$token", result["request"]["headers"]["token"])
self.assertEqual("abc123", result["request"]["headers"]["token"])
self.assertEqual("POST", result["request"]["method"])
self.assertIsNone(result["request"]["data"]["null"])
self.assertTrue(result["request"]["data"]["true"])
self.assertFalse(result["request"]["data"]["false"])
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
)

218
tests/test_runner.py
View File

@@ -1,7 +1,7 @@
import os
import time
from httprunner import HttpRunner, exceptions, loader, runner
from httprunner import exceptions, loader, runner
from httprunner.utils import deep_update_dict
from tests.api_server import HTTPBIN_SERVER
from tests.base import ApiServerUnittest
@@ -10,38 +10,47 @@ from tests.base import ApiServerUnittest
class TestRunner(ApiServerUnittest):
def setUp(self):
self.test_runner = runner.Runner()
project_dir = os.path.join(os.getcwd(), "tests")
loader.load_project_tests(project_dir)
loader.load_debugtalk_module(project_dir)
self.debugtalk_module = loader.project_mapping["debugtalk"]
config_dict = {
"variables": self.debugtalk_module["variables"],
"functions": self.debugtalk_module["functions"]
}
self.test_runner = runner.Runner(config_dict)
self.reset_all()
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):
url = "%s/api/reset-all" % self.host
headers = self.get_authenticated_headers()
return self.api_client.get(url, headers=headers)
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)
config_dict = {
"path": testcase_file_path
"variables": self.debugtalk_module["variables"],
"functions": self.debugtalk_module["functions"]
}
self.test_runner.init_config(config_dict, "testset")
test_runner = runner.Runner(config_dict)
test = testcases[0]["test"]
self.test_runner.run_test(test)
test_runner.run_test(test)
test = testcases[1]["test"]
self.test_runner.run_test(test)
test_runner.run_test(test)
test = testcases[2]["test"]
self.test_runner.run_test(test)
test_runner.run_test(test)
def test_run_single_testcase_fail(self):
test = {
@@ -75,6 +84,8 @@ class TestRunner(ApiServerUnittest):
config_dict = {
"path": os.path.join(os.getcwd(), __file__),
"name": "basic test with httpbin",
"variables": self.debugtalk_module["variables"],
"functions": self.debugtalk_module["functions"],
"request": {
"base_url": HTTPBIN_SERVER
},
@@ -123,6 +134,8 @@ class TestRunner(ApiServerUnittest):
config_dict = {
"path": os.path.join(os.getcwd(), __file__),
"name": "basic test with httpbin",
"variables": self.debugtalk_module["variables"],
"functions": self.debugtalk_module["functions"],
"request": {
"base_url": HTTPBIN_SERVER
}
@@ -152,110 +165,6 @@ class TestRunner(ApiServerUnittest):
test_runner = runner.Runner(config_dict)
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):
test = {
"name": "get token",
@@ -281,7 +190,7 @@ class TestRunner(ApiServerUnittest):
config_dict = {
"path": os.path.join(os.getcwd(), __file__)
}
self.test_runner.init_config(config_dict, "testset")
self.test_runner.init_config(config_dict, "testcase")
start_time = time.time()
self.test_runner.run_test(test)
@@ -314,7 +223,7 @@ class TestRunner(ApiServerUnittest):
config_dict = {
"path": os.path.join(os.getcwd(), __file__)
}
self.test_runner.init_config(config_dict, "testset")
self.test_runner.init_config(config_dict, "testcase")
start_time = time.time()
self.test_runner.run_test(test)
@@ -322,62 +231,6 @@ class TestRunner(ApiServerUnittest):
# check if teardown function executed
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):
testcase_file_path = os.path.join(
os.getcwd(), 'tests/data/test_bugfix.yml')
@@ -398,20 +251,11 @@ class TestRunner(ApiServerUnittest):
config_dict = {
"path": testcase_file_path
}
self.test_runner.init_config(config_dict, "testset")
self.test_runner.init_config(config_dict, "testcase")
test = testcases[2]["test"]
self.test_runner.run_test(test)
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):
test = {
"name": "get token",

80
tests/test_task.py
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)

10
tests/test_utils.py
View File

@@ -210,7 +210,7 @@ class TestUtils(ApiServerUnittest):
{"a": 1},
{"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.assertIn("a", ordered_dict)
@@ -219,7 +219,7 @@ class TestUtils(ApiServerUnittest):
{"a": 1},
{"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}
new_dict = utils.update_ordered_dict(ordered_dict, override_mapping)
self.assertEqual(3, new_dict["a"])
@@ -231,7 +231,7 @@ class TestUtils(ApiServerUnittest):
{"b": 2}
]
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(4, new_dict["c"])
@@ -242,14 +242,14 @@ class TestUtils(ApiServerUnittest):
}
)
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(4, new_dict["c"])
map_list = "invalid"
override_mapping = {"a": 3, "c": 4}
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):
project_path = os.path.join(os.getcwd(), "projectABC")