"""Unit tests for pure helpers in ``app/domains/ai_spending/service.py``.

Helpers covered:
* ``_parse_owner_from_resource_id`` — extracts the AWS principal-tag
  Owner from the synthetic ``aws-principal:Owner=<x>|service=<svc>``
  resource_id the driver writes; this is how Bedrock bucket rows
  get attributed back to a specific employee even when CUR's
  ``user_resource_tag_user_owner`` is empty.
* ``_money`` — every dashboard amount goes through this; if the CNY
  side ever drifts off the configured FX rate we'd silently invoice
  in the wrong currency.
* ``_pct_change`` — MoM / YoY computation, including the divide-by-zero
  branch (must return None, never raise or yield infinity).
* ``_shift_period`` — period arithmetic wrapping across year boundaries;
  ``_current_period - 12`` must be the same month last year, not
  whatever month-arithmetic produces if you naively subtract 12 from
  the month integer.
* ``_hash_user_id`` — PII pseudonymisation; LLM prompt must never
  carry the raw user id. Output must be stable (same input -> same
  hash; deterministic for de-pseudonymisation at render time) and
  prefixed with ``u`` (so an accidental raw-id leak shows up as a
  format mismatch downstream).
* ``_sum_usd`` / ``_distinct_users`` — used to compose KPIs; tests
  pin the Bailian exclusion in ``_distinct_users``.
"""
from __future__ import annotations

from datetime import date
from decimal import Decimal
from unittest.mock import AsyncMock, patch

import pytest

from app.domains.ai_spending.service import (
    BUCKET_BAILIAN,
    BUCKET_BEDROCK,
    BUCKET_FOUNDRY,
    BUCKET_SEATS,
    AiSpendingService,
    BucketRow,
    _distinct_users,
    _hash_user_id,
    _money,
    _parse_owner_from_resource_id,
    _pct_change,
    _shift_period,
    _sum_gross_usd,
    _sum_usd,
)


class TestParseOwner:
    def test_well_formed_resource_id(self) -> None:
        rid = "aws-principal:Owner=alice@example.com|service=bedrock"
        assert _parse_owner_from_resource_id(rid) == "alice@example.com"

    def test_unallocated_returns_none(self) -> None:
        rid = "aws-principal:Owner=(unallocated)|service=bedrock"
        assert _parse_owner_from_resource_id(rid) is None

    def test_missing_owner_section_returns_none(self) -> None:
        rid = "aws-principal:service=bedrock"
        assert _parse_owner_from_resource_id(rid) is None

    def test_non_principal_prefix_returns_none(self) -> None:
        # Azure resource_ids look different — must not false-match.
        rid = "azure:rg=NexusCRM"
        assert _parse_owner_from_resource_id(rid) is None

    def test_empty_returns_none(self) -> None:
        assert _parse_owner_from_resource_id(None) is None
        assert _parse_owner_from_resource_id("") is None

    def test_owner_without_service_segment(self) -> None:
        rid = "aws-principal:Owner=bob"
        assert _parse_owner_from_resource_id(rid) == "bob"

    def test_lowercase_owner_key_returns_none(self) -> None:
        # The AWS driver writes the literal key ``Owner`` (capital O).
        # A future driver rewrite that emits ``owner=...`` lower-case
        # must NOT silently start surfacing attribution under the
        # old contract — the parser is case-sensitive on the key on
        # purpose so the surface-mismatch shows up as "missing
        # attribution" loud, not as wrong attribution silent.
        rid = "aws-principal:owner=alice"
        assert _parse_owner_from_resource_id(rid) is None

    def test_multi_segment_owner_preserved(self) -> None:
        # The first ``|`` segment is what we read; extra segments
        # after the IAM-user are ignored. Regression guard against a
        # future driver that adds a third KV (e.g. region tag) and
        # silently breaks attribution.
        rid = "aws-principal:Owner=alice@x.com|region=us-east-1|service=bedrock"
        assert _parse_owner_from_resource_id(rid) == "alice@x.com"


