sam
d88ec60ef4
- 在MarketingCode聚合中新增品类ID和品类名称字段，完善产品信息结构 - 迁移生成营销码命令，支持传入品类ID和品类名称参数 - 积分发放失败时发送积分获得失败通知集成事件 - 新增通知发送及积分失败通知的集成事件处理器，使用SSE推送通知 - 在积分相关集成事件处理器中添加发送积分变动通知功能 - 移除Notification聚合，相关数据库表删除 - 新增分页结果类型PagedResult，支持营销码查询分页返回 - 营销码查询支持分页参数，返回分页结果数据
230 lines
7.7 KiB
Python
230 lines
7.7 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
SSE通知功能集成测试脚本
|
||
验证前端页面是否能正确接收和处理积分变动通知
|
||
"""
|
||
|
||
import requests
|
||
import json
|
||
import time
|
||
from datetime import datetime
|
||
|
||
# 配置
|
||
BASE_URL = "http://localhost:5000"
|
||
ADMIN_EMAIL = "admin@example.com"
|
||
ADMIN_PASSWORD = "Admin123!"
|
||
|
||
def log_step(step_num, description, status="INFO"):
|
||
"""记录测试步骤"""
|
||
timestamp = datetime.now().strftime("%H:%M:%S")
|
||
status_icon = {
|
||
"PASS": "✅",
|
||
"FAIL": "❌",
|
||
"INFO": "ℹ️",
|
||
"WARN": "⚠️"
|
||
}
|
||
print(f"[{timestamp}] {status_icon.get(status, 'ℹ️')} 步骤 {step_num}: {description}")
|
||
|
||
def get_admin_token():
|
||
"""获取管理员token"""
|
||
try:
|
||
response = requests.post(
|
||
f"{BASE_URL}/api/admins/login",
|
||
json={
|
||
"email": ADMIN_EMAIL,
|
||
"password": ADMIN_PASSWORD
|
||
}
|
||
)
|
||
if response.status_code == 200:
|
||
data = response.json()
|
||
return data.get('data', {}).get('token')
|
||
return None
|
||
except Exception as e:
|
||
print(f"获取token失败: {e}")
|
||
return None
|
||
|
||
def create_test_member(admin_token):
|
||
"""创建测试会员"""
|
||
try:
|
||
response = requests.post(
|
||
f"{BASE_URL}/api/members",
|
||
headers={"Authorization": f"Bearer {admin_token}"},
|
||
json={
|
||
"phone": f"13800138{int(time.time()) % 10000:04d}",
|
||
"nickname": f"测试用户{int(time.time()) % 1000}",
|
||
"initialPoints": 1000
|
||
}
|
||
)
|
||
if response.status_code == 200:
|
||
data = response.json()
|
||
member_id = data.get('data', {}).get('id')
|
||
log_step(1, f"创建测试会员成功: {member_id}", "PASS")
|
||
return member_id
|
||
else:
|
||
log_step(1, f"创建测试会员失败: {response.text}", "FAIL")
|
||
return None
|
||
except Exception as e:
|
||
log_step(1, f"创建测试会员异常: {e}", "FAIL")
|
||
return None
|
||
|
||
def get_member_token(member_id):
|
||
"""模拟会员登录获取token"""
|
||
try:
|
||
# 这里简化处理，实际应该有会员登录接口
|
||
# 暂时返回None，因为我们主要测试SSE通知机制
|
||
log_step(2, "获取会员token (简化处理)", "INFO")
|
||
return "test_member_token"
|
||
except Exception as e:
|
||
log_step(2, f"获取会员token失败: {e}", "FAIL")
|
||
return None
|
||
|
||
def trigger_points_event(admin_token, member_id, event_type, amount):
|
||
"""触发积分事件"""
|
||
try:
|
||
event_mapping = {
|
||
"earned": "PointsEarnedSuccess",
|
||
"consumed": "PointsConsumed",
|
||
"expired": "PointsExpired",
|
||
"refunded": "PointsRefunded"
|
||
}
|
||
|
||
event_data = {
|
||
"memberId": member_id,
|
||
"amount": amount,
|
||
"relatedId": f"test_{int(time.time())}",
|
||
"source": "测试系统"
|
||
}
|
||
|
||
if event_type == "consumed":
|
||
event_data["reason"] = "测试消费"
|
||
event_data["orderId"] = f"order_{int(time.time())}"
|
||
|
||
response = requests.post(
|
||
f"{BASE_URL}/api/test/trigger-points-event",
|
||
headers={"Authorization": f"Bearer {admin_token}"},
|
||
json={
|
||
"eventType": event_mapping[event_type],
|
||
"eventData": event_data
|
||
}
|
||
)
|
||
|
||
if response.status_code == 200:
|
||
log_step(3, f"触发积分{event_type}事件成功: {amount}积分", "PASS")
|
||
return True
|
||
else:
|
||
# 如果测试接口不存在，尝试直接调用积分API
|
||
log_step(3, f"触发积分{event_type}事件失败，尝试备用方案", "WARN")
|
||
return trigger_points_via_api(admin_token, member_id, event_type, amount)
|
||
except Exception as e:
|
||
log_step(3, f"触发积分事件异常: {e}", "FAIL")
|
||
return False
|
||
|
||
def trigger_points_via_api(admin_token, member_id, event_type, amount):
|
||
"""通过API触发积分变动"""
|
||
try:
|
||
if event_type == "earned":
|
||
# 通过营销码扫码获得积分
|
||
response = requests.post(
|
||
f"{BASE_URL}/api/marketing-codes/mock-scan",
|
||
headers={"Authorization": f"Bearer {admin_token}"},
|
||
json={
|
||
"memberId": member_id,
|
||
"points": amount
|
||
}
|
||
)
|
||
elif event_type == "consumed":
|
||
# 通过创建兑换订单消费积分
|
||
response = requests.post(
|
||
f"{BASE_URL}/api/redemption-orders/mock-consume",
|
||
headers={"Authorization": f"Bearer {admin_token}"},
|
||
json={
|
||
"memberId": member_id,
|
||
"points": amount,
|
||
"reason": "测试消费"
|
||
}
|
||
)
|
||
else:
|
||
log_step(3, f"暂不支持的事件类型: {event_type}", "WARN")
|
||
return True
|
||
|
||
if response.status_code in [200, 201]:
|
||
log_step(3, f"通过API触发积分{event_type}成功: {amount}积分", "PASS")
|
||
return True
|
||
else:
|
||
log_step(3, f"通过API触发积分{event_type}失败: {response.text}", "FAIL")
|
||
return False
|
||
except Exception as e:
|
||
log_step(3, f"通过API触发积分事件异常: {e}", "FAIL")
|
||
return False
|
||
|
||
def verify_frontend_updates():
|
||
"""验证前端页面更新（模拟）"""
|
||
log_step(4, "验证前端SSE通知接收", "INFO")
|
||
print(" 请在前端页面观察以下变化:")
|
||
print(" - 首页: 积分余额应该自动更新")
|
||
print(" - 积分明细: 应该显示新的积分记录")
|
||
print(" - 个人中心: 积分信息应该同步更新")
|
||
print(" - 购物车/结算页面: 可用积分应该更新")
|
||
print(" - 通知气泡: 应该显示积分变动提醒")
|
||
|
||
input(" 手动验证完成后按回车继续...")
|
||
|
||
def test_sse_connection():
|
||
"""测试SSE连接"""
|
||
log_step(5, "测试SSE连接建立", "INFO")
|
||
print(" 请在浏览器开发者工具中检查:")
|
||
print(" - Network标签页应该显示到/api/notifications/sse的连接")
|
||
print(" - 连接状态应该是CONNECTED")
|
||
print(" - 应该定期收到心跳消息")
|
||
|
||
input(" 连接验证完成后按回车继续...")
|
||
|
||
def main():
|
||
"""主测试流程"""
|
||
print("=" * 60)
|
||
print("SSE通知功能集成测试")
|
||
print("=" * 60)
|
||
|
||
# 步骤1: 获取管理员权限
|
||
log_step(0, "获取管理员权限")
|
||
admin_token = get_admin_token()
|
||
if not admin_token:
|
||
log_step(0, "无法获取管理员权限，测试终止", "FAIL")
|
||
return
|
||
|
||
# 步骤2: 创建测试会员
|
||
member_id = create_test_member(admin_token)
|
||
if not member_id:
|
||
return
|
||
|
||
# 步骤3: 测试各种积分事件
|
||
events_to_test = [
|
||
("earned", 100),
|
||
("consumed", 50),
|
||
("earned", 200),
|
||
("expired", 10)
|
||
]
|
||
|
||
for event_type, amount in events_to_test:
|
||
print(f"\n--- 测试积分{event_type}事件 ---")
|
||
success = trigger_points_event(admin_token, member_id, event_type, amount)
|
||
if success:
|
||
# 等待一段时间让前端处理通知
|
||
time.sleep(3)
|
||
verify_frontend_updates()
|
||
else:
|
||
log_step(3, f"积分{event_type}事件测试失败", "FAIL")
|
||
|
||
# 步骤4: 测试SSE连接
|
||
print(f"\n--- SSE连接测试 ---")
|
||
test_sse_connection()
|
||
|
||
# 步骤5: 总结
|
||
print("\n" + "=" * 60)
|
||
print("测试完成!")
|
||
print("请检查前端页面是否正确显示了所有积分变动通知")
|
||
print("确保通知气泡正常显示且页面数据及时更新")
|
||
print("=" * 60)
|
||
|
||
if __name__ == "__main__":
|
||
main() |