import unittest from unittest.mock import patch from examples.demand.evidence_pack_builder import build_evidence_pack def _execution(): return {"id": 581, "status": "success"} def _itemset(**overrides): data = { "id": 1607313, "execution_id": 581, "mining_config_id": 2081, "mining_config_execution_id": 581, "mining_config_scope": "topic", "scope": "topic", "item_count": 1, "support": 0.1, "absolute_support": 2, "matched_post_ids": ["p1", "p2"], } data.update(overrides) return data def _item(**overrides): data = { "itemset_item_id": 1, "itemset_id": 1607313, "category_id": 10, "bound_category_id": 10, "category_found": True, "category_execution_id": 581, "category_name": "综合性腐败", "category_path": "/事件行为/违纪违法/综合性腐败", "category_full_path": "/事件行为/违纪违法/综合性腐败", "dimension": "实质", "point_type": "关键点", "element_name": None, } data.update(overrides) return data def _binding(**overrides): data = { "itemset_id": 1607313, "itemset_item_id": 1, "category_id": 10, "dimension": "实质", "point_type": "关键点", "element_name": None, "matched_element_count": 1, "matched_post_count": 1, "matched_post_ids": ["p1"], "sample_elements": [{"name": "综合性腐败", "category_path": "/事件行为/违纪违法/综合性腐败"}], } data.update(overrides) return data def _empty_binding(post_ids=None): binding = _binding() binding["matched_element_count"] = 0 binding["matched_post_count"] = 0 binding["matched_post_ids"] = post_ids or [] binding["sample_elements"] = [] return binding class PgEvidenceBuilderTest(unittest.TestCase): def _build(self, evidence_refs): return build_evidence_pack( 581, {"element_names": ["综合性腐败"], "evidence_refs": evidence_refs}, trace_id="trace-test", demand_task_id=1, demand_content_id=1, ) def _patch_success(self, itemsets=None): patches = [ patch("examples.demand.evidence_pack_builder.query_execution_for_evidence", return_value=_execution()), patch("examples.demand.evidence_pack_builder.query_itemset_evidence", return_value=itemsets or [_itemset()]), patch("examples.demand.evidence_pack_builder.query_itemset_items_with_categories", return_value=[_item()]), patch("examples.demand.evidence_pack_builder.query_element_bindings_for_items", return_value=[_binding()]), patch("examples.demand.evidence_pack_builder.query_case_ids_by_post_ids", return_value=[]), patch("examples.demand.evidence_pack_builder.query_seed_points_for_sources", return_value=[]), ] for patcher in patches: patcher.start() self.addCleanup(patcher.stop) def test_topic_itemset_builds_pg_evidence_pack(self): self._patch_success() result = self._build( { "source_kind": "pattern_itemset", "source_tool": "get_itemset_detail", "itemset_ids": [1607313], "source_post_id": "p1", "seed_terms": ["LLM伪造词"], } ) self.assertTrue(result["success"]) evidence_pack = result["evidence_pack"] self.assertEqual(evidence_pack["pattern_source_system"], "pg_pattern_v2") self.assertEqual(evidence_pack["validation_status"], "passed") self.assertEqual(evidence_pack["source_certainty"], "db_validated") self.assertEqual(evidence_pack["matched_post_ids"], ["p1", "p2"]) self.assertEqual(evidence_pack["seed_terms"], ["综合性腐败"]) self.assertIn("query_seed_points", evidence_pack) self.assertIn("demand_scope", evidence_pack) def test_demand_scope_platform_is_canonicalized(self): self._patch_success() result = build_evidence_pack( 581, { "element_names": ["综合性腐败"], "evidence_refs": { "source_kind": "pattern_itemset", "source_tool": "get_itemset_detail", "itemset_ids": [1607313], "source_post_id": "p1", }, }, trace_id="trace-test", demand_task_id=1, demand_content_id=1, demand_scope={"platform": "票圈"}, ) self.assertTrue(result["success"]) self.assertEqual(result["evidence_pack"]["demand_scope"]["platform"], "piaoquan") def test_defaults_to_piaoquan_scope_and_caps_query_seed_points(self): with ( patch("examples.demand.evidence_pack_builder.query_execution_for_evidence", return_value=_execution()), patch("examples.demand.evidence_pack_builder.query_itemset_evidence", return_value=[_itemset(matched_post_ids=["pq1", "pq2"])]) as itemset_evidence, patch("examples.demand.evidence_pack_builder.query_itemset_items_with_categories", return_value=[_item()]), patch("examples.demand.evidence_pack_builder.query_element_bindings_for_items", return_value=[_binding(matched_post_ids=["pq1"])]), patch("examples.demand.evidence_pack_builder.query_case_ids_by_post_ids", return_value=[]), patch("examples.demand.evidence_pack_builder.query_seed_points_for_sources", return_value=[]) as seed_points, patch.dict("os.environ", {"DEMAND_QUERY_SEED_POINTS_TOP_K": "100"}), ): result = self._build( { "source_kind": "pattern_itemset", "source_tool": "get_itemset_detail", "itemset_ids": [1607313], "source_post_id": "pq1", } ) self.assertTrue(result["success"]) self.assertEqual(result["evidence_pack"]["demand_scope"]["platform"], "piaoquan") self.assertEqual(itemset_evidence.call_args.kwargs["platform"], "piaoquan") seed_points.assert_called_once() self.assertEqual(seed_points.call_args.kwargs["matched_post_ids"], ["pq1", "pq2"]) self.assertEqual(seed_points.call_args.kwargs["top_k"], 30) def test_non_topic_scope_is_rejected(self): self._patch_success(itemsets=[_itemset(scope="topic_element")]) result = self._build( { "source_kind": "pattern_itemset", "itemset_ids": [1607313], "source_post_id": "p1", } ) self.assertFalse(result["success"]) self.assertIn("scope=topic_element", result["reject_reason"]) def test_source_post_outside_matched_posts_is_rejected(self): self._patch_success() result = self._build( { "source_kind": "pattern_itemset", "itemset_ids": [1607313], "source_post_id": "not-a-source-post", } ) self.assertFalse(result["success"]) self.assertIn("not in matched_post_ids", result["reject_reason"]) def test_falls_back_to_db_validated_source_post_with_element_binding(self): bindings = [ [_empty_binding()], [{**_binding(), "matched_post_ids": ["p2"]}], [{**_binding(), "matched_post_ids": ["p2"]}], ] with ( patch("examples.demand.evidence_pack_builder.query_execution_for_evidence", return_value=_execution()), patch("examples.demand.evidence_pack_builder.query_itemset_evidence", return_value=[_itemset()]), patch("examples.demand.evidence_pack_builder.query_itemset_items_with_categories", return_value=[_item()]), patch("examples.demand.evidence_pack_builder.query_element_bindings_for_items", side_effect=bindings), patch("examples.demand.evidence_pack_builder.query_case_ids_by_post_ids", return_value=[]), patch("examples.demand.evidence_pack_builder.query_seed_points_for_sources", return_value=[]), ): result = self._build( { "source_kind": "pattern_itemset", "itemset_ids": [1607313], "source_post_id": "p1", } ) self.assertTrue(result["success"]) self.assertEqual(result["evidence_pack"]["source_post_id"], "p2") def test_multiple_itemsets_are_supported(self): patches = [ patch("examples.demand.evidence_pack_builder.query_execution_for_evidence", return_value=_execution()), patch( "examples.demand.evidence_pack_builder.query_itemset_evidence", return_value=[ _itemset(), _itemset(id=1607314, mining_config_id=2082, absolute_support=2), ], ), patch( "examples.demand.evidence_pack_builder.query_itemset_items_with_categories", return_value=[ _item(), _item(itemset_item_id=2, itemset_id=1607314, category_id=11, bound_category_id=11), ], ), patch( "examples.demand.evidence_pack_builder.query_element_bindings_for_items", return_value=[ _binding(), _binding(itemset_id=1607314, itemset_item_id=2, category_id=11), ], ), patch("examples.demand.evidence_pack_builder.query_case_ids_by_post_ids", return_value=[]), patch("examples.demand.evidence_pack_builder.query_seed_points_for_sources", return_value=[]), ] for patcher in patches: patcher.start() self.addCleanup(patcher.stop) result = self._build( { "source_kind": "pattern_itemset", "itemset_ids": [1607313, 1607314], "source_post_id": "p1", } ) self.assertTrue(result["success"]) self.assertEqual(result["evidence_pack"]["itemset_ids"], [1607313, 1607314]) self.assertEqual(result["evidence_pack"]["source_kind"], "pattern_itemset") def test_high_weight_element_source_builds_pg_evidence_pack(self): rows = [ { "id": 1, "post_id": "p1", "source_table": "post_decode_topic_point_element", "source_element_id": 11, "point_type": "灵感点", "point_text": "反腐故事", "element_type": "实质", "name": "综合性腐败", "category_id": 10, "category_name": "综合性腐败", "category_path": "/事件行为/违纪违法/综合性腐败", "category_full_path": "/事件行为/违纪违法/综合性腐败", "topic_point_id": 101, } ] with ( patch("examples.demand.evidence_pack_builder.query_execution_for_evidence", return_value={**_execution(), "post_count": 10}), patch("examples.demand.evidence_pack_builder.query_source_elements", return_value=rows), patch("examples.demand.evidence_pack_builder.query_case_ids_by_post_ids", return_value=[]), patch("examples.demand.evidence_pack_builder.query_seed_points_for_sources", return_value=[]), ): result = self._build( { "sources": [ { "source_kind": "high_weight_element", "source_tool": "get_weight_score_topn", "element_names": ["综合性腐败"], "element_type": "实质", } ] } ) self.assertTrue(result["success"]) pack = result["evidence_pack"] self.assertEqual(pack["source_kind"], "high_weight_element") self.assertEqual(pack["itemset_ids"], []) self.assertEqual(pack["matched_post_ids"], ["p1"]) self.assertEqual(pack["seed_terms"], ["综合性腐败"]) self.assertEqual(pack["evidence_sources"][0]["source_kind"], "high_weight_element") def test_co_occurrence_alias_is_normalized(self): rows = [ { "id": 1, "post_id": "p1", "source_table": "post_decode_topic_point_element", "source_element_id": 11, "point_type": "灵感点", "point_text": "反腐故事", "element_type": "实质", "name": "综合性腐败", "category_id": 10, "category_name": "综合性腐败", "category_path": "/事件行为/违纪违法/综合性腐败", "category_full_path": "/事件行为/违纪违法/综合性腐败", "topic_point_id": 101, } ] with ( patch("examples.demand.evidence_pack_builder.query_execution_for_evidence", return_value={**_execution(), "post_count": 10}), patch("examples.demand.evidence_pack_builder.query_source_elements", return_value=rows), patch("examples.demand.evidence_pack_builder.query_case_ids_by_post_ids", return_value=[]), patch("examples.demand.evidence_pack_builder.query_seed_points_for_sources", return_value=[]), ): result = self._build( { "sources": [ { "source_kind": "co-occurrence", "source_tool": "get_element_co_occurrences", "element_names": ["综合性腐败"], } ] } ) self.assertTrue(result["success"]) self.assertEqual(result["evidence_pack"]["source_kind"], "element_co_occurrence") def test_high_weight_category_source_builds_pg_evidence_pack(self): rows = [ { "id": 1, "post_id": "p1", "source_table": "post_decode_topic_point_element", "source_element_id": 11, "point_type": "灵感点", "point_text": "反腐故事", "element_type": "实质", "name": "综合性腐败", "category_id": 10, "category_name": "综合性腐败", "category_path": "/事件行为/违纪违法/综合性腐败", "category_full_path": "/事件行为/违纪违法/综合性腐败", "topic_point_id": 101, } ] with ( patch("examples.demand.evidence_pack_builder.query_execution_for_evidence", return_value={**_execution(), "post_count": 10}), patch("examples.demand.evidence_pack_builder.query_source_elements", return_value=rows), patch("examples.demand.evidence_pack_builder.query_case_ids_by_post_ids", return_value=[]), patch("examples.demand.evidence_pack_builder.query_seed_points_for_sources", return_value=[]), ): result = self._build( { "sources": [ { "source_kind": "high_weight_category", "source_tool": "get_weight_score_topn", "category_ids": [10], } ] } ) self.assertTrue(result["success"]) pack = result["evidence_pack"] self.assertEqual(pack["source_kind"], "high_weight_category") self.assertEqual(pack["matched_post_ids"], ["p1"]) self.assertEqual(pack["seed_terms"], ["综合性腐败"]) self.assertEqual(pack["evidence_sources"][0]["source_kind"], "high_weight_category") def test_category_co_occurrence_source_builds_pg_evidence_pack(self): rows = [ { "id": 1, "post_id": "p1", "source_table": "post_decode_topic_point_element", "source_element_id": 11, "point_type": "灵感点", "point_text": "反腐故事", "element_type": "实质", "name": "综合性腐败", "category_id": 10, "category_name": "综合性腐败", "category_path": "/事件行为/违纪违法/综合性腐败", "category_full_path": "/事件行为/违纪违法/综合性腐败", "topic_point_id": 101, } ] with ( patch("examples.demand.evidence_pack_builder.query_execution_for_evidence", return_value={**_execution(), "post_count": 10}), patch("examples.demand.evidence_pack_builder.query_source_elements", return_value=rows), patch("examples.demand.evidence_pack_builder.query_case_ids_by_post_ids", return_value=[]), patch("examples.demand.evidence_pack_builder.query_seed_points_for_sources", return_value=[]), ): result = self._build( { "sources": [ { "source_kind": "category_co_occurrence", "source_tool": "get_category_co_occurrences", "category_names": ["综合性腐败"], } ] } ) self.assertTrue(result["success"]) self.assertEqual(result["evidence_pack"]["source_kind"], "category_co_occurrence") if __name__ == "__main__": unittest.main()