comi-logic/mxivideo
7
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix

Browse Source
This commit is contained in:
root
2025-07-12 18:19:55 +08:00
parent 97d41587ba
commit 050f9820b8
7 changed files with 662 additions and 362 deletions
Download Patch File Download Diff File Expand all files Collapse all files

68
python_core/cli/commands/auth.py
View File

@@ -24,7 +24,8 @@ def register_user(
email: str = typer.Argument(..., help="邮箱地址"),
display_name: Optional[str] = typer.Option(None, "--display-name", "-d", help="显示名称"),
password: Optional[str] = typer.Option(None, "--password", "-p", help="密码（不提供则交互式输入）"),
verbose: bool = typer.Option(False, "--verbose", "-v", help="详细输出")
verbose: bool = typer.Option(False, "--verbose", "-v", help="详细输出"),
json_output: bool = typer.Option(False, "--json", help="JSON格式输出")
):
"""注册新用户"""
@@ -51,13 +52,17 @@ def register_user(
})
if result["success"]:
console.print(f"\n✅ [bold green]注册成功![/bold green]")
console.print(f"用户ID: {result['data']['user']['id']}")
console.print(f"创建时间: {result['data']['user']['created_at']}")
if verbose:
# 显示详细信息
user_info = f"""
if json_output:
# JSON格式输出
print(json.dumps(result, ensure_ascii=False, indent=2))
else:
console.print(f"\n✅ [bold green]注册成功![/bold green]")
console.print(f"用户ID: {result['data']['user']['id']}")
console.print(f"创建时间: {result['data']['user']['created_at']}")
if verbose:
# 显示详细信息
user_info = f"""
🆔 用户ID: {result['data']['user']['id']}
👤 用户名: {result['data']['user']['username']}
📧 邮箱: {result['data']['user']['email']}
@@ -65,12 +70,15 @@ def register_user(
📅 创建时间: {result['data']['user']['created_at']}
🔑 Token: {result['data']['token'][:50]}...
⏰ 过期时间: {result['data']['expires_at']}
"""
panel = Panel(user_info.strip(), title="用户详情", border_style="green")
console.print(panel)
"""
panel = Panel(user_info.strip(), title="用户详情", border_style="green")
console.print(panel)
else:
console.print(f"[red]❌ 注册失败: {result['message']}[/red]")
if json_output:
print(json.dumps(result, ensure_ascii=False, indent=2))
else:
console.print(f"[red]❌ 注册失败: {result['message']}[/red]")
raise typer.Exit(1)
except Exception as e:
@@ -82,7 +90,8 @@ def register_user(
def login_user(
username_or_email: str = typer.Argument(..., help="用户名或邮箱"),
password: Optional[str] = typer.Option(None, "--password", "-p", help="密码（不提供则交互式输入）"),
verbose: bool = typer.Option(False, "--verbose", "-v", help="详细输出")
verbose: bool = typer.Option(False, "--verbose", "-v", help="详细输出"),
json_output: bool = typer.Option(False, "--json", help="JSON格式输出")
):
"""用户登录"""
@@ -104,13 +113,17 @@ def login_user(
})
if result["success"]:
console.print(f"\n✅ [bold green]登录成功![/bold green]")
console.print(f"欢迎回来，{result['data']['user']['display_name']}!")
console.print(f"最后登录: {result['data']['user']['last_login']}")
if verbose:
# 显示详细信息
user_info = f"""
if json_output:
# JSON格式输出
print(json.dumps(result, ensure_ascii=False, indent=2))
else:
console.print(f"\n✅ [bold green]登录成功![/bold green]")
console.print(f"欢迎回来，{result['data']['user']['display_name']}!")
console.print(f"最后登录: {result['data']['user']['last_login']}")
if verbose:
# 显示详细信息
user_info = f"""
🆔 用户ID: {result['data']['user']['id']}
👤 用户名: {result['data']['user']['username']}
📧 邮箱: {result['data']['user']['email']}
@@ -118,12 +131,15 @@ def login_user(
🕒 最后登录: {result['data']['user']['last_login']}
🔑 Token: {result['data']['token'][:50]}...
⏰ 过期时间: {result['data']['expires_at']}
"""
panel = Panel(user_info.strip(), title="登录信息", border_style="green")
console.print(panel)
"""
panel = Panel(user_info.strip(), title="登录信息", border_style="green")
console.print(panel)
else:
console.print(f"[red]❌ 登录失败: {result['message']}[/red]")
if json_output:
print(json.dumps(result, ensure_ascii=False, indent=2))
else:
console.print(f"[red]❌ 登录失败: {result['message']}[/red]")
raise typer.Exit(1)
except Exception as e:

2
src-tauri/src/commands/mod.rs
View File

@@ -13,6 +13,7 @@ pub mod resource_category;
pub mod python_core_demo;
pub mod python_env_manager;
pub mod mix_edit;
pub mod python_cli;
// Re-export all commands for easy access
pub use basic::*;
@@ -23,3 +24,4 @@ pub use project::*;
pub use template::*;
pub use resource_category::*;
pub use python_core_demo::*;
pub use python_cli::*;

278
src-tauri/src/commands/python_cli.rs Normal file
View File

@@ -0,0 +1,278 @@
/**
* Python CLI 命令封装
* 将 python -m python_core.cli 封装成 Rust 命令
*/
use serde::{Deserialize, Serialize};
use tauri::{command, AppHandle};
use crate::python_executor::{execute_python_command, PythonExecutorConfig};
#[derive(Debug, Serialize, Deserialize)]
pub struct PythonCliResponse {
pub success: bool,
pub message: String,
pub data: Option<serde_json::Value>,
pub output: Option<String>,
pub error: Option<String>,
}
#[derive(Debug, Serialize, Deserialize)]
pub struct AuthLoginRequest {
pub username_or_email: String,
pub password: Option<String>,
pub verbose: Option<bool>,
pub json_output: Option<bool>,
}
#[derive(Debug, Serialize, Deserialize)]
pub struct AuthRegisterRequest {
pub username: String,
pub email: String,
pub display_name: Option<String>,
pub password: Option<String>,
pub verbose: Option<bool>,
pub json_output: Option<bool>,
}
#[derive(Debug, Serialize, Deserialize)]
pub struct AuthVerifyRequest {
pub token: String,
pub verbose: Option<bool>,
}
/// 执行Python CLI命令的通用函数
async fn execute_python_cli_command(
app: AppHandle,
command_args: Vec<String>,
config: Option<PythonExecutorConfig>,
) -> Result<PythonCliResponse, String> {
// 构建完整的命令参数
let mut args = vec!["-m".to_string(), "python_core.cli".to_string()];
args.extend(command_args);
println!("Executing Python CLI: python {}", args.join(" "));
match execute_python_command(app, &args, config).await {
Ok(output) => {
// 尝试解析JSON响应
if let Ok(json_value) = serde_json::from_str::<serde_json::Value>(&output) {
Ok(PythonCliResponse {
success: true,
message: "命令执行成功".to_string(),
data: Some(json_value),
output: Some(output),
error: None,
})
} else {
// 如果不是JSON，直接返回文本输出
Ok(PythonCliResponse {
success: true,
message: "命令执行成功".to_string(),
data: None,
output: Some(output),
error: None,
})
}
}
Err(error) => {
Ok(PythonCliResponse {
success: false,
message: "命令执行失败".to_string(),
data: None,
output: None,
error: Some(error),
})
}
}
}
/// 用户登录
#[command]
pub async fn python_cli_auth_login(
app: AppHandle,
request: AuthLoginRequest,
) -> Result<PythonCliResponse, String> {
let mut args = vec!["auth".to_string(), "login".to_string(), request.username_or_email];
if let Some(password) = request.password {
args.push("--password".to_string());
args.push(password);
}
if request.verbose.unwrap_or(false) {
args.push("--verbose".to_string());
}
if request.json_output.unwrap_or(false) {
args.push("--json".to_string());
}
execute_python_cli_command(app, args, None).await
}
/// 用户注册
#[command]
pub async fn python_cli_auth_register(
app: AppHandle,
request: AuthRegisterRequest,
) -> Result<PythonCliResponse, String> {
let mut args = vec!["auth".to_string(), "register".to_string(), request.username, request.email];
if let Some(display_name) = request.display_name {
args.push("--display-name".to_string());
args.push(display_name);
}
if let Some(password) = request.password {
args.push("--password".to_string());
args.push(password);
}
if request.verbose.unwrap_or(false) {
args.push("--verbose".to_string());
}
if request.json_output.unwrap_or(false) {
args.push("--json".to_string());
}
execute_python_cli_command(app, args, None).await
}
/// 验证Token
#[command]
pub async fn python_cli_auth_verify(
app: AppHandle,
request: AuthVerifyRequest,
) -> Result<PythonCliResponse, String> {
let mut args = vec!["auth".to_string(), "verify".to_string(), request.token];
if request.verbose.unwrap_or(false) {
args.push("--verbose".to_string());
}
execute_python_cli_command(app, args, None).await
}
/// 获取用户列表
#[command]
pub async fn python_cli_auth_list(
app: AppHandle,
include_inactive: Option<bool>,
limit: Option<i32>,
verbose: Option<bool>,
) -> Result<PythonCliResponse, String> {
let mut args = vec!["auth".to_string(), "list".to_string()];
if include_inactive.unwrap_or(false) {
args.push("--include-inactive".to_string());
}
if let Some(limit_val) = limit {
args.push("--limit".to_string());
args.push(limit_val.to_string());
}
if verbose.unwrap_or(false) {
args.push("--verbose".to_string());
}
execute_python_cli_command(app, args, None).await
}
/// 获取用户统计信息
#[command]
pub async fn python_cli_auth_stats(app: AppHandle) -> Result<PythonCliResponse, String> {
let args = vec!["auth".to_string(), "stats".to_string()];
execute_python_cli_command(app, args, None).await
}
/// 获取模板列表
#[command]
pub async fn python_cli_template_list(
app: AppHandle,
limit: Option<i32>,
verbose: Option<bool>,
) -> Result<PythonCliResponse, String> {
let mut args = vec!["template".to_string(), "list".to_string()];
if let Some(limit_val) = limit {
args.push("--limit".to_string());
args.push(limit_val.to_string());
}
if verbose.unwrap_or(false) {
args.push("--verbose".to_string());
}
execute_python_cli_command(app, args, None).await
}
/// 批量导入模板
#[command]
pub async fn python_cli_template_batch_import(
app: AppHandle,
folder_path: String,
verbose: Option<bool>,
) -> Result<PythonCliResponse, String> {
let mut args = vec!["template".to_string(), "batch-import".to_string(), folder_path];
if verbose.unwrap_or(false) {
args.push("--verbose".to_string());
}
execute_python_cli_command(app, args, None).await
}
/// 获取分类列表
#[command]
pub async fn python_cli_category_list(
app: AppHandle,
verbose: Option<bool>,
) -> Result<PythonCliResponse, String> {
let mut args = vec!["category".to_string(), "list".to_string()];
if verbose.unwrap_or(false) {
args.push("--verbose".to_string());
}
execute_python_cli_command(app, args, None).await
}
/// 场景检测
#[command]
pub async fn python_cli_scene_detect(
app: AppHandle,
video_path: String,
threshold: Option<f64>,
min_scene_len: Option<f64>,
output_dir: Option<String>,
) -> Result<PythonCliResponse, String> {
let mut args = vec!["scene".to_string(), "detect".to_string(), video_path];
if let Some(threshold_val) = threshold {
args.push("--threshold".to_string());
args.push(threshold_val.to_string());
}
if let Some(min_len) = min_scene_len {
args.push("--min-scene-len".to_string());
args.push(min_len.to_string());
}
if let Some(output) = output_dir {
args.push("--output".to_string());
args.push(output);
}
execute_python_cli_command(app, args, None).await
}
/// 执行任意Python CLI命令
#[command]
pub async fn python_cli_execute(
app: AppHandle,
command_args: Vec<String>,
) -> Result<PythonCliResponse, String> {
execute_python_cli_command(app, command_args, None).await
}

13
src-tauri/src/lib.rs
View File

@@ -109,7 +109,18 @@ pub fn run() {
commands::python_env_manager::setup_embedded_python,
commands::python_env_manager::setup_embedded_python_simple,
// Mix Edit Commands
commands::mix_edit::start_mix_edit
commands::mix_edit::start_mix_edit,
// Python CLI Commands
commands::python_cli::python_cli_auth_login,
commands::python_cli::python_cli_auth_register,
commands::python_cli::python_cli_auth_verify,
commands::python_cli::python_cli_auth_list,
commands::python_cli::python_cli_auth_stats,
commands::python_cli::python_cli_template_list,
commands::python_cli::python_cli_template_batch_import,
commands::python_cli::python_cli_category_list,
commands::python_cli::python_cli_scene_detect,
commands::python_cli::python_cli_execute
])
.run(tauri::generate_context!())
.expect("error while running tauri application");

309
src/services/pythonCliAuth.ts Normal file
View File

@@ -0,0 +1,309 @@
/**
* Python CLI 认证服务
* 通过Rust调用Python CLI实现用户认证功能
*/
import { invoke } from '@tauri-apps/api/core';
export interface LoginCredentials {
email?: string;
username?: string;
password: string;
}
export interface RegisterCredentials {
email: string;
username: string;
password: string;
displayName?: string;
}
export interface UserProfile {
id: string;
username: string;
displayName: string;
email?: string;
avatarUrl?: string;
createTime: string;
updateTime: string;
lastLogin?: string;
isActive: boolean;
}
export interface AuthResponse {
success: boolean;
message: string;
data?: any;
output?: string;
error?: string;
}
export interface PythonCliAuthResponse {
success: boolean;
message: string;
data?: {
user?: UserProfile;
token?: string;
expires_at?: string;
};
output?: string;
error?: string;
}
class PythonCliAuth {
private token: string | null = null;
private user: UserProfile | null = null;
constructor() {
// 从localStorage恢复认证状态
this.loadStoredAuth();
}
/**
* 用户登录
*/
async login(credentials: LoginCredentials): Promise<UserProfile> {
try {
const username_or_email = credentials.email || credentials.username;
if (!username_or_email) {
throw new Error('Email or username is required');
}
const response: PythonCliAuthResponse = await invoke('python_cli_auth_login', {
request: {
username_or_email,
password: credentials.password,
json_output: true
}
});
if (!response.success) {
throw new Error(response.error || response.message || 'Login failed');
}
// 解析输出中的用户信息和token
if (response.success && response.output) {
// 尝试从输出中解析JSON
const jsonMatch = response.output.match(/\{[\s\S]*\}/);
if (jsonMatch) {
try {
const authData = JSON.parse(jsonMatch[0]);
if (authData.success && authData.data?.user && authData.data?.token) {
this.token = authData.data.token;
this.user = {
id: authData.data.user.id,
username: authData.data.user.username,
displayName: authData.data.user.display_name,
email: authData.data.user.email,
avatarUrl: authData.data.user.avatar_url,
createTime: authData.data.user.created_at,
updateTime: authData.data.user.updated_at,
lastLogin: authData.data.user.last_login,
isActive: authData.data.user.is_active
};
// 保存到localStorage
this.saveAuthToStorage();
return this.user;
}
} catch (parseError) {
console.error('Failed to parse auth response JSON:', parseError);
}
}
}
throw new Error(response.error || response.message || 'Login failed');
} catch (error) {
console.error('Login failed:', error);
throw error;
}
}
/**
* 用户注册
*/
async register(credentials: RegisterCredentials): Promise<UserProfile> {
try {
const response: PythonCliAuthResponse = await invoke('python_cli_auth_register', {
request: {
username: credentials.username,
email: credentials.email,
password: credentials.password,
display_name: credentials.displayName,
json_output: true
}
});
if (!response.success) {
throw new Error(response.error || response.message || 'Registration failed');
}
// 解析输出中的用户信息和token
if (response.success && response.output) {
// 尝试从输出中解析JSON
const jsonMatch = response.output.match(/\{[\s\S]*\}/);
if (jsonMatch) {
try {
const authData = JSON.parse(jsonMatch[0]);
if (authData.success && authData.data?.user && authData.data?.token) {
this.token = authData.data.token;
this.user = {
id: authData.data.user.id,
username: authData.data.user.username,
displayName: authData.data.user.display_name,
email: authData.data.user.email,
avatarUrl: authData.data.user.avatar_url,
createTime: authData.data.user.created_at,
updateTime: authData.data.user.updated_at,
lastLogin: authData.data.user.last_login,
isActive: authData.data.user.is_active
};
// 保存到localStorage
this.saveAuthToStorage();
return this.user;
}
} catch (parseError) {
console.error('Failed to parse auth response JSON:', parseError);
}
}
}
throw new Error(response.error || response.message || 'Registration failed');
} catch (error) {
console.error('Registration failed:', error);
throw error;
}
}
/**
* 验证Token
*/
async verifyToken(token?: string): Promise<boolean> {
try {
const tokenToVerify = token || this.token;
if (!tokenToVerify) {
return false;
}
const response: PythonCliAuthResponse = await invoke('python_cli_auth_verify', {
request: {
token: tokenToVerify,
verbose: true
}
});
return response.success;
} catch (error) {
console.error('Token verification failed:', error);
return false;
}
}
/**
* 用户登出
*/
async logout(): Promise<void> {
this.token = null;
this.user = null;
this.clearStoredAuth();
}
/**
* 获取当前用户信息
*/
async getCurrentUser(): Promise<UserProfile | null> {
if (!this.token) {
return null;
}
try {
// 验证当前token是否有效
const isValid = await this.verifyToken();
if (!isValid) {
this.logout();
return null;
}
return this.user;
} catch (error) {
console.error('Failed to get current user:', error);
this.logout();
return null;
}
}
/**
* 更新用户资料
*/
async updateProfile(updates: Partial<Pick<UserProfile, 'displayName' | 'avatarUrl'>>): Promise<void> {
// TODO: 实现用户资料更新
console.log('Update profile:', updates);
throw new Error('Profile update not implemented yet');
}
/**
* 检查是否已认证
*/
isAuthenticated(): boolean {
return this.token !== null && this.user !== null;
}
/**
* 获取当前token
*/
getToken(): string | null {
return this.token;
}
/**
* 获取当前用户
*/
getUser(): UserProfile | null {
return this.user;
}
/**
* 保存认证信息到localStorage
*/
private saveAuthToStorage(): void {
if (this.token && this.user) {
localStorage.setItem('auth_token', this.token);
localStorage.setItem('auth_user', JSON.stringify(this.user));
}
}
/**
* 从localStorage加载认证信息
*/
private loadStoredAuth(): void {
try {
const token = localStorage.getItem('auth_token');
const userStr = localStorage.getItem('auth_user');
if (token && userStr) {
this.token = token;
this.user = JSON.parse(userStr);
}
} catch (error) {
console.error('Failed to load stored auth:', error);
this.clearStoredAuth();
}
}
/**
* 清除存储的认证信息
*/
private clearStoredAuth(): void {
localStorage.removeItem('auth_token');
localStorage.removeItem('auth_user');
}
}
// 创建全局实例
export const pythonCliAuth = new PythonCliAuth();
// 导出默认实例
export default pythonCliAuth;

38
src/stores/useAuthStore.ts
View File

@@ -5,7 +5,7 @@
import { create } from 'zustand';
import { persist } from 'zustand/middleware';
import { nakamaAuth, UserProfile, LoginCredentials, RegisterCredentials } from '../services/nakamaAuth';
import { pythonCliAuth, UserProfile, LoginCredentials, RegisterCredentials } from '../services/pythonCliAuth';
interface AuthState {
// 状态
@@ -38,16 +38,16 @@ export const useAuthStore = create<AuthState>()(
set({ loading: true, error: null });
try {
const user = await nakamaAuth.login(credentials);
set({
isAuthenticated: true,
user,
const user = await pythonCliAuth.login(credentials);
set({
isAuthenticated: true,
user,
loading: false,
error: null
error: null
});
} catch (error) {
set({
loading: false,
set({
loading: false,
error: error instanceof Error ? error.message : '登录失败',
isAuthenticated: false,
user: null
@@ -61,16 +61,16 @@ export const useAuthStore = create<AuthState>()(
set({ loading: true, error: null });
try {
const user = await nakamaAuth.register(credentials);
set({
isAuthenticated: true,
user,
const user = await pythonCliAuth.register(credentials);
set({
isAuthenticated: true,
user,
loading: false,
error: null
error: null
});
} catch (error) {
set({
loading: false,
set({
loading: false,
error: error instanceof Error ? error.message : '注册失败',
isAuthenticated: false,
user: null
@@ -84,7 +84,7 @@ export const useAuthStore = create<AuthState>()(
set({ loading: true });
try {
await nakamaAuth.logout();
await pythonCliAuth.logout();
set({
isAuthenticated: false,
user: null,
@@ -108,7 +108,7 @@ export const useAuthStore = create<AuthState>()(
set({ loading: true, error: null });
try {
const user = await nakamaAuth.getCurrentUser();
const user = await pythonCliAuth.getCurrentUser();
if (user) {
set({
isAuthenticated: true,
@@ -137,7 +137,7 @@ export const useAuthStore = create<AuthState>()(
set({ loading: true, error: null });
try {
await nakamaAuth.updateProfile(updates);
await pythonCliAuth.updateProfile(updates);
// 更新本地用户信息
const currentUser = get().user;
@@ -181,7 +181,7 @@ export const initializeAuth = async () => {
const { getCurrentUser } = useAuthStore.getState();
// 检查是否有有效的会话
if (nakamaAuth.isAuthenticated()) {
if (pythonCliAuth.isAuthenticated()) {
await getCurrentUser();
} else {
// 清除过期的状态

316
test_auth_system.py
View File

@@ -1,316 +0,0 @@
#!/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()