998211c483
- 从服务器拉取完整代码 - 按框架规范整理项目结构 - 配置 Drone CI 测试环境部署 - 包含后端(FastAPI)、前端(Vue3)、管理端 技术栈: Vue3 + TypeScript + FastAPI + MySQL
273 lines
9.0 KiB
Python
273 lines
9.0 KiB
Python
"""
|
||
团队管理API测试脚本
|
||
用于验证团队管理相关接口的功能
|
||
"""
|
||
import asyncio
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
# 添加项目根目录到Python路径
|
||
sys.path.insert(0, str(Path(__file__).parent))
|
||
|
||
import httpx
|
||
|
||
|
||
BASE_URL = "http://localhost:8000/api/v1"
|
||
TOKEN = None # 需要先登录获取token
|
||
|
||
|
||
async def login():
|
||
"""登录获取token"""
|
||
global TOKEN
|
||
async with httpx.AsyncClient() as client:
|
||
response = await client.post(
|
||
f"{BASE_URL}/auth/login",
|
||
json={
|
||
"username": "admin",
|
||
"password": "admin123"
|
||
}
|
||
)
|
||
if response.status_code == 200:
|
||
data = response.json()
|
||
TOKEN = data.get("data", {}).get("access_token")
|
||
print(f"✅ 登录成功，获取token: {TOKEN[:20]}...")
|
||
return True
|
||
else:
|
||
print(f"❌ 登录失败: {response.text}")
|
||
return False
|
||
|
||
|
||
async def test_team_statistics():
|
||
"""测试团队统计接口"""
|
||
print("\n" + "="*60)
|
||
print("测试1: GET /team/management/statistics")
|
||
print("="*60)
|
||
|
||
async with httpx.AsyncClient() as client:
|
||
response = await client.get(
|
||
f"{BASE_URL}/team/management/statistics",
|
||
headers={"Authorization": f"Bearer {TOKEN}"}
|
||
)
|
||
|
||
print(f"状态码: {response.status_code}")
|
||
data = response.json()
|
||
print(f"响应数据: {data}")
|
||
|
||
if response.status_code == 200:
|
||
stats = data.get("data", {})
|
||
print(f"\n📊 团队统计:")
|
||
print(f" - 团队总人数: {stats.get('teamCount')}")
|
||
print(f" - 活跃成员: {stats.get('activeMembers')}")
|
||
print(f" - 平均学习进度: {stats.get('avgProgress')}%")
|
||
print(f" - 团队平均分: {stats.get('avgScore')}")
|
||
print("✅ 测试通过")
|
||
else:
|
||
print(f"❌ 测试失败: {data.get('message')}")
|
||
|
||
|
||
async def test_team_members():
|
||
"""测试团队成员列表接口"""
|
||
print("\n" + "="*60)
|
||
print("测试2: GET /team/management/members")
|
||
print("="*60)
|
||
|
||
async with httpx.AsyncClient() as client:
|
||
# 测试基础查询
|
||
response = await client.get(
|
||
f"{BASE_URL}/team/management/members",
|
||
params={"page": 1, "size": 5},
|
||
headers={"Authorization": f"Bearer {TOKEN}"}
|
||
)
|
||
|
||
print(f"状态码: {response.status_code}")
|
||
data = response.json()
|
||
|
||
if response.status_code == 200:
|
||
result = data.get("data", {})
|
||
print(f"\n👥 成员列表:")
|
||
print(f" - 总数: {result.get('total')}")
|
||
print(f" - 当前页: {result.get('page')}")
|
||
print(f" - 每页数量: {result.get('page_size')}")
|
||
print(f" - 总页数: {result.get('pages')}")
|
||
|
||
items = result.get("items", [])
|
||
print(f"\n 成员列表 (前{len(items)}条):")
|
||
for member in items:
|
||
print(f" - ID: {member['id']}, 姓名: {member['name']}, "
|
||
f"岗位: {member['position']}, 状态: {member['status']}, "
|
||
f"进度: {member['progress']}%, 平均分: {member['avgScore']}")
|
||
print("✅ 测试通过")
|
||
else:
|
||
print(f"❌ 测试失败: {data.get('message')}")
|
||
|
||
# 测试搜索功能
|
||
print("\n测试搜索功能...")
|
||
response = await client.get(
|
||
f"{BASE_URL}/team/management/members",
|
||
params={"page": 1, "size": 5, "search_text": "admin"},
|
||
headers={"Authorization": f"Bearer {TOKEN}"}
|
||
)
|
||
if response.status_code == 200:
|
||
data = response.json()
|
||
result = data.get("data", {})
|
||
print(f" 搜索'admin'结果数: {result.get('total')}")
|
||
print("✅ 搜索测试通过")
|
||
|
||
|
||
async def test_member_detail():
|
||
"""测试成员详情接口"""
|
||
print("\n" + "="*60)
|
||
print("测试3: GET /team/management/members/{id}/detail")
|
||
print("="*60)
|
||
|
||
# 先获取一个成员ID
|
||
async with httpx.AsyncClient() as client:
|
||
response = await client.get(
|
||
f"{BASE_URL}/team/management/members",
|
||
params={"page": 1, "size": 1},
|
||
headers={"Authorization": f"Bearer {TOKEN}"}
|
||
)
|
||
|
||
if response.status_code != 200:
|
||
print("❌ 无法获取成员列表")
|
||
return
|
||
|
||
data = response.json()
|
||
items = data.get("data", {}).get("items", [])
|
||
if not items:
|
||
print("⚠️ 没有成员数据，跳过测试")
|
||
return
|
||
|
||
member_id = items[0]["id"]
|
||
print(f"测试成员ID: {member_id}")
|
||
|
||
# 获取成员详情
|
||
response = await client.get(
|
||
f"{BASE_URL}/team/management/members/{member_id}/detail",
|
||
headers={"Authorization": f"Bearer {TOKEN}"}
|
||
)
|
||
|
||
print(f"状态码: {response.status_code}")
|
||
data = response.json()
|
||
|
||
if response.status_code == 200:
|
||
detail = data.get("data", {})
|
||
print(f"\n📋 成员详情:")
|
||
print(f" - 姓名: {detail.get('name')}")
|
||
print(f" - 岗位: {detail.get('position')}")
|
||
print(f" - 状态: {detail.get('status')}")
|
||
print(f" - 学习时长: {detail.get('studyTime')}小时")
|
||
print(f" - 完成课程: {detail.get('completedCourses')}门")
|
||
print(f" - 平均成绩: {detail.get('avgScore')}分")
|
||
print(f" - 通过率: {detail.get('passRate')}%")
|
||
|
||
records = detail.get("recentRecords", [])
|
||
print(f"\n 最近学习记录 ({len(records)}条):")
|
||
for record in records[:3]:
|
||
print(f" - {record['time']}: {record['content']}")
|
||
|
||
print("✅ 测试通过")
|
||
else:
|
||
print(f"❌ 测试失败: {data.get('message')}")
|
||
|
||
|
||
async def test_member_report():
|
||
"""测试成员学习报告接口"""
|
||
print("\n" + "="*60)
|
||
print("测试4: GET /team/management/members/{id}/report")
|
||
print("="*60)
|
||
|
||
# 先获取一个成员ID
|
||
async with httpx.AsyncClient() as client:
|
||
response = await client.get(
|
||
f"{BASE_URL}/team/management/members",
|
||
params={"page": 1, "size": 1},
|
||
headers={"Authorization": f"Bearer {TOKEN}"}
|
||
)
|
||
|
||
if response.status_code != 200:
|
||
print("❌ 无法获取成员列表")
|
||
return
|
||
|
||
data = response.json()
|
||
items = data.get("data", {}).get("items", [])
|
||
if not items:
|
||
print("⚠️ 没有成员数据，跳过测试")
|
||
return
|
||
|
||
member_id = items[0]["id"]
|
||
print(f"测试成员ID: {member_id}")
|
||
|
||
# 获取学习报告
|
||
response = await client.get(
|
||
f"{BASE_URL}/team/management/members/{member_id}/report",
|
||
headers={"Authorization": f"Bearer {TOKEN}"}
|
||
)
|
||
|
||
print(f"状态码: {response.status_code}")
|
||
data = response.json()
|
||
|
||
if response.status_code == 200:
|
||
report = data.get("data", {})
|
||
|
||
# 概览
|
||
overview = report.get("overview", [])
|
||
print(f"\n📊 报告概览:")
|
||
for item in overview:
|
||
print(f" - {item['label']}: {item['value']}")
|
||
|
||
# 进度趋势
|
||
trend = report.get("progressTrend", {})
|
||
dates = trend.get("dates", [])
|
||
data_points = trend.get("data", [])
|
||
print(f"\n📈 学习进度趋势 (30天):")
|
||
print(f" - 数据点数: {len(dates)}")
|
||
if dates and data_points:
|
||
print(f" - 起始: {dates[0]} -> {data_points[0]}%")
|
||
print(f" - 结束: {dates[-1]} -> {data_points[-1]}%")
|
||
|
||
# 能力评估
|
||
abilities = report.get("abilities", [])
|
||
print(f"\n🎯 能力评估:")
|
||
for ability in abilities:
|
||
print(f" - {ability['name']}: {ability['score']}分 - {ability['description']}")
|
||
|
||
# 详细记录
|
||
records = report.get("records", [])
|
||
print(f"\n📚 详细学习记录 ({len(records)}条)")
|
||
|
||
print("✅ 测试通过")
|
||
else:
|
||
print(f"❌ 测试失败: {data.get('message')}")
|
||
|
||
|
||
async def main():
|
||
"""主测试函数"""
|
||
print("="*60)
|
||
print("团队管理API测试")
|
||
print("="*60)
|
||
|
||
# 登录
|
||
if not await login():
|
||
return
|
||
|
||
# 运行测试
|
||
try:
|
||
await test_team_statistics()
|
||
await test_team_members()
|
||
await test_member_detail()
|
||
await test_member_report()
|
||
|
||
print("\n" + "="*60)
|
||
print("✅ 所有测试完成")
|
||
print("="*60)
|
||
except Exception as e:
|
||
print(f"\n❌ 测试过程中出现错误: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
asyncio.run(main())
|
||
|