Spaces:
Runtime error
Runtime error
File size: 12,818 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 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 | """
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']) |