"""Incremental pull — /billing/changes and /billing/daily?updated_since=. The pins: * **the watermark columns are the ones that actually move** — the two source tables are mirror images (``billing_daily.fetched_at`` vs ``cost_facts.updated_at``); reading the frozen one yields a feed that silently returns nothing, so both directions are pinned here * both sources are consulted: a line-item-only re-ingest and an aggregate-only re-fetch must each show up * an unchanged re-fetch still reports (ingest watermark, not amount diff) * `?updated_since=` filters the daily series to exactly those cells, before paging * the horizon is enforced, and a `since` older than it is refused with a message that names the alternative — rather than quietly returning a partial answer * a future `since` is refused (an empty page would read as "nothing changed") * whitelist still applies; every read is audited """ from __future__ import annotations import uuid from datetime import UTC, date, datetime, timedelta from decimal import Decimal import pytest from sqlalchemy import select, text, update from sqlalchemy.ext.asyncio import AsyncSession from app.api_clients.models import ApiClient from app.api_clients.scopes import SCOPE_CLOUD_BILLING_READ from app.api_clients.tokens import ApiClientTokenService from app.audit.models import AuditLog from app.domains.cmdb.models import ( CmdbCloudBillingDaily, CmdbCloudScope, CmdbCostFact, ) from app.identity.person_models import IdentityPerson from app.integrations.models import Integration pytestmark = pytest.mark.asyncio CHANGES = "/api/v1/open/cloud/billing/changes" DAILY = "/api/v1/open/cloud/billing/daily" def _headers(token: str) -> dict: return {"Authorization": f"Bearer {token}"} async def _mk_client( db: AsyncSession, *, visibility: dict | None = None, ) -> str: owner = IdentityPerson( id=uuid.uuid4(), work_email=f"co-{uuid.uuid4().hex[:8]}@pingcap.com", display_name="Owner", employment_status="active", source_system="successfactors", hris_person_id=f"sf-{uuid.uuid4().hex[:10]}", ) db.add(owner) await db.flush() client = ApiClient( id=uuid.uuid4(), name="BI", slug=f"bi-{uuid.uuid4().hex[:6]}", owner_person_id=owner.id, scopes=[SCOPE_CLOUD_BILLING_READ], cloud_visibility=visibility, ) db.add(client) await db.flush() await db.refresh(client) _row, plaintext = await ApiClientTokenService(db).mint( client=client, name="t", expires_in_days=None, created_by=None, ) await db.commit() return plaintext async def _mk_scope( db: AsyncSession, *, provider: str = "aws", name: str = "RD Compute", ) -> tuple[CmdbCloudScope, Integration]: integ = Integration( id=uuid.uuid4(), kind="cloud", provider=provider, name=f"{provider}-{uuid.uuid4().hex[:6]}", slug=f"{provider}-{uuid.uuid4().hex[:6]}", enabled=True, is_primary=False, capabilities=["servers"], ) db.add(integ) await db.flush() scope = CmdbCloudScope( id=uuid.uuid4(), integration_id=str(integ.id), scope_type=f"{provider}_account", external_id=f"{provider}-{uuid.uuid4().hex[:10]}", display_name=name, status="active", environment="prod", first_seen_at=datetime.now(UTC), last_synced_at=datetime.now(UTC), ) db.add(scope) await db.flush() return scope, integ async def _mk_fact( db: AsyncSession, *, scope: CmdbCloudScope, integ: Integration, day: date, amount: str = "10", updated_at: datetime | None = None, ) -> CmdbCostFact: fact = CmdbCostFact( fact_key=uuid.uuid4().hex + uuid.uuid4().hex[:32], date=day, integration_id=str(integ.id), scope_id=str(scope.id), service="Amazon EC2", amount_usd=Decimal(amount) / 2, amount_native=Decimal(amount) / 2, amount_usd_list=Decimal(amount), currency="USD", ) db.add(fact) await db.flush() if updated_at is not None: await db.execute( update(CmdbCostFact) .where(CmdbCostFact.id == fact.id) .values(updated_at=updated_at), ) await db.flush() return fact async def _mk_daily( db: AsyncSession, *, scope: CmdbCloudScope, integ: Integration, day: date, amount: str = "9", fetched_at: datetime | None = None, ) -> CmdbCloudBillingDaily: row = CmdbCloudBillingDaily( id=uuid.uuid4(), integration_id=str(integ.id), scope_id=str(scope.id), linked_account_id=scope.external_id or "", date=day, total_cost_usd=Decimal(amount), total_cost_usd_list=Decimal(amount), cost_by_service={}, currency="USD", ) db.add(row) await db.flush() if fetched_at is not None: await db.execute( update(CmdbCloudBillingDaily) .where(CmdbCloudBillingDaily.id == row.id) .values(fetched_at=fetched_at), ) await db.flush() return row def _iso(dt: datetime) -> str: """URL-safe cursor. A raw ``+00:00`` in a query string decodes to a space, which is exactly the integrator mistake ``_parse_since`` forgives — pinned separately in :func:`test_unescaped_plus_offset_is_accepted`.""" from urllib.parse import quote return quote(dt.replace(microsecond=0).isoformat()) # ── The watermark columns ───────────────────────────────────────────── async def test_watermark_columns_are_the_ones_that_advance( db_session: AsyncSession, ): """Guards the choice that makes the whole feature work. ``cmdb_cloud_billing_daily`` writes ``fetched_at`` on re-ingest and leaves ``updated_at`` frozen; ``cmdb_cost_facts`` does the exact opposite. Reading the frozen column of either table produces an incremental feed that silently returns nothing. """ for table, advancing, frozen in ( ("cmdb_cloud_billing_daily", "fetched_at", "updated_at"), ("cmdb_cost_facts", "updated_at", "fetched_at"), ): cols = (await db_session.execute( text(f"SHOW COLUMNS FROM {table}"), )).all() by_name = {row[0]: row for row in cols} assert advancing in by_name, f"{table}.{advancing} missing" assert frozen in by_name, f"{table}.{frozen} missing" # Neither table advances the frozen column via DDL either — the # only writer is the ingest's SET list. assert "on update" not in (by_name[frozen][5] or "").lower() # ── /billing/changes ────────────────────────────────────────────────── async def test_changes_reports_reingested_cells_from_both_sources( client, db_session: AsyncSession, ): facts_scope, facts_integ = await _mk_scope(db_session, name="Facts Side") agg_scope, agg_integ = await _mk_scope(db_session, name="Aggregate Side") today = datetime.now(UTC).date() day = today - timedelta(days=2) old = datetime.now(UTC) - timedelta(days=5) fresh = datetime.now(UTC) - timedelta(minutes=5) # Stale on both sides… await _mk_fact(db_session, scope=facts_scope, integ=facts_integ, day=day - timedelta(days=1), updated_at=old) await _mk_daily(db_session, scope=agg_scope, integ=agg_integ, day=day - timedelta(days=1), fetched_at=old) # …and one fresh cell in each table. await _mk_fact(db_session, scope=facts_scope, integ=facts_integ, day=day, updated_at=fresh) await _mk_daily(db_session, scope=agg_scope, integ=agg_integ, day=day, fetched_at=fresh) await db_session.commit() token = await _mk_client(db_session) since = datetime.now(UTC) - timedelta(hours=1) resp = await client.get( f"{CHANGES}?since={_iso(since)}", headers=_headers(token), ) assert resp.status_code == 200, resp.text body = resp.json() got = {(r["name"], r["date"]) for r in body["data"]} assert got == { ("Facts Side", day.isoformat()), ("Aggregate Side", day.isoformat()), } assert set(body["meta"]["signals"]) == { "billing_daily.fetched_at", "cost_facts.updated_at", } assert body["meta"]["horizon"] assert "does_not_detect" in body["meta"] async def test_changes_includes_unchanged_reingest( client, db_session: AsyncSession, ): """An ingest watermark, not an amount diff: a day re-fetched with the same number is still reported, because a consumer overwriting by key does not care and under-reporting would lose data silently.""" scope, integ = await _mk_scope(db_session) day = datetime.now(UTC).date() - timedelta(days=1) fact = await _mk_fact(db_session, scope=scope, integ=integ, day=day, amount="10", updated_at=datetime.now(UTC) - timedelta(days=3)) await db_session.commit() token = await _mk_client(db_session) since = datetime.now(UTC) - timedelta(hours=1) before = await client.get( f"{CHANGES}?since={_iso(since)}", headers=_headers(token), ) assert before.json()["data"] == [] # Re-ingest with the identical amount — only the watermark moves. await db_session.execute( update(CmdbCostFact) .where(CmdbCostFact.id == fact.id) .values(updated_at=datetime.now(UTC)), ) await db_session.commit() after = await client.get( f"{CHANGES}?since={_iso(since)}", headers=_headers(token), ) assert [r["date"] for r in after.json()["data"]] == [day.isoformat()] async def test_changes_respects_visibility(client, db_session: AsyncSession): mine, mine_integ = await _mk_scope(db_session, provider="aws") hidden, hidden_integ = await _mk_scope(db_session, provider="gcp") day = datetime.now(UTC).date() - timedelta(days=1) fresh = datetime.now(UTC) await _mk_fact(db_session, scope=mine, integ=mine_integ, day=day, updated_at=fresh) await _mk_fact(db_session, scope=hidden, integ=hidden_integ, day=day, updated_at=fresh) await db_session.commit() token = await _mk_client(db_session, visibility={"providers": ["aws"]}) resp = await client.get( f"{CHANGES}?since={_iso(datetime.now(UTC) - timedelta(hours=1))}", headers=_headers(token), ) assert [r["external_id"] for r in resp.json()["data"]] == [mine.external_id] async def test_since_older_than_horizon_is_refused_with_the_alternative( client, db_session: AsyncSession, ): token = await _mk_client(db_session) stale = datetime.now(UTC) - timedelta(days=200) resp = await client.get( f"{CHANGES}?since={_iso(stale)}", headers=_headers(token), ) assert resp.status_code in (400, 422) assert "full pull" in resp.text async def test_future_since_is_refused(client, db_session: AsyncSession): token = await _mk_client(db_session) future = datetime.now(UTC) + timedelta(days=1) resp = await client.get( f"{CHANGES}?since={_iso(future)}", headers=_headers(token), ) assert resp.status_code in (400, 422) assert "clock" in resp.text async def test_unescaped_plus_offset_is_accepted( client, db_session: AsyncSession, ): """``+00:00`` pasted into a URL arrives as a space. The intent is unambiguous, so it is parsed rather than 422'd.""" scope, integ = await _mk_scope(db_session) day = datetime.now(UTC).date() - timedelta(days=1) await _mk_fact(db_session, scope=scope, integ=integ, day=day, updated_at=datetime.now(UTC)) await db_session.commit() token = await _mk_client(db_session) raw = (datetime.now(UTC) - timedelta(hours=1)).replace( microsecond=0, ).isoformat() # unencoded: the "+" becomes a space on the wire resp = await client.get(f"{CHANGES}?since={raw}", headers=_headers(token)) assert resp.status_code == 200, resp.text assert len(resp.json()["data"]) == 1 async def test_naive_since_is_read_as_utc(client, db_session: AsyncSession): scope, integ = await _mk_scope(db_session) day = datetime.now(UTC).date() - timedelta(days=1) await _mk_fact(db_session, scope=scope, integ=integ, day=day, updated_at=datetime.now(UTC)) await db_session.commit() token = await _mk_client(db_session) naive = (datetime.now(UTC) - timedelta(hours=1)).replace( tzinfo=None, microsecond=0, ) resp = await client.get( f"{CHANGES}?since={naive.isoformat()}", headers=_headers(token), ) assert resp.status_code == 200 assert len(resp.json()["data"]) == 1 async def test_changes_is_audited(client, db_session: AsyncSession): token = await _mk_client(db_session) await client.get( f"{CHANGES}?since={_iso(datetime.now(UTC) - timedelta(hours=1))}", headers=_headers(token), ) rows = (await db_session.execute( select(AuditLog).where(AuditLog.action == "cloud.billing.read_changes"), )).scalars().all() assert len(rows) == 1 assert rows[0].actor_type == "service" assert (rows[0].detail or {}).get("since") # ── /billing/daily?updated_since= ───────────────────────────────────── async def test_daily_accepts_the_same_horizon_knob_as_changes( client, db_session: AsyncSession, ): """The two surfaces share one cursor, so they must share its limits: a horizon reachable on /billing/changes but not on /billing/daily would make the same `since` value valid on one and rejected on the other.""" token = await _mk_client(db_session) today = datetime.now(UTC).date() stale = datetime.now(UTC) - timedelta(days=60) # Outside the default 32-day horizon… default_horizon = await client.get( f"{DAILY}?updated_since={_iso(stale)}", headers=_headers(token), ) assert default_horizon.status_code in (400, 422) # …reachable once the caller widens it, exactly as on /changes. widened = await client.get( f"{DAILY}?start={today - timedelta(days=70)}&end={today}" f"&updated_since={_iso(stale)}&horizon_days=90", headers=_headers(token), ) assert widened.status_code == 200, widened.text assert widened.json()["meta"]["incremental"]["horizon"] async def test_daily_updated_since_filters_to_changed_cells( client, db_session: AsyncSession, ): scope, integ = await _mk_scope(db_session) today = datetime.now(UTC).date() stale_day = today - timedelta(days=6) fresh_day = today - timedelta(days=2) await _mk_fact(db_session, scope=scope, integ=integ, day=stale_day, amount="5", updated_at=datetime.now(UTC) - timedelta(days=4)) await _mk_fact(db_session, scope=scope, integ=integ, day=fresh_day, amount="7", updated_at=datetime.now(UTC)) await db_session.commit() token = await _mk_client(db_session) full = await client.get( f"{DAILY}?start={today - timedelta(days=10)}&end={today}", headers=_headers(token), ) assert {r["date"] for r in full.json()["data"]} == { stale_day.isoformat(), fresh_day.isoformat(), } since = _iso(datetime.now(UTC) - timedelta(hours=1)) incremental = await client.get( f"{DAILY}?start={today - timedelta(days=10)}&end={today}" f"&updated_since={since}", headers=_headers(token), ) body = incremental.json() assert [r["date"] for r in body["data"]] == [fresh_day.isoformat()] # Total reflects the filtered set, not the unfiltered one. assert body["meta"]["total"] == 1 assert body["meta"]["incremental"]["changed_cells"] == 1 assert body["meta"]["incremental"]["since"]