mirror of
https://github.com/httprunner/httprunner.git
synced 2026-05-12 02:21:29 +08:00
273 lines
8.5 KiB
Python
273 lines
8.5 KiB
Python
# encoding: utf-8
|
|
|
|
import os
|
|
import unittest
|
|
|
|
from httprunner import (exceptions, loader, logger, parser, report, runner,
|
|
utils, validator)
|
|
|
|
|
|
class HttpRunner(object):
|
|
|
|
def __init__(self, failfast=False, save_tests=False, report_template=None, report_dir=None,
|
|
log_level="INFO", log_file=None):
|
|
""" initialize HttpRunner.
|
|
|
|
Args:
|
|
failfast (bool): stop the test run on the first error or failure.
|
|
save_tests (bool): save loaded/parsed tests to JSON file.
|
|
report_template (str): report template file path, template should be in Jinja2 format.
|
|
report_dir (str): html report save directory.
|
|
log_level (str): logging level.
|
|
log_file (str): log file path.
|
|
|
|
"""
|
|
self.exception_stage = "initialize HttpRunner()"
|
|
kwargs = {
|
|
"failfast": failfast,
|
|
"resultclass": report.HtmlTestResult
|
|
}
|
|
self.unittest_runner = unittest.TextTestRunner(**kwargs)
|
|
self.test_loader = unittest.TestLoader()
|
|
self.save_tests = save_tests
|
|
self.report_template = report_template
|
|
self.report_dir = report_dir
|
|
self._summary = None
|
|
if log_file:
|
|
logger.setup_logger(log_level, log_file)
|
|
|
|
def _add_tests(self, tests_mapping):
|
|
""" initialize testcase with Runner() and add to test suite.
|
|
|
|
Args:
|
|
tests_mapping (dict): project info and testcases list.
|
|
|
|
Returns:
|
|
unittest.TestSuite()
|
|
|
|
"""
|
|
def _add_test(test_runner, test_dict):
|
|
""" add test to testcase.
|
|
"""
|
|
def test(self):
|
|
try:
|
|
test_runner.run_test(test_dict)
|
|
except exceptions.MyBaseFailure as ex:
|
|
self.fail(str(ex))
|
|
finally:
|
|
self.meta_datas = test_runner.meta_datas
|
|
|
|
if "config" in test_dict:
|
|
# run nested testcase
|
|
test.__doc__ = test_dict["config"].get("name")
|
|
else:
|
|
# run api test
|
|
test.__doc__ = test_dict.get("name")
|
|
|
|
return test
|
|
|
|
test_suite = unittest.TestSuite()
|
|
functions = tests_mapping.get("project_mapping", {}).get("functions", {})
|
|
|
|
for testcase in tests_mapping["testcases"]:
|
|
config = testcase.get("config", {})
|
|
test_runner = runner.Runner(config, functions)
|
|
TestSequense = type('TestSequense', (unittest.TestCase,), {})
|
|
|
|
tests = testcase.get("teststeps", [])
|
|
for index, test_dict in enumerate(tests):
|
|
for times_index in range(int(test_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 = _add_test(test_runner, test_dict)
|
|
setattr(TestSequense, test_method_name, test_method)
|
|
|
|
loaded_testcase = self.test_loader.loadTestsFromTestCase(TestSequense)
|
|
setattr(loaded_testcase, "config", config)
|
|
setattr(loaded_testcase, "teststeps", tests)
|
|
setattr(loaded_testcase, "runner", test_runner)
|
|
test_suite.addTest(loaded_testcase)
|
|
|
|
return test_suite
|
|
|
|
def _run_suite(self, test_suite):
|
|
""" run tests in test_suite
|
|
|
|
Args:
|
|
test_suite: unittest.TestSuite()
|
|
|
|
Returns:
|
|
list: tests_results
|
|
|
|
"""
|
|
tests_results = []
|
|
|
|
for testcase in test_suite:
|
|
testcase_name = testcase.config.get("name")
|
|
logger.log_info("Start to run testcase: {}".format(testcase_name))
|
|
|
|
result = self.unittest_runner.run(testcase)
|
|
tests_results.append((testcase, result))
|
|
|
|
return tests_results
|
|
|
|
def _aggregate(self, tests_results):
|
|
""" aggregate results
|
|
|
|
Args:
|
|
tests_results (list): list of (testcase, result)
|
|
|
|
"""
|
|
summary = {
|
|
"success": True,
|
|
"stat": {
|
|
"testcases": {
|
|
"total": len(tests_results),
|
|
"success": 0,
|
|
"fail": 0
|
|
},
|
|
"teststeps": {}
|
|
},
|
|
"time": {},
|
|
"platform": report.get_platform(),
|
|
"details": []
|
|
}
|
|
|
|
for tests_result in tests_results:
|
|
testcase, result = tests_result
|
|
testcase_summary = report.get_summary(result)
|
|
|
|
if testcase_summary["success"]:
|
|
summary["stat"]["testcases"]["success"] += 1
|
|
else:
|
|
summary["stat"]["testcases"]["fail"] += 1
|
|
|
|
summary["success"] &= testcase_summary["success"]
|
|
testcase_summary["name"] = testcase.config.get("name")
|
|
testcase_summary["in_out"] = utils.get_testcase_io(testcase)
|
|
|
|
report.aggregate_stat(summary["stat"]["teststeps"], testcase_summary["stat"])
|
|
report.aggregate_stat(summary["time"], testcase_summary["time"])
|
|
|
|
summary["details"].append(testcase_summary)
|
|
|
|
return summary
|
|
|
|
def run_tests(self, tests_mapping):
|
|
""" run testcase/testsuite data
|
|
"""
|
|
# parse tests
|
|
self.exception_stage = "parse tests"
|
|
parsed_tests_mapping = parser.parse_tests(tests_mapping)
|
|
|
|
if self.save_tests:
|
|
utils.dump_tests(parsed_tests_mapping, "parsed")
|
|
|
|
# add tests to test suite
|
|
self.exception_stage = "add tests to test suite"
|
|
test_suite = self._add_tests(parsed_tests_mapping)
|
|
|
|
# run test suite
|
|
self.exception_stage = "run test suite"
|
|
results = self._run_suite(test_suite)
|
|
|
|
# aggregate results
|
|
self.exception_stage = "aggregate results"
|
|
self._summary = self._aggregate(results)
|
|
|
|
# generate html report
|
|
self.exception_stage = "generate html report"
|
|
report.stringify_summary(self._summary)
|
|
|
|
if self.save_tests:
|
|
utils.dump_summary(self._summary, tests_mapping["project_mapping"])
|
|
|
|
report_path = report.render_html_report(
|
|
self._summary,
|
|
self.report_template,
|
|
self.report_dir
|
|
)
|
|
|
|
return report_path
|
|
|
|
def run_path(self, path, dot_env_path=None, mapping=None):
|
|
""" run testcase/testsuite file or folder.
|
|
|
|
Args:
|
|
path (str): testcase/testsuite file/foler path.
|
|
dot_env_path (str): specified .env file path.
|
|
mapping (dict): if mapping is specified, it will override variables in config block.
|
|
|
|
Returns:
|
|
instance: HttpRunner() instance
|
|
|
|
"""
|
|
# load tests
|
|
self.exception_stage = "load tests"
|
|
tests_mapping = loader.load_tests(path, dot_env_path)
|
|
tests_mapping["project_mapping"]["test_path"] = path
|
|
|
|
if mapping:
|
|
tests_mapping["project_mapping"]["variables"] = mapping
|
|
|
|
if self.save_tests:
|
|
utils.dump_tests(tests_mapping, "loaded")
|
|
|
|
return self.run_tests(tests_mapping)
|
|
|
|
def run(self, path_or_tests, dot_env_path=None, mapping=None):
|
|
""" main interface.
|
|
|
|
Args:
|
|
path_or_tests:
|
|
str: testcase/testsuite file/foler path
|
|
dict: valid testcase/testsuite data
|
|
|
|
"""
|
|
if validator.is_testcase_path(path_or_tests):
|
|
return self.run_path(path_or_tests, dot_env_path, mapping)
|
|
elif validator.is_testcases(path_or_tests):
|
|
return self.run_tests(path_or_tests)
|
|
else:
|
|
raise exceptions.ParamsError("Invalid testcase path or testcases: {}".format(path_or_tests))
|
|
|
|
@property
|
|
def summary(self):
|
|
""" get test reuslt summary.
|
|
"""
|
|
return self._summary
|
|
|
|
|
|
def prepare_locust_tests(path):
|
|
""" prepare locust testcases
|
|
|
|
Args:
|
|
path (str): testcase file path.
|
|
|
|
Returns:
|
|
dict: locust tests data
|
|
|
|
{
|
|
"functions": {},
|
|
"tests": []
|
|
}
|
|
|
|
"""
|
|
tests_mapping = loader.load_tests(path)
|
|
parsed_tests_mapping = parser.parse_tests(tests_mapping)
|
|
|
|
functions = parsed_tests_mapping.get("project_mapping", {}).get("functions", {})
|
|
|
|
tests = []
|
|
|
|
for testcase in parsed_tests_mapping["testcases"]:
|
|
testcase_weight = testcase.get("config", {}).pop("weight", 1)
|
|
for _ in range(testcase_weight):
|
|
tests.append(testcase)
|
|
|
|
return {
|
|
"functions": functions,
|
|
"tests": tests
|
|
}
|