Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
19 changes: 19 additions & 0 deletions src/tenantq/search.py
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,23 @@
Mode = Literal["dense", "sparse", "hybrid"]


class UnscopedTenantError(ValueError):
"""Raised when a search is attempted without a real tenant scope."""


def require_tenant_id(tenant_id: str) -> str:
"""Reject empty/whitespace tenant ids so they never become MatchValue("").

An empty tenant looks like "no documents" to callers; the truth is the
request was never scoped. Call this at the filter boundary.
"""
if tenant_id is None or not str(tenant_id).strip():
raise UnscopedTenantError(
"request was not scoped: tenant_id is empty or whitespace-only"
)
return tenant_id


@dataclass
class SearchHit:
id: int
Expand All @@ -44,6 +61,7 @@ def build_filter(
created_before: Optional[int] = None,
) -> models.Filter:
"""Build a tenant-scoped filter, optionally narrowed by metadata."""
tenant_id = require_tenant_id(tenant_id)
must: List[models.FieldCondition] = [
models.FieldCondition(key=TENANT_FIELD, match=models.MatchValue(value=tenant_id))
]
Expand Down Expand Up @@ -91,6 +109,7 @@ def search(
prefetch_limit: int = 50,
) -> List[SearchHit]:
"""Run a tenant-isolated search in the requested retrieval mode."""
tenant_id = require_tenant_id(tenant_id)
qfilter = build_filter(tenant_id, category, created_after, created_before)
params = models.SearchParams(hnsw_ef=settings.hnsw.hnsw_ef)

Expand Down
29 changes: 29 additions & 0 deletions tests/test_search.py
Original file line number Diff line number Diff line change
Expand Up @@ -38,3 +38,32 @@ def test_created_at_range_filter(ingested, settings, embedder, dataset):
by_id = {d.id: d for d in dataset.documents}
for h in hits:
assert lo <= by_id[h.id].created_at <= hi


def test_build_filter_rejects_empty_tenant_id():
import pytest
from tenantq.search import UnscopedTenantError, build_filter

with pytest.raises(UnscopedTenantError, match="not scoped"):
build_filter("")
with pytest.raises(UnscopedTenantError, match="not scoped"):
build_filter(" ")


def test_build_filter_allows_tenant_id_with_internal_spaces():
from tenantq.search import build_filter

# Valid id that happens to contain spaces must not be stripped to empty
f = build_filter("acme corp")
assert f.must is not None
assert f.must[0].match.value == "acme corp"


def test_search_rejects_empty_tenant_id(ingested, settings, embedder):
import pytest
from tenantq.search import UnscopedTenantError, search

with pytest.raises(UnscopedTenantError, match="not scoped"):
search(ingested, settings, embedder, "query", tenant_id="")
with pytest.raises(UnscopedTenantError, match="not scoped"):
search(ingested, settings, embedder, "query", tenant_id=" \t")
Loading