优化代码
This commit is contained in:
chenyongzhiaaron
parent
88d9fe7beb
commit
4265f28e38
|
|
@ -2,7 +2,7 @@
|
|||
<project version="4">
|
||||
<component name="dataSourceStorageLocal" created-in="PY-231.9161.41">
|
||||
<data-source name="@localhost" uuid="49b6f686-3676-4df5-9645-cd7a2fe91d80">
|
||||
<database-info product="MySQL" version="8.0.32" jdbc-version="4.2" driver-name="MySQL Connector/J" driver-version="mysql-connector-java-8.0.25 (Revision: 08be9e9b4cba6aa115f9b27b215887af40b159e0)" dbms="MYSQL" exact-version="8.0.32" exact-driver-version="8.0">
|
||||
<database-info product="MySQL" version="8.0.26" jdbc-version="4.2" driver-name="MySQL Connector/J" driver-version="mysql-connector-java-8.0.25 (Revision: 08be9e9b4cba6aa115f9b27b215887af40b159e0)" dbms="MYSQL" exact-version="8.0.26" exact-driver-version="8.0">
|
||||
<extra-name-characters>#@</extra-name-characters>
|
||||
<identifier-quote-string>`</identifier-quote-string>
|
||||
</database-info>
|
||||
|
|
|
|||
108
common/action.py
108
common/action.py
|
|
@ -10,60 +10,70 @@
|
|||
import ast
|
||||
|
||||
from common.crypto.encrypt_data import EncryptData
|
||||
from common.database.mysql_client import MysqlClient
|
||||
from common.utils.decorators import singleton
|
||||
from common.utils.load_and_execute_script import LoadScript
|
||||
from common.validation.extractor import Extractor
|
||||
from common.validation.validator import Validator
|
||||
from common.database.mysql_client import MysqlClient
|
||||
|
||||
|
||||
@singleton
|
||||
class Action(Extractor, LoadScript, Validator, MysqlClient):
|
||||
def __init__(self, initialize_data=None, bif_functions=None, db_config=None):
|
||||
super().__init__()
|
||||
MysqlClient.__init__(self, db_config)
|
||||
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, '<string>', 'exec')
|
||||
exec(compiled, {"pm": self})
|
||||
except SyntaxError as e:
|
||||
error_message = f'动态代码语法异常: {e}'
|
||||
self._handle_error(error_message)
|
||||
except Exception as e:
|
||||
error_message = f"动态代码执行异常: {e}"
|
||||
self._handle_error(error_message)
|
||||
return self.__variables
|
||||
|
||||
def _handle_error(self, error_message):
|
||||
print(f'发现异常: {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):
|
||||
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)
|
||||
# 提取请求参数信息
|
||||
if extract_request_data is not None and request_data is not None:
|
||||
self.substitute_data(request_data, jp_dict=extract_request_data)
|
||||
return headers, request_data
|
||||
|
||||
def send_request(self, host, url, method, teardown_script, **kwargs):
|
||||
response = self.http_client(host, url, method, **kwargs)
|
||||
self.update_environments("responseTime", response.elapsed.total_seconds() * 1000) # 存响应时间
|
||||
self.execute_dynamic_code(response, teardown_script)
|
||||
return response
|
||||
def __init__(self, initialize_data=None, bif_functions=None, db_config=None):
|
||||
super().__init__()
|
||||
MysqlClient.__init__(self, db_config)
|
||||
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)
|
||||
self.variables = item
|
||||
if code is not None:
|
||||
try:
|
||||
ast_obj = ast.parse(code, mode='exec')
|
||||
compiled = compile(ast_obj, '<string>', 'exec')
|
||||
exec(compiled, {"pm": self})
|
||||
except SyntaxError as e:
|
||||
error_message = f'动态代码语法异常: {e}'
|
||||
self._handle_error(error_message)
|
||||
except Exception as e:
|
||||
error_message = f"动态代码执行异常: {e}"
|
||||
self._handle_error(error_message)
|
||||
return self.variables
|
||||
# return self.__variables
|
||||
|
||||
def _handle_error(self, error_message):
|
||||
print(f'发现异常: {error_message}')
|
||||
|
||||
@property
|
||||
def variables(self, key=None):
|
||||
return self.__variables if not key else self.__variables.get(key)
|
||||
|
||||
@variables.setter
|
||||
def variables(self, item):
|
||||
self.__variables = item
|
||||
|
||||
# def update_variables(self, key, value):
|
||||
# self.__variables[f"{{{{{key}}}}}"] = value
|
||||
|
||||
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)
|
||||
# 提取请求参数信息
|
||||
if extract_request_data is not None and request_data is not None:
|
||||
self.substitute_data(request_data, jp_dict=extract_request_data)
|
||||
return headers, request_data
|
||||
|
||||
def send_request(self, host, url, method, teardown_script, **kwargs):
|
||||
response = self.http_client(host, url, method, **kwargs)
|
||||
self.update_environments("responseTime", response.elapsed.total_seconds() * 1000) # 存响应时间
|
||||
self.execute_dynamic_code(response, teardown_script)
|
||||
return response
|
||||
|
||||
# Request = property(_get_variables, _set_variables)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
print(Action.__mro__)
|
||||
|
|
|
|||
|
|
@ -0,0 +1,10 @@
|
|||
#!/usr/bin/env python
|
||||
# encoding: utf-8
|
||||
"""
|
||||
@author: kira
|
||||
@contact: 262667641@qq.com
|
||||
@file: get_set.py
|
||||
@time: 2023/7/21 17:07
|
||||
@desc:
|
||||
"""
|
||||
|
||||
Loading…
Reference in New Issue