"""Unit tests for the GCP cloud driver. Google API discovery chains are deep (``crm.projects().search().execute()``), so instead of mocking every intermediate we build a small helper that returns a stubbed service client keyed by ``(service, version)``. Each test only stubs the endpoints it cares about. """ from __future__ import annotations import json from datetime import UTC, datetime from typing import Any from unittest.mock import AsyncMock, MagicMock, patch import pytest from app.integrations.driver_base import ( CloudAccessContext, CloudAccountRevokeContext, CloudPrincipalNotFoundError, CloudScopeInfo, CloudScopePartitionStatus, CloudScopeRef, ProvisioningResult, ) from app.integrations.drivers.gcp_cloud import GcpCloudDriver _SA_JSON = json.dumps({ "type": "service_account", "project_id": "test-project", "private_key_id": "x", "private_key": "-----BEGIN PRIVATE KEY-----\nfake\n-----END PRIVATE KEY-----\n", "client_email": "sa@test-project.iam.gserviceaccount.com", "client_id": "0", "auth_uri": "https://accounts.google.com/o/oauth2/auth", "token_uri": "https://oauth2.googleapis.com/token", }) _ORG_ID = "478164292948" def _driver( *, with_impersonation: bool = False, **config_overrides: Any, ) -> GcpCloudDriver: cfg = { "organization_id": _ORG_ID, "service_account_json": _SA_JSON, "billing_scope_mode": "projects_only", } if with_impersonation: cfg["admin_impersonation_email"] = "admin@pingcap.com" cfg.update(config_overrides) return GcpCloudDriver(integration_id="test-gcp", config=cfg) class _FakeHttpError(Exception): """Mimics googleapiclient.errors.HttpError.resp.status.""" def __init__(self, status: int, content: bytes = b"") -> None: super().__init__(f"HTTP {status}") self.resp = MagicMock() self.resp.status = status self.content = content def _attach_call(obj: MagicMock, path: list[str], final_return: Any) -> MagicMock: """Wire ``obj.a().b().c(...).execute() == final_return`` for any chain. We want each intermediate call (service accessor like ``.projects()``) to be a MagicMock that returns a new MagicMock, and the terminal call to have a deterministic ``.execute()`` value. googleapiclient chains like ``crm.projects().search(...).execute()`` need exactly this shape. """ node = obj for step in path: child = MagicMock() # Accessor attr returns a callable returning `child` call = MagicMock(return_value=child) setattr(node, step, call) node = child node.execute = MagicMock(return_value=final_return) return node def _with_svc(svcs: dict[tuple[str, str], MagicMock]): """Patch GcpCloudDriver._svc to return the matching mock.""" crm = svcs.get(("cloudresourcemanager", "v3")) if crm is not None: # Keep legacy fixtures concise while exercising the production # parent-bounded traversal. Existing search side effects supply each # parent page; the adapter adds the parent field required by the # fail-closed helper and exposes an empty folder collection. def _list_projects(**kwargs: Any) -> MagicMock: request = crm.projects.return_value.search( query="state:ACTIVE", pageToken=kwargs.get("pageToken"), ) original_execute = request.execute def _execute(*, num_retries: int = 0) -> Any: # noqa: ARG001 response = original_execute() for project in response.get("projects", []): if isinstance(project, dict): project.setdefault("parent", kwargs["parent"]) return response request.execute = MagicMock(side_effect=_execute) return request crm.projects.return_value.list.side_effect = _list_projects empty_folders = MagicMock() empty_folders.execute.return_value = {"folders": []} crm.folders.return_value.list.return_value = empty_folders def _factory(self, service: str, version: str, *, admin: bool = False): # noqa: ARG001 key = (service, version) if key not in svcs: raise AssertionError( f"Unexpected _svc({service!r}, {version!r}, admin={admin})" ) return svcs[key] return patch.object(GcpCloudDriver, "_svc", new=_factory) # ── metadata ──────────────────────────────────────────────────── def test_driver_metadata() -> None: m = GcpCloudDriver.metadata() assert m.kind == "cloud" assert m.provider == "gcp" assert "servers" in m.capabilities assert "user_revoke" in m.capabilities # ── list_scopes ───────────────────────────────────────────────── @pytest.mark.asyncio async def test_list_scopes_uses_parent_bounded_hierarchy() -> None: crm = MagicMock() search_call = MagicMock() search_call.execute.return_value = { "projects": [ {"projectId": "proj-a", "displayName": "Proj A", "parent": f"organizations/{_ORG_ID}", "labels": {}}, {"projectId": "proj-b", "displayName": "Proj B", "parent": f"organizations/{_ORG_ID}"}, ], } crm.projects.return_value.search.return_value = search_call with _with_svc({("cloudresourcemanager", "v3"): crm}): scopes = await _driver().list_scopes() kwargs = crm.projects.return_value.list.call_args.kwargs assert kwargs["parent"] == f"organizations/{_ORG_ID}" assert kwargs["showDeleted"] is False assert len(scopes) == 2 assert scopes[0].external_id == "proj-a" assert scopes[0].raw["organization_id"] == _ORG_ID assert scopes[1].raw["parent"] == f"organizations/{_ORG_ID}" @pytest.mark.asyncio async def test_list_scopes_paginates() -> None: crm = MagicMock() # Build two pages: first returns nextPageToken, second returns None. page1 = MagicMock() page1.execute.return_value = { "projects": [{"projectId": "p1"}], "nextPageToken": "tok", } page2 = MagicMock() page2.execute.return_value = { "projects": [{"projectId": "p2"}], } crm.projects.return_value.search.side_effect = [page1, page2] with _with_svc({("cloudresourcemanager", "v3"): crm}): scopes = await _driver().list_scopes() assert [s.external_id for s in scopes] == ["p1", "p2"] assert crm.projects.return_value.search.call_count == 2 # ── list_scope_roles ──────────────────────────────────────────── @pytest.mark.asyncio async def test_list_scope_roles_merges_defaults_and_custom() -> None: iam = MagicMock() list_call = MagicMock() list_call.execute.return_value = { "roles": [ { "name": "projects/proj-a/roles/customReader", "title": "Custom Reader", "description": "Read-only for data team", "stage": "GA", }, ], } iam.projects.return_value.roles.return_value.list.return_value = list_call iam.projects.return_value.roles.return_value.list_next.return_value = None with _with_svc({("iam", "v1"): iam}), \ patch( "app.integrations.drivers._cloud_defaults.load_default_roles", return_value=[{"role_key": "roles/viewer", "display_name": "V"}], ): ref = CloudScopeRef(scope_type="gcp_project", external_id="proj-a") roles = await _driver().list_scope_roles(ref) # One default + one custom assert len(roles) == 2 assert roles[0].role_key == "roles/viewer" assert roles[0].is_custom is False assert roles[1].role_key == "projects/proj-a/roles/customReader" assert roles[1].is_custom is True @pytest.mark.asyncio async def test_list_scope_roles_drops_billing_account_only_roles() -> None: # A stale platform_setting may still carry roles/billing.viewer — it # can't bind on a project, so the dropdown must not offer it. iam = MagicMock() list_call = MagicMock() list_call.execute.return_value = {"roles": []} iam.projects.return_value.roles.return_value.list.return_value = list_call iam.projects.return_value.roles.return_value.list_next.return_value = None with _with_svc({("iam", "v1"): iam}), \ patch( "app.integrations.drivers._cloud_defaults.load_default_roles", return_value=[ {"role_key": "roles/viewer", "display_name": "V"}, {"role_key": "roles/billing.viewer", "display_name": "Billing Viewer"}, ], ): ref = CloudScopeRef(scope_type="gcp_project", external_id="proj-a") roles = await _driver().list_scope_roles(ref) keys = {r.role_key for r in roles} assert "roles/viewer" in keys assert "roles/billing.viewer" not in keys # ── billing-account scope (PR5) ────────────────────────────────── def test_list_billing_accounts_sync_emits_open_accounts_only() -> None: billing = MagicMock() accounts = { "billingAccounts/AAAA-BBBB-CCCC": { "name": "billingAccounts/AAAA-BBBB-CCCC", "displayName": "Corp", "open": True, "parent": f"organizations/{_ORG_ID}", }, "billingAccounts/DDDD-EEEE-FFFF": { "name": "billingAccounts/DDDD-EEEE-FFFF", "displayName": "Closed", "open": False, "parent": f"organizations/{_ORG_ID}", }, } def _get(*, name: str) -> MagicMock: request = MagicMock() request.execute.return_value = accounts[name] return request billing.billingAccounts.return_value.get.side_effect = _get with _with_svc({("cloudbilling", "v1"): billing}): scopes = _driver( billing_scope_mode="allowlist", billing_account_allowlist="\n".join(accounts), )._list_billing_accounts_sync() assert len(scopes) == 1 assert scopes[0].scope_type == "gcp_billing_account" assert scopes[0].external_id == "billingAccounts/AAAA-BBBB-CCCC" assert scopes[0].display_name == "Corp" def test_list_billing_accounts_sync_verifies_reseller_parent_chain() -> None: billing = MagicMock() accounts = { "billingAccounts/AAAA-BBBB-CCCC": { "name": "billingAccounts/AAAA-BBBB-CCCC", "open": True, "parent": "billingAccounts/PARENT-ACCOUNT-001", }, "billingAccounts/PARENT-ACCOUNT-001": { "name": "billingAccounts/PARENT-ACCOUNT-001", "open": True, "parent": f"organizations/{_ORG_ID}", }, } def _get(*, name: str) -> MagicMock: request = MagicMock() request.execute.return_value = accounts[name] return request billing.billingAccounts.return_value.get.side_effect = _get with _with_svc({("cloudbilling", "v1"): billing}): scopes = _driver( billing_scope_mode="allowlist", billing_account_allowlist="billingAccounts/AAAA-BBBB-CCCC", )._list_billing_accounts_sync() assert [s.external_id for s in scopes] == ["billingAccounts/AAAA-BBBB-CCCC"] assert billing.billingAccounts.return_value.get.call_count == 2 assert not billing.billingAccounts.return_value.list.called @pytest.mark.parametrize("status", [403, 429, 503]) @pytest.mark.asyncio async def test_list_scopes_marks_billing_partition_incomplete_on_api_error( status: int, ) -> None: driver = _driver() project = CloudScopeInfo( scope_type="gcp_project", external_id="project-ok", ) with ( patch.object(driver, "_list_projects_sync", return_value=[project]), patch.object( driver, "_list_billing_accounts_sync", side_effect=_FakeHttpError(status), ), ): scopes = await driver.list_scopes() assert scopes == [project] partitions = driver.scope_snapshot_partitions() assert partitions["gcp_project"].complete is True assert partitions["gcp_billing_account"].complete is False assert f"HTTP {status}" in (partitions["gcp_billing_account"].error or "") @pytest.mark.asyncio @pytest.mark.parametrize( "account", [ {"displayName": "missing-id"}, {"name": "billingAccounts/"}, {"name": "not-a-billing-account"}, "bad-item", ], ) async def test_list_scopes_marks_malformed_billing_item_incomplete( account: object, ) -> None: driver = _driver( billing_scope_mode="allowlist", billing_account_allowlist="billingAccounts/AAAA-BBBB-CCCC", ) project = CloudScopeInfo( scope_type="gcp_project", external_id="project-ok", ) billing = MagicMock() billing.billingAccounts.return_value.get.return_value.execute.return_value = account with ( patch.object(driver, "_list_projects_sync", return_value=[project]), _with_svc({("cloudbilling", "v1"): billing}), ): scopes = await driver.list_scopes() assert scopes == [project] partition = driver.scope_snapshot_partitions()["gcp_billing_account"] assert partition.complete is False assert "snapshot is incomplete" in (partition.error or "") def test_list_billing_accounts_sync_rejects_foreign_organization() -> None: billing = MagicMock() billing.billingAccounts.return_value.get.return_value.execute.return_value = { "name": "billingAccounts/AAAA-BBBB-CCCC", "open": True, "parent": "organizations/999999999999", } with _with_svc({("cloudbilling", "v1"): billing}): with pytest.raises(RuntimeError, match="not configured"): _driver( billing_scope_mode="allowlist", billing_account_allowlist="billingAccounts/AAAA-BBBB-CCCC", )._list_billing_accounts_sync() def test_list_billing_accounts_sync_requires_explicit_policy() -> None: with pytest.raises(RuntimeError, match="must explicitly be"): GcpCloudDriver( integration_id="test-gcp", config={ "organization_id": _ORG_ID, "service_account_json": _SA_JSON, }, )._list_billing_accounts_sync() def test_list_projects_sync_walks_every_page() -> None: crm = MagicMock() execute = crm.projects.return_value.search.return_value.execute execute.side_effect = [ { "projects": [{"projectId": "project-a"}], "nextPageToken": "page-2", }, {"projects": [{"projectId": "project-b"}]}, ] with _with_svc({("cloudresourcemanager", "v3"): crm}): scopes = _driver()._list_projects_sync(_ORG_ID) assert [scope.external_id for scope in scopes] == [ "project-a", "project-b", ] def test_list_projects_sync_repeated_token_fails_snapshot() -> None: crm = MagicMock() crm.projects.return_value.search.return_value.execute.side_effect = [ {"projects": [], "nextPageToken": "repeat"}, {"projects": [], "nextPageToken": "repeat"}, ] with _with_svc({("cloudresourcemanager", "v3"): crm}): with pytest.raises(RuntimeError, match="repeated nextPageToken"): _driver()._list_projects_sync(_ORG_ID) @pytest.mark.asyncio @pytest.mark.parametrize("project", [{"projectId": None}, {"projectId": ""}, "bad"]) async def test_list_scopes_marks_malformed_project_item_incomplete( project: object, ) -> None: driver = _driver() crm = MagicMock() crm.projects.return_value.search.return_value.execute.return_value = { "projects": [project], } with _with_svc({("cloudresourcemanager", "v3"): crm}): with pytest.raises(RuntimeError, match="snapshot is incomplete"): await driver.list_scopes() partition = driver.scope_snapshot_partitions()["gcp_project"] assert partition.complete is False assert "snapshot is incomplete" in (partition.error or "") def test_list_projects_sync_page_cap_fails_snapshot() -> None: crm = MagicMock() crm.projects.return_value.search.return_value.execute.side_effect = [ {"projects": [], "nextPageToken": "page-2"}, {"projects": [], "nextPageToken": "page-3"}, ] with ( _with_svc({("cloudresourcemanager", "v3"): crm}), patch("app.domains.cmdb.providers.gcp._MAX_ORG_HIERARCHY_PAGES", 2), ): with pytest.raises(RuntimeError, match="page cap"): _driver()._list_projects_sync(_ORG_ID) @pytest.mark.asyncio async def test_list_scopes_marks_project_partition_incomplete_and_raises() -> None: driver = _driver() with patch.object( driver, "_list_projects_sync", side_effect=RuntimeError("partial pager"), ): with pytest.raises(RuntimeError, match="partial pager"): await driver.list_scopes() partition = driver.scope_snapshot_partitions()["gcp_project"] assert partition.complete is False assert "partial pager" in (partition.error or "") @pytest.mark.asyncio async def test_connection_is_red_when_billing_partition_is_incomplete() -> None: driver = _driver() project = CloudScopeInfo(scope_type="gcp_project", external_id="project-a") with ( patch.object(driver, "_list_projects_sync", return_value=[project]), patch.object( driver, "_list_billing_accounts_sync", side_effect=RuntimeError("billing HTTP 503"), ), ): result = await driver.test_connection() assert result.ok is False assert "gcp_billing_account" in result.message assert "HTTP 503" in result.message assert result.detail["dry_run"] is True assert result.detail["discovered_scopes"] == [ {"scope_type": "gcp_project", "external_id": "project-a"}, ] assert result.detail["scope_partitions"]["gcp_project"]["complete"] is True assert ( result.detail["scope_partitions"]["gcp_billing_account"]["complete"] is False ) @pytest.mark.asyncio async def test_connection_redacts_scope_probe_error() -> None: driver = _driver() secret = "access_token=ya29.secret-provider-token" with patch.object( driver, "list_scopes", new=AsyncMock(side_effect=RuntimeError(f"request failed: {secret}")), ): result = await driver.test_connection() assert result.ok is False assert secret not in result.message assert "access_token=" in result.message @pytest.mark.asyncio async def test_connection_redacts_admin_sdk_probe_error() -> None: driver = _driver(with_impersonation=True) secret = "access_token=ya29.secret-admin-token" with ( patch.object( driver, "list_scopes", new=AsyncMock(return_value=[]), ), patch.object( driver, "_probe_admin_sdk_sync", side_effect=RuntimeError(f"admin request failed: {secret}"), ), ): result = await driver.test_connection() assert result.ok is False assert secret not in result.message assert "access_token=" in result.message assert "Admin SDK impersonation FAILED" in result.message @pytest.mark.asyncio async def test_revoke_account_refuses_partial_billing_without_iam_mutation() -> None: driver = _driver() project = CloudScopeInfo(scope_type="gcp_project", external_id="project-a") driver._scope_snapshot_partitions = { "gcp_project": CloudScopePartitionStatus(complete=True), "gcp_billing_account": CloudScopePartitionStatus( complete=False, error="billing HTTP 503", ), } revoke_sync = MagicMock() with ( patch.object(driver, "list_scopes", new=AsyncMock(return_value=[project])), patch.object(driver, "_revoke_account_sync", new=revoke_sync), ): result = await driver.revoke_account( CloudAccountRevokeContext( provider="gcp", account_id="leaver@example.com", ), ) assert result.ok is False assert "gcp_billing_account" in (result.error_message or "") assert "No IAM changes" in (result.error_message or "") revoke_sync.assert_not_called() @pytest.mark.asyncio async def test_revoke_account_auditable_failure_on_project_enumeration_error() -> None: driver = _driver() revoke_sync = MagicMock() with ( patch.object( driver, "list_scopes", new=AsyncMock(side_effect=RuntimeError("projects HTTP 503")), ), patch.object(driver, "_revoke_account_sync", new=revoke_sync), ): result = await driver.revoke_account( CloudAccountRevokeContext( provider="gcp", account_id="leaver@example.com", ), ) assert result.ok is False assert "scope enumeration failed" in (result.error_message or "") assert "HTTP 503" in (result.error_message or "") assert "No IAM changes" in (result.error_message or "") revoke_sync.assert_not_called() @pytest.mark.asyncio async def test_revoke_account_drift_baseline_only_counts_active_synced_identities() -> None: driver = _driver() scopes = [ CloudScopeInfo(scope_type="gcp_project", external_id=f"project-{i}") for i in range(5) ] captured_statements: list[object] = [] result_proxy = MagicMock() result_proxy.scalar.return_value = 5 class _DriftSession: async def execute(self, statement: object) -> object: captured_statements.append(statement) return result_proxy class _Factory: async def __aenter__(self) -> _DriftSession: return _DriftSession() async def __aexit__(self, *args: object) -> None: return None revoke_sync = MagicMock(return_value=ProvisioningResult(ok=True)) with ( patch.object(driver, "list_scopes", new=AsyncMock(return_value=scopes)), patch.object(driver, "_revoke_account_sync", new=revoke_sync), patch("app.core.database.async_session_factory", return_value=_Factory()), ): result = await driver.revoke_account( CloudAccountRevokeContext( provider="gcp", account_id="leaver@example.com", ), ) assert result.ok is True revoke_sync.assert_called_once_with("leaver@example.com", scopes) query = str(captured_statements[0]) assert "cmdb_cloud_scopes.status =" in query assert "cmdb_cloud_scopes.deleted_at IS NULL" in query assert "cmdb_cloud_scopes.last_seen_sync_job_id IS NOT NULL" in query assert "GROUP BY cmdb_cloud_scopes.scope_type" in query assert "cmdb_cloud_scopes.external_id" in query @pytest.mark.asyncio async def test_revoke_account_drift_guard_deduplicates_current_scope_snapshot() -> None: driver = _driver() unique_scopes = [ CloudScopeInfo(scope_type="gcp_project", external_id=f"project-{i}") for i in range(5) ] scopes = unique_scopes + unique_scopes result_proxy = MagicMock() result_proxy.scalar.return_value = 6 class _DriftSession: async def execute(self, _statement: object) -> object: return result_proxy class _Factory: async def __aenter__(self) -> _DriftSession: return _DriftSession() async def __aexit__(self, *args: object) -> None: return None revoke_sync = MagicMock() with ( patch.object(driver, "list_scopes", new=AsyncMock(return_value=scopes)), patch.object(driver, "_revoke_account_sync", new=revoke_sync), patch("app.core.database.async_session_factory", return_value=_Factory()), ): result = await driver.revoke_account( CloudAccountRevokeContext( provider="gcp", account_id="leaver@example.com", ), ) assert result.ok is False assert result.raw_response["scopes_today"] == 5 assert result.raw_response["scopes_prior"] == 6 revoke_sync.assert_not_called() @pytest.mark.asyncio async def test_grant_access_billing_account_uses_cloudbilling() -> None: billing = MagicMock() billing.billingAccounts.return_value.getIamPolicy.return_value.execute.return_value = { "bindings": [], "etag": "x", } billing.billingAccounts.return_value.setIamPolicy.return_value.execute.return_value = {} crm = MagicMock() with _with_svc({("cloudbilling", "v1"): billing, ("cloudresourcemanager", "v3"): crm}): ctx = CloudAccessContext( principal_email="alice@pingcap.com", principal_display_name="Alice", scope=CloudScopeRef( scope_type="gcp_billing_account", external_id="billingAccounts/AAAA-BBBB-CCCC", ), role_key="roles/billing.viewer", ) result = await _driver().grant_access(ctx) assert result.ok # bound on the billing account, NOT a project sent = billing.billingAccounts.return_value.setIamPolicy.call_args.kwargs assert sent["resource"] == "billingAccounts/AAAA-BBBB-CCCC" members = sent["body"]["policy"]["bindings"][0]["members"] assert members == ["user:alice@pingcap.com"] assert not crm.projects.return_value.setIamPolicy.called @pytest.mark.asyncio async def test_grant_billing_account_getiampolicy_requests_v3() -> None: """#1654: the billing-account read-modify-write MUST request policy version 3 on getIamPolicy — without it, setIamPolicy silently DROPS any condition-bearing (v3) binding on the billing account. The project path already did this; billing was the holdout.""" billing = MagicMock() getpol = billing.billingAccounts.return_value.getIamPolicy getpol.return_value.execute.return_value = {"bindings": [], "etag": "x"} billing.billingAccounts.return_value.setIamPolicy.return_value.execute.return_value = {} crm = MagicMock() with _with_svc({("cloudbilling", "v1"): billing, ("cloudresourcemanager", "v3"): crm}): ctx = CloudAccessContext( principal_email="alice@pingcap.com", principal_display_name="Alice", scope=CloudScopeRef( scope_type="gcp_billing_account", external_id="billingAccounts/AAAA-BBBB-CCCC", ), role_key="roles/billing.viewer", ) result = await _driver().grant_access(ctx) assert result.ok kwargs = getpol.call_args.kwargs assert kwargs["resource"] == "billingAccounts/AAAA-BBBB-CCCC" # cloudbilling v1 getIamPolicy is a GET: the version is the query kwarg # options_requestedPolicyVersion, NOT a body= (which raised a TypeError # in prod — #1654 regression). See test_billing_getiampolicy_rejects_body. assert kwargs["options_requestedPolicyVersion"] == 3 assert "body" not in kwargs def test_revoke_account_billing_scope_uses_cloudbilling() -> None: """#1642: revoke_account must route a gcp_billing_account scope through the cloudbilling API — the sweep previously ran projects().getIamPolicy against a ``billingAccounts/…`` external_id, which errored, so the leaver kept their billing-account IAM and the whole revoke returned ok=False (502).""" billing = MagicMock() getpol = billing.billingAccounts.return_value.getIamPolicy getpol.return_value.execute.return_value = { "bindings": [ { "role": "roles/billing.viewer", "members": ["user:alice@pingcap.com", "user:keep@pingcap.com"], }, ], "etag": "x", } billing.billingAccounts.return_value.setIamPolicy.return_value.execute.return_value = {} crm = MagicMock() with _with_svc({("cloudbilling", "v1"): billing, ("cloudresourcemanager", "v3"): crm}): result = _driver()._revoke_account_sync( "alice@pingcap.com", [CloudScopeInfo( scope_type="gcp_billing_account", external_id="billingAccounts/AAAA-BBBB-CCCC", )], ) assert result.ok # Removed via cloudbilling; the project IAM API was never touched. assert not crm.projects.return_value.getIamPolicy.called sent = billing.billingAccounts.return_value.setIamPolicy.call_args.kwargs assert sent["resource"] == "billingAccounts/AAAA-BBBB-CCCC" remaining = sent["body"]["policy"]["bindings"][0]["members"] assert remaining == ["user:keep@pingcap.com"] # only the leaver removed # v3 requested on the billing read so conditional bindings survive. assert getpol.call_args.kwargs["options_requestedPolicyVersion"] == 3 assert "body" not in getpol.call_args.kwargs @pytest.mark.asyncio async def test_list_billing_grants_getiampolicy_requests_v3() -> None: """#1654: reverse discovery on a billing account must ALSO request policy version 3 so conditional bindings are enumerated, not silently collapsed to their non-conditional form during drift comparison.""" billing = MagicMock() getpol = billing.billingAccounts.return_value.getIamPolicy getpol.return_value.execute.return_value = { "bindings": [ {"role": "roles/billing.viewer", "members": ["user:bob@pingcap.com"]}, ], } with _with_svc({("cloudbilling", "v1"): billing}): ref = CloudScopeRef( scope_type="gcp_billing_account", external_id="billingAccounts/AAAA-BBBB-CCCC", ) grants = await _driver().list_scope_grants(ref) assert len(grants) == 1 kwargs = getpol.call_args.kwargs assert kwargs["options_requestedPolicyVersion"] == 3 assert "body" not in kwargs class _StrictBillingGetIamPolicy: """Mimics the real googleapiclient cloudbilling v1 ``getIamPolicy``: it is a GET with NO request body, so the generated method accepts ``resource=`` and ``options_requestedPolicyVersion=`` ONLY. Passing ``body=`` raises ``TypeError: Got an unexpected keyword argument body`` — exactly the prod failure (#1654 regression). Plain ``MagicMock`` swallows any kwarg, which is why the original tests passed while ``cmdb.grants_sync`` broke in prod. This fake enforces the signature.""" def __init__(self, policy: dict[str, Any]) -> None: self._policy = policy self.last_kwargs: dict[str, Any] | None = None def __call__(self, **kwargs: Any) -> Any: if "body" in kwargs: raise TypeError("Got an unexpected keyword argument body") allowed = {"resource", "options_requestedPolicyVersion"} extra = set(kwargs) - allowed if extra: raise TypeError( f"Got an unexpected keyword argument {next(iter(extra))}" ) self.last_kwargs = kwargs req = MagicMock() req.execute = MagicMock(return_value=self._policy) return req @pytest.mark.asyncio async def test_list_billing_grants_getiampolicy_rejects_body_kwarg() -> None: """Regression for cmdb.grants_sync partial_failure: ``TypeError: Got an unexpected keyword argument body`` on ``billingAccounts/…``. The billing getIamPolicy is a GET — driver must pass options_requestedPolicyVersion, never body=. Uses a strict fake that enforces the real signature so the bug cannot slip past MagicMock again.""" strict = _StrictBillingGetIamPolicy( {"bindings": [ {"role": "roles/billing.viewer", "members": ["user:bob@pingcap.com"]}, ]}, ) billing = MagicMock() billing.billingAccounts.return_value.getIamPolicy = strict with _with_svc({("cloudbilling", "v1"): billing}): ref = CloudScopeRef( scope_type="gcp_billing_account", external_id="billingAccounts/AAAA-BBBB-CCCC", ) grants = await _driver().list_scope_grants(ref) assert [g.principal_email for g in grants] == ["bob@pingcap.com"] assert strict.last_kwargs == { "resource": "billingAccounts/AAAA-BBBB-CCCC", "options_requestedPolicyVersion": 3, } @pytest.mark.asyncio async def test_list_scope_roles_billing_account_returns_billing_viewer() -> None: ref = CloudScopeRef( scope_type="gcp_billing_account", external_id="billingAccounts/AAAA-BBBB-CCCC", ) roles = await _driver().list_scope_roles(ref) keys = {r.role_key: r for r in roles} assert "roles/billing.viewer" in keys assert keys["roles/billing.viewer"].is_promoted is True def test_billing_viewer_normal_on_billing_account_but_ungrantable_on_project() -> None: d = _driver() assert d.classify_role( "roles/billing.viewer", "Billing Viewer", "gcp_billing_account", ).tier == "normal" assert d.classify_role( "roles/billing.viewer", "Billing Viewer", "gcp_project", ).tier == "ungrantable" @pytest.mark.asyncio async def test_list_scope_grants_billing_account_reads_cloudbilling() -> None: billing = MagicMock() billing.billingAccounts.return_value.getIamPolicy.return_value.execute.return_value = { "bindings": [ {"role": "roles/billing.viewer", "members": [ "user:bob@pingcap.com", "group:fin@pingcap.com", ]}, ], } with _with_svc({("cloudbilling", "v1"): billing}): ref = CloudScopeRef( scope_type="gcp_billing_account", external_id="billingAccounts/AAAA-BBBB-CCCC", ) grants = await _driver().list_scope_grants(ref) assert len(grants) == 1 # group: filtered out assert grants[0].principal_email == "bob@pingcap.com" assert grants[0].role_key == "roles/billing.viewer" # ── grant_access ──────────────────────────────────────────────── def test_transport_build_has_socket_timeout() -> None: driver = _driver() credentials = MagicMock() http = MagicMock() authorized = MagicMock() service = MagicMock() with ( patch.object(driver, "_credentials", return_value=credentials), patch("app.integrations.drivers.gcp_cloud.httplib2.Http", return_value=http) as http_cls, patch( "app.integrations.drivers.gcp_cloud.google_auth_httplib2.AuthorizedHttp", return_value=authorized, ) as authorized_cls, patch("app.integrations.drivers.gcp_cloud.build", return_value=service) as build_mock, ): assert driver._svc("cloudresourcemanager", "v3") is service http_cls.assert_called_once_with(timeout=30) authorized_cls.assert_called_once_with(credentials, http=http) build_mock.assert_called_once_with( "cloudresourcemanager", "v3", http=authorized, cache_discovery=False, ) def test_read_and_write_execute_use_distinct_retry_budgets() -> None: read_request = MagicMock() write_request = MagicMock() GcpCloudDriver._execute_read(read_request) GcpCloudDriver._execute_write(write_request) read_request.execute.assert_called_once_with(num_retries=3) write_request.execute.assert_called_once_with(num_retries=0) def test_create_parent_folder_ancestry_must_reach_configured_org() -> None: crm = MagicMock() crm.folders.return_value.get.return_value.execute.side_effect = [ {"name": "folders/222", "parent": "folders/111"}, { "name": "folders/111", "parent": f"organizations/{_ORG_ID}", }, ] _driver()._assert_resource_parent_in_organization( crm, "folders/222", organization=f"organizations/{_ORG_ID}", ) assert crm.folders.return_value.get.call_count == 2 for call in crm.folders.return_value.get.return_value.execute.call_args_list: assert call.kwargs == {"num_retries": 3} def test_create_parent_rejects_foreign_organization_before_write() -> None: crm = MagicMock() with pytest.raises(RuntimeError, match="outside configured"): _driver()._assert_resource_parent_in_organization( crm, "organizations/999999999999", organization=f"organizations/{_ORG_ID}", ) assert not crm.projects.return_value.create.called @pytest.mark.asyncio async def test_grant_reconciles_unknown_write_as_applied_without_replay() -> None: crm = MagicMock() member = "user:alice@pingcap.com" crm.projects.return_value.getIamPolicy.return_value.execute.side_effect = [ {"bindings": [], "etag": "before"}, { "bindings": [ {"role": "roles/viewer", "members": [member]}, ], "etag": "after", }, ] write_execute = crm.projects.return_value.setIamPolicy.return_value.execute write_execute.side_effect = TimeoutError("socket timed out after commit") ctx = CloudAccessContext( principal_email="alice@pingcap.com", principal_display_name="Alice", scope=CloudScopeRef(scope_type="gcp_project", external_id="proj-a"), role_key="roles/viewer", ) with _with_svc({("cloudresourcemanager", "v3"): crm}): result = await _driver().grant_access(ctx) assert result.ok is True assert result.detail["write_outcome"] == "reconciled_applied" assert write_execute.call_count == 1 write_execute.assert_called_once_with(num_retries=0) @pytest.mark.asyncio async def test_grant_reconciliation_confirms_unknown_write_not_applied() -> None: crm = MagicMock() crm.projects.return_value.getIamPolicy.return_value.execute.side_effect = [ {"bindings": [], "etag": "before"}, {"bindings": [], "etag": "after"}, ] write_execute = crm.projects.return_value.setIamPolicy.return_value.execute write_execute.side_effect = TimeoutError("socket timed out") ctx = CloudAccessContext( principal_email="alice@pingcap.com", principal_display_name="Alice", scope=CloudScopeRef(scope_type="gcp_project", external_id="proj-a"), role_key="roles/viewer", ) with _with_svc({("cloudresourcemanager", "v3"): crm}): result = await _driver().grant_access(ctx) assert result.ok is False assert result.detail["write_outcome"] == "confirmed_not_applied" assert write_execute.call_count == 1 @pytest.mark.asyncio async def test_grant_access_refuses_billing_account_role_without_api_call() -> None: # roles/billing.viewer is a billing-account/org-level role — GCP # rejects it on a project setIamPolicy with a 400. The driver must # fail fast with an actionable message and never touch the API. crm = MagicMock() with _with_svc({("cloudresourcemanager", "v3"): crm}): ctx = CloudAccessContext( principal_email="alex.quan@pingcap.com", principal_display_name="Alex Quan", scope=CloudScopeRef(scope_type="gcp_project", external_id="gcp-compute-dev"), role_key="roles/billing.viewer", ) result = await _driver().grant_access(ctx) assert result.ok is False assert "roles/billing.viewer" in (result.error_message or "") assert not crm.projects.return_value.getIamPolicy.called assert not crm.projects.return_value.setIamPolicy.called @pytest.mark.asyncio async def test_grant_access_refuses_org_level_role_without_api_call() -> None: # Org/folder-level predefined roles also reject on a project IAM # policy (same 400 class as billing.*). The guard now uses the full # ungrantable set, so they fail fast with a clean message, never the # raw Google 400. crm = MagicMock() with _with_svc({("cloudresourcemanager", "v3"): crm}): ctx = CloudAccessContext( principal_email="alex.quan@pingcap.com", principal_display_name="Alex Quan", scope=CloudScopeRef(scope_type="gcp_project", external_id="proj-a"), role_key="roles/resourcemanager.organizationAdmin", ) result = await _driver().grant_access(ctx) assert result.ok is False assert "roles/resourcemanager.organizationAdmin" in (result.error_message or "") assert not crm.projects.return_value.getIamPolicy.called assert not crm.projects.return_value.setIamPolicy.called @pytest.mark.asyncio async def test_grant_access_adds_binding() -> None: crm = MagicMock() get_call = MagicMock() get_call.execute.return_value = {"bindings": [], "etag": "abc"} set_call = MagicMock() set_call.execute.return_value = {} crm.projects.return_value.getIamPolicy.return_value = get_call crm.projects.return_value.setIamPolicy.return_value = set_call with _with_svc({("cloudresourcemanager", "v3"): crm}): ctx = CloudAccessContext( principal_email="alice@pingcap.com", principal_display_name="Alice", scope=CloudScopeRef(scope_type="gcp_project", external_id="proj-a"), role_key="roles/viewer", ) result = await _driver().grant_access(ctx) assert result.ok assert not result.already_exists # Verify the payload we sent includes the new binding sent = crm.projects.return_value.setIamPolicy.call_args.kwargs["body"] members = sent["policy"]["bindings"][0]["members"] assert members == ["user:alice@pingcap.com"] @pytest.mark.asyncio async def test_grant_access_idempotent_when_binding_exists() -> None: crm = MagicMock() crm.projects.return_value.getIamPolicy.return_value.execute.return_value = { "bindings": [ {"role": "roles/viewer", "members": ["user:alice@pingcap.com"]}, ], "etag": "abc", } # No setIamPolicy call expected with _with_svc({("cloudresourcemanager", "v3"): crm}): ctx = CloudAccessContext( principal_email="alice@pingcap.com", principal_display_name="Alice", scope=CloudScopeRef(scope_type="gcp_project", external_id="proj-a"), role_key="roles/viewer", ) result = await _driver().grant_access(ctx) assert result.ok assert result.already_exists assert not crm.projects.return_value.setIamPolicy.called @pytest.mark.asyncio async def test_grant_access_retries_on_etag_conflict() -> None: from googleapiclient.errors import HttpError crm = MagicMock() # Each getIamPolicy call must return a FRESH dict — the driver mutates # policy["bindings"] in-place between retries. get1 = MagicMock() get1.execute.return_value = {"bindings": [], "etag": "1"} get2 = MagicMock() get2.execute.return_value = {"bindings": [], "etag": "2"} crm.projects.return_value.getIamPolicy.side_effect = [get1, get2] err = HttpError( resp=MagicMock(status=409), content=b'{"error":"aborted"}', ) set_fail = MagicMock() set_fail.execute.side_effect = err set_ok = MagicMock() set_ok.execute.return_value = {} crm.projects.return_value.setIamPolicy.side_effect = [set_fail, set_ok] with _with_svc({("cloudresourcemanager", "v3"): crm}): ctx = CloudAccessContext( principal_email="alice@pingcap.com", principal_display_name="Alice", scope=CloudScopeRef(scope_type="gcp_project", external_id="proj-a"), role_key="roles/viewer", ) result = await _driver().grant_access(ctx) assert result.ok assert result.detail["attempts"] == 2 @pytest.mark.asyncio async def test_grant_access_invalid_principal_raises_not_found() -> None: from googleapiclient.errors import HttpError crm = MagicMock() crm.projects.return_value.getIamPolicy.return_value.execute.return_value = { "bindings": [], "etag": "abc", } err = HttpError( resp=MagicMock(status=400), content=b'{"error":"User contractor@x.com does not exist"}', ) set_fail = MagicMock() set_fail.execute.side_effect = err crm.projects.return_value.setIamPolicy.return_value = set_fail with _with_svc({("cloudresourcemanager", "v3"): crm}): ctx = CloudAccessContext( principal_email="contractor@x.com", principal_display_name="X", scope=CloudScopeRef(scope_type="gcp_project", external_id="proj-a"), role_key="roles/viewer", ) with pytest.raises(CloudPrincipalNotFoundError): await _driver().grant_access(ctx) def _deleted_account_http_error(member: str): """Prod-shaped setIamPolicy rejection for a DELETED Google account (cloudresourcemanager v3, observed 2026-07 on gcp-storage-datarepl-dev): generic top-level message — neither "does not exist" nor "not a valid" appears — with the machine-readable signal only in ``error.details``. """ from googleapiclient.errors import HttpError return HttpError( resp=MagicMock(status=400), content=json.dumps({ "error": { "code": 400, "message": "Request contains an invalid argument.", "status": "INVALID_ARGUMENT", "details": [{ "@type": ( "type.googleapis.com/" "google.cloudresourcemanager.v1.ProjectIamPolicyError" ), "type": "SOLO_REQUIRE_TOS_ACCEPTER", "role": "roles/owner", "member": member, }], }, }).encode(), ) @pytest.mark.asyncio async def test_grant_access_deleted_account_policy_error_raises_not_found() -> None: """Regression (prod 2026-07): a deleted Google account fails setIamPolicy with 400 "Request contains an invalid argument." + a ProjectIamPolicyError detail naming the member. The old substring classification missed it → generic ok=False → grant stranded PENDING → backfill cron retried the doomed IAM call every 10 min (31+ failure audits). Must classify as CloudPrincipalNotFoundError.""" crm = MagicMock() crm.projects.return_value.getIamPolicy.return_value.execute.return_value = { "bindings": [], "etag": "abc", } set_fail = MagicMock() set_fail.execute.side_effect = _deleted_account_http_error( "user:weiwan@pingcap.com", ) crm.projects.return_value.setIamPolicy.return_value = set_fail with _with_svc({("cloudresourcemanager", "v3"): crm}): ctx = CloudAccessContext( principal_email="weiwan@pingcap.com", principal_display_name="Wei Wan", scope=CloudScopeRef( scope_type="gcp_project", external_id="gcp-storage-datarepl-dev", ), role_key="roles/owner", ) with pytest.raises(CloudPrincipalNotFoundError): await _driver().grant_access(ctx) @pytest.mark.asyncio async def test_grant_access_policy_error_naming_other_member_stays_generic() -> None: """The member match is load-bearing: setIamPolicy validates the WHOLE policy, so a ProjectIamPolicyError naming some *other* stale member means the policy has a pre-existing problem — our principal may be fine. Must stay a generic ok=False failure (retryable), NOT a principal-not-found raise.""" crm = MagicMock() crm.projects.return_value.getIamPolicy.return_value.execute.return_value = { "bindings": [], "etag": "abc", } set_fail = MagicMock() set_fail.execute.side_effect = _deleted_account_http_error( "user:someone-else@pingcap.com", ) crm.projects.return_value.setIamPolicy.return_value = set_fail with _with_svc({("cloudresourcemanager", "v3"): crm}): ctx = CloudAccessContext( principal_email="alice@pingcap.com", principal_display_name="Alice", scope=CloudScopeRef(scope_type="gcp_project", external_id="proj-a"), role_key="roles/viewer", ) result = await _driver().grant_access(ctx) assert not result.ok assert "invalid argument" in (result.error_message or "").lower() # ── revoke_access ─────────────────────────────────────────────── @pytest.mark.asyncio async def test_revoke_access_removes_member() -> None: crm = MagicMock() crm.projects.return_value.getIamPolicy.return_value.execute.return_value = { "bindings": [ {"role": "roles/viewer", "members": [ "user:alice@pingcap.com", "user:bob@pingcap.com", ]}, ], "etag": "abc", } crm.projects.return_value.setIamPolicy.return_value.execute.return_value = {} with _with_svc({("cloudresourcemanager", "v3"): crm}): ctx = CloudAccessContext( principal_email="alice@pingcap.com", principal_display_name="Alice", scope=CloudScopeRef(scope_type="gcp_project", external_id="proj-a"), role_key="roles/viewer", ) result = await _driver().revoke_access(ctx) assert result.ok # Verify only bob remains sent = crm.projects.return_value.setIamPolicy.call_args.kwargs["body"] assert sent["policy"]["bindings"][0]["members"] == ["user:bob@pingcap.com"] @pytest.mark.asyncio async def test_revoke_access_idempotent_when_already_absent() -> None: crm = MagicMock() crm.projects.return_value.getIamPolicy.return_value.execute.return_value = { "bindings": [ {"role": "roles/viewer", "members": ["user:bob@pingcap.com"]}, ], "etag": "abc", } with _with_svc({("cloudresourcemanager", "v3"): crm}): ctx = CloudAccessContext( principal_email="alice@pingcap.com", principal_display_name="Alice", scope=CloudScopeRef(scope_type="gcp_project", external_id="proj-a"), role_key="roles/viewer", ) result = await _driver().revoke_access(ctx) assert result.ok assert result.already_absent assert not crm.projects.return_value.setIamPolicy.called # ── revoke_account (sweep across projects) ────────────────────── @pytest.mark.asyncio async def test_revoke_account_sweeps_projects_and_reports() -> None: crm = MagicMock() billing = MagicMock() billing.billingAccounts.return_value.list.return_value.execute.return_value = { "billingAccounts": [], } # list_scopes returns 2 projects crm.projects.return_value.search.return_value.execute.return_value = { "projects": [ {"projectId": "proj-a"}, {"projectId": "proj-b"}, ], } # getIamPolicy called twice; each returns different bindings proj_a_policy = { "bindings": [ {"role": "roles/viewer", "members": [ "user:alice@pingcap.com", "user:bob@pingcap.com", ]}, ], "etag": "a", } proj_b_policy = { "bindings": [ {"role": "roles/editor", "members": ["user:bob@pingcap.com"]}, ], "etag": "b", } get_a = MagicMock() get_a.execute.return_value = proj_a_policy get_b = MagicMock() get_b.execute.return_value = proj_b_policy crm.projects.return_value.getIamPolicy.side_effect = [get_a, get_b] crm.projects.return_value.setIamPolicy.return_value.execute.return_value = {} with _with_svc({ ("cloudresourcemanager", "v3"): crm, ("cloudbilling", "v1"): billing, }): ctx = CloudAccountRevokeContext( provider="gcp", account_id="alice@pingcap.com", ) result = await _driver().revoke_account(ctx) assert result.ok # Only proj-a should be modified (proj-b has no alice binding) assert result.raw_response["projects_modified"] == ["proj-a"] assert result.raw_response["projects_scanned"] == 2 assert result.raw_response["projects_failed"] == [] @pytest.mark.asyncio async def test_revoke_account_rejects_non_email() -> None: result = await _driver().revoke_account(CloudAccountRevokeContext( provider="gcp", account_id="not-an-email", )) assert not result.ok assert "email" in (result.error_message or "").lower() # ── get_last_login ────────────────────────────────────────────── @pytest.mark.asyncio async def test_get_last_login_returns_none_without_impersonation() -> None: # No admin_impersonation_email configured → Admin SDK unavailable. ctx = CloudAccessContext( principal_email="alice@pingcap.com", principal_display_name="Alice", scope=CloudScopeRef(scope_type="gcp_project", external_id="proj-a"), role_key="roles/viewer", ) result = await _driver().get_last_login(ctx) assert result is None @pytest.mark.asyncio async def test_get_last_login_parses_workspace_iso() -> None: admin = MagicMock() admin.users.return_value.get.return_value.execute.return_value = { "lastLoginTime": "2026-03-09T12:34:56.000Z", } with _with_svc({("admin", "directory_v1"): admin}): ctx = CloudAccessContext( principal_email="alice@pingcap.com", principal_display_name="Alice", scope=CloudScopeRef(scope_type="gcp_project", external_id="proj-a"), role_key="roles/viewer", ) result = await _driver(with_impersonation=True).get_last_login(ctx) assert result == datetime(2026, 3, 9, 12, 34, 56, tzinfo=UTC) @pytest.mark.asyncio async def test_get_last_login_epoch_treated_as_never() -> None: admin = MagicMock() admin.users.return_value.get.return_value.execute.return_value = { "lastLoginTime": "1970-01-01T00:00:00.000Z", } with _with_svc({("admin", "directory_v1"): admin}): ctx = CloudAccessContext( principal_email="neverused@pingcap.com", principal_display_name="X", scope=CloudScopeRef(scope_type="gcp_project", external_id="proj-a"), role_key="roles/viewer", ) result = await _driver(with_impersonation=True).get_last_login(ctx) assert result is None @pytest.mark.asyncio async def test_grant_access_validates_principal_when_impersonation_configured() -> None: from googleapiclient.errors import HttpError admin = MagicMock() admin.users.return_value.get.return_value.execute.side_effect = HttpError( resp=MagicMock(status=404), content=b'{"error":"Resource Not Found"}', ) # getIamPolicy should NEVER be reached — we fail fast in validation. crm = MagicMock() with _with_svc({ ("admin", "directory_v1"): admin, ("cloudresourcemanager", "v3"): crm, }): ctx = CloudAccessContext( principal_email="contractor@x.com", principal_display_name="X", scope=CloudScopeRef(scope_type="gcp_project", external_id="proj-a"), role_key="roles/viewer", ) with pytest.raises(CloudPrincipalNotFoundError): await _driver(with_impersonation=True).grant_access(ctx) assert not crm.projects.return_value.getIamPolicy.called # ── list_domains / list_load_balancers / list_cdn_domains ─────── @pytest.mark.asyncio async def test_domain_surfaces_empty_to_honor_gcp_semantics() -> None: # GCP has no "registered domains" concept comparable to Route 53 Registered # Domains. Driver returns empty without hitting any API. d = _driver() assert await d.list_domains() == [] assert await d.list_load_balancers() == [] assert await d.list_cdn_domains() == [] # ── billing surfaces config errors loudly ──────────────────────── @pytest.mark.asyncio async def test_billing_daily_no_bq_table_raises() -> None: """Direct callers (tests / debug paths) that bypass ``billing_daily_precheck`` and call ``list_billing_daily`` on a misconfigured integration get a RuntimeError instead of an empty list. The lifecycle service uses ``billing_daily_precheck`` to short-circuit cleanly; everything else should fail loud rather than silently emit zero rows that look like "no spend today".""" from datetime import date as _date with pytest.raises(RuntimeError, match="billing_bq_table is not configured"): await _driver().list_billing_daily( start_date=_date(2026, 4, 1), end_date=_date(2026, 4, 7), ) @pytest.mark.asyncio async def test_list_billing_daily_unions_across_comma_separated_tables() -> None: """The line-item ``list_cost_facts`` path was taught to split a comma-separated ``billing_bq_table`` across multiple billing accounts in PR #232. The daily-billing path used the same config field but kept interpolating the raw string, so the same multi- billing-account orgs crashed daily-billing with ``400 Invalid table name`` even though their cost-facts sync passed. Pin the parity fix: the single ``billing_bq_table`` config can be comma-separated, the driver splits + shape-validates, and the SQL ``UNION ALL``s with an outer aggregator to merge any project-day overlap from a billing-account migration. """ from datetime import date as _date d = GcpCloudDriver(integration_id="t", config={ "organization_id": _ORG_ID, "service_account_json": _SA_JSON, "billing_bq_table": ( "proj-a.bill.gcp_billing_export_v1_AAA," "proj-b.bill.gcp_billing_export_v1_BBB" ), }) captured: dict[str, Any] = {} class _FakeQueryJob: def result(self) -> list[Any]: return [] class _FakeBQClient: def __init__(self, *_a: Any, **_k: Any) -> None: pass def query(self, sql: str, *, job_config: Any = None) -> _FakeQueryJob: captured.setdefault("sql", sql) return _FakeQueryJob() fake_bq = MagicMock() fake_bq.Client = _FakeBQClient fake_bq.QueryJobConfig = lambda **kw: kw fake_bq.ScalarQueryParameter = lambda *a, **kw: ("p", a, kw) with patch.dict( "sys.modules", {"google.cloud.bigquery": fake_bq, "google.cloud": MagicMock(bigquery=fake_bq)}, ), patch.object( GcpCloudDriver, "_credentials", lambda self: None, ): out = await d.list_billing_daily( start_date=_date(2026, 5, 1), end_date=_date(2026, 5, 7), ) assert out == [] sql = captured["sql"] assert "`proj-a.bill.gcp_billing_export_v1_AAA`" in sql assert "`proj-b.bill.gcp_billing_export_v1_BBB`" in sql assert sql.count("UNION ALL") == 1 # Outer aggregator merges duplicates from billing-account migration assert "SUM(total_cost)" in sql # Regression guard for the original concatenation bug. assert ",gcp_billing_export" not in sql @pytest.mark.asyncio async def test_list_billing_daily_emits_per_currency_rows() -> None: """Multi-billing-account GCP orgs can have one account in USD and another in CNY (or EUR / JPY). Pre-Round-3 the SQL did ``ANY_VALUE(currency)`` and the Python aggregator keyed on ``(day, project_id)``, so two rows with different currencies for the same (day, project) collapsed into one bucket — ``SUM`` blended raw amounts across currencies and ``ANY_VALUE`` arbitrarily picked one tag, silently producing mixed-unit totals. Post-fix the SQL groups by ``currency`` and the Python loop keys on ``(day, project_id, currency)``, so the driver emits one ``CloudDailyCost`` per (day × project × currency) and the downstream consumer attributes USD per row. """ from datetime import date as _date d = GcpCloudDriver(integration_id="t", config={ "organization_id": _ORG_ID, "service_account_json": _SA_JSON, "billing_bq_table": "proj.bill.gcp_billing_export_v1_AAA", }) class _FakeRow(dict): # BQ Row supports both ``r["col"]`` and ``r.get("col")``. pass fake_rows = [ _FakeRow({ "day": _date(2026, 5, 5), "project_id": "proj-x", "service_name": "Compute Engine", "total_cost": 100.0, "currency": "USD", }), _FakeRow({ "day": _date(2026, 5, 5), "project_id": "proj-x", "service_name": "Compute Engine", "total_cost": 700.0, "currency": "CNY", }), _FakeRow({ "day": _date(2026, 5, 5), "project_id": "proj-x", "service_name": "Cloud Storage", "total_cost": 50.0, "currency": "USD", }), ] class _FakeQueryJob: def result(self) -> list[Any]: return fake_rows class _FakeBQClient: def __init__(self, *_a: Any, **_k: Any) -> None: pass def query(self, sql: str, *, job_config: Any = None) -> _FakeQueryJob: return _FakeQueryJob() fake_bq = MagicMock() fake_bq.Client = _FakeBQClient fake_bq.QueryJobConfig = lambda **kw: kw fake_bq.ScalarQueryParameter = lambda *a, **kw: ("p", a, kw) with patch.dict( "sys.modules", {"google.cloud.bigquery": fake_bq, "google.cloud": MagicMock(bigquery=fake_bq)}, ), patch.object( GcpCloudDriver, "_credentials", lambda self: None, ): out = await d.list_billing_daily( start_date=_date(2026, 5, 5), end_date=_date(2026, 5, 5), ) # Two rows: one USD bucket (100 + 50 = 150), one CNY bucket (700). by_currency = {r.currency: r for r in out} assert set(by_currency.keys()) == {"USD", "CNY"} assert by_currency["USD"].total_cost_usd == pytest.approx(150.0) assert by_currency["CNY"].total_cost_usd == pytest.approx(700.0) # Per-service breakdown must split per currency. assert by_currency["USD"].cost_by_service == { "Compute Engine": pytest.approx(100.0), "Cloud Storage": pytest.approx(50.0), } assert by_currency["CNY"].cost_by_service == { "Compute Engine": pytest.approx(700.0), } @pytest.mark.asyncio async def test_list_billing_daily_rejects_sql_injection() -> None: """Same shape validator the cost-facts path uses must apply to the daily-billing config too — the regex refuses backticks / semicolons / SQL comments before any BigQuery call.""" from datetime import date as _date d = GcpCloudDriver(integration_id="t", config={ "organization_id": _ORG_ID, "service_account_json": _SA_JSON, "billing_bq_table": "proj.ds.t`; DROP TABLE secrets; --", }) with pytest.raises(RuntimeError, match="does not match"): await d.list_billing_daily( start_date=_date(2026, 5, 1), end_date=_date(2026, 5, 7), ) @pytest.mark.asyncio async def test_billing_daily_precheck_flags_missing_bq_table() -> None: reason = await _driver().billing_daily_precheck() assert reason is not None assert "billing_bq_table" in reason @pytest.mark.asyncio async def test_billing_daily_precheck_passes_with_bq_table() -> None: d = GcpCloudDriver(integration_id="t", config={ "organization_id": _ORG_ID, "service_account_json": _SA_JSON, "billing_bq_table": "proj.dataset.gcp_billing_export_v1_XXX", }) reason = await d.billing_daily_precheck() # Either None (bigquery installed) or complains about the missing # optional bigquery dep; never the table-missing reason. if reason is not None: assert "billing_bq_table" not in reason assert "bigquery" in reason.lower() @pytest.mark.asyncio async def test_list_cost_facts_unions_across_comma_separated_tables() -> None: """Multi-billing-account orgs configure ``billing_bq_table`` as a comma-separated list of fully-qualified table ids. Pre-fix, the driver interpolated the raw string (commas and all) into ``FROM `{table}` ``, so BigQuery returned ``400 Invalid dataset ID ""``. Post-fix, the driver splits + shape-validates the list, then builds one ``SELECT ... GROUP BY`` per table joined by ``UNION ALL`` with an outer ``GROUP BY`` to merge any project-overlap from a billing-account migration. Pin the SQL shape via a mocked BigQuery client so a future refactor can't regress to the single-string interpolation. """ from datetime import date as _date d = GcpCloudDriver(integration_id="t", config={ "organization_id": _ORG_ID, "service_account_json": _SA_JSON, "billing_bq_table": ( "proj-a.bill.gcp_billing_export_v1_AAA," "proj-b.bill.gcp_billing_export_v1_BBB," "proj-c.bill.gcp_billing_export_v1_CCC" ), }) captured: dict[str, Any] = {} class _FakeQueryJob: def result(self) -> list[Any]: return [] class _FakeBQClient: def __init__(self, *_a: Any, **_k: Any) -> None: pass def query(self, sql: str, *, job_config: Any = None) -> _FakeQueryJob: captured.setdefault("sql", sql) return _FakeQueryJob() fake_bq = MagicMock() fake_bq.Client = _FakeBQClient fake_bq.QueryJobConfig = lambda **kw: kw fake_bq.ScalarQueryParameter = lambda *a, **kw: ("p", a, kw) with patch.dict( "sys.modules", {"google.cloud.bigquery": fake_bq, "google.cloud": MagicMock(bigquery=fake_bq)}, ), patch.object( GcpCloudDriver, "_credentials", lambda self: None, ): out = await d.list_cost_facts( start_date=_date(2026, 5, 1), end_date=_date(2026, 5, 7), ) assert out == [] sql = captured["sql"] # Each configured table must appear in the FROM clause. assert "`proj-a.bill.gcp_billing_export_v1_AAA`" in sql assert "`proj-b.bill.gcp_billing_export_v1_BBB`" in sql assert "`proj-c.bill.gcp_billing_export_v1_CCC`" in sql # 3 tables → 2 UNION ALL joiners. assert sql.count("UNION ALL") == 2 # No raw comma sneaking through into the FROM clause (regression # guard for the original bug). assert ",gcp_billing_export" not in sql assert "AAA,proj-b" not in sql # Outer aggregator wraps the per-table selects so duplicate # (project, day, resource) rows during a billing-account # migration get merged. assert "SUM(native_cost)" in sql assert "SUM(usd_cost)" in sql @pytest.mark.asyncio async def test_list_cost_facts_handles_standard_export_without_resource_struct() -> None: """GCP Cloud Billing exports come in two flavours: * **Standard usage cost** — ``gcp_billing_export_v1_``, no ``resource`` STRUCT. * **Detailed usage cost** — ``gcp_billing_export_resource_v1_``, adds ``resource`` STRUCT. Pre-fix the driver unconditionally referenced ``resource.name`` / ``resource.global_name``, so every per-scope cost-facts sync against a tenant on the standard export hard-failed with ``400 Unrecognized name: resource at [21:22]`` (observed 2026-05-10 across ``gcp-qa-perf-testing-dev`` / ``gcp-serverless-engine-dev`` / ``gcp-pm-preview``). Post-fix the driver detects the table flavour and substitutes empty-string literals on standard exports — the sync still returns (day × project × service × sku × region) rows, just without the per-resource breakdown the standard export doesn't carry. Pin the substitution via SQL inspection so a future refactor can't regress to the unconditional ``resource.name`` reference. """ from datetime import date as _date d = GcpCloudDriver(integration_id="t", config={ "organization_id": _ORG_ID, "service_account_json": _SA_JSON, "billing_bq_table": "proj.bill.gcp_billing_export_v1_AAA", }) captured: dict[str, Any] = {} class _FakeQueryJob: def result(self) -> list[Any]: return [] class _FakeBQClient: def __init__(self, *_a: Any, **_k: Any) -> None: pass def query(self, sql: str, *, job_config: Any = None) -> _FakeQueryJob: captured.setdefault("sql", sql) return _FakeQueryJob() fake_bq = MagicMock() fake_bq.Client = _FakeBQClient fake_bq.QueryJobConfig = lambda **kw: kw fake_bq.ScalarQueryParameter = lambda *a, **kw: ("p", a, kw) with patch.dict( "sys.modules", {"google.cloud.bigquery": fake_bq, "google.cloud": MagicMock(bigquery=fake_bq)}, ), patch.object( GcpCloudDriver, "_credentials", lambda self: None, ): out = await d.list_cost_facts( start_date=_date(2026, 5, 1), end_date=_date(2026, 5, 7), ) assert out == [] sql = captured["sql"] # Standard export must NOT reference resource.name / resource.global_name. assert "resource.name" not in sql assert "resource.global_name" not in sql # Empty-string literal substitution preserves the column shape so # the outer (potential) UNION ALL aggregator still type-aligns. assert "'' AS resource_name" in sql assert "'' AS resource_global" in sql @pytest.mark.asyncio async def test_list_cost_facts_mixed_standard_and_detailed_exports() -> None: """An org may have one billing account on the standard export and another on the detailed export (e.g. mid-migration, or different cost teams configured them at different times). Each per-table SELECT must use the right ``resource`` expression independently so the UNION ALL stays type-aligned and BigQuery doesn't fail the whole query because of one mis-shaped sub-select. """ from datetime import date as _date d = GcpCloudDriver(integration_id="t", config={ "organization_id": _ORG_ID, "service_account_json": _SA_JSON, "billing_bq_table": ( "proj-a.bill.gcp_billing_export_v1_AAA," # standard "proj-b.bill.gcp_billing_export_resource_v1_BBB" # detailed ), }) captured: dict[str, Any] = {} class _FakeQueryJob: def result(self) -> list[Any]: return [] class _FakeBQClient: def __init__(self, *_a: Any, **_k: Any) -> None: pass def query(self, sql: str, *, job_config: Any = None) -> _FakeQueryJob: captured.setdefault("sql", sql) return _FakeQueryJob() fake_bq = MagicMock() fake_bq.Client = _FakeBQClient fake_bq.QueryJobConfig = lambda **kw: kw fake_bq.ScalarQueryParameter = lambda *a, **kw: ("p", a, kw) with patch.dict( "sys.modules", {"google.cloud.bigquery": fake_bq, "google.cloud": MagicMock(bigquery=fake_bq)}, ), patch.object( GcpCloudDriver, "_credentials", lambda self: None, ): await d.list_cost_facts( start_date=_date(2026, 5, 1), end_date=_date(2026, 5, 7), ) sql = captured["sql"] # Both tables present. assert "`proj-a.bill.gcp_billing_export_v1_AAA`" in sql assert "`proj-b.bill.gcp_billing_export_resource_v1_BBB`" in sql # Detailed export keeps the IFNULL(resource.*) references. assert "IFNULL(resource.name, '')" in sql assert "IFNULL(resource.global_name, '')" in sql # Standard export sub-select substitutes empty literals. assert "'' AS resource_name" in sql assert "'' AS resource_global" in sql # Single UNION ALL between the two per-table selects. assert sql.count("UNION ALL") == 1 @pytest.mark.asyncio async def test_list_cost_facts_rejects_sql_injection_via_billing_bq_table() -> None: """``billing_bq_table`` is admin-controlled config. Privileged- admin compromise (or typo) must not be able to inject backticks / semicolons / SQL comments into the BigQuery query template. The shape validator refuses anything outside ``project.dataset.table`` with a tight char class. """ from datetime import date as _date d = GcpCloudDriver(integration_id="t", config={ "organization_id": _ORG_ID, "service_account_json": _SA_JSON, "billing_bq_table": "proj.ds.t`; DROP TABLE secrets; --", }) with pytest.raises(RuntimeError, match="does not match"): await d.list_cost_facts( start_date=_date(2026, 5, 1), end_date=_date(2026, 5, 7), ) @pytest.mark.asyncio async def test_list_cost_facts_computes_usd_via_safe_divide_not_cost_in_usd_column() -> None: """GCP's BigQuery billing export does NOT carry a ``cost_in_usd`` column — that's AWS CUR. An earlier driver version (PR #226) referenced ``SUM(cost_in_usd) AS usd_cost`` as if parity held; on 2026-05-10 every cost-facts sync crashed with ``400 Unrecognized name: cost_in_usd at [25:19]`` across ``gcp-qa-perf-testing-dev`` / ``gcp-serverless-engine-dev`` / ``gcp-pm-preview``. Post-fix the driver computes USD from the real schema columns: ``SAFE_DIVIDE(cost, currency_conversion_rate)`` (the rate converts USD → local at billing time, so USD = local / rate; equals 1.0 for USD-billed accounts). Pin the SQL shape so a future refactor can't regress to the non-existent column. """ from datetime import date as _date d = GcpCloudDriver(integration_id="t", config={ "organization_id": _ORG_ID, "service_account_json": _SA_JSON, "billing_bq_table": "proj.bill.gcp_billing_export_v1_AAA", }) captured: dict[str, Any] = {} class _FakeQueryJob: def result(self) -> list[Any]: return [] class _FakeBQClient: def __init__(self, *_a: Any, **_k: Any) -> None: pass def query(self, sql: str, *, job_config: Any = None) -> _FakeQueryJob: captured.setdefault("sql", sql) return _FakeQueryJob() fake_bq = MagicMock() fake_bq.Client = _FakeBQClient fake_bq.QueryJobConfig = lambda **kw: kw fake_bq.ScalarQueryParameter = lambda *a, **kw: ("p", a, kw) with patch.dict( "sys.modules", {"google.cloud.bigquery": fake_bq, "google.cloud": MagicMock(bigquery=fake_bq)}, ), patch.object( GcpCloudDriver, "_credentials", lambda self: None, ): await d.list_cost_facts( start_date=_date(2026, 5, 1), end_date=_date(2026, 5, 7), ) sql = captured["sql"] # The non-existent AWS-style column must not appear anywhere in # the per-table SELECT. assert "cost_in_usd" not in sql # USD is computed from the real GCP export columns via SAFE_DIVIDE # so a 0 / NULL rate row contributes NULL (skipped by SUM) instead # of crashing the whole batch. assert ( "SUM(SAFE_DIVIDE(cost, currency_conversion_rate)) AS usd_cost" in sql ) @pytest.mark.asyncio async def test_list_cost_facts_groups_by_currency_and_usage_unit() -> None: """Multi-billing-account orgs can have one billing account in USD and another in CNY (e.g. mid-migration). Pre-Round-2 fix the SQL used ``ANY_VALUE(currency)`` and didn't include ``currency`` in the GROUP BY — so the outer aggregator collapsed rows from the two currencies under one arbitrary tag, with ``SUM(usd_cost)`` blending native amounts across currencies. The same issue affected ``ANY_VALUE(usage.unit)`` blending byte-seconds with bytes within a SKU group. Both must be GROUP BY keys, not ANY_VALUE picks. """ from datetime import date as _date d = GcpCloudDriver(integration_id="t", config={ "organization_id": _ORG_ID, "service_account_json": _SA_JSON, "billing_bq_table": ( "proj-a.bill.gcp_billing_export_v1_AAA," "proj-b.bill.gcp_billing_export_v1_BBB" ), }) captured: dict[str, Any] = {} class _FakeQueryJob: def result(self) -> list[Any]: return [] class _FakeBQClient: def __init__(self, *_a: Any, **_k: Any) -> None: pass def query(self, sql: str, *, job_config: Any = None) -> _FakeQueryJob: captured.setdefault("sql", sql) return _FakeQueryJob() def close(self) -> None: pass fake_bq = MagicMock() fake_bq.Client = _FakeBQClient fake_bq.QueryJobConfig = lambda **kw: kw fake_bq.ScalarQueryParameter = lambda *a, **kw: ("p", a, kw) with patch.dict( "sys.modules", {"google.cloud.bigquery": fake_bq, "google.cloud": MagicMock(bigquery=fake_bq)}, ), patch.object( GcpCloudDriver, "_credentials", lambda self: None, ): await d.list_cost_facts( start_date=_date(2026, 5, 1), end_date=_date(2026, 5, 7), ) sql = captured["sql"] # No more ANY_VALUE on currency / usage_unit — they're GROUP BY # keys now. assert "ANY_VALUE(currency)" not in sql assert "ANY_VALUE(usage_unit)" not in sql assert "ANY_VALUE(usage.unit)" not in sql # Both inner per-table SELECT GROUP BY clause and outer aggregator # GROUP BY clause must include currency + usage_unit to keep each # (day × resource × sku × currency × unit) atomic. assert sql.count("currency, usage_unit") >= 2 # inner + outer # IFNULL for inner SELECT (NULL is OK for empty unit/currency rows). assert "IFNULL(currency, '') AS currency" in sql assert "IFNULL(usage.unit, '') AS usage_unit" in sql @pytest.mark.asyncio async def test_list_cost_facts_uses_sku_id_for_natural_key_uniqueness() -> None: """``compute_fact_key`` keys on ``usage_type``. Pre-Round-2 fix the driver mapped ``usage_type=sku.description`` — but SKU descriptions are NOT unique cross-region. Two regional SKUs sharing one description (e.g. "N1 Predefined Instance Core") collapsed to one fact_key on standard exports (where resource_id is NULL), and the upsert silently overwrote, undercounting cross-region cost. Post-fix uses ``f"{sku_id}@{region}"`` as ``usage_type`` so the natural key disambiguates correctly. Pin via constructed rows. """ from datetime import date as _date d = GcpCloudDriver(integration_id="t", config={ "organization_id": _ORG_ID, "service_account_json": _SA_JSON, "billing_bq_table": "proj.bill.gcp_billing_export_v1_AAA", }) class _FakeRow(dict): def __getitem__(self, k: str) -> Any: return dict.__getitem__(self, k) def get(self, k: str, default: Any = None) -> Any: return dict.get(self, k, default) base_day = _date(2026, 5, 1) # Two rows differing ONLY in region — same sku_description. fake_rows = [ _FakeRow({ "day": base_day, "project_id": "proj-1", "project_name": "Proj 1", "service_name": "Compute Engine", "sku_description": "N1 Predefined Instance Core running Americas", "sku_id": "0001-AAAA-1111", "resource_name": "", "resource_global": "", "region": "us-east1", "currency": "USD", "usage_unit": "seconds", "native_cost": 12.5, "usd_cost": 12.5, "usage_amount": 3600, "labels_agg": [], }), _FakeRow({ "day": base_day, "project_id": "proj-1", "project_name": "Proj 1", "service_name": "Compute Engine", "sku_description": "N1 Predefined Instance Core running Americas", "sku_id": "0002-BBBB-2222", "resource_name": "", "resource_global": "", "region": "us-west1", "currency": "USD", "usage_unit": "seconds", "native_cost": 8.75, "usd_cost": 8.75, "usage_amount": 1800, "labels_agg": [], }), ] class _FakeQueryJob: def result(self) -> list[Any]: return fake_rows class _FakeBQClient: def __init__(self, *_a: Any, **_k: Any) -> None: pass def query(self, sql: str, *, job_config: Any = None) -> _FakeQueryJob: return _FakeQueryJob() def close(self) -> None: pass fake_bq = MagicMock() fake_bq.Client = _FakeBQClient fake_bq.QueryJobConfig = lambda **kw: kw fake_bq.ScalarQueryParameter = lambda *a, **kw: ("p", a, kw) with patch.dict( "sys.modules", {"google.cloud.bigquery": fake_bq, "google.cloud": MagicMock(bigquery=fake_bq)}, ), patch.object( GcpCloudDriver, "_credentials", lambda self: None, ): out = await d.list_cost_facts( start_date=_date(2026, 5, 1), end_date=_date(2026, 5, 7), ) assert len(out) == 2 # Each row's usage_type must include the sku_id so they have # distinct natural keys despite identical sku_description. assert out[0].usage_type != out[1].usage_type assert "0001-AAAA-1111" in (out[0].usage_type or "") assert "0002-BBBB-2222" in (out[1].usage_type or "") # Both rows still expose the human-readable description on # resource_type / raw for dashboard / CSV / drill-down. for fact in out: assert fact.resource_type == ( "N1 Predefined Instance Core running Americas" ) assert fact.raw["sku_description"] == ( "N1 Predefined Instance Core running Americas" ) @pytest.mark.asyncio async def test_list_cost_facts_drops_or_usd_default_on_null_currency() -> None: """Pre-Round-2 fix the row builder did ``currency = str(r.get("currency") or "USD")`` — silently labelling a NULL-currency row as USD. If the row's ``cost`` was actually CNY, the row's ``amount_usd`` would be set to ``native`` (USD identity), inflating USD totals. Post-fix routes empty/NULL currency to ``"UNKNOWN"`` and ``amount_usd=0`` so the row still persists for native-currency analytics but doesn't fake USD totals. """ from datetime import date as _date d = GcpCloudDriver(integration_id="t", config={ "organization_id": _ORG_ID, "service_account_json": _SA_JSON, "billing_bq_table": "proj.bill.gcp_billing_export_v1_AAA", }) class _FakeRow(dict): def __getitem__(self, k: str) -> Any: return dict.__getitem__(self, k) def get(self, k: str, default: Any = None) -> Any: return dict.get(self, k, default) fake_rows = [ _FakeRow({ "day": _date(2026, 5, 1), "project_id": "proj-1", "project_name": "Proj 1", "service_name": "Storage", "sku_description": "Some SKU", "sku_id": "1234-AAAA", "resource_name": "", "resource_global": "", "region": "us-east1", "currency": "", # NULL emitted as empty string by IFNULL "usage_unit": "", "native_cost": 100.0, # SAFE_DIVIDE returns NULL when rate is NULL/0. "usd_cost": None, "usage_amount": 1, "labels_agg": [], }), ] class _FakeQueryJob: def result(self) -> list[Any]: return fake_rows class _FakeBQClient: def __init__(self, *_a: Any, **_k: Any) -> None: pass def query(self, sql: str, *, job_config: Any = None) -> _FakeQueryJob: return _FakeQueryJob() def close(self) -> None: pass fake_bq = MagicMock() fake_bq.Client = _FakeBQClient fake_bq.QueryJobConfig = lambda **kw: kw fake_bq.ScalarQueryParameter = lambda *a, **kw: ("p", a, kw) with patch.dict( "sys.modules", {"google.cloud.bigquery": fake_bq, "google.cloud": MagicMock(bigquery=fake_bq)}, ), patch.object( GcpCloudDriver, "_credentials", lambda self: None, ): out = await d.list_cost_facts( start_date=_date(2026, 5, 1), end_date=_date(2026, 5, 7), ) assert len(out) == 1 fact = out[0] # NULL currency → "UNKNOWN", NOT silently "USD". assert fact.currency == "UNKNOWN" # NULL usd_cost + non-USD currency → amount_usd stays at 0.0 so # the row doesn't fake USD totals; native preserved. assert fact.amount_usd == 0.0 assert fact.amount_native == 100.0 @pytest.mark.asyncio async def test_list_cost_facts_pins_partition_window_and_bq_client_safety() -> None: """SQL hygiene pins post-Round-2: * Partition window padded ±3d/+7d (was ±1d/+1d) so back-dated SUD/CUD credit lines that GCP back-dates 30-45 days don't get silently missed by the nightly cost-facts run. * Partition predicate is strict ``_PARTITIONTIME >= ... AND < ...`` (not ``IS NULL OR ...``) so BQ can prune partitions — the ``IS NULL OR`` variant defeated pruning and blew past the 50 GB ``maximum_bytes_billed`` cap on multi-year exports. GCP billing exports are always ingestion-time-partitioned so the strict form is safe. * BQ Client is constructed with ``project=`` from the first table's project so the SA's home project doesn't get billed for jobs issued against an export in a different project. * QueryJobConfig has ``maximum_bytes_billed`` and ``job_timeout_ms`` defensive caps. """ from datetime import date as _date d = GcpCloudDriver(integration_id="t", config={ "organization_id": _ORG_ID, "service_account_json": _SA_JSON, "billing_bq_table": "billing-host-proj.bill.gcp_billing_export_v1_AAA", }) captured: dict[str, Any] = {} class _FakeQueryJob: def result(self) -> list[Any]: return [] class _FakeBQClient: def __init__(self, *_a: Any, **_k: Any) -> None: captured["client_project_kw"] = _k.get("project") def query(self, sql: str, *, job_config: Any = None) -> _FakeQueryJob: captured.setdefault("sql", sql) captured["job_config"] = job_config return _FakeQueryJob() def close(self) -> None: captured["closed"] = True fake_bq = MagicMock() fake_bq.Client = _FakeBQClient fake_bq.QueryJobConfig = lambda **kw: kw fake_bq.ScalarQueryParameter = lambda *a, **kw: ("p", a, kw) with patch.dict( "sys.modules", {"google.cloud.bigquery": fake_bq, "google.cloud": MagicMock(bigquery=fake_bq)}, ), patch.object( GcpCloudDriver, "_credentials", lambda self: None, ): await d.list_cost_facts( start_date=_date(2026, 5, 10), end_date=_date(2026, 5, 10), ) # Client constructed with project= extracted from the first table. assert captured["client_project_kw"] == "billing-host-proj" sql = captured["sql"] # Strict partition predicate (no `IS NULL OR ...` — that would # defeat partition pruning). assert "_PARTITIONTIME IS NULL" not in sql assert "_PARTITIONTIME >= TIMESTAMP(@partition_lo)" in sql assert "_PARTITIONTIME < TIMESTAMP(@partition_hi)" in sql # JobConfig has defensive caps. job_cfg = captured["job_config"] assert job_cfg.get("maximum_bytes_billed") == 50_000_000_000 assert job_cfg.get("job_timeout_ms") == 300_000 # Client.close() called via try/finally. assert captured.get("closed") is True @pytest.mark.asyncio async def test_list_cost_facts_sql_unnests_credits_for_net_cost() -> None: """GCP's ``cost`` column is the PRE-credit list amount; SUD/CUD discounts, committed-use credits, and promotions live in the per-row ``credits`` STRUCT array as negative amounts. The cost-facts SQL must unnest + sum them so the emitted amounts are NET, not gross. Pin the SQL shape so a future refactor can't silently regress to gross. """ from datetime import date as _date d = GcpCloudDriver(integration_id="t", config={ "organization_id": _ORG_ID, "service_account_json": _SA_JSON, "billing_bq_table": "proj.bill.gcp_billing_export_v1_AAA", }) captured: dict[str, Any] = {} class _FakeQueryJob: def result(self) -> list[Any]: return [] class _FakeBQClient: def __init__(self, *_a: Any, **_k: Any) -> None: pass def query(self, sql: str, *, job_config: Any = None) -> _FakeQueryJob: captured.setdefault("sql", sql) return _FakeQueryJob() fake_bq = MagicMock() fake_bq.Client = _FakeBQClient fake_bq.QueryJobConfig = lambda **kw: kw fake_bq.ScalarQueryParameter = lambda *a, **kw: ("p", a, kw) with patch.dict( "sys.modules", {"google.cloud.bigquery": fake_bq, "google.cloud": MagicMock(bigquery=fake_bq)}, ), patch.object( GcpCloudDriver, "_credentials", lambda self: None, ): await d.list_cost_facts( start_date=_date(2026, 5, 1), end_date=_date(2026, 5, 7), ) sql = captured["sql"] # Native + USD credit aggregates unnested from the credits array. assert ( "SUM((SELECT SUM(c.amount) FROM UNNEST(credits) c)) AS credit_native" in sql ) assert "AS credit_usd" in sql assert "UNNEST(credits)" in sql @pytest.mark.asyncio async def test_list_cost_facts_nets_credits_and_preserves_gross_in_raw() -> None: """2026-07 成本口径改造后的核心契约(替代旧的「credit 折进行内」): * usage 行改存 **税前毛额**(cost 列,PRE-credit)—— $100 毛额 + −$30 credit 的组,fact 金额 = 100(不再是 70);credit 聚合列仅 留在 ``raw`` 供审计; * credits 由第二个查询按 (day × project × type) 合成 **独立负数 fact 行**(bucket 按类型映射)—— 发票口径合计不变(100 − 30 = 70); * 白名单 list 度量(cost_at_list)落在 usage 行上,credit 行恒 None。 """ from datetime import date as _date d = GcpCloudDriver(integration_id="t", config={ "organization_id": _ORG_ID, "service_account_json": _SA_JSON, "billing_bq_table": "proj.bill.gcp_billing_export_v1_AAA", }) class _FakeRow(dict): def __getitem__(self, k: str) -> Any: return dict.__getitem__(self, k) def get(self, k: str, default: Any = None) -> Any: return dict.get(self, k, default) fact_rows = [ _FakeRow({ "day": _date(2026, 5, 1), "project_id": "proj-1", "project_name": "Proj 1", "service_name": "Vertex AI", "sku_description": "Vertex AI Online Prediction", "sku_id": "0001-AAAA-1111", "resource_name": "", "resource_global": "", "region": "us-central1", "currency": "USD", "usage_unit": "count", "native_cost": 100.0, "usd_cost": 100.0, "credit_native": -30.0, "credit_usd": -30.0, "list_native": 90.0, "list_usd": 90.0, "usage_amount": 10, "labels_agg": [], }), ] credit_rows = [ _FakeRow({ "day": _date(2026, 5, 1), "project_id": "proj-1", "project_name": "Proj 1", "credit_type": "PROMOTION", "currency": "USD", "credit_native": -30.0, "credit_usd": -30.0, }), ] class _FakeQueryJob: def __init__(self, rows: list[Any]) -> None: self._rows = rows def result(self) -> list[Any]: return self._rows class _FakeBQClient: def __init__(self, *_a: Any, **_k: Any) -> None: pass def query(self, sql: str, *, job_config: Any = None) -> _FakeQueryJob: # 第二个查询 = credits 明细(按 SQL 形状分发)。 if "UNNEST(credits) AS c" in sql and "credit_type" in sql: return _FakeQueryJob(credit_rows) return _FakeQueryJob(fact_rows) def close(self) -> None: pass fake_bq = MagicMock() fake_bq.Client = _FakeBQClient fake_bq.QueryJobConfig = lambda **kw: kw fake_bq.ScalarQueryParameter = lambda *a, **kw: ("p", a, kw) with patch.dict( "sys.modules", {"google.cloud.bigquery": fake_bq, "google.cloud": MagicMock(bigquery=fake_bq)}, ), patch.object( GcpCloudDriver, "_credentials", lambda self: None, ): out = await d.list_cost_facts( start_date=_date(2026, 5, 1), end_date=_date(2026, 5, 7), ) assert len(out) == 2 usage = next(f for f in out if f.service == "Vertex AI") credit = next(f for f in out if f.service == "Credits") # usage 行 = 税前毛额(不再折 credit)。 assert usage.amount_native == 100.0 assert usage.amount_usd == 100.0 assert usage.amount_usd_gross == 100.0 assert usage.amount_native_gross == 100.0 assert usage.cost_bucket == "usage" # 白名单 list 度量(cost_at_list)+ GCP 牌价 == list。 assert usage.amount_usd_list == 90.0 assert usage.amount_native_list == 90.0 assert usage.amount_usd_public == 90.0 # credit 聚合列仅留 raw 审计。 assert usage.raw["credit_native"] == -30.0 assert usage.raw["credit_usd"] == -30.0 # 合成 credit fact:独立负数行、按类型分桶、list 恒 None。 assert credit.amount_usd == -30.0 assert credit.amount_native == -30.0 assert credit.cost_bucket == "credit" # PROMOTION → credit 桶 assert credit.amount_usd_list is None assert credit.usage_type == "PROMOTION" assert credit.resource_id == "gcp-project:proj-1:credit:PROMOTION" # 发票口径合计不变:100 − 30 = 70。 assert sum(f.amount_usd for f in out) == 70.0 @pytest.mark.asyncio async def test_list_cost_facts_gross_honesty_guard() -> None: """``amount_usd_gross`` honesty guard(2026-07 口径改造后): * 正常正数行 → gross == amount(税前毛额),如实呈现; * 负数冲正行(amount ≤ 0)→ gross 落 NULL(guard 拒绝无意义 gross); * 毛额为 0 的组(纯 credit 载体行)→ 从 usage facts 跳过——它的 金额内容由合成 credit facts 承载(本测试第二查询返回空,故不出现)。 """ from datetime import date as _date d = GcpCloudDriver(integration_id="t", config={ "organization_id": _ORG_ID, "service_account_json": _SA_JSON, "billing_bq_table": "proj.bill.gcp_billing_export_v1_AAA", }) class _FakeRow(dict): def get(self, k: str, default: Any = None) -> Any: return dict.get(self, k, default) def _row(sku: str, native: float, usd: float, cn: float, cu: float) -> Any: return _FakeRow({ "day": _date(2026, 5, 1), "project_id": "proj-1", "project_name": "Proj 1", "service_name": "Vertex AI", "sku_description": "Vertex AI", "sku_id": sku, "resource_name": "", "resource_global": "", "region": "us-central1", "currency": "USD", "usage_unit": "count", "native_cost": native, "usd_cost": usd, "credit_native": cn, "credit_usd": cu, "usage_amount": 1, "labels_agg": [], }) fake_rows = [ # 正数行 → gross == amount,如实呈现。 _row("SKU-NOCREDIT", 50.0, 50.0, 0.0, 0.0), # 带正向 credit 调整的行:金额 = 税前毛额 50(credit 不再折入)。 _row("SKU-ADJUST", 50.0, 50.0, 10.0, 10.0), # 负数冲正行 → 保留(白名单允许负冲正),gross 落 NULL。 _row("SKU-REFUND", -5.0, -5.0, 0.0, 0.0), # 毛额为 0 的纯 credit 载体行 → 从 usage facts 跳过。 _row("SKU-ZEROGROSS", 0.0, 0.0, -40.0, -40.0), ] class _FakeQueryJob: def __init__(self, rows: list[Any]) -> None: self._rows = rows def result(self) -> list[Any]: return self._rows class _FakeBQClient: def __init__(self, *_a: Any, **_k: Any) -> None: pass def query(self, sql: str, *, job_config: Any = None) -> _FakeQueryJob: if "UNNEST(credits) AS c" in sql and "credit_type" in sql: return _FakeQueryJob([]) return _FakeQueryJob(fake_rows) def close(self) -> None: pass fake_bq = MagicMock() fake_bq.Client = _FakeBQClient fake_bq.QueryJobConfig = lambda **kw: kw fake_bq.ScalarQueryParameter = lambda *a, **kw: ("p", a, kw) with patch.dict( "sys.modules", {"google.cloud.bigquery": fake_bq, "google.cloud": MagicMock(bigquery=fake_bq)}, ), patch.object( GcpCloudDriver, "_credentials", lambda self: None, ): out = await d.list_cost_facts( start_date=_date(2026, 5, 1), end_date=_date(2026, 5, 7), ) by_sku = {f.usage_type.split("@")[0]: f for f in out} # 毛额为 0 的行从 usage facts 跳过。 assert "SKU-ZEROGROSS" not in by_sku # 正数行:gross == amount(税前毛额)。 assert by_sku["SKU-NOCREDIT"].amount_usd == 50.0 assert by_sku["SKU-NOCREDIT"].amount_usd_gross == 50.0 # credit 不再折入金额:amount = 毛额 50(旧契约是 60)。 assert by_sku["SKU-ADJUST"].amount_usd == 50.0 assert by_sku["SKU-ADJUST"].amount_usd_gross == 50.0 # 负数冲正行保留,gross guard 拒绝(amount ≤ 0 → gross NULL)。 assert by_sku["SKU-REFUND"].amount_usd == -5.0 assert by_sku["SKU-REFUND"].amount_usd_gross is None assert by_sku["SKU-REFUND"].amount_native_gross is None @pytest.mark.asyncio async def test_list_billing_daily_nets_credits_into_total_cost() -> None: """The daily-billing path (spike-detection / reconciliation) shares the same gross-not-net bug. ``total_cost`` must net the unnested credit sum so a $100 day with a -$30 credit reports $70. """ from datetime import date as _date d = GcpCloudDriver(integration_id="t", config={ "organization_id": _ORG_ID, "service_account_json": _SA_JSON, "billing_bq_table": "proj.bill.gcp_billing_export_v1_AAA", }) captured: dict[str, Any] = {} class _FakeRow(dict): def get(self, k: str, default: Any = None) -> Any: return dict.get(self, k, default) fake_rows = [ _FakeRow({ "day": _date(2026, 5, 5), "project_id": "proj-x", "service_name": "Vertex AI", "total_cost": 70.0, "currency": "USD", }), ] class _FakeQueryJob: def result(self) -> list[Any]: return fake_rows class _FakeBQClient: def __init__(self, *_a: Any, **_k: Any) -> None: pass def query(self, sql: str, *, job_config: Any = None) -> _FakeQueryJob: captured.setdefault("sql", sql) return _FakeQueryJob() def close(self) -> None: pass fake_bq = MagicMock() fake_bq.Client = _FakeBQClient fake_bq.QueryJobConfig = lambda **kw: kw fake_bq.ScalarQueryParameter = lambda *a, **kw: ("p", a, kw) with patch.dict( "sys.modules", {"google.cloud.bigquery": fake_bq, "google.cloud": MagicMock(bigquery=fake_bq)}, ), patch.object( GcpCloudDriver, "_credentials", lambda self: None, ): out = await d.list_billing_daily( start_date=_date(2026, 5, 1), end_date=_date(2026, 5, 7), ) # SQL nets the unnested credit sum into total_cost. sql = captured["sql"] assert "UNNEST(credits)" in sql assert "AS total_cost" in sql # The fake driver returns the already-netted total; assert it flows # through unchanged so the netting isn't double-applied in Python. assert len(out) == 1 assert out[0].total_cost_usd == 70.0 @pytest.mark.asyncio async def test_list_billing_daily_emits_business_list_cost() -> None: """双口径（成本口径改造 PR2）：daily SQL 携带 business list cost 白名单列（cost_at_list；cost_type='regular'；Support/Invoice 排除； credits 不参与），行发射把 total_cost_usd_list / _public 落到 CloudDailyCost（GCP 牌价 == list，两者同值）。""" from datetime import date as _date d = GcpCloudDriver(integration_id="t", config={ "organization_id": _ORG_ID, "service_account_json": _SA_JSON, "billing_bq_table": "proj.bill.gcp_billing_export_v1_AAA", }) captured: dict[str, Any] = {} class _FakeRow(dict): def get(self, k: str, default: Any = None) -> Any: return dict.get(self, k, default) fake_rows = [ _FakeRow({ "day": _date(2026, 5, 5), "project_id": "proj-x", "service_name": "Compute Engine", "total_cost": 70.0, "total_cost_list": 100.0, "currency": "USD", }), # Support 组：list 白名单排除 → SQL 端 NULL → 计 0，不影响 list。 _FakeRow({ "day": _date(2026, 5, 5), "project_id": "proj-x", "service_name": "Support", "total_cost": 8.0, "total_cost_list": None, "currency": "USD", }), ] class _FakeQueryJob: def result(self) -> list[Any]: return fake_rows class _FakeBQClient: def __init__(self, *_a: Any, **_k: Any) -> None: pass def query(self, sql: str, *, job_config: Any = None) -> _FakeQueryJob: captured.setdefault("sql", sql) return _FakeQueryJob() def close(self) -> None: pass fake_bq = MagicMock() fake_bq.Client = _FakeBQClient fake_bq.QueryJobConfig = lambda **kw: kw fake_bq.ScalarQueryParameter = lambda *a, **kw: ("p", a, kw) with patch.dict( "sys.modules", {"google.cloud.bigquery": fake_bq, "google.cloud": MagicMock(bigquery=fake_bq)}, ), patch.object( GcpCloudDriver, "_credentials", lambda self: None, ): out = await d.list_billing_daily( start_date=_date(2026, 5, 1), end_date=_date(2026, 5, 7), ) sql = captured["sql"] # 白名单形状：cost_at_list + regular + Support/Invoice 排除。 assert "cost_at_list" in sql assert "cost_type = 'regular'" in sql assert "NOT IN ('Support', 'Invoice')" in sql assert len(out) == 1 row = out[0] # 发票口径（净额）不变；研发口径 = 白名单 list 合计（Support 组 # 的 NULL 计 0）；GCP 牌价 == list。 assert row.total_cost_usd == 78.0 assert row.total_cost_usd_list == 100.0 assert row.total_cost_usd_public == 100.0 @pytest.mark.asyncio async def test_cost_at_list_missing_degrades_to_null_measures() -> None: """cost_at_list 缺列（老导出 schema）→ 单次降级重试：发票口径照常 产出，list/public 落 None（=口径未接入），绝不伪造 $0。daily 与 cost_facts 两条路径同一策略；此处 pin daily。""" from datetime import date as _date d = GcpCloudDriver(integration_id="t", config={ "organization_id": _ORG_ID, "service_account_json": _SA_JSON, "billing_bq_table": "proj.bill.gcp_billing_export_v1_AAA", }) class _FakeRow(dict): def get(self, k: str, default: Any = None) -> Any: return dict.get(self, k, default) legacy_rows = [ _FakeRow({ "day": _date(2026, 5, 5), "project_id": "proj-x", "service_name": "Compute Engine", "total_cost": 70.0, "currency": "USD", }), ] calls: list[str] = [] class _FakeQueryJob: def __init__(self, rows: list[Any] | None) -> None: self._rows = rows def result(self) -> list[Any]: if self._rows is None: raise RuntimeError( "400 Unrecognized name: cost_at_list at [7:24]", ) return self._rows class _FakeBQClient: def __init__(self, *_a: Any, **_k: Any) -> None: pass def query(self, sql: str, *, job_config: Any = None) -> _FakeQueryJob: calls.append(sql) if "cost_at_list" in sql: return _FakeQueryJob(None) # 老 schema：列不存在。 return _FakeQueryJob(legacy_rows) def close(self) -> None: pass fake_bq = MagicMock() fake_bq.Client = _FakeBQClient fake_bq.QueryJobConfig = lambda **kw: kw fake_bq.ScalarQueryParameter = lambda *a, **kw: ("p", a, kw) with patch.dict( "sys.modules", {"google.cloud.bigquery": fake_bq, "google.cloud": MagicMock(bigquery=fake_bq)}, ), patch.object( GcpCloudDriver, "_credentials", lambda self: None, ): out = await d.list_billing_daily( start_date=_date(2026, 5, 1), end_date=_date(2026, 5, 7), ) # 两次查询：带 list 列的失败 → 降级重试成功。 assert len(calls) == 2 assert "cost_at_list" in calls[0] assert "cost_at_list" not in calls[1] assert len(out) == 1 assert out[0].total_cost_usd == 70.0 assert out[0].total_cost_usd_list is None assert out[0].total_cost_usd_public is None