263 lines
9.0 KiB
Python
263 lines
9.0 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
cntoai 对话相关接口联调测试（dev.opencomputing.cn）
|
||
|
||
用法：
|
||
pip install requests
|
||
set CNTOAI_USERID=你的用户id
|
||
set CNTOAI_API_KEY=你的api_key
|
||
python test_chat.py
|
||
|
||
可选环境变量：
|
||
CNTOAI_BASE_URL 默认 https://dev.opencomputing.cn
|
||
CNTOAI_MODEL 默认 qwen3.6-plus
|
||
CNTOAI_LLM_API_URL 默认 https://ai.atvoe.com/llmage/v1/chat/completions
|
||
CNTOAI_COOKIE 浏览器 Cookie（未传 userid 时用于鉴权）
|
||
|
||
单测：
|
||
python test_chat.py --only models
|
||
python test_chat.py --only completions
|
||
python test_chat.py --only send
|
||
python test_chat.py --only session
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import os
|
||
import sys
|
||
import time
|
||
from typing import Any, Dict, Optional, Tuple
|
||
|
||
try:
|
||
import requests
|
||
except ImportError:
|
||
print("请先安装依赖: pip install requests")
|
||
sys.exit(1)
|
||
|
||
|
||
BASE_URL = "https://dev.opencomputing.cn"
|
||
USERID = "hSqZuekZ1yKmhKmCN9UAK"
|
||
API_KEY = "sk-c22d6573e85a4d3fa8ab932386cf2909"
|
||
# API_URL = "https://ai.atvoe.com/llmage/v1/chat/completions"
|
||
API_URL = "https://api.deepseek.com/chat/completions"
|
||
# MODEL = "qwen3.6-plus"
|
||
MODEL = "deepseek-v4-pro"
|
||
COOKIE = "".strip()
|
||
TIMEOUT = int(120)
|
||
|
||
|
||
class ChatApiClient:
|
||
def __init__(self, base_url: str = BASE_URL):
|
||
self.base_url = base_url.rstrip("/")
|
||
self.session = requests.Session()
|
||
self.session.headers.update({"Accept": "application/json"})
|
||
if COOKIE:
|
||
self.session.headers["Cookie"] = COOKIE
|
||
|
||
def _url(self, path: str) -> str:
|
||
path = path if path.startswith("/") else f"/{path}"
|
||
return f"{self.base_url}{path}"
|
||
|
||
def _auth(self, extra: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
|
||
params: Dict[str, Any] = {}
|
||
if USERID:
|
||
params["userid"] = USERID
|
||
if API_KEY:
|
||
params["api_key"] = API_KEY
|
||
if API_URL:
|
||
params["api_url"] = API_URL
|
||
if extra:
|
||
params.update(extra)
|
||
return params
|
||
|
||
def get(self, path: str, params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
|
||
resp = self.session.get(self._url(path), params=self._auth(params), timeout=TIMEOUT)
|
||
return self._parse(resp)
|
||
|
||
def post(self, path: str, data: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
|
||
resp = self.session.post(
|
||
self._url(path),
|
||
json=self._auth(data),
|
||
headers={"Content-Type": "application/json"},
|
||
timeout=TIMEOUT,
|
||
)
|
||
return self._parse(resp)
|
||
|
||
@staticmethod
|
||
def _parse(resp: requests.Response) -> Dict[str, Any]:
|
||
print(f" HTTP {resp.status_code} {resp.url[:120]}...")
|
||
try:
|
||
return resp.json()
|
||
except Exception:
|
||
return {"status": False, "msg": f"非 JSON 响应: {resp.text[:300]}"}
|
||
|
||
|
||
def ok(name: str, data: Dict[str, Any]) -> bool:
|
||
passed = data.get("status") is True
|
||
tag = "PASS" if passed else "FAIL"
|
||
print(f"\n[{tag}] {name}")
|
||
print(json.dumps(data, ensure_ascii=False, indent=2)[:2000])
|
||
if not passed:
|
||
print(f" -> {data.get('msg', 'unknown error')}")
|
||
return passed
|
||
|
||
|
||
def test_model_list(client: ChatApiClient) -> bool:
|
||
print("\n=== GET /cntoai/model_management_customer_search.dspy ===")
|
||
data = client.get("/cntoai/model_management_customer_search.dspy", {
|
||
"page_size": 20,
|
||
"current_page": 1,
|
||
})
|
||
if ok("模型列表", data) and data.get("data"):
|
||
models = data["data"].get("model_list") or []
|
||
print(f" 共 {len(models)} 个模型")
|
||
if models:
|
||
m0 = models[0]
|
||
print(f" 首个: {m0.get('model_name')} / {m0.get('display_name')}")
|
||
return data.get("status") is True
|
||
|
||
|
||
def test_llm_chat_completions(client: ChatApiClient) -> bool:
|
||
print("\n=== POST /cntoai/llm_chat_completions.dspy ===")
|
||
data = client.post("/cntoai/llm_chat_completions.dspy", {
|
||
"model": MODEL,
|
||
"message": "用一句话介绍你自己",
|
||
"stream": True,
|
||
})
|
||
if ok("直连模型", data) and data.get("data"):
|
||
print(f" 回复摘要: {(data['data'].get('reply') or '')[:200]}")
|
||
return data.get("status") is True
|
||
|
||
|
||
def test_chat_send(
|
||
client: ChatApiClient,
|
||
session_id: Optional[str] = None,
|
||
) -> Tuple[bool, Optional[str]]:
|
||
print("\n=== POST /cntoai/chat_send.dspy ===")
|
||
payload: Dict[str, Any] = {
|
||
"model": MODEL,
|
||
"message": (
|
||
"你好，这是 test_chat.py 自动化测试"
|
||
if not session_id
|
||
else "继续，用一句话回复我"
|
||
),
|
||
"stream": True,
|
||
}
|
||
if session_id:
|
||
payload["session_id"] = session_id
|
||
data = client.post("/cntoai/chat_send.dspy", payload)
|
||
if ok("发送消息", data) and data.get("data"):
|
||
sid = data["data"].get("session_id")
|
||
print(f" session_id: {sid}")
|
||
print(f" 回复摘要: {(data['data'].get('reply') or '')[:200]}")
|
||
return True, sid
|
||
return False, session_id
|
||
|
||
|
||
def test_chat_session_list(client: ChatApiClient) -> bool:
|
||
print("\n=== GET /cntoai/chat_session_list.dspy ===")
|
||
data = client.get("/cntoai/chat_session_list.dspy", {"page_size": 10})
|
||
if ok("会话列表", data) and data.get("data"):
|
||
sessions = data["data"].get("sessions") or []
|
||
print(f" 共 {data['data'].get('total_count', len(sessions))} 条会话")
|
||
for s in sessions[:3]:
|
||
print(f" - {s.get('id')} | {s.get('title')}")
|
||
return data.get("status") is True
|
||
|
||
|
||
def test_chat_session_messages(client: ChatApiClient, session_id: str) -> bool:
|
||
print("\n=== GET /cntoai/chat_session_messages.dspy ===")
|
||
data = client.get("/cntoai/chat_session_messages.dspy", {"session_id": session_id})
|
||
if ok("会话消息", data) and data.get("data"):
|
||
msgs = data["data"].get("messages") or []
|
||
print(f" 消息数: {len(msgs)}")
|
||
for m in msgs:
|
||
print(f" [{m.get('role')}] {str(m.get('content') or '')[:80]}")
|
||
return data.get("status") is True
|
||
|
||
|
||
def test_chat_session_delete(client: ChatApiClient, session_id: str) -> bool:
|
||
print("\n=== GET /cntoai/chat_session_delete.dspy ===")
|
||
data = client.get("/cntoai/chat_session_delete.dspy", {"session_id": session_id})
|
||
return ok("删除会话", data)
|
||
|
||
|
||
def check_config(require_userid: bool = True) -> bool:
|
||
print("配置:")
|
||
print(f" BASE_URL = {BASE_URL}")
|
||
print(f" MODEL = {MODEL}")
|
||
print(f" API_URL = {API_URL or '(走服务端配置)'}")
|
||
print(f" USERID = {USERID or '(未设置)'}")
|
||
print(f" API_KEY = {'已设置' if API_KEY else '(未设置)'}")
|
||
print(f" COOKIE = {'已设置' if COOKIE else '(未设置)'}")
|
||
|
||
if require_userid and not USERID and not COOKIE:
|
||
print("\n错误: 持久化接口需要 CNTOAI_USERID 或 CNTOAI_COOKIE")
|
||
return False
|
||
if not API_KEY:
|
||
print("\n警告: 未设置 CNTOAI_API_KEY，将依赖服务端 Key")
|
||
return True
|
||
|
||
|
||
def main() -> int:
|
||
parser = argparse.ArgumentParser(description="cntoai chat API 联调测试")
|
||
parser.add_argument(
|
||
"--only",
|
||
choices=["models", "completions", "send", "session", "delete", "all"],
|
||
default="all",
|
||
)
|
||
parser.add_argument("--keep-session", action="store_true", help="不删除测试会话")
|
||
parser.add_argument("--base-url", default=BASE_URL)
|
||
args = parser.parse_args()
|
||
|
||
client = ChatApiClient(base_url=args.base_url)
|
||
results = []
|
||
session_id: Optional[str] = None
|
||
|
||
if args.only in ("all", "models"):
|
||
results.append(("models", test_model_list(client)))
|
||
|
||
if args.only in ("all", "completions"):
|
||
if check_config(require_userid=False):
|
||
results.append(("completions", test_llm_chat_completions(client)))
|
||
else:
|
||
results.append(("completions", False))
|
||
|
||
if args.only in ("all", "send", "session", "delete"):
|
||
if not check_config(require_userid=True):
|
||
return 1
|
||
|
||
if args.only in ("all", "send"):
|
||
passed, session_id = test_chat_send(client)
|
||
results.append(("send_1", passed))
|
||
if passed and session_id:
|
||
time.sleep(1)
|
||
passed2, session_id = test_chat_send(client, session_id=session_id)
|
||
results.append(("send_2_multiturn", passed2))
|
||
|
||
if args.only in ("all", "session") and session_id:
|
||
results.append(("session_list", test_chat_session_list(client)))
|
||
results.append(("session_messages", test_chat_session_messages(client, session_id)))
|
||
elif args.only == "session":
|
||
results.append(("session_list", test_chat_session_list(client)))
|
||
|
||
if args.only in ("all", "delete") and session_id and not args.keep_session:
|
||
results.append(("delete", test_chat_session_delete(client, session_id)))
|
||
elif session_id and args.keep_session:
|
||
print(f"\n保留测试会话: {session_id}")
|
||
|
||
print("\n" + "=" * 50)
|
||
print("汇总:")
|
||
failed = sum(1 for _, p in results if not p)
|
||
for name, passed in results:
|
||
print(f" {'OK' if passed else 'FAIL'} {name}")
|
||
print("=" * 50)
|
||
return 1 if failed else 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|