Add more base chat cases (#3203)

* add test base class

* fix codestyle

* fix codestyle

* add base chat

test/ce/server/core/__init__.py

@@ -9,10 +9,16 @@ from .logger import Logger
 
 base_logger = Logger(loggername="FDSentry", save_level="channel", log_path="./fd_logs").get_logger()
 base_logger.setLevel("INFO")
-from .request_template import TEMPLATES
-from .utils import build_request_payload, get_stream_chunks, send_request
 
-__all__ = ["build_request_payload", "send_request", "TEMPLATES", "get_stream_chunks"]
+from .request_template import TEMPLATES
+from .utils import (
+    build_request_payload,
+    get_stream_chunks,
+    get_token_list,
+    send_request,
+)
+
+__all__ = ["build_request_payload", "send_request", "TEMPLATES", "get_stream_chunks", "get_token_list"]
 
 # 检查环境变量是否存在
 URL = os.environ.get("URL")
@@ -23,8 +29,8 @@ if not URL:
     missing_vars.append("URL")
 if not TEMPLATE:
     missing_vars.append("TEMPLATE")
-if missing_vars:
-    if not URL:
+
+if not URL:
     msg = (
         f"❌ 缺少环境变量：{', '.join(missing_vars)}，请先设置，例如：\n"
         f"   export URL=http://localhost:8000/v1/chat/completions\n"
@@ -32,5 +38,6 @@ if missing_vars:
     )
     base_logger.error(msg)
     sys.exit(33)  # 终止程序
-    if not TEMPLATE:
-        base_logger.warning("未启用请求模板，请在Case中自行设置请求模板")
+
+if not TEMPLATE:
+    base_logger.warning("⚠️ 未设置 TEMPLATE，请确保在用例中显式传入请求模板。")

test/ce/server/core/utils.py

@@ -37,10 +37,6 @@ def send_request(url, payload, timeout=600, stream=False):
 
     Returns:
         response: 请求的响应结果，如果请求失败则返回None。
-
-    Raises:
-        None
-
     """
     headers = {
         "Content-Type": "application/json",
@@ -60,7 +56,7 @@ def send_request(url, payload, timeout=600, stream=False):
 
 
 def get_stream_chunks(response):
-    """解析流式返回，生成chunk List[dict]"""
+    """解析流式返回，生成 chunk List[dict]"""
     chunks = []
 
     if response.status_code == 200:
@@ -82,3 +78,22 @@ def get_stream_chunks(response):
         base_logger.error("返回内容：", response.text)
 
     return chunks
+
+
+def get_token_list(response):
+    """解析 response 中的 token 文本列表"""
+    token_list = []
+
+    try:
+        content_logprobs = response["choices"][0]["logprobs"]["content"]
+    except (KeyError, IndexError, TypeError) as e:
+        base_logger.error(f"解析失败：{e}")
+        return []
+
+    for token_info in content_logprobs:
+        token = token_info.get("token")
+        if token is not None:
+            token_list.append(token)
+
+    base_logger.info(f"Token List:{token_list}")
+    return token_list

test/ce/server/test_base_chat.py

@@ -9,7 +9,7 @@ some basic check for fd web api
 
 import json
 
-from core import TEMPLATE, URL, build_request_payload, send_request
+from core import TEMPLATE, URL, build_request_payload, get_token_list, send_request
 
 
 def test_stream_response():
@@ -76,9 +76,12 @@ def test_logprobs_enabled():
 def test_stop_sequence():
     data = {
         "stream": False,
-        "stop": ["果冻"],
+        "stop": ["。"],
         "messages": [
-            {"role": "user", "content": "你要严格按照我接下来的话输出，输出冒号后面的内容，请输出：这是第一段。果冻这是第二段啦啦啦啦啦。"},
+            {
+                "role": "user",
+                "content": "你要严格按照我接下来的话输出，输出冒号后面的内容，请输出：这是第一段。这是第二段啦啦啦啦啦。",
+            },
         ],
         "max_tokens": 20,
         "top_p": 0,
@@ -86,8 +89,11 @@ def test_stop_sequence():
     payload = build_request_payload(TEMPLATE, data)
     resp = send_request(URL, payload).json()
     content = resp["choices"][0]["message"]["content"]
+    token_list = get_token_list(resp)
     print("截断输出:", content)
     assert "第二段" not in content
+    assert "第二段" not in token_list
+    assert "。" in token_list, "没有找到。符号"
 
 
 def test_sampling_parameters():
@@ -125,7 +131,7 @@ def test_multi_turn_conversation():
 
 
 def test_bad_words_filtering():
-    banned_tokens = ["和", "呀"]
+    banned_tokens = ["香"]
 
     data = {
         "stream": False,
@@ -140,36 +146,14 @@ def test_bad_words_filtering():
 
     payload = build_request_payload(TEMPLATE, data)
     response = send_request(URL, payload).json()
-
     content = response["choices"][0]["message"]["content"]
     print("生成内容:", content)
+    token_list = get_token_list(response)
 
     for word in banned_tokens:
-        assert word not in content, f"bad_word '{word}' 不应出现在生成结果中"
+        assert word not in token_list, f"bad_word '{word}' 不应出现在生成结果中"
 
-    print("test_bad_words_filtering 通过：生成结果未包含被禁词")
-
-    data = {
-        "stream": False,
-        "messages": [
-            {"role": "system", "content": "你是一个助手，回答简洁清楚"},
-            {"role": "user", "content": "请输出冒号后面的字，一模一样: 我爱吃果冻，苹果，香蕉，和荔枝呀呀呀"},
-        ],
-        "top_p": 0,
-        "max_tokens": 69,
-        # "bad_words": banned_tokens,
-    }
-
-    payload = build_request_payload(TEMPLATE, data)
-    response = send_request(URL, payload).json()
-
-    content = response["choices"][0]["message"]["content"]
-    print("生成内容:", content)
-
-    for word in banned_tokens:
-        assert word not in content, f"bad_word '{word}' 不应出现在生成结果中"
-
-    print("test_bad_words_filtering 通过：生成结果未包含被禁词")
+    print("test_bad_words_filtering 正例验证通过")
 
 
 def test_bad_words_filtering1():
@@ -195,8 +179,10 @@ def test_bad_words_filtering1():
     for word in banned_tokens:
         assert word not in content, f"bad_word '{word}' 不应出现在生成结果中"
 
-    print("test_bad_words_filtering 通过：生成结果未包含被禁词")
-    word = "呀呀"
+    print("test_bad_words_filtering1 通过：生成结果未包含被禁词")
+
+    # 正例验证
+    word = "呀"
     data = {
         "stream": False,
         "messages": [
@@ -212,7 +198,7 @@ def test_bad_words_filtering1():
 
     content = response["choices"][0]["message"]["content"]
     print("生成内容:", content)
+    token_list = get_token_list(response)
+    assert word in token_list, f"'{word}' 应出现在生成结果中"
 
-    assert word in content, f" '{word}' 应出现在生成结果中"
-
-    print("test_bad_words_filtering 通过：生成结果未包含被禁词")
+    print("test_bad_words_filtering1 正例验证通过")