From 46e3880c0bd91a5f5c01df7fe5d89e148cadb069 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B5=85=E5=BF=B5?= Date: Sat, 1 Aug 2026 14:51:46 +0800 Subject: [PATCH] feat: anti-risk-control stealth layer (browser headers, jittered throttle, backoff, volume cap) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Centralize scraping stealth in YunkeClient.request() so automated traffic resembles the normal CRM web app and is paced conservatively to avoid triggering 云客 risk control (pulls M6 rate-limit work forward): - app/config.py: add 6 tunable settings with conservative defaults (request_min_interval=1.0, request_jitter=0.8, max_requests_per_run=0, risk_backoff_{base=2.0, max=60.0, attempts=4}). - app/scraper/base.py: RateLimiter gains optional jitter + a new wait() method (fixed min+jitter per-call sleep); gap-based acquire() is unchanged for backward compat (default jitter=0.0). - app/session/yunke_client.py: every request now carries browser-identical headers (UA/sec-ch-ua/accept/accept-language/referer) merged with the auth headers (per-request overrides win); awaits the shared jittered limiter before sending; enforces a per-run request cap (RequestCapReached); and backs off exponentially on risk signals (HTTP 429 or 200 body carrying 频繁/风控/限制/稍后), distinct from the existing 401 relogin-once path. - tests/conftest.py: zero the throttle env in the test suite so the conservative prod defaults don't slow unit tests. - tests/test_yunke_client_stealth.py: 9 new tests (browser headers, override, limiter spy, 429 backoff, 200-marker backoff, persistent-429 raise, volume cap, unlimited, 401-relogin preserved). TDD red->green. Suite: 171 -> 181 passing. ruff clean. Co-Authored-By: Claude --- yunke-server/app/config.py | 21 ++ yunke-server/app/scraper/base.py | 47 ++- yunke-server/app/session/yunke_client.py | 201 ++++++++++- yunke-server/tests/conftest.py | 19 ++ yunke-server/tests/test_config.py | 18 + .../tests/test_yunke_client_stealth.py | 313 ++++++++++++++++++ 6 files changed, 603 insertions(+), 16 deletions(-) create mode 100644 yunke-server/tests/test_yunke_client_stealth.py diff --git a/yunke-server/app/config.py b/yunke-server/app/config.py index 1d85efe..b188858 100644 --- a/yunke-server/app/config.py +++ b/yunke-server/app/config.py @@ -28,6 +28,27 @@ class Settings(BaseSettings): # `` or ``?token=``. See app/api/auth.py. api_token: str = "" + # --- Anti-risk-control / stealth tuning --------------------------------- + # These are the knobs that pace automated scraping so it resembles the + # normal web app. Defaults are deliberately CONSERVATIVE (slow + jittered): + # roughly one request every ~1.0..1.8s. Loosen these only once we have + # evidence the CRM tolerates a faster pace. See app.session.yunke_client. + # Lower bound on seconds between successive upstream requests. + request_min_interval: float = 1.0 + # Random extra 0..jitter seconds added on top of ``request_min_interval`` + # per request, so the cadence is not a perfectly periodic signature. + request_jitter: float = 0.8 + # Hard cap on the number of upstream requests a single client may send in + # one run. 0 = unlimited; >0 = enforce, then raise RequestCapReached. + max_requests_per_run: int = 0 + # Exponential backoff parameters applied when the CRM signals risk control + # (HTTP 429 or a 200 body carrying a 风控/限流 marker). Delay for retry N + # (0-indexed) is ``risk_backoff_base * 2**N`` clamped to ``risk_backoff_max``. + risk_backoff_base: float = 2.0 + risk_backoff_max: float = 60.0 + # Number of retries attempted after the initial request before giving up. + risk_backoff_attempts: int = 4 + @lru_cache def get_settings() -> Settings: diff --git a/yunke-server/app/scraper/base.py b/yunke-server/app/scraper/base.py index 77bfcf0..875ae0d 100644 --- a/yunke-server/app/scraper/base.py +++ b/yunke-server/app/scraper/base.py @@ -34,11 +34,16 @@ job is just to advance the cursor and stop on ``has_more == False``. from __future__ import annotations import asyncio +import random import time from collections.abc import AsyncIterator, Awaitable, Callable -from typing import Any +from typing import TYPE_CHECKING, Any -from app.session.yunke_client import YunkeClient +if TYPE_CHECKING: + # ``YunkeClient`` is referenced only in the paginator signatures. Kept + # under TYPE_CHECKING so ``app.session.yunke_client`` may import + # :class:`RateLimiter` at runtime without forming an import cycle. + from app.session.yunke_client import YunkeClient __all__ = ["RateLimiter", "paginate_by_cursor", "paginate_by_page"] @@ -175,21 +180,49 @@ async def paginate_by_cursor( class RateLimiter: """Minimal asyncio limiter enforcing a minimum interval between calls. - Call ``await limiter.acquire()`` before each upstream request. The first - call never waits. Successive calls wait just long enough that the gap - between the previous call's dispatch and this one is at least - ``min_interval`` seconds. + Two pacing modes are supported: + + * :meth:`acquire` — gap-based spacing. Call it before each upstream + request; the first call never waits and successive calls wait just long + enough that the gap between dispatches is at least ``min_interval`` + seconds. Used by the list paginators (:func:`paginate_by_page` / + :func:`paginate_by_cursor`). + + * :meth:`wait` — fixed per-call throttle. Sleeps ``min_interval`` plus a + uniform random jitter in ``[0, jitter)`` every time it is awaited. Used + by :class:`app.session.yunke_client.YunkeClient` to pace every upstream + request with a conservative, slightly randomized cadence (the jitter + prevents a perfectly periodic signature that risk control could finger). This is deliberately tiny: no tokens, no concurrency cap, no skew handling — just a floor on inter-call spacing, which is all the yunke scrapers need to stay polite to the CRM. If you need a semaphore or token bucket, that belongs in a future helper, not here. + + ``jitter`` defaults to ``0.0`` so existing callers that pass only + ``min_interval`` keep their exact prior behaviour (gap-based ``acquire`` + never consults ``jitter``). """ - def __init__(self, min_interval: float = 0.0) -> None: + def __init__(self, min_interval: float = 0.0, jitter: float = 0.0) -> None: self._min_interval = float(min_interval) + self._jitter = float(jitter) self._last: float = 0.0 + async def wait(self) -> None: + """Sleep ``min_interval`` plus a random ``[0, jitter)`` every call. + + Unlike :meth:`acquire` this is not gap-aware — it is a flat, slightly + randomized per-call delay, which is what the yunke client wants ahead + of each upstream request. With ``min_interval == 0`` and + ``jitter == 0`` it is a no-op. + """ + delay = self._min_interval + if self._jitter > 0: + delay += random.uniform(0.0, self._jitter) + if delay > 0: + await asyncio.sleep(delay) + async def acquire(self) -> None: """Block until at least ``min_interval`` has elapsed since the last call.""" if self._min_interval <= 0: diff --git a/yunke-server/app/session/yunke_client.py b/yunke-server/app/session/yunke_client.py index 6c3f1ec..6476002 100644 --- a/yunke-server/app/session/yunke_client.py +++ b/yunke-server/app/session/yunke_client.py @@ -7,9 +7,18 @@ CRM (see design doc §6.1 会话层). It wraps an :class:`httpx.AsyncClient` and session is present, or when constructed with credentials. * Injects the 5 known auth headers (``userid`` / ``company`` / ``source`` / ``channel`` / ``content-type``) on every request. +* **Anti-risk-control / stealth** (see §6.1 风控规避): every request also + carries browser-identical headers (UA / sec-ch-ua / accept / accept-language + / referer) so automated traffic looks like the normal web app; a shared + jittered :class:`app.scraper.base.RateLimiter` paces each request; a + per-run volume cap guards against runaway loops; and risk-control signals + (HTTP 429 or a 200 body carrying a 风控/限流 marker) trigger exponential + backoff + retry. * Detects HTTP 401 (or a body-level session-expired marker), re-logs-in *once*, and retries. A persistent 401 raises after the single retry — no - infinite loop. + infinite loop. The 401 path is deliberately distinct from the 429/limit + path: a 401 is a session death (-> relogin), a 429 is throttling + (-> backoff); they are never confused. Phase-0 note: the session is held in-memory only. Future work will persist the encrypted session blob to ``YkAccount.session_blob_enc`` (see @@ -19,11 +28,14 @@ session instead of always logging in fresh. from __future__ import annotations +import asyncio from collections.abc import Mapping from typing import Any import httpx +from app.config import get_settings +from app.scraper.base import RateLimiter from app.session.login import ( LoginResult, get_secure_key, @@ -38,6 +50,46 @@ _STATIC_AUTH_HEADERS: dict[str, str] = { "content-type": "application/json", } +# Browser-identical headers captured from the real CRM web app. Sent on EVERY +# request (merged ahead of the auth headers; per-request overrides win) so the +# automated client presents the same fingerprint as a normal Chrome-on-Windows +# user. ``referer`` defaults to the app root; callers can override it per call +# when they know the precise cms path. See §6.1 风控规避. +_BROWSER_HEADERS: dict[str, str] = { + "user-agent": ( + "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " + "(KHTML, like Gecko) Chrome/150.0.0.0 Safari/537.36" + ), + "accept": "application/json, text/plain, */*", + "accept-language": "zh-CN", + "sec-ch-ua": '"Not;A=Brand";v="8", "Chromium";v="150", "Google Chrome";v="150"', + "sec-ch-ua-mobile": "?0", + "sec-ch-ua-platform": '"Windows"', + "referer": "https://crm2.yunkecn.com/", +} + +# Substrings that, when found in a HTTP 200 response body, indicate the CRM is +# throttling / risk-controlling us even though it returned 200 (a common +# Chinese-CRM idiom). Kept deliberately conservative — the explicit markers +# observed in 风控 messages. See §6.1 风控规避. +_RISK_BODY_MARKERS: tuple[str, ...] = ("频繁", "风控", "限制", "稍后") + + +class RequestCapReached(Exception): + """Raised when a run exceeds ``max_requests_per_run`` upstream calls. + + Carries the cap and the count so callers can surface a useful message. + The exception is a control-flow signal for orchestrators (a run that hits + the cap should stop scheduling more work, not crash the process). + """ + + def __init__(self, cap: int, count: int) -> None: + self.cap = cap + self.count = count + super().__init__( + f"request cap reached: {count} requests issued, cap is {cap}" + ) + class YunkeClient: """Authenticated async HTTP client for the yunke CRM API.""" @@ -52,6 +104,12 @@ class YunkeClient: company: str | None = None, cookies: Mapping[str, str] | None = None, client: httpx.AsyncClient | None = None, + # --- stealth / anti-risk-control knobs (override Settings defaults) --- + rate_limiter: RateLimiter | None = None, + max_requests_per_run: int | None = None, + backoff_base: float | None = None, + backoff_max: float | None = None, + backoff_attempts: int | None = None, ) -> None: """Construct a client either from credentials or an existing session. @@ -68,6 +126,10 @@ class YunkeClient: :class:`httpx.AsyncClient` (e.g. for custom transports). When omitted a new client bound to ``base_url`` is created and owned by this instance (closed by :meth:`aclose`). + + Stealth knobs default to the conservative values declared on + :class:`app.config.Settings`; pass any of them explicitly to override + (tests inject a no-op ``rate_limiter`` and small backoff to stay fast). """ has_creds = username is not None and password is not None has_session = userid is not None and company is not None @@ -93,6 +155,28 @@ class YunkeClient: for name, value in (cookies or {}).items(): self._client.cookies.set(name, value) + # --- stealth configuration ---------------------------------------- + # Defaults come from Settings; explicit args win. Tests pass explicit + # no-op values so the suite doesn't sleep on the conservative prod + # defaults. + s = get_settings() + self._rate_limiter = ( + rate_limiter + if rate_limiter is not None + else RateLimiter( + min_interval=s.request_min_interval, jitter=s.request_jitter + ) + ) + self._max_requests = ( + s.max_requests_per_run if max_requests_per_run is None else max_requests_per_run + ) + self._backoff_base = s.risk_backoff_base if backoff_base is None else backoff_base + self._backoff_max = s.risk_backoff_max if backoff_max is None else backoff_max + self._backoff_attempts = ( + s.risk_backoff_attempts if backoff_attempts is None else backoff_attempts + ) + self._request_count = 0 + # -- properties --------------------------------------------------------- @property @@ -104,6 +188,11 @@ class YunkeClient: def logged_in(self) -> bool: return self.login_result is not None + @property + def request_count(self) -> int: + """Number of upstream requests issued through this client so far.""" + return self._request_count + # -- login -------------------------------------------------------------- async def login(self) -> LoginResult: @@ -133,20 +222,89 @@ class YunkeClient: path: str, *, headers: Mapping[str, str] | None = None, + **kwargs: Any, + ) -> httpx.Response: + """Send an authed request with the full stealth layer applied. + + Per request: + + 1. **Volume cap** — if ``max_requests_per_run > 0`` and the counter + has reached it, raise :class:`RequestCapReached` (fail fast, before + pacing or sending anything). + 2. **Throttle** — await the shared jittered :class:`RateLimiter` once, + so successive calls are spaced at the conservative cadence. + 3. **Send + 401 handling** — :meth:`_send_with_relogin` performs the + actual request (with browser + auth headers) and re-logs-in once on + a 401/session-expired signal, exactly as before. + 4. **Backoff on risk-control signals** — if the response is a 429 or a + 200 carrying a 风控/限流 marker, sleep + ``backoff_base * 2**attempt`` (capped at ``backoff_max``) and retry + up to ``backoff_attempts`` times; raise on persistent throttling. + This is intentionally separate from the 401 path. + + Per-request ``headers`` override both the browser and the auth + defaults for this call only (per-request values win). + """ + # 1. Volume cap. + if self._max_requests > 0 and self._request_count >= self._max_requests: + raise RequestCapReached(cap=self._max_requests, count=self._request_count) + self._request_count += 1 + + # 2. Throttle ahead of every request. + await self._rate_limiter.wait() + + user_headers = dict(headers or {}) + + # 3 + 4. Send, then back off on risk-control signals. The 401-relogin + # behaviour lives inside ``_send_with_relogin`` and is unaffected by + # this loop — a 401 is never treated as a rate-limit signal. + last_resp: httpx.Response | None = None + for attempt in range(self._backoff_attempts + 1): + resp = await self._send_with_relogin(method, path, user_headers, **kwargs) + if not _is_rate_limited(resp): + return resp + last_resp = resp + if attempt >= self._backoff_attempts: + break # retries exhausted -> raise below + delay = min( + self._backoff_base * (2 ** attempt), self._backoff_max + ) + await asyncio.sleep(delay) + + # Persistent risk-control signal after all retries — surface it. + assert last_resp is not None # loop ran at least once + raise httpx.HTTPStatusError( + "yunke risk-control signal persisted after" + f" {self._backoff_attempts} backoff retries", + request=last_resp.request, + response=last_resp, + ) + + async def _send_with_relogin( + self, + method: str, + path: str, + user_headers: Mapping[str, str], + *, _retried: bool = False, **kwargs: Any, ) -> httpx.Response: - """Send an authed request, auto-logging-in and re-logging-in on 401. + """Perform one actual send with browser + auth headers and 401-relogin. - Per-request ``headers`` override the auth defaults for this call - only. ``_retried`` is an internal guard against infinite re-login - loops and must not be set by callers. + Auto-logs-in on the first call when no session is present. On a + session-expired signal (401 or body code 401) re-logs-in *once* and + retries; a persistent 401 raises so a dead session cannot loop. This + preserves the pre-stealth behaviour verbatim — the only addition is + the browser headers merged into the outbound request. """ if self.login_result is None: await self.login() - user_headers = dict(headers or {}) - merged = {**self._auth_headers(), **user_headers} + merged = { + **_BROWSER_HEADERS, + **self._auth_headers(), + **user_headers, + } resp = await self._client.request(method, path, headers=merged, **kwargs) @@ -160,8 +318,8 @@ class YunkeClient: response=resp, ) await self.login() - return await self.request( - method, path, headers=user_headers, _retried=True, **kwargs + return await self._send_with_relogin( + method, path, user_headers, _retried=True, **kwargs ) return resp @@ -198,6 +356,9 @@ def _is_session_expired(resp: httpx.Response) -> bool: idiom) -> yes. We deliberately stay narrow here — only the explicit auth-failed code, not generic non-zero codes, to avoid re-login storms on ordinary business errors. + + Note: this is deliberately disjoint from :func:`_is_rate_limited`. A 401 + is relogin territory; a 429/limit is backoff territory. """ if resp.status_code == 401: return True @@ -206,3 +367,25 @@ def _is_session_expired(resp: httpx.Response) -> bool: except (ValueError, httpx.DecodingError): return False return isinstance(body, dict) and body.get("code") in (401, "401") + + +def _is_rate_limited(resp: httpx.Response) -> bool: + """Heuristic: does this response indicate risk-control / throttling? + + Triggers backoff when: + + * HTTP 429 (the canonical "too many requests" status), OR + * HTTP 200 with a body whose text contains any of the conservative + risk-control markers (频繁/风控/限制/稍后). Many Chinese CRMs signal + throttling inline in a 200 envelope rather than via a proper status. + + Deliberately conservative: we do NOT flag generic non-zero business codes + (those are ordinary errors, not throttling) and we never flag 401 (that + is session-expiry, handled by :func:`_is_session_expired` + relogin). + """ + if resp.status_code == 429: + return True + if resp.status_code != 200: + return False + text = resp.text + return any(marker in text for marker in _RISK_BODY_MARKERS) diff --git a/yunke-server/tests/conftest.py b/yunke-server/tests/conftest.py index 2d44b35..5fcc42f 100644 --- a/yunke-server/tests/conftest.py +++ b/yunke-server/tests/conftest.py @@ -15,3 +15,22 @@ def reset_settings_cache(): """ yield get_settings.cache_clear() + + +@pytest.fixture(autouse=True) +def _disable_request_throttle(monkeypatch): + """Zero out the conservative request-pace settings for the test suite. + + Production defaults (``request_min_interval=1.0``, ``request_jitter=0.8``) + would make every mocked request sleep ~1-1.8s and balloon the suite's + runtime. In tests we want the *behaviour* (header stealth, backoff, cap) + without the wall-clock cost, so we pin the throttle env vars to 0. The + defaults themselves are still asserted directly via ``Settings`` field + introspection in ``test_config.py``. The cache is cleared on either side + so the zeroed env is picked up the first time a test reads settings. + """ + get_settings.cache_clear() + monkeypatch.setenv("REQUEST_MIN_INTERVAL", "0") + monkeypatch.setenv("REQUEST_JITTER", "0") + yield + get_settings.cache_clear() diff --git a/yunke-server/tests/test_config.py b/yunke-server/tests/test_config.py index 699ca6f..09f3584 100644 --- a/yunke-server/tests/test_config.py +++ b/yunke-server/tests/test_config.py @@ -67,3 +67,21 @@ def test_settings_constructable_without_env(): s = Settings() assert isinstance(s.db_url, str) assert s.encryption_key == "" + + +def test_stealth_defaults_are_conservative(): + """The anti-risk-control settings declare conservative defaults. + + Asserted via Pydantic field-introspection rather than ``Settings()`` so it + is immune to the test-env throttle-zeroing in conftest (which intentionally + pins ``REQUEST_MIN_INTERVAL``/``REQUEST_JITTER`` to 0 for suite speed). + Production deployments that don't set these env vars get the slow, + jittered defaults declared here. + """ + f = Settings.model_fields + assert f["request_min_interval"].default == 1.0 + assert f["request_jitter"].default == 0.8 + assert f["max_requests_per_run"].default == 0 + assert f["risk_backoff_base"].default == 2.0 + assert f["risk_backoff_max"].default == 60.0 + assert f["risk_backoff_attempts"].default == 4 diff --git a/yunke-server/tests/test_yunke_client_stealth.py b/yunke-server/tests/test_yunke_client_stealth.py new file mode 100644 index 0000000..e742ff3 --- /dev/null +++ b/yunke-server/tests/test_yunke_client_stealth.py @@ -0,0 +1,313 @@ +"""Tests for the anti-risk-control / stealth layer in :class:`YunkeClient`. + +These exercise the four stealth concerns that ``request()`` now centralizes: + +* **Browser-identical headers** on every request (UA / sec-ch-ua / accept + / accept-language / referer) merged with the existing auth headers, with + per-request overrides winning. +* A shared jittered :class:`RateLimiter` awaited ahead of each upstream + request. +* **Exponential backoff + retry** on risk-control signals (HTTP 429, or a 200 + body carrying a 风控/限流 marker such as 频繁/风控/限制/稍后) — distinct from + 401 session-expiry, which keeps its existing relogin-once behaviour. +* A hard per-run request cap that raises :class:`RequestCapReached`. + +Every test injects explicit stealth params (no-op throttle, small backoff) so +the suite stays fast and deterministic regardless of the process-wide +:class:`Settings` defaults (which are deliberately slow in production). +""" + +from __future__ import annotations + +from unittest.mock import AsyncMock + +import httpx +import pytest +import respx +from cryptography.hazmat.primitives.asymmetric import rsa + +from app.scraper.base import RateLimiter +from app.session.yunke_client import RequestCapReached, YunkeClient + +BASE = "https://crm2.yunkecn.com" +USERID = "a" * 32 +COMPANY = "8m0atd" + +DATA_PATH = "/wechatweb/wxFriends/friend_list/es" +DATA_URL = f"{BASE}{DATA_PATH}" + +# The exact browser-captured header values the client must reproduce. +UA = ( + "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " + "(KHTML, like Gecko) Chrome/150.0.0.0 Safari/537.36" +) +SEC_CH_UA = '"Not;A=Brand";v="8", "Chromium";v="150", "Google Chrome";v="150"' + +# Backoff schedule uses base=2.0 so expected delays are 2.0, 4.0, 8.0, ... +BACKOFF_BASE = 2.0 + +# A real RSA keypair backing the mocked ``getSecureKey`` endpoint so the real +# RSA step in ``pc_login`` has a valid key to encrypt against (mirrors the +# pattern in tests/test_yunke_client.py). Generated once at import. +_TEST_PRIVATE_KEY = rsa.generate_private_key(public_exponent=65537, key_size=2048) + + +def _secure_key_from_test_public_key() -> dict[str, str]: + nums = _TEST_PRIVATE_KEY.public_key().public_numbers() + return { + "public_exponent": format(nums.e, "x"), + "modulus": format(nums.n, "x"), + } + + +_TEST_SECURE_KEY = _secure_key_from_test_public_key() + + +def _patch_backoff_sleep(monkeypatch, sink: list[float]) -> None: + """Replace ``asyncio.sleep`` in yunke_client with an instant recorder.""" + + async def fake_sleep(delay: float) -> None: + sink.append(delay) + + monkeypatch.setattr("app.session.yunke_client.asyncio.sleep", fake_sleep) + + +@pytest.fixture() +async def client(): + """A logged-in (session-mode) client with a no-op throttle. + + Session mode (userid + company) skips the login flow entirely. Tests that + need relogin construct their own credentials-mode client inline. + """ + c = YunkeClient( + base_url=BASE, + userid=USERID, + company=COMPANY, + rate_limiter=RateLimiter(min_interval=0.0, jitter=0.0), + ) + try: + yield c + finally: + await c.aclose() + + +# --------------------------------------------------------------------------- +# 1. Browser-identical headers on every request +# --------------------------------------------------------------------------- + + +@respx.mock +async def test_browser_headers_present_on_every_request(client: YunkeClient): + route = respx.post(DATA_URL).mock( + return_value=httpx.Response(200, json={"ok": True}) + ) + + await client.post(DATA_PATH, json={"pageNum": 1}) + + sent = route.calls.last.request.headers + assert sent["user-agent"] == UA + assert sent["accept"] == "application/json, text/plain, */*" + assert sent["accept-language"] == "zh-CN" + assert sent["sec-ch-ua"] == SEC_CH_UA + assert sent["sec-ch-ua-mobile"] == "?0" + assert sent["sec-ch-ua-platform"] == '"Windows"' + assert sent["referer"] == "https://crm2.yunkecn.com/" + # The existing auth headers still ride along unchanged. + assert sent["source"] == "yunkepc" + assert sent["channel"] == "4" + assert sent["userid"] == USERID + assert sent["company"] == COMPANY + assert sent["content-type"] == "application/json" + + +@respx.mock +async def test_per_request_header_overrides_browser_defaults(client: YunkeClient): + """A caller-supplied header wins over the browser + auth defaults.""" + route = respx.post(DATA_URL).mock( + return_value=httpx.Response(200, json={"ok": True}) + ) + + await client.post( + DATA_PATH, json={}, headers={"referer": "https://crm2.yunkecn.com/cms/x"} + ) + + assert route.calls.last.request.headers["referer"] == ( + "https://crm2.yunkecn.com/cms/x" + ) + + +# --------------------------------------------------------------------------- +# 2. Rate limiter awaited once per request +# --------------------------------------------------------------------------- + + +@respx.mock +async def test_rate_limiter_wait_awaited_once_per_request(client: YunkeClient): + respx.post(DATA_URL).mock(return_value=httpx.Response(200, json={"ok": True})) + + spy = AsyncMock() + client._rate_limiter.wait = spy # type: ignore[method-assign] + + await client.post(DATA_PATH, json={}) + + assert spy.await_count == 1 + + +# --------------------------------------------------------------------------- +# 3. Risk-control signals -> exponential backoff + retry +# --------------------------------------------------------------------------- + + +@respx.mock +async def test_429_triggers_backoff_retry_then_success( + monkeypatch, client: YunkeClient +): + delays: list[float] = [] + _patch_backoff_sleep(monkeypatch, delays) + + route = respx.post(DATA_URL).mock( + side_effect=[ + httpx.Response(429, json={"msg": "请求过于频繁"}), + httpx.Response(429, json={"msg": "请求过于频繁"}), + httpx.Response(200, json={"data": [{"id": 1}]}), + ] + ) + + resp = await client.post(DATA_PATH, json={}) + + assert resp.status_code == 200 + assert resp.json() == {"data": [{"id": 1}]} + # Initial try + two retries after backoff. + assert route.call_count == 3 + # base * 2**0 = 2.0, base * 2**1 = 4.0 + assert delays == [2.0, 4.0] + + +@respx.mock +async def test_200_with_risk_marker_triggers_backoff(monkeypatch, client: YunkeClient): + """A 200 body carrying a 风控/限流 marker must also trigger backoff — many + Chinese CRMs signal throttling inline rather than via a real 429.""" + _patch_backoff_sleep(monkeypatch, []) + + route = respx.post(DATA_URL).mock( + side_effect=[ + httpx.Response(200, json={"code": 500, "msg": "操作过于频繁"}), + httpx.Response(200, json={"data": []}), + ] + ) + + resp = await client.post(DATA_PATH, json={}) + + assert resp.status_code == 200 + assert resp.json() == {"data": []} + assert route.call_count == 2 + + +@respx.mock +async def test_persistent_429_raises_after_backoff_attempts(monkeypatch): + """When the upstream keeps signaling risk past ``risk_backoff_attempts``, + the client gives up and raises an :class:`httpx.HTTPStatusError`.""" + delays: list[float] = [] + _patch_backoff_sleep(monkeypatch, delays) + + c = YunkeClient( + base_url=BASE, + userid=USERID, + company=COMPANY, + rate_limiter=RateLimiter(min_interval=0.0, jitter=0.0), + backoff_attempts=2, + backoff_base=BACKOFF_BASE, + ) + try: + respx.post(DATA_URL).mock( + return_value=httpx.Response(429, json={"msg": "请求过于频繁"}) + ) + with pytest.raises(httpx.HTTPStatusError): + await c.post(DATA_PATH, json={}) + # 1 initial attempt + 2 retries = 3 upstream calls, 2 backoff sleeps. + assert delays == [2.0, 4.0] + finally: + await c.aclose() + + +# --------------------------------------------------------------------------- +# 4. Volume cap -> RequestCapReached +# --------------------------------------------------------------------------- + + +@respx.mock +async def test_volume_cap_raises_on_nth_request(): + respx.post(DATA_URL).mock(return_value=httpx.Response(200, json={"ok": True})) + capped = YunkeClient( + base_url=BASE, + userid=USERID, + company=COMPANY, + rate_limiter=RateLimiter(min_interval=0.0, jitter=0.0), + max_requests_per_run=2, + ) + try: + await capped.post(DATA_PATH, json={}) # 1 -> ok + await capped.post(DATA_PATH, json={}) # 2 -> ok + with pytest.raises(RequestCapReached): + await capped.post(DATA_PATH, json={}) # 3 -> cap hit + finally: + await capped.aclose() + + +@respx.mock +async def test_volume_cap_zero_means_unlimited(client: YunkeClient): + """``max_requests_per_run=0`` (the default) disables the cap entirely.""" + respx.post(DATA_URL).mock(return_value=httpx.Response(200, json={"ok": True})) + for _ in range(5): + await client.post(DATA_PATH, json={}) + + +# --------------------------------------------------------------------------- +# 5. 401 session-expiry still relogs exactly once (existing behaviour preserved) +# --------------------------------------------------------------------------- + + +@respx.mock +async def test_401_still_relogs_once_with_stealth_active(): + """Stealth must not break the existing 401 -> relogin-once -> retry path. + + A 401 is a session-death (handled by relogin), NOT a risk-control signal, + so it must not enter the backoff loop. + """ + secure = respx.get(f"{BASE}/newuc/login/getSecureKey").mock( + return_value=httpx.Response(200, json={"data": _TEST_SECURE_KEY}) + ) + login = respx.post(f"{BASE}/usercenter/login/pcLogin").mock( + return_value=httpx.Response( + 200, + json={"code": 0, "data": {"userid": USERID, "company": COMPANY}}, + headers={"Set-Cookie": "JSESSIONID=deadbeef; Path=/"}, + ) + ) + data = respx.post(DATA_URL).mock( + side_effect=[ + httpx.Response(401, json={"msg": "session expired"}), + httpx.Response(200, json={"data": [{"id": 1}]}), + ] + ) + + c = YunkeClient( + base_url=BASE, + username="tester", + password="hunter2", + rate_limiter=RateLimiter(min_interval=0.0, jitter=0.0), + ) + # Prime logged-in so only the 401-driven relogin is counted (mirrors the + # pattern in test_yunke_client.py). + from app.session.login import LoginResult + + c.login_result = LoginResult(userid=USERID, company=COMPANY) + try: + resp = await c.post(DATA_PATH, json={}) + assert resp.status_code == 200 + assert resp.json() == {"data": [{"id": 1}]} + assert secure.called + assert login.call_count == 1 + assert data.call_count == 2 + finally: + await c.aclose()