forked from innovacion/searchbox
Add testing
This commit is contained in:
288
tests/test_engine/test_factory.py
Normal file
288
tests/test_engine/test_factory.py
Normal file
@@ -0,0 +1,288 @@
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from vector_search_mcp.engine import Backend, get_engine
|
||||
from vector_search_mcp.engine.base_engine import BaseEngine
|
||||
from vector_search_mcp.engine.qdrant_engine import QdrantEngine
|
||||
|
||||
|
||||
class TestEngineFactory:
|
||||
"""Test suite for get_engine factory function"""
|
||||
|
||||
def test_engine_type_enum_values(self):
|
||||
"""Test that EngineType enum has expected values"""
|
||||
assert Backend.QDRANT == "qdrant"
|
||||
assert len(Backend) == 2 # QDRANT and COSMOS engine types
|
||||
|
||||
def test_get_engine_qdrant(self):
|
||||
"""Test get_engine returns QdrantEngine for QDRANT type"""
|
||||
with (
|
||||
patch(
|
||||
"vector_search_mcp.engine.qdrant_engine.Settings"
|
||||
) as mock_settings_class,
|
||||
patch(
|
||||
"vector_search_mcp.engine.qdrant_engine.AsyncQdrantClient"
|
||||
) as mock_client_class,
|
||||
):
|
||||
# Setup mocks
|
||||
mock_settings = MagicMock()
|
||||
mock_settings.url = "http://localhost:6333"
|
||||
mock_settings.api_key = "test_key"
|
||||
mock_settings_class.return_value = mock_settings
|
||||
|
||||
mock_client = MagicMock()
|
||||
mock_client_class.return_value = mock_client
|
||||
|
||||
# Test factory function
|
||||
engine = get_engine(Backend.QDRANT)
|
||||
|
||||
# Verify return type
|
||||
assert isinstance(engine, QdrantEngine)
|
||||
assert isinstance(engine, BaseEngine)
|
||||
|
||||
# Verify initialization was called correctly
|
||||
mock_settings_class.assert_called_once()
|
||||
mock_client_class.assert_called_once_with(
|
||||
url=mock_settings.url, api_key=mock_settings.api_key
|
||||
)
|
||||
|
||||
def test_get_engine_invalid_type(self):
|
||||
"""Test get_engine raises ValueError for unknown engine type"""
|
||||
# Create an invalid engine type (bypassing enum validation)
|
||||
invalid_type = "invalid_engine"
|
||||
|
||||
with pytest.raises(ValueError, match="Unknown engine type: invalid_engine"):
|
||||
# We need to cast to bypass type checking
|
||||
get_engine(invalid_type) # type: ignore
|
||||
|
||||
def test_get_engine_typing_literal_qdrant(self):
|
||||
"""Test that get_engine with literal QDRANT returns correct type"""
|
||||
with (
|
||||
patch("vector_search_mcp.engine.qdrant_engine.Settings"),
|
||||
patch("vector_search_mcp.engine.qdrant_engine.AsyncQdrantClient"),
|
||||
):
|
||||
# When using literal Backend.QDRANT, mypy should know it's QdrantEngine
|
||||
engine = get_engine(Backend.QDRANT)
|
||||
|
||||
# Runtime verification that it's the correct type
|
||||
assert type(engine).__name__ == "QdrantEngine"
|
||||
assert hasattr(engine, "client") # QdrantEngine specific attribute
|
||||
assert hasattr(engine, "settings") # QdrantEngine specific attribute
|
||||
|
||||
def test_get_engine_typing_variable(self):
|
||||
"""Test that get_engine with variable returns BaseEngine type"""
|
||||
with (
|
||||
patch("vector_search_mcp.engine.qdrant_engine.Settings"),
|
||||
patch("vector_search_mcp.engine.qdrant_engine.AsyncQdrantClient"),
|
||||
):
|
||||
# When using a variable, mypy should see it as BaseEngine
|
||||
engine_type: Backend = Backend.QDRANT
|
||||
engine = get_engine(engine_type)
|
||||
|
||||
# Runtime verification - it's still a QdrantEngine but typed as BaseEngine
|
||||
assert isinstance(engine, BaseEngine)
|
||||
assert isinstance(engine, QdrantEngine)
|
||||
|
||||
def test_get_engine_uses_cache(self):
|
||||
"""Test that get_engine uses cache and returns same instances"""
|
||||
with (
|
||||
patch(
|
||||
"vector_search_mcp.engine.qdrant_engine.Settings"
|
||||
) as mock_settings_class,
|
||||
patch(
|
||||
"vector_search_mcp.engine.qdrant_engine.AsyncQdrantClient"
|
||||
) as mock_client_class,
|
||||
):
|
||||
# Setup mocks
|
||||
mock_settings_class.return_value = MagicMock()
|
||||
mock_client_class.return_value = MagicMock()
|
||||
|
||||
# Create multiple engines
|
||||
engine1 = get_engine(Backend.QDRANT)
|
||||
engine2 = get_engine(Backend.QDRANT)
|
||||
|
||||
# Verify they are the same instance due to @cache decorator
|
||||
assert engine1 is engine2
|
||||
assert id(engine1) == id(engine2)
|
||||
|
||||
# But they are the same type
|
||||
assert type(engine1) is type(engine2)
|
||||
assert isinstance(engine1, QdrantEngine)
|
||||
assert isinstance(engine2, QdrantEngine)
|
||||
|
||||
# Verify initialization was called only once due to caching
|
||||
mock_settings_class.assert_called_once()
|
||||
mock_client_class.assert_called_once()
|
||||
|
||||
def test_engine_type_string_values(self):
|
||||
"""Test EngineType string representations"""
|
||||
assert str(Backend.QDRANT) == "qdrant"
|
||||
assert str(Backend.COSMOS) == "cosmos"
|
||||
|
||||
# Test that it can be used in string contexts
|
||||
engine_name = f"engine_{Backend.QDRANT}"
|
||||
assert engine_name == "engine_qdrant"
|
||||
|
||||
def test_engine_type_iteration(self):
|
||||
"""Test that EngineType can be iterated over"""
|
||||
engine_types = list(Backend)
|
||||
assert len(engine_types) == 2
|
||||
assert Backend.QDRANT in engine_types
|
||||
assert Backend.COSMOS in engine_types
|
||||
|
||||
def test_engine_factory_integration(self):
|
||||
"""Test complete factory integration with engine functionality"""
|
||||
with (
|
||||
patch(
|
||||
"vector_search_mcp.engine.qdrant_engine.Settings"
|
||||
) as mock_settings_class,
|
||||
patch(
|
||||
"vector_search_mcp.engine.qdrant_engine.AsyncQdrantClient"
|
||||
) as mock_client_class,
|
||||
):
|
||||
# Setup mocks
|
||||
mock_settings = MagicMock()
|
||||
mock_settings_class.return_value = mock_settings
|
||||
mock_client_class.return_value = MagicMock()
|
||||
|
||||
# Create engine through factory
|
||||
engine = get_engine(Backend.QDRANT)
|
||||
|
||||
# Verify engine has all required methods from BaseEngine
|
||||
assert hasattr(engine, "transform_conditions")
|
||||
assert hasattr(engine, "transform_response")
|
||||
assert hasattr(engine, "run_similarity_query")
|
||||
assert hasattr(engine, "semantic_search")
|
||||
|
||||
# Verify methods are callable
|
||||
assert callable(engine.transform_conditions)
|
||||
assert callable(engine.transform_response)
|
||||
assert callable(engine.run_similarity_query)
|
||||
assert callable(engine.semantic_search)
|
||||
|
||||
def test_future_engine_extensibility(self):
|
||||
"""Test structure supports future engine additions"""
|
||||
# Verify that EngineType is a StrEnum and can be extended
|
||||
assert issubclass(Backend, str)
|
||||
|
||||
# Verify the factory function structure can handle new engines
|
||||
# (This is more of a design verification)
|
||||
import inspect
|
||||
|
||||
sig = inspect.signature(get_engine)
|
||||
|
||||
# Should take Backend and return BaseEngine
|
||||
params = list(sig.parameters.values())
|
||||
assert len(params) == 1
|
||||
assert params[0].name == "backend"
|
||||
|
||||
|
||||
class TestEngineTypeEnum:
|
||||
"""Test suite specifically for Backend enum"""
|
||||
|
||||
def test_engine_type_is_str_enum(self):
|
||||
"""Test that Backend is a StrEnum"""
|
||||
from enum import StrEnum
|
||||
|
||||
assert issubclass(Backend, StrEnum)
|
||||
|
||||
# Should behave like strings
|
||||
assert Backend.QDRANT == "qdrant"
|
||||
assert f"{Backend.QDRANT}" == "qdrant"
|
||||
|
||||
def test_engine_type_comparison(self):
|
||||
"""Test EngineType comparison operations"""
|
||||
# Should equal string value
|
||||
assert Backend.QDRANT == "qdrant"
|
||||
|
||||
# Should not equal other strings
|
||||
assert Backend.QDRANT != "other"
|
||||
assert Backend.QDRANT != "QDRANT" # Case sensitive
|
||||
|
||||
def test_engine_type_in_collections(self):
|
||||
"""Test EngineType works in collections"""
|
||||
engine_list = [Backend.QDRANT]
|
||||
assert Backend.QDRANT in engine_list
|
||||
assert "qdrant" in engine_list # StrEnum benefit
|
||||
|
||||
engine_set = {Backend.QDRANT}
|
||||
assert Backend.QDRANT in engine_set
|
||||
|
||||
def test_engine_type_json_serializable(self):
|
||||
"""Test that Backend can be JSON serialized"""
|
||||
import json
|
||||
|
||||
data = {"engine": Backend.QDRANT}
|
||||
json_str = json.dumps(data, default=str)
|
||||
assert '"engine": "qdrant"' in json_str
|
||||
|
||||
def test_engine_type_immutable(self):
|
||||
"""Test that Backend values cannot be modified"""
|
||||
original_value = Backend.QDRANT
|
||||
|
||||
# Enum values should be immutable
|
||||
with pytest.raises(AttributeError):
|
||||
Backend.QDRANT = "modified" # type: ignore
|
||||
|
||||
# Original should be unchanged
|
||||
assert Backend.QDRANT == original_value
|
||||
|
||||
|
||||
class TestEngineFactoryErrorHandling:
|
||||
"""Test suite for error handling in engine factory"""
|
||||
|
||||
def test_none_backend_type(self):
|
||||
"""Test get_engine with None raises appropriate error"""
|
||||
with pytest.raises((TypeError, ValueError)):
|
||||
get_engine(None) # type: ignore
|
||||
|
||||
def test_empty_string_backend_type(self):
|
||||
"""Test get_engine with empty string"""
|
||||
with pytest.raises(ValueError, match="Unknown engine type"):
|
||||
get_engine("") # type: ignore
|
||||
|
||||
def test_numeric_backend_type(self):
|
||||
"""Test get_engine with numeric input"""
|
||||
with pytest.raises((TypeError, ValueError)):
|
||||
get_engine(123) # type: ignore
|
||||
|
||||
def test_boolean_backend_type(self):
|
||||
"""Test get_engine with boolean input"""
|
||||
with pytest.raises((TypeError, ValueError)):
|
||||
get_engine(True) # type: ignore
|
||||
|
||||
def test_get_engine_cosmos_not_implemented(self):
|
||||
"""Test that COSMOS engine raises NotImplementedError"""
|
||||
with pytest.raises(
|
||||
NotImplementedError, match="Cosmos engine is not implemented yet"
|
||||
):
|
||||
get_engine(Backend.COSMOS)
|
||||
|
||||
def test_engine_initialization_failure(self):
|
||||
"""Test handling of engine initialization failures"""
|
||||
with (
|
||||
patch("vector_search_mcp.engine.qdrant_engine.Settings") as mock_settings,
|
||||
patch("vector_search_mcp.engine.qdrant_engine.AsyncQdrantClient"),
|
||||
):
|
||||
# Make Settings initialization raise an exception
|
||||
mock_settings.side_effect = Exception("Settings initialization failed")
|
||||
|
||||
with pytest.raises(Exception, match="Settings initialization failed"):
|
||||
get_engine(Backend.QDRANT)
|
||||
|
||||
def test_case_sensitive_backend_type(self):
|
||||
"""Test that backend type matching is case sensitive"""
|
||||
with pytest.raises(ValueError, match="Unknown engine type"):
|
||||
get_engine("QDRANT") # type: ignore
|
||||
|
||||
with pytest.raises(ValueError, match="Unknown engine type"):
|
||||
get_engine("Qdrant") # type: ignore
|
||||
|
||||
def test_whitespace_backend_type(self):
|
||||
"""Test backend type with whitespace"""
|
||||
with pytest.raises(ValueError, match="Unknown engine type"):
|
||||
get_engine(" qdrant ") # type: ignore
|
||||
|
||||
with pytest.raises(ValueError, match="Unknown engine type"):
|
||||
get_engine("\tqdrant\n") # type: ignore
|
||||
Reference in New Issue
Block a user