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():
    return {
        "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,
    }


def _binding():
    return {
        "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": "/事件行为/违纪违法/综合性腐败"}],
    }


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_itemsets", 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_itemsets", 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_rejected(self):
        self._patch_success()
        result = self._build(
            {
                "source_kind": "pattern_itemset",
                "itemset_ids": [1607313, 1607314],
                "source_post_id": "p1",
            }
        )

        self.assertFalse(result["success"])
        self.assertIn("exactly one itemset_id", result["reject_reason"])


if __name__ == "__main__":
    unittest.main()