#!/usr/bin/env python3
"""测试所有功能."""
import sys
from src.config_manager import ConfigManager
from src.graphiti_client import GraphitiClient
def test_config():
"""测试配置功能."""
print("🧪 测试配置功能...")
cm = ConfigManager()
# 测试配置 Neo4j
result = cm.configure_neo4j(
uri="bolt://localhost:7687",
username="neo4j",
password="password"
)
assert result['success'], "Neo4j 配置失败"
print(" ✅ Neo4j 配置成功")
# 测试检查配置
status = cm.get_config_status()
assert status['neo4j_configured'], "Neo4j 未配置"
print(" ✅ 配置检查成功")
return cm
def test_client_connection(cm):
"""测试客户端连接."""
print("\n🧪 测试客户端连接...")
client = GraphitiClient(cm)
if client.connect():
print(" ✅ 连接成功")
return client
else:
print(" ❌ 连接失败")
return None
def test_add_episode(client):
"""测试添加 Episode."""
print("\n🧪 测试添加 Episode...")
result = client.add_episode(
content="测试 Episode：我喜欢 Python",
metadata={"type": "test"}
)
assert result['success'], "添加 Episode 失败"
print(" ✅ Episode 添加成功")
return result.get('episode_id')
def test_search(client):
"""测试搜索功能."""
print("\n🧪 测试搜索功能...")
# 搜索实体
result = client.search_entities(query="Python", limit=5)
assert result['success'], "搜索实体失败"
print(f" ✅ 实体搜索成功：找到 {len(result['results'])} 个结果")
# 时间范围查询
result = client.query_by_time_range(days=7)
assert result['success'], "时间范围查询失败"
print(f" ✅ 时间查询成功：找到 {len(result['results'])} 个结果")
def test_bulk_operations(client):
"""测试批量操作."""
print("\n🧪 测试批量操作...")
episodes = [
{"content": "批量测试 1", "metadata": {"type": "test"}},
{"content": "批量测试 2", "metadata": {"type": "test"}},
]
result = client.add_episodes_bulk(episodes)
assert result['success'], "批量添加失败"
print(f" ✅ 批量操作成功：{result['success_count']} 个成功")
def test_delete(client):
"""测试删除功能."""
print("\n🧪 测试删除功能...")
result = client.delete_episode(content="测试 Episode：我喜欢 Python")
assert result['success'], "删除失败"
print(" ✅ 删除成功")
def main():
"""运行所有测试."""
print("🚀 Graphiti MCP Server 功能测试\n")
print("=" * 50)
try:
# 测试配置
cm = test_config()
# 测试连接
client = test_client_connection(cm)
if not client:
print("\n❌ 无法继续测试：连接失败")
return
# 测试添加
episode_id = test_add_episode(client)
# 测试搜索
test_search(client)
# 测试批量操作
test_bulk_operations(client)
# 测试删除
test_delete(client)
# 清理
client.disconnect()
print("\n" + "=" * 50)
print("✅ 所有测试通过！")
except AssertionError as e:
print(f"\n❌ 测试失败：{e}")
sys.exit(1)
except Exception as e:
print(f"\n❌ 测试出错：{e}")
import traceback
traceback.print_exc()
sys.exit(1)
if __name__ == "__main__":
main()
