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

bugfix: adjust functions location to avoid cross reference

Browse Source
This commit is contained in:
debugtalk
2017-09-18 16:01:19 +08:00
parent cd20b0a8dc
commit 611acf528a
9 changed files with 251 additions and 243 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
ate/locustfile_template
View File

@@ -2,7 +2,7 @@
import zmq
import os
from locust import HttpLocust, TaskSet, task
from ate import utils, runner, exception
from ate import testcase, runner, exception
class WebPageTasks(TaskSet):
def on_start(self):
@@ -22,5 +22,5 @@ class WebPageUser(HttpLocust):
min_wait = 1000
max_wait = 5000
testsets = utils.load_testcases_by_path("$TESTCASE_FILE")
testsets = testcase.load_testcases_by_path("$TESTCASE_FILE")
testset = testsets[0]

2
ate/locusts.py
View File

@@ -3,7 +3,7 @@ import multiprocessing
import os
import sys
from ate.utils import load_testcases_by_path
from ate.testcase import load_testcases_by_path
from locust.main import main

4
ate/task.py
View File

@@ -1,6 +1,6 @@
import unittest
from ate import runner, utils
from ate import runner, testcase, utils
class ApiTestCase(unittest.TestCase):
@@ -43,7 +43,7 @@ def create_task(testcase_path):
each task suite may include one or several test suite.
"""
task_suite = unittest.TestSuite()
testsets = utils.load_testcases_by_path(testcase_path)
testsets = testcase.load_testcases_by_path(testcase_path)
for testset in testsets:
suite = create_suite(testset)

117
ate/testcase.py
View File

@@ -7,6 +7,7 @@ from ate import exception, utils
variable_regexp = r"\$([\w_]+)"
function_regexp = r"\$\{([\w_]+\([\$\w_ =,]*\))\}"
function_regexp_compile = re.compile(r"^([\w_]+)\(([\$\w_ =,]*)\)$")
api_overall_dict = {}
def extract_variables(content):
@@ -87,6 +88,122 @@ def parse_function(content):
return function_meta
def load_testcases_by_path(path):
""" load testcases from file 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)
@return testcase sets list, each testset is corresponding to a file
[
{"name": "desc1", "config": {}, "testcases": [testcase11, testcase12]},
{"name": "desc2", "config": {}, "testcases": [testcase21, testcase22, testcase23]},
]
"""
if isinstance(path, (list, set)):
testsets_list = []
for file_path in set(path):
_testsets_list = load_testcases_by_path(file_path)
testsets_list.extend(_testsets_list)
return testsets_list
if not os.path.isabs(path):
path = os.path.join(os.getcwd(), path)
if os.path.isdir(path):
files_list = utils.load_folder_files(path, file_type="test", recursive=True)
return load_testcases_by_path(files_list)
elif os.path.isfile(path):
testset = {
"name": "",
"config": {
"path": path
},
"testcases": []
}
testcases_list = utils.load_testcases(path)
dir_path = os.path.dirname(os.path.abspath(path))
for item in testcases_list:
for key in item:
if key == "config":
testset["config"].update(item["config"])
testset["name"] = item["config"].get("name", "")
elif key == "test":
test_dict = item["test"]
if "api" in test_dict:
update_test_info(test_dict, dir_path)
testset["testcases"].append(test_dict)
return [testset] if testset["testcases"] else []
else:
return []
def update_test_info(test_dict, dir_path):
api_call = test_dict["api"]
function_meta = parse_function(api_call)
func_name = function_meta["func_name"]
api_info = get_api_definition(func_name, dir_path)
test_dict.update(api_info)
def get_api_definition(name, dir_path):
""" get expected api from dir_path upward recursively
@param
name: api name
dir_path: start search dir path
@return
expected api info if found, otherwise raise ApiNotFound exception
"""
api_dir_dict = api_overall_dict.get(dir_path)
if not api_dir_dict:
api_dir_dict = load_api_definition(dir_path)
api_overall_dict[dir_path] = api_dir_dict
api_info = api_dir_dict.get(name)
if api_info:
return api_info
parent_dir_path = os.path.dirname(dir_path)
if dir_path == parent_dir_path:
# system root path
err_msg = "{} not found in recursive upward path!".format(name)
raise exception.ApiNotFound(err_msg)
return get_api_definition(name, parent_dir_path)
def load_api_definition(dir_path):
""" load all api definitions in specified dir path
@param (str) dir_path
@return (dict) all api definitions in dir_path merged in one dict
"""
api_files = utils.load_folder_files(dir_path, file_type="api", recursive=False)
api_def_list = []
for api_file in api_files:
api_def_list.extend(utils.load_testcases(api_file))
api_dir_dict = {}
for item in api_def_list:
for key in item:
if key == "api":
api_def = item["api"].pop("def")
function_meta = parse_function(api_def)
func_name = function_meta["func_name"]
api_info = {}
api_info["function_meta"] = function_meta
api_info.update(item["api"])
api_dir_dict[func_name] = api_info
return api_dir_dict
class TestcaseParser(object):

120
ate/utils.py
View File

@@ -11,7 +11,7 @@ import string
import types
import yaml
from ate import exception, testcase
from ate import exception
from requests.structures import CaseInsensitiveDict
try:
@@ -24,7 +24,7 @@ except NameError:
PYTHON_VERSION = 3
SECRET_KEY = "DebugTalk"
api_overall_dict = {}
def gen_random_string(str_len):
return ''.join(
@@ -88,122 +88,6 @@ def load_folder_files(folder_path, file_type, recursive=False):
return file_list
def load_testcases_by_path(path):
""" load testcases from file 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)
@return testcase sets list, each testset is corresponding to a file
[
{"name": "desc1", "config": {}, "testcases": [testcase11, testcase12]},
{"name": "desc2", "config": {}, "testcases": [testcase21, testcase22, testcase23]},
]
"""
if isinstance(path, (list, set)):
testsets_list = []
for file_path in set(path):
_testsets_list = load_testcases_by_path(file_path)
testsets_list.extend(_testsets_list)
return testsets_list
if not os.path.isabs(path):
path = os.path.join(os.getcwd(), path)
if os.path.isdir(path):
files_list = load_folder_files(path, file_type="test", recursive=True)
return load_testcases_by_path(files_list)
elif os.path.isfile(path):
testset = {
"name": "",
"config": {
"path": path
},
"testcases": []
}
testcases_list = load_testcases(path)
dir_path = os.path.dirname(os.path.abspath(path))
for item in testcases_list:
for key in item:
if key == "config":
testset["config"].update(item["config"])
testset["name"] = item["config"].get("name", "")
elif key == "test":
test_dict = item["test"]
if "api" in test_dict:
update_test_info(test_dict, dir_path)
testset["testcases"].append(test_dict)
return [testset] if testset["testcases"] else []
else:
return []
def update_test_info(test_dict, dir_path):
api_call = test_dict["api"]
function_meta = testcase.parse_function(api_call)
func_name = function_meta["func_name"]
api_info = get_api_definition(func_name, dir_path)
test_dict.update(api_info)
def get_api_definition(name, dir_path):
""" get expected api from dir_path upward recursively
@param
name: api name
dir_path: start search dir path
@return
expected api info if found, otherwise raise ApiNotFound exception
"""
api_dir_dict = api_overall_dict.get(dir_path)
if not api_dir_dict:
api_dir_dict = load_api_definition(dir_path)
api_overall_dict[dir_path] = api_dir_dict
api_info = api_dir_dict.get(name)
if api_info:
return api_info
parent_dir_path = os.path.dirname(dir_path)
if dir_path == parent_dir_path:
# system root path
err_msg = "{} not found in recursive upward path!".format(name)
raise exception.ApiNotFound(err_msg)
return get_api_definition(name, parent_dir_path)
def load_api_definition(dir_path):
""" load all api definitions in specified dir path
@param (str) dir_path
@return (dict) all api definitions in dir_path merged in one dict
"""
api_files = load_folder_files(dir_path, file_type="api", recursive=False)
api_def_list = []
for api_file in api_files:
api_def_list.extend(load_testcases(api_file))
api_dir_dict = {}
for item in api_def_list:
for key in item:
if key == "api":
api_def = item["api"].pop("def")
function_meta = testcase.parse_function(api_def)
func_name = function_meta["func_name"]
api_info = {}
api_info["function_meta"] = function_meta
api_info.update(item["api"])
api_dir_dict[func_name] = api_info
return api_dir_dict
def query_json(json_content, query, delimiter='.'):
""" Do an xpath-like query with json_content.
@param (json_content) json_content

