"""两条云账单链路的日级互查。 背景(2026-08-16 生产实测):每个云账号每天被写进两张表,走的是**两个不同 的厂商调用** —— ``cmdb_cloud_billing_daily`` 是日总额,``cmdb_cost_facts`` 是逐行明细,而页面上的每个折前数字都来自后者。明细是逐日翻页拉的,厂商提前 返回一个空 ``NextToken`` 就等于「这天到此为止」,**少拉的部分没有任何痕迹**: 任务写 ``status=ok``,run summary 里的行数与实际入库行数完全相等(拉到多少 写多少)。只有把两条链路对一下才看得出来。 2026 全年 3 起,全在阿里云国际站,而且正是此前两次昂贵排查的答案: * dbaas-prod-dataplane-001 2026-05-06:日总 12,424.40 vs 明细 **200.83** * dbaas-prod-dataplane-001 2026-05-05:12,209.62 vs 8,520.61 —— 上面两条合计就是 5 月 BQ 对账剩下的 $15,913 * dbaas-staging-dev-nextgen-dataplane 2026-07-31:473.88 vs 91.93 —— 就是 7 月 BQ 对账报出的 $381.95 这批用例钉住的不是「能不能发现差异」(那太容易),而是**阈值为什么是这个 阈值**:比值和绝对额两个条件缺一不可,而且必须容忍 AWS 两条链路天然的几个 百分点差(不同数据源),否则会刷出 224 条噪声 —— 一个训练人忽略告警的闸, 比没有闸更坏。 """ from __future__ import annotations import uuid from datetime import date from decimal import Decimal import pytest import pytest_asyncio from sqlalchemy.ext.asyncio import AsyncSession from app.domains.cmdb.cost_chain_crosscheck import ( MAX_HEALTHY_RATIO, MIN_MATERIAL_USD, crosscheck_recent, find_chain_divergences, repair_windows, ) from app.domains.cmdb.models import ( CmdbCloudBillingDaily, CmdbCloudScope, CmdbCostFact, ) from app.integrations.models import Integration DAY = date(2026, 5, 6) @pytest_asyncio.fixture async def env(db_session: AsyncSession): integ = Integration( id=str(uuid.uuid4()), kind="cloud", provider="alicloud_intl", name="crosscheck test", slug=f"cc-{uuid.uuid4().hex[:8]}", enabled=True, ) db_session.add(integ) await db_session.flush() scope = CmdbCloudScope( id=str(uuid.uuid4()), integration_id=str(integ.id), scope_type="aliyun_account", external_id="5901860086223871", display_name="dbaas-prod-dataplane-001", status="active", ) db_session.add(scope) await db_session.commit() return {"integration": integ, "scope": scope} async def _seed( db: AsyncSession, env, *, day: date, daily_list, facts, bucket: str = "usage", ) -> None: """写一天:``daily_list`` 是日总额链路,``facts`` 是明细链路的一行。""" if daily_list is not None: db.add(CmdbCloudBillingDaily( id=str(uuid.uuid4()), integration_id=str(env["integration"].id), scope_id=str(env["scope"].id), date=day, total_cost_usd=Decimal(daily_list), total_cost_usd_list=Decimal(daily_list), currency="USD", )) if facts is not None: db.add(CmdbCostFact( fact_key=uuid.uuid4().hex, date=day, integration_id=str(env["integration"].id), scope_id=str(env["scope"].id), service="ECS", amount_usd=Decimal(facts), amount_native=Decimal(facts), amount_usd_list=Decimal(facts), cost_bucket=bucket, currency="USD", )) await db.commit() async def _find(db: AsyncSession, **kw): return await find_chain_divergences( db, start=date(2026, 5, 1), end=date(2026, 5, 31), **kw, ) # ── 真实事故的形状 ─────────────────────────────────────────────── @pytest.mark.asyncio async def test_catches_the_2026_05_06_truncation( db_session: AsyncSession, env, ) -> None: """生产实测的最严重一起:日总 12,424.40,明细只有 200.83。""" await _seed(db_session, env, day=DAY, daily_list="12424.40", facts="200.83") found = await _find(db_session) assert len(found) == 1 d = found[0] assert d.day == DAY assert d.gap_usd == Decimal("12223.57") assert round(d.ratio, 3) == 0.016 assert d.scope_name == "dbaas-prod-dataplane-001" @pytest.mark.asyncio async def test_day_with_zero_facts_rows_is_flagged( db_session: AsyncSession, env, ) -> None: """整天一行都没拉到 —— 最严重的形态,必须报。 明细侧的 ``GROUP BY`` 只为「至少有一行」的 (scope, date) 产出分组,所以 用 INNER JOIN 会把这一天**整个丢掉**:``divergence_count`` 在损失最大的 时候恰好是 0,给出「明细链路健康」的假保证 —— 闸在最该响的地方静默放行。 这条用例就是钉住那个外连接。 """ await _seed(db_session, env, day=DAY, daily_list="12424.40", facts=None) found = await _find(db_session) assert len(found) == 1, "有日总额却没有任何明细行的一天被静默跳过了" assert found[0].facts_total_usd == Decimal("0") assert found[0].gap_usd == Decimal("12424.40") assert found[0].ratio == 0.0 @pytest.mark.asyncio async def test_matching_day_is_not_flagged( db_session: AsyncSession, env, ) -> None: """相邻的正常日必须干净 —— 否则这个闸只会被整体关掉。""" await _seed(db_session, env, day=date(2026, 5, 7), daily_list="12536.91", facts="12536.91") assert await _find(db_session) == [] # ── 阈值:两个条件缺一不可 ─────────────────────────────────────── @pytest.mark.asyncio async def test_small_account_total_loss_is_not_flagged( db_session: AsyncSession, env, ) -> None: """比值 0.0 但只差 $5 —— 小账号的分币噪声不该刷屏。 没有绝对额下限的话,``$0.01 vs $0.05`` 这种行会天天报。 """ await _seed(db_session, env, day=DAY, daily_list="5.00", facts="0.00") assert await _find(db_session) == [] @pytest.mark.asyncio async def test_large_but_shallow_gap_is_not_flagged( db_session: AsyncSession, env, ) -> None: """缺 $1,000 但比值 0.95 —— 这是口径差,不是整段丢失。 AWS 两条链路来自不同数据源(Cost Explorer vs BigQuery 里的 CUR), 天然差几个百分点;2026 全年按 5% 报会刷出 224 条。**闸报出噪声就等于 没有闸** —— 人会学会忽略它。 """ await _seed(db_session, env, day=DAY, daily_list="20000.00", facts="19000.00") assert await _find(db_session) == [] @pytest.mark.asyncio async def test_both_conditions_together_are_required( db_session: AsyncSession, env, ) -> None: """同时满足才报 —— 把两个否定用例的边界钉死在一次断言里。""" await _seed(db_session, env, day=date(2026, 5, 2), daily_list="300.00", facts="30.00") # 比 0.1、缺 $270 → 报 found = await _find(db_session) assert [d.day for d in found] == [date(2026, 5, 2)] assert float(MIN_MATERIAL_USD) == 200.0 assert MAX_HEALTHY_RATIO == 0.85 # ── 明细侧必须复用页面在用的口径 ──────────────────────────────── @pytest.mark.asyncio async def test_non_usage_rows_do_not_mask_a_truncation( db_session: AsyncSession, env, ) -> None: """退款/调账行不得被算进明细侧。 否则一笔大额调账会把当天的缺口「填平」,让真正的截断隐形 —— 而页面上 的折前数字从来不含这些行(``fact_engineering_amount()`` 的行白名单)。 自己重写一遍聚合就会踩这个坑,所以模块必须复用那个表达式。 """ await _seed(db_session, env, day=DAY, daily_list="12424.40", facts="200.83") await _seed(db_session, env, day=DAY, daily_list=None, facts="12000.00", bucket="adjustment") found = await _find(db_session) assert len(found) == 1, "调账行被算进明细侧 = 没走行白名单" assert found[0].facts_total_usd == Decimal("200.83") @pytest.mark.asyncio async def test_scope_without_list_measure_is_skipped( db_session: AsyncSession, env, ) -> None: """没写折前列的厂商(azure / tencent / ucloud)不参与。 拿实付去比折前只会得到一整片假漂移。 """ db_session.add(CmdbCloudBillingDaily( id=str(uuid.uuid4()), integration_id=str(env["integration"].id), scope_id=str(env["scope"].id), date=DAY, total_cost_usd=Decimal("12424.40"), total_cost_usd_list=None, currency="USD", )) await db_session.commit() await _seed(db_session, env, day=DAY, daily_list=None, facts="200.83") assert await _find(db_session) == [] # ── 报告 ───────────────────────────────────────────────────────── def test_repair_windows_merge_consecutive_days() -> None: """连着的天合成一个窗口,断开的分开 —— 一次翻页故障可能跨日。""" from app.domains.cmdb.cost_chain_crosscheck import ChainDivergence def d(day: date, slug: str = "alicloud-intl-primary"): return ChainDivergence( provider="alicloud_intl", integration_slug=slug, scope_external_id="x", scope_name=None, day=day, daily_total_usd=Decimal("1000"), facts_total_usd=Decimal("1"), ) got = repair_windows([ d(date(2026, 5, 6)), d(date(2026, 5, 5)), d(date(2026, 7, 31)), d(date(2026, 5, 5), slug="other-integration"), ]) assert { (w["integration"], w["start_date"], w["end_date"]) for w in got } == { ("alicloud-intl-primary", "2026-05-05", "2026-05-06"), ("alicloud-intl-primary", "2026-07-31", "2026-07-31"), ("other-integration", "2026-05-05", "2026-05-05"), } @pytest.mark.asyncio async def test_report_excludes_today(db_session: AsyncSession, env) -> None: """今天还在累积,两条链路天然不齐 —— 比它必然是假阳性。""" await _seed(db_session, env, day=date(2026, 5, 6), daily_list="12424.40", facts="200.83") report = await crosscheck_recent( db_session, today=date(2026, 5, 6), lookback_days=30, ) assert report["window"]["end"] == "2026-05-05" assert report["divergence_count"] == 0 @pytest.mark.asyncio async def test_report_carries_totals_and_repair_windows( db_session: AsyncSession, env, ) -> None: await _seed(db_session, env, day=date(2026, 5, 5), daily_list="12209.62", facts="8520.61") await _seed(db_session, env, day=date(2026, 5, 6), daily_list="12424.40", facts="200.83") report = await crosscheck_recent( db_session, today=date(2026, 5, 20), lookback_days=30, ) assert report["divergence_count"] == 2 # 12,223.57 + 3,689.01 —— 5 月 BQ 对账剩下的那 $15,913。 assert round(report["total_gap_usd"], 2) == 15912.58 # 最大的缺口排在前面:先看那个几乎整天丢光的。 assert report["divergences"][0]["date"] == "2026-05-06" assert report["repair_windows"] == [{ "integration": env["integration"].slug, "start_date": "2026-05-05", "end_date": "2026-05-06", }]