使用 Mock 测试类模拟 OpenAI GPT‑5.6 系列 API 调用

在开发基于 OpenAI GPT‑5.6 系列模型（Sol、Terra、Luna）的应用程序时，单元测试和集成测试是保障代码质量的关键环节。然而，直接调用真实的 API 会带来成本、延迟以及外部依赖的不稳定性。为了解决这一问题，本文介绍一个 无需安装官方 openai 包 的 Mock 测试类，它提供了与真实 SDK 完全一致的接口，支持纯文本、多模态（图片、文件附件）、同步、异步以及流式输出。测试代码只需将导入语句从 openai 替换为 mock_openai，其余调用逻辑与生产环境完全相同。它是一个功能全面、易于集成的 Mock 测试类，用于模拟 OpenAI GPT‑5.6 系列模型的 API 调用，并支持同步/异步、流式输出、多模态（图片、文件附件）以及自定义回复，并且 完全不需要安装官方 openai 包。通过简单地替换导入语句，开发者可以在测试环境中安全、高效地验证业务逻辑，而无需担心 API 成本或网络问题。该 Mock 类极大地降低了测试代码与生产代码的差异，是开发 GPT‑5.6 应用时不可或缺的测试工具。

一、设计目标与特性

该 Mock 测试类的核心目标是 让测试代码与真实调用代码保持一致，从而最大程度降低测试与生产代码之间的差异。具体特性如下：

二、实现细节

mock_openai.py 主要包含以下部分：

• 响应对象：MockMessage、MockChoice、MockResponse、MockStreamChunk 分别模拟官方 SDK 返回结构中的对应对象。
• 内容提取：MockChatCompletions._extract_text_and_attachments 方法负责解析 messages 中的 content 字段。若 content 为字符串则视为纯文本；若为列表则遍历其中的元素，根据 type 字段（text、image_url、file、file_url）分别提取文本、图片 URL 和文件信息。
• 默认回复生成：_default_generate 根据模型名称、提取到的文本/图片/文件信息以及请求参数（如 temperature、max_tokens、reasoning_effort）生成一个可读的模拟回复字符串。
• 流式处理：在 create 方法中，若 stream=True，则将完整回复按固定大小（默认 20 字符）切分为多个数据块，以生成器形式返回。

三、完整代码

下面是完整的 mock_openai.py 文件内容：

"""
独立的 OpenAI Mock 模块，提供与 openai SDK 相同接口的 OpenAI 和 AsyncOpenAI 类。
支持多模态消息：图片、文件附件。
只需将此文件放在测试代码可导入的路径下，然后在测试代码中：
from mock_openai import OpenAI # 替代 from openai import OpenAI
from mock_openai import AsyncOpenAI # 替代 from openai import AsyncOpenAI
其他调用代码与真实 openai 完全一致。
"""
import asyncio
from typing import List, Dict, Any, Iterator, Optional, Callable

# ———- 模拟响应对象 ———-
class MockMessage:
def __init__(self, content: str):
self.content = content

class MockChoice:
def __init__(self, content: str, is_delta: bool = False):
if is_delta:
self.delta = MockMessage(content)
else:
self.message = MockMessage(content)

class MockResponse:
def __init__(self, choices: List[MockChoice]):
self.choices = choices

class MockStreamChunk:
def __init__(self, content: str):
self.choices = [MockChoice(content, is_delta=True)]

# ———- 模拟同步聊天补全 ———-
class MockChatCompletions:
def __init__(self, response_generator: Optional[Callable] = None):
self.response_generator = response_generator or self._default_generate

@staticmethod
def _extract_text_and_attachments(messages: List[Dict[str, Any]]) –> tuple:
"""
从 messages 中提取纯文本内容、图片 URL 列表和文件信息列表。
支持 content 为字符串或列表（多模态格式）。
返回 (文本, 图片URL列表, 文件描述列表)
"""
all_text = []
all_image_urls = []
all_file_infos = []

for msg in messages:
content = msg.get("content", "")
if isinstance(content, str):
if msg.get("role") == "user":
all_text.append(content)
elif isinstance(content, list):
for part in content:
if not isinstance(part, dict):
continue
part_type = part.get("type")
if part_type == "text":
all_text.append(part.get("text", ""))
elif part_type == "image_url":
image_url_obj = part.get("image_url", {})
url = image_url_obj.get("url", "")
if url:
all_image_urls.append(url)
elif part_type in ("file", "file_url"):
file_obj = part.get("file", {}) or part.get("file_url", {})
if isinstance(file_obj, dict):
file_id = file_obj.get("file_id")
url = file_obj.get("url")
if file_id:
all_file_infos.append(f"file_id={file_id}")
elif url:
all_file_infos.append(f"file_url={url}")
else:
all_file_infos.append("file (unknown)")
else:
all_file_infos.append(f"file={file_obj}")
user_text = "\\n".join(all_text)
return user_text, all_image_urls, all_file_infos

