""" End-to-end tests for complete user chat workflows. Tests the entire flow from session creation to chat completion. """ import pytest import asyncio import json import time from unittest.mock import patch, MagicMock import socketio from flask import Flask 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 TestCompleteUserChatWorkflow: """Test complete user chat workflows from start to finish.""" @pytest.fixture def app(self): """Create test Flask app.""" app = Flask(__name__) app.config['TESTING'] = True app.config['SECRET_KEY'] = 'test-secret-key' return app @pytest.fixture def client(self, app): """Create test client.""" return app.test_client() @pytest.fixture def mock_groq_client(self): """Mock Groq client for testing.""" with patch('chat_agent.services.groq_client.GroqClient') as mock: mock_instance = MagicMock() mock_instance.generate_response.return_value = "This is a test response about Python programming." mock_instance.stream_response.return_value = iter([ "This is ", "a test ", "response about ", "Python programming." ]) mock.return_value = mock_instance yield mock_instance @pytest.fixture def session_manager(self): """Create session manager for testing.""" return SessionManager() @pytest.fixture def language_context_manager(self): """Create language context manager for testing.""" return LanguageContextManager() @pytest.fixture def chat_history_manager(self): """Create chat history manager for testing.""" return ChatHistoryManager() @pytest.fixture def chat_agent(self, mock_groq_client, session_manager, language_context_manager, chat_history_manager): """Create chat agent for testing.""" return ChatAgent( groq_client=mock_groq_client, session_manager=session_manager, language_context_manager=language_context_manager, chat_history_manager=chat_history_manager ) def test_complete_python_chat_workflow(self, chat_agent, session_manager, language_context_manager, chat_history_manager): """Test complete workflow: create session, send messages, get responses, verify history.""" user_id = "test-user-123" # Step 1: Create new chat session session = session_manager.create_session(user_id, language="python") assert session is not None assert session['user_id'] == user_id assert session['language'] == "python" assert session['message_count'] == 0 # Step 2: Verify default language context language = language_context_manager.get_language(session['session_id']) assert language == "python" # Step 3: Send first message user_message_1 = "Hello, can you help me with Python lists?" response_1 = chat_agent.process_message( session_id=session['session_id'], message=user_message_1, language="python" ) assert response_1 is not None assert "Python" in response_1 or "python" in response_1 # Step 4: Verify message was stored in history history = chat_history_manager.get_recent_history(session['session_id'], limit=10) assert len(history) == 2 # User message + assistant response assert history[0]['role'] == 'user' assert history[0]['content'] == user_message_1 assert history[1]['role'] == 'assistant' # Step 5: Send follow-up message user_message_2 = "Can you show me an example of list comprehension?" response_2 = chat_agent.process_message( session_id=session['session_id'], message=user_message_2, language="python" ) assert response_2 is not None # Step 6: Verify conversation history includes context history = chat_history_manager.get_recent_history(session['session_id'], limit=10) assert len(history) == 4 # 2 user messages + 2 assistant responses # Step 7: Verify session was updated updated_session = session_manager.get_session(session['session_id']) assert updated_session['message_count'] > 0 assert updated_session['last_active'] > session['created_at'] def test_language_switching_workflow(self, chat_agent, session_manager, language_context_manager, chat_history_manager): """Test workflow with language switching mid-conversation.""" user_id = "test-user-456" # Step 1: Create session with Python session = session_manager.create_session(user_id, language="python") # Step 2: Send Python-related message python_message = "How do I create a dictionary in Python?" response_1 = chat_agent.process_message( session_id=session['session_id'], message=python_message, language="python" ) assert response_1 is not None # Step 3: Switch to JavaScript chat_agent.switch_language(session['session_id'], "javascript") # Step 4: Verify language context changed current_language = language_context_manager.get_language(session['session_id']) assert current_language == "javascript" # Step 5: Send JavaScript-related message js_message = "How do I create an object in JavaScript?" response_2 = chat_agent.process_message( session_id=session['session_id'], message=js_message, language="javascript" ) assert response_2 is not None # Step 6: Verify history maintains both conversations history = chat_history_manager.get_recent_history(session['session_id'], limit=10) assert len(history) == 4 # Verify language context in messages python_messages = [msg for msg in history if msg['language'] == 'python'] js_messages = [msg for msg in history if msg['language'] == 'javascript'] assert len(python_messages) == 2 # User message + response assert len(js_messages) == 2 # User message + response def test_error_recovery_workflow(self, chat_agent, session_manager, mock_groq_client): """Test workflow with API errors and recovery.""" user_id = "test-user-789" # Step 1: Create session session = session_manager.create_session(user_id, language="python") # Step 2: Simulate API error mock_groq_client.generate_response.side_effect = Exception("API Error") # Step 3: Send message and expect graceful error handling user_message = "What is a Python function?" response = chat_agent.process_message( session_id=session['session_id'], message=user_message, language="python" ) # Should return fallback response, not raise exception assert response is not None assert "error" in response.lower() or "sorry" in response.lower() # Step 4: Restore API functionality mock_groq_client.generate_response.side_effect = None mock_groq_client.generate_response.return_value = "A function is a reusable block of code." # Step 5: Send another message and verify recovery user_message_2 = "Can you give me an example?" response_2 = chat_agent.process_message( session_id=session['session_id'], message=user_message_2, language="python" ) assert response_2 is not None assert "function" in response_2.lower() def test_session_cleanup_workflow(self, session_manager, chat_history_manager): """Test session cleanup and data persistence.""" user_id = "test-user-cleanup" # Step 1: Create multiple sessions session_1 = session_manager.create_session(user_id, language="python") session_2 = session_manager.create_session(user_id, language="javascript") # Step 2: Add messages to sessions chat_history_manager.store_message( session_1['session_id'], 'user', 'Test message 1', 'python' ) chat_history_manager.store_message( session_2['session_id'], 'user', 'Test message 2', 'javascript' ) # Step 3: Simulate session inactivity with patch('time.time', return_value=time.time() + 3600): # 1 hour later inactive_sessions = session_manager.get_inactive_sessions(timeout=1800) # 30 min timeout assert len(inactive_sessions) >= 2 # Step 4: Clean up inactive sessions session_manager.cleanup_inactive_sessions(timeout=1800) # Step 5: Verify sessions are marked inactive but history preserved history_1 = chat_history_manager.get_full_history(session_1['session_id']) history_2 = chat_history_manager.get_full_history(session_2['session_id']) assert len(history_1) > 0 # History should be preserved assert len(history_2) > 0 # History should be preserved def test_concurrent_user_workflow(self, chat_agent, session_manager): """Test workflow with multiple concurrent users.""" user_ids = ["user-1", "user-2", "user-3"] sessions = [] # Step 1: Create sessions for multiple users for user_id in user_ids: session = session_manager.create_session(user_id, language="python") sessions.append(session) # Step 2: Send messages concurrently messages = [ "What is Python?", "How do I use loops?", "Explain functions please" ] responses = [] for i, session in enumerate(sessions): response = chat_agent.process_message( session_id=session['session_id'], message=messages[i], language="python" ) responses.append(response) # Step 3: Verify all responses received assert len(responses) == 3 for response in responses: assert response is not None assert len(response) > 0 # Step 4: Verify session isolation for session in sessions: history = chat_history_manager.get_recent_history(session['session_id']) assert len(history) == 2 # Only user message + response for this session @pytest.mark.asyncio async def test_streaming_response_workflow(self, chat_agent, session_manager, mock_groq_client): """Test streaming response workflow.""" user_id = "test-user-streaming" # Step 1: Create session session = session_manager.create_session(user_id, language="python") # Step 2: Configure streaming response mock_groq_client.stream_response.return_value = iter([ "Python is ", "a high-level ", "programming language ", "that is easy to learn." ]) # Step 3: Process message with streaming user_message = "What is Python?" response_stream = chat_agent.stream_response( session_id=session['session_id'], message=user_message, language="python" ) # Step 4: Collect streamed response full_response = "" async for chunk in response_stream: full_response += chunk # Step 5: Verify complete response assert "Python is a high-level programming language that is easy to learn." in full_response # Step 6: Verify message was stored history = chat_history_manager.get_recent_history(session['session_id']) assert len(history) == 2 assert history[1]['content'] == full_response.strip() if __name__ == '__main__': pytest.main([__file__, '-v'])