from __future__ import annotations import uuid import pytest from sqlalchemy import select from sqlalchemy.exc import IntegrityError from app.domains.data_access import constants as c from app.domains.data_access.models import ( ENTITLEMENT_STATUS_ACTIVE, ENTITLEMENT_VERSION_V1, OWNER_RESOLUTION_RESOLVED, OWNER_SOURCE_BOOTSTRAP, REQUEST_ITEM_STATUS_PENDING, REQUEST_STATUS_PENDING_FEISHU, RESOURCE_STATUS_ACTIVE, DataAccessEntitlement, DataAccessRequest, DataAccessRequestItem, DataAccessResource, ) from app.domains.email_group.models import EmailGroup from app.integrations.models import Integration async def _integration_id(db_session) -> str: row = Integration( kind=c.DOMAIN_DATA_ACCESS, provider=c.PROVIDER_BIGQUERY, name="Data Access BigQuery", slug=f"dax-bq-{uuid.uuid4().hex[:8]}", enabled=True, is_primary=False, ) db_session.add(row) await db_session.flush() return str(row.id) def _entitlement() -> DataAccessEntitlement: return 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, ) def _resource( *, integration_id: str, owner_person_id: uuid.UUID, dataset_id: str, ) -> DataAccessResource: return DataAccessResource( integration_id=integration_id, provider=c.PROVIDER_BIGQUERY, resource_type=c.RESOURCE_TYPE_DATASET, project_id="helix-data-prod", dataset_id=dataset_id, display_name=f"helix-data-prod.{dataset_id}", requestable=True, status=RESOURCE_STATUS_ACTIVE, owner_person_id=str(owner_person_id), owner_resolution_status=OWNER_RESOLUTION_RESOLVED, owner_source=OWNER_SOURCE_BOOTSTRAP, source_version="bootstrap-2026-06-10", source_etag=f"etag-{dataset_id}", ) async def test_request_stores_semantic_group_and_many_dataset_items( db_session, employee_user, ) -> None: requester, _ = employee_user group = EmailGroup( email="analytics-readers@example.com", name="Analytics Readers", status="active", google_group_id="google-group-analytics", aliases=["alias-readers@example.com"], ) entitlement = _entitlement() integration_id = await _integration_id(db_session) finance = _resource( integration_id=integration_id, owner_person_id=requester.id, dataset_id="finance", ) people = _resource( integration_id=integration_id, owner_person_id=requester.id, dataset_id="people", ) db_session.add_all([group, entitlement, finance, people]) await db_session.flush() request = 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="dax-feishu-instance-001", idempotency_key="dax-request-idempotency-001", business_reason="Quarterly analytics read access", status=REQUEST_STATUS_PENDING_FEISHU, trace_id="trace-dax-008", items=[ DataAccessRequestItem( resource_id=finance.id, line_no=1, resource_project_id_snapshot=finance.project_id, resource_dataset_id_snapshot=finance.dataset_id, resource_source_version_snapshot=finance.source_version, resource_source_etag_snapshot=finance.source_etag, resource_owner_person_id_snapshot=requester.id, status=REQUEST_ITEM_STATUS_PENDING, ), DataAccessRequestItem( resource_id=people.id, line_no=2, resource_project_id_snapshot=people.project_id, resource_dataset_id_snapshot=people.dataset_id, resource_source_version_snapshot=people.source_version, resource_source_etag_snapshot=people.source_etag, resource_owner_person_id_snapshot=requester.id, status=REQUEST_ITEM_STATUS_PENDING, ), ], ) db_session.add(request) await db_session.flush() rows = ( await db_session.scalars( select(DataAccessRequestItem) .where(DataAccessRequestItem.request_id == request.id) .order_by(DataAccessRequestItem.line_no), ) ).all() assert request.principal_email_group_id == group.id assert request.principal_google_group_id_snapshot == "google-group-analytics" assert request.principal_email_snapshot == "analytics-readers@example.com" assert request.principal_email_snapshot != "alias-readers@example.com" assert request.feishu_instance_code == "dax-feishu-instance-001" assert request.approval_flow_key == c.APPROVAL_FLOW_DATA_ACCESS_GROUP_GRANT assert request.entitlement_snapshot == c.ENTITLEMENT_READ assert request.entitlement_version_snapshot == ENTITLEMENT_VERSION_V1 assert request.status == REQUEST_STATUS_PENDING_FEISHU assert [row.resource_dataset_id_snapshot for row in rows] == [ "finance", "people", ] assert {row.resource_id for row in rows} == {finance.id, people.id} async def test_duplicate_resource_inside_one_request_is_rejected( db_session, employee_user, ) -> None: requester, _ = employee_user group = EmailGroup( email="security-readers@example.com", name="Security Readers", status="active", google_group_id="google-group-security", ) entitlement = _entitlement() integration_id = await _integration_id(db_session) resource = _resource( integration_id=integration_id, owner_person_id=requester.id, dataset_id="security", ) db_session.add_all([group, entitlement, resource]) await db_session.flush() db_session.add( 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="dax-feishu-instance-duplicate", status=REQUEST_STATUS_PENDING_FEISHU, items=[ DataAccessRequestItem( 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, ), DataAccessRequestItem( resource_id=resource.id, line_no=2, 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, ), ], ), ) with pytest.raises(IntegrityError): await db_session.flush() def test_request_tables_have_no_alias_or_provider_role_override_path() -> None: request_columns = set(DataAccessRequest.__table__.columns.keys()) item_columns = set(DataAccessRequestItem.__table__.columns.keys()) all_columns = request_columns | item_columns assert "principal_email_group_id" in request_columns assert "principal_google_group_id_snapshot" in request_columns assert "principal_email_snapshot" in request_columns assert not any("alias" in column for column in all_columns) assert all_columns.isdisjoint({ "provider_member", "provider_member_snapshot", "provider_role", "provider_role_override", "resource_provider_role", "entitlement_override", })