196 lines
11 KiB
Python
196 lines
11 KiB
Python
"""Dual-Claw 回归哨兵 —— R-DUAL-*（设计书 leagent_accel_belt_design.md §6，任务 #873）。
|
||
|
||
硬约束：Web Claw 与 IM Bot Claw 必须消费**同一份**能力清单与**同一套**工具命名。
|
||
本文件是「活体哨兵」：任何把 IM Claw 能力改回退（如 #866 覆盖 #764/#765 导致
|
||
mcp=[]/agents=[] 写死）的提交，会在 pre-push 阶段立即变红。
|
||
|
||
历史教训（见设计书 §2）：双端对齐若靠「在第二份实现里再写一遍」必然被下一次覆盖
|
||
回退。**唯一根治法是消灭第二份实现**——即 im_llm 必须复用统一的 capability_discovery。
|
||
|
||
测试分两层：
|
||
A. 始终运行（确定性、无网络）：源码契约守卫 + DB 级能力一致性 + 命名同源守卫。
|
||
B. 可选运行（需联网 MCP/LLM）：置 RUN_DUAL_INTEGRATION=1 时执行真实工具调用，
|
||
作为持续哨兵；默认 skip，避免 CI 网关因外部网络抖动而误杀正常提交。
|
||
"""
|
||
import os
|
||
import re
|
||
import glob
|
||
import sys
|
||
|
||
import pytest
|
||
|
||
_HERE = os.path.dirname(os.path.abspath(__file__))
|
||
APP_DIR = os.path.abspath(os.path.join(_HERE, "..", "app"))
|
||
API_DIR = os.path.join(APP_DIR, "api")
|
||
|
||
sys.path.insert(0, os.path.join(_HERE, ".."))
|
||
|
||
from app.database import SessionLocal # noqa: E402
|
||
from app.core.capability_discovery import ( # noqa: E402
|
||
discover_skills,
|
||
discover_mcp_servers,
|
||
_safe_name,
|
||
)
|
||
|
||
|
||
def _read_api_sources():
|
||
parts = []
|
||
for p in sorted(glob.glob(os.path.join(API_DIR, "chat*.py"))) + \
|
||
[os.path.join(API_DIR, "im_llm.py")]:
|
||
if os.path.exists(p):
|
||
with open(p, "r", encoding="utf-8") as f:
|
||
parts.append(f.read())
|
||
return "\n".join(parts)
|
||
|
||
|
||
@pytest.fixture(scope="module")
|
||
def db():
|
||
session = SessionLocal()
|
||
yield session
|
||
session.close()
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# R-DUAL-2 (源码守卫)：IM Claw 必须复用统一能力发现，禁止写死 mcp=[]
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
def test_rdual2_im_reuses_unified_discovery():
|
||
"""IM Bot 引擎必须从 capability_discovery 取能力，且不得残留 mcp=[] 写死。
|
||
|
||
这是 #866 覆盖 #764/#765 的精确回归签名：当时 im_llm 把 mcp=[] 硬编码，
|
||
导致 IM 永远拿不到 MCP 工具。本测试捕获该字面量复现。
|
||
"""
|
||
im_src = ""
|
||
im_path = os.path.join(API_DIR, "im_llm.py")
|
||
with open(im_path, "r", encoding="utf-8") as f:
|
||
im_src = f.read()
|
||
|
||
# 1) 必须导入统一发现层
|
||
assert "from app.core.capability_discovery import" in im_src, \
|
||
"im_llm 未从 capability_discovery 导入——IM 可能又退化成第二份实现"
|
||
assert "discover_skills" in im_src, "im_llm 未使用 discover_skills"
|
||
assert "discover_mcp_servers" in im_src, "im_llm 未使用 discover_mcp_servers"
|
||
|
||
# 2) 禁止写死 mcp=[]（agents=[] 是允许的：当前无已发布 agent）
|
||
assert "mcp=[]" not in im_src, \
|
||
"检测到 im_llm 写死 mcp=[]——IM Claw 的 MCP 能力已回退（#866 类回归）"
|
||
|
||
# 3) _build_system_prompt 调用必须传入真实 mcp（而非空列表）
|
||
assert "mcp=mcp_servers" in im_src or "mcp=mcp_servers," in im_src, \
|
||
"im_llm 未把发现的 MCP 服务器注入 system prompt"
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# R-DUAL-1 (DB 级一致性)：双 Claw 返回同一份能力集合
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
def test_rdual1_skill_set_parity(db):
|
||
"""同一用户，Web 与 IM 必须发现相同的技能集合（统一发现层 cliff 保证）。"""
|
||
web = discover_skills(db, "dual-claw-sentinel", claw="web")
|
||
im = discover_skills(db, "dual-claw-sentinel", claw="im")
|
||
|
||
assert len(web) > 0, "Web Claw 未发现任何技能（published fallback 可能失效）"
|
||
web_names = {s.get("name") for s in web}
|
||
im_names = {s.get("name") for s in im}
|
||
assert web_names == im_names, (
|
||
f"Web 与 IM 技能集合不一致：\n web={sorted(web_names)}\n im={sorted(im_names)}"
|
||
)
|
||
|
||
|
||
def test_rdual1_mcp_server_parity(db):
|
||
"""双 Claw 必须挂载相同的 MCP 服务器集合（IM 不再丢 MCP）。"""
|
||
web = discover_mcp_servers(db, None, claw="web")
|
||
im = discover_mcp_servers(db, None, claw="im")
|
||
|
||
assert len(web) > 0, "未发现有启用的 MCP 服务器"
|
||
web_names = sorted(s.get("name") for s in web)
|
||
im_names = sorted(s.get("name") for s in im)
|
||
assert web_names == im_names, (
|
||
f"Web 与 IM 的 MCP 服务器不一致：\n web={web_names}\n im={im_names}"
|
||
)
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# R-DUAL-4 (命名同源)：双 Claw 工具命名必须同源，杜绝第二份实现发散
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
def test_rdual4_tool_naming_single_source():
|
||
"""capability_discovery._safe_name 必须与 Web Claw 的命名公式完全一致。
|
||
|
||
Web Claw (chat_service.py:3428) 用：
|
||
_sname = re.sub(r'[^A-Za-z0-9_-]', '_', name[:30])
|
||
_tname = re.sub(r'[^A-Za-z0-9_-]', '_', name[:50])
|
||
_fname = f"mcp__{_sname}__{_tname}__{mcp_idx}"
|
||
统一层必须用同一个公式，否则两端工具名对不上 → 工具调用在 IM 端失效。
|
||
"""
|
||
# 统一层的公式
|
||
for raw, maxlen in [("My Server!", 30), ("some tool.name", 50), ("中文-工具", 50)]:
|
||
expected = re.sub(r"[^A-Za-z0-9_-]", "_", raw[:maxlen])
|
||
assert _safe_name(raw, maxlen) == expected, \
|
||
f"_safe_name 公式与 Web Claw 不一致: {_safe_name(raw, maxlen)!r} != {expected!r}"
|
||
|
||
# Web Claw 源码必须仍使用同一公式（[:30] 给 server，[:50] 给 tool）
|
||
cs_path = os.path.join(API_DIR, "chat_service.py")
|
||
with open(cs_path, "r", encoding="utf-8") as f:
|
||
cs_src = f.read()
|
||
assert "re.sub(r'[^A-Za-z0-9_-]', '_', " in cs_src, \
|
||
"chat_service.py 工具命名公式缺失/被改写"
|
||
assert "[:30]" in cs_src and "[:50]" in cs_src, \
|
||
"chat_service.py 的 server/tool 截断长度与统一层不匹配（应为 30/50）"
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# R-DUAL-5 (媒体结构契约)：IM 必须保留 __media__ 结构化返回
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
def test_rdual5_im_preserves_media_struct():
|
||
"""微信内联图片依赖 __media__ 结构化返回，禁止在 IM 端退化成纯文本。"""
|
||
im_path = os.path.join(API_DIR, "im_llm.py")
|
||
with open(im_path, "r", encoding="utf-8") as f:
|
||
im_src = f.read()
|
||
assert "__media__" in im_src, \
|
||
"im_llm 缺失 __media__ 结构化返回——微信内联图片将失效（R-DUAL-5）"
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# R-DUAL-2 / R-DUAL-3（真实集成，需联网，默认 skip）
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
_RUN_LIVE = os.environ.get("RUN_DUAL_INTEGRATION") == "1"
|
||
|
||
|
||
@pytest.mark.skipif(not _RUN_LIVE, reason="set RUN_DUAL_INTEGRATION=1 to run live MCP/LLM integration")
|
||
@pytest.mark.asyncio
|
||
async def test_rdual2_live_im_mcp_tools_reachable(db):
|
||
"""真实拉取启用 MCP 服务器的工具列表，证明 IM 端代理绕过已生效（对照 #P0 修复）。"""
|
||
from app.mcp_client import list_mcp_tools
|
||
|
||
servers = discover_mcp_servers(db, None, claw="im")
|
||
assert servers, "没有启用 MCP 服务器可供实时验证"
|
||
|
||
total = 0
|
||
for srv in servers:
|
||
try:
|
||
tools = await list_mcp_tools(srv, timeout=30, raise_on_error=False)
|
||
except Exception:
|
||
tools = []
|
||
total += len(tools or [])
|
||
assert total > 0, "实时拉取 MCP 工具为 0——IM 端 MCP 代理可能再次被代理劫持"
|
||
|
||
|
||
@pytest.mark.skipif(not _RUN_LIVE, reason="set RUN_DUAL_INTEGRATION=1 to run live agent integration")
|
||
def test_rdual3_im_agent_capability(db):
|
||
"""若已发布 agent，IM 端必须能纳入；当前无已发布 agent 则 skip（待 #873 薄壳化）。"""
|
||
from app.models import MarketApp
|
||
|
||
agent_count = db.query(MarketApp).filter(
|
||
MarketApp.type == "agent", MarketApp.status == "published"
|
||
).count()
|
||
if agent_count == 0:
|
||
pytest.skip("尚无已发布 agent（market_apps.type='agent'=0），IM agent 能力待 #873 薄壳化后激活")
|
||
# 待 agent 接入后补充：断言 im_llm 经 EngineCore 获得 agent 编排能力
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# R-DUAL-6（OTel 标签，待 #876）：双 Claw span 须携带 tenant_id + claw_type
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
@pytest.mark.skip(reason="pending #876 OTel enhancement: spans must carry tenant_id + claw_type")
|
||
def test_rdual6_spans_carry_tenant_and_claw():
|
||
"""双 Claw 链路的 span 必须带有 tenant_id 与 claw_type 标签（双端横向对比看板依赖）。"""
|
||
pytest.skip("pending #876")
|