class TestMoney:
    def test_usd_passes_through(self) -> None:
        m = _money(Decimal("100.0"), rate=7.2)
        assert m.usd == Decimal("100.00")

    def test_cny_computed_from_rate(self) -> None:
        # Contract: ``rate`` is USD-per-CNY (small number, ≈0.14 in
        # prod, refreshed weekly from frankfurter.dev). CNY = USD / rate
        # ⇒ 100 USD ÷ 0.14 ≈ 714.29 CNY. This pins the direction so a
        # flipped ``cny = usd * rate`` regression (which would render
        # $100 as ¥14 — visually believable but ~50× wrong) bites here.
        m = _money(Decimal("100.00"), rate=0.14)
        assert float(m.cny) == pytest.approx(714.29, abs=0.01)
        # Cross-check at a different rate.
        m2 = _money(Decimal("50.00"), rate=0.139)
        assert float(m2.cny) == pytest.approx(359.71, abs=0.01)

    def test_accepts_int_and_float(self) -> None:
        assert _money(100, 7.2).usd == Decimal("100.00")
        assert _money(100.5, 7.2).usd == Decimal("100.50")


class TestPctChange:
    def test_basic(self) -> None:
        assert _pct_change(Decimal("100"), Decimal("110")) == 10.0

    def test_decrease(self) -> None:
        assert _pct_change(Decimal("200"), Decimal("100")) == -50.0

    def test_zero_previous_returns_none(self) -> None:
        # Critical: divide-by-zero must not raise. None is the
        # honest answer when the baseline is zero (any non-zero
        # current would be "+infinity").
        assert _pct_change(Decimal("0"), Decimal("100")) is None

    def test_none_previous_returns_none(self) -> None:
        assert _pct_change(None, Decimal("100")) is None


class TestShiftPeriod:
    def test_minus_1_same_year(self) -> None:
        assert _shift_period("2026-05", -1) == "2026-04"

    def test_minus_1_year_boundary(self) -> None:
        assert _shift_period("2026-01", -1) == "2025-12"

    def test_minus_12_yoy(self) -> None:
        # Critical for the YoY KPI — must roll the year, not produce
        # "2026-(-7)" or wraparound bugs.
        assert _shift_period("2026-05", -12) == "2025-05"

    def test_plus_1(self) -> None:
        assert _shift_period("2025-12", 1) == "2026-01"

    def test_plus_24_multi_year(self) -> None:
        assert _shift_period("2024-03", 24) == "2026-03"


class TestHashUserId:
    def test_stable(self) -> None:
        # Same input must produce same hash — server-side de-pseudonymises
        # at render time by re-hashing each known user and matching.
        assert _hash_user_id("alice@example.com") == _hash_user_id("alice@example.com")

    def test_prefix(self) -> None:
        h = _hash_user_id("alice@example.com")
        assert h is not None and h.startswith("u")

    def test_different_inputs_differ(self) -> None:
        assert _hash_user_id("alice@example.com") != _hash_user_id("bob@example.com")

    def test_none_in_none_out(self) -> None:
        assert _hash_user_id(None) is None
        assert _hash_user_id("") is None

    def test_does_not_leak_input(self) -> None:
        # Critical security property: the hash must not contain the
        # raw input (regression guard if someone "improves" this to
        # ``"u" + uid + hash[:4]``).
        raw = "alice@pingcap.com"
        h = _hash_user_id(raw)
        assert h is not None
        assert raw not in h
        # Local part alone must also not leak.
        assert "alice" not in h