19
tests/test_runner.py
View File

@@ -1,8 +1,11 @@
import os
import requests
from ate import runner, exception, utils
from ate import exception, runner, utils
from ate.testcase import load_testcases_by_path
from tests.base import ApiServerUnittest
class TestRunner(ApiServerUnittest):
def setUp(self):
@@ -64,14 +67,14 @@ class TestRunner(ApiServerUnittest):
def test_run_testset_hardcode(self):
for testcase_file_path in self.testcase_file_path_list:
testsets = utils.load_testcases_by_path(testcase_file_path)
testsets = load_testcases_by_path(testcase_file_path)
results = self.test_runner.run_testset(testsets[0])
self.assertEqual(len(results), 3)
self.assertEqual(results, [True] * 3)
def test_run_testsets_hardcode(self):
for testcase_file_path in self.testcase_file_path_list:
testsets = utils.load_testcases_by_path(testcase_file_path)
testsets = load_testcases_by_path(testcase_file_path)
results = self.test_runner.run_testsets(testsets)
self.assertEqual(len(results), 1)
self.assertEqual(results, [[True] * 3])
@@ -79,7 +82,7 @@ class TestRunner(ApiServerUnittest):
def test_run_testset_template_variables(self):
testcase_file_path = os.path.join(
os.getcwd(), 'tests/data/demo_testset_variables.yml')
testsets = utils.load_testcases_by_path(testcase_file_path)
testsets = load_testcases_by_path(testcase_file_path)
results = self.test_runner.run_testset(testsets[0])
self.assertEqual(len(results), 3)
self.assertEqual(results, [True] * 3)
@@ -87,7 +90,7 @@ class TestRunner(ApiServerUnittest):
def test_run_testset_template_import_functions(self):
testcase_file_path = os.path.join(
os.getcwd(), 'tests/data/demo_testset_template_import_functions.yml')
testsets = utils.load_testcases_by_path(testcase_file_path)
testsets = load_testcases_by_path(testcase_file_path)
results = self.test_runner.run_testset(testsets[0])
self.assertEqual(len(results), 3)
self.assertEqual(results, [True] * 3)
@@ -95,7 +98,7 @@ class TestRunner(ApiServerUnittest):
def test_run_testsets_template_import_functions(self):
testcase_file_path = os.path.join(
os.getcwd(), 'tests/data/demo_testset_template_import_functions.yml')
testsets = utils.load_testcases_by_path(testcase_file_path)
testsets = load_testcases_by_path(testcase_file_path)
results = self.test_runner.run_testsets(testsets)
self.assertEqual(len(results), 1)
self.assertEqual(results, [[True] * 3])
@@ -103,7 +106,7 @@ class TestRunner(ApiServerUnittest):
def test_run_testsets_template_lambda_functions(self):
testcase_file_path = os.path.join(
os.getcwd(), 'tests/data/demo_testset_template_lambda_functions.yml')
testsets = utils.load_testcases_by_path(testcase_file_path)
testsets = load_testcases_by_path(testcase_file_path)
results = self.test_runner.run_testsets(testsets)
self.assertEqual(len(results), 1)
self.assertEqual(results, [[True] * 3])
@@ -111,7 +114,7 @@ class TestRunner(ApiServerUnittest):
def test_run_testset_layered(self):
testcase_file_path = os.path.join(
os.getcwd(), 'tests/data/demo_testset_layer.yml')
testsets = utils.load_testcases_by_path(testcase_file_path)
testsets = load_testcases_by_path(testcase_file_path)
results = self.test_runner.run_testsets(testsets)
self.assertEqual(len(results), 1)
self.assertEqual(results, [[True] * 3])

