From 611acf528a4f82e5e8799d30f8823fdf1838a52c Mon Sep 17 00:00:00 2001 From: debugtalk Date: Mon, 18 Sep 2017 16:01:19 +0800 Subject: [PATCH] bugfix: adjust functions location to avoid cross reference --- ate/locustfile_template | 4 +- ate/locusts.py | 2 +- ate/task.py | 4 +- ate/testcase.py | 117 +++++++++++++++++++++++++++++++++++++++ ate/utils.py | 120 +--------------------------------------- tests/test_runner.py | 19 ++++--- tests/test_task.py | 7 ++- tests/test_testcase.py | 112 ++++++++++++++++++++++++++++++++++++- tests/test_utils.py | 109 ------------------------------------ 9 files changed, 251 insertions(+), 243 deletions(-) diff --git a/ate/locustfile_template b/ate/locustfile_template index 11a7c478..235c21a7 100644 --- a/ate/locustfile_template +++ b/ate/locustfile_template @@ -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] diff --git a/ate/locusts.py b/ate/locusts.py index c3299b0d..7bc5a129 100644 --- a/ate/locusts.py +++ b/ate/locusts.py @@ -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 diff --git a/ate/task.py b/ate/task.py index 5ae997ab..be29358e 100644 --- a/ate/task.py +++ b/ate/task.py @@ -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) diff --git a/ate/testcase.py b/ate/testcase.py index 33a93967..17337ff0 100644 --- a/ate/testcase.py +++ b/ate/testcase.py @@ -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): diff --git a/ate/utils.py b/ate/utils.py index 0a5cbfce..428ef9ec 100644 --- a/ate/utils.py +++ b/ate/utils.py @@ -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 diff --git a/tests/test_runner.py b/tests/test_runner.py index 58430d65..0f30d28e 100644 --- a/tests/test_runner.py +++ b/tests/test_runner.py @@ -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]) diff --git a/tests/test_task.py b/tests/test_task.py index ffecc6fc..d2082827 100644 --- a/tests/test_task.py +++ b/tests/test_task.py @@ -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: diff --git a/tests/test_testcase.py b/tests/test_testcase.py index 1f6af65b..674abed0 100644 --- a/tests/test_testcase.py +++ b/tests/test_testcase.py @@ -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) diff --git a/tests/test_utils.py b/tests/test_utils.py index 8e330e04..d535cd55 100644 --- a/tests/test_utils.py +++ b/tests/test_utils.py @@ -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],