#!/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())