kyfx
/
apitest_unittest
mirror of https://gitee.com/chenyongzhiaaron_admin/apitest.git
5
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

兼容处理上传文件的接口。允许多个文件一起上传。

This commit is contained in:
chenyongzhiaaron 2023-07-12 17:43:38 +08:00
parent 2eae02a410
commit 3d41b5f62a
10 changed files with 172 additions and 112 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.idea/dataSources.local.xml
View File

@ -1,6 +1,6 @@
<?xml version="1.0" encoding="UTF-8"?>
<project version="4">
<component name="dataSourceStorageLocal" created-in="PY-231.9011.38">
<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.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>

25
OutPut/log/2023-07-12.log Normal file
View File

@ -0,0 +1,25 @@
2023-07-12 11:17:52 | ERROR | key:<re.Match object; span=(282, 289), match='{{999}}'>,在关联参数表中查询不到,请检查关联参数字段提取及填写是否正常
2023-07-12 11:17:52 | ERROR | 异常用例: 安全纯净大屏_2_登录_非BIP用户登录
Expecting value: line 1 column 1 (char 0)
2023-07-12 11:17:54 | ERROR | key:<re.Match object; span=(367, 374), match='{{999}}'>,在关联参数表中查询不到,请检查关联参数字段提取及填写是否正常
2023-07-12 11:17:54 | ERROR | 异常用例: 安全纯净大屏_4_劳务基础配置_绑定TV
Expecting value: line 1 column 1 (char 0)
2023-07-12 11:17:57 | ERROR | key:<re.Match object; span=(320, 327), match='{{999}}'>,在关联参数表中查询不到,请检查关联参数字段提取及填写是否正常
2023-07-12 11:17:57 | ERROR | 被提取对象非字典、非字符串、非列表，不执行jsonpath提取，被提取对象: None
2023-07-12 11:17:57 | ERROR | 异常用例: 安全纯净大屏_5_劳务基础配置_查询配置
Expecting value: line 1 column 1 (char 0)
2023-07-12 15:06:03 | ERROR | 发送请求失败: list indices must be integers or slices, not str
2023-07-12 15:12:59 | ERROR | 发送请求失败: list indices must be integers or slices, not str
2023-07-12 15:13:12 | ERROR | 发送请求失败: list indices must be integers or slices, not str
2023-07-12 15:13:35 | ERROR | 发送请求失败: list indices must be integers or slices, not str
2023-07-12 15:14:04 | ERROR | 发送请求失败: list indices must be integers or slices, not str
2023-07-12 15:14:27 | ERROR | 发送请求失败: list indices must be integers or slices, not str
2023-07-12 15:15:51 | ERROR | 发送请求失败: read of closed file
2023-07-12 15:24:51 | ERROR | 发送请求失败: read of closed file
2023-07-12 16:07:16 | ERROR | 发送请求失败: read of closed file
2023-07-12 16:07:38 | ERROR | 发送请求失败: read of closed file
2023-07-12 16:24:30 | ERROR | 发送请求失败: URL cannot be None
2023-07-12 16:25:45 | ERROR | 发送请求失败: Data must not be a string.

1
README.md
View File

@ -377,6 +377,7 @@ None，貌似没有对 None 特殊处理
![img.png](/image/img_1.png)
- 这个MD5方法，一般都没有使用，一般都是直接excel中是否使用【参数加密方式字段开关】来处理加密
![](image/img_31.png)
![img.png](/image/img.png)
- 内置函数使用，统一是 `{{xxx()}}`，可以传参数到()内,比如 `{{token(999)}}`
![img.png](/image/img_2.png)

BIN
cases/cases/test_cases.xlsx
View File

Binary file not shown.

0
cases/files/test.txt Normal file
View File

0
cases/files/test2.txt Normal file
View File

57
common/config.py
View File

