# ============================================================ # 方向 F 黄金测试：MCP 插件管理总线（矩阵 75 行） # 覆盖：注册/manifest、工具契约、权限拒绝+审计、权限开关、 # 启停、版本兼容、健康历史、Skill 桥接（只增不删） # ============================================================ from __future__ import annotations import pytest from server.agent_core.mcp_bus import ( BUS_VERSION, McpBus, ToolExecutionError, ) from server.agent_core.skills import SkillRegistry def _sample_manifest(**overrides): base = { "plugin_id": "mcp.mes", "name": "MES 桥接插件", "version": "2.1.0", "system": "MES", "endpoint": "local://stub", "max_power": "P3", "min_bus_version": "1.0", "tools": [ { "name": "mes.fetch_progress", "description": "查询工单执行进度", "power": "P0", "input_schema": {"type": "object", "properties": {"woId": {"type": "string"}}}, "output_schema": {"type": "object", "properties": {"progress": {"type": "number"}}}, "idempotent": True, "transport": "local", }, { "name": "mes.release_order", "description": "下发工单（写操作）", "power": "P2", "input_schema": {"type": "object", "properties": {"woId": {"type": "string"}}}, "output_schema": {"type": "object"}, "idempotent": False, "transport": "local", }, ], } base.update(overrides) return base def test_register_and_tool_contracts(tmp_path): """注册后 manifest 完整、工具契约（入出参 schema/权力）可查、缺省权限显式化。""" bus = McpBus(path=str(tmp_path / "plugins.json")) plugin = bus.register(_sample_manifest(), actor="planner") assert plugin["plugin_id"] == "mcp.mes" assert plugin["version"] == "2.1.0" tools = {t["name"]: t for t in plugin["tools"]} assert tools["mes.fetch_progress"]["power"] == "P0" assert tools["mes.fetch_progress"]["input_schema"]["properties"]["woId"] == {"type": "string"} assert tools["mes.release_order"]["output_schema"] == {"type": "object"} # 缺省权限：P0/P1 放行、P2/P3 拒绝（安全优先） assert plugin["permissions"]["mes.fetch_progress"]["allow"] is True assert plugin["permissions"]["mes.release_order"]["allow"] is False # 落盘可重读 again = McpBus(path=str(tmp_path / "plugins.json")) assert any(p["plugin_id"] == "mcp.mes" for p in again.list()) def test_permission_denied_rejected_and_audited(tmp_path): """越权测试：工具未授权（P2 缺省 deny）→ 拒绝 + DENIED 审计 + 统计。""" bus = McpBus(path=str(tmp_path / "plugins.json")) bus.register(_sample_manifest()) bus.register_handler("mcp.mes", "mes.release_order", lambda args: {"ok": True}) with pytest.raises(PermissionError): bus.call_tool("mcp.mes", "mes.release_order", {"woId": "W-1"}, actor="planner") denied = [e for e in bus.audit("mcp.mes") if e["action"] == "mcp.tool.denied"] assert denied, "越权调用必须写 DENIED 审计" assert denied[0]["result"] == "DENIED" assert denied[0]["target"]["tool"] == "mes.release_order" assert denied[0]["power"] == "P2" stats = bus.stats("mcp.mes")[0] assert stats["tools"]["mes.release_order"]["denied"] == 1 assert stats["tools"]["mes.release_order"]["allowed"] == 0 def test_permission_switch_allows_tool(tmp_path): """权限开关放行后调用成功，调用/变更均有审计。""" bus = McpBus(path=str(tmp_path / "plugins.json")) bus.register(_sample_manifest()) bus.register_handler("mcp.mes", "mes.release_order", lambda args: {"ok": True, **args}) bus.set_permission("mcp.mes", "mes.release_order", True, actor="planner", reason="测试放行") result = bus.call_tool("mcp.mes", "mes.release_order", {"woId": "W-2"}, actor="planner") assert result == {"ok": True, "woId": "W-2"} events = bus.audit("mcp.mes") assert any(e["action"] == "mcp.permission.update" and e["rationale"]["allow"] for e in events) assert any(e["action"] == "mcp.tool.run" and e["result"] == "SUCCESS" for e in events) def test_enable_stop_blocks_tool_and_audits(tmp_path): """停用后 call_tool 一律拒绝并审计；重新启用恢复。""" bus = McpBus(path=str(tmp_path / "plugins.json")) bus.register(_sample_manifest()) bus.register_handler("mcp.mes", "mes.fetch_progress", lambda args: {"progress": 0.5}) assert bus.call_tool("mcp.mes", "mes.fetch_progress", {}, actor="planner") == {"progress": 0.5} bus.set_enabled("mcp.mes", False, actor="planner") with pytest.raises(PermissionError): bus.call_tool("mcp.mes", "mes.fetch_progress", {}, actor="planner") events = bus.audit("mcp.mes") assert any(e["action"] == "mcp.plugin.stop" for e in events) assert any( e["action"] == "mcp.tool.denied" and e["rationale"]["reason"] == "plugin-disabled" for e in events ) bus.set_enabled("mcp.mes", True, actor="planner") assert bus.call_tool("mcp.mes", "mes.fetch_progress", {}, actor="planner") == {"progress": 0.5} def test_version_compatibility_gate(tmp_path): """min_bus_version 高于总线版本 → 拒绝登记；兼容插件正常登记并出报告。""" bus = McpBus(path=str(tmp_path / "plugins.json")) with pytest.raises(ValueError, match="版本不兼容"): bus.register(_sample_manifest(min_bus_version="9.9.0")) ok_plugin = bus.register(_sample_manifest(plugin_id="mcp.ems", min_bus_version="0.5")) assert ok_plugin["plugin_id"] == "mcp.ems" report = {r["plugin_id"]: r for r in bus.version_report()} assert report["mcp.stub"]["compatible"] is True assert report["mcp.ems"]["compatible"] is True assert report["mcp.ems"]["bus_version"] == BUS_VERSION assert all("compatible" in r for r in bus.version_report()) def test_health_probe_records_history(tmp_path): """健康探测：local://stub 恒健康，且写探测历史。""" bus = McpBus(path=str(tmp_path / "plugins.json")) bus.health("mcp.stub") bus.health("mcp.stub") hist = bus.history("mcp.stub") assert len(hist) == 2 assert all(h["ok"] for h in hist) def test_local_tool_without_handler_fails(tmp_path): """本地工具未注册处理器 → 显式报错，不静默。""" bus = McpBus(path=str(tmp_path / "plugins.json")) bus.register(_sample_manifest()) with pytest.raises(ToolExecutionError): bus.call_tool("mcp.mes", "mes.fetch_progress", {}, actor="planner") def test_seed_from_skills_bridges_without_touching_skills(tmp_path): """既有 SkillRegistry → MCP 插件桥接（skill.，向后兼容只增不删）。""" reg = SkillRegistry(path=str(tmp_path / "skills.json")) reg.upsert({ "skill_id": "algo.opt", "name": "外部优化器", "endpoint": "http://127.0.0.1:9000", "version": "2.1", "max_power": "P2", }) bus = McpBus(path=str(tmp_path / "plugins.json")) count = bus.seed_from_skills(reg, actor="system") assert count >= 1 plugin = bus.get("skill.algo.opt") assert plugin is not None tools = {t["name"]: t for t in plugin["tools"]} assert tools["flex.schedule"]["transport"] == "http" assert tools["flex.schedule"]["power"] == "P2" assert plugin["system"] == "ALGO" # SkillRegistry 不受影响 assert reg.get("algo.opt")["version"] == "2.1" # ---------------- 真实 MES 适配器（矩阵 75）：默认注册 + fail-closed ---------------- def test_default_bus_registers_mes_http_adapter_fail_closed(tmp_path, monkeypatch): """默认总线注册 mes.http（manifest + power 门禁）；未配置 base_url 时 fail-closed。""" from server.integrations.mes_http import MesHttpError, reset_http_mes_client monkeypatch.delenv("MES_HTTP_BASE_URL", raising=False) reset_http_mes_client(base_url="", token="") bus = McpBus(path=str(tmp_path / "plugins.json")) plugin = bus.get("mes.http") assert plugin is not None assert plugin["system"] == "MES" assert plugin["enabled"] is True tools = {t["name"]: t for t in plugin["tools"]} assert tools["mes.http_dispatch"]["power"] == "P3" assert tools["mes.http_dispatch"]["idempotent"] is True assert tools["mes.http_status"]["power"] == "P0" assert tools["mes.http_status"]["idempotent"] is True # P3 写工具默认 deny（power 门禁） with pytest.raises(PermissionError): bus.call_tool("mes.http", "mes.http_dispatch", {"idemKey": "k"}, actor="planner") bus.set_permission("mes.http", "mes.http_dispatch", True, actor="planner", reason="golden") # fail-closed：未配置 base_url → 明确报错（MesHttpError MES_HTTP_NOT_CONFIGURED） with pytest.raises(MesHttpError) as ei: bus.call_tool("mes.http", "mes.http_dispatch", {"idemKey": "k"}, actor="planner") assert ei.value.code == "MES_HTTP_NOT_CONFIGURED"