vent_deep_agent/tests/test_mcp_interfaces.py

# -*- coding: utf-8 -*-
"""
通风模型算法 MCP 接口全面测试脚本
测试 server: vent-model-tools @ http://39.97.59.228:8071/mcp
"""

import asyncio
import json
import time
import os
from datetime import datetime
from fastmcp import Client

MCP_URL = "http://39.97.59.228:8071/mcp"
MODEL_ID = "2012326636757958658"

# ── 测试用例定义 ──
# 按文档中的 15 个接口定义
TEST_CASES = [
    # ═══ 故障诊断类 ═══
    {
        "id": "TC-01",
        "category": "故障诊断",
        "tool": "check_model_connect_status",
        "params": {"model_id": MODEL_ID},
        "desc": "模型网络连通检查",
    },
    {
        "id": "TC-02",
        "category": "故障诊断",
        "tool": "check_model_one_dir_cycle",
        "params": {"model_id": MODEL_ID},
        "desc": "模型循环风路检查",
    },
    {
        "id": "TC-03",
        "category": "故障诊断",
        "tool": "check_model_one_dir_node",
        "params": {"model_id": MODEL_ID},
        "desc": "模型单向节点检查",
    },
    {
        "id": "TC-04",
        "category": "故障诊断",
        "tool": "check_model_diagonal_structure",
        "params": {"model_id": MODEL_ID},
        "desc": "模型角联结构诊断（计算重，耗时较长）",
    },
    {
        "id": "TC-05",
        "category": "故障诊断",
        "tool": "get_model_fault_diagnosis",
        "params": {"model_id": MODEL_ID, "include_diagonal": False},
        "desc": "模型故障诊断（聚合，不含角联）",
    },
    # ═══ 避灾路线类 ═══
    {
        "id": "TC-06",
        "category": "避灾路线",
        "tool": "get_escape_path",
        "params": {"model_id": MODEL_ID, "fire_tun_id": "4", "person_tun_id": "6"},
        "desc": "避灾路线模拟",
        "note": "使用硬编码隧道ID 4/6，实际应取自 get_out_shafts",
    },
    {
        "id": "TC-07",
        "category": "避灾路线",
        "tool": "get_escape_path_each_exit",
        "params": {"model_id": MODEL_ID, "fire_tun_id": "4", "person_tun_id": "6", "co_per": 2000.0, "during_time": 30.0},
        "desc": "避灾路线模拟（各出口，含CO参数）",
    },
    # ═══ 关键阻力类 ═══
    {
        "id": "TC-08",
        "category": "关键阻力",
        "tool": "get_out_shafts",
        "params": {"model_id": MODEL_ID},
        "desc": "获取回风井巷道ID列表",
    },
    {
        "id": "TC-09",
        "category": "关键阻力",
        "tool": "get_in_shafts",
        "params": {"model_id": MODEL_ID},
        "desc": "获取进风井巷道ID列表",
    },
    # get_max_resistance_path - need node_id, will be filled dynamically
    {
        "id": "TC-10",
        "category": "关键阻力",
        "tool": "get_max_resistance_path",
        "params": {"model_id": MODEL_ID, "node_id": None},  # filled dynamically
        "desc": "最大阻力路线",
        "dynamic": True,
    },
    {
        "id": "TC-11",
        "category": "关键阻力",
        "tool": "get_three_area_distribution",
        "params": {"model_id": MODEL_ID},
        "desc": "三区阻力分布",
    },
    {
        "id": "TC-12",
        "category": "关键阻力",
        "tool": "get_key_path_decision",
        "params": {"model_id": MODEL_ID},
        "desc": "关键路径决策",
    },
    # ═══ 压能/解算类 ═══
    # get_path_press_power - need node IDs, will be filled dynamically
    {
        "id": "TC-13",
        "category": "压能/解算",
        "tool": "get_path_press_power",
        "params": {"model_id": MODEL_ID, "id_from": None, "id_to": None},  # filled dynamically
        "desc": "节点压能图",
        "dynamic": True,
    },
    {
        "id": "TC-14",
        "category": "压能/解算",
        "tool": "net_cal",
        "params": {"model_id": MODEL_ID},
        "desc": "网络解算（耗时较长）",
    },
    {
        "id": "TC-15",
        "category": "压能/解算",
        "tool": "net_cal_for_plan",
        "params": {"model_id": MODEL_ID, "plan": "{}"},
        "desc": "方案模拟解算",
        "note": "plan 参数不能为空字符串，使用空JSON对象作为最小有效方案",
    },
]


