taiji-AI-PAD/scripts/test_auth.py

#!/usr/bin/env python3
"""
测试账号注册和登录功能
"""

import requests
import json
from typing import Dict, Any

BASE_URL = "http://localhost:8002"

def print_section(title: str):
    """打印分节标题"""
    print(f"\n{'='*60}")
    print(f"  {title}")
    print(f"{'='*60}\n")

def print_result(title: str, data: Any):
    """打印结果"""
    print(f"✓ {title}")
    print(json.dumps(data, indent=2, ensure_ascii=False))
    print()

def test_health():
    """测试健康检查"""
    print_section("1. 健康检查")
    try:
        response = requests.get(f"{BASE_URL}/health")
        response.raise_for_status()
        print_result("Health Check", response.json())
        return True
    except Exception as e:
        print(f"✗ 健康检查失败: {e}")
        return False

def test_admin_login():
    """测试管理员登录（使用frontend-integration端点）"""
    print_section("2. 测试管理员登录（自动创建）")
    
    # 使用frontend-integration的admin login端点，会自动创建用户
    try:
        response = requests.post(
            f"{BASE_URL}/api/admin/auth/login",
            json={
                "email": "admin@test.com",
                "password": "admin123"
            }
        )
        response.raise_for_status()
        result = response.json()
        print_result("管理员登录成功", result)
        return result.get("token")
    except Exception as e:
        print(f"✗ 管理员登录失败: {e}")
        if hasattr(e, 'response') and e.response:
            print(f"  响应: {e.response.text}")
        return None

def test_channel_login():
    """测试渠道管理员登录"""
    print_section("3. 测试渠道管理员登录（自动创建）")
    
    try:
        response = requests.post(
            f"{BASE_URL}/api/channel/auth/login",
            json={
                "email": "channel@test.com",
                "password": "channel123"
            }
        )
        response.raise_for_status()
        result = response.json()
        print_result("渠道管理员登录成功", result)
        return result.get("token")
    except Exception as e:
        print(f"✗ 渠道管理员登录失败: {e}")
        if hasattr(e, 'response') and e.response:
            print(f"  响应: {e.response.text}")
        return None

def test_provider_login():
    """测试供应商登录"""
    print_section("4. 测试供应商登录（自动创建）")
    
    try:
        response = requests.post(
            f"{BASE_URL}/api/providers/auth/login",
            json={
                "email": "provider@test.com",
                "password": "provider123"
            }
        )
        response.raise_for_status()
        result = response.json()
        print_result("供应商登录成功", result)
        return result.get("token")
    except Exception as e:
        print(f"✗ 供应商登录失败: {e}")
        if hasattr(e, 'response') and e.response:
            print(f"  响应: {e.response.text}")
        return None

def test_user_login():
    """测试普通用户登录"""
    print_section("5. 测试普通用户登录")
    
    # 先尝试使用标准auth端点登录
    try:
        response = requests.post(
            f"{BASE_URL}/api/auth/login",
            json={
                "email": "user@test.com",
                "password": "user123",
                "role": "user"
            }
        )
        
        if response.status_code == 401:
            print("用户不存在，需要先创建")
            return None
        
        response.raise_for_status()
        result = response.json()
        print_result("用户登录成功", result)
        return result.get("data", {}).get("token")
    except Exception as e:
        print(f"✗ 用户登录失败: {e}")
        if hasattr(e, 'response') and e.response:
            print(f"  响应: {e.response.text}")
        return None

def test_authenticated_request(token: str, endpoint: str, title: str):
    """测试需要认证的请求"""
    print_section(f"测试认证请求: {title}")
    
    try:
        response = requests.get(
            f"{BASE_URL}{endpoint}",
            headers={"Authorization": f"Bearer {token}"}
        )
        response.raise_for_status()
        result = response.json()
        print_result(f"{title} - 成功", result)
        return True
    except Exception as e:
        print(f"✗ {title} 失败: {e}")
        if hasattr(e, 'response') and e.response:
            print(f"  响应: {e.response.text}")
        return False

def main():
    """主测试流程"""
    print("\n" + "="*60)
    print("  Taiji AI-PAD 账号注册和登录功能测试")
    print("="*60)
    
    # 1. 健康检查
    if not test_health():
        print("\n服务未就绪，退出测试")
        return
    
    # 2. 测试管理员登录
    admin_token = test_admin_login()
    if admin_token:
        test_authenticated_request(
            admin_token,
            "/api/admin/dashboard/stats",
            "管理员仪表板"
        )
    
    # 3. 测试渠道管理员登录
    channel_token = test_channel_login()
    if channel_token:
        test_authenticated_request(
            channel_token,
            "/api/channel/dashboard/stats",
            "渠道仪表板"
        )
    
    # 4. 测试供应商登录
    provider_token = test_provider_login()
    if provider_token:
        test_authenticated_request(
            provider_token,
            "/api/providers/models",
            "供应商模型列表"
        )
    
    # 5. 测试普通用户登录
    user_token = test_user_login()
    
    # 总结
    print_section("测试总结")
    results = {
        "管理员登录": "✓" if admin_token else "✗",
        "渠道管理员登录": "✓" if channel_token else "✗",
        "供应商登录": "✓" if provider_token else "✗",
        "普通用户登录": "✓" if user_token else "✗ (需要先创建)",
    }
    
    for test_name, status in results.items():
        print(f"{status} {test_name}")
    
    print("\n" + "="*60)
    print("  测试完成")
    print("="*60 + "\n")

if __name__ == "__main__":
    main()