"""
集成测试 - DoWhy MCP v2.0
验证服务器启动、工具调用、错误处理等完整流程
"""
import asyncio
import json
import sys
import os
from pathlib import Path
# 添加项目根目录到路径
project_root = Path(__file__).parent.parent
sys.path.insert(0, str(project_root / "src"))
def test_server_info():
"""测试服务器信息获取"""
print("🔍 测试服务器信息获取...")
try:
from dowhy_mcp_v2.mcp.server import DoWhyMCPServer
server = DoWhyMCPServer()
info = server.get_server_info()
# 验证基本信息
assert info["name"] == "dowhy-mcp-v2", "服务器名称不正确"
assert info["version"] == "2.0.0", "版本号不正确"
assert "capabilities" in info, "缺少capabilities信息"
assert "tools" in info["capabilities"], "缺少工具数量信息"
assert info["capabilities"]["tools"] == 42, f"工具数量不正确，期望42，实际{info['capabilities']['tools']}"
print(f" ✅ 服务器信息正确: {info['capabilities']['tools']}个工具")
return True
except Exception as e:
print(f" ❌ 服务器信息获取失败: {e}")
return False
def test_cli_functionality():
"""测试CLI功能"""
print("🖥️ 测试CLI功能...")
try:
# 测试CLI info命令
import subprocess
result = subprocess.run(
[sys.executable, "-m", "src.dowhy_mcp_v2.cli", "info"],
cwd=project_root,
capture_output=True,
text=True,
timeout=30
)
if result.returncode == 0:
output = result.stdout
# 检查输出是否包含预期内容
if "DoWhy MCP v2.0 服务器信息" in output and "42" in output:
print(" ✅ CLI info命令正常工作")
return True
else:
print(f" ⚠️ CLI输出不完整: {output[:200]}...")
return False
else:
print(f" ❌ CLI命令失败: {result.stderr}")
return False
except Exception as e:
print(f" ❌ CLI测试失败: {e}")
return False
def test_tool_registration():
"""测试工具注册"""
print("🔧 测试工具注册...")
try:
from dowhy_mcp_v2.mcp.server import DoWhyMCPServer
server = DoWhyMCPServer()
# 检查工具注册数量
tools_count = len(server.server.tools)
expected_count = 42
if tools_count == expected_count:
print(f" ✅ 工具注册正确: {tools_count}个工具")
return True
else:
print(f" ⚠️ 工具注册数量不匹配: 期望{expected_count}，实际{tools_count}")
return False
except Exception as e:
print(f" ❌ 工具注册测试失败: {e}")
return False
def test_handler_coverage():
"""测试处理函数覆盖率"""
print("📊 测试处理函数覆盖率...")
try:
# 检查各个模块的处理函数
coverage_results = {}
# 建模工具
try:
from dowhy_mcp_v2.mcp.tools.modeling import MODELING_HANDLERS
coverage_results["建模工具"] = len(MODELING_HANDLERS)
except:
coverage_results["建模工具"] = 0
# 估计工具
try:
from dowhy_mcp_v2.mcp.tools.estimation import ESTIMATION_HANDLERS
coverage_results["估计工具"] = len(ESTIMATION_HANDLERS)
except:
coverage_results["估计工具"] = 0
# 敏感性分析工具
try:
from dowhy_mcp_v2.tools.sensitivity import (
handle_e_value_analyzer,
handle_rosenbaum_bounds_analyzer,
handle_refutation_test_suite,
handle_stability_analyzer,
handle_tipping_point_analyzer,
handle_unobserved_confounder_analyzer
)
coverage_results["敏感性分析工具"] = 6
except:
coverage_results["敏感性分析工具"] = 0
# 因果发现工具
try:
from dowhy_mcp_v2.tools.discovery import (
handle_pc_algorithm,
handle_ges_algorithm,
handle_fcm_discovery
)
coverage_results["因果发现工具"] = 3
except:
coverage_results["因果发现工具"] = 0
# 归因分析工具
try:
from dowhy_mcp_v2.tools.attribution import (
handle_shapley_value_attribution,
handle_direct_causal_influence,
handle_total_causal_influence,
handle_path_specific_effects,
handle_mechanism_attribution,
handle_causal_contribution_analysis
)
coverage_results["归因分析工具"] = 6
except:
coverage_results["归因分析工具"] = 0
# 根本原因分析工具
try:
from dowhy_mcp_v2.tools.root_cause import (
handle_anomaly_attribution_analyzer,
handle_distribution_change_attribution,
handle_causal_chain_tracer,
handle_root_cause_identifier,
handle_event_attribution_analyzer
)
coverage_results["根本原因分析工具"] = 5
except:
coverage_results["根本原因分析工具"] = 0
# 反事实分析工具
try:
from dowhy_mcp_v2.tools.counterfactual import (
handle_individual_counterfactual,
handle_population_counterfactual,
handle_intervention_simulator,
handle_what_if_analyzer,
handle_scenario_comparator,
handle_counterfactual_fairness
)
coverage_results["反事实分析工具"] = 6
except:
coverage_results["反事实分析工具"] = 0
# 计算总覆盖率
total_handlers = sum(coverage_results.values())
expected_total = 42 # 期望的总工具数
print(f" 📈 处理函数覆盖情况:")
for category, count in coverage_results.items():
print(f" - {category}: {count}个")
coverage_rate = (total_handlers / expected_total) * 100
print(f" 🎯 总覆盖率: {total_handlers}/{expected_total} ({coverage_rate:.1f}%)")
if coverage_rate >= 80:
print(" ✅ 处理函数覆盖率良好")
return True
else:
print(" ⚠️ 处理函数覆盖率需要提升")
return False
except Exception as e:
print(f" ❌ 处理函数覆盖率测试失败: {e}")
return False
def test_error_handling():
"""测试错误处理"""
print("🛡️ 测试错误处理...")
try:
# 测试一个简单的工具调用错误处理
from dowhy_mcp_v2.mcp.tools.modeling import MODELING_HANDLERS
if "causal_graph_builder" in MODELING_HANDLERS:
handler = MODELING_HANDLERS["causal_graph_builder"]
# 创建一个错误的请求（缺少必要参数）
class MockRequest:
def __init__(self):
self.name = "causal_graph_builder"
self.arguments = {} # 空参数，应该导致错误
# 测试错误处理
async def test_error():
try:
result = await handler(MockRequest())
# 检查是否返回了错误信息
if len(result) > 0:
result_text = result[0].text
if "error" in result_text.lower() or "失败" in result_text:
return True
return False
except:
return True # 异常也算是正确的错误处理
error_handled = asyncio.run(test_error())
if error_handled:
print(" ✅ 错误处理机制正常")
return True
else:
print(" ⚠️ 错误处理可能有问题")
return False
else:
print(" ⚠️ 无法找到测试用的处理函数")
return False
except Exception as e:
print(f" ❌ 错误处理测试失败: {e}")
return False
def run_integration_tests():
"""运行集成测试"""
print("🚀 DoWhy MCP v2.0 集成测试")
print("=" * 60)
# 切换到项目根目录
os.chdir(project_root)
tests = [
("服务器信息", test_server_info),
("CLI功能", test_cli_functionality),
("工具注册", test_tool_registration),
("处理函数覆盖", test_handler_coverage),
("错误处理", test_error_handling)
]
results = []
for test_name, test_func in tests:
print(f"\n{test_name}测试:")
try:
result = test_func()
results.append(result)
except Exception as e:
print(f" ❌ {test_name}测试执行失败: {e}")
results.append(False)
# 汇总结果
passed = sum(results)
total = len(results)
print("\n" + "=" * 60)
print("📊 集成测试结果汇总")
print("=" * 60)
for i, (test_name, _) in enumerate(tests):
status = "✅" if results[i] else "❌"
print(f"{status} {test_name}测试")
success_rate = (passed / total) * 100
print(f"\n🎯 总体结果: {passed}/{total} ({success_rate:.1f}%)")
if success_rate >= 80:
print("🎉 DoWhy MCP v2.0 集成测试通过！系统基本可用！")
return True
elif success_rate >= 60:
print("⚠️ DoWhy MCP v2.0 部分功能可用，建议进一步优化")
return False
else:
print("❌ DoWhy MCP v2.0 需要重大修复")
return False
if __name__ == "__main__":
success = run_integration_tests()
if success:
print("\n✅ 集成测试完成 - 系统可用")
exit(0)
else:
print("\n⚠️ 集成测试完成 - 需要优化")
exit(1)
