File size: 5,859 Bytes
9639483 0cdf561 9639483 0cdf561 9639483 0cdf561 9639483 0cdf561 9639483 0cdf561 7baf8ba 0cdf561 9639483 7baf8ba 9639483 0cdf561 |
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 |
from unittest.mock import MagicMock, patch
from src.utils.service_loader import (
get_embedding_service_if_available,
)
class TestGetEmbeddingServiceIfAvailable:
"""Test get_embedding_service_if_available() safety wrapper."""
def test_returns_service_when_available(self):
"""Test successful loading of embedding service (free tier fallback)."""
mock_service = MagicMock()
# Patch settings to disable premium tier, then patch the local service
with patch("src.utils.service_loader.settings") as mock_settings:
mock_settings.has_openai_key = False
with patch("src.services.embeddings.get_embedding_service", return_value=mock_service):
service = get_embedding_service_if_available()
assert service is mock_service
def test_returns_none_when_no_service_available(self):
"""Test handling of ImportError when loading embedding service."""
# Disable premium tier, then make local service fail
with patch("src.utils.service_loader.settings") as mock_settings:
mock_settings.has_openai_key = False
with patch(
"src.services.embeddings.get_embedding_service",
side_effect=ImportError("Missing deps"),
):
service = get_embedding_service_if_available()
assert service is None
def test_get_embedding_service_generic_error():
"""Test handling of generic Exception when loading embedding service."""
# Disable premium tier, then make local service fail
with patch("src.utils.service_loader.settings") as mock_settings:
mock_settings.has_openai_key = False
with patch(
"src.services.embeddings.get_embedding_service",
side_effect=ValueError("Boom"),
):
service = get_embedding_service_if_available()
assert service is None
class TestGetEmbeddingService:
"""Test get_embedding_service() logic."""
def test_uses_llamaindex_when_openai_key_present(self):
"""OpenAI key (env) → LlamaIndex."""
with patch("src.utils.service_loader.settings") as mock_settings:
mock_settings.has_openai_key = True
mock_settings.openai_api_key = "sk-env"
# Mock LlamaIndex dependencies and factory
with patch.dict(
"sys.modules",
{
"src.services.llamaindex_rag": MagicMock(),
"chromadb": MagicMock(),
"llama_index": MagicMock(),
},
):
mock_rag_service = MagicMock()
with patch(
"src.services.llamaindex_rag.get_rag_service", return_value=mock_rag_service
):
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):
"""BYOK key → LlamaIndex."""
with patch("src.utils.service_loader.settings") as mock_settings:
mock_settings.has_openai_key = False
with patch.dict(
"sys.modules",
{
"src.services.llamaindex_rag": MagicMock(),
},
):
mock_rag_service = MagicMock()
with patch(
"src.services.llamaindex_rag.get_rag_service", return_value=mock_rag_service
):
from src.utils.service_loader import get_embedding_service
service = get_embedding_service(api_key="sk-test")
assert service is mock_rag_service
def test_falls_back_to_local_when_no_openai_key(self):
"""No OpenAI key → Local embeddings."""
with patch("src.utils.service_loader.settings") as mock_settings:
mock_settings.has_openai_key = False
mock_local_service = MagicMock()
with patch(
"src.services.embeddings.get_embedding_service", return_value=mock_local_service
):
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):
"""LlamaIndex fails import → Local embeddings."""
with patch("src.utils.service_loader.settings") as mock_settings:
mock_settings.has_openai_key = True
# Mock ImportError for LlamaIndex
with patch(
"src.services.llamaindex_rag.get_rag_service", side_effect=ImportError("No deps")
):
mock_local_service = MagicMock()
with patch(
"src.services.embeddings.get_embedding_service", return_value=mock_local_service
):
from src.utils.service_loader import get_embedding_service
service = get_embedding_service()
assert service is mock_local_service
def test_raises_when_no_embedding_service_available(self):
"""All services fail → ImportError."""
with patch("src.utils.service_loader.settings") as mock_settings:
mock_settings.has_openai_key = False
with patch(
"src.services.embeddings.get_embedding_service", side_effect=ImportError("No deps")
):
import pytest
from src.utils.service_loader import get_embedding_service
with pytest.raises(ImportError) as exc:
get_embedding_service()
assert "No embedding service available" in str(exc.value)
|