sam
d88ec60ef4
- 在MarketingCode聚合中新增品类ID和品类名称字段，完善产品信息结构 - 迁移生成营销码命令，支持传入品类ID和品类名称参数 - 积分发放失败时发送积分获得失败通知集成事件 - 新增通知发送及积分失败通知的集成事件处理器，使用SSE推送通知 - 在积分相关集成事件处理器中添加发送积分变动通知功能 - 移除Notification聚合，相关数据库表删除 - 新增分页结果类型PagedResult，支持营销码查询分页返回 - 营销码查询支持分页参数，返回分页结果数据
229 lines
7.9 KiB
Python
229 lines
7.9 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
积分实时通知机制验证测试脚本
|
||
测试完整的积分获得流程和通知推送
|
||
"""
|
||
|
||
import requests
|
||
import json
|
||
import time
|
||
import threading
|
||
from datetime import datetime
|
||
|
||
# 配置
|
||
BASE_URL = "http://localhost:5511"
|
||
ADMIN_TOKEN = "admin-token" # 假设的管理员token
|
||
MEMBER_ID = "019c4c9d-13d9-70e8-a275-a0b941d91fae" # 测试会员ID
|
||
|
||
def log(message):
|
||
"""打印带时间戳的日志"""
|
||
timestamp = datetime.now().strftime("%H:%M:%S")
|
||
print(f"[{timestamp}] {message}")
|
||
|
||
def test_member_login():
|
||
"""测试会员登录获取token"""
|
||
log("=== 测试会员登录 ===")
|
||
try:
|
||
response = requests.get(f"{BASE_URL}/api/members/current")
|
||
if response.status_code == 200:
|
||
member_data = response.json()
|
||
log(f"✅ 会员登录成功: {member_data.get('nickname', 'Unknown')}")
|
||
log(f" 当前积分: {member_data.get('availablePoints', 0)}")
|
||
return True
|
||
else:
|
||
log(f"❌ 会员登录失败: {response.status_code}")
|
||
return False
|
||
except Exception as e:
|
||
log(f"❌ 会员登录异常: {str(e)}")
|
||
return False
|
||
|
||
def test_get_member_points():
|
||
"""获取会员当前积分"""
|
||
log("=== 查询会员积分 ===")
|
||
try:
|
||
response = requests.get(f"{BASE_URL}/api/members/current")
|
||
if response.status_code == 200:
|
||
member_data = response.json()
|
||
points = member_data.get('availablePoints', 0)
|
||
log(f"✅ 当前可用积分: {points}")
|
||
return points
|
||
else:
|
||
log(f"❌ 查询积分失败: {response.status_code}")
|
||
return None
|
||
except Exception as e:
|
||
log(f"❌ 查询积分异常: {str(e)}")
|
||
return None
|
||
|
||
def test_scan_marketing_code(code="TEST123"):
|
||
"""测试扫描营销码获得积分"""
|
||
log("=== 测试扫描营销码 ===")
|
||
try:
|
||
payload = {
|
||
"code": code,
|
||
"scanTime": datetime.now().isoformat()
|
||
}
|
||
response = requests.post(
|
||
f"{BASE_URL}/api/marketing-codes/scan",
|
||
json=payload,
|
||
headers={"Content-Type": "application/json"}
|
||
)
|
||
|
||
log(f"请求URL: {BASE_URL}/api/marketing-codes/scan")
|
||
log(f"请求体: {json.dumps(payload, indent=2, ensure_ascii=False)}")
|
||
log(f"响应状态: {response.status_code}")
|
||
log(f"响应内容: {response.text}")
|
||
|
||
if response.status_code == 200:
|
||
result = response.json()
|
||
log(f"✅ 扫码成功: {result}")
|
||
return True, result
|
||
else:
|
||
log(f"❌ 扫码失败: {response.status_code} - {response.text}")
|
||
return False, response.text
|
||
except Exception as e:
|
||
log(f"❌ 扫码异常: {str(e)}")
|
||
return False, str(e)
|
||
|
||
def test_sse_connection():
|
||
"""测试SSE连接"""
|
||
log("=== 测试SSE实时通知连接 ===")
|
||
try:
|
||
url = f"{BASE_URL}/api/notifications/stream?memberId={MEMBER_ID}"
|
||
log(f"连接SSE: {url}")
|
||
|
||
def sse_listener():
|
||
try:
|
||
with requests.get(url, stream=True) as response:
|
||
log(f"SSE连接状态: {response.status_code}")
|
||
if response.status_code == 200:
|
||
log("✅ SSE连接建立成功")
|
||
for line in response.iter_lines():
|
||
if line:
|
||
decoded_line = line.decode('utf-8')
|
||
if decoded_line.startswith('data:'):
|
||
data = decoded_line[5:].strip()
|
||
if data:
|
||
try:
|
||
notification = json.loads(data)
|
||
log(f"🔔 收到通知: {notification}")
|
||
except json.JSONDecodeError:
|
||
log(f"🔔 原始数据: {data}")
|
||
else:
|
||
log(f"❌ SSE连接失败: {response.status_code}")
|
||
except Exception as e:
|
||
log(f"❌ SSE监听异常: {str(e)}")
|
||
|
||
# 启动SSE监听线程
|
||
sse_thread = threading.Thread(target=sse_listener, daemon=True)
|
||
sse_thread.start()
|
||
time.sleep(2) # 等待连接建立
|
||
return sse_thread
|
||
except Exception as e:
|
||
log(f"❌ SSE连接异常: {str(e)}")
|
||
return None
|
||
|
||
def test_create_test_marketing_code():
|
||
"""创建测试用的营销码"""
|
||
log("=== 创建测试营销码 ===")
|
||
try:
|
||
# 先查询现有批次
|
||
response = requests.get(f"{BASE_URL}/api/admin/marketing-codes/batches")
|
||
if response.status_code == 200:
|
||
batches = response.json()
|
||
if batches and len(batches) > 0:
|
||
batch_no = batches[0]['batchNo']
|
||
log(f"使用现有批次: {batch_no}")
|
||
|
||
# 创建单个测试码
|
||
payload = {
|
||
"batchNo": batch_no,
|
||
"count": 1,
|
||
"prefix": "TEST"
|
||
}
|
||
create_response = requests.post(
|
||
f"{BASE_URL}/api/admin/marketing-codes/generate",
|
||
json=payload,
|
||
headers={"Content-Type": "application/json"}
|
||
)
|
||
|
||
if create_response.status_code == 200:
|
||
codes = create_response.json()
|
||
if codes and len(codes) > 0:
|
||
test_code = codes[0]
|
||
log(f"✅ 创建测试码成功: {test_code}")
|
||
return test_code
|
||
else:
|
||
log(f"❌ 创建测试码失败: {create_response.status_code}")
|
||
log(f"响应内容: {create_response.text}")
|
||
else:
|
||
log("❌ 没有可用的营销码批次")
|
||
else:
|
||
log(f"❌ 查询批次失败: {response.status_code}")
|
||
return None
|
||
except Exception as e:
|
||
log(f"❌ 创建测试码异常: {str(e)}")
|
||
return None
|
||
|
||
def main():
|
||
"""主测试流程"""
|
||
log("🚀 开始积分实时通知机制验证测试")
|
||
log("=" * 50)
|
||
|
||
# 1. 测试基础连接
|
||
if not test_member_login():
|
||
log("❌ 基础连接测试失败，退出测试")
|
||
return
|
||
|
||
# 2. 获取初始积分
|
||
initial_points = test_get_member_points()
|
||
if initial_points is None:
|
||
log("❌ 无法获取初始积分，退出测试")
|
||
return
|
||
|
||
# 3. 建立SSE连接监听通知
|
||
sse_thread = test_sse_connection()
|
||
if not sse_thread:
|
||
log("❌ SSE连接失败，继续其他测试")
|
||
|
||
# 4. 等待一段时间让SSE连接稳定
|
||
log("⏳ 等待SSE连接稳定...")
|
||
time.sleep(3)
|
||
|
||
# 5. 测试扫码获得积分
|
||
log("\n" + "=" * 30)
|
||
log("开始积分获得测试")
|
||
log("=" * 30)
|
||
|
||
success, result = test_scan_marketing_code("TEST123")
|
||
|
||
# 6. 等待通知处理
|
||
log("⏳ 等待通知处理...")
|
||
time.sleep(5)
|
||
|
||
# 7. 验证积分是否更新
|
||
log("\n" + "=" * 30)
|
||
log("验证积分更新")
|
||
log("=" * 30)
|
||
|
||
final_points = test_get_member_points()
|
||
if final_points is not None and initial_points is not None:
|
||
points_diff = final_points - initial_points
|
||
log(f"📊 积分变化: {initial_points} -> {final_points} (变化: {points_diff})")
|
||
if points_diff > 0:
|
||
log("✅ 积分成功增加")
|
||
elif points_diff == 0:
|
||
log("⚠️ 积分无变化")
|
||
else:
|
||
log("❌ 积分减少")
|
||
|
||
# 8. 总结
|
||
log("\n" + "=" * 50)
|
||
log("🎯 测试总结")
|
||
log("=" * 50)
|
||
log(f"初始积分: {initial_points}")
|
||
log(f"最终积分: {final_points}")
|
||
log(f"扫码结果: {'成功' if success else '失败'}")
|
||
log("测试完成!")
|
||
|
||
if __name__ == "__main__":
|
||
main() |