"""Unit tests for the rule-first graph query planner.""" from __future__ import annotations from unittest.mock import AsyncMock, MagicMock import pytest from app.core.graph_query_planner import ( GraphQueryPlanner, QueryIntent, QueryPlan, ) @pytest.mark.parametrize( ("question", "keywords", "expected"), [ ("这个文件的上位法依据是什么", ["上位法", "依据"], QueryIntent.POLICY_CHAIN), ("XX规定被哪些文件修改过", ["XX规定", "修改"], QueryIntent.REVISION_HISTORY), ("自然资源局发布了哪些政策", ["自然资源局", "政策"], QueryIntent.ENTITY_DOCS), ("建设用地审批需要什么材料", ["建设用地审批", "材料"], QueryIntent.MATTER_DETAIL), ("营商环境相关政策有哪些", ["营商环境", "政策"], QueryIntent.THEME_EXPLORE), ("清远市2024年经济发展情况", ["清远市", "2024年", "经济发展"], QueryIntent.GENERAL), ], ) def test_plan_classifies_intent( question: str, keywords: list[str], expected: QueryIntent, ): planner = GraphQueryPlanner() plan = planner.plan(question, keywords) assert plan.intent == expected def test_plan_extracts_doc_code() -> None: planner = GraphQueryPlanner() plan = planner.plan("清政办〔2024〕3号的上位法依据是什么？", ["清政办〔2024〕3号"]) assert plan.intent == QueryIntent.POLICY_CHAIN assert plan.doc_code == "清政办〔2024〕3号" @pytest.mark.asyncio async def test_collect_matter_detail_evidence() -> None: planner = GraphQueryPlanner() graph = MagicMock() graph.search_matters = AsyncMock( return_value=[ { "matter_id": "matter_001", "name": "建设用地审批", } ] ) graph.get_matter_card = AsyncMock( return_value={ "matter_id": "matter_001", "name": "建设用地审批", "conditions": [{"name": "申请人应具有法人资格"}], "materials": [{"name": "营业执照副本"}], "time_limits": [{"name": "15个工作日"}], "handled_by": [{"name": "自然资源局"}], "governing_docs": [ { "doc_id": "doc-1", "title": "建设用地管理办法", "doc_code": "清政办〔2024〕3号", } ], } ) evidence = await planner.collect_evidence( QueryPlan(intent=QueryIntent.MATTER_DETAIL, matter_query="建设用地审批"), graph, ) assert "### 事项：建设用地审批" in evidence.text assert "营业执照副本" in evidence.text assert "自然资源局" in evidence.text assert evidence.doc_ids == ["doc-1"] @pytest.mark.asyncio async def test_collect_matter_detail_evidence_passes_acl_tokens() -> None: planner = GraphQueryPlanner() graph = MagicMock() graph.search_matters = AsyncMock( return_value=[ { "matter_id": "matter_001", "name": "建设用地审批", } ] ) graph.get_matter_card = AsyncMock( return_value={ "matter_id": "matter_001", "name": "建设用地审批", "conditions": [], "materials": [], "time_limits": [], "handled_by": [], "governing_docs": [], } ) await planner.collect_evidence( QueryPlan(intent=QueryIntent.MATTER_DETAIL, matter_query="建设用地审批"), graph, acl_tokens=["D_01"], ) graph.search_matters.assert_awaited_once_with( "建设用地审批", limit=5, acl_tokens=["D_01"], ) graph.get_matter_card.assert_awaited_once_with( "matter_001", acl_tokens=["D_01"], ) @pytest.mark.asyncio async def test_collect_policy_chain_evidence() -> None: planner = GraphQueryPlanner() graph = MagicMock() graph.find_doc_by_code = AsyncMock(return_value="doc-1") graph.get_policy_chain = AsyncMock( return_value={ "chain": [ { "doc_id": "doc-1", "title": "XX实施细则", "doc_code": "清政办〔2024〕3号", }, { "doc_id": "doc-2", "title": "YY管理办法", "doc_code": "国发〔2023〕15号", }, ], "edges": [ { "from_doc_id": "doc-1", "to_doc_id": "doc-2", "rel_type": "BASED_ON", } ], } ) evidence = await planner.collect_evidence( QueryPlan(intent=QueryIntent.POLICY_CHAIN, doc_code="清政办〔2024〕3号"), graph, ) assert "### 政策依据链" in evidence.text assert "XX实施细则" in evidence.text assert "YY管理办法" in evidence.text assert evidence.doc_ids == ["doc-1", "doc-2"] @pytest.mark.asyncio async def test_collect_general_returns_empty() -> None: planner = GraphQueryPlanner() evidence = await planner.collect_evidence( QueryPlan(intent=QueryIntent.GENERAL), MagicMock(), ) assert evidence.text == "" assert evidence.doc_ids == [] @pytest.mark.asyncio async def test_collect_evidence_degrades_on_graph_error() -> None: planner = GraphQueryPlanner() graph = MagicMock() graph.find_doc_by_code = AsyncMock(side_effect=RuntimeError("boom")) evidence = await planner.collect_evidence( QueryPlan(intent=QueryIntent.POLICY_CHAIN, doc_code="清政办〔2024〕3号"), graph, ) assert evidence.text == "" assert evidence.doc_ids == [] # =========================================================================== # Revision history evidence # =========================================================================== @pytest.mark.asyncio async def test_collect_revision_history_evidence() -> None: planner = GraphQueryPlanner() graph = MagicMock() graph.find_doc_by_code = AsyncMock(return_value="doc-1") graph.get_revision_history = AsyncMock( return_value={ "documents": [ {"doc_id": "doc-1", "title": "旧版管理办法", "doc_code": "清政办〔2020〕5号"}, {"doc_id": "doc-2", "title": "新版管理办法", "doc_code": "清政办〔2024〕3号"}, ], "edges": [ { "from_doc_id": "doc-2", "to_doc_id": "doc-1", "rel_type": "AMENDS", } ], } ) evidence = await planner.collect_evidence( QueryPlan(intent=QueryIntent.REVISION_HISTORY, doc_code="清政办〔2024〕3号"), graph, ) assert "### 修订历史" in evidence.text assert "旧版管理办法" in evidence.text assert "修订" in evidence.text assert set(evidence.doc_ids) == {"doc-1", "doc-2"} # =========================================================================== # Entity docs evidence # =========================================================================== @pytest.mark.asyncio async def test_collect_entity_docs_evidence() -> None: planner = GraphQueryPlanner() graph = MagicMock() graph.search_entities = AsyncMock( return_value=[ { "id": "eid-1", "labels": ["Organization"], "properties": {"name": "自然资源局"}, } ] ) graph.get_docs_by_entity = AsyncMock( return_value=[ {"doc_id": "doc-a", "title": "用地管理办法", "doc_number": "", "rel_type": "ISSUED_BY"}, {"doc_id": "doc-b", "title": "土地规划通知", "doc_number": "", "rel_type": "ISSUED_BY"}, ] ) evidence = await planner.collect_evidence( QueryPlan( intent=QueryIntent.ENTITY_DOCS, entity_name="自然资源局", entity_label="Organization", ), graph, ) assert "### 自然资源局 相关文件" in evidence.text assert "用地管理办法" in evidence.text assert "土地规划通知" in evidence.text assert evidence.doc_ids == ["doc-a", "doc-b"] @pytest.mark.asyncio async def test_collect_entity_docs_no_entity_name_returns_empty() -> None: planner = GraphQueryPlanner() evidence = await planner.collect_evidence( QueryPlan(intent=QueryIntent.ENTITY_DOCS, entity_name=None), MagicMock(), ) assert evidence.text == "" assert evidence.doc_ids == [] # =========================================================================== # Theme explore evidence # =========================================================================== @pytest.mark.asyncio async def test_collect_theme_docs_via_entity_resolve() -> None: """Theme explore without doc_code falls back to search_entities for PolicyTheme.""" planner = GraphQueryPlanner() graph = MagicMock() graph.search_entities = AsyncMock( return_value=[ { "id": "eid-t1", "labels": ["PolicyTheme"], "properties": {"name": "营商环境优化"}, } ] ) graph.get_docs_by_entity = AsyncMock( return_value=[ {"doc_id": "doc-x", "title": "营商环境实施方案", "doc_number": "", "rel_type": "BELONGS_TO_THEME"}, ] ) evidence = await planner.collect_evidence( QueryPlan( intent=QueryIntent.THEME_EXPLORE, entity_name="营商环境", entity_label="PolicyTheme", ), graph, ) assert "主题相关文件" in evidence.text assert "营商环境实施方案" in evidence.text assert evidence.doc_ids == ["doc-x"] # Verify it called search_entities with PolicyTheme label graph.search_entities.assert_awaited_once() call_kwargs = graph.search_entities.call_args assert call_kwargs.kwargs.get("label") == "PolicyTheme" @pytest.mark.asyncio async def test_collect_theme_docs_via_doc_code() -> None: """Theme explore with doc_code uses get_same_theme_documents.""" planner = GraphQueryPlanner() graph = MagicMock() graph.find_doc_by_code = AsyncMock(return_value="doc-1") graph.get_same_theme_documents = AsyncMock( return_value=[ {"doc_id": "doc-y", "title": "相关文件A", "theme_name": "营商环境", "doc_code": ""}, ] ) evidence = await planner.collect_evidence( QueryPlan( intent=QueryIntent.THEME_EXPLORE, doc_code="清政办〔2024〕3号", entity_name="营商环境", entity_label="PolicyTheme", ), graph, ) assert "主题相关文件" in evidence.text assert "相关文件A" in evidence.text assert evidence.doc_ids == ["doc-y"] graph.get_same_theme_documents.assert_awaited_once_with("doc-1", limit=8) # =========================================================================== # Edge cases # =========================================================================== def test_plan_with_empty_keywords() -> None: planner = GraphQueryPlanner() plan = planner.plan("建设用地审批需要什么材料", []) assert plan.intent == QueryIntent.MATTER_DETAIL def test_pick_best_entity_prefers_exact_match() -> None: planner = GraphQueryPlanner() entities = [ {"labels": ["Organization"], "properties": {"name": "自然资源和规划局"}}, {"labels": ["Organization"], "properties": {"name": "自然资源局"}}, ] best = planner._pick_best_entity(entities, "自然资源局") assert best["properties"]["name"] == "自然资源局" def test_pick_best_entity_falls_back_to_first() -> None: planner = GraphQueryPlanner() entities = [ {"labels": ["Organization"], "properties": {"name": "自然资源和规划局"}}, ] best = planner._pick_best_entity(entities, "自然资源局") assert best["properties"]["name"] == "自然资源和规划局" # =========================================================================== # V1.5: New entity detail intent classification # =========================================================================== @pytest.mark.parametrize( ("question", "keywords", "expected_intent", "expected_label"), [ ("一体化政务服务平台的运维情况", ["一体化政务服务平台", "运维"], QueryIntent.SYSTEM_DETAIL, "System"), ("人口基础信息数据资源的详细信息", ["人口基础信息数据资源", "详细信息"], QueryIntent.DATA_RESOURCE_DETAIL, "DataResource"), ("这笔预算资金的使用情况", ["预算资金", "使用"], QueryIntent.BUDGET_DETAIL, "Budget"), ("GDP增速指标的详情", ["GDP增速指标", "详情"], QueryIntent.INDICATOR_DETAIL, "Indicator"), ("新能源汽车产业的发展情况", ["新能源汽车产业", "发展"], QueryIntent.INDUSTRY_DETAIL, "Industry"), ], ) def test_plan_classifies_v15_detail_intents( question: str, keywords: list[str], expected_intent: QueryIntent, expected_label: str, ): planner = GraphQueryPlanner() plan = planner.plan(question, keywords) assert plan.intent == expected_intent assert plan.entity_label == expected_label @pytest.mark.asyncio async def test_collect_system_detail_evidence() -> None: planner = GraphQueryPlanner() graph = MagicMock() graph.resolve_entity_by_exact_name = AsyncMock(return_value="一体化政务服务平台") graph.get_system_card = AsyncMock( return_value={ "name": "一体化政务服务平台", "operated_by": ["数字政府中心"], "managed_data": ["办件数据库"], "technologies": ["微服务架构"], "related_docs": [], } ) evidence = await planner.collect_evidence( QueryPlan(intent=QueryIntent.SYSTEM_DETAIL, entity_name="一体化政务服务平台", entity_label="System"), graph, ) assert "### 系统：一体化政务服务平台" in evidence.text assert "数字政府中心" in evidence.text @pytest.mark.asyncio async def test_collect_budget_detail_evidence_with_related_docs() -> None: planner = GraphQueryPlanner() graph = MagicMock() graph.resolve_entity_by_exact_name = AsyncMock(return_value="2024年度专项资金") graph.get_budget_card = AsyncMock( return_value={ "name": "2024年度专项资金", "funded_tasks": [{"name": "数据治理任务", "task_id": "t-1"}], "funded_projects": [], "related_docs": [ {"doc_id": "doc-1", "title": "资金管理办法", "doc_code": "清财〔2024〕5号"}, ], } ) evidence = await planner.collect_evidence( QueryPlan(intent=QueryIntent.BUDGET_DETAIL, entity_name="2024年度专项资金", entity_label="Budget"), graph, ) assert "### 预算：2024年度专项资金" in evidence.text assert "数据治理任务" in evidence.text assert evidence.doc_ids == ["doc-1"] @pytest.mark.asyncio async def test_collect_detail_returns_empty_on_resolve_failure() -> None: """All new detail intents degrade gracefully when entity not found.""" planner = GraphQueryPlanner() graph = MagicMock() graph.resolve_entity_by_exact_name = AsyncMock(return_value=None) for intent in ( QueryIntent.SYSTEM_DETAIL, QueryIntent.DATA_RESOURCE_DETAIL, QueryIntent.BUDGET_DETAIL, QueryIntent.INDICATOR_DETAIL, QueryIntent.INDUSTRY_DETAIL, ): evidence = await planner.collect_evidence( QueryPlan(intent=intent, entity_name="不存在的实体", entity_label="System"), graph, ) assert evidence.text == "" assert evidence.doc_ids == [] def test_pick_best_named_item_exact_then_contains_then_first() -> None: planner = GraphQueryPlanner() # Exact match items = [{"name": "建设用地审批管理"}, {"name": "建设用地审批"}] best = planner._pick_best_named_item(items, "建设用地审批") assert best["name"] == "建设用地审批" # Contains fallback items = [{"name": "建设用地审批管理"}, {"name": "土地管理"}] best = planner._pick_best_named_item(items, "建设用地审批") assert best["name"] == "建设用地审批管理" # First fallback items = [{"name": "完全不相关"}, {"name": "另一个"}] best = planner._pick_best_named_item(items, "建设用地审批") assert best["name"] == "完全不相关"