从 0 到 1：基于 Python + Pytest 的 AI SDK 接口自动化测试实战

1. 项目背景

最近我在学习 AI 大模型相关技术时，发现很多 AI 服务都通过 HTTP 接口对外提供能力。为了深入理解这类服务的质量保障方法，我基于一个开源的 C++ AI 大模型接入 SDK（ChatSDK） 在本地部署了服务（ChatServer），它对外提供模型查询、会话管理、消息发送等功能。

作为一名准测试开发，我决定以这个 AI 服务为被测对象，基于 Python + Pytest + Requests + Allure 搭建一套完整的接口自动化测试框架，对它进行全流程的“质量体检”。

最终，我独立设计了 13 个自动化测试用例，覆盖 7 个核心接口，在测试过程中成功发现并定位了 4 个真实缺陷。

被测系统地址：http://localhost:8080

2. 先搞懂两个专业名词：SSE 和流式响应

在开始测试之前，我们先搞清楚这个项目的核心——流式响应（SSE）。

2.1 什么是“流式响应”？

你平时访问网页，比如百度搜索，是一次性把整页结果显示给你。
但 AI 聊天不一样。如果 AI 回答一篇 500 字的文章，它不可能等整段话都想好了才一次性给你看，那样用户要等 10 秒以上。

流式响应，就是让 AI 边想边说，一个字一个字往外蹦。就像打字机一样，你看它一个字一个字地出现在屏幕上，体验感极好。

2.2 什么是 SSE？

SSE 的全称是 Server-Sent Events（服务器发送事件）。
它是一种基于 HTTP 协议的技术，允许服务器主动向客户端“推”数据。

关键点：

• 单向：数据只能从服务器流向客户端（AI 服务器 → 你）。你不需要通过 SSE 回复它。
• 基于 HTTP：不需要额外开端口，不需要特殊协议，浏览器天然支持。
• 格式固定：服务器推送的每条数据，必须以 data: 开头，以两个换行符 \n\n 结尾。

举个例子，如果 AI 要回答“你好”，SSE 推送的数据长这样：

data: {"content": "你"}

data: {"content": "好"}

data: {"content": "！"}

data: [DONE]

• 前面三条是 AI 一个字一个字说的话。
• 最后那条 data: [DONE] 是“结束标记”，告诉客户端“我说完了”。

2.3 为什么 AI 聊天喜欢用 SSE，而不是 WebSocket？

你可能听过另一个技术叫 WebSocket（双向通信，可以聊天、打游戏）。
那为什么大模型 API 几乎全都用 SSE 呢？

因为大模型只需要“单向推送”。

• 你问问题 → 用普通的 POST 请求发过去。
• AI 回答你 → 用 SSE 一个字一个字推给你。
• 在这个回答过程中，你不需要往服务器再发数据。

所以，SSE 这种“轻量级、单向推送”的技术，完美契合了大模型的需求。用 WebSocket 属于“杀鸡用牛刀”，反而更复杂。

2.4 测开视角：流式响应怎么测？

测开要验证的是：

1. 数据块完整性：有没有收到完整的数据块？会不会断流？
2. 结束标记：有没有收到 [DONE] 标记？
3. 异常场景：服务端报错时，有没有返回规范的错误格式，而不是直接断开连接？

这次的测试项目，恰恰发现了 SSE 相关的一个 Bug（后面会讲到）。

---

3. 分层设计（框架结构）

ai-sdk-test/
├── api/chat_api.py          # 接口封装层
├── testcases/               # 测试用例层
│   ├── test_models.py       # 模型接口测试
│   ├── test_session.py      # 会话接口测试
│   └── test_message.py      # 消息接口测试
├── conftest.py              # 全局 Fixture
├── utils/db_utils.py        # 数据库工具
├── data/test_data.yaml      # 测试数据
├── pytest.ini               # Pytest 配置
└── requirements.txt         # 依赖清单

为什么这么分层？

• api/：只负责“怎么发请求”，接口地址变了只改这个文件。
• data/：测试数据跟代码分离，改数据不用改代码。
• testcases/：只负责“断言什么”，关注业务逻辑。
• utils/：只负责“查数据库”这种辅助操作。
• conftest.py：只负责“测试环境准备和清理”。

---

4. 接口封装层 api/chat_api.py

完整代码

import requests

BASE_URL = "http://localhost:8080"


