tests/test_vector_store.py

from __future__ import annotations

import pytest

from app.config import Settings
from app.services.vector_store import VectorStore


class FakeIndex:
    def __init__(self, name: str):
        self.name = name

    def describe_index_stats(self):
        return {"total_vector_count": 42}


class FakePineconeClient:
    def __init__(self, indexes):
        self._indexes = list(indexes)
        self.created = []
        self.index_requests = []

    def list_indexes(self):
        return list(self._indexes)

    def create_index(self, name, dimension, metric, spec):
        del metric, spec
        self.created.append((name, dimension))
        self._indexes.append({"name": name, "dimension": dimension})

    def Index(self, name):
        self.index_requests.append(name)
        return FakeIndex(name)


def test_ensure_index_fails_fast_when_dimension_mismatches():
    settings = Settings(
        hf_token="token",
        pinecone_api_key="pinecone",
        pinecone_index="gapura-rag",
        embedding_model="intfloat/multilingual-e5-large",
        embedding_dim=1024,
    )
    store = VectorStore(settings)
    store._client = FakePineconeClient([{"name": "gapura-rag", "dimension": 768}])

    with pytest.raises(ValueError, match="has dimension 768"):
        store.ensure_index()

    assert store._client.created == []


def test_ensure_index_keeps_configured_index_when_dimension_matches():
    settings = Settings(
        hf_token="token",
        pinecone_api_key="pinecone",
        pinecone_index="gapura-rag",
        embedding_dim=768,
    )
    store = VectorStore(settings)
    store._client = FakePineconeClient([{"name": "gapura-rag", "dimension": 768}])

    store.ensure_index()

    assert store.index_name == "gapura-rag"
    assert store.index_dimension == 768
    assert store._client.created == []
    assert store._client.index_requests[-1] == "gapura-rag"


def test_get_index_binding_reports_active_index_metadata():
    settings = Settings(
        hf_token="token",
        pinecone_api_key="pinecone",
        pinecone_index="gapura-rag",
        embedding_dim=768,
        pinecone_metric="cosine",
    )
    store = VectorStore(settings)
    store._client = FakePineconeClient([{"name": "gapura-rag", "dimension": 768}])

    store.ensure_index()

    assert store.get_index_binding() == {
        "configured_index": "gapura-rag",
        "active_index": "gapura-rag",
        "embedding_dim": 768,
        "index_dimension": 768,
        "metric": "cosine",
    }


def test_merge_overlapping_text_removes_chunk_boundary_overlap():
    merged = VectorStore._merge_overlapping_text(
        [
            "Petugas menyampaikan informasi kepada penumpang setiap perkembangan delay.",
            "setiap perkembangan delay. Petugas menyiapkan kompensasi sesuai ketentuan.",
        ]
    )

    assert merged == (
        "Petugas menyampaikan informasi kepada penumpang setiap perkembangan delay. "
        "Petugas menyiapkan kompensasi sesuai ketentuan."
    )