feat: 抽取 scripts/core.py 公共模块，重构检测脚本
将 anthropic_detect.py 和 openai_detect.py 中的公共功能抽取到 core.py 模块，包括： - HTTP 请求（普通/流式）及重试逻辑 - SSL 上下文管理 - 测试用例/结果数据结构 (TestCase, TestResult) - 错误分类 (ErrorType) - 响应验证辅助函数 (validate_response_structure 等) - 测试执行框架 (run_test, run_test_suite) 两个检测脚本重构后更聚焦于各自 API 的测试用例定义。
This commit is contained in:
File diff suppressed because it is too large
Load Diff
471
scripts/core.py
Normal file
471
scripts/core.py
Normal file
@@ -0,0 +1,471 @@
|
||||
#!/usr/bin/env python3
|
||||
"""兼容性测试脚本的核心公共函数
|
||||
|
||||
提供 HTTP 请求、SSL 上下文、JSON 格式化、验证辅助等通用功能。
|
||||
"""
|
||||
|
||||
import json
|
||||
import time
|
||||
import ssl
|
||||
import urllib.request
|
||||
import urllib.error
|
||||
from dataclasses import dataclass
|
||||
from typing import Optional, Dict, Any, Tuple, List, Union, Type
|
||||
from enum import Enum
|
||||
|
||||
TIMEOUT = 30
|
||||
MAX_RETRIES = 2 # 最大重试次数
|
||||
|
||||
|
||||
class ErrorType(Enum):
|
||||
"""错误类型分类"""
|
||||
NETWORK = "network" # 网络错误
|
||||
CLIENT = "client" # 4xx 错误
|
||||
SERVER = "server" # 5xx 错误
|
||||
SUCCESS = "success" # 成功
|
||||
|
||||
|
||||
@dataclass
|
||||
class TestCase:
|
||||
"""测试用例数据结构"""
|
||||
desc: str # 测试描述
|
||||
method: str # HTTP 方法
|
||||
url: str # 请求 URL
|
||||
headers: Dict[str, str] # 请求头
|
||||
body: Optional[Any] = None # 请求体
|
||||
stream: bool = False # 是否流式请求
|
||||
validator: Optional[Any] = None # 响应验证函数（可选）
|
||||
|
||||
|
||||
@dataclass
|
||||
class TestResult:
|
||||
"""测试结果数据结构"""
|
||||
status: Optional[int] # HTTP 状态码
|
||||
elapsed: float # 耗时（秒）
|
||||
error_type: ErrorType # 错误类型
|
||||
response: str # 响应内容
|
||||
|
||||
|
||||
def create_ssl_context() -> ssl.SSLContext:
|
||||
"""创建不验证证书的 SSL 上下文（用于测试环境）"""
|
||||
ctx = ssl.create_default_context()
|
||||
ctx.check_hostname = False
|
||||
ctx.verify_mode = ssl.CERT_NONE
|
||||
return ctx
|
||||
|
||||
|
||||
def classify_error(status: Optional[int]) -> ErrorType:
|
||||
"""根据状态码分类错误类型"""
|
||||
if status is None:
|
||||
return ErrorType.NETWORK
|
||||
if 200 <= status < 300:
|
||||
return ErrorType.SUCCESS
|
||||
if 400 <= status < 500:
|
||||
return ErrorType.CLIENT
|
||||
if status >= 500:
|
||||
return ErrorType.SERVER
|
||||
return ErrorType.NETWORK
|
||||
|
||||
|
||||
def http_request(
|
||||
url: str,
|
||||
method: str = "GET",
|
||||
headers: Optional[Dict[str, str]] = None,
|
||||
body: Optional[Any] = None,
|
||||
ssl_ctx: Optional[ssl.SSLContext] = None,
|
||||
retries: int = MAX_RETRIES
|
||||
) -> TestResult:
|
||||
"""执行普通 HTTP 请求（支持重试）
|
||||
|
||||
Args:
|
||||
url: 请求 URL
|
||||
method: HTTP 方法 (GET/POST/PUT/DELETE)
|
||||
headers: 请求头字典
|
||||
body: 请求体 (dict 或 str)
|
||||
ssl_ctx: SSL 上下文
|
||||
retries: 重试次数
|
||||
|
||||
Returns:
|
||||
TestResult 对象
|
||||
"""
|
||||
req = urllib.request.Request(url, method=method)
|
||||
if headers:
|
||||
for k, v in headers.items():
|
||||
req.add_header(k, v)
|
||||
if body is not None:
|
||||
if isinstance(body, str):
|
||||
req.data = body.encode("utf-8")
|
||||
else:
|
||||
req.data = json.dumps(body).encode("utf-8")
|
||||
|
||||
start = time.time()
|
||||
last_error = None
|
||||
|
||||
for attempt in range(retries + 1):
|
||||
try:
|
||||
resp = urllib.request.urlopen(req, timeout=TIMEOUT, context=ssl_ctx)
|
||||
elapsed = time.time() - start
|
||||
status = resp.getcode()
|
||||
return TestResult(
|
||||
status=status,
|
||||
elapsed=elapsed,
|
||||
error_type=classify_error(status),
|
||||
response=resp.read().decode("utf-8")
|
||||
)
|
||||
except urllib.error.HTTPError as e:
|
||||
elapsed = time.time() - start
|
||||
return TestResult(
|
||||
status=e.code,
|
||||
elapsed=elapsed,
|
||||
error_type=classify_error(e.code),
|
||||
response=e.read().decode("utf-8")
|
||||
)
|
||||
except Exception as e:
|
||||
last_error = str(e)
|
||||
if attempt < retries:
|
||||
time.sleep(0.5 * (attempt + 1)) # 递增延迟
|
||||
continue
|
||||
|
||||
elapsed = time.time() - start
|
||||
return TestResult(
|
||||
status=None,
|
||||
elapsed=elapsed,
|
||||
error_type=ErrorType.NETWORK,
|
||||
response=last_error or "Unknown error"
|
||||
)
|
||||
|
||||
|
||||
def http_stream_request(
|
||||
url: str,
|
||||
headers: Optional[Dict[str, str]] = None,
|
||||
body: Optional[Any] = None,
|
||||
ssl_ctx: Optional[ssl.SSLContext] = None,
|
||||
retries: int = MAX_RETRIES
|
||||
) -> TestResult:
|
||||
"""执行流式 HTTP 请求 (SSE，支持重试)
|
||||
|
||||
Args:
|
||||
url: 请求 URL
|
||||
headers: 请求头字典
|
||||
body: 请求体 (dict)
|
||||
ssl_ctx: SSL 上下文
|
||||
retries: 重试次数
|
||||
|
||||
Returns:
|
||||
TestResult 对象
|
||||
"""
|
||||
req = urllib.request.Request(url, method="POST")
|
||||
if headers:
|
||||
for k, v in headers.items():
|
||||
req.add_header(k, v)
|
||||
if body is not None:
|
||||
req.data = json.dumps(body).encode("utf-8")
|
||||
|
||||
start = time.time()
|
||||
last_error = None
|
||||
|
||||
for attempt in range(retries + 1):
|
||||
try:
|
||||
resp = urllib.request.urlopen(req, timeout=TIMEOUT, context=ssl_ctx)
|
||||
status = resp.getcode()
|
||||
lines = []
|
||||
for raw_line in resp:
|
||||
line = raw_line.decode("utf-8").rstrip("\n\r")
|
||||
if line:
|
||||
lines.append(line)
|
||||
elapsed = time.time() - start
|
||||
return TestResult(
|
||||
status=status,
|
||||
elapsed=elapsed,
|
||||
error_type=classify_error(status),
|
||||
response="\n".join(lines)
|
||||
)
|
||||
except urllib.error.HTTPError as e:
|
||||
elapsed = time.time() - start
|
||||
return TestResult(
|
||||
status=e.code,
|
||||
elapsed=elapsed,
|
||||
error_type=classify_error(e.code),
|
||||
response=e.read().decode("utf-8")
|
||||
)
|
||||
except Exception as e:
|
||||
last_error = str(e)
|
||||
if attempt < retries:
|
||||
time.sleep(0.5 * (attempt + 1))
|
||||
continue
|
||||
|
||||
elapsed = time.time() - start
|
||||
return TestResult(
|
||||
status=None,
|
||||
elapsed=elapsed,
|
||||
error_type=ErrorType.NETWORK,
|
||||
response=last_error or "Unknown error"
|
||||
)
|
||||
|
||||
|
||||
def format_json(text: str) -> str:
|
||||
"""格式化 JSON 文本（用于美化输出）
|
||||
|
||||
Args:
|
||||
text: JSON 字符串或任意文本
|
||||
|
||||
Returns:
|
||||
格式化后的 JSON 字符串，或原文本（如果不是有效 JSON）
|
||||
"""
|
||||
try:
|
||||
parsed = json.loads(text)
|
||||
return json.dumps(parsed, ensure_ascii=False, indent=2)
|
||||
except (json.JSONDecodeError, TypeError):
|
||||
return text
|
||||
|
||||
|
||||
def run_test(
|
||||
index: int,
|
||||
total: int,
|
||||
test_case: TestCase,
|
||||
ssl_ctx: ssl.SSLContext
|
||||
) -> TestResult:
|
||||
"""执行单个测试用例并打印结果
|
||||
|
||||
Args:
|
||||
index: 测试序号
|
||||
total: 总测试数
|
||||
test_case: 测试用例对象
|
||||
ssl_ctx: SSL 上下文
|
||||
|
||||
Returns:
|
||||
TestResult 对象
|
||||
"""
|
||||
print(f"\n[{index}/{total}] {test_case.desc}")
|
||||
print(f">>> {test_case.method} {test_case.url}")
|
||||
if test_case.body is not None:
|
||||
if isinstance(test_case.body, str):
|
||||
print(test_case.body)
|
||||
else:
|
||||
print(format_json(json.dumps(test_case.body, ensure_ascii=False)))
|
||||
|
||||
if test_case.stream:
|
||||
result = http_stream_request(
|
||||
test_case.url,
|
||||
test_case.headers,
|
||||
test_case.body,
|
||||
ssl_ctx
|
||||
)
|
||||
else:
|
||||
result = http_request(
|
||||
test_case.url,
|
||||
test_case.method,
|
||||
test_case.headers,
|
||||
test_case.body,
|
||||
ssl_ctx
|
||||
)
|
||||
|
||||
if result.status is not None:
|
||||
print(f"状态码: {result.status} | 耗时: {result.elapsed:.2f}s")
|
||||
else:
|
||||
print(f"请求失败 | 耗时: {result.elapsed:.2f}s")
|
||||
|
||||
if test_case.stream and result.status and result.status < 300:
|
||||
# 流式响应按 SSE 行逐行输出
|
||||
for line in result.response.split("\n"):
|
||||
print(line)
|
||||
else:
|
||||
print(format_json(result.response))
|
||||
|
||||
# 执行响应验证
|
||||
if test_case.validator and result.status and 200 <= result.status < 300:
|
||||
is_valid, errors = test_case.validator(result.response)
|
||||
if is_valid:
|
||||
print("✓ 响应验证通过")
|
||||
else:
|
||||
print("✗ 响应验证失败:")
|
||||
for error in errors:
|
||||
print(f" - {error}")
|
||||
|
||||
return result
|
||||
|
||||
|
||||
def run_test_suite(
|
||||
cases: List[TestCase],
|
||||
ssl_ctx: ssl.SSLContext,
|
||||
title: str,
|
||||
base_url: str,
|
||||
model: str,
|
||||
flags: Optional[List[str]] = None
|
||||
) -> Tuple[int, int, int, int]:
|
||||
"""执行测试套件并打印总结
|
||||
|
||||
Args:
|
||||
cases: 测试用例列表
|
||||
ssl_ctx: SSL 上下文
|
||||
title: 测试标题
|
||||
base_url: API 基础地址
|
||||
model: 模型名称
|
||||
flags: 扩展测试标记列表
|
||||
|
||||
Returns:
|
||||
(总数, 成功数, 客户端错误数, 服务端错误数)
|
||||
"""
|
||||
total = len(cases)
|
||||
count_success = 0
|
||||
count_client_error = 0
|
||||
count_server_error = 0
|
||||
count_network_error = 0
|
||||
|
||||
print("=" * 60)
|
||||
print(title)
|
||||
print(f"目标: {base_url}")
|
||||
print(f"模型: {model}")
|
||||
print(f"时间: {time.strftime('%Y-%m-%d %H:%M:%S')}")
|
||||
if flags:
|
||||
print(f"用例: {total} 个 | 扩展: {', '.join(flags)}")
|
||||
else:
|
||||
print(f"用例: {total} 个")
|
||||
print("=" * 60)
|
||||
|
||||
for i, test_case in enumerate(cases, 1):
|
||||
result = run_test(i, total, test_case, ssl_ctx)
|
||||
|
||||
if result.error_type == ErrorType.SUCCESS:
|
||||
count_success += 1
|
||||
elif result.error_type == ErrorType.CLIENT:
|
||||
count_client_error += 1
|
||||
elif result.error_type == ErrorType.SERVER:
|
||||
count_server_error += 1
|
||||
else:
|
||||
count_network_error += 1
|
||||
|
||||
print()
|
||||
print("=" * 60)
|
||||
print(f"测试完成 | 总计: {total} | 成功: {count_success} | "
|
||||
f"客户端错误: {count_client_error} | 服务端错误: {count_server_error} | "
|
||||
f"网络错误: {count_network_error}")
|
||||
print("=" * 60)
|
||||
|
||||
return total, count_success, count_client_error, count_server_error
|
||||
|
||||
|
||||
# ==================== 通用验证辅助函数 ====================
|
||||
|
||||
def check_required_fields(data: Dict[str, Any], required_fields: List[str]) -> Tuple[bool, List[str]]:
|
||||
"""检查必需字段是否存在
|
||||
|
||||
Args:
|
||||
data: 待检查的数据字典
|
||||
required_fields: 必需字段列表
|
||||
|
||||
Returns:
|
||||
(是否全部存在, 缺失字段列表)
|
||||
"""
|
||||
missing = []
|
||||
for field in required_fields:
|
||||
if field not in data:
|
||||
missing.append(field)
|
||||
return len(missing) == 0, missing
|
||||
|
||||
|
||||
def check_field_type(value: Any, expected_type: Union[Type, tuple]) -> bool:
|
||||
"""检查字段类型是否正确
|
||||
|
||||
Args:
|
||||
value: 待检查的值
|
||||
expected_type: 期望的类型（可以是类型元组）
|
||||
|
||||
Returns:
|
||||
类型是否匹配
|
||||
"""
|
||||
if value is None:
|
||||
return True # None值通常表示可选字段，允许
|
||||
return isinstance(value, expected_type)
|
||||
|
||||
|
||||
def check_enum_value(value: Any, allowed_values: List[Any]) -> bool:
|
||||
"""检查值是否在允许的枚举值列表中
|
||||
|
||||
Args:
|
||||
value: 待检查的值
|
||||
allowed_values: 允许的值列表
|
||||
|
||||
Returns:
|
||||
值是否合法
|
||||
"""
|
||||
if value is None:
|
||||
return True # None值通常表示可选字段，允许
|
||||
return value in allowed_values
|
||||
|
||||
|
||||
def check_array_items_type(arr: List[Any], expected_item_type: Union[Type, tuple]) -> bool:
|
||||
"""检查数组中所有元素的类型
|
||||
|
||||
Args:
|
||||
arr: 待检查的数组
|
||||
expected_item_type: 期望的元素类型
|
||||
|
||||
Returns:
|
||||
所有元素类型是否匹配
|
||||
"""
|
||||
if not isinstance(arr, list):
|
||||
return False
|
||||
return all(check_field_type(item, expected_item_type) for item in arr)
|
||||
|
||||
|
||||
def format_validation_errors(errors: List[str]) -> str:
|
||||
"""格式化验证错误信息
|
||||
|
||||
Args:
|
||||
errors: 错误信息列表
|
||||
|
||||
Returns:
|
||||
格式化后的错误字符串
|
||||
"""
|
||||
if not errors:
|
||||
return "验证通过"
|
||||
return "验证失败:\n - " + "\n - ".join(errors)
|
||||
|
||||
|
||||
def validate_response_structure(
|
||||
response_text: str,
|
||||
required_fields: List[str],
|
||||
field_types: Optional[Dict[str, Union[Type, tuple]]] = None,
|
||||
enum_values: Optional[Dict[str, List[Any]]] = None
|
||||
) -> Tuple[bool, List[str]]:
|
||||
"""验证响应结构（通用验证函数）
|
||||
|
||||
Args:
|
||||
response_text: 响应文本
|
||||
required_fields: 必需字段列表
|
||||
field_types: 字段类型映射 {字段名: 期望类型}
|
||||
enum_values: 枚举值映射 {字段名: 允许值列表}
|
||||
|
||||
Returns:
|
||||
(是否验证通过, 错误信息列表)
|
||||
"""
|
||||
errors = []
|
||||
|
||||
# 尝试解析JSON
|
||||
try:
|
||||
data = json.loads(response_text)
|
||||
except json.JSONDecodeError as e:
|
||||
errors.append(f"响应不是有效的JSON: {e}")
|
||||
return False, errors
|
||||
|
||||
# 检查必需字段
|
||||
has_required, missing = check_required_fields(data, required_fields)
|
||||
if not has_required:
|
||||
errors.append(f"缺少必需字段: {', '.join(missing)}")
|
||||
|
||||
# 检查字段类型
|
||||
if field_types:
|
||||
for field, expected_type in field_types.items():
|
||||
if field in data and not check_field_type(data[field], expected_type):
|
||||
actual_type = type(data[field]).__name__
|
||||
expected_name = expected_type.__name__ if isinstance(expected_type, type) else str(expected_type)
|
||||
errors.append(f"字段 '{field}' 类型错误: 期望 {expected_name}, 实际 {actual_type}")
|
||||
|
||||
# 检查枚举值
|
||||
if enum_values:
|
||||
for field, allowed in enum_values.items():
|
||||
if field in data and not check_enum_value(data[field], allowed):
|
||||
errors.append(f"字段 '{field}' 值非法: {data[field]}, 允许值: {allowed}")
|
||||
|
||||
return len(errors) == 0, errors
|
||||
@@ -11,115 +11,157 @@
|
||||
"""
|
||||
|
||||
import json
|
||||
import time
|
||||
import ssl
|
||||
import argparse
|
||||
import urllib.request
|
||||
import urllib.error
|
||||
|
||||
TIMEOUT = 30
|
||||
from typing import Dict, List, Tuple, Any
|
||||
from core import (
|
||||
create_ssl_context,
|
||||
TestCase,
|
||||
run_test_suite,
|
||||
validate_response_structure,
|
||||
format_validation_errors
|
||||
)
|
||||
|
||||
|
||||
def create_ssl_context():
|
||||
ctx = ssl.create_default_context()
|
||||
ctx.check_hostname = False
|
||||
ctx.verify_mode = ssl.CERT_NONE
|
||||
return ctx
|
||||
|
||||
|
||||
def http_request(url, method="GET", headers=None, body=None, ssl_ctx=None):
|
||||
req = urllib.request.Request(url, method=method)
|
||||
if headers:
|
||||
for k, v in headers.items():
|
||||
req.add_header(k, v)
|
||||
if body is not None:
|
||||
if isinstance(body, str):
|
||||
req.data = body.encode("utf-8")
|
||||
else:
|
||||
req.data = json.dumps(body).encode("utf-8")
|
||||
|
||||
start = time.time()
|
||||
try:
|
||||
resp = urllib.request.urlopen(req, timeout=TIMEOUT, context=ssl_ctx)
|
||||
elapsed = time.time() - start
|
||||
return resp.getcode(), resp.read().decode("utf-8"), elapsed
|
||||
except urllib.error.HTTPError as e:
|
||||
elapsed = time.time() - start
|
||||
return e.code, e.read().decode("utf-8"), elapsed
|
||||
except Exception as e:
|
||||
elapsed = time.time() - start
|
||||
return None, str(e), elapsed
|
||||
|
||||
|
||||
def http_stream_request(url, headers=None, body=None, ssl_ctx=None):
|
||||
req = urllib.request.Request(url, method="POST")
|
||||
if headers:
|
||||
for k, v in headers.items():
|
||||
req.add_header(k, v)
|
||||
if body is not None:
|
||||
req.data = json.dumps(body).encode("utf-8")
|
||||
|
||||
start = time.time()
|
||||
try:
|
||||
resp = urllib.request.urlopen(req, timeout=TIMEOUT, context=ssl_ctx)
|
||||
status = resp.getcode()
|
||||
lines = []
|
||||
for raw_line in resp:
|
||||
line = raw_line.decode("utf-8").rstrip("\n\r")
|
||||
if line:
|
||||
lines.append(line)
|
||||
elapsed = time.time() - start
|
||||
return status, "\n".join(lines), elapsed
|
||||
except urllib.error.HTTPError as e:
|
||||
elapsed = time.time() - start
|
||||
return e.code, e.read().decode("utf-8"), elapsed
|
||||
except Exception as e:
|
||||
elapsed = time.time() - start
|
||||
return None, str(e), elapsed
|
||||
|
||||
|
||||
def format_json(text):
|
||||
try:
|
||||
parsed = json.loads(text)
|
||||
return json.dumps(parsed, ensure_ascii=False, indent=2)
|
||||
except (json.JSONDecodeError, TypeError):
|
||||
return text
|
||||
|
||||
|
||||
def build_headers(api_key):
|
||||
def build_headers(api_key: str) -> Dict[str, str]:
|
||||
"""构建 OpenAI API 请求头"""
|
||||
h = {"Content-Type": "application/json"}
|
||||
if api_key:
|
||||
h["Authorization"] = f"Bearer {api_key}"
|
||||
return h
|
||||
|
||||
|
||||
def run_test(index, total, desc, url, method, headers, body, stream, ssl_ctx):
|
||||
print(f"\n[{index}/{total}] {desc}")
|
||||
print(f">>> {method} {url}")
|
||||
if body is not None:
|
||||
if isinstance(body, str):
|
||||
print(body)
|
||||
# ==================== OpenAI 响应验证函数 ====================
|
||||
|
||||
def validate_openai_models_list_response(response_text: str) -> Tuple[bool, List[str]]:
|
||||
"""验证 OpenAI Models List 响应
|
||||
|
||||
根据API文档，响应应包含：
|
||||
- object: "list"
|
||||
- data: array of Model objects
|
||||
"""
|
||||
errors = []
|
||||
|
||||
try:
|
||||
data = json.loads(response_text)
|
||||
except json.JSONDecodeError as e:
|
||||
return False, [f"响应不是有效的JSON: {e}"]
|
||||
|
||||
# 检查必需字段
|
||||
if "object" not in data:
|
||||
errors.append("缺少必需字段: object")
|
||||
elif data["object"] != "list":
|
||||
errors.append(f"字段 'object' 值错误: 期望 'list', 实际 '{data['object']}'")
|
||||
|
||||
if "data" not in data:
|
||||
errors.append("缺少必需字段: data")
|
||||
elif not isinstance(data["data"], list):
|
||||
errors.append(f"字段 'data' 类型错误: 期望 list, 实际 {type(data['data']).__name__}")
|
||||
else:
|
||||
# 验证每个 model 对象
|
||||
for i, model in enumerate(data["data"]):
|
||||
if not isinstance(model, dict):
|
||||
errors.append(f"data[{i}] 不是对象")
|
||||
continue
|
||||
|
||||
# 检查 model 对象的必需字段
|
||||
model_required = ["id", "object", "created", "owned_by"]
|
||||
for field in model_required:
|
||||
if field not in model:
|
||||
errors.append(f"data[{i}] 缺少必需字段: {field}")
|
||||
|
||||
# 检查 object 字段值
|
||||
if "object" in model and model["object"] != "model":
|
||||
errors.append(f"data[{i}].object 值错误: 期望 'model', 实际 '{model['object']}'")
|
||||
|
||||
return len(errors) == 0, errors
|
||||
|
||||
|
||||
def validate_openai_model_retrieve_response(response_text: str) -> Tuple[bool, List[str]]:
|
||||
"""验证 OpenAI Model Retrieve 响应
|
||||
|
||||
根据API文档，响应应包含：
|
||||
- id: string
|
||||
- object: "model"
|
||||
- created: number
|
||||
- owned_by: string
|
||||
"""
|
||||
required_fields = ["id", "object", "created", "owned_by"]
|
||||
field_types = {
|
||||
"id": str,
|
||||
"object": str,
|
||||
"created": (int, float),
|
||||
"owned_by": str
|
||||
}
|
||||
enum_values = {
|
||||
"object": ["model"]
|
||||
}
|
||||
|
||||
return validate_response_structure(response_text, required_fields, field_types, enum_values)
|
||||
|
||||
|
||||
def validate_openai_chat_completion_response(response_text: str) -> Tuple[bool, List[str]]:
|
||||
"""验证 OpenAI Chat Completion 响应
|
||||
|
||||
根据API文档，响应应包含：
|
||||
- id: string
|
||||
- object: "chat.completion"
|
||||
- created: number
|
||||
- model: string
|
||||
- choices: array
|
||||
- usage: object (可选)
|
||||
"""
|
||||
errors = []
|
||||
|
||||
try:
|
||||
data = json.loads(response_text)
|
||||
except json.JSONDecodeError as e:
|
||||
return False, [f"响应不是有效的JSON: {e}"]
|
||||
|
||||
# 检查必需字段
|
||||
required_fields = ["id", "object", "created", "model", "choices"]
|
||||
for field in required_fields:
|
||||
if field not in data:
|
||||
errors.append(f"缺少必需字段: {field}")
|
||||
|
||||
# 检查 object 字段值
|
||||
if "object" in data and data["object"] != "chat.completion":
|
||||
errors.append(f"字段 'object' 值错误: 期望 'chat.completion', 实际 '{data['object']}'")
|
||||
|
||||
# 检查 choices 数组
|
||||
if "choices" in data:
|
||||
if not isinstance(data["choices"], list):
|
||||
errors.append(f"字段 'choices' 类型错误: 期望 list, 实际 {type(data['choices']).__name__}")
|
||||
else:
|
||||
print(format_json(json.dumps(body, ensure_ascii=False)))
|
||||
for i, choice in enumerate(data["choices"]):
|
||||
if not isinstance(choice, dict):
|
||||
errors.append(f"choices[{i}] 不是对象")
|
||||
continue
|
||||
|
||||
if stream:
|
||||
status, data, elapsed = http_stream_request(url, headers, body, ssl_ctx)
|
||||
else:
|
||||
status, data, elapsed = http_request(url, method, headers, body, ssl_ctx)
|
||||
# 检查 choice 对象的必需字段
|
||||
choice_required = ["index", "message", "finish_reason"]
|
||||
for field in choice_required:
|
||||
if field not in choice:
|
||||
errors.append(f"choices[{i}] 缺少必需字段: {field}")
|
||||
|
||||
if status is not None:
|
||||
print(f"状态码: {status} | 耗时: {elapsed:.2f}s")
|
||||
else:
|
||||
print(f"请求失败 | 耗时: {elapsed:.2f}s")
|
||||
# 检查 message 对象
|
||||
if "message" in choice and isinstance(choice["message"], dict):
|
||||
msg = choice["message"]
|
||||
if "role" not in msg:
|
||||
errors.append(f"choices[{i}].message 缺少必需字段: role")
|
||||
elif msg["role"] != "assistant":
|
||||
errors.append(f"choices[{i}].message.role 值错误: 期望 'assistant', 实际 '{msg['role']}'")
|
||||
|
||||
if stream and status and status < 300:
|
||||
# 流式响应按 SSE 行逐行输出
|
||||
for line in data.split("\n"):
|
||||
print(line)
|
||||
else:
|
||||
print(format_json(data))
|
||||
# 检查 usage 对象（可选）
|
||||
if "usage" in data and data["usage"] is not None:
|
||||
if not isinstance(data["usage"], dict):
|
||||
errors.append(f"字段 'usage' 类型错误: 期望 object, 实际 {type(data['usage']).__name__}")
|
||||
else:
|
||||
usage_required = ["prompt_tokens", "completion_tokens", "total_tokens"]
|
||||
for field in usage_required:
|
||||
if field not in data["usage"]:
|
||||
errors.append(f"usage 缺少必需字段: {field}")
|
||||
|
||||
return status
|
||||
return len(errors) == 0, errors
|
||||
|
||||
|
||||
def main():
|
||||
@@ -154,57 +196,79 @@ def main():
|
||||
|
||||
chat_url = f"{base_url}/chat/completions"
|
||||
|
||||
# --- 收集用例: (描述, 方法, URL, 请求头, 请求体, 是否流式) ---
|
||||
cases = []
|
||||
# --- 收集测试用例 ---
|
||||
cases: List[TestCase] = []
|
||||
|
||||
# ---- Models API ----
|
||||
cases.append((
|
||||
"获取模型列表 (GET /models)",
|
||||
"GET", f"{base_url}/models", headers, None, False
|
||||
cases.append(TestCase(
|
||||
desc="获取模型列表 (GET /models)",
|
||||
method="GET",
|
||||
url=f"{base_url}/models",
|
||||
headers=headers,
|
||||
validator=validate_openai_models_list_response
|
||||
))
|
||||
cases.append((
|
||||
"获取指定模型详情 (GET /models/{model})",
|
||||
"GET", f"{base_url}/models/{model}", headers, None, False
|
||||
cases.append(TestCase(
|
||||
desc="获取指定模型详情 (GET /models/{model})",
|
||||
method="GET",
|
||||
url=f"{base_url}/models/{model}",
|
||||
headers=headers,
|
||||
validator=validate_openai_model_retrieve_response
|
||||
))
|
||||
cases.append((
|
||||
"获取不存在的模型 (GET /models/nonexistent-model-xxx)",
|
||||
"GET", f"{base_url}/models/nonexistent-model-xxx", headers, None, False
|
||||
cases.append(TestCase(
|
||||
desc="获取不存在的模型 (GET /models/nonexistent-model-xxx)",
|
||||
method="GET",
|
||||
url=f"{base_url}/models/nonexistent-model-xxx",
|
||||
headers=headers
|
||||
))
|
||||
|
||||
# ---- Chat Completions: 正面用例 ----
|
||||
cases.append((
|
||||
"基本对话（仅 user）",
|
||||
"POST", chat_url, headers, {
|
||||
cases.append(TestCase(
|
||||
desc="基本对话（仅 user）",
|
||||
method="POST",
|
||||
url=chat_url,
|
||||
headers=headers,
|
||||
body={
|
||||
"model": model,
|
||||
"messages": [{"role": "user", "content": "Hi"}],
|
||||
"max_tokens": 5
|
||||
}, False
|
||||
},
|
||||
validator=validate_openai_chat_completion_response
|
||||
))
|
||||
cases.append((
|
||||
"system + user 对话",
|
||||
"POST", chat_url, headers, {
|
||||
cases.append(TestCase(
|
||||
desc="system + user 对话",
|
||||
method="POST",
|
||||
url=chat_url,
|
||||
headers=headers,
|
||||
body={
|
||||
"model": model,
|
||||
"messages": [
|
||||
{"role": "system", "content": "You are a helpful assistant."},
|
||||
{"role": "user", "content": "1+1="}
|
||||
],
|
||||
"max_tokens": 5
|
||||
}, False
|
||||
},
|
||||
validator=validate_openai_chat_completion_response
|
||||
))
|
||||
cases.append((
|
||||
"developer + user 对话",
|
||||
"POST", chat_url, headers, {
|
||||
cases.append(TestCase(
|
||||
desc="developer + user 对话",
|
||||
method="POST",
|
||||
url=chat_url,
|
||||
headers=headers,
|
||||
body={
|
||||
"model": model,
|
||||
"messages": [
|
||||
{"role": "developer", "content": "You are a helpful assistant."},
|
||||
{"role": "user", "content": "1+1="}
|
||||
],
|
||||
"max_tokens": 5
|
||||
}, False
|
||||
}
|
||||
))
|
||||
cases.append((
|
||||
"多轮对话（含 assistant 历史）",
|
||||
"POST", chat_url, headers, {
|
||||
cases.append(TestCase(
|
||||
desc="多轮对话（含 assistant 历史）",
|
||||
method="POST",
|
||||
url=chat_url,
|
||||
headers=headers,
|
||||
body={
|
||||
"model": model,
|
||||
"messages": [
|
||||
{"role": "user", "content": "Hi"},
|
||||
@@ -212,74 +276,98 @@ def main():
|
||||
{"role": "user", "content": "1+1="}
|
||||
],
|
||||
"max_tokens": 5
|
||||
}, False
|
||||
}
|
||||
))
|
||||
cases.append((
|
||||
"temperature + top_p",
|
||||
"POST", chat_url, headers, {
|
||||
cases.append(TestCase(
|
||||
desc="temperature + top_p",
|
||||
method="POST",
|
||||
url=chat_url,
|
||||
headers=headers,
|
||||
body={
|
||||
"model": model,
|
||||
"messages": [{"role": "user", "content": "Hi"}],
|
||||
"max_tokens": 5,
|
||||
"temperature": 0.5,
|
||||
"top_p": 0.9
|
||||
}, False
|
||||
}
|
||||
))
|
||||
cases.append((
|
||||
"max_tokens 限制",
|
||||
"POST", chat_url, headers, {
|
||||
cases.append(TestCase(
|
||||
desc="max_tokens 限制",
|
||||
method="POST",
|
||||
url=chat_url,
|
||||
headers=headers,
|
||||
body={
|
||||
"model": model,
|
||||
"messages": [{"role": "user", "content": "讲一个故事"}],
|
||||
"max_tokens": 10
|
||||
}, False
|
||||
}
|
||||
))
|
||||
cases.append((
|
||||
"stop sequences",
|
||||
"POST", chat_url, headers, {
|
||||
cases.append(TestCase(
|
||||
desc="stop sequences",
|
||||
method="POST",
|
||||
url=chat_url,
|
||||
headers=headers,
|
||||
body={
|
||||
"model": model,
|
||||
"messages": [{"role": "user", "content": "数数: 1,2,3,"}],
|
||||
"max_tokens": 20,
|
||||
"stop": ["5"]
|
||||
}, False
|
||||
}
|
||||
))
|
||||
cases.append((
|
||||
"n=2 多候选",
|
||||
"POST", chat_url, headers, {
|
||||
cases.append(TestCase(
|
||||
desc="n=2 多候选",
|
||||
method="POST",
|
||||
url=chat_url,
|
||||
headers=headers,
|
||||
body={
|
||||
"model": model,
|
||||
"messages": [{"role": "user", "content": "Hi"}],
|
||||
"max_tokens": 5,
|
||||
"n": 2
|
||||
}, False
|
||||
}
|
||||
))
|
||||
cases.append((
|
||||
"seed 参数",
|
||||
"POST", chat_url, headers, {
|
||||
cases.append(TestCase(
|
||||
desc="seed 参数",
|
||||
method="POST",
|
||||
url=chat_url,
|
||||
headers=headers,
|
||||
body={
|
||||
"model": model,
|
||||
"messages": [{"role": "user", "content": "Hi"}],
|
||||
"max_tokens": 5,
|
||||
"seed": 42
|
||||
}, False
|
||||
}
|
||||
))
|
||||
cases.append((
|
||||
"frequency_penalty + presence_penalty",
|
||||
"POST", chat_url, headers, {
|
||||
cases.append(TestCase(
|
||||
desc="frequency_penalty + presence_penalty",
|
||||
method="POST",
|
||||
url=chat_url,
|
||||
headers=headers,
|
||||
body={
|
||||
"model": model,
|
||||
"messages": [{"role": "user", "content": "Hi"}],
|
||||
"max_tokens": 5,
|
||||
"frequency_penalty": 0.5,
|
||||
"presence_penalty": 0.5
|
||||
}, False
|
||||
}
|
||||
))
|
||||
cases.append((
|
||||
"max_completion_tokens 参数",
|
||||
"POST", chat_url, headers, {
|
||||
cases.append(TestCase(
|
||||
desc="max_completion_tokens 参数",
|
||||
method="POST",
|
||||
url=chat_url,
|
||||
headers=headers,
|
||||
body={
|
||||
"model": model,
|
||||
"messages": [{"role": "user", "content": "讲一个故事"}],
|
||||
"max_completion_tokens": 10
|
||||
}, False
|
||||
}
|
||||
))
|
||||
cases.append((
|
||||
"JSON mode (response_format: json_object)",
|
||||
"POST", chat_url, headers, {
|
||||
cases.append(TestCase(
|
||||
desc="JSON mode (response_format: json_object)",
|
||||
method="POST",
|
||||
url=chat_url,
|
||||
headers=headers,
|
||||
body={
|
||||
"model": model,
|
||||
"messages": [
|
||||
{"role": "system", "content": "以 JSON 格式回复: {\"answer\": \"ok\"}"},
|
||||
@@ -287,51 +375,69 @@ def main():
|
||||
],
|
||||
"max_tokens": 10,
|
||||
"response_format": {"type": "json_object"}
|
||||
}, False
|
||||
}
|
||||
))
|
||||
|
||||
# ---- Chat Completions: 负面用例 ----
|
||||
cases.append((
|
||||
"缺少 model 参数",
|
||||
"POST", chat_url, headers, {
|
||||
cases.append(TestCase(
|
||||
desc="缺少 model 参数",
|
||||
method="POST",
|
||||
url=chat_url,
|
||||
headers=headers,
|
||||
body={
|
||||
"messages": [{"role": "user", "content": "Hi"}],
|
||||
"max_tokens": 5
|
||||
}, False
|
||||
}
|
||||
))
|
||||
cases.append((
|
||||
"缺少 messages 参数",
|
||||
"POST", chat_url, headers, {
|
||||
cases.append(TestCase(
|
||||
desc="缺少 messages 参数",
|
||||
method="POST",
|
||||
url=chat_url,
|
||||
headers=headers,
|
||||
body={
|
||||
"model": model,
|
||||
"max_tokens": 5
|
||||
}, False
|
||||
}
|
||||
))
|
||||
cases.append((
|
||||
"messages 为空数组",
|
||||
"POST", chat_url, headers, {
|
||||
cases.append(TestCase(
|
||||
desc="messages 为空数组",
|
||||
method="POST",
|
||||
url=chat_url,
|
||||
headers=headers,
|
||||
body={
|
||||
"model": model,
|
||||
"messages": [],
|
||||
"max_tokens": 5
|
||||
}, False
|
||||
}
|
||||
))
|
||||
cases.append((
|
||||
"无效 API key",
|
||||
"POST", chat_url, headers_bad_auth, {
|
||||
cases.append(TestCase(
|
||||
desc="无效 API key",
|
||||
method="POST",
|
||||
url=chat_url,
|
||||
headers=headers_bad_auth,
|
||||
body={
|
||||
"model": model,
|
||||
"messages": [{"role": "user", "content": "Hi"}],
|
||||
"max_tokens": 5
|
||||
}, False
|
||||
}
|
||||
))
|
||||
cases.append((
|
||||
"不存在的模型 (chat)",
|
||||
"POST", chat_url, headers, {
|
||||
cases.append(TestCase(
|
||||
desc="不存在的模型 (chat)",
|
||||
method="POST",
|
||||
url=chat_url,
|
||||
headers=headers,
|
||||
body={
|
||||
"model": "nonexistent-model-xxx",
|
||||
"messages": [{"role": "user", "content": "Hi"}],
|
||||
"max_tokens": 5
|
||||
}, False
|
||||
}
|
||||
))
|
||||
cases.append((
|
||||
"畸形 JSON body",
|
||||
"POST", chat_url, headers, "invalid json{", False
|
||||
cases.append(TestCase(
|
||||
desc="畸形 JSON body",
|
||||
method="POST",
|
||||
url=chat_url,
|
||||
headers=headers,
|
||||
body="invalid json{"
|
||||
))
|
||||
|
||||
# ---- --vision ----
|
||||
@@ -341,9 +447,12 @@ def main():
|
||||
"Gfp-wisconsin-madison-the-nature-boardwalk.jpg/"
|
||||
"2560px-Gfp-wisconsin-madison-the-nature-boardwalk.jpg"
|
||||
)
|
||||
cases.append((
|
||||
"图片 URL 输入 + detail 参数 (--vision)",
|
||||
"POST", chat_url, headers, {
|
||||
cases.append(TestCase(
|
||||
desc="图片 URL 输入 + detail 参数 (--vision)",
|
||||
method="POST",
|
||||
url=chat_url,
|
||||
headers=headers,
|
||||
body={
|
||||
"model": model,
|
||||
"messages": [
|
||||
{"role": "system", "content": "简短描述图片"},
|
||||
@@ -355,39 +464,51 @@ def main():
|
||||
]}
|
||||
],
|
||||
"max_tokens": 10
|
||||
}, False
|
||||
}
|
||||
))
|
||||
|
||||
# ---- --stream ----
|
||||
if args.stream:
|
||||
cases.append((
|
||||
"基本流式 (--stream)",
|
||||
"POST", chat_url, headers, {
|
||||
cases.append(TestCase(
|
||||
desc="基本流式 (--stream)",
|
||||
method="POST",
|
||||
url=chat_url,
|
||||
headers=headers,
|
||||
body={
|
||||
"model": model,
|
||||
"messages": [{"role": "user", "content": "Hi"}],
|
||||
"max_tokens": 5,
|
||||
"stream": True
|
||||
}, True
|
||||
},
|
||||
stream=True
|
||||
))
|
||||
cases.append((
|
||||
"流式 + include_usage (--stream)",
|
||||
"POST", chat_url, headers, {
|
||||
cases.append(TestCase(
|
||||
desc="流式 + include_usage (--stream)",
|
||||
method="POST",
|
||||
url=chat_url,
|
||||
headers=headers,
|
||||
body={
|
||||
"model": model,
|
||||
"messages": [{"role": "user", "content": "Hi"}],
|
||||
"max_tokens": 5,
|
||||
"stream": True,
|
||||
"stream_options": {"include_usage": True}
|
||||
}, True
|
||||
},
|
||||
stream=True
|
||||
))
|
||||
cases.append((
|
||||
"流式 + stop sequences (--stream)",
|
||||
"POST", chat_url, headers, {
|
||||
cases.append(TestCase(
|
||||
desc="流式 + stop sequences (--stream)",
|
||||
method="POST",
|
||||
url=chat_url,
|
||||
headers=headers,
|
||||
body={
|
||||
"model": model,
|
||||
"messages": [{"role": "user", "content": "数数: 1,2,3,"}],
|
||||
"max_tokens": 20,
|
||||
"stream": True,
|
||||
"stop": ["5"]
|
||||
}, True
|
||||
},
|
||||
stream=True
|
||||
))
|
||||
|
||||
# ---- --tools ----
|
||||
@@ -406,29 +527,38 @@ def main():
|
||||
}
|
||||
}
|
||||
}
|
||||
cases.append((
|
||||
"工具调用 tool_choice: auto (--tools)",
|
||||
"POST", chat_url, headers, {
|
||||
cases.append(TestCase(
|
||||
desc="工具调用 tool_choice: auto (--tools)",
|
||||
method="POST",
|
||||
url=chat_url,
|
||||
headers=headers,
|
||||
body={
|
||||
"model": model,
|
||||
"messages": [{"role": "user", "content": "北京天气怎么样？"}],
|
||||
"max_tokens": 50,
|
||||
"tools": [tool_weather],
|
||||
"tool_choice": "auto"
|
||||
}, False
|
||||
}
|
||||
))
|
||||
cases.append((
|
||||
"工具调用 tool_choice: required (--tools)",
|
||||
"POST", chat_url, headers, {
|
||||
cases.append(TestCase(
|
||||
desc="工具调用 tool_choice: required (--tools)",
|
||||
method="POST",
|
||||
url=chat_url,
|
||||
headers=headers,
|
||||
body={
|
||||
"model": model,
|
||||
"messages": [{"role": "user", "content": "北京天气怎么样？"}],
|
||||
"max_tokens": 50,
|
||||
"tools": [tool_weather],
|
||||
"tool_choice": "required"
|
||||
}, False
|
||||
}
|
||||
))
|
||||
cases.append((
|
||||
"指定函数调用 tool_choice: {name} (--tools)",
|
||||
"POST", chat_url, headers, {
|
||||
cases.append(TestCase(
|
||||
desc="指定函数调用 tool_choice: {name} (--tools)",
|
||||
method="POST",
|
||||
url=chat_url,
|
||||
headers=headers,
|
||||
body={
|
||||
"model": model,
|
||||
"messages": [{"role": "user", "content": "北京天气怎么样？"}],
|
||||
"max_tokens": 50,
|
||||
@@ -437,11 +567,14 @@ def main():
|
||||
"type": "function",
|
||||
"function": {"name": "get_weather"}
|
||||
}
|
||||
}, False
|
||||
}
|
||||
))
|
||||
cases.append((
|
||||
"多轮工具调用（构造 tool 结果）(--tools)",
|
||||
"POST", chat_url, headers, {
|
||||
cases.append(TestCase(
|
||||
desc="多轮工具调用（构造 tool 结果）(--tools)",
|
||||
method="POST",
|
||||
url=chat_url,
|
||||
headers=headers,
|
||||
body={
|
||||
"model": model,
|
||||
"messages": [
|
||||
{"role": "user", "content": "北京天气怎么样？"},
|
||||
@@ -457,38 +590,47 @@ def main():
|
||||
],
|
||||
"max_tokens": 20,
|
||||
"tools": [tool_weather]
|
||||
}, False
|
||||
}
|
||||
))
|
||||
cases.append((
|
||||
"parallel_tool_calls: false (--tools)",
|
||||
"POST", chat_url, headers, {
|
||||
cases.append(TestCase(
|
||||
desc="parallel_tool_calls: false (--tools)",
|
||||
method="POST",
|
||||
url=chat_url,
|
||||
headers=headers,
|
||||
body={
|
||||
"model": model,
|
||||
"messages": [{"role": "user", "content": "北京和上海的天气怎么样？"}],
|
||||
"max_tokens": 50,
|
||||
"tools": [tool_weather],
|
||||
"tool_choice": "auto",
|
||||
"parallel_tool_calls": False
|
||||
}, False
|
||||
}
|
||||
))
|
||||
|
||||
# ---- --logprobs ----
|
||||
if args.logprobs:
|
||||
cases.append((
|
||||
"logprobs + top_logprobs (--logprobs)",
|
||||
"POST", chat_url, headers, {
|
||||
cases.append(TestCase(
|
||||
desc="logprobs + top_logprobs (--logprobs)",
|
||||
method="POST",
|
||||
url=chat_url,
|
||||
headers=headers,
|
||||
body={
|
||||
"model": model,
|
||||
"messages": [{"role": "user", "content": "Hi"}],
|
||||
"max_tokens": 5,
|
||||
"logprobs": True,
|
||||
"top_logprobs": 2
|
||||
}, False
|
||||
}
|
||||
))
|
||||
|
||||
# ---- --json-schema ----
|
||||
if args.json_schema:
|
||||
cases.append((
|
||||
"Structured Output json_schema (--json_schema)",
|
||||
"POST", chat_url, headers, {
|
||||
cases.append(TestCase(
|
||||
desc="Structured Output json_schema (--json_schema)",
|
||||
method="POST",
|
||||
url=chat_url,
|
||||
headers=headers,
|
||||
body={
|
||||
"model": model,
|
||||
"messages": [{"role": "user", "content": "1+1等于几？"}],
|
||||
"max_tokens": 20,
|
||||
@@ -508,19 +650,67 @@ def main():
|
||||
}
|
||||
}
|
||||
}
|
||||
}, False
|
||||
}
|
||||
))
|
||||
|
||||
# ---- 执行测试 ----
|
||||
total = len(cases)
|
||||
count_2xx = 0
|
||||
count_other = 0
|
||||
# ---- 高级参数测试 ----
|
||||
# logit_bias: 修改特定token的似然
|
||||
cases.append(TestCase(
|
||||
desc="logit_bias 参数测试",
|
||||
method="POST",
|
||||
url=chat_url,
|
||||
headers=headers,
|
||||
body={
|
||||
"model": model,
|
||||
"messages": [{"role": "user", "content": "Hello"}],
|
||||
"max_tokens": 5,
|
||||
"logit_bias": {"1234": -100, "5678": 50} # token_id: bias
|
||||
}
|
||||
))
|
||||
|
||||
print("=" * 60)
|
||||
print("OpenAI 兼容性测试")
|
||||
print(f"目标: {base_url}")
|
||||
print(f"模型: {model}")
|
||||
print(f"时间: {time.strftime('%Y-%m-%d %H:%M:%S')}")
|
||||
# reasoning_effort: 推理努力级别（需要模型支持）
|
||||
cases.append(TestCase(
|
||||
desc="reasoning_effort: medium",
|
||||
method="POST",
|
||||
url=chat_url,
|
||||
headers=headers,
|
||||
body={
|
||||
"model": model,
|
||||
"messages": [{"role": "user", "content": "1+1=?"}],
|
||||
"max_tokens": 10,
|
||||
"reasoning_effort": "medium"
|
||||
}
|
||||
))
|
||||
|
||||
# service_tier: 服务层级
|
||||
cases.append(TestCase(
|
||||
desc="service_tier: auto",
|
||||
method="POST",
|
||||
url=chat_url,
|
||||
headers=headers,
|
||||
body={
|
||||
"model": model,
|
||||
"messages": [{"role": "user", "content": "Hi"}],
|
||||
"max_tokens": 5,
|
||||
"service_tier": "auto"
|
||||
}
|
||||
))
|
||||
|
||||
# verbosity: 冗长程度
|
||||
cases.append(TestCase(
|
||||
desc="verbosity: low",
|
||||
method="POST",
|
||||
url=chat_url,
|
||||
headers=headers,
|
||||
body={
|
||||
"model": model,
|
||||
"messages": [{"role": "user", "content": "介绍一下Python"}],
|
||||
"max_tokens": 50,
|
||||
"verbosity": "low"
|
||||
}
|
||||
))
|
||||
|
||||
# ---- 执行测试 ----
|
||||
flags = []
|
||||
if args.vision:
|
||||
flags.append("vision")
|
||||
@@ -532,20 +722,15 @@ def main():
|
||||
flags.append("logprobs")
|
||||
if args.json_schema:
|
||||
flags.append("json-schema")
|
||||
print(f"用例: {total} 个" + (f" | 扩展: {', '.join(flags)}" if flags else ""))
|
||||
print("=" * 60)
|
||||
|
||||
for i, (desc, method, url, hdrs, body, stream) in enumerate(cases, 1):
|
||||
status = run_test(i, total, desc, url, method, hdrs, body, stream, ssl_ctx)
|
||||
if status is not None and 200 <= status < 300:
|
||||
count_2xx += 1
|
||||
else:
|
||||
count_other += 1
|
||||
|
||||
print()
|
||||
print("=" * 60)
|
||||
print(f"测试完成 | 总计: {total} | HTTP 2xx: {count_2xx} | 非 2xx: {count_other}")
|
||||
print("=" * 60)
|
||||
run_test_suite(
|
||||
cases=cases,
|
||||
ssl_ctx=ssl_ctx,
|
||||
title="OpenAI 兼容性测试",
|
||||
base_url=base_url,
|
||||
model=model,
|
||||
flags=flags
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
Reference in New Issue
Block a user