@ -2,34 +2,35 @@ import os
class Config:
# 根目录路径
# *****************************************************************
base_path = os.path.dirname(os.path.dirname(__file__))
current_path = os.path.dirname(__file__)
# *****************************************************************
# 测试数据所在路径
# *****************************************************************
templates = os.path.join(base_path, "cases", "templates", "template.xlsx") # 模板文件
test_case = os.path.join(base_path, "cases", "cases", "test_cases.xlsx")
# *****************************************************************
# 测试用例脚本目录
# *****************************************************************
script = os.path.join(base_path, "test_script")
# *****************************************************************
# 测试报告及 logger 所在路径
# *****************************************************************
test_report = os.path.join(base_path, "output", "reports")
log_path = os.path.join(base_path, "output", "log")
SCRIPTS_DIR = os.path.join(base_path, "scripts")
# 根目录路径
# *****************************************************************
base_path = os.path.dirname(os.path.dirname(__file__))
current_path = os.path.dirname(__file__)
# *****************************************************************
# 测试数据所在路径
# *****************************************************************
templates = os.path.join(base_path, "cases", "templates", "template.xlsx") # 模板文件
test_case = os.path.join(base_path, "cases", "cases", "test_cases.xlsx")
test_files = os.path.join(base_path, 'cases', 'files')
# *****************************************************************
# 测试用例脚本目录
# *****************************************************************
script = os.path.join(base_path, "test_script")
# *****************************************************************
# 测试报告及 logger 所在路径
# *****************************************************************
test_report = os.path.join(base_path, "output", "reports")
log_path = os.path.join(base_path, "output", "log")
SCRIPTS_DIR = os.path.join(base_path, "scripts")
if __name__ == '__main__':
test = Config()
print(test.base_path)
print(test.test_case)
print(test.test_report)
print(test.test_case)
print(test.current_path)
print(test.SCRIPTS_DIR)
test = Config()
print(test.base_path)
print(test.test_case)
print(test.test_report)
print(test.test_case)
print(test.current_path)
print(test.SCRIPTS_DIR)

31
common/file_handling/file_utils.py
View File

@ -14,6 +14,8 @@ from configparser import RawConfigParser
import yaml
from common.config import Config
class FileUtils:
@staticmethod
@ -30,7 +32,7 @@ class FileUtils:
for root, dirs, files in os.walk(open_file_path):
path_list.extend([os.path.join(root, file) for file in files])
return path_list
@staticmethod
def get_files_in_folder(folder_path):
"""
@ -49,7 +51,16 @@ class FileUtils:
if os.path.isfile(file_path):
file_list.append(filename)
return file_list
@staticmethod
def get_file_path(file_name):
"""根据文件名获取指定目录下的文件路径"""
for root, dirs, files in os.walk(Config.test_files):
for file in files:
if file == file_name:
return os.path.join(root, file)
return
@staticmethod
def get_folders_in_path(dir_path):
"""
@ -68,7 +79,7 @@ class FileUtils:
if os.path.isdir(folder_path):
folder_list.append(foldername)
return folder_list
@staticmethod
def read_file(file_path):
"""
@ -83,7 +94,7 @@ class FileUtils:
raise ValueError("Invalid file path")
with open(file_path, "r", encoding="utf-8") as f:
return f.read()
@staticmethod
def read_json_file(file_path):
"""
@ -99,7 +110,7 @@ class FileUtils:
return json.loads(content)
except json.JSONDecodeError as e:
raise ValueError("Invalid JSON file: {}".format(e))
@staticmethod
def read_yaml_file(file_path):
"""
@ -112,7 +123,7 @@ class FileUtils:
"""
with open(file_path, "r", encoding="utf-8") as f:
return yaml.safe_load(f)
@staticmethod
def get_value_from_dict(data, key_path):
"""
@ -126,15 +137,15 @@ class FileUtils:
"""
if isinstance(key_path, str):
key_path = key_path.split('.')
for key in key_path:
if isinstance(data, dict) and key in data:
data = data[key]
else:
return None
return data
@staticmethod
def read_config_data(file_path, section, option):
"""
@ -150,7 +161,7 @@ class FileUtils:
cf = RawConfigParser()
cf.read(file_path, encoding="UTF-8")
return eval(cf.get(section, option))
@staticmethod
def read_json_data(file_path):
"""

168
common/http_client/http_client.py
View File