def safe_parse_json(text: str) -> dict:
    """安全解析 JSON，支持多层嵌套"""
    try:
        return json.loads(text) if isinstance(text, str) else text
    except (json.JSONDecodeError, TypeError):
        return {"raw": str(text)[:2000]}


def extract_deep_result(data: dict) -> dict:
    """尝试提取深层 result"""
    # 尝试多层解包
    for _ in range(5):
        if isinstance(data, dict):
            if "result" in data and isinstance(data["result"], str):
                try:
                    data = json.loads(data["result"])
                    continue
                except (json.JSONDecodeError, TypeError):
                    pass
            if "result" in data and isinstance(data["result"], dict):
                data = data["result"]
                continue
        break
    return data


def count_result_size(data) -> str:
    """估算结果大小"""
    s = json.dumps(data, ensure_ascii=False, default=str)
    size = len(s)
    if size < 1024:
        return f"{size} B"
    elif size < 1024 * 1024:
        return f"{size / 1024:.1f} KB"
    else:
        return f"{size / (1024 * 1024):.1f} MB"


def summarize_structure(data, depth=0) -> str:
    """概括数据结构"""
    if depth > 5:
        return "..."
    if isinstance(data, dict):
        keys = list(data.keys())
        if len(keys) <= 8:
            parts = []
            for k in keys:
                v = data[k]
                if isinstance(v, (dict, list)):
                    parts.append(f"{k}: {summarize_structure(v, depth + 1)}")
                elif isinstance(v, str) and len(v) > 100:
                    parts.append(f"{k}: str({len(v)})")
                else:
                    parts.append(f"{k}: {type(v).__name__}")
            return "{" + ", ".join(parts[:10]) + ("..." if len(parts) > 10 else "") + "}"
        else:
            return f"{{...{len(keys)} keys...}}"
    elif isinstance(data, list):
        if len(data) == 0:
            return "[]"
        return f"[{len(data)} items, first: {summarize_structure(data[0], depth + 1)}]"
    elif isinstance(data, str):
        return f'str({len(data)})'
    else:
        return type(data).__name__


