"""阿里云 BSS 双口径写侧契约（成本口径改造 PR3）。 Pins: * ``_classify_bss_bucket`` — ``Item`` 字段主分类（SubscriptionOrder / PayAsYouGoBill → usage；Refund / Adjustment → adjustment；未知新 类型 → other 绝不误入 usage；缺失回退按符号）。 * ``_bss_gross_amount`` — ``PretaxGrossAmount``（牌价）缺失/坏值 → None，绝不用折后实付兜底。 * ``list_cost_facts`` 行发射：usage 行 list=牌价（与实付同一 FX 钩子 换算）、调账行 list=None + bucket=adjustment、缺牌价行计数 WARN。 * ``list_billing_daily``：日聚合携带 total_cost_usd_list/_public （CNY raw，公一致 = 阿里的牌价即 PretaxGrossAmount）。 """ from __future__ import annotations from datetime import date from typing import Any import pytest from app.integrations.driver_base import CloudScopeRef from app.integrations.drivers.aliyun_cloud import AliyunCloudDriver class _StubDriver(AliyunCloudDriver): def __init__(self, items: list[dict[str, Any]]) -> None: self.integration_id = "test-aliyun" self._items = items async def _acs3_call(self, *, host: str, version: str, action: str, params: dict[str, Any]) -> dict[str, Any]: assert action == "DescribeInstanceBill" return {"Data": {"Items": self._items, "NextToken": None}} async def _master_account_id(self) -> str: return "5000000000000001" # ── 纯函数 ────────────────────────────────────────────────────────── @pytest.mark.parametrize("item,native,expect", [ ({"Item": "PayAsYouGoBill"}, 10.0, "usage"), ({"Item": "SubscriptionOrder"}, 10.0, "usage"), ({"Item": "Refund"}, -5.0, "adjustment"), ({"Item": "Adjustment"}, -5.0, "adjustment"), # 未知新类型 → other（白名单思路），绝不误入 usage。 ({"Item": "SomeNewType"}, 10.0, "other"), # 缺 Item 字段：按符号保守回退。 ({}, 10.0, "usage"), ({}, -3.0, "adjustment"), ]) def test_classify_bss_bucket(item, native, expect) -> None: assert AliyunCloudDriver._classify_bss_bucket(item, native) == expect @pytest.mark.parametrize("raw,expect", [ ("12.5", 12.5), (12.5, 12.5), (0, 0.0), (None, None), ("", None), ("not-a-number", None), ]) def test_bss_gross_amount_never_falls_back(raw, expect) -> None: item = {"PretaxGrossAmount": raw} if raw is not None else {} assert AliyunCloudDriver._bss_gross_amount(item) == expect # ── cost_facts 行发射 ─────────────────────────────────────────────── @pytest.mark.asyncio async def test_cost_facts_emit_list_and_buckets(caplog) -> None: scope = CloudScopeRef( scope_type="aliyun_account", external_id="5000000000000001", ) items = [ # 正常按量行：牌价 20 折后 14（合同折扣在行内字段，非独立行）。 { "Item": "PayAsYouGoBill", "OwnerID": "5000000000000001", "PretaxAmount": "14.0", "PretaxGrossAmount": "20.0", "ProductName": "云服务器 ECS", "InstanceID": "i-abc", "BillingItem": "cpu", }, # 调账负数行（5/25、7/21 的形状）：list 恒 None。 { "Item": "Refund", "OwnerID": "5000000000000001", "PretaxAmount": "-6.0", "PretaxGrossAmount": "0", "ProductName": "对象存储 OSS", "InstanceID": "oss-x", }, # usage 行缺牌价 → list None + 计数 WARN，绝不用实付兜底。 { "Item": "PayAsYouGoBill", "OwnerID": "5000000000000001", "PretaxAmount": "3.0", "ProductName": "云数据库 RDS", "InstanceID": "rds-y", }, ] drv = _StubDriver(items) facts = await drv.list_cost_facts( date(2026, 7, 21), date(2026, 7, 21), scope=scope, cny_per_usd=7.0, ) by_service = {f.service: f for f in facts} ecs = by_service["云服务器 ECS"] assert ecs.amount_native == pytest.approx(14.0) assert ecs.amount_native_list == pytest.approx(20.0) assert ecs.amount_usd_list == pytest.approx(20.0 / 7.0) assert ecs.amount_usd_public == ecs.amount_usd_list assert ecs.cost_bucket == "usage" refund = by_service["对象存储 OSS"] assert refund.amount_native == pytest.approx(-6.0) assert refund.amount_native_list is None assert refund.amount_usd_list is None assert refund.cost_bucket == "adjustment" rds = by_service["云数据库 RDS"] assert rds.amount_native_list is None assert rds.cost_bucket == "usage" assert any( "missing" in r.message and "PretaxGrossAmount" in r.message for r in caplog.records ) # ── billing_daily 聚合 ───────────────────────────────────────────── @pytest.mark.asyncio async def test_billing_daily_emits_list_measures() -> None: scope = CloudScopeRef( scope_type="aliyun_account", external_id="5000000000000001", ) items = [ { "Item": "PayAsYouGoBill", "OwnerID": "5000000000000001", "PretaxAmount": "14.0", "PretaxGrossAmount": "20.0", "ProductName": "云服务器 ECS", }, { "Item": "Refund", "OwnerID": "5000000000000001", "PretaxAmount": "-6.0", "PretaxGrossAmount": "0", "ProductName": "对象存储 OSS", }, ] drv = _StubDriver(items) rows = await drv.list_billing_daily( date(2026, 7, 21), date(2026, 7, 21), scope=scope, ) assert len(rows) == 1 r = rows[0] # 发票口径：净额（含调账）。 assert r.total_cost_usd == pytest.approx(8.0) assert r.currency == "CNY" # 研发口径：仅 usage 行的牌价；调账剔除。public == list。 assert r.total_cost_usd_list == pytest.approx(20.0) assert r.total_cost_usd_public == pytest.approx(20.0) # ── #2222 / #2223（07-31 审计）────────────────────────────────────── @pytest.mark.asyncio async def test_fully_discounted_row_keeps_its_list_price() -> None: """#2223：``PretaxAmount == 0`` 的 skip 跑在牌价累计之前。 被代金券 / 满减 / RI 全额覆盖的用量行,实付为 0 而牌价 > 0 —— 正是 #2175（AWS BQ）与 #2180（principal-tag）在同一个 PR 里修掉的 「invoice 零但 list 非零」形状。先 skip 就把这类行从**工程口径**里 静默丢掉了,而折扣恰恰不该让工程口径动。 """ scope = CloudScopeRef( scope_type="aliyun_account", external_id="5000000000000001", ) items = [ # 全额抵扣：实付 0、牌价 20。 { "Item": "PayAsYouGoBill", "OwnerID": "5000000000000001", "PretaxAmount": "0", "PretaxGrossAmount": "20.0", "ProductName": "云服务器 ECS", "InstanceID": "i-free", }, # 纯计量噪声：两个口径都没有钱 —— 仍旧不落行。 { "Item": "PayAsYouGoBill", "OwnerID": "5000000000000001", "PretaxAmount": "0", "ProductName": "无用行", "InstanceID": "i-noise", }, ] drv = _StubDriver(items) facts = await drv.list_cost_facts( date(2026, 7, 21), date(2026, 7, 21), scope=scope, cny_per_usd=7.0, ) by_service = {f.service: f for f in facts} assert "无用行" not in by_service, "两个口径都为空的行不应落库" ecs = by_service["云服务器 ECS"] assert ecs.amount_native == pytest.approx(0.0) assert ecs.amount_native_list == pytest.approx(20.0) assert ecs.amount_usd_list == pytest.approx(20.0 / 7.0) @pytest.mark.asyncio async def test_daily_aggregate_keeps_fully_discounted_list_money() -> None: """#2223 的日聚合孪生：实付 0、牌价 20 的一天,list 必须是 20。""" scope = CloudScopeRef( scope_type="aliyun_account", external_id="5000000000000001", ) items = [{ "Item": "PayAsYouGoBill", "OwnerID": "5000000000000001", "PretaxAmount": "0", "PretaxGrossAmount": "20.0", "ProductName": "云服务器 ECS", }] rows = await _StubDriver(items).list_billing_daily( date(2026, 7, 21), date(2026, 7, 21), scope=scope, ) assert len(rows) == 1 assert rows[0].total_cost_usd == pytest.approx(0.0) assert rows[0].total_cost_usd_list == pytest.approx(20.0) class _EndlessTokenDriver(AliyunCloudDriver): """BSS 交替返回两个 token —— 「同 token 闸」看不见,页封顶才拦得住。""" def __init__(self) -> None: self.integration_id = "test-aliyun-loop" self.calls = 0 async def _acs3_call(self, *, host: str, version: str, action: str, params: dict[str, Any]) -> dict[str, Any]: self.calls += 1 return {"Data": { "Items": [], # A→B→A→B：每页都在「前进」,连续相等检查永远不成立。 "NextToken": "A" if self.calls % 2 else "B", }} async def _master_account_id(self) -> str: return "5000000000000001" @pytest.mark.asyncio async def test_cost_facts_walk_has_a_page_ceiling() -> None: """#2222：``list_cost_facts`` 只有同 token 闸、没有页封顶。 交替 token(或每页造一个新假 token)即可击穿连续相等检查 —— 每次调用 都很快成功,30s 单请求超时不会触发,worker watchdog 按其自身注释也看 不见挂死的循环,这个 job 就无限期占着一个并发槽。intl 子类继承本 方法,生产 intl cost-facts 路径同样中招。 """ from app.integrations.drivers.aliyun_cloud import _MAX_BILL_PAGES drv = _EndlessTokenDriver() with pytest.raises(RuntimeError, match="page ceiling"): await drv.list_cost_facts( date(2026, 7, 21), date(2026, 7, 21), scope=CloudScopeRef( scope_type="aliyun_account", external_id="5000000000000001", ), cny_per_usd=7.0, ) # 第 501 次进入循环时抢在调用之前抛出 —— 正好 500 次外呼。 assert drv.calls == _MAX_BILL_PAGES def test_both_monthly_providers_cap_their_walk() -> None: """#2222：两个 provider 的 QueryAccountBill 月度 walk 是裸 while True。""" import inspect from app.domains.cmdb.providers.alicloud_intl import AlicloudIntlProvider from app.domains.cmdb.providers.aliyun import AliyunProvider for cls in (AliyunProvider, AlicloudIntlProvider): src = inspect.getsource(cls.get_monthly_billing) assert "_MAX_BILL_PAGES" in src, cls.__name__ assert "non-advancing NextToken" in src, cls.__name__ # 子类继承父类的封顶常量,不许各自定义一份漂移的。 assert ( AlicloudIntlProvider._MAX_BILL_PAGES is AliyunProvider._MAX_BILL_PAGES )