tests/test_search_cache.py

import pytest

from src.config import COLLECTION_NAME, VECTOR_DB_PERSIST_PATH
from src.embedding.embedding_service import EmbeddingService
from src.search.search_service import SearchService
from src.vector_store.vector_db import VectorDatabase


@pytest.mark.integration
def test_search_cache_basic():
    vdb = VectorDatabase(VECTOR_DB_PERSIST_PATH, COLLECTION_NAME)
    emb = EmbeddingService()
    svc = SearchService(vdb, emb)
    q = "remote work policy"
    r1 = svc.search(q, top_k=3)
    stats_after_first = svc.get_cache_stats()
    r2 = svc.search(q, top_k=3)
    stats_after_second = svc.get_cache_stats()
    assert r1 == r2
    # After first search: one miss
    assert stats_after_first["misses"] >= 1
    # After second search: at least one hit
    assert stats_after_second["hits"] >= 1


@pytest.mark.integration
def test_search_cache_eviction():
    vdb = VectorDatabase(VECTOR_DB_PERSIST_PATH, COLLECTION_NAME)
    emb = EmbeddingService()
    svc = SearchService(vdb, emb)
    for i in range(60):
        svc.search(f"query {i}", top_k=1)
    stats = svc.get_cache_stats()
    assert stats["size"] <= stats["capacity"]