demand-agent-new/examples/demand/tests/test_pg_evidence_builder.py

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_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()