""" 团队管理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())