Files

renjianbo 7b9e0826de feat: 向量记忆 RAG、工具市场、SSE 流式响应、前端集成与测试覆盖

- 新增 embedding_service（语义检索）、knowledge_service（RAG）、text_chunker、document_parser
- 新增 tool_registry（自定义工具注册表）并完善工具市场 API（CRUD + code/http 执行）
- 新增 agent_vector_memory / knowledge_base 模型及对应数据库表
- 实现 SSE 流式响应与 Agent 预算控制
- AgentChat.vue 集成 MainLayout 导航布局
- 完善测试体系：7 个新测试文件共 110 个测试覆盖
- 修复 conftest.py SQLite 内存数据库连接隔离问题

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

2026-05-01 22:30:46 +08:00

5.0 KiB

Raw Blame History

单元测试说明

测试框架

本项目使用 pytest 作为测试框架，支持异步测试和数据库测试。

运行测试

运行所有测试

cd backend
pytest

运行特定测试文件

pytest tests/test_auth.py

运行特定测试类或函数

pytest tests/test_auth.py::TestAuth::test_register_user

运行带标记的测试

# 只运行单元测试
pytest -m unit

# 只运行工作流相关测试
pytest -m workflow

# 只运行认证相关测试
pytest -m auth

运行并显示覆盖率

pytest --cov=app --cov-report=html

测试标记

@pytest.mark.unit - 纯单元测试（不依赖网络/数据库）
@pytest.mark.integration - 集成测试（需要网络或可选依赖）
@pytest.mark.slow - 慢速测试（需要网络或数据库）
@pytest.mark.api - API测试
@pytest.mark.workflow - 工作流测试
@pytest.mark.auth - 认证测试
@pytest.mark.asyncio - 异步测试（使用 pytest-asyncio）

测试结构

tests/
├── __init__.py
├── conftest.py                  # 共享fixtures和配置
├── test_auth.py                 # 认证API测试
├── test_workflows.py            # 工作流API测试
├── test_workflow_engine.py      # 工作流引擎测试
├── test_workflow_validator.py   # 工作流验证器测试
├── test_text_chunker.py         # 文本分块器单元测试
├── test_document_parser.py      # 文档解析器单元测试
├── test_tool_registry.py        # 工具注册表单元测试
├── test_tools_api.py            # 工具市场API测试
├── test_agent_memory.py         # Agent记忆/上下文/Schema测试
├── test_embedding_service.py    # Embedding服务单元测试
└── test_knowledge_base.py       # 知识库RAG单元测试

Fixtures

`db_session`

创建测试数据库会话，每个测试函数都会获得一个独立的会话。

`client`

创建FastAPI测试客户端，用于API测试。

`authenticated_client`

创建已认证的测试客户端，自动注册用户并登录。

`test_user_data`

提供测试用户数据。

`sample_workflow_data`

提供示例工作流数据。

测试数据库

测试使用SQLite临时文件数据库（而非 :memory:），每个测试函数都会：

创建所有表
执行测试
删除所有表

这样可以确保测试之间的隔离性。

注意：使用临时文件而非 :memory: 是因为 FastAPI + TestClient 在异步/多线程请求中，SQLite 内存数据库会发生"连接隔离"问题——每个连接看到不同的数据库实例。文件数据库确保了所有连接共享同一份数据。

注意：纯服务层测试（如 test_tool_registry.py、test_embedding_service.py、test_text_chunker.py）不依赖 db_session fixture，它们直接对服务类打桩或使用临时文件，运行速度更快。

异步测试配置

使用 pytest-asyncio 支持异步测试。所有被 @pytest.mark.asyncio 标记的测试函数必须使用 async def 声明：

@pytest.mark.unit
@pytest.mark.asyncio
async def test_my_async_method(self):
    result = await my_service.my_method()
    assert result is not None

注意：conftest.py 中的 fixtures 默认使用 function 作用域，异步 fixture 不需要额外配置。

编写新测试

示例：API测试

@pytest.mark.unit
@pytest.mark.api
class TestMyAPI:
    def test_my_endpoint(self, authenticated_client):
        response = authenticated_client.get("/api/v1/my-endpoint")
        assert response.status_code == 200

示例：服务测试

@pytest.mark.unit
class TestMyService:
    @pytest.mark.asyncio
    async def test_my_service_method(self):
        service = MyService()
        result = await service.my_method()
        assert result is not None

注意事项

测试隔离：每个测试函数都应该独立，不依赖其他测试的执行顺序。
数据库清理：使用 db_session fixture 确保每个测试都有干净的数据库。
异步测试：使用 @pytest.mark.asyncio 标记异步测试函数。
标记测试：使用适当的标记（@pytest.mark.unit 等）来组织测试。
测试数据：使用 fixtures 提供测试数据，避免硬编码。
已知问题：有 9 个历史遗留测试失败（主要在 test_auth.py、test_workflow_engine.py、test_workflow_validator.py、test_nodes_all.py、test_nodes_phase4.py 中），这些失败与本次改造无关，可忽略检查新建的测试文件时排除它们：
```
pytest tests/test_text_chunker.py tests/test_document_parser.py \
        tests/test_tool_registry.py tests/test_tools_api.py \
        tests/test_agent_memory.py tests/test_embedding_service.py \
        tests/test_knowledge_base.py -v
```

CI/CD集成

在CI/CD流程中运行测试：

# .github/workflows/test.yml
- name: Run tests
  run: |
    cd backend
    pytest --cov=app --cov-report=xml

5.0 KiB Raw Blame History Unescape Escape