test_orders.py
10.4 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
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()