Spaces:
Runtime error
Runtime error
| """ | |
| Shared pytest configuration and fixtures for all tests. | |
| """ | |
| import pytest | |
| import os | |
| import tempfile | |
| import shutil | |
| from unittest.mock import Mock, MagicMock, patch | |
| from datetime import datetime, timedelta | |
| # Test configuration | |
| def test_config(): | |
| """Test configuration settings.""" | |
| return { | |
| 'database_url': 'sqlite:///:memory:', | |
| 'redis_url': 'redis://localhost:6379/15', # Use test database | |
| 'groq_api_key': 'test-api-key', | |
| 'session_timeout': 300, # 5 minutes for tests | |
| 'rate_limit_enabled': False, | |
| 'log_level': 'DEBUG' | |
| } | |
| # Database fixtures | |
| def temp_database(): | |
| """Create temporary database for testing.""" | |
| temp_dir = tempfile.mkdtemp() | |
| db_path = os.path.join(temp_dir, 'test_chat.db') | |
| yield f'sqlite:///{db_path}' | |
| # Cleanup | |
| shutil.rmtree(temp_dir, ignore_errors=True) | |
| # Mock fixtures | |
| def mock_groq_client(): | |
| """Mock Groq client for testing.""" | |
| with patch('chat_agent.services.groq_client.GroqClient') as mock: | |
| mock_instance = MagicMock() | |
| # Default responses | |
| mock_instance.generate_response.return_value = "This is a test response from the LLM." | |
| mock_instance.stream_response.return_value = iter([ | |
| "This is ", "a test ", "streaming ", "response." | |
| ]) | |
| mock_instance.test_connection.return_value = True | |
| mock.return_value = mock_instance | |
| yield mock_instance | |
| def mock_redis(): | |
| """Mock Redis client for testing.""" | |
| with patch('redis.from_url') as mock_redis: | |
| mock_client = MagicMock() | |
| # Mock Redis operations | |
| mock_client.get.return_value = None | |
| mock_client.set.return_value = True | |
| mock_client.setex.return_value = True | |
| mock_client.delete.return_value = 1 | |
| mock_client.ping.return_value = True | |
| mock_redis.return_value = mock_client | |
| yield mock_client | |
| def mock_database(): | |
| """Mock database operations for testing.""" | |
| with patch('chat_agent.models.db') as mock_db: | |
| mock_session = MagicMock() | |
| mock_db.session = mock_session | |
| # Mock database operations | |
| mock_session.add.return_value = None | |
| mock_session.commit.return_value = None | |
| mock_session.rollback.return_value = None | |
| mock_session.query.return_value = mock_session | |
| mock_session.filter.return_value = mock_session | |
| mock_session.first.return_value = None | |
| mock_session.all.return_value = [] | |
| yield mock_session | |
| # Service fixtures | |
| def session_manager(mock_database, mock_redis): | |
| """Create session manager with mocked dependencies.""" | |
| from chat_agent.services.session_manager import SessionManager | |
| return SessionManager() | |
| def language_context_manager(): | |
| """Create language context manager.""" | |
| from chat_agent.services.language_context import LanguageContextManager | |
| return LanguageContextManager() | |
| def chat_history_manager(mock_database, mock_redis): | |
| """Create chat history manager with mocked dependencies.""" | |
| from chat_agent.services.chat_history import ChatHistoryManager | |
| return ChatHistoryManager() | |
| def chat_agent(mock_groq_client, session_manager, language_context_manager, chat_history_manager): | |
| """Create chat agent with all dependencies.""" | |
| from chat_agent.services.chat_agent import ChatAgent | |
| return ChatAgent( | |
| groq_client=mock_groq_client, | |
| session_manager=session_manager, | |
| language_context_manager=language_context_manager, | |
| chat_history_manager=chat_history_manager | |
| ) | |
| # Test data fixtures | |
| def sample_user_id(): | |
| """Sample user ID for testing.""" | |
| return "test-user-12345" | |
| def sample_session_data(): | |
| """Sample session data for testing.""" | |
| return { | |
| 'session_id': 'test-session-12345', | |
| 'user_id': 'test-user-12345', | |
| 'language': 'python', | |
| 'created_at': datetime.utcnow(), | |
| 'last_active': datetime.utcnow(), | |
| 'message_count': 0, | |
| 'is_active': True, | |
| 'metadata': {} | |
| } | |
| def sample_messages(): | |
| """Sample chat messages for testing.""" | |
| return [ | |
| { | |
| 'id': 'msg-1', | |
| 'session_id': 'test-session-12345', | |
| 'role': 'user', | |
| 'content': 'What is Python?', | |
| 'language': 'python', | |
| 'timestamp': datetime.utcnow() - timedelta(minutes=5), | |
| 'metadata': {} | |
| }, | |
| { | |
| 'id': 'msg-2', | |
| 'session_id': 'test-session-12345', | |
| 'role': 'assistant', | |
| 'content': 'Python is a high-level programming language.', | |
| 'language': 'python', | |
| 'timestamp': datetime.utcnow() - timedelta(minutes=4), | |
| 'metadata': {'tokens': 12} | |
| }, | |
| { | |
| 'id': 'msg-3', | |
| 'session_id': 'test-session-12345', | |
| 'role': 'user', | |
| 'content': 'How do I create a list?', | |
| 'language': 'python', | |
| 'timestamp': datetime.utcnow() - timedelta(minutes=2), | |
| 'metadata': {} | |
| } | |
| ] | |
| # Flask app fixtures | |
| def app(): | |
| """Create Flask app for testing.""" | |
| from flask import Flask | |
| app = Flask(__name__) | |
| app.config['TESTING'] = True | |
| app.config['SECRET_KEY'] = 'test-secret-key' | |
| app.config['WTF_CSRF_ENABLED'] = False | |
| return app | |
| def client(app): | |
| """Create test client.""" | |
| return app.test_client() | |
| def auth_headers(): | |
| """Authentication headers for API testing.""" | |
| return { | |
| 'X-User-ID': 'test-user-12345', | |
| 'Content-Type': 'application/json' | |
| } | |
| # Performance testing fixtures | |
| def performance_config(): | |
| """Configuration for performance tests.""" | |
| return { | |
| 'light_load_users': 10, | |
| 'medium_load_users': 25, | |
| 'heavy_load_users': 50, | |
| 'messages_per_user': 3, | |
| 'max_response_time': 2.0, | |
| 'min_success_rate': 0.8 | |
| } | |
| # Cleanup fixtures | |
| def cleanup_test_data(): | |
| """Automatically cleanup test data after each test.""" | |
| yield | |
| # Cleanup logic here if needed | |
| # For example, clear test caches, reset mocks, etc. | |
| pass | |
| # Markers for test categorization | |
| def pytest_configure(config): | |
| """Configure pytest with custom markers.""" | |
| config.addinivalue_line( | |
| "markers", "unit: Unit tests for individual components" | |
| ) | |
| config.addinivalue_line( | |
| "markers", "integration: Integration tests for component interactions" | |
| ) | |
| config.addinivalue_line( | |
| "markers", "e2e: End-to-end tests for complete workflows" | |
| ) | |
| config.addinivalue_line( | |
| "markers", "performance: Performance and load tests" | |
| ) | |
| config.addinivalue_line( | |
| "markers", "slow: Tests that take longer to run" | |
| ) | |
| # Test collection customization | |
| def pytest_collection_modifyitems(config, items): | |
| """Modify test collection to add markers automatically.""" | |
| for item in items: | |
| # Add markers based on test file location | |
| if "unit" in str(item.fspath): | |
| item.add_marker(pytest.mark.unit) | |
| elif "integration" in str(item.fspath): | |
| item.add_marker(pytest.mark.integration) | |
| elif "e2e" in str(item.fspath): | |
| item.add_marker(pytest.mark.e2e) | |
| elif "performance" in str(item.fspath): | |
| item.add_marker(pytest.mark.performance) | |
| item.add_marker(pytest.mark.slow) | |
| # Skip conditions | |
| def pytest_runtest_setup(item): | |
| """Setup conditions for running tests.""" | |
| # Skip performance tests in CI unless explicitly requested | |
| if "performance" in item.keywords and not item.config.getoption("--run-performance", default=False): | |
| pytest.skip("Performance tests skipped (use --run-performance to run)") | |
| def pytest_addoption(parser): | |
| """Add custom command line options.""" | |
| parser.addoption( | |
| "--run-performance", | |
| action="store_true", | |
| default=False, | |
| help="Run performance tests" | |
| ) | |
| parser.addoption( | |
| "--run-slow", | |
| action="store_true", | |
| default=False, | |
| help="Run slow tests" | |
| ) |