From ca7570443f10f96922b5c4aedfd9879c6f2df938 Mon Sep 17 00:00:00 2001 From: debugtalk Date: Fri, 15 May 2020 15:49:50 +0800 Subject: [PATCH] refactor: replace loader --- httprunner/ext/make/__init__.py | 9 +- httprunner/ext/make/make_test.py | 5 +- httprunner/{new_loader.py => loader.py} | 3 + httprunner/loader/__init__.py | 27 -- httprunner/loader/buildup.py | 418 ------------------------ httprunner/loader/buildup_test.py | 203 ------------ httprunner/loader/check.py | 43 --- httprunner/loader/load.py | 218 ------------ httprunner/loader/load_test.py | 124 ------- httprunner/loader/locate.py | 123 ------- httprunner/loader/locate_test.py | 36 -- httprunner/loader_test.py | 138 +++++++- httprunner/parser.py | 6 +- httprunner/runner.py | 2 +- 14 files changed, 147 insertions(+), 1208 deletions(-) rename httprunner/{new_loader.py => loader.py} (98%) delete mode 100644 httprunner/loader/__init__.py delete mode 100644 httprunner/loader/buildup.py delete mode 100644 httprunner/loader/buildup_test.py delete mode 100644 httprunner/loader/check.py delete mode 100644 httprunner/loader/load.py delete mode 100644 httprunner/loader/load_test.py delete mode 100644 httprunner/loader/locate.py delete mode 100644 httprunner/loader/locate_test.py diff --git a/httprunner/ext/make/__init__.py b/httprunner/ext/make/__init__.py index 2414bc73..a42afb91 100644 --- a/httprunner/ext/make/__init__.py +++ b/httprunner/ext/make/__init__.py @@ -7,7 +7,7 @@ from loguru import logger from httprunner import exceptions from httprunner.exceptions import TestCaseFormatError -from httprunner.new_loader import load_testcase_file, load_folder_files +from httprunner.loader import load_testcase_file, load_folder_files __TMPL__ = """# NOTICE: Generated By HttpRunner. DO'NOT EDIT! import unittest @@ -96,9 +96,10 @@ def init_make_parser(subparsers): """ make testcases: parse command line options and run commands. """ parser = subparsers.add_parser( - "make", - help="Convert YAML/JSON testcases to Python unittests.", + "make", help="Convert YAML/JSON testcases to Python unittests.", + ) + parser.add_argument( + "testcase_path", nargs="?", help="Specify YAML/JSON testcase path" ) - parser.add_argument("testcase_path", nargs="?", help="Specify YAML/JSON testcase path") return parser diff --git a/httprunner/ext/make/make_test.py b/httprunner/ext/make/make_test.py index e1bcf391..eaba229c 100644 --- a/httprunner/ext/make/make_test.py +++ b/httprunner/ext/make/make_test.py @@ -3,13 +3,12 @@ from httprunner.ext.make import make_testcase, main_make class TestLoader(unittest.TestCase): - def test_make_testcase(self): path = "examples/postman_echo/request_methods/request_with_variables.yml" testcase_python_path = make_testcase(path) self.assertEqual( testcase_python_path, - "examples/postman_echo/request_methods/request_with_variables_test.py" + "examples/postman_echo/request_methods/request_with_variables_test.py", ) def test_make_testcase_folder(self): @@ -17,5 +16,5 @@ class TestLoader(unittest.TestCase): testcase_python_list = main_make(path) self.assertIn( "examples/postman_echo/request_methods/request_with_functions_test.py", - testcase_python_list + testcase_python_list, ) diff --git a/httprunner/new_loader.py b/httprunner/loader.py similarity index 98% rename from httprunner/new_loader.py rename to httprunner/loader.py index c2b23e9b..96de4c8f 100644 --- a/httprunner/new_loader.py +++ b/httprunner/loader.py @@ -56,6 +56,9 @@ def _load_json_file(json_file: Text) -> Dict: def load_testcase_file(testcase_file: Text) -> Tuple[Dict, TestCase]: """load testcase file and validate with pydantic model""" + if not os.path.isfile(testcase_file): + raise exceptions.FileNotFound(f"testcase file not exists: {testcase_file}") + file_suffix = os.path.splitext(testcase_file)[1].lower() if file_suffix == ".json": testcase_content = _load_json_file(testcase_file) diff --git a/httprunner/loader/__init__.py b/httprunner/loader/__init__.py deleted file mode 100644 index 9ddfa747..00000000 --- a/httprunner/loader/__init__.py +++ /dev/null @@ -1,27 +0,0 @@ -""" -HttpRunner loader - -- check: validate api/testcase/testsuite data structure with JSON schema -- locate: locate debugtalk.py, make it's dir as project root path -- load: load testcase files and relevant data, including debugtalk.py, .env, yaml/json api/testcases, csv, etc. -- buildup: assemble loaded content to httprunner testcase/testsuite data structure - -""" - -from httprunner.loader.buildup import load_cases, load_project_data -from httprunner.loader.check import is_test_path -from httprunner.loader.load import load_csv_file, load_builtin_functions -from httprunner.loader.locate import ( - get_project_working_directory as get_pwd, - init_project_working_directory as init_pwd, -) - -__all__ = [ - "is_test_path", - "get_pwd", - "init_pwd", - "load_csv_file", - "load_builtin_functions", - "load_project_data", - "load_cases", -] diff --git a/httprunner/loader/buildup.py b/httprunner/loader/buildup.py deleted file mode 100644 index 44cbd548..00000000 --- a/httprunner/loader/buildup.py +++ /dev/null @@ -1,418 +0,0 @@ -import importlib -import os - -from loguru import logger - -from httprunner import exceptions, utils -from httprunner.loader.load import ( - load_module_functions, - load_file, - load_dot_env_file, - load_folder_files, -) -from httprunner.loader.locate import ( - init_project_working_directory, - get_project_working_directory, -) - -tests_def_mapping = {"api": {}, "testcases": {}} - - -def load_debugtalk_functions(): - """ load project debugtalk.py module functions - debugtalk.py should be located in project working directory. - - Returns: - dict: debugtalk module functions mapping - { - "func1_name": func1, - "func2_name": func2 - } - - """ - # load debugtalk.py module - imported_module = importlib.import_module("debugtalk") - return load_module_functions(imported_module) - - -def __extend_with_api_ref(raw_testinfo): - """ extend with api reference - - Raises: - exceptions.ApiNotFound: api not found - - """ - api_name = raw_testinfo["api"] - - # api maybe defined in two types: - # 1, individual file: each file is corresponding to one api definition - # 2, api sets file: one file contains a list of api definitions - if not os.path.isabs(api_name): - # make compatible with Windows/Linux - pwd = get_project_working_directory() - api_path = os.path.join(pwd, *api_name.split("/")) - if os.path.isfile(api_path): - # type 1: api is defined in individual file - api_name = api_path - - if api_name in tests_def_mapping["api"]: - block = tests_def_mapping["api"][api_name] - elif not os.path.isfile(api_name): - raise exceptions.ApiNotFound(f"{api_name} not found!") - else: - block = load_file(api_name) - - # NOTICE: avoid project_meta been changed during iteration. - raw_testinfo["api_def"] = utils.deepcopy_dict(block) - tests_def_mapping["api"][api_name] = block - - -def __extend_with_testcase_ref(raw_testinfo): - """ extend with testcase reference - """ - testcase_path = raw_testinfo["testcase"] - - if testcase_path not in tests_def_mapping["testcases"]: - # make compatible with Windows/Linux - pwd = get_project_working_directory() - testcase_path = os.path.join(pwd, *testcase_path.split("/")) - loaded_testcase = load_file(testcase_path) - - # TODO: validate with pydantic - if isinstance(loaded_testcase, dict) and "teststeps" in loaded_testcase: - testcase_dict = load_testcase(loaded_testcase) - else: - raise exceptions.FileFormatError( - f"Invalid format testcase: {testcase_path}" - ) - - tests_def_mapping["testcases"][testcase_path] = testcase_dict - else: - testcase_dict = tests_def_mapping["testcases"][testcase_path] - - raw_testinfo["testcase_def"] = testcase_dict - - -def load_teststep(raw_testinfo): - """ load testcase step content. - teststep maybe defined directly, or reference api/testcase. - - Args: - raw_testinfo (dict): test data, maybe in 3 formats. - # api reference - { - "name": "add product to cart", - "api": "/path/to/api", - "variables": {}, - "validate": [], - "extract": {} - } - # testcase reference - { - "name": "add product to cart", - "testcase": "/path/to/testcase", - "variables": {} - } - # define directly - { - "name": "checkout cart", - "request": {}, - "variables": {}, - "validate": [], - "extract": {} - } - - Returns: - dict: loaded teststep content - - """ - # reference api - if "api" in raw_testinfo: - __extend_with_api_ref(raw_testinfo) - - # TODO: reference proc functions - # elif "func" in raw_testinfo: - # pass - - # reference testcase - elif "testcase" in raw_testinfo: - __extend_with_testcase_ref(raw_testinfo) - - # define directly - else: - pass - - return raw_testinfo - - -def load_testcase(raw_testcase): - """ load testcase. - - Args: - raw_testcase (dict): raw testcase content loaded from JSON/YAML file: - { - "config": { - "name": "xxx", - "variables": {} - } - "teststeps": [ - { - "name": "teststep 1", - "request" {...} - }, - { - "name": "teststep 2", - "request" {...} - }, - ] - } - - Returns: - dict: loaded testcase content - { - "config": {}, - "teststeps": [test11, test12] - } - - """ - raw_teststeps = raw_testcase.pop("teststeps") - raw_testcase["teststeps"] = [load_teststep(teststep) for teststep in raw_teststeps] - return raw_testcase - - -def load_testsuite(raw_testsuite): - """ load testsuite with testcase references. - support two different formats. - - Args: - raw_testsuite (dict): raw testsuite content loaded from JSON/YAML file: - { - "config": { - "name": "xxx", - "variables": {} - } - "testcases": [ - { - "name": "testcase1", - "testcase": "/path/to/testcase", - "variables": {...}, - "parameters": {...} - }, - {} - ] - } - - Returns: - dict: loaded testsuite content - { - "config": {}, - "testcases": [testcase1, testcase2] - } - - """ - raw_testcases = raw_testsuite["testcases"] - - # TODO: validate with pydantic - if not isinstance(raw_testcases, list): - # invalid format - raise exceptions.FileFormatError("Invalid testsuite format!") - - raw_testsuite["testcases"] = {} - for raw_testcase in raw_testcases: - __extend_with_testcase_ref(raw_testcase) - testcase_name = raw_testcase["name"] - raw_testsuite["testcases"][testcase_name] = raw_testcase - - return raw_testsuite - - -def load_test_file(path: str) -> dict: - """ load test file, file maybe testcase/testsuite/api - - Args: - path (str): test file path - - Returns: - dict: loaded test content - - # api - { - "path": path, - "type": "api", - "name": "", - "request": {} - } - - # testcase - { - "path": path, - "type": "testcase", - "config": {}, - "teststeps": [] - } - - # testsuite - { - "path": path, - "type": "testsuite", - "config": {}, - "testcases": {} - } - - """ - raw_content = load_file(path) - - if not isinstance(raw_content, dict): - # invalid format - raise exceptions.FileFormatError("Invalid test file format!") - - if "testcases" in raw_content: - # file_type: testsuite - loaded_content = load_testsuite(raw_content) - loaded_content["path"] = path - loaded_content["type"] = "testsuite" - - elif "teststeps" in raw_content: - # file_type: testcase - loaded_content = load_testcase(raw_content) - loaded_content["path"] = path - loaded_content["type"] = "testcase" - - elif "request" in raw_content: - # file_type: api - loaded_content = raw_content - loaded_content["path"] = path - loaded_content["type"] = "api" - - else: - # invalid format - raise exceptions.FileFormatError("Invalid test file format!") - - return loaded_content - - -def load_project_data(test_path, dot_env_path=None): - """ load api, testcases, .env, debugtalk.py functions. - api/testcases folder is relative to project_working_directory - - Args: - test_path (str): test file/folder path, locate pwd from this path. - dot_env_path (str): specified .env file path - - Returns: - dict: project loaded api/testcases definitions, - environments and debugtalk.py functions. - - """ - debugtalk_path, project_working_directory = init_project_working_directory( - test_path - ) - - project_meta = {} - - # load .env file - # NOTICE: - # environment variable maybe loaded in debugtalk.py - # thus .env file should be loaded before loading debugtalk.py - dot_env_path = dot_env_path or os.path.join(project_working_directory, ".env") - project_meta["env"] = load_dot_env_file(dot_env_path) - - if debugtalk_path: - # load debugtalk.py functions - debugtalk_functions = load_debugtalk_functions() - else: - debugtalk_functions = {} - - # locate PWD and load debugtalk.py functions - project_meta["PWD"] = project_working_directory - project_meta["functions"] = debugtalk_functions - project_meta["test_path"] = os.path.abspath(test_path)[ - len(project_working_directory) + 1 : - ] - - return project_meta - - -def load_cases(path: str, dot_env_path: str = None): - """ load testcases from file path, extend and merge with api/testcase definitions. - - Args: - path (str): testcase/testsuite file/foler path. - path could be in 2 types: - - absolute/relative file path - - absolute/relative folder path - dot_env_path (str): specified .env file path - - Returns: - dict: tests mapping, include project_meta and testcases. - each testcase is corresponding to a file. - { - "project_meta": { - "PWD": "XXXXX", - "functions": {}, - "env": {} - }, - "testcases": [ - { # testcase data structure - "config": { - "name": "desc1", - "path": "testcase1_path", - "variables": [], # optional - }, - "teststeps": [ - # test data structure - { - 'name': 'test desc1', - 'variables': [], # optional - 'extract': [], # optional - 'validate': [], - 'request': {} - }, - test_dict_2 # another test dict - ] - }, - testcase_2_dict # another testcase dict - ], - "testsuites": [ - { # testsuite data structure - "config": {}, - "testcases": { - "testcase1": {}, - "testcase2": {}, - } - }, - testsuite_2_dict - ] - } - - """ - - tests_mapping = {"project_meta": load_project_data(path, dot_env_path)} - - def __load_file_content(path): - loaded_content = None - try: - loaded_content = load_test_file(path) - except exceptions.ApiNotFound as ex: - logger.warning(f"Invalid api reference in {path}: {ex}") - except exceptions.FileFormatError: - logger.warning(f"Invalid test file format: {path}") - - if not loaded_content: - pass - elif loaded_content["type"] == "testsuite": - tests_mapping.setdefault("testsuites", []).append(loaded_content) - elif loaded_content["type"] == "testcase": - tests_mapping.setdefault("testcases", []).append(loaded_content) - elif loaded_content["type"] == "api": - tests_mapping.setdefault("apis", []).append(loaded_content) - - if os.path.isdir(path): - files_list = load_folder_files(path) - for path in files_list: - __load_file_content(path) - - elif os.path.isfile(path): - __load_file_content(path) - - return tests_mapping diff --git a/httprunner/loader/buildup_test.py b/httprunner/loader/buildup_test.py deleted file mode 100644 index 35598070..00000000 --- a/httprunner/loader/buildup_test.py +++ /dev/null @@ -1,203 +0,0 @@ -import os -import unittest - -from httprunner import exceptions, loader -from httprunner.loader import buildup - - -class TestModuleLoader(unittest.TestCase): - def test_filter_module_functions(self): - module_functions = buildup.load_module_functions(buildup) - self.assertIn("load_module_functions", module_functions) - self.assertNotIn("is_py3", module_functions) - - def test_load_debugtalk_module(self): - project_meta = buildup.load_project_data( - os.path.join(os.getcwd(), "httprunner") - ) - self.assertNotIn("alter_response", project_meta["functions"]) - - project_meta = buildup.load_project_data(os.path.join(os.getcwd(), "tests")) - self.assertIn("alter_response", project_meta["functions"]) - - is_status_code_200 = project_meta["functions"]["is_status_code_200"] - self.assertTrue(is_status_code_200(200)) - self.assertFalse(is_status_code_200(500)) - - def test_load_debugtalk_py(self): - project_meta = buildup.load_project_data("tests/data/demo_testcase.yml") - project_working_directory = project_meta["PWD"] - debugtalk_functions = project_meta["functions"] - self.assertEqual(project_working_directory, os.path.join(os.getcwd(), "tests")) - self.assertIn("gen_md5", debugtalk_functions) - - project_meta = buildup.load_project_data("tests/base.py") - project_working_directory = project_meta["PWD"] - debugtalk_functions = project_meta["functions"] - self.assertEqual(project_working_directory, os.path.join(os.getcwd(), "tests")) - self.assertIn("gen_md5", debugtalk_functions) - - project_meta = buildup.load_project_data("httprunner/__init__.py") - project_working_directory = project_meta["PWD"] - debugtalk_functions = project_meta["functions"] - self.assertEqual(project_working_directory, os.getcwd()) - self.assertEqual(debugtalk_functions, {}) - - -class TestSuiteLoader(unittest.TestCase): - @classmethod - def setUpClass(cls): - cls.project_meta = buildup.load_project_data(os.path.join(os.getcwd(), "tests")) - cls.tests_def_mapping = buildup.tests_def_mapping - - def test_load_teststep_api(self): - raw_test = { - "name": "create user (override).", - "api": "api/create_user.yml", - "variables": [{"uid": "999"}], - } - teststep = buildup.load_teststep(raw_test) - self.assertEqual("create user (override).", teststep["name"]) - self.assertIn("api_def", teststep) - api_def = teststep["api_def"] - self.assertEqual(api_def["name"], "create user") - self.assertEqual(api_def["request"]["url"], "/api/users/$uid") - - def test_load_teststep_testcase(self): - raw_test = { - "name": "setup and reset all (override).", - "testcase": "testcases/setup.yml", - "variables": [{"device_sn": "$device_sn"}], - } - testcase = buildup.load_teststep(raw_test) - self.assertEqual("setup and reset all (override).", testcase["name"]) - tests = testcase["testcase_def"]["teststeps"] - self.assertEqual(len(tests), 2) - self.assertEqual(tests[0]["name"], "get token (setup)") - self.assertEqual(tests[1]["name"], "reset all users") - - def test_load_test_file_api(self): - loaded_content = buildup.load_test_file("tests/api/create_user.yml") - self.assertEqual(loaded_content["type"], "api") - self.assertIn("path", loaded_content) - self.assertIn("request", loaded_content) - self.assertEqual(loaded_content["request"]["url"], "/api/users/$uid") - - def test_load_test_file_testcase(self): - for loaded_content in [ - buildup.load_test_file("tests/testcases/setup.yml"), - buildup.load_test_file("tests/testcases/setup.json"), - ]: - self.assertEqual(loaded_content["type"], "testcase") - self.assertIn("path", loaded_content) - self.assertIn("config", loaded_content) - self.assertEqual(loaded_content["config"]["name"], "setup and reset all.") - self.assertIn("teststeps", loaded_content) - self.assertEqual(len(loaded_content["teststeps"]), 2) - - def test_load_test_file_testsuite(self): - for loaded_content in [ - buildup.load_test_file("tests/testsuites/create_users.yml"), - buildup.load_test_file("tests/testsuites/create_users.json"), - ]: - self.assertEqual(loaded_content["type"], "testsuite") - - testcases = loaded_content["testcases"] - self.assertEqual(len(testcases), 2) - self.assertIn("create user 1000 and check result.", testcases) - self.assertIn( - "testcase_def", testcases["create user 1000 and check result."] - ) - self.assertEqual( - testcases["create user 1000 and check result."]["testcase_def"][ - "config" - ]["name"], - "create user and check result.", - ) - - def test_load_tests_api_file(self): - path = os.path.join(os.getcwd(), "tests/api/create_user.yml") - tests_mapping = loader.load_cases(path) - api_list = tests_mapping["apis"] - self.assertEqual(len(api_list), 1) - self.assertEqual(api_list[0]["request"]["url"], "/api/users/$uid") - - def test_load_tests_testcase_file_2(self): - testcase_file_path = os.path.join(os.getcwd(), "tests/data/demo_testcase.yml") - tests_mapping = loader.load_cases(testcase_file_path) - testcases = tests_mapping["testcases"] - self.assertIsInstance(testcases, list) - self.assertEqual(testcases[0]["config"]["name"], "123t$var_a") - self.assertIn("sum_two", tests_mapping["project_meta"]["functions"]) - self.assertEqual( - testcases[0]["config"]["variables"]["var_c"], "${sum_two($var_a, $var_b)}" - ) - self.assertEqual( - testcases[0]["config"]["variables"]["PROJECT_KEY"], "${ENV(PROJECT_KEY)}" - ) - - def test_load_tests_testcase_file_with_api_ref(self): - path = os.path.join(os.getcwd(), "tests/data/demo_testcase_layer.yml") - tests_mapping = loader.load_cases(path) - project_meta = tests_mapping["project_meta"] - testcases_list = tests_mapping["testcases"] - self.assertIn("device_sn", testcases_list[0]["config"]["variables"]) - self.assertIn("gen_md5", project_meta["functions"]) - self.assertIn("base_url", testcases_list[0]["config"]) - test_dict0 = testcases_list[0]["teststeps"][0] - self.assertEqual("get token with $user_agent, $app_version", test_dict0["name"]) - self.assertIn("/api/get-token", test_dict0["api_def"]["request"]["url"]) - self.assertIn({"eq": ["status_code", 200]}, test_dict0["validate"]) - - def test_load_tests_testsuite_file_with_testcase_ref(self): - path = os.path.join(os.getcwd(), "tests/testsuites/create_users.yml") - tests_mapping = loader.load_cases(path) - project_meta = tests_mapping["project_meta"] - testsuites_list = tests_mapping["testsuites"] - - self.assertEqual("create users with uid", testsuites_list[0]["config"]["name"]) - self.assertEqual( - "${gen_random_string(15)}", - testsuites_list[0]["config"]["variables"]["device_sn"], - ) - self.assertIn( - "create user 1000 and check result.", testsuites_list[0]["testcases"] - ) - - self.assertEqual( - testsuites_list[0]["testcases"]["create user 1000 and check result."][ - "testcase_def" - ]["config"]["name"], - "create user and check result.", - ) - - def test_load_tests_folder_path(self): - # absolute folder path - path = os.path.join(os.getcwd(), "tests/data") - tests_mapping = loader.load_cases(path) - testcase_list_1 = tests_mapping["testcases"] - self.assertGreater(len(testcase_list_1), 4) - - # relative folder path - path = "tests/data/" - tests_mapping = loader.load_cases(path) - testcase_list_2 = tests_mapping["testcases"] - self.assertEqual(len(testcase_list_1), len(testcase_list_2)) - - def test_load_tests_path_not_exist(self): - # absolute folder path - path = os.path.join(os.getcwd(), "tests/data_not_exist") - with self.assertRaises(exceptions.FileNotFound): - loader.load_cases(path) - - # relative folder path - path = "tests/data_not_exist" - with self.assertRaises(exceptions.FileNotFound): - loader.load_cases(path) - - def test_load_project_tests(self): - buildup.load_project_data(os.path.join(os.getcwd(), "tests")) - self.assertIn("gen_md5", self.project_meta["functions"]) - self.assertEqual(self.project_meta["env"]["PROJECT_KEY"], "ABCDEFGH") - self.assertEqual(os.path.basename(self.project_meta["PWD"]), "tests") - self.assertEqual(os.path.basename(self.project_meta["test_path"]), "") # FIXME diff --git a/httprunner/loader/check.py b/httprunner/loader/check.py deleted file mode 100644 index 56b1f473..00000000 --- a/httprunner/loader/check.py +++ /dev/null @@ -1,43 +0,0 @@ -import os - - -def is_test_path(path): - """ check if path is valid json/yaml file path or a existed directory. - - Args: - path (str/list/tuple): file path/directory or file path list. - - Returns: - bool: True if path is valid file path or path list, otherwise False. - - """ - if not isinstance(path, (str, list, tuple)): - return False - - elif isinstance(path, (list, tuple)): - for p in path: - if not is_test_path(p): - return False - - return True - - else: - # path is string - if not os.path.exists(path): - return False - - # path exists - if os.path.isfile(path): - # path is a file - file_suffix = os.path.splitext(path)[1].lower() - if file_suffix not in [".json", ".yaml", ".yml"]: - # path is not json/yaml file - return False - else: - return True - elif os.path.isdir(path): - # path is a directory - return True - else: - # path is neither a folder nor a file, maybe a symbol link or something else - return False diff --git a/httprunner/loader/load.py b/httprunner/loader/load.py deleted file mode 100644 index 43e6a19c..00000000 --- a/httprunner/loader/load.py +++ /dev/null @@ -1,218 +0,0 @@ -import csv -import io -import json -import os -import types - -import yaml -from loguru import logger - -from httprunner import builtin -from httprunner import exceptions, utils -from httprunner.loader.locate import get_project_working_directory - -try: - # PyYAML version >= 5.1 - # ref: https://github.com/yaml/pyyaml/wiki/PyYAML-yaml.load(input)-Deprecation - yaml.warnings({"YAMLLoadWarning": False}) -except AttributeError: - pass - - -def _load_yaml_file(yaml_file): - """ load yaml file and check file content format - """ - with io.open(yaml_file, "r", encoding="utf-8") as stream: - try: - yaml_content = yaml.load(stream) - except yaml.YAMLError as ex: - logger.error(str(ex)) - raise exceptions.FileFormatError - - return yaml_content - - -def _load_json_file(json_file): - """ load json file and check file content format - """ - with io.open(json_file, encoding="utf-8") as data_file: - try: - json_content = json.load(data_file) - except json.JSONDecodeError: - err_msg = f"JSONDecodeError: JSON file format error: {json_file}" - logger.error(err_msg) - raise exceptions.FileFormatError(err_msg) - - return json_content - - -def load_csv_file(csv_file): - """ load csv file and check file content format - - Args: - csv_file (str): csv file path, csv file content is like below: - - Returns: - list: list of parameters, each parameter is in dict format - - Examples: - >>> cat csv_file - username,password - test1,111111 - test2,222222 - test3,333333 - - >>> load_csv_file(csv_file) - [ - {'username': 'test1', 'password': '111111'}, - {'username': 'test2', 'password': '222222'}, - {'username': 'test3', 'password': '333333'} - ] - - """ - if not os.path.isabs(csv_file): - pwd = get_project_working_directory() - # make compatible with Windows/Linux - csv_file = os.path.join(pwd, *csv_file.split("/")) - - if not os.path.isfile(csv_file): - # file path not exist - raise exceptions.CSVNotFound(csv_file) - - csv_content_list = [] - - with io.open(csv_file, encoding="utf-8") as csvfile: - reader = csv.DictReader(csvfile) - for row in reader: - csv_content_list.append(row) - - return csv_content_list - - -def load_file(file_path): - if not os.path.isfile(file_path): - raise exceptions.FileNotFound(f"{file_path} does not exist.") - - file_suffix = os.path.splitext(file_path)[1].lower() - if file_suffix == ".json": - return _load_json_file(file_path) - elif file_suffix in [".yaml", ".yml"]: - return _load_yaml_file(file_path) - elif file_suffix == ".csv": - return load_csv_file(file_path) - else: - # '' or other suffix - logger.warning(f"Unsupported file format: {file_path}") - return [] - - -def load_folder_files(folder_path, recursive=True): - """ load folder path, return all files endswith yml/yaml/json in list. - - Args: - folder_path (str): specified folder path to load - recursive (bool): load files recursively if True - - Returns: - list: files endswith yml/yaml/json - """ - if isinstance(folder_path, (list, set)): - files = [] - for path in set(folder_path): - files.extend(load_folder_files(path, recursive)) - - return files - - if not os.path.exists(folder_path): - return [] - - file_list = [] - - for dirpath, dirnames, filenames in os.walk(folder_path): - filenames_list = [] - - for filename in filenames: - if not filename.endswith((".yml", ".yaml", ".json")): - continue - - filenames_list.append(filename) - - for filename in filenames_list: - file_path = os.path.join(dirpath, filename) - file_list.append(file_path) - - if not recursive: - break - - return file_list - - -def load_dot_env_file(dot_env_path): - """ load .env file. - - Args: - dot_env_path (str): .env file path - - Returns: - dict: environment variables mapping - - { - "UserName": "debugtalk", - "Password": "123456", - "PROJECT_KEY": "ABCDEFGH" - } - - Raises: - exceptions.FileFormatError: If .env file format is invalid. - - """ - if not os.path.isfile(dot_env_path): - return {} - - logger.info(f"Loading environment variables from {dot_env_path}") - env_variables_mapping = {} - - with io.open(dot_env_path, "r", encoding="utf-8") as fp: - for line in fp: - # maxsplit=1 - if "=" in line: - variable, value = line.split("=", 1) - elif ":" in line: - variable, value = line.split(":", 1) - else: - raise exceptions.FileFormatError(".env format error") - - env_variables_mapping[variable.strip()] = value.strip() - - utils.set_os_environ(env_variables_mapping) - return env_variables_mapping - - -def load_module_functions(module): - """ load python module functions. - - Args: - module: python module - - Returns: - dict: functions mapping for specified python module - - { - "func1_name": func1, - "func2_name": func2 - } - - """ - module_functions = {} - - for name, item in vars(module).items(): - if isinstance(item, types.FunctionType): - module_functions[name] = item - - return module_functions - - -def load_builtin_functions(): - """ load builtin module functions - """ - return load_module_functions(builtin) diff --git a/httprunner/loader/load_test.py b/httprunner/loader/load_test.py deleted file mode 100644 index d9c96c8d..00000000 --- a/httprunner/loader/load_test.py +++ /dev/null @@ -1,124 +0,0 @@ -import os -import unittest - -from httprunner import exceptions -from httprunner.loader import load -from httprunner.loader.buildup import load_test_file - - -class TestFileLoader(unittest.TestCase): - def test_load_yaml_file_file_format_error(self): - yaml_tmp_file = "tests/data/tmp.yml" - # create empty yaml file - with open(yaml_tmp_file, "w") as f: - f.write("") - - with self.assertRaises(exceptions.FileFormatError): - load_test_file(yaml_tmp_file) - - os.remove(yaml_tmp_file) - - # create invalid format yaml file - with open(yaml_tmp_file, "w") as f: - f.write("abc") - - with self.assertRaises(exceptions.FileFormatError): - load_test_file(yaml_tmp_file) - - os.remove(yaml_tmp_file) - - def test_load_json_file_file_format_error(self): - json_tmp_file = "tests/data/tmp.json" - # create empty file - with open(json_tmp_file, "w") as f: - f.write("") - - with self.assertRaises(exceptions.FileFormatError): - load_test_file(json_tmp_file) - - os.remove(json_tmp_file) - - # create empty json file - with open(json_tmp_file, "w") as f: - f.write("{}") - - with self.assertRaises(exceptions.FileFormatError): - load_test_file(json_tmp_file) - - os.remove(json_tmp_file) - - # create invalid format json file - with open(json_tmp_file, "w") as f: - f.write("abc") - - with self.assertRaises(exceptions.FileFormatError): - load_test_file(json_tmp_file) - - os.remove(json_tmp_file) - - def test_load_testcases_bad_filepath(self): - testcase_file_path = os.path.join(os.getcwd(), "tests/data/demo") - with self.assertRaises(exceptions.FileNotFound): - load.load_file(testcase_file_path) - - def test_load_csv_file_one_parameter(self): - csv_file_path = os.path.join(os.getcwd(), "tests/data/user_agent.csv") - csv_content = load.load_file(csv_file_path) - self.assertEqual( - csv_content, - [ - {"user_agent": "iOS/10.1"}, - {"user_agent": "iOS/10.2"}, - {"user_agent": "iOS/10.3"}, - ], - ) - - def test_load_csv_file_multiple_parameters(self): - csv_file_path = os.path.join(os.getcwd(), "tests/data/account.csv") - csv_content = load.load_file(csv_file_path) - self.assertEqual( - csv_content, - [ - {"username": "test1", "password": "111111"}, - {"username": "test2", "password": "222222"}, - {"username": "test3", "password": "333333"}, - ], - ) - - def test_load_folder_files(self): - folder = os.path.join(os.getcwd(), "tests") - file1 = os.path.join(os.getcwd(), "tests", "test_utils.py") - file2 = os.path.join(os.getcwd(), "tests", "api", "reset_all.yml") - - files = load.load_folder_files(folder, recursive=False) - self.assertEqual(files, []) - - files = load.load_folder_files(folder) - self.assertIn(file2, files) - self.assertNotIn(file1, files) - - files = load.load_folder_files("not_existed_foulder", recursive=False) - self.assertEqual([], files) - - files = load.load_folder_files(file2, recursive=False) - self.assertEqual([], files) - - def test_load_dot_env_file(self): - dot_env_path = os.path.join(os.getcwd(), "tests", ".env") - env_variables_mapping = load.load_dot_env_file(dot_env_path) - self.assertIn("PROJECT_KEY", env_variables_mapping) - self.assertEqual(env_variables_mapping["UserName"], "debugtalk") - - def test_load_custom_dot_env_file(self): - dot_env_path = os.path.join(os.getcwd(), "tests", "data", "test.env") - env_variables_mapping = load.load_dot_env_file(dot_env_path) - self.assertIn("PROJECT_KEY", env_variables_mapping) - self.assertEqual(env_variables_mapping["UserName"], "test") - self.assertEqual( - env_variables_mapping["content_type"], "application/json; charset=UTF-8" - ) - - def test_load_env_path_not_exist(self): - dot_env_path = os.path.join(os.getcwd(), "tests", "data",) - env_variables_mapping = load.load_dot_env_file(dot_env_path) - self.assertEqual(env_variables_mapping, {}) diff --git a/httprunner/loader/locate.py b/httprunner/loader/locate.py deleted file mode 100644 index 0d4dd056..00000000 --- a/httprunner/loader/locate.py +++ /dev/null @@ -1,123 +0,0 @@ -import os -import sys - -from loguru import logger - -from httprunner import exceptions - -project_working_directory = None - - -def locate_file(start_path, file_name): - """ locate filename and return absolute file path. - searching will be recursive upward until current working directory or system root dir. - - Args: - file_name (str): target locate file name - start_path (str): start locating path, maybe file path or directory path - - Returns: - str: located file path. None if file not found. - - Raises: - exceptions.FileNotFound: If failed to locate file. - - """ - if os.path.isfile(start_path): - start_dir_path = os.path.dirname(start_path) - elif os.path.isdir(start_path): - start_dir_path = start_path - else: - raise exceptions.FileNotFound(f"invalid path: {start_path}") - - file_path = os.path.join(start_dir_path, file_name) - if os.path.isfile(file_path): - return os.path.abspath(file_path) - - # current working directory - if os.path.abspath(start_dir_path) == os.getcwd(): - raise exceptions.FileNotFound(f"{file_name} not found in {start_path}") - - # system root dir - # Windows, e.g. 'E:\\' - # Linux/Darwin, '/' - parent_dir = os.path.dirname(start_dir_path) - if parent_dir == start_dir_path: - raise exceptions.FileNotFound(f"{file_name} not found in {start_path}") - - # locate recursive upward - return locate_file(parent_dir, file_name) - - -def locate_debugtalk_py(start_path): - """ locate debugtalk.py file - - Args: - start_path (str): start locating path, - maybe testcase file path or directory path - - Returns: - str: debugtalk.py file path, None if not found - - """ - try: - # locate debugtalk.py file. - debugtalk_path = locate_file(start_path, "debugtalk.py") - except exceptions.FileNotFound: - debugtalk_path = None - - return debugtalk_path - - -def init_project_working_directory(test_path): - """ this should be called at startup - - run test file: - run_path -> load_cases -> load_project_data -> init_project_working_directory - or run passed in data structure: - run -> init_project_working_directory - - Args: - test_path: specified testfile path - - Returns: - (str, str): debugtalk.py path, project_working_directory - - """ - - def prepare_path(path): - if not os.path.exists(path): - err_msg = f"path not exist: {path}" - logger.error(err_msg) - raise exceptions.FileNotFound(err_msg) - - if not os.path.isabs(path): - path = os.path.join(os.getcwd(), path) - - return path - - test_path = prepare_path(test_path) - - # locate debugtalk.py file - debugtalk_path = locate_debugtalk_py(test_path) - - global project_working_directory - if debugtalk_path: - # The folder contains debugtalk.py will be treated as PWD. - project_working_directory = os.path.dirname(debugtalk_path) - else: - # debugtalk.py not found, use os.getcwd() as PWD. - project_working_directory = os.getcwd() - - # add PWD to sys.path - sys.path.insert(0, project_working_directory) - - return debugtalk_path, project_working_directory - - -def get_project_working_directory(): - global project_working_directory - if project_working_directory is None: - raise exceptions.MyBaseFailure("loader.load_cases() has not been called!") - - return project_working_directory diff --git a/httprunner/loader/locate_test.py b/httprunner/loader/locate_test.py deleted file mode 100644 index df26aad7..00000000 --- a/httprunner/loader/locate_test.py +++ /dev/null @@ -1,36 +0,0 @@ -import os -import unittest - -from httprunner import exceptions -from httprunner.loader import locate - - -class TestLoaderLocate(unittest.TestCase): - def test_locate_file(self): - with self.assertRaises(exceptions.FileNotFound): - locate.locate_file(os.getcwd(), "debugtalk.py") - - with self.assertRaises(exceptions.FileNotFound): - locate.locate_file("", "debugtalk.py") - - start_path = os.path.join(os.getcwd(), "tests") - self.assertEqual( - locate.locate_file(start_path, "debugtalk.py"), - os.path.join(os.getcwd(), "tests/debugtalk.py"), - ) - self.assertEqual( - locate.locate_file("tests/", "debugtalk.py"), - os.path.join(os.getcwd(), "tests", "debugtalk.py"), - ) - self.assertEqual( - locate.locate_file("tests", "debugtalk.py"), - os.path.join(os.getcwd(), "tests", "debugtalk.py"), - ) - self.assertEqual( - locate.locate_file("tests/base.py", "debugtalk.py"), - os.path.join(os.getcwd(), "tests", "debugtalk.py"), - ) - self.assertEqual( - locate.locate_file("tests/data/demo_testcase.yml", "debugtalk.py"), - os.path.join(os.getcwd(), "tests", "debugtalk.py"), - ) diff --git a/httprunner/loader_test.py b/httprunner/loader_test.py index 1296ee84..4defb674 100644 --- a/httprunner/loader_test.py +++ b/httprunner/loader_test.py @@ -1,13 +1,141 @@ +import os import unittest -from httprunner.new_loader import load_testcase_file + +from httprunner import exceptions, loader class TestLoader(unittest.TestCase): - def test_load_testcase_file(self): path = "examples/postman_echo/request_methods/request_with_variables.yml" - testcase_json, testcase_obj = load_testcase_file(path) - self.assertEqual(testcase_json["config"]["name"], "request methods testcase with variables") - self.assertEqual(testcase_obj.config.name, "request methods testcase with variables") + testcase_json, testcase_obj = loader.load_testcase_file(path) + self.assertEqual( + testcase_json["config"]["name"], "request methods testcase with variables" + ) + self.assertEqual( + testcase_obj.config.name, "request methods testcase with variables" + ) self.assertEqual(len(testcase_json["teststeps"]), 3) self.assertEqual(len(testcase_obj.teststeps), 3) + + def test_load_json_file_file_format_error(self): + json_tmp_file = "tests/data/tmp.json" + # create empty file + with open(json_tmp_file, "w") as f: + f.write("") + + with self.assertRaises(exceptions.FileFormatError): + loader._load_json_file(json_tmp_file) + + os.remove(json_tmp_file) + + # create empty json file + with open(json_tmp_file, "w") as f: + f.write("{}") + + loader._load_json_file(json_tmp_file) + os.remove(json_tmp_file) + + # create invalid format json file + with open(json_tmp_file, "w") as f: + f.write("abc") + + with self.assertRaises(exceptions.FileFormatError): + loader._load_json_file(json_tmp_file) + + os.remove(json_tmp_file) + + def test_load_testcases_bad_filepath(self): + testcase_file_path = os.path.join(os.getcwd(), "tests/data/demo") + with self.assertRaises(exceptions.FileNotFound): + loader.load_testcase_file(testcase_file_path) + + def test_load_csv_file_one_parameter(self): + csv_file_path = os.path.join(os.getcwd(), "tests/data/user_agent.csv") + csv_content = loader.load_csv_file(csv_file_path) + self.assertEqual( + csv_content, + [ + {"user_agent": "iOS/10.1"}, + {"user_agent": "iOS/10.2"}, + {"user_agent": "iOS/10.3"}, + ], + ) + + def test_load_csv_file_multiple_parameters(self): + csv_file_path = os.path.join(os.getcwd(), "tests/data/account.csv") + csv_content = loader.load_csv_file(csv_file_path) + self.assertEqual( + csv_content, + [ + {"username": "test1", "password": "111111"}, + {"username": "test2", "password": "222222"}, + {"username": "test3", "password": "333333"}, + ], + ) + + def test_load_folder_files(self): + folder = os.path.join(os.getcwd(), "tests") + file1 = os.path.join(os.getcwd(), "tests", "test_utils.py") + file2 = os.path.join(os.getcwd(), "tests", "api", "reset_all.yml") + + files = loader.load_folder_files(folder, recursive=False) + self.assertEqual(files, []) + + files = loader.load_folder_files(folder) + self.assertIn(file2, files) + self.assertNotIn(file1, files) + + files = loader.load_folder_files("not_existed_foulder", recursive=False) + self.assertEqual([], files) + + files = loader.load_folder_files(file2, recursive=False) + self.assertEqual([], files) + + def test_load_dot_env_file(self): + dot_env_path = os.path.join(os.getcwd(), "tests", ".env") + env_variables_mapping = loader.load_dot_env_file(dot_env_path) + self.assertIn("PROJECT_KEY", env_variables_mapping) + self.assertEqual(env_variables_mapping["UserName"], "debugtalk") + + def test_load_custom_dot_env_file(self): + dot_env_path = os.path.join(os.getcwd(), "tests", "data", "test.env") + env_variables_mapping = loader.load_dot_env_file(dot_env_path) + self.assertIn("PROJECT_KEY", env_variables_mapping) + self.assertEqual(env_variables_mapping["UserName"], "test") + self.assertEqual( + env_variables_mapping["content_type"], "application/json; charset=UTF-8" + ) + + def test_load_env_path_not_exist(self): + dot_env_path = os.path.join(os.getcwd(), "tests", "data",) + env_variables_mapping = loader.load_dot_env_file(dot_env_path) + self.assertEqual(env_variables_mapping, {}) + + def test_locate_file(self): + with self.assertRaises(exceptions.FileNotFound): + loader.locate_file(os.getcwd(), "debugtalk.py") + + with self.assertRaises(exceptions.FileNotFound): + loader.locate_file("", "debugtalk.py") + + start_path = os.path.join(os.getcwd(), "tests") + self.assertEqual( + loader.locate_file(start_path, "debugtalk.py"), + os.path.join(os.getcwd(), "tests/debugtalk.py"), + ) + self.assertEqual( + loader.locate_file("tests/", "debugtalk.py"), + os.path.join(os.getcwd(), "tests", "debugtalk.py"), + ) + self.assertEqual( + loader.locate_file("tests", "debugtalk.py"), + os.path.join(os.getcwd(), "tests", "debugtalk.py"), + ) + self.assertEqual( + loader.locate_file("tests/base.py", "debugtalk.py"), + os.path.join(os.getcwd(), "tests", "debugtalk.py"), + ) + self.assertEqual( + loader.locate_file("tests/data/demo_testcase.yml", "debugtalk.py"), + os.path.join(os.getcwd(), "tests", "debugtalk.py"), + ) diff --git a/httprunner/parser.py b/httprunner/parser.py index 01aeec14..a43023e6 100644 --- a/httprunner/parser.py +++ b/httprunner/parser.py @@ -3,7 +3,7 @@ import builtins import re from typing import Any, Set, Text, Callable, List, Dict -from httprunner import new_loader, utils, exceptions +from httprunner import loader, utils, exceptions from httprunner.schema import VariablesMapping, FunctionsMapping absolute_http_url_regexp = re.compile(r"^https?://", re.I) @@ -222,7 +222,7 @@ def get_mapping_function( return functions_mapping[function_name] elif function_name in ["parameterize", "P"]: - return new_loader.load_csv_file + return loader.load_csv_file elif function_name in ["environ", "ENV"]: return utils.get_os_environ @@ -235,7 +235,7 @@ def get_mapping_function( try: # check if HttpRunner builtin functions - built_in_functions = new_loader.load_builtin_functions() + built_in_functions = loader.load_builtin_functions() return built_in_functions[function_name] except KeyError: pass diff --git a/httprunner/runner.py b/httprunner/runner.py index 235fdb26..ec6505af 100644 --- a/httprunner/runner.py +++ b/httprunner/runner.py @@ -8,7 +8,7 @@ from loguru import logger from httprunner import utils, exceptions from httprunner.client import HttpSession from httprunner.exceptions import ValidationFailure, ParamsError -from httprunner.new_loader import load_project_meta, load_testcase_file +from httprunner.loader import load_project_meta, load_testcase_file from httprunner.parser import build_url, parse_data, parse_variables_mapping from httprunner.response import ResponseObject from httprunner.schema import (