932 lines
31 KiB
Python
932 lines
31 KiB
Python
#!/usr/bin/env python
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
测试图片管理接口性能并生成CSV报告和接口文档
|
||
"""
|
||
import requests
|
||
import time
|
||
import csv
|
||
from datetime import datetime
|
||
import io
|
||
from PIL import Image
|
||
|
||
# API配置
|
||
API_BASE_URL = "http://127.0.0.1:8216"
|
||
USERNAME = "13621242430"
|
||
PASSWORD = "admin123"
|
||
|
||
# 测试结果存储
|
||
test_results = []
|
||
api_docs = []
|
||
|
||
def login():
|
||
"""登录获取token"""
|
||
api_docs.append({
|
||
'endpoint': '/api/auth/login',
|
||
'method': 'POST',
|
||
'description': '用户登录',
|
||
'auth': '无需认证',
|
||
'params': 'username, password',
|
||
'response': '返回token和用户信息'
|
||
})
|
||
|
||
print("\n" + "="*80)
|
||
print("正在登录...")
|
||
url = f"{API_BASE_URL}/api/auth/login"
|
||
data = {"username": USERNAME, "password": PASSWORD}
|
||
|
||
start_time = time.time()
|
||
response = requests.post(url, json=data, timeout=30)
|
||
elapsed_time = (time.time() - start_time) * 1000
|
||
|
||
if response.status_code == 200:
|
||
token = response.json()['data']['token']
|
||
print(f"✓ 登录成功，耗时: {elapsed_time:.2f}ms")
|
||
return token
|
||
else:
|
||
print(f"✗ 登录失败: {response.status_code}")
|
||
return None
|
||
|
||
def test_get_images_list(token):
|
||
"""测试获取图片列表"""
|
||
print("\n" + "-"*80)
|
||
print("[1/16] 测试获取图片列表...")
|
||
|
||
api_docs.append({
|
||
'endpoint': '/api/images/list',
|
||
'method': 'GET',
|
||
'description': '获取图片列表（支持分页和筛选）',
|
||
'auth': '需要认证',
|
||
'params': 'page, pageSize, keyword, product_id, type',
|
||
'response': '返回图片列表（含标签信息、产品信息）'
|
||
})
|
||
|
||
url = f"{API_BASE_URL}/api/images/list"
|
||
headers = {"Authorization": f"Bearer {token}"}
|
||
|
||
test_cases = [
|
||
{"page": 1, "pageSize": 20, "desc": "第1页"},
|
||
{"page": 1, "pageSize": 10, "desc": "每页10条"},
|
||
]
|
||
|
||
for i, params in enumerate(test_cases, 1):
|
||
desc = params.pop('desc')
|
||
try:
|
||
start_time = time.time()
|
||
response = requests.get(url, headers=headers, params=params, timeout=30)
|
||
elapsed_time = (time.time() - start_time) * 1000
|
||
|
||
success = response.status_code == 200
|
||
error = '' if success else response.text[:100]
|
||
|
||
print(f" [{i}] {desc}: {response.status_code} - {elapsed_time:.2f}ms")
|
||
|
||
test_results.append({
|
||
'endpoint': f'/api/images/list ({desc})',
|
||
'method': 'GET',
|
||
'status_code': response.status_code,
|
||
'elapsed_time_ms': f"{elapsed_time:.2f}",
|
||
'success': success,
|
||
'error': error
|
||
})
|
||
|
||
time.sleep(0.1)
|
||
except Exception as e:
|
||
print(f" [{i}] {desc}: 失败 - {str(e)}")
|
||
test_results.append({
|
||
'endpoint': f'/api/images/list ({desc})',
|
||
'method': 'GET',
|
||
'status_code': 0,
|
||
'elapsed_time_ms': '0',
|
||
'success': False,
|
||
'error': str(e)
|
||
})
|
||
|
||
def test_get_images_dashboard(token):
|
||
"""测试获取图片库仪表盘数据"""
|
||
print("\n" + "-"*80)
|
||
print("[2/16] 测试获取图片库仪表盘数据...")
|
||
|
||
api_docs.append({
|
||
'endpoint': '/api/images/list_dashboard',
|
||
'method': 'GET',
|
||
'description': '获取图片库仪表盘统计数据',
|
||
'auth': '需要认证',
|
||
'params': '无',
|
||
'response': '返回图片总数、产品图片总数、产品类型总数、场景图片总数'
|
||
})
|
||
|
||
url = f"{API_BASE_URL}/api/images/list_dashboard"
|
||
headers = {"Authorization": f"Bearer {token}"}
|
||
|
||
try:
|
||
start_time = time.time()
|
||
response = requests.get(url, headers=headers, timeout=30)
|
||
elapsed_time = (time.time() - start_time) * 1000
|
||
|
||
success = response.status_code == 200
|
||
error = '' if success else response.text[:100]
|
||
|
||
print(f" 状态码: {response.status_code} - 耗时: {elapsed_time:.2f}ms")
|
||
|
||
test_results.append({
|
||
'endpoint': '/api/images/list_dashboard',
|
||
'method': 'GET',
|
||
'status_code': response.status_code,
|
||
'elapsed_time_ms': f"{elapsed_time:.2f}",
|
||
'success': success,
|
||
'error': error
|
||
})
|
||
except Exception as e:
|
||
print(f" 失败: {str(e)}")
|
||
test_results.append({
|
||
'endpoint': '/api/images/list_dashboard',
|
||
'method': 'GET',
|
||
'status_code': 0,
|
||
'elapsed_time_ms': '0',
|
||
'success': False,
|
||
'error': str(e)
|
||
})
|
||
|
||
def test_upload_image(token):
|
||
"""测试上传图片"""
|
||
print("\n" + "-"*80)
|
||
print("[3/16] 测试上传图片...")
|
||
|
||
api_docs.append({
|
||
'endpoint': '/api/images/upload',
|
||
'method': 'POST',
|
||
'description': '上传图片（支持真实文件上传、图片压缩、生成缩略图）',
|
||
'auth': '需要认证',
|
||
'params': 'image (file), image_name, image_type_name, image_type_id, tag_keywords, product_id, product_name, description',
|
||
'response': '返回图片ID、URL、缩略图URL、宽高、文件大小'
|
||
})
|
||
|
||
url = f"{API_BASE_URL}/api/images/upload"
|
||
headers = {"Authorization": f"Bearer {token}"}
|
||
|
||
# 创建一个测试图片
|
||
try:
|
||
img = Image.new('RGB', (100, 100))
|
||
img_bytes = io.BytesIO()
|
||
img.save(img_bytes, format='PNG')
|
||
img_bytes.seek(0)
|
||
|
||
files = {'image': ('test.png', img_bytes, 'image/png')}
|
||
data = {
|
||
'image_name': '测试图片',
|
||
'image_type_name': '测试类型',
|
||
'tag_keywords': '测试,性能'
|
||
}
|
||
|
||
start_time = time.time()
|
||
response = requests.post(url, headers=headers, files=files, data=data, timeout=30)
|
||
elapsed_time = (time.time() - start_time) * 1000
|
||
|
||
success = response.status_code == 200
|
||
error = '' if success else response.text[:100]
|
||
|
||
print(f" 状态码: {response.status_code} - 耗时: {elapsed_time:.2f}ms")
|
||
|
||
test_results.append({
|
||
'endpoint': '/api/images/upload',
|
||
'method': 'POST',
|
||
'status_code': response.status_code,
|
||
'elapsed_time_ms': f"{elapsed_time:.2f}",
|
||
'success': success,
|
||
'error': error
|
||
})
|
||
except Exception as e:
|
||
print(f" 失败: {str(e)}")
|
||
test_results.append({
|
||
'endpoint': '/api/images/upload',
|
||
'method': 'POST',
|
||
'status_code': 0,
|
||
'elapsed_time_ms': '0',
|
||
'success': False,
|
||
'error': str(e)
|
||
})
|
||
|
||
def test_delete_image(token):
|
||
"""测试删除图片"""
|
||
print("\n" + "-"*80)
|
||
print("[4/16] 测试删除图片...")
|
||
|
||
api_docs.append({
|
||
'endpoint': '/api/images/<image_id>',
|
||
'method': 'DELETE',
|
||
'description': '删除图片（软删除）',
|
||
'auth': '需要认证',
|
||
'params': 'image_id (路径参数)',
|
||
'response': '返回删除成功消息'
|
||
})
|
||
|
||
url = f"{API_BASE_URL}/api/images/999999"
|
||
headers = {"Authorization": f"Bearer {token}"}
|
||
|
||
try:
|
||
start_time = time.time()
|
||
response = requests.delete(url, headers=headers, timeout=30)
|
||
elapsed_time = (time.time() - start_time) * 1000
|
||
|
||
success = response.status_code in [200, 404]
|
||
error = '' if success else response.text[:100]
|
||
|
||
print(f" 状态码: {response.status_code} - 耗时: {elapsed_time:.2f}ms")
|
||
|
||
test_results.append({
|
||
'endpoint': '/api/images/<image_id>',
|
||
'method': 'DELETE',
|
||
'status_code': response.status_code,
|
||
'elapsed_time_ms': f"{elapsed_time:.2f}",
|
||
'success': success,
|
||
'error': error
|
||
})
|
||
except Exception as e:
|
||
print(f" 失败: {str(e)}")
|
||
test_results.append({
|
||
'endpoint': '/api/images/<image_id>',
|
||
'method': 'DELETE',
|
||
'status_code': 0,
|
||
'elapsed_time_ms': '0',
|
||
'success': False,
|
||
'error': str(e)
|
||
})
|
||
|
||
def test_tag_names_list(token):
|
||
"""测试获取标签名称列表"""
|
||
print("\n" + "-"*80)
|
||
print("[5/16] 测试获取标签名称列表...")
|
||
|
||
api_docs.append({
|
||
'endpoint': '/api/images/tags/names/list',
|
||
'method': 'GET',
|
||
'description': '获取标签名称列表（ai_image_tags_name表）',
|
||
'auth': '需要认证',
|
||
'params': 'page, pageSize, keyword, status',
|
||
'response': '返回标签名称列表'
|
||
})
|
||
|
||
url = f"{API_BASE_URL}/api/images/tags/names/list"
|
||
headers = {"Authorization": f"Bearer {token}"}
|
||
params = {"page": 1, "pageSize": 20}
|
||
|
||
try:
|
||
start_time = time.time()
|
||
response = requests.get(url, headers=headers, params=params, timeout=30)
|
||
elapsed_time = (time.time() - start_time) * 1000
|
||
|
||
success = response.status_code == 200
|
||
error = '' if success else response.text[:100]
|
||
|
||
print(f" 状态码: {response.status_code} - 耗时: {elapsed_time:.2f}ms")
|
||
|
||
test_results.append({
|
||
'endpoint': '/api/images/tags/names/list',
|
||
'method': 'GET',
|
||
'status_code': response.status_code,
|
||
'elapsed_time_ms': f"{elapsed_time:.2f}",
|
||
'success': success,
|
||
'error': error
|
||
})
|
||
except Exception as e:
|
||
print(f" 失败: {str(e)}")
|
||
test_results.append({
|
||
'endpoint': '/api/images/tags/names/list',
|
||
'method': 'GET',
|
||
'status_code': 0,
|
||
'elapsed_time_ms': '0',
|
||
'success': False,
|
||
'error': str(e)
|
||
})
|
||
|
||
def test_create_tag_name(token):
|
||
"""测试创建标签名称"""
|
||
print("\n" + "-"*80)
|
||
print("[6/16] 测试创建标签名称...")
|
||
|
||
api_docs.append({
|
||
'endpoint': '/api/images/tags/names/create',
|
||
'method': 'POST',
|
||
'description': '创建标签名称',
|
||
'auth': '需要认证',
|
||
'params': 'tag_name, tag_category, department, description, status',
|
||
'response': '返回新创建的标签ID'
|
||
})
|
||
|
||
url = f"{API_BASE_URL}/api/images/tags/names/create"
|
||
headers = {"Authorization": f"Bearer {token}"}
|
||
timestamp = int(time.time())
|
||
data = {
|
||
"tag_name": f"测试标签{timestamp}",
|
||
"description": "测试标签描述"
|
||
}
|
||
|
||
try:
|
||
start_time = time.time()
|
||
response = requests.post(url, headers=headers, json=data, timeout=30)
|
||
elapsed_time = (time.time() - start_time) * 1000
|
||
|
||
success = response.status_code == 200
|
||
error = '' if success else response.text[:100]
|
||
|
||
print(f" 状态码: {response.status_code} - 耗时: {elapsed_time:.2f}ms")
|
||
|
||
test_results.append({
|
||
'endpoint': '/api/images/tags/names/create',
|
||
'method': 'POST',
|
||
'status_code': response.status_code,
|
||
'elapsed_time_ms': f"{elapsed_time:.2f}",
|
||
'success': success,
|
||
'error': error
|
||
})
|
||
except Exception as e:
|
||
print(f" 失败: {str(e)}")
|
||
test_results.append({
|
||
'endpoint': '/api/images/tags/names/create',
|
||
'method': 'POST',
|
||
'status_code': 0,
|
||
'elapsed_time_ms': '0',
|
||
'success': False,
|
||
'error': str(e)
|
||
})
|
||
|
||
def test_update_tag_name(token):
|
||
"""测试更新标签名称"""
|
||
print("\n" + "-"*80)
|
||
print("[7/16] 测试更新标签名称...")
|
||
|
||
api_docs.append({
|
||
'endpoint': '/api/images/tags/names/<tag_id>',
|
||
'method': 'PUT',
|
||
'description': '更新标签名称',
|
||
'auth': '需要认证',
|
||
'params': 'tag_id (路径参数), tag_name, tag_category, department, description, status',
|
||
'response': '返回更新成功消息'
|
||
})
|
||
|
||
url = f"{API_BASE_URL}/api/images/tags/names/999999"
|
||
headers = {"Authorization": f"Bearer {token}"}
|
||
data = {"description": "更新后的描述"}
|
||
|
||
try:
|
||
start_time = time.time()
|
||
response = requests.put(url, headers=headers, json=data, timeout=30)
|
||
elapsed_time = (time.time() - start_time) * 1000
|
||
|
||
success = response.status_code in [200, 404]
|
||
error = '' if success else response.text[:100]
|
||
|
||
print(f" 状态码: {response.status_code} - 耗时: {elapsed_time:.2f}ms")
|
||
|
||
test_results.append({
|
||
'endpoint': '/api/images/tags/names/<tag_id>',
|
||
'method': 'PUT',
|
||
'status_code': response.status_code,
|
||
'elapsed_time_ms': f"{elapsed_time:.2f}",
|
||
'success': success,
|
||
'error': error
|
||
})
|
||
except Exception as e:
|
||
print(f" 失败: {str(e)}")
|
||
test_results.append({
|
||
'endpoint': '/api/images/tags/names/<tag_id>',
|
||
'method': 'PUT',
|
||
'status_code': 0,
|
||
'elapsed_time_ms': '0',
|
||
'success': False,
|
||
'error': str(e)
|
||
})
|
||
|
||
def test_delete_tag_name(token):
|
||
"""测试删除标签名称"""
|
||
print("\n" + "-"*80)
|
||
print("[8/16] 测试删除标签名称...")
|
||
|
||
api_docs.append({
|
||
'endpoint': '/api/images/tags/names/<tag_id>',
|
||
'method': 'DELETE',
|
||
'description': '删除标签名称（物理删除）',
|
||
'auth': '需要认证',
|
||
'params': 'tag_id (路径参数)',
|
||
'response': '返回删除成功消息'
|
||
})
|
||
|
||
url = f"{API_BASE_URL}/api/images/tags/names/999999"
|
||
headers = {"Authorization": f"Bearer {token}"}
|
||
|
||
try:
|
||
start_time = time.time()
|
||
response = requests.delete(url, headers=headers, timeout=30)
|
||
elapsed_time = (time.time() - start_time) * 1000
|
||
|
||
success = response.status_code in [200, 404]
|
||
error = '' if success else response.text[:100]
|
||
|
||
print(f" 状态码: {response.status_code} - 耗时: {elapsed_time:.2f}ms")
|
||
|
||
test_results.append({
|
||
'endpoint': '/api/images/tags/names/<tag_id> (DELETE)',
|
||
'method': 'DELETE',
|
||
'status_code': response.status_code,
|
||
'elapsed_time_ms': f"{elapsed_time:.2f}",
|
||
'success': success,
|
||
'error': error
|
||
})
|
||
except Exception as e:
|
||
print(f" 失败: {str(e)}")
|
||
test_results.append({
|
||
'endpoint': '/api/images/tags/names/<tag_id> (DELETE)',
|
||
'method': 'DELETE',
|
||
'status_code': 0,
|
||
'elapsed_time_ms': '0',
|
||
'success': False,
|
||
'error': str(e)
|
||
})
|
||
|
||
def test_tag_relations_list(token):
|
||
"""测试获取图片标签关系列表"""
|
||
print("\n" + "-"*80)
|
||
print("[9/16] 测试获取图片标签关系列表...")
|
||
|
||
api_docs.append({
|
||
'endpoint': '/api/images/tags/relations/list',
|
||
'method': 'GET',
|
||
'description': '获取图片标签关系列表',
|
||
'auth': '需要认证',
|
||
'params': 'page, pageSize, image_id, tag_id',
|
||
'response': '返回图片标签关系列表'
|
||
})
|
||
|
||
url = f"{API_BASE_URL}/api/images/tags/relations/list"
|
||
headers = {"Authorization": f"Bearer {token}"}
|
||
params = {"page": 1, "pageSize": 20}
|
||
|
||
try:
|
||
start_time = time.time()
|
||
response = requests.get(url, headers=headers, params=params, timeout=30)
|
||
elapsed_time = (time.time() - start_time) * 1000
|
||
|
||
success = response.status_code == 200
|
||
error = '' if success else response.text[:100]
|
||
|
||
print(f" 状态码: {response.status_code} - 耗时: {elapsed_time:.2f}ms")
|
||
|
||
test_results.append({
|
||
'endpoint': '/api/images/tags/relations/list',
|
||
'method': 'GET',
|
||
'status_code': response.status_code,
|
||
'elapsed_time_ms': f"{elapsed_time:.2f}",
|
||
'success': success,
|
||
'error': error
|
||
})
|
||
except Exception as e:
|
||
print(f" 失败: {str(e)}")
|
||
test_results.append({
|
||
'endpoint': '/api/images/tags/relations/list',
|
||
'method': 'GET',
|
||
'status_code': 0,
|
||
'elapsed_time_ms': '0',
|
||
'success': False,
|
||
'error': str(e)
|
||
})
|
||
|
||
def test_image_types_list(token):
|
||
"""测试获取图片类型列表"""
|
||
print("\n" + "-"*80)
|
||
print("[10/16] 测试获取图片类型列表...")
|
||
|
||
api_docs.append({
|
||
'endpoint': '/api/images/types/list',
|
||
'method': 'GET',
|
||
'description': '获取图片类型列表（ai_image_type表）',
|
||
'auth': '需要认证',
|
||
'params': 'page, pageSize, keyword',
|
||
'response': '返回图片类型列表'
|
||
})
|
||
|
||
url = f"{API_BASE_URL}/api/images/types/list"
|
||
headers = {"Authorization": f"Bearer {token}"}
|
||
params = {"page": 1, "pageSize": 20}
|
||
|
||
try:
|
||
start_time = time.time()
|
||
response = requests.get(url, headers=headers, params=params, timeout=30)
|
||
elapsed_time = (time.time() - start_time) * 1000
|
||
|
||
success = response.status_code == 200
|
||
error = '' if success else response.text[:100]
|
||
|
||
print(f" 状态码: {response.status_code} - 耗时: {elapsed_time:.2f}ms")
|
||
|
||
test_results.append({
|
||
'endpoint': '/api/images/types/list',
|
||
'method': 'GET',
|
||
'status_code': response.status_code,
|
||
'elapsed_time_ms': f"{elapsed_time:.2f}",
|
||
'success': success,
|
||
'error': error
|
||
})
|
||
except Exception as e:
|
||
print(f" 失败: {str(e)}")
|
||
test_results.append({
|
||
'endpoint': '/api/images/types/list',
|
||
'method': 'GET',
|
||
'status_code': 0,
|
||
'elapsed_time_ms': '0',
|
||
'success': False,
|
||
'error': str(e)
|
||
})
|
||
|
||
def test_create_image_type(token):
|
||
"""测试创建图片类型"""
|
||
print("\n" + "-"*80)
|
||
print("[11/16] 测试创建图片类型...")
|
||
|
||
api_docs.append({
|
||
'endpoint': '/api/images/types/create',
|
||
'method': 'POST',
|
||
'description': '创建图片类型',
|
||
'auth': '需要认证',
|
||
'params': 'type_name, keywords_id, keywords_name, department_id, department_name',
|
||
'response': '返回新创建的类型ID'
|
||
})
|
||
|
||
url = f"{API_BASE_URL}/api/images/types/create"
|
||
headers = {"Authorization": f"Bearer {token}"}
|
||
timestamp = int(time.time())
|
||
data = {
|
||
"type_name": f"测试类型{timestamp}"
|
||
}
|
||
|
||
try:
|
||
start_time = time.time()
|
||
response = requests.post(url, headers=headers, json=data, timeout=30)
|
||
elapsed_time = (time.time() - start_time) * 1000
|
||
|
||
success = response.status_code == 200
|
||
error = '' if success else response.text[:100]
|
||
|
||
print(f" 状态码: {response.status_code} - 耗时: {elapsed_time:.2f}ms")
|
||
|
||
test_results.append({
|
||
'endpoint': '/api/images/types/create',
|
||
'method': 'POST',
|
||
'status_code': response.status_code,
|
||
'elapsed_time_ms': f"{elapsed_time:.2f}",
|
||
'success': success,
|
||
'error': error
|
||
})
|
||
except Exception as e:
|
||
print(f" 失败: {str(e)}")
|
||
test_results.append({
|
||
'endpoint': '/api/images/types/create',
|
||
'method': 'POST',
|
||
'status_code': 0,
|
||
'elapsed_time_ms': '0',
|
||
'success': False,
|
||
'error': str(e)
|
||
})
|
||
|
||
def test_update_image_type(token):
|
||
"""测试更新图片类型"""
|
||
print("\n" + "-"*80)
|
||
print("[12/16] 测试更新图片类型...")
|
||
|
||
api_docs.append({
|
||
'endpoint': '/api/images/types/<type_id>',
|
||
'method': 'PUT',
|
||
'description': '更新图片类型',
|
||
'auth': '需要认证',
|
||
'params': 'type_id (路径参数), type_name, keywords_id, keywords_name, department_id, department_name',
|
||
'response': '返回更新成功消息'
|
||
})
|
||
|
||
url = f"{API_BASE_URL}/api/images/types/999999"
|
||
headers = {"Authorization": f"Bearer {token}"}
|
||
data = {"type_name": "更新后的类型名称"}
|
||
|
||
try:
|
||
start_time = time.time()
|
||
response = requests.put(url, headers=headers, json=data, timeout=30)
|
||
elapsed_time = (time.time() - start_time) * 1000
|
||
|
||
success = response.status_code in [200, 404]
|
||
error = '' if success else response.text[:100]
|
||
|
||
print(f" 状态码: {response.status_code} - 耗时: {elapsed_time:.2f}ms")
|
||
|
||
test_results.append({
|
||
'endpoint': '/api/images/types/<type_id>',
|
||
'method': 'PUT',
|
||
'status_code': response.status_code,
|
||
'elapsed_time_ms': f"{elapsed_time:.2f}",
|
||
'success': success,
|
||
'error': error
|
||
})
|
||
except Exception as e:
|
||
print(f" 失败: {str(e)}")
|
||
test_results.append({
|
||
'endpoint': '/api/images/types/<type_id>',
|
||
'method': 'PUT',
|
||
'status_code': 0,
|
||
'elapsed_time_ms': '0',
|
||
'success': False,
|
||
'error': str(e)
|
||
})
|
||
|
||
def test_delete_image_type(token):
|
||
"""测试删除图片类型"""
|
||
print("\n" + "-"*80)
|
||
print("[13/16] 测试删除图片类型...")
|
||
|
||
api_docs.append({
|
||
'endpoint': '/api/images/types/<type_id>',
|
||
'method': 'DELETE',
|
||
'description': '删除图片类型（物理删除）',
|
||
'auth': '需要认证',
|
||
'params': 'type_id (路径参数)',
|
||
'response': '返回删除成功消息'
|
||
})
|
||
|
||
url = f"{API_BASE_URL}/api/images/types/999999"
|
||
headers = {"Authorization": f"Bearer {token}"}
|
||
|
||
try:
|
||
start_time = time.time()
|
||
response = requests.delete(url, headers=headers, timeout=30)
|
||
elapsed_time = (time.time() - start_time) * 1000
|
||
|
||
success = response.status_code in [200, 404]
|
||
error = '' if success else response.text[:100]
|
||
|
||
print(f" 状态码: {response.status_code} - 耗时: {elapsed_time:.2f}ms")
|
||
|
||
test_results.append({
|
||
'endpoint': '/api/images/types/<type_id> (DELETE)',
|
||
'method': 'DELETE',
|
||
'status_code': response.status_code,
|
||
'elapsed_time_ms': f"{elapsed_time:.2f}",
|
||
'success': success,
|
||
'error': error
|
||
})
|
||
except Exception as e:
|
||
print(f" 失败: {str(e)}")
|
||
test_results.append({
|
||
'endpoint': '/api/images/types/<type_id> (DELETE)',
|
||
'method': 'DELETE',
|
||
'status_code': 0,
|
||
'elapsed_time_ms': '0',
|
||
'success': False,
|
||
'error': str(e)
|
||
})
|
||
|
||
def test_image_tags_list(token):
|
||
"""测试获取图片标签列表"""
|
||
print("\n" + "-"*80)
|
||
print("[14/16] 测试获取图片标签列表（综合表）...")
|
||
|
||
api_docs.append({
|
||
'endpoint': '/api/images/tags/list',
|
||
'method': 'GET',
|
||
'description': '获取图片标签列表（ai_image_tags综合表）',
|
||
'auth': '需要认证',
|
||
'params': 'page, pageSize, image_id, tag_id, department_id',
|
||
'response': '返回图片标签列表（含图片信息、标签信息、部门信息）'
|
||
})
|
||
|
||
url = f"{API_BASE_URL}/api/images/tags/list"
|
||
headers = {"Authorization": f"Bearer {token}"}
|
||
params = {"page": 1, "pageSize": 20}
|
||
|
||
try:
|
||
start_time = time.time()
|
||
response = requests.get(url, headers=headers, params=params, timeout=30)
|
||
elapsed_time = (time.time() - start_time) * 1000
|
||
|
||
success = response.status_code == 200
|
||
error = '' if success else response.text[:100]
|
||
|
||
print(f" 状态码: {response.status_code} - 耗时: {elapsed_time:.2f}ms")
|
||
|
||
test_results.append({
|
||
'endpoint': '/api/images/tags/list',
|
||
'method': 'GET',
|
||
'status_code': response.status_code,
|
||
'elapsed_time_ms': f"{elapsed_time:.2f}",
|
||
'success': success,
|
||
'error': error
|
||
})
|
||
except Exception as e:
|
||
print(f" 失败: {str(e)}")
|
||
test_results.append({
|
||
'endpoint': '/api/images/tags/list',
|
||
'method': 'GET',
|
||
'status_code': 0,
|
||
'elapsed_time_ms': '0',
|
||
'success': False,
|
||
'error': str(e)
|
||
})
|
||
|
||
def test_create_image_tag(token):
|
||
"""测试创建图片标签"""
|
||
print("\n" + "-"*80)
|
||
print("[15/16] 测试创建图片标签...")
|
||
|
||
api_docs.append({
|
||
'endpoint': '/api/images/tags/create',
|
||
'method': 'POST',
|
||
'description': '创建图片标签（关联图片和标签）',
|
||
'auth': '需要认证',
|
||
'params': 'image_id, tag_id, image_name, image_url, tag_name, keywords_id, keywords_name, department_id, department_name',
|
||
'response': '返回新创建的标签关联ID'
|
||
})
|
||
|
||
url = f"{API_BASE_URL}/api/images/tags/create"
|
||
headers = {"Authorization": f"Bearer {token}"}
|
||
data = {
|
||
"image_id": 1,
|
||
"tag_id": 1
|
||
}
|
||
|
||
try:
|
||
start_time = time.time()
|
||
response = requests.post(url, headers=headers, json=data, timeout=30)
|
||
elapsed_time = (time.time() - start_time) * 1000
|
||
|
||
success = response.status_code in [200, 409]
|
||
error = '' if success else response.text[:100]
|
||
|
||
print(f" 状态码: {response.status_code} - 耗时: {elapsed_time:.2f}ms")
|
||
|
||
test_results.append({
|
||
'endpoint': '/api/images/tags/create',
|
||
'method': 'POST',
|
||
'status_code': response.status_code,
|
||
'elapsed_time_ms': f"{elapsed_time:.2f}",
|
||
'success': success,
|
||
'error': error
|
||
})
|
||
except Exception as e:
|
||
print(f" 失败: {str(e)}")
|
||
test_results.append({
|
||
'endpoint': '/api/images/tags/create',
|
||
'method': 'POST',
|
||
'status_code': 0,
|
||
'elapsed_time_ms': '0',
|
||
'success': False,
|
||
'error': str(e)
|
||
})
|
||
|
||
def test_delete_image_tag(token):
|
||
"""测试删除图片标签"""
|
||
print("\n" + "-"*80)
|
||
print("[16/16] 测试删除图片标签...")
|
||
|
||
api_docs.append({
|
||
'endpoint': '/api/images/tags/<tag_record_id>',
|
||
'method': 'DELETE',
|
||
'description': '删除图片标签（物理删除）',
|
||
'auth': '需要认证',
|
||
'params': 'tag_record_id (路径参数)',
|
||
'response': '返回删除成功消息'
|
||
})
|
||
|
||
url = f"{API_BASE_URL}/api/images/tags/999999"
|
||
headers = {"Authorization": f"Bearer {token}"}
|
||
|
||
try:
|
||
start_time = time.time()
|
||
response = requests.delete(url, headers=headers, timeout=30)
|
||
elapsed_time = (time.time() - start_time) * 1000
|
||
|
||
success = response.status_code in [200, 404]
|
||
error = '' if success else response.text[:100]
|
||
|
||
print(f" 状态码: {response.status_code} - 耗时: {elapsed_time:.2f}ms")
|
||
|
||
test_results.append({
|
||
'endpoint': '/api/images/tags/<tag_record_id>',
|
||
'method': 'DELETE',
|
||
'status_code': response.status_code,
|
||
'elapsed_time_ms': f"{elapsed_time:.2f}",
|
||
'success': success,
|
||
'error': error
|
||
})
|
||
except Exception as e:
|
||
print(f" 失败: {str(e)}")
|
||
test_results.append({
|
||
'endpoint': '/api/images/tags/<tag_record_id>',
|
||
'method': 'DELETE',
|
||
'status_code': 0,
|
||
'elapsed_time_ms': '0',
|
||
'success': False,
|
||
'error': str(e)
|
||
})
|
||
|
||
def save_to_csv():
|
||
"""保存测试结果到CSV文件"""
|
||
timestamp = datetime.now().strftime('%Y%m%d')
|
||
filename = f'图片接口性能测试_{timestamp}.csv'
|
||
|
||
with open(filename, 'w', newline='', encoding='utf-8-sig') as f:
|
||
writer = csv.DictWriter(f, fieldnames=[
|
||
'endpoint', 'method', 'status_code', 'elapsed_time_ms', 'success', 'error'
|
||
])
|
||
writer.writeheader()
|
||
writer.writerows(test_results)
|
||
|
||
print(f"\n✓ 性能测试结果已保存到: {filename}")
|
||
return filename
|
||
|
||
def save_api_docs():
|
||
"""保存接口文档到CSV文件"""
|
||
timestamp = datetime.now().strftime('%Y%m%d')
|
||
filename = f'图片接口文档_{timestamp}.csv'
|
||
|
||
with open(filename, 'w', newline='', encoding='utf-8-sig') as f:
|
||
writer = csv.DictWriter(f, fieldnames=[
|
||
'endpoint', 'method', 'description', 'auth', 'params', 'response'
|
||
])
|
||
writer.writeheader()
|
||
writer.writerows(api_docs)
|
||
|
||
print(f"✓ 接口文档已保存到: {filename}")
|
||
return filename
|
||
|
||
def print_summary():
|
||
"""打印测试摘要"""
|
||
print("\n" + "="*80)
|
||
print("测试摘要")
|
||
print("="*80)
|
||
|
||
total = len(test_results)
|
||
success = sum(1 for r in test_results if r['success'])
|
||
failed = total - success
|
||
|
||
times = [float(r['elapsed_time_ms']) for r in test_results if r['elapsed_time_ms'] != '0']
|
||
avg_time = sum(times) / len(times) if times else 0
|
||
max_time = max(times) if times else 0
|
||
min_time = min(times) if times else 0
|
||
|
||
print(f"总测试数: {total}")
|
||
print(f"成功: {success} ({success/total*100:.1f}%)")
|
||
print(f"失败: {failed} ({failed/total*100:.1f}%)")
|
||
print(f"平均耗时: {avg_time:.2f}ms")
|
||
print(f"最大耗时: {max_time:.2f}ms")
|
||
print(f"最小耗时: {min_time:.2f}ms")
|
||
print("="*80)
|
||
|
||
def main():
|
||
"""主函数"""
|
||
print("="*80)
|
||
print("图片管理接口性能测试工具")
|
||
print(f"测试时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
|
||
print(f"API地址: {API_BASE_URL}")
|
||
print("="*80)
|
||
|
||
# 登录
|
||
token = login()
|
||
if not token:
|
||
print("\n✗ 登录失败，无法继续测试")
|
||
return
|
||
|
||
# 执行所有测试
|
||
test_get_images_list(token)
|
||
test_get_images_dashboard(token)
|
||
test_upload_image(token)
|
||
test_delete_image(token)
|
||
test_tag_names_list(token)
|
||
test_create_tag_name(token)
|
||
test_update_tag_name(token)
|
||
test_delete_tag_name(token)
|
||
test_tag_relations_list(token)
|
||
test_image_types_list(token)
|
||
test_create_image_type(token)
|
||
test_update_image_type(token)
|
||
test_delete_image_type(token)
|
||
test_image_tags_list(token)
|
||
test_create_image_tag(token)
|
||
test_delete_image_tag(token)
|
||
|
||
# 打印摘要
|
||
print_summary()
|
||
|
||
# 保存结果
|
||
csv_file = save_to_csv()
|
||
doc_file = save_api_docs()
|
||
|
||
print(f"\n✓ 测试完成！")
|
||
print(f" - 性能数据: {csv_file}")
|
||
print(f" - 接口文档: {doc_file}")
|
||
|
||
if __name__ == "__main__":
|
||
try:
|
||
main()
|
||
except KeyboardInterrupt:
|
||
print("\n\n✗ 测试被用户中断")
|
||
except Exception as e:
|
||
print(f"\n✗ 测试异常: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|