7
tests/test_task.py
View File

@@ -1,6 +1,9 @@
import os
from ate import task
from ate.testcase import load_testcases_by_path
from tests.base import ApiServerUnittest
from ate import task, utils
class TestTask(ApiServerUnittest):
@@ -14,7 +17,7 @@ class TestTask(ApiServerUnittest):
def test_create_suite(self):
testcase_file_path = os.path.join(os.getcwd(), 'tests/data/demo_testset_variables.yml')
testsets = utils.load_testcases_by_path(testcase_file_path)
testsets = load_testcases_by_path(testcase_file_path)
suite = task.create_suite(testsets[0])
self.assertEqual(suite.countTestCases(), 3)
for testcase in suite:

112
tests/test_testcase.py
View File

@@ -1,8 +1,9 @@
import os
import time
import unittest
from ate import testcase
from ate.exception import ParamsError
from ate.exception import ParamsError, ApiNotFound
class TestcaseParserUnittest(unittest.TestCase):
@@ -332,3 +333,112 @@ class TestcaseParserUnittest(unittest.TestCase):
parsed_testcase["headers"]["sum"],
3
)
def test_load_testcases_by_path_files(self):
testsets_list = []
# absolute file path
path = os.path.join(
os.getcwd(), 'tests/data/demo_testset_hardcode.json')
testset_list = testcase.load_testcases_by_path(path)
self.assertEqual(len(testset_list), 1)
self.assertIn("path", testset_list[0]["config"])
self.assertEqual(testset_list[0]["config"]["path"], path)
self.assertEqual(len(testset_list[0]["testcases"]), 3)
testsets_list.extend(testset_list)
# relative file path
path = 'tests/data/demo_testset_hardcode.yml'
testset_list = testcase.load_testcases_by_path(path)
self.assertEqual(len(testset_list), 1)
self.assertIn("path", testset_list[0]["config"])
self.assertIn(path, testset_list[0]["config"]["path"])
self.assertEqual(len(testset_list[0]["testcases"]), 3)
testsets_list.extend(testset_list)
# list/set container with file(s)
path = [
os.path.join(os.getcwd(), 'tests/data/demo_testset_hardcode.json'),
'tests/data/demo_testset_hardcode.yml'
]
testset_list = testcase.load_testcases_by_path(path)
self.assertEqual(len(testset_list), 2)
self.assertEqual(len(testset_list[0]["testcases"]), 3)
self.assertEqual(len(testset_list[1]["testcases"]), 3)
testsets_list.extend(testset_list)
self.assertEqual(len(testsets_list), 4)
for testset in testsets_list:
for test in testset["testcases"]:
self.assertIn('name', test)
self.assertIn('request', test)
self.assertIn('url', test['request'])
self.assertIn('method', test['request'])
def test_load_testcases_by_path_folder(self):
# absolute folder path
path = os.path.join(os.getcwd(), 'tests/data')
testset_list_1 = testcase.load_testcases_by_path(path)
self.assertGreater(len(testset_list_1), 4)
# relative folder path
path = 'tests/data/'
testset_list_2 = testcase.load_testcases_by_path(path)
self.assertEqual(len(testset_list_1), len(testset_list_2))
# list/set container with file(s)
path = [
os.path.join(os.getcwd(), 'tests/data'),
'tests/data/'
]
testset_list_3 = testcase.load_testcases_by_path(path)
self.assertEqual(len(testset_list_3), 2 * len(testset_list_1))
def test_load_testcases_by_path_not_exist(self):
# absolute folder path
path = os.path.join(os.getcwd(), 'tests/data_not_exist')
testset_list_1 = testcase.load_testcases_by_path(path)
self.assertEqual(testset_list_1, [])
# relative folder path
path = 'tests/data_not_exist'
testset_list_2 = testcase.load_testcases_by_path(path)
self.assertEqual(testset_list_2, [])
# list/set container with file(s)
path = [
os.path.join(os.getcwd(), 'tests/data_not_exist'),
'tests/data_not_exist/'
]
testset_list_3 = testcase.load_testcases_by_path(path)
self.assertEqual(testset_list_3, [])
def test_load_testcases_by_path_layered(self):
path = os.path.join(
os.getcwd(), 'tests/data/demo_testset_layer.yml')
testsets_list = testcase.load_testcases_by_path(path)
self.assertIn("variable_binds", testsets_list[0]["config"])
self.assertIn("request", testsets_list[0]["config"])
print(testsets_list[0]["testcases"][0])
self.assertIn("request", testsets_list[0]["testcases"][0])
self.assertIn("url", testsets_list[0]["testcases"][0]["request"])
self.assertIn("validators", testsets_list[0]["testcases"][0])
def test_load_api_definition(self):
path = os.path.join(
os.getcwd(), 'tests/data')
api_dir_dict = testcase.load_api_definition(path)
self.assertIn("get_token", api_dir_dict)
self.assertEqual("/api/get-token", api_dir_dict["get_token"]["request"]["url"])
self.assertIn("$user_name", api_dir_dict["get_token"]["function_meta"]["args"])
self.assertIn("create_user", api_dir_dict)
def test_get_api_definition(self):
path = os.path.join(
os.getcwd(), 'tests/data')
api_info = testcase.get_api_definition("get_token", path)
self.assertEqual("/api/get-token", api_info["request"]["url"])
self.assertIn(path, testcase.api_overall_dict)
with self.assertRaises(ApiNotFound):
testcase.get_api_definition("api_not_exist", path)

