"""工具语义搜索所有分支测试."""
import pytest
from unittest.mock import Mock, MagicMock, patch
from src.tools import handle_tool_call
from src.config_manager import ConfigManager
class TestToolsSemanticSearchAllBranches:
"""工具语义搜索所有分支测试类."""
@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_semantic_search_enhanced_keyword_branch(self, config_manager):
"""测试语义搜索增强关键词分支."""
from src.graphiti_client import GraphitiClient
from unittest.mock import AsyncMock
mock_client = Mock(spec=GraphitiClient)
mock_client.is_connected.return_value = True
mock_client.semantic_search = AsyncMock(return_value={
"success": True,
"search_type": "enhanced_keyword",
"results": [
{"content": "Test content", "name": "Entity1"}
]
})
result = await handle_tool_call(
tool_name="semantic_search",
arguments={"query": "test"},
config_manager=config_manager,
graphiti_client=mock_client
)
assert result is not None
assert isinstance(result, list)
result_text = "".join([str(item.text) if hasattr(item, 'text') else str(item) for item in result])
assert "增强关键词" in result_text or "enhanced" in result_text.lower() or len(result_text) > 0
@pytest.mark.asyncio
async def test_semantic_search_semantic_branch(self, config_manager):
"""测试语义搜索语义分支."""
from src.graphiti_client import GraphitiClient
from unittest.mock import AsyncMock
mock_client = Mock(spec=GraphitiClient)
mock_client.is_connected.return_value = True
mock_client.semantic_search = AsyncMock(return_value={
"success": True,
"search_type": "semantic",
"results": [
{
"from_node": {"name": "Node1"},
"to_node": {"name": "Node2"},
"relationship": "RELATES_TO"
}
]
})
result = await handle_tool_call(
tool_name="semantic_search",
arguments={"query": "test"},
config_manager=config_manager,
graphiti_client=mock_client
)
assert result is not None
assert isinstance(result, list)
result_text = "".join([str(item.text) if hasattr(item, 'text') else str(item) for item in result])
assert "向量搜索" in result_text or "semantic" in result_text.lower() or len(result_text) > 0
@pytest.mark.asyncio
async def test_semantic_search_else_branch(self, config_manager):
"""测试语义搜索 else 分支（关键词类型）."""
from src.graphiti_client import GraphitiClient
from unittest.mock import AsyncMock
mock_client = Mock(spec=GraphitiClient)
mock_client.is_connected.return_value = True
mock_client.semantic_search = AsyncMock(return_value={
"success": True,
"search_type": "keyword", # 关键词类型，会走 else 分支
"results": [
{"content": "Test content", "name": "Entity1"}
]
})
result = await handle_tool_call(
tool_name="semantic_search",
arguments={"query": "test"},
config_manager=config_manager,
graphiti_client=mock_client
)
assert result is not None
assert isinstance(result, list)
result_text = "".join([str(item.text) if hasattr(item, 'text') else str(item) for item in result])
assert "关键词搜索" in result_text or "keyword" in result_text.lower() or len(result_text) > 0
