Spaces:
Sleeping
Sleeping
File size: 3,921 Bytes
dc1b199 32c4506 dc1b199 d611f45 dc1b199 d611f45 dc1b199 d611f45 dc1b199 d611f45 dc1b199 d611f45 dc1b199 d611f45 dc1b199 d611f45 dc1b199 d611f45 dc1b199 d611f45 dc1b199 d611f45 dc1b199 d611f45 dc1b199 d611f45 dc1b199 d611f45 dc1b199 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 | """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
|