class TestSumAndDistinct:
    @pytest.fixture
    def rows(self) -> list[BucketRow]:
        d = date(2026, 5, 1)
        return [
            BucketRow(bucket=BUCKET_SEATS, date=d, cost_usd=Decimal("10"),
                      user_id="alice"),
            BucketRow(bucket=BUCKET_SEATS, date=d, cost_usd=Decimal("20"),
                      user_id="bob"),
            BucketRow(bucket=BUCKET_FOUNDRY, date=d, cost_usd=Decimal("30"),
                      user_id="alice"),  # same person, different bucket
            BucketRow(bucket=BUCKET_BEDROCK, date=d, cost_usd=Decimal("40"),
                      user_id=None),  # unattributed
            BucketRow(bucket=BUCKET_BAILIAN, date=d, cost_usd=Decimal("50"),
                      user_id="carol"),  # attributed via the AK ladder
        ]

    def test_sum_usd(self, rows: list[BucketRow]) -> None:
        assert _sum_usd(rows) == Decimal("150")

    def test_sum_usd_empty(self) -> None:
        assert _sum_usd([]) == Decimal("0")

    def test_distinct_users_counts_attributed_bailian(
        self, rows: list[BucketRow],
    ) -> None:
        # alice (seats + foundry = 1) + bob (seats) + carol (Bailian).
        #
        # 2026-08-15: carol used to be excluded "by design (BSS gives no
        # per-user split, so any value would be a guess)". That stopped
        # being true at migration 299 — ``bailian_usage_daily`` is
        # per-(day × API key × model) and ``_bucket_bailian`` resolves an
        # owner per row, so carol is a measured attribution, not a guess.
        # Excluding her hid real people from 活跃 AI 用户 and their
        # dollars from 人均 AI 成本. Rows with no owner are still
        # excluded — by the ``user_id`` guard, not by bucket name.
        assert _distinct_users(rows) == 3

    def test_distinct_users_ignores_none(self) -> None:
        d = date(2026, 5, 1)
        rows = [
            BucketRow(bucket=BUCKET_FOUNDRY, date=d, cost_usd=Decimal("1"),
                      user_id=None),
            BucketRow(bucket=BUCKET_FOUNDRY, date=d, cost_usd=Decimal("1"),
                      user_id=None),
        ]
        assert _distinct_users(rows) == 0


