"""Tests for native GCP monthly billing via BigQuery billing export.""" from __future__ import annotations from datetime import date, timedelta from unittest.mock import MagicMock, patch import pytest from app.domains.cmdb.providers.base import ProviderCredentials from app.domains.cmdb.providers.exceptions import ProviderAuthError from app.domains.cmdb.providers.gcp import GcpProvider # Minimal but valid SA shape — google-auth's # from_service_account_info() is patched in each test so the contents # never need to validate against real Google APIs. _FAKE_SA = { "type": "service_account", "project_id": "host", "private_key": ( "-----BEGIN PRIVATE KEY-----\nfake\n-----END PRIVATE KEY-----\n" ), "client_email": "test@host.iam.gserviceaccount.com", "token_uri": "https://oauth2.googleapis.com/token", } def _creds(*, billing_bq_table: str | None = None) -> ProviderCredentials: fields = {"service_account_json": "{}"} if billing_bq_table is not None: fields["billing_bq_table"] = billing_bq_table return ProviderCredentials( provider_id="x", provider_type="gcp", account_alias="x", fields=fields, ) @pytest.mark.asyncio async def test_get_monthly_billing_aggregates_with_credits(): """Net cost = ``cost + credits.amount`` (credits are typically negative in GCP's billing export — they REDUCE total).""" with patch( "app.domains.cmdb.providers.gcp.service_account." "Credentials.from_service_account_info", return_value=MagicMock(), ): p = GcpProvider(_creds(billing_bq_table="proj.ds.gcp_billing_export_v1_X")) captured_sql: list[str] = [] captured_params: list = [] class FakeRow(dict): def __getitem__(self, key): return super().__getitem__(key) fake_rows = [ FakeRow(svc="Compute Engine", cost_sum=1000.0, credit_sum=-100.0, currency="USD"), FakeRow(svc="BigQuery", cost_sum=200.0, credit_sum=0, currency="USD"), FakeRow(svc="Cloud Storage", cost_sum=50.0, credit_sum=-5.0, currency="USD"), ] fake_job = MagicMock() fake_job.result.return_value = iter(fake_rows) fake_client = MagicMock() fake_client.query.side_effect = lambda sql, job_config=None: ( captured_sql.append(sql), captured_params.append( job_config.query_parameters if job_config else None, ), fake_job, )[2] with patch( "google.cloud.bigquery.Client", return_value=fake_client, ): out = await p.get_monthly_billing(date(2026, 3, 15)) await p.close() # 1000-100 + 200 + 50-5 = 1145 assert out["total_cost_usd"] == pytest.approx(1145.0) assert out["cost_by_service"]["Compute Engine"] == pytest.approx(900.0) assert out["cost_by_service"]["BigQuery"] == pytest.approx(200.0) assert out["cost_by_service"]["Cloud Storage"] == pytest.approx(45.0) assert out["currency"] == "USD" assert out["month"] == date(2026, 3, 1) # SQL pivots on invoice.month (not usage_start_time) assert "invoice.month = @invoice_month" in captured_sql[0] # Param value uses YYYYMM format params = captured_params[0] assert any( getattr(p, "value", None) == "202603" for p in params ) def test_split_bq_tables_handles_comma_and_whitespace(): """The ``billing_bq_table`` config accepts multiple tables for orgs with multiple billing accounts.""" from app.domains.cmdb.providers.gcp import GcpProvider as G assert G._split_bq_tables("p.d.t1") == ["p.d.t1"] assert G._split_bq_tables("p.d.t1,p.d.t2") == ["p.d.t1", "p.d.t2"] assert G._split_bq_tables("p.d.t1, p.d.t2 p.d.t3") == [ "p.d.t1", "p.d.t2", "p.d.t3", ] # Trailing/empty fragments stripped silently assert G._split_bq_tables(",p.d.t1,,") == ["p.d.t1"] assert G._split_bq_tables("") == [] assert G._split_bq_tables(" ") == [] @pytest.mark.asyncio async def test_multiple_bq_tables_union_all_in_query(): """Orgs with multiple billing accounts have one export table per billing account. The driver must UNION ALL across them in a single query (verified safer than wildcard ``gcp_billing_export_v1_*`` which BigQuery hung on for 20+ min in prod).""" with patch( "app.domains.cmdb.providers.gcp.service_account." "Credentials.from_service_account_info", return_value=MagicMock(), ): p = GcpProvider(_creds(billing_bq_table=( "proj.ds.gcp_billing_export_v1_AAA," "proj.ds.gcp_billing_export_v1_BBB," "proj.ds.gcp_billing_export_v1_CCC" ))) captured_sql: list[str] = [] fake_job = MagicMock() fake_job.result.return_value = iter([]) fake_client = MagicMock() fake_client.query.side_effect = lambda sql, job_config=None: ( captured_sql.append(sql), fake_job, )[1] with patch("google.cloud.bigquery.Client", return_value=fake_client): out = await p.get_monthly_billing(date(2026, 3, 1)) await p.close() sql = captured_sql[0] # All three tables appear in the SQL assert "gcp_billing_export_v1_AAA" in sql assert "gcp_billing_export_v1_BBB" in sql assert "gcp_billing_export_v1_CCC" in sql # UNION ALL between them (not wildcard) assert sql.count("UNION ALL") == 2 assert "gcp_billing_export_v1_*" not in sql # CRITICAL: must include _PARTITIONTIME predicate or BQ does a # full table scan on every history partition. Verified prod # 2026-04-26 query hung 20+ min without it; with it, ~3s. assert sql.count("_PARTITIONTIME >= TIMESTAMP") == 3 assert sql.count("_PARTITIONTIME < TIMESTAMP") == 3 # Pre-launch hardening 2026-05-05: partition window widened from # ±15 to ±45 days so SUD / committed-use credits that GCP back- # dates 30-45 days after invoice month aren't silently truncated. # March 2026 → lo = (Mar 1 - 45d) = Jan 15; hi = (Apr 1 + 45d) = May 16. assert "TIMESTAMP('2026-01-15')" in sql assert "TIMESTAMP('2026-05-16')" in sql # raw_response captures the full table list assert out["raw_response"]["bq_tables"] == [ "proj.ds.gcp_billing_export_v1_AAA", "proj.ds.gcp_billing_export_v1_BBB", "proj.ds.gcp_billing_export_v1_CCC", ] def test_partition_window_handles_year_boundary(): """Partition window for January queries the previous November's last partitions; for December the next February. Pin so a future refactor doesn't introduce off-by-one or wrap bugs. Pre-launch hardening 2026-05-05: window widened from ±15 to ±45 days so SUD / committed-use credits that GCP back-dates 30-45 days after invoice month aren't silently truncated. """ from datetime import date, timedelta # Mirror the partition-lo/hi calculation in the implementation — # this is a code-side regression guard. def window(target: date) -> tuple[str, str]: anchor = target.replace(day=1) lo = anchor - timedelta(days=45) if anchor.month == 12: next_first = anchor.replace(year=anchor.year + 1, month=1, day=1) else: next_first = anchor.replace(month=anchor.month + 1, day=1) hi = next_first + timedelta(days=45) return lo.isoformat(), hi.isoformat() # 2026-01-01 - 45d = 2025-11-17 ; 2026-02-01 + 45d = 2026-03-18 assert window(date(2026, 1, 1)) == ("2025-11-17", "2026-03-18") # 2026-12-01 - 45d = 2026-10-17 ; 2027-01-01 + 45d = 2027-02-15 assert window(date(2026, 12, 1)) == ("2026-10-17", "2027-02-15") # 2026-06-01 - 45d = 2026-04-17 ; 2026-07-01 + 45d = 2026-08-15 assert window(date(2026, 6, 1)) == ("2026-04-17", "2026-08-15") @pytest.mark.asyncio async def test_missing_bq_table_raises_with_help_text(): """If ``billing_bq_table`` isn't set, surface a clear error pointing at the GCP doc — silent $0 rows would be much worse.""" with patch( "app.domains.cmdb.providers.gcp.service_account." "Credentials.from_service_account_info", return_value=MagicMock(), ): p = GcpProvider(_creds()) # no billing_bq_table with pytest.raises(ProviderAuthError, match="billing_bq_table"): await p.get_monthly_billing(date(2026, 3, 1)) await p.close() @pytest.mark.asyncio async def test_default_month_is_previous_full_month(): with patch( "app.domains.cmdb.providers.gcp.service_account." "Credentials.from_service_account_info", return_value=MagicMock(), ): p = GcpProvider(_creds(billing_bq_table="p.d.t")) fake_job = MagicMock() fake_job.result.return_value = iter([]) fake_client = MagicMock() fake_client.query.return_value = fake_job with patch("google.cloud.bigquery.Client", return_value=fake_client): out = await p.get_monthly_billing() await p.close() today = date.today() expected = (today.replace(day=1) - timedelta(days=1)).replace(day=1) assert out["month"] == expected