"""DAX-023 BigQuery data-access driver read-path tests.""" from __future__ import annotations import json import re from pathlib import Path from typing import Any import pytest from app.domains.data_access.constants import ( BIGQUERY_DATA_VIEWER_ROLE, BIGQUERY_GROUP_MEMBER_PREFIX, BIGQUERY_SERVICE_ACCOUNT_MEMBER_PREFIX, BIGQUERY_USER_MEMBER_PREFIX, ) from app.integrations.driver_base import ( DataAccessDatasetFieldSchema, DataAccessDatasetIamBinding, DataAccessDatasetIamPolicy, DataAccessDatasetIamPolicyPatch, DataAccessDatasetTablePage, DataAccessDatasetTableSchema, DataAccessDatasetTableSummary, ) from app.integrations.drivers import bigquery_data_access as driver_module from app.integrations.drivers.bigquery_data_access import BigQueryDataAccessDriver _SA_INFO = { "type": "service_account", "client_email": "dax-reader@example.iam.gserviceaccount.com", "private_key": "-----BEGIN PRIVATE KEY-----\nabc\n-----END PRIVATE KEY-----\n", "project_id": "helix-dax-admin", } def _driver(config: dict[str, Any] | None = None) -> BigQueryDataAccessDriver: return BigQueryDataAccessDriver( integration_id="dummy", config=config or {"service_account_json": json.dumps(_SA_INFO)}, ) def _enabled_apply_config() -> dict[str, Any]: return { "apply_mode_enabled": True, } def _dataset_response(etag: str, access: list[dict[str, Any]]) -> dict[str, Any]: return { "datasetReference": { "projectId": "analytics-prod", "datasetId": "warehouse", }, "etag": etag, "access": access, } class _FakeAccessEntry: def __init__( self, role: str, entity_type: str, entity_id: str, *, condition: dict[str, str] | None = None, ) -> None: self.role = role self.entity_type = entity_type self.entity_id = entity_id self.condition = condition def to_api_repr(self) -> dict[str, Any]: out: dict[str, Any] = { "role": self.role, "entity_type": self.entity_type, "entity_id": self.entity_id, } if self.condition is not None: out["condition"] = self.condition return out class _FakeDataset: etag = "dataset-etag-1" access_entries = ( _FakeAccessEntry("READER", "groupByEmail", "analytics@example.com"), _FakeAccessEntry("READER", "iamMember", "group:finance@example.com"), _FakeAccessEntry("WRITER", "userByEmail", "owner@example.com"), _FakeAccessEntry("OWNER", "specialGroup", "projectOwners"), _FakeAccessEntry("READER", "view", "view-ref-is-not-a-principal"), _FakeAccessEntry( "READER", "groupByEmail", "conditional@example.com", condition={ "title": "expires", "expression": "request.time < timestamp('2026-07-01T00:00:00Z')", }, ), ) @pytest.mark.asyncio async def test_get_dataset_iam_reads_only_requested_dataset_locator(monkeypatch) -> None: calls: list[tuple[str, str, str, str]] = [] class _FakeClient: def get_dataset(self, dataset_ref, *, dataset_view=None): calls.append(( dataset_ref.project, dataset_ref.dataset_id, dataset_ref.path, str(getattr(dataset_view, "value", dataset_view)), )) return _FakeDataset() def list_datasets(self, *_a, **_kw): # pragma: no cover - fail-fast guard raise AssertionError("DAX read path must not list datasets") def query(self, *_a, **_kw): # pragma: no cover - fail-fast guard raise AssertionError("DAX read path must not run BigQuery jobs") def close(self) -> None: calls.append(("closed", "", "", "")) driver = _driver() monkeypatch.setattr(driver, "_bigquery_client", lambda: _FakeClient()) policy = await driver.get_dataset_iam("analytics-prod", "finance_2026") assert calls == [ ( "analytics-prod", "finance_2026", "/projects/analytics-prod/datasets/finance_2026", "ACL", ), ("closed", "", "", ""), ] assert policy.project_id == "analytics-prod" assert policy.dataset_id == "finance_2026" assert policy.etag == "dataset-etag-1" assert policy.bindings == ( DataAccessDatasetIamBinding( role="roles/bigquery.dataEditor", members=("user:owner@example.com",), ), DataAccessDatasetIamBinding( role="roles/bigquery.dataOwner", members=("specialGroup:projectOwners",), ), DataAccessDatasetIamBinding( role="roles/bigquery.dataViewer", members=("group:analytics@example.com", "group:finance@example.com"), ), ) assert policy.raw_policy == { "access_entries": [ { "role": "READER", "entity_type": "groupByEmail", "entity_id": "analytics@example.com", }, { "role": "READER", "entity_type": "iamMember", "entity_id": "group:finance@example.com", }, { "role": "WRITER", "entity_type": "userByEmail", "entity_id": "owner@example.com", }, { "role": "OWNER", "entity_type": "specialGroup", "entity_id": "projectOwners", }, { "role": "READER", "entity_type": "view", "entity_id": "view-ref-is-not-a-principal", }, { "role": "READER", "entity_type": "groupByEmail", "entity_id": "conditional@example.com", "condition": { "title": "expires", "expression": "request.time < timestamp('2026-07-01T00:00:00Z')", }, }, ], } @pytest.mark.asyncio async def test_get_dataset_iam_fails_closed_on_auth_errors(monkeypatch) -> None: class _FakeClient: def get_dataset(self, dataset_ref): raise PermissionError(f"denied {dataset_ref.project}.{dataset_ref.dataset_id}") def list_datasets(self, *_a, **_kw): # pragma: no cover - fail-fast guard raise AssertionError("DAX read path must not list datasets") def close(self) -> None: pass driver = _driver() monkeypatch.setattr(driver, "_bigquery_client", lambda: _FakeClient()) with pytest.raises(RuntimeError, match="BigQuery dataset IAM read failed"): await driver.get_dataset_iam("analytics-prod", "finance_2026") @pytest.mark.parametrize( ("project_id", "dataset_id", "message"), [ ("AnalyticsProd", "finance_2026", "project_id"), ("analytics-prod", "finance.table", "dataset only"), ("analytics-prod", "finance/table", "dataset only"), ], ) @pytest.mark.asyncio async def test_get_dataset_iam_rejects_non_dataset_locators( project_id: str, dataset_id: str, message: str, ) -> None: with pytest.raises(ValueError, match=message): await _driver().get_dataset_iam(project_id, dataset_id) @pytest.mark.asyncio async def test_list_dataset_tables_is_scoped_to_one_dataset(monkeypatch) -> None: calls: list[tuple[str, str]] = [] class _Table: def __init__(self, table_id: str, table_type: str) -> None: self.table_id = table_id self.table_type = table_type class _FakeClient: def list_tables(self, dataset_ref): calls.append((dataset_ref.project, dataset_ref.dataset_id)) return [ _Table("z_monthly_view", "VIEW"), _Table("account_daily", "TABLE"), ] def list_datasets(self, *_a, **_kw): # pragma: no cover - fail-fast guard raise AssertionError("schema browsing must not list datasets") def query(self, *_a, **_kw): # pragma: no cover - fail-fast guard raise AssertionError("schema browsing must not run query jobs") def close(self) -> None: calls.append(("closed", "")) driver = _driver() monkeypatch.setattr(driver, "_bigquery_client", lambda: _FakeClient()) tables = await driver.list_dataset_tables("analytics-prod", "warehouse") assert calls == [("analytics-prod", "warehouse"), ("closed", "")] assert tables == ( DataAccessDatasetTableSummary( table_id="account_daily", table_type="TABLE", ), DataAccessDatasetTableSummary( table_id="z_monthly_view", table_type="VIEW", ), ) @pytest.mark.asyncio async def test_list_dataset_tables_page_uses_provider_cursor_without_full_scan( monkeypatch, ) -> None: calls: list[object] = [] class _Table: def __init__(self, table_id: str, table_type: str) -> None: self.table_id = table_id self.table_type = table_type class _Page: raw_page = {"totalItems": "174"} def __iter__(self): return iter([ _Table("nexuscrm_account", "TABLE"), _Table("nexuscrm_contact", "TABLE"), ]) class _Iterator: next_page_token = "provider-next-page" @property def pages(self): return iter([_Page()]) class _FakeClient: def list_tables(self, dataset_ref, *, max_results, page_token): calls.append(( dataset_ref.project, dataset_ref.dataset_id, max_results, page_token, )) return _Iterator() def list_datasets(self, *_a, **_kw): # pragma: no cover - fail-fast guard raise AssertionError("schema browsing must not list datasets") def query(self, *_a, **_kw): # pragma: no cover - fail-fast guard raise AssertionError("schema browsing must not run query jobs") def close(self) -> None: calls.append("closed") driver = _driver() monkeypatch.setattr(driver, "_bigquery_client", lambda: _FakeClient()) page = await driver.list_dataset_tables_page( "analytics-prod", "warehouse", page_size=20, page_token="provider-current-page", ) assert page == DataAccessDatasetTablePage( items=( DataAccessDatasetTableSummary("nexuscrm_account", "TABLE"), DataAccessDatasetTableSummary("nexuscrm_contact", "TABLE"), ), next_page_token="provider-next-page", total_rows=174, ) assert calls == [ ("analytics-prod", "warehouse", 20, "provider-current-page"), "closed", ] @pytest.mark.asyncio async def test_get_dataset_table_schema_flattens_nested_fields(monkeypatch) -> None: calls: list[tuple[str, str, str]] = [] class _Field: def __init__( self, name: str, field_type: str, *, mode: str = "NULLABLE", description: str | None = None, fields: tuple[Any, ...] = (), ) -> None: self.name = name self.field_type = field_type self.mode = mode self.description = description self.fields = fields class _Table: table_id = "account_daily" table_type = "TABLE" description = "Daily account snapshot" schema = ( _Field("account_id", "STRING", mode="REQUIRED", description="Account key"), _Field( "owner", "RECORD", fields=( _Field("email", "STRING", description="Current owner email"), _Field("department", "STRING"), ), ), ) class _FakeClient: def get_table(self, table_ref): calls.append(( table_ref.project, table_ref.dataset_id, table_ref.table_id, )) return _Table() def query(self, *_a, **_kw): # pragma: no cover - fail-fast guard raise AssertionError("schema browsing must not run query jobs") def close(self) -> None: calls.append(("closed", "", "")) driver = _driver() monkeypatch.setattr(driver, "_bigquery_client", lambda: _FakeClient()) schema = await driver.get_dataset_table_schema( "analytics-prod", "warehouse", "account_daily", ) assert calls == [ ("analytics-prod", "warehouse", "account_daily"), ("closed", "", ""), ] assert schema == DataAccessDatasetTableSchema( project_id="analytics-prod", dataset_id="warehouse", table_id="account_daily", table_type="TABLE", description="Daily account snapshot", fields=( DataAccessDatasetFieldSchema( name="account_id", field_path="account_id", field_type="STRING", mode="REQUIRED", description="Account key", depth=0, ordinal_position=1, ), DataAccessDatasetFieldSchema( name="owner", field_path="owner", field_type="RECORD", mode="NULLABLE", description=None, depth=0, ordinal_position=2, ), DataAccessDatasetFieldSchema( name="email", field_path="owner.email", field_type="STRING", mode="NULLABLE", description="Current owner email", depth=1, ordinal_position=1, ), DataAccessDatasetFieldSchema( name="department", field_path="owner.department", field_type="STRING", mode="NULLABLE", description=None, depth=1, ordinal_position=2, ), ), ) def test_service_account_json_is_required_and_validated() -> None: with pytest.raises(RuntimeError, match="service_account_json not configured"): _driver({"service_account_json": ""})._service_account_info() with pytest.raises(ValueError, match="not valid JSON"): _driver({"service_account_json": "not json"})._service_account_info() with pytest.raises(ValueError, match="missing fields: private_key"): _driver({ "service_account_json": json.dumps({ "type": "service_account", "client_email": "dax-reader@example.iam.gserviceaccount.com", "project_id": "helix-dax-admin", }), })._service_account_info() def test_bigquery_client_uses_service_account_json_and_bigquery_scope(monkeypatch) -> None: from google.cloud import bigquery from google.oauth2 import service_account captured: dict[str, Any] = {} def _fake_credentials_from_info(info: dict[str, Any], *, scopes: list[str]): captured["info"] = info captured["scopes"] = scopes return "fake-credentials" class _FakeClient: def __init__(self, *, credentials: Any, project: str) -> None: captured["credentials"] = credentials captured["project"] = project monkeypatch.setattr( service_account.Credentials, "from_service_account_info", staticmethod(_fake_credentials_from_info), ) monkeypatch.setattr(bigquery, "Client", _FakeClient) client = _driver()._bigquery_client() assert isinstance(client, _FakeClient) assert captured["info"] == _SA_INFO assert captured["scopes"] == [driver_module.BIGQUERY_DATA_ACCESS_OAUTH_SCOPE] assert captured["credentials"] == "fake-credentials" assert captured["project"] == "helix-dax-admin" def test_least_privilege_contract_pins_acl_and_schema_metadata_permissions() -> None: assert driver_module.BIGQUERY_DATA_ACCESS_OAUTH_SCOPE == ( "https://www.googleapis.com/auth/bigquery" ) assert driver_module.BIGQUERY_DATA_ACCESS_REQUIRED_PERMISSIONS == ( "bigquery.datasets.get", "bigquery.datasets.getIamPolicy", "bigquery.datasets.update", "bigquery.datasets.setIamPolicy", "bigquery.tables.get", "bigquery.tables.getIamPolicy", "bigquery.tables.list", "bigquery.tables.setIamPolicy", ) assert "readonly" not in driver_module.BIGQUERY_DATA_ACCESS_OAUTH_SCOPE assert { "bigquery.datasets.create", "bigquery.datasets.delete", "bigquery.datasets.list", "bigquery.jobs.create", "bigquery.tables.getData", "bigquery.tables.update", "bigquery.tables.updateData", }.isdisjoint(driver_module.BIGQUERY_DATA_ACCESS_REQUIRED_PERMISSIONS) terraform = ( Path(__file__).parents[3] / "infra/gcp-dax/main.tf" ).read_text(encoding="utf-8") role_block = re.search( r'resource "google_project_iam_custom_role" "dax_dataset_iam_reconciler" ' r"\{(?P.*?)\n\}", terraform, re.DOTALL, ) assert role_block is not None assert "for_each = var.managed_dataset_project_ids" in role_block.group("body") assert "project = each.value" in role_block.group("body") terraform_permissions = tuple(re.findall( r'"(bigquery\.[^"]+)"', role_block.group("body"), )) assert terraform_permissions == driver_module.BIGQUERY_DATA_ACCESS_REQUIRED_PERMISSIONS def test_dataset_acl_patch_body_only_changes_exact_dax_group_grants() -> None: assert driver_module._DATASET_UPDATE_MODE_ACL == "UPDATE_ACL" raw_access = [ {"role": "OWNER", "specialGroup": "projectOwners"}, {"role": "READER", "specialGroup": "projectReaders"}, {"role": "WRITER", "groupByEmail": "writers@example.com"}, {"role": "READER", "userByEmail": "person@example.com"}, { "role": "READER", "view": { "projectId": "analytics-prod", "datasetId": "shared_views", "tableId": "v_finance", }, }, { "role": "READER", "routine": { "projectId": "analytics-prod", "datasetId": "shared_routines", "routineId": "mask_email", }, }, { "role": "roles/bigquery.jobUser", "iamMember": "group:batch@example.com", }, {"role": "READER", "groupByEmail": "remove@example.com"}, { "role": "READER", "groupByEmail": "remove@example.com", "condition": { "title": "temporary remove", "expression": ( "request.time < timestamp('2026-12-31T00:00:00Z')" ), }, }, {"role": "READER", "iamMember": "group:already@example.com"}, { "role": "READER", "groupByEmail": "conditional-only@example.com", "condition": { "title": "temporary existing", "expression": ( "request.time < timestamp('2026-12-31T00:00:00Z')" ), }, }, ] policy = DataAccessDatasetIamPolicy( project_id="analytics-prod", dataset_id="warehouse", etag="dataset-etag-canary", bindings=(), raw_policy={"etag": "dataset-etag-canary", "access": raw_access}, ) patch = DataAccessDatasetIamPolicyPatch( add_bindings=( DataAccessDatasetIamBinding( role=BIGQUERY_DATA_VIEWER_ROLE, members=( f"{BIGQUERY_GROUP_MEMBER_PREFIX}already@example.com", f"{BIGQUERY_GROUP_MEMBER_PREFIX}conditional-only@example.com", f"{BIGQUERY_GROUP_MEMBER_PREFIX}new@example.com", ), ), ), remove_bindings=( DataAccessDatasetIamBinding( role=BIGQUERY_DATA_VIEWER_ROLE, members=(f"{BIGQUERY_GROUP_MEMBER_PREFIX}remove@example.com",), ), ), ) body = driver_module._dataset_access_update_body(policy, patch) assert body == { "etag": "dataset-etag-canary", "access": [ {"role": "OWNER", "specialGroup": "projectOwners"}, {"role": "READER", "specialGroup": "projectReaders"}, {"role": "WRITER", "groupByEmail": "writers@example.com"}, {"role": "READER", "userByEmail": "person@example.com"}, { "role": "READER", "view": { "projectId": "analytics-prod", "datasetId": "shared_views", "tableId": "v_finance", }, }, { "role": "READER", "routine": { "projectId": "analytics-prod", "datasetId": "shared_routines", "routineId": "mask_email", }, }, { "role": "roles/bigquery.jobUser", "iamMember": "group:batch@example.com", }, { "role": "READER", "groupByEmail": "remove@example.com", "condition": { "title": "temporary remove", "expression": ( "request.time < timestamp('2026-12-31T00:00:00Z')" ), }, }, {"role": "READER", "iamMember": "group:already@example.com"}, { "role": "READER", "groupByEmail": "conditional-only@example.com", "condition": { "title": "temporary existing", "expression": ( "request.time < timestamp('2026-12-31T00:00:00Z')" ), }, }, {"role": "READER", "groupByEmail": "conditional-only@example.com"}, {"role": "READER", "groupByEmail": "new@example.com"}, ], } assert {"role": "READER", "groupByEmail": "remove@example.com"} in raw_access def test_dataset_acl_patch_body_accepts_current_access_entry_shape() -> None: policy = DataAccessDatasetIamPolicy( project_id="analytics-prod", dataset_id="warehouse", etag="dataset-etag-current-shape", bindings=(), raw_policy={ "access_entries": [ { "role": "READER", "entity_type": "groupByEmail", "entity_id": "remove@example.com", }, { "role": "READER", "entity_type": "groupByEmail", "entity_id": "keep@example.com", "condition": {"title": "temporary"}, }, ], }, ) patch = DataAccessDatasetIamPolicyPatch( add_bindings=( DataAccessDatasetIamBinding( role=BIGQUERY_DATA_VIEWER_ROLE, members=(f"{BIGQUERY_GROUP_MEMBER_PREFIX}new@example.com",), ), ), remove_bindings=( DataAccessDatasetIamBinding( role=BIGQUERY_DATA_VIEWER_ROLE, members=(f"{BIGQUERY_GROUP_MEMBER_PREFIX}remove@example.com",), ), ), ) assert driver_module._dataset_access_update_body(policy, patch) == { "etag": "dataset-etag-current-shape", "access": [ { "role": "READER", "groupByEmail": "keep@example.com", "condition": {"title": "temporary"}, }, {"role": "READER", "groupByEmail": "new@example.com"}, ], } def test_dataset_acl_patch_body_rejects_non_dax_grant_shapes() -> None: policy = DataAccessDatasetIamPolicy( project_id="analytics-prod", dataset_id="warehouse", etag="dataset-etag-canary", bindings=(), raw_policy={"etag": "dataset-etag-canary", "access": []}, ) with pytest.raises(ValueError, match="roles/bigquery.dataViewer"): driver_module._dataset_access_update_body( policy, DataAccessDatasetIamPolicyPatch( add_bindings=( DataAccessDatasetIamBinding( role="roles/bigquery.metadataViewer", members=( f"{BIGQUERY_GROUP_MEMBER_PREFIX}analytics@example.com", ), ), ), ), ) with pytest.raises(ValueError, match="user:"): driver_module._dataset_access_update_body( policy, DataAccessDatasetIamPolicyPatch( add_bindings=( DataAccessDatasetIamBinding( role=BIGQUERY_DATA_VIEWER_ROLE, members=("user:not-an-email",), ), ), ), ) def test_dataset_acl_patch_body_manages_exact_service_account_entries_only() -> None: """A DAX SA effect is a BigQuery ``userByEmail`` entry, not a user grant. The provider can surface the same SA through ``iamMember``. Both read shapes are intentionally recognised for the exact service-account email, while unmanaged user entries and conditional entries remain untouched. """ service_account = "etl@analytics-prod.iam.gserviceaccount.com" policy = DataAccessDatasetIamPolicy( project_id="analytics-prod", dataset_id="warehouse", etag="dataset-etag-service-account", bindings=(), raw_policy={ "access": [ {"role": "READER", "userByEmail": service_account}, {"role": "READER", "userByEmail": "person@example.com"}, { "role": "READER", "iamMember": f"serviceAccount:{service_account}", "condition": {"title": "temporary"}, }, {"role": "READER", "iamMember": "user:person@example.com"}, ], }, ) patch = DataAccessDatasetIamPolicyPatch( add_bindings=( DataAccessDatasetIamBinding( role=BIGQUERY_DATA_VIEWER_ROLE, members=( f"{BIGQUERY_SERVICE_ACCOUNT_MEMBER_PREFIX}{service_account}", f"{BIGQUERY_SERVICE_ACCOUNT_MEMBER_PREFIX}new-etl@analytics-prod.iam.gserviceaccount.com", ), ), ), remove_bindings=( DataAccessDatasetIamBinding( role=BIGQUERY_DATA_VIEWER_ROLE, members=( f"{BIGQUERY_SERVICE_ACCOUNT_MEMBER_PREFIX}{service_account}", ), ), ), ) assert driver_module._dataset_access_update_body(policy, patch) == { "etag": "dataset-etag-service-account", "access": [ {"role": "READER", "userByEmail": "person@example.com"}, { "role": "READER", "iamMember": f"serviceAccount:{service_account}", "condition": {"title": "temporary"}, }, {"role": "READER", "iamMember": "user:person@example.com"}, { "role": "READER", "userByEmail": "new-etl@analytics-prod.iam.gserviceaccount.com", }, ], } def test_dataset_acl_patch_body_manages_exact_personal_user_entries_only() -> None: """Personal DAX effects map only their exact userByEmail ACL entry.""" user_email = "employee@example.com" policy = DataAccessDatasetIamPolicy( project_id="analytics-prod", dataset_id="warehouse", etag="dataset-etag-canary", bindings=(), raw_policy={ "access": [ {"role": "READER", "userByEmail": user_email}, {"role": "READER", "userByEmail": "other@example.com"}, { "role": "READER", "iamMember": f"user:{user_email}", "condition": {"title": "temporary"}, }, { "role": "READER", "userByEmail": "etl@analytics-prod.iam.gserviceaccount.com", }, ], }, ) body = driver_module._dataset_access_update_body( policy, DataAccessDatasetIamPolicyPatch( add_bindings=( DataAccessDatasetIamBinding( role=BIGQUERY_DATA_VIEWER_ROLE, members=(f"{BIGQUERY_USER_MEMBER_PREFIX}new.employee@example.com",), ), ), remove_bindings=( DataAccessDatasetIamBinding( role=BIGQUERY_DATA_VIEWER_ROLE, members=(f"{BIGQUERY_USER_MEMBER_PREFIX}{user_email}",), ), ), ), ) assert body == { "etag": "dataset-etag-canary", "access": [ {"role": "READER", "userByEmail": "other@example.com"}, { "role": "READER", "iamMember": f"user:{user_email}", "condition": {"title": "temporary"}, }, { "role": "READER", "userByEmail": "etl@analytics-prod.iam.gserviceaccount.com", }, {"role": "READER", "userByEmail": "new.employee@example.com"}, ], } @pytest.mark.asyncio async def test_set_dataset_iam_requires_apply_mode_and_expected_etag( monkeypatch, ) -> None: def _unexpected_client(self): raise AssertionError("disabled apply path must not build a BigQuery client") monkeypatch.setattr(BigQueryDataAccessDriver, "_bigquery_client", _unexpected_client) patch = DataAccessDatasetIamPolicyPatch() disabled = await _driver().set_dataset_iam( "analytics-prod", "warehouse", "dataset-etag-1", patch, ) missing_etag = await _driver(_enabled_apply_config()).set_dataset_iam( "analytics-prod", "warehouse", None, patch, ) assert disabled.ok is False assert disabled.changed is False assert disabled.error_message == "BigQuery DAX apply mode is disabled" assert missing_etag.ok is False assert missing_etag.changed is False assert "expected dataset etag" in (missing_etag.error_message or "") @pytest.mark.asyncio async def test_get_table_iam_uses_bigquery_supported_policy_version( monkeypatch, ) -> None: from google.api_core.iam import Policy calls: list[tuple[str, str]] = [] class _FakeClient: def get_iam_policy(self, table_ref, requested_policy_version=1): calls.append(("get", table_ref.path)) assert requested_policy_version == 1 return Policy.from_api_repr({ "version": 1, "etag": "table-etag-1", "bindings": [], }) def close(self) -> None: calls.append(("close", "")) monkeypatch.setattr( BigQueryDataAccessDriver, "_bigquery_client", lambda self: _FakeClient(), ) policy = await _driver().get_table_iam( "analytics-prod", "warehouse", "account_daily", ) assert policy.etag == "table-etag-1" assert policy.bindings == () assert calls == [ ("get", "/projects/analytics-prod/datasets/warehouse/tables/account_daily"), ("close", ""), ] @pytest.mark.asyncio async def test_set_table_iam_patches_only_the_exact_table_policy( monkeypatch, ) -> None: from google.api_core.iam import Policy calls: list[tuple[str, str]] = [] current = Policy.from_api_repr({ "version": 1, "etag": "table-etag-1", "bindings": [ { "role": BIGQUERY_DATA_VIEWER_ROLE, "members": [ "group:remove@example.com", "user:keep@example.com", ], }, { "role": "roles/bigquery.dataOwner", "members": ["group:owners@example.com"], }, ], }) class _FakeClient: def get_iam_policy(self, table_ref, requested_policy_version=1): calls.append(("get", table_ref.path)) assert requested_policy_version == 1 return current def set_iam_policy(self, table_ref, policy, fields=None): calls.append(("set", table_ref.path)) assert fields == ["bindings", "etag", "version"] assert policy.to_api_repr() == { "etag": "table-etag-1", "version": 1, "bindings": [ { "role": "roles/bigquery.dataOwner", "members": ["group:owners@example.com"], }, { "role": BIGQUERY_DATA_VIEWER_ROLE, "members": [ "group:new@example.com", "user:keep@example.com", ], }, ], } policy.etag = "table-etag-2" return policy def close(self) -> None: calls.append(("close", "")) monkeypatch.setattr( BigQueryDataAccessDriver, "_bigquery_client", lambda self: _FakeClient(), ) patch = DataAccessDatasetIamPolicyPatch( add_bindings=(DataAccessDatasetIamBinding( role=BIGQUERY_DATA_VIEWER_ROLE, members=("group:new@example.com",), ),), remove_bindings=(DataAccessDatasetIamBinding( role=BIGQUERY_DATA_VIEWER_ROLE, members=("group:remove@example.com",), ),), ) result = await _driver(_enabled_apply_config()).set_table_iam( "analytics-prod", "warehouse", "account_daily", "table-etag-1", patch, ) assert result.ok is True, result.error_message assert result.changed is True assert result.etag == "table-etag-2" assert calls == [ ("get", "/projects/analytics-prod/datasets/warehouse/tables/account_daily"), ("set", "/projects/analytics-prod/datasets/warehouse/tables/account_daily"), ("close", ""), ] @pytest.mark.asyncio async def test_set_dataset_iam_patches_only_exact_dax_apply_and_revoke( monkeypatch, ) -> None: calls: list[dict[str, Any]] = [] current_access = [ {"role": "OWNER", "specialGroup": "projectOwners"}, {"role": "WRITER", "groupByEmail": "writers@example.com"}, {"role": "READER", "groupByEmail": "remove@example.com"}, { "role": "READER", "groupByEmail": "remove@example.com", "condition": { "title": "temporary", "expression": "request.time < timestamp('2026-12-31T00:00:00Z')", }, }, {"role": "READER", "iamMember": "group:already@example.com"}, {"role": "READER", "userByEmail": "person@example.com"}, ] patched_access = [ {"role": "OWNER", "specialGroup": "projectOwners"}, {"role": "WRITER", "groupByEmail": "writers@example.com"}, { "role": "READER", "groupByEmail": "remove@example.com", "condition": { "title": "temporary", "expression": "request.time < timestamp('2026-12-31T00:00:00Z')", }, }, {"role": "READER", "iamMember": "group:already@example.com"}, {"role": "READER", "userByEmail": "person@example.com"}, {"role": "READER", "groupByEmail": "new@example.com"}, ] class _FakeClient: def _call_api(self, retry, **kwargs): del retry calls.append(kwargs) if kwargs["method"] == "GET": return _dataset_response("dataset-etag-1", current_access) if kwargs["method"] == "PATCH": assert kwargs["data"] == { "etag": "dataset-etag-1", "access": patched_access, } return _dataset_response("dataset-etag-2", patched_access) raise AssertionError(f"unexpected BigQuery method {kwargs['method']}") def close(self) -> None: calls.append({"method": "CLOSE"}) monkeypatch.setattr( BigQueryDataAccessDriver, "_bigquery_client", lambda self: _FakeClient(), ) patch = DataAccessDatasetIamPolicyPatch( add_bindings=( DataAccessDatasetIamBinding( role=BIGQUERY_DATA_VIEWER_ROLE, members=( f"{BIGQUERY_GROUP_MEMBER_PREFIX}already@example.com", f"{BIGQUERY_GROUP_MEMBER_PREFIX}new@example.com", ), ), ), remove_bindings=( DataAccessDatasetIamBinding( role=BIGQUERY_DATA_VIEWER_ROLE, members=(f"{BIGQUERY_GROUP_MEMBER_PREFIX}remove@example.com",), ), ), ) result = await _driver(_enabled_apply_config()).set_dataset_iam( "analytics-prod", "warehouse", "dataset-etag-1", patch, ) assert result.ok is True assert result.changed is True assert result.conflict is False assert result.etag == "dataset-etag-2" assert result.raw_response == _dataset_response("dataset-etag-2", patched_access) assert calls == [ { "span_name": "BigQuery.getDataset", "span_attributes": { "path": "/projects/analytics-prod/datasets/warehouse", }, "method": "GET", "path": "/projects/analytics-prod/datasets/warehouse", "query_params": { "datasetView": "ACL", "accessPolicyVersion": 3, }, }, { "span_name": "BigQuery.updateDataset", "span_attributes": { "path": "/projects/analytics-prod/datasets/warehouse", "fields": ("access",), }, "method": "PATCH", "path": "/projects/analytics-prod/datasets/warehouse", "data": { "etag": "dataset-etag-1", "access": patched_access, }, "headers": {"If-Match": "dataset-etag-1"}, "query_params": { "updateMode": "UPDATE_ACL", "accessPolicyVersion": 3, }, }, {"method": "CLOSE"}, ] @pytest.mark.asyncio async def test_set_dataset_iam_returns_conflict_without_write_on_stale_etag( monkeypatch, ) -> None: calls: list[str] = [] class _FakeClient: def _call_api(self, retry, **kwargs): del retry calls.append(kwargs["method"]) if kwargs["method"] == "GET": return _dataset_response("dataset-etag-new", []) raise AssertionError("stale etag must not patch BigQuery") def close(self) -> None: calls.append("CLOSE") monkeypatch.setattr( BigQueryDataAccessDriver, "_bigquery_client", lambda self: _FakeClient(), ) result = await _driver(_enabled_apply_config()).set_dataset_iam( "analytics-prod", "warehouse", "dataset-etag-old", DataAccessDatasetIamPolicyPatch( add_bindings=( DataAccessDatasetIamBinding( role=BIGQUERY_DATA_VIEWER_ROLE, members=(f"{BIGQUERY_GROUP_MEMBER_PREFIX}new@example.com",), ), ), ), ) assert result.ok is False assert result.changed is False assert result.conflict is True assert result.etag == "dataset-etag-new" assert "expected 'dataset-etag-old'" in (result.error_message or "") assert calls == ["GET", "CLOSE"] @pytest.mark.asyncio async def test_set_dataset_iam_returns_conflict_on_provider_etag_error( monkeypatch, ) -> None: from google.api_core import exceptions class _FakeClient: def _call_api(self, retry, **kwargs): del retry if kwargs["method"] == "GET": return _dataset_response("dataset-etag-1", []) if kwargs["method"] == "PATCH": raise exceptions.PreconditionFailed("etag conflict") raise AssertionError(f"unexpected BigQuery method {kwargs['method']}") def close(self) -> None: pass monkeypatch.setattr( BigQueryDataAccessDriver, "_bigquery_client", lambda self: _FakeClient(), ) result = await _driver(_enabled_apply_config()).set_dataset_iam( "analytics-prod", "warehouse", "dataset-etag-1", DataAccessDatasetIamPolicyPatch( add_bindings=( DataAccessDatasetIamBinding( role=BIGQUERY_DATA_VIEWER_ROLE, members=(f"{BIGQUERY_GROUP_MEMBER_PREFIX}new@example.com",), ), ), ), ) assert result.ok is False assert result.changed is False assert result.conflict is True assert "etag conflict" in (result.error_message or "") @pytest.mark.asyncio async def test_set_dataset_iam_returns_already_applied_without_write( monkeypatch, ) -> None: calls: list[str] = [] current_access = [ {"role": "READER", "groupByEmail": "already@example.com"}, ] class _FakeClient: def _call_api(self, retry, **kwargs): del retry calls.append(kwargs["method"]) if kwargs["method"] == "GET": return _dataset_response("dataset-etag-1", current_access) raise AssertionError("idempotent no-op must not patch BigQuery") def close(self) -> None: calls.append("CLOSE") monkeypatch.setattr( BigQueryDataAccessDriver, "_bigquery_client", lambda self: _FakeClient(), ) result = await _driver(_enabled_apply_config()).set_dataset_iam( "analytics-prod", "warehouse", "dataset-etag-1", DataAccessDatasetIamPolicyPatch( add_bindings=( DataAccessDatasetIamBinding( role=BIGQUERY_DATA_VIEWER_ROLE, members=(f"{BIGQUERY_GROUP_MEMBER_PREFIX}already@example.com",), ), ), ), ) assert result.ok is True assert result.changed is False assert result.already_applied is True assert result.etag == "dataset-etag-1" assert result.raw_response == { "access_entries": current_access, } assert calls == ["GET", "CLOSE"] def test_metadata_marks_service_account_json_as_secret_and_least_privilege() -> None: metadata = BigQueryDataAccessDriver.metadata() field = next(f for f in metadata.fields if f.key == "service_account_json") assert field.required is True assert field.is_secret is True assert field.is_textarea is True assert "HelixDaxDatasetIamReconciler" in (field.hint or "") assert "DATA_ACCESS.md" in (field.hint or "") def test_public_error_message_redacts_private_key_material() -> None: msg = BigQueryDataAccessDriver._public_error_message( RuntimeError( "bad key -----BEGIN PRIVATE KEY-----\nsecret\n-----END PRIVATE KEY-----", ), ) assert "secret" not in msg assert "" in msg