"""阿里云 OpenAPI 限流重试（``_acs3_call``）。 2026-07-29 生产回填实测：阿里云用 **HTTP 400 + ``Code=Throttling.User``** 表达限流（不是 429），而 ``_acs3_call`` 对任何 >=400 直接抛 —— 4 路并行 拉 BSS ``DescribeInstanceBill``（25 个子账号 × 按天分页）触发 219 次 ``Throttling.User``，绝大多数子账号当块无数据落库。本仓库其它云驱动 （AWS/Azure/腾讯）都有限流重试，阿里云是唯一缺口。 Pins： * ``Throttling.User`` 400 → 退避重试，成功后正常返回数据。 * 持续限流 → 用尽预算后抛原样 RuntimeError（不吞错）。 * **非**限流的 400（如签名错误）→ 立即抛，不浪费退避。 * 429（若 BSS 将来改用标准状态码）同样按限流处理。 * 判定只看 ``Code`` 字段，不对 Message 做模糊匹配（避免业务错误 里带 throttling 字样被误判）。 """ from __future__ import annotations import json from unittest.mock import AsyncMock, patch import httpx import pytest from app.integrations.drivers import aliyun_cloud from app.integrations.drivers.aliyun_cloud import ( AliyunCloudDriver, _is_acs3_retryable_5xx, _is_acs3_throttled, ) _THROTTLE_BODY = json.dumps({ "RequestId": "019FAB90-6C5F-327E-984A-44593F5AB9E4", "Message": "Request was denied due to user flow control.", "Code": "Throttling.User", "HostId": "business.ap-southeast-1.aliyuncs.com", }) _SIGNATURE_BODY = json.dumps({ "Code": "SignatureDoesNotMatch", "Message": "signature mismatch", }) class _StubDriver(AliyunCloudDriver): def __init__(self) -> None: self.integration_id = "aliyun-throttle-test" self.config = { "access_key_id": "LTAI-test", "access_key_secret": "secret-test", } def _resp(status: int, text: str) -> httpx.Response: return httpx.Response( status_code=status, request=httpx.Request("POST", "https://business.aliyuncs.com/"), content=text.encode(), ) # ── 判定函数 ────────────────────────────────────────────────────── def test_detector_matches_throttling_codes() -> None: assert _is_acs3_throttled(400, _THROTTLE_BODY) assert _is_acs3_throttled(403, json.dumps({"Code": "Throttling"})) assert _is_acs3_throttled(400, json.dumps({"Code": "Throttling.Api"})) assert _is_acs3_throttled(429, "") # 未来若改标准状态码 def test_detector_rejects_non_throttle_errors() -> None: assert not _is_acs3_throttled(400, _SIGNATURE_BODY) assert not _is_acs3_throttled(500, json.dumps({"Code": "InternalError"})) assert not _is_acs3_throttled(400, "not-json-at-all") # 只看 Code,不拿 Message 模糊匹配 assert not _is_acs3_throttled(400, json.dumps({ "Code": "InvalidParameter", "Message": "param 'throttling' unsupported", })) # ── 重试行为 ────────────────────────────────────────────────────── @pytest.mark.asyncio async def test_retries_then_succeeds() -> None: drv = _StubDriver() calls: list[int] = [] async def _request(*_a, **_kw): calls.append(1) if len(calls) <= 2: return _resp(400, _THROTTLE_BODY) return _resp(200, json.dumps({"Data": {"Items": [{"ok": 1}]}})) slept: list[float] = [] async def _sleep(d: float) -> None: slept.append(d) with patch.object( httpx.AsyncClient, "request", new=AsyncMock(side_effect=_request), ), patch.object(aliyun_cloud.asyncio, "sleep", new=_sleep): out = await drv._acs3_call( host="business.ap-southeast-1.aliyuncs.com", version="2017-12-14", action="DescribeInstanceBill", ) assert out["Data"]["Items"] == [{"ok": 1}] assert len(calls) == 3 assert slept == [1.0, 2.0] # 阶梯退避,不引随机 @pytest.mark.asyncio async def test_persistent_throttle_raises_after_budget() -> None: drv = _StubDriver() calls: list[int] = [] async def _request(*_a, **_kw): calls.append(1) return _resp(400, _THROTTLE_BODY) async def _sleep(_d: float) -> None: return None with patch.object( httpx.AsyncClient, "request", new=AsyncMock(side_effect=_request), ), patch.object( aliyun_cloud.asyncio, "sleep", new=_sleep, ), pytest.raises(RuntimeError, match="Throttling.User"): await drv._acs3_call( host="business.ap-southeast-1.aliyuncs.com", version="2017-12-14", action="DescribeInstanceBill", ) assert len(calls) == aliyun_cloud._ACS3_THROTTLE_MAX_ATTEMPTS @pytest.mark.asyncio async def test_non_throttle_error_fails_fast_without_backoff() -> None: drv = _StubDriver() calls: list[int] = [] slept: list[float] = [] async def _request(*_a, **_kw): calls.append(1) return _resp(400, _SIGNATURE_BODY) async def _sleep(d: float) -> None: slept.append(d) with patch.object( httpx.AsyncClient, "request", new=AsyncMock(side_effect=_request), ), patch.object( aliyun_cloud.asyncio, "sleep", new=_sleep, ), pytest.raises(RuntimeError, match="SignatureDoesNotMatch"): await drv._acs3_call( host="business.ap-southeast-1.aliyuncs.com", version="2017-12-14", action="DescribeInstanceBill", ) assert len(calls) == 1 # 不重试 assert slept == [] # ── 临时 5xx：只对只读 action 重试 ──────────────────────────────── # # 5xx 语义是「结果未知」(请求可能已在服务端生效)，所以变更类 action # 不能重试（AttachPolicyToUser 收到 503 但实际已生效 → 重试会重复 # 下发）。限流不同：那是明确「被拒、未执行」。 # 生产实测（2026-07-29 阿里云 9 个月回填）：1 次 503 就让某子账号 # 整个 30 天窗口没有数据落库。 _SERVICE_UNAVAILABLE_BODY = json.dumps({ "RequestId": "019FAE14-39AA-35C9-A073-9520DBBFDCE8", "Message": "The request has failed due to a temporary failure of the server.", "Code": "ServiceUnavailable", }) def test_readonly_5xx_is_retryable_mutating_is_not() -> None: for action in ( "DescribeInstanceBill", "ListPolicies", "QueryAccountBill", "GetUser", ): assert _is_acs3_retryable_5xx(503, action), action for action in ( "AttachPolicyToUser", "CreateUser", "DeleteUser", "UpdateUser", "SetDefaultPolicyVersion", "AddUserToGroup", ): assert not _is_acs3_retryable_5xx(503, action), action # 非 5xx 不走这条 assert not _is_acs3_retryable_5xx(400, "DescribeInstanceBill") assert not _is_acs3_retryable_5xx(404, "DescribeInstanceBill") # 其余可重试状态码 for code in (500, 502, 504): assert _is_acs3_retryable_5xx(code, "DescribeInstanceBill"), code @pytest.mark.asyncio async def test_readonly_503_retries_then_succeeds() -> None: drv = _StubDriver() calls: list[int] = [] slept: list[float] = [] async def _request(*_a, **_kw): calls.append(1) if len(calls) == 1: return _resp(503, _SERVICE_UNAVAILABLE_BODY) return _resp(200, json.dumps({"Data": {"Items": []}})) async def _sleep(d: float) -> None: slept.append(d) with patch.object( httpx.AsyncClient, "request", new=AsyncMock(side_effect=_request), ), patch.object(aliyun_cloud.asyncio, "sleep", new=_sleep): out = await drv._acs3_call( host="business.ap-southeast-1.aliyuncs.com", version="2017-12-14", action="DescribeInstanceBill", ) assert out["Data"]["Items"] == [] assert len(calls) == 2 assert slept == [1.0] @pytest.mark.asyncio async def test_mutating_503_fails_fast_without_retry() -> None: """变更类 action 收到 503 立即抛 —— 绝不重试(可能已生效)。""" drv = _StubDriver() calls: list[int] = [] slept: list[float] = [] async def _request(*_a, **_kw): calls.append(1) return _resp(503, _SERVICE_UNAVAILABLE_BODY) async def _sleep(d: float) -> None: slept.append(d) with patch.object( httpx.AsyncClient, "request", new=AsyncMock(side_effect=_request), ), patch.object( aliyun_cloud.asyncio, "sleep", new=_sleep, ), pytest.raises(RuntimeError, match="ServiceUnavailable"): await drv._acs3_call( host="ram.aliyuncs.com", version="2015-05-01", action="AttachPolicyToUser", ) assert len(calls) == 1 assert slept == []