"""Aliyun BSS empty-OwnerID attribution (master scope, no fan-out). DescribeInstanceBill on a Resource Directory master returns the consolidated bill; each row carries ``OwnerID`` (resource owner = sub-account). EMPTY-OwnerID rows are master-account-allocated charges. The per-scope billing methods fan out per scope and filter client-side, so an empty-OwnerID row must be attributed to EXACTLY ONE scope (the master) — pre-fix it passed the ``not in ("", owner_filter)`` filter for EVERY scope and over-reported tenant spend (prod 2026-06-17: ~$16.8K / 30d duplicated across 23 alicloud_intl scopes). These tests pin the shared keep-filter + the master-scope registration so a future refactor cannot silently re-introduce the fan-out. """ from __future__ import annotations import pytest from app.integrations.drivers.aliyun_cloud import AliyunCloudDriver class _StubDriver(AliyunCloudDriver): """Bypass ``__init__`` so we can exercise pure logic + cache.""" def __init__(self) -> None: self.integration_id = "test-aliyun" MASTER = "5962436752113096" SUB_A = "5108739859105002" SUB_B = "5901860086223871" def test_keep_filter_owner_match_and_master_only_empties(): keep = AliyunCloudDriver._billing_row_belongs_to_scope # Tenant-wide call (scope=None / owner_filter falsy) keeps everything # — single pass, no fan-out possible. assert keep("", None, MASTER) is True assert keep(SUB_A, None, MASTER) is True # Sub-account scope: keep only its own rows. assert keep(SUB_A, SUB_A, MASTER) is True assert keep(SUB_B, SUB_A, MASTER) is False # Empty-OwnerID (master-allocated): kept ONLY for the master scope. assert keep("", MASTER, MASTER) is True assert keep("", SUB_A, MASTER) is False assert keep("", SUB_B, MASTER) is False def test_keep_filter_skips_master_allocated_when_master_unknown(): """Round-3 fix: STS failed → master_id None → we cannot identify the master scope, so master-allocated (empty-OwnerID) rows must be SKIPPED for every scope, NOT kept for every scope. Keeping fans the row out N× → the org total / budget alerts over-report by ~(N-1)× and the moving re-ingest window only partially self-heals. An outage-window under-count that self-heals on the next clean sync is far safer than a silent, partially-permanent over-count.""" keep = AliyunCloudDriver._billing_row_belongs_to_scope assert keep("", SUB_A, None) is False assert keep("", SUB_B, None) is False # When the master IS known, the row is kept only for the master scope. assert keep("", SUB_A, SUB_A) is True assert keep("", SUB_B, SUB_A) is False # Non-empty owner still filtered normally regardless of master. assert keep(SUB_B, SUB_A, None) is False @pytest.mark.asyncio async def test_master_account_id_cached_single_resolution(monkeypatch): drv = _StubDriver() calls = {"n": 0} async def fake_own(): calls["n"] += 1 return MASTER monkeypatch.setattr(drv, "_own_account_id", fake_own) assert await drv._master_account_id() == MASTER assert await drv._master_account_id() == MASTER # Resolved once, cached thereafter (cron calls per-scope 23×). assert calls["n"] == 1 @pytest.mark.asyncio async def test_master_account_id_none_on_failure(monkeypatch): drv = _StubDriver() async def boom(): raise RuntimeError("STS down") monkeypatch.setattr(drv, "_own_account_id", boom) # Degrades to None rather than raising (don't drop all billing). assert await drv._master_account_id() is None @pytest.mark.asyncio async def test_list_scopes_appends_master_when_absent(monkeypatch): """RD ListAccounts enumerates members and may omit the management account — list_scopes must register the master so empty-OwnerID rows have a scope home.""" drv = _StubDriver() drv.config = {} # Force RD mode (non-iam) so list_scopes hits the ListAccounts path. monkeypatch.setattr( type(drv), "account_mode", property(lambda self: "sso"), ) async def fake_acs3(*, host, version, action, params=None, method=None): assert action == "ListAccounts" return { "Accounts": {"Account": [ {"AccountId": SUB_A, "DisplayName": "sub-a", "Status": "CreateSuccess", "Type": "ResourceAccount"}, {"AccountId": SUB_B, "DisplayName": "sub-b", "Status": "CreateSuccess", "Type": "ResourceAccount"}, ]}, "TotalCount": 2, } async def fake_own(): return MASTER monkeypatch.setattr(drv, "_acs3_call", fake_acs3) monkeypatch.setattr(drv, "_own_account_id", fake_own) scopes = await drv.list_scopes() ids = {s.external_id for s in scopes} assert ids == {SUB_A, SUB_B, MASTER} master_scope = next(s for s in scopes if s.external_id == MASTER) assert master_scope.raw.get("rd_master") is True @pytest.mark.asyncio async def test_list_scopes_no_duplicate_master_when_present(monkeypatch): """If ListAccounts already returns the master, don't add a dup.""" drv = _StubDriver() drv.config = {} monkeypatch.setattr( type(drv), "account_mode", property(lambda self: "sso"), ) async def fake_acs3(*, host, version, action, params=None, method=None): return { "Accounts": {"Account": [ {"AccountId": MASTER, "DisplayName": "master", "Status": "CreateSuccess", "Type": "CloudAccount"}, {"AccountId": SUB_A, "DisplayName": "sub-a", "Status": "CreateSuccess", "Type": "ResourceAccount"}, ]}, "TotalCount": 2, } async def fake_own(): return MASTER monkeypatch.setattr(drv, "_acs3_call", fake_acs3) monkeypatch.setattr(drv, "_own_account_id", fake_own) scopes = await drv.list_scopes() master_rows = [s for s in scopes if s.external_id == MASTER] assert len(master_rows) == 1 # When ListAccounts returns the master (matched by STS id), it must # still carry the rd_master marker — else owner-governance surfaces # (which key on raw.rd_master) would flag it as unowned. assert master_rows[0].raw.get("rd_master") is True @pytest.mark.asyncio async def test_list_scopes_rejects_directory_master_that_disagrees_with_sts( monkeypatch, ): """A provider type label cannot override the STS identity authority.""" drv = _StubDriver() drv.config = {} monkeypatch.setattr( type(drv), "account_mode", property(lambda self: "sso"), ) async def fake_acs3(*, host, version, action, params=None, method=None): return { "Accounts": {"Account": [ {"AccountId": MASTER, "DisplayName": "master", "Status": "CreateSuccess", "Type": "ResourceDirectoryMaster"}, {"AccountId": SUB_A, "DisplayName": "sub-a", "Status": "CreateSuccess", "Type": "ResourceAccount"}, ]}, "TotalCount": 2, } async def fake_own(): return SUB_B monkeypatch.setattr(drv, "_acs3_call", fake_acs3) monkeypatch.setattr(drv, "_own_account_id", fake_own) with pytest.raises(RuntimeError, match="disagrees with STS"): await drv.list_scopes()