sam
d88ec60ef4
- 在MarketingCode聚合中新增品类ID和品类名称字段，完善产品信息结构 - 迁移生成营销码命令，支持传入品类ID和品类名称参数 - 积分发放失败时发送积分获得失败通知集成事件 - 新增通知发送及积分失败通知的集成事件处理器，使用SSE推送通知 - 在积分相关集成事件处理器中添加发送积分变动通知功能 - 移除Notification聚合，相关数据库表删除 - 新增分页结果类型PagedResult，支持营销码查询分页返回 - 营销码查询支持分页参数，返回分页结果数据
472 lines
17 KiB
Python
472 lines
17 KiB
Python
"""
|
||
一物一码会员营销系统 - API 业务流程测试
|
||
直接测试后端API，不依赖前端UI
|
||
"""
|
||
|
||
import requests
|
||
import json
|
||
import time
|
||
from datetime import datetime, timedelta
|
||
from typing import Optional, Dict, Any
|
||
|
||
class APITester:
|
||
def __init__(self, base_url: str = "http://localhost:5511"):
|
||
self.base_url = base_url
|
||
self.admin_token: Optional[str] = None
|
||
self.member_token: Optional[str] = None
|
||
self.test_results = []
|
||
self.marketing_codes = []
|
||
self.gift_id: Optional[str] = None
|
||
self.member_id: Optional[str] = None
|
||
self.order_id: Optional[str] = None
|
||
|
||
def log_test(self, test_name: str, status: str, message: str = "", data: Any = None):
|
||
"""记录测试结果"""
|
||
result = {
|
||
"test": test_name,
|
||
"status": status,
|
||
"message": message,
|
||
"timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
|
||
"data": data
|
||
}
|
||
self.test_results.append(result)
|
||
|
||
status_emoji = "✅" if status == "PASS" else "❌" if status == "FAIL" else "⏭️"
|
||
print(f"{status_emoji} [{test_name}] {status}: {message}")
|
||
if data:
|
||
print(f" 数据: {json.dumps(data, ensure_ascii=False, indent=2)}")
|
||
|
||
def request(self, method: str, endpoint: str, **kwargs) -> Optional[Dict]:
|
||
"""发送HTTP请求"""
|
||
url = f"{self.base_url}{endpoint}"
|
||
headers = kwargs.get('headers', {})
|
||
|
||
# 添加认证头
|
||
if 'admin' in endpoint and self.admin_token:
|
||
headers['Authorization'] = f'Bearer {self.admin_token}'
|
||
elif 'members' in endpoint and self.member_token:
|
||
headers['Authorization'] = f'Bearer {self.member_token}'
|
||
|
||
kwargs['headers'] = headers
|
||
|
||
try:
|
||
response = requests.request(method, url, **kwargs)
|
||
|
||
if response.status_code >= 400:
|
||
print(f" HTTP {response.status_code}: {response.text}")
|
||
return None
|
||
|
||
return response.json() if response.text else {}
|
||
|
||
except Exception as e:
|
||
print(f" 请求失败: {e}")
|
||
return None
|
||
|
||
def test_health_check(self):
|
||
"""测试健康检查"""
|
||
print("\n=== 测试：健康检查 ===")
|
||
|
||
response = self.request('GET', '/health')
|
||
if response:
|
||
self.log_test("健康检查", "PASS", "后端服务正常")
|
||
return True
|
||
else:
|
||
self.log_test("健康检查", "FAIL", "后端服务不可用")
|
||
return False
|
||
|
||
def test_generate_marketing_codes(self):
|
||
"""测试生成营销码"""
|
||
print("\n=== 测试：生成营销码 ===")
|
||
|
||
batch_no = f"BATCH-{int(time.time())}"
|
||
product_id = "PROD-001"
|
||
|
||
payload = {
|
||
"batchNo": batch_no,
|
||
"productId": product_id,
|
||
"productName": "测试产品A",
|
||
"quantity": 10,
|
||
"expirationTime": (datetime.now() + timedelta(days=30)).isoformat()
|
||
}
|
||
|
||
response = self.request('POST', '/api/admin/marketing-codes/generate', json=payload)
|
||
|
||
if response and response.get('success'):
|
||
data = response.get('data', {})
|
||
self.marketing_codes = data.get('marketingCodes', [])
|
||
self.log_test("生成营销码", "PASS",
|
||
f"成功生成 {len(self.marketing_codes)} 个营销码",
|
||
{"批次号": batch_no, "数量": len(self.marketing_codes)})
|
||
return True
|
||
else:
|
||
self.log_test("生成营销码", "FAIL", "生成失败")
|
||
return False
|
||
|
||
def test_create_points_rule(self):
|
||
"""测试创建积分规则"""
|
||
print("\n=== 测试：创建积分规则 ===")
|
||
|
||
payload = {
|
||
"ruleName": "测试产品积分规则",
|
||
"ruleType": 1, # Product
|
||
"pointsValue": 100,
|
||
"rewardMultiplier": 1.0,
|
||
"productId": "PROD-001",
|
||
"startTime": datetime.now().isoformat(),
|
||
"endTime": (datetime.now() + timedelta(days=365)).isoformat()
|
||
}
|
||
|
||
response = self.request('POST', '/api/admin/points-rules', json=payload)
|
||
|
||
if response and response.get('success'):
|
||
self.log_test("创建积分规则", "PASS",
|
||
"积分规则创建成功",
|
||
{"规则名": payload['ruleName'], "积分值": payload['pointsValue']})
|
||
return True
|
||
else:
|
||
self.log_test("创建积分规则", "FAIL", "创建失败")
|
||
return False
|
||
|
||
def test_create_gift(self):
|
||
"""测试创建礼品"""
|
||
print("\n=== 测试：创建礼品 ===")
|
||
|
||
payload = {
|
||
"name": "测试礼品-电子产品",
|
||
"giftType": 1, # Physical
|
||
"description": "这是一个测试用的电子产品礼品",
|
||
"imageUrl": "https://via.placeholder.com/400",
|
||
"requiredPoints": 500,
|
||
"totalStock": 100,
|
||
"redemptionLimit": 2
|
||
}
|
||
|
||
response = self.request('POST', '/api/admin/gifts', json=payload)
|
||
|
||
if response and response.get('success'):
|
||
data = response.get('data', {})
|
||
self.gift_id = data.get('giftId')
|
||
self.log_test("创建礼品", "PASS",
|
||
"礼品创建成功",
|
||
{"礼品ID": self.gift_id, "名称": payload['name']})
|
||
return True
|
||
else:
|
||
self.log_test("创建礼品", "FAIL", "创建失败")
|
||
return False
|
||
|
||
def test_put_gift_on_shelf(self):
|
||
"""测试礼品上架"""
|
||
print("\n=== 测试：礼品上架 ===")
|
||
|
||
if not self.gift_id:
|
||
self.log_test("礼品上架", "SKIP", "没有礼品ID")
|
||
return False
|
||
|
||
response = self.request('POST', f'/api/admin/gifts/{self.gift_id}/putonshelf')
|
||
|
||
if response and response.get('success'):
|
||
self.log_test("礼品上架", "PASS", "礼品上架成功")
|
||
return True
|
||
else:
|
||
self.log_test("礼品上架", "FAIL", "上架失败")
|
||
return False
|
||
|
||
def test_member_register(self):
|
||
"""测试会员注册"""
|
||
print("\n=== 测试：会员注册 ===")
|
||
|
||
timestamp = int(time.time()) % 100000000
|
||
phone = f"138{timestamp}"
|
||
|
||
payload = {
|
||
"phoneNumber": phone,
|
||
"password": "123456",
|
||
"nickname": f"测试用户{timestamp}"
|
||
}
|
||
|
||
response = self.request('POST', '/api/members/register', json=payload)
|
||
|
||
if response and response.get('success'):
|
||
data = response.get('data', {})
|
||
self.member_id = data.get('memberId')
|
||
self.member_token = data.get('token')
|
||
self.log_test("会员注册", "PASS",
|
||
"会员注册成功",
|
||
{"会员ID": self.member_id, "手机号": phone})
|
||
return True
|
||
else:
|
||
self.log_test("会员注册", "FAIL", "注册失败")
|
||
return False
|
||
|
||
def test_member_login(self):
|
||
"""测试会员登录"""
|
||
print("\n=== 测试：会员登录 ===")
|
||
|
||
# 如果已经注册过，跳过
|
||
if self.member_token:
|
||
self.log_test("会员登录", "SKIP", "已有Token")
|
||
return True
|
||
|
||
self.log_test("会员登录", "SKIP", "需要先注册")
|
||
return False
|
||
|
||
def test_scan_marketing_code(self):
|
||
"""测试扫码获取积分"""
|
||
print("\n=== 测试：扫码获取积分 ===")
|
||
|
||
if not self.marketing_codes:
|
||
self.log_test("扫码获取积分", "SKIP", "没有可用的营销码")
|
||
return False
|
||
|
||
marketing_code = self.marketing_codes[0]
|
||
|
||
payload = {
|
||
"marketingCode": marketing_code
|
||
}
|
||
|
||
response = self.request('POST', '/api/members/scan', json=payload)
|
||
|
||
if response and response.get('success'):
|
||
data = response.get('data', {})
|
||
points = data.get('pointsAwarded', 0)
|
||
self.log_test("扫码获取积分", "PASS",
|
||
f"成功获得 {points} 积分",
|
||
{"营销码": marketing_code, "积分": points})
|
||
return True
|
||
else:
|
||
self.log_test("扫码获取积分", "FAIL", "扫码失败")
|
||
return False
|
||
|
||
def test_duplicate_scan(self):
|
||
"""测试重复扫码（验证唯一性）"""
|
||
print("\n=== 测试：重复扫码验证 ===")
|
||
|
||
if not self.marketing_codes:
|
||
self.log_test("重复扫码验证", "SKIP", "没有可用的营销码")
|
||
return False
|
||
|
||
marketing_code = self.marketing_codes[0]
|
||
|
||
payload = {
|
||
"marketingCode": marketing_code
|
||
}
|
||
|
||
response = self.request('POST', '/api/members/scan', json=payload)
|
||
|
||
# 应该失败
|
||
if response and not response.get('success'):
|
||
self.log_test("重复扫码验证", "PASS",
|
||
"正确阻止重复扫码",
|
||
{"消息": response.get('message')})
|
||
return True
|
||
else:
|
||
self.log_test("重复扫码验证", "FAIL", "未能阻止重复扫码")
|
||
return False
|
||
|
||
def test_get_member_points(self):
|
||
"""测试查询会员积分"""
|
||
print("\n=== 测试：查询会员积分 ===")
|
||
|
||
if not self.member_id:
|
||
self.log_test("查询会员积分", "SKIP", "没有会员ID")
|
||
return False
|
||
|
||
response = self.request('GET', f'/api/members/{self.member_id}')
|
||
|
||
if response and response.get('success'):
|
||
data = response.get('data', {})
|
||
points = data.get('availablePoints', 0)
|
||
self.log_test("查询会员积分", "PASS",
|
||
f"当前积分: {points}",
|
||
{"会员ID": self.member_id, "积分": points})
|
||
return True
|
||
else:
|
||
self.log_test("查询会员积分", "FAIL", "查询失败")
|
||
return False
|
||
|
||
def test_get_gifts_list(self):
|
||
"""测试获取礼品列表"""
|
||
print("\n=== 测试：获取礼品列表 ===")
|
||
|
||
response = self.request('GET', '/api/gifts')
|
||
|
||
if response and response.get('success'):
|
||
data = response.get('data', [])
|
||
on_shelf = [g for g in data if g.get('isOnShelf')]
|
||
self.log_test("获取礼品列表", "PASS",
|
||
f"共 {len(data)} 个礼品，{len(on_shelf)} 个已上架",
|
||
{"总数": len(data), "上架": len(on_shelf)})
|
||
return True
|
||
else:
|
||
self.log_test("获取礼品列表", "FAIL", "查询失败")
|
||
return False
|
||
|
||
def test_redeem_gift(self):
|
||
"""测试兑换礼品"""
|
||
print("\n=== 测试：兑换礼品 ===")
|
||
|
||
if not self.gift_id:
|
||
self.log_test("兑换礼品", "SKIP", "没有礼品ID")
|
||
return False
|
||
|
||
payload = {
|
||
"giftId": self.gift_id,
|
||
"quantity": 1,
|
||
"deliveryAddress": {
|
||
"consignee": "张三",
|
||
"phoneNumber": "13800138000",
|
||
"address": "北京市朝阳区xxx街道xxx号"
|
||
}
|
||
}
|
||
|
||
response = self.request('POST', '/api/members/redeem', json=payload)
|
||
|
||
if response and response.get('success'):
|
||
data = response.get('data', {})
|
||
self.order_id = data.get('orderId')
|
||
self.log_test("兑换礼品", "PASS",
|
||
"礼品兑换成功",
|
||
{"订单ID": self.order_id})
|
||
return True
|
||
else:
|
||
message = response.get('message', '兑换失败') if response else '兑换失败'
|
||
self.log_test("兑换礼品", "FAIL", message)
|
||
return False
|
||
|
||
def test_get_member_orders(self):
|
||
"""测试查询会员订单"""
|
||
print("\n=== 测试：查询会员订单 ===")
|
||
|
||
response = self.request('GET', '/api/members/orders')
|
||
|
||
if response and response.get('success'):
|
||
data = response.get('data', [])
|
||
self.log_test("查询会员订单", "PASS",
|
||
f"共 {len(data)} 个订单",
|
||
{"订单数": len(data)})
|
||
return True
|
||
else:
|
||
self.log_test("查询会员订单", "FAIL", "查询失败")
|
||
return False
|
||
|
||
def test_admin_get_orders(self):
|
||
"""测试管理员查询订单"""
|
||
print("\n=== 测试：管理员查询订单 ===")
|
||
|
||
response = self.request('GET', '/api/admin/redemption-orders')
|
||
|
||
if response and response.get('success'):
|
||
data = response.get('data', [])
|
||
pending = [o for o in data if o.get('status') == 1]
|
||
self.log_test("管理员查询订单", "PASS",
|
||
f"共 {len(data)} 个订单，{len(pending)} 个待处理",
|
||
{"总数": len(data), "待处理": len(pending)})
|
||
return True
|
||
else:
|
||
self.log_test("管理员查询订单", "FAIL", "查询失败")
|
||
return False
|
||
|
||
def test_dispatch_order(self):
|
||
"""测试订单发货"""
|
||
print("\n=== 测试：订单发货 ===")
|
||
|
||
if not self.order_id:
|
||
self.log_test("订单发货", "SKIP", "没有订单ID")
|
||
return False
|
||
|
||
payload = {
|
||
"trackingNumber": "SF1234567890"
|
||
}
|
||
|
||
response = self.request('POST', f'/api/admin/redemption-orders/{self.order_id}/dispatch',
|
||
json=payload)
|
||
|
||
if response and response.get('success'):
|
||
self.log_test("订单发货", "PASS",
|
||
"订单发货成功",
|
||
{"物流单号": payload['trackingNumber']})
|
||
return True
|
||
else:
|
||
self.log_test("订单发货", "FAIL", "发货失败")
|
||
return False
|
||
|
||
def generate_report(self):
|
||
"""生成测试报告"""
|
||
print("\n" + "="*60)
|
||
print("API 测试报告")
|
||
print("="*60)
|
||
|
||
total = len(self.test_results)
|
||
passed = sum(1 for r in self.test_results if r['status'] == 'PASS')
|
||
failed = sum(1 for r in self.test_results if r['status'] == 'FAIL')
|
||
skipped = sum(1 for r in self.test_results if r['status'] == 'SKIP')
|
||
|
||
print(f"\n总测试数: {total}")
|
||
print(f"通过: {passed} ✅")
|
||
print(f"失败: {failed} ❌")
|
||
print(f"跳过: {skipped} ⏭️")
|
||
if total > 0:
|
||
print(f"成功率: {(passed/total*100):.2f}%")
|
||
|
||
print("\n详细结果:")
|
||
print("-"*60)
|
||
for result in self.test_results:
|
||
status_emoji = "✅" if result['status'] == 'PASS' else "❌" if result['status'] == 'FAIL' else "⏭️"
|
||
print(f"{status_emoji} {result['test']}: {result['message']}")
|
||
|
||
# 保存JSON报告
|
||
with open('test_api_report.json', 'w', encoding='utf-8') as f:
|
||
json.dump(self.test_results, f, ensure_ascii=False, indent=2)
|
||
|
||
print(f"\n测试报告已保存到: test_api_report.json")
|
||
|
||
# 返回是否有失败
|
||
return failed == 0
|
||
|
||
def run_api_tests():
|
||
"""运行所有API测试"""
|
||
print("="*60)
|
||
print("一物一码会员营销系统 - API 业务流程测试")
|
||
print("="*60)
|
||
|
||
tester = APITester()
|
||
|
||
# 1. 健康检查
|
||
if not tester.test_health_check():
|
||
print("\n❌ 后端服务不可用，测试终止")
|
||
return False
|
||
|
||
# 2. 管理后台功能测试
|
||
print("\n【测试管理后台功能】")
|
||
tester.test_generate_marketing_codes()
|
||
tester.test_create_points_rule()
|
||
tester.test_create_gift()
|
||
tester.test_put_gift_on_shelf()
|
||
|
||
# 3. 会员功能测试
|
||
print("\n【测试会员功能】")
|
||
if tester.test_member_register():
|
||
tester.test_scan_marketing_code()
|
||
tester.test_duplicate_scan() # 验证唯一性
|
||
tester.test_get_member_points()
|
||
|
||
# 4. 积分商城测试
|
||
print("\n【测试积分商城】")
|
||
tester.test_get_gifts_list()
|
||
tester.test_redeem_gift() # 可能因积分不足失败
|
||
tester.test_get_member_orders()
|
||
|
||
# 5. 订单管理测试
|
||
print("\n【测试订单管理】")
|
||
tester.test_admin_get_orders()
|
||
tester.test_dispatch_order()
|
||
|
||
# 生成报告
|
||
success = tester.generate_report()
|
||
|
||
return success
|
||
|
||
if __name__ == "__main__":
|
||
success = run_api_tests()
|
||
exit(0 if success else 1)
|