class TestBuildDetailGroups:
    """``_build_detail_groups`` powers the drill-drawer 按人/按子账号/按服务
    rollups. It must fold same-key rows, sort by amount desc, keep money
    tied to the FX rate, bucket NULL owners as an unattributed row, and
    drop axes that have no identifying value at all."""

    @pytest.fixture
    def rows(self) -> list[BucketRow]:
        d1 = date(2026, 5, 1)
        d2 = date(2026, 5, 2)
        # Foundry-style rows: owner + scope + service all present.
        return [
            BucketRow(bucket=BUCKET_FOUNDRY, date=d1, cost_usd=Decimal("30"),
                      user_id="alice", integration_id="az",
                      scope_id="sub-1", service_name="Azure OpenAI"),
            BucketRow(bucket=BUCKET_FOUNDRY, date=d2, cost_usd=Decimal("20"),
                      user_id="alice", integration_id="az",
                      scope_id="sub-1", service_name="Cognitive Search"),
            BucketRow(bucket=BUCKET_FOUNDRY, date=d1, cost_usd=Decimal("40"),
                      user_id="bob", integration_id="az",
                      scope_id="sub-2", service_name="Azure OpenAI"),
            BucketRow(bucket=BUCKET_FOUNDRY, date=d1, cost_usd=Decimal("10"),
                      user_id=None, integration_id="az",
                      resource_group="rg-x", service_name="Azure OpenAI"),
        ]

    def _groups(self, rows: list[BucketRow]) -> dict:
        gs = AiSpendingService._build_detail_groups(
            rows, rate=0.1, total_usd=_sum_usd(rows),
            u_map={"alice": "Alice", "bob": "Bob"},
            d_map={"alice": "Eng", "bob": "Sales"},
            s_map={}, sc_map={"sub-1": "Prod Sub", "sub-2": "Dev Sub"},
            i_map={"az": "Microsoft Azure"},
        )
        return {g.dimension: g for g in gs}

    def test_owner_rollup_folds_and_sorts(self, rows: list[BucketRow]) -> None:
        owner = self._groups(rows)["owner"]
        # bob 40 > alice 50? alice=30+20=50, bob=40, none=10 → alice first.
        assert [(r.key, r.amount.usd) for r in owner.rows] == [
            ("alice", Decimal("50.00")),
            ("bob", Decimal("40.00")),
            (None, Decimal("10.00")),
        ]
        alice = owner.rows[0]
        assert alice.label == "Alice"
        assert alice.secondary == "Eng"
        assert alice.row_count == 2
        # share of 100 total.
        assert abs(alice.share - 0.5) < 1e-9
        # CNY ties to rate (usd / 0.1).
        assert alice.amount.cny == Decimal("500.00")

    def test_unattributed_label_is_none(self, rows: list[BucketRow]) -> None:
        owner = self._groups(rows)["owner"]
        last = owner.rows[-1]
        assert last.key is None
        assert last.label is None  # FE renders the "unattributed" tag

    def test_scope_uses_label_and_rg_fallback(
        self, rows: list[BucketRow],
    ) -> None:
        scope = self._groups(rows)["scope"]
        by_key = {r.key: r for r in scope.rows}
        assert by_key["sub-1"].label == "Prod Sub"
        assert by_key["sub-1"].secondary == "Microsoft Azure"
        # The owner-less row falls back to its resource group as key.
        assert by_key["rg-x"].label == "rg-x"

    def test_service_rollup(self, rows: list[BucketRow]) -> None:
        service = self._groups(rows)["service"]
        by_key = {r.key: r for r in service.rows}
        # Azure OpenAI = 30 + 40 + 10 = 80, Cognitive Search = 20.
        assert by_key["Azure OpenAI"].amount.usd == Decimal("80.00")
        assert by_key["Cognitive Search"].amount.usd == Decimal("20.00")

    def test_axis_dropped_when_no_identifying_value(self) -> None:
        # Seat rows with no scope/service → only the owner axis survives.
        d = date(2026, 5, 1)
        rows = [
            BucketRow(bucket=BUCKET_SEATS, date=d, cost_usd=Decimal("5"),
                      user_id="alice"),
        ]
        dims = {
            g.dimension
            for g in AiSpendingService._build_detail_groups(
                rows, rate=0.1, total_usd=Decimal("5"),
                u_map={"alice": "Alice"}, d_map={}, s_map={},
                sc_map={}, i_map={},
            )
        }
        assert dims == {"owner"}

    def test_zero_total_yields_zero_share(self) -> None:
        d = date(2026, 5, 1)
        rows = [
            BucketRow(bucket=BUCKET_FOUNDRY, date=d, cost_usd=Decimal("0"),
                      user_id="alice", scope_id="s"),
        ]
        owner = {
            g.dimension: g
            for g in AiSpendingService._build_detail_groups(
                rows, rate=0.1, total_usd=Decimal("0"),
                u_map={"alice": "Alice"}, d_map={}, s_map={},
                sc_map={}, i_map={},
            )
        }["owner"]
        assert owner.rows[0].share == 0.0


