#!/usr/bin/env python3 """ Unit tests for session management system """ import unittest import tempfile import shutil import time import os from session_manager import SessionManager, UserSession, SessionResource from flask import Flask, g, session class TestSessionManager(unittest.TestCase): def setUp(self): """Set up test fixtures""" self.temp_dir = tempfile.mkdtemp() self.config = { 'max_session_duration': 3600, 'max_idle_time': 900, 'max_resources_per_session': 5, # Small limit for testing 'max_bytes_per_session': 1024 * 1024, # 1MB for testing 'cleanup_interval': 1, # 1 second for faster testing 'session_storage_path': self.temp_dir } self.manager = SessionManager(self.config) def tearDown(self): """Clean up test fixtures""" shutil.rmtree(self.temp_dir, ignore_errors=True) def test_create_session(self): """Test session creation""" session = self.manager.create_session( session_id='test-123', user_id='user-1', ip_address='127.0.0.1', user_agent='Test Agent' ) self.assertEqual(session.session_id, 'test-123') self.assertEqual(session.user_id, 'user-1') self.assertEqual(session.ip_address, '127.0.0.1') self.assertEqual(session.user_agent, 'Test Agent') self.assertEqual(len(session.resources), 0) def test_get_session(self): """Test session retrieval""" self.manager.create_session(session_id='test-456') session = self.manager.get_session('test-456') self.assertIsNotNone(session) self.assertEqual(session.session_id, 'test-456') # Non-existent session session = self.manager.get_session('non-existent') self.assertIsNone(session) def test_add_resource(self): """Test adding resources to session""" self.manager.create_session(session_id='test-789') # Add a resource resource = self.manager.add_resource( session_id='test-789', resource_type='audio_file', resource_id='audio-1', path='/tmp/test.wav', size_bytes=1024, metadata={'format': 'wav'} ) self.assertIsNotNone(resource) self.assertEqual(resource.resource_id, 'audio-1') self.assertEqual(resource.resource_type, 'audio_file') self.assertEqual(resource.size_bytes, 1024) # Check session updated session = self.manager.get_session('test-789') self.assertEqual(len(session.resources), 1) self.assertEqual(session.total_bytes_used, 1024) def test_resource_limits(self): """Test resource limit enforcement""" self.manager.create_session(session_id='test-limits') # Add resources up to limit for i in range(5): self.manager.add_resource( session_id='test-limits', resource_type='temp_file', resource_id=f'file-{i}', size_bytes=100 ) session = self.manager.get_session('test-limits') self.assertEqual(len(session.resources), 5) # Add one more - should remove oldest self.manager.add_resource( session_id='test-limits', resource_type='temp_file', resource_id='file-new', size_bytes=100 ) session = self.manager.get_session('test-limits') self.assertEqual(len(session.resources), 5) # Still 5 self.assertNotIn('file-0', session.resources) # Oldest removed self.assertIn('file-new', session.resources) # New one added def test_size_limits(self): """Test size limit enforcement""" self.manager.create_session(session_id='test-size') # Add a large resource self.manager.add_resource( session_id='test-size', resource_type='audio_file', resource_id='large-1', size_bytes=500 * 1024 # 500KB ) # Add another large resource self.manager.add_resource( session_id='test-size', resource_type='audio_file', resource_id='large-2', size_bytes=600 * 1024 # 600KB - would exceed 1MB limit ) session = self.manager.get_session('test-size') # First resource should be removed to make space self.assertNotIn('large-1', session.resources) self.assertIn('large-2', session.resources) self.assertLessEqual(session.total_bytes_used, 1024 * 1024) def test_remove_resource(self): """Test resource removal""" self.manager.create_session(session_id='test-remove') self.manager.add_resource( session_id='test-remove', resource_type='temp_file', resource_id='to-remove', size_bytes=1000 ) # Remove resource success = self.manager.remove_resource('test-remove', 'to-remove') self.assertTrue(success) # Check it's gone session = self.manager.get_session('test-remove') self.assertEqual(len(session.resources), 0) self.assertEqual(session.total_bytes_used, 0) def test_cleanup_session(self): """Test session cleanup""" # Create session with resources self.manager.create_session(session_id='test-cleanup') # Create actual temp file temp_file = os.path.join(self.temp_dir, 'test-file.txt') with open(temp_file, 'w') as f: f.write('test content') self.manager.add_resource( session_id='test-cleanup', resource_type='temp_file', path=temp_file, size_bytes=12 ) # Cleanup session success = self.manager.cleanup_session('test-cleanup') self.assertTrue(success) # Check session is gone session = self.manager.get_session('test-cleanup') self.assertIsNone(session) # Check file is deleted self.assertFalse(os.path.exists(temp_file)) def test_session_info(self): """Test session info retrieval""" self.manager.create_session( session_id='test-info', ip_address='192.168.1.1' ) self.manager.add_resource( session_id='test-info', resource_type='audio_file', size_bytes=2048 ) info = self.manager.get_session_info('test-info') self.assertIsNotNone(info) self.assertEqual(info['session_id'], 'test-info') self.assertEqual(info['ip_address'], '192.168.1.1') self.assertEqual(info['resource_count'], 1) self.assertEqual(info['total_bytes_used'], 2048) def test_stats(self): """Test statistics calculation""" # Create multiple sessions for i in range(3): self.manager.create_session(session_id=f'test-stats-{i}') self.manager.add_resource( session_id=f'test-stats-{i}', resource_type='temp_file', size_bytes=1000 ) stats = self.manager.get_stats() self.assertEqual(stats['active_sessions'], 3) self.assertEqual(stats['active_resources'], 3) self.assertEqual(stats['active_bytes'], 3000) self.assertEqual(stats['total_sessions_created'], 3) def test_metrics_export(self): """Test metrics export""" self.manager.create_session(session_id='test-metrics') metrics = self.manager.export_metrics() self.assertIn('sessions', metrics) self.assertIn('resources', metrics) self.assertIn('limits', metrics) self.assertEqual(metrics['sessions']['active'], 1) class TestFlaskIntegration(unittest.TestCase): def setUp(self): """Set up Flask app for testing""" self.app = Flask(__name__) self.app.config['TESTING'] = True self.app.config['SECRET_KEY'] = 'test-secret' self.temp_dir = tempfile.mkdtemp() self.app.config['UPLOAD_FOLDER'] = self.temp_dir # Initialize session manager from session_manager import init_app init_app(self.app) self.client = self.app.test_client() self.ctx = self.app.test_request_context() self.ctx.push() def tearDown(self): """Clean up""" self.ctx.pop() shutil.rmtree(self.temp_dir, ignore_errors=True) def test_before_request_handler(self): """Test Flask before_request integration""" with self.client: # Make a request response = self.client.get('/') # Session should be created with self.client.session_transaction() as sess: self.assertIn('session_id', sess) if __name__ == '__main__': unittest.main()