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

"""V3-M1A: shared crawapi HTTP base unit tests."""

from __future__ import annotations

import httpx
import pytest

from content_agent.errors import ContentAgentError, ErrorCode
from content_agent.integrations.crawapi_http import (
    CrawapiBusinessError,
    RateLimiter,
    is_rate_limit_business_error,
    post_crawapi_json,
)


class FakeHttpClient:
    def __init__(self, responses):
        self.responses = list(responses)
        self.requests = []

    def post(self, url, json, headers, timeout):
        self.requests.append({"url": url, "json": json})
        response = self.responses.pop(0)
        if isinstance(response, Exception):
            raise response
        return response


def _response(status_code, data, headers=None):
    return httpx.Response(
        status_code,
        json=data,
        headers=headers,
        request=httpx.Request("POST", "http://crawapi.test/x"),
    )


def _text_response(status_code, text):
    return httpx.Response(
        status_code,
        text=text,
        headers={"content-type": "text/plain"},
        request=httpx.Request("POST", "http://crawapi.test/x"),
    )


def _post(responses, **kwargs):
    return post_crawapi_json(
        http_client=FakeHttpClient(responses),
        base_url="http://crawapi.test/",
        path="x",
        payload={},
        operation="probe",
        timeout_seconds=60.0,
        business_codes=kwargs.get("business_codes", set()),
        rate_limiter=kwargs.get("rate_limiter"),
        rate_limit_bucket=kwargs.get("rate_limit_bucket"),
    )


def test_rate_limiter_waits_min_interval_between_same_bucket():
    clock = {"now": 0.0}
    sleeps: list[float] = []
    limiter = RateLimiter(
        min_interval_seconds=30.0,
        now_fn=lambda: clock["now"],
        sleep_fn=lambda s: (sleeps.append(s), clock.__setitem__("now", clock["now"] + s)),
    )
    limiter.wait("b")
    limiter.wait("b")
    assert sleeps == [30.0]


def test_rate_limiter_supports_random_interval_between_same_bucket():
    clock = {"now": 0.0}
    sleeps: list[float] = []
    limiter = RateLimiter(
        min_interval_seconds=10.0,
        max_interval_seconds=12.0,
        now_fn=lambda: clock["now"],
        sleep_fn=lambda s: (sleeps.append(s), clock.__setitem__("now", clock["now"] + s)),
        random_fn=lambda lo, hi: (lo + hi) / 2,
    )
    limiter.wait("b")
    limiter.wait("b")
    assert sleeps == [11.0]


def test_http_429_maps_to_platform_rate_limited():
    with pytest.raises(ContentAgentError) as exc:
        _post([_response(429, {"msg": "slow down"})])
    assert exc.value.error_code == ErrorCode.PLATFORM_RATE_LIMITED


def test_message_token_maps_to_platform_rate_limited():
    with pytest.raises(ContentAgentError) as exc:
        _post([_response(200, {"code": 50000, "msg": "请求频繁"})])
    assert exc.value.error_code == ErrorCode.PLATFORM_RATE_LIMITED


def test_bad_response_non_dict_raises_runtime_error():
    with pytest.raises(RuntimeError, match="bad_response"):
        _post([_response(200, ["not", "a", "dict"])])


def test_http_error_includes_response_summary():
    with pytest.raises(RuntimeError) as exc:
        _post([_text_response(502, "Bad Gateway from crawapi")])

    message = str(exc.value)
    assert "HTTP 502" in message
    assert "Bad Gateway from crawapi" in message


def test_business_error_includes_code_and_message():
    with pytest.raises(RuntimeError) as exc:
        _post([_response(200, {"code": 50001, "msg": "source url expired"})])

    message = str(exc.value)
    assert "business_error" in message
    assert "code=50001" in message
    assert "source url expired" in message


def test_business_error_carries_full_response_detail():
    with pytest.raises(CrawapiBusinessError) as exc:
        post_crawapi_json(
            http_client=FakeHttpClient(
                [
                    _response(
                        200,
                        {
                            "code": 10000,
                            "msg": "未知错误",
                            "request_id": "body-request",
                            "data": {
                                "trace_id": "body-trace",
                                "reason": "upstream opaque failure",
                                "retry_after": 30,
                            },
                        },
                        headers={"x-request-id": "header-request"},
                    )
                ]
            ),
            base_url="http://crawapi.test/",
            path="x",
            payload={
                "keyword": "历史人物评价",
                "cursor": "0",
                "account_id": "should-not-persist",
            },
            operation="keyword_search",
            timeout_seconds=60.0,
            business_codes=set(),
        )

    detail = exc.value.detail
    assert detail["operation"] == "keyword_search"
    assert detail["business_code"] == "10000"
    assert detail["business_message"] == "未知错误"
    assert detail["response_summary"]["code"] == "10000"
    assert detail["response_json_keys"] == ["code", "data", "msg", "request_id"]
    assert detail["business_data"] == {
        "trace_id": "body-trace",
        "reason": "upstream opaque failure",
        "retry_after": 30,
    }
    assert detail["trace_refs"]["request_id"] == "body-request"
    assert detail["trace_refs"]["trace_id"] == "body-trace"
    assert detail["trace_refs"]["x-request-id"] == "header-request"
    assert detail["request_payload_summary"] == {
        "keyword": "历史人物评价",
        "cursor": "0",
        "account_id": "<redacted>",
    }


def test_business_codes_param_classifies_rate_limit():
    assert is_rate_limit_business_error("30005", {}, business_codes={"30005"}) is True
    assert is_rate_limit_business_error("30005", {}, business_codes=set()) is False