Spaces:
Runtime error
Runtime error
File size: 8,945 Bytes
330b6e4 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 | """
Unit tests for ChatAgent service.
Tests the core functionality of the ChatAgent class including message processing,
language switching, streaming responses, and error handling.
"""
import unittest
from unittest.mock import Mock, patch
from datetime import datetime
from chat_agent.services.chat_agent import ChatAgent, ChatAgentError, create_chat_agent
from chat_agent.services.groq_client import GroqClient, ChatMessage, LanguageContext
from chat_agent.services.language_context import LanguageContextManager
from chat_agent.services.session_manager import SessionManager, SessionNotFoundError
from chat_agent.services.chat_history import ChatHistoryManager
class TestChatAgent(unittest.TestCase):
"""Test cases for ChatAgent class."""
def setUp(self):
"""Set up test fixtures."""
# Create mock dependencies
self.mock_groq_client = Mock(spec=GroqClient)
self.mock_groq_client.generate_response.return_value = "Test LLM response"
self.mock_groq_client.stream_response.return_value = iter(["Test ", "stream ", "response"])
self.mock_groq_client.get_model_info.return_value = {"model": "test-model"}
self.language_context_manager = LanguageContextManager()
self.mock_session_manager = Mock(spec=SessionManager)
mock_session = Mock()
mock_session.id = "test-session"
mock_session.language = "python"
mock_session.message_count = 0
self.mock_session_manager.get_session.return_value = mock_session
self.mock_chat_history_manager = Mock(spec=ChatHistoryManager)
mock_message = Mock()
mock_message.id = "test-message-id"
mock_message.role = "user"
mock_message.content = "test content"
mock_message.language = "python"
mock_message.timestamp = datetime.utcnow()
mock_message.message_metadata = {}
self.mock_chat_history_manager.store_message.return_value = mock_message
self.mock_chat_history_manager.get_recent_history.return_value = [mock_message]
self.mock_chat_history_manager.get_message_count.return_value = 1
self.mock_chat_history_manager.get_cache_stats.return_value = {"cached_messages": 1}
# Create ChatAgent instance
self.chat_agent = ChatAgent(
self.mock_groq_client,
self.language_context_manager,
self.mock_session_manager,
self.mock_chat_history_manager
)
def test_initialization(self):
"""Test ChatAgent initialization."""
self.assertEqual(self.chat_agent.groq_client, self.mock_groq_client)
self.assertEqual(self.chat_agent.language_context_manager, self.language_context_manager)
self.assertEqual(self.chat_agent.session_manager, self.mock_session_manager)
self.assertEqual(self.chat_agent.chat_history_manager, self.mock_chat_history_manager)
def test_process_message_success(self):
"""Test successful message processing."""
result = self.chat_agent.process_message("test-session", "Hello, world!")
self.assertEqual(result['response'], "Test LLM response")
self.assertEqual(result['language'], "python")
self.assertEqual(result['session_id'], "test-session")
self.assertIn('message_id', result)
self.assertIn('timestamp', result)
self.assertIn('metadata', result)
# Verify service calls
self.mock_session_manager.get_session.assert_called_with("test-session")
self.mock_session_manager.update_session_activity.assert_called_with("test-session")
self.mock_session_manager.increment_message_count.assert_called_with("test-session")
self.assertEqual(self.mock_chat_history_manager.store_message.call_count, 2)
self.mock_groq_client.generate_response.assert_called_once()
def test_process_message_with_language_override(self):
"""Test message processing with language override."""
result = self.chat_agent.process_message("test-session", "Hello!", "javascript")
self.assertEqual(result['language'], "javascript")
self.mock_session_manager.set_session_language.assert_called_with("test-session", "javascript")
def test_process_message_session_not_found(self):
"""Test message processing with session not found."""
self.mock_session_manager.get_session.side_effect = SessionNotFoundError("Not found")
with self.assertRaises(ChatAgentError) as context:
self.chat_agent.process_message("invalid-session", "Hello!")
self.assertIn("Session error", str(context.exception))
def test_switch_language_success(self):
"""Test successful language switching."""
result = self.chat_agent.switch_language("test-session", "javascript")
self.assertTrue(result['success'])
self.assertEqual(result['new_language'], "javascript")
self.assertEqual(result['previous_language'], "python")
self.assertEqual(result['session_id'], "test-session")
self.assertIn('message', result)
# Verify service calls
self.mock_session_manager.set_session_language.assert_called_with("test-session", "javascript")
self.mock_chat_history_manager.store_message.assert_called_once()
def test_switch_language_invalid_language(self):
"""Test language switching with invalid language."""
with self.assertRaises(ChatAgentError) as context:
self.chat_agent.switch_language("test-session", "invalid-lang")
self.assertIn("Unsupported language", str(context.exception))
def test_stream_response_success(self):
"""Test successful streaming response."""
stream_results = list(self.chat_agent.stream_response("test-session", "Hello!"))
self.assertGreaterEqual(len(stream_results), 5) # start + chunks + complete
# Check start event
start_event = stream_results[0]
self.assertEqual(start_event['type'], 'start')
self.assertEqual(start_event['session_id'], "test-session")
# Check chunk events
chunk_events = [event for event in stream_results if event['type'] == 'chunk']
self.assertEqual(len(chunk_events), 3)
# Check complete event
complete_event = stream_results[-1]
self.assertEqual(complete_event['type'], 'complete')
self.assertEqual(complete_event['session_id'], "test-session")
def test_get_chat_history_success(self):
"""Test successful chat history retrieval."""
history = self.chat_agent.get_chat_history("test-session", 10)
self.assertIsInstance(history, list)
self.assertEqual(len(history), 1)
message = history[0]
self.assertIn('id', message)
self.assertIn('role', message)
self.assertIn('content', message)
self.assertIn('language', message)
self.assertIn('timestamp', message)
self.mock_chat_history_manager.get_recent_history.assert_called_with("test-session", 10)
def test_get_session_info_success(self):
"""Test successful session info retrieval."""
info = self.chat_agent.get_session_info("test-session")
self.assertIn('session', info)
self.assertIn('language_context', info)
self.assertIn('statistics', info)
self.assertIn('supported_languages', info)
# Verify service calls
self.mock_session_manager.get_session.assert_called_with("test-session")
self.mock_chat_history_manager.get_message_count.assert_called_with("test-session")
self.mock_chat_history_manager.get_cache_stats.assert_called_with("test-session")
class TestChatAgentFactory(unittest.TestCase):
"""Test cases for ChatAgent factory function."""
def test_create_chat_agent(self):
"""Test ChatAgent factory function."""
groq_client = Mock(spec=GroqClient)
language_manager = LanguageContextManager()
session_manager = Mock(spec=SessionManager)
history_manager = Mock(spec=ChatHistoryManager)
agent = create_chat_agent(groq_client, language_manager, session_manager, history_manager)
self.assertIsInstance(agent, ChatAgent)
self.assertEqual(agent.groq_client, groq_client)
self.assertEqual(agent.language_context_manager, language_manager)
self.assertEqual(agent.session_manager, session_manager)
self.assertEqual(agent.chat_history_manager, history_manager)
if __name__ == '__main__':
unittest.main() |