class TestBucketsFromRowsGross:
    """折扣前(gross)→ 折扣额 / 折扣率 / 覆盖率 聚合。

    ``_buckets_from_rows`` 是纯计算(不碰 self.db),用 ``__new__`` 跳过
    __init__ 直接测。rate=0.1 → CNY = USD × 10,数值整齐便于断言。
    """

    def _svc(self) -> AiSpendingService:
        return object.__new__(AiSpendingService)  # type: ignore[no-any-return]

    def test_bailian_full_coverage_discount(self) -> None:
        # 模拟 2026-05 Bailian 形状:净 38506.81 / 原价 106503.57 → 63.8% off。
        d = date(2026, 5, 1)
        cur = [
            BucketRow(
                bucket=BUCKET_BAILIAN, date=d,
                cost_usd=Decimal("38506.81"),
                cost_usd_gross=Decimal("106503.57"),
                aliyun_account_id="acc-1",
            ),
        ]
        out = self._svc()._buckets_from_rows(cur, [], rate=0.145694)
        b = next(x for x in out if x.key == BUCKET_BAILIAN)
        assert b.amount_gross is not None
        assert b.amount_gross.usd == pytest.approx(106503.57, abs=0.01)
        assert b.discount_amount is not None
        assert b.discount_amount.usd == pytest.approx(67996.76, abs=0.01)
        assert b.discount_pct == pytest.approx(63.85, abs=0.1)
        assert b.gross_coverage_pct == pytest.approx(100.0, abs=0.01)

    def test_partial_coverage_discount_only_on_covered(self) -> None:
        # 一行有原价(net 40 / gross 100),一行无原价(net 20)。
        # 折扣率只看可溯部分:(100-40)/100 = 60%;覆盖率 40/60 = 66.67%。
        d = date(2026, 5, 1)
        cur = [
            BucketRow(bucket=BUCKET_FOUNDRY, date=d,
                      cost_usd=Decimal("40"), cost_usd_gross=Decimal("100"),
                      resource_group="rg-a", integration_id="i1"),
            BucketRow(bucket=BUCKET_FOUNDRY, date=d,
                      cost_usd=Decimal("20"), cost_usd_gross=None,
                      resource_group="rg-b", integration_id="i1"),
        ]
        out = self._svc()._buckets_from_rows(cur, [], rate=0.1)
        b = next(x for x in out if x.key == BUCKET_FOUNDRY)
        assert b.amount.usd == pytest.approx(60.0)
        assert b.amount_gross is not None
        assert b.amount_gross.usd == pytest.approx(100.0)
        assert b.discount_pct == pytest.approx(60.0)
        assert b.gross_coverage_pct == pytest.approx(66.67, abs=0.01)

    def test_no_gross_leaves_fields_null(self) -> None:
        # Seats:没有原价口径 → gross 相关字段全空,覆盖率 0。
        d = date(2026, 5, 1)
        cur = [
            BucketRow(bucket=BUCKET_SEATS, date=d, cost_usd=Decimal("10"),
                      user_id="u1"),
        ]
        out = self._svc()._buckets_from_rows(cur, [], rate=0.1)
        b = next(x for x in out if x.key == BUCKET_SEATS)
        assert b.amount_gross is None
        assert b.discount_amount is None
        assert b.discount_pct is None
        assert b.gross_coverage_pct == 0.0

    def test_negative_discount_when_net_exceeds_gross(self) -> None:
        # 退款/credit 致 net > gross → 折扣率为负,如实显示不 clamp。
        d = date(2026, 5, 1)
        cur = [
            BucketRow(bucket=BUCKET_BEDROCK, date=d,
                      cost_usd=Decimal("120"), cost_usd_gross=Decimal("100"),
                      iam_user="bob", integration_id="i1"),
        ]
        out = self._svc()._buckets_from_rows(cur, [], rate=0.1)
        b = next(x for x in out if x.key == BUCKET_BEDROCK)
        assert b.discount_pct == pytest.approx(-20.0)


