|
from unittest.mock import MagicMock
|
|
|
|
from rp.core.enhanced_assistant import EnhancedAssistant
|
|
|
|
|
|
def test_enhanced_assistant_init():
|
|
mock_base = MagicMock()
|
|
assistant = EnhancedAssistant(mock_base)
|
|
assert assistant.base == mock_base
|
|
assert assistant.current_conversation_id is not None
|
|
|
|
|
|
def test_enhanced_call_api_with_cache():
|
|
mock_base = MagicMock()
|
|
mock_base.model = "test-model"
|
|
mock_base.api_url = "http://test"
|
|
mock_base.api_key = "key"
|
|
mock_base.use_tools = False
|
|
mock_base.verbose = False
|
|
|
|
assistant = EnhancedAssistant(mock_base)
|
|
assistant.api_cache = MagicMock()
|
|
assistant.api_cache.get.return_value = {"cached": True}
|
|
|
|
result = assistant.enhanced_call_api([{"role": "user", "content": "test"}])
|
|
assert result == {"cached": True}
|
|
assistant.api_cache.get.assert_called_once()
|
|
|
|
|
|
def test_enhanced_call_api_without_cache():
|
|
mock_base = MagicMock()
|
|
mock_base.model = "test-model"
|
|
mock_base.api_url = "http://test"
|
|
mock_base.api_key = "key"
|
|
mock_base.use_tools = False
|
|
mock_base.verbose = False
|
|
|
|
assistant = EnhancedAssistant(mock_base)
|
|
assistant.api_cache = None
|
|
|
|
# It will try to call API and fail with network error, but that's expected
|
|
result = assistant.enhanced_call_api([{"role": "user", "content": "test"}])
|
|
assert "error" in result
|
|
|
|
|
|
def test_execute_workflow_not_found():
|
|
mock_base = MagicMock()
|
|
assistant = EnhancedAssistant(mock_base)
|
|
assistant.workflow_storage = MagicMock()
|
|
assistant.workflow_storage.load_workflow_by_name.return_value = None
|
|
|
|
result = assistant.execute_workflow("nonexistent")
|
|
assert "error" in result
|
|
|
|
|
|
def test_create_agent():
|
|
mock_base = MagicMock()
|
|
assistant = EnhancedAssistant(mock_base)
|
|
assistant.agent_manager = MagicMock()
|
|
assistant.agent_manager.create_agent.return_value = "agent_id"
|
|
|
|
result = assistant.create_agent("role")
|
|
assert result == "agent_id"
|
|
|
|
|
|
def test_search_knowledge():
|
|
mock_base = MagicMock()
|
|
assistant = EnhancedAssistant(mock_base)
|
|
assistant.knowledge_store = MagicMock()
|
|
assistant.knowledge_store.search_entries.return_value = [{"result": True}]
|
|
|
|
result = assistant.search_knowledge("query")
|
|
assert result == [{"result": True}]
|
|
|
|
|
|
def test_get_cache_statistics():
|
|
mock_base = MagicMock()
|
|
assistant = EnhancedAssistant(mock_base)
|
|
assistant.api_cache = MagicMock()
|
|
assistant.api_cache.get_statistics.return_value = {"total_cache_hits": 10}
|
|
assistant.tool_cache = MagicMock()
|
|
assistant.tool_cache.get_statistics.return_value = {"total_cache_hits": 5}
|
|
|
|
stats = assistant.get_cache_statistics()
|
|
assert "api_cache" in stats
|
|
assert "tool_cache" in stats
|
|
|
|
|
|
def test_clear_caches():
|
|
mock_base = MagicMock()
|
|
assistant = EnhancedAssistant(mock_base)
|
|
assistant.api_cache = MagicMock()
|
|
assistant.tool_cache = MagicMock()
|
|
|
|
assistant.clear_caches()
|
|
assistant.api_cache.clear_all.assert_called_once()
|
|
assistant.tool_cache.clear_all.assert_called_once()
|