"""Pydantic-shape tests for the new 4-bucket AI Spending schemas. Round-tripping the canonical wire shapes lets the front-end depend on a stable contract. These pin: * ``AiServiceWhitelist`` rejects unknown providers at the schema layer (``extra="forbid"``) — admin payload validation happens twice (Pydantic shape + ``validate_whitelist_payload`` semantic), this test pins the first layer. * ``AiSpendingBucket`` shape with optional + nullable fields rendering consistently — front-end ``BucketKey`` enum and ``MoneyPair`` consumer code depends on these. * ``AiBucketAttribution`` rows + grand-total — chargeback table data flows through this schema; renaming a field silently here would break the table without a TS error. * ``AiSpendingSettingsUpdate`` partial-update semantics — every field is optional; an empty body is valid (no-op) and a complete body triggers every code-path in ``update_settings``. """ from __future__ import annotations from decimal import Decimal import pytest from pydantic import ValidationError from app.domains.ai_spending.schemas import ( AiBucketAttribution, AiBucketAttributionRow, AiServiceWhitelist, AiSpendingBucket, AiSpendingSettingsUpdate, MoneyPair, ) class TestAiServiceWhitelist: def test_empty_valid(self) -> None: w = AiServiceWhitelist() assert w.azure == [] assert w.aws == [] assert w.gcp == [] def test_full_valid(self) -> None: w = AiServiceWhitelist( azure=["Foundry Models", "Azure OpenAI"], aws=["Amazon Bedrock"], gcp=["Vertex AI"], ) assert w.azure == ["Foundry Models", "Azure OpenAI"] def test_unknown_provider_rejected(self) -> None: # ``extra="forbid"`` — adding ``aliyun`` would silently # bypass the whitelist semantic by routing it through neither # Bailian nor cloud-bucket. Schema rejects it loudly. with pytest.raises(ValidationError): AiServiceWhitelist(azure=["Foundry"], aliyun=["Bailian"]) # type: ignore[call-arg] def test_non_list_value_rejected(self) -> None: with pytest.raises(ValidationError): AiServiceWhitelist(azure="Foundry") # type: ignore[arg-type] class TestMoneyPair: def test_simple(self) -> None: m = MoneyPair(usd=Decimal("100.00"), cny=Decimal("720.00"), usd_per_cny=7.2) assert m.usd == Decimal("100.00") assert m.cny == Decimal("720.00") def test_serializes(self) -> None: m = MoneyPair(usd=Decimal("100.00"), cny=Decimal("720.00"), usd_per_cny=7.2) d = m.model_dump() assert "usd" in d and "cny" in d and "usd_per_cny" in d class TestAiSpendingBucket: def test_minimal_required(self) -> None: # Bucket card on the dashboard renders even when the bucket # has no data — confirms required fields can be zero/0. b = AiSpendingBucket( key="seats", label_en="AI Seats", label_zh="AI 坐席", amount=MoneyPair(usd=Decimal("0"), cny=Decimal("0"), usd_per_cny=7.2), share_pct=0.0, fact_count=0, distinct_subjects=0, ) assert b.previous_amount is None assert b.mom_change_pct is None assert b.data_caveat is None def test_with_caveat(self) -> None: b = AiSpendingBucket( key="bailian", label_en="Aliyun Bailian", label_zh="阿里云百炼", amount=MoneyPair(usd=Decimal("8000"), cny=Decimal("57600"), usd_per_cny=7.2), share_pct=42.0, fact_count=1, distinct_subjects=1, data_caveat=( "BSS gives account-level totals only — no per-API-key " "split. Treat this number as ground truth but expect " "attribution rows to show a single Aliyun account." ), ) assert b.data_caveat is not None assert "BSS" in b.data_caveat class TestAiBucketAttribution: def test_empty(self) -> None: a = AiBucketAttribution( period="2026-05", unattributed_amount=MoneyPair( usd=Decimal("0"), cny=Decimal("0"), usd_per_cny=7.2, ), rows=[], ) assert a.rows == [] assert a.unattributed_amount.usd == Decimal("0") def test_with_foundry_row(self) -> None: # Mirrors the NexusCRM case: per-RG Foundry attribution with # scope-owner fallback when the AK request was not tagged # to a specific person. row = AiBucketAttributionRow( bucket="foundry", integration_name="Azure-PingCAP", scope_display_name="NexusCRM", resource_group="NexusCRM", label_en="Azure Foundry · NexusCRM", label_zh="Azure Foundry · NexusCRM", owner_label="alice@example.com", auto_owner_source="scope_owner", period_amount=MoneyPair( usd=Decimal("510.16"), cny=Decimal("3673.15"), usd_per_cny=7.2, ), ) a = AiBucketAttribution( period="2026-05", unattributed_amount=MoneyPair( usd=Decimal("0"), cny=Decimal("0"), usd_per_cny=7.2, ), rows=[row], ) assert len(a.rows) == 1 assert a.rows[0].auto_owner_source == "scope_owner" assert a.rows[0].bucket == "foundry" assert a.rows[0].resource_group == "NexusCRM" def test_with_unattributed(self) -> None: # A bedrock row with no Owner tag — surfaces via unattributed_amount. row = AiBucketAttributionRow( bucket="bedrock", iam_user="ai-eng-team-bot", label_en="AWS Bedrock · ai-eng-team-bot", label_zh="AWS Bedrock · ai-eng-team-bot", owner_person_id=None, owner_label=None, auto_owner_source=None, period_amount=MoneyPair( usd=Decimal("42.00"), cny=Decimal("302.40"), usd_per_cny=7.2, ), ) a = AiBucketAttribution( period="2026-05", unattributed_amount=MoneyPair( usd=Decimal("42.00"), cny=Decimal("302.40"), usd_per_cny=7.2, ), rows=[row], ) assert a.unattributed_amount.usd == Decimal("42.00") assert a.rows[0].owner_person_id is None assert a.rows[0].owner_label is None class TestSettingsUpdate: def test_empty_update_valid(self) -> None: # No-op PATCH is allowed — confirms admin can hit ``Save`` # with no changes without a 422. u = AiSpendingSettingsUpdate() assert u.cny_to_usd_rate is None assert u.service_whitelist is None def test_whitelist_subfield(self) -> None: u = AiSpendingSettingsUpdate( service_whitelist=AiServiceWhitelist(azure=["Foundry"]), ) assert u.service_whitelist is not None assert u.service_whitelist.azure == ["Foundry"] def test_partial_only_rate(self) -> None: u = AiSpendingSettingsUpdate(cny_to_usd_rate=7.3) assert u.cny_to_usd_rate == 7.3 assert u.llm_region is None assert u.service_whitelist is None def test_regen_limit_range(self) -> None: # Schema-level ge=1, le=100 AiSpendingSettingsUpdate(regenerate_per_user_per_day=50) with pytest.raises(ValidationError): AiSpendingSettingsUpdate(regenerate_per_user_per_day=0) with pytest.raises(ValidationError): AiSpendingSettingsUpdate(regenerate_per_user_per_day=101)