@ -11,82 +11,104 @@ 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
class Pyt(LoadModulesFromFolder):
# 类属性，保存一个会话对象，防止每次都创建一个新的会话，节省资源
session = requests.Session()
def __init__(self):
super().__init__()
self.request = None
self.response = None
def log_decorator(msg="请求异常"):
def decorator(func):
def wrapper(*args, **kwargs):
try:
print(f"发送请求的参数： {kwargs}")
response = func(*args, **kwargs)
print(f"请求地址 --> {response.request.url}")
print(f"请求头 --> {response.request.headers}")
print(f"请求 body --> {response.request.body}")
print(f"接口状态--> {response.status_code}")
print(f"接口耗时--> {response.elapsed}")
print(f"接口响应--> {response.text}")
return response
except Exception as e:
logger.error(f"发送请求失败: {e}")
return
return wrapper
return decorator
# def pre_script(self,name):
# def decorator(func):
# self.variables = func
# return decorator
@log_decorator()
def http_client(self, host, url, method, **kwargs):
"""
发送 http 请求
@param host: 域名
@param url: 接口 __url
@param method: http 请求方法
@param kwargs: 接受 requests 原生的关键字参数
@return: 响应对象
"""
# 关闭 https 警告信息
urllib3.disable_warnings()
if not url:
raise ValueError("URL cannot be None")
__url = f'{host}{url}' if not re.match(r"https?", url) else url
# 增加兼容
# 处理 headers 参数为字符串类型的情况
if 'headers' in kwargs and isinstance(kwargs['headers'], str):
kwargs['headers'] = json.loads(kwargs['headers'])
# 处理 json 参数为字符串类型的情况
if 'json' in kwargs and isinstance(kwargs['json'], str):
kwargs['json'] = json.loads(kwargs['json'])
# 发送请求
self.request = requests.Request(method, __url, **kwargs)
self.response = self.session.send(self.request.prepare(), timeout=30, verify=True)
return self.response
# 类属性，保存一个会话对象，防止每次都创建一个新的会话，节省资源
session = requests.Session()
file_utils = FileUtils()
def __init__(self):
super().__init__()
self.request = None
self.response = None
def log_decorator(msg="请求异常"):
def decorator(func):
def wrapper(*args, **kwargs):
try:
print(f"发送请求的参数： {kwargs}")
response = func(*args, **kwargs)
print(f"请求地址 --> {response.request.url}")
print(f"请求头 --> {response.request.headers}")
print(f"请求 body --> {response.request.body}")
print(f"接口状态--> {response.status_code}")
print(f"接口耗时--> {response.elapsed}")
print(f"接口响应--> {response.text}")
return response
except Exception as e:
logger.error(f"发送请求失败: {e}")
return
return wrapper
return decorator
# def pre_script(self,name):
# def decorator(func):
# self.variables = func
# return decorator
@log_decorator()
def http_client(self, host, url, method, **kwargs):
"""
发送 http 请求
@param host: 域名
@param url: 接口 __url
@param method: http 请求方法
@param kwargs: 接受 requests 原生的关键字参数
@return: 响应对象
"""
# 关闭 https 警告信息
urllib3.disable_warnings()
if not url:
raise ValueError("URL cannot be None")
__url = f'{host}{url}' if not re.match(r"https?", url) else url
# 增加兼容
# 处理 headers 参数为字符串类型的情况
if 'headers' in kwargs and isinstance(kwargs['headers'], str):
kwargs['headers'] = json.loads(kwargs['headers'])
# 处理 json 参数为字符串类型的情况
if 'json' in kwargs and isinstance(kwargs['json'], str):
kwargs['json'] = json.loads(kwargs['json'])
# 处理 files 参数的情况
if 'files' in kwargs:
file_paths = kwargs['files']
if isinstance(file_paths, str):
file_paths = json.loads(file_paths)
files = []
fs = []
for i, file_path in enumerate(file_paths):
file_path_completion = self.file_utils.get_file_path(file_path)
f = open(file_path_completion, 'rb')
fs.append(f)
files.append(
('file', (f'{file_path}', f))
)
kwargs['files'] = files
print(kwargs)
# 发送请求
self.request = requests.Request(method, __url, **kwargs)
self.response = self.session.send(self.request.prepare(), timeout=30, verify=True)
[i.close() for i in fs if len(fs) > 0]
return self.response
if __name__ == '__main__':
hst = 'https://baidu.com'
url = '/login?t=168672334'
method = 'post'
kwargs = {
'json': '{"account": "kira","password": "9999999"}',
'headers': None}
pyt = Pyt()
pyt.http_client(hst, url, method, **kwargs)
hst = 'htt.com'
# url = '/File'
url = '/ge'
method = 'post'
kwargs = {
'headers': {},
'data': {},
'files': ['test.txt']
}
pyt = Pyt()
pyt.http_client(hst, url, method, **kwargs)

BIN
image/img_31.png Normal file
View File

Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 329 KiB