Spaces:
Runtime error
Runtime error
File size: 20,547 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 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452 453 454 455 456 457 458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484 485 486 487 488 489 490 491 492 493 494 495 496 497 498 499 500 501 502 503 504 505 506 507 508 509 510 511 512 513 514 515 516 517 518 519 520 521 522 523 524 525 | """
Unit tests for SessionManager service.
Tests cover session creation, retrieval, activity updates, cleanup operations,
Redis caching, and error handling scenarios.
"""
import pytest
import json
from datetime import datetime, timedelta
from unittest.mock import Mock, patch, MagicMock
from uuid import uuid4
from chat_agent.services.session_manager import (
SessionManager,
SessionManagerError,
SessionNotFoundError,
SessionExpiredError,
create_session_manager
)
from chat_agent.models.chat_session import ChatSession
class TestSessionManager:
"""Test suite for SessionManager class"""
@pytest.fixture
def mock_redis_client(self):
"""Mock Redis client for testing"""
mock_redis = Mock()
mock_redis.setex = Mock()
mock_redis.get = Mock()
mock_redis.delete = Mock()
mock_redis.sadd = Mock()
mock_redis.srem = Mock()
mock_redis.expire = Mock()
mock_redis.keys = Mock()
return mock_redis
@pytest.fixture
def session_manager(self, mock_redis_client):
"""Create SessionManager instance for testing"""
return SessionManager(mock_redis_client, session_timeout=3600)
@pytest.fixture
def sample_session_data(self):
"""Sample session data for testing"""
session_id = str(uuid4())
user_id = str(uuid4())
return {
'session_id': session_id,
'user_id': user_id,
'language': 'python',
'session_metadata': {'test': 'data'}
}
@pytest.fixture
def mock_chat_session(self, sample_session_data):
"""Mock ChatSession instance"""
session = Mock(spec=ChatSession)
session.id = sample_session_data['session_id']
session.user_id = sample_session_data['user_id']
session.language = sample_session_data['language']
session.created_at = datetime.utcnow()
session.last_active = datetime.utcnow()
session.message_count = 0
session.is_active = True
session.session_metadata = sample_session_data['session_metadata']
session.is_expired = Mock(return_value=False)
session.update_activity = Mock()
session.increment_message_count = Mock()
session.set_language = Mock()
session.deactivate = Mock()
return session
class TestSessionCreation:
"""Test session creation functionality"""
@patch('chat_agent.services.session_manager.ChatSession')
def test_create_session_success(self, mock_chat_session_class, session_manager,
mock_redis_client, sample_session_data):
"""Test successful session creation"""
# Setup
mock_session = Mock()
mock_session.id = sample_session_data['session_id']
mock_session.user_id = sample_session_data['user_id']
mock_session.language = sample_session_data['language']
mock_session.created_at = datetime.utcnow()
mock_session.last_active = datetime.utcnow()
mock_session.message_count = 0
mock_session.is_active = True
mock_session.session_metadata = sample_session_data['session_metadata']
mock_chat_session_class.create_session.return_value = mock_session
# Execute
result = session_manager.create_session(
user_id=sample_session_data['user_id'],
language=sample_session_data['language'],
session_metadata=sample_session_data['session_metadata']
)
# Verify
assert result == mock_session
mock_chat_session_class.create_session.assert_called_once_with(
user_id=sample_session_data['user_id'],
language=sample_session_data['language'],
session_metadata=sample_session_data['session_metadata']
)
# Verify Redis caching
mock_redis_client.setex.assert_called_once()
mock_redis_client.sadd.assert_called_once()
@patch('chat_agent.services.session_manager.ChatSession')
def test_create_session_default_language(self, mock_chat_session_class,
session_manager, sample_session_data):
"""Test session creation with default language"""
# Setup
mock_session = Mock()
mock_chat_session_class.create_session.return_value = mock_session
# Execute
session_manager.create_session(user_id=sample_session_data['user_id'])
# Verify default language is used
mock_chat_session_class.create_session.assert_called_once_with(
user_id=sample_session_data['user_id'],
language='python',
session_metadata={}
)
@patch('chat_agent.services.session_manager.ChatSession')
def test_create_session_database_error(self, mock_chat_session_class, session_manager):
"""Test session creation with database error"""
# Setup
from sqlalchemy.exc import SQLAlchemyError
mock_chat_session_class.create_session.side_effect = SQLAlchemyError("DB Error")
# Execute & Verify
with pytest.raises(SessionManagerError, match="Failed to create session"):
session_manager.create_session(user_id="test_user")
class TestSessionRetrieval:
"""Test session retrieval functionality"""
def test_get_session_from_cache(self, session_manager, mock_redis_client,
sample_session_data):
"""Test getting session from Redis cache"""
# Setup
cached_data = {
'id': sample_session_data['session_id'],
'user_id': sample_session_data['user_id'],
'language': sample_session_data['language'],
'created_at': datetime.utcnow().isoformat(),
'last_active': datetime.utcnow().isoformat(),
'message_count': 0,
'is_active': True,
'session_metadata': sample_session_data['session_metadata']
}
mock_redis_client.get.return_value = json.dumps(cached_data)
# Execute
result = session_manager.get_session(sample_session_data['session_id'])
# Verify
assert result.id == sample_session_data['session_id']
assert result.user_id == sample_session_data['user_id']
assert result.language == sample_session_data['language']
mock_redis_client.get.assert_called_once()
@patch('chat_agent.services.session_manager.db')
def test_get_session_from_database(self, mock_db, session_manager,
mock_redis_client, mock_chat_session):
"""Test getting session from database when not in cache"""
# Setup
mock_redis_client.get.return_value = None
mock_db.session.query.return_value.filter.return_value.first.return_value = mock_chat_session
# Execute
result = session_manager.get_session(mock_chat_session.id)
# Verify
assert result == mock_chat_session
mock_redis_client.setex.assert_called_once() # Should cache the result
@patch('chat_agent.services.session_manager.db')
def test_get_session_not_found(self, mock_db, session_manager, mock_redis_client):
"""Test getting non-existent session"""
# Setup
mock_redis_client.get.return_value = None
mock_db.session.query.return_value.filter.return_value.first.return_value = None
# Execute & Verify
with pytest.raises(SessionNotFoundError):
session_manager.get_session("non_existent_id")
def test_get_session_expired_from_cache(self, session_manager, mock_redis_client):
"""Test getting expired session from cache"""
# Setup - expired session
expired_time = datetime.utcnow() - timedelta(hours=2)
cached_data = {
'id': 'test_session',
'user_id': 'test_user',
'language': 'python',
'created_at': expired_time.isoformat(),
'last_active': expired_time.isoformat(),
'message_count': 0,
'is_active': True,
'session_metadata': {}
}
mock_redis_client.get.return_value = json.dumps(cached_data)
# Execute & Verify
with pytest.raises(SessionExpiredError):
session_manager.get_session("test_session")
class TestSessionActivity:
"""Test session activity management"""
def test_update_session_activity(self, session_manager, mock_chat_session):
"""Test updating session activity"""
# Setup
with patch.object(session_manager, 'get_session', return_value=mock_chat_session):
# Execute
session_manager.update_session_activity(mock_chat_session.id)
# Verify
mock_chat_session.update_activity.assert_called_once()
def test_increment_message_count(self, session_manager, mock_chat_session):
"""Test incrementing message count"""
# Setup
with patch.object(session_manager, 'get_session', return_value=mock_chat_session):
# Execute
session_manager.increment_message_count(mock_chat_session.id)
# Verify
mock_chat_session.increment_message_count.assert_called_once()
def test_set_session_language(self, session_manager, mock_chat_session):
"""Test setting session language"""
# Setup
with patch.object(session_manager, 'get_session', return_value=mock_chat_session):
# Execute
session_manager.set_session_language(mock_chat_session.id, 'javascript')
# Verify
mock_chat_session.set_language.assert_called_once_with('javascript')
class TestSessionCleanup:
"""Test session cleanup functionality"""
@patch('chat_agent.services.session_manager.ChatSession')
def test_cleanup_inactive_sessions(self, mock_chat_session_class, session_manager):
"""Test cleaning up inactive sessions"""
# Setup
mock_chat_session_class.cleanup_expired_sessions.return_value = 5
# Execute
result = session_manager.cleanup_inactive_sessions()
# Verify
assert result == 5
mock_chat_session_class.cleanup_expired_sessions.assert_called_once_with(3600)
@patch('chat_agent.services.session_manager.db')
def test_delete_session(self, mock_db, session_manager, mock_chat_session):
"""Test deleting a session"""
# Setup
mock_db.session.query.return_value.filter.return_value.first.return_value = mock_chat_session
# Execute
session_manager.delete_session(mock_chat_session.id)
# Verify
mock_db.session.delete.assert_called_once_with(mock_chat_session)
mock_db.session.commit.assert_called_once()
@patch('chat_agent.services.session_manager.db')
def test_delete_session_not_found(self, mock_db, session_manager):
"""Test deleting non-existent session"""
# Setup
mock_db.session.query.return_value.filter.return_value.first.return_value = None
# Execute & Verify
with pytest.raises(SessionNotFoundError):
session_manager.delete_session("non_existent_id")
class TestUserSessions:
"""Test user session management"""
@patch('chat_agent.services.session_manager.db')
def test_get_user_sessions(self, mock_db, session_manager, mock_chat_session):
"""Test getting all sessions for a user"""
# Setup
mock_db.session.query.return_value.filter.return_value.filter.return_value.order_by.return_value.all.return_value = [mock_chat_session]
# Execute
result = session_manager.get_user_sessions("test_user")
# Verify
assert result == [mock_chat_session]
@patch('chat_agent.services.session_manager.db')
def test_get_user_sessions_filters_expired(self, mock_db, session_manager):
"""Test that expired sessions are filtered out"""
# Setup
expired_session = Mock()
expired_session.is_expired.return_value = True
expired_session.deactivate = Mock()
active_session = Mock()
active_session.is_expired.return_value = False
mock_db.session.query.return_value.filter.return_value.filter.return_value.order_by.return_value.all.return_value = [
expired_session, active_session
]
# Execute
result = session_manager.get_user_sessions("test_user", active_only=True)
# Verify
assert result == [active_session]
expired_session.deactivate.assert_called_once()
class TestCacheOperations:
"""Test Redis cache operations"""
def test_cache_session(self, session_manager, mock_redis_client, mock_chat_session):
"""Test caching a session"""
# Execute
session_manager._cache_session(mock_chat_session)
# Verify
mock_redis_client.setex.assert_called_once()
args = mock_redis_client.setex.call_args
assert args[0][0] == f"session:{mock_chat_session.id}"
assert args[0][1] == 3900 # timeout + buffer
def test_get_cached_session_success(self, session_manager, mock_redis_client):
"""Test successfully getting cached session"""
# Setup
cached_data = {
'id': 'test_session',
'user_id': 'test_user',
'language': 'python',
'created_at': datetime.utcnow().isoformat(),
'last_active': datetime.utcnow().isoformat(),
'message_count': 0,
'is_active': True,
'session_metadata': {}
}
mock_redis_client.get.return_value = json.dumps(cached_data)
# Execute
result = session_manager._get_cached_session('test_session')
# Verify
assert result is not None
assert result.id == 'test_session'
assert result.user_id == 'test_user'
def test_get_cached_session_not_found(self, session_manager, mock_redis_client):
"""Test getting non-existent cached session"""
# Setup
mock_redis_client.get.return_value = None
# Execute
result = session_manager._get_cached_session('test_session')
# Verify
assert result is None
def test_get_cached_session_invalid_data(self, session_manager, mock_redis_client):
"""Test getting cached session with invalid JSON"""
# Setup
mock_redis_client.get.return_value = "invalid json"
# Execute
result = session_manager._get_cached_session('test_session')
# Verify
assert result is None
def test_remove_from_cache(self, session_manager, mock_redis_client):
"""Test removing session from cache"""
# Execute
session_manager._remove_from_cache('test_session')
# Verify
mock_redis_client.delete.assert_called_once_with('session:test_session')
def test_cleanup_expired_cache_sessions(self, session_manager, mock_redis_client):
"""Test cleaning up expired cache sessions"""
# Setup
expired_time = datetime.utcnow() - timedelta(hours=2)
valid_time = datetime.utcnow()
mock_redis_client.keys.return_value = ['session:expired', 'session:valid']
mock_redis_client.get.side_effect = [
json.dumps({
'id': 'expired',
'user_id': 'user1',
'language': 'python',
'created_at': expired_time.isoformat(),
'last_active': expired_time.isoformat(),
'message_count': 0,
'is_active': True,
'session_metadata': {}
}),
json.dumps({
'id': 'valid',
'user_id': 'user2',
'language': 'python',
'created_at': valid_time.isoformat(),
'last_active': valid_time.isoformat(),
'message_count': 0,
'is_active': True,
'session_metadata': {}
})
]
# Execute
session_manager._cleanup_expired_cache_sessions()
# Verify
mock_redis_client.delete.assert_called_once_with('session:expired')
class TestErrorHandling:
"""Test error handling scenarios"""
def test_redis_error_during_caching(self, session_manager, mock_redis_client, mock_chat_session):
"""Test handling Redis errors during caching"""
# Setup
import redis
mock_redis_client.setex.side_effect = redis.RedisError("Connection failed")
# Execute - should not raise exception
session_manager._cache_session(mock_chat_session)
# Verify - error is logged but doesn't propagate
assert True # Test passes if no exception is raised
@patch('chat_agent.services.session_manager.db')
def test_database_error_during_get_user_sessions(self, mock_db, session_manager):
"""Test handling database errors during user session retrieval"""
# Setup
from sqlalchemy.exc import SQLAlchemyError
mock_db.session.query.side_effect = SQLAlchemyError("DB Connection failed")
# Execute & Verify
with pytest.raises(SessionManagerError, match="Failed to get user sessions"):
session_manager.get_user_sessions("test_user")
class TestFactoryFunction:
"""Test factory function"""
def test_create_session_manager(self, mock_redis_client):
"""Test creating SessionManager with factory function"""
# Execute
manager = create_session_manager(mock_redis_client, session_timeout=7200)
# Verify
assert isinstance(manager, SessionManager)
assert manager.redis_client == mock_redis_client
assert manager.session_timeout == 7200
def test_create_session_manager_default_timeout(self, mock_redis_client):
"""Test creating SessionManager with default timeout"""
# Execute
manager = create_session_manager(mock_redis_client)
# Verify
assert manager.session_timeout == 3600 # Default timeout
class TestSessionExpiration:
"""Test session expiration logic"""
def test_is_session_expired_true(self, session_manager):
"""Test session expiration check returns True for expired session"""
# Setup
expired_session = Mock()
expired_session.is_expired.return_value = True
# Execute
result = session_manager._is_session_expired(expired_session)
# Verify
assert result is True
expired_session.is_expired.assert_called_once_with(3600)
def test_is_session_expired_false(self, session_manager):
"""Test session expiration check returns False for active session"""
# Setup
active_session = Mock()
active_session.is_expired.return_value = False
# Execute
result = session_manager._is_session_expired(active_session)
# Verify
assert result is False
active_session.is_expired.assert_called_once_with(3600)
@patch('chat_agent.services.session_manager.db')
def test_expire_session(self, mock_db, session_manager, mock_chat_session):
"""Test expiring a session"""
# Setup
mock_db.session.query.return_value.filter.return_value.first.return_value = mock_chat_session
# Execute
session_manager._expire_session(mock_chat_session.id)
# Verify
mock_chat_session.deactivate.assert_called_once() |