class ChatAPI:
    """AI SDK 接口封装"""

    @staticmethod
    def get_models():
        """获取模型列表"""
        return requests.get(f"{BASE_URL}/api/models")

    @staticmethod
    def create_session(model="deepseek-chat"):
        """创建会话"""
        return requests.post(f"{BASE_URL}/api/session", json={"model": model})

    @staticmethod
    def get_sessions():
        """获取会话列表"""
        return requests.get(f"{BASE_URL}/api/sessions")

    @staticmethod
    def delete_session(session_id):
        """删除会话"""
        return requests.delete(f"{BASE_URL}/api/session/{session_id}")

    @staticmethod
    def get_history(session_id):
        """获取历史消息"""
        return requests.get(f"{BASE_URL}/api/session/{session_id}/history")

    @staticmethod
    def send_message(session_id, message):
        """发送消息（全量返回）"""
        return requests.post(
            f"{BASE_URL}/api/message",
            json={"session_id": session_id, "message": message}
        )

    @staticmethod
    def send_message_stream(session_id, message):
        """发送消息（流式返回 SSE）"""
        return requests.post(
            f"{BASE_URL}/api/message/async",
            json={"session_id": session_id, "message": message},
            stream=True
        )

逐段讲解

① 为什么要定义 BASE_URL？

BASE_URL = "http://localhost:8080"

把接口地址抽出来定义一次。如果服务器换 IP，只改这一行就行。

② 为什么用 class ChatAPI 封装？
把相关的接口都“打包”在一个类里，看起来更整齐，也方便后续维护。

③ 为什么用 @staticmethod？
表示这是静态方法，不需要创建 ChatAPI() 实例就能直接调用 ChatAPI.get_models()。

④ requests.post(..., stream=True) 里的 stream=True 是什么意思？
告诉 requests 库不要一次性把响应读完，而是“边收边处理”。这是专门为流式响应设计的。如果不加这个参数，requests 会一直等到服务器把所有数据都发完，才把结果返回给你——这就不叫流式了。

---

5. 全局 Fixture conftest.py

完整代码

import pytest
from api.chat_api import ChatAPI


@pytest.fixture(scope="function")
def session_id():
    """前置：创建一个新会话；后置：自动删除该会话"""
    # 前置
    res = ChatAPI.create_session("deepseek-chat")
    assert res.status_code == 200, f"Fixture 创建会话失败: {res.text}"
    sid = res.json()["data"]["session_id"]
    print(f"\n[Fixture] ✅ 创建会话成功: {sid}")

    yield sid

    # 后置（清理数据，保证测试环境干净）
    ChatAPI.delete_session(sid)
    print(f"\n[Fixture] 🧹 删除会话完成: {sid}")

逐段讲解

① @pytest.fixture(scope="function") 是什么？

• @pytest.fixture：告诉 Pytest 这是一个“前置/后置钩子”。
• scope="function"：作用范围是“每个测试函数”。意思是每个测试用例执行前都会重新跑一遍这个 Fixture，保证数据独立。

② yield sid 是什么？

• yield 之前的代码 = 前置操作（准备数据）。
• yield 的值 = 交给测试用例用（这里是 sid）。
• yield 之后的代码 = 后置操作（清理数据）。无论测试通过还是失败，后置代码都会执行。

---

6. 测试用例层 testcases/test_models.py

完整代码

from api.chat_api import ChatAPI


class TestModels:

    def test_get_models_success(self):
        """正常：获取模型列表"""
        res = ChatAPI.get_models()
        assert res.status_code == 200
        data = res.json()
        assert data["success"] is True
        assert data["message"] == "get model lists success"

        model_names = [m["name"] for m in data["data"]]
        assert "deepseek-chat" in model_names
        print(f"\n✅ 模型列表: {model_names}")

    def test_models_structure(self):
        """校验每个模型都包含 name 和 desc 字段"""
        data = ChatAPI.get_models().json()
        for model in data["data"]:
            assert "name" in model, "模型缺少 name 字段"
            assert "desc" in model, "模型缺少 desc 字段"
        print(f"\n✅ 共 {len(data['data'])} 个模型，结构完整")

逐段讲解

① assert 是什么？
assert 是 Python 的断言语句。如果条件为假，就会抛出 AssertionError，Pytest 就会把这个用例标记为 FAILED。

② [m["name"] for m in data["data"]] 是什么？
这是列表推导式，把每个模型的 name 提取出来，生成一个新的列表。

---

7. 测试用例层 testcases/test_session.py

完整代码

import pytest
from api.chat_api import ChatAPI


