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

"""M9E:抖音作者画像 client——请求体(account_id=sec_uid + need_*)、限流 bucket、错误上抛。"""

import httpx
import pytest

from content_agent.errors import ContentAgentError
from content_agent.integrations.douyin import (
    RAW_AUTHOR_ACCOUNT_KEY,
    CrawapiDouyinClient,
    RateLimiter,
)


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/p")
    )


def _client(responses, rate_limiter=None):
    return CrawapiDouyinClient(
        base_url="http://crawapi.test",
        keyword_path="/crawler/dou_yin/keyword",
        portrait_path="/crawler/dou_yin/re_dian_bao/account_fans_portrait",
        http_client=FakeHttpClient(responses),
        rate_limiter=rate_limiter,
    )


def test_portrait_request_body_and_inner_data_unwrap():
    fans = {"age": {"data": {"50-": {"percentage": "30%", "preference": "130"}}}}
    client = _client([_response(200, {"code": 0, "data": {"data": {"fans": fans}}})])
    result = client.fetch_account_fans_portrait("MS4wLjABAAAAsecuid")

    req = client.http_client.requests[0]
    assert req["url"].endswith("/crawler/dou_yin/re_dian_bao/account_fans_portrait")
    assert req["json"][RAW_AUTHOR_ACCOUNT_KEY] == "MS4wLjABAAAAsecuid"
    assert req["json"]["need_age"] is True
    assert result == {"fans": fans}  # 双层 data.data 解包为内层


def test_portrait_does_not_use_rate_limiter_bucket():
    limiter = RateLimiter(min_interval_seconds=0)
    client = _client([_response(200, {"code": 0, "data": {"data": {}}})], rate_limiter=limiter)
    client.fetch_account_fans_portrait("MS4wLjAsec")
    assert limiter._last_call_by_bucket == {}


def test_portrait_business_error_raises():
    client = _client([_response(200, {"code": 100, "msg": "blocked"})])
    with pytest.raises((ContentAgentError, RuntimeError)):
        client.fetch_account_fans_portrait("MS4wLjAsec")