@staticmethod
def _default_generate(model: str, messages: List[Dict[str, Any]], **kwargs) –> str:
user_text, image_urls, file_infos = MockChatCompletions._extract_text_and_attachments(messages)

model_prefix = {
"gpt-5.6-sol": "[Sol 旗舰]",
"gpt-5.6-terra": "[Terra 均衡]",
"gpt-5.6-luna": "[Luna 快速]",
}.get(model, "[GPT-5.6]")

temp = kwargs.get("temperature", 1.0)
max_tokens = kwargs.get("max_tokens", 100)
reasoning = kwargs.get("reasoning_effort", "none")
reasoning_note = f" (推理力度: {reasoning})" if reasoning != "none" else ""

content = f"{model_prefix} 模拟回复{reasoning_note}。\\n"
if image_urls:
content += f"检测到 {len(image_urls)} 张图片输入，但模拟环境无法真正分析图像内容。\\n"
content += f"图片 URL(s): {', '.join(image_urls[:3])}{'…' if len(image_urls) > 3 else ''}\\n"
if file_infos:
content += f"检测到 {len(file_infos)} 个文件附件，但模拟环境无法读取文件内容。\\n"
content += f"文件信息: {', '.join(file_infos[:3])}{'…' if len(file_infos) > 3 else ''}\\n"
content += f"用户文本：'{user_text}'\\n"
content += f"请求参数：temperature={temp}, max_tokens={max_tokens}\\n"
content += "这是一个 Mock 响应，用于测试目的。"

if max_tokens < len(content):
content = content[:max_tokens]
return content

def create(self, model: str, messages: List[Dict[str, Any]], stream: bool = False, **kwargs) –> Any:
full_content = self.response_generator(model, messages, **kwargs)

if stream:
chunk_size = 20
def stream_generator() –> Iterator[MockStreamChunk]:
for i in range(0, len(full_content), chunk_size):
yield MockStreamChunk(full_content[i:i+chunk_size])
return stream_generator()
else:
return MockResponse([MockChoice(full_content)])

class MockChat:
def __init__(self, response_generator=None):
self.completions = MockChatCompletions(response_generator)

class OpenAI:
"""
模拟同步 OpenAI 客户端，与官方 openai.OpenAI 接口一致。
"""
def __init__(self, api_key: Optional[str] = None, base_url: Optional[str] = None,
response_generator: Optional[Callable] = None, **kwargs):
self.api_key = api_key
self.base_url = base_url
self.chat = MockChat(response_generator)

# ———- 模拟异步聊天补全 ———-
class MockAsyncChatCompletions:
def __init__(self, response_generator=None):
self._sync_completions = MockChatCompletions(response_generator)

async def create(self, model: str, messages: List[Dict[str, Any]], stream: bool = False, **kwargs) –> Any:
await asyncio.sleep(0.01) # 模拟异步延迟

if stream:
sync_stream = self._sync_completions.create(model, messages, stream=True, **kwargs)
async def async_stream_generator():
for chunk in sync_stream:
yield chunk
await asyncio.sleep(0.01)
return async_stream_generator()
else:
return self._sync_completions.create(model, messages, stream=False, **kwargs)

class MockAsyncChat:
def __init__(self, response_generator=None):
self.completions = MockAsyncChatCompletions(response_generator)

class AsyncOpenAI:
"""
模拟异步 OpenAI 客户端，与官方 openai.AsyncOpenAI 接口一致。
"""
def __init__(self, api_key: Optional[str] = None, base_url: Optional[str] = None,
response_generator: Optional[Callable] = None, **kwargs):
self.api_key = api_key
self.base_url = base_url
self.chat = MockAsyncChat(response_generator)

四、调用示例

以下示例涵盖了纯文本、多模态（图片、文件附件）以及所有同步/异步、普通/流式组合。所有示例仅需将导入语句从 openai 改为 mock_openai，其余代码与真实调用完全一致。

1. 同步 – 纯文本调用

from mock_openai import OpenAI

client = OpenAI()
response = client.chat.completions.create(
model="gpt-5.6-sol",
messages=[
{"role": "system", "content": "You are a helpful assistant."},
{"role": "user", "content": "解释一下什么是量子计算。"}
],
temperature=0.7,
max_tokens=1024,
)
print(response.choices[0].message.content)

模拟输出：

