File size: 6,446 Bytes
7baf8ba 915b009 7baf8ba 915b009 7baf8ba e70a3b7 7baf8ba e70a3b7 7baf8ba |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 |
"""Tests for service loader embedding service selection.
TDD: These tests define the expected behavior of get_embedding_service().
"""
from unittest.mock import MagicMock, patch
import pytest
class TestGetEmbeddingService:
"""Tests for get_embedding_service() tiered selection."""
def test_uses_llamaindex_when_openai_key_present(self):
"""Should return LlamaIndexRAGService when OPENAI_API_KEY is set."""
mock_rag_service = MagicMock()
# Patch at the point of use (inside service_loader)
with patch("src.utils.service_loader.settings") as mock_settings:
mock_settings.has_openai_key = True
with patch(
"src.utils.service_loader.get_rag_service",
return_value=mock_rag_service,
create=True,
):
# Also need to prevent the actual import from failing
# Update lambda to accept **kwargs (api_key)
mock_module = MagicMock(get_rag_service=lambda **kwargs: mock_rag_service)
with patch.dict("sys.modules", {"src.services.llamaindex_rag": mock_module}):
from src.utils.service_loader import get_embedding_service
service = get_embedding_service()
assert service is mock_rag_service
def test_uses_llamaindex_when_byok_key_present(self):
"""Should return LlamaIndexRAGService when valid BYOK key passed."""
mock_rag_service = MagicMock()
with patch("src.utils.service_loader.settings") as mock_settings:
mock_settings.has_openai_key = False # Env key missing
# Update lambda to accept **kwargs
mock_module = MagicMock(get_rag_service=lambda **kwargs: mock_rag_service)
with patch.dict("sys.modules", {"src.services.llamaindex_rag": mock_module}):
from src.utils.service_loader import get_embedding_service
# Pass valid BYOK key
service = get_embedding_service(api_key="sk-test-key")
assert service is mock_rag_service
def test_falls_back_to_local_when_no_openai_key(self):
"""Should return EmbeddingService when no OpenAI key."""
mock_local_service = MagicMock()
with patch("src.utils.service_loader.settings") as mock_settings:
mock_settings.has_openai_key = False
# Patch the embeddings module
mock_embed_mod = MagicMock(get_embedding_service=lambda: mock_local_service)
with patch.dict("sys.modules", {"src.services.embeddings": mock_embed_mod}):
from src.utils.service_loader import get_embedding_service
service = get_embedding_service()
assert service is mock_local_service
def test_falls_back_when_llamaindex_import_fails(self):
"""Should fallback to local if LlamaIndex deps missing."""
mock_local_service = MagicMock()
with patch("src.utils.service_loader.settings") as mock_settings:
mock_settings.has_openai_key = True
# LlamaIndex import fails
def raise_import_error(*args, **kwargs):
raise ImportError("llama_index not installed")
# Make llamaindex_rag module raise ImportError on import
import sys
original_modules = dict(sys.modules)
# Remove llamaindex_rag if it exists
if "src.services.llamaindex_rag" in sys.modules:
del sys.modules["src.services.llamaindex_rag"]
try:
# Patch to raise ImportError
mock_embed_module = MagicMock(get_embedding_service=lambda: mock_local_service)
with patch.dict(
"sys.modules",
{
"src.services.llamaindex_rag": None, # None causes ImportError
"src.services.embeddings": mock_embed_module,
},
):
from src.utils.service_loader import get_embedding_service
service = get_embedding_service()
assert service is mock_local_service
finally:
# Restore original modules
sys.modules.update(original_modules)
def test_raises_when_no_embedding_service_available(self):
"""Should raise ImportError when no embedding service can be loaded."""
with patch("src.utils.service_loader.settings") as mock_settings:
mock_settings.has_openai_key = False
# Make embeddings module raise ImportError
with patch.dict(
"sys.modules",
{"src.services.embeddings": None}, # None causes ImportError
):
from src.utils.service_loader import get_embedding_service
with pytest.raises(ImportError) as exc_info:
get_embedding_service()
assert "No embedding service available" in str(exc_info.value)
class TestGetEmbeddingServiceIfAvailable:
"""Tests for get_embedding_service_if_available() safe wrapper."""
def test_returns_none_when_no_service_available(self):
"""Should return None instead of raising when no service available."""
with patch("src.utils.service_loader.settings") as mock_settings:
mock_settings.has_openai_key = False
# Make embeddings module raise ImportError
with patch.dict(
"sys.modules",
{"src.services.embeddings": None},
):
from src.utils.service_loader import get_embedding_service_if_available
result = get_embedding_service_if_available()
assert result is None
def test_returns_service_when_available(self):
"""Should return the service when available."""
mock_service = MagicMock()
with patch("src.utils.service_loader.settings") as mock_settings:
mock_settings.has_openai_key = False
with patch.dict(
"sys.modules",
{"src.services.embeddings": MagicMock(get_embedding_service=lambda: mock_service)},
):
from src.utils.service_loader import get_embedding_service_if_available
result = get_embedding_service_if_available()
assert result is mock_service
|