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

from content_agent.business_modules import platform_access
from content_agent.errors import ContentAgentError, ErrorCode


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"