from __future__ import annotations import json import uuid from datetime import UTC, datetime from types import SimpleNamespace from typing import Any import pytest from sqlalchemy import func, select from app.approval.models import ApprovalRequest from app.audit.models import AuditLog from app.domains.data_access import constants as c from app.domains.data_access import router as router_module from app.domains.data_access.models import ( BINDING_STATUS_ACTIVE, BINDING_STATUS_REVOKED, BINDING_STATUS_REVOKED_PENDING, EFFECT_DESIRED_ABSENT, EFFECT_DESIRED_PRESENT, EFFECT_MANAGED_REASON_GRANT_REQUEST, EFFECT_MANAGED_REASON_REVOKE_REQUEST, EFFECT_STATUS_APPLIED, EFFECT_STATUS_FAILED, EFFECT_STATUS_PENDING_RECONCILE, EFFECT_STATUS_RETRYABLE, ENTITLEMENT_STATUS_ACTIVE, ENTITLEMENT_VERSION_V1, OWNER_RESOLUTION_FALLBACK, OWNER_RESOLUTION_RESOLVED, OWNER_RESOLUTION_UNRESOLVED, OWNER_SOURCE_BOOTSTRAP, OWNER_SOURCE_BOOTSTRAP_TRANSFERRED, OWNER_SOURCE_CATALOG, OWNER_SOURCE_CATALOG_TRANSFERRED, RECONCILE_STATUS_APPLIED, RECONCILE_STATUS_FAILED, RECONCILE_STATUS_RETRYABLE, REQUEST_ITEM_STATUS_APPROVED, REQUEST_STATUS_APPROVED, REQUEST_STATUS_PENDING_FEISHU, RESOURCE_STATUS_ACTIVE, RESOURCE_STATUS_SUSPENDED, DataAccessBinding, DataAccessCompiledEffect, DataAccessEffectRun, DataAccessEntitlement, DataAccessRequest, DataAccessRequestItem, DataAccessResource, ) from app.domains.data_access.reconciler import APPLY_PRESENT_OUTCOME_APPLIED from app.domains.data_access.service import ( DATA_ACCESS_COMPILER_VERSION, DataAccessService, ) from app.domains.email_group.models import EmailGroup, EmailGroupOwner from app.domains.email_group.service import EmailGroupService from app.integrations.driver_base import ( DataAccessDatasetIamBinding, DataAccessDatasetIamPolicy, DataAccessDatasetIamPolicyPatch, DataAccessDatasetIamUpdateResult, DataAccessDatasetTableSummary, DataAccessDriver, ) from app.integrations.models import Integration, IntegrationApprovalFlow class _FakeFeishuClient: def __init__(self) -> None: self.instances: list[dict[str, Any]] = [] self.cc_instances: list[dict[str, Any]] = [] async def get_approval_definition(self, approval_code: str) -> dict[str, Any]: return { "form": json.dumps([ { "id": "widget-request-summary", "name": "Request summary", "type": "textarea", }, { "id": "widget-requested-access", "name": "Access to grant", "type": "textarea", }, { "id": "widget-dataset-list", "name": "Datasets", "type": "textarea", }, { "id": "widget-business-reason", "name": "Requester note / business reason", "type": "textarea", }, {"id": "widget-detail-url", "name": "Detail URL", "type": "input"}, ]), } async def resolve_user_id_by_email(self, email: str) -> str: localpart = email.split("@", 1)[0].replace(".", "-") return f"feishu-user-{localpart}" async def create_instance( self, *, approval_code: str, user_id: str | None = None, open_id: str | None = None, form: str, node_approver_user_id_list: list[dict[str, Any]] | None = None, node_approver_open_id_list: list[dict[str, Any]] | None = None, node_cc_user_id_list: list[dict[str, Any]] | None = None, node_cc_open_id_list: list[dict[str, Any]] | None = None, instance_uuid: str | None = None, ) -> dict[str, Any]: instance_code = f"dax-router-{len(self.instances) + 1:03d}" self.instances.append({ "approval_code": approval_code, "user_id": user_id, "open_id": open_id, "form": form, "node_approver_user_id_list": node_approver_user_id_list, "node_approver_open_id_list": node_approver_open_id_list, "node_cc_user_id_list": node_cc_user_id_list, "node_cc_open_id_list": node_cc_open_id_list, "instance_uuid": instance_uuid, }) return { "instance_code": instance_code, "instance_status": "PENDING", } async def cc_instance( self, *, approval_code: str, instance_code: str, user_id: str | None = None, open_id: str | None = None, cc_user_ids: list[str], comment: str | None = None, ) -> dict[str, Any]: self.cc_instances.append({ "approval_code": approval_code, "instance_code": instance_code, "user_id": user_id, "open_id": open_id, "cc_user_ids": cc_user_ids, "comment": comment, }) return {"ok": True, "instance_code": instance_code} async def cancel_instance( self, *, instance_code: str, user_id: str, approval_code: str, ) -> dict[str, Any]: return {"ok": True} class _FakeApplyDriver(DataAccessDriver): provider = c.PROVIDER_BIGQUERY def __init__( self, *, policies: dict[tuple[str, str], DataAccessDatasetIamPolicy], apply_results: list[DataAccessDatasetIamUpdateResult], ) -> None: super().__init__( integration_id="fake-data-access-integration", config={}, ) self.policies = policies self.apply_results = list(apply_results) self.get_calls: list[tuple[str, str]] = [] self.set_calls: list[tuple[str, str, str | None, DataAccessDatasetIamPolicyPatch]] = [] async def get_dataset_iam( self, project_id: str, dataset_id: str, ) -> DataAccessDatasetIamPolicy: self.get_calls.append((project_id, dataset_id)) return self.policies[(project_id, dataset_id)] async def set_dataset_iam( self, project_id: str, dataset_id: str, expected_etag: str | None, policy_patch: DataAccessDatasetIamPolicyPatch, ) -> DataAccessDatasetIamUpdateResult: self.set_calls.append((project_id, dataset_id, expected_etag, policy_patch)) if self.apply_results: return self.apply_results.pop(0) return DataAccessDatasetIamUpdateResult( ok=True, changed=True, etag=f"{expected_etag}-applied", ) class _FakeTableSelectionDriver(DataAccessDriver): provider = c.PROVIDER_BIGQUERY def __init__(self) -> None: super().__init__( integration_id="fake-data-access-table-selection", config={}, ) async def list_dataset_tables( self, project_id: str, dataset_id: str, ) -> tuple[DataAccessDatasetTableSummary, ...]: return (DataAccessDatasetTableSummary("account_daily", "TABLE"),) async def _add_chat_flow( db_session, *, flow_key: str = c.APPROVAL_FLOW_DATA_ACCESS_GROUP_GRANT, ) -> None: integration = Integration( kind="chat_approval", provider="feishu", name="Primary Feishu", slug=f"feishu-dax-router-{uuid.uuid4().hex[:8]}", enabled=True, is_primary=True, config_public={"app_id": "unused-in-test"}, ) flow = IntegrationApprovalFlow( integration=integration, flow_key=flow_key, label="Data Access Grant", external_code="approval-code-dax-router", payload_schema={ "request_summary": { "id": "widget-request-summary", "type": "textarea", }, "requested_access": { "id": "widget-requested-access", "type": "textarea", }, "dataset_list": { "id": "widget-dataset-list", "type": "textarea", }, "business_reason": { "id": "widget-business-reason", "type": "textarea", }, "detail_url": "widget-detail-url", }, active=True, ) db_session.add(flow) await db_session.flush() async def _add_entitlement(db_session) -> DataAccessEntitlement: row = DataAccessEntitlement( domain=c.DOMAIN_DATA_ACCESS, provider=c.PROVIDER_BIGQUERY, resource_type=c.RESOURCE_TYPE_DATASET, principal_type=c.PRINCIPAL_TYPE_GOOGLE_GROUP, entitlement=c.ENTITLEMENT_READ, version=ENTITLEMENT_VERSION_V1, provider_member_prefix=c.BIGQUERY_GROUP_MEMBER_PREFIX, provider_role=c.BIGQUERY_DATA_VIEWER_ROLE, requestable=True, status=ENTITLEMENT_STATUS_ACTIVE, ) db_session.add(row) await db_session.flush() return row async def _add_resource( db_session, *, owner_person_id: uuid.UUID | None, dataset_id: str, integration: Integration | None = None, project_id: str = "helix-data-prod", catalog_resource_id: str | None = None, requestable: bool = True, status: str = RESOURCE_STATUS_ACTIVE, owner_resolution_status: str = OWNER_RESOLUTION_RESOLVED, owner_source: str = OWNER_SOURCE_BOOTSTRAP, source_version: str = "bootstrap-2026-06-10", source_etag: str | None = None, notes: str | None = None, ) -> DataAccessResource: if integration is None: integration = Integration( kind=c.DOMAIN_DATA_ACCESS, provider=c.PROVIDER_BIGQUERY, name="Data Access BigQuery", slug=f"dax-bq-router-{uuid.uuid4().hex[:8]}", enabled=True, is_primary=False, ) db_session.add(integration) await db_session.flush() row = DataAccessResource( integration_id=integration.id, provider=c.PROVIDER_BIGQUERY, resource_type=c.RESOURCE_TYPE_DATASET, project_id=project_id, dataset_id=dataset_id, display_name=f"{project_id}.{dataset_id}", notes=notes, catalog_resource_id=catalog_resource_id, requestable=requestable, status=status, owner_person_id=str(owner_person_id) if owner_person_id else None, owner_resolution_status=owner_resolution_status, owner_source=owner_source, source_version=source_version, source_etag=source_etag or f"etag-{dataset_id}", ) db_session.add(row) await db_session.flush() return row async def _add_group( db_session, *, owner_person_id: uuid.UUID | None, email: str, status: str = "active", google_group_id: str | None = None, ) -> EmailGroup: row = EmailGroup( email=email, name=email.split("@", 1)[0], status=status, google_group_id=google_group_id or f"google-{email}", aliases=["alias-" + email], direct_members_count=12, ) db_session.add(row) await db_session.flush() if owner_person_id is not None: db_session.add( EmailGroupOwner(email_group_id=row.id, person_id=owner_person_id), ) await db_session.flush() return row async def _add_binding_effect( db_session, *, resource: DataAccessResource, group: EmailGroup, entitlement: DataAccessEntitlement, status: str, desired_state: str = EFFECT_DESIRED_PRESENT, binding_status: str = BINDING_STATUS_ACTIVE, reconcile_status: str = RECONCILE_STATUS_RETRYABLE, source_request: DataAccessRequest | None = None, source_item: DataAccessRequestItem | None = None, managed_reason: str = EFFECT_MANAGED_REASON_GRANT_REQUEST, last_provider_outcome: str | None = "auth_error", last_error: str | None = "403 permission denied", ) -> tuple[DataAccessBinding, DataAccessCompiledEffect]: binding = DataAccessBinding( resource_id=resource.id, principal_email_group_id=group.id, principal_google_group_id_snapshot=group.google_group_id, principal_email_snapshot=group.email, entitlement_id=entitlement.id, entitlement_snapshot=c.ENTITLEMENT_READ, entitlement_version_snapshot=ENTITLEMENT_VERSION_V1, source_request_id=source_request.id if source_request else None, source_request_item_id=source_item.id if source_item else None, status=binding_status, reconcile_status=reconcile_status, last_reconcile_error=last_error, ) db_session.add(binding) await db_session.flush() effect = DataAccessCompiledEffect( binding_id=binding.id, resource_id=resource.id, principal_email_group_id=group.id, entitlement_id=entitlement.id, provider=c.PROVIDER_BIGQUERY, resource_type=c.RESOURCE_TYPE_DATASET, target_project_id=resource.project_id, target_dataset_id=resource.dataset_id, provider_member=f"{c.BIGQUERY_GROUP_MEMBER_PREFIX}{group.email}", provider_role=c.BIGQUERY_DATA_VIEWER_ROLE, desired_state=desired_state, entitlement_version_snapshot=ENTITLEMENT_VERSION_V1, resource_source_version_snapshot=resource.source_version, resource_source_etag_snapshot=resource.source_etag, compiler_version=DATA_ACCESS_COMPILER_VERSION, managed_reason=managed_reason, status=status, last_provider_outcome=last_provider_outcome, last_error=last_error, ) db_session.add(effect) await db_session.flush() return binding, effect async def _add_effect_run( db_session, *, effect: DataAccessCompiledEffect, binding: DataAccessBinding, resource: DataAccessResource, mode: str = "apply_present", effect_status: str | None = None, provider_outcome: str = "apply_present_error", error_message: str | None = "403 permission denied", trace_id: str | None = "trace-dax-run-history", ) -> DataAccessEffectRun: run = DataAccessEffectRun( run_id=uuid.uuid4(), effect_id=effect.id, binding_id=binding.id, request_id=binding.source_request_id, resource_id=resource.id, integration_id=str(resource.integration_id), trace_id=trace_id, mode=mode, provider=effect.provider, resource_type=effect.resource_type, target_project_id=effect.target_project_id, target_dataset_id=effect.target_dataset_id, provider_member=effect.provider_member, provider_role=effect.provider_role, desired_state=effect.desired_state, effect_status=effect_status or effect.status, provider_outcome=provider_outcome, provider_etag=effect.last_provider_etag, error_message=error_message, started_at=datetime.now(UTC), completed_at=datetime.now(UTC), ) db_session.add(run) await db_session.flush() return run def test_router_imports_capability_endpoint() -> None: from app.domains.data_access.router import router assert any( route.path.endswith("/capabilities") and "GET" in route.methods for route in router.routes ) def test_service_returns_current_authority_boundary() -> None: data = DataAccessService().get_capabilities() assert data.provider_mutation_enabled is False assert data.contract.domain == c.DOMAIN_DATA_ACCESS assert data.contract.providers == (c.PROVIDER_BIGQUERY,) assert data.contract.resource_types == ( c.RESOURCE_TYPE_DATASET, c.RESOURCE_TYPE_TABLE, ) assert data.contract.principal_types == ( c.PRINCIPAL_TYPE_GOOGLE_GROUP, c.PRINCIPAL_TYPE_GCP_SERVICE_ACCOUNT, c.PRINCIPAL_TYPE_GOOGLE_USER, ) assert data.contract.entitlements == (c.ENTITLEMENT_READ,) assert data.contract.approval_flows == ( c.APPROVAL_FLOW_DATA_ACCESS_GROUP_GRANT, c.APPROVAL_FLOW_DATA_ACCESS_SERVICE_ACCOUNT_GRANT, c.APPROVAL_FLOW_DATA_ACCESS_PERSONAL_GRANT, ) assert data.authority_boundary.request_handlers_mutate_provider is False assert data.authority_boundary.feishu_callbacks_mutate_provider is False assert ( data.authority_boundary.provider_mutation_owner == "data_access_reconciler" ) async def test_capabilities_endpoint_returns_standard_envelope_for_users( client, employee_headers, admin_headers, ) -> None: for headers in (employee_headers, admin_headers): response = await client.get( "/api/v1/data-access/capabilities", headers=headers, ) assert response.status_code == 200 payload = response.json() assert set(payload) == {"success", "data", "meta", "errors"} assert payload["success"] is True assert payload["errors"] == [] assert payload["meta"]["trace_id"] assert payload["meta"]["timestamp"] data = payload["data"] assert data["provider_mutation_enabled"] is False assert data["contract"]["domain"] == c.DOMAIN_DATA_ACCESS assert data["contract"]["providers"] == [c.PROVIDER_BIGQUERY] assert data["contract"]["resource_types"] == [ c.RESOURCE_TYPE_DATASET, c.RESOURCE_TYPE_TABLE, ] assert data["contract"]["principal_types"] == [ c.PRINCIPAL_TYPE_GOOGLE_GROUP, c.PRINCIPAL_TYPE_GCP_SERVICE_ACCOUNT, c.PRINCIPAL_TYPE_GOOGLE_USER, ] assert data["contract"]["entitlements"] == [c.ENTITLEMENT_READ] assert data["contract"]["approval_flows"] == [ c.APPROVAL_FLOW_DATA_ACCESS_GROUP_GRANT, c.APPROVAL_FLOW_DATA_ACCESS_SERVICE_ACCOUNT_GRANT, c.APPROVAL_FLOW_DATA_ACCESS_PERSONAL_GRANT, ] assert ( data["contract"]["provider_effect"]["role"] == c.BIGQUERY_DATA_VIEWER_ROLE ) assert ( data["authority_boundary"]["semantic_principal_field"] == c.SEMANTIC_PRINCIPAL_FIELD ) assert ( data["authority_boundary"]["provider_mutation_owner"] == "data_access_reconciler" ) assert ( data["authority_boundary"]["request_handlers_mutate_provider"] is False ) assert ( data["authority_boundary"]["feishu_callbacks_mutate_provider"] is False ) async def test_requestable_reads_expose_only_authorized_dax_inputs( client, db_session, employee_user, employee_headers, admin_headers, ) -> None: employee, _ = employee_user owned_group = await _add_group( db_session, owner_person_id=employee.id, email="analytics-readers@example.com", ) other_group = await _add_group( db_session, owner_person_id=None, email="platform-readers@example.com", ) await _add_group( db_session, owner_person_id=employee.id, email="deleted-readers@example.com", status="deleted", ) no_google_id = await _add_group( db_session, owner_person_id=employee.id, email="missing-anchor@example.com", ) no_google_id.google_group_id = None visible_resource = await _add_resource( db_session, owner_person_id=employee.id, dataset_id="finance", notes=( "Finance reporting dataset with cloud billing and revenue data." ), ) suspended_resource = await _add_resource( db_session, owner_person_id=employee.id, dataset_id="suspended", status=RESOURCE_STATUS_SUSPENDED, ) await _add_resource( db_session, owner_person_id=employee.id, dataset_id="fallback-owner", owner_resolution_status=OWNER_RESOLUTION_FALLBACK, ) not_requestable_resource = await _add_resource( db_session, owner_person_id=employee.id, dataset_id="not-requestable", requestable=False, ) principals = await client.get( "/api/v1/data-access/requestable-principals", headers=employee_headers, ) assert principals.status_code == 200 principal_rows = principals.json()["data"] assert [row["id"] for row in principal_rows] == [str(owned_group.id)] assert principal_rows[0]["status"] == "active" assert principal_rows[0]["caller_is_owner"] is True assert principal_rows[0]["caller_is_member"] is False assert principal_rows[0]["google_group_id"] == owned_group.google_group_id admin_principals = await client.get( "/api/v1/data-access/requestable-principals", headers=admin_headers, ) assert admin_principals.status_code == 200 admin_ids = {row["id"] for row in admin_principals.json()["data"]} assert str(owned_group.id) in admin_ids assert str(other_group.id) in admin_ids assert str(no_google_id.id) not in admin_ids resources = await client.get( "/api/v1/data-access/requestable-resources", headers=employee_headers, ) assert resources.status_code == 200 resource_rows = resources.json()["data"] assert [row["id"] for row in resource_rows] == [ str(visible_resource.id), str(not_requestable_resource.id), ] assert resource_rows[0]["provider"] == c.PROVIDER_BIGQUERY assert resource_rows[0]["resource_type"] == c.RESOURCE_TYPE_DATASET assert resource_rows[0]["status"] == RESOURCE_STATUS_ACTIVE assert resource_rows[0]["requestable"] is True assert resource_rows[0]["availability"] == "requestable" assert resource_rows[0]["notice_code"] is None assert resource_rows[0]["description"] == ( "Finance reporting dataset with cloud billing and revenue data." ) assert resource_rows[0]["owner_authority_enabled"] is True assert resource_rows[0]["caller_can_manage_owner"] is True assert "Employee User" in resource_rows[0]["owner_label"] assert resource_rows[1]["requestable"] is False assert resource_rows[1]["availability"] == "not_requestable" assert resource_rows[1]["notice_code"] is None admin_resources = await client.get( "/api/v1/data-access/requestable-resources", headers=admin_headers, ) assert admin_resources.status_code == 200, admin_resources.text admin_resource_rows = admin_resources.json()["data"] admin_resources_by_id = { row["id"]: row for row in admin_resource_rows } assert set(admin_resources_by_id) == { str(visible_resource.id), str(suspended_resource.id), str(not_requestable_resource.id), } suspended_row = admin_resources_by_id[str(suspended_resource.id)] assert admin_resources_by_id[str(visible_resource.id)][ "caller_can_manage_owner" ] is False assert suspended_row["status"] == RESOURCE_STATUS_SUSPENDED assert suspended_row["requestable"] is True assert suspended_row["availability"] == "catalog_missing_suspended" assert suspended_row["notice_code"] == "catalog_missing_suspension" assert "fallback-owner" not in admin_resources.text assert "not-requestable" in admin_resources.text visible_resource.notes = "Notes-backed warehouse description." await db_session.flush() legacy_search = await client.get( "/api/v1/data-access/requestable-resources?search=notes-backed", headers=employee_headers, ) assert legacy_search.status_code == 200 assert legacy_search.json()["data"][0]["description"] == ( "Notes-backed warehouse description." ) async def test_requestable_resources_allow_active_employee_for_personal_grants( client, db_session, employee_user, employee_headers, ) -> None: employee, _ = employee_user resource = await _add_resource( db_session, owner_person_id=employee.id, dataset_id="finance-sensitive", ) response = await client.get( "/api/v1/data-access/requestable-resources", headers=employee_headers, ) assert response.status_code == 200 rows = response.json()["data"] assert [row["id"] for row in rows] == [str(resource.id)] assert rows[0]["dataset_id"] == "finance-sensitive" async def test_requestable_resources_searches_catalog_table_name_and_description( client, db_session, employee_user, employee_headers, monkeypatch, ) -> None: employee, _ = employee_user resource = await _add_resource( db_session, owner_person_id=employee.id, dataset_id="finance", ) await _add_resource( db_session, owner_person_id=employee.id, dataset_id="unrelated", ) table_match = router_module.DataAccessLineageCatalogSearchMatch( project_id=resource.project_id, dataset_id=resource.dataset_id, table=router_module.DataAccessLineageCatalogTable( table_id="account_daily", table_type="TABLE", description="Daily account snapshot for Finance.", ), ) def _search(search: str): assert search in {"account_daily", "daily account snapshot"} return (table_match,) monkeypatch.setattr(router_module, "search_lineage_catalog_tables", _search) for search in ("account_daily", "daily account snapshot"): response = await client.get( "/api/v1/data-access/requestable-resources", params={"search": search}, headers=employee_headers, ) assert response.status_code == 200, response.text payload = response.json() assert payload["meta"]["total"] == 1 assert [row["id"] for row in payload["data"]] == [str(resource.id)] assert payload["data"][0]["matched_table_count"] == 1 assert payload["data"][0]["matched_tables"] == [{ "table_id": "account_daily", "table_type": "TABLE", "description": "Daily account snapshot for Finance.", "catalog_tier": "lineage", }] async def test_requestable_resources_searches_managed_table_metadata( client, db_session, employee_user, employee_headers, monkeypatch, ) -> None: employee, _ = employee_user dataset = await _add_resource( db_session, owner_person_id=employee.id, dataset_id="finance", ) db_session.add(DataAccessResource( integration_id=dataset.integration_id, provider=c.PROVIDER_BIGQUERY, resource_type=c.RESOURCE_TYPE_TABLE, project_id=dataset.project_id, dataset_id=dataset.dataset_id, parent_resource_id=dataset.id, table_id="manual_margin_view", table_type="VIEW", display_name="Manual margin view", requestable=True, status=RESOURCE_STATUS_ACTIVE, owner_person_id=str(employee.id), owner_resolution_status=OWNER_RESOLUTION_RESOLVED, owner_source=OWNER_SOURCE_BOOTSTRAP, source_version="managed-table-fixture", source_etag="managed-table-etag", notes="Quarterly margin review source.", )) await db_session.flush() monkeypatch.setattr( router_module, "search_lineage_catalog_tables", lambda _search: (), ) response = await client.get( "/api/v1/data-access/requestable-resources", params={"search": "quarterly margin"}, headers=employee_headers, ) assert response.status_code == 200, response.text payload = response.json() assert payload["meta"]["total"] == 1 assert payload["data"][0]["id"] == str(dataset.id) assert payload["data"][0]["matched_table_count"] == 1 assert payload["data"][0]["matched_tables"] == [{ "table_id": "manual_margin_view", "table_type": "VIEW", "description": "Quarterly margin review source.", "catalog_tier": "managed", }] async def test_dataset_owner_can_transfer_their_dataset( client, db_session, employee_user, admin_user, employee_headers, ) -> None: employee, _ = employee_user next_owner, _ = admin_user resource = await _add_resource( db_session, owner_person_id=employee.id, dataset_id="finance-transferable", ) response = await client.put( f"/api/v1/data-access/requestable-resources/{resource.id}/owner", headers=employee_headers, json={"owner_person_id": str(next_owner.id)}, ) assert response.status_code == 200, response.text data = response.json()["data"] assert data["id"] == str(resource.id) assert data["owner_person_id"] == str(next_owner.id) assert data["caller_can_manage_owner"] is False async def test_requestable_resources_filter_requestability_for_admin( client, db_session, employee_user, employee_headers, admin_headers, ) -> None: owner, _ = employee_user requestable_resource = await _add_resource( db_session, owner_person_id=owner.id, dataset_id="requestable-filter-visible", ) unavailable_resource = await _add_resource( db_session, owner_person_id=owner.id, dataset_id="requestable-filter-suspended", status=RESOURCE_STATUS_SUSPENDED, ) active_unavailable_resource = await _add_resource( db_session, owner_person_id=owner.id, dataset_id="requestable-filter-active-disabled", requestable=False, ) requestable_response = await client.get( "/api/v1/data-access/requestable-resources?requestability=requestable", headers=admin_headers, ) unavailable_response = await client.get( "/api/v1/data-access/requestable-resources?requestability=not_requestable", headers=admin_headers, ) employee_unavailable_response = await client.get( "/api/v1/data-access/requestable-resources?requestability=not_requestable", headers=employee_headers, ) assert requestable_response.status_code == 200 assert requestable_response.json()["meta"]["total"] == 1 assert [row["id"] for row in requestable_response.json()["data"]] == [ str(requestable_resource.id), ] assert unavailable_response.status_code == 200 assert unavailable_response.json()["meta"]["total"] == 2 assert {row["id"] for row in unavailable_response.json()["data"]} == { str(active_unavailable_resource.id), str(unavailable_resource.id), } assert employee_unavailable_response.status_code == 200 assert employee_unavailable_response.json()["meta"]["total"] == 1 assert [row["id"] for row in employee_unavailable_response.json()["data"]] == [ str(active_unavailable_resource.id), ] async def test_requestable_resources_order_requestable_before_unavailable_across_pages( client, db_session, employee_user, employee_headers, ) -> None: owner, _ = employee_user requestable_resource = await _add_resource( db_session, owner_person_id=owner.id, dataset_id="zzz-requestable-after-unavailable-alphabetically", ) unavailable_resource = await _add_resource( db_session, owner_person_id=owner.id, dataset_id="aaa-unavailable-before-requestable-alphabetically", requestable=False, ) first_page = await client.get( "/api/v1/data-access/requestable-resources?page=1&page_size=1", headers=employee_headers, ) second_page = await client.get( "/api/v1/data-access/requestable-resources?page=2&page_size=1", headers=employee_headers, ) assert first_page.status_code == 200 assert first_page.json()["meta"]["total"] == 2 assert [row["id"] for row in first_page.json()["data"]] == [ str(requestable_resource.id), ] assert second_page.status_code == 200 assert [row["id"] for row in second_page.json()["data"]] == [ str(unavailable_resource.id), ] async def test_requestable_reads_allow_group_member( client, db_session, employee_user, employee_headers, monkeypatch: pytest.MonkeyPatch, ) -> None: employee, _ = employee_user member_group = await _add_group( db_session, owner_person_id=None, email="member-readers@example.com", ) visible_resource = await _add_resource( db_session, owner_person_id=employee.id, dataset_id="member-visible", ) async def fake_member_groups(self, *, person): assert person.id == employee.id return SimpleNamespace( groups=[SimpleNamespace(group_id=member_group.id)], ) monkeypatch.setattr( EmailGroupService, "list_member_groups_for_person", fake_member_groups, ) principals = await client.get( "/api/v1/data-access/requestable-principals", headers=employee_headers, ) assert principals.status_code == 200 principal_rows = principals.json()["data"] assert [row["id"] for row in principal_rows] == [str(member_group.id)] assert principal_rows[0]["caller_is_owner"] is False assert principal_rows[0]["caller_is_member"] is True resources = await client.get( "/api/v1/data-access/requestable-resources", headers=employee_headers, ) assert resources.status_code == 200 resource_rows = resources.json()["data"] assert [row["id"] for row in resource_rows] == [str(visible_resource.id)] async def test_group_grant_submit_endpoint_is_not_registered( client, employee_headers, ) -> None: response = await client.post( "/api/v1/data-access/requests/group-grants", headers=employee_headers, json={}, ) assert response.status_code in {404, 405} async def test_personal_grant_submit_endpoint_derives_the_authenticated_employee( client, db_session, employee_user, employee_headers, monkeypatch: pytest.MonkeyPatch, ) -> None: requester, _ = employee_user await _add_chat_flow( db_session, flow_key=c.APPROVAL_FLOW_DATA_ACCESS_PERSONAL_GRANT, ) await _add_entitlement(db_session) resource = await _add_resource( db_session, owner_person_id=requester.id, dataset_id="personal_finance", ) fake_client = _FakeFeishuClient() async def fake_service_client(self, db): return fake_client monkeypatch.setattr( DataAccessService, "_data_access_feishu_client", fake_service_client, ) response = await client.post( "/api/v1/data-access/requests/personal-grants", headers={ **employee_headers, "Idempotency-Key": "dax-personal-router-submit", }, json={ "resource_ids": [str(resource.id)], "business_reason": "Validate the monthly finance close metrics", "detail_url": "https://helix.example/data-access/requests/new", }, ) assert response.status_code == 201, response.text payload = response.json() assert payload["success"] is True assert payload["data"]["dataset_count"] == 1 request = await db_session.get(DataAccessRequest, payload["data"]["request_id"]) assert request is not None assert request.principal_type_snapshot == c.PRINCIPAL_TYPE_GOOGLE_USER assert request.principal_subject_snapshot == str(requester.id) assert request.principal_email_snapshot == requester.work_email assert request.principal_id is not None assert len(fake_client.instances) == 1 assert "employee@test.com" in "\n".join( row["value"] for row in json.loads(fake_client.instances[0]["form"]) ) async def test_personal_grant_submit_endpoint_accepts_dataset_and_table_resources_together( client, db_session, employee_user, employee_headers, monkeypatch: pytest.MonkeyPatch, ) -> None: requester, _ = employee_user await _add_chat_flow( db_session, flow_key=c.APPROVAL_FLOW_DATA_ACCESS_PERSONAL_GRANT, ) await _add_entitlement(db_session) whole_dataset = await _add_resource( db_session, owner_person_id=requester.id, dataset_id="whole_finance", ) table_parent = await _add_resource( db_session, owner_person_id=requester.id, dataset_id="warehouse", ) fake_client = _FakeFeishuClient() async def fake_service_client(self, db): return fake_client async def fake_schema_driver(db): return _FakeTableSelectionDriver() monkeypatch.setattr( DataAccessService, "_data_access_feishu_client", fake_service_client, ) monkeypatch.setattr( router_module, "_dataset_schema_driver", fake_schema_driver, ) response = await client.post( "/api/v1/data-access/requests/personal-grants", headers={ **employee_headers, "Idempotency-Key": "dax-personal-mixed-resource-submit", }, json={ "resource_ids": [str(whole_dataset.id)], "table_selections": [{ "dataset_resource_id": str(table_parent.id), "table_id": "account_daily", }], "business_reason": "Compare finance totals with the daily account snapshot", "detail_url": "https://helix.example/data-access/requests/new", }, ) assert response.status_code == 201, response.text payload = response.json()["data"] assert payload["dataset_count"] == 2 request_id = uuid.UUID(payload["request_id"]) resource_ids = list((await db_session.scalars( select(DataAccessRequestItem.resource_id).where( DataAccessRequestItem.request_id == request_id, ), )).all()) resources = list((await db_session.scalars( select(DataAccessResource).where(DataAccessResource.id.in_(resource_ids)), )).all()) assert { (resource.resource_type, resource.dataset_id, resource.table_id) for resource in resources } == { (c.RESOURCE_TYPE_DATASET, "whole_finance", None), (c.RESOURCE_TYPE_TABLE, "warehouse", "account_daily"), } async def test_request_detail_endpoint_returns_feishu_preview_and_apply_status( client, db_session, employee_user, employee_headers, ) -> None: requester, _ = employee_user entitlement = await _add_entitlement(db_session) resource = await _add_resource( db_session, owner_person_id=requester.id, dataset_id="finance_sensitive", ) group = await _add_group( db_session, owner_person_id=requester.id, email="finance-analytics@example.com", ) now = datetime.now(UTC) run_id = uuid.uuid4() approval = ApprovalRequest( external_id="dax-detail-feishu-001", approval_type=c.APPROVAL_FLOW_DATA_ACCESS_GROUP_GRANT, applicant_id=requester.id, status="approved", source_payload={"data_access_request_id": "pending-flush"}, ) db_session.add(approval) await db_session.flush() request_row = DataAccessRequest( requester_person_id=requester.id, principal_email_group_id=group.id, principal_google_group_id_snapshot=group.google_group_id, principal_email_snapshot=group.email, entitlement_id=entitlement.id, entitlement_snapshot=c.ENTITLEMENT_READ, entitlement_version_snapshot=ENTITLEMENT_VERSION_V1, approval_flow_key=c.APPROVAL_FLOW_DATA_ACCESS_GROUP_GRANT, feishu_instance_code=approval.external_id, approval_request_id=approval.id, business_reason="Quarter-end finance review", status=REQUEST_STATUS_APPROVED, reconcile_status=RECONCILE_STATUS_RETRYABLE, last_reconcile_run_id=run_id, last_reconcile_at=now, last_reconcile_error="provider auth denied", trace_id="trace-dax-detail", decided_at=now, ) db_session.add(request_row) await db_session.flush() approval.source_payload = {"data_access_request_id": str(request_row.id)} item = DataAccessRequestItem( request_id=request_row.id, resource_id=resource.id, line_no=1, resource_project_id_snapshot=resource.project_id, resource_dataset_id_snapshot=resource.dataset_id, resource_source_version_snapshot=resource.source_version, resource_source_etag_snapshot=resource.source_etag, resource_owner_person_id_snapshot=requester.id, status=REQUEST_ITEM_STATUS_APPROVED, ) db_session.add(item) await db_session.flush() binding = DataAccessBinding( resource_id=resource.id, principal_email_group_id=group.id, principal_google_group_id_snapshot=group.google_group_id, principal_email_snapshot=group.email, entitlement_id=entitlement.id, entitlement_snapshot=c.ENTITLEMENT_READ, entitlement_version_snapshot=ENTITLEMENT_VERSION_V1, source_request_id=request_row.id, source_request_item_id=item.id, status="active", reconcile_status=RECONCILE_STATUS_RETRYABLE, last_reconcile_run_id=run_id, last_reconcile_at=now, last_reconcile_error="provider auth denied", ) db_session.add(binding) await db_session.flush() effect = DataAccessCompiledEffect( binding_id=binding.id, resource_id=resource.id, principal_email_group_id=group.id, entitlement_id=entitlement.id, provider=c.PROVIDER_BIGQUERY, resource_type=c.RESOURCE_TYPE_DATASET, target_project_id=resource.project_id, target_dataset_id=resource.dataset_id, provider_member=f"{c.BIGQUERY_GROUP_MEMBER_PREFIX}{group.email}", provider_role=c.BIGQUERY_DATA_VIEWER_ROLE, desired_state=EFFECT_DESIRED_PRESENT, entitlement_version_snapshot=ENTITLEMENT_VERSION_V1, resource_source_version_snapshot=resource.source_version, resource_source_etag_snapshot=resource.source_etag, compiler_version=DATA_ACCESS_COMPILER_VERSION, managed_reason=EFFECT_MANAGED_REASON_GRANT_REQUEST, status=EFFECT_STATUS_FAILED, last_provider_outcome="auth_error", last_error="403 permission denied", ) db_session.add(effect) await db_session.flush() db_session.add( DataAccessEffectRun( run_id=run_id, effect_id=effect.id, binding_id=binding.id, request_id=request_row.id, resource_id=resource.id, integration_id=resource.integration_id, trace_id="trace-dax-detail", mode="apply", provider=c.PROVIDER_BIGQUERY, resource_type=c.RESOURCE_TYPE_DATASET, target_project_id=resource.project_id, target_dataset_id=resource.dataset_id, provider_member=effect.provider_member, provider_role=effect.provider_role, desired_state=effect.desired_state, effect_status=EFFECT_STATUS_FAILED, provider_outcome="auth_error", error_message="403 permission denied", started_at=now, completed_at=now, ), ) await db_session.flush() response = await client.get( f"/api/v1/data-access/requests/{request_row.id}", headers=employee_headers, ) assert response.status_code == 200, response.text payload = response.json() assert payload["success"] is True data = payload["data"] assert data["id"] == str(request_row.id) assert data["status"] == REQUEST_STATUS_APPROVED assert data["approval"]["external_id"] == approval.external_id assert data["approval"]["status"] == "approved" assert data["principal_email_snapshot"] == group.email assert data["provider_role"] == c.BIGQUERY_DATA_VIEWER_ROLE assert data["last_reconcile_error"] == "provider auth denied" assert data["item_count"] == 1 detail_item = data["items"][0] assert detail_item["display_state"] == "failed" assert detail_item["locator"] == "helix-data-prod.finance_sensitive" assert detail_item["item_status"] == REQUEST_ITEM_STATUS_APPROVED assert detail_item["binding_reconcile_status"] == RECONCILE_STATUS_RETRYABLE assert detail_item["compiled_preview"]["provider_member"] == ( f"{c.BIGQUERY_GROUP_MEMBER_PREFIX}{group.email}" ) assert detail_item["compiled_preview"]["provider_role"] == ( c.BIGQUERY_DATA_VIEWER_ROLE ) assert detail_item["effect"]["status"] == EFFECT_STATUS_FAILED assert detail_item["effect"]["last_error"] == "403 permission denied" assert detail_item["effect"]["runs"][0]["provider_outcome"] == "auth_error" async def test_request_detail_endpoint_hides_unrelated_request_from_employee( client, db_session, employee_user, manager_and_subordinate, ) -> None: requester, _ = employee_user manager, _, manager_token = manager_and_subordinate entitlement = await _add_entitlement(db_session) resource = await _add_resource( db_session, owner_person_id=requester.id, dataset_id="payroll_sensitive", ) group = await _add_group( db_session, owner_person_id=requester.id, email="payroll-readers@example.com", ) request_row = DataAccessRequest( requester_person_id=requester.id, principal_email_group_id=group.id, principal_google_group_id_snapshot=group.google_group_id, principal_email_snapshot=group.email, entitlement_id=entitlement.id, entitlement_snapshot=c.ENTITLEMENT_READ, entitlement_version_snapshot=ENTITLEMENT_VERSION_V1, approval_flow_key=c.APPROVAL_FLOW_DATA_ACCESS_GROUP_GRANT, business_reason="Payroll analytics", status=REQUEST_STATUS_PENDING_FEISHU, ) db_session.add(request_row) await db_session.flush() db_session.add( DataAccessRequestItem( request_id=request_row.id, resource_id=resource.id, line_no=1, resource_project_id_snapshot=resource.project_id, resource_dataset_id_snapshot=resource.dataset_id, resource_source_version_snapshot=resource.source_version, resource_source_etag_snapshot=resource.source_etag, resource_owner_person_id_snapshot=requester.id, ), ) await db_session.flush() response = await client.get( f"/api/v1/data-access/requests/{request_row.id}", headers={"Authorization": f"Bearer {manager_token}"}, ) assert manager.id != requester.id assert response.status_code == 404 assert response.json()["errors"][0]["code"] == "DATA_ACCESS_REQUEST_NOT_FOUND" assert "payroll_sensitive" not in response.text assert "payroll-readers" not in response.text async def test_request_detail_endpoint_scopes_resource_owner_to_owned_items( client, db_session, employee_user, manager_and_subordinate, ) -> None: requester, _ = employee_user resource_owner, _, resource_owner_token = manager_and_subordinate entitlement = await _add_entitlement(db_session) visible_resource = await _add_resource( db_session, owner_person_id=resource_owner.id, dataset_id="finance_visible", ) hidden_resource = await _add_resource( db_session, owner_person_id=requester.id, dataset_id="payroll_hidden", ) group = await _add_group( db_session, owner_person_id=requester.id, email="mixed-request-readers@example.com", ) now = datetime.now(UTC) visible_run_id = uuid.uuid4() hidden_run_id = uuid.uuid4() approval = ApprovalRequest( external_id="dax-detail-feishu-mixed", approval_type=c.APPROVAL_FLOW_DATA_ACCESS_GROUP_GRANT, applicant_id=requester.id, status="approved", source_payload={"data_access_request_id": "pending-flush"}, ) db_session.add(approval) await db_session.flush() request_row = DataAccessRequest( requester_person_id=requester.id, principal_email_group_id=group.id, principal_google_group_id_snapshot=group.google_group_id, principal_email_snapshot=group.email, entitlement_id=entitlement.id, entitlement_snapshot=c.ENTITLEMENT_READ, entitlement_version_snapshot=ENTITLEMENT_VERSION_V1, approval_flow_key=c.APPROVAL_FLOW_DATA_ACCESS_GROUP_GRANT, feishu_instance_code=approval.external_id, approval_request_id=approval.id, business_reason="Need finance_visible and payroll_hidden analytics", status=REQUEST_STATUS_APPROVED, reconcile_status=RECONCILE_STATUS_RETRYABLE, last_reconcile_run_id=hidden_run_id, last_reconcile_at=now, last_reconcile_error="payroll_hidden provider denied", trace_id="trace-dax-mixed-detail", decided_at=now, ) db_session.add(request_row) await db_session.flush() approval.source_payload = {"data_access_request_id": str(request_row.id)} visible_item = DataAccessRequestItem( request_id=request_row.id, resource_id=visible_resource.id, line_no=1, resource_project_id_snapshot=visible_resource.project_id, resource_dataset_id_snapshot=visible_resource.dataset_id, resource_source_version_snapshot=visible_resource.source_version, resource_source_etag_snapshot=visible_resource.source_etag, resource_owner_person_id_snapshot=resource_owner.id, status=REQUEST_ITEM_STATUS_APPROVED, ) hidden_item = DataAccessRequestItem( request_id=request_row.id, resource_id=hidden_resource.id, line_no=2, resource_project_id_snapshot=hidden_resource.project_id, resource_dataset_id_snapshot=hidden_resource.dataset_id, resource_source_version_snapshot=hidden_resource.source_version, resource_source_etag_snapshot=hidden_resource.source_etag, resource_owner_person_id_snapshot=requester.id, status=REQUEST_ITEM_STATUS_APPROVED, ) db_session.add_all([visible_item, hidden_item]) await db_session.flush() visible_binding = DataAccessBinding( resource_id=visible_resource.id, principal_email_group_id=group.id, principal_google_group_id_snapshot=group.google_group_id, principal_email_snapshot=group.email, entitlement_id=entitlement.id, entitlement_snapshot=c.ENTITLEMENT_READ, entitlement_version_snapshot=ENTITLEMENT_VERSION_V1, source_request_id=request_row.id, source_request_item_id=visible_item.id, status="active", reconcile_status=RECONCILE_STATUS_APPLIED, last_reconcile_run_id=visible_run_id, last_reconcile_at=now, ) hidden_binding = DataAccessBinding( resource_id=hidden_resource.id, principal_email_group_id=group.id, principal_google_group_id_snapshot=group.google_group_id, principal_email_snapshot=group.email, entitlement_id=entitlement.id, entitlement_snapshot=c.ENTITLEMENT_READ, entitlement_version_snapshot=ENTITLEMENT_VERSION_V1, source_request_id=request_row.id, source_request_item_id=hidden_item.id, status="active", reconcile_status=RECONCILE_STATUS_RETRYABLE, last_reconcile_run_id=hidden_run_id, last_reconcile_at=now, last_reconcile_error="payroll_hidden binding denied", ) db_session.add_all([visible_binding, hidden_binding]) await db_session.flush() visible_effect = DataAccessCompiledEffect( binding_id=visible_binding.id, resource_id=visible_resource.id, principal_email_group_id=group.id, entitlement_id=entitlement.id, provider=c.PROVIDER_BIGQUERY, resource_type=c.RESOURCE_TYPE_DATASET, target_project_id=visible_resource.project_id, target_dataset_id=visible_resource.dataset_id, provider_member=f"{c.BIGQUERY_GROUP_MEMBER_PREFIX}{group.email}", provider_role=c.BIGQUERY_DATA_VIEWER_ROLE, desired_state=EFFECT_DESIRED_PRESENT, entitlement_version_snapshot=ENTITLEMENT_VERSION_V1, resource_source_version_snapshot=visible_resource.source_version, resource_source_etag_snapshot=visible_resource.source_etag, compiler_version=DATA_ACCESS_COMPILER_VERSION, managed_reason=EFFECT_MANAGED_REASON_GRANT_REQUEST, status=EFFECT_STATUS_APPLIED, last_provider_outcome="applied", updated_at=now, ) hidden_effect = DataAccessCompiledEffect( binding_id=hidden_binding.id, resource_id=hidden_resource.id, principal_email_group_id=group.id, entitlement_id=entitlement.id, provider=c.PROVIDER_BIGQUERY, resource_type=c.RESOURCE_TYPE_DATASET, target_project_id=hidden_resource.project_id, target_dataset_id=hidden_resource.dataset_id, provider_member=f"{c.BIGQUERY_GROUP_MEMBER_PREFIX}{group.email}", provider_role=c.BIGQUERY_DATA_VIEWER_ROLE, desired_state=EFFECT_DESIRED_PRESENT, entitlement_version_snapshot=ENTITLEMENT_VERSION_V1, resource_source_version_snapshot=hidden_resource.source_version, resource_source_etag_snapshot=hidden_resource.source_etag, compiler_version=DATA_ACCESS_COMPILER_VERSION, managed_reason=EFFECT_MANAGED_REASON_GRANT_REQUEST, status=EFFECT_STATUS_FAILED, last_provider_outcome="auth_error", last_error="payroll_hidden provider denied", updated_at=now, ) db_session.add_all([visible_effect, hidden_effect]) await db_session.flush() db_session.add_all([ DataAccessEffectRun( run_id=visible_run_id, effect_id=visible_effect.id, binding_id=visible_binding.id, request_id=request_row.id, resource_id=visible_resource.id, integration_id=visible_resource.integration_id, trace_id="trace-dax-visible", mode="apply", provider=c.PROVIDER_BIGQUERY, resource_type=c.RESOURCE_TYPE_DATASET, target_project_id=visible_resource.project_id, target_dataset_id=visible_resource.dataset_id, provider_member=visible_effect.provider_member, provider_role=visible_effect.provider_role, desired_state=visible_effect.desired_state, effect_status=EFFECT_STATUS_APPLIED, provider_outcome="applied", error_message=None, started_at=now, completed_at=now, ), DataAccessEffectRun( run_id=hidden_run_id, effect_id=hidden_effect.id, binding_id=hidden_binding.id, request_id=request_row.id, resource_id=hidden_resource.id, integration_id=hidden_resource.integration_id, trace_id="trace-dax-hidden", mode="apply", provider=c.PROVIDER_BIGQUERY, resource_type=c.RESOURCE_TYPE_DATASET, target_project_id=hidden_resource.project_id, target_dataset_id=hidden_resource.dataset_id, provider_member=hidden_effect.provider_member, provider_role=hidden_effect.provider_role, desired_state=hidden_effect.desired_state, effect_status=EFFECT_STATUS_FAILED, provider_outcome="auth_error", error_message="payroll_hidden provider denied", started_at=now, completed_at=now, ), ]) await db_session.flush() response = await client.get( f"/api/v1/data-access/requests/{request_row.id}", headers={"Authorization": f"Bearer {resource_owner_token}"}, ) assert response.status_code == 200, response.text data = response.json()["data"] assert resource_owner.id != requester.id assert data["item_count"] == 1 assert data["reconcile_status"] == RECONCILE_STATUS_APPLIED assert data["last_reconcile_run_id"] == str(visible_run_id) assert data["last_reconcile_error"] is None assert data["business_reason"] is None assert data["trace_id"] is None assert [item["locator"] for item in data["items"]] == [ "helix-data-prod.finance_visible", ] assert data["items"][0]["display_state"] == "active" assert data["items"][0]["effect"]["runs"][0]["run_id"] == str(visible_run_id) assert "payroll_hidden" not in response.text assert "trace-dax-mixed-detail" not in response.text assert "trace-dax-hidden" not in response.text assert str(hidden_run_id) not in response.text async def test_group_access_endpoint_returns_active_revoked_history_for_owned_group_and_admin( client, db_session, employee_user, employee_headers, admin_headers, ) -> None: requester, _ = employee_user entitlement = await _add_entitlement(db_session) active_resource = await _add_resource( db_session, owner_person_id=requester.id, dataset_id="finance_active", ) revoke_pending_resource = await _add_resource( db_session, owner_person_id=requester.id, dataset_id="finance_revoke_pending", ) history_resource = await _add_resource( db_session, owner_person_id=requester.id, dataset_id="finance_history", ) group = await _add_group( db_session, owner_person_id=requester.id, email="finance-readers@example.com", ) now = datetime.now(UTC) run_id = uuid.uuid4() active_binding = DataAccessBinding( resource_id=active_resource.id, principal_email_group_id=group.id, principal_google_group_id_snapshot=group.google_group_id, principal_email_snapshot=group.email, entitlement_id=entitlement.id, entitlement_snapshot=c.ENTITLEMENT_READ, entitlement_version_snapshot=ENTITLEMENT_VERSION_V1, status=BINDING_STATUS_ACTIVE, reconcile_status=RECONCILE_STATUS_APPLIED, last_reconcile_run_id=run_id, last_reconcile_at=now, ) revoke_pending_binding = DataAccessBinding( resource_id=revoke_pending_resource.id, principal_email_group_id=group.id, principal_google_group_id_snapshot=group.google_group_id, principal_email_snapshot=group.email, entitlement_id=entitlement.id, entitlement_snapshot=c.ENTITLEMENT_READ, entitlement_version_snapshot=ENTITLEMENT_VERSION_V1, status=BINDING_STATUS_REVOKED_PENDING, reconcile_status="pending_reconcile", last_reconcile_at=now, ) history_binding = DataAccessBinding( resource_id=history_resource.id, principal_email_group_id=group.id, principal_google_group_id_snapshot=group.google_group_id, principal_email_snapshot=group.email, entitlement_id=entitlement.id, entitlement_snapshot=c.ENTITLEMENT_READ, entitlement_version_snapshot=ENTITLEMENT_VERSION_V1, status=BINDING_STATUS_REVOKED, reconcile_status=RECONCILE_STATUS_APPLIED, last_reconcile_at=now, revoked_at=now, revoke_reason="owner_request", ) db_session.add_all([ active_binding, revoke_pending_binding, history_binding, ]) await db_session.flush() active_effect = DataAccessCompiledEffect( binding_id=active_binding.id, resource_id=active_resource.id, principal_email_group_id=group.id, entitlement_id=entitlement.id, provider=c.PROVIDER_BIGQUERY, resource_type=c.RESOURCE_TYPE_DATASET, target_project_id=active_resource.project_id, target_dataset_id=active_resource.dataset_id, provider_member=f"{c.BIGQUERY_GROUP_MEMBER_PREFIX}{group.email}", provider_role=c.BIGQUERY_DATA_VIEWER_ROLE, desired_state=EFFECT_DESIRED_PRESENT, entitlement_version_snapshot=ENTITLEMENT_VERSION_V1, resource_source_version_snapshot=active_resource.source_version, resource_source_etag_snapshot=active_resource.source_etag, compiler_version=DATA_ACCESS_COMPILER_VERSION, managed_reason=EFFECT_MANAGED_REASON_GRANT_REQUEST, already_present_at_first_apply=True, status=EFFECT_STATUS_APPLIED, last_provider_outcome="applied", updated_at=now, ) revoke_pending_effect = DataAccessCompiledEffect( binding_id=revoke_pending_binding.id, resource_id=revoke_pending_resource.id, principal_email_group_id=group.id, entitlement_id=entitlement.id, provider=c.PROVIDER_BIGQUERY, resource_type=c.RESOURCE_TYPE_DATASET, target_project_id=revoke_pending_resource.project_id, target_dataset_id=revoke_pending_resource.dataset_id, provider_member=f"{c.BIGQUERY_GROUP_MEMBER_PREFIX}{group.email}", provider_role=c.BIGQUERY_DATA_VIEWER_ROLE, desired_state=EFFECT_DESIRED_ABSENT, entitlement_version_snapshot=ENTITLEMENT_VERSION_V1, resource_source_version_snapshot=revoke_pending_resource.source_version, resource_source_etag_snapshot=revoke_pending_resource.source_etag, compiler_version=DATA_ACCESS_COMPILER_VERSION, managed_reason=EFFECT_MANAGED_REASON_REVOKE_REQUEST, status=EFFECT_STATUS_PENDING_RECONCILE, updated_at=now, ) history_effect = DataAccessCompiledEffect( binding_id=history_binding.id, resource_id=history_resource.id, principal_email_group_id=group.id, entitlement_id=entitlement.id, provider=c.PROVIDER_BIGQUERY, resource_type=c.RESOURCE_TYPE_DATASET, target_project_id=history_resource.project_id, target_dataset_id=history_resource.dataset_id, provider_member=f"{c.BIGQUERY_GROUP_MEMBER_PREFIX}{group.email}", provider_role=c.BIGQUERY_DATA_VIEWER_ROLE, desired_state=EFFECT_DESIRED_ABSENT, entitlement_version_snapshot=ENTITLEMENT_VERSION_V1, resource_source_version_snapshot=history_resource.source_version, resource_source_etag_snapshot=history_resource.source_etag, compiler_version=DATA_ACCESS_COMPILER_VERSION, managed_reason=EFFECT_MANAGED_REASON_REVOKE_REQUEST, status=EFFECT_STATUS_APPLIED, last_provider_outcome="removed", updated_at=now, ) db_session.add_all([active_effect, revoke_pending_effect, history_effect]) await db_session.flush() db_session.add( DataAccessEffectRun( run_id=run_id, effect_id=active_effect.id, binding_id=active_binding.id, request_id=None, resource_id=active_resource.id, integration_id=active_resource.integration_id, trace_id="trace-dax-group-access", mode="apply", provider=c.PROVIDER_BIGQUERY, resource_type=c.RESOURCE_TYPE_DATASET, target_project_id=active_resource.project_id, target_dataset_id=active_resource.dataset_id, provider_member=active_effect.provider_member, provider_role=active_effect.provider_role, desired_state=active_effect.desired_state, effect_status=EFFECT_STATUS_APPLIED, provider_outcome="applied", provider_etag="etag-active", started_at=now, completed_at=now, ), ) await db_session.flush() response = await client.get( f"/api/v1/data-access/principals/{group.id}/access", headers=employee_headers, ) assert response.status_code == 200, response.text data = response.json()["data"] assert data["principal"]["email"] == group.email assert data["principal"]["caller_is_owner"] is True assert data["summary"] == { "total": 3, "active": 1, "pending_apply": 0, "failed": 0, "revoked_pending": 1, "revoked_history": 1, } assert [item["access_state"] for item in data["items"]] == [ "active", "revoked_pending", "revoked_history", ] assert {item["principal_type"] for item in data["items"]} == { c.PRINCIPAL_TYPE_GOOGLE_GROUP, } assert {item["entitlement"] for item in data["items"]} == { c.ENTITLEMENT_READ, } assert data["items"][0]["latest_run"]["trace_id"] == "trace-dax-group-access" assert "user:" not in response.text admin_response = await client.get( f"/api/v1/data-access/principals/{group.id}/access", headers=admin_headers, ) assert admin_response.status_code == 200, admin_response.text assert admin_response.json()["data"]["principal"]["caller_is_owner"] is False async def test_my_dataset_grants_are_requester_scoped_and_identify_principal_kind( client, db_session, employee_user, employee_headers, manager_and_subordinate, ) -> None: requester, _ = employee_user other_requester, _, other_token = manager_and_subordinate entitlement = await _add_entitlement(db_session) group = await _add_group( db_session, owner_person_id=requester.id, email="finance-readers@example.com", ) async def add_grant( *, requester_person_id: uuid.UUID, dataset_id: str, principal_type: str, principal_email: str, principal_email_group_id: uuid.UUID | None = None, principal_subject: str | None = None, ) -> DataAccessBinding: resource = await _add_resource( db_session, owner_person_id=requester.id, dataset_id=dataset_id, notes=f"Business description for {dataset_id}", ) request_row = DataAccessRequest( requester_person_id=requester_person_id, principal_email_group_id=principal_email_group_id, principal_type_snapshot=principal_type, principal_subject_snapshot=principal_subject, principal_google_group_id_snapshot=( group.google_group_id if principal_email_group_id else None ), principal_email_snapshot=principal_email, entitlement_id=entitlement.id, entitlement_snapshot=c.ENTITLEMENT_READ, entitlement_version_snapshot=ENTITLEMENT_VERSION_V1, status=REQUEST_STATUS_APPROVED, reconcile_status=RECONCILE_STATUS_APPLIED, ) db_session.add(request_row) await db_session.flush() binding = DataAccessBinding( resource_id=resource.id, principal_email_group_id=principal_email_group_id, principal_type_snapshot=principal_type, principal_subject_snapshot=principal_subject, principal_google_group_id_snapshot=( group.google_group_id if principal_email_group_id else None ), principal_email_snapshot=principal_email, entitlement_id=entitlement.id, entitlement_snapshot=c.ENTITLEMENT_READ, entitlement_version_snapshot=ENTITLEMENT_VERSION_V1, source_request_id=request_row.id, status=BINDING_STATUS_ACTIVE, reconcile_status=RECONCILE_STATUS_APPLIED, ) db_session.add(binding) await db_session.flush() return binding await add_grant( requester_person_id=requester.id, dataset_id="finance_personal", principal_type=c.PRINCIPAL_TYPE_GOOGLE_USER, principal_email=requester.work_email, principal_subject=str(requester.id), ) await add_grant( requester_person_id=requester.id, dataset_id="finance_group", principal_type=c.PRINCIPAL_TYPE_GOOGLE_GROUP, principal_email=group.email, principal_email_group_id=group.id, principal_subject=group.google_group_id, ) await add_grant( requester_person_id=requester.id, dataset_id="finance_service_account", principal_type=c.PRINCIPAL_TYPE_GCP_SERVICE_ACCOUNT, principal_email="finance-reader@analytics-prod.iam.gserviceaccount.com", principal_subject="finance-reader@analytics-prod.iam.gserviceaccount.com", ) await add_grant( requester_person_id=other_requester.id, dataset_id="other_requester_only", principal_type=c.PRINCIPAL_TYPE_GCP_SERVICE_ACCOUNT, principal_email="other-reader@analytics-prod.iam.gserviceaccount.com", principal_subject="other-reader@analytics-prod.iam.gserviceaccount.com", ) response = await client.get( "/api/v1/data-access/me/grants?page=1&page_size=2", headers=employee_headers, ) assert response.status_code == 200, response.text body = response.json() assert body["meta"]["total"] == 3 assert body["meta"]["page"] == 1 assert body["meta"]["page_size"] == 2 assert len(body["data"]) == 2 all_response = await client.get( "/api/v1/data-access/me/grants?page=1&page_size=20", headers=employee_headers, ) assert all_response.status_code == 200, all_response.text rows = all_response.json()["data"] assert {row["dataset_id"] for row in rows} == { "finance_personal", "finance_group", "finance_service_account", } assert {row["principal_type"] for row in rows} == { c.PRINCIPAL_TYPE_GOOGLE_USER, c.PRINCIPAL_TYPE_GOOGLE_GROUP, c.PRINCIPAL_TYPE_GCP_SERVICE_ACCOUNT, } assert {row["provider_member"].split(":", 1)[0] for row in rows} == { "user", "group", "serviceAccount", } assert all(row["source_request_id"] for row in rows) assert all(row["source_request_submitted_at"] for row in rows) assert all(row["access_state"] == "active" for row in rows) assert all(row["description"].startswith("Business description") for row in rows) assert "other_requester_only" not in all_response.text service_account_response = await client.get( "/api/v1/data-access/me/grants", params={ "principal_type": c.PRINCIPAL_TYPE_GCP_SERVICE_ACCOUNT, "search": "finance-reader@analytics-prod", }, headers=employee_headers, ) assert service_account_response.status_code == 200 assert service_account_response.json()["meta"]["total"] == 1 assert service_account_response.json()["data"][0]["dataset_id"] == ( "finance_service_account" ) other_response = await client.get( "/api/v1/data-access/me/grants", headers={"Authorization": f"Bearer {other_token}"}, ) assert other_response.status_code == 200 assert other_response.json()["meta"]["total"] == 1 assert other_response.json()["data"][0]["dataset_id"] == ( "other_requester_only" ) async def test_group_access_endpoint_marks_suspended_resource_notice_for_owner( client, db_session, employee_user, employee_headers, ) -> None: requester, _ = employee_user entitlement = await _add_entitlement(db_session) resource = await _add_resource( db_session, owner_person_id=requester.id, dataset_id="catalog_missing_group_access", status=RESOURCE_STATUS_SUSPENDED, ) group = await _add_group( db_session, owner_person_id=requester.id, email="suspended-access-readers@example.com", ) await _add_binding_effect( db_session, resource=resource, group=group, entitlement=entitlement, status=EFFECT_STATUS_APPLIED, reconcile_status=RECONCILE_STATUS_APPLIED, last_provider_outcome="applied", last_error=None, ) response = await client.get( f"/api/v1/data-access/principals/{group.id}/access", headers=employee_headers, ) assert response.status_code == 200, response.text item = response.json()["data"]["items"][0] assert item["resource_status"] == RESOURCE_STATUS_SUSPENDED assert item["resource_requestable"] is True assert item["revoke_reason"] is None assert item["notice_code"] == "catalog_missing_suspension" assert item["access_state"] == "active" async def test_group_access_endpoint_marks_group_deleted_revocation_admin_only( client, db_session, employee_user, employee_headers, admin_headers, ) -> None: requester, _ = employee_user entitlement = await _add_entitlement(db_session) resource = await _add_resource( db_session, owner_person_id=requester.id, dataset_id="deleted_group_revocation", ) group = await _add_group( db_session, owner_person_id=requester.id, email="deleted-dax-readers@example.com", status="deleted", ) binding, _ = await _add_binding_effect( db_session, resource=resource, group=group, entitlement=entitlement, status=EFFECT_STATUS_PENDING_RECONCILE, desired_state=EFFECT_DESIRED_ABSENT, binding_status=BINDING_STATUS_REVOKED_PENDING, reconcile_status="pending_reconcile", managed_reason=EFFECT_MANAGED_REASON_REVOKE_REQUEST, ) binding.revoke_reason = "group_deleted_without_bridge" binding.revoked_at = datetime.now(UTC) await db_session.flush() admin_principals = await client.get( "/api/v1/data-access/requestable-principals?include_access_history=true", headers=admin_headers, ) assert admin_principals.status_code == 200 principal_rows = { row["id"]: row for row in admin_principals.json()["data"] } assert str(group.id) in principal_rows assert principal_rows[str(group.id)]["status"] == "deleted" request_principals = await client.get( "/api/v1/data-access/requestable-principals", headers=admin_headers, ) assert request_principals.status_code == 200 assert str(group.id) not in { row["id"] for row in request_principals.json()["data"] } employee_principals = await client.get( "/api/v1/data-access/requestable-principals?include_access_history=true", headers=employee_headers, ) assert employee_principals.status_code == 200 assert str(group.id) not in { row["id"] for row in employee_principals.json()["data"] } employee_response = await client.get( f"/api/v1/data-access/principals/{group.id}/access", headers=employee_headers, ) assert employee_response.status_code == 404 assert employee_response.json()["errors"][0]["code"] == ( "DATA_ACCESS_GROUP_ACCESS_NOT_FOUND" ) assert "deleted_group_revocation" not in employee_response.text admin_response = await client.get( f"/api/v1/data-access/principals/{group.id}/access", headers=admin_headers, ) assert admin_response.status_code == 200, admin_response.text data = admin_response.json()["data"] assert data["principal"]["status"] == "deleted" assert data["principal"]["caller_is_owner"] is False item = data["items"][0] assert item["resource_status"] == RESOURCE_STATUS_ACTIVE assert item["resource_requestable"] is True assert item["revoke_reason"] == "group_deleted_without_bridge" assert item["notice_code"] == "group_deleted_revocation" assert item["access_state"] == "revoked_pending" async def test_group_access_endpoint_hides_unowned_group_from_employee( client, db_session, employee_user, manager_and_subordinate, ) -> None: requester, _ = employee_user viewer, _, viewer_token = manager_and_subordinate entitlement = await _add_entitlement(db_session) resource = await _add_resource( db_session, owner_person_id=requester.id, dataset_id="executive_history", ) group = await _add_group( db_session, owner_person_id=requester.id, email="executive-readers@example.com", ) binding = DataAccessBinding( resource_id=resource.id, principal_email_group_id=group.id, principal_google_group_id_snapshot=group.google_group_id, principal_email_snapshot=group.email, entitlement_id=entitlement.id, entitlement_snapshot=c.ENTITLEMENT_READ, entitlement_version_snapshot=ENTITLEMENT_VERSION_V1, status=BINDING_STATUS_ACTIVE, reconcile_status=RECONCILE_STATUS_APPLIED, ) db_session.add(binding) await db_session.flush() response = await client.get( f"/api/v1/data-access/principals/{group.id}/access", headers={"Authorization": f"Bearer {viewer_token}"}, ) assert viewer.id != requester.id assert response.status_code == 404 assert response.json()["errors"][0]["code"] == ( "DATA_ACCESS_GROUP_ACCESS_NOT_FOUND" ) assert "executive_history" not in response.text assert "executive-readers" not in response.text assert str(binding.id) not in response.text async def test_failed_apply_queue_lists_failed_retryable_effects_for_admin( client, db_session, employee_user, employee_headers, admin_headers, ) -> None: owner, _ = employee_user entitlement = await _add_entitlement(db_session) group = await _add_group( db_session, owner_person_id=owner.id, email="ops-readers@example.com", ) failed_resource = await _add_resource( db_session, owner_person_id=owner.id, dataset_id="finance_apply_failed", ) retryable_resource = await _add_resource( db_session, owner_person_id=owner.id, dataset_id="finance_revoke_retryable", ) applied_resource = await _add_resource( db_session, owner_person_id=owner.id, dataset_id="finance_already_applied", ) stale_present_resource = await _add_resource( db_session, owner_person_id=owner.id, dataset_id="finance_stale_present_after_revoke", ) await _add_binding_effect( db_session, resource=failed_resource, group=group, entitlement=entitlement, status=EFFECT_STATUS_FAILED, ) await _add_binding_effect( db_session, resource=retryable_resource, group=group, entitlement=entitlement, status=EFFECT_STATUS_RETRYABLE, desired_state=EFFECT_DESIRED_ABSENT, binding_status=BINDING_STATUS_REVOKED_PENDING, managed_reason=EFFECT_MANAGED_REASON_REVOKE_REQUEST, last_provider_outcome="conflict", last_error="etag changed before apply", ) await _add_binding_effect( db_session, resource=applied_resource, group=group, entitlement=entitlement, status=EFFECT_STATUS_APPLIED, reconcile_status=RECONCILE_STATUS_APPLIED, last_provider_outcome="applied", last_error=None, ) await _add_binding_effect( db_session, resource=stale_present_resource, group=group, entitlement=entitlement, status=EFFECT_STATUS_FAILED, desired_state=EFFECT_DESIRED_PRESENT, binding_status=BINDING_STATUS_REVOKED_PENDING, reconcile_status=RECONCILE_STATUS_FAILED, last_provider_outcome="stale_present_error", last_error="stale present effect after semantic revoke", ) response = await client.get( "/api/v1/data-access/admin/failed-apply", headers=admin_headers, ) assert response.status_code == 200, response.text data = response.json()["data"] assert data["summary"] == { "total": 2, "failed": 1, "retryable": 1, "desired_present": 1, "desired_absent": 1, } assert data["total"] == 2 datasets = {item["dataset_id"] for item in data["items"]} assert datasets == {"finance_apply_failed", "finance_revoke_retryable"} assert {item["principal_email"] for item in data["items"]} == { "ops-readers@example.com", } assert {item["provider_role"] for item in data["items"]} == { c.BIGQUERY_DATA_VIEWER_ROLE, } assert "finance_already_applied" not in response.text assert "finance_stale_present_after_revoke" not in response.text assert "user:" not in response.text employee_response = await client.get( "/api/v1/data-access/admin/failed-apply", headers=employee_headers, ) assert employee_response.status_code == 403 assert employee_response.json()["errors"][0]["code"] == ( "DATA_ACCESS_ADMIN_REQUIRED" ) assert "finance_apply_failed" not in employee_response.text async def test_admin_overview_aggregates_governance_counts_for_admin( client, db_session, employee_user, employee_headers, admin_headers, ) -> None: owner, _ = employee_user entitlement = await _add_entitlement(db_session) integration = Integration( kind=c.DOMAIN_DATA_ACCESS, provider=c.PROVIDER_BIGQUERY, name="DAX BigQuery Overview", slug=f"dax-bq-overview-{uuid.uuid4().hex[:8]}", enabled=True, is_primary=False, ) db_session.add(integration) await db_session.flush() group_a = await _add_group( db_session, owner_person_id=owner.id, email="analytics-readers@example.com", ) group_b = await _add_group( db_session, owner_person_id=owner.id, email="finance-readers@example.com", ) resource_1 = await _add_resource( db_session, owner_person_id=owner.id, dataset_id="sales", integration=integration, ) resource_2 = await _add_resource( db_session, owner_person_id=owner.id, dataset_id="marketing", integration=integration, ) resource_3 = await _add_resource( db_session, owner_person_id=owner.id, dataset_id="ledger", integration=integration, ) resource_4 = await _add_resource( db_session, owner_person_id=owner.id, dataset_id="archived", integration=integration, ) # group_a → 2 active datasets (applied) await _add_binding_effect( db_session, resource=resource_1, group=group_a, entitlement=entitlement, status=EFFECT_STATUS_APPLIED, reconcile_status=RECONCILE_STATUS_APPLIED, last_provider_outcome="applied", last_error=None, ) await _add_binding_effect( db_session, resource=resource_2, group=group_a, entitlement=entitlement, status=EFFECT_STATUS_APPLIED, reconcile_status=RECONCILE_STATUS_APPLIED, last_provider_outcome="applied", last_error=None, ) # group_b → 1 active dataset (failed effect) + 1 revoked-history binding await _add_binding_effect( db_session, resource=resource_3, group=group_b, entitlement=entitlement, status=EFFECT_STATUS_FAILED, reconcile_status=RECONCILE_STATUS_FAILED, ) await _add_binding_effect( db_session, resource=resource_4, group=group_b, entitlement=entitlement, status=EFFECT_STATUS_APPLIED, desired_state=EFFECT_DESIRED_ABSENT, binding_status=BINDING_STATUS_REVOKED, reconcile_status=RECONCILE_STATUS_APPLIED, managed_reason=EFFECT_MANAGED_REASON_REVOKE_REQUEST, last_provider_outcome="applied", last_error=None, ) response = await client.get( "/api/v1/data-access/admin/overview", headers=admin_headers, ) assert response.status_code == 200, response.text data = response.json()["data"] assert data["datasets"]["total"] == 4 assert data["datasets"]["active"] == 4 assert data["datasets"]["requestable"] == 4 assert data["bindings"]["active"] == 3 assert data["bindings"]["revoked_history"] == 1 assert data["bindings"]["groups_with_access"] == 2 assert data["bindings"]["datasets_with_access"] == 3 # both active groups seed direct_members_count=12 → reach 24 assert data["people"]["reach"] == 24 assert data["people"]["groups_with_unknown_size"] == 0 assert data["effects"]["failed"] == 1 assert data["effects"]["applied"] == 3 assert data["apply_mode"]["integration_count"] == 1 assert data["apply_mode"]["apply_enabled_count"] == 0 assert data["apply_mode"]["provider_mutation_enabled"] is False top_groups = {row["email"]: row["active_dataset_count"] for row in data["top_groups"]} assert top_groups["analytics-readers@example.com"] == 2 assert top_groups["finance-readers@example.com"] == 1 top_datasets = {row["dataset_id"] for row in data["top_datasets"]} assert top_datasets == {"sales", "marketing", "ledger"} employee_response = await client.get( "/api/v1/data-access/admin/overview", headers=employee_headers, ) assert employee_response.status_code == 403 assert employee_response.json()["errors"][0]["code"] == ( "DATA_ACCESS_ADMIN_REQUIRED" ) async def test_failed_apply_retry_runs_reconciler_without_reapproval( client, db_session, employee_user, admin_headers, monkeypatch: pytest.MonkeyPatch, ) -> None: requester, _ = employee_user entitlement = await _add_entitlement(db_session) resource = await _add_resource( db_session, owner_person_id=requester.id, dataset_id="finance_retry", ) integration = await db_session.get(Integration, resource.integration_id) assert integration is not None integration.config_public = { "apply_mode_enabled": True, } group = await _add_group( db_session, owner_person_id=requester.id, email="retry-readers@example.com", ) approval = ApprovalRequest( external_id=f"approval-dax-retry-{uuid.uuid4()}", approval_type=c.APPROVAL_FLOW_DATA_ACCESS_GROUP_GRANT, applicant_id=requester.id, status="approved", source_payload={"source": "test"}, ) db_session.add(approval) await db_session.flush() request_row = DataAccessRequest( requester_person_id=requester.id, principal_email_group_id=group.id, principal_google_group_id_snapshot=group.google_group_id, principal_email_snapshot=group.email, entitlement_id=entitlement.id, entitlement_snapshot=c.ENTITLEMENT_READ, entitlement_version_snapshot=ENTITLEMENT_VERSION_V1, approval_flow_key=c.APPROVAL_FLOW_DATA_ACCESS_GROUP_GRANT, approval_request_id=approval.id, business_reason="Retry failed provider apply", status=REQUEST_STATUS_APPROVED, reconcile_status=RECONCILE_STATUS_FAILED, ) db_session.add(request_row) await db_session.flush() item = DataAccessRequestItem( request_id=request_row.id, resource_id=resource.id, line_no=1, resource_project_id_snapshot=resource.project_id, resource_dataset_id_snapshot=resource.dataset_id, resource_source_version_snapshot=resource.source_version, resource_source_etag_snapshot=resource.source_etag, resource_owner_person_id_snapshot=requester.id, status=REQUEST_ITEM_STATUS_APPROVED, ) db_session.add(item) await db_session.flush() binding, effect = await _add_binding_effect( db_session, resource=resource, group=group, entitlement=entitlement, status=EFFECT_STATUS_FAILED, reconcile_status=RECONCILE_STATUS_FAILED, source_request=request_row, source_item=item, last_provider_outcome="auth_error", last_error="403 permission denied", ) driver = _FakeApplyDriver( policies={ ("helix-data-prod", "finance_retry"): DataAccessDatasetIamPolicy( project_id="helix-data-prod", dataset_id="finance_retry", etag="provider-etag-retry-1", bindings=(), raw_policy={"etag": "provider-etag-retry-1"}, ), }, apply_results=[ DataAccessDatasetIamUpdateResult( ok=True, changed=True, etag="provider-etag-retry-2", ), ], ) async def fake_build_driver_by_id(integration_id, session=None): del integration_id, session return driver import app.domains.data_access.reconciler as reconciler monkeypatch.setattr(reconciler, "build_driver_by_id", fake_build_driver_by_id) approval_count_before = await db_session.scalar( select(func.count()).select_from(ApprovalRequest), ) response = await client.post( f"/api/v1/data-access/admin/failed-apply/{effect.id}/retry", headers=admin_headers, ) assert response.status_code == 200, response.text data = response.json()["data"] assert data["effect_id"] == str(effect.id) assert data["previous_status"] == EFFECT_STATUS_FAILED assert data["status"] == EFFECT_STATUS_APPLIED assert data["apply_mode"] == "apply_present" assert data["provider_write_count"] == 1 assert data["result_count"] == 1 assert data["blocked_effect_count"] == 0 assert data["config_error"] is None assert data["item"]["latest_run"]["mode"] == "apply_present" assert data["item"]["latest_run"]["provider_outcome"] == ( APPLY_PRESENT_OUTCOME_APPLIED ) assert driver.get_calls == [("helix-data-prod", "finance_retry")] assert len(driver.set_calls) == 1 _, _, expected_etag, patch = driver.set_calls[0] assert expected_etag == "provider-etag-retry-1" assert patch == DataAccessDatasetIamPolicyPatch( add_bindings=( DataAccessDatasetIamBinding( role=c.BIGQUERY_DATA_VIEWER_ROLE, members=("group:retry-readers@example.com",), ), ), remove_bindings=(), ) await db_session.refresh(effect) await db_session.refresh(binding) await db_session.refresh(request_row) assert effect.status == EFFECT_STATUS_APPLIED assert effect.last_provider_outcome == APPLY_PRESENT_OUTCOME_APPLIED assert effect.last_error is None assert binding.reconcile_status == RECONCILE_STATUS_APPLIED assert request_row.reconcile_status == RECONCILE_STATUS_APPLIED approval_count_after = await db_session.scalar( select(func.count()).select_from(ApprovalRequest), ) assert approval_count_after == approval_count_before audit = await db_session.scalar( select(AuditLog) .where(AuditLog.action == "data_access.failed_apply.retry") .order_by(AuditLog.created_at.desc()), ) assert audit is not None assert audit.actor_user_id is not None assert audit.detail["approval_request_created"] is False assert audit.detail["provider_mutation_owner"] == "data_access_reconciler" async def test_failed_apply_retry_rejects_stale_present_effect_after_revoke( client, db_session, employee_user, admin_headers, monkeypatch: pytest.MonkeyPatch, ) -> None: requester, _ = employee_user entitlement = await _add_entitlement(db_session) resource = await _add_resource( db_session, owner_person_id=requester.id, dataset_id="finance_stale_present_after_revoke", ) group = await _add_group( db_session, owner_person_id=requester.id, email="stale-present-readers@example.com", ) _, effect = await _add_binding_effect( db_session, resource=resource, group=group, entitlement=entitlement, status=EFFECT_STATUS_FAILED, desired_state=EFFECT_DESIRED_PRESENT, binding_status=BINDING_STATUS_REVOKED_PENDING, reconcile_status=RECONCILE_STATUS_FAILED, last_provider_outcome="auth_error", last_error="present apply failed before revoke", ) async def fail_if_driver_resolved(integration_id, session=None): del integration_id, session raise AssertionError("stale present retry must not reach provider driver") import app.domains.data_access.reconciler as reconciler monkeypatch.setattr(reconciler, "build_driver_by_id", fail_if_driver_resolved) queue_response = await client.get( "/api/v1/data-access/admin/failed-apply", headers=admin_headers, ) assert queue_response.status_code == 200, queue_response.text assert "finance_stale_present_after_revoke" not in queue_response.text retry_response = await client.post( f"/api/v1/data-access/admin/failed-apply/{effect.id}/retry", headers=admin_headers, ) assert retry_response.status_code == 404 assert retry_response.json()["errors"][0]["code"] == ( "DATA_ACCESS_FAILED_APPLY_NOT_FOUND" ) async def test_reconcile_run_history_lists_admin_detail_for_v1_runs( client, db_session, employee_user, employee_headers, admin_headers, ) -> None: owner, _ = employee_user entitlement = await _add_entitlement(db_session) group = await _add_group( db_session, owner_person_id=owner.id, email="history-readers@example.com", ) resource = await _add_resource( db_session, owner_person_id=owner.id, dataset_id="finance_history", ) resource.display_name = "Finance History" await db_session.flush() binding, effect = await _add_binding_effect( db_session, resource=resource, group=group, entitlement=entitlement, status=EFFECT_STATUS_FAILED, reconcile_status=RECONCILE_STATUS_FAILED, last_provider_outcome="apply_present_error", last_error="403 permission denied", ) run = await _add_effect_run( db_session, effect=effect, binding=binding, resource=resource, mode="apply_present", effect_status=EFFECT_STATUS_FAILED, provider_outcome="apply_present_error", error_message="403 permission denied", trace_id="trace-dax-042-history", ) user_resource = await _add_resource( db_session, owner_person_id=owner.id, dataset_id="finance_user_member", ) user_binding, user_effect = await _add_binding_effect( db_session, resource=user_resource, group=group, entitlement=entitlement, status=EFFECT_STATUS_FAILED, reconcile_status=RECONCILE_STATUS_FAILED, ) user_effect.provider_member = "user:direct@example.com" await db_session.flush() await _add_effect_run( db_session, effect=user_effect, binding=user_binding, resource=user_resource, mode="apply_present", effect_status=EFFECT_STATUS_FAILED, provider_outcome="out_of_scope_user_member", error_message="must stay hidden", trace_id="trace-user-member-hidden", ) response = await client.get( "/api/v1/data-access/admin/reconcile-runs", headers=admin_headers, ) assert response.status_code == 200, response.text data = response.json()["data"] assert data["summary"] == { "total": 1, "failed": 1, "retryable": 0, "applied": 0, "desired_present": 1, "desired_absent": 0, } assert data["total"] == 1 assert len(data["items"]) == 1 item = data["items"][0] assert item["id"] == str(run.id) assert item["run_id"] == str(run.run_id) assert item["effect_id"] == str(effect.id) assert item["binding_id"] == str(binding.id) assert item["dataset_id"] == "finance_history" assert item["display_name"] == "Finance History" assert item["principal_email"] == "history-readers@example.com" assert item["provider_member"] == "group:history-readers@example.com" assert item["provider_role"] == c.BIGQUERY_DATA_VIEWER_ROLE assert item["mode"] == "apply_present" assert item["effect_status"] == EFFECT_STATUS_FAILED assert item["provider_outcome"] == "apply_present_error" assert item["error_message"] == "403 permission denied" assert item["trace_id"] == "trace-dax-042-history" assert item["current_effect_status"] == EFFECT_STATUS_FAILED assert item["binding_status"] == BINDING_STATUS_ACTIVE assert item["binding_reconcile_status"] == RECONCILE_STATUS_FAILED assert item["current_provider_outcome"] == "apply_present_error" assert item["current_error"] == "403 permission denied" assert "finance_user_member" not in response.text assert "user:direct@example.com" not in response.text search_response = await client.get( "/api/v1/data-access/admin/reconcile-runs", params={"search": "trace-dax-042-history"}, headers=admin_headers, ) assert search_response.status_code == 200, search_response.text assert search_response.json()["data"]["total"] == 1 employee_response = await client.get( "/api/v1/data-access/admin/reconcile-runs", headers=employee_headers, ) assert employee_response.status_code == 403 assert employee_response.json()["errors"][0]["code"] == ( "DATA_ACCESS_ADMIN_REQUIRED" ) assert "finance_history" not in employee_response.text async def test_cutover_validation_report_lists_required_gate_categories( client, db_session, employee_user, employee_headers, admin_headers, ) -> None: owner, _ = employee_user integration = Integration( kind=c.DOMAIN_DATA_ACCESS, provider=c.PROVIDER_BIGQUERY, name="Data Access BigQuery Cutover", slug=f"dax-cutover-{uuid.uuid4().hex[:8]}", enabled=True, is_primary=False, ) db_session.add(integration) await db_session.flush() catalog_active = await _add_resource( db_session, integration=integration, owner_person_id=owner.id, dataset_id="catalog_active", catalog_resource_id="catalog-active", owner_source=OWNER_SOURCE_CATALOG, source_version="catalog-2026-06-11", ) catalog_transferred = await _add_resource( db_session, integration=integration, owner_person_id=owner.id, dataset_id="catalog_transferred", catalog_resource_id="catalog-transferred", owner_source=OWNER_SOURCE_CATALOG_TRANSFERRED, source_version="catalog-2026-08-25", ) await _add_resource( db_session, integration=integration, owner_person_id=owner.id, dataset_id="catalog_suspended", catalog_resource_id="catalog-suspended", status=RESOURCE_STATUS_SUSPENDED, owner_source=OWNER_SOURCE_CATALOG, source_version="catalog-2026-06-11", ) await _add_resource( db_session, integration=integration, owner_person_id=owner.id, dataset_id="bootstrap_requestable", ) await _add_resource( db_session, integration=integration, owner_person_id=owner.id, dataset_id="bootstrap_transferred", owner_source=OWNER_SOURCE_BOOTSTRAP_TRANSFERRED, ) historical_bootstrap = await _add_resource( db_session, integration=integration, owner_person_id=owner.id, dataset_id="bootstrap_history", status=RESOURCE_STATUS_SUSPENDED, requestable=False, ) await _add_resource( db_session, integration=integration, owner_person_id=None, dataset_id="owner_mismatch", catalog_resource_id="catalog-owner-mismatch", status=RESOURCE_STATUS_SUSPENDED, requestable=False, owner_resolution_status=OWNER_RESOLUTION_UNRESOLVED, ) locator_drift = await _add_resource( db_session, integration=integration, owner_person_id=owner.id, dataset_id="locator_drift_old", catalog_resource_id="catalog-locator-drift", status=RESOURCE_STATUS_SUSPENDED, requestable=False, ) await _add_resource( db_session, integration=integration, owner_person_id=owner.id, dataset_id="duplicate_locator", catalog_resource_id="catalog-duplicate-locator-a", status=RESOURCE_STATUS_SUSPENDED, requestable=False, owner_source=OWNER_SOURCE_CATALOG, source_version="catalog-2026-06-11", ) await _add_resource( db_session, integration=integration, owner_person_id=owner.id, dataset_id="duplicate_locator", catalog_resource_id="catalog-duplicate-locator-b", status=RESOURCE_STATUS_SUSPENDED, requestable=False, owner_source=OWNER_SOURCE_CATALOG, source_version="catalog-2026-06-11", ) await _add_resource( db_session, integration=integration, owner_person_id=owner.id, dataset_id="duplicate_catalog_a", catalog_resource_id="catalog-duplicate-id", status=RESOURCE_STATUS_SUSPENDED, requestable=False, owner_source=OWNER_SOURCE_CATALOG, source_version="catalog-2026-06-11", ) await _add_resource( db_session, integration=integration, owner_person_id=owner.id, dataset_id="duplicate_catalog_b", catalog_resource_id="catalog-duplicate-id", status=RESOURCE_STATUS_SUSPENDED, requestable=False, owner_source=OWNER_SOURCE_CATALOG, source_version="catalog-2026-06-11", ) await _add_resource( db_session, integration=integration, owner_person_id=owner.id, dataset_id="invalid_status", status="archived", owner_source=OWNER_SOURCE_CATALOG, source_version="catalog-2026-06-11", ) invalid_provider = await _add_resource( db_session, integration=integration, owner_person_id=owner.id, dataset_id="invalid_provider", status=RESOURCE_STATUS_SUSPENDED, requestable=False, owner_source=OWNER_SOURCE_CATALOG, source_version="catalog-2026-06-11", ) invalid_provider.provider = "gcp" await db_session.flush() await _add_resource( db_session, integration=integration, owner_person_id=None, dataset_id="fallback_owner", status=RESOURCE_STATUS_SUSPENDED, requestable=False, owner_resolution_status=OWNER_RESOLUTION_FALLBACK, owner_source=OWNER_SOURCE_CATALOG, source_version="catalog-2026-06-11", ) entitlement = await _add_entitlement(db_session) group = await _add_group( db_session, owner_person_id=owner.id, email="cutover-report-readers@example.com", ) await _add_binding_effect( db_session, resource=locator_drift, group=group, entitlement=entitlement, status=EFFECT_STATUS_APPLIED, reconcile_status=RECONCILE_STATUS_APPLIED, last_provider_outcome="applied", last_error=None, ) binding, _ = await _add_binding_effect( db_session, resource=historical_bootstrap, group=group, entitlement=entitlement, status=EFFECT_STATUS_APPLIED, reconcile_status=RECONCILE_STATUS_APPLIED, last_provider_outcome="applied", last_error=None, ) response = await client.get( "/api/v1/data-access/admin/cutover-validation", params={"integration_id": str(integration.id), "page_size": 50}, headers=admin_headers, ) assert response.status_code == 200, response.text data = response.json()["data"] summary = data["summary"] counts = summary["category_counts"] assert summary["cutover_ready"] is False assert summary["provider_mutation_attempted"] is False assert summary["request_handlers_mutate_provider"] is False assert summary["feishu_callbacks_mutate_provider"] is False assert counts["bootstrap_remaining"] >= 4 assert counts["bootstrap_requestable"] == 2 assert counts["bootstrap_historical"] >= 2 assert counts["catalog_fed_active"] == 2 assert counts["suspended"] >= 8 assert counts["duplicate_locator"] == 2 assert counts["duplicate_catalog_id"] == 2 assert counts["locator_drift_with_history"] == 1 assert counts["owner_mismatch"] == 1 assert counts["invalid"] == 2 assert counts["unresolved_owner_fallback"] == 2 assert counts["unsafe_unmatched_bootstrap_revoke"] == 0 assert set(summary["blocking_category_counts"]) >= { "bootstrap_requestable", "duplicate_locator", "duplicate_catalog_id", "locator_drift_with_history", "owner_mismatch", "unresolved_owner_fallback", "invalid", } by_dataset = {item["dataset_id"]: item for item in data["items"]} assert by_dataset["catalog_active"]["resource_id"] == str(catalog_active.id) assert "catalog_fed_active" in by_dataset["catalog_active"]["findings"] assert by_dataset["catalog_transferred"]["resource_id"] == str( catalog_transferred.id, ) assert "catalog_fed_active" in by_dataset[ "catalog_transferred" ]["findings"] assert "bootstrap_requestable" in by_dataset[ "bootstrap_requestable" ]["blocking_findings"] assert "owner_mismatch" in by_dataset["owner_mismatch"]["blocking_findings"] assert "locator_drift_with_history" in by_dataset[ "locator_drift_old" ]["blocking_findings"] assert by_dataset["invalid_status"]["invalid_reasons"] == ["invalid_status"] assert by_dataset["invalid_provider"]["invalid_reasons"] == [ "non_bigquery_provider", ] assert by_dataset["bootstrap_history"]["history"]["active_binding_count"] == 1 await db_session.refresh(binding) assert binding.status == BINDING_STATUS_ACTIVE assert binding.revoke_reason is None assert await db_session.scalar( select(func.count()) .select_from(DataAccessCompiledEffect) .where( DataAccessCompiledEffect.resource_id == historical_bootstrap.id, DataAccessCompiledEffect.desired_state == EFFECT_DESIRED_ABSENT, ), ) == 0 employee_response = await client.get( "/api/v1/data-access/admin/cutover-validation", params={"integration_id": str(integration.id)}, headers=employee_headers, ) assert employee_response.status_code == 403 assert employee_response.json()["errors"][0]["code"] == ( "DATA_ACCESS_ADMIN_REQUIRED" ) assert "catalog_active" not in employee_response.text async def test_cutover_validation_report_allows_historical_bootstrap_provenance( client, db_session, employee_user, admin_headers, ) -> None: owner, _ = employee_user integration = Integration( kind=c.DOMAIN_DATA_ACCESS, provider=c.PROVIDER_BIGQUERY, name="Data Access BigQuery Cutover Ready", slug=f"dax-cutover-ready-{uuid.uuid4().hex[:8]}", enabled=True, is_primary=False, ) db_session.add(integration) await db_session.flush() await _add_resource( db_session, integration=integration, owner_person_id=owner.id, dataset_id="catalog_ready", catalog_resource_id="catalog-ready", owner_source=OWNER_SOURCE_CATALOG, source_version="catalog-2026-06-11", ) historical = await _add_resource( db_session, integration=integration, owner_person_id=owner.id, dataset_id="bootstrap_provenance", status=RESOURCE_STATUS_SUSPENDED, requestable=False, ) response = await client.get( "/api/v1/data-access/admin/cutover-validation", params={"integration_id": str(integration.id)}, headers=admin_headers, ) assert response.status_code == 200, response.text data = response.json()["data"] assert data["summary"]["cutover_ready"] is True assert data["summary"]["blocking_count"] == 0 assert data["summary"]["category_counts"]["bootstrap_remaining"] == 1 assert data["summary"]["category_counts"]["bootstrap_historical"] == 1 assert data["summary"]["category_counts"]["bootstrap_requestable"] == 0 assert data["summary"]["category_counts"]["catalog_fed_active"] == 1 item = next( row for row in data["items"] if row["resource_id"] == str(historical.id) ) assert item["findings"] == [ "bootstrap_remaining", "bootstrap_historical", "unmatched_bootstrap", "suspended", ] assert item["blocking_findings"] == []