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

refactor: pipeline

Browse Source
This commit is contained in:
debugtalk
2018-08-21 17:39:12 +08:00
parent 32f78dd2bb
commit 6f3dbacf67
20 changed files with 870 additions and 852 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
httprunner/__about__.py
View File

@@ -1,7 +1,7 @@
__title__ = 'HttpRunner'
__description__ = 'One-stop solution for HTTP(S) testing.'
__url__ = 'https://github.com/HttpRunner/HttpRunner'
__version__ = '1.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

309
httprunner/api.py Normal file
View File

@@ -0,0 +1,309 @@
# 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.
"""
self.kwargs = kwargs
dot_env_path = self.kwargs.pop("dot_env_path", None)
self.project_mapping = self.__loader(dot_env_path)
self.http_client_session = self.kwargs.pop("http_client_session", None)
def __loader(self, dot_env_path=None):
""" load project dependent files, including api/testcase definitions,
environment variables and debugtalk.py module.
Args:
dot_env_path (str): .env file path
Returns:
dict: project dependent info mapping.
{
"debugtalk": {},
"env": {},
"def-api": {},
"def-testcase": {}
}
"""
# load .env
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)
project_mapping = loader.project_mapping
utils.set_os_environ(project_mapping["env"])
return project_mapping
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):
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"]
)
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:
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)

2
httprunner/context.py
View File

@@ -365,7 +365,7 @@ class Context(object):
})
"""
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)

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)

355
httprunner/task.py
View File

@@ -1,355 +0,0 @@
# encoding: utf-8
import copy
import os
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:
# testcase 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)
# teststep 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 teststep 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 teststep 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 variables 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)
self.project_mapping = self.loader(dot_env_path)
kwargs.setdefault("resultclass", HtmlTestResult)
self.runner = unittest.TextTestRunner(**kwargs)
def loader(self, dot_env_path=None):
""" load project files, including api/testcase definitions, testcases,
environment variables and debugtalk.py module.
Args:
dot_env_path (str): .env file path
Returns:
dict: project tests info mapping.
"""
# load .env
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)
project_mapping = loader.project_mapping
utils.set_os_environ(project_mapping["env"])
return project_mapping
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
"""
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>

170
httprunner/utils.py
View File

@@ -158,25 +158,33 @@ 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.
@@ -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,14 +255,82 @@ 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.context.evaluated_validators
test_runner.http_client_session.init_meta_data()
if is_py2:
test.__func__.__doc__ = teststep_dict["name"]
else:
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 = ""
@@ -234,21 +342,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 +440,7 @@ def validate_json_file(file_list):
print("OK")
def prettify_json_file(file_list):
""" prettify JSON testset format
"""
@@ -362,6 +467,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()

2
httprunner/validator.py
View File

@@ -39,7 +39,7 @@ def is_testcase(data_structure):
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

361
tests/test_api.py Normal file
View File

@@ -0,0 +1,361 @@
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_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]["in_out"]["out"])
self.assertIn("user_agent", summary["details"][0]["in_out"]["in"])
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]["in_out"]["out"])
self.assertEqual(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.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"])
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)

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)

16
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):
@@ -476,16 +477,3 @@ class TestSuiteLoader(unittest.TestCase):
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"])

187
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
@@ -13,20 +13,20 @@ class TestRunner(ApiServerUnittest):
self.test_runner = runner.Runner()
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 = {
@@ -152,110 +152,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",
@@ -322,62 +218,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')
@@ -403,15 +243,6 @@ class TestRunner(ApiServerUnittest):
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")