class TestGetDetailBailianExpansion:
    """``get_detail`` expands the Bailian monthly aggregate BucketRow into the
    per-(sub-account × commodity) lines the loader kept in ``raw``, so the
    drill-down shows "which account spent how much on which product" instead
    of one opaque ``aggregate`` row. DB helpers are patched so this stays a
    pure unit test (``__new__`` skips ``__init__`` → no real ``self.db``).
    """

    def _svc(self) -> AiSpendingService:
        return object.__new__(AiSpendingService)  # type: ignore[no-any-return]

    @pytest.mark.asyncio
    async def test_bailian_aggregate_expands_into_per_account_rows(self) -> None:
        d = date(2026, 5, 1)
        row = BucketRow(
            bucket=BUCKET_BAILIAN, date=d,
            cost_usd=Decimal("4727.71"),
            integration_id="int-ali",
            raw={"bss_overview_items": [
                {"bill_account_name": "sa@pingcap.com",
                 "commodity_code": "sfm_inferenceglobal_public_intl",
                 "product_detail": "Model Studio Foundation Model Inference",
                 "biz_type": "trusteeship",
                 "net_usd": 374.21, "net_cny": 2672.93, "gross_usd": 1069.19},
                {"bill_account_name": "ml-team@pingcap.com",
                 "commodity_code": "sfm_inferenceglobal_public_intl",
                 "product_detail": "Model Studio Foundation Model Inference",
                 "biz_type": "",
                 "net_usd": 4353.50, "net_cny": 31096.43, "gross_usd": 12831.34},
            ]},
        )
        svc = self._svc()
        with patch.object(svc, "_collect_period_rows", AsyncMock(return_value=[row])), \
             patch.object(svc, "_get_cny_rate", AsyncMock(return_value=0.14)), \
             patch.object(svc, "_person_label_map", AsyncMock(return_value={})), \
             patch.object(svc, "_person_department_map", AsyncMock(return_value={})), \
             patch.object(svc, "_software_label_map", AsyncMock(return_value={})), \
             patch.object(svc, "_scope_label_map", AsyncMock(return_value={})), \
             patch.object(svc, "_integration_label_map",
                          AsyncMock(return_value={"int-ali": "AliCloud (海外)"})), \
             patch.object(svc, "_build_ak_link_map", AsyncMock(return_value={})):
            detail = await svc.get_detail(
                period="2026-05", dimension="bucket", dimension_key="bailian",
            )
        # Two real rows, not one opaque aggregate.
        assert len(detail.rows) == 2
        accounts = {r.resource_group for r in detail.rows}
        assert accounts == {"sa@pingcap.com", "ml-team@pingcap.com"}
        assert all(
            r.service_name == "Model Studio Foundation Model Inference"
            for r in detail.rows
        )
        assert all(r.integration_name == "AliCloud (海外)" for r in detail.rows)
        # Per-item net sums back to the bucket total (header stays exact).
        assert sum(float(r.cost_usd) for r in detail.rows) == pytest.approx(
            4727.71, abs=0.01,
        )
        assert detail.total.usd == pytest.approx(4727.71, abs=0.01)

    @pytest.mark.asyncio
    async def test_bailian_without_items_falls_back_to_single_row(self) -> None:
        # Pre-resync rows (only a ``sample``, no item list) must not crash —
        # they degrade to the old single aggregate row.
        d = date(2026, 5, 1)
        row = BucketRow(
            bucket=BUCKET_BAILIAN, date=d, cost_usd=Decimal("4727.71"),
            integration_id="int-ali", raw={"bss_overview_sample": {"x": 1}},
        )
        svc = self._svc()
        with patch.object(svc, "_collect_period_rows", AsyncMock(return_value=[row])), \
             patch.object(svc, "_get_cny_rate", AsyncMock(return_value=0.14)), \
             patch.object(svc, "_person_label_map", AsyncMock(return_value={})), \
             patch.object(svc, "_person_department_map", AsyncMock(return_value={})), \
             patch.object(svc, "_software_label_map", AsyncMock(return_value={})), \
             patch.object(svc, "_scope_label_map", AsyncMock(return_value={})), \
             patch.object(svc, "_integration_label_map", AsyncMock(return_value={})), \
             patch.object(svc, "_build_ak_link_map", AsyncMock(return_value={})):
            detail = await svc.get_detail(
                period="2026-05", dimension="bucket", dimension_key="bailian",
            )
        assert len(detail.rows) == 1