async def call_single_tool(client: Client, case: dict) -> dict:
    """调用单个工具并返回统一格式结果"""
    result = {
        "id": case["id"],
        "tool": case["tool"],
        "category": case["category"],
        "desc": case["desc"],
        "params": {k: v for k, v in case["params"].items() if v is not None},
        "timestamp": datetime.now().isoformat(),
    }
    
    start = time.perf_counter()
    try:
        raw = await client.call_tool(case["tool"], result["params"])
        elapsed = time.perf_counter() - start
        
        # 提取文本内容
        if raw.content and len(raw.content) > 0:
            text = raw.content[0].text
            data = safe_parse_json(text)
        else:
            text = ""
            data = {}
        
        result.update({
            "success": True,
            "elapsed_ms": round(elapsed * 1000, 2),
            "has_content": len(raw.content) > 0,
            "content_len": len(text),
            "size": count_result_size(data),
            "structure": summarize_structure(data),
            "data_preview": json.dumps(data, ensure_ascii=False, default=str)[:500],
            "deep_success": None,  # will check below
        })
        
        # 检查深层 success 字段
        # 注意: bSucced=0 对于故障检测类接口表示"未发现故障"(正常), bSucced=1 表示"检测到故障"
        # 对于计算类接口, bSucced=1 表示计算成功, bSucced=0 表示计算失败
        deep = extract_deep_result(data)
        if isinstance(deep, dict):
            # 优先检查 outer success + code 200
            outer_ok = data.get("success") and data.get("code") == 200
            # 检查 inner result 是否有 bSucced 字段 (1=成功/发现, 0=失败/未发现)
            has_bSucced = "bSucced" in deep
            bSucced_val = deep.get("bSucced")
            # 如果有 error 字段且非空，标记业务异常
            has_error = bool(deep.get("error") or data.get("error"))
            if has_error:
                result["deep_success"] = False
            elif outer_ok:
                result["deep_success"] = True
            elif has_bSucced:
                # bSucced 存在时，都视为业务层面正常（0或1都有意义）
                result["deep_success"] = True
            else:
                result["deep_success"] = data.get("code") == 200
        
    except Exception as e:
        elapsed = time.perf_counter() - start
        result.update({
            "success": False,
            "elapsed_ms": round(elapsed * 1000, 2),
            "error": str(e),
            "error_type": type(e).__name__,
        })
    
    return result


async def discover_node_and_tunnel_ids(client: Client) -> dict:
    """发现可用的 node_id 和 tunnel_id，用于后续动态测试"""
    discovered = {"node_ids": [], "out_shaft_tuns": [], "in_shaft_tuns": []}
    
    # 获取回风井
    try:
        r = await client.call_tool("get_out_shafts", {"model_id": MODEL_ID})
        text = r.content[0].text
        data = safe_parse_json(text)
        deep = extract_deep_result(data)
        # 尝试找到隧道ID列表
        if isinstance(deep, dict):
            for key in ["outShafts", "tunIds", "ids", "data"]:
                if key in deep and isinstance(deep[key], list):
                    discovered["out_shaft_tuns"] = deep[key][:5]
                    break
        discovered["out_shafts_raw"] = summarize_structure(data)
    except Exception as e:
        discovered["out_shafts_error"] = str(e)
    
    # 获取进风井
    try:
        r = await client.call_tool("get_in_shafts", {"model_id": MODEL_ID})
        text = r.content[0].text
        data = safe_parse_json(text)
        deep = extract_deep_result(data)
        if isinstance(deep, dict):
            for key in ["inShafts", "tunIds", "ids", "data"]:
                if key in deep and isinstance(deep[key], list):
                    discovered["in_shaft_tuns"] = deep[key][:5]
                    break
        discovered["in_shafts_raw"] = summarize_structure(data)
    except Exception as e:
        discovered["in_shafts_error"] = str(e)
    
    return discovered


