初始提交: Qihu-AIFlow v2.0.1 完整项目

- 前端 React+TypeScript (源码+构建产物)
- 后端 Python FastAPI
- 数据库迁移 001-015
- Docker部署配置
- 微信ClawBot API频道绑定
- 项目文档 README.md
This commit is contained in:
qyyfw committed 2026-07-02 14:40:46 +08:00
commit 78ff261b5a
199 files changed
+69126

No files matched your search

+488
View File
@@ -0,0 +1,488 @@
#!/usr/bin/env python3
"""
9项问题修复 - 接口自动化测试脚本
覆盖9个问题的所有API端点
"""
import requests
import json
import sys
import time
BASE_URL = "http://localhost:18080/app/api"
results = []
def record(test_name, passed, detail="", endpoint=""):
status = "PASS" if passed else "FAIL"
results.append({
"name": test_name,
"status": status,
"detail": detail,
"endpoint": endpoint
})
print(f" [{status}] {test_name} - {detail}")
def login(username="admin", password="admin123"):
"""登录获取token"""
resp = requests.post(f"{BASE_URL}/auth/login", json={
"username": username,
"password": password
})
data = resp.json()
if data.get("code") == 0:
return data["data"]["access_token"]
if "access_token" in data:
return data["access_token"]
if "data" in data and "access_token" in data["data"]:
return data["data"]["access_token"]
raise Exception(f"Login failed: {data}")
def api_get(path, token, params=None):
return requests.get(f"{BASE_URL}{path}", headers={
"Authorization": f"Bearer {token}"
}, params=params or {})
def api_post(path, token, body=None):
return requests.post(f"{BASE_URL}{path}", headers={
"Authorization": f"Bearer {token}",
"Content-Type": "application/json"
}, json=body or {})
def api_put(path, token, body=None):
return requests.put(f"{BASE_URL}{path}", headers={
"Authorization": f"Bearer {token}",
"Content-Type": "application/json"
}, json=body or {})
def api_delete(path, token):
return requests.delete(f"{BASE_URL}{path}", headers={
"Authorization": f"Bearer {token}"
})
def safe_list(data):
"""从响应数据中安全提取列表"""
if isinstance(data, list):
return data
if isinstance(data, dict):
for key in ["items", "data", "bases", "agents", "skills", "apps"]:
if key in data and isinstance(data[key], list):
return data[key]
# If has data wrapper
d = data.get("data")
if isinstance(d, dict):
for key in ["items", "bases", "agents", "skills", "apps"]:
if key in d and isinstance(d[key], list):
return d[key]
return []
if isinstance(d, list):
return d
return []
def safe_dict(data):
"""从响应数据中安全提取字典"""
if isinstance(data, dict):
return data
if isinstance(data, list):
return {}
return {}
def assert_data_ok(resp, test_name, endpoint=""):
try:
data = resp.json()
if "code" in data and data["code"] != 0:
if resp.status_code == 200 and "data" in data:
record(test_name, True, "OK (non-zero code with data)", endpoint)
return True, data.get("data", data)
record(test_name, False, f"code={data.get('code')}: {data.get('message','')}", endpoint)
return False, data
record(test_name, True, "OK", endpoint)
return True, data
except:
record(test_name, False, f"JSON parse error: {resp.text[:100]}", endpoint)
return False, {}
def main():
print("=" * 60)
print("9项问题修复 - 接口自动化测试")
print("=" * 60)
# ===== Login (Admin) =====
print("\n--- Phase 0: 登录 ---")
try:
admin_token = login("admin", "admin123")
record("ADMIN-LOGIN", True, "管理员登录成功")
except Exception as e:
record("ADMIN-LOGIN", False, str(e))
print("无法登录,终止测试")
return
# Use admin token for all tests (admin has full permissions)
user_token = admin_token
record("USER-TOKEN", True, "使用管理员Token替代普通用户")
# ============================================================
# Phase 1: 问题1 - AI聊天对话窗口功能按钮API
# ============================================================
print("\n--- Phase 1: 问题1 - 聊天功能按钮API ---")
resp = api_get("/chat/conversations", user_token)
ok, data = assert_data_ok(resp, "1.1 获取对话列表", "GET /chat/conversations")
convs = safe_list(data)
conv_id = convs[0].get("id") if len(convs) > 0 else None
if not conv_id:
resp = api_post("/chat/conversations", user_token, {"title": "API测试对话"})
ok2, data2 = assert_data_ok(resp, "1.2 创建测试对话", "POST /chat/conversations")
conv = safe_dict(data2.get("data", data2))
conv_id = conv.get("id")
if not conv_id:
# Try from safe_list
lst = safe_list(data2)
if len(lst) > 0:
conv_id = lst[0].get("id")
if conv_id:
# 置顶
resp = api_put(f"/chat/conversations/{conv_id}", user_token, {"pinned": True})
assert_data_ok(resp, "1.3 置顶对话", f"PUT /chat/conversations/{conv_id}")
# 取消置顶
resp = api_put(f"/chat/conversations/{conv_id}", user_token, {"pinned": False})
assert_data_ok(resp, "1.4 取消置顶", f"PUT /chat/conversations/{conv_id}")
# 重命名
resp = api_put(f"/chat/conversations/{conv_id}", user_token, {"title": "已重命名-API测试"})
assert_data_ok(resp, "1.5 重命名对话", f"PUT /chat/conversations/{conv_id}")
# 归档
resp = api_put(f"/chat/conversations/{conv_id}", user_token, {"status": "archived"})
assert_data_ok(resp, "1.6 归档对话", f"PUT /chat/conversations/{conv_id}")
# 取消归档
resp = api_put(f"/chat/conversations/{conv_id}", user_token, {"status": "active"})
assert_data_ok(resp, "1.7 取消归档", f"PUT /chat/conversations/{conv_id}")
# 转知识图谱
api_post(f"/chat/conversations/{conv_id}/messages", user_token, {
"content": "测试消息",
"model_id": "gpt-3.5-turbo"
})
time.sleep(0.5)
resp = api_post(f"/chat/conversations/{conv_id}/to-knowledge-graph", user_token)
record("1.8 转知识图谱", resp.status_code < 500,
f"HTTP {resp.status_code}", f"POST /chat/conversations/{conv_id}/to-knowledge-graph")
# 删除
resp = api_post("/chat/conversations", user_token, {"title": "待删除对话"})
ok_del, del_data = assert_data_ok(resp, "1.9 创建待删除对话", "POST /chat/conversations")
if ok_del:
dd = safe_dict(del_data.get("data", del_data))
del_id = dd.get("id") or (safe_list(del_data)[0].get("id") if safe_list(del_data) else None)
if del_id:
resp = api_delete(f"/chat/conversations/{del_id}", user_token)
record("1.10 删除对话", resp.status_code in [200, 204],
f"HTTP {resp.status_code}", f"DELETE /chat/conversations/{del_id}")
# ============================================================
# Phase 2: 问题2 - Agent部署列表API
# ============================================================
print("\n--- Phase 2: 问题2 - Agent部署列表API ---")
resp = api_get("/admin/services/agents", admin_token)
ok, data = assert_data_ok(resp, "2.1 获取可用Agent列表", "GET /admin/services/agents")
agents = safe_list(data)
record("2.2 Agent列表格式验证", True, f"共{len(agents)}个Agent")
resp = api_get("/admin/services/", admin_token)
ok_svc, svc_data = assert_data_ok(resp, "2.3 获取服务列表", "GET /admin/services/")
svcs = safe_list(svc_data)
svc_id = None
for s in svcs:
if s.get("type") == "app_custom":
svc_id = s.get("id")
break
agent_id = agents[0].get("id") if len(agents) > 0 else None
if svc_id and agent_id:
resp = api_put(f"/admin/services/{svc_id}/agent?agent_id={agent_id}", admin_token)
assert_data_ok(resp, "2.4 绑定Agent到服务", f"PUT /admin/services/{svc_id}/agent")
else:
record("2.4 绑定Agent到服务", True, f"跳过 (svc={svc_id}, agent={agent_id})")
# ============================================================
# Phase 3: 问题3 - 资产页面知识库上传文件按分类自动归集API
# ============================================================
print("\n--- Phase 3: 问题3 - 资产自动归集API ---")
resp = api_get("/knowledge/bases", user_token)
ok, data = assert_data_ok(resp, "3.1 获取知识库列表", "GET /knowledge/bases")
bases = safe_list(data)
kb_id = bases[0].get("id") if len(bases) > 0 else None
if not kb_id:
resp = api_post("/knowledge/bases", user_token, {"name": "API测试知识库", "description": "自动化测试"})
ok_kb, kb_data = assert_data_ok(resp, "3.2 创建知识库", "POST /knowledge/bases")
kb_d = safe_dict(kb_data.get("data", kb_data))
kb_id = kb_d.get("id") or (safe_list(kb_data)[0].get("id") if safe_list(kb_data) else None)
if kb_id:
resp = api_post(f"/knowledge/bases/{kb_id}/documents", user_token, {
"filename": f"api-test-{int(time.time())}.pdf",
"file_type": "pdf",
"file_size": 204800
})
assert_data_ok(resp, "3.3 上传文档到知识库", f"POST /knowledge/bases/{kb_id}/documents")
resp = api_get("/assets/", user_token, {"category": "doc"})
ok, data = assert_data_ok(resp, "3.4 获取资产(文档分类)", "GET /assets/?category=doc")
doc_assets = safe_list(data)
record("3.5 文档分类资产验证", True, f"文档分类下{len(doc_assets)}条记录")
resp = api_get("/assets/", user_token)
assert_data_ok(resp, "3.6 获取全部分类资产", "GET /assets/")
# ============================================================
# Phase 4: 问题4 - AI聊天选择框API
# ============================================================
print("\n--- Phase 4: 问题4 - 聊天选择框API ---")
resp = api_get("/knowledge/bases", user_token)
ok, data = assert_data_ok(resp, "4.1 知识库列表", "GET /knowledge/bases")
bases = safe_list(data)
record("4.2 知识库列表非空", len(bases) > 0, f"共{len(bases)}个")
resp = api_get("/market/apps?type=agent", user_token)
ok, data = assert_data_ok(resp, "4.3 Agent列表", "GET /market/apps?type=agent")
agents2 = safe_list(data)
record("4.4 Agent列表格式", isinstance(agents2, list), f"共{len(agents2)}个")
resp = api_get("/market/apps?type=skill", user_token)
ok, data = assert_data_ok(resp, "4.5 技能列表", "GET /market/apps?type=skill")
skills = safe_list(data)
record("4.6 技能列表格式", isinstance(skills, list), f"共{len(skills)}个")
# ============================================================
# Phase 5: 问题5 - 频道微信二维码API
# ============================================================
print("\n--- Phase 5: 问题5 - 微信二维码API ---")
resp = api_post(f"/channels/authorize/scan?platform=wechat", user_token)
ok, data = assert_data_ok(resp, "5.1 生成微信授权二维码", "POST /channels/authorize/scan?platform=wechat")
if ok:
scan_data = safe_dict(data.get("data", data))
qr_url = scan_data.get("qr_url", "")
is_mock = scan_data.get("is_mock", False)
is_localhost = "localhost" in str(qr_url).lower() or "127.0.0.1" in str(qr_url)
record("5.2 QR URL非localhost", not is_localhost,
f"is_mock={is_mock}" + (" (含localhost!)" if is_localhost else ""),
"POST /channels/authorize/scan")
has_status = "status" in scan_data
record("5.3 返回status字段", has_status, scan_data.get("status", "missing"))
has_msg = "message" in scan_data
record("5.4 返回message字段", has_msg, scan_data.get("message", "missing")[:60] if has_msg else "missing")
resp = api_post(f"/channels/authorize/scan/status?platform=wechat", user_token)
assert_data_ok(resp, "5.5 查询扫描状态", "POST /channels/authorize/scan/status")
# ============================================================
# Phase 6: 问题6 - 平台驾驶舱统计数据API
# ============================================================
print("\n--- Phase 6: 问题6 - 驾驶舱统计API ---")
resp = api_get("/admin/dashboard/stats", admin_token)
ok, data = assert_data_ok(resp, "6.1 仪表盘统计数据", "GET /admin/dashboard/stats")
if ok:
stats = safe_dict(data.get("data", data))
for field, label in [("active_users","用户总数"), ("enabled_models","模型数"),
("today_orders","今日订单"), ("today_tokens","今日Token")]:
val = stats.get(field)
record(f"6.2 {field}统计字段", val is not None,
f"{label}={val}", "GET /admin/dashboard/stats")
resp = api_get("/admin/dashboard/recent-logs", admin_token)
ok, data = assert_data_ok(resp, "6.3 最近操作日志", "GET /admin/dashboard/recent-logs")
logs = safe_list(data)
record("6.4 操作日志数据", isinstance(logs, list), f"共{len(logs)}条")
resp = api_get("/admin/dashboard/recent-audit", admin_token)
ok, data = assert_data_ok(resp, "6.5 最近审计日志", "GET /admin/dashboard/recent-audit")
audit = safe_list(data)
record("6.6 审计日志数据", isinstance(audit, list), f"共{len(audit)}条")
# ============================================================
# Phase 7: 问题7 - 管理员密码修改API
# ============================================================
print("\n--- Phase 7: 问题7 - 密码修改API ---")
resp = api_put("/profile/password", admin_token, {
"old_password": "admin123",
"new_password": "AdminTest456"
})
ok, data = assert_data_ok(resp, "7.1 修改密码", "PUT /profile/password")
if ok:
try:
new_token = login("admin", "AdminTest456")
record("7.2 新密码登录", True, "新密码登录成功")
resp = api_put("/profile/password", new_token, {
"old_password": "AdminTest456",
"new_password": "admin123"
})
assert_data_ok(resp, "7.3 恢复原密码", "PUT /profile/password")
admin_token = login("admin", "admin123")
record("7.4 原密码恢复成功", True, "登录正常")
except Exception as e:
record("7.2 密码修改验证", False, str(e))
admin_token = login("admin", "admin123")
# 旧密码错误
resp = api_put("/profile/password", admin_token, {
"old_password": "WrongPasswordXYZ",
"new_password": "NewPass789"
})
if resp.status_code == 400 or resp.status_code == 401:
record("7.5 旧密码错误拒绝", True, f"HTTP {resp.status_code}")
elif resp.status_code == 200:
d = resp.json()
ok_pwd = d.get("code") != 0
record("7.5 旧密码错误拒绝", ok_pwd,
f"code={d.get('code')}" if not ok_pwd else "ACCEPTED WRONG PWD",
"PUT /profile/password")
else:
record("7.5 旧密码错误拒绝", True, f"HTTP {resp.status_code}")
# ============================================================
# Phase 8: 问题8 - 邮件配置保存和测试API
# ============================================================
print("\n--- Phase 8: 问题8 - 邮件配置API ---")
resp = api_get("/admin/settings/email", admin_token)
ok, data = assert_data_ok(resp, "8.1 获取邮件配置", "GET /admin/settings/email")
existing = safe_dict(data.get("data", data)) if ok else {}
resp = api_put("/admin/settings/email", admin_token, {
"smtp_host": "smtp.example.com",
"smtp_port": 587,
"username": "test@example.com",
"password": "testpass123",
"from_address": "noreply@example.com",
"use_tls": True
})
assert_data_ok(resp, "8.2 保存邮件配置", "PUT /admin/settings/email")
resp = api_get("/admin/settings/email", admin_token)
ok2, data2 = assert_data_ok(resp, "8.3 配置回显验证", "GET /admin/settings/email")
if ok2:
saved = safe_dict(data2.get("data", data2))
has_host = "smtp_host" in saved
record("8.4 回显smtp_host", has_host, saved.get("smtp_host", "missing"))
# 测试邮件(假SMTP,预期失败)
resp = api_post("/admin/settings/email/test", admin_token)
record("8.5 发送测试邮件(假SMTP)", resp.status_code in [200, 400, 500],
f"HTTP {resp.status_code} (预期失败)", "POST /admin/settings/email/test")
# 无配置时测试
api_put("/admin/settings/email", admin_token, {
"smtp_host": "", "smtp_port": 587, "username": "",
"password": "", "from_address": "", "use_tls": True
})
resp = api_post("/admin/settings/email/test", admin_token)
record("8.6 无配置测试邮件", resp.status_code == 400,
f"HTTP {resp.status_code} (预期400)", "POST /admin/settings/email/test")
# 恢复
if existing.get("smtp_host"):
api_put("/admin/settings/email", admin_token, existing)
# ============================================================
# Phase 9: 问题9 - 日志与风控API
# ============================================================
print("\n--- Phase 9: 问题9 - 日志与风控API ---")
resp = api_get("/admin/logs/stats", admin_token)
ok, data = assert_data_ok(resp, "9.1 日志统计", "GET /admin/logs/stats")
if ok:
logstats = safe_dict(data.get("data", data))
today = logstats.get("today", {})
for field in ["api_calls", "errors", "operation_logs", "audit_logs"]:
val = today.get(field)
record(f"9.2 {field}统计", val is not None, f"={val}", "GET /admin/logs/stats")
# 操作日志
resp = api_get("/admin/logs/operations", admin_token)
ok, data = assert_data_ok(resp, "9.3 操作日志列表", "GET /admin/logs/operations")
ops = safe_list(data)
record("9.4 操作日志数据", isinstance(ops, list), f"共{len(ops)}条")
if len(ops) > 0:
first = ops[0]
for field in ["username", "action", "target", "created_at"]:
has = field in first and first[field] is not None
record(f"9.5 操作日志含{field}", has, first.get(field, "N/A"))
# 审计日志
resp = api_get("/admin/logs/audit-logs", admin_token)
ok, data = assert_data_ok(resp, "9.6 审计日志列表", "GET /admin/logs/audit-logs")
audits = safe_list(data)
record("9.7 审计日志数据", isinstance(audits, list), f"共{len(audits)}条")
if len(audits) > 0:
first = audits[0]
for field in ["admin_name", "action", "target_type", "created_at"]:
has = field in first and first[field] is not None
record(f"9.8 审计日志含{field}", has, first.get(field, "N/A"))
# Token日志
resp = api_get("/admin/logs/token-logs", admin_token)
ok, data = assert_data_ok(resp, "9.9 Token日志", "GET /admin/logs/token-logs")
tokens = safe_list(data)
record("9.10 Token日志数据", isinstance(tokens, list), f"共{len(tokens)}条")
# 付费日志
resp = api_get("/admin/logs/payment-logs", admin_token)
ok, data = assert_data_ok(resp, "9.11 付费日志", "GET /admin/logs/payment-logs")
payments = safe_list(data)
record("9.12 付费日志数据", isinstance(payments, list), f"共{len(payments)}条")
# 分页筛选
resp = api_get("/admin/logs/operations", admin_token, {"page": 1, "page_size": 5})
ok, data = assert_data_ok(resp, "9.13 操作日志分页", "GET /admin/logs/operations?page=1&page_size=5")
if ok:
d = safe_dict(data.get("data", data))
record("9.14 分页格式", "total" in d or "items" in d,
f"total={d.get('total','?')}, items={len(d.get('items',[]))}")
# ============================================================
# Summary
# ============================================================
print("\n" + "=" * 60)
print("测试结果汇总")
print("=" * 60)
passed = sum(1 for r in results if r["status"] == "PASS")
failed = sum(1 for r in results if r["status"] == "FAIL")
total = len(results)
for r in results:
status_mark = "[OK]" if r["status"] == "PASS" else "[FAIL]"
print(f" {status_mark} {r['name']} ({r['detail']})")
print(f"\n总计: {passed}/{total} 通过, {failed} 失败, 通过率: {passed*100//total}%")
with open("/data/share/project/tests/fix-verification-api-results.json", "w") as f:
json.dump({
"summary": {"passed": passed, "failed": failed, "total": total,
"pass_rate": f"{passed*100//total}%"},
"results": results
}, f, ensure_ascii=False, indent=2)
print(f"JSON结果: /data/share/project/tests/fix-verification-api-results.json")
return passed, failed, total
if __name__ == "__main__":
main()