diff --git a/.idea/api_project.iml b/.idea/api_project.iml
index 8bbb12f..d79ef17 100644
--- a/.idea/api_project.iml
+++ b/.idea/api_project.iml
@@ -5,7 +5,7 @@
-
+
diff --git a/.idea/dataSources.local.xml b/.idea/dataSources.local.xml
index 37e6fc8..3b44b4e 100644
--- a/.idea/dataSources.local.xml
+++ b/.idea/dataSources.local.xml
@@ -1,6 +1,6 @@
-
+
#@
diff --git a/.idea/misc.xml b/.idea/misc.xml
index 8978351..93e7776 100644
--- a/.idea/misc.xml
+++ b/.idea/misc.xml
@@ -1,6 +1,6 @@
-
+
diff --git a/cases/cases/test_api.xlsx b/cases/cases/test_api.xlsx
index f25e7f7..d2adeb1 100644
Binary files a/cases/cases/test_api.xlsx and b/cases/cases/test_api.xlsx differ
diff --git a/common/action.py b/common/action.py
index 55e0652..75a7acb 100644
--- a/common/action.py
+++ b/common/action.py
@@ -18,46 +18,46 @@ from common.validation.validator import Validator
@singleton
class Action(Extractor, LoadScript, Validator):
- def __init__(self, initialize_data=None, bif_functions=None):
- super().__init__()
- self.encrypt = EncryptData()
- self.__variables = {}
- self.set_environments(initialize_data)
- self.set_bif_fun(bif_functions)
-
- def execute_dynamic_code(self, item, code):
- self.set_variables(item)
- if code is not None:
- try:
- ast_obj = ast.parse(code, mode='exec')
- compiled = compile(ast_obj, '', 'exec')
- exec(compiled, {"pm": self})
- except SyntaxError as e:
- error_message = f'Syntax error in dynamic code: {e}'
- self._handle_error(error_message)
- except Exception as e:
- error_message = f"Error executing dynamic code: {e}"
- self._handle_error(error_message)
- finally:
- return self.__variables
- return item
-
- def _handle_error(self, error_message):
- print(f'Error occurred: {error_message}')
-
- def set_variables(self, item):
- self.__variables = item
-
- def update_variables(self, key, value):
- self.__variables[f"{{{{{key}}}}}"] = value
-
- def get_variables(self, key=None):
- """获取依赖表 或 依赖表中key对应的值"""
- return self.__variables if not key else self.__variables.get(key)
-
- def analysis_request(self, request_data, headers_crypto, headers, request_crypto, extract_request_data):
- # 请求头及body加密或者加签
- headers, request_data = self.encrypt.encrypts(headers_crypto, headers, request_crypto, request_data)
- # 提取请求参数信息
- self.substitute_data(request_data, jp_dict=extract_request_data)
- return headers, request_data
+ def __init__(self, initialize_data=None, bif_functions=None):
+ super().__init__()
+ self.encrypt = EncryptData()
+ self.__variables = {}
+ self.set_environments(initialize_data)
+ self.set_bif_fun(bif_functions)
+
+ def execute_dynamic_code(self, item, code):
+ self.set_variables(item)
+ if code is not None:
+ try:
+ ast_obj = ast.parse(code, mode='exec')
+ compiled = compile(ast_obj, '', 'exec')
+ exec(compiled, {"pm": self})
+ except SyntaxError as e:
+ error_message = f'Syntax error in dynamic code: {e}'
+ self._handle_error(error_message)
+ except Exception as e:
+ error_message = f"Error executing dynamic code: {e}"
+ self._handle_error(error_message)
+ finally:
+ return self.__variables
+ return item
+
+ def _handle_error(self, error_message):
+ print(f'Error occurred: {error_message}')
+
+ def set_variables(self, item):
+ self.__variables = item
+
+ def update_variables(self, key, value):
+ self.__variables[f"{{{{{key}}}}}"] = value
+
+ def get_variables(self, key=None):
+ """获取依赖表 或 依赖表中key对应的值"""
+ return self.__variables if not key else self.__variables.get(key)
+
+ def analysis_request(self, request_data, headers_crypto, headers, request_crypto, extract_request_data):
+ # 请求头及body加密或者加签
+ headers, request_data = self.encrypt.encrypts(headers_crypto, headers, request_crypto, request_data)
+ # 提取请求参数信息
+ self.substitute_data(request_data, jp_dict=extract_request_data)
+ return headers, request_data
diff --git a/common/data_extraction/data_extractor.py b/common/data_extraction/data_extractor.py
index e5851ab..b795641 100644
--- a/common/data_extraction/data_extractor.py
+++ b/common/data_extraction/data_extractor.py
@@ -124,7 +124,6 @@ class DataExtractor(Environments):
Returns: 字符串或者list
"""
-
json_path_dict = json_path_dict if isinstance(json_path_dict, dict) else json.loads(json_path_dict)
for key, expression in json_path_dict.items():
try:
diff --git a/common/data_extraction/dependent_parameter.py b/common/data_extraction/dependent_parameter.py
index b0ce61c..b579ec3 100644
--- a/common/data_extraction/dependent_parameter.py
+++ b/common/data_extraction/dependent_parameter.py
@@ -46,7 +46,6 @@ class DependentParameter(DataExtractor):
# 函数替换
key = self.pattern_fun.search(jst).group()
if key in self.get_environments().keys():
-
# 如果参数名称存在于关联参数表中，则调用相应的函数获取返回值，并替换字符串中的参数
value_ = self.get_environments(key)()
jst = jst.replace(key, str(value_))
diff --git a/common/http_client/http_client.py b/common/http_client/http_client.py
index 727d7e4..15d913e 100644
--- a/common/http_client/http_client.py
+++ b/common/http_client/http_client.py
@@ -10,7 +10,6 @@ import urllib3
sys.path.append("../")
sys.path.append("./common")
-from common.http_client import logger
from common.validation.load_modules_from_folder import LoadModulesFromFolder
from common.file_handling.file_utils import FileUtils
from common.utils.decorators import request_decorator
diff --git a/test_script/test_api.py b/test_script/test_api.py
index 8dba591..0a192d4 100644
--- a/test_script/test_api.py
+++ b/test_script/test_api.py
@@ -25,173 +25,174 @@ host = init_data.get('host', "") + init_data.get("path", "")
@ddt
class TestProjectApi(unittest.TestCase):
- maxDiff = None
- action = Action(initialize_data, bif_functions)
-
- @classmethod
- def setUpClass(cls) -> None:
- pass
-
- def setUp(self) -> None:
- self.action.set_bif_fun(dynamic_scaling_methods)
-
- @data(*test_case)
- def test_api(self, item):
-
- sheet, iid, condition, st, name, desc, h_crypto, r_crypto, method = self.__base_info(item)
- regex, keys, deps, jp_dict, extract_request_data = self.__extractor_info(item)
- setup_script, teardown_script = self.script(item)
-
- if self.__is_run(condition):
- return
-
- self.__pause_execution(st)
-
- # 首执行 sql
- self.__exc_sql(item, method)
-
- # 执行动态代码
- item = self.action.execute_dynamic_code(item, setup_script)
-
- # prepost_script = f"prepost_script_{sheet}_{iid}.py"
- # item = self.action.load_and_execute_script(Config.SCRIPTS_DIR, prepost_script, "setup", item)
-
- # 修正参数
- item = self.action.replace_dependent_parameter(item)
- url, query_str, request_data, headers, expected, request_data_type = self.__request_info(item)
-
- # 分析请求参数信息
- headers, request_data = self.action.analysis_request(request_data, h_crypto, headers, r_crypto,extract_request_data)
- result_tuple = None
- result = "PASS"
- response = None
-
- try:
- # 执行请求操作
- kwargs = {request_data_type: request_data, 'headers': headers, "params": query_str}
- response = self.action.http_client(host, url, method, **kwargs)
-
- # 执行后置代码片段
- self.action.execute_dynamic_code(response, teardown_script)
-
- # 执行断言 返回结果元组
- result_tuple = self.action.run_validate(expected, response.json())
- self.assertNotIn("FAIL", result_tuple, "FAIL 存在结果元组中")
- try:
- # 提取响应
- self.action.substitute_data(response.json(), regex=regex, keys=keys, deps=deps, jp_dict=jp_dict)
-
- except Exception as err:
- logger.error(f"提取响应失败：{sheet}_{iid}_{name}_{desc}"
- f"\nregex={regex};"
- f" \nkeys={keys};"
- f"\ndeps={deps};"
- f"\njp_dict={jp_dict}"
- f"\n{err}")
- except Exception as e:
- result = "FAIL"
- logger.error(f'异常用例: {sheet}_{iid}_{name}_{desc}\n{e}')
- raise e
- finally:
- response = response.text if response is not None else str(response)
- # 响应结果及测试结果回写 excel
- excel.write_back(sheet_name=sheet, i=iid, response=response, test_result=result,
- assert_log=str(result_tuple))
-
- @classmethod
- def tearDownClass(cls) -> None:
- excel.close_excel()
- logger.info(f"所有用例执行完毕")
-
- @staticmethod
- def __base_info(item):
- """
- 获取基础信息
- """
- sheet = item.pop("sheet")
- item_id = item.pop("Id")
- condition = item.pop("Run")
- sleep_time = item.pop("Time")
- name = item.pop("Name")
- desc = item.pop("Description")
- headers_crypto = item.pop("Headers Crypto")
- request_data_crypto = item.pop("Request Data Crypto")
- method = item.pop("Method")
- return sheet, item_id, condition, sleep_time, name, desc, headers_crypto, request_data_crypto, method
-
- @staticmethod
- def __sql_info(item):
- sql = item.pop("SQL")
- sql_params_dict = item.pop("Sql Params Dict")
- return sql, sql_params_dict
-
- @staticmethod
- def __extractor_info(item):
- """
- 获取提取参数的基本字段信息
- Args:
- item:
-
- Returns:
-
- """
- regex = item.pop("Regex")
- keys = item.pop("Regex Params List")
- deps = item.pop("Retrieve Value")
- jp_dict = item.pop("Jsonpath")
- extract_request_data = item.pop("Extract Request Data")
- return regex, keys, deps, jp_dict, extract_request_data
-
- @staticmethod
- def __request_info(item):
- """
- 请求数据
- """
- url = item.pop("Url")
- query_str = item.pop("Query Str")
- request_data = item.pop("Request Data")
- headers = item.pop("Headers")
- expected = item.pop("Expected")
- request_data_type = item.pop("Request Data Type") if item.get("Request Data Type") else 'params'
-
- return url, query_str, request_data, headers, expected, request_data_type
-
- @staticmethod
- def script(item):
- setup_script = item.pop("Setup Script")
- teardown_script = item.pop("Teardown Script")
- return setup_script, teardown_script
-
- @staticmethod
- def __is_run(condition):
- is_run = condition
- if not is_run or is_run.upper() != "YES":
- return True
-
- @staticmethod
- def __pause_execution(sleep_time):
- if sleep_time:
- try:
- time.sleep(sleep_time)
- except Exception as e:
- logger.error("暂时时间必须是数字")
- raise e
-
- def __exc_sql(self, item, method):
- sql, sql_params_dict = self.__sql_info(item)
- sql = self.action.replace_dependent_parameter(sql)
- if sql:
- try:
- execute_sql_results = mysql.do_mysql(sql)
- if execute_sql_results and sql_params_dict:
- self.action.extract_request_data(execute_sql_results, jp_dict=sql_params_dict)
- if method == "SQL" and mysql:
- return None
- except Exception as e:
- logger.error(f'执行 sql 失败:{sql},异常信息:{e}')
- raise e
- return sql
+ maxDiff = None
+ action = Action(initialize_data, bif_functions)
+
+ @classmethod
+ def setUpClass(cls) -> None:
+ pass
+
+ def setUp(self) -> None:
+ self.action.set_bif_fun(dynamic_scaling_methods)
+
+ @data(*test_case)
+ def test_api(self, item):
+
+ sheet, iid, condition, st, name, desc, h_crypto, r_crypto, method = self.__base_info(item)
+ regex, keys, deps, jp_dict, extract_request_data = self.__extractor_info(item)
+ setup_script, teardown_script = self.script(item)
+
+ if self.__is_run(condition):
+ return
+
+ self.__pause_execution(st)
+
+ # 首执行 sql
+ self.__exc_sql(item, method)
+
+ # 执行动态代码
+ item = self.action.execute_dynamic_code(item, setup_script)
+
+ # prepost_script = f"prepost_script_{sheet}_{iid}.py"
+ # item = self.action.load_and_execute_script(Config.SCRIPTS_DIR, prepost_script, "setup", item)
+
+ # 修正参数
+ item = self.action.replace_dependent_parameter(item)
+ url, query_str, request_data, headers, expected, request_data_type = self.__request_info(item)
+
+ # 分析请求参数信息
+ headers, request_data = self.action.analysis_request(request_data, h_crypto, headers, r_crypto,
+ extract_request_data)
+ result_tuple = None
+ result = "PASS"
+ response = None
+
+ try:
+ # 执行请求操作
+ kwargs = {request_data_type: request_data, 'headers': headers, "params": query_str}
+ response = self.action.http_client(host, url, method, **kwargs)
+
+ # 执行后置代码片段
+ self.action.execute_dynamic_code(response, teardown_script)
+
+ # 执行断言 返回结果元组
+ result_tuple = self.action.run_validate(expected, response.json())
+ self.assertNotIn("FAIL", result_tuple, "FAIL 存在结果元组中")
+ try:
+ # 提取响应
+ self.action.substitute_data(response.json(), regex=regex, keys=keys, deps=deps, jp_dict=jp_dict)
+
+ except Exception as err:
+ logger.error(f"提取响应失败：{sheet}_{iid}_{name}_{desc}"
+ f"\nregex={regex};"
+ f" \nkeys={keys};"
+ f"\ndeps={deps};"
+ f"\njp_dict={jp_dict}"
+ f"\n{err}")
+ except Exception as e:
+ result = "FAIL"
+ logger.error(f'异常用例: {sheet}_{iid}_{name}_{desc}\n{e}')
+ raise e
+ finally:
+ response = response.text if response is not None else str(response)
+ # 响应结果及测试结果回写 excel
+ excel.write_back(sheet_name=sheet, i=iid, response=response, test_result=result,
+ assert_log=str(result_tuple))
+
+ @classmethod
+ def tearDownClass(cls) -> None:
+ excel.close_excel()
+ logger.info(f"所有用例执行完毕")
+
+ @staticmethod
+ def __base_info(item):
+ """
+ 获取基础信息
+ """
+ sheet = item.pop("sheet")
+ item_id = item.pop("Id")
+ condition = item.pop("Run")
+ sleep_time = item.pop("Time")
+ name = item.pop("Name")
+ desc = item.pop("Description")
+ headers_crypto = item.pop("Headers Crypto")
+ request_data_crypto = item.pop("Request Data Crypto")
+ method = item.pop("Method")
+ return sheet, item_id, condition, sleep_time, name, desc, headers_crypto, request_data_crypto, method
+
+ @staticmethod
+ def __sql_info(item):
+ sql = item.pop("SQL")
+ sql_params_dict = item.pop("Sql Params Dict")
+ return sql, sql_params_dict
+
+ @staticmethod
+ def __extractor_info(item):
+ """
+ 获取提取参数的基本字段信息
+ Args:
+ item:
+
+ Returns:
+
+ """
+ regex = item.pop("Regex")
+ keys = item.pop("Regex Params List")
+ deps = item.pop("Retrieve Value")
+ jp_dict = item.pop("Jsonpath")
+ extract_request_data = item.pop("Extract Request Data")
+ return regex, keys, deps, jp_dict, extract_request_data
+
+ @staticmethod
+ def __request_info(item):
+ """
+ 请求数据
+ """
+ url = item.pop("Url")
+ query_str = item.pop("Query Str")
+ request_data = item.pop("Request Data")
+ headers = item.pop("Headers")
+ expected = item.pop("Expected")
+ request_data_type = item.pop("Request Data Type") if item.get("Request Data Type") else 'params'
+
+ return url, query_str, request_data, headers, expected, request_data_type
+
+ @staticmethod
+ def script(item):
+ setup_script = item.pop("Setup Script")
+ teardown_script = item.pop("Teardown Script")
+ return setup_script, teardown_script
+
+ @staticmethod
+ def __is_run(condition):
+ is_run = condition
+ if not is_run or is_run.upper() != "YES":
+ return True
+
+ @staticmethod
+ def __pause_execution(sleep_time):
+ if sleep_time:
+ try:
+ time.sleep(sleep_time)
+ except Exception as e:
+ logger.error("暂时时间必须是数字")
+ raise e
+
+ def __exc_sql(self, item, method):
+ sql, sql_params_dict = self.__sql_info(item)
+ sql = self.action.replace_dependent_parameter(sql)
+ if sql:
+ try:
+ execute_sql_results = mysql.do_mysql(sql)
+ if execute_sql_results and sql_params_dict:
+ self.action.extract_request_data(execute_sql_results, jp_dict=sql_params_dict)
+ if method == "SQL" and mysql:
+ return None
+ except Exception as e:
+ logger.error(f'执行 sql 失败:{sql},异常信息:{e}')
+ raise e
+ return sql
if __name__ == '__main__':
- unittest.main()
+ unittest.main()