196 lines
9.0 KiB
Python
196 lines
9.0 KiB
Python
# ============================================================
|
||
# 方向 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.<id>，向后兼容只增不删）。"""
|
||
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"
|