| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234 |
- """
- Global pytest configuration and fixtures for RAG System tests.
- This module provides shared test fixtures and configuration that are available
- to all test modules across the test suite.
- """
- import asyncio
- import pytest
- from typing import AsyncGenerator, Generator
- from pathlib import Path
- # ============================================================================
- # Event Loop Configuration
- # ============================================================================
- @pytest.fixture(scope="session")
- def event_loop() -> Generator[asyncio.AbstractEventLoop, None, None]:
- """
- Create an event loop for the entire test session.
-
- This fixture ensures that async tests have access to a properly configured
- event loop. The loop is created once per test session and cleaned up after
- all tests complete.
-
- Yields:
- asyncio.AbstractEventLoop: The event loop for async tests
- """
- loop = asyncio.get_event_loop_policy().new_event_loop()
- yield loop
- loop.close()
- # ============================================================================
- # Singleton Reset
- # ============================================================================
- @pytest.fixture(autouse=True)
- def reset_singletons() -> Generator[None, None, None]:
- """
- Reset singleton instances between tests.
-
- This fixture automatically runs before each test to ensure that singleton
- instances (like configuration objects) are reset, preventing test pollution.
-
- Yields:
- None
- """
- yield
- # Clear any cached configuration instances
- from functools import lru_cache
- # Note: Specific singleton reset logic will be added as needed
- # For example: get_settings.cache_clear() when settings are implemented
- # ============================================================================
- # Test Data Paths
- # ============================================================================
- @pytest.fixture(scope="session")
- def test_data_dir() -> Path:
- """
- Get the path to the test data directory.
-
- Returns:
- Path: Path to tests/fixtures directory
- """
- return Path(__file__).parent / "fixtures"
- @pytest.fixture(scope="session")
- def sample_documents_dir(test_data_dir: Path) -> Path:
- """
- Get the path to sample documents for testing.
-
- Args:
- test_data_dir: Path to test data directory
-
- Returns:
- Path: Path to sample documents directory
- """
- return test_data_dir / "documents"
- # ============================================================================
- # Mock Configuration
- # ============================================================================
- @pytest.fixture
- def mock_settings() -> dict:
- """
- Provide mock settings for testing.
-
- Returns:
- dict: Mock configuration settings
- """
- return {
- "app_name": "RAG System Test",
- "debug": True,
- "vector_db_type": "infinity",
- "database": {
- "host": "localhost",
- "port": 5432,
- "database": "test_rag_system",
- "username": "test_user",
- "password": "test_password",
- },
- "infinity": {
- "host": "localhost",
- "port": 23817,
- },
- "elasticsearch": {
- "host": "localhost",
- "port": 9200,
- },
- }
- # ============================================================================
- # Test Database Fixtures
- # ============================================================================
- @pytest.fixture
- async def test_db_session() -> AsyncGenerator:
- """
- Provide a test database session.
-
- This fixture will be implemented when the database infrastructure is ready.
- It should provide an isolated database session for each test.
-
- Yields:
- Database session for testing
- """
- # TODO: Implement when database infrastructure is ready
- # Example:
- # async with get_test_session_factory()() as session:
- # yield session
- # await session.rollback()
- yield None
- # ============================================================================
- # Mock External Services
- # ============================================================================
- @pytest.fixture
- def mock_vector_db():
- """
- Provide a mock vector database for testing.
-
- This fixture will be implemented when vector database infrastructure is ready.
-
- Returns:
- Mock vector database instance
- """
- # TODO: Implement when vector database infrastructure is ready
- return None
- @pytest.fixture
- def mock_embedding_service():
- """
- Provide a mock embedding service for testing.
-
- This fixture generates fake embeddings for testing purposes.
-
- Returns:
- Mock embedding service instance
- """
- # TODO: Implement when embedding service is ready
- return None
- # ============================================================================
- # Test Markers Configuration
- # ============================================================================
- def pytest_configure(config):
- """
- Configure custom pytest markers.
-
- Args:
- config: pytest configuration object
- """
- config.addinivalue_line(
- "markers", "unit: Unit tests that test individual components in isolation"
- )
- config.addinivalue_line(
- "markers", "integration: Integration tests that test component interactions"
- )
- config.addinivalue_line(
- "markers", "e2e: End-to-end tests that test complete user workflows"
- )
- config.addinivalue_line(
- "markers", "slow: Tests that take a long time to run (> 1 second)"
- )
- config.addinivalue_line(
- "markers", "requires_db: Tests that require a database connection"
- )
- config.addinivalue_line(
- "markers", "requires_vector_db: Tests that require a vector database connection"
- )
- config.addinivalue_line(
- "markers", "requires_external_service: Tests that require external services"
- )
- config.addinivalue_line(
- "markers", "property: Property-based tests using Hypothesis"
- )
- config.addinivalue_line(
- "markers", "asyncio: Async tests using pytest-asyncio"
- )
- # ============================================================================
- # Test Collection Configuration
- # ============================================================================
- def pytest_collection_modifyitems(config, items):
- """
- Modify test collection to add markers based on test location.
-
- Args:
- config: pytest configuration object
- items: List of collected test items
- """
- 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)
|