class TestSumGrossUsd:
    """``_sum_gross_usd`` backs the dashboard's pre-discount grand total
    + coverage. covered_net (not the full total) is the discount
    denominator so partial coverage never inflates the discount."""

    def test_no_gross_rows(self) -> None:
        d = date(2026, 5, 1)
        rows = [BucketRow(bucket=BUCKET_SEATS, date=d, cost_usd=Decimal("10"))]
        gross, covered, has = _sum_gross_usd(rows)
        assert (gross, covered, has) == (Decimal("0"), Decimal("0"), False)

    def test_partial_coverage(self) -> None:
        # One row has gross (net 40 / gross 100), one doesn't (net 20).
        # gross_sum=100, covered_net=40 (NOT 60) → discount over covered
        # only; the 20 uncovered shows up via coverage, not discount.
        d = date(2026, 5, 1)
        rows = [
            BucketRow(bucket=BUCKET_BEDROCK, date=d,
                      cost_usd=Decimal("40"), cost_usd_gross=Decimal("100")),
            BucketRow(bucket=BUCKET_FOUNDRY, date=d, cost_usd=Decimal("20")),
        ]
        gross, covered, has = _sum_gross_usd(rows)
        assert has is True
        assert gross == Decimal("100")
        assert covered == Decimal("40")


class TestRollupFingerprint:
    """The rollup fingerprint must cover everything a live recompute reads
    that isn't frozen in the source rows: whitelist + FX + seat-plan
    price/currency/cycle + semantics. FX is the subtle one — the seats
    bucket reprices CNY plan costs at the LIVE rate, so an FX change must
    restale closed-month rollups or they'd serve stale seat USD. Regression
    guard against dropping FX from the hash. (The seat-plan dimension is
    pinned by ``test_seat_plan_fingerprint.py`` against a real DB; here it
    is stubbed so these whitelist/FX cases run on a bare mock db.)"""

    def _svc(self):
        svc = object.__new__(AiSpendingService)
        svc.db = AsyncMock()
        # Stub the seat-plan digest (needs a real DB) so these focused
        # whitelist/FX cases don't exercise its query against the mock.
        svc._seat_plan_fingerprint = AsyncMock(return_value="")
        return svc

    @pytest.mark.asyncio
    async def test_fx_change_changes_fingerprint(self) -> None:
        svc = self._svc()
        with patch(
            "app.domains.ai_spending.service.load_whitelist",
            AsyncMock(return_value={"aws": ["bedrock"]}),
        ):
            with patch.object(svc, "_get_cny_rate", AsyncMock(return_value=0.14)):
                fp1 = await svc._rollup_fingerprint()
            with patch.object(svc, "_get_cny_rate", AsyncMock(return_value=0.139)):
                fp2 = await svc._rollup_fingerprint()
        assert fp1 != fp2

    @pytest.mark.asyncio
    async def test_whitelist_change_changes_fingerprint(self) -> None:
        svc = self._svc()
        with patch.object(svc, "_get_cny_rate", AsyncMock(return_value=0.14)):
            with patch(
                "app.domains.ai_spending.service.load_whitelist",
                AsyncMock(return_value={"aws": ["bedrock"]}),
            ):
                fp1 = await svc._rollup_fingerprint()
            with patch(
                "app.domains.ai_spending.service.load_whitelist",
                AsyncMock(return_value={"aws": ["bedrock", "sagemaker"]}),
            ):
                fp2 = await svc._rollup_fingerprint()
        assert fp1 != fp2

    @pytest.mark.asyncio
    async def test_same_inputs_same_fingerprint(self) -> None:
        svc = self._svc()
        with patch(
            "app.domains.ai_spending.service.load_whitelist",
            AsyncMock(return_value={"aws": ["bedrock"], "gcp": ["vertex"]}),
        ), patch.object(svc, "_get_cny_rate", AsyncMock(return_value=0.14)):
            fp1 = await svc._rollup_fingerprint()
            fp2 = await svc._rollup_fingerprint()
        assert fp1 == fp2 and len(fp1) == 64