""" Integration tests for language switching and chat history persistence. Tests the complete integration between language context, chat history, and session management. """ import pytest import time from unittest.mock import patch, MagicMock from chat_agent.services.chat_agent import ChatAgent from chat_agent.services.session_manager import SessionManager from chat_agent.services.language_context import LanguageContextManager from chat_agent.services.chat_history import ChatHistoryManager from chat_agent.services.groq_client import GroqClient class TestLanguageSwitchingIntegration: """Integration tests for language switching functionality.""" @pytest.fixture def mock_groq_client(self): """Mock Groq client with language-specific responses.""" with patch('chat_agent.services.groq_client.GroqClient') as mock: mock_instance = MagicMock() def mock_generate_response(prompt, chat_history=None, language_context=None, **kwargs): # Return language-specific responses if language_context and 'python' in language_context.lower(): return "This is a Python-specific response about programming concepts." elif language_context and 'javascript' in language_context.lower(): return "This is a JavaScript-specific response about web development." elif language_context and 'java' in language_context.lower(): return "This is a Java-specific response about object-oriented programming." else: return "This is a general programming response." mock_instance.generate_response.side_effect = mock_generate_response mock.return_value = mock_instance yield mock_instance @pytest.fixture def integrated_system(self, mock_groq_client): """Create fully integrated chat system.""" session_manager = SessionManager() language_context_manager = LanguageContextManager() chat_history_manager = ChatHistoryManager() chat_agent = ChatAgent( groq_client=mock_groq_client, session_manager=session_manager, language_context_manager=language_context_manager, chat_history_manager=chat_history_manager ) return { 'chat_agent': chat_agent, 'session_manager': session_manager, 'language_context_manager': language_context_manager, 'chat_history_manager': chat_history_manager, 'groq_client': mock_groq_client } def test_language_switching_preserves_history(self, integrated_system): """Test that language switching preserves chat history.""" user_id = "test-user-lang-switch" system = integrated_system # Create session with Python session = system['session_manager'].create_session(user_id, language="python") session_id = session['session_id'] # Send Python message python_message = "How do I create a list in Python?" python_response = system['chat_agent'].process_message( session_id=session_id, message=python_message, language="python" ) # Verify Python response assert "Python" in python_response # Check history after Python message history_after_python = system['chat_history_manager'].get_recent_history(session_id) assert len(history_after_python) == 2 assert history_after_python[0]['content'] == python_message assert history_after_python[0]['language'] == 'python' assert history_after_python[1]['language'] == 'python' # Switch to JavaScript system['chat_agent'].switch_language(session_id, "javascript") # Verify language context changed current_language = system['language_context_manager'].get_language(session_id) assert current_language == "javascript" # Send JavaScript message js_message = "How do I create an array in JavaScript?" js_response = system['chat_agent'].process_message( session_id=session_id, message=js_message, language="javascript" ) # Verify JavaScript response assert "JavaScript" in js_response # Check complete history complete_history = system['chat_history_manager'].get_recent_history(session_id, limit=10) assert len(complete_history) == 4 # Verify history contains both languages python_messages = [msg for msg in complete_history if msg['language'] == 'python'] js_messages = [msg for msg in complete_history if msg['language'] == 'javascript'] assert len(python_messages) == 2 # User message + response assert len(js_messages) == 2 # User message + response # Verify message order is preserved assert complete_history[0]['content'] == python_message assert complete_history[1]['content'] == python_response assert complete_history[2]['content'] == js_message assert complete_history[3]['content'] == js_response def test_multiple_language_switches_in_conversation(self, integrated_system): """Test multiple language switches within a single conversation.""" user_id = "test-user-multi-switch" system = integrated_system # Create session session = system['session_manager'].create_session(user_id, language="python") session_id = session['session_id'] # Conversation flow: Python -> JavaScript -> Java -> Python conversation_flow = [ ("python", "What is a Python dictionary?"), ("javascript", "How do I create an object in JavaScript?"), ("java", "Explain Java classes"), ("python", "What is list comprehension in Python?") ] for language, message in conversation_flow: # Switch language system['chat_agent'].switch_language(session_id, language) # Verify language context current_lang = system['language_context_manager'].get_language(session_id) assert current_lang == language # Send message response = system['chat_agent'].process_message( session_id=session_id, message=message, language=language ) # Verify response is language-specific assert language.title() in response or language.lower() in response.lower() # Verify complete conversation history complete_history = system['chat_history_manager'].get_full_history(session_id) assert len(complete_history) == 8 # 4 user messages + 4 responses # Verify language distribution language_counts = {} for msg in complete_history: lang = msg['language'] language_counts[lang] = language_counts.get(lang, 0) + 1 assert language_counts['python'] == 4 # 2 messages + 2 responses assert language_counts['javascript'] == 2 # 1 message + 1 response assert language_counts['java'] == 2 # 1 message + 1 response def test_language_context_affects_llm_prompts(self, integrated_system): """Test that language context properly affects LLM prompts.""" user_id = "test-user-context-prompts" system = integrated_system # Create session session = system['session_manager'].create_session(user_id, language="python") session_id = session['session_id'] # Send message in Python context message = "How do I handle errors?" system['chat_agent'].process_message( session_id=session_id, message=message, language="python" ) # Verify Groq client was called with Python context call_args = system['groq_client'].generate_response.call_args assert call_args is not None # Check that language context was passed if len(call_args) > 1: # positional args language_context = call_args[1] if len(call_args) > 1 else None else: # keyword args language_context = call_args.kwargs.get('language_context') # Switch to JavaScript and send same message system['chat_agent'].switch_language(session_id, "javascript") system['chat_agent'].process_message( session_id=session_id, message=message, language="javascript" ) # Verify second call had different context second_call_args = system['groq_client'].generate_response.call_args assert second_call_args is not None def test_chat_history_persistence_across_language_switches(self, integrated_system): """Test that chat history persists correctly across language switches.""" user_id = "test-user-history-persistence" system = integrated_system # Create session session = system['session_manager'].create_session(user_id, language="python") session_id = session['session_id'] # Build conversation with language switches messages_and_languages = [ ("python", "What is a variable?"), ("python", "How do I create a function?"), ("javascript", "What is a closure?"), ("javascript", "How do I use async/await?"), ("java", "What is inheritance?") ] for language, message in messages_and_languages: system['chat_agent'].switch_language(session_id, language) system['chat_agent'].process_message( session_id=session_id, message=message, language=language ) # Test different history retrieval methods # 1. Recent history (last 6 messages) recent_history = system['chat_history_manager'].get_recent_history(session_id, limit=6) assert len(recent_history) == 6 # Should include last 3 conversations (user + assistant messages) expected_languages = ["javascript", "javascript", "java", "java"] actual_languages = [msg['language'] for msg in recent_history[-4:]] assert actual_languages == expected_languages # 2. Full history full_history = system['chat_history_manager'].get_full_history(session_id) assert len(full_history) == 10 # 5 user messages + 5 responses # 3. Language-specific filtering (if implemented) python_messages = [msg for msg in full_history if msg['language'] == 'python'] js_messages = [msg for msg in full_history if msg['language'] == 'javascript'] java_messages = [msg for msg in full_history if msg['language'] == 'java'] assert len(python_messages) == 4 # 2 user + 2 assistant assert len(js_messages) == 4 # 2 user + 2 assistant assert len(java_messages) == 2 # 1 user + 1 assistant def test_session_language_state_persistence(self, integrated_system): """Test that session language state persists correctly.""" user_id = "test-user-session-state" system = integrated_system # Create session with default language session = system['session_manager'].create_session(user_id, language="python") session_id = session['session_id'] # Verify initial language initial_language = system['language_context_manager'].get_language(session_id) assert initial_language == "python" # Switch languages multiple times languages = ["javascript", "java", "python", "cpp"] for language in languages: system['chat_agent'].switch_language(session_id, language) # Verify immediate state change current_language = system['language_context_manager'].get_language(session_id) assert current_language == language # Send a message to ensure state is used response = system['chat_agent'].process_message( session_id=session_id, message=f"Test message in {language}", language=language ) assert response is not None # Verify final state final_language = system['language_context_manager'].get_language(session_id) assert final_language == "cpp" # Simulate session retrieval (as if from database) retrieved_session = system['session_manager'].get_session(session_id) assert retrieved_session is not None # Language context should still be accessible persistent_language = system['language_context_manager'].get_language(session_id) assert persistent_language == "cpp" def test_concurrent_language_switches(self, integrated_system): """Test concurrent language switches across multiple sessions.""" user_ids = ["concurrent-user-1", "concurrent-user-2", "concurrent-user-3"] system = integrated_system # Create multiple sessions sessions = [] for user_id in user_ids: session = system['session_manager'].create_session(user_id, language="python") sessions.append(session) # Perform concurrent language switches import threading def switch_and_chat(session_id, target_language, message): system['chat_agent'].switch_language(session_id, target_language) response = system['chat_agent'].process_message( session_id=session_id, message=message, language=target_language ) return response # Create threads for concurrent operations threads = [] results = {} operations = [ (sessions[0]['session_id'], "javascript", "JS question"), (sessions[1]['session_id'], "java", "Java question"), (sessions[2]['session_id'], "python", "Python question") ] for session_id, language, message in operations: thread = threading.Thread( target=lambda sid=session_id, lang=language, msg=message: results.update({sid: switch_and_chat(sid, lang, msg)}) ) threads.append(thread) thread.start() # Wait for all threads to complete for thread in threads: thread.join() # Verify all operations completed successfully assert len(results) == 3 # Verify each session has correct language context for i, (session_id, expected_language, _) in enumerate(operations): current_language = system['language_context_manager'].get_language(session_id) assert current_language == expected_language # Verify response was generated assert session_id in results assert results[session_id] is not None def test_language_switching_with_chat_context(self, integrated_system): """Test that language switching maintains proper chat context for LLM.""" user_id = "test-user-context-maintenance" system = integrated_system # Create session session = system['session_manager'].create_session(user_id, language="python") session_id = session['session_id'] # Start conversation in Python system['chat_agent'].process_message( session_id=session_id, message="I'm learning about data structures", language="python" ) system['chat_agent'].process_message( session_id=session_id, message="Can you explain Python lists?", language="python" ) # Switch to JavaScript but ask follow-up question system['chat_agent'].switch_language(session_id, "javascript") response = system['chat_agent'].process_message( session_id=session_id, message="What's the equivalent in JavaScript?", language="javascript" ) # Verify that the LLM received chat history as context # The mock should have been called with chat history call_args = system['groq_client'].generate_response.call_args assert call_args is not None # Check that chat history was provided if len(call_args) > 0: # History should be in the call arguments chat_history = None if len(call_args) > 1: chat_history = call_args[1] if isinstance(call_args[1], list) else None # Or in keyword arguments if not chat_history: chat_history = call_args.kwargs.get('chat_history') # Verify context was maintained history = system['chat_history_manager'].get_recent_history(session_id, limit=5) assert len(history) >= 4 # Previous conversation + current message # Verify response acknowledges context assert response is not None assert "JavaScript" in response if __name__ == '__main__': pytest.main([__file__, '-v'])