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

refactor: make test runner unified

Browse Source
This commit is contained in:
debugtalk
2018-02-14 20:28:28 +08:00
parent 39ca2cdaa8
commit a0ace0ee76
6 changed files with 161 additions and 182 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
httprunner/__init__.py
View File

@@ -1 +1 @@
__version__ = '0.9.0'
__version__ = '0.9.1'

88
httprunner/cli.py
View File

@@ -3,15 +3,42 @@ import logging
import multiprocessing
import os
import sys
import unittest
from collections import OrderedDict
from httprunner import __version__ as ate_version
from httprunner import exception
from httprunner.task import TaskSuite
from httprunner.utils import create_scaffold, string_type
from httprunner import __version__ as hrun_version
from httprunner.utils import create_scaffold, print_output, string_type
from pyunitreport import __version__ as pyu_version
from pyunitreport import HTMLTestRunner
from .exception import TestcaseNotFound
from .task import Result, TaskSuite
def run_suite_path(path, mapping=None, runner=None):
""" run suite with YAML/JSON file path
@params:
- path: testset path
- mapping: passed in variables mapping, it will override variables in config block
- runner: HTMLTestRunner() or TextTestRunner()
"""
try:
mapping = mapping or {}
task_suite = TaskSuite(path, mapping)
except TestcaseNotFound:
sys.exit(1)
test_runner = runner or unittest.TextTestRunner()
result = test_runner.run(task_suite)
output = {}
for task in task_suite.tasks:
output.update(task.output)
return Result(
result.wasSuccessful(),
output
)
def main_hrun():
""" API test: parse command line options and run commands.
@@ -27,9 +54,6 @@ def main_hrun():
parser.add_argument(
'--log-level', default='INFO',
help="Specify logging level, default is INFO.")
parser.add_argument(
'--report-name',
help="Specify report name, default is generated time.")
parser.add_argument(
'--failfast', action='store_true', default=False,
help="Stop the test run on the first error or failure.")
@@ -40,7 +64,7 @@ def main_hrun():
args = parser.parse_args()
if args.version:
print("HttpRunner version: {}".format(ate_version))
print("HttpRunner version: {}".format(hrun_version))
print("PyUnitReport version: {}".format(pyu_version))
exit(0)
@@ -53,47 +77,15 @@ def main_hrun():
create_scaffold(project_path)
exit(0)
report_name = args.report_name
if report_name and len(args.testset_paths) > 1:
report_name = None
logging.warning("More than one testset paths specified, \
report name is ignored, use generated time instead.")
kwargs = {
"output": os.path.join(os.getcwd(), "reports"),
"failfast": args.failfast
}
test_runner = HTMLTestRunner(**kwargs)
result = run_suite_path(args.testset_paths, {}, test_runner)
print_output(result.output)
results = {}
success = True
for testset_path in set(args.testset_paths):
testset_path = testset_path.rstrip('/')
try:
task_suite = TaskSuite(testset_path)
except exception.TestcaseNotFound:
success = False
continue
output_folder_name = os.path.basename(os.path.splitext(testset_path)[0])
kwargs = {
"output": output_folder_name,
"report_name": report_name,
"failfast": args.failfast
}
result = HTMLTestRunner(**kwargs).run(task_suite)
results[testset_path] = OrderedDict({
"total": result.testsRun,
"successes": len(result.successes),
"failures": len(result.failures),
"errors": len(result.errors),
"skipped": len(result.skipped)
})
if len(result.successes) != result.testsRun:
success = False
for task in task_suite.tasks:
task.print_output()
return 0 if success is True else 1
return 0 if result.success else 1
def main_locust():
""" Performance test with locust: parse command line options and run commands.

7
httprunner/locustfile_template
View File

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

105
httprunner/runner.py
View File

