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

from content_agent.business_modules.content_discovery.pattern_recall.decode import (
    decode_content,
    extract_decode_elements,
    normalize_decode_status,
)
from content_agent.integrations.decode_api import AigcDecodeClient, redact_sensitive_payload
from tests.p4_helpers import (
    FakeDecodeClient,
    fake_decode_bad_shape,
    fake_decode_pending,
    fake_decode_success,
)


def test_decode_status_normalizes_external_uppercase_values():
    assert normalize_decode_status({"decode_status": "SUCCESS"}) == "success"
    assert normalize_decode_status({"decode_status": "RUNNING"}) == "running"
    assert normalize_decode_status({"decode_status": "PENDING"}) == "pending"
    assert normalize_decode_status({"decode_status": "FAILED"}) == "failed"


def test_decode_extracts_only_strong_terms_from_decode_content():
    elements = extract_decode_elements(
        {
            "目的点": [{"点": "爱国情感", "实质": [{"名称": "人物故事"}]}],
            "关键点": [{"点": "标题提示", "类型": "形式"}, {"点": "国家认同", "类型": "实质"}],
            "分词结果": [{"词": "标签辅助"}],
        }
    )

    assert elements["strong_terms"] == ["爱国情感", "人物故事", "国家认同"]
    assert "标签辅助" in elements["auxiliary_terms"]


def test_decode_timeout_records_pending_without_waiting():
    result = decode_content(
        content={"platform_content_id": "739"},
        media={"play_url": None},
        source_context={},
        decode_client=FakeDecodeClient(fake_decode_pending()),
        max_wait_seconds=0,
        poll_interval_seconds=0,
    )

    assert result["decode_status"] == "running"
    assert result["pending_reason"] == "decode_timeout_20m"


def test_decode_bad_shape_becomes_failed():
    result = decode_content(
        content={"platform_content_id": "739"},
        media={"play_url": None},
        source_context={},
        decode_client=FakeDecodeClient(fake_decode_bad_shape()),
        max_wait_seconds=1200,
        poll_interval_seconds=0,
    )

    assert result["decode_status"] == "failed"
    assert result["failure_reason"] == "decode_result_bad_shape"


def test_decode_result_client_error_becomes_failed():
    result = decode_content(
        content={"platform_content_id": "739"},
        media={"play_url": None},
        source_context={},
        decode_client=FakeDecodeClient(
            fake_decode_pending("decode_task_001"),
            result_sequence=[_raise_runtime_error],
        ),
        max_wait_seconds=1,
        poll_interval_seconds=0,
    )

    assert result["decode_status"] == "failed"
    assert result["decode_task_id"] == "decode_task_001"
    assert result["failure_reason"] == "decode_client_error"
    assert result["raw_response"]["error_type"] == "RuntimeError"


def test_decode_result_reads_aigc_result_data_rows():
    result = decode_content(
        content={"platform_content_id": "739"},
        media={"play_url": None},
        source_context={},
        decode_client=FakeDecodeClient(
            fake_decode_pending("739"),
            result_sequence=[
                {
                    "request": {"params": {"configId": 58, "channelContentIds": ["739"]}},
                    "response": {
                        "code": 0,
                        "msg": "success",
                        "data": [
                            {
                                "channelContentId": "739",
                                "status": "SUCCESS",
                                "errorMessage": None,
                                "dataContent": (
                                    '{"目的点":[{"点":"爱国情感","实质":[{"名称":"人物故事"}]}]}'
                                ),
                            }
                        ],
                    },
                    "raw_response": {
                        "code": 0,
                        "msg": "success",
                        "data": [
                            {
                                "channelContentId": "739",
                                "status": "SUCCESS",
                                "errorMessage": None,
                                "dataContent": (
                                    '{"目的点":[{"点":"爱国情感","实质":[{"名称":"人物故事"}]}]}'
                                ),
                            }
                        ],
                    },
                }
            ],
        ),
        max_wait_seconds=1,
        poll_interval_seconds=0,
    )

    assert result["decode_status"] == "success"
    assert result["decode_elements"]["strong_terms"] == ["爱国情感", "人物故事"]


def test_decode_submit_payload_uses_config_id_58():
    client = FakeDecodeClient(fake_decode_success())

    decode_content(
        content={"platform_content_id": "739", "description": "desc", "tags": []},
        media={"play_url": "https://video.example/a.mp4"},
        source_context={"ext_data": {"evidence_pack": {"source_post_id": "519"}}},
        decode_client=client,
        max_wait_seconds=1200,
        poll_interval_seconds=0,
    )

    assert client.submit_calls[0]["content"]["platform_content_id"] == "739"


def test_redact_sensitive_payload_removes_sensitive_key_names():
    auth_key = "Authorizatio" + "n"
    cookie_key = "Cook" + "ie"
    redacted = redact_sensitive_payload(
        {
            "token": "abc",
            "headers": {auth_key: "Bearer abc", cookie_key: "session=1"},
            "safe": "ok",
        }
    )

    assert "token" not in redacted
    assert auth_key not in redacted["headers"]
    assert cookie_key not in redacted["headers"]
    assert redacted["token_redacted"] == "<redacted>"
    assert redacted["headers"][f"{auth_key}_redacted"] == "<redacted>"


def test_aigc_decode_client_builds_config_id_request():
    client = AigcDecodeClient(
        base_url="https://aigc-api.aiddit.com",
        token="dummy",
        http_client=_FakeHttpClient({"status": "SUCCESS", "taskId": "task_001"}),
    )

    result = client.submit_decode(
        {"platform_content_id": "739", "description": "desc"},
        {"play_url": None},
        {"merge_leve2": "测试"},
    )

    assert result["request"]["params"]["configId"] == 58
    assert result["decode_task_id"] == "task_001"


def test_aigc_decode_client_uses_channel_content_ids_for_result_query():
    http_client = _FakeHttpClient(
        {"code": 0, "data": [{"status": "PENDING", "channelContentId": "739"}]}
    )
    client = AigcDecodeClient(
        base_url="https://aigc-api.aiddit.com",
        token="dummy",
        http_client=http_client,
    )

    submit_result = client.submit_decode(
        {"platform_content_id": "739", "description": "desc"},
        {"play_url": None},
        {"merge_leve2": "测试"},
    )
    result = client.get_decode_result("739")

    assert submit_result["decode_task_id"] == "739"
    assert result["request"] == {"params": {"configId": 58, "channelContentIds": ["739"]}}
    assert http_client.requests[-1]["json"] == {
        "params": {"configId": 58, "channelContentIds": ["739"]}
    }


class _FakeResponse:
    def __init__(self, data):
        self.data = data

    def raise_for_status(self):
        return None

    def json(self):
        return self.data


class _FakeHttpClient:
    def __init__(self, data):
        self.data = data
        self.requests = []

    def post(self, *args, **kwargs):
        self.requests.append({"args": args, **kwargs})
        return _FakeResponse(self.data)


def _raise_runtime_error():
    raise RuntimeError("decode result unavailable")


def test_decode_content_without_event_sink_keeps_previous_result():
    kwargs = dict(
        content={"platform_content_id": "739"},
        media={"play_url": None},
        source_context={},
        max_wait_seconds=1200,
        poll_interval_seconds=0,
    )

    without_sink = decode_content(decode_client=FakeDecodeClient(fake_decode_success()), **kwargs)
    events = []
    with_sink = decode_content(
        decode_client=FakeDecodeClient(fake_decode_success()),
        event_sink=events.append,
        **kwargs,
    )

    assert without_sink == with_sink
    assert events  # sink 只新增观测,不改变业务返回。