#!/usr/bin/env python3 """探索工具 server（read_tools）的离线测试：不连真实库。 固定工具合同：五个只读工具按登记表执行、未知工具与非法参数拒绝、 结果有界截断、登记表 CLI 输出结构稳定；并校验派发器的工具开关门禁 （探索工具未启用工具 server 时失败关闭）与依赖清单产出。 """ from __future__ import annotations import json import pathlib import subprocess import sys import unittest PROJECT_ROOT = pathlib.Path(__file__).resolve().parents[3] SKILL_DIR = PROJECT_ROOT / "muse" / "authority" / "tools" / "read" FRAMEWORK_EXTENSION = PROJECT_ROOT / "framework" / "adapters" / "pi" / "mcp_bridge.ts" for path in (PROJECT_ROOT, SKILL_DIR): if str(path) not in sys.path: sys.path.insert(0, str(path)) from read_tools import TOOL_REGISTRY, execute_tool # noqa: E402 READ_TOOLS_PY = SKILL_DIR / "read_tools.py" PYTHON = sys.executable class FakeCursor: def __init__(self, rows): self._rows = rows def fetchone(self): return self._rows[0] if self._rows else None def fetchall(self): return self._rows class FakeConn: """记录每条 SQL 与参数、按构造行返回的假连接。""" def __init__(self, rows): self.rows = rows self.queries = [] def execute(self, sql, params=None): self.queries.append((sql, params)) return FakeCursor(self.rows) def close(self): pass class ToolBehaviourTest(unittest.TestCase): def test_read_fine_outline_hit_and_miss(self): conn = FakeConn([({"chapterGoal": "g"}, 3)]) result = execute_tool( "read_fine_outline", {"work_id": 12, "target_chapter": 4}, connect=lambda: conn ) self.assertTrue(result["found"]) self.assertEqual(result["version"], 3) self.assertEqual(result["payload"], {"chapterGoal": "g"}) sql, params = conn.queries[0] self.assertIn("state='confirmed'", sql) self.assertEqual(params, (12, 4)) conn = FakeConn([]) result = execute_tool( "read_fine_outline", {"work_id": 12, "target_chapter": 9}, connect=lambda: conn ) self.assertFalse(result["found"]) def test_read_style_constraints_dict_projection(self): conn = FakeConn([({"句式": "短句", "用词质感": "冷"},)]) result = execute_tool("read_style_constraints", {"work_id": 12}, connect=lambda: conn) self.assertEqual(result["constraints"], ["句式：短句", "用词质感：冷"]) def test_read_style_falls_back_to_setting_row(self): class TwoStepConn: def __init__(self): self._step = 0 def execute(self, sql, params=None): self._step += 1 rows = [] if self._step == 1 else [({"style": "一句话文风"},)] return FakeCursor(rows) def close(self): pass result = execute_tool("read_style_constraints", {"work_id": 12}, connect=TwoStepConn) self.assertEqual(result["constraints"], ["一句话文风"]) def test_read_pattern_bindings(self): conn = FakeConn([({"patternReferences": [{"id": 1}, {"x": 2}]},)]) result = execute_tool("read_pattern_bindings", {"work_id": 12}, connect=lambda: conn) self.assertEqual(result["bindings"], [{"id": 1}, {"x": 2}]) conn = FakeConn([]) result = execute_tool("read_pattern_bindings", {"work_id": 12}, connect=lambda: conn) self.assertEqual(result["bindings"], []) def test_read_chapter_text_truncates(self): class ChapterConn: def __init__(self): self._step = 0 def execute(self, sql, params=None): self._step += 1 rows = [(1, 3, "第三章")] if self._step == 1 else [("字" * 40000,)] return FakeCursor(rows) def close(self): pass result = execute_tool( "read_chapter_text", {"work_id": 12, "chapter_order": 3}, connect=ChapterConn ) self.assertTrue(result["found"]) self.assertTrue(result["truncated"]) self.assertEqual(len(result["text"]), 30000) self.assertEqual(result["totalChars"], 40000) def test_search_entities_like_and_bounds(self): conn = FakeConn([("character", "林深", "主角", "active")]) result = execute_tool( "search_entities", {"work_id": 12, "keyword": "林"}, connect=lambda: conn ) self.assertEqual(result["entities"][0]["name"], "林深") sql, params = conn.queries[0] self.assertIn("ILIKE", sql) self.assertEqual(params, (12, "%林%", 50)) def test_unknown_tool_and_bad_args_rejected(self): with self.assertRaises(KeyError): execute_tool("drop_table", {}, connect=lambda: FakeConn([])) with self.assertRaises(ValueError): execute_tool( "search_entities", {"work_id": 12, "keyword": ""}, connect=lambda: FakeConn([]) ) with self.assertRaises(ValueError): execute_tool( "read_fine_outline", {"work_id": "x", "target_chapter": 1}, connect=lambda: FakeConn([]), ) class RegistryCliTest(unittest.TestCase): def test_registry_shape_and_cli_list(self): self.assertEqual( set(TOOL_REGISTRY), { "read_fine_outline", "read_style_constraints", "read_pattern_bindings", "read_chapter_text", "search_entities", }, ) for entry in TOOL_REGISTRY.values(): self.assertTrue(entry["description"]) self.assertTrue(entry["tables"]) self.assertTrue(entry["args"]) proc = subprocess.run( [PYTHON, str(READ_TOOLS_PY), "--list"], capture_output=True, text=True, timeout=30, ) self.assertEqual(proc.returncode, 0, proc.stderr) listing = json.loads(proc.stdout) self.assertEqual(set(listing), set(TOOL_REGISTRY)) for name, item in listing.items(): self.assertEqual(item["args"], TOOL_REGISTRY[name]["args"]) def test_cli_rejects_unknown_tool_with_contract_exit_code(self): proc = subprocess.run( [PYTHON, str(READ_TOOLS_PY), "drop_table", "{}"], capture_output=True, text=True, timeout=30, ) self.assertEqual(proc.returncode, 2) class ExtensionSyncTest(unittest.TestCase): """扩展不自带工具清单：它必须从登记表 --list 动态装载，防止双源漂移。""" def test_extension_loads_registry_dynamically(self): extension = FRAMEWORK_EXTENSION.read_text(encoding="utf-8") self.assertIn("--list", extension) self.assertIn("registerTool", extension) self.assertIn("MUSE_READ_TOOLS_PYTHON", extension) self.assertIn("MUSE_READ_TOOLS_SCRIPT", extension) # 扩展不得硬编码登记表之外的工具名（只能出现动态注册语句）。 for name in ("read_fine_outline", "search_entities"): self.assertNotIn(f'name: "{name}"', extension) if __name__ == "__main__": unittest.main()