Project.Fengling.QoderVersion/tests/sse_notification_tests/sse_notification_verification.py

#!/usr/bin/env python3
"""
SSE通知发送验证测试
专门验证后端是否正确发送积分变动通知
"""

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 test_sse_notification_sending():
    """测试SSE通知发送机制"""
    log("=== SSE通知发送机制验证测试 ===", "INFO")
    
    # 使用已知的测试账号
    log("1. 使用测试账号登录...", "INFO")
    login_data = {
        "phone": "15921072307",
        "password": "Sl52788542"
    }
    
    try:
        # 登录获取token
        login_resp = requests.post(
            f"{BASE_URL}/api/members/login",
            json=login_data,
            timeout=10
        )
        
        if login_resp.status_code != 200:
            log(f"❌ 登录失败: {login_resp.status_code}", "FAIL")
            return
            
        login_result = login_resp.json()
        token = login_result.get("data", {}).get("token")
        member_id = login_result.get("data", {}).get("memberId")
        log(f"✅ 登录成功 - Member ID: {member_id}", "PASS")
        
        headers = {"Authorization": f"Bearer {token}"}
        
        # 2. 检查初始通知历史
        log("2. 检查初始通知历史...", "INFO")
        try:
            history_resp = requests.get(
                f"{BASE_URL}/api/notifications/history?pageSize=5&pageNumber=1",
                headers=headers,
                timeout=10
            )
            
            if history_resp.status_code == 200:
                history_data = history_response.json()
                initial_count = len(history_data.get('data', {}).get('items', []))
                log(f"✅ 初始通知数量: {initial_count}", "INFO")
            else:
                log(f"⚠️  获取通知历史失败: {history_resp.status_code}", "WARN")
                initial_count = 0
        except Exception as e:
            log(f"⚠️  检查通知历史异常: {e}", "WARN")
            initial_count = 0
            
        # 3. 获取可用营销码并扫码
        log("3. 执行扫码操作...", "INFO")
        codes_resp = requests.get(
            f"{BASE_URL}/api/admin/marketing-codes?batchNo=001&pageSize=3&pageNumber=1"
        )
        
        if codes_resp.status_code == 200:
            codes_data = codes_resp.json()
            available_codes = [
                item for item in codes_data.get('data', {}).get('items', [])
                if not item.get('isUsed', True)
            ]
            
            if available_codes:
                test_code = available_codes[0]['code']
                log(f"✅ 使用营销码: {test_code}", "INFO")
                
                # 执行扫码
                scan_resp = requests.post(
                    f"{BASE_URL}/api/marketing-codes/scan",
                    json={"code": test_code},
                    headers=headers,
                    timeout=10
                )
                
                if scan_resp.status_code == 200:
                    scan_result = scan_resp.json()
                    message = scan_result.get('data', {}).get('message', '')
                    log(f"✅ 扫码成功: {message}", "PASS")
                else:
                    log(f"❌ 扫码失败: {scan_resp.status_code}", "FAIL")
                    return
            else:
                log("❌ 没有可用的营销码", "FAIL")
                return
        else:
            log(f"❌ 获取营销码失败: {codes_resp.status_code}", "FAIL")
            return
            
        # 4. 等待并检查通知是否发送
        log("4. 等待通知发送并验证...", "INFO")
        time.sleep(10)  # 给足够时间让通知发送
        
        # 检查通知历史变化
        try:
            history_resp2 = requests.get(
                f"{BASE_URL}/api/notifications/history?pageSize=10&pageNumber=1",
                headers=headers,
                timeout=10
            )
            
            if history_resp2.status_code == 200:
                history_data2 = history_resp2.json()
                final_items = history_data2.get('data', {}).get('items', [])
                final_count = len(final_items)
                
                log(f"✅ 最终通知数量: {final_count}", "INFO")
                log(f"✅ 通知数量变化: {final_count - initial_count}", "INFO")
                
                # 查找积分相关通知
                points_notifications = [
                    item for item in final_items
                    if '积分' in item.get('title', '') or '积分' in item.get('content', '')
                ]
                
                if points_notifications:
                    log(f"✅ 找到 {len(points_notifications)} 条积分通知:", "PASS")
                    for i, noti in enumerate(points_notifications[:3]):
                        log(f"   通知 {i+1}: {noti.get('title')} - {noti.get('content')}", "INFO")
                        log(f"            时间: {noti.get('createdAt')}", "INFO")
                else:
                    log("⚠️  未找到积分相关通知", "WARN")
                    log("   可能的原因:", "INFO")
                    log("   1. 通知发送机制需要调试", "INFO")
                    log("   2. 积分事件处理未触发通知", "INFO")
                    log("   3. 通知过滤条件需要调整", "INFO")
                    
            else:
                log(f"❌ 获取最终通知历史失败: {history_resp2.status_code}", "FAIL")
                
        except Exception as e:
            log(f"❌ 检查通知历史异常: {e}", "FAIL")
            
        # 5. 验证积分是否实际增加
        log("5. 验证积分实际变化...", "INFO")
        current_resp = requests.get(f"{BASE_URL}/api/members/current", headers=headers)
        if current_resp.status_code == 200:
            current_data = current_resp.json()
            final_points = current_data.get('availablePoints', 0)
            log(f"✅ 当前积分: {final_points}", "INFO")
            
            if final_points > 0:
                log("✅ 积分确实增加了", "PASS")
            else:
                log("⚠️  积分未增加，需要检查积分处理逻辑", "WARN")
        
        # 6. 总结
        log("=== 测试总结 ===", "INFO")
        if points_notifications:
            log("✅ SSE通知发送机制工作正常", "PASS")
        else:
            log("❌ SSE通知发送机制可能存在问题", "FAIL")
            
        if final_points > 0:
            log("✅ 积分处理机制工作正常", "PASS")
        else:
            log("❌ 积分处理机制可能存在问题", "FAIL")
            
    except Exception as e:
        log(f"❌ 测试过程中出现异常: {e}", "FAIL")

if __name__ == "__main__":
    test_sse_notification_sending()