test: Add test for verbose search

This commit is contained in:
Igor Ilic 2025-12-18 13:04:17 +01:00
parent f2bc7ca992
commit 31e491bc88

View file

@ -0,0 +1,100 @@
import types
from uuid import uuid4
import pytest
from cognee.modules.search.types import SearchType
def _make_user(user_id: str = "u1", tenant_id=None):
return types.SimpleNamespace(id=user_id, tenant_id=tenant_id)
def _make_dataset(*, name="ds", tenant_id="t1", dataset_id=None, owner_id=None):
return types.SimpleNamespace(
id=dataset_id or uuid4(),
name=name,
tenant_id=tenant_id,
owner_id=owner_id or uuid4(),
)
@pytest.fixture
def search_mod():
import importlib
return importlib.import_module("cognee.modules.search.methods.search")
@pytest.fixture(autouse=True)
def _patch_side_effect_boundaries(monkeypatch, search_mod):
"""
Keep production logic; patch only unavoidable side-effect boundaries.
"""
async def dummy_log_query(_query_text, _query_type, _user_id):
return types.SimpleNamespace(id="qid-1")
async def dummy_log_result(*_args, **_kwargs):
return None
async def dummy_prepare_search_result(search_result):
if isinstance(search_result, tuple) and len(search_result) == 3:
result, context, datasets = search_result
return {"result": result, "context": context, "graphs": {}, "datasets": datasets}
return {"result": None, "context": None, "graphs": {}, "datasets": []}
monkeypatch.setattr(search_mod, "send_telemetry", lambda *a, **k: None)
monkeypatch.setattr(search_mod, "log_query", dummy_log_query)
monkeypatch.setattr(search_mod, "log_result", dummy_log_result)
monkeypatch.setattr(search_mod, "prepare_search_result", dummy_prepare_search_result)
yield
@pytest.mark.asyncio
async def test_search_access_control_returns_dataset_shaped_dicts(monkeypatch, search_mod):
user = _make_user()
ds = _make_dataset(name="ds1", tenant_id="t1")
async def dummy_authorized_search(**kwargs):
assert kwargs["dataset_ids"] == [ds.id]
return [("r", ["ctx"], [ds])]
monkeypatch.setattr(search_mod, "backend_access_control_enabled", lambda: True)
monkeypatch.setattr(search_mod, "authorized_search", dummy_authorized_search)
out_non_verbose = await search_mod.search(
query_text="q",
query_type=SearchType.CHUNKS,
dataset_ids=[ds.id],
user=user,
verbose=False,
)
assert out_non_verbose == [
{
"search_result": ["r"],
"dataset_id": ds.id,
"dataset_name": "ds1",
"dataset_tenant_id": "t1",
}
]
out_verbose = await search_mod.search(
query_text="q",
query_type=SearchType.CHUNKS,
dataset_ids=[ds.id],
user=user,
verbose=True,
)
assert out_verbose == [
{
"search_result": ["r"],
"dataset_id": ds.id,
"dataset_name": "ds1",
"dataset_tenant_id": "t1",
"graphs": {},
}
]