from __future__ import annotations from pathlib import Path from unittest.mock import AsyncMock, MagicMock import pytest from app.config import settings from app.core.ingest_pipeline import IngestPipeline, _NoOpRecorder from app.core.service_guide_extractor import ServiceGuideExtractionOutput from app.schemas.service_guide_profile import StandardServiceGuideProfile class _FakeChunk: def __init__(self, content: str) -> None: self.content = content self.metadata: dict[str, object] = {} def to_dict(self) -> dict[str, object]: return { "chunk_id": "chunk-1", "content_hash": "hash-1", "doc_ids": ["doc-1"], "chunk_index": 0, "content": self.content, "acl_ids": [], } @pytest.mark.asyncio async def test_ingest_pipeline_runs_service_guide_stage_on_unified_analysis_path( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: source_file = tmp_path / "guide.txt" source_file.write_text("标准办事指南测试内容", encoding="utf-8") es_client = MagicMock() es_client.find_by_content_hash = AsyncMock(return_value=None) es_client.bulk_index_chunks = AsyncMock(return_value=(1, [])) es_client.recompute_chunk_acl = AsyncMock() es_client.raw = MagicMock() es_client.raw.update = AsyncMock() es_client.raw.index = AsyncMock() embedding = MagicMock() embedding.embed_chunks = AsyncMock(return_value=[[0.1, 0.2]]) chunker = MagicMock() chunker.chunk_document.return_value = [_FakeChunk("标准办事指南测试内容")] document_analyzer = MagicMock() document_analyzer.analyze = AsyncMock( return_value=( { "title": "自动抽取标题", "doc_number": "AUTO-001", "document_scene_type": "standard_service_guide", }, "自动摘要", {"system_prompt": "", "user_prompt_template": ""}, ) ) monkeypatch.setattr(settings, "summary_enabled", True) monkeypatch.setattr(settings, "graph_build_enabled", False) monkeypatch.setattr(settings, "file_storage_path", tmp_path / "storage") pipeline = IngestPipeline( es_client=es_client, embedding_service=embedding, chunker=chunker, document_analyzer=document_analyzer, service_guide_extractor=MagicMock(), ) pipeline._run_service_guide_stage = AsyncMock() # type: ignore[method-assign] result = await pipeline.ingest_document( "doc-1", source_file, {"acl_ids": [], "auto_extract": True}, ) assert result["status"] == "completed" pipeline._run_service_guide_stage.assert_awaited_once() @pytest.mark.asyncio async def test_ingest_pipeline_syncs_service_guide_matter_bindings() -> None: es_client = MagicMock() es_client.bind_service_guide_matters = AsyncMock(return_value=1) graph_query_service = MagicMock() graph_query_service.get_governed_matters = AsyncMock( return_value=[ {"matter_id": "matter_1", "name": "事项一"}, {"matter_id": "matter_2", "name": "事项二"}, ] ) pipeline = IngestPipeline( es_client=es_client, embedding_service=MagicMock(), graph_query_service=graph_query_service, ) matter_ids = await pipeline._sync_service_guide_matter_bindings( guide_doc_id="guide-doc-1", source_doc_id="source-doc-1", ) assert matter_ids == ["matter_1", "matter_2"] graph_query_service.get_governed_matters.assert_awaited_once_with("source-doc-1") es_client.bind_service_guide_matters.assert_awaited_once_with( "guide-doc-1", ["matter_1", "matter_2"], ) @pytest.mark.asyncio async def test_ingest_pipeline_serializes_typed_service_guide_profile_before_indexing() -> None: es_client = MagicMock() es_client.delete_service_guides_by_doc_id = AsyncMock() es_client.raw = MagicMock() es_client.raw.index = AsyncMock() extractor = MagicMock() extractor.extract = AsyncMock( return_value=ServiceGuideExtractionOutput( detected=True, scene_type="standard_service_guide", detection_confidence=0.96, profile=StandardServiceGuideProfile( profile_id="guide-typed-1", doc_id="doc-typed-1", matter_name="办理普通护照", materials=[{"material_name": "居民身份证"}], service_windows=[{"window_name": "出入境窗口"}], ), root_fields={"matter_name": "办理普通护照"}, ) ) pipeline = IngestPipeline( es_client=es_client, embedding_service=MagicMock(), service_guide_extractor=extractor, ) metadata = {"acl_ids": [], "title": "办理普通护照办事指南"} await pipeline._run_service_guide_stage( doc_id="doc-typed-1", content_hash="hash-typed-1", metadata=metadata, full_text="事项名称：办理普通护照", rec=_NoOpRecorder(), patch_chunks=False, ) assert metadata["service_guide_status"] == "completed" assert metadata["guide_profile_id"] == "guide-typed-1" assert metadata["guide_matter_name"] == "办理普通护照" body = es_client.raw.index.await_args.kwargs["body"] assert isinstance(body, dict) assert body["profile_id"] == "guide-typed-1" assert body["matter_name"] == "办理普通护照" assert body["materials"][0]["material_name"] == "居民身份证" @pytest.mark.asyncio async def test_ingest_pipeline_preserves_existing_guide_when_detected_profile_is_missing() -> None: es_client = MagicMock() es_client.find_service_guide_by_doc_id = AsyncMock( return_value={ "profile_id": "guide-existing-1", "matter_name": "既有事项", "scene_type": "standard_service_guide", } ) es_client.delete_service_guides_by_doc_id = AsyncMock() es_client.patch_chunks_by_content_hash = AsyncMock() es_client.raw = MagicMock() es_client.raw.index = AsyncMock() extractor = MagicMock() extractor.extract = AsyncMock( return_value=ServiceGuideExtractionOutput( detected=True, scene_type="standard_service_guide", detection_confidence=0.81, detection_reasons=["包含事项名称与申请材料栏目"], warnings=["材料表格缺少结构化行"], profile=None, ) ) rec = MagicMock(spec=_NoOpRecorder) rec.record_stage_start = AsyncMock() rec.record_stage_complete = AsyncMock() rec.record_stage_failed = AsyncMock() rec.has_stage_failure = False pipeline = IngestPipeline( es_client=es_client, embedding_service=MagicMock(), service_guide_extractor=extractor, ) metadata = {"acl_ids": [], "title": "既有事项办事指南"} await pipeline._run_service_guide_stage( doc_id="doc-existing-1", content_hash="hash-existing-1", metadata=metadata, full_text="事项名称：既有事项", rec=rec, patch_chunks=False, ) assert metadata["service_guide_status"] == "failed" assert metadata["guide_profile_id"] == "guide-existing-1" assert metadata["guide_matter_name"] == "既有事项" assert metadata["document_scene_type"] == "standard_service_guide" es_client.find_service_guide_by_doc_id.assert_awaited_once_with( "doc-existing-1", source_fields=["profile_id", "matter_name", "scene_type"], ) es_client.delete_service_guides_by_doc_id.assert_not_awaited() rec.record_stage_failed.assert_awaited_once() assert rec.record_stage_failed.await_args.kwargs["error_code"] == "INGEST_SERVICE_GUIDE_PROFILE_UNAVAILABLE"