""" 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()