content-find-agent-new/tests/test_pattern_recall_decision.py

import copy

from content_agent.business_modules.content_discovery.content_discovery_builder import run as build_content
from content_agent.business_modules.content_discovery.pattern_recall.recall_decision import run
from content_agent.integrations.runtime_files import LocalRuntimeFileStore
from tests.p1_helpers import real_source_payload
from tests.p4_helpers import (
    FailingCategoryMatchClient,
    FailingDecodeClient,
    FakeCategoryMatchClient,
    FakeDecodeClient,
    fake_decode_bad_shape,
    fake_decode_pending,
    fake_decode_success,
    fake_match_paths_hit,
    fake_match_paths_no_hit,
)


def _build_state(tmp_path):
    runtime = LocalRuntimeFileStore(tmp_path / "runtime")
    runtime.prepare_run("run_001")
    source_context = real_source_payload()
    platform_results = [
        {
            "content_discovery_id": "content_001",
            "search_query_id": "q_001",
            "platform": "douyin",
            "platform_content_id": "7390000000000000000",
            "platform_content_format": "video",
            "description": "爱国情感类人物故事观察",
            "platform_author_id": "author_001",
            "author_display_name": "作者",
            "statistics": {"digg_count": 1},
            "tags": ["#人物故事"],
            "score": 72,
            "portrait_available": True,
            "age_50_plus_level": "medium",
            "risk_level": "low",
            "discovery_relation": "derived_from_pattern_demand",
            "discovery_start_source": "pattern_itemset",
            "previous_discovery_step": "search_query_direct",
            "play_url": "https://video.example/a.mp4",
        }
    ]
    built = build_content("run_001", "policy_001", platform_results, source_context, runtime)
    pattern_seed_pack = {
        "seed_terms": ["爱国情感", "人物故事"],
        "category_bindings": [
            {"category_path": "/理念/观念/个人观念/情感认同/国家民族认同/爱国情感"}
        ],
    }
    return runtime, source_context, pattern_seed_pack, built


def test_recall_decision_matched_writes_evidence_and_updates_bundle(tmp_path):
    runtime, source_context, pattern_seed_pack, built = _build_state(tmp_path)

    result = run(
        "run_001",
        "policy_001",
        built["discovered_content_items"],
        built["content_media_records"],
        built["evidence_bundles"],
        source_context,
        pattern_seed_pack,
        runtime,
        FakeDecodeClient(fake_decode_success()),
        FakeCategoryMatchClient(),
        poll_interval_seconds=0,
    )

    evidence = result["pattern_recall_evidence"][0]
    bundle = result["evidence_bundles"][0]
    item = result["discovered_content_items"][0]
    assert evidence["recall_status"] == "matched"
    assert evidence["matched_terms"]
    assert evidence["matched_category_paths"]
    assert bundle["pattern_match_result"]["pattern_recall"] == "matched"
    assert bundle["pattern_match_result"]["score"] == 72
    assert item["pattern_match_result"]["pattern_recall_evidence_id"] == "recall_001"


def test_recall_decision_pending_does_not_fail_or_match(tmp_path):
    runtime, source_context, pattern_seed_pack, built = _build_state(tmp_path)

    result = run(
        "run_001",
        "policy_001",
        built["discovered_content_items"],
        built["content_media_records"],
        built["evidence_bundles"],
        source_context,
        pattern_seed_pack,
        runtime,
        FakeDecodeClient(fake_decode_pending()),
        FakeCategoryMatchClient(),
        max_wait_seconds=0,
        poll_interval_seconds=0,
    )

    assert result["pattern_recall_evidence"][0]["recall_status"] == "pending"
    assert result["evidence_bundles"][0]["pattern_match_result"]["pattern_recall"] == (
        "pattern_recall_pending"
    )


def test_recall_decision_bad_decode_is_rejected(tmp_path):
    runtime, source_context, pattern_seed_pack, built = _build_state(tmp_path)

    result = run(
        "run_001",
        "policy_001",
        built["discovered_content_items"],
        built["content_media_records"],
        built["evidence_bundles"],
        source_context,
        pattern_seed_pack,
        runtime,
        FakeDecodeClient(fake_decode_bad_shape()),
        FakeCategoryMatchClient(),
        poll_interval_seconds=0,
    )

    assert result["pattern_recall_evidence"][0]["recall_status"] == "rejected"


def test_recall_decision_decode_client_error_is_content_failed(tmp_path):
    runtime, source_context, pattern_seed_pack, built = _build_state(tmp_path)

    result = run(
        "run_001",
        "policy_001",
        built["discovered_content_items"],
        built["content_media_records"],
        built["evidence_bundles"],
        source_context,
        pattern_seed_pack,
        runtime,
        FailingDecodeClient(),
        FakeCategoryMatchClient(),
        poll_interval_seconds=0,
    )

    evidence = result["pattern_recall_evidence"][0]
    assert evidence["recall_status"] == "failed"
    assert evidence["evidence_summary"]["failure_reason"] == "decode_client_error"
    assert result["evidence_bundles"][0]["pattern_match_result"]["pattern_recall"] == (
        "pattern_recall_failed"
    )


def test_recall_decision_category_client_error_is_content_failed(tmp_path):
    runtime, source_context, pattern_seed_pack, built = _build_state(tmp_path)

    result = run(
        "run_001",
        "policy_001",
        built["discovered_content_items"],
        built["content_media_records"],
        built["evidence_bundles"],
        source_context,
        pattern_seed_pack,
        runtime,
        FakeDecodeClient(fake_decode_success()),
        FailingCategoryMatchClient(),
        poll_interval_seconds=0,
    )

    evidence = result["pattern_recall_evidence"][0]
    assert evidence["recall_status"] == "failed"
    assert evidence["evidence_summary"]["failure_reason"] == "category_match_client_error"


def test_recall_decision_no_match_when_category_has_no_hit(tmp_path):
    runtime, source_context, pattern_seed_pack, built = _build_state(tmp_path)

    result = run(
        "run_001",
        "policy_001",
        built["discovered_content_items"],
        built["content_media_records"],
        built["evidence_bundles"],
        source_context,
        pattern_seed_pack,
        runtime,
        FakeDecodeClient(fake_decode_success(terms=["不相关主题"])),
        FakeCategoryMatchClient(fake_match_paths_no_hit()),
        poll_interval_seconds=0,
    )

    assert result["pattern_recall_evidence"][0]["recall_status"] == "no_match"


def test_recall_decision_preserves_upstream_source_evidence(tmp_path):
    runtime, source_context, pattern_seed_pack, built = _build_state(tmp_path)
    before = copy.deepcopy(built["evidence_bundles"][0]["source_evidence"])

    result = run(
        "run_001",
        "policy_001",
        built["discovered_content_items"],
        built["content_media_records"],
        built["evidence_bundles"],
        source_context,
        pattern_seed_pack,
        runtime,
        FakeDecodeClient(fake_decode_success()),
        FakeCategoryMatchClient(
            fake_match_paths_hit(
                paths=[
                    "/理念/观念/个人观念/情感认同/国家民族认同/爱国情感",
                    "/理念/观念/个人观念/情感认同/国家民族认同/人物故事",
                ]
            )
        ),
        poll_interval_seconds=0,
    )

    after = result["evidence_bundles"][0]["source_evidence"]
    assert after == before
    evidence = result["pattern_recall_evidence"][0]
    assert len(evidence["matched_category_paths"]) == 2
    assert evidence["evidence_summary"]["primary_matched_category_path"]