from __future__ import annotations import uuid from datetime import UTC, datetime from sqlalchemy import select from app.domains.data_access import constants as c from app.domains.data_access.models import ( BINDING_STATUS_ACTIVE, BINDING_STATUS_REVOKED, EFFECT_DESIRED_ABSENT, EFFECT_DESIRED_PRESENT, EFFECT_MANAGED_REASON_GRANT_REQUEST, EFFECT_MANAGED_REASON_GROUP_RENAME_BRIDGE, EFFECT_STATUS_OBSERVED, EFFECT_STATUS_PENDING_RECONCILE, ENTITLEMENT_STATUS_ACTIVE, ENTITLEMENT_VERSION_V1, OWNER_RESOLUTION_RESOLVED, OWNER_SOURCE_BOOTSTRAP, RESOURCE_STATUS_ACTIVE, DataAccessBinding, DataAccessCompiledEffect, DataAccessEffectRun, DataAccessEntitlement, 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, ) -> DataAccessResource: return DataAccessResource( integration_id=integration_id, provider=c.PROVIDER_BIGQUERY, resource_type=c.RESOURCE_TYPE_DATASET, project_id="helix-data-prod", dataset_id="finance", display_name="helix-data-prod.finance", 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="etag-finance", ) def _binding( *, resource: DataAccessResource, group: EmailGroup, entitlement: DataAccessEntitlement, email_snapshot: str | None = None, status: str = BINDING_STATUS_ACTIVE, ) -> DataAccessBinding: return DataAccessBinding( resource_id=resource.id, principal_email_group_id=group.id, principal_google_group_id_snapshot=group.google_group_id, principal_email_snapshot=email_snapshot or group.email, entitlement_id=entitlement.id, entitlement_snapshot=c.ENTITLEMENT_READ, entitlement_version_snapshot=ENTITLEMENT_VERSION_V1, status=status, ) def _effect( *, binding: DataAccessBinding, resource: DataAccessResource, group: EmailGroup, entitlement: DataAccessEntitlement, desired_state: str = EFFECT_DESIRED_PRESENT, managed_reason: str = EFFECT_MANAGED_REASON_GRANT_REQUEST, ) -> DataAccessCompiledEffect: return 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}{binding.principal_email_snapshot}" ), 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="dax-compiler-v1", managed_reason=managed_reason, already_present_at_first_apply=True, status=EFFECT_STATUS_PENDING_RECONCILE, ) async def test_compiled_effect_persists_provider_member_and_provenance( db_session, employee_user, ) -> None: owner, _ = 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) resource = _resource(integration_id=integration_id, owner_person_id=owner.id) db_session.add_all([group, entitlement, resource]) await db_session.flush() binding = _binding(resource=resource, group=group, entitlement=entitlement) db_session.add(binding) await db_session.flush() effect = _effect( binding=binding, resource=resource, group=group, entitlement=entitlement, ) db_session.add(effect) await db_session.flush() row = await db_session.scalar( select(DataAccessCompiledEffect).where( DataAccessCompiledEffect.id == effect.id, ), ) assert row is not None assert row.principal_email_group_id == group.id assert row.provider_member == "group:analytics-readers@example.com" assert row.provider_member != "group:alias-readers@example.com" assert row.provider_role == c.BIGQUERY_DATA_VIEWER_ROLE assert row.target_project_id == "helix-data-prod" assert row.target_dataset_id == "finance" assert row.entitlement_version_snapshot == ENTITLEMENT_VERSION_V1 assert row.resource_source_version_snapshot == "bootstrap-2026-06-10" assert row.resource_source_etag_snapshot == "etag-finance" assert row.compiler_version == "dax-compiler-v1" assert row.managed_reason == EFFECT_MANAGED_REASON_GRANT_REQUEST assert row.already_present_at_first_apply is True assert row.status == EFFECT_STATUS_PENDING_RECONCILE async def test_effect_run_history_persists_provider_outcome( db_session, employee_user, ) -> None: owner, _ = employee_user group = EmailGroup( email="analytics-history@example.com", name="Analytics History", status="active", google_group_id="google-group-history", ) entitlement = _entitlement() integration_id = await _integration_id(db_session) resource = _resource(integration_id=integration_id, owner_person_id=owner.id) db_session.add_all([group, entitlement, resource]) await db_session.flush() binding = _binding(resource=resource, group=group, entitlement=entitlement) db_session.add(binding) await db_session.flush() effect = _effect( binding=binding, resource=resource, group=group, entitlement=entitlement, ) effect.status = EFFECT_STATUS_OBSERVED effect.last_provider_outcome = "observe_would_add" effect.last_provider_etag = "provider-etag-history" db_session.add(effect) await db_session.flush() started_at = datetime.now(UTC) run = DataAccessEffectRun( run_id=uuid.uuid4(), effect_id=effect.id, binding_id=binding.id, request_id=None, resource_id=resource.id, integration_id=integration_id, trace_id="dax-effect-history-test", mode="observe", 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, provider_outcome=effect.last_provider_outcome, provider_etag=effect.last_provider_etag, error_message=None, started_at=started_at, completed_at=datetime.now(UTC), ) db_session.add(run) await db_session.flush() stored = await db_session.get(DataAccessEffectRun, run.id) assert stored is not None assert stored.effect_id == effect.id assert stored.binding_id == binding.id assert stored.resource_id == resource.id assert stored.integration_id == integration_id assert stored.provider_outcome == "observe_would_add" assert stored.effect_status == EFFECT_STATUS_OBSERVED assert stored.provider_etag == "provider-etag-history" async def test_absent_effect_can_preserve_old_email_for_rename_bridge( db_session, employee_user, ) -> None: owner, _ = employee_user group = EmailGroup( email="analytics-new@example.com", name="Analytics Readers", status="active", google_group_id="google-group-analytics", ) entitlement = _entitlement() integration_id = await _integration_id(db_session) resource = _resource(integration_id=integration_id, owner_person_id=owner.id) db_session.add_all([group, entitlement, resource]) await db_session.flush() old_binding = _binding( resource=resource, group=group, entitlement=entitlement, email_snapshot="analytics-old@example.com", status=BINDING_STATUS_REVOKED, ) db_session.add(old_binding) await db_session.flush() absent = _effect( binding=old_binding, resource=resource, group=group, entitlement=entitlement, desired_state=EFFECT_DESIRED_ABSENT, managed_reason=EFFECT_MANAGED_REASON_GROUP_RENAME_BRIDGE, ) db_session.add(absent) await db_session.flush() assert absent.principal_email_group_id == group.id assert absent.provider_member == "group:analytics-old@example.com" assert absent.desired_state == EFFECT_DESIRED_ABSENT assert absent.managed_reason == EFFECT_MANAGED_REASON_GROUP_RENAME_BRIDGE def test_effect_table_has_no_alias_or_provider_mutation_path() -> None: columns = set(DataAccessCompiledEffect.__table__.columns.keys()) run_columns = set(DataAccessEffectRun.__table__.columns.keys()) assert "principal_email_group_id" in columns assert "provider_member" in columns assert "provider_role" in columns assert ( DataAccessCompiledEffect.__table__.c.target_project_id.type.collation == "utf8mb4_bin" ) assert ( DataAccessCompiledEffect.__table__.c.target_dataset_id.type.collation == "utf8mb4_bin" ) assert ( DataAccessEffectRun.__table__.c.target_project_id.type.collation == "utf8mb4_bin" ) assert ( DataAccessEffectRun.__table__.c.target_dataset_id.type.collation == "utf8mb4_bin" ) assert not any("alias" in column for column in columns) assert not any("alias" in column for column in run_columns) assert columns.isdisjoint({ "project_iam_role", "bigquery_job_user", "policy_patch", "provider_policy", "raw_provider_secret", })