app/tests/test_isolation.py

"""Tenant isolation acceptance tests.

These are the hard-gating acceptance tests that must pass before any PR is merged:

- Ingest a document under tenant_A and another under tenant_B.
- Searching as tenant_A must NEVER return chunks that belong to tenant_B.
- Searching as tenant_B must NEVER return chunks that belong to tenant_A.

The vector store under test is a LangChain FAISS instance with an empty
on-disk path (from conftest).  Documents carry ``tenant_id`` and ``chunk_id`` metadata.
"""

from __future__ import annotations

from langchain_core.documents import Document


def _make_docs(tenant_id: str, doc_id: str, n: int = 3) -> list[Document]:
    """Build ``n`` LangChain Documents for ``tenant_id``."""
    return [
        Document(
            page_content=f"FIXTURE {tenant_id} document chunk number {i}.",
            metadata={
                "tenant_id": tenant_id,
                "doc_id": doc_id,
                "chunk_id": f"c-{tenant_id}-{i}",
            },
        )
        for i in range(n)
    ]


def _index_docs(vs: object, docs: list[Document]) -> None:
    """Add documents to the vector store using the LangChain API."""
    vs.add_documents(docs)


def test_tenant_a_results_only_contain_tenant_a_chunks(
    in_memory_vs: object,
) -> None:
    """tenant_A search must not return any chunk from tenant_B.

    This is a hard acceptance criterion — any failure here means tenant data
    isolation is broken and the PR must not be merged.
    """
    _index_docs(in_memory_vs, _make_docs("tenant_A", "doc-a"))
    _index_docs(in_memory_vs, _make_docs("tenant_B", "doc-b"))

    results = in_memory_vs.search("FIXTURE tenant_A document chunk", tenant_id="tenant_A", k=20)

    assert len(results) > 0, "tenant_A should have matching results"
    for r in results:
        assert r.tenant_id == "tenant_A", (
            f"Cross-tenant leak: got tenant_id='{r.tenant_id}' "
            f"(chunk_id='{r.chunk_id}') in tenant_A query"
        )


def test_tenant_b_results_only_contain_tenant_b_chunks(
    in_memory_vs: object,
) -> None:
    """tenant_B search must not return any chunk from tenant_A."""
    _index_docs(in_memory_vs, _make_docs("tenant_A", "doc-a"))
    _index_docs(in_memory_vs, _make_docs("tenant_B", "doc-b"))

    results = in_memory_vs.search("FIXTURE tenant_B document chunk", tenant_id="tenant_B", k=20)

    assert len(results) > 0, "tenant_B should have matching results"
    for r in results:
        assert r.tenant_id == "tenant_B", (
            f"Cross-tenant leak: got tenant_id='{r.tenant_id}' "
            f"(chunk_id='{r.chunk_id}') in tenant_B query"
        )


def test_unknown_tenant_returns_empty_results(
    in_memory_vs: object,
) -> None:
    """A tenant with no ingested data should receive an empty result set."""
    _index_docs(in_memory_vs, _make_docs("tenant_A", "doc-a"))

    results = in_memory_vs.search("some query", tenant_id="tenant_UNKNOWN", k=10)

    assert results == [], (
        f"Expected empty results for unknown tenant, got {len(results)} results"
    )


def test_count_is_per_tenant(
    in_memory_vs: object,
) -> None:
    """count() should return the correct per-tenant chunk count."""
    docs_a = _make_docs("tenant_A", "doc-a", n=3)
    docs_b = _make_docs("tenant_B", "doc-b", n=2)
    _index_docs(in_memory_vs, docs_a)
    _index_docs(in_memory_vs, docs_b)

    count_a = in_memory_vs.count("tenant_A")
    count_b = in_memory_vs.count("tenant_B")

    assert count_a == len(docs_a)
    assert count_b == len(docs_b)


def test_delete_document_removes_chunks(
    in_memory_vs: object,
) -> None:
    """delete_document should remove all chunks for that doc_id."""
    docs_a = _make_docs("tenant_A", "doc-a", n=3)
    _index_docs(in_memory_vs, docs_a)

    assert in_memory_vs.count("tenant_A") == len(docs_a)
    in_memory_vs.delete_document("doc-a")
    assert in_memory_vs.count("tenant_A") == 0