LQAdminPlatform/test_sliding_token.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
测试滑动过期Token机制
"""

import sys
import os
import asyncio
from datetime import datetime, timedelta, timezone

# 添加src目录到Python路径
sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'src'))

from app.services.jwt_token import (
    create_access_token, 
    verify_token, 
    verify_and_refresh_token,
    refresh_token_sliding_expiration
)
from app.core.config import config_handler

def test_token_creation():
    """测试token创建"""
    print("🧪 测试1: Token创建")
    
    # 创建测试数据
    test_data = {
        "sub": "test_user_123",
        "username": "testuser",
        "email": "test@example.com",
        "is_superuser": False
    }
    
    # 创建token
    token = create_access_token(test_data)
    print(f"✅ Token创建成功: {token[:50]}...")
    
    # 验证token
    payload = verify_token(token)
    if payload:
        print(f"✅ Token验证成功: 用户={payload.get('username')}")
        print(f"   过期时间: {datetime.fromtimestamp(payload.get('exp'), tz=timezone.utc)}")
    else:
        print("❌ Token验证失败")
    
    return token

def test_sliding_expiration(token):
    """测试滑动过期机制"""
    print("\n🧪 测试2: 滑动过期机制")
    
    # 第一次验证（应该不需要刷新）
    payload, new_token = verify_and_refresh_token(token)
    if payload:
        print(f"✅ 第一次验证成功: 用户={payload.get('username')}")
        if new_token:
            print(f"🔄 Token被刷新: {new_token[:50]}...")
            return new_token
        else:
            print("ℹ️  Token无需刷新")
            return token
    else:
        print("❌ Token验证失败")
        return None

def test_token_refresh(token):
    """测试token刷新"""
    print("\n🧪 测试3: 手动Token刷新")
    
    new_token = refresh_token_sliding_expiration(token)
    if new_token:
        print(f"✅ Token刷新成功: {new_token[:50]}...")
        
        # 验证新token
        payload = verify_token(new_token)
        if payload:
            print(f"✅ 新Token验证成功: 用户={payload.get('username')}")
            print(f"   新过期时间: {datetime.fromtimestamp(payload.get('exp'), tz=timezone.utc)}")
        else:
            print("❌ 新Token验证失败")
        
        return new_token
    else:
        print("❌ Token刷新失败")
        return None

def test_config_loading():
    """测试配置加载"""
    print("\n🧪 测试4: 配置加载")
    
    admin_expire = config_handler.get_int("admin_app", "ADMIN_TOKEN_EXPIRE_MINUTES", None)
    access_expire = config_handler.get_int("admin_app", "ACCESS_TOKEN_EXPIRE_MINUTES", 30)
    refresh_expire = config_handler.get_int("admin_app", "ADMIN_REFRESH_TOKEN_EXPIRE_HOURS", None)
    
    print(f"✅ 后台管理Token过期时间: {admin_expire} 分钟")
    print(f"✅ 通用Token过期时间: {access_expire} 分钟")
    print(f"✅ 刷新Token过期时间: {refresh_expire} 小时")
    
    # 显示实际使用的过期时间
    actual_expire = admin_expire if admin_expire is not None else access_expire
    print(f"🎯 实际使用的过期时间: {actual_expire} 分钟")

def simulate_user_activity():
    """模拟用户活动场景"""
    print("\n🧪 测试5: 模拟用户活动场景")
    
    # 创建初始token
    test_data = {
        "sub": "active_user_456",
        "username": "activeuser",
        "email": "active@example.com",
        "is_superuser": True
    }
    
    current_token = create_access_token(test_data)
    print(f"📅 初始Token创建: {datetime.now()}")
    
    # 模拟多次API调用
    for i in range(3):
        print(f"\n🔄 第{i+1}次API调用:")
        payload, new_token = verify_and_refresh_token(current_token)
        
        if payload:
            print(f"   ✅ 验证成功: {payload.get('username')}")
            if new_token:
                print(f"   🔄 Token已刷新")
                current_token = new_token
            else:
                print(f"   ℹ️  Token无需刷新")
        else:
            print(f"   ❌ 验证失败")
            break
        
        # 模拟时间间隔
        import time
        time.sleep(1)
    
    return current_token

def main():
    """主测试函数"""
    print("🚀 开始测试滑动过期Token机制")
    print("=" * 60)
    
    try:
        # 测试配置加载
        test_config_loading()
        
        # 测试token创建
        token = test_token_creation()
        if not token:
            print("❌ Token创建失败，终止测试")
            return
        
        # 测试滑动过期
        token = test_sliding_expiration(token)
        if not token:
            print("❌ 滑动过期测试失败，终止测试")
            return
        
        # 测试token刷新
        new_token = test_token_refresh(token)
        if not new_token:
            print("❌ Token刷新测试失败")
        
        # 模拟用户活动
        simulate_user_activity()
        
        print("\n" + "=" * 60)
        print("🎉 所有测试完成！")
        
    except Exception as e:
        print(f"\n❌ 测试过程中发生错误: {e}")
        import traceback
        traceback.print_exc()

if __name__ == "__main__":
    main()