sam
d88ec60ef4
- 在MarketingCode聚合中新增品类ID和品类名称字段，完善产品信息结构 - 迁移生成营销码命令，支持传入品类ID和品类名称参数 - 积分发放失败时发送积分获得失败通知集成事件 - 新增通知发送及积分失败通知的集成事件处理器，使用SSE推送通知 - 在积分相关集成事件处理器中添加发送积分变动通知功能 - 移除Notification聚合，相关数据库表删除 - 新增分页结果类型PagedResult，支持营销码查询分页返回 - 营销码查询支持分页参数，返回分页结果数据
126 lines
4.3 KiB
Python
126 lines
4.3 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
简单验证测试 - 检查CAP修复后的效果
|
||
"""
|
||
|
||
import requests
|
||
import json
|
||
import time
|
||
from datetime import datetime
|
||
|
||
BASE_URL = "http://localhost:5511"
|
||
|
||
def log(msg, status="INFO"):
|
||
timestamp = datetime.now().strftime("%H:%M:%S")
|
||
icons = {"PASS": "✅", "FAIL": "❌", "INFO": "ℹ️", "WARN": "⚠️"}
|
||
print(f"[{timestamp}] {icons.get(status, 'ℹ️')} {msg}")
|
||
|
||
def quick_verification_test():
|
||
"""快速验证测试"""
|
||
log("=== CAP修复验证测试 ===", "INFO")
|
||
|
||
try:
|
||
# 1. 用户登录
|
||
log("1. 用户登录...", "INFO")
|
||
user_login_data = {
|
||
"phone": "15921072307",
|
||
"password": "Sl52788542"
|
||
}
|
||
|
||
user_resp = requests.post(
|
||
f"{BASE_URL}/api/members/login",
|
||
json=user_login_data,
|
||
timeout=10
|
||
)
|
||
|
||
if user_resp.status_code != 200:
|
||
log("❌ 用户登录失败", "FAIL")
|
||
return
|
||
|
||
user_token = user_resp.json()['data']['token']
|
||
user_headers = {"Authorization": f"Bearer {user_token}"}
|
||
log("✅ 用户登录成功", "PASS")
|
||
|
||
# 2. 建立SSE连接
|
||
log("2. 建立SSE连接...", "INFO")
|
||
notifications_received = []
|
||
|
||
def collect_notifications():
|
||
try:
|
||
sse_resp = requests.get(
|
||
f"{BASE_URL}/api/notifications/sse",
|
||
headers=user_headers,
|
||
stream=True,
|
||
timeout=20
|
||
)
|
||
|
||
if sse_resp.status_code == 200:
|
||
for line in sse_resp.iter_lines():
|
||
if line:
|
||
try:
|
||
line_str = line.decode('utf-8')
|
||
if line_str.startswith('data: '):
|
||
data_str = line_str[6:]
|
||
if data_str.strip():
|
||
notification = json.loads(data_str)
|
||
notifications_received.append(notification)
|
||
notification_type = notification.get('type', 'unknown')
|
||
title = notification.get('title', '')
|
||
log(f" 收到 [{notification_type}]: {title}", "INFO")
|
||
except:
|
||
pass
|
||
except:
|
||
pass
|
||
|
||
# 启动SSE监听线程
|
||
import threading
|
||
sse_thread = threading.Thread(target=collect_notifications, daemon=True)
|
||
sse_thread.start()
|
||
|
||
time.sleep(2) # 等待连接建立
|
||
|
||
# 3. 执行扫码
|
||
log("3. 执行扫码...", "INFO")
|
||
scan_data = {"code": "011-000049-20260213075254-2875"}
|
||
scan_resp = requests.post(
|
||
f"{BASE_URL}/api/marketing-codes/scan",
|
||
json=scan_data,
|
||
headers=user_headers,
|
||
timeout=15
|
||
)
|
||
|
||
if scan_resp.status_code == 200:
|
||
log("✅ 扫码成功", "PASS")
|
||
else:
|
||
log(f"❌ 扫码失败: {scan_resp.status_code}", "FAIL")
|
||
return
|
||
|
||
# 4. 等待并检查通知
|
||
log("4. 等待通知...", "INFO")
|
||
initial_count = len(notifications_received)
|
||
|
||
time.sleep(15) # 等待15秒
|
||
|
||
final_count = len(notifications_received)
|
||
new_notifications = notifications_received[initial_count:] if final_count > initial_count else []
|
||
|
||
points_notifications = [
|
||
n for n in new_notifications
|
||
if '积分' in n.get('title', '') or '积分' in n.get('message', '')
|
||
]
|
||
|
||
log(f" 收到新通知: {len(new_notifications)} 条", "INFO")
|
||
log(f" 积分相关通知: {len(points_notifications)} 条", "INFO")
|
||
|
||
# 5. 结论
|
||
if len(points_notifications) > 0:
|
||
log("🎉 积分通知推送成功！CAP修复生效！", "PASS")
|
||
else:
|
||
log("❌ 仍然没有收到积分通知", "FAIL")
|
||
log(" 可能需要进一步检查CAP配置", "WARN")
|
||
|
||
except Exception as e:
|
||
log(f"❌ 测试异常: {e}", "FAIL")
|
||
|
||
if __name__ == "__main__":
|
||
quick_verification_test() |