sam
d88ec60ef4
- 在MarketingCode聚合中新增品类ID和品类名称字段，完善产品信息结构 - 迁移生成营销码命令，支持传入品类ID和品类名称参数 - 积分发放失败时发送积分获得失败通知集成事件 - 新增通知发送及积分失败通知的集成事件处理器，使用SSE推送通知 - 在积分相关集成事件处理器中添加发送积分变动通知功能 - 移除Notification聚合，相关数据库表删除 - 新增分页结果类型PagedResult，支持营销码查询分页返回 - 营销码查询支持分页参数，返回分页结果数据
108 lines
4.3 KiB
Python
108 lines
4.3 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
简单测试SSE修复效果
|
||
"""
|
||
|
||
import requests
|
||
import json
|
||
import time
|
||
|
||
BASE_URL = "http://localhost:5511"
|
||
|
||
def test_basic_flow():
|
||
print("=== 基础流程测试 ===")
|
||
|
||
# 1. 登录
|
||
print("1. 测试登录...")
|
||
login_data = {
|
||
"phone": "15921072307",
|
||
"password": "Sl52788542"
|
||
}
|
||
|
||
try:
|
||
response = requests.post(
|
||
f"{BASE_URL}/api/members/login",
|
||
json=login_data,
|
||
timeout=10
|
||
)
|
||
|
||
if response.status_code == 200:
|
||
result = response.json()
|
||
token = result.get("data", {}).get("token")
|
||
member_id = result.get("data", {}).get("memberId")
|
||
print(f"✅ 登录成功 - Member ID: {member_id}")
|
||
print(f" Token: {token[:30]}...")
|
||
|
||
headers = {"Authorization": f"Bearer {token}"}
|
||
|
||
# 2. 检查当前积分
|
||
print("\n2. 检查当前积分...")
|
||
current_resp = requests.get(f"{BASE_URL}/api/members/current", headers=headers)
|
||
if current_resp.status_code == 200:
|
||
current_data = current_resp.json()
|
||
points = current_data.get('availablePoints', 0)
|
||
print(f"✅ 当前积分: {points}")
|
||
|
||
# 3. 测试扫码接口
|
||
print("\n3. 测试扫码接口...")
|
||
scan_data = {"code": "001-000099-20260211095706-3756"} # 使用有效的未使用营销码测试
|
||
|
||
scan_resp = requests.post(
|
||
f"{BASE_URL}/api/marketing-codes/scan",
|
||
json=scan_data,
|
||
headers=headers,
|
||
timeout=10
|
||
)
|
||
|
||
print(f"状态码: {scan_resp.status_code}")
|
||
print(f"响应内容: {scan_resp.text}")
|
||
|
||
if scan_resp.status_code == 200:
|
||
scan_result = scan_resp.json()
|
||
data = scan_result.get('data', {})
|
||
earned_points = data.get('earnedPoints', 'N/A')
|
||
message = data.get('message', 'N/A')
|
||
product_name = data.get('productName', 'N/A')
|
||
|
||
print(f"✅ 扫码响应分析:")
|
||
print(f" earnedPoints: {earned_points}")
|
||
print(f" message: {message}")
|
||
print(f" productName: {product_name}")
|
||
|
||
# 验证修复效果
|
||
if str(earned_points) == '0' and '发放' in str(message):
|
||
print("✅ 修复验证通过: 返回正确的处理中状态")
|
||
|
||
# 额外验证：检查积分是否真的增加了
|
||
print("\n4. 验证积分是否实际增加...")
|
||
time.sleep(5) # 延长等待时间让事件处理完成
|
||
|
||
final_resp = requests.get(f"{BASE_URL}/api/members/current", headers=headers)
|
||
if final_resp.status_code == 200:
|
||
final_data = final_resp.json()
|
||
final_points = final_data.get('availablePoints', 0)
|
||
print(f" 扫码后积分: {final_points}")
|
||
|
||
if final_points > points:
|
||
print(f"✅ 积分确实增加了 {final_points - points} 分")
|
||
print("✅ 后台处理机制正常工作")
|
||
else:
|
||
print("⚠️ 积分未增加，可能需要检查事件处理")
|
||
|
||
elif earned_points == 0:
|
||
print("⚠️ 积分值为0，但消息内容需要确认")
|
||
else:
|
||
print(f"❌ 仍返回具体积分值: {earned_points}")
|
||
|
||
else:
|
||
print(f"❌ 获取当前积分失败: {current_resp.status_code}")
|
||
|
||
else:
|
||
print(f"❌ 登录失败: {response.status_code}")
|
||
print(f"响应: {response.text}")
|
||
|
||
except Exception as e:
|
||
print(f"❌ 测试异常: {e}")
|
||
|
||
if __name__ == "__main__":
|
||
test_basic_flow() |