forked from xiaohei/taiji-AI-PAD
463 lines
15 KiB
Python
463 lines
15 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
完整工作流测试脚本
|
|
测试：超级管理员创建渠道、创建计费/运维管理员、创建租户、权限验证
|
|
"""
|
|
|
|
import requests
|
|
import sys
|
|
import os
|
|
import time
|
|
import subprocess
|
|
from typing import Optional, Dict, Any
|
|
|
|
BASE_URL = "http://localhost:8002"
|
|
|
|
# 测试账户信息
|
|
SUPER_ADMIN = {
|
|
"email": "superadmin@taiji-ai.com",
|
|
"password": "Admin@123456",
|
|
"role": "super_admin"
|
|
}
|
|
|
|
BILLING_ADMIN = {
|
|
"email": "newbilling@test.com",
|
|
"password": "Billing@123456",
|
|
"role": "billing_admin"
|
|
}
|
|
|
|
OPS_ADMIN = {
|
|
"email": "newops@test.com",
|
|
"password": "Ops@123456",
|
|
"role": "operations_admin"
|
|
}
|
|
|
|
CHANNEL_ADMIN = {
|
|
"email": "channel-a@test.com",
|
|
"password": "ChannelA@123456",
|
|
"role": "channel_admin"
|
|
}
|
|
|
|
|
|
def print_header(title: str):
|
|
"""打印标题"""
|
|
print("\n" + "="*80)
|
|
print(f" {title}")
|
|
print("="*80)
|
|
|
|
|
|
def print_step(step: str):
|
|
"""打印步骤"""
|
|
print(f"\n[步骤] {step}")
|
|
print("-" * 80)
|
|
|
|
|
|
def wait_for_service(url: str, max_retries: int = 30, delay: int = 2) -> bool:
|
|
"""等待服务启动"""
|
|
print(f"等待服务启动: {url}")
|
|
for i in range(max_retries):
|
|
try:
|
|
resp = requests.get(f"{url}/health", timeout=2)
|
|
if resp.status_code == 200:
|
|
print(f" ✓ 服务已启动")
|
|
return True
|
|
except:
|
|
pass
|
|
if i < max_retries - 1:
|
|
print(f" 等待中... ({i+1}/{max_retries})")
|
|
time.sleep(delay)
|
|
print(f" ✗ 服务启动超时")
|
|
return False
|
|
|
|
|
|
def login(email: str, password: str, role: str = None) -> Optional[str]:
|
|
"""登录并获取token"""
|
|
try:
|
|
data = {"email": email, "password": password}
|
|
if role:
|
|
data["role"] = role
|
|
|
|
resp = requests.post(
|
|
f"{BASE_URL}/api/auth/login",
|
|
json=data,
|
|
timeout=10
|
|
)
|
|
|
|
if resp.status_code == 200:
|
|
result = resp.json()
|
|
token = result.get("data", {}).get("token")
|
|
if token:
|
|
print(f" ✓ 登录成功: {email}")
|
|
return token
|
|
else:
|
|
print(f" ✗ 登录失败: 响应中未找到token")
|
|
return None
|
|
else:
|
|
error = resp.json().get("detail", resp.text)
|
|
print(f" ✗ 登录失败: {error}")
|
|
return None
|
|
except Exception as e:
|
|
print(f" ✗ 登录出错: {e}")
|
|
return None
|
|
|
|
|
|
def test_create_channel(token: str) -> Optional[str]:
|
|
"""测试创建渠道"""
|
|
print_step("测试：超级管理员创建渠道")
|
|
|
|
channel_data = {
|
|
"name": "测试渠道A",
|
|
"email": "channel-a-test@test.com",
|
|
"password": "Channel@123456",
|
|
"commissionRate": 15.0
|
|
}
|
|
|
|
try:
|
|
resp = requests.post(
|
|
f"{BASE_URL}/api/admin/channels/create",
|
|
headers={
|
|
"Authorization": f"Bearer {token}",
|
|
"Content-Type": "application/json"
|
|
},
|
|
json=channel_data,
|
|
timeout=10
|
|
)
|
|
|
|
if resp.status_code == 200:
|
|
result = resp.json()
|
|
channel_id = result.get("data", {}).get("id")
|
|
print(f" ✓ 渠道创建成功: {channel_data['name']} (ID: {channel_id})")
|
|
return channel_id
|
|
else:
|
|
error = resp.json().get("detail", resp.text)
|
|
if "邮箱已被使用" in error or "already exists" in error.lower():
|
|
print(f" ⚠ 渠道已存在: {channel_data['email']}")
|
|
# 尝试获取已存在的渠道
|
|
return get_channel_by_email(token, channel_data["email"])
|
|
else:
|
|
print(f" ✗ 创建失败: {error}")
|
|
return None
|
|
except Exception as e:
|
|
print(f" ✗ 创建出错: {e}")
|
|
return None
|
|
|
|
|
|
def get_channel_by_email(token: str, email: str) -> Optional[str]:
|
|
"""通过邮箱获取渠道ID"""
|
|
try:
|
|
resp = requests.get(
|
|
f"{BASE_URL}/api/admin/channels",
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
timeout=10
|
|
)
|
|
|
|
if resp.status_code == 200:
|
|
result = resp.json()
|
|
channels = result.get("data", {}).get("channels", [])
|
|
for channel in channels:
|
|
if channel.get("email") == email:
|
|
return channel.get("id")
|
|
return None
|
|
except:
|
|
return None
|
|
|
|
|
|
def test_list_channels(token: str) -> bool:
|
|
"""测试获取渠道列表"""
|
|
print_step("测试：获取渠道列表")
|
|
|
|
try:
|
|
resp = requests.get(
|
|
f"{BASE_URL}/api/admin/channels",
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
timeout=10
|
|
)
|
|
|
|
if resp.status_code == 200:
|
|
result = resp.json()
|
|
channels = result.get("data", {}).get("channels", [])
|
|
print(f" ✓ 获取成功，共 {len(channels)} 个渠道")
|
|
for ch in channels[:3]: # 只显示前3个
|
|
print(f" - {ch.get('name')} ({ch.get('email')})")
|
|
return True
|
|
else:
|
|
error = resp.json().get("detail", resp.text)
|
|
print(f" ✗ 获取失败: {error}")
|
|
return False
|
|
except Exception as e:
|
|
print(f" ✗ 获取出错: {e}")
|
|
return False
|
|
|
|
|
|
def test_create_admin(token: str, admin_info: dict, channel_id: str = None) -> bool:
|
|
"""测试创建管理员"""
|
|
role_name = {
|
|
"billing_admin": "计费管理员",
|
|
"operations_admin": "运维管理员"
|
|
}.get(admin_info["role"], admin_info["role"])
|
|
|
|
print_step(f"测试：创建{role_name}")
|
|
|
|
admin_data = {
|
|
"name": admin_info.get("name", role_name),
|
|
"email": admin_info["email"],
|
|
"password": admin_info["password"],
|
|
"role": admin_info["role"]
|
|
}
|
|
|
|
if channel_id:
|
|
admin_data["channelId"] = channel_id
|
|
|
|
try:
|
|
resp = requests.post(
|
|
f"{BASE_URL}/api/admin/admins/create",
|
|
headers={
|
|
"Authorization": f"Bearer {token}",
|
|
"Content-Type": "application/json"
|
|
},
|
|
json=admin_data,
|
|
timeout=10
|
|
)
|
|
|
|
if resp.status_code == 200:
|
|
result = resp.json()
|
|
print(f" ✓ {role_name}创建成功: {admin_info['email']}")
|
|
return True
|
|
else:
|
|
error = resp.json().get("detail", resp.text)
|
|
if "邮箱已被使用" in error or "already exists" in error.lower():
|
|
print(f" ⚠ {role_name}已存在: {admin_info['email']}")
|
|
return True # 已存在也算成功
|
|
else:
|
|
print(f" ✗ 创建失败: {error}")
|
|
return False
|
|
except Exception as e:
|
|
print(f" ✗ 创建出错: {e}")
|
|
return False
|
|
|
|
|
|
def test_list_admins(token: str) -> bool:
|
|
"""测试获取管理员列表"""
|
|
print_step("测试：获取管理员列表")
|
|
|
|
try:
|
|
resp = requests.get(
|
|
f"{BASE_URL}/api/admin/admins",
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
timeout=10
|
|
)
|
|
|
|
if resp.status_code == 200:
|
|
result = resp.json()
|
|
admins = result.get("data", {}).get("admins", [])
|
|
print(f" ✓ 获取成功，共 {len(admins)} 个管理员")
|
|
for admin in admins[:5]: # 只显示前5个
|
|
print(f" - {admin.get('name')} ({admin.get('email')}) - {admin.get('role')}")
|
|
return True
|
|
else:
|
|
error = resp.json().get("detail", resp.text)
|
|
print(f" ✗ 获取失败: {error}")
|
|
return False
|
|
except Exception as e:
|
|
print(f" ✗ 获取出错: {e}")
|
|
return False
|
|
|
|
|
|
def test_create_tenant(token: str, channel_token: str = None) -> Optional[str]:
|
|
"""测试创建租户"""
|
|
print_step("测试：创建租户")
|
|
|
|
# 使用渠道管理员token或提供的token
|
|
use_token = channel_token or token
|
|
|
|
tenant_data = {
|
|
"name": "测试租户A",
|
|
"email": "tenant-a@test.com",
|
|
"password": "Tenant@123456",
|
|
"subscriptionTier": "pro"
|
|
}
|
|
|
|
try:
|
|
resp = requests.post(
|
|
f"{BASE_URL}/api/channel/tenants/create",
|
|
headers={
|
|
"Authorization": f"Bearer {use_token}",
|
|
"Content-Type": "application/json"
|
|
},
|
|
json=tenant_data,
|
|
timeout=10
|
|
)
|
|
|
|
if resp.status_code == 200:
|
|
result = resp.json()
|
|
tenant_id = result.get("data", {}).get("id")
|
|
print(f" ✓ 租户创建成功: {tenant_data['name']} (ID: {tenant_id})")
|
|
return tenant_id
|
|
else:
|
|
error = resp.json().get("detail", resp.text)
|
|
if "邮箱已被使用" in error or "already exists" in error.lower():
|
|
print(f" ⚠ 租户已存在: {tenant_data['email']}")
|
|
return "existing"
|
|
else:
|
|
print(f" ✗ 创建失败: {error}")
|
|
return None
|
|
except Exception as e:
|
|
print(f" ✗ 创建出错: {e}")
|
|
return None
|
|
|
|
|
|
def test_list_tenants(token: str) -> bool:
|
|
"""测试获取租户列表"""
|
|
print_step("测试：获取租户列表")
|
|
|
|
try:
|
|
resp = requests.get(
|
|
f"{BASE_URL}/api/channel/tenants",
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
timeout=10
|
|
)
|
|
|
|
if resp.status_code == 200:
|
|
result = resp.json()
|
|
tenants = result.get("data", {}).get("tenants", [])
|
|
print(f" ✓ 获取成功，共 {len(tenants)} 个租户")
|
|
for tenant in tenants[:5]: # 只显示前5个
|
|
print(f" - {tenant.get('name')} ({tenant.get('email')}) - {tenant.get('subscriptionTier')}")
|
|
return True
|
|
else:
|
|
error = resp.json().get("detail", resp.text)
|
|
print(f" ✗ 获取失败: {error}")
|
|
return False
|
|
except Exception as e:
|
|
print(f" ✗ 获取出错: {e}")
|
|
return False
|
|
|
|
|
|
def test_permission_verification():
|
|
"""测试权限验证"""
|
|
print_header("权限验证测试")
|
|
|
|
# 测试1: 计费管理员不能创建渠道
|
|
print_step("测试1: 计费管理员尝试创建渠道（应该失败）")
|
|
billing_token = login(BILLING_ADMIN["email"], BILLING_ADMIN["password"], BILLING_ADMIN["role"])
|
|
if billing_token:
|
|
try:
|
|
resp = requests.post(
|
|
f"{BASE_URL}/api/admin/channels/create",
|
|
headers={
|
|
"Authorization": f"Bearer {billing_token}",
|
|
"Content-Type": "application/json"
|
|
},
|
|
json={
|
|
"name": "未授权渠道",
|
|
"email": "unauthorized@test.com",
|
|
"password": "Test@123456",
|
|
"commissionRate": 10.0
|
|
},
|
|
timeout=10
|
|
)
|
|
if resp.status_code == 403:
|
|
print(f" ✓ 权限验证正确：计费管理员无法创建渠道")
|
|
else:
|
|
print(f" ✗ 权限验证失败：计费管理员不应该能创建渠道")
|
|
except Exception as e:
|
|
print(f" ✗ 测试出错: {e}")
|
|
|
|
# 测试2: 运维管理员不能创建租户
|
|
print_step("测试2: 运维管理员尝试创建租户（应该失败）")
|
|
ops_token = login(OPS_ADMIN["email"], OPS_ADMIN["password"], OPS_ADMIN["role"])
|
|
if ops_token:
|
|
try:
|
|
resp = requests.post(
|
|
f"{BASE_URL}/api/channel/tenants/create",
|
|
headers={
|
|
"Authorization": f"Bearer {ops_token}",
|
|
"Content-Type": "application/json"
|
|
},
|
|
json={
|
|
"name": "未授权租户",
|
|
"email": "unauthorized-tenant@test.com",
|
|
"password": "Test@123456",
|
|
"subscriptionTier": "free"
|
|
},
|
|
timeout=10
|
|
)
|
|
if resp.status_code == 403:
|
|
print(f" ✓ 权限验证正确：运维管理员无法创建租户")
|
|
else:
|
|
print(f" ✗ 权限验证失败：运维管理员不应该能创建租户")
|
|
except Exception as e:
|
|
print(f" ✗ 测试出错: {e}")
|
|
|
|
# 测试3: 计费管理员可以创建租户
|
|
print_step("测试3: 计费管理员尝试创建租户（应该成功）")
|
|
if billing_token:
|
|
tenant_id = test_create_tenant(billing_token, billing_token)
|
|
if tenant_id:
|
|
print(f" ✓ 权限验证正确：计费管理员可以创建租户")
|
|
else:
|
|
print(f" ⚠ 创建租户失败（可能是其他原因）")
|
|
|
|
# 测试4: 超级管理员可以访问所有资源
|
|
print_step("测试4: 超级管理员访问所有资源（应该成功）")
|
|
super_token = login(SUPER_ADMIN["email"], SUPER_ADMIN["password"], SUPER_ADMIN["role"])
|
|
if super_token:
|
|
success = True
|
|
success = success and test_list_channels(super_token)
|
|
success = success and test_list_admins(super_token)
|
|
if success:
|
|
print(f" ✓ 权限验证正确：超级管理员可以访问所有资源")
|
|
else:
|
|
print(f" ✗ 部分资源访问失败")
|
|
|
|
|
|
def main():
|
|
"""主函数"""
|
|
print_header("完整工作流测试")
|
|
|
|
# 步骤1: 等待服务启动
|
|
print_step("等待服务启动")
|
|
if not wait_for_service(BASE_URL):
|
|
print(" ✗ 服务未启动，请先启动Docker服务")
|
|
return
|
|
|
|
# 步骤2: 登录超级管理员
|
|
print_step("登录超级管理员")
|
|
super_token = login(SUPER_ADMIN["email"], SUPER_ADMIN["password"], SUPER_ADMIN["role"])
|
|
if not super_token:
|
|
print(" ✗ 无法登录超级管理员，请先运行 test_cre_admins.py 创建管理员")
|
|
return
|
|
|
|
# 步骤3: 测试创建渠道
|
|
channel_id = test_create_channel(super_token)
|
|
|
|
# 步骤4: 测试获取渠道列表
|
|
test_list_channels(super_token)
|
|
|
|
# 步骤5: 测试创建管理员
|
|
if channel_id:
|
|
test_create_admin(super_token, BILLING_ADMIN, channel_id)
|
|
test_create_admin(super_token, OPS_ADMIN, channel_id)
|
|
|
|
# 步骤6: 测试获取管理员列表
|
|
test_list_admins(super_token)
|
|
|
|
# 步骤7: 测试创建租户（使用计费管理员）
|
|
print_header("租户管理测试")
|
|
billing_token = login(BILLING_ADMIN["email"], BILLING_ADMIN["password"], BILLING_ADMIN["role"])
|
|
if billing_token:
|
|
test_create_tenant(billing_token, billing_token)
|
|
test_list_tenants(billing_token)
|
|
|
|
# 步骤8: 权限验证
|
|
test_permission_verification()
|
|
|
|
# 总结
|
|
print_header("测试完成")
|
|
print("\n所有测试已完成！")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|
|
|