Compare commits

...
@@ -0,0 +1,82 @@
"""Phase 0 回归用例 R1/R2/R3 — 固化三个反复出现的问题。
R1: 技能列表不得被硬性截断为 15 条(微信 Bot 曾只返回 15 个技能)。
R2: 显式选中技能/MCP 时,system prompt 必须注入「最高优先级覆盖规则」。
R3: 媒体生成(图像/视频)必须有超时保护,超时后回写用户可懂的提示而非永久挂起。
说明:在 chat.py 巨石(10070 行)尚未拆分前,R1/R2 采用「源码契约」守卫,
这是最低风险且能切实挡住回归的方式;R3 为行为级单元测试(mock 内部生成调用模拟超时)。
Phase 1 抽出 chat_media / chat_mcp / chat_skill 干净 seam 后,R1/R2 应升级为行为测试。
"""
import asyncio
import os
from unittest.mock import MagicMock
import pytest
_HERE = os.path.dirname(os.path.abspath(__file__))
CHAT_PY = os.path.abspath(os.path.join(_HERE, "..", "app", "api", "chat.py"))
def _read_chat_source():
with open(CHAT_PY, "r", encoding="utf-8") as f:
return f.read()
# ---------- R1 ----------
def test_r1_skill_list_not_hard_capped_at_15():
"""微信 Bot 技能列表曾因 limit(15) 只返回 15 个,必须防止回归。"""
src = _read_chat_source()
assert "limit(15)" not in src, "检测到 limit(15),技能列表可能被硬性截断"
assert "LIMIT 15" not in src, "检测到 LIMIT 15,技能列表可能被硬性截断"
# ---------- R2 ----------
def test_r2_explicit_selection_override_rule_present():
"""显式选中技能/MCP 时,system prompt 必须包含最高优先级覆盖规则。"""
src = _read_chat_source()
assert "最高优先级覆盖规则" in src, "显式选中工具的覆盖规则文本缺失"
assert "只基于上文提供的该工具具体资料作答" in src, "覆盖规则应要求只基于已选工具资料作答"
assert "平台知识库" in src and "不要引用" in src, "覆盖规则应明确禁止引用平台知识库通用描述"
# ---------- R3 ----------
@pytest.mark.asyncio
async def test_r3_media_generation_timeout_returns_user_message(monkeypatch):
"""媒体生成超时必须回写用户可懂的超时提示,而非让前端永远『生成中…』。
注:_generate_media_task 是后台 fire-and-forget 任务(通过 asyncio.create_task 调用),
本身返回 None,超时保护体现在把提示回写到消息 content。故此处断言副作用。
"""
from app.api import chat as chat_mod
class FakeModel:
id = "m_test"
capabilities = ["image"] # 图像 -> 90s 超时
msg_mock = MagicMock() # 捕获被回写的消息对象
fake_session = MagicMock()
def _query_side_effect(model_cls):
q = MagicMock()
if model_cls is chat_mod.AIModel:
q.filter.return_value.first.return_value = FakeModel()
elif model_cls is chat_mod.Message:
q.filter.return_value.first.return_value = msg_mock
return q
fake_session.query.side_effect = _query_side_effect
monkeypatch.setattr(chat_mod, "SessionLocal", lambda: fake_session)
# 内部生成调用直接抛出 TimeoutError,等价模拟「生成耗时超过超时阈值」
async def _hang(db, ai_model, prompt, user_id):
raise asyncio.TimeoutError()
monkeypatch.setattr(chat_mod, "_generate_media", _hang)
await chat_mod._generate_media_task("msg1", "m_test", "画一只猫", "u1")
# 副作用:消息 content 被回写为用户可懂的超时提示
assert msg_mock.content is not None, "超时后必须回写消息内容,不能让前端永远『生成中…』"
assert "超时" in str(msg_mock.content), "媒体生成超时应回写用户可懂的超时提示"