From 050f9820b8e84038781461325875e98c12e5e018 Mon Sep 17 00:00:00 2001 From: root Date: Sat, 12 Jul 2025 18:19:55 +0800 Subject: [PATCH] fix --- python_core/cli/commands/auth.py | 68 +++--- src-tauri/src/commands/mod.rs | 2 + src-tauri/src/commands/python_cli.rs | 278 +++++++++++++++++++++++ src-tauri/src/lib.rs | 13 +- src/services/pythonCliAuth.ts | 309 ++++++++++++++++++++++++++ src/stores/useAuthStore.ts | 38 ++-- test_auth_system.py | 316 --------------------------- 7 files changed, 662 insertions(+), 362 deletions(-) create mode 100644 src-tauri/src/commands/python_cli.rs create mode 100644 src/services/pythonCliAuth.ts delete mode 100644 test_auth_system.py diff --git a/python_core/cli/commands/auth.py b/python_core/cli/commands/auth.py index d433b15..59702a7 100644 --- a/python_core/cli/commands/auth.py +++ b/python_core/cli/commands/auth.py @@ -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: diff --git a/src-tauri/src/commands/mod.rs b/src-tauri/src/commands/mod.rs index 2fa6f30..fab43d7 100644 --- a/src-tauri/src/commands/mod.rs +++ b/src-tauri/src/commands/mod.rs @@ -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::*; diff --git a/src-tauri/src/commands/python_cli.rs b/src-tauri/src/commands/python_cli.rs new file mode 100644 index 0000000..a854aa3 --- /dev/null +++ b/src-tauri/src/commands/python_cli.rs @@ -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, + pub output: Option, + pub error: Option, +} + +#[derive(Debug, Serialize, Deserialize)] +pub struct AuthLoginRequest { + pub username_or_email: String, + pub password: Option, + pub verbose: Option, + pub json_output: Option, +} + +#[derive(Debug, Serialize, Deserialize)] +pub struct AuthRegisterRequest { + pub username: String, + pub email: String, + pub display_name: Option, + pub password: Option, + pub verbose: Option, + pub json_output: Option, +} + +#[derive(Debug, Serialize, Deserialize)] +pub struct AuthVerifyRequest { + pub token: String, + pub verbose: Option, +} + +/// 执行Python CLI命令的通用函数 +async fn execute_python_cli_command( + app: AppHandle, + command_args: Vec, + config: Option, +) -> Result { + // 构建完整的命令参数 + 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::(&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 { + 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 { + 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 { + 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, + limit: Option, + verbose: Option, +) -> Result { + 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 { + 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, + verbose: Option, +) -> Result { + 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, +) -> Result { + 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, +) -> Result { + 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, + min_scene_len: Option, + output_dir: Option, +) -> Result { + 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, +) -> Result { + execute_python_cli_command(app, command_args, None).await +} diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index 8746ae0..333bf14 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -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"); diff --git a/src/services/pythonCliAuth.ts b/src/services/pythonCliAuth.ts new file mode 100644 index 0000000..4f3fd13 --- /dev/null +++ b/src/services/pythonCliAuth.ts @@ -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 { + 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 { + 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 { + 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 { + this.token = null; + this.user = null; + this.clearStoredAuth(); + } + + /** + * 获取当前用户信息 + */ + async getCurrentUser(): Promise { + 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>): Promise { + // 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; diff --git a/src/stores/useAuthStore.ts b/src/stores/useAuthStore.ts index 5864c7c..78cd09a 100644 --- a/src/stores/useAuthStore.ts +++ b/src/stores/useAuthStore.ts @@ -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()( 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()( 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()( set({ loading: true }); try { - await nakamaAuth.logout(); + await pythonCliAuth.logout(); set({ isAuthenticated: false, user: null, @@ -108,7 +108,7 @@ export const useAuthStore = create()( 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()( 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 { // 清除过期的状态 diff --git a/test_auth_system.py b/test_auth_system.py deleted file mode 100644 index f906836..0000000 --- a/test_auth_system.py +++ /dev/null @@ -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 ") - 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()