class TestSession:

    def test_create_session_success(self):
        """正常：创建会话"""
        res = ChatAPI.create_session("deepseek-chat")
        assert res.status_code == 200
        data = res.json()
        assert data["success"] is True
        assert "session_id" in data["data"]
        assert data["data"]["model"] == "deepseek-chat"
        ChatAPI.delete_session(data["data"]["session_id"])
        print(f"\n✅ 创建会话成功: {data['data']['session_id']}")

    def test_get_sessions(self):
        """正常：获取会话列表"""
        res = ChatAPI.get_sessions()
        assert res.status_code == 200
        data = res.json()
        assert data["success"] is True
        assert isinstance(data["data"], list)
        print(f"\n✅ 会话总数: {len(data['data'])}")

    def test_delete_session_success(self):
        """正常：删除会话"""
        res = ChatAPI.create_session("deepseek-chat")
        sid = res.json()["data"]["session_id"]
        del_res = ChatAPI.delete_session(sid)
        assert del_res.status_code == 200
        assert del_res.json()["success"] is True
        print(f"\n✅ 删除会话成功: {sid}")

    def test_delete_session_not_found(self):
        """异常：删除不存在的会话，返回 404"""
        res = ChatAPI.delete_session("session_not_exist_999")
        assert res.status_code == 404
        assert res.json()["success"] is False
        print(f"\n✅ 删除不存在会话返回 404")

    def test_get_history_empty(self, session_id):
        """正常：新建会话的历史消息应为空"""
        res = ChatAPI.get_history(session_id)
        assert res.status_code == 200
        data = res.json()
        assert data["data"] == []
        print(f"\n✅ 新建会话历史为空")

    @pytest.mark.xfail(reason="已知Bug：查询不存在会话返回500而非404")
    def test_get_history_not_found(self):
        """异常：查询不存在会话的历史，返回 404"""
        res = ChatAPI.get_history("session_fake_999")
        assert res.status_code == 404

逐段讲解

① isinstance(data["data"], list) 是什么？
判断 data["data"] 是不是一个列表，验证返回的数据类型。

② @pytest.mark.xfail 是什么？
xfail = expected to fail（预期失败）。这个用例肯定会失败（服务端返回 500 而不是 404），但我们提前声明“我知道它要挂，而且知道原因”。Pytest 会把它标记为 XFAIL，而不是 FAILED。

③ 为什么这个用例用了 xfail？
因为我们真实发现了一个 Bug：查询不存在的会话历史时，服务端返回了 500（服务器错误），而不是 404（资源不存在）。

---

8. 测试用例层 testcases/test_message.py

完整代码

import pytest
from api.chat_api import ChatAPI


class TestMessage:

    @pytest.mark.xfail(reason="已知Bug：temperature 精度问题导致服务端返回500")
    def test_send_message_success(self, session_id):
        """正常：发送消息（全量返回）"""
        res = ChatAPI.send_message(session_id, "你好，请用一句话介绍你自己")
        assert res.status_code == 200
        data = res.json()
        assert data["success"] is True
        assert "response" in data["data"]
        assert len(data["data"]["response"]) > 0
        print(f"\n✅ AI回复: {data['data']['response'][:80]}...")

    def test_send_message_empty(self, session_id):
        """异常：发送空消息，返回 400"""
        res = ChatAPI.send_message(session_id, "")
        assert res.status_code == 400
        assert res.json()["success"] is False
        print(f"\n✅ 空消息正确返回 400")

    def test_send_message_invalid_session(self):
        """异常：使用不存在的 session_id，返回 500"""
        res = ChatAPI.send_message("session_fake_999", "你好")
        assert res.status_code == 500
        print(f"\n✅ 非法session正确返回 500")

    @pytest.mark.xfail(reason="已知Bug：流式接口异常中断，未返回SSE错误格式")
    def test_send_message_stream(self, session_id):
        """正常：流式发送消息，验证 SSE 数据块及 [DONE] 标记"""
        res = ChatAPI.send_message_stream(session_id, "你好")
        assert res.status_code == 200

        chunks = []
        for line in res.iter_lines(decode_unicode=True):
            if line:
                chunks.append(line)

        print(f"\n✅ 收到 {len(chunks)} 个流式数据块")
        assert any("[DONE]" in c for c in chunks), "流式响应缺少 [DONE] 标记"

    @pytest.mark.xfail(reason="已知Bug：消息发送失败导致持久化未执行")
    def test_send_message_and_check_history(self, session_id):
        """端到端：发消息后，历史记录应包含 user 和 assistant 两条消息"""
        ChatAPI.send_message(session_id, "今天天气怎么样？")
        res = ChatAPI.get_history(session_id)
        assert res.status_code == 200
        data = res.json()
        assert len(data["data"]) >= 2, f"历史消息应至少2条，实际: {len(data['data'])}"

        roles = [m["role"] for m in data["data"]]
        assert "user" in roles, "历史消息缺少 user 角色"
        assert "assistant" in roles, "历史消息缺少 assistant 角色"
        print(f"\n✅ 历史消息条数: {len(data['data'])}, 角色: {roles}")

