aps-agent/tests/golden/test_mcp_bus.py

# ============================================================
# 方向 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"