52 lines
1.8 KiB
Python
52 lines
1.8 KiB
Python
|
from __future__ import annotations
|
||
|
|
|
||
|
|
from types import SimpleNamespace
|
||
|
|
from unittest.mock import Mock, patch
|
||
|
|
|
||
|
|
from django.test import SimpleTestCase
|
||
|
|
|
||
|
|
from rag.embedding import embed_texts
|
||
|
|
from rag.ingest import ingest
|
||
|
|
from rag.observability import METRICS
|
||
|
|
from rag.retrieve import search_with_query
|
||
|
|
|
||
|
|
|
||
|
|
class RAGObservabilityTests(SimpleTestCase):
|
||
|
|
def tearDown(self):
|
||
|
|
METRICS.clear()
|
||
|
|
|
||
|
|
def test_embed_texts_records_empty_input_metric(self):
|
||
|
|
result = embed_texts([])
|
||
|
|
|
||
|
|
self.assertEqual(result, [])
|
||
|
|
self.assertEqual(METRICS["rag.embedding.empty_input|operation=embed_texts"], 1)
|
||
|
|
|
||
|
|
@patch("rag.retrieve.QdrantVectorStore")
|
||
|
|
@patch("rag.retrieve.embed_single", return_value=[0.1, 0.2])
|
||
|
|
@patch("rag.retrieve.load_rag_config")
|
||
|
|
def test_search_with_query_records_empty_result_metric(self, mock_load_config, _mock_embed, mock_store_cls):
|
||
|
|
mock_load_config.return_value = SimpleNamespace(
|
||
|
|
embedding=SimpleNamespace(provider="gapgpt"),
|
||
|
|
)
|
||
|
|
mock_store = Mock()
|
||
|
|
mock_store.search.return_value = []
|
||
|
|
mock_store_cls.return_value = mock_store
|
||
|
|
|
||
|
|
result = search_with_query("query")
|
||
|
|
|
||
|
|
self.assertEqual(result, [])
|
||
|
|
self.assertEqual(METRICS["rag.retrieve.empty_result|operation=search_with_query,service_id=None"], 1)
|
||
|
|
|
||
|
|
@patch("rag.ingest.load_sources", return_value=[])
|
||
|
|
@patch("rag.ingest.QdrantVectorStore")
|
||
|
|
@patch("rag.ingest.load_rag_config")
|
||
|
|
def test_ingest_records_empty_sources_metric(self, mock_load_config, _mock_store_cls, _mock_sources):
|
||
|
|
mock_load_config.return_value = SimpleNamespace(
|
||
|
|
embedding=SimpleNamespace(provider="gapgpt"),
|
||
|
|
)
|
||
|
|
|
||
|
|
result = ingest()
|
||
|
|
|
||
|
|
self.assertEqual(result["chunks_added"], 0)
|
||
|
|
self.assertEqual(METRICS["rag.ingest.empty_sources|kb_name=None"], 1)
|