from __future__ import annotations from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock import pytest import app.api.v1.ingest as ingest_api import app.api.v1.service_guide as service_guide_api from app.api.schemas.service_guide import StandardServiceGuideProfile from app.core.permission import PermissionContext def _fake_es_client(search_result: dict) -> tuple[SimpleNamespace, MagicMock]: raw = SimpleNamespace(search=AsyncMock(return_value=search_result)) es_client = SimpleNamespace(raw=raw) return es_client, raw @pytest.mark.asyncio async def test_list_service_guides_uses_public_filter_for_anonymous_requests() -> None: es_client, raw = _fake_es_client({"hits": {"hits": [], "total": {"value": 0}}}) response = await service_guide_api.list_service_guides( es_client=es_client, perm=None, q="护照", doc_id="", implementation_code="", matter_name="", page=1, page_size=20, ) assert response.total == 0 body = raw.search.await_args.kwargs["body"] assert body["query"]["bool"]["filter"][0] == { "bool": {"must_not": [{"exists": {"field": "acl_ids"}}]} } @pytest.mark.asyncio async def test_list_service_guides_supports_prd_query_filters() -> None: es_client, raw = _fake_es_client({"hits": {"hits": [], "total": {"value": 0}}}) await service_guide_api.list_service_guides( es_client=es_client, perm=PermissionContext(user_id="user-1", acl_tokens=["U_user-1"]), q="普通护照", legacy_q="", doc_id="doc-1", implementation_code="impl-1", basic_code="basic-1", matter_type="行政许可", handled_org="广州市公安局", region="广州市", express_supported=True, reservation_supported=False, must_onsite=False, matter_name="办理普通护照", page=2, page_size=5, ) body = raw.search.await_args.kwargs["body"] bool_query = body["query"]["bool"] assert any( clause.get("multi_match", {}).get("query") == "普通护照" for clause in bool_query["must"] ) assert {"match_phrase": {"matter_name": "办理普通护照"}} in bool_query["must"] assert {"term": {"doc_id": "doc-1"}} in bool_query["filter"] assert {"term": {"implementation_code": "impl-1"}} in bool_query["filter"] assert {"term": {"basic_code": "basic-1"}} in bool_query["filter"] assert {"term": {"matter_type": "行政许可"}} in bool_query["filter"] assert {"term": {"handled_org_names": "广州市公安局"}} in bool_query["filter"] assert {"term": {"region_names": "广州市"}} in bool_query["filter"] assert {"term": {"express_supported": True}} in bool_query["filter"] assert {"term": {"reservation_supported": False}} in bool_query["filter"] assert {"term": {"must_onsite": False}} in bool_query["filter"] assert body["from"] == 5 assert body["size"] == 5 @pytest.mark.asyncio async def test_list_service_guides_accepts_legacy_q_parameter() -> None: es_client, raw = _fake_es_client({"hits": {"hits": [], "total": {"value": 0}}}) await service_guide_api.list_service_guides( es_client=es_client, perm=None, q="", legacy_q="旧参数关键词", doc_id="", implementation_code="", basic_code="", matter_type="", handled_org="", region="", express_supported=None, reservation_supported=None, must_onsite=None, matter_name="", page=1, page_size=20, ) body = raw.search.await_args.kwargs["body"] assert any( clause.get("multi_match", {}).get("query") == "旧参数关键词" for clause in body["query"]["bool"]["must"] ) @pytest.mark.asyncio async def test_list_service_guides_by_matter_falls_back_to_matter_name_match(monkeypatch: pytest.MonkeyPatch) -> None: es_client, raw = _fake_es_client( { "hits": { "total": {"value": 1}, "hits": [ { "_id": "guide-1", "_source": { "profile_id": "guide-1", "doc_id": "doc-1", "matter_name": "办理普通护照", "colloquial_names": ["护照办理"], "implementation_code": "12345", "matter_version": "v1", "handled_org_names": ["出入境管理局"], "region_names": ["广州市"], "express_supported": True, "reservation_supported": True, "needs_review": False, }, } ], } } ) class FakeGraphQueryService: def __init__(self, neo4j) -> None: self._neo4j = neo4j async def get_matter_card(self, matter_id: str, *, acl_tokens=None): return {"matter_id": matter_id, "name": "办理普通护照"} monkeypatch.setattr(service_guide_api, "GraphQueryService", FakeGraphQueryService) response = await service_guide_api.list_service_guides_by_matter( matter_id="matter-1", es_client=es_client, perm=PermissionContext(user_id="user-1", acl_tokens=["U_user-1"]), neo4j=SimpleNamespace(), page=1, page_size=20, ) assert response.total == 1 assert response.items[0].matter_name == "办理普通护照" body = raw.search.await_args.kwargs["body"] assert {"term": {"linked_matter_ids": "matter-1"}} in body["query"]["bool"]["should"] assert {"match_phrase": {"matter_name": "办理普通护照"}} in body["query"]["bool"]["should"] @pytest.mark.asyncio async def test_get_service_guide_detail_returns_404_when_acl_denied() -> None: raw = SimpleNamespace( get=AsyncMock( return_value={ "_id": "guide-1", "_source": { "profile_id": "guide-1", "doc_id": "doc-secret", "scene_type": "standard_service_guide", "guide_version": "v1", "acl_ids": ["D_secret"], }, } ) ) es_client = SimpleNamespace(raw=raw) with pytest.raises(service_guide_api.HTTPException) as exc_info: await service_guide_api.get_service_guide_detail( profile_id="guide-1", es_client=es_client, perm=PermissionContext(user_id="user-1", acl_tokens=["U_user-1"]), ) assert exc_info.value.status_code == 404 @pytest.mark.asyncio async def test_get_service_guide_by_doc_returns_typed_profile() -> None: raw = SimpleNamespace( search=AsyncMock( return_value={ "hits": { "hits": [ { "_id": "guide-1", "_source": { "profile_id": "guide-1", "doc_id": "doc-1", "scene_type": "standard_service_guide", "guide_version": "v1", "matter_name": "办理普通护照", "materials": [{"material_name": "居民身份证"}], "consultation_and_supervision": { "consultation_phones": ["12367"] }, "experimental_field": {"stage": "api"}, }, } ] } } ) ) es_client = SimpleNamespace(raw=raw) response = await service_guide_api.get_service_guide_by_doc( doc_id="doc-1", es_client=es_client, perm=None, ) assert isinstance(response.profile, StandardServiceGuideProfile) assert response.profile.matter_name == "办理普通护照" assert response.profile.materials[0].material_name == "居民身份证" assert response.profile.consultation_and_supervision.consultation_phones == ["12367"] assert response.profile.model_dump()["experimental_field"] == {"stage": "api"} @pytest.mark.asyncio async def test_get_service_guide_by_doc_tolerates_legacy_malformed_profile() -> None: raw = SimpleNamespace( search=AsyncMock( return_value={ "hits": { "hits": [ { "_id": "guide-legacy-1", "_source": { "profile_id": "guide-legacy-1", "doc_id": "doc-legacy-1", "scene_type": "standard_service_guide", "guide_version": "v1", "matter_name": "办理普通护照", "materials": "居民身份证", "consultation_and_supervision": "12345", "document_info": ["broken-shape"], }, } ] } } ) ) es_client = SimpleNamespace(raw=raw) response = await service_guide_api.get_service_guide_by_doc( doc_id="doc-legacy-1", es_client=es_client, perm=None, ) assert isinstance(response.profile, StandardServiceGuideProfile) assert response.profile.profile_id == "guide-legacy-1" assert response.profile.matter_name == "办理普通护照" assert response.profile.materials == [] assert response.profile.consultation_and_supervision.consultation_phones == [] @pytest.mark.asyncio async def test_get_service_guide_detail_tolerates_legacy_malformed_profile() -> None: raw = SimpleNamespace( get=AsyncMock( return_value={ "_id": "guide-legacy-2", "_source": { "profile_id": "guide-legacy-2", "doc_id": "doc-legacy-2", "scene_type": "standard_service_guide", "guide_version": "v2", "matter_name": "建设用地审批", "fees": "工本费120元", "basic_info": "bad-shape", "acl_ids": [], }, } ) ) es_client = SimpleNamespace(raw=raw) response = await service_guide_api.get_service_guide_detail( profile_id="guide-legacy-2", es_client=es_client, perm=None, ) assert isinstance(response.profile, StandardServiceGuideProfile) assert response.profile.profile_id == "guide-legacy-2" assert response.profile.matter_name == "建设用地审批" assert response.profile.fees == [] assert response.profile.basic_info.service_object == [] @pytest.mark.asyncio async def test_permission_webhook_queues_empty_acl_as_public(monkeypatch: pytest.MonkeyPatch) -> None: fake_delay = MagicMock(return_value=SimpleNamespace(id="task-public")) monkeypatch.setattr(ingest_api.update_permissions_task, "delay", fake_delay) response = await ingest_api.webhook_update_permission( request=SimpleNamespace(), body=ingest_api.PermissionUpdateRequest(doc_id="doc-public", acl_ids=[]), ) assert response["status"] == "queued" assert response["task_id"] == "task-public" fake_delay.assert_called_once_with(doc_id="doc-public", acl_ids=[])