test_orders.py 10.4 KB
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()