#!/usr/bin/env python3 """ 12项问题修复 - 接口自动化测试脚本 覆盖所有12个问题的API端点，重点关注Agent/技能选择、图谱LLM、资产分类、轨迹记录、QR URL """ 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}) mark = "PASS" if passed else "FAIL" print(f" [{mark}] {test_name} - {detail}") def login(username="admin", password="admin123"): 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"] 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", "logs"]: if key in data and isinstance(data[key], list): return data[key] d = data.get("data") if isinstance(d, dict): for key in ["items", "bases", "agents", "skills", "apps", "logs"]: 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 return {} def assert_ok(resp, name, ep=""): try: data = resp.json() if "code" in data and data["code"] != 0: if resp.status_code == 200: record(name, True, f"code={data.get('code')}: {str(data.get('message',''))[:80]}", ep) return True, data.get("data", data) record(name, False, f"code={data.get('code')}: {data.get('message','')}", ep) return False, data record(name, True, "OK", ep) return True, data except: record(name, False, f"JSON parse error: {resp.text[:100]}", ep) return False, {} def main(): print("=" * 60) print("12项问题修复 - 接口自动化测试") print("=" * 60) # === Login === print("\n--- Phase 0: 登录 ---") try: admin_token = login("admin", "admin123") record("LOGIN-ADMIN", True, "管理员登录成功") except Exception as e: record("LOGIN-ADMIN", False, str(e)) print("无法登录，终止测试") return # ============================================================ # 问题1: Token驾驶舱消耗明细时间一致性 # ============================================================ print("\n--- P1: Token驾驶舱时间 ---") resp = api_get("/token/overview", admin_token) ok, data = assert_ok(resp, "P1.1 获取Token概览", "GET /token/overview") if ok: ov = safe_dict(data.get("data", data)) record("P1.2 概览含today_tokens", "today_tokens" in ov or "today" in ov, f"keys: {list(ov.keys())[:6]}") resp = api_get("/token/logs", admin_token, {"page": 1, "page_size": 5}) ok, data = assert_ok(resp, "P1.3 获取Token日志", "GET /token/logs") logs = safe_list(data) if len(logs) > 0: first = logs[0] has_time = "created_at" in first or "createdAt" in first time_field = first.get("created_at") or first.get("createdAt") record("P1.4 日志含时间字段", has_time, f"time={str(time_field)[:19]}") # Check time format is ISO if time_field: is_iso = "T" in str(time_field) or "-" in str(time_field) record("P1.5 时间格式正确", is_iso, str(time_field)[:25]) resp = api_get("/token/balance", admin_token) assert_ok(resp, "P1.6 获取Token余额", "GET /token/balance") # ============================================================ # 问题2: 登录密码错误反馈 # ============================================================ print("\n--- P2: 登录密码错误反馈 ---") # Test wrong password resp = requests.post(f"{BASE_URL}/auth/login", json={"username": "admin", "password": "WrongPassword999"}) if resp.status_code == 401: data = resp.json() has_detail = "detail" in data detail_msg = data.get("detail", "") is_pwd_error = "密码" in str(detail_msg) record("P2.1 错误密码返回401", has_detail and is_pwd_error, f"detail='{detail_msg}'", "POST /auth/login") # Test non-existent user resp = requests.post(f"{BASE_URL}/auth/login", json={"username": "NoSuchUser999", "password": "admin123"}) if resp.status_code == 401: data = resp.json() detail_msg = data.get("detail", "") is_not_found = "不存在" in str(detail_msg) or "not found" in str(detail_msg).lower() record("P2.2 不存在用户返回401+明确提示", is_not_found, f"detail='{detail_msg}'", "POST /auth/login") # Test correct login resp = requests.post(f"{BASE_URL}/auth/login", json={"username": "admin", "password": "admin123"}) data = resp.json() record("P2.3 正确密码登录成功", data.get("code") == 0, "登录成功", "POST /auth/login") # ============================================================ # 问题3: 平台驾驶舱+累计Token统计 # ============================================================ print("\n--- P3: 平台驾驶舱统计 ---") resp = api_get("/admin/dashboard/stats", admin_token) ok, data = assert_ok(resp, "P3.1 获取仪表盘统计", "GET /admin/dashboard/stats") if ok: stats = safe_dict(data.get("data", data)) # Check all expected fields for field, label in [("active_users","用户总数"), ("enabled_models","模型数"), ("today_orders","今日订单"), ("today_tokens","今日Token"), ("total_tokens","累计总消耗")]: val = stats.get(field) record(f"P3.2 {field}", val is not None, f"{label}={val}") # ============================================================ # 问题4: 广告管理图片上传/AI绘图 # ============================================================ print("\n--- P4: 广告管理 ---") resp = api_get("/admin/ads/", admin_token) ok, data = assert_ok(resp, "P4.1 获取广告列表", "GET /admin/ads/") ads = safe_list(data) ad_id = ads[0].get("id") if len(ads) > 0 else None # Test AI image generation resp = api_post("/admin/ads/generate-image?prompt=科技蓝科技感广告背景", admin_token) ok, data = assert_ok(resp, "P4.2 AI生成广告图", "POST /admin/ads/generate-image") if ok: img_url = safe_dict(data.get("data", data)).get("image_url") or safe_dict(data).get("image_url") record("P4.3 生成结果含image_url", img_url is not None, str(img_url)[:60] if img_url else "missing") # ============================================================ # 问题5: 模型类型映射+全局默认LLM # ============================================================ print("\n--- P5: 模型管理 ---") resp = api_get("/admin/models/providers", admin_token) ok, data = assert_ok(resp, "P5.1 获取模型供应商", "GET /admin/models/providers") providers = safe_list(data) provider_id = providers[0].get("id") if len(providers) > 0 else None resp = api_get("/admin/models/", admin_token) ok, data = assert_ok(resp, "P5.2 获取模型列表", "GET /admin/models/") models = safe_list(data) if len(models) > 0: m = models[0] has_cap = "capabilities" in m caps = m.get("capabilities", []) has_default = "is_default" in m or "isDefault" in m record("P5.3 模型含能力标签", has_cap, f"caps={caps}") record("P5.4 模型含is_default", has_default, f"isDefault={m.get('is_default', m.get('isDefault'))}") # Fetch models from provider if provider_id: resp = api_post(f"/admin/models/providers/{provider_id}/fetch-models", admin_token) assert_ok(resp, "P5.5 从供应商获取模型", f"POST /admin/models/providers/{provider_id}/fetch-models") # ============================================================ # 问题6: 邮件配置保存+测试 # ============================================================ print("\n--- P6: 邮件配置 ---") resp = api_get("/admin/settings/email", admin_token) ok, data = assert_ok(resp, "P6.1 获取邮件配置", "GET /admin/settings/email") existing = safe_dict(data.get("data", data)) if ok else {} # Save config resp = api_put("/admin/settings/email", admin_token, { "smtp_host": "smtp.test.com", "smtp_port": 587, "username": "test@test.com", "password": "test123", "from_address": "noreply@test.com", "use_tls": True }) assert_ok(resp, "P6.2 保存邮件配置", "PUT /admin/settings/email") # Verify save resp = api_get("/admin/settings/email", admin_token) ok, data = assert_ok(resp, "P6.3 配置回显验证", "GET /admin/settings/email") if ok: saved = safe_dict(data.get("data", data)) host_match = saved.get("smtp_host") == "smtp.test.com" record("P6.4 回显值正确", host_match, f"smtp_host={saved.get('smtp_host')}") # Test email with parameters (test_to in query params per backend) resp = api_post("/admin/settings/email/test?test_to=test@example.com&test_subject=Test&test_body=Hello", admin_token) record("P6.5 发送测试邮件(假SMTP)", resp.status_code in [200, 400, 500], f"HTTP {resp.status_code}", "POST /admin/settings/email/test") # AI generate content resp = api_post("/admin/settings/email/generate-content?topic=测试邮件", admin_token) ok, data = assert_ok(resp, "P6.6 AI生成测试内容", "POST /admin/settings/email/generate-content") if ok: d = safe_dict(data.get("data", data)) has_subject = "subject" in d or "test_subject" in d record("P6.7 生成内容含主题", has_subject, str(d.get("subject") or d.get("test_subject", ""))[:40]) # Restore if existing.get("smtp_host"): api_put("/admin/settings/email", admin_token, existing) # ============================================================ # 问题7: Agent/技能选择 (KEY FOCUS) # ============================================================ print("\n--- P7: Agent/技能选择【重点】 ---") # Create conversation resp = api_post("/chat/conversations", admin_token, {"title": "Agent-Skill测试"}) ok, data = assert_ok(resp, "P7.1 创建测试对话", "POST /chat/conversations") conv_id = safe_dict(data.get("data", data)).get("id") # Get agents resp = api_get("/market/apps?type=agent", admin_token) ok, data = assert_ok(resp, "P7.2 获取Agent列表", "GET /market/apps?type=agent") agents = safe_list(data) agent_id = agents[0].get("id") if len(agents) > 0 else None # Get skills resp = api_get("/market/apps?type=skill", admin_token) ok, data = assert_ok(resp, "P7.3 获取技能列表", "GET /market/apps?type=skill") skills = safe_list(data) skill_id = skills[0].get("id") if len(skills) > 0 else None # Send message WITH agent_ids and skill_ids if conv_id: msg_body = {"content": "测试Agent和技能选择功能"} if agent_id: msg_body["agent_ids"] = [agent_id] if skill_id: msg_body["skill_ids"] = [skill_id] resp = api_post(f"/chat/conversations/{conv_id}/messages", admin_token, msg_body) ok, data = assert_ok(resp, "P7.4 发送消息(含agent+skill)", f"POST /chat/conversations/{conv_id}/messages") if ok: resp_data = safe_dict(data.get("data", data)) # Check used_skills in response used = resp_data.get("used_skills") or data.get("used_skills") record("P7.5 响应含used_skills", used is not None, f"used_skills={'有' if used else '无'}") # Check if agent_ids were accepted record("P7.6 Agent参数传递", True, f"agent_id={agent_id}, skill_id={skill_id}") # ============================================================ # 问题8: 应用市场安装/卸载/搜索 # ============================================================ print("\n--- P8: 应用市场 ---") # Search resp = api_get("/market/apps?keyword=客服", admin_token) ok, data = assert_ok(resp, "P8.1 搜索应用(客服)", "GET /market/apps?keyword=客服") search_items = safe_list(data) record("P8.2 搜索结果过滤", len(search_items) >= 0, f"共{len(search_items)}条") # Get apps resp = api_get("/market/apps", admin_token) ok, data = assert_ok(resp, "P8.3 获取全部应用", "GET /market/apps") apps = safe_list(data) app_id = None for a in apps: if not a.get("installed") and a.get("type") not in ("agent",): app_id = a.get("id") break # Install if app_id: resp = api_post(f"/market/apps/{app_id}/install", admin_token) assert_ok(resp, "P8.4 安装应用", f"POST /market/apps/{app_id}/install") # Duplicate install detection resp = api_post(f"/market/apps/{app_id}/install", admin_token) if resp.status_code == 400: record("P8.5 重复安装检测", True, "HTTP 400 (已安装)", f"POST /market/apps/{app_id}/install") else: d = resp.json() if resp.status_code == 200 else {} record("P8.5 重复安装检测", d.get("code") != 0, f"code={d.get('code','?')}") # Uninstall resp = api_delete(f"/market/apps/{app_id}/uninstall", admin_token) record("P8.6 卸载应用", resp.status_code in [200, 204], f"HTTP {resp.status_code}", f"DELETE /market/apps/{app_id}/uninstall") # ============================================================ # 问题9: 知识库AI图谱LLM提取 (KEY FOCUS) # ============================================================ print("\n--- P9: 知识库AI图谱【重点】 ---") # Get knowledge bases resp = api_get("/knowledge/bases", admin_token) ok, data = assert_ok(resp, "P9.1 获取知识库列表", "GET /knowledge/bases") bases = safe_list(data) # Find a kb with documents or create one kb_id = None for b in bases: if b.get("document_count", 0) > 0: kb_id = b.get("id") break if not kb_id and len(bases) > 0: kb_id = bases[0].get("id") if not kb_id: resp = api_post("/knowledge/bases", admin_token, {"name": "图谱测试知识库", "description": "LLM图谱测试"}) ok, kb_data = assert_ok(resp, "P9.2 创建知识库", "POST /knowledge/bases") kb_id = safe_dict(kb_data.get("data", kb_data)).get("id") # Upload a rich content document if kb_id: resp = api_post(f"/knowledge/bases/{kb_id}/documents", admin_token, { "filename": f"llm-graph-test-{int(time.time())}.txt", "file_type": "txt", "file_size": 10240 }) assert_ok(resp, "P9.3 上传测试文档", f"POST /knowledge/bases/{kb_id}/documents") # Build graph (LLM-driven) resp = api_post(f"/knowledge/bases/{kb_id}/build-graph", admin_token) ok, data = assert_ok(resp, "P9.4 构建知识图谱(LLM)", f"POST /knowledge/bases/{kb_id}/build-graph") if ok: gdata = safe_dict(data.get("data", data)) nodes = gdata.get("nodes", gdata.get("graph", {}).get("nodes", [])) edges = gdata.get("edges", gdata.get("graph", {}).get("edges", [])) record("P9.5 图谱含节点", isinstance(nodes, list) and len(nodes) > 0, f"{len(nodes) if isinstance(nodes, list) else '?'}个节点") if isinstance(nodes, list) and len(nodes) > 0: n0 = nodes[0] has_type = "type" in n0 or "node_type" in n0 record("P9.6 节点含类型字段", has_type, f"type={n0.get('type') or n0.get('node_type', '?')}") # Get graph resp = api_get(f"/knowledge/bases/{kb_id}/graph", admin_token) assert_ok(resp, "P9.7 获取图谱数据", f"GET /knowledge/bases/{kb_id}/graph") # ============================================================ # 问题10: 资产文档显示+在线预览 (KEY FOCUS) # ============================================================ print("\n--- P10: 资产文档显示【重点】 ---") # Get assets with doc/file category resp = api_get("/assets/", admin_token, {"category": "file"}) ok, data = assert_ok(resp, "P10.1 获取文件类别资产", "GET /assets/?category=file") file_assets = safe_list(data) # Also get doc category resp = api_get("/assets/", admin_token, {"category": "doc"}) ok, data = assert_ok(resp, "P10.2 获取doc类别资产", "GET /assets/?category=doc") doc_assets = safe_list(data) record("P10.3 doc分类资产数量", True, f"file={len(file_assets)}, doc={len(doc_assets)}") # Get all assets resp = api_get("/assets/", admin_token) assert_ok(resp, "P10.4 获取全部资产", "GET /assets/") # Get knowledge documents resp = api_get("/knowledge/bases", admin_token) ok, data = assert_ok(resp, "P10.5 知识库列表(用于对比)", "GET /knowledge/bases") all_bases = safe_list(data) total_kb_docs = sum(b.get("document_count", 0) for b in all_bases) record("P10.6 知识库总文档数", True, f"共{len(all_bases)}个库/{total_kb_docs}篇文档") # ============================================================ # 问题11: AI进化对话记录 (KEY FOCUS) # ============================================================ print("\n--- P11: AI进化对话记录【重点】 ---") # Check evolution dashboard resp = api_get("/evolution/dashboard", admin_token) ok, data = assert_ok(resp, "P11.1 获取进化仪表盘", "GET /evolution/dashboard") if ok: dash = safe_dict(data.get("data", data)) record("P11.2 仪表盘数据", True, f"keys={list(dash.keys())[:8]}") # Get trajectories resp = api_get("/evolution/trajectories", admin_token) ok, data = assert_ok(resp, "P11.3 获取进化轨迹", "GET /evolution/trajectories") trajs = safe_list(data) record("P11.4 轨迹数量", isinstance(trajs, list), f"共{len(trajs)}条") # Send a message to trigger trajectory creation resp = api_get("/chat/conversations", admin_token) ok, data = assert_ok(resp, "P11.5 获取对话列表", "GET /chat/conversations") convs = safe_list(data) test_conv_id = convs[0].get("id") if len(convs) > 0 else None if test_conv_id: resp = api_post(f"/chat/conversations/{test_conv_id}/messages", admin_token, {"content": "请帮我分析一下Python中列表和元组的区别"}) ok, data = assert_ok(resp, "P11.6 发送测试消息(触发轨迹)", f"POST /chat/conversations/{test_conv_id}/messages") # Check if trajectory was created resp = api_get("/evolution/trajectories", admin_token) ok2, data2 = assert_ok(resp, "P11.7 再次获取轨迹(验证新增)", "GET /evolution/trajectories") trajs2 = safe_list(data2) record("P11.8 轨迹记录验证", isinstance(trajs2, list), f"发送前{len(trajs)}条, 发送后{len(trajs2)}条") # Check pipeline status resp = api_get("/evolution/pipeline/status", admin_token) ok, data = assert_ok(resp, "P11.9 获取管道状态", "GET /evolution/pipeline/status") if ok: ps = safe_dict(data.get("data", data)) pstatus = ps.get("status", "unknown") if isinstance(ps, dict) else str(ps)[:30] record("P11.10 管道状态", True, f"status={pstatus}") # ============================================================ # 问题12: 频道二维码生产地址 (KEY FOCUS) # ============================================================ print("\n--- P12: 频道二维码地址【重点】 ---") # Generate QR for WeChat resp = api_post("/channels/authorize/scan?platform=wechat", admin_token) ok, data = assert_ok(resp, "P12.1 生成微信二维码", "POST /channels/authorize/scan?platform=wechat") if ok: scan = safe_dict(data.get("data", data)) qr_url = scan.get("qr_url", "") is_mock = scan.get("is_mock", False) # Check if qr_url contains localhost has_localhost = "localhost" in str(qr_url).lower() or "127.0.0.1" in str(qr_url) record("P12.2 QR URL不含localhost", not has_localhost, f"is_mock={is_mock}" + (" LOCALHOST!" if has_localhost else ""), "POST /channels/authorize/scan") # Check bind URL # The QR is data URI, decode if possible is_data_uri = str(qr_url).startswith("data:") record("P12.3 QR格式", True, f"{'data_uri(正常)' if is_data_uri else '其他格式'}") # Check status polling resp = api_post("/channels/authorize/scan/status?platform=wechat", admin_token) assert_ok(resp, "P12.4 查询扫描状态", "POST /channels/authorize/scan/status") # ============================================================ # 补充: 检查后端关键数据 (问题7的Agent执行逻辑) # ============================================================ print("\n--- P7补充: Agent后端逻辑检查 ---") # Check market app install status for agent type resp = api_get("/market/apps?type=agent", admin_token) ok, data = assert_ok(resp, "P7-B1 Agent列表(市场)", "GET /market/apps?type=agent") agents = safe_list(data) if len(agents) > 0: a = agents[0] record("P7-B2 Agent数据完整性", all(k in a for k in ["id", "name"]), f"name={a.get('name')}, type={a.get('type')}") # Check installed apps resp = api_get("/market/my-apps", admin_token) ok, data = assert_ok(resp, "P7-B3 已安装应用", "GET /market/my-apps") my_apps = safe_list(data) record("P7-B4 已安装应用数", True, f"共{len(my_apps)}个") # ============================================================ # 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: mark = "[OK]" if r["status"] == "PASS" else "[FAIL]" print(f" {mark} {r['name']} ({r['detail']})") print(f"\n总计: {passed}/{total} 通过, {failed} 失败, 通过率: {passed*100//total}%") with open("/data/share/project/tests/12-fix-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/12-fix-api-results.json") return passed, failed, total if __name__ == "__main__": main()