sam
d88ec60ef4
- 在MarketingCode聚合中新增品类ID和品类名称字段，完善产品信息结构 - 迁移生成营销码命令，支持传入品类ID和品类名称参数 - 积分发放失败时发送积分获得失败通知集成事件 - 新增通知发送及积分失败通知的集成事件处理器，使用SSE推送通知 - 在积分相关集成事件处理器中添加发送积分变动通知功能 - 移除Notification聚合，相关数据库表删除 - 新增分页结果类型PagedResult，支持营销码查询分页返回 - 营销码查询支持分页参数，返回分页结果数据
228 lines
7.5 KiB
Python
228 lines
7.5 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
积分通知系统测试脚本
|
||
用于验证实时通知机制是否正常工作
|
||
"""
|
||
|
||
import requests
|
||
import json
|
||
import time
|
||
from datetime import datetime
|
||
|
||
# 配置
|
||
BASE_URL = "http://localhost:5511"
|
||
API_TIMEOUT = 10
|
||
|
||
def test_health_check():
|
||
"""测试API健康检查"""
|
||
print("🔍 测试API健康检查...")
|
||
try:
|
||
response = requests.get(f"{BASE_URL}/health", timeout=API_TIMEOUT)
|
||
if response.status_code == 200:
|
||
print("✅ API服务正常运行")
|
||
return True
|
||
else:
|
||
print(f"❌ API健康检查失败: {response.status_code}")
|
||
return False
|
||
except Exception as e:
|
||
print(f"❌ 无法连接到API服务: {e}")
|
||
return False
|
||
|
||
def test_member_login():
|
||
"""测试会员登录获取token"""
|
||
print("\n🔐 测试会员登录...")
|
||
login_data = {
|
||
"phoneNumber": "13800138000",
|
||
"verificationCode": "123456"
|
||
}
|
||
|
||
try:
|
||
response = requests.post(
|
||
f"{BASE_URL}/api/members/login",
|
||
json=login_data,
|
||
timeout=API_TIMEOUT
|
||
)
|
||
|
||
if response.status_code == 200:
|
||
result = response.json()
|
||
token = result.get("data", {}).get("token")
|
||
member_id = result.get("data", {}).get("member", {}).get("id")
|
||
if token and member_id:
|
||
print(f"✅ 登录成功")
|
||
print(f" Token: {token[:20]}...")
|
||
print(f" Member ID: {member_id}")
|
||
return token, member_id
|
||
else:
|
||
print("❌ 登录响应格式不正确")
|
||
return None, None
|
||
else:
|
||
print(f"❌ 登录失败: {response.status_code}")
|
||
print(f" 响应: {response.text}")
|
||
return None, None
|
||
except Exception as e:
|
||
print(f"❌ 登录请求失败: {e}")
|
||
return None, None
|
||
|
||
def test_sse_connection(token):
|
||
"""测试SSE连接"""
|
||
print("\n📡 测试SSE实时通知连接...")
|
||
headers = {"Authorization": f"Bearer {token}"}
|
||
|
||
try:
|
||
# 使用requests.Session保持连接
|
||
with requests.Session() as session:
|
||
response = session.get(
|
||
f"{BASE_URL}/api/notifications/sse",
|
||
headers=headers,
|
||
stream=True,
|
||
timeout=30
|
||
)
|
||
|
||
if response.status_code == 200:
|
||
print("✅ SSE连接建立成功")
|
||
print(" 开始监听实时通知...")
|
||
|
||
# 读取几条消息进行测试
|
||
message_count = 0
|
||
for line in response.iter_lines():
|
||
if line:
|
||
message_count += 1
|
||
print(f" 收到消息 #{message_count}: {line.decode('utf-8')}")
|
||
if message_count >= 3: # 只读取前3条消息
|
||
break
|
||
|
||
return True
|
||
else:
|
||
print(f"❌ SSE连接失败: {response.status_code}")
|
||
return False
|
||
|
||
except Exception as e:
|
||
print(f"❌ SSE连接异常: {e}")
|
||
return False
|
||
|
||
def test_marketing_code_scan(member_id, token):
|
||
"""测试营销码扫描触发通知"""
|
||
print("\n📱 测试营销码扫描...")
|
||
|
||
# 模拟营销码扫描数据
|
||
scan_data = {
|
||
"code": "TEST-CODE-001",
|
||
"productId": "PROD-001",
|
||
"productName": "测试产品",
|
||
"categoryId": "CAT-001"
|
||
}
|
||
|
||
headers = {
|
||
"Authorization": f"Bearer {token}",
|
||
"Content-Type": "application/json"
|
||
}
|
||
|
||
try:
|
||
response = requests.post(
|
||
f"{BASE_URL}/api/marketing-codes/scan",
|
||
json=scan_data,
|
||
headers=headers,
|
||
timeout=API_TIMEOUT
|
||
)
|
||
|
||
print(f" 请求状态: {response.status_code}")
|
||
if response.status_code == 200:
|
||
result = response.json()
|
||
print(f"✅ 营销码扫描成功")
|
||
print(f" 响应: {json.dumps(result, indent=2, ensure_ascii=False)}")
|
||
return True
|
||
else:
|
||
print(f"⚠️ 营销码扫描返回: {response.status_code}")
|
||
print(f" 响应内容: {response.text}")
|
||
# 即使返回非200状态，也可能触发通知机制
|
||
return True
|
||
except Exception as e:
|
||
print(f"❌ 营销码扫描请求失败: {e}")
|
||
return False
|
||
|
||
def test_notification_history(member_id, token):
|
||
"""测试通知历史查询"""
|
||
print("\n📜 测试通知历史查询...")
|
||
|
||
headers = {"Authorization": f"Bearer {token}"}
|
||
|
||
try:
|
||
response = requests.get(
|
||
f"{BASE_URL}/api/notifications/history?pageSize=10&pageNumber=1",
|
||
headers=headers,
|
||
timeout=API_TIMEOUT
|
||
)
|
||
|
||
if response.status_code == 200:
|
||
result = response.json()
|
||
notifications = result.get("data", {}).get("items", [])
|
||
print(f"✅ 获取通知历史成功")
|
||
print(f" 找到 {len(notifications)} 条通知")
|
||
|
||
for i, notification in enumerate(notifications[:3]): # 显示前3条
|
||
print(f" 通知 #{i+1}:")
|
||
print(f" 类型: {notification.get('type')}")
|
||
print(f" 标题: {notification.get('title')}")
|
||
print(f" 内容: {notification.get('content')}")
|
||
print(f" 时间: {notification.get('createdAt')}")
|
||
|
||
return True
|
||
else:
|
||
print(f"❌ 获取通知历史失败: {response.status_code}")
|
||
return False
|
||
except Exception as e:
|
||
print(f"❌ 通知历史查询失败: {e}")
|
||
return False
|
||
|
||
def main():
|
||
"""主测试函数"""
|
||
print("=" * 50)
|
||
print("🚀 积分通知系统测试开始")
|
||
print("=" * 50)
|
||
print(f"时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
|
||
|
||
# 1. 健康检查
|
||
if not test_health_check():
|
||
print("\n❌ 基础服务检查失败，退出测试")
|
||
return
|
||
|
||
# 2. 会员登录
|
||
token, member_id = test_member_login()
|
||
if not token:
|
||
print("\n❌ 无法获取认证信息，退出测试")
|
||
return
|
||
|
||
# 3. 测试SSE连接
|
||
sse_success = test_sse_connection(token)
|
||
|
||
# 4. 测试营销码扫描
|
||
scan_success = test_marketing_code_scan(member_id, token)
|
||
|
||
# 等待一段时间让事件处理完成
|
||
print("\n⏳ 等待事件处理完成...")
|
||
time.sleep(3)
|
||
|
||
# 5. 测试通知历史
|
||
history_success = test_notification_history(member_id, token)
|
||
|
||
# 测试总结
|
||
print("\n" + "=" * 50)
|
||
print("📊 测试结果总结:")
|
||
print("=" * 50)
|
||
print(f"✅ 健康检查: {'通过' if True else '失败'}")
|
||
print(f"✅ 会员登录: {'通过' if token else '失败'}")
|
||
print(f"✅ SSE连接: {'通过' if sse_success else '失败'}")
|
||
print(f"✅ 营销码扫描: {'通过' if scan_success else '失败'}")
|
||
print(f"✅ 通知历史: {'通过' if history_success else '失败'}")
|
||
|
||
success_count = sum([True, bool(token), sse_success, scan_success, history_success])
|
||
total_tests = 5
|
||
print(f"\n🎯 总体成功率: {success_count}/{total_tests} ({success_count/total_tests*100:.1f}%)")
|
||
|
||
if success_count == total_tests:
|
||
print("🎉 所有测试通过！通知系统工作正常")
|
||
else:
|
||
print("⚠️ 部分测试未通过，请检查相关功能")
|
||
|
||
if __name__ == "__main__":
|
||
main() |