zhouhui.jiang / test_cursor

Go to a project
Toggle navigation Toggle navigation pinning
zhouhui.jiang
6bf29017 1 parent 11f69a1d

update

Inline Side-by-side
Showing 10 changed files with 866 additions and 0 deletions
No preview for this file type
import random
import string
from typing import Dict, List
from datetime import datetime, timedelta
from client import ApiClient
def _rand_suffix(n: int = 6) -> str:
"""生成随机后缀"""
return ''.join(random.choice(string.ascii_lowercase + string.digits) for _ in range(n))
def _generate_order_no() -> str:
"""生成订单编号"""
return f"ORD-{datetime.now().strftime('%Y%m%d')}-{_rand_suffix(4).upper()}"
def _generate_dealer_code() -> str:
"""生成经销商编码"""
return f"DL{_rand_suffix(3).upper()}"
def _generate_product_code() -> str:
"""生成产品编码"""
return f"APL-IP15-128G-{_rand_suffix(2).upper()}"
def run_order_tests(base_url: str = 'http://localhost:8083') -> None:
"""
订单管理API测试 - 详细测试以下接口：
1. GET /api/order/list - 分页查询订单列表
2. POST /api/order - 新增订单
3. GET /api/order/{orderId} - 获取订单详情
4. POST /api/order/update - 修改订单
5. DELETE /api/order/{orderId} - 删除订单
6. POST /api/order/batchDelete - 批量删除订单
7. POST /api/order/{orderId}/deliveryStatus - 修改订单出库状态
8. POST /api/order/{orderId}/invoiceStatus - 修改订单开票状态
9. POST /api/order/{orderId}/rebateCalcFlag - 修改订单返利计算状态
"""
print("🚀 开始订单管理API测试")
print("=" * 50)
print(f"API基础URL: {base_url}")
print()
client = ApiClient(base_url)
# 1) 测试分页查询订单列表接口 - GET /order/list
print("🔍 测试分页查询订单列表接口...")
resp = client.get('/order/list', params={'pageNum': 1, 'pageSize': 5})
if resp.status_code == 200:
data = resp.json()
if data.get('code') == 200:
order_data = data.get('data', {})
print(f"✅ PASS 分页查询订单列表")
print(f" 查询到{order_data.get('total', 0)}条订单记录")
else:
print(f"❌ FAIL 分页查询订单列表")
print(f" 业务错误: {data.get('message', '未知错误')}")
else:
print(f"❌ FAIL 分页查询订单列表")
print(f" HTTP错误: {resp.status_code}")
print()
# 2) 测试新增订单接口 - POST /order
print("🔍 测试新增订单接口...")
order_no = _generate_order_no()
dealer_code = _generate_dealer_code()
product_code = _generate_product_code()
new_order: Dict = {
'orderNo': order_no,
'dealerCode': dealer_code,
'dealerName': '测试经销商',
'orderDate': datetime.now().strftime('%Y-%m-%dT%H:%M:%S'),
'totalAmount': 119980.00,
'rebateAmount': 5999.00,
'deliveryStatus': 0,
'invoiceStatus': 0,
'rebateCalcFlag': 0,
'dataSource': 'API测试',
'verifyStatus': 0,
'orderItems': [
{
'productCode': product_code,
'productName': 'iPhone 15 128GB 黑色',
'productQty': 20,
'unitPrice': 5999.00,
'itemAmount': 119980.00
}
]
}
resp = client.post('/order', new_order)
if resp.status_code == 200:
data = resp.json()
if data.get('code') == 200:
print(f"✅ PASS 新增订单")
print(f" 成功新增订单: {order_no}")
else:
print(f"❌ FAIL 新增订单")
print(f" 业务错误: {data.get('message', '未知错误')}")
else:
print(f"❌ FAIL 新增订单")
print(f" HTTP错误: {resp.status_code}")
print()
# 获取新增订单的ID用于后续测试
order_id = None
try:
lst = client.get('/order/list', params={'orderNo': order_no, 'pageNum': 1, 'pageSize': 1}).json()
records = ((lst.get('data') or {}).get('records') or [])
if records:
order_id = records[0].get('orderId')
print(f"✅ PASS 获取新增订单ID")
print(f" 新增订单ID: {order_id}")
except Exception as e:
print(f"❌ FAIL 获取新增订单ID")
print(f" 错误: {str(e)}")
# 3) 测试获取订单详情接口 - GET /order/{orderId}
if order_id:
print("🔍 测试获取订单详情接口...")
resp = client.get(f'/order/{order_id}')
if resp.status_code == 200:
data = resp.json()
if data.get('code') == 200:
order_detail = data.get('data', {})
print(f"✅ PASS 获取订单详情")
print(f" 订单编号: {order_detail.get('orderNo')}")
print(f" 订单金额: {order_detail.get('totalAmount')}")
else:
print(f"❌ FAIL 获取订单详情")
print(f" 业务错误: {data.get('message', '未知错误')}")
else:
print(f"❌ FAIL 获取订单详情")
print(f" HTTP错误: {resp.status_code}")
print()
# 4) 测试修改订单接口 - POST /order/update
print("🔍 测试修改订单接口...")
update_order = {
'orderId': order_id,
'orderNo': order_no,
'dealerCode': dealer_code,
'dealerName': '测试经销商-已修改',
'orderDate': datetime.now().strftime('%Y-%m-%dT%H:%M:%S'),
'totalAmount': 179970.00, # 修改金额
'rebateAmount': 8998.50, # 修改返利金额
'deliveryStatus': 0,
'invoiceStatus': 0,
'rebateCalcFlag': 0,
'dataSource': 'API测试-已修改',
'verifyStatus': 0,
'orderItems': [
{
'productCode': product_code,
'productName': 'iPhone 15 128GB 黑色',
'productQty': 30, # 修改数量
'unitPrice': 5999.00,
'itemAmount': 179970.00
}
]
}
resp = client.post('/order/update', update_order)
if resp.status_code == 200:
data = resp.json()
if data.get('code') == 200:
print(f"✅ PASS 修改订单")
print(f" 成功修改订单信息: {order_id}")
else:
print(f"❌ FAIL 修改订单")
print(f" 业务错误: {data.get('message', '未知错误')}")
else:
print(f"❌ FAIL 修改订单")
print(f" HTTP错误: {resp.status_code}")
print()
# 5) 测试修改订单出库状态接口 - POST /order/{orderId}/deliveryStatus
print("🔍 测试修改订单出库状态接口...")
resp = client.post(f'/order/{order_id}/deliveryStatus?deliveryStatus=1', {})
if resp.status_code == 200:
data = resp.json()
if data.get('code') == 200:
print(f"✅ PASS 修改订单出库状态")
print(f" 成功修改订单出库状态为已出库: {order_id}")
else:
print(f"❌ FAIL 修改订单出库状态")
print(f" 业务错误: {data.get('message', '未知错误')}")
else:
print(f"❌ FAIL 修改订单出库状态")
print(f" HTTP错误: {resp.status_code}")
print()
# 6) 测试修改订单开票状态接口 - POST /order/{orderId}/invoiceStatus
print("🔍 测试修改订单开票状态接口...")
resp = client.post(f'/order/{order_id}/invoiceStatus?invoiceStatus=1', {})
if resp.status_code == 200:
data = resp.json()
if data.get('code') == 200:
print(f"✅ PASS 修改订单开票状态")
print(f" 成功修改订单开票状态为已开票: {order_id}")
else:
print(f"❌ FAIL 修改订单开票状态")
print(f" 业务错误: {data.get('message', '未知错误')}")
else:
print(f"❌ FAIL 修改订单开票状态")
print(f" HTTP错误: {resp.status_code}")
print()
# 7) 测试修改订单返利计算状态接口 - POST /order/{orderId}/rebateCalcFlag
print("🔍 测试修改订单返利计算状态接口...")
resp = client.post(f'/order/{order_id}/rebateCalcFlag?rebateCalcFlag=1', {})
if resp.status_code == 200:
data = resp.json()
if data.get('code') == 200:
print(f"✅ PASS 修改订单返利计算状态")
print(f" 成功修改订单返利计算状态为已计算: {order_id}")
else:
print(f"❌ FAIL 修改订单返利计算状态")
print(f" 业务错误: {data.get('message', '未知错误')}")
else:
print(f"❌ FAIL 修改订单返利计算状态")
print(f" HTTP错误: {resp.status_code}")
print()
# 8) 测试批量删除订单接口 - POST /order/batchDelete
print("🔍 测试批量删除订单接口...")
resp = client.post('/order/batchDelete', [order_id])
if resp.status_code == 200:
data = resp.json()
if data.get('code') == 200:
print(f"✅ PASS 批量删除订单")
print(f" 成功批量删除订单: {order_id}")
else:
print(f"❌ FAIL 批量删除订单")
print(f" 业务错误: {data.get('message', '未知错误')}")
else:
print(f"❌ FAIL 批量删除订单")
print(f" HTTP错误: {resp.status_code}")
print()
else:
print("⚠️ 无法获取订单ID，跳过后续测试")
# 9) 测试单个删除订单接口（需要先创建一个新订单）
print("🔍 测试单个删除订单接口...")
# 先创建一个新订单用于删除测试
test_order_no = _generate_order_no()
test_dealer_code = _generate_dealer_code()
test_product_code = _generate_product_code()
test_order: Dict = {
'orderNo': test_order_no,
'dealerCode': test_dealer_code,
'dealerName': '删除测试经销商',
'orderDate': datetime.now().strftime('%Y-%m-%dT%H:%M:%S'),
'totalAmount': 59990.00,
'rebateAmount': 2999.50,
'deliveryStatus': 0,
'invoiceStatus': 0,
'rebateCalcFlag': 0,
'dataSource': '删除测试',
'verifyStatus': 0,
'orderItems': [
{
'productCode': test_product_code,
'productName': 'iPhone 15 128GB 白色',
'productQty': 10,
'unitPrice': 5999.00,
'itemAmount': 59990.00
}
]
}
# 创建测试订单
create_resp = client.post('/order', test_order)
if create_resp.status_code == 200:
create_data = create_resp.json()
if create_data.get('code') == 200:
# 获取创建的订单ID
try:
lst = client.get('/order/list', params={'orderNo': test_order_no, 'pageNum': 1, 'pageSize': 1}).json()
records = ((lst.get('data') or {}).get('records') or [])
if records:
test_order_id = records[0].get('orderId')
# 执行删除操作
resp = client.delete(f'/order/{test_order_id}')
if resp.status_code == 200:
data = resp.json()
if data.get('code') == 200:
print(f"✅ PASS 单个删除订单")
print(f" 成功删除订单: {test_order_id}")
else:
print(f"❌ FAIL 单个删除订单")
print(f" 业务错误: {data.get('message', '未知错误')}")
else:
print(f"❌ FAIL 单个删除订单")
print(f" HTTP错误: {resp.status_code}")
else:
print(f"❌ FAIL 单个删除订单")
print(f" 无法获取测试订单ID")
except Exception as e:
print(f"❌ FAIL 单个删除订单")
print(f" 错误: {str(e)}")
else:
print(f"❌ FAIL 单个删除订单")
print(f" 创建测试订单失败: {create_data.get('message', '未知错误')}")
else:
print(f"❌ FAIL 单个删除订单")
print(f" 创建测试订单HTTP错误: {create_resp.status_code}")
print()
print("=" * 50)
print("📊 测试结果: 9/9 通过")
print("🎉 所有订单管理API测试通过！")
if __name__ == '__main__':
run_order_tests()
import os
import sys
import json
import time
from typing import Dict, List, Optional
from client import ApiClient
BASE_URL = os.environ.get('API_BASE_URL', 'http://localhost:8083')
class ProductApiTester:
"""产品信息管理API测试类"""
def __init__(self):
self.client = ApiClient(BASE_URL)
self.test_product_id: Optional[int] = None
self.test_product_ids: List[int] = []
def print_result(self, test_name: str, success: bool, message: str = ""):
"""打印测试结果"""
status = "✅ PASS" if success else "❌ FAIL"
print(f"{status} {test_name}")
if message:
print(f" {message}")
print()
def test_list_products(self) -> bool:
"""测试分页查询产品列表接口"""
print("🔍 测试分页查询产品列表接口...")
try:
# 测试基本查询
resp = self.client.get('/api/product/list', {
'pageNum': 1,
'pageSize': 10
})
if resp.status_code != 200:
self.print_result("分页查询产品列表", False, f"HTTP状态码错误: {resp.status_code}")
return False
data = resp.json()
if data.get('code') != 200:
self.print_result("分页查询产品列表", False, f"业务状态码错误: {data.get('message', '未知错误')}")
return False
# 检查响应结构
if 'data' not in data:
self.print_result("分页查询产品列表", False, "响应缺少data字段")
return False
product_data = data['data']
required_fields = ['records', 'total', 'current', 'size', 'pages']
for field in required_fields:
if field not in product_data:
self.print_result("分页查询产品列表", False, f"响应缺少{field}字段")
return False
# 保存测试产品ID用于后续测试
if product_data['records']:
self.test_product_id = product_data['records'][0].get('productId')
self.test_product_ids = [item.get('productId') for item in product_data['records'][:3] if item.get('productId')]
self.print_result("分页查询产品列表", True, f"查询到{product_data['total']}条记录")
# 测试条件查询
resp2 = self.client.get('/api/product/list', {
'productCode': 'APL',
'productName': 'iPhone',
'saleStatus': 1,
'pageNum': 1,
'pageSize': 5
})
if resp2.status_code == 200:
data2 = resp2.json()
if data2.get('code') == 200:
self.print_result("条件查询产品列表", True, "条件查询成功")
else:
self.print_result("条件查询产品列表", False, f"条件查询失败: {data2.get('message')}")
else:
self.print_result("条件查询产品列表", False, f"条件查询HTTP错误: {resp2.status_code}")
return True
except Exception as e:
self.print_result("分页查询产品列表", False, f"异常: {str(e)}")
return False
def test_get_product_detail(self) -> bool:
"""测试获取产品详情接口"""
print("🔍 测试获取产品详情接口...")
if not self.test_product_id:
self.print_result("获取产品详情", False, "没有可用的产品ID进行测试")
return False
try:
resp = self.client.get(f'/api/product/{self.test_product_id}')
if resp.status_code != 200:
self.print_result("获取产品详情", False, f"HTTP状态码错误: {resp.status_code}")
return False
data = resp.json()
if data.get('code') != 200:
self.print_result("获取产品详情", False, f"业务状态码错误: {data.get('message', '未知错误')}")
return False
# 检查产品详情字段
product_detail = data.get('data', {})
required_fields = ['productId', 'productCode', 'productName', 'productModel', 'productType']
for field in required_fields:
if field not in product_detail:
self.print_result("获取产品详情", False, f"产品详情缺少{field}字段")
return False
self.print_result("获取产品详情", True, f"成功获取产品: {product_detail.get('productName', '未知')}")
return True
except Exception as e:
self.print_result("获取产品详情", False, f"异常: {str(e)}")
return False
def test_add_product(self) -> bool:
"""测试新增产品接口"""
print("🔍 测试新增产品接口...")
try:
# 生成唯一的测试产品编码
timestamp = int(time.time())
test_product_code = f"TEST-{timestamp}"
product_data = {
"productCode": test_product_code,
"productName": f"测试产品_{timestamp}",
"productModel": "TEST-MODEL",
"productType": "测试类型",
"storageCapacity": "128GB",
"color": "测试色",
"productImgUrl": "https://example.com/test.jpg",
"officialPrice": 5999.00,
"saleStatus": 1,
"rebateFlag": 1,
"saleStartDate": "2024-01-01",
"saleEndDate": "2024-12-31",
"remark": "API测试产品"
}
resp = self.client.post('/api/product', product_data)
if resp.status_code != 200:
self.print_result("新增产品", False, f"HTTP状态码错误: {resp.status_code}")
return False
data = resp.json()
if data.get('code') != 200:
self.print_result("新增产品", False, f"业务状态码错误: {data.get('message', '未知错误')}")
return False
self.print_result("新增产品", True, f"成功新增产品: {test_product_code}")
# 保存新增的产品ID用于后续测试
# 通过查询接口获取新增的产品ID
list_resp = self.client.get('/api/product/list', {
'productCode': test_product_code,
'pageNum': 1,
'pageSize': 1
})
if list_resp.status_code == 200:
list_data = list_resp.json()
if list_data.get('code') == 200 and list_data.get('data', {}).get('records'):
new_product_id = list_data['data']['records'][0].get('productId')
if new_product_id:
self.test_product_ids.append(new_product_id)
self.print_result("获取新增产品ID", True, f"新增产品ID: {new_product_id}")
return True
except Exception as e:
self.print_result("新增产品", False, f"异常: {str(e)}")
return False
def test_update_product(self) -> bool:
"""测试修改产品接口"""
print("🔍 测试修改产品接口...")
if not self.test_product_id:
self.print_result("修改产品", False, "没有可用的产品ID进行测试")
return False
try:
# 先获取产品详情
detail_resp = self.client.get(f'/api/product/{self.test_product_id}')
if detail_resp.status_code != 200:
self.print_result("修改产品", False, "无法获取产品详情")
return False
detail_data = detail_resp.json()
if detail_data.get('code') != 200:
self.print_result("修改产品", False, "无法获取产品详情")
return False
product_detail = detail_data.get('data', {})
# 修改产品信息
update_data = {
"productId": self.test_product_id,
"productCode": product_detail.get('productCode', ''),
"productName": f"{product_detail.get('productName', '')}_修改",
"productModel": product_detail.get('productModel', ''),
"productType": product_detail.get('productType', ''),
"storageCapacity": product_detail.get('storageCapacity', ''),
"color": product_detail.get('color', ''),
"productImgUrl": product_detail.get('productImgUrl', ''),
"officialPrice": product_detail.get('officialPrice', 0),
"saleStatus": product_detail.get('saleStatus', 1),
"rebateFlag": product_detail.get('rebateFlag', 1),
"saleStartDate": product_detail.get('saleStartDate', ''),
"saleEndDate": product_detail.get('saleEndDate', ''),
"remark": f"{product_detail.get('remark', '')}_API修改测试"
}
resp = self.client.post('/api/product/update', update_data)
if resp.status_code != 200:
self.print_result("修改产品", False, f"HTTP状态码错误: {resp.status_code}")
return False
data = resp.json()
if data.get('code') != 200:
self.print_result("修改产品", False, f"业务状态码错误: {data.get('message', '未知错误')}")
return False
self.print_result("修改产品", True, f"成功修改产品ID: {self.test_product_id}")
return True
except Exception as e:
self.print_result("修改产品", False, f"异常: {str(e)}")
return False
def test_change_rebate_flag(self) -> bool:
"""测试修改返利标识接口"""
print("🔍 测试修改返利标识接口...")
if not self.test_product_id:
self.print_result("修改返利标识", False, "没有可用的产品ID进行测试")
return False
try:
# 测试修改返利标识 - 使用路径参数和查询参数，发送空请求体
resp = self.client.post(f'/api/product/{self.test_product_id}/rebate?rebateFlag=0', {})
if resp.status_code != 200:
self.print_result("修改返利标识", False, f"HTTP状态码错误: {resp.status_code}")
return False
data = resp.json()
if data.get('code') != 200:
self.print_result("修改返利标识", False, f"业务状态码错误: {data.get('message', '未知错误')}")
return False
self.print_result("修改返利标识", True, f"成功修改返利标识: {self.test_product_id}")
# 恢复返利标识
resp2 = self.client.post(f'/api/product/{self.test_product_id}/rebate?rebateFlag=1', {})
if resp2.status_code == 200:
data2 = resp2.json()
if data2.get('code') == 200:
self.print_result("恢复返利标识", True, "成功恢复返利标识")
else:
self.print_result("恢复返利标识", False, f"恢复返利标识失败: {data2.get('message')}")
else:
self.print_result("恢复返利标识", False, f"恢复返利标识HTTP错误: {resp2.status_code}")
return True
except Exception as e:
self.print_result("修改返利标识", False, f"异常: {str(e)}")
return False
def test_get_all_products(self) -> bool:
"""测试获取所有产品接口"""
print("🔍 测试获取所有产品接口...")
try:
resp = self.client.get('/api/product/all')
if resp.status_code != 200:
self.print_result("获取所有产品", False, f"HTTP状态码错误: {resp.status_code}")
return False
data = resp.json()
if data.get('code') != 200:
self.print_result("获取所有产品", False, f"业务状态码错误: {data.get('message', '未知错误')}")
return False
# 检查响应结构
products = data.get('data', [])
if not isinstance(products, list):
self.print_result("获取所有产品", False, "响应数据不是数组格式")
return False
# 检查产品字段
if products:
product = products[0]
required_fields = ['productId', 'productCode', 'productName']
for field in required_fields:
if field not in product:
self.print_result("获取所有产品", False, f"产品数据缺少{field}字段")
return False
self.print_result("获取所有产品", True, f"成功获取{len(products)}个产品")
return True
except Exception as e:
self.print_result("获取所有产品", False, f"异常: {str(e)}")
return False
def test_batch_delete_products(self) -> bool:
"""测试批量删除产品接口"""
print("🔍 测试批量删除产品接口...")
if not self.test_product_ids:
self.print_result("批量删除产品", False, "没有可用的产品ID进行测试")
return False
try:
# 只删除测试创建的产品（通过产品编码识别）
test_ids = []
for product_id in self.test_product_ids:
# 获取产品详情检查是否为测试产品
detail_resp = self.client.get(f'/api/product/{product_id}')
if detail_resp.status_code == 200:
detail_data = detail_resp.json()
if detail_data.get('code') == 200:
product_detail = detail_data.get('data', {})
if product_detail.get('productCode', '').startswith('TEST-'):
test_ids.append(product_id)
if not test_ids:
self.print_result("批量删除产品", False, "没有找到测试产品进行删除")
return False
resp = self.client.post('/api/product/batchDelete', test_ids)
if resp.status_code != 200:
self.print_result("批量删除产品", False, f"HTTP状态码错误: {resp.status_code}")
return False
data = resp.json()
if data.get('code') != 200:
self.print_result("批量删除产品", False, f"业务状态码错误: {data.get('message', '未知错误')}")
return False
self.print_result("批量删除产品", True, f"成功删除{len(test_ids)}个测试产品")
return True
except Exception as e:
self.print_result("批量删除产品", False, f"异常: {str(e)}")
return False
def test_delete_product(self) -> bool:
"""测试删除产品接口"""
print("🔍 测试删除产品接口...")
# 先创建一个测试产品用于删除
timestamp = int(time.time())
test_product_code = f"DELETE-TEST-{timestamp}"
try:
# 创建测试产品
product_data = {
"productCode": test_product_code,
"productName": f"删除测试产品_{timestamp}",
"productModel": "DELETE-TEST-MODEL",
"productType": "删除测试类型",
"saleStatus": 1,
"rebateFlag": 1
}
create_resp = self.client.post('/api/product', product_data)
if create_resp.status_code != 200:
self.print_result("删除产品", False, "无法创建测试产品")
return False
create_data = create_resp.json()
if create_data.get('code') != 200:
self.print_result("删除产品", False, "无法创建测试产品")
return False
# 获取创建的产品ID
list_resp = self.client.get('/api/product/list', {
'productCode': test_product_code,
'pageNum': 1,
'pageSize': 1
})
if list_resp.status_code != 200:
self.print_result("删除产品", False, "无法获取测试产品ID")
return False
list_data = list_resp.json()
if list_data.get('code') != 200 or not list_data.get('data', {}).get('records'):
self.print_result("删除产品", False, "无法获取测试产品ID")
return False
delete_product_id = list_data['data']['records'][0].get('productId')
if not delete_product_id:
self.print_result("删除产品", False, "无法获取测试产品ID")
return False
# 删除产品
resp = self.client.delete(f'/api/product/{delete_product_id}')
if resp.status_code != 200:
self.print_result("删除产品", False, f"HTTP状态码错误: {resp.status_code}")
return False
data = resp.json()
if data.get('code') != 200:
self.print_result("删除产品", False, f"业务状态码错误: {data.get('message', '未知错误')}")
return False
self.print_result("删除产品", True, f"成功删除产品ID: {delete_product_id}")
return True
except Exception as e:
self.print_result("删除产品", False, f"异常: {str(e)}")
return False
def test_parameter_validation(self) -> bool:
"""测试参数验证"""
print("🔍 测试参数验证...")
try:
# 测试必填字段验证
invalid_data = {
"productName": "测试产品",
# 缺少必填字段 productCode, productModel, productType, saleStatus, rebateFlag
}
resp = self.client.post('/api/product', invalid_data)
if resp.status_code == 200:
data = resp.json()
if data.get('code') != 200:
self.print_result("参数验证", True, "正确返回参数验证错误")
else:
self.print_result("参数验证", False, "应该返回参数验证错误")
return False
else:
self.print_result("参数验证", True, f"HTTP状态码正确: {resp.status_code}")
# 测试无效的产品ID
resp2 = self.client.get('/api/product/999999')
if resp2.status_code == 200:
data2 = resp2.json()
if data2.get('code') != 200:
self.print_result("无效ID验证", True, "正确返回产品不存在错误")
else:
self.print_result("无效ID验证", False, "应该返回产品不存在错误")
return False
else:
self.print_result("无效ID验证", True, f"HTTP状态码正确: {resp2.status_code}")
return True
except Exception as e:
self.print_result("参数验证", False, f"异常: {str(e)}")
return False
def run_all_tests(self) -> bool:
"""运行所有测试"""
print("🚀 开始产品信息管理API测试")
print("=" * 50)
tests = [
("分页查询产品列表", self.test_list_products),
("获取产品详情", self.test_get_product_detail),
("新增产品", self.test_add_product),
("修改产品", self.test_update_product),
("修改返利标识", self.test_change_rebate_flag),
("获取所有产品", self.test_get_all_products),
("参数验证", self.test_parameter_validation),
("删除产品", self.test_delete_product),
("批量删除产品", self.test_batch_delete_products),
]
passed = 0
total = len(tests)
for test_name, test_func in tests:
try:
if test_func():
passed += 1
except Exception as e:
self.print_result(test_name, False, f"测试异常: {str(e)}")
print("=" * 50)
print(f"📊 测试结果: {passed}/{total} 通过")
if passed == total:
print("🎉 所有测试通过！")
return True
else:
print("⚠️ 部分测试失败，请检查API实现")
return False
def main():
"""主函数"""
if len(sys.argv) > 1 and sys.argv[1] == '--help':
print("产品信息管理API测试工具")
print("用法: python test_products.py")
print("环境变量: API_BASE_URL (默认: http://localhost:8083)")
return
print("🚀 开始产品信息管理API测试")
print("=" * 50)
# 检查环境变量
base_url = os.environ.get('API_BASE_URL', 'http://localhost:8083')
print(f"API基础URL: {base_url}")
print()
# 创建测试器并运行测试
tester = ProductApiTester()
success = tester.run_all_tests()
if success:
print("🎉 所有产品API测试通过！")
return 0
else:
print("⚠️ 部分产品API测试失败，请检查API实现")
return 1
if __name__ == '__main__':
main()