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

from content_agent.business_modules import platform_access
from content_agent.errors import ContentAgentError, ErrorCode
from content_agent.integrations.crawapi_http import CrawapiBusinessError


class DuplicateContentClient:
    def search(self, search_query):
        return [
            {
                "content_discovery_id": f"{search_query['search_query_id']}_content_001",
                "search_query_id": search_query["search_query_id"],
                "platform_content_id": "7601814454925298994",
                "description": "重复内容",
            }
        ]


def test_platform_access_deduplicates_same_content_across_search_queries():
    search_queries = [
        {
            "search_query_id": "q_001",
            "search_query": "对比分析",
            "search_query_generation_method": "item_single",
        },
        {
            "search_query_id": "q_002",
            "search_query": "对比分析 人物故事",
            "search_query_generation_method": "item_single",
        },
    ]

    result = platform_access.run(search_queries, DuplicateContentClient())
    results = result["platform_results"]

    assert len(results) == 1
    assert results[0]["search_query_id"] == "q_001"
    assert results[0]["search_query"] == "对比分析"
    assert results[0]["matched_search_query_ids"] == ["q_001", "q_002"]
    assert results[0]["matched_search_queries"] == ["对比分析", "对比分析 人物故事"]


class UniqueContentClient:
    def search(self, search_query):
        return [
            {
                "content_discovery_id": f"{search_query['search_query_id']}_content_001",
                "search_query_id": search_query["search_query_id"],
                "platform_content_id": f"platform_{search_query['search_query_id']}",
                "description": search_query["search_query"],
            }
        ]


def test_platform_access_preserves_llm_variant_generation_method():
    search_queries = [
        {
            "search_query_id": "q_001",
            "search_query": "对比分析",
            "search_query_generation_method": "item_single",
        },
        {
            "search_query_id": "q_002",
            "search_query": "人物叙事素材",
            "search_query_generation_method": "llm_variant",
            "llm_variant_of": "q_001",
        },
    ]

    results = platform_access.run(search_queries, UniqueContentClient())["platform_results"]

    assert [result["search_query_generation_method"] for result in results] == [
        "item_single",
        "llm_variant",
    ]
    assert results[1]["search_query_id"] == "q_002"
    assert results[1]["search_query"] == "人物叙事素材"


class FailingThenSuccessfulClient:
    def search(self, search_query):
        if search_query["search_query_id"] == "q_001":
            raise RuntimeError("temporary platform failure")
        return [
            {
                "content_discovery_id": f"{search_query['search_query_id']}_content_001",
                "search_query_id": search_query["search_query_id"],
                "platform_content_id": "7601814454925298995",
                "description": "成功内容",
            }
        ]


def test_platform_access_keeps_running_after_single_query_failure():
    search_queries = [
        {
            "search_query_id": "q_001",
            "search_query": "接口失败",
            "search_query_generation_method": "item_single",
        },
        {
            "search_query_id": "q_002",
            "search_query": "正常召回",
            "search_query_generation_method": "llm_variant",
        },
    ]

    result = platform_access.run(search_queries, FailingThenSuccessfulClient())

    assert [item["platform_content_id"] for item in result["platform_results"]] == [
        "7601814454925298995"
    ]
    assert result["query_failures"][0]["search_query_id"] == "q_001"
    assert result["query_failures"][0]["error_code"] == ErrorCode.PLATFORM_REQUEST_FAILED.value


class AlwaysFailingClient:
    def search(self, search_query):
        raise RuntimeError("platform unavailable")


def test_platform_access_fails_run_when_all_queries_fail():
    search_queries = [
        {
            "search_query_id": "q_001",
            "search_query": "接口失败",
            "search_query_generation_method": "item_single",
        },
        {
            "search_query_id": "q_002",
            "search_query": "仍然失败",
            "search_query_generation_method": "llm_variant",
        }
    ]

    try:
        platform_access.run(search_queries, AlwaysFailingClient())
    except ContentAgentError as exc:
        assert exc.error_code == ErrorCode.PLATFORM_REQUEST_FAILED
        assert [failure["search_query_id"] for failure in exc.detail["query_failures"]] == [
            "q_001",
            "q_002",
        ]
        for failure in exc.detail["query_failures"]:
            assert failure["status"] == "failed"
            assert failure["error_code"] == ErrorCode.PLATFORM_REQUEST_FAILED.value
            assert failure["message"] == "platform query failed"
            assert failure["error_detail"]["exception_type"] == "RuntimeError"
    else:
        raise AssertionError("expected platform request failure")


class RateLimitedClient:
    def search(self, search_query):
        raise ContentAgentError(
            ErrorCode.PLATFORM_RATE_LIMITED,
            "crawapi keyword_search failed: rate_limited",
            {"operation": "keyword_search", "status_code": 429},
        )


class HealthyClient:
    def search(self, search_query):
        return [
            {
                "content_discovery_id": f"{search_query['search_query_id']}_content_001",
                "search_query_id": search_query["search_query_id"],
                "platform_content_id": "7601814454925298001",
                "description": "正常内容",
            }
        ]


