| name | write-tests |
| description | 编写新测试文件或为现有模块补充测试时使用,覆盖命名约定、速度要求、Mock 模式与断言规范。关键词:测试编写、单元测试、pytest、MockProvider、测试速度、精准测试。 |
测试编写指南
新测试应针对 EmotionalGroupChatEngine 及其子系统(BasicMemoryManager、DiaryManager、SemanticMemoryManager、ContextAssembler、UserManager 等)编写。
目标
为 Sirius Pulse 编写正确、快速、可维护的 pytest 测试。测试必须从业务侧出发,依照业务逻辑和用户实际使用路径设计,优先验证用户输入、公开入口、系统处理与可观察结果之间的闭环。所有测试必须满足四个核心要求:
- 快(Fast):单个测试用例应在 1 秒以内 完成;整个测试套件目标 30 秒以内。
- 业务侧(Business-first):测试场景应对应用户实际使用方式,避免只验证私有方法、内部字段或临时实现细节。
- 准(Precise):一个测试函数只验证一个业务概念,断言直接指向用户可感知的行为。
- 稳(Stable):无真实网络调用、无随机时间依赖、无副作用残留。
零、业务侧测试原则
测试用例应先回答“用户如何使用这个能力”,再决定调用哪个入口和断言什么结果。优先从公开 API、引擎入口、CLI/配置入口、WebUI API、持久化结果等用户可触达路径验证行为。
推荐覆盖
- 用户发送消息后,引擎是否依据人格、上下文、记忆和策略给出合理响应。
- 用户配置 Provider、人格、技能或平台适配后,系统是否产生符合预期的业务行为。
- 记忆、事件、技能执行等内部模块是否在真实调用链中协作正确。
- 失败、空输入、非法配置、Provider 异常等场景是否返回用户可理解的降级结果。
禁止倾向
- 只为了覆盖率直接测试私有方法或内部字段。
- 把实现步骤逐行复刻成测试,导致重构时业务没变但测试大量失败。
- 断言无业务意义的中间变量、调用次数或临时数据结构。
一、速度红线
禁止的写法
import asyncio
await asyncio.sleep(8)
time.sleep(3)
engine.start_background_tasks()
标准配置模板(绝大多数测试应使用此配置)
engine = EmotionalGroupChatEngine(
work_path=tmp_path,
persona=PersonaProfile(name="TestBot"),
config={
"sensitivity": 0.0,
},
)
二、标准 Provider 与引擎初始化
from sirius_pulse.core.emotional_engine import EmotionalGroupChatEngine
from sirius_pulse.models.persona import PersonaProfile
from sirius_pulse.models.models import Message, Participant
from sirius_pulse.providers.mock import MockProvider
provider = MockProvider(
responses=[
"第一条回复",
"第二条回复",
]
)
engine = EmotionalGroupChatEngine(
work_path=tmp_path,
persona=PersonaProfile(name="TestBot"),
provider_async=provider,
)
MockProvider 按序返回 responses,用完后循环最后一条。可通过 provider.requests 检查引擎向 LLM 发出的所有请求。
三、标准 process_message 测试模式
@pytest.mark.asyncio
async def test_process_message_replies_to_greeting(tmp_path):
provider = MockProvider(responses=["你好呀!"])
engine = EmotionalGroupChatEngine(
work_path=tmp_path,
persona=PersonaProfile(name="TestBot"),
provider_async=provider,
)
participant = Participant(name="u1", user_id="u1")
msg = Message(role="human", content="大家好", speaker="u1")
reply = await engine.process_message(msg, [participant], "group_a")
assert reply is not None
assert "你好" in reply.content
四、异步测试写法
所有涉及引擎的测试必须使用 pytest.mark.asyncio:
import pytest
@pytest.mark.asyncio
async def test_something(tmp_path):
engine = EmotionalGroupChatEngine(work_path=tmp_path, ...)
...
非异步逻辑(纯数据模型、工具函数)可用普通 def。
五、断言规范
推荐断言模式
| 场景 | 推荐写法 | 不推荐 |
|---|
| 验证回复内容 | assert "关键词" in reply.content | 完整字符串匹配(易因微调而失败) |
| 验证事件类型 | assert SessionEventType.X in types | 仅验证事件数量 |
| 验证记忆写入 | assert len(manager.entries) == 1 | 隐式依赖其他测试的数据 |
| 验证异常 | pytest.raises(ValueError, match="...") | try/except + assert False |
断言应优先落在业务结果上,例如回复内容、事件类型、记忆可检索性、配置生效结果、错误提示和持久化文件内容。只有当调用次数、内部字段或中间结构本身是业务契约时,才应直接断言这些实现细节。
六、文件组织规范
测试文件应按被测模块命名:
| 被测模块 | 测试文件 |
|---|
memory/basic/manager.py | tests/test_basic_memory.py |
memory/diary/manager.py | tests/test_diary_memory.py |
memory/semantic/manager.py | tests/test_semantic_memory.py |
memory/context_assembler.py | tests/test_context_assembler.py |
core/emotional_engine.py | tests/test_engine_event_stream.py |
core/prompt_factory.py | tests/test_response_assembler.py |
core/cognition.py | tests/test_cognition_analyzer.py |
core/response_strategy.py | tests/test_response_strategy.py |
skills/registry.py | tests/test_skill_system.py |
skills/executor.py | tests/test_skill_system.py |
providers/routing.py | tests/test_provider_routing.py |
providers/base.py | tests/test_providers.py |
token/utils.py | tests/test_token_utils.py |
token/store.py | tests/test_token_store.py |
config/manager.py | tests/test_config_manager.py |
models/models.py | tests/test_public_api.py |
session/store.py | tests/test_api_integrity.py |
persona_manager.py | tests/test_persona_integration.py |
persona_worker.py | tests/test_persona_integration.py |
identity_resolver.py | tests/test_identity_resolver.py |
logging_config.py | tests/test_logging_config.py |
exceptions.py | tests/test_exceptions.py |
七、Mock 技巧速查
MockProvider
from sirius_pulse.providers.mock import MockProvider
provider = MockProvider(responses=["reply1", "reply2"])
assert len(provider.requests) == 2
assert "system" in provider.requests[0]["messages"][0]["role"]
手动 patch 依赖
from unittest.mock import patch
with patch("sirius_pulse.memory.diary.generator.DiaryGenerator.generate") as mock_gen:
mock_gen.return_value = DiaryEntry(...)
八、完整示例
"""Tests for BasicMemoryManager heat tracking."""
from __future__ import annotations
import pytest
from sirius_pulse.memory.basic.manager import BasicMemoryManager
class TestBasicMemoryHeat:
def test_new_group_has_zero_heat(self):
mgr = BasicMemoryManager()
assert mgr.compute_heat("g1") == 0.0
def test_heat_increases_after_messages(self):
mgr = BasicMemoryManager()
for i in range(5):
mgr.add_entry("g1", f"u{i}", "user", f"msg{i}")
assert mgr.compute_heat("g1") > 0.0