from __future__ import annotations from types import SimpleNamespace from unittest.mock import AsyncMock import pytest import app.api.v1.search as search_api from app.api.deps import UserContext from app.api.schemas.search import SearchRequest from app.core.permission import PermissionContext def _fake_request(search_result: dict) -> tuple[SimpleNamespace, SimpleNamespace]: raw = SimpleNamespace(search=AsyncMock(return_value=search_result)) es_client = SimpleNamespace(raw=raw) request = SimpleNamespace(app=SimpleNamespace(state=SimpleNamespace(es_client=es_client))) return request, raw @pytest.mark.asyncio async def test_filter_options_returns_subject_word_suggestions() -> None: request, raw = _fake_request( { "aggregations": { "subject_words": { "buckets": [ {"key": "政务服务", "doc_count": 3}, {"key": "智慧城市", "doc_count": 5}, {"key": "数字政府", "doc_count": 5}, {"key": "", "doc_count": 1}, ] } } } ) perm_service = SimpleNamespace( resolve=AsyncMock( return_value=PermissionContext( user_id="user-1", acl_tokens=["U_user-1", "A_01"], ) ) ) user = UserContext( user_id="user-1", office_id="O_01", dept_id="D_01", area_id="A_01", role_ids=[], ) result = await search_api.filter_options( user=user, request=request, perm_service=perm_service, ) assert result["subject_words"] == [ {"label": "数字政府", "value": "数字政府"}, {"label": "智慧城市", "value": "智慧城市"}, {"label": "政务服务", "value": "政务服务"}, ] body = raw.search.await_args.kwargs["body"] assert body["aggs"]["subject_words"]["terms"]["field"] == "subject_words" assert body["aggs"]["subject_words"]["terms"]["size"] == 50 assert { "bool": { "should": [ {"term": {"status": "completed"}}, {"bool": {"must_not": [{"exists": {"field": "status"}}]}}, ], "minimum_should_match": 1, } } in body["query"]["bool"]["filter"] assert { "bool": { "should": [ {"bool": {"must_not": [{"exists": {"field": "acl_ids"}}]}}, {"terms": {"acl_ids": ["U_user-1", "A_01"]}}, ], "minimum_should_match": 1, } } in body["query"]["bool"]["filter"] @pytest.mark.asyncio async def test_filter_options_falls_back_to_chunk_index_when_meta_has_no_subject_words() -> None: request, raw = _fake_request({"aggregations": {"subject_words": {"buckets": []}}}) raw.search = AsyncMock( side_effect=[ {"aggregations": {"subject_words": {"buckets": []}}}, { "aggregations": { "subject_words": { "buckets": [ {"key": "数字政府", "doc_count": 4}, {"key": "政务服务", "doc_count": 2}, ] } } }, ] ) request.app.state.es_client.raw = raw perm_service = SimpleNamespace( resolve=AsyncMock( return_value=PermissionContext( user_id="user-1", acl_tokens=["U_user-1", "A_01"], ) ) ) user = UserContext( user_id="user-1", office_id="O_01", dept_id="D_01", area_id="A_01", role_ids=[], ) result = await search_api.filter_options( user=user, request=request, perm_service=perm_service, ) assert result["subject_words"] == [ {"label": "数字政府", "value": "数字政府"}, {"label": "政务服务", "value": "政务服务"}, ] assert raw.search.await_count == 2 assert raw.search.await_args_list[0].kwargs["index"] == search_api.settings.es_meta_index assert raw.search.await_args_list[1].kwargs["index"] == search_api.settings.es_chunk_index def _html_search_result() -> dict: return { "total": 1, "documents": [ { "doc_id": "doc-1", "content_hash": "hash-1", "version_count": 1, "title": "广东省人民政府 ", "highlights": [ "关于深化产教融合 " ], "matched_chunks": [ { "text": "推进产教融合 ", "chunk_index": 2, } ], } ], } async def _run_hybrid_search_with_body(body: SearchRequest, monkeypatch: pytest.MonkeyPatch): monkeypatch.setattr(search_api.settings, "search_cache_ttl", 0) search_engine = SimpleNamespace(search=AsyncMock(return_value=_html_search_result())) perm_service = SimpleNamespace( resolve=AsyncMock( return_value=PermissionContext( user_id="user-1", acl_tokens=["U_user-1"], ) ) ) user = UserContext( user_id="user-1", office_id="O_01", dept_id="D_01", area_id="A_01", role_ids=[], ) redis_client = SimpleNamespace(raw=None) return await search_api.hybrid_search( body=body, user=user, search_engine=search_engine, perm_service=perm_service, redis_client=redis_client, ) @pytest.mark.asyncio async def test_search_passes_llm_flag_to_engine_and_cache( monkeypatch: pytest.MonkeyPatch, ) -> None: monkeypatch.setattr(search_api.settings, "search_cache_ttl", 120) get_cached = AsyncMock(return_value=None) set_cached = AsyncMock() monkeypatch.setattr(search_api, "get_cached_search", get_cached) monkeypatch.setattr(search_api, "set_cached_search", set_cached) search_engine = SimpleNamespace(search=AsyncMock(return_value=_html_search_result())) perm_service = SimpleNamespace( resolve=AsyncMock( return_value=PermissionContext( user_id="user-1", acl_tokens=["U_user-1"], ) ) ) user = UserContext( user_id="user-1", office_id="O_01", dept_id="D_01", area_id="A_01", role_ids=[], ) redis_client = SimpleNamespace(raw=None) await search_api.hybrid_search( body=SearchRequest(query="产教融合", page_size=7, llm=True), user=user, search_engine=search_engine, perm_service=perm_service, redis_client=redis_client, ) assert search_engine.search.await_args.kwargs["llm"] is True assert get_cached.await_args.args == (None, "user-1", "产教融合", {}, 1) assert get_cached.await_args.kwargs == {"page_size": 7, "llm": True} assert set_cached.await_args.args[:5] == (None, "user-1", "产教融合", {}, 1) assert set_cached.await_args.kwargs == {"page_size": 7, "llm": True} @pytest.mark.asyncio async def test_search_strips_highlight_html_by_default(monkeypatch: pytest.MonkeyPatch) -> None: result = await _run_hybrid_search_with_body( SearchRequest(query="产教融合"), monkeypatch, ) doc = result.documents[0] assert doc.title == "广东省人民政府 " assert doc.highlights == ["关于深化产教融合 "] assert doc.matched_chunks[0].text == "推进产教融合 " @pytest.mark.asyncio async def test_search_preserves_highlight_html_when_escape_html_false( monkeypatch: pytest.MonkeyPatch, ) -> None: result = await _run_hybrid_search_with_body( SearchRequest(query="产教融合", escapeHtml=False), monkeypatch, ) doc = result.documents[0] assert doc.title == "广东省人民政府 " assert doc.highlights == [ "关于深化产教融合 " ] assert doc.matched_chunks[0].text == ( "推进产教融合 " )