| """ |
| Unit tests for Long-Term User Memory, Reciprocal Rank Fusion (RRF), and Semantic Cache. |
| """ |
| from __future__ import annotations |
|
|
| import pytest |
| import numpy as np |
| from services.rrf_service import compute_rrf |
| from services.reranker import RankedChunk |
| from services.cache import cache_service |
|
|
|
|
| def test_rrf_fusion(): |
| c1 = RankedChunk("Text A", 0.9, "doc1", "a.txt", 1, "hr") |
| c2 = RankedChunk("Text B", 0.8, "doc2", "b.txt", 1, "hr") |
|
|
| dense = [c1, c2] |
| sparse = [c2, c1] |
|
|
| merged = compute_rrf(dense, sparse, k=60, top_n=2) |
| assert len(merged) == 2 |
| assert merged[0].score > 0.0 |
|
|
|
|
| def test_semantic_cache(): |
| v1 = [1.0, 0.0, 0.0] |
| v2 = [0.99, 0.01, 0.0] |
| v3 = [0.0, 1.0, 0.0] |
|
|
| res_data = {"response": "Cached answer"} |
| cache_service.set_semantic(v1, res_data) |
|
|
| |
| hit = cache_service.get_semantic(v2, similarity_threshold=0.90) |
| assert hit is not None |
| assert hit["response"] == "Cached answer" |
|
|
| |
| miss = cache_service.get_semantic(v3, similarity_threshold=0.90) |
| assert miss is None |
|
|