@@ -8,11 +8,13 @@ from httprunner.context import Context
class Runner(object):
def __init__(self, http_client_session=None, request_failure_hook=None):
def __init__(self, config_dict=None, http_client_session=None):
self.http_client_session = http_client_session
self.context = Context()
testcase.load_test_dependencies()
self.request_failure_hook = request_failure_hook
config_dict = config_dict or {}
self.init_config(config_dict, "testset")
def init_config(self, config_dict, level):
""" create/update context variables binds
@@ -166,101 +168,8 @@ class Runner(object):
return True
def _run_testset(self, testset, variables_mapping=None):
""" run single testset, including one or several testcases.
@param
(dict) testset
{
"name": "testset description",
"config": {
"name": "testset description",
"requires": [],
"function_binds": {},
"variables": [],
"request": {}
},
"testcases": [
{
"name": "testcase description",
"variables": [], # optional, override
"request": {},
"extract": {}, # optional
"validate": {} # optional
},
testcase12
]
}
(dict) variables_mapping:
passed in variables mapping, it will override variables in config block
@return (dict) test result of testset
{
"success": True,
"output": {} # variables mapping
}
"""
success = True
config_dict = testset.get("config", {})
variables = config_dict.get("variables", [])
variables_mapping = variables_mapping or {}
config_dict["variables"] = utils.override_variables_binds(variables, variables_mapping)
self.init_config(config_dict, level="testset")
testcases = testset.get("testcases", [])
for testcase_dict in testcases:
try:
self._run_test(testcase_dict)
except exception.MyBaseError as ex:
success = False
if self.request_failure_hook:
self.request_failure_hook.fire(
request_type=testcase_dict.get("request", {}).get("method"),
name=testcase_dict.get("request", {}).get("url"),
response_time=0,
exception=ex
)
else:
logging.exception(
"Exception occured in testcase: {}".format(testcase_dict.get("name")))
break
output_variables_list = config_dict.get("output", [])
return {
"success": success,
"output": self.generate_output(output_variables_list)
}
def run(self, path, mapping=None):
""" run specified testset path or folder path.
@param
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)
(dict) mapping:
passed in variables mapping, it will override variables in config block
"""
success = True
mapping = mapping or {}
output = {}
testsets = testcase.load_testcases_by_path(path)
for testset in testsets:
try:
result = self._run_testset(testset, mapping)
assert result["success"]
output.update(result["output"])
except AssertionError:
success = False
return {
"success": success,
"output": output
}
def generate_output(self, output_variables_list):
""" generate and print output
def extract_output(self, output_variables_list):
""" extract output variables
"""
variables_mapping = self.context.get_testcase_variables_mapping()
@@ -275,6 +184,4 @@ class Runner(object):
output[variable] = variables_mapping[variable]
utils.print_output(output)
return output

99
httprunner/task.py
View File

@@ -20,12 +20,43 @@ class ApiTestCase(unittest.TestCase):
class ApiTestSuite(unittest.TestSuite):
""" create test suite with a testset, it may include one or several testcases.
each suite should initialize a separate Runner() with testset config.
@param
(dict) testset
{
"name": "testset description",
"config": {
"name": "testset description",
"requires": [],
"function_binds": {},
"variables": [],
"request": {}
},
"testcases": [
{
"name": "testcase description",
"variables": [], # optional, override
"request": {},
"extract": {}, # optional
"validate": {} # optional
},
testcase12
]
}
(dict) variables_mapping:
passed in variables mapping, it will override variables in config block
@return (instance) test result of testset
Result(success, output)
"""
def __init__(self, testset):
def __init__(self, testset, variables_mapping=None, http_client_session=None):
super(ApiTestSuite, self).__init__()
self.test_runner = runner.Runner()
self.config_dict = testset.get("config", {})
self.test_runner.init_config(self.config_dict, level="testset")
variables = self.config_dict.get("variables", [])
variables_mapping = variables_mapping or {}
self.config_dict["variables"] = utils.override_variables_binds(variables, variables_mapping)
self.test_runner = runner.Runner(self.config_dict, http_client_session)
testcases = testset.get("testcases", [])
self._add_tests_to_suite(testcases)
@@ -39,26 +70,76 @@ class ApiTestSuite(unittest.TestSuite):
test = ApiTestCase(self.test_runner, testcase_dict)
[self.addTest(test) for _ in range(int(testcase_dict.get("times", 1)))]
def print_output(self):
@property
def output(self):
output_variables_list = self.config_dict.get("output", [])
self.test_runner.generate_output(output_variables_list)
return self.test_runner.extract_output(output_variables_list)
class TaskSuite(unittest.TestSuite):
""" create test task suite with specified testcase path.
each task suite may include one or several test suite.
"""
def __init__(self, testcase_path):
def __init__(self, path, mapping=None, http_client_session=None):
"""
@params
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)
(dict) mapping:
passed in variables mapping, it will override variables in config block
"""
super(TaskSuite, self).__init__()
self.suite_list = []
testsets = testcase.load_testcases_by_path(testcase_path)
mapping = mapping or {}
if not isinstance(path, list):
# absolute/relative file/folder path
path = [path]
# remove duplicate path
path = set(path)
testsets = testcase.load_testcases_by_path(path)
if not testsets:
raise exception.TestcaseNotFound
self.suite_list = []
for testset in testsets:
suite = ApiTestSuite(testset)
suite = ApiTestSuite(testset, mapping, http_client_session)
self.addTest(suite)
self.suite_list.append(suite)
@property
def tasks(self):
return self.suite_list
class Result(object):
def __init__(self, success, output):
self.success = success
self.output = output
class LocustTask(object):
def __init__(self, path, locust_client, mapping=None):
mapping = mapping or {}
self.task_suite = TaskSuite(path, mapping, locust_client)
def run(self):
for suite in self.task_suite:
for test in suite:
try:
test.runTest()
except exception.MyBaseError as ex:
try:
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
)
except ImportError:
logging.exception(
"Exception occured in testcase: {}".format(test.testcase_dict.get("name")))

