"""最终覆盖率补全测试 - 覆盖剩余未测试代码."""
import pytest
import os
import json
from unittest.mock import Mock, MagicMock, patch, AsyncMock
from pathlib import Path
from src.config_manager import ConfigManager
from src.graphiti_client import GraphitiClient
class TestGraphitiClientClearGraph:
"""测试 GraphitiClient 清空图谱功能."""
@pytest.fixture
def config_manager(self, temp_config_dir):
"""创建配置管理器."""
return ConfigManager(config_path=temp_config_dir / ".graphitiace" / "config.json")
@pytest.fixture
def connected_client(self, config_manager):
"""创建已连接的客户端."""
client = GraphitiClient(config_manager)
mock_driver = MagicMock()
mock_session = MagicMock()
mock_driver.session.return_value.__enter__ = MagicMock(return_value=mock_session)
mock_driver.session.return_value.__exit__ = MagicMock(return_value=None)
client.driver = mock_driver
client._connected = True
return client, mock_session
def test_clear_graph_with_group_id(self, connected_client, config_manager):
"""测试清空指定组的图谱."""
client, mock_session = connected_client
# 设置 group_id
config_manager.set_group_id("test_group")
# Mock 查询结果
mock_result = MagicMock()
mock_record = {"deleted_count": 5}
mock_result.single.return_value = mock_record
mock_result.peek.return_value = mock_record
mock_session.run.return_value = mock_result
result = client.clear_graph()
assert result is not None
assert result["success"] is True
assert "test_group" in result["message"]
class TestGraphitiClientStatisticsCache:
"""测试 GraphitiClient 统计信息缓存."""
@pytest.fixture
def config_manager(self, temp_config_dir):
"""创建配置管理器."""
return ConfigManager(config_path=temp_config_dir / ".graphitiace" / "config.json")
@pytest.fixture
def connected_client(self, config_manager):
"""创建已连接的客户端."""
client = GraphitiClient(config_manager)
mock_driver = MagicMock()
mock_session = MagicMock()
mock_driver.session.return_value.__enter__ = MagicMock(return_value=mock_session)
mock_driver.session.return_value.__exit__ = MagicMock(return_value=None)
client.driver = mock_driver
client._connected = True
return client, mock_session
def test_get_statistics_from_cache(self, connected_client, config_manager):
"""测试从缓存获取统计信息."""
client, mock_session = connected_client
# 设置缓存
group_id = config_manager.get_group_id() or "default"
cache_key = f"statistics:{group_id}"
cached_result = {
"success": True,
"message": "统计信息获取成功",
"statistics": {"total_nodes": 10}
}
client.cache.set(cache_key, cached_result, ttl=60)
# 从缓存获取
result = client.get_statistics(use_cache=True)
assert result is not None
assert result["success"] is True
class TestGraphitiClientInitializeGraphiti:
"""测试 GraphitiClient 初始化 Graphiti."""
@pytest.fixture
def fresh_config_manager(self, temp_config_dir):
"""创建全新的配置管理器."""
fresh_path = temp_config_dir / ".graphitiace_final" / "config.json"
fresh_path.parent.mkdir(parents=True, exist_ok=True)
return ConfigManager(config_path=fresh_path)
def test_initialize_graphiti_with_openai_base_url(self, fresh_config_manager):
"""测试使用 OpenAI base_url 初始化 Graphiti."""
client = GraphitiClient(fresh_config_manager)
# 配置 Neo4j
fresh_config_manager.configure_neo4j(
uri="bolt://localhost:7687",
username="neo4j",
password="test"
)
# 配置 API 包含 base_url
fresh_config_manager.configure_api(
provider="openai",
api_key="test_key",
base_url="https://api.example.com",
model="gpt-4"
)
# 尝试初始化（可能失败，因为没有真实的 Graphiti）
result = client._initialize_graphiti()
# 结果取决于 graphiti-core 是否安装
assert isinstance(result, bool)
def test_initialize_graphiti_with_anthropic(self, fresh_config_manager):
"""测试使用 Anthropic 初始化 Graphiti."""
client = GraphitiClient(fresh_config_manager)
# 配置 Neo4j
fresh_config_manager.configure_neo4j(
uri="bolt://localhost:7687",
username="neo4j",
password="test"
)
# 配置 Anthropic API
fresh_config_manager.configure_api(
provider="anthropic",
api_key="test_key",
model="claude-3"
)
# 尝试初始化
result = client._initialize_graphiti()
assert isinstance(result, bool)
class TestServerACEManagerExceptions:
"""测试 server.py 中 ACE Manager 异常处理."""
def test_create_server_ace_import_error(self, temp_config_dir):
"""测试 ACE Manager 导入失败."""
# 直接导入并创建服务器
# ACE Manager 导入可能成功也可能失败
from src.server import create_server
server = create_server()
assert server is not None
def test_create_server_with_test_mode(self, temp_config_dir):
"""测试测试模式下创建服务器."""
with patch.dict(os.environ, {'GRAPHITIACE_TEST_MODE': 'true'}):
from src.server import create_server
server = create_server()
assert server is not None
class TestServerCallToolEdgeCases:
"""测试 server.py 工具调用边缘情况."""
@pytest.fixture
def config_manager(self, temp_config_dir):
"""创建配置管理器."""
return ConfigManager(config_path=temp_config_dir / ".graphitiace" / "config.json")
@pytest.mark.asyncio
async def test_handle_tool_call_empty_args(self, config_manager):
"""测试 handle_tool_call 空参数."""
from src.tools import handle_tool_call
result = await handle_tool_call(
tool_name="check_configuration",
arguments={},
config_manager=config_manager
)
assert result is not None
class TestServerResourceStrategyHeatmap:
"""测试 strategy-heatmap 资源."""
def test_strategy_heatmap_logic_ace_none(self):
"""测试 strategy-heatmap 逻辑 - ACE 为 None."""
_ace_manager = None
if _ace_manager is None or not _ace_manager.is_enabled():
content = json.dumps({
"success": False,
"message": "ACE Manager 未启用或未安装，无法生成策略热力图。"
}, indent=2, ensure_ascii=False)
result = json.loads(content)
assert result["success"] is False
def test_strategy_heatmap_logic_ace_disabled(self):
"""测试 strategy-heatmap 逻辑 - ACE 未启用."""
_ace_manager = MagicMock()
_ace_manager.is_enabled.return_value = False
if _ace_manager is None or not _ace_manager.is_enabled():
content = json.dumps({
"success": False,
"message": "ACE Manager 未启用或未安装，无法生成策略热力图。"
}, indent=2, ensure_ascii=False)
result = json.loads(content)
assert result["success"] is False
def test_strategy_heatmap_logic_success(self):
"""测试 strategy-heatmap 逻辑 - 成功."""
_ace_manager = MagicMock()
_ace_manager.is_enabled.return_value = True
_ace_manager.get_strategy_heatmap.return_value = {
"success": True,
"heatmap": {"tool1": 5}
}
if _ace_manager is not None and _ace_manager.is_enabled():
heatmap = _ace_manager.get_strategy_heatmap(limit=30, group_by="tool")
if heatmap and heatmap.get("success"):
content = json.dumps(heatmap, indent=2, ensure_ascii=False, default=str)
result = json.loads(content)
assert result["success"] is True
def test_strategy_heatmap_logic_no_data(self):
"""测试 strategy-heatmap 逻辑 - 无数据."""
_ace_manager = MagicMock()
_ace_manager.is_enabled.return_value = True
_ace_manager.get_strategy_heatmap.return_value = {
"success": False,
"message": "暂无数据"
}
if _ace_manager is not None and _ace_manager.is_enabled():
heatmap = _ace_manager.get_strategy_heatmap(limit=30, group_by="tool")
if not (heatmap and heatmap.get("success")):
message = (heatmap or {}).get("message", "暂无可用的策略数据。")
content = json.dumps({
"success": False,
"message": message
}, indent=2, ensure_ascii=False)
result = json.loads(content)
assert result["success"] is False
assert "暂无" in result["message"]
class TestHealthCheckDegradedStatus:
"""测试健康检查 degraded 状态."""
@pytest.fixture
def config_manager(self, temp_config_dir):
"""创建配置管理器."""
return ConfigManager(config_path=temp_config_dir / ".graphitiace" / "config.json")
def test_health_check_with_warning(self, config_manager):
"""测试健康检查返回 warning 时的处理."""
# 模拟健康检查结果处理逻辑
result = {
"status": "healthy",
"checks": {
"neo4j": {"status": "healthy"},
"config": {"status": "warning"} # warning 状态
}
}
check_statuses = [check.get("status") for check in result["checks"].values()]
if "error" in check_statuses:
result["status"] = "unhealthy"
elif "warning" in check_statuses:
result["status"] = "degraded"
assert result["status"] == "degraded"
class TestNeo4jDateTimeImport:
"""测试 Neo4jDateTime 导入."""
def test_neo4j_datetime_import_success(self):
"""测试 Neo4jDateTime 成功导入."""
try:
from neo4j.time import DateTime as Neo4jDateTime
assert Neo4jDateTime is not None
except ImportError:
# 如果导入失败，Neo4jDateTime 应该是 None
pass
def test_neo4j_to_dict_with_none_datetime(self):
"""测试 _neo4j_to_dict 处理 None Neo4jDateTime."""
from src.graphiti_client import _neo4j_to_dict
# 测试普通对象
result = _neo4j_to_dict("test")
assert result == "test"
# 测试 None
result = _neo4j_to_dict(None)
assert result is None
class TestGraphitiClientConnectionFailure:
"""测试 GraphitiClient 连接失败."""
@pytest.fixture
def config_manager(self, temp_config_dir):
"""创建配置管理器."""
return ConfigManager(config_path=temp_config_dir / ".graphitiace" / "config.json")
def test_connect_failure_cleans_driver(self, config_manager):
"""测试连接失败时清理 driver."""
client = GraphitiClient(config_manager)
config_manager.configure_neo4j(
uri="bolt://localhost:7687",
username="neo4j",
password="wrong_password"
)
# Mock GraphDatabase.driver 抛出异常
with patch('src.graphiti_client.GraphDatabase.driver', side_effect=Exception("Connection failed")):
result = client.connect()
assert result is False
assert client.driver is None
