998211c483
- 从服务器拉取完整代码 - 按框架规范整理项目结构 - 配置 Drone CI 测试环境部署 - 包含后端(FastAPI)、前端(Vue3)、管理端 技术栈: Vue3 + TypeScript + FastAPI + MySQL
571 lines
22 KiB
Python
571 lines
22 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
考培练系统 AI 功能完整测试脚本
|
||
|
||
测试所有 AI 功能是否正常：
|
||
1. AI 基础服务连通性（4sapi + OpenRouter）
|
||
2. 试题生成功能（V2 引擎）
|
||
3. 答案判断功能（V2 引擎）
|
||
4. 知识点分析功能（V2 引擎）
|
||
5. 课程对话功能（V2 引擎）
|
||
6. 陪练场景提取功能（V2 引擎）
|
||
7. 能力分析功能（V2 引擎）
|
||
|
||
测试方式：通过 HTTP API 调用后端服务
|
||
"""
|
||
|
||
import asyncio
|
||
import httpx
|
||
import json
|
||
import sys
|
||
import time
|
||
from datetime import datetime
|
||
from typing import Optional, Dict, Any, List
|
||
|
||
# ==================== 配置 ====================
|
||
|
||
# 测试基础URL（演示版后端）
|
||
BASE_URL = "http://localhost:8000/api/v1"
|
||
|
||
# 测试账户（从数据库获取有效用户）
|
||
TEST_USERNAME = "admin"
|
||
TEST_PASSWORD = "admin123"
|
||
|
||
# 测试超时时间（秒）
|
||
TIMEOUT = 120
|
||
|
||
# 测试结果统计
|
||
test_results: List[Dict[str, Any]] = []
|
||
|
||
# ==================== 工具函数 ====================
|
||
|
||
def print_header(title: str):
|
||
"""打印标题"""
|
||
print("\n" + "=" * 60)
|
||
print(f" {title}")
|
||
print("=" * 60)
|
||
|
||
def print_result(test_name: str, success: bool, message: str, duration_ms: int = 0):
|
||
"""打印测试结果"""
|
||
status = "✅ 通过" if success else "❌ 失败"
|
||
duration_str = f" ({duration_ms}ms)" if duration_ms > 0 else ""
|
||
print(f" {status} | {test_name}{duration_str}")
|
||
if not success:
|
||
print(f" └─ {message}")
|
||
|
||
# 记录结果
|
||
test_results.append({
|
||
"name": test_name,
|
||
"success": success,
|
||
"message": message,
|
||
"duration_ms": duration_ms,
|
||
"timestamp": datetime.now().isoformat()
|
||
})
|
||
|
||
async def get_auth_token(client: httpx.AsyncClient) -> Optional[str]:
|
||
"""获取认证 token"""
|
||
try:
|
||
response = await client.post(
|
||
f"{BASE_URL}/auth/login",
|
||
data={"username": TEST_USERNAME, "password": TEST_PASSWORD}
|
||
)
|
||
if response.status_code == 200:
|
||
data = response.json()
|
||
return data.get("data", {}).get("access_token")
|
||
else:
|
||
print(f" ⚠️ 登录失败: {response.status_code} - {response.text}")
|
||
return None
|
||
except Exception as e:
|
||
print(f" ⚠️ 登录异常: {e}")
|
||
return None
|
||
|
||
# ==================== 测试用例 ====================
|
||
|
||
async def test_1_ai_basic_connectivity():
|
||
"""测试1: AI 基础服务连通性"""
|
||
print_header("测试1: AI 基础服务连通性")
|
||
|
||
# 直接测试 4sapi
|
||
async with httpx.AsyncClient(timeout=30) as client:
|
||
# 4sapi 测试
|
||
start = time.time()
|
||
try:
|
||
response = await client.post(
|
||
"https://4sapi.com/v1/chat/completions",
|
||
headers={
|
||
"Authorization": "Bearer sk-9yMCXjRGANbacz20kJY8doSNy6Rf446aYwmgGIuIXQ7DAyBw",
|
||
"Content-Type": "application/json"
|
||
},
|
||
json={
|
||
"model": "gemini-3-flash-preview",
|
||
"messages": [{"role": "user", "content": "你好，请回复'测试成功'"}],
|
||
"max_tokens": 50
|
||
}
|
||
)
|
||
duration = int((time.time() - start) * 1000)
|
||
|
||
if response.status_code == 200:
|
||
data = response.json()
|
||
content = data.get("choices", [{}])[0].get("message", {}).get("content", "")
|
||
print_result("4sapi.com 连通性", True, f"响应: {content[:50]}...", duration)
|
||
else:
|
||
print_result("4sapi.com 连通性", False, f"HTTP {response.status_code}: {response.text[:100]}", duration)
|
||
except Exception as e:
|
||
duration = int((time.time() - start) * 1000)
|
||
print_result("4sapi.com 连通性", False, str(e), duration)
|
||
|
||
# OpenRouter 测试
|
||
start = time.time()
|
||
try:
|
||
response = await client.post(
|
||
"https://openrouter.ai/api/v1/chat/completions",
|
||
headers={
|
||
"Authorization": "Bearer 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",
|
||
"Content-Type": "application/json"
|
||
},
|
||
json={
|
||
"model": "google/gemini-3-flash-preview",
|
||
"messages": [{"role": "user", "content": "你好，请回复'测试成功'"}],
|
||
"max_tokens": 50
|
||
}
|
||
)
|
||
duration = int((time.time() - start) * 1000)
|
||
|
||
if response.status_code == 200:
|
||
data = response.json()
|
||
content = data.get("choices", [{}])[0].get("message", {}).get("content", "")
|
||
print_result("OpenRouter 连通性", True, f"响应: {content[:50]}...", duration)
|
||
else:
|
||
print_result("OpenRouter 连通性", False, f"HTTP {response.status_code}: {response.text[:100]}", duration)
|
||
except Exception as e:
|
||
duration = int((time.time() - start) * 1000)
|
||
print_result("OpenRouter 连通性", False, str(e), duration)
|
||
|
||
async def test_2_exam_generation():
|
||
"""测试2: 试题生成功能"""
|
||
print_header("测试2: 试题生成功能 (V2 引擎)")
|
||
|
||
async with httpx.AsyncClient(timeout=TIMEOUT) as client:
|
||
# 获取 token
|
||
token = await get_auth_token(client)
|
||
if not token:
|
||
print_result("试题生成", False, "无法获取认证 token")
|
||
return
|
||
|
||
headers = {"Authorization": f"Bearer {token}"}
|
||
|
||
# 先查询一个有知识点的课程
|
||
try:
|
||
response = await client.get(f"{BASE_URL}/courses", headers=headers)
|
||
if response.status_code != 200:
|
||
print_result("试题生成 - 查询课程", False, f"HTTP {response.status_code}")
|
||
return
|
||
|
||
courses = response.json().get("data", {}).get("items", [])
|
||
if not courses:
|
||
print_result("试题生成", False, "没有可用的课程")
|
||
return
|
||
|
||
course_id = courses[0].get("id")
|
||
print(f" 📝 使用课程ID: {course_id}")
|
||
except Exception as e:
|
||
print_result("试题生成 - 查询课程", False, str(e))
|
||
return
|
||
|
||
# 测试试题生成 API
|
||
start = time.time()
|
||
try:
|
||
response = await client.post(
|
||
f"{BASE_URL}/exams/generate?engine=v2",
|
||
headers=headers,
|
||
json={
|
||
"course_id": course_id,
|
||
"single_choice_count": 2,
|
||
"multiple_choice_count": 1,
|
||
"true_false_count": 1,
|
||
"fill_blank_count": 0,
|
||
"essay_count": 0,
|
||
"difficulty_level": 3,
|
||
"current_round": 1
|
||
}
|
||
)
|
||
duration = int((time.time() - start) * 1000)
|
||
|
||
if response.status_code == 200:
|
||
data = response.json()
|
||
result = data.get("data", {}).get("result", "")
|
||
try:
|
||
questions = json.loads(result) if result else []
|
||
print_result("试题生成 (V2)", True, f"生成了 {len(questions)} 道题目", duration)
|
||
except:
|
||
print_result("试题生成 (V2)", True, f"返回结果: {result[:100]}...", duration)
|
||
else:
|
||
print_result("试题生成 (V2)", False, f"HTTP {response.status_code}: {response.text[:200]}", duration)
|
||
except Exception as e:
|
||
duration = int((time.time() - start) * 1000)
|
||
print_result("试题生成 (V2)", False, str(e), duration)
|
||
|
||
async def test_3_answer_judge():
|
||
"""测试3: 答案判断功能"""
|
||
print_header("测试3: 答案判断功能 (V2 引擎)")
|
||
|
||
async with httpx.AsyncClient(timeout=TIMEOUT) as client:
|
||
# 获取 token
|
||
token = await get_auth_token(client)
|
||
if not token:
|
||
print_result("答案判断", False, "无法获取认证 token")
|
||
return
|
||
|
||
headers = {"Authorization": f"Bearer {token}"}
|
||
|
||
# 测试正确答案
|
||
start = time.time()
|
||
try:
|
||
response = await client.post(
|
||
f"{BASE_URL}/exams/judge-answer?engine=v2",
|
||
headers=headers,
|
||
json={
|
||
"question": "玻尿酸的主要作用是什么？",
|
||
"correct_answer": "保湿、填充、塑形",
|
||
"user_answer": "保湿和填充",
|
||
"analysis": "玻尿酸具有强大的保水能力，可用于皮肤保湿和面部填充"
|
||
}
|
||
)
|
||
duration = int((time.time() - start) * 1000)
|
||
|
||
if response.status_code == 200:
|
||
data = response.json()
|
||
is_correct = data.get("data", {}).get("is_correct")
|
||
print_result("答案判断 (V2)", True, f"判断结果: {'正确' if is_correct else '错误'}", duration)
|
||
else:
|
||
print_result("答案判断 (V2)", False, f"HTTP {response.status_code}: {response.text[:200]}", duration)
|
||
except Exception as e:
|
||
duration = int((time.time() - start) * 1000)
|
||
print_result("答案判断 (V2)", False, str(e), duration)
|
||
|
||
async def test_4_knowledge_analysis():
|
||
"""测试4: 知识点分析功能"""
|
||
print_header("测试4: 知识点分析功能 (V2 引擎)")
|
||
|
||
async with httpx.AsyncClient(timeout=TIMEOUT) as client:
|
||
# 获取 token
|
||
token = await get_auth_token(client)
|
||
if not token:
|
||
print_result("知识点分析", False, "无法获取认证 token")
|
||
return
|
||
|
||
headers = {"Authorization": f"Bearer {token}"}
|
||
|
||
# 先查询课程和资料
|
||
try:
|
||
response = await client.get(f"{BASE_URL}/courses", headers=headers)
|
||
if response.status_code != 200:
|
||
print_result("知识点分析 - 查询课程", False, f"HTTP {response.status_code}")
|
||
return
|
||
|
||
courses = response.json().get("data", {}).get("items", [])
|
||
if not courses:
|
||
print_result("知识点分析", False, "没有可用的课程")
|
||
return
|
||
|
||
course_id = courses[0].get("id")
|
||
|
||
# 查询课程资料
|
||
response = await client.get(f"{BASE_URL}/courses/{course_id}/materials", headers=headers)
|
||
if response.status_code != 200:
|
||
print_result("知识点分析 - 查询资料", False, f"HTTP {response.status_code}")
|
||
return
|
||
|
||
materials = response.json().get("data", {}).get("items", [])
|
||
if not materials:
|
||
print_result("知识点分析", False, "课程没有可用的资料")
|
||
return
|
||
|
||
material_id = materials[0].get("id")
|
||
print(f" 📝 使用课程ID: {course_id}, 资料ID: {material_id}")
|
||
except Exception as e:
|
||
print_result("知识点分析 - 查询资料", False, str(e))
|
||
return
|
||
|
||
# 测试知识点分析 API
|
||
start = time.time()
|
||
try:
|
||
response = await client.post(
|
||
f"{BASE_URL}/courses/{course_id}/materials/{material_id}/analyze?engine=v2",
|
||
headers=headers
|
||
)
|
||
duration = int((time.time() - start) * 1000)
|
||
|
||
if response.status_code == 200:
|
||
data = response.json()
|
||
kps = data.get("data", {}).get("knowledge_points", [])
|
||
print_result("知识点分析 (V2)", True, f"提取了 {len(kps)} 个知识点", duration)
|
||
else:
|
||
# 可能返回 400 因为没有文件内容
|
||
error_detail = response.json().get("detail", response.text[:200])
|
||
print_result("知识点分析 (V2)", False, f"HTTP {response.status_code}: {error_detail}", duration)
|
||
except Exception as e:
|
||
duration = int((time.time() - start) * 1000)
|
||
print_result("知识点分析 (V2)", False, str(e), duration)
|
||
|
||
async def test_5_course_chat():
|
||
"""测试5: 课程对话功能"""
|
||
print_header("测试5: 课程对话功能 (V2 引擎)")
|
||
|
||
async with httpx.AsyncClient(timeout=TIMEOUT) as client:
|
||
# 获取 token
|
||
token = await get_auth_token(client)
|
||
if not token:
|
||
print_result("课程对话", False, "无法获取认证 token")
|
||
return
|
||
|
||
headers = {"Authorization": f"Bearer {token}"}
|
||
|
||
# 先查询课程
|
||
try:
|
||
response = await client.get(f"{BASE_URL}/courses", headers=headers)
|
||
if response.status_code != 200:
|
||
print_result("课程对话 - 查询课程", False, f"HTTP {response.status_code}")
|
||
return
|
||
|
||
courses = response.json().get("data", {}).get("items", [])
|
||
if not courses:
|
||
print_result("课程对话", False, "没有可用的课程")
|
||
return
|
||
|
||
course_id = courses[0].get("id")
|
||
print(f" 📝 使用课程ID: {course_id}")
|
||
except Exception as e:
|
||
print_result("课程对话 - 查询课程", False, str(e))
|
||
return
|
||
|
||
# 测试课程对话 API（流式响应）
|
||
start = time.time()
|
||
try:
|
||
# 注意：课程对话是流式响应，需要特殊处理
|
||
async with client.stream(
|
||
"POST",
|
||
f"{BASE_URL}/course/chat?engine=v2",
|
||
headers=headers,
|
||
json={
|
||
"course_id": course_id,
|
||
"query": "请简单介绍一下这门课程的主要内容"
|
||
}
|
||
) as response:
|
||
if response.status_code == 200:
|
||
content = ""
|
||
async for line in response.aiter_lines():
|
||
if line.startswith("data:"):
|
||
try:
|
||
data = json.loads(line[5:].strip())
|
||
if data.get("event") == "message_chunk":
|
||
content += data.get("content", "")
|
||
except:
|
||
pass
|
||
|
||
duration = int((time.time() - start) * 1000)
|
||
if content:
|
||
print_result("课程对话 (V2)", True, f"响应: {content[:100]}...", duration)
|
||
else:
|
||
print_result("课程对话 (V2)", True, "收到流式响应", duration)
|
||
else:
|
||
duration = int((time.time() - start) * 1000)
|
||
body = await response.aread()
|
||
print_result("课程对话 (V2)", False, f"HTTP {response.status_code}: {body.decode()[:200]}", duration)
|
||
except Exception as e:
|
||
duration = int((time.time() - start) * 1000)
|
||
print_result("课程对话 (V2)", False, str(e), duration)
|
||
|
||
async def test_6_practice_scene():
|
||
"""测试6: 陪练场景提取功能"""
|
||
print_header("测试6: 陪练场景提取功能 (V2 引擎)")
|
||
|
||
async with httpx.AsyncClient(timeout=TIMEOUT) as client:
|
||
# 获取 token
|
||
token = await get_auth_token(client)
|
||
if not token:
|
||
print_result("陪练场景提取", False, "无法获取认证 token")
|
||
return
|
||
|
||
headers = {"Authorization": f"Bearer {token}"}
|
||
|
||
# 先查询课程
|
||
try:
|
||
response = await client.get(f"{BASE_URL}/courses", headers=headers)
|
||
if response.status_code != 200:
|
||
print_result("陪练场景提取 - 查询课程", False, f"HTTP {response.status_code}")
|
||
return
|
||
|
||
courses = response.json().get("data", {}).get("items", [])
|
||
if not courses:
|
||
print_result("陪练场景提取", False, "没有可用的课程")
|
||
return
|
||
|
||
course_id = courses[0].get("id")
|
||
print(f" 📝 使用课程ID: {course_id}")
|
||
except Exception as e:
|
||
print_result("陪练场景提取 - 查询课程", False, str(e))
|
||
return
|
||
|
||
# 测试陪练场景提取 API
|
||
start = time.time()
|
||
try:
|
||
response = await client.post(
|
||
f"{BASE_URL}/practice/extract-scene?engine=v2",
|
||
headers=headers,
|
||
json={
|
||
"course_id": course_id,
|
||
"scene_type": "consultation" # 咨询场景
|
||
}
|
||
)
|
||
duration = int((time.time() - start) * 1000)
|
||
|
||
if response.status_code == 200:
|
||
data = response.json()
|
||
scenes = data.get("data", {}).get("scenes", [])
|
||
print_result("陪练场景提取 (V2)", True, f"提取了 {len(scenes)} 个场景", duration)
|
||
else:
|
||
error_detail = response.json().get("detail", response.text[:200])
|
||
print_result("陪练场景提取 (V2)", False, f"HTTP {response.status_code}: {error_detail}", duration)
|
||
except Exception as e:
|
||
duration = int((time.time() - start) * 1000)
|
||
print_result("陪练场景提取 (V2)", False, str(e), duration)
|
||
|
||
async def test_7_ability_analysis():
|
||
"""测试7: 能力分析功能"""
|
||
print_header("测试7: 能力分析功能 (V2 引擎)")
|
||
|
||
async with httpx.AsyncClient(timeout=TIMEOUT) as client:
|
||
# 获取 token
|
||
token = await get_auth_token(client)
|
||
if not token:
|
||
print_result("能力分析", False, "无法获取认证 token")
|
||
return
|
||
|
||
headers = {"Authorization": f"Bearer {token}"}
|
||
|
||
# 测试能力分析 API（需要员工的言迹数据）
|
||
start = time.time()
|
||
try:
|
||
# 先查询是否有员工数据
|
||
response = await client.get(f"{BASE_URL}/users/me", headers=headers)
|
||
if response.status_code != 200:
|
||
print_result("能力分析 - 获取用户信息", False, f"HTTP {response.status_code}")
|
||
return
|
||
|
||
user = response.json().get("data", {})
|
||
user_id = user.get("id")
|
||
print(f" 📝 当前用户ID: {user_id}")
|
||
|
||
# 测试能力分析 API
|
||
response = await client.get(
|
||
f"{BASE_URL}/ability/user/{user_id}/analysis?engine=v2",
|
||
headers=headers
|
||
)
|
||
duration = int((time.time() - start) * 1000)
|
||
|
||
if response.status_code == 200:
|
||
data = response.json()
|
||
dimensions = data.get("data", {}).get("dimensions", [])
|
||
print_result("能力分析 (V2)", True, f"分析了 {len(dimensions)} 个能力维度", duration)
|
||
elif response.status_code == 404:
|
||
# 可能没有言迹数据
|
||
print_result("能力分析 (V2)", True, "接口正常（无言迹数据）", duration)
|
||
else:
|
||
error_detail = response.json().get("detail", response.text[:200])
|
||
print_result("能力分析 (V2)", False, f"HTTP {response.status_code}: {error_detail}", duration)
|
||
except Exception as e:
|
||
duration = int((time.time() - start) * 1000)
|
||
print_result("能力分析 (V2)", False, str(e), duration)
|
||
|
||
async def test_backend_health():
|
||
"""测试后端健康状态"""
|
||
print_header("后端服务健康检查")
|
||
|
||
backends = [
|
||
("演示版 (aiedu)", "http://localhost:8000"),
|
||
("瑞小美 (kpl)", "http://localhost:8001"),
|
||
("华尔倍丽 (hua)", "http://localhost:8010"),
|
||
("杨扬宠物 (yy)", "http://localhost:8011"),
|
||
("武汉禾丽 (hl)", "http://localhost:8012"),
|
||
("芯颜定制 (xy)", "http://localhost:8013"),
|
||
("飞沃 (fw)", "http://localhost:8014"),
|
||
("恩喜成都 (ex)", "http://localhost:8015"),
|
||
("陪练试用版 (pl)", "http://localhost:8020"),
|
||
]
|
||
|
||
async with httpx.AsyncClient(timeout=10) as client:
|
||
for name, url in backends:
|
||
start = time.time()
|
||
try:
|
||
response = await client.get(f"{url}/health")
|
||
duration = int((time.time() - start) * 1000)
|
||
|
||
if response.status_code == 200:
|
||
print_result(name, True, "健康", duration)
|
||
else:
|
||
print_result(name, False, f"HTTP {response.status_code}", duration)
|
||
except Exception as e:
|
||
duration = int((time.time() - start) * 1000)
|
||
print_result(name, False, str(e), duration)
|
||
|
||
def print_summary():
|
||
"""打印测试汇总"""
|
||
print_header("测试汇总")
|
||
|
||
total = len(test_results)
|
||
passed = sum(1 for r in test_results if r["success"])
|
||
failed = total - passed
|
||
|
||
print(f"\n 总计: {total} 项测试")
|
||
print(f" ✅ 通过: {passed}")
|
||
print(f" ❌ 失败: {failed}")
|
||
|
||
if failed > 0:
|
||
print("\n 失败项目:")
|
||
for r in test_results:
|
||
if not r["success"]:
|
||
print(f" - {r['name']}: {r['message']}")
|
||
|
||
# 计算平均延迟
|
||
durations = [r["duration_ms"] for r in test_results if r["duration_ms"] > 0]
|
||
if durations:
|
||
avg_duration = sum(durations) / len(durations)
|
||
print(f"\n 平均响应时间: {avg_duration:.0f}ms")
|
||
|
||
print("\n" + "=" * 60)
|
||
|
||
return failed == 0
|
||
|
||
async def main():
|
||
"""主测试入口"""
|
||
print("\n" + "=" * 60)
|
||
print(" 考培练系统 AI 功能完整测试")
|
||
print(f" 时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
|
||
print("=" * 60)
|
||
|
||
# 检查后端健康状态
|
||
await test_backend_health()
|
||
|
||
# 运行所有测试
|
||
await test_1_ai_basic_connectivity()
|
||
await test_2_exam_generation()
|
||
await test_3_answer_judge()
|
||
await test_4_knowledge_analysis()
|
||
await test_5_course_chat()
|
||
await test_6_practice_scene()
|
||
await test_7_ability_analysis()
|
||
|
||
# 打印汇总
|
||
success = print_summary()
|
||
|
||
sys.exit(0 if success else 1)
|
||
|
||
if __name__ == "__main__":
|
||
asyncio.run(main())
|
||
|
||
|
||
|
||
|
||
|