逐段讲解

① res.iter_lines(decode_unicode=True) 是什么？
这是 requests 库里专门处理流式响应的方法。它会一行一行地读取服务器推送过来的数据块，而不是一次性读完整。

② 为什么最后要验证 [DONE]？
SSE 协议规定，流式数据传完后，服务器会发一个 data: [DONE] 表示“我讲完了”。如果客户端没收到它，就说明连接被异常中断了。

③ 为什么第一个用例和最后一个用例都 xfail 了？
因为它们都依赖 send_message 接口，而这个接口因为 temperature 精度 Bug 会返回 500，所以：

• 第一个用例：预期 200，实际 500。
• 最后一个用例：发消息失败，数据库里当然没记录。

---

9. 数据库校验工具 utils/db_utils.py

完整代码

import os
import sqlite3


def get_db_path():
    """根据服务器启动目录，定位 chatDB.db"""
    possible_paths = [
        os.path.expanduser(
            "~/workspace/ai-model-acess-tech/AIModelAcessTech/ChatServer/build_new/chatDB.db"
        ),
        "chatDB.db",
    ]
    for p in possible_paths:
        if os.path.exists(p):
            return p
    return None


def query_messages_by_session(session_id):
    """从 messages 表查询指定会话的所有消息"""
    db_path = get_db_path()
    if not db_path:
        return None
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()
    cursor.execute(
        "SELECT message_id, role, content FROM messages WHERE session_id = ?",
        (session_id,),
    )
    rows = cursor.fetchall()
    conn.close()
    return rows

逐段讲解

① 为什么要用 WHERE session_id = ? 而不是拼接字符串？
这是参数化查询，可以防止 SQL 注入攻击。

② 这个工具在测试里怎么用？

from utils.db_utils import query_messages_by_session

def test_message_persisted(self, session_id):
    ChatAPI.send_message(session_id, "你好")
    rows = query_messages_by_session(session_id)
    assert len(rows) >= 1, "消息没有落库！"

---

10. 测试结果与 Allure 报告

运行命令

# 1. 安装依赖
pip3 install -r requirements.txt

# 2. 执行测试（生成 Allure 数据）
pytest --alluredir=./report

# 3. 启动 Allure 报告服务
allure serve ./report

测试结果汇总

测试文件	用例数	通过	XFAIL（已知缺陷）
test_models.py	2	2	0
test_session.py	6	5	1
test_message.py	5	2	3
合计	13	9	4

11. 发现的 4 个 Bug

Bug 1：temperature 浮点数精度问题导致请求失败（500）

现象：发送消息接口返回 500，服务端日志报 Connection handling canceled。
定位：抓包发现请求体里的 temperature 被序列化成 0.69999999999999996，DeepSeek API 拒绝了该精度。
修复建议：使用 std::setprecision(2) 格式化后再发送。

Bug 2：流式接口异常中断，未返回规范 SSE 错误格式

现象：调用 /api/message/async 时，客户端抛出 ChunkedEncodingError: IncompleteRead(0 bytes read)。
定位：服务端请求第三方 API 失败后，直接断开了连接，没有返回 data: {"error": "..."}。
影响：前端无法优雅地提示用户“服务暂时不可用”。

Bug 3：消息发送失败导致持久化未执行

现象：发消息后查历史记录，数据库里是空的。
定位：因为 Bug 1 导致发送失败，后续的 insertMessage 根本没执行。

Bug 4：查询不存在会话返回 500 而非 404

现象：查询一个不存在的 session_id 的历史记录，服务端返回了 500，而不是 404。
定位：ChatServer.cpp 里虽然写了 response.status = 404，但 getSession 内部抛出了 C++ 异常，在进入 404 分支前就崩溃了。
修复建议：在 getSession 里用 find() 代替 .at()，找不到时返回 nullptr 而不是抛异常。

12. 项目链接

Gitee 仓库：https://gitee.com/lin-kailuoluo/ai-sdk-test

13. 总结

通过这个项目，我不仅掌握了 Pytest + Requests + Allure 的实战用法，更锻炼了从源码层面定位 Bug 根因的能力。

我还搞懂了流式响应（SSE）的测试方法，这是当前 AI 测试领域非常核心的技能。

测开的核心价值，不是跑通接口，而是保障系统的稳定性与健壮性。

如果你也在学习接口自动化，希望这篇博客能给你一些启发。欢迎在评论区交流！

转载自 CSDN-专业IT技术社区

原文链接：https://blog.csdn.net/zoelinkailuo/article/details/165491520