async def main():
    print("=" * 80)
    print("  通风模型算法 MCP 接口测试")
    print(f"  Server: {MCP_URL}")
    print(f"  Model:  {MODEL_ID}")
    print(f"  时间:    {datetime.now().isoformat()}")
    print("=" * 80)
    print()
    
    client = Client(MCP_URL)
    results = []
    
    async with client:
        # ── 第一阶段：发现动态参数 ──
        print("🔍 第一阶段：发现动态参数（回风井/进风井/节点ID）...")
        discovered = await discover_node_and_tunnel_ids(client)
        print(f"   回风井: {json.dumps(discovered.get('out_shaft_tuns', []), ensure_ascii=False)}")
        print(f"   进风井: {json.dumps(discovered.get('in_shaft_tuns', []), ensure_ascii=False)}")
        
        # 尝试用已知隧道ID获取更多节点信息（需要调用 get_tun_list_by_modelid 或 net_cal）
        node_id_from = None
        node_id_to = None
        node_id_max_res = None
        
        # 先尝试从回风井结果中提取节点ID
        if discovered.get("out_shaft_tuns"):
            # 假设可以从 net_cal 结果中获取节点信息
            pass
        
        # 使用固定节点ID（文档中的示例值）
        node_id_max_res = "3958"
        node_id_from = "2"
        node_id_to = "3958"
        
        # ── 更新动态测试用例 ──
        for case in TEST_CASES:
            if case["tool"] == "get_max_resistance_path" and case.get("dynamic"):
                case["params"]["node_id"] = node_id_max_res
                print(f"\n   📍 TC-10 使用 node_id={node_id_max_res}")
            elif case["tool"] == "get_path_press_power" and case.get("dynamic"):
                case["params"]["id_from"] = node_id_from
                case["params"]["id_to"] = node_id_to
                print(f"   📍 TC-13 使用 id_from={node_id_from}, id_to={node_id_to}")
        
        # ── 第二阶段：逐个测试 ──
        print(f"\n{'=' * 80}")
        print(f"  第二阶段：执行 {len(TEST_CASES)} 个测试用例")
        print(f"{'=' * 80}\n")
        
        for i, case in enumerate(TEST_CASES):
            print(f"[{i+1:02d}/{len(TEST_CASES)}] {case['tool']} ... ", end="", flush=True)
            result = await call_single_tool(client, case)
            results.append(result)
            
            status = "✅" if result["success"] else "❌"
            elapsed = result.get("elapsed_ms", 0)
            size = result.get("size", "N/A")
            print(f"{status} {elapsed:.0f}ms {size}")
            
            if not result["success"]:
                print(f"     ⚠️  错误: {result.get('error', 'N/A')[:120]}")
            elif result.get("data_preview"):
                preview = result["data_preview"][:120].replace("\n", " ")
                print(f"     📄 {preview}")
        
    # ── 第三阶段：生成报告 ──
    print(f"\n{'=' * 80}")
    print(f"  第三阶段：生成测试报告")
    print(f"{'=' * 80}\n")
    
    # 统计
    total = len(results)
    passed = sum(1 for r in results if r["success"])
    failed = sum(1 for r in results if not r["success"])
    # 业务层面统计
    biz_ok = sum(1 for r in results if r["success"] and r.get("deep_success"))
    biz_timeout = sum(1 for r in results if r["success"] and r.get("data_preview", "").find('"code": 408') > 0)
    biz_500 = sum(1 for r in results if r["success"] and r.get("data_preview", "").find('"code": 500') > 0)
    total_time = sum(r.get("elapsed_ms", 0) for r in results)
    
    # 生成 JSON 报告
    report_json_path = os.path.join(os.path.dirname(__file__), "mcp_test_report.json")
    report = {
        "title": "通风模型算法 MCP 接口测试报告",
        "server": MCP_URL,
        "model_id": MODEL_ID,
        "timestamp": datetime.now().isoformat(),
        "summary": {
            "total": total,
            "passed": passed,
            "failed": failed,
            "pass_rate": f"{passed / total * 100:.1f}%" if total > 0 else "N/A",
            "biz_ok": biz_ok,
            "biz_timeout": biz_timeout,
            "biz_500": biz_500,
            "total_elapsed_ms": round(total_time, 2),
            "avg_elapsed_ms": round(total_time / total, 2) if total > 0 else 0,
        },
        "results": results,
    }
    
    with open(report_json_path, "w", encoding="utf-8") as f:
        json.dump(report, f, ensure_ascii=False, indent=2, default=str)
    print(f"   JSON 报告已保存: {report_json_path}")
    
    # 生成 Markdown 报告
    md_path = os.path.join(os.path.dirname(__file__), "mcp_test_report.md")
    md_lines = []
    md_lines.append("# 通风模型算法 MCP 接口测试报告\n")
    md_lines.append(f"**测试时间**: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}  \n")
    md_lines.append(f"**MCP 服务地址**: `{MCP_URL}`  \n")
    md_lines.append(f"**测试模型 ID**: `{MODEL_ID}`  \n")
    md_lines.append(f"**传输协议**: `streamable-http`  \n")
    md_lines.append(f"**客户端库**: `fastmcp`  \n")
    md_lines.append("")
    
    # 汇总
    md_lines.append("## 📊 测试汇总\n")
    md_lines.append("### 传输层结果\n")
    md_lines.append("| 指标 | 数值 |")
    md_lines.append("|:---|---:|")
    md_lines.append(f"| 接口总数 | {total} |")
    md_lines.append(f"| ✅ 传输层通过 | {passed} |")
    md_lines.append(f"| ❌ 传输层失败 | {failed} |")
    md_lines.append(f"| 传输层通过率 | {passed / total * 100:.1f}% |")
    md_lines.append(f"| 总耗时 | {total_time:.0f} ms |")
    md_lines.append(f"| 平均响应 | {total_time / total:.0f} ms |")
    md_lines.append("")
    md_lines.append("### 业务层结果\n")
    md_lines.append("| 指标 | 数值 |")
    md_lines.append("|:---|---:|")
    md_lines.append(f"| ✅ 业务正常 | {biz_ok} |")
    md_lines.append(f"| ⏱️ 服务端超时 (408) | {biz_timeout} |")
    md_lines.append(f"| 🔴 服务端错误 (500) | {biz_500} |")
    md_lines.append(f"| ❌ 客户端校验失败 | {failed} |")
    md_lines.append(f"| 综合可用率 | {biz_ok / total * 100:.1f}% |")
    md_lines.append("")
    
    # 分类统计
    md_lines.append("### 分类统计\n")
    md_lines.append("| 分类 | 总数 | 通过 | 失败 | 平均耗时 |")
    md_lines.append("|:---|---:|---:|---:|---:|")
    categories = {}
    for r in results:
        cat = r["category"]
        if cat not in categories:
            categories[cat] = {"total": 0, "passed": 0, "failed": 0, "times": []}
        categories[cat]["total"] += 1
        if r["success"]:
            categories[cat]["passed"] += 1
        else:
            categories[cat]["failed"] += 1
        categories[cat]["times"].append(r.get("elapsed_ms", 0))
    
    for cat, stats in categories.items():
        avg_t = sum(stats["times"]) / len(stats["times"]) if stats["times"] else 0
        md_lines.append(f"| {cat} | {stats['total']} | {stats['passed']} | {stats['failed']} | {avg_t:.0f} ms |")
    md_lines.append("")
    
    # 详细结果
    md_lines.append("## 📋 详细测试结果\n")
    
    for r in results:
        # 确定状态图标
        if not r["success"]:
            status_icon = "❌"
            biz_label = "客户端参数校验失败"
        elif r.get("deep_success") is False:
            preview = r.get("data_preview", "")
            if '"code": 408' in preview or "408" in preview:
                status_icon = "⏱️"
                biz_label = "服务端超时 (408)"
            elif '"code": 500' in preview or "500" in preview:
                status_icon = "🔴"
                biz_label = "服务端内部错误 (500)"
            elif '"code": 503' in preview or "503" in preview:
                status_icon = "🔴"
                biz_label = "服务不可用 (503)"
            else:
                status_icon = "⚠️"
                biz_label = "业务异常"
        else:
            status_icon = "✅"
            biz_label = "正常"
        
        md_lines.append(f"### {status_icon} {r['id']}: {r['desc']} (`{r['tool']}`)\n")
        md_lines.append(f"- **分类**: {r['category']}")
        md_lines.append(f"- **传输状态**: {'通过' if r['success'] else '失败'} | **业务状态**: {biz_label}")
        md_lines.append(f"- **耗时**: {r.get('elapsed_ms', 'N/A')} ms")
        
        if r["success"]:
            md_lines.append(f"- **结果大小**: {r.get('size', 'N/A')}")
            md_lines.append(f"- **数据结构**: `{r.get('structure', 'N/A')}`")
        else:
            md_lines.append(f"- **错误类型**: `{r.get('error_type', 'N/A')}`")
            md_lines.append(f"- **错误信息**: {r.get('error', 'N/A')[:300]}")
        
        # 参数
        params_str = ", ".join(f"`{k}={v}`" for k, v in r["params"].items())
        md_lines.append(f"- **调用参数**: {params_str}")
        
        # 数据预览
        if r.get("data_preview"):
            preview = r["data_preview"]
            if len(preview) > 500:
                preview = preview[:500] + "..."
            md_lines.append(f"\n<details>\n<summary>数据预览</summary>\n\n```json\n{preview}\n```\n</details>")
        
        md_lines.append("")
    
    # 建议
    md_lines.append("## 💡 建议与注意事项\n")
    md_lines.append(f"### ⚠️ 需要修复的问题（{biz_timeout + biz_500} 个）\n")
    md_lines.append("| 接口 | 问题 | 建议 |")
    md_lines.append("|:---|:---|:---|")
    
    for r in results:
        preview = r.get("data_preview", "")
        if '"code": 408' in preview or "408" in preview:
            md_lines.append(f"| `{r['tool']}` | 服务端超时 (408)，耗时 {r.get('elapsed_ms', 0):.0f}ms | 优化后端算法或增加超时时间；前端设置 180s+ 超时并显示进度 |")
        elif '"code": 500' in preview or "500" in preview:
            md_lines.append(f"| `{r['tool']}` | 服务端内部错误 (500) - Connection prematurely closed | 检查后端 netty 连接池配置，可能需要增大响应超时 |")
        elif '"code": 503' in preview or "503" in preview:
            md_lines.append(f"| `{r['tool']}` | 服务不可用 (503) - Unable to find instance | 检查微服务 `ventanaly-model` 实例是否在线/注册中心状态 |")
    
    md_lines.append("")
    md_lines.append("### 📝 通用建议\n")
    md_lines.append("1. **响应时间差异大**：快速接口（进/回风井查询、循环风路检查）约 0.8-1.5s，重计算接口（压能图、避灾路线）约 12-35s，超时接口约 180s。建议按接口类型设置差异化超时。\n")
    md_lines.append("2. **数据量大**：`get_three_area_distribution` (366KB) 和 `get_path_press_power` (3.8MB) 返回数据量大，移动端需考虑分页或压缩。\n")
    md_lines.append("3. **参数校验**：`net_cal_for_plan` 的 `plan` 参数不能为空字符串，建议后端明确 `plan` 格式规范或提供示例。\n")
    md_lines.append("4. **返回值统一性**：外层统一为 `{success, code, result}`，但 `get_model_fault_diagnosis` 直接在外层返回 `modelID`，略有不一致。\n")
    md_lines.append("5. **节点 vs 隧道 ID**：`get_max_resistance_path` 和 `get_path_press_power` 使用 `node_id`（节点ID），其他接口使用 `tun_id`（隧道ID），文档已说明，但调用时容易混淆。\n")
    md_lines.append("6. **CO 参数**：`get_escape_path_each_exit` 的 `co_per` 默认 2000.0，`during_time` 默认 0.0，合理默认值便于调用。\n")
    
    with open(md_path, "w", encoding="utf-8") as f:
        f.write("\n".join(md_lines))
    print(f"   Markdown 报告已保存: {md_path}")
    
    # ── 终端输出汇总 ──
    print(f"\n{'=' * 80}")
    print(f"  📊 测试汇总")
    print(f"  总计: {total} | ✅ 通过: {passed} | ❌ 失败: {failed} | 通过率: {passed / total * 100:.1f}%")
    print(f"  总耗时: {total_time:.0f} ms | 平均: {total_time / total:.0f} ms")
    print(f"{'=' * 80}")


if __name__ == "__main__":
    asyncio.run(main())