[Sol 旗舰] 模拟回复。
您的问题是：'解释一下什么是量子计算。'
请求参数：temperature=0.7, max_tokens=1024
这是一个 Mock 响应，用于测试目的。

2. 同步 – 流式输出

from mock_openai import OpenAI

client = OpenAI()
stream = client.chat.completions.create(
model="gpt-5.6-luna",
messages=[{"role": "user", "content": "讲个笑话"}],
stream=True,
)
for chunk in stream:
if chunk.choices[0].delta.content is not None:
print(chunk.choices[0].delta.content, end="")

模拟输出（内容分块打印）：

[Luna 快速] 模拟回复。
您的问题是：'讲个笑话'
请求参数：temperature=1.0, max_tokens=100
这是一个 Mock 响应，用于测试目的。

3. 异步 – 纯文本调用

import asyncio
from mock_openai import AsyncOpenAI

async def main():
client = AsyncOpenAI()
response = await client.chat.completions.create(
model="gpt-5.6-terra",
messages=[{"role": "user", "content": "Hello!"}],
)
print(response.choices[0].message.content)

asyncio.run(main())

模拟输出：

[Terra 均衡] 模拟回复。
您的问题是：'Hello!'
请求参数：temperature=1.0, max_tokens=100
这是一个 Mock 响应，用于测试目的。

4. 异步 – 流式输出

import asyncio
from mock_openai import AsyncOpenAI

async def main():
client = AsyncOpenAI()
stream = await client.chat.completions.create(
model="gpt-5.6-luna",
messages=[{"role": "user", "content": "讲个笑话"}],
stream=True,
)
async for chunk in stream:
if chunk.choices[0].delta.content is not None:
print(chunk.choices[0].delta.content, end="")

asyncio.run(main())

模拟输出（内容分块打印）：

[Luna 快速] 模拟回复。
您的问题是：'讲个笑话'
请求参数：temperature=1.0, max_tokens=100
这是一个 Mock 响应，用于测试目的。

5. 多模态 – 同步调用（图片输入）

from mock_openai import OpenAI

client = OpenAI()
response = client.chat.completions.create(
model="gpt-5.6-sol",
messages=[
{
"role": "user",
"content": [
{"type": "text", "text": "这张图片里有什么？"},
{"type": "image_url", "image_url": {"url": "https://example.com/image.png"}}
]
}
],
max_tokens=300,
)
print(response.choices[0].message.content)

模拟输出：

[Sol 旗舰] 模拟回复。
检测到 1 张图片输入，但模拟环境无法真正分析图像内容。
图片 URL(s): https://example.com/image.png
用户文本：'这张图片里有什么？'
请求参数：temperature=1.0, max_tokens=300
这是一个 Mock 响应，用于测试目的。

6. 多模态 – 同步流式输出

from mock_openai import OpenAI

client = OpenAI()
stream = client.chat.completions.create(
model="gpt-5.6-terra",
messages=[
{
"role": "user",
"content": [
{"type": "text", "text": "描述这张图片"},
{"type": "image_url", "image_url": {"url": "data:image/png;base64,…."}}
]
}
],
stream=True,
)
for chunk in stream:
if chunk.choices[0].delta.content is not None:
print(chunk.choices[0].delta.content, end="")

模拟输出（内容分块打印）：

[Terra 均衡] 模拟回复。
检测到 1 张图片输入，但模拟环境无法真正分析图像内容。
图片 URL(s): data:image/png;base64,….
用户文本：'描述这张图片'
请求参数：temperature=1.0, max_tokens=100
这是一个 Mock 响应，用于测试目的。

7. 多模态 – 异步调用

import asyncio
from mock_openai import AsyncOpenAI

async def main():
client = AsyncOpenAI()
response = await client.chat.completions.create(
model="gpt-5.6-luna",
messages=[
{
"role": "user",
"content": [
{"type": "text", "text": "分析这张图片"},
{"type": "image_url", "image_url": {"url": "https://example.com/photo.jpg"}}
]
}
]
)
print(response.choices[0].message.content)

asyncio.run(main())

模拟输出：

[Luna 快速] 模拟回复。
检测到 1 张图片输入，但模拟环境无法真正分析图像内容。
图片 URL(s): https://example.com/photo.jpg
用户文本：'分析这张图片'
请求参数：temperature=1.0, max_tokens=100
这是一个 Mock 响应，用于测试目的。

8. 多模态 – 异步流式输出

import asyncio
from mock_openai import AsyncOpenAI