class RuntimeErrorClient:
    def search(self, search_query):
        raise RuntimeError("crawapi keyword_search failed: HTTP 500")


class SplitClient:
    """First query rate limited, second query succeeds."""

    def search(self, search_query):
        if search_query["search_query_id"] == "q_001":
            return RateLimitedClient().search(search_query)
        return HealthyClient().search(search_query)


def test_platform_access_preserves_rate_limited_error_code():
    search_queries = [
        {"search_query_id": "q_001", "search_query": "被限流", "search_query_generation_method": "item_single"},
        {"search_query_id": "q_002", "search_query": "正常", "search_query_generation_method": "item_single"},
    ]

    result = platform_access.run(search_queries, SplitClient())

    failure = result["query_failures"][0]
    assert failure["search_query_id"] == "q_001"
    assert failure["error_code"] == ErrorCode.PLATFORM_RATE_LIMITED.value
    assert failure["message"] == "crawapi keyword_search failed: rate_limited"
    assert failure["error_detail"]["operation"] == "keyword_search"
    assert len(result["platform_results"]) == 1


def test_platform_access_counts_runtime_error_as_platform_request_failed():
    search_queries = [
        {"search_query_id": "q_001", "search_query": "普通失败", "search_query_generation_method": "item_single"},
        {"search_query_id": "q_002", "search_query": "正常", "search_query_generation_method": "item_single"},
    ]

    class MixedClient:
        def search(self, search_query):
            if search_query["search_query_id"] == "q_001":
                raise RuntimeError("crawapi keyword_search failed: HTTP 500")
            return HealthyClient().search(search_query)

    result = platform_access.run(search_queries, MixedClient())

    failure = result["query_failures"][0]
    assert failure["error_code"] == ErrorCode.PLATFORM_REQUEST_FAILED.value
    assert failure["error_detail"]["exception_type"] == "RuntimeError"
    assert (
        failure["error_detail"]["exception_message"]
        == "crawapi keyword_search failed: HTTP 500"
    )


def test_platform_access_preserves_structured_crawapi_business_detail():
    search_queries = [
        {"search_query_id": "q_001", "search_query": "历史人物评价", "search_query_generation_method": "item_single"},
        {"search_query_id": "q_002", "search_query": "正常", "search_query_generation_method": "item_single"},
    ]

    class MixedClient:
        def search(self, search_query):
            if search_query["search_query_id"] == "q_001":
                raise CrawapiBusinessError(
                    "crawapi keyword_search failed: business_error code=10000 message=未知错误",
                    {
                        "operation": "keyword_search",
                        "business_code": "10000",
                        "business_message": "未知错误",
                        "business_data": {"trace_id": "trace-001", "upstream": "opaque"},
                        "trace_refs": {"x-request-id": "req-001"},
                    },
                )
            return HealthyClient().search(search_query)

    result = platform_access.run(search_queries, MixedClient())

    failure = result["query_failures"][0]
    assert failure["error_code"] == ErrorCode.PLATFORM_REQUEST_FAILED.value
    assert failure["message"] == "platform query failed"
    assert failure["error_detail"]["exception_type"] == "CrawapiBusinessError"
    assert failure["error_detail"]["operation"] == "keyword_search"
    assert failure["error_detail"]["business_code"] == "10000"
    assert failure["error_detail"]["business_data"] == {
        "trace_id": "trace-001",
        "upstream": "opaque",
    }
    assert failure["error_detail"]["trace_refs"] == {"x-request-id": "req-001"}


def test_platform_access_records_runtime_error_cause_message():
    class CausedFailureClient:
        def search(self, search_query):
            cause = ValueError("upstream body was not json")
            raise RuntimeError("crawapi keyword_search failed: bad_json") from cause

    search_queries = [
        {
            "search_query_id": "q_001",
            "search_query": "坏响应",
            "search_query_generation_method": "item_single",
        }
    ]

    try:
        platform_access.run(search_queries, CausedFailureClient())
    except ContentAgentError as exc:
        failure = exc.detail["query_failures"][0]
        assert failure["error_detail"]["exception_type"] == "RuntimeError"
        assert (
            failure["error_detail"]["exception_message"]
            == "crawapi keyword_search failed: bad_json"
        )
        assert failure["error_detail"]["cause_exception_type"] == "ValueError"
        assert failure["error_detail"]["cause_message"] == "upstream body was not json"
    else:
        raise AssertionError("expected platform request failure")


def test_platform_access_accepts_search_only_client():
    class SearchOnlyClient:
        def search(self, search_query):
            return [
                {
                    "content_discovery_id": f"{search_query['search_query_id']}_content_001",
                    "search_query_id": search_query["search_query_id"],
                    "platform_content_id": "ks_001",
                    "description": "搜索协议只要求 search",
                }
            ]

    search_queries = [
        {
            "search_query_id": "q_001",
            "search_query": "快手祝福",
            "search_query_generation_method": "item_single",
        }
    ]

    result = platform_access.run(search_queries, SearchOnlyClient())

    assert result["platform_results"][0]["platform_content_id"] == "ks_001"
    assert result["query_failures"] == []