109
tests/test_utils.py
View File

@@ -47,115 +47,6 @@ class TestUtils(ApiServerUnittest):
api_file = os.path.join(os.getcwd(), 'tests', 'data', 'api.yml')
self.assertEqual(files[0], api_file)
def test_load_testcases_by_path_files(self):
testsets_list = []
# absolute file path
path = os.path.join(
os.getcwd(), 'tests/data/demo_testset_hardcode.json')
testset_list = utils.load_testcases_by_path(path)
self.assertEqual(len(testset_list), 1)
self.assertIn("path", testset_list[0]["config"])
self.assertEqual(testset_list[0]["config"]["path"], path)
self.assertEqual(len(testset_list[0]["testcases"]), 3)
testsets_list.extend(testset_list)
# relative file path
path = 'tests/data/demo_testset_hardcode.yml'
testset_list = utils.load_testcases_by_path(path)
self.assertEqual(len(testset_list), 1)
self.assertIn("path", testset_list[0]["config"])
self.assertIn(path, testset_list[0]["config"]["path"])
self.assertEqual(len(testset_list[0]["testcases"]), 3)
testsets_list.extend(testset_list)
# list/set container with file(s)
path = [
os.path.join(os.getcwd(), 'tests/data/demo_testset_hardcode.json'),
'tests/data/demo_testset_hardcode.yml'
]
testset_list = utils.load_testcases_by_path(path)
self.assertEqual(len(testset_list), 2)
self.assertEqual(len(testset_list[0]["testcases"]), 3)
self.assertEqual(len(testset_list[1]["testcases"]), 3)
testsets_list.extend(testset_list)
self.assertEqual(len(testsets_list), 4)
for testset in testsets_list:
for testcase in testset["testcases"]:
self.assertIn('name', testcase)
self.assertIn('request', testcase)
self.assertIn('url', testcase['request'])
self.assertIn('method', testcase['request'])
def test_load_testcases_by_path_folder(self):
# absolute folder path
path = os.path.join(os.getcwd(), 'tests/data')
testset_list_1 = utils.load_testcases_by_path(path)
self.assertGreater(len(testset_list_1), 4)
# relative folder path
path = 'tests/data/'
testset_list_2 = utils.load_testcases_by_path(path)
self.assertEqual(len(testset_list_1), len(testset_list_2))
# list/set container with file(s)
path = [
os.path.join(os.getcwd(), 'tests/data'),
'tests/data/'
]
testset_list_3 = utils.load_testcases_by_path(path)
self.assertEqual(len(testset_list_3), 2 * len(testset_list_1))
def test_load_testcases_by_path_not_exist(self):
# absolute folder path
path = os.path.join(os.getcwd(), 'tests/data_not_exist')
testset_list_1 = utils.load_testcases_by_path(path)
self.assertEqual(testset_list_1, [])
# relative folder path
path = 'tests/data_not_exist'
testset_list_2 = utils.load_testcases_by_path(path)
self.assertEqual(testset_list_2, [])
# list/set container with file(s)
path = [
os.path.join(os.getcwd(), 'tests/data_not_exist'),
'tests/data_not_exist/'
]
testset_list_3 = utils.load_testcases_by_path(path)
self.assertEqual(testset_list_3, [])
def test_load_testcases_by_path_layered(self):
path = os.path.join(
os.getcwd(), 'tests/data/demo_testset_layer.yml')
testsets_list = utils.load_testcases_by_path(path)
self.assertIn("variable_binds", testsets_list[0]["config"])
self.assertIn("request", testsets_list[0]["config"])
print(testsets_list[0]["testcases"][0])
self.assertIn("request", testsets_list[0]["testcases"][0])
self.assertIn("url", testsets_list[0]["testcases"][0]["request"])
self.assertIn("validators", testsets_list[0]["testcases"][0])
def test_load_api_definition(self):
path = os.path.join(
os.getcwd(), 'tests/data')
api_dir_dict = utils.load_api_definition(path)
self.assertIn("get_token", api_dir_dict)
self.assertEqual("/api/get-token", api_dir_dict["get_token"]["request"]["url"])
self.assertIn("$user_name", api_dir_dict["get_token"]["function_meta"]["args"])
self.assertIn("create_user", api_dir_dict)
def test_get_api_definition(self):
path = os.path.join(
os.getcwd(), 'tests/data')
api_info = utils.get_api_definition("get_token", path)
self.assertEqual("/api/get-token", api_info["request"]["url"])
self.assertIn(path, utils.api_overall_dict)
with self.assertRaises(exception.ApiNotFound):
utils.get_api_definition("api_not_exist", path)
def test_query_json(self):
json_content = {
"ids": [1, 2, 3, 4],