async def main():
client = AsyncOpenAI()
stream = await client.chat.completions.create(
model="gpt-5.6-sol",
messages=[
{
"role": "user",
"content": [
{"type": "text", "text": "总结图片内容"},
{"type": "image_url", "image_url": {"url": "https://example.com/img1.png"}},
{"type": "image_url", "image_url": {"url": "https://example.com/img2.png"}}
]
}
],
stream=True,
)
async for chunk in stream:
if chunk.choices[0].delta.content is not None:
print(chunk.choices[0].delta.content, end="")

asyncio.run(main())

模拟输出（内容分块打印）：

[Sol 旗舰] 模拟回复。
检测到 2 张图片输入，但模拟环境无法真正分析图像内容。
图片 URL(s): https://example.com/img1.png, https://example.com/img2.png
用户文本：'总结图片内容'
请求参数：temperature=1.0, max_tokens=100
这是一个 Mock 响应，用于测试目的。

9. 文件附件 – 同步调用（使用 file_id）

from mock_openai import OpenAI

client = OpenAI()
response = client.chat.completions.create(
model="gpt-5.6-sol",
messages=[
{
"role": "user",
"content": [
{"type": "text", "text": "请总结这个PDF文件的内容。"},
{"type": "file", "file": {"file_id": "file-abc123"}}
]
}
],
max_tokens=300,
)
print(response.choices[0].message.content)

模拟输出：

[Sol 旗舰] 模拟回复。
检测到 1 个文件附件，但模拟环境无法读取文件内容。
文件信息: file_id=file-abc123
用户文本：'请总结这个PDF文件的内容。'
请求参数：temperature=1.0, max_tokens=300
这是一个 Mock 响应，用于测试目的。

10. 文件附件 – 同步流式输出（使用 file_url）

from mock_openai import OpenAI

client = OpenAI()
stream = client.chat.completions.create(
model="gpt-5.6-terra",
messages=[
{
"role": "user",
"content": [
{"type": "text", "text": "解析这个文档"},
{"type": "file_url", "file_url": {"url": "https://example.com/report.docx"}}
]
}
],
stream=True,
)
for chunk in stream:
if chunk.choices[0].delta.content is not None:
print(chunk.choices[0].delta.content, end="")

模拟输出（内容分块打印）：

[Terra 均衡] 模拟回复。
检测到 1 个文件附件，但模拟环境无法读取文件内容。
文件信息: file_url=https://example.com/report.docx
用户文本：'解析这个文档'
请求参数：temperature=1.0, max_tokens=100
这是一个 Mock 响应，用于测试目的。

11. 混合附件（图片 + 文件）

from mock_openai import OpenAI

client = OpenAI()
response = client.chat.completions.create(
model="gpt-5.6-sol",
messages=[
{
"role": "user",
"content": [
{"type": "text", "text": "结合图片和文档回答问题"},
{"type": "image_url", "image_url": {"url": "https://example.com/chart.png"}},
{"type": "file", "file": {"file_id": "file-report-000"}}
]
}
]
)
print(response.choices[0].message.content)

模拟输出：

[Sol 旗舰] 模拟回复。
检测到 1 张图片输入，但模拟环境无法真正分析图像内容。
图片 URL(s): https://example.com/chart.png
检测到 1 个文件附件，但模拟环境无法读取文件内容。
文件信息: file_id=file-report-000
用户文本：'结合图片和文档回答问题'
请求参数：temperature=1.0, max_tokens=100
这是一个 Mock 响应，用于测试目的。

12. 带 reasoning_effort 参数的调用

from mock_openai import OpenAI

client = OpenAI()
response = client.chat.completions.create(
model="gpt-5.6-sol",
messages=[{"role": "user", "content": "请证明哥德巴赫猜想。"}],
reasoning_effort="high"
)
print(response.choices[0].message.content)

模拟输出：

[Sol 旗舰] 模拟回复 (推理力度: high)。
您的问题是：'请证明哥德巴赫猜想。'
请求参数：temperature=1.0, max_tokens=100
这是一个 Mock 响应，用于测试目的。

13. 自定义回复生成器

from mock_openai import OpenAI

def custom_generator(model, messages, **kwargs):
# 自定义逻辑，完全控制返回内容
user_msg = ""
for m in messages:
if m.get("role") == "user":
if isinstance(m.get("content"), str):
user_msg = m["content"]
elif isinstance(m.get("content"), list):
for part in m["content"]:
if part.get("type") == "text":
user_msg = part.get("text")
break
return f"自定义回复：模型 {model}，用户说 '{user_msg}'"

client = OpenAI(response_generator=custom_generator)
resp = client.chat.completions.create(
model="gpt-5.6-terra",
messages=[{"role": "user", "content": "测试一下"}]
)
print(resp.choices[0].message.content)

输出：

自定义回复：模型 gpt-5.6-terra，用户说 '测试一下'

五、集成到测试项目