"""Tests for ``GET /api/v1/apikeys/assignments/{id}/spend`` and the backing ``ApiKeyService.get_assignment_spend`` service method. Production-grade per CLAUDE.md: no DB mocks, real cmdb_cost_facts rows, real ApiKeyAssignment, real auth gating. Contract under test (PR #251 — per-AK product-scoped): * The MTD sum dispatches per upstream product bucket: foundry → cost_facts in the AK's RG, Foundry-family services bedrock → cost_facts tagged with the AK's IAM user vertex → cost_facts in the AK's GCP project scope bailian → data_available=False (Aliyun BSS account-level only) * Auth: applicant themselves OR an asset_admin / it_admin. Anyone else: 403. * Threshold ladder mirrors the budget cron's 80%/100%. * ``daily_series`` is zero-filled across the requested month. * The same per-AK query backs both this panel AND the ``apikey.budget_check`` cron's threshold compare — both go through ``compute_assignment_spend`` so the bar and the alert agree to the cent. See ``tests/test_apikey/test_budget_alerts.py`` for the cron-side coverage of that contract. Pre-PR-251 the panel summed every fact whose ``cost_owner_person_id == applicant.id``; that conflated unrelated scope-owner fallback attribution into a single key's bar and produced the NexusCRM AK-20260426-204D 127.5% false-positive over-budget incident (May 2026, prod). The old test cases (which exercised the per-applicant path) are replaced here by bucket-keyed cases. """ from __future__ import annotations import uuid from datetime import date, datetime from decimal import Decimal from types import SimpleNamespace from zoneinfo import ZoneInfo import pytest from sqlalchemy.ext.asyncio import AsyncSession from app.core.exceptions import ForbiddenError from app.domains.apikey.models import ApiKeyAssignment, ApiKeyProvider from app.domains.apikey.service import ApiKeyService from app.domains.cmdb.models import ( CmdbCloudScope, CmdbCostFact, CmdbIamUser, ) from app.identity.person_models import IdentityPerson from app.integrations.models import Integration # ── Fixtures-as-helpers (kept inline so each test reads end-to-end # without bouncing into conftest fixtures the reader hasn't seen). async def _make_provider( db: AsyncSession, *, name: str = "Azure OpenAI", ) -> ApiKeyProvider: p = ApiKeyProvider( id=uuid.uuid4(), name=name, description="x", is_active=True, sort_order=0, ) db.add(p) await db.commit() return p async def _make_person( db: AsyncSession, *, email: str, ) -> IdentityPerson: person = IdentityPerson( id=uuid.uuid4(), source_system="successfactors", hris_person_id=f"sf-{uuid.uuid4().hex[:8]}", display_name=email.split("@")[0], work_email=email, employment_status="active", ) db.add(person) await db.commit() return person async def _make_integration( db: AsyncSession, *, provider: str = "azure", ) -> Integration: integ = Integration( id=uuid.uuid4(), kind="cloud", provider=provider, name=f"{provider.upper()} Test", slug=f"{provider}-test-{uuid.uuid4().hex[:6]}", enabled=True, is_primary=False, capabilities=[], config_public={}, ) db.add(integ) await db.commit() return integ async def _make_rg_scope( db: AsyncSession, *, integration: Integration, rg_name: str, owner: IdentityPerson | None = None, ) -> CmdbCloudScope: """Create an azure_resource_group-typed scope (the kind Azure Foundry auto-grant mints per AK).""" scope = CmdbCloudScope( id=uuid.uuid4(), integration_id=str(integration.id), scope_type="azure_resource_group", external_id=rg_name, display_name=f"Foundry · {rg_name}", status="active", owner_person_id=str(owner.id) if owner else None, ) db.add(scope) await db.commit() return scope async def _make_assignment( db: AsyncSession, *, provider: ApiKeyProvider, applicant: IdentityPerson, monthly_budget_usd: Decimal, cloud_scope: CmdbCloudScope | None = None, iam_user_arn: str | None = None, iam_integration: Integration | None = None, iam_user: CmdbIamUser | None = None, status: str = "active", created_at: datetime | None = None, request_number: str | None = None, ) -> ApiKeyAssignment: a = ApiKeyAssignment( id=uuid.uuid4(), provider_id=provider.id, assigned_to=applicant.id, status=status, request_number=request_number or f"AK-{uuid.uuid4().hex[:8].upper()}", project_name="LLM eval", reason="test", monthly_budget_usd=monthly_budget_usd, cloud_scope_id=str(cloud_scope.id) if cloud_scope else None, iam_user_arn=iam_user_arn, iam_integration_id=( str(iam_integration.id) if iam_integration else None ), iam_user_id=iam_user.id if iam_user else None, ) if created_at is not None: a.created_at = created_at db.add(a) await db.commit() return a async def _make_iam_user( db: AsyncSession, *, integration: Integration, user_name: str, tags: dict | None = None, owner_person: IdentityPerson | None = None, ) -> CmdbIamUser: """Create a cmdb_iam_users row — the mirror of an AWS IAM user as the AWS IAM-sync writes it. ``tags`` carries the AWS IAM user's user-defined tags including ``Owner`` (the cost-allocation tag).""" arn = f"arn:aws:iam::000000000000:user/{user_name}" iam = CmdbIamUser( id=uuid.uuid4(), integration_id=str(integration.id), user_name=user_name, sub_account_id="", arn=arn, tags=tags, owner_person_id=str(owner_person.id) if owner_person else None, ) db.add(iam) await db.commit() return iam async def _add_cost_fact( db: AsyncSession, *, integration: Integration, fact_date: date, amount_usd: Decimal, service: str, resource_group: str | None = None, resource_id: str | None = None, tags: dict | None = None, scope_id: str | None = None, cost_owner_person_id: str | None = None, ) -> CmdbCostFact: fact = CmdbCostFact( fact_key=uuid.uuid4().hex, date=fact_date, integration_id=str(integration.id), service=service, amount_usd=amount_usd, amount_native=amount_usd, currency="USD", resource_group=resource_group, resource_id=resource_id, tags=tags, scope_id=scope_id, cost_owner_person_id=cost_owner_person_id, auto_owner_source="scope_owner", ) db.add(fact) await db.commit() return fact def _auth(person: IdentityPerson, *, admin: bool = False) -> SimpleNamespace: """Minimal AuthUser stand-in — the service only reads ``.id`` and ``.role_names`` off ``current_user``.""" return SimpleNamespace( id=person.id, role_names=["asset_admin"] if admin else ["employee"], ) # ── Foundry bucket ───────────────────────────────────────────────── @pytest.mark.asyncio async def test_foundry_ak_sums_facts_in_its_rg(db_session: AsyncSession): """Foundry AK's MTD = sum of cost_facts whose resource_group equals the AK's RG external_id, filtered to Foundry-family services. Facts in OTHER resource groups on the same integration must NOT count.""" provider = await _make_provider(db_session, name="Azure OpenAI") integ = await _make_integration(db_session, provider="azure") applicant = await _make_person(db_session, email="alice@pingcap.com") rg = await _make_rg_scope( db_session, integration=integ, rg_name="NexusCRM", owner=applicant, ) assignment = await _make_assignment( db_session, provider=provider, applicant=applicant, monthly_budget_usd=Decimal("400.00"), cloud_scope=rg, ) # NOTE: Use Shanghai TZ to match ``ApiKeyService.get_assignment_spend``'s # MTD reference (service.py L4775 uses ``Asia/Shanghai``). UTC vs # Shanghai diverge by date at the UTC→CST midnight boundary, causing # every ``mtd_usd`` assertion in this file to flip to 0 for ~16h after # UTC reset. Pre-existing bug surfaced 2026-06-01 by PR-E rebase — # fixed inline because it was blocking the PR-E merge. today = datetime.now(ZoneInfo("Asia/Shanghai")).date() # Fact attributable to THIS AK's RG, Foundry service — should count. await _add_cost_fact( db_session, integration=integ, fact_date=today, amount_usd=Decimal("12.00"), service="Foundry Models", resource_group="NexusCRM", ) # Fact in a DIFFERENT RG (same subscription) — must NOT count. await _add_cost_fact( db_session, integration=integ, fact_date=today, amount_usd=Decimal("99.99"), service="Foundry Models", resource_group="some-other-rg", ) # Non-AI service in the AK's RG — must NOT count (infra spend # belongs to the RG owner via the cloud billing dashboard, not # to the AK's per-product budget). await _add_cost_fact( db_session, integration=integ, fact_date=today, amount_usd=Decimal("3.00"), service="Storage", resource_group="NexusCRM", ) out = await ApiKeyService(db_session).get_assignment_spend( assignment.id, current_user=_auth(applicant), ) assert out["bucket"] == "foundry" assert out["data_available"] is True assert out["data_caveat"] is None assert out["mtd_usd"] == pytest.approx(12.00) assert out["budget_usd"] == pytest.approx(400.00) assert out["percent"] == pytest.approx(3.0) assert out["threshold_state"] == "ok" assert out["last_fact_date"] == today.isoformat() @pytest.mark.asyncio async def test_foundry_ak_without_rg_binding_reports_unavailable( db_session: AsyncSession, ): """A Foundry AK whose ``cloud_scope_id`` was never set (legacy AK, auto-grant skipped, …) cannot resolve per-AK spend. The response must signal ``data_available=False`` and include a caveat — the panel must NOT render a "0% (ok)" bar that misleads the reader into thinking spend is healthy.""" provider = await _make_provider(db_session, name="Azure OpenAI") applicant = await _make_person(db_session, email="bob@pingcap.com") assignment = await _make_assignment( db_session, provider=provider, applicant=applicant, monthly_budget_usd=Decimal("400.00"), cloud_scope=None, ) out = await ApiKeyService(db_session).get_assignment_spend( assignment.id, current_user=_auth(applicant), ) assert out["bucket"] == "foundry" assert out["data_available"] is False assert out["data_caveat"] assert "resource group" in out["data_caveat"].lower() assert out["mtd_usd"] == pytest.approx(0.0) assert out["threshold_state"] == "unknown" @pytest.mark.asyncio async def test_foundry_threshold_over_at_100_pct(db_session: AsyncSession): """A Foundry AK that genuinely exceeded 100% of its budget must report ``threshold_state='over'`` — the cron and the panel agree to the cent so the user sees the same number their Feishu alert quotes.""" provider = await _make_provider(db_session, name="Azure OpenAI") integ = await _make_integration(db_session, provider="azure") applicant = await _make_person(db_session, email="carol@pingcap.com") rg = await _make_rg_scope( db_session, integration=integ, rg_name="HighSpend-RG", owner=applicant, ) assignment = await _make_assignment( db_session, provider=provider, applicant=applicant, monthly_budget_usd=Decimal("100.00"), cloud_scope=rg, ) today = datetime.now(ZoneInfo("Asia/Shanghai")).date() await _add_cost_fact( db_session, integration=integ, fact_date=today, amount_usd=Decimal("125.00"), service="Foundry Models", resource_group="HighSpend-RG", ) out = await ApiKeyService(db_session).get_assignment_spend( assignment.id, current_user=_auth(applicant), ) assert out["threshold_state"] == "over" assert out["percent"] == pytest.approx(125.0) @pytest.mark.asyncio async def test_foundry_does_not_leak_other_applicants_spend( db_session: AsyncSession, ): """The PR #251 incident in one assertion: applicant ALICE owns subscription S; the Foundry AK with RG=AK_RG sits in S. Another applicant BOB's RG=BOB_RG also sits in S and burned $507. Pre-fix the panel summed by applicant and Alice saw Bob's $507 on Alice's AK budget bar — false 127.5% over-budget. Post-fix Alice's panel must show only her RG's spend.""" provider = await _make_provider(db_session, name="Azure OpenAI") integ = await _make_integration(db_session, provider="azure") alice = await _make_person(db_session, email="alice2@pingcap.com") rg_for_alice = await _make_rg_scope( db_session, integration=integ, rg_name="alice-rg", owner=alice, ) assignment = await _make_assignment( db_session, provider=provider, applicant=alice, monthly_budget_usd=Decimal("400.00"), cloud_scope=rg_for_alice, ) today = datetime.now(ZoneInfo("Asia/Shanghai")).date() # Bob's RG burned $507 in the same subscription — must NOT show # up under Alice's AK. await _add_cost_fact( db_session, integration=integ, fact_date=today, amount_usd=Decimal("507.00"), service="Foundry Models", resource_group="bob-rg", ) out = await ApiKeyService(db_session).get_assignment_spend( assignment.id, current_user=_auth(alice), ) assert out["mtd_usd"] == pytest.approx(0.0) assert out["percent"] == pytest.approx(0.0) assert out["threshold_state"] == "ok" # ── Bedrock bucket ───────────────────────────────────────────────── @pytest.mark.asyncio async def test_bedrock_ak_matches_on_iam_mirror_owner_tag( db_session: AsyncSession, ): """Path 1 of the Owner-tag cascade: ``cmdb_iam_users.tags['Owner']`` is authoritative — the value AWS actually has on the IAM user is what Cost Explorer's principal-tag grouping uses. AK setup mirrors a real Bedrock autogrant: the IAM user is tagged ``Owner=`` (since PR #220) and the autogrant upserts the mirror in cmdb_iam_users with the same tags. The spend query joins on the mirror's Owner value, not on the IAM username (the bug we're fixing in this PR: AWS resource_id rows are keyed by the Owner tag's VALUE, not the username). """ provider = await _make_provider(db_session, name="AWS BedRock") integ = await _make_integration(db_session, provider="aws") applicant = await _make_person(db_session, email="dan@pingcap.com") iam = await _make_iam_user( db_session, integration=integ, user_name="bedrock-LLMEval-20260512-AAAA", tags={ "Owner": "dan@pingcap.com", "helix-owner-email": "dan@pingcap.com", "helix-project-name": "LLM eval", }, owner_person=applicant, ) assignment = await _make_assignment( db_session, provider=provider, applicant=applicant, monthly_budget_usd=Decimal("300.00"), iam_user_arn=iam.arn, iam_integration=integ, iam_user=iam, ) today = datetime.now(ZoneInfo("Asia/Shanghai")).date() # This AK's Owner tag value (email) — should count. await _add_cost_fact( db_session, integration=integ, fact_date=today, amount_usd=Decimal("42.00"), service="Claude Sonnet 4.6 (Amazon Bedrock Edition)", resource_id=( "aws-principal:Owner=dan@pingcap.com|service=" "Claude Sonnet 4.6 (Amazon Bedrock Edition)" ), tags={"Owner": "dan@pingcap.com"}, ) # Different Owner tag value, same account — must NOT count. await _add_cost_fact( db_session, integration=integ, fact_date=today, amount_usd=Decimal("500.00"), service="Claude Sonnet 4.6 (Amazon Bedrock Edition)", resource_id=( "aws-principal:Owner=other@pingcap.com|service=" "Claude Sonnet 4.6 (Amazon Bedrock Edition)" ), tags={"Owner": "other@pingcap.com"}, ) # Non-Bedrock service tagged with this user — must NOT count # (whitelist filter rejects non-Bedrock-family services). await _add_cost_fact( db_session, integration=integ, fact_date=today, amount_usd=Decimal("17.00"), service="Amazon EC2", resource_id=( "aws-principal:Owner=dan@pingcap.com|service=Amazon EC2" ), tags={"Owner": "dan@pingcap.com"}, ) # (unallocated) bucket — AWS CE returns this when the principal # had no Owner tag at line-item time (Owner tag not yet activated # in Billing Console, or principal-tag value empty). Must NOT # silently roll into this AK even though they share the integration. await _add_cost_fact( db_session, integration=integ, fact_date=today, amount_usd=Decimal("128.00"), service="Amazon Bedrock", resource_id=( "aws-principal:Owner=(unallocated)|service=Amazon Bedrock" ), tags={}, ) out = await ApiKeyService(db_session).get_assignment_spend( assignment.id, current_user=_auth(applicant), ) assert out["bucket"] == "bedrock" assert out["data_available"] is True assert out["mtd_usd"] == pytest.approx(42.00) @pytest.mark.asyncio async def test_bedrock_ak_falls_back_to_person_work_email( db_session: AsyncSession, ): """Path 2 of the cascade: when the cmdb_iam_users mirror is missing the Owner tag (e.g. mirror row written before the autogrant added the tag step, or the cmdb iam sync overwrote tags from a console- edited user without the Owner key), the resolver falls through to the assignee's ``work_email``. This is what bedrock_autogrant.py:362 writes to AWS today, so for autogrant'd AKs without a current mirror, the email path lands on the same value AWS Cost Explorer is grouping by. """ provider = await _make_provider(db_session, name="AWS BedRock") integ = await _make_integration(db_session, provider="aws") applicant = await _make_person(db_session, email="path2@pingcap.com") # Mirror exists but has NO Owner tag — represents a stale sync. iam = await _make_iam_user( db_session, integration=integ, user_name="bedrock-Path2-20260512-BBBB", tags={"helix-owner-email": "path2@pingcap.com"}, owner_person=applicant, ) assignment = await _make_assignment( db_session, provider=provider, applicant=applicant, monthly_budget_usd=Decimal("100.00"), iam_user_arn=iam.arn, iam_integration=integ, iam_user=iam, ) today = datetime.now(ZoneInfo("Asia/Shanghai")).date() await _add_cost_fact( db_session, integration=integ, fact_date=today, amount_usd=Decimal("19.00"), service="Amazon Bedrock", resource_id="aws-principal:Owner=path2@pingcap.com|service=Amazon Bedrock", tags={"Owner": "path2@pingcap.com"}, ) out = await ApiKeyService(db_session).get_assignment_spend( assignment.id, current_user=_auth(applicant), ) assert out["bucket"] == "bedrock" assert out["data_available"] is True assert out["mtd_usd"] == pytest.approx(19.00) @pytest.mark.asyncio async def test_bedrock_ak_legacy_fallback_to_iam_username( db_session: AsyncSession, ): """Path 3 of the cascade: pre-PR-220 rows where AWS was tagged ``Owner=`` (the convention before email tagging landed). Mirror has no Owner key; the bound person has no work_email; resolver returns the IAM username and the AK still sees its historic spend. """ provider = await _make_provider(db_session, name="AWS BedRock") integ = await _make_integration(db_session, provider="aws") # Bind a person with empty work_email so path 2 misses (schema # is NOT NULL; empty string is the legal "missing" signal here). applicant = IdentityPerson( id=uuid.uuid4(), source_system="successfactors", hris_person_id=f"sf-{uuid.uuid4().hex[:8]}", display_name="legacy-user", work_email=f"legacy-empty-{uuid.uuid4().hex[:8]}@nodomain.invalid", employment_status="active", ) db_session.add(applicant) await db_session.commit() # Now blank out work_email so the cascade falls through path 2. # We commit a non-empty placeholder first to satisfy the unique # constraint, then UPDATE to empty in a separate statement — # mirrors how a real "person has no contact email yet" row could # look in the wild. from sqlalchemy import update await db_session.execute( update(IdentityPerson) .where(IdentityPerson.id == applicant.id) .values(work_email=""), ) await db_session.commit() iam = await _make_iam_user( db_session, integration=integ, user_name="bedrock-Legacy-20260101-CCCC", tags=None, owner_person=applicant, ) assignment = await _make_assignment( db_session, provider=provider, applicant=applicant, monthly_budget_usd=Decimal("80.00"), iam_user_arn=iam.arn, iam_integration=integ, iam_user=iam, ) today = datetime.now(ZoneInfo("Asia/Shanghai")).date() await _add_cost_fact( db_session, integration=integ, fact_date=today, amount_usd=Decimal("7.50"), service="Amazon Bedrock", resource_id=( "aws-principal:Owner=bedrock-Legacy-20260101-CCCC|" "service=Amazon Bedrock" ), tags={"Owner": "bedrock-Legacy-20260101-CCCC"}, ) out = await ApiKeyService(db_session).get_assignment_spend( assignment.id, current_user=_auth(applicant), ) assert out["mtd_usd"] == pytest.approx(7.50) @pytest.mark.asyncio async def test_bedrock_ak_handles_owner_tag_with_like_metachars( db_session: AsyncSession, ): """Path 1 with a LIKE metachar in the Owner value (rare but possible — emails can theoretically contain ``%`` in the local part per RFC 5321, and historic Owner= rows could have arbitrary text). The LIKE pattern in spend.py escapes ``%``, ``_``, ``\\`` before splicing, so an Owner=``a%pct`` must match EXACTLY ``a%pct`` — NOT every Owner starting with ``a``. """ provider = await _make_provider(db_session, name="AWS BedRock") integ = await _make_integration(db_session, provider="aws") applicant = await _make_person(db_session, email="meta@pingcap.com") iam = await _make_iam_user( db_session, integration=integ, user_name="bedrock-Meta-Demo", tags={"Owner": "a%pct"}, owner_person=applicant, ) assignment = await _make_assignment( db_session, provider=provider, applicant=applicant, monthly_budget_usd=Decimal("100.00"), iam_user_arn=iam.arn, iam_integration=integ, iam_user=iam, ) today = datetime.now(ZoneInfo("Asia/Shanghai")).date() # Literal match — should count. await _add_cost_fact( db_session, integration=integ, fact_date=today, amount_usd=Decimal("7.00"), service="Amazon Bedrock", resource_id="aws-principal:Owner=a%pct|service=Amazon Bedrock", tags={"Owner": "a%pct"}, ) # Would over-match if ``%`` were treated as a wildcard. await _add_cost_fact( db_session, integration=integ, fact_date=today, amount_usd=Decimal("500.00"), service="Amazon Bedrock", resource_id=( "aws-principal:Owner=alarmpctbignumber|service=Amazon Bedrock" ), tags={"Owner": "alarmpctbignumber"}, ) out = await ApiKeyService(db_session).get_assignment_spend( assignment.id, current_user=_auth(applicant), ) assert out["mtd_usd"] == pytest.approx(7.0) # NOT 507.0 @pytest.mark.asyncio async def test_bedrock_ak_without_iam_binding_reports_unavailable( db_session: AsyncSession, ): """A Bedrock AK whose iam_user_arn is unset AND iam_user_id is unset (auto-grant fell back, legacy row, manual fill without rebind) cannot resolve per-AK spend — the per-user cost-allocation tag chain requires SOME IAM binding to know which Owner tag value AWS is keying rows under.""" provider = await _make_provider(db_session, name="AWS BedRock") applicant = await _make_person(db_session, email="erin@pingcap.com") assignment = await _make_assignment( db_session, provider=provider, applicant=applicant, monthly_budget_usd=Decimal("300.00"), iam_user_arn=None, ) out = await ApiKeyService(db_session).get_assignment_spend( assignment.id, current_user=_auth(applicant), ) assert out["bucket"] == "bedrock" assert out["data_available"] is False assert out["data_caveat"] assert "iam user" in out["data_caveat"].lower() assert out["threshold_state"] == "unknown" @pytest.mark.asyncio async def test_bedrock_ak_with_arn_but_no_resolvable_owner_returns_distinct_caveat( db_session: AsyncSession, ): """A Bedrock AK whose iam_user_arn is set but ALL three resolver paths miss (no mirror row → no path 1; person has no work_email → no path 2; ARN un-parseable → no path 3) must surface the no-owner-tag caveat — distinct from the no-iam-binding caveat so the admin sees actionable text (re-tag in AWS / run backfill, not rebind). """ provider = await _make_provider(db_session, name="AWS BedRock") integ = await _make_integration(db_session, provider="aws") # Bind a person with empty work_email (column is NOT NULL, so we # insert a placeholder then blank it). applicant = IdentityPerson( id=uuid.uuid4(), source_system="successfactors", hris_person_id=f"sf-{uuid.uuid4().hex[:8]}", display_name="no-email", work_email=f"placeholder-{uuid.uuid4().hex[:8]}@nodomain.invalid", employment_status="active", ) db_session.add(applicant) await db_session.commit() from sqlalchemy import update await db_session.execute( update(IdentityPerson) .where(IdentityPerson.id == applicant.id) .values(work_email=""), ) await db_session.commit() # ARN whose path 3 parser returns None (no ``:user/`` segment). assignment = await _make_assignment( db_session, provider=provider, applicant=applicant, monthly_budget_usd=Decimal("50.00"), iam_user_arn="arn:aws:iam::000:role/SomeRoleNotAUser", iam_integration=integ, ) out = await ApiKeyService(db_session).get_assignment_spend( assignment.id, current_user=_auth(applicant), ) assert out["bucket"] == "bedrock" assert out["data_available"] is False # Must NOT collapse with the no-binding caveat — admin needs to # see "backfill the Owner tag" vs "rebind the IAM user". assert "owner" in out["data_caveat"].lower() assert "backfill" in out["data_caveat"].lower() or \ "cost-allocation" in out["data_caveat"].lower() @pytest.mark.asyncio async def test_bedrock_ak_surfaces_unallocated_caveat_when_zero_with_unallocated_facts( db_session: AsyncSession, ): """The 2026-05-12 production scenario: AK resolves an Owner tag value correctly (path 1 or 2), but AWS Cost Explorer has never seen the ``Owner`` cost-allocation tag activated in Billing Console, so every cost fact lands in the ``(unallocated)`` bucket. The AK matches NO facts → mtd_usd = $0, but the integration HAS Bedrock cost in (unallocated). Without a caveat the panel renders an under-budget green check and the user thinks they spend $0 on AI. """ provider = await _make_provider(db_session, name="AWS BedRock") integ = await _make_integration(db_session, provider="aws") applicant = await _make_person(db_session, email="unalloc@pingcap.com") iam = await _make_iam_user( db_session, integration=integ, user_name="bedrock-Unalloc-DEMO", tags={"Owner": "unalloc@pingcap.com"}, owner_person=applicant, ) assignment = await _make_assignment( db_session, provider=provider, applicant=applicant, monthly_budget_usd=Decimal("100.00"), iam_user_arn=iam.arn, iam_integration=integ, iam_user=iam, ) today = datetime.now(ZoneInfo("Asia/Shanghai")).date() # No facts for this AK's Owner value; everything is in (unallocated). await _add_cost_fact( db_session, integration=integ, fact_date=today, amount_usd=Decimal("128.00"), service="Claude Sonnet 4.6 (Amazon Bedrock Edition)", resource_id=( "aws-principal:Owner=(unallocated)|" "service=Claude Sonnet 4.6 (Amazon Bedrock Edition)" ), tags={}, ) out = await ApiKeyService(db_session).get_assignment_spend( assignment.id, current_user=_auth(applicant), ) # mtd is correctly $0 for this AK (the AWS Owner tag never reached # CE) but the user sees the actionable caveat — "activate the tag # in Billing Console" — instead of a silent green check. assert out["bucket"] == "bedrock" assert out["mtd_usd"] == pytest.approx(0.0) assert out["data_caveat"] caveat_text = out["data_caveat"].lower() assert "unallocated" in caveat_text assert "billing console" in caveat_text @pytest.mark.asyncio async def test_bedrock_ak_no_unallocated_caveat_when_real_spend_present( db_session: AsyncSession, ): """Sibling of the prior test: the integration has BOTH a matching fact (real spend for this AK's Owner) AND (unallocated) rows. The AK's mtd_usd is non-zero, so the panel must NOT show the unallocated caveat — the AK itself is healthy; the (unallocated) bucket belongs on the global dashboard's "未分摊" KPI, not on a per-AK panel. """ provider = await _make_provider(db_session, name="AWS BedRock") integ = await _make_integration(db_session, provider="aws") applicant = await _make_person(db_session, email="mixed@pingcap.com") iam = await _make_iam_user( db_session, integration=integ, user_name="bedrock-Mixed-DEMO", tags={"Owner": "mixed@pingcap.com"}, owner_person=applicant, ) assignment = await _make_assignment( db_session, provider=provider, applicant=applicant, monthly_budget_usd=Decimal("100.00"), iam_user_arn=iam.arn, iam_integration=integ, iam_user=iam, ) today = datetime.now(ZoneInfo("Asia/Shanghai")).date() await _add_cost_fact( db_session, integration=integ, fact_date=today, amount_usd=Decimal("5.00"), service="Amazon Bedrock", resource_id="aws-principal:Owner=mixed@pingcap.com|service=Amazon Bedrock", tags={"Owner": "mixed@pingcap.com"}, ) await _add_cost_fact( db_session, integration=integ, fact_date=today, amount_usd=Decimal("99.00"), service="Amazon Bedrock", resource_id="aws-principal:Owner=(unallocated)|service=Amazon Bedrock", tags={}, ) out = await ApiKeyService(db_session).get_assignment_spend( assignment.id, current_user=_auth(applicant), ) assert out["mtd_usd"] == pytest.approx(5.0) assert out["data_caveat"] is None @pytest.mark.asyncio async def test_bedrock_ak_unallocated_owner_value_is_never_returned( db_session: AsyncSession, ): """When the mirror row has ``Owner='(unallocated)'`` (e.g. the iam-sync wrote whatever Cost Explorer reported back), the resolver must NOT return the sentinel — that would attribute every untagged AWS principal's cost to this AK. Falls through to subsequent cascade steps. """ provider = await _make_provider(db_session, name="AWS BedRock") integ = await _make_integration(db_session, provider="aws") applicant = await _make_person( db_session, email="real-owner@pingcap.com", ) iam = await _make_iam_user( db_session, integration=integ, user_name="bedrock-Unalloc", # Bogus mirror — sentinel snuck in via sync; resolver must skip. tags={"Owner": "(unallocated)"}, owner_person=applicant, ) assignment = await _make_assignment( db_session, provider=provider, applicant=applicant, monthly_budget_usd=Decimal("60.00"), iam_user_arn=iam.arn, iam_integration=integ, iam_user=iam, ) today = datetime.now(ZoneInfo("Asia/Shanghai")).date() # Email-owned cost fact — must match via path 2 fallback, NOT via # the sentinel in path 1. await _add_cost_fact( db_session, integration=integ, fact_date=today, amount_usd=Decimal("3.00"), service="Amazon Bedrock", resource_id=( "aws-principal:Owner=real-owner@pingcap.com|" "service=Amazon Bedrock" ), tags={"Owner": "real-owner@pingcap.com"}, ) # If the resolver had returned ``(unallocated)``, this big row # would have been wrongly attributed to the AK. await _add_cost_fact( db_session, integration=integ, fact_date=today, amount_usd=Decimal("999.00"), service="Amazon Bedrock", resource_id="aws-principal:Owner=(unallocated)|service=Amazon Bedrock", tags={}, ) out = await ApiKeyService(db_session).get_assignment_spend( assignment.id, current_user=_auth(applicant), ) assert out["mtd_usd"] == pytest.approx(3.0) # NOT 1002.0 # ── Bailian bucket — honest "not available from BSS" ────────────── @pytest.mark.asyncio async def test_bailian_ak_no_linked_key_reports_unavailable( db_session: AsyncSession, ): """A Bailian assignment with no synced Bailian key yet can't resolve per-key spend → ``data_available=False`` with the no-key caveat (the panel renders the caveat, not a false "0% — under budget" check).""" provider = await _make_provider( db_session, name="Alibaba Cloud Model Studio", ) applicant = await _make_person(db_session, email="frank@pingcap.com") assignment = await _make_assignment( db_session, provider=provider, applicant=applicant, monthly_budget_usd=Decimal("100.00"), ) out = await ApiKeyService(db_session).get_assignment_spend( assignment.id, current_user=_auth(applicant), ) assert out["bucket"] == "bailian" assert out["data_available"] is False assert "no synced api key" in (out["data_caveat"] or "").lower() assert out["mtd_usd"] == pytest.approx(0.0) assert out["threshold_state"] == "unknown" @pytest.mark.asyncio async def test_bailian_ak_per_key_spend_resolved( db_session: AsyncSession, ): """When the assignment has a synced Bailian key (migration 299), per-AK spend sums ``bailian_usage_daily`` for that key — data_available=True, MTD reconciles, threshold evaluates against the budget.""" from app.domains.bailian.models import BailianApiKey, BailianUsageDaily provider = await _make_provider( db_session, name="Alibaba Cloud Model Studio", ) applicant = await _make_person(db_session, email="grace@pingcap.com") integ = await _make_integration(db_session, provider="alicloud_intl") assignment = await _make_assignment( db_session, provider=provider, applicant=applicant, monthly_budget_usd=Decimal("100.00"), ) key = BailianApiKey( id=uuid.uuid4(), integration_id=str(integ.id), api_key_id="998877", display_name="Helix · grace", owner_person_id=str(applicant.id), assignment_id=assignment.id, status="active", ) db_session.add(key) await db_session.flush() today = datetime.now(ZoneInfo("Asia/Shanghai")).date() mstart = today.replace(day=1) # Aggregate by (date, model) so the rows never collide on # bailian_usage_daily's (integration, date, account, key, model) unique # key — on the 1st of the month ``today == mstart`` and the two # qwen3.6-plus rows would otherwise be a duplicate insert (1062). MTD # still sums to 6.0 regardless of the day. by_key: dict[tuple, Decimal] = {} for d, model, usd in ( (mstart, "qwen3.6-plus", "3.000000"), (mstart, "deepseek-v4-pro", "2.000000"), (today, "qwen3.6-plus", "1.000000"), ): by_key[(d, model)] = by_key.get((d, model), Decimal("0")) + Decimal(usd) for (d, model), usd in by_key.items(): db_session.add(BailianUsageDaily( integration_id=str(integ.id), api_key_pk=key.id, bailian_api_key_id="998877", aliyun_account_id="5962436752113096", usage_date=d, model_name=model, cost_usd=usd, cost_cny=Decimal("0"), )) await db_session.commit() out = await ApiKeyService(db_session).get_assignment_spend( assignment.id, current_user=_auth(applicant), ) assert out["bucket"] == "bailian" assert out["data_available"] is True assert out["data_caveat"] is None assert out["mtd_usd"] == pytest.approx(6.0) assert out["percent"] == pytest.approx(6.0) # 6 / 100 budget assert out["threshold_state"] == "ok" # ── Vertex AI bucket ─────────────────────────────────────────────── @pytest.mark.asyncio async def test_vertex_ak_matches_on_gcp_project_scope( db_session: AsyncSession, ): """Vertex AI AK's MTD = sum of cost_facts in the AK's GCP project scope filtered to the Vertex / Generative AI service family.""" provider = await _make_provider(db_session, name="Google Vertex AI") integ = await _make_integration(db_session, provider="gcp") applicant = await _make_person(db_session, email="gina@pingcap.com") project_scope = CmdbCloudScope( id=uuid.uuid4(), integration_id=str(integ.id), scope_type="gcp_project", external_id="my-gcp-project", display_name="my-gcp-project", status="active", owner_person_id=str(applicant.id), ) db_session.add(project_scope) await db_session.commit() assignment = await _make_assignment( db_session, provider=provider, applicant=applicant, monthly_budget_usd=Decimal("100.00"), cloud_scope=project_scope, ) today = datetime.now(ZoneInfo("Asia/Shanghai")).date() await _add_cost_fact( db_session, integration=integ, fact_date=today, amount_usd=Decimal("8.40"), service="Vertex AI", scope_id=str(project_scope.id), ) # Vertex spend in a DIFFERENT project — must NOT count. other_scope = CmdbCloudScope( id=uuid.uuid4(), integration_id=str(integ.id), scope_type="gcp_project", external_id="other-project", display_name="other-project", status="active", ) db_session.add(other_scope) await db_session.commit() await _add_cost_fact( db_session, integration=integ, fact_date=today, amount_usd=Decimal("88.00"), service="Vertex AI", scope_id=str(other_scope.id), ) # Non-Vertex service in this project — must NOT count. await _add_cost_fact( db_session, integration=integ, fact_date=today, amount_usd=Decimal("11.00"), service="Compute Engine", scope_id=str(project_scope.id), ) out = await ApiKeyService(db_session).get_assignment_spend( assignment.id, current_user=_auth(applicant), ) assert out["bucket"] == "vertex" assert out["data_available"] is True assert out["mtd_usd"] == pytest.approx(8.40) # ── Daily series + month window ─────────────────────────────────── @pytest.mark.asyncio async def test_daily_series_truncated_at_last_fact_date( db_session: AsyncSession, ): """``daily_series`` ends at ``last_fact_date`` — trailing days for which cost_facts has no rows are NOT zero-filled. Pre-2026-05-18 behaviour zero-filled the entire month (31 rows for March even when the latest fact was on 3/20). That rendered as $0 sparkline bars on un-ingested trailing days, misleading users into thinking they had spent nothing on those days. The series now stops at the last day with data (3/20 → 20 rows for 3/1..3/20), with the drawer's data-lag banner / ``last_fact_date`` field communicating the freshness gap explicitly. Mid-month gaps (no fact on 3/10 but facts on 3/5 + 3/20) still zero-fill INSIDE the kept range — those days are 0 contribution to the sum (mathematically correct) and a chart "hole" would mislead worse than a single $0 dip. """ provider = await _make_provider(db_session, name="Azure OpenAI") integ = await _make_integration(db_session, provider="azure") applicant = await _make_person(db_session, email="hank@pingcap.com") rg = await _make_rg_scope( db_session, integration=integ, rg_name="series-rg", owner=applicant, ) assignment = await _make_assignment( db_session, provider=provider, applicant=applicant, monthly_budget_usd=Decimal("100.00"), cloud_scope=rg, ) # Two facts in March 2026; last is 3/20. await _add_cost_fact( db_session, integration=integ, fact_date=date(2026, 3, 5), amount_usd=Decimal("12.00"), service="Foundry Models", resource_group="series-rg", ) await _add_cost_fact( db_session, integration=integ, fact_date=date(2026, 3, 20), amount_usd=Decimal("8.00"), service="Foundry Models", resource_group="series-rg", ) out = await ApiKeyService(db_session).get_assignment_spend( assignment.id, current_user=_auth(applicant), month="2026-03", ) # Series runs 3/1..3/20 (20 days), NOT 3/1..3/31. Trailing un- # ingested days are dropped so the sparkline doesn't draw false $0. assert len(out["daily_series"]) == 20 assert out["daily_series"][-1]["date"] == "2026-03-20" assert out["last_fact_date"] == "2026-03-20" assert out["mtd_usd"] == pytest.approx(20.0) march_5 = next( r for r in out["daily_series"] if r["date"] == "2026-03-05" ) march_20 = next( r for r in out["daily_series"] if r["date"] == "2026-03-20" ) assert march_5["usd"] == pytest.approx(12.0) assert march_20["usd"] == pytest.approx(8.0) # Mid-month gap (3/10 — no fact, between 3/5 and 3/20) IS zero- # filled (still inside the kept range). march_10 = next( r for r in out["daily_series"] if r["date"] == "2026-03-10" ) assert march_10["usd"] == pytest.approx(0.0) @pytest.mark.asyncio async def test_explicit_month_isolates_other_months( db_session: AsyncSession, ): """``month=YYYY-MM`` restricts to that month — passing 2026-03 must not include facts on 2026-04-01. Series length follows ``last_fact_date`` (truncation contract; see ``test_daily_series_truncated_at_last_fact_date``): March's only fact is on 3/15 → 15 rows; April's only fact is on 4/1 → 1 row. Pre-2026-05-18 the series was zero-filled to month_end (31 / 30 rows) regardless of where the data stopped. """ provider = await _make_provider(db_session, name="Azure OpenAI") integ = await _make_integration(db_session, provider="azure") applicant = await _make_person(db_session, email="iris@pingcap.com") rg = await _make_rg_scope( db_session, integration=integ, rg_name="iso-rg", owner=applicant, ) assignment = await _make_assignment( db_session, provider=provider, applicant=applicant, monthly_budget_usd=Decimal("100.00"), cloud_scope=rg, ) await _add_cost_fact( db_session, integration=integ, fact_date=date(2026, 3, 15), amount_usd=Decimal("30.00"), service="Foundry Models", resource_group="iso-rg", ) await _add_cost_fact( db_session, integration=integ, fact_date=date(2026, 4, 1), amount_usd=Decimal("70.00"), service="Foundry Models", resource_group="iso-rg", ) march = await ApiKeyService(db_session).get_assignment_spend( assignment.id, current_user=_auth(applicant), month="2026-03", ) assert march["mtd_usd"] == pytest.approx(30.0) # March: only fact is on 3/15 → series 3/1..3/15 = 15 rows. assert len(march["daily_series"]) == 15 assert march["last_fact_date"] == "2026-03-15" april = await ApiKeyService(db_session).get_assignment_spend( assignment.id, current_user=_auth(applicant), month="2026-04", ) assert april["mtd_usd"] == pytest.approx(70.0) # April: only fact is on 4/1 → series 4/1..4/1 = 1 row. assert len(april["daily_series"]) == 1 assert april["last_fact_date"] == "2026-04-01" # ── Empty / no-data ──────────────────────────────────────────────── @pytest.mark.asyncio async def test_empty_window_returns_structured_zero( db_session: AsyncSession, ): """No facts → structured zero (not 404 / null) so the UI still renders the bar at 0%. Foundry bucket with binding present → data_available=True, mtd=0.""" provider = await _make_provider(db_session, name="Azure OpenAI") integ = await _make_integration(db_session, provider="azure") applicant = await _make_person(db_session, email="jane@pingcap.com") rg = await _make_rg_scope( db_session, integration=integ, rg_name="empty-rg", owner=applicant, ) assignment = await _make_assignment( db_session, provider=provider, applicant=applicant, monthly_budget_usd=Decimal("100.00"), cloud_scope=rg, ) out = await ApiKeyService(db_session).get_assignment_spend( assignment.id, current_user=_auth(applicant), ) assert out["data_available"] is True assert out["mtd_usd"] == pytest.approx(0.0) assert out["percent"] == pytest.approx(0.0) assert out["threshold_state"] == "ok" assert out["last_fact_date"] is None # ── Auth ─────────────────────────────────────────────────────────── @pytest.mark.asyncio async def test_rejects_unauthorised_caller(db_session: AsyncSession): """Non-owner non-admin → 403.""" provider = await _make_provider(db_session, name="Azure OpenAI") applicant = await _make_person(db_session, email="kate@pingcap.com") snooper = await _make_person(db_session, email="snooper@pingcap.com") assignment = await _make_assignment( db_session, provider=provider, applicant=applicant, monthly_budget_usd=Decimal("100.00"), ) with pytest.raises(ForbiddenError): await ApiKeyService(db_session).get_assignment_spend( assignment.id, current_user=_auth(snooper), ) @pytest.mark.asyncio async def test_admin_can_view_any_assignment(db_session: AsyncSession): """Asset / IT admins can read any assignment's spend — required for the admin Key list.""" provider = await _make_provider(db_session, name="Azure OpenAI") applicant = await _make_person(db_session, email="leo@pingcap.com") admin = await _make_person(db_session, email="admin@pingcap.com") assignment = await _make_assignment( db_session, provider=provider, applicant=applicant, monthly_budget_usd=Decimal("100.00"), ) out = await ApiKeyService(db_session).get_assignment_spend( assignment.id, current_user=_auth(admin, admin=True), ) assert out["assignment_id"] == str(assignment.id) @pytest.mark.asyncio async def test_invalid_month_format_raises(db_session: AsyncSession): from app.core.exceptions import ValidationError provider = await _make_provider(db_session, name="Azure OpenAI") applicant = await _make_person(db_session, email="mike@pingcap.com") assignment = await _make_assignment( db_session, provider=provider, applicant=applicant, monthly_budget_usd=Decimal("100.00"), ) with pytest.raises(ValidationError): await ApiKeyService(db_session).get_assignment_spend( assignment.id, current_user=_auth(applicant), month="2026/05", ) @pytest.mark.asyncio async def test_single_digit_month_format_rejected( db_session: AsyncSession, ): """Strict ``YYYY-MM`` validation: ``2026-5`` (single-digit month) must reject. Pre-fix the int-cast accepted it and the response rendered ``month="2026-05"`` — a display drift silently masking a malformed caller (CR1 #11).""" from app.core.exceptions import ValidationError provider = await _make_provider(db_session, name="Azure OpenAI") applicant = await _make_person(db_session, email="nan@pingcap.com") assignment = await _make_assignment( db_session, provider=provider, applicant=applicant, monthly_budget_usd=Decimal("100.00"), ) with pytest.raises(ValidationError): await ApiKeyService(db_session).get_assignment_spend( assignment.id, current_user=_auth(applicant), month="2026-5", ) # ── Bucket classifier edge cases ─────────────────────────────────── @pytest.mark.asyncio async def test_bare_openai_provider_does_not_classify_as_foundry( db_session: AsyncSession, ): """A provider literally named ``OpenAI`` (no Azure prefix) is NOT Foundry — it represents a direct OpenAI SaaS connection that doesn't live in any cloud's cost_facts table. The bucket classifier must return ``unknown`` so the panel renders an honest caveat instead of running a Foundry query that would silently match nothing AND look like "$0 / 0% / under budget" (CR1 #3 — bare-OpenAI over-match was the original concern). """ from app.domains.apikey.spend import ( BUCKET_FOUNDRY, BUCKET_UNKNOWN, classify_bucket, ) assert classify_bucket("OpenAI") == BUCKET_UNKNOWN assert classify_bucket("openai") == BUCKET_UNKNOWN assert classify_bucket("Direct OpenAI API") == BUCKET_UNKNOWN # Sanity: actual Azure OpenAI provider still classifies as Foundry. assert classify_bucket("Azure OpenAI") == BUCKET_FOUNDRY assert classify_bucket("Azure OpenAI") == BUCKET_FOUNDRY # 2 spaces @pytest.mark.asyncio async def test_two_aks_sharing_one_rg_both_report_full_rg_spend( db_session: AsyncSession, ): """Two Foundry AKs bound to the SAME ``cloud_scope_id`` (unusual — autogrant mints one RG per AK; happens only via admin rebind / legacy data) BOTH report the RG's full spend. Documented behaviour — there is no "split this RG's cost between two AKs" semantics in the data model; admins re-binding two AKs to one RG opt into a shared view. CR2 #5.""" provider = await _make_provider(db_session, name="Azure OpenAI") integ = await _make_integration(db_session, provider="azure") alice = await _make_person(db_session, email="alice-share@pingcap.com") bob = await _make_person(db_session, email="bob-share@pingcap.com") rg = await _make_rg_scope( db_session, integration=integ, rg_name="shared-rg", owner=alice, ) ak_alice = await _make_assignment( db_session, provider=provider, applicant=alice, monthly_budget_usd=Decimal("200.00"), cloud_scope=rg, ) ak_bob = await _make_assignment( db_session, provider=provider, applicant=bob, monthly_budget_usd=Decimal("300.00"), cloud_scope=rg, ) today = datetime.now(ZoneInfo("Asia/Shanghai")).date() await _add_cost_fact( db_session, integration=integ, fact_date=today, amount_usd=Decimal("50.00"), service="Foundry Models", resource_group="shared-rg", ) out_alice = await ApiKeyService(db_session).get_assignment_spend( ak_alice.id, current_user=_auth(alice), ) out_bob = await ApiKeyService(db_session).get_assignment_spend( ak_bob.id, current_user=_auth(bob), ) # Both AKs see the full $50 — the RG is shared, the cost is # shared. Each percent differs because budgets differ. assert out_alice["mtd_usd"] == pytest.approx(50.0) assert out_bob["mtd_usd"] == pytest.approx(50.0) assert out_alice["percent"] == pytest.approx(25.0) # 50/200 assert out_bob["percent"] == pytest.approx(16.67, abs=0.01) # 50/300 # ── #1776: shared Owner-tag collision across a person's Bedrock keys ── # AWS Cost Explorer attributes Bedrock spend at the Owner tag (= # work_email), NOT per IAM user / per key. When one person holds >1 live # Bedrock key, every key resolves to the SAME Owner total; reporting that # total on each key N×-inflates the fleet rollup and trips false budget # breaches. The total is pinned to a single deterministic canonical key # (earliest-created); siblings report data_available=False + a caveat. async def _make_bedrock_key( db: AsyncSession, *, provider: ApiKeyProvider, integ: Integration, applicant: IdentityPerson, user_name: str, owner_email: str, created_at: datetime, request_number: str, budget: Decimal = Decimal("300.00"), ) -> ApiKeyAssignment: iam = await _make_iam_user( db, integration=integ, user_name=user_name, tags={"Owner": owner_email}, owner_person=applicant, ) return await _make_assignment( db, provider=provider, applicant=applicant, monthly_budget_usd=budget, iam_user_arn=iam.arn, iam_integration=integ, iam_user=iam, created_at=created_at, request_number=request_number, ) @pytest.mark.asyncio async def test_bedrock_shared_owner_tag_pins_total_to_canonical( db_session: AsyncSession, ): """One person, two live Bedrock keys, one shared Owner tag: the earliest-created key reports the full Owner total; the later key reports data_available=False with the shared-owner caveat naming the canonical key — never the same total twice.""" provider = await _make_provider(db_session, name="AWS BedRock") integ = await _make_integration(db_session, provider="aws") dan = await _make_person(db_session, email="dan@pingcap.com") canonical = await _make_bedrock_key( db_session, provider=provider, integ=integ, applicant=dan, user_name="bedrock-first-20260101-AAAA", owner_email="dan@pingcap.com", created_at=datetime(2026, 1, 1, 9, 0, 0), request_number="AK-FIRST01", ) sibling = await _make_bedrock_key( db_session, provider=provider, integ=integ, applicant=dan, user_name="bedrock-second-20260601-BBBB", owner_email="dan@pingcap.com", created_at=datetime(2026, 6, 1, 9, 0, 0), request_number="AK-SECOND02", ) today = datetime.now(ZoneInfo("Asia/Shanghai")).date() await _add_cost_fact( db_session, integration=integ, fact_date=today, amount_usd=Decimal("42.00"), service="Claude Sonnet 4.6 (Amazon Bedrock Edition)", resource_id=( "aws-principal:Owner=dan@pingcap.com|service=" "Claude Sonnet 4.6 (Amazon Bedrock Edition)" ), tags={"Owner": "dan@pingcap.com"}, ) out_canon = await ApiKeyService(db_session).get_assignment_spend( canonical.id, current_user=_auth(dan), ) out_sib = await ApiKeyService(db_session).get_assignment_spend( sibling.id, current_user=_auth(dan), ) # Canonical (earliest) carries the full Owner total. assert out_canon["bucket"] == "bedrock" assert out_canon["data_available"] is True assert out_canon["mtd_usd"] == pytest.approx(42.00) # Sibling reports no isolated number + the shared-owner caveat. assert out_sib["bucket"] == "bedrock" assert out_sib["data_available"] is False assert out_sib["mtd_usd"] == pytest.approx(0.0) assert "Owner cost-allocation tag" in (out_sib["data_caveat"] or "") assert "AK-FIRST01" in (out_sib["data_caveat"] or "") @pytest.mark.asyncio async def test_bedrock_fleet_sum_not_double_counted(db_session: AsyncSession): """The fleet rollup (analytics/budget-cron pattern) sums compute_assignment_spend over every key. With the collision pin the two colliding keys sum to the single Owner total, not 2×.""" from app.domains.apikey.spend import compute_assignment_spend provider = await _make_provider(db_session, name="AWS BedRock") integ = await _make_integration(db_session, provider="aws") dan = await _make_person(db_session, email="dan@pingcap.com") k1 = await _make_bedrock_key( db_session, provider=provider, integ=integ, applicant=dan, user_name="bedrock-a-20260101-AAAA", owner_email="dan@pingcap.com", created_at=datetime(2026, 1, 1, 9, 0, 0), request_number="AK-A01", ) k2 = await _make_bedrock_key( db_session, provider=provider, integ=integ, applicant=dan, user_name="bedrock-b-20260601-BBBB", owner_email="dan@pingcap.com", created_at=datetime(2026, 6, 1, 9, 0, 0), request_number="AK-B02", ) today = datetime.now(ZoneInfo("Asia/Shanghai")).date() month_start = today.replace(day=1) await _add_cost_fact( db_session, integration=integ, fact_date=today, amount_usd=Decimal("100.00"), service="Amazon Bedrock", resource_id="aws-principal:Owner=dan@pingcap.com|service=Amazon Bedrock", tags={"Owner": "dan@pingcap.com"}, ) total = Decimal(0) for k in (k1, k2): s = await compute_assignment_spend( db_session, assignment=k, month_start=month_start, month_end=today, ) total += s["mtd_usd"] assert total == pytest.approx(100.00) # once, not 200 @pytest.mark.asyncio async def test_bedrock_single_key_no_collision_unchanged( db_session: AsyncSession, ): """Regression guard: the common single-key case is untouched — full total, data_available=True, no shared caveat.""" provider = await _make_provider(db_session, name="AWS BedRock") integ = await _make_integration(db_session, provider="aws") eve = await _make_person(db_session, email="eve@pingcap.com") only = await _make_bedrock_key( db_session, provider=provider, integ=integ, applicant=eve, user_name="bedrock-solo-20260101-AAAA", owner_email="eve@pingcap.com", created_at=datetime(2026, 1, 1, 9, 0, 0), request_number="AK-SOLO01", ) today = datetime.now(ZoneInfo("Asia/Shanghai")).date() await _add_cost_fact( db_session, integration=integ, fact_date=today, amount_usd=Decimal("30.00"), service="Amazon Bedrock", resource_id="aws-principal:Owner=eve@pingcap.com|service=Amazon Bedrock", tags={"Owner": "eve@pingcap.com"}, ) out = await ApiKeyService(db_session).get_assignment_spend( only.id, current_user=_auth(eve), ) assert out["data_available"] is True assert out["mtd_usd"] == pytest.approx(30.00) assert "share one Owner" not in (out["data_caveat"] or "") # ── Ownership-transfer Owner-tag epoch stitching ─────────────────── # AWS Cost Explorer attributes each day's Bedrock spend to the Owner # cost-allocation tag value AT USAGE TIME — the transfer flow's re-tag # is forward-looking only, so days before the hand-over stay recorded # under the previous holder's tag forever. Pre-fix, a transferred key's # month-to-date silently dropped everything before the transfer date # (prod: AK-20260428-B9C1 showed $210 instead of ~$1,588 the moment its # 2026-07-23 transfer completed — the pre-transfer $1,287 the budget # cron had alerted on two hours earlier just vanished from the panel). async def _make_completed_transfer( db: AsyncSession, *, assignment: ApiKeyAssignment, from_person: IdentityPerson, to_person: IdentityPerson, completed_at: datetime, ): from app.domains.offboarding.models import ( XFER_COMPLETED, AssetTransferRequest, ) xfer = AssetTransferRequest( id=uuid.uuid4(), initiator_id=from_person.id, recipient_id=to_person.id, asset_type="api_key", asset_ref_id=assignment.id, asset_label=assignment.request_number, status=XFER_COMPLETED, responded_at=completed_at, completed_at=completed_at, ) db.add(xfer) await db.commit() return xfer @pytest.mark.asyncio async def test_bedrock_transfer_stitches_previous_owner_epoch( db_session: AsyncSession, ): """A key transferred mid-month keeps its pre-transfer spend: facts under the PREVIOUS holder's Owner tag up to (and including) the transfer-completion date are stitched into the total, facts under that tag AFTER the boundary are not (they belong to whatever still runs under the old tag — e.g. the previous holder's personal IAM user), and the payload carries the transfer-epoch info caveat while staying data_available=True.""" from app.domains.apikey.spend import ( CAVEAT_BEDROCK_TRANSFER_EPOCHS, compute_assignment_spend, ) provider = await _make_provider(db_session, name="AWS BedRock") integ = await _make_integration(db_session, provider="aws") old_owner = await _make_person(db_session, email="prev@pingcap.com") new_owner = await _make_person(db_session, email="curr@pingcap.com") # Post-transfer state: the IAM mirror already carries the NEW # holder's Owner tag (the transfer flow re-tags on completion). iam = await _make_iam_user( db_session, integration=integ, user_name="investibook", tags={"Owner": "curr@pingcap.com"}, owner_person=new_owner, ) key = await _make_assignment( db_session, provider=provider, applicant=new_owner, monthly_budget_usd=Decimal("1650.00"), iam_user_arn=iam.arn, iam_integration=integ, iam_user=iam, request_number="AK-XFER01", ) await _make_completed_transfer( db_session, assignment=key, from_person=old_owner, to_person=new_owner, completed_at=datetime(2026, 7, 23, 1, 49, 55), ) def _rid(email: str) -> str: return f"aws-principal:Owner={email}|service=Amazon Bedrock" # Pre-transfer epoch (counted): previous holder's tag. await _add_cost_fact( db_session, integration=integ, fact_date=date(2026, 7, 10), amount_usd=Decimal("100.00"), service="Amazon Bedrock", resource_id=_rid("prev@pingcap.com"), ) # Boundary day (counted — CE can record the hand-over day under # either tag; distinct prefixes make double-count impossible). await _add_cost_fact( db_session, integration=integ, fact_date=date(2026, 7, 23), amount_usd=Decimal("5.00"), service="Amazon Bedrock", resource_id=_rid("prev@pingcap.com"), ) # Old tag AFTER the boundary (NOT counted). await _add_cost_fact( db_session, integration=integ, fact_date=date(2026, 7, 24), amount_usd=Decimal("20.00"), service="Amazon Bedrock", resource_id=_rid("prev@pingcap.com"), ) # Current holder's tag (counted, unbounded). await _add_cost_fact( db_session, integration=integ, fact_date=date(2026, 7, 25), amount_usd=Decimal("30.00"), service="Amazon Bedrock", resource_id=_rid("curr@pingcap.com"), ) out = await compute_assignment_spend( db_session, assignment=key, month_start=date(2026, 7, 1), month_end=date(2026, 7, 31), ) assert out["data_available"] is True assert out["mtd_usd"] == pytest.approx(135.00) # 100 + 5 + 30 assert out["data_caveat_code"] == CAVEAT_BEDROCK_TRANSFER_EPOCHS assert out["data_caveat_params"]["from_emails"] == ["prev@pingcap.com"] assert out["data_caveat_params"]["until"] == "2026-07-23" @pytest.mark.asyncio async def test_bedrock_transfer_epoch_contested_is_skipped( db_session: AsyncSession, ): """When the previous holder STILL has their own live Bedrock key resolving the same Owner tag, the epoch is skipped (that key claims the tag's rows — stitching would double-count them in the fleet rollup) and the transferred key reports only current-tag facts.""" from app.domains.apikey.spend import compute_assignment_spend provider = await _make_provider(db_session, name="AWS BedRock") integ = await _make_integration(db_session, provider="aws") old_owner = await _make_person(db_session, email="busy@pingcap.com") new_owner = await _make_person(db_session, email="taker@pingcap.com") iam = await _make_iam_user( db_session, integration=integ, user_name="xfer-svc", tags={"Owner": "taker@pingcap.com"}, owner_person=new_owner, ) key = await _make_assignment( db_session, provider=provider, applicant=new_owner, monthly_budget_usd=Decimal("500.00"), iam_user_arn=iam.arn, iam_integration=integ, iam_user=iam, request_number="AK-XFER02", ) # Previous holder keeps their OWN live Bedrock key on the same # integration whose Owner tag is their email → epoch contested. await _make_bedrock_key( db_session, provider=provider, integ=integ, applicant=old_owner, user_name="bedrock-own-key", owner_email="busy@pingcap.com", created_at=datetime(2026, 1, 5, 9, 0, 0), request_number="AK-OWN01", ) await _make_completed_transfer( db_session, assignment=key, from_person=old_owner, to_person=new_owner, completed_at=datetime(2026, 7, 20, 8, 0, 0), ) await _add_cost_fact( db_session, integration=integ, fact_date=date(2026, 7, 10), amount_usd=Decimal("77.00"), service="Amazon Bedrock", resource_id="aws-principal:Owner=busy@pingcap.com|service=Amazon Bedrock", ) await _add_cost_fact( db_session, integration=integ, fact_date=date(2026, 7, 25), amount_usd=Decimal("11.00"), service="Amazon Bedrock", resource_id="aws-principal:Owner=taker@pingcap.com|service=Amazon Bedrock", ) out = await compute_assignment_spend( db_session, assignment=key, month_start=date(2026, 7, 1), month_end=date(2026, 7, 31), ) assert out["mtd_usd"] == pytest.approx(11.00) # old epoch skipped assert out["data_caveat_code"] is None @pytest.mark.asyncio async def test_bedrock_delayed_retag_extends_previous_owner_epoch( db_session: AsyncSession, ): """When the transfer's cloud-side retag FAILED at accept and only a later ``retry_iam_retag`` landed it, cost-fact days between transfer completion and the actual successful retag still carry the OLD Owner tag. The previous holder's epoch must extend to the retag-success date — pre-fix those days were matched by NEITHER clause (old-tag rows past ``completed_at``, new-tag clause is a different prefix) and the spend silently vanished with no caveat.""" from app.domains.apikey.spend import ( CAVEAT_BEDROCK_TRANSFER_EPOCHS, compute_assignment_spend, ) provider = await _make_provider(db_session, name="AWS BedRock") integ = await _make_integration(db_session, provider="aws") old_owner = await _make_person(db_session, email="prevlag@pingcap.com") new_owner = await _make_person(db_session, email="currlag@pingcap.com") iam = await _make_iam_user( db_session, integration=integ, user_name="xfer-lag", tags={"Owner": "currlag@pingcap.com"}, owner_person=new_owner, ) key = await _make_assignment( db_session, provider=provider, applicant=new_owner, monthly_budget_usd=Decimal("1000.00"), iam_user_arn=iam.arn, iam_integration=integ, iam_user=iam, request_number="AK-XFERLAG1", ) # Transfer completed day D=2026-07-20; the accept-time tag push # failed and an admin retry only succeeded on D+3 (2026-07-23) — # exactly the shape ``retry_iam_retag`` records on # ``detail.last_transfer``. xfer = await _make_completed_transfer( db_session, assignment=key, from_person=old_owner, to_person=new_owner, completed_at=datetime(2026, 7, 20, 2, 0, 0), ) key.detail = { "last_transfer": { "transfer_request_id": str(xfer.id), "from_person_id": str(old_owner.id), "to_person_id": str(new_owner.id), "completed_at": "2026-07-20T02:00:00+00:00", "cloud_tag_summary": {"bedrock": { "status": "tagged", "user_name": "xfer-lag", "new_email": "currlag@pingcap.com", }}, "retag_attempts": [ { "at": "2026-07-21T09:00:00+00:00", "by_person_id": str(new_owner.id), "summary": {"bedrock": { "status": "failed", "reason": "AccessDenied", }}, }, { "at": "2026-07-23T09:00:00+00:00", "by_person_id": str(new_owner.id), "summary": {"bedrock": { "status": "tagged", "user_name": "xfer-lag", "new_email": "currlag@pingcap.com", }}, }, ], }, } db_session.add(key) await db_session.commit() def _rid(email: str) -> str: return f"aws-principal:Owner={email}|service=Amazon Bedrock" # Pre-transfer (counted). await _add_cost_fact( db_session, integration=integ, fact_date=date(2026, 7, 10), amount_usd=Decimal("100.00"), service="Amazon Bedrock", resource_id=_rid("prevlag@pingcap.com"), ) # D+2: after completion, BEFORE the successful retag — old tag # still live on AWS. Pre-fix this row vanished (counted now). await _add_cost_fact( db_session, integration=integ, fact_date=date(2026, 7, 22), amount_usd=Decimal("40.00"), service="Amazon Bedrock", resource_id=_rid("prevlag@pingcap.com"), ) # Retag-success day itself (counted — boundary day, same rule as # the transfer-completion boundary). await _add_cost_fact( db_session, integration=integ, fact_date=date(2026, 7, 23), amount_usd=Decimal("5.00"), service="Amazon Bedrock", resource_id=_rid("prevlag@pingcap.com"), ) # Old tag AFTER the retag flipped (NOT counted). await _add_cost_fact( db_session, integration=integ, fact_date=date(2026, 7, 24), amount_usd=Decimal("20.00"), service="Amazon Bedrock", resource_id=_rid("prevlag@pingcap.com"), ) # Current holder's tag (counted, unbounded). await _add_cost_fact( db_session, integration=integ, fact_date=date(2026, 7, 25), amount_usd=Decimal("30.00"), service="Amazon Bedrock", resource_id=_rid("currlag@pingcap.com"), ) out = await compute_assignment_spend( db_session, assignment=key, month_start=date(2026, 7, 1), month_end=date(2026, 7, 31), ) assert out["data_available"] is True assert out["mtd_usd"] == pytest.approx(175.00) # 100 + 40 + 5 + 30 assert out["data_caveat_code"] == CAVEAT_BEDROCK_TRANSFER_EPOCHS assert out["data_caveat_params"]["until"] == "2026-07-23" @pytest.mark.asyncio async def test_bedrock_pending_retag_keeps_previous_owner_epoch_open( db_session: AsyncSession, ): """When the retag failed at accept and NO retry has succeeded yet, the old Owner value is still live on AWS — the previous holder's epoch stays open (through today) so post-completion days recorded under the old tag are stitched in rather than silently dropped, and the transfer caveat still fires.""" from app.domains.apikey.spend import ( CAVEAT_BEDROCK_TRANSFER_EPOCHS, compute_assignment_spend, ) provider = await _make_provider(db_session, name="AWS BedRock") integ = await _make_integration(db_session, provider="aws") old_owner = await _make_person(db_session, email="prevpend@pingcap.com") new_owner = await _make_person(db_session, email="currpend@pingcap.com") # No Owner tag on the IAM mirror → the current owner tag resolves # via the assignee's work_email (the new holder) while AWS keeps # emitting facts under the OLD holder's value until the retag # finally lands — the exact vanishing-band shape. iam = await _make_iam_user( db_session, integration=integ, user_name="xfer-pend", tags=None, owner_person=new_owner, ) key = await _make_assignment( db_session, provider=provider, applicant=new_owner, monthly_budget_usd=Decimal("1000.00"), iam_user_arn=iam.arn, iam_integration=integ, iam_user=iam, request_number="AK-XFERPEND1", ) xfer = await _make_completed_transfer( db_session, assignment=key, from_person=old_owner, to_person=new_owner, completed_at=datetime(2026, 7, 20, 2, 0, 0), ) key.detail = { "last_transfer": { "transfer_request_id": str(xfer.id), "from_person_id": str(old_owner.id), "to_person_id": str(new_owner.id), "completed_at": "2026-07-20T02:00:00+00:00", "cloud_tag_summary": {"bedrock": { "status": "failed", "user_name": "xfer-pend", "reason": "AccessDenied", }}, }, } db_session.add(key) await db_session.commit() def _rid(email: str) -> str: return f"aws-principal:Owner={email}|service=Amazon Bedrock" # Pre-transfer (counted). await _add_cost_fact( db_session, integration=integ, fact_date=date(2026, 7, 10), amount_usd=Decimal("100.00"), service="Amazon Bedrock", resource_id=_rid("prevpend@pingcap.com"), ) # Post-completion, retag still pending → old tag, counted. await _add_cost_fact( db_session, integration=integ, fact_date=date(2026, 7, 24), amount_usd=Decimal("40.00"), service="Amazon Bedrock", resource_id=_rid("prevpend@pingcap.com"), ) # Current holder's tag (counted, unbounded). await _add_cost_fact( db_session, integration=integ, fact_date=date(2026, 7, 25), amount_usd=Decimal("30.00"), service="Amazon Bedrock", resource_id=_rid("currpend@pingcap.com"), ) out = await compute_assignment_spend( db_session, assignment=key, month_start=date(2026, 7, 1), month_end=date(2026, 7, 31), ) assert out["data_available"] is True assert out["mtd_usd"] == pytest.approx(170.00) # 100 + 40 + 30 assert out["data_caveat_code"] == CAVEAT_BEDROCK_TRANSFER_EPOCHS @pytest.mark.asyncio async def test_bedrock_transfer_epoch_scoped_to_queried_window( db_session: AsyncSession, ): """The transfer-epoch stitch is scoped to the queried window: a month entirely AFTER the hand-over behaves exactly like a never-transferred key (no epoch clauses, no transfer caveat — pre-fix the 'this key changed owner' note stuck to every later month forever), while the transfer month itself keeps the stitch and the caveat.""" from app.domains.apikey.spend import ( CAVEAT_BEDROCK_TRANSFER_EPOCHS, _bedrock_owner_epochs, compute_assignment_spend, ) provider = await _make_provider(db_session, name="AWS BedRock") integ = await _make_integration(db_session, provider="aws") old_owner = await _make_person(db_session, email="prevjun@pingcap.com") new_owner = await _make_person(db_session, email="curraug@pingcap.com") iam = await _make_iam_user( db_session, integration=integ, user_name="xfer-jun", tags={"Owner": "curraug@pingcap.com"}, owner_person=new_owner, ) key = await _make_assignment( db_session, provider=provider, applicant=new_owner, monthly_budget_usd=Decimal("1000.00"), iam_user_arn=iam.arn, iam_integration=integ, iam_user=iam, request_number="AK-XFERJUN1", ) await _make_completed_transfer( db_session, assignment=key, from_person=old_owner, to_person=new_owner, completed_at=datetime(2026, 6, 15, 8, 0, 0), ) def _rid(email: str) -> str: return f"aws-principal:Owner={email}|service=Amazon Bedrock" # June: pre-transfer old-tag spend + post-transfer new-tag spend. await _add_cost_fact( db_session, integration=integ, fact_date=date(2026, 6, 10), amount_usd=Decimal("50.00"), service="Amazon Bedrock", resource_id=_rid("prevjun@pingcap.com"), ) await _add_cost_fact( db_session, integration=integ, fact_date=date(2026, 6, 20), amount_usd=Decimal("25.00"), service="Amazon Bedrock", resource_id=_rid("prevjun@pingcap.com"), ) await _add_cost_fact( db_session, integration=integ, fact_date=date(2026, 6, 22), amount_usd=Decimal("25.00"), service="Amazon Bedrock", resource_id=_rid("curraug@pingcap.com"), ) # August: only the current tag belongs to this key. await _add_cost_fact( db_session, integration=integ, fact_date=date(2026, 8, 5), amount_usd=Decimal("99.00"), service="Amazon Bedrock", resource_id=_rid("prevjun@pingcap.com"), ) await _add_cost_fact( db_session, integration=integ, fact_date=date(2026, 8, 10), amount_usd=Decimal("30.00"), service="Amazon Bedrock", resource_id=_rid("curraug@pingcap.com"), ) # June window: epoch overlaps → stitched total + caveat. out_jun = await compute_assignment_spend( db_session, assignment=key, month_start=date(2026, 6, 1), month_end=date(2026, 6, 30), ) assert out_jun["mtd_usd"] == pytest.approx(75.00) # 50 + 25(curr) assert out_jun["data_caveat_code"] == CAVEAT_BEDROCK_TRANSFER_EPOCHS assert out_jun["data_caveat_params"]["until"] == "2026-06-15" # August window: the June epoch is out of scope → behaves like a # never-transferred key: no epochs resolved at all, no caveat, and # the old tag's August rows never leak in. assert await _bedrock_owner_epochs( db_session, assignment=key, current_owner_tag="curraug@pingcap.com", window_start=date(2026, 8, 1), window_end=date(2026, 8, 31), eligibility_month_start=date(2026, 8, 1), ) == [] out_aug = await compute_assignment_spend( db_session, assignment=key, month_start=date(2026, 8, 1), month_end=date(2026, 8, 31), ) assert out_aug["mtd_usd"] == pytest.approx(30.00) assert out_aug["data_caveat"] is None assert out_aug["data_caveat_code"] is None @pytest.mark.asyncio async def test_monthly_series_keeps_previous_holder_spend_in_earlier_bars( db_session: AsyncSession, ): """The 6-month bar chart must resolve transfer epochs against ITS OWN window, not the requested month. The requested-month filter gates epochs on [month_start, month_end], so reusing it for the series evaluates the five earlier bars with a gate that excludes their own hand-over: a key transferred in June, read in August, rendered the June bar as the current holder's share only — silently re-writing a number the user had already seen in July, with no caveat anywhere saying so.""" from app.domains.apikey.spend import compute_assignment_spend_detail provider = await _make_provider(db_session, name="AWS BedRock") integ = await _make_integration(db_session, provider="aws") old_owner = await _make_person(db_session, email="prevser@pingcap.com") new_owner = await _make_person(db_session, email="currser@pingcap.com") iam = await _make_iam_user( db_session, integration=integ, user_name="xfer-series", tags={"Owner": "currser@pingcap.com"}, owner_person=new_owner, ) key = await _make_assignment( db_session, provider=provider, applicant=new_owner, monthly_budget_usd=Decimal("1000.00"), iam_user_arn=iam.arn, iam_integration=integ, iam_user=iam, request_number="AK-XFERSER1", ) await _make_completed_transfer( db_session, assignment=key, from_person=old_owner, to_person=new_owner, completed_at=datetime(2026, 6, 15, 8, 0, 0), ) def _rid(email: str) -> str: return f"aws-principal:Owner={email}|service=Amazon Bedrock" # June, pre-transfer: the previous holder's tag — this key's spend. await _add_cost_fact( db_session, integration=integ, fact_date=date(2026, 6, 10), amount_usd=Decimal("50.00"), service="Amazon Bedrock", resource_id=_rid("prevser@pingcap.com"), ) # June, post-transfer old tag: NOT this key's (belongs to whatever # still runs under the old tag) — the epoch bound still applies # inside the series window. await _add_cost_fact( db_session, integration=integ, fact_date=date(2026, 6, 20), amount_usd=Decimal("25.00"), service="Amazon Bedrock", resource_id=_rid("prevser@pingcap.com"), ) # June, post-transfer new tag. await _add_cost_fact( db_session, integration=integ, fact_date=date(2026, 6, 22), amount_usd=Decimal("25.00"), service="Amazon Bedrock", resource_id=_rid("currser@pingcap.com"), ) # August: only the current tag belongs to this key. await _add_cost_fact( db_session, integration=integ, fact_date=date(2026, 8, 5), amount_usd=Decimal("99.00"), service="Amazon Bedrock", resource_id=_rid("prevser@pingcap.com"), ) await _add_cost_fact( db_session, integration=integ, fact_date=date(2026, 8, 10), amount_usd=Decimal("30.00"), service="Amazon Bedrock", resource_id=_rid("currser@pingcap.com"), ) out = await compute_assignment_spend_detail( db_session, assignment=key, month_start=date(2026, 8, 1), month_end=date(2026, 8, 31), ) bars = {r["month"]: r["usd"] for r in out["monthly_series"]} assert list(bars)[-1] == "2026-08" # 50 (previous holder, pre-hand-over) + 25 (current holder) — the # same number July's page showed. assert bars["2026-06"] == pytest.approx(75.00) assert bars["2026-08"] == pytest.approx(30.00) # The requested month is unchanged: no epoch overlaps August, so the # old tag's August rows stay out and no transfer caveat is attached. assert out["mtd_usd"] == pytest.approx(30.00) assert out["data_caveat"] is None assert out["data_caveat_code"] is None @pytest.mark.asyncio async def test_retag_attempt_after_a_success_does_not_extend_the_epoch( db_session: AsyncSession, ): """The epoch ends at the FIRST retag attempt that reported ``tagged`` — attempts after it say nothing new about when the AWS tag flipped. Three attempts, seeded so first-wins and last-wins give DIFFERENT answers (one ``tagged`` attempt alone only proves the attempts are read at all): * 7/22 ``tagged`` — the real flip; * 7/25 ``tagged`` — an admin re-running the retag on an already-correct tag. Last-wins would end the epoch here and swallow the 7/25 old-tag row, reporting 90.00 / ``until=2026-07-25``; * 7/27 ``failed`` (IAM user gone) — this also overwrote ``cloud_tag_summary``, which on its own reads as "the old value is still live" and pushed the epoch end, and the caveat's ``until``, to today, drifting one day further every day.""" from app.domains.apikey.spend import ( CAVEAT_BEDROCK_TRANSFER_EPOCHS, compute_assignment_spend, ) provider = await _make_provider(db_session, name="AWS BedRock") integ = await _make_integration(db_session, provider="aws") old_owner = await _make_person(db_session, email="prevgra@pingcap.com") new_owner = await _make_person(db_session, email="currgra@pingcap.com") iam = await _make_iam_user( db_session, integration=integ, user_name="xfer-gra", tags={"Owner": "currgra@pingcap.com"}, owner_person=new_owner, ) key = await _make_assignment( db_session, provider=provider, applicant=new_owner, monthly_budget_usd=Decimal("1000.00"), iam_user_arn=iam.arn, iam_integration=integ, iam_user=iam, request_number="AK-XFERGRA1", ) xfer = await _make_completed_transfer( db_session, assignment=key, from_person=old_owner, to_person=new_owner, completed_at=datetime(2026, 7, 20, 2, 0, 0), ) key.detail = { "last_transfer": { "transfer_request_id": str(xfer.id), "from_person_id": str(old_owner.id), "to_person_id": str(new_owner.id), "completed_at": "2026-07-20T02:00:00+00:00", # ``retry_iam_retag`` OVERWRITES cloud_tag_summary with the # latest attempt's outcome, so this is the failed re-run, not # the accept-time push. "cloud_tag_summary": {"bedrock": { "status": "failed", "reason": "NoSuchEntity", }}, "retag_attempts": [ { "at": "2026-07-22T09:00:00+00:00", "by_person_id": str(new_owner.id), "summary": {"bedrock": { "status": "tagged", "user_name": "xfer-gra", "new_email": "currgra@pingcap.com", }}, }, { "at": "2026-07-25T09:00:00+00:00", "by_person_id": str(new_owner.id), "summary": {"bedrock": { "status": "tagged", "user_name": "xfer-gra", "new_email": "currgra@pingcap.com", }}, }, { "at": "2026-07-27T09:00:00+00:00", "by_person_id": str(new_owner.id), "summary": {"bedrock": { "status": "failed", "reason": "NoSuchEntity", }}, }, ], }, } db_session.add(key) await db_session.commit() def _rid(email: str) -> str: return f"aws-principal:Owner={email}|service=Amazon Bedrock" # Retag-success day (counted — boundary day). await _add_cost_fact( db_session, integration=integ, fact_date=date(2026, 7, 22), amount_usd=Decimal("40.00"), service="Amazon Bedrock", resource_id=_rid("prevgra@pingcap.com"), ) # Old tag after the tag actually flipped (NOT counted) — and on the # date of the SECOND ``tagged`` attempt, so a last-wins reader would # pull it in. await _add_cost_fact( db_session, integration=integ, fact_date=date(2026, 7, 25), amount_usd=Decimal("20.00"), service="Amazon Bedrock", resource_id=_rid("prevgra@pingcap.com"), ) await _add_cost_fact( db_session, integration=integ, fact_date=date(2026, 7, 26), amount_usd=Decimal("30.00"), service="Amazon Bedrock", resource_id=_rid("currgra@pingcap.com"), ) out = await compute_assignment_spend( db_session, assignment=key, month_start=date(2026, 7, 1), month_end=date(2026, 7, 31), ) # 40 (7/22, the real flip) + 30 (new tag). A last-tagged-wins reader # ends the epoch at 7/25 and adds the 20.00 old-tag row → 90.00. assert float(out["mtd_usd"]) == pytest.approx(70.00) assert out["data_caveat_code"] == CAVEAT_BEDROCK_TRANSFER_EPOCHS assert out["data_caveat_params"]["until"] == "2026-07-22", ( "the epoch must end at the FIRST tagged attempt, not the last" ) @pytest.mark.asyncio async def test_pending_retag_open_epoch_is_anchored_on_shanghai_today( db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch, ): """The open (never-retagged) epoch's end must use the SAME timezone the window bounds come from. Callers derive ``month_start`` / ``month_end`` from Asia/Shanghai. A UTC "today" is up to 8h behind: between 00:00 and 08:00 CST it names yesterday, and on the 1st of a month that is before ``month_start``, so ``end_d < month_start`` dropped the whole open epoch — the previous holder's spend for that day vanished, and reappeared 8h later.""" from datetime import UTC import app.domains.apikey.spend as spend_mod from app.domains.apikey.spend import ( CAVEAT_BEDROCK_TRANSFER_EPOCHS, compute_assignment_spend, ) class _FrozenDatetime(datetime): """2026-08-01 01:30 Asia/Shanghai == 2026-07-31 17:30 UTC.""" _AT = datetime(2026, 7, 31, 17, 30, tzinfo=UTC) @classmethod def now(cls, tz=None): # type: ignore[override] return cls._AT if tz is None else cls._AT.astimezone(tz) monkeypatch.setattr(spend_mod, "datetime", _FrozenDatetime) provider = await _make_provider(db_session, name="AWS BedRock") integ = await _make_integration(db_session, provider="aws") old_owner = await _make_person(db_session, email="prevtz@pingcap.com") new_owner = await _make_person(db_session, email="currtz@pingcap.com") iam = await _make_iam_user( db_session, integration=integ, user_name="xfer-tz", tags={"Owner": "currtz@pingcap.com"}, owner_person=new_owner, ) key = await _make_assignment( db_session, provider=provider, applicant=new_owner, monthly_budget_usd=Decimal("1000.00"), iam_user_arn=iam.arn, iam_integration=integ, iam_user=iam, request_number="AK-XFERTZ1", ) xfer = await _make_completed_transfer( db_session, assignment=key, from_person=old_owner, to_person=new_owner, completed_at=datetime(2026, 7, 20, 2, 0, 0), ) key.detail = { "last_transfer": { "transfer_request_id": str(xfer.id), "from_person_id": str(old_owner.id), "to_person_id": str(new_owner.id), "completed_at": "2026-07-20T02:00:00+00:00", "cloud_tag_summary": {"bedrock": { "status": "failed", "reason": "AccessDenied", }}, }, } db_session.add(key) await db_session.commit() def _rid(email: str) -> str: return f"aws-principal:Owner={email}|service=Amazon Bedrock" # The Shanghai "today" — day 1 of the queried month. Under a UTC # anchor the epoch ended 2026-07-31, before month_start, so this row # (and the caveat) disappeared. await _add_cost_fact( db_session, integration=integ, fact_date=date(2026, 8, 1), amount_usd=Decimal("40.00"), service="Amazon Bedrock", resource_id=_rid("prevtz@pingcap.com"), ) await _add_cost_fact( db_session, integration=integ, fact_date=date(2026, 8, 1), amount_usd=Decimal("10.00"), service="Amazon Bedrock", resource_id=_rid("currtz@pingcap.com"), ) # The window the caller builds at that instant: [1st, today_shanghai]. out = await compute_assignment_spend( db_session, assignment=key, month_start=date(2026, 8, 1), month_end=date(2026, 8, 1), ) assert out["mtd_usd"] == pytest.approx(50.00) assert out["data_caveat_code"] == CAVEAT_BEDROCK_TRANSFER_EPOCHS assert out["data_caveat_params"]["until"] == "2026-08-01" @pytest.mark.asyncio async def test_caveat_codes_surface_in_service_payload( db_session: AsyncSession, ): """The service payload carries the machine-readable caveat code + params the FE localizes — the #1776 sibling pin must read as ``bedrock_owner_shared`` (naming the canonical key), NOT as the static per-bucket 'iam_user_arn is empty' copy the UI showed pre-fix for a key whose ARN was set.""" from app.domains.apikey.spend import ( CAVEAT_BEDROCK_NO_IAM, CAVEAT_BEDROCK_OWNER_SHARED, ) provider = await _make_provider(db_session, name="AWS BedRock") integ = await _make_integration(db_session, provider="aws") dan = await _make_person(db_session, email="dan2@pingcap.com") await _make_bedrock_key( db_session, provider=provider, integ=integ, applicant=dan, user_name="bedrock-c1", owner_email="dan2@pingcap.com", created_at=datetime(2026, 1, 1, 9, 0, 0), request_number="AK-CANON1", ) sibling = await _make_bedrock_key( db_session, provider=provider, integ=integ, applicant=dan, user_name="bedrock-c2", owner_email="dan2@pingcap.com", created_at=datetime(2026, 6, 1, 9, 0, 0), request_number="AK-SIB2", ) out_sib = await ApiKeyService(db_session).get_assignment_spend( sibling.id, current_user=_auth(dan), ) assert out_sib["data_caveat_code"] == CAVEAT_BEDROCK_OWNER_SHARED assert out_sib["data_caveat_params"]["canonical_ref"] == "AK-CANON1" assert out_sib["data_caveat_params"]["sibling_count"] == 1 # Missing IAM binding keeps its own distinct code. unbound = await _make_assignment( db_session, provider=provider, applicant=dan, monthly_budget_usd=Decimal("100.00"), request_number="AK-UNBOUND1", ) out_unbound = await ApiKeyService(db_session).get_assignment_spend( unbound.id, current_user=_auth(dan), ) assert out_unbound["data_caveat_code"] == CAVEAT_BEDROCK_NO_IAM @pytest.mark.asyncio async def test_the_requested_month_bar_agrees_with_the_mtd_headline( db_session: AsyncSession, ): """The 6-month series and the MTD headline must keep the SAME transfer epochs for the requested month. The series widens its epoch-OVERLAP window to six months (correct — an earlier bar has to carry the holder who owned the tag in ITS month), but the sibling-liveness scan behind the contested guard used to widen with it. A sibling key revoked INSIDE the six months but BEFORE the requested month then read as live for the series and dead for the month, so the leaver's epoch was kept for the MTD and dropped for the bar: the detail page rendered "$130 MTD, includes pre-transfer spend" directly above a $30 August bar in its own chart. Shape below is that exact case — a leaver's own Bedrock key revoked at offboarding in May, another of their keys force-transferred to the manager on Aug 10. """ from app.domains.apikey.spend import ( CAVEAT_BEDROCK_TRANSFER_EPOCHS, compute_assignment_spend_detail, ) provider = await _make_provider(db_session, name="AWS BedRock") integ = await _make_integration(db_session, provider="aws") leaver = await _make_person(db_session, email="asymprev@pingcap.com") manager = await _make_person(db_session, email="asymcurr@pingcap.com") key = await _make_bedrock_key( db_session, provider=provider, integ=integ, applicant=manager, user_name="bedrock-asym", owner_email="asymcurr@pingcap.com", created_at=datetime(2026, 1, 5, 9, 0, 0), request_number="AK-ASYM01", ) await _make_completed_transfer( db_session, assignment=key, from_person=leaver, to_person=manager, completed_at=datetime(2026, 8, 10, 4, 0, 0), ) # The leaver's OWN Bedrock key, revoked at offboarding in May: inside # the series' 6-month span, outside the requested month. It must not # count as a live claimant of the leaver's Owner tag for EITHER # surface. stale_sibling = await _make_bedrock_key( db_session, provider=provider, integ=integ, applicant=leaver, user_name="bedrock-asym-stale", owner_email="asymprev@pingcap.com", created_at=datetime(2026, 1, 6, 9, 0, 0), request_number="AK-ASYM02", ) stale_sibling.status = "revoked" stale_sibling.revoked_at = datetime(2026, 5, 20, 8, 0, 0) db_session.add(stale_sibling) await db_session.commit() def _rid(email: str) -> str: return f"aws-principal:Owner={email}|service=Amazon Bedrock" # Pre-hand-over August days, still under the leaver's tag. await _add_cost_fact( db_session, integration=integ, fact_date=date(2026, 8, 4), amount_usd=Decimal("100.00"), service="Amazon Bedrock", resource_id=_rid("asymprev@pingcap.com"), ) # Post-hand-over, under the manager's tag. await _add_cost_fact( db_session, integration=integ, fact_date=date(2026, 8, 12), amount_usd=Decimal("30.00"), service="Amazon Bedrock", resource_id=_rid("asymcurr@pingcap.com"), ) out = await compute_assignment_spend_detail( db_session, assignment=key, month_start=date(2026, 8, 1), month_end=date(2026, 8, 31), ) assert out["data_caveat_code"] == CAVEAT_BEDROCK_TRANSFER_EPOCHS assert float(out["mtd_usd"]) == pytest.approx(130.00) aug_bar = next(r for r in out["monthly_series"] if r["month"] == "2026-08") assert aug_bar["usd"] == pytest.approx(float(out["mtd_usd"])), ( "the requested month's bar disagrees with the MTD headline above it" ) def test_trimming_retag_attempts_never_drops_the_first_success(): """The 20-attempt cap must not move the "first tagged" datum. ``service.py`` keeps only the newest attempts; ``spend.py`` bounds the previous holder's Owner-tag epoch — and the user-facing caveat's ``until`` — at the FIRST attempt reporting ``tagged``. Trimming purely by recency silently promotes a LATER attempt to "first", extending the previous holder's window and shifting money between two people's segments. Needs >20 retries on one hand-over, so it is a coupling guard, not a live bug. """ from app.domains.apikey.service import ( _RETAG_ATTEMPTS_CAP, _trim_retag_attempts, ) def attempt(day: int, status: str) -> dict: return { "at": f"2026-07-{day:02d}T09:00:00+00:00", "by_person_id": "someone", "summary": {"bedrock": {"status": status}}, } # First attempt succeeded, then 30 failures — the shape where a # newest-N trim loses the success entirely. attempts = [attempt(1, "tagged")] + [ attempt(2 + i, "failed") for i in range(30) ] trimmed = _trim_retag_attempts(attempts) assert len(trimmed) == _RETAG_ATTEMPTS_CAP assert trimmed[0] == attempts[0], ( "the first successful attempt was trimmed away — the epoch end " "would jump to a later attempt" ) # Still chronological, and still the newest attempts after the pin. assert [a["at"] for a in trimmed] == sorted(a["at"] for a in trimmed) assert trimmed[-1] == attempts[-1] # Under the cap nothing is touched. short = [attempt(1, "failed"), attempt(2, "tagged")] assert _trim_retag_attempts(short) == short # No success at all → plain newest-N. all_failed = [attempt(1 + i, "failed") for i in range(25)] assert _trim_retag_attempts(all_failed) == all_failed[-20:] # A bucket that only succeeds LATE keeps its own first success, even # when another bucket succeeded earlier. mixed = [ {"at": "2026-07-01T00:00:00+00:00", "summary": {"azure": {"status": "tagged"}}}, {"at": "2026-07-02T00:00:00+00:00", "summary": {"bedrock": {"status": "tagged"}}}, ] + [attempt(3 + i, "failed") for i in range(30)] trimmed_mixed = _trim_retag_attempts(mixed) assert trimmed_mixed[:2] == mixed[:2] def test_same_owner_epoch_ranges_are_merged_before_they_reach_sql(): """Two ranges naming ONE Owner tag must not both be OR-ed into the sum. Distinct tag values match disjoint ``resource_id`` prefixes, so the deliberate hand-over-day overlap between ADJACENT epochs is free. Two epochs for the SAME value are not: they are the same rows, summed twice. Reachable only via a chain that hands the key back to an earlier holder. """ from app.domains.apikey.spend import _coalesced_epochs # Untouched when every owner appears once — order preserved, oldest # first, the open-ended epoch leading. chain = [ ("a@pingcap.com", None, date(2026, 7, 8)), ("b@pingcap.com", date(2026, 7, 8), date(2026, 7, 20)), ] assert _coalesced_epochs(chain) == chain # A→B→C→B: B's two ranges touch on 7/12 and collapse into one. assert _coalesced_epochs([ ("a@pingcap.com", None, date(2026, 7, 5)), ("b@pingcap.com", date(2026, 7, 5), date(2026, 7, 12)), ("c@pingcap.com", date(2026, 7, 12), date(2026, 7, 12)), ("b@pingcap.com", date(2026, 7, 12), date(2026, 7, 20)), ]) == [ ("a@pingcap.com", None, date(2026, 7, 5)), ("b@pingcap.com", date(2026, 7, 5), date(2026, 7, 20)), ("c@pingcap.com", date(2026, 7, 12), date(2026, 7, 12)), ] # Same owner, genuinely disjoint tenures — kept as two clauses so the # gap in between stays out of the sum. Output is oldest-first (see the # function's closing sort), NOT grouped by owner: c's tenure sorts # between b's two. assert _coalesced_epochs([ ("b@pingcap.com", date(2026, 7, 1), date(2026, 7, 5)), ("c@pingcap.com", date(2026, 7, 5), date(2026, 7, 15)), ("b@pingcap.com", date(2026, 7, 15), date(2026, 7, 20)), ]) == [ ("b@pingcap.com", date(2026, 7, 1), date(2026, 7, 5)), ("c@pingcap.com", date(2026, 7, 5), date(2026, 7, 15)), ("b@pingcap.com", date(2026, 7, 15), date(2026, 7, 20)), ] # Case-insensitive: AWS tag values are matched with ILIKE, so two # casings of one email are one tag. assert _coalesced_epochs([ ("B@pingcap.com", None, date(2026, 7, 10)), ("b@pingcap.com", date(2026, 7, 10), date(2026, 7, 20)), ]) == [("B@pingcap.com", None, date(2026, 7, 20))] @pytest.mark.asyncio async def test_a_returning_holders_shared_day_is_not_counted_twice( db_session: AsyncSession, ): """A→B→C→B→D with B→C and C→B on the SAME day. B holds two epochs whose inclusive ranges share 7/12. Without the same-owner merge both clauses match B's 7/12 rows and the OR inside the ``SUM`` counts them twice — the #1776 failure mode, re-introduced by making the epoch start inclusive. """ from app.domains.apikey.spend import compute_assignment_spend provider = await _make_provider(db_session, name="AWS BedRock") integ = await _make_integration(db_session, provider="aws") a = await _make_person(db_session, email="ret1@pingcap.com") b = await _make_person(db_session, email="ret2@pingcap.com") c = await _make_person(db_session, email="ret3@pingcap.com") d = await _make_person(db_session, email="ret4@pingcap.com") key = await _make_bedrock_key( db_session, provider=provider, integ=integ, applicant=d, user_name="bedrock-ret", owner_email="ret4@pingcap.com", created_at=datetime(2026, 1, 5, 9, 0, 0), request_number="AK-RET01", ) for frm, to, when in ( (a, b, datetime(2026, 7, 5, 3, 0, 0)), (b, c, datetime(2026, 7, 12, 8, 0, 0)), (c, b, datetime(2026, 7, 12, 20, 0, 0)), (b, d, datetime(2026, 7, 20, 4, 0, 0)), ): await _make_completed_transfer( db_session, assignment=key, from_person=frm, to_person=to, completed_at=when, ) def _rid(email: str) -> str: return f"aws-principal:Owner={email}|service=Amazon Bedrock" await _add_cost_fact( db_session, integration=integ, fact_date=date(2026, 7, 12), amount_usd=Decimal("11.00"), service="Amazon Bedrock", resource_id=_rid("ret2@pingcap.com"), ) out = await compute_assignment_spend( db_session, assignment=key, month_start=date(2026, 7, 1), month_end=date(2026, 7, 31), ) assert float(out["mtd_usd"]) == pytest.approx(11.00), ( "the shared day was counted once per epoch naming the same tag" )