42
tests/test_runner.py
View File

@@ -2,6 +2,7 @@ import os
import time
from httprunner import exception, runner, testcase, utils
from httprunner.cli import run_suite_path
from tests.base import ApiServerUnittest
@@ -77,50 +78,49 @@ class TestRunner(ApiServerUnittest):
def test_run_testset_hardcode(self):
for testcase_file_path in self.testcase_file_path_list:
result = self.test_runner.run(testcase_file_path)
self.assertTrue(result["success"])
result = run_suite_path(testcase_file_path)
self.assertTrue(result.success)
def test_run_testsets_hardcode(self):
for testcase_file_path in self.testcase_file_path_list:
result = self.test_runner.run(testcase_file_path)
self.assertTrue(result["success"])
result = run_suite_path(self.testcase_file_path_list)
self.assertTrue(result.success)
def test_run_testset_template_variables(self):
testcase_file_path = os.path.join(
os.getcwd(), 'tests/data/demo_testset_variables.yml')
result = self.test_runner.run(testcase_file_path)
self.assertTrue(result["success"])
result = run_suite_path(testcase_file_path)
self.assertTrue(result.success)
def test_run_testset_template_import_functions(self):
testcase_file_path = os.path.join(
os.getcwd(), 'tests/data/demo_testset_template_import_functions.yml')
result = self.test_runner.run(testcase_file_path)
self.assertTrue(result["success"])
result = run_suite_path(testcase_file_path)
self.assertTrue(result.success)
def test_run_testsets_template_import_functions(self):
testcase_file_path = os.path.join(
os.getcwd(), 'tests/data/demo_testset_template_import_functions.yml')
result = self.test_runner.run(testcase_file_path)
self.assertTrue(result["success"])
result = run_suite_path(testcase_file_path)
self.assertTrue(result.success)
def test_run_testsets_template_lambda_functions(self):
testcase_file_path = os.path.join(
os.getcwd(), 'tests/data/demo_testset_template_lambda_functions.yml')
result = self.test_runner.run(testcase_file_path)
self.assertTrue(result["success"])
result = run_suite_path(testcase_file_path)
self.assertTrue(result.success)
def test_run_testset_layered(self):
testcase_file_path = os.path.join(
os.getcwd(), 'tests/data/demo_testset_layer.yml')
result = self.test_runner.run(testcase_file_path)
self.assertTrue(result["success"])
result = run_suite_path(testcase_file_path)
self.assertTrue(result.success)
def test_run_testset_output(self):
testcase_file_path = os.path.join(
os.getcwd(), 'tests/data/demo_testset_layer.yml')
result = self.test_runner.run(testcase_file_path)
self.assertTrue(result["success"])
self.assertIn("token", result["output"])
result = run_suite_path(testcase_file_path)
self.assertTrue(result.success)
self.assertIn("token", result.output)
def test_run_testset_with_variables_mapping(self):
testcase_file_path = os.path.join(
@@ -128,9 +128,9 @@ class TestRunner(ApiServerUnittest):
variables_mapping = {
"app_version": '2.9.7'
}
result = self.test_runner.run(testcase_file_path, variables_mapping)
self.assertTrue(result["success"])
self.assertIn("token", result["output"])
result = run_suite_path(testcase_file_path, variables_mapping)
self.assertTrue(result.success)
self.assertIn("token", result.output)
def test_run_testcase_with_empty_header(self):
testcase_file_path = os.path.join(