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 (
    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):
    return httpx.Response(
        status_code, json=data, 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=12.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 == [12.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_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