imetting/backend/test/test_redis_jwt.py

#!/usr/bin/env python3
"""
Redis JWT Token 验证脚本
用于检查JWT token是否正确存储在Redis中

运行方法:
cd /Users/jiliu/工作/projects/imeeting/backend
source venv/bin/activate  # 激活虚拟环境
python test/test_redis_jwt.py
"""
import sys
import os
import redis
import json

# 添加项目根目录到Python路径
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

try:
    from app.core.config import REDIS_CONFIG
    print("✅ 成功导入项目配置")
except ImportError as e:
    print(f"❌ 导入项目配置失败: {e}")
    print("请确保在 backend 目录下运行: python test/test_redis_jwt.py")
    sys.exit(1)

def check_jwt_in_redis():
    """检查Redis中的JWT token"""
    try:
        # 使用项目配置连接Redis
        r = redis.Redis(**REDIS_CONFIG)
        
        # 测试连接
        r.ping()
        print("✅ Redis连接成功")
        print(f"连接配置: {REDIS_CONFIG}")
        
        # 获取所有token相关的keys
        token_keys = r.keys("token:*")
        
        if not token_keys:
            print("❌ Redis中没有找到JWT token")
            print("提示: 请先通过前端登录以生成token")
            return False
        
        print(f"✅ 找到 {len(token_keys)} 个token记录:")
        
        for key in token_keys:
            # 解析key格式: token:user_id:jwt_token
            key_str = key.decode('utf-8') if isinstance(key, bytes) else key
            parts = key_str.split(":", 2)
            if len(parts) >= 3:
                user_id = parts[1]
                token_preview = parts[2][:20] + "..."
                ttl = r.ttl(key)
                value = r.get(key)
                value_str = value.decode('utf-8') if isinstance(value, bytes) else value
                
                print(f"  - 用户ID: {user_id}")
                print(f"    Token预览: {token_preview}")
                if ttl > 0:
                    print(f"    剩余时间: {ttl}秒 ({ttl/3600:.1f}小时)")
                else:
                    print(f"    TTL: {ttl} (永不过期)" if ttl == -1 else f"    TTL: {ttl} (已过期)")
                print(f"    状态: {value_str}")
                print()
        
        return True
        
    except redis.ConnectionError:
        print("❌ 无法连接到Redis服务器")
        print("请确保Redis服务正在运行:")
        print("  brew services start redis  # macOS")
        print("  或 redis-server            # 直接启动")
        return False
    except Exception as e:
        print(f"❌ 检查失败: {e}")
        return False

def test_token_operations():
    """测试token操作"""
    try:
        r = redis.Redis(**REDIS_CONFIG)
        
        print("\n=== Token操作测试 ===")
        
        # 模拟创建token
        test_key = "token:999:test_token_12345"
        r.setex(test_key, 60, "active")
        print(f"✅ 创建测试token: {test_key}")
        
        # 检查token存在
        if r.exists(test_key):
            print("✅ Token存在性验证通过")
        
        # 检查TTL
        ttl = r.ttl(test_key)
        print(f"✅ Token TTL: {ttl}秒")
        
        # 删除测试token
        r.delete(test_key)
        print("✅ 清理测试token")
        
        return True
        
    except Exception as e:
        print(f"❌ Token操作测试失败: {e}")
        return False

def test_jwt_service():
    """测试JWT服务"""
    try:
        from app.services.jwt_service import jwt_service
        
        print("\n=== JWT服务测试 ===")
        
        # 测试创建token
        test_data = {
            "user_id": 999,
            "username": "test_user",
            "caption": "测试用户"
        }
        
        token = jwt_service.create_access_token(test_data)
        print(f"✅ 创建JWT token: {token[:30]}...")
        
        # 测试验证token
        payload = jwt_service.verify_token(token)
        if payload:
            print(f"✅ Token验证成功: 用户ID={payload['user_id']}, 用户名={payload['username']}")
        else:
            print("❌ Token验证失败")
            return False
        
        # 测试撤销token
        revoked = jwt_service.revoke_token(token, test_data["user_id"])
        print(f"✅ 撤销token: {'成功' if revoked else '失败'}")
        
        # 验证撤销后token失效
        payload_after_revoke = jwt_service.verify_token(token)
        if not payload_after_revoke:
            print("✅ Token撤销后验证失败，符合预期")
        else:
            print("❌ Token撤销后仍然有效，不符合预期")
            return False
        
        return True
        
    except Exception as e:
        print(f"❌ JWT服务测试失败: {e}")
        return False

if __name__ == "__main__":
    print("JWT + Redis 认证系统测试")
    print("=" * 50)
    print(f"工作目录: {os.getcwd()}")
    print(f"测试脚本路径: {__file__}")
    
    # 检查Redis中的JWT tokens
    redis_ok = check_jwt_in_redis()
    
    # 测试token操作
    operations_ok = test_token_operations()
    
    # 测试JWT服务
    jwt_service_ok = test_jwt_service()
    
    print("=" * 50)
    if redis_ok and operations_ok and jwt_service_ok:
        print("✅ JWT + Redis 认证系统工作正常!")
    else:
        print("❌ JWT + Redis 认证系统存在问题")
        print("\n故障排除建议:")
        print("1. 确保在 backend 目录下运行测试")
        print("2. 确保Redis服务正在运行")
        print("3. 确保已安装所有依赖: pip install -r requirements.txt")
        print("4. 尝试先通过前端登录生成token")
        sys.exit(1)