317 lines
9.4 KiB
Python
317 lines
9.4 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
测试Python用户认证系统
|
||
"""
|
||
|
||
import sys
|
||
import os
|
||
sys.path.insert(0, '/root/projects/mixvideo_v2')
|
||
|
||
from python_core.api.auth_api import auth_api
|
||
from python_core.utils.jwt_auth import jwt_auth
|
||
from python_core.services.user_storage import user_storage
|
||
|
||
|
||
def test_user_registration():
|
||
"""测试用户注册"""
|
||
print("🧪 测试用户注册...")
|
||
|
||
try:
|
||
# 注册测试用户
|
||
result = auth_api.register({
|
||
"username": "testuser",
|
||
"email": "test@example.com",
|
||
"password": "password123",
|
||
"display_name": "测试用户"
|
||
})
|
||
|
||
if result["success"]:
|
||
print(f"✅ 注册成功: {result['data']['user']['username']}")
|
||
print(f" 用户ID: {result['data']['user']['id']}")
|
||
print(f" Token: {result['data']['token'][:50]}...")
|
||
return result['data']['token']
|
||
else:
|
||
print(f"❌ 注册失败: {result['message']}")
|
||
return None
|
||
|
||
except Exception as e:
|
||
print(f"❌ 注册测试失败: {e}")
|
||
return None
|
||
|
||
|
||
def test_user_login():
|
||
"""测试用户登录"""
|
||
print("\n🧪 测试用户登录...")
|
||
|
||
try:
|
||
# 登录测试用户
|
||
result = auth_api.login({
|
||
"username_or_email": "testuser",
|
||
"password": "password123"
|
||
})
|
||
|
||
if result["success"]:
|
||
print(f"✅ 登录成功: {result['data']['user']['username']}")
|
||
print(f" 显示名称: {result['data']['user']['display_name']}")
|
||
print(f" Token: {result['data']['token'][:50]}...")
|
||
return result['data']['token']
|
||
else:
|
||
print(f"❌ 登录失败: {result['message']}")
|
||
return None
|
||
|
||
except Exception as e:
|
||
print(f"❌ 登录测试失败: {e}")
|
||
return None
|
||
|
||
|
||
def test_token_verification(token):
|
||
"""测试Token验证"""
|
||
print("\n🧪 测试Token验证...")
|
||
|
||
try:
|
||
# 验证token
|
||
result = auth_api.verify_token({
|
||
"token": token
|
||
})
|
||
|
||
if result["success"]:
|
||
user = result['data']['user']
|
||
print(f"✅ Token验证成功: {user['username']}")
|
||
print(f" 用户ID: {user['user_id']}")
|
||
print(f" 邮箱: {user['email']}")
|
||
|
||
# 获取token详细信息
|
||
token_info = jwt_auth.get_token_info(token)
|
||
print(f" Token有效性: {token_info['valid']}")
|
||
print(f" 剩余时间: {token_info['time_remaining']}")
|
||
|
||
return True
|
||
else:
|
||
print(f"❌ Token验证失败: {result['message']}")
|
||
return False
|
||
|
||
except Exception as e:
|
||
print(f"❌ Token验证测试失败: {e}")
|
||
return False
|
||
|
||
|
||
def test_get_current_user(token):
|
||
"""测试获取当前用户"""
|
||
print("\n🧪 测试获取当前用户...")
|
||
|
||
try:
|
||
# 获取当前用户
|
||
result = auth_api.get_current_user({
|
||
"token": token
|
||
})
|
||
|
||
if result["success"]:
|
||
user = result['data']['user']
|
||
print(f"✅ 获取用户成功: {user['username']}")
|
||
print(f" 显示名称: {user['display_name']}")
|
||
print(f" 创建时间: {user['created_at']}")
|
||
print(f" 最后登录: {user['last_login']}")
|
||
return True
|
||
else:
|
||
print(f"❌ 获取用户失败: {result['message']}")
|
||
return False
|
||
|
||
except Exception as e:
|
||
print(f"❌ 获取用户测试失败: {e}")
|
||
return False
|
||
|
||
|
||
def test_user_storage():
|
||
"""测试用户存储"""
|
||
print("\n🧪 测试用户存储...")
|
||
|
||
try:
|
||
# 获取所有用户
|
||
users = user_storage.get_all_users()
|
||
print(f"✅ 获取到 {len(users)} 个用户")
|
||
|
||
# 获取用户统计
|
||
stats = user_storage.get_user_count()
|
||
print(f" 总用户数: {stats['total']}")
|
||
print(f" 活跃用户: {stats['active']}")
|
||
print(f" 禁用用户: {stats['inactive']}")
|
||
|
||
# 搜索用户
|
||
search_results = user_storage.search_users("test")
|
||
print(f" 搜索'test'找到 {len(search_results)} 个用户")
|
||
|
||
return True
|
||
|
||
except Exception as e:
|
||
print(f"❌ 用户存储测试失败: {e}")
|
||
return False
|
||
|
||
|
||
def test_duplicate_registration():
|
||
"""测试重复注册"""
|
||
print("\n🧪 测试重复注册...")
|
||
|
||
try:
|
||
# 尝试重复注册
|
||
result = auth_api.register({
|
||
"username": "testuser",
|
||
"email": "test@example.com",
|
||
"password": "password123",
|
||
"display_name": "重复用户"
|
||
})
|
||
|
||
if not result["success"]:
|
||
print(f"✅ 正确阻止重复注册: {result['message']}")
|
||
return True
|
||
else:
|
||
print(f"❌ 未能阻止重复注册")
|
||
return False
|
||
|
||
except Exception as e:
|
||
print(f"❌ 重复注册测试失败: {e}")
|
||
return False
|
||
|
||
|
||
def test_invalid_login():
|
||
"""测试无效登录"""
|
||
print("\n🧪 测试无效登录...")
|
||
|
||
try:
|
||
# 尝试错误密码登录
|
||
result = auth_api.login({
|
||
"username_or_email": "testuser",
|
||
"password": "wrongpassword"
|
||
})
|
||
|
||
if not result["success"]:
|
||
print(f"✅ 正确拒绝错误密码: {result['message']}")
|
||
else:
|
||
print(f"❌ 未能拒绝错误密码")
|
||
return False
|
||
|
||
# 尝试不存在的用户登录
|
||
result = auth_api.login({
|
||
"username_or_email": "nonexistentuser",
|
||
"password": "password123"
|
||
})
|
||
|
||
if not result["success"]:
|
||
print(f"✅ 正确拒绝不存在用户: {result['message']}")
|
||
return True
|
||
else:
|
||
print(f"❌ 未能拒绝不存在用户")
|
||
return False
|
||
|
||
except Exception as e:
|
||
print(f"❌ 无效登录测试失败: {e}")
|
||
return False
|
||
|
||
|
||
def test_jwt_features():
|
||
"""测试JWT特性"""
|
||
print("\n🧪 测试JWT特性...")
|
||
|
||
try:
|
||
# 生成token
|
||
token_info = jwt_auth.generate_token("test_user_id", "testuser", "test@example.com")
|
||
token = token_info["token"]
|
||
print(f"✅ 生成Token成功")
|
||
print(f" 过期时间: {token_info['expires_at']}")
|
||
print(f" 有效期: {token_info['expires_in']} 秒")
|
||
|
||
# 验证token
|
||
payload = jwt_auth.verify_token(token)
|
||
if payload:
|
||
print(f"✅ Token验证成功")
|
||
print(f" 用户ID: {payload['user_id']}")
|
||
print(f" 用户名: {payload['username']}")
|
||
print(f" 签发者: {payload['iss']}")
|
||
else:
|
||
print(f"❌ Token验证失败")
|
||
return False
|
||
|
||
# 获取token信息
|
||
info = jwt_auth.get_token_info(token)
|
||
print(f"✅ Token信息获取成功")
|
||
print(f" 有效性: {info['valid']}")
|
||
print(f" 剩余时间: {info['time_remaining']}")
|
||
|
||
return True
|
||
|
||
except Exception as e:
|
||
print(f"❌ JWT特性测试失败: {e}")
|
||
return False
|
||
|
||
|
||
def main():
|
||
"""主测试函数"""
|
||
print("🚀 开始测试Python用户认证系统...")
|
||
|
||
test_results = []
|
||
|
||
# 测试JWT特性
|
||
test_results.append(("JWT特性", test_jwt_features()))
|
||
|
||
# 测试用户注册
|
||
token = test_user_registration()
|
||
test_results.append(("用户注册", token is not None))
|
||
|
||
if token:
|
||
# 测试Token验证
|
||
test_results.append(("Token验证", test_token_verification(token)))
|
||
|
||
# 测试获取当前用户
|
||
test_results.append(("获取当前用户", test_get_current_user(token)))
|
||
|
||
# 测试用户登录
|
||
login_token = test_user_login()
|
||
test_results.append(("用户登录", login_token is not None))
|
||
|
||
# 测试用户存储
|
||
test_results.append(("用户存储", test_user_storage()))
|
||
|
||
# 测试重复注册
|
||
test_results.append(("重复注册检查", test_duplicate_registration()))
|
||
|
||
# 测试无效登录
|
||
test_results.append(("无效登录检查", test_invalid_login()))
|
||
|
||
# 显示测试结果
|
||
print("\n📊 测试结果汇总:")
|
||
print("=" * 50)
|
||
|
||
passed = 0
|
||
total = len(test_results)
|
||
|
||
for test_name, result in test_results:
|
||
status = "✅ 通过" if result else "❌ 失败"
|
||
print(f"{test_name:20} {status}")
|
||
if result:
|
||
passed += 1
|
||
|
||
print("=" * 50)
|
||
print(f"总计: {passed}/{total} 个测试通过")
|
||
|
||
if passed == total:
|
||
print("\n🎉 所有测试通过！Python用户认证系统工作正常！")
|
||
|
||
print("\n📖 使用方法:")
|
||
print(" # 注册用户")
|
||
print(" python3 -m python_core.cli auth register username email@example.com")
|
||
print(" # 登录用户")
|
||
print(" python3 -m python_core.cli auth login username")
|
||
print(" # 验证Token")
|
||
print(" python3 -m python_core.cli auth verify <token>")
|
||
print(" # 查看用户列表")
|
||
print(" python3 -m python_core.cli auth list")
|
||
print(" # 查看统计信息")
|
||
print(" python3 -m python_core.cli auth stats")
|
||
|
||
else:
|
||
print(f"\n❌ {total - passed} 个测试失败！")
|
||
sys.exit(1)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|