Loading...
Loading...
Create factory fixture patterns for customizable test setup with variations. Use when building reusable test fixtures with multiple configurations, creating parameterizable mocks, or implementing test data builders. Works with pytest fixtures, mock objects, and test utilities. Enables DRY test setup while maintaining flexibility for edge cases.
npx skill4agent add dawiddutoit/custom-claude test-implement-factory-fixtures# Create a factory fixture
@pytest.fixture
def mock_service_factory():
def create_service(dimensions=384, success=True):
service = AsyncMock()
service.generate = AsyncMock(return_value=[0.1] * dimensions)
return service
return create_service
# Use in tests
def test_with_custom_dimensions(mock_service_factory):
service = mock_service_factory(dimensions=1536)
result = service.generate()
assert len(result) == 1536@pytest.fixture
def mock_thing_factory():
"""Create a factory for custom mock things.
Returns:
callable: Function that creates things with custom parameters
Example:
def test_something(mock_thing_factory):
thing = mock_thing_factory(param1=value1, param2=value2)
# Use thing in test
"""
def create_thing(param1=default1, param2=default2):
"""Create custom mock thing.
Args:
param1: Description (default: default1)
param2: Description (default: default2)
Returns:
Type: Custom thing instance
"""
# Implementation
return thing
return create_thing@pytest.fixture
def mock_embedding_service_factory():
def create_service(dimensions=384, success=True, error_message=None):
service = AsyncMock()
if success:
service.generate_embeddings = AsyncMock(
return_value=MagicMock(
success=True,
data=[[0.1] * dimensions],
)
)
else:
service.generate_embeddings = AsyncMock(
return_value=MagicMock(
success=False,
error=error_message or "Embedding service error"
)
)
return service
return create_service@pytest.fixture
def mock_settings_factory():
def create_settings(**kwargs):
settings = MagicMock()
# Set default structure
settings.project = MagicMock()
settings.neo4j = MagicMock()
settings.chunking = MagicMock()
# Apply defaults with kwargs overrides
settings.project.project_name = kwargs.get("project_name", "test_project")
settings.chunking.chunk_size = kwargs.get("chunk_size", 50)
settings.neo4j.database_name = kwargs.get("database_name", "test_db")
return settings
return create_settings@pytest.fixture
def mock_indexing_module_factory(
mock_neo4j_driver,
mock_embedding_service_factory,
mock_settings_factory
):
def create_module(**kwargs):
from project_watch_mcp.infrastructure.neo4j.graphrag.indexing import IndexingModule
settings = mock_settings_factory(**kwargs)
embedder = mock_embedding_service_factory(
dimensions=kwargs.get("embedding_dimensions", 384)
)
return IndexingModule(
settings=settings,
db=mock_neo4j_driver,
embedder=embedder,
project_name=kwargs.get("project_name", "test_project"),
database=kwargs.get("database", "test_db"),
)
return create_module@pytest.fixture
def mock_service_result():
def create_result(success=True, data=None, error=None):
result = MagicMock()
result.is_success = success
result.is_failure = not success
if success:
result.data = data
result.unwrap = MagicMock(return_value=data)
result.error = None
else:
result.data = None
result.error = error or "Operation failed"
result.unwrap = MagicMock(side_effect=ValueError(result.error))
return result
return create_resulttests/utils/
├── mock_services.py # Service-layer factories
├── mock_settings.py # Configuration factories
├── mock_drivers.py # Infrastructure factories
└── __init__.py # Exports (if needed)mock_{thing}_factorymock_real_{thing}_factorydef test_with_custom_dimensions(mock_embedding_service_factory):
service = mock_embedding_service_factory(dimensions=1536)
result = await service.generate_embeddings(["text"])
assert len(result.data[0]) == 1536def test_handles_service_error(mock_embedding_service_factory):
service = mock_embedding_service_factory(
success=False,
error_message="API rate limit"
)
result = await service.generate_embeddings(["text"])
assert result.is_failure
assert "rate limit" in result.errordef test_indexing_with_custom_settings(
mock_indexing_module_factory,
mock_settings_factory
):
module = mock_indexing_module_factory(
chunk_size=100,
project_name="custom_project",
embedding_dimensions=1536
)
# Test with custom configurationtests/utils/mock_services.py@pytest.fixture
def mock_embedding_service_factory():
"""Create a factory for custom mock embedding services.
Returns:
callable: Function that creates embedding services with custom dimensions
Example:
def test_something(mock_embedding_service_factory):
service = mock_embedding_service_factory(dimensions=1536)
# Use service in test
"""
def create_service(dimensions=384, success=True, error_message=None):
"""Create custom mock embedding service.
Args:
dimensions: Number of dimensions for embeddings
success: Whether the service should return success
error_message: Error message for failure cases
Returns:
AsyncMock: Custom embedding service
"""
service = AsyncMock()
if success:
service.generate_embeddings = AsyncMock(
return_value=MagicMock(
success=True,
data=[[0.1] * dimensions],
)
)
service.get_embeddings = AsyncMock(return_value=[[0.1] * dimensions])
service.get_embedding = AsyncMock(return_value=[0.1] * dimensions)
else:
service.generate_embeddings = AsyncMock(
return_value=MagicMock(
success=False, error=error_message or "Embedding service error"
)
)
service.get_embeddings = AsyncMock(side_effect=Exception(error_message or "Error"))
service.get_embedding = AsyncMock(side_effect=Exception(error_message or "Error"))
return service
return create_servicedef test_with_384_dimensions(mock_embedding_service_factory):
service = mock_embedding_service_factory(dimensions=384)
result = await service.generate_embeddings(["text"])
assert len(result.data[0]) == 384
def test_with_1536_dimensions(mock_embedding_service_factory):
service = mock_embedding_service_factory(dimensions=1536)
result = await service.generate_embeddings(["text"])
assert len(result.data[0]) == 1536
def test_service_failure(mock_embedding_service_factory):
service = mock_embedding_service_factory(
success=False,
error_message="Rate limit exceeded"
)
result = await service.generate_embeddings(["text"])
assert not result.success
assert "Rate limit" in result.errortests/utils/mock_settings.py@pytest.fixture
def mock_settings_factory():
"""Create a factory for custom mock settings.
Returns:
callable: Function that creates settings with custom attributes
Example:
def test_something(mock_settings_factory):
settings = mock_settings_factory(
project_name="custom_project",
chunk_size=100,
database_name="custom_db"
)
# Use settings in test
"""
def create_settings(**kwargs):
"""Create custom mock settings with specified attributes.
Args:
**kwargs: Attributes to set on the settings object.
Use dot notation for nested attributes.
Returns:
MagicMock: Custom settings object
"""
settings = MagicMock()
# Set default structure
settings.project = MagicMock()
settings.neo4j = MagicMock()
settings.chunking = MagicMock()
settings.indexing = MagicMock()
# Apply defaults
settings.project.project_name = kwargs.get("project_name", "test_project")
settings.project.repository_path = Path(kwargs.get("repository_path", "/test/repo"))
settings.neo4j.database_name = kwargs.get("database_name", "test_db")
settings.neo4j.uri = kwargs.get("neo4j_uri", "bolt://192.168.68.122:7687")
settings.chunking.chunk_size = kwargs.get("chunk_size", 50)
settings.chunking.chunk_overlap = kwargs.get("chunk_overlap", 5)
settings.indexing.batch_size = kwargs.get("batch_size", 100)
settings.indexing.max_workers = kwargs.get("max_workers", 4)
return settings
return create_settingsdef test_with_custom_chunk_size(mock_settings_factory):
settings = mock_settings_factory(chunk_size=100)
assert settings.chunking.chunk_size == 100
def test_with_multiple_overrides(mock_settings_factory):
settings = mock_settings_factory(
project_name="my_project",
chunk_size=200,
batch_size=50,
database_name="prod_db"
)
assert settings.project.project_name == "my_project"
assert settings.chunking.chunk_size == 200
assert settings.indexing.batch_size == 50
assert settings.neo4j.database_name == "prod_db"tests/utils/mock_services.py@pytest.fixture
def mock_indexing_module_factory(
mock_neo4j_driver,
mock_embedding_service_factory,
mock_settings_factory
):
"""Create a factory for real IndexingModule instances with custom settings.
Args:
mock_neo4j_driver: Mock Neo4j driver fixture
mock_embedding_service_factory: Mock embedding service factory
mock_settings_factory: Mock settings factory
Returns:
callable: Function that creates IndexingModule instances
Example:
def test_something(mock_indexing_module_factory):
module = mock_indexing_module_factory(
chunk_size=100,
project_name="custom_project"
)
# Use module in test
"""
def create_module(**kwargs):
"""Create IndexingModule with custom settings.
Args:
**kwargs: Settings to override
Returns:
IndexingModule: Real IndexingModule instance with mocks
"""
from project_watch_mcp.infrastructure.neo4j.graphrag.indexing import IndexingModule
settings = mock_settings_factory(**kwargs)
embedder = mock_embedding_service_factory(
dimensions=kwargs.get("embedding_dimensions", 384)
)
return IndexingModule(
settings=settings,
db=mock_neo4j_driver,
embedder=embedder,
project_name=kwargs.get("project_name", "test_project"),
database=kwargs.get("database", "test_db"),
)
return create_moduledef test_indexing_with_large_chunks(mock_indexing_module_factory):
module = mock_indexing_module_factory(chunk_size=500)
# Test real IndexingModule logic with large chunks
def test_indexing_with_custom_embeddings(mock_indexing_module_factory):
module = mock_indexing_module_factory(
embedding_dimensions=1536,
project_name="ml_project"
)
# Test with different embedding dimensions# Already included in project dependencies
uv pip install -e .# ❌ WRONG: Parametrize for complex factory-like needs
@pytest.mark.parametrize("dimensions,success,error", [
(384, True, None),
(1536, True, None),
(384, False, "Error"),
])
def test_embedding_service(dimensions, success, error):
# Complex mock setup repeated in test
service = create_complex_mock(dimensions, success, error)
# ...
# ✅ CORRECT: Factory fixture for complex setup
def test_embedding_service_384(mock_embedding_service_factory):
service = mock_embedding_service_factory(dimensions=384)
# ...
def test_embedding_service_failure(mock_embedding_service_factory):
service = mock_embedding_service_factory(success=False, error="Error")
# ...def create_service(
param1, param2, param3, param4, param5,
param6, param7, param8, param9, param10
):
# Too complex!def create_service(**kwargs):
# Flexible and readabledef create_service(dimensions, success, error_message):
# Forces every test to specify everythingdef create_service(dimensions=384, success=True, error_message=None):
# Common case works with no args@pytest.fixture
def mock_service_factory():
call_count = 0 # Shared across calls!
def create_service():
nonlocal call_count
call_count += 1
# State leaks between tests@pytest.fixture
def mock_service_factory():
def create_service(call_count=0): # Pass as parameter
service = AsyncMock()
service.call_count = call_count
return service
return create_servicedef create_service(dimensions=384):
# What type is returned?def create_service(dimensions: int = 384) -> AsyncMock:
"""Create custom mock embedding service.
Args:
dimensions: Number of dimensions for embeddings
Returns:
AsyncMock: Custom embedding service
"""/Users/dawiddutoit/projects/play/project-watch-mcp/tests/utils/mock_services.py/Users/dawiddutoit/projects/play/project-watch-mcp/tests/utils/mock_settings.py/Users/dawiddutoit/projects/play/project-watch-mcp/tests/utils/mock_drivers